Add log index to Vat move

db/migrations/1538396815_create_vat_move.up.sql

@@ -4,9 +4,10 @@ CREATE TABLE maker.vat_move (
   src           TEXT NOT NULL,
   dst           TEXT NOT NULL,
   rad           NUMERIC NOT NULL,
+  log_idx       INTEGER NOT NULL,
   tx_idx        INTEGER NOT NULL,
   raw_log       JSONB,
-  UNIQUE (header_id, tx_idx)
+  UNIQUE (header_id, tx_idx, log_idx)
 );
 
 ALTER TABLE public.checked_headers

db/schema.sql

@@ -904,6 +904,7 @@ CREATE TABLE maker.vat_move (
     src text NOT NULL,
     dst text NOT NULL,
     rad numeric NOT NULL,
+    log_idx integer NOT NULL,
     tx_idx integer NOT NULL,
     raw_log jsonb
 );
@@ -2095,11 +2096,11 @@ ALTER TABLE ONLY maker.vat_init
 
 
 --
--- Name: vat_move vat_move_header_id_tx_idx_key; Type: CONSTRAINT; Schema: maker; Owner: -
+-- Name: vat_move vat_move_header_id_tx_idx_log_idx_key; Type: CONSTRAINT; Schema: maker; Owner: -
 --
 
 ALTER TABLE ONLY maker.vat_move
-    ADD CONSTRAINT vat_move_header_id_tx_idx_key UNIQUE (header_id, tx_idx);
+    ADD CONSTRAINT vat_move_header_id_tx_idx_log_idx_key UNIQUE (header_id, tx_idx, log_idx);
 
 
 --

pkg/transformers/test_data/vat_move.go

@@ -38,7 +38,7 @@ var EthVatMoveLog = types.Log{
 	TxHash:      common.HexToHash("0xe8f39fbb7fea3621f543868f19b1114e305aff6a063a30d32835ff1012526f91"),
 	TxIndex:     7,
 	BlockHash:   common.HexToHash("0xe3dd2e05bd8b92833e20ed83e2171bbc06a9ec823232eca1730a807bd8f5edc0"),
-	Index:       0,
+	Index:       8,
 	Removed:     false,
 }
 
@@ -47,6 +47,7 @@ var VatMoveModel = vat_move.VatMoveModel{
 	Src:              "0xA730d1FF8B6Bc74a26d54c20a9dda539909BaB0e",
 	Dst:              "0xB730D1fF8b6BC74a26D54c20a9ddA539909BAb0e",
 	Rad:              "42",
+	LogIndex:         EthVatMoveLog.Index,
 	TransactionIndex: EthVatMoveLog.TxIndex,
 	Raw:              rawVatMoveLog,
 }

pkg/transformers/vat_move/converter.go

@@ -43,6 +43,7 @@ func (VatMoveConverter) ToModels(ethLogs []types.Log) ([]interface{}, error) {
 			Src:              src.String(),
 			Dst:              dst.String(),
 			Rad:              rad.String(),
+			LogIndex:         ethLog.Index,
 			TransactionIndex: ethLog.TxIndex,
 			Raw:              raw,
 		})

pkg/transformers/vat_move/converter_test.go

@@ -41,6 +41,7 @@ var _ = Describe("Vat move converter", func() {
 		models, err := converter.ToModels([]types.Log{test_data.EthVatMoveLog})
 
 		Expect(err).NotTo(HaveOccurred())
+		Expect(len(models)).To(Equal(1))
 		Expect(models[0]).To(Equal(test_data.VatMoveModel))
 	})
 })

pkg/transformers/vat_move/model.go

@@ -18,6 +18,7 @@ type VatMoveModel struct {
 	Src              string
 	Dst              string
 	Rad              string
+	LogIndex         uint   `db:"log_idx"`
 	TransactionIndex uint   `db:"tx_idx"`
 	Raw              []byte `db:"raw_log"`
 }

pkg/transformers/vat_move/repository.go

