wip

pipeline/pipeline/publisher.go

@@ -49,7 +49,8 @@ func (p *Publisher) Publish(ctx context.Context, e *watcher.Event) {
 		TxHash:           e.TxHash,
 		Version:          e.Version,
 		Revision:         e.Revision,
-		Hash:             e.Hash,
+		Digest:           e.Digest,
+		Overwrite:        e.DuplicatedFixed,
 	}
 
 	// In some scenarios the fly component that inserts the VAA documents does not have the txhash field available,

pipeline/topic/topic.go

@@ -19,7 +19,8 @@ type Event struct {
 	TxHash           string     `json:"txHash"`
 	Version          uint16     `json:"version"`
 	Revision         uint16     `json:"revision"`
-	Hash             []byte     `json:"hash"`
+	Digest           string     `json:"digest"`
+	Overwrite        bool       `json:"overwrite"`
 }
 
 // PushFunc is a function to push VAAEvent.

pipeline/watcher/watcher.go

@@ -50,8 +50,9 @@ type Event struct {
 	TxHash           string     `bson:"txHash"`
 	Version          uint16     `bson:"version"`
 	Revision         uint16     `bson:"revision"`
-	Hash             []byte     `bson:"hash"`
+	Digest           string     `bson:"digest"`
 	IsDuplicated     bool       `bson:"isDuplicated"`
+	DuplicatedFixed  bool       `bson:"duplicatedFixed"`
 }
 
 const queryTemplate = `

tx-tracker/consumer/consumer.go

@@ -114,7 +114,7 @@ func (c *Consumer) processSourceTx(ctx context.Context, msg queue.ConsumerMessag
 		Vaa:         event.Vaa,
 		IsVaaSigned: event.IsVaaSigned,
 		Metrics:     c.metrics,
-		Overwrite:   false, // avoid processing the same transaction twice
+		Overwrite:   event.Overwrite, // avoid processing the same transaction twice
 		Source:      event.Source,
 	}
 	_, err := ProcessSourceTx(ctx, c.logger, c.rpcpool, c.wormchainRpcPool, c.repository, &p, c.p2pNetwork)

tx-tracker/queue/converter.go

@@ -25,6 +25,7 @@ type VaaEvent struct {
 	TxHash           string      `json:"txHash"`
 	Version          uint16      `json:"version"`
 	Revision         uint16      `json:"revision"`
+	Overwrite        bool        `json:"overwrite"`
 }
 
 // VaaConverter converts a message from a VAAEvent.
@@ -49,6 +50,7 @@ func NewVaaConverter(log *zap.Logger) ConverterFunc {
 			Vaa:            vaaEvent.Vaa,
 			IsVaaSigned:    true,
 			TxHash:         vaaEvent.TxHash,
+			Overwrite:      vaaEvent.Overwrite,
 		}, nil
 	}
 }

tx-tracker/queue/types.go

@@ -51,6 +51,7 @@ type Event struct {
 	Vaa            []byte
 	IsVaaSigned    bool
 	Attributes     any
+	Overwrite      bool
 }
 
 func GetAttributes[T EventAttributes](e *Event) (T, bool) {