@@ -38,9 +38,9 @@ func (repository VatMoveRepository) Create(headerID int64, models []interface{})
 		}
 
 		_, err = tx.Exec(
-			`INSERT INTO maker.vat_move (header_id, src, dst, rad, tx_idx, raw_log)
+			`INSERT INTO maker.vat_move (header_id, src, dst, rad, log_idx, tx_idx, raw_log)
-        	VALUES ($1, $2, $3, $4::NUMERIC, $5, $6)`,
+				VALUES ($1, $2, $3, $4::NUMERIC, $5, $6, $7)`,
-			headerID, vatMove.Src, vatMove.Dst, vatMove.Rad, vatMove.TransactionIndex, vatMove.Raw,
+			headerID, vatMove.Src, vatMove.Dst, vatMove.Rad, vatMove.LogIndex, vatMove.TransactionIndex, vatMove.Raw,
 		)
 
 		if err != nil {

pkg/transformers/vat_move/repository_test.go

@@ -49,22 +49,40 @@ var _ = Describe("Vat Move", func() {
 		BeforeEach(func() {
 			headerID, err = headerRepository.CreateOrUpdateHeader(fakes.FakeHeader)
 			Expect(err).NotTo(HaveOccurred())
-			err = vatMoveRepository.Create(headerID, []interface{}{test_data.VatMoveModel})
-			Expect(err).NotTo(HaveOccurred())
 		})
 
 		It("adds a vat move event", func() {
+			err = vatMoveRepository.Create(headerID, []interface{}{test_data.VatMoveModel})
+
+			Expect(err).NotTo(HaveOccurred())
 			var dbVatMove vat_move.VatMoveModel
-			err = db.Get(&dbVatMove, `SELECT src, dst, rad, tx_idx, raw_log FROM maker.vat_move WHERE header_id = $1`, headerID)
+			err = db.Get(&dbVatMove, `SELECT src, dst, rad, log_idx, tx_idx, raw_log FROM maker.vat_move WHERE header_id = $1`, headerID)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(dbVatMove.Src).To(Equal(test_data.VatMoveModel.Src))
 			Expect(dbVatMove.Dst).To(Equal(test_data.VatMoveModel.Dst))
 			Expect(dbVatMove.Rad).To(Equal(test_data.VatMoveModel.Rad))
+			Expect(dbVatMove.LogIndex).To(Equal(test_data.VatMoveModel.LogIndex))
 			Expect(dbVatMove.TransactionIndex).To(Equal(test_data.VatMoveModel.TransactionIndex))
 			Expect(dbVatMove.Raw).To(MatchJSON(test_data.VatMoveModel.Raw))
 		})
 
 		It("marks header id as checked for vat move logs", func() {
+			err = vatMoveRepository.Create(headerID, []interface{}{test_data.VatMoveModel})
+
+			Expect(err).NotTo(HaveOccurred())
+			var headerChecked bool
+			err = db.Get(&headerChecked, `SELECT vat_move_checked FROM public.checked_headers WHERE header_id = $1`, headerID)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(headerChecked).To(BeTrue())
+		})
+
+		It("updates the header to checked if checked headers row already exists", func() {
+			_, err = db.Exec(`INSERT INTO public.checked_headers (header_id) VALUES ($1)`, headerID)
+			Expect(err).NotTo(HaveOccurred())
+
+			err = vatMoveRepository.Create(headerID, []interface{}{test_data.VatMoveModel})
+
+			Expect(err).NotTo(HaveOccurred())
 			var headerChecked bool
 			err = db.Get(&headerChecked, `SELECT vat_move_checked FROM public.checked_headers WHERE header_id = $1`, headerID)
 			Expect(err).NotTo(HaveOccurred())
@@ -73,16 +91,23 @@ var _ = Describe("Vat Move", func() {
 
 		It("returns an error if insertion fails", func() {
 			err = vatMoveRepository.Create(headerID, []interface{}{test_data.VatMoveModel})
+			Expect(err).NotTo(HaveOccurred())
+
+			err = vatMoveRepository.Create(headerID, []interface{}{test_data.VatMoveModel})
+
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("pq: duplicate key value violates unique constraint"))
 		})
 
 		It("removes vat move event if corresponding header is deleted", func() {
-			_, err = db.Exec(`DELETE FROM headers WHERE id = $1`, headerID)
+			err = vatMoveRepository.Create(headerID, []interface{}{test_data.VatMoveModel})
 			Expect(err).NotTo(HaveOccurred())
 
+			_, err = db.Exec(`DELETE FROM headers WHERE id = $1`, headerID)
+
+			Expect(err).NotTo(HaveOccurred())
 			var dbVatMove vat_move.VatMoveModel
-			err = db.Get(&dbVatMove, `SELECT src, dst, rad, tx_idx, raw_log FROM maker.vat_move WHERE header_id = $1`, headerID)
+			err = db.Get(&dbVatMove, `SELECT src, dst, rad, log_idx, tx_idx, raw_log FROM maker.vat_move WHERE header_id = $1`, headerID)
 			Expect(err).To(HaveOccurred())
 			Expect(err).To(MatchError(sql.ErrNoRows))
 		})