Add log index to various events (#78)

* Add log_index field to flip kick

* Add log index to flop kick

* Add repo tests for Cat.file

* Add log_index to vat_flux

* Add log_index to vat_slip

* Add log_index to vat_toll

* Add log_index to vat_tune

* Add test to repos for updating checked headers

db/migrations/1532468319_create_flip_kick_table.up.sql

@@ -4,11 +4,12 @@ CREATE TABLE maker.flip_kick (
   bid_id        NUMERIC NOT NULL,
   lot           NUMERIC,
   bid           NUMERIC,
-  gal           VARCHAR,
+  gal           TEXT,
 	"end"         TIMESTAMP WITH TIME ZONE,
-  urn           VARCHAR,
+  urn           TEXT,
   tab           NUMERIC,
   tx_idx        INTEGER NOT NUll,
+  log_idx       INTEGER NOT NUll,
   raw_log       JSONB,
-  UNIQUE (header_id, tx_idx)
+  UNIQUE (header_id, tx_idx, log_idx)
 );

db/migrations/1536942529_create_flop_kick.up.sql

@@ -4,11 +4,12 @@ CREATE TABLE maker.flop_kick (
   bid_id        NUMERIC NOT NULL,
   lot           NUMERIC NOT NULL,
   bid           NUMERIC NOT NULL,
-  gal           VARCHAR,
+  gal           TEXT,
   "end"         TIMESTAMP WITH TIME ZONE,
   tx_idx        INTEGER NOT NULL,
+  log_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/migrations/1538679495_create_vat_toll.up.sql

@@ -5,8 +5,9 @@ CREATE TABLE maker.vat_toll (
   urn           TEXT,
   take          NUMERIC,
   tx_idx        INTEGER NOT NULL,
+  log_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/migrations/1538689084_create_vat_tune.up.sql

@@ -8,8 +8,9 @@ CREATE TABLE maker.vat_tune (
   dink          NUMERIC,
   dart          NUMERIC,
   tx_idx        INTEGER NOT NULL,
+  log_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/migrations/1539276289_create_vat_flux_table.up.sql

@@ -6,8 +6,9 @@ CREATE TABLE maker.vat_flux (
   dst           TEXT,
   rad           numeric,
   tx_idx        INTEGER NOT NULL,
+  log_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/migrations/1539638109_create_vat_slip.up.sql

@@ -5,8 +5,9 @@ CREATE TABLE maker.vat_slip (
   guy           TEXT,
   rad           NUMERIC,
   tx_idx        INTEGER NOT NULL,
+  log_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

@@ -427,11 +427,12 @@ CREATE TABLE maker.flip_kick (
     bid_id numeric NOT NULL,
     lot numeric,
     bid numeric,
-    gal character varying,
+    gal text,
     "end" timestamp with time zone,
-    urn character varying,
+    urn text,
     tab numeric,
     tx_idx integer NOT NULL,
+    log_idx integer NOT NULL,
     raw_log jsonb
 );
 
@@ -466,9 +467,10 @@ CREATE TABLE maker.flop_kick (
     bid_id numeric NOT NULL,
     lot numeric NOT NULL,
     bid numeric NOT NULL,
-    gal character varying,
+    gal text,
     "end" timestamp with time zone,
     tx_idx integer NOT NULL,
+    log_idx integer NOT NULL,
     raw_log jsonb
 );
 
@@ -725,6 +727,7 @@ CREATE TABLE maker.vat_flux (
     dst text,
     rad numeric,
     tx_idx integer NOT NULL,
+    log_idx integer NOT NULL,
     raw_log jsonb
 );
 
@@ -941,6 +944,7 @@ CREATE TABLE maker.vat_slip (
     guy text,
     rad numeric,
     tx_idx integer NOT NULL,
+    log_idx integer NOT NULL,
     raw_log jsonb
 );
 
@@ -976,6 +980,7 @@ CREATE TABLE maker.vat_toll (
     urn text,
     take numeric,
     tx_idx integer NOT NULL,
+    log_idx integer NOT NULL,
     raw_log jsonb
 );
 
@@ -1014,6 +1019,7 @@ CREATE TABLE maker.vat_tune (
     dink numeric,
     dart numeric,
     tx_idx integer NOT NULL,
+    log_idx integer NOT NULL,
     raw_log jsonb
 );
 
@@ -1888,11 +1894,11 @@ ALTER TABLE ONLY maker.drip_file_vow
 
 
 --
--- Name: flip_kick flip_kick_header_id_tx_idx_key; Type: CONSTRAINT; Schema: maker; Owner: -
+-- Name: flip_kick flip_kick_header_id_tx_idx_log_idx_key; Type: CONSTRAINT; Schema: maker; Owner: -
 --
 
 ALTER TABLE ONLY maker.flip_kick
-    ADD CONSTRAINT flip_kick_header_id_tx_idx_key UNIQUE (header_id, tx_idx);
+    ADD CONSTRAINT flip_kick_header_id_tx_idx_log_idx_key UNIQUE (header_id, tx_idx, log_idx);
 
 
 --
@@ -1904,11 +1910,11 @@ ALTER TABLE ONLY maker.flip_kick
 
 
 --
--- Name: flop_kick flop_kick_header_id_tx_idx_key; Type: CONSTRAINT; Schema: maker; Owner: -
+-- Name: flop_kick flop_kick_header_id_tx_idx_log_idx_key; Type: CONSTRAINT; Schema: maker; Owner: -
 --
 
 ALTER TABLE ONLY maker.flop_kick
-    ADD CONSTRAINT flop_kick_header_id_tx_idx_key UNIQUE (header_id, tx_idx);
+    ADD CONSTRAINT flop_kick_header_id_tx_idx_log_idx_key UNIQUE (header_id, tx_idx, log_idx);
 
 
 --
@@ -2016,11 +2022,11 @@ ALTER TABLE ONLY maker.tend
 
 
 --
--- Name: vat_flux vat_flux_header_id_tx_idx_key; Type: CONSTRAINT; Schema: maker; Owner: -
+-- Name: vat_flux vat_flux_header_id_tx_idx_log_idx_key; Type: CONSTRAINT; Schema: maker; Owner: -
 --
 
 ALTER TABLE ONLY maker.vat_flux
-    ADD CONSTRAINT vat_flux_header_id_tx_idx_key UNIQUE (header_id, tx_idx);
+    ADD CONSTRAINT vat_flux_header_id_tx_idx_log_idx_key UNIQUE (header_id, tx_idx, log_idx);
 
 
 --
@@ -2112,11 +2118,11 @@ ALTER TABLE ONLY maker.vat_move
 
 
 --
--- Name: vat_slip vat_slip_header_id_tx_idx_key; Type: CONSTRAINT; Schema: maker; Owner: -
+-- Name: vat_slip vat_slip_header_id_tx_idx_log_idx_key; Type: CONSTRAINT; Schema: maker; Owner: -
 --
 
 ALTER TABLE ONLY maker.vat_slip
-    ADD CONSTRAINT vat_slip_header_id_tx_idx_key UNIQUE (header_id, tx_idx);
+    ADD CONSTRAINT vat_slip_header_id_tx_idx_log_idx_key UNIQUE (header_id, tx_idx, log_idx);
 
 
 --
@@ -2128,11 +2134,11 @@ ALTER TABLE ONLY maker.vat_slip
 
 
 --
--- Name: vat_toll vat_toll_header_id_tx_idx_key; Type: CONSTRAINT; Schema: maker; Owner: -
+-- Name: vat_toll vat_toll_header_id_tx_idx_log_idx_key; Type: CONSTRAINT; Schema: maker; Owner: -
 --
 
 ALTER TABLE ONLY maker.vat_toll
-    ADD CONSTRAINT vat_toll_header_id_tx_idx_key UNIQUE (header_id, tx_idx);
+    ADD CONSTRAINT vat_toll_header_id_tx_idx_log_idx_key UNIQUE (header_id, tx_idx, log_idx);
 
 
 --
@@ -2144,11 +2150,11 @@ ALTER TABLE ONLY maker.vat_toll
 
 
 --
--- Name: vat_tune vat_tune_header_id_tx_idx_key; Type: CONSTRAINT; Schema: maker; Owner: -
+-- Name: vat_tune vat_tune_header_id_tx_idx_log_idx_key; Type: CONSTRAINT; Schema: maker; Owner: -
 --
 
 ALTER TABLE ONLY maker.vat_tune
-    ADD CONSTRAINT vat_tune_header_id_tx_idx_key UNIQUE (header_id, tx_idx);
+    ADD CONSTRAINT vat_tune_header_id_tx_idx_log_idx_key UNIQUE (header_id, tx_idx, log_idx);
 
 
 --

pkg/transformers/cat_file/chop_lump/repository_test.go

@@ -58,12 +58,13 @@ var _ = Describe("Cat file chop lump repository", func() {
 
 		It("adds a cat file chop lump event", func() {
 			var dbResult chop_lump.CatFileChopLumpModel
-			err = db.Get(&dbResult, `SELECT ilk, what, data, tx_idx, raw_log FROM maker.cat_file_chop_lump WHERE header_id = $1`, headerID)
+			err = db.Get(&dbResult, `SELECT ilk, what, data, tx_idx, log_idx, raw_log FROM maker.cat_file_chop_lump WHERE header_id = $1`, headerID)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(dbResult.Ilk).To(Equal(test_data.CatFileChopLumpModel.Ilk))
 			Expect(dbResult.What).To(Equal(test_data.CatFileChopLumpModel.What))
 			Expect(dbResult.Data).To(Equal(test_data.CatFileChopLumpModel.Data))
 			Expect(dbResult.TransactionIndex).To(Equal(test_data.CatFileChopLumpModel.TransactionIndex))
+			Expect(dbResult.LogIndex).To(Equal(test_data.CatFileChopLumpModel.LogIndex))
 			Expect(dbResult.Raw).To(MatchJSON(test_data.CatFileChopLumpModel.Raw))
 		})
 
@@ -81,7 +82,7 @@ var _ = Describe("Cat file chop lump repository", func() {
 			Expect(err.Error()).To(ContainSubstring("pq: duplicate key value violates unique constraint"))
 		})
 
-		It("allows for multiple cat file chop lump events if they have different log indexes", func() {
+		It("allows for multiple cat file chop lump events in one transaction if they have different log indexes", func() {
 			newCatFileChopLump := test_data.CatFileChopLumpModel
 			newCatFileChopLump.LogIndex = newCatFileChopLump.LogIndex + 1
 			err = catFileRepository.Create(headerID, []chop_lump.CatFileChopLumpModel{newCatFileChopLump})

pkg/transformers/cat_file/flip/repository_test.go

@@ -82,6 +82,14 @@ var _ = Describe("Cat file flip repository", func() {
 			Expect(err.Error()).To(ContainSubstring("pq: duplicate key value violates unique constraint"))
 		})
 
+		It("allows for multiple cat file flip events in one transaction if they have different log indexes", func() {
+			catFileFlip := test_data.CatFileFlipModel
+			catFileFlip.LogIndex = catFileFlip.LogIndex + 1
+			err = catFileRepository.Create(headerID, []flip.CatFileFlipModel{catFileFlip})
+
+			Expect(err).NotTo(HaveOccurred())
+		})
+
 		It("removes cat file flip if corresponding header is deleted", func() {
 			_, err = db.Exec(`DELETE FROM headers WHERE id = $1`, headerID)
 

pkg/transformers/cat_file/pit_vow/repository_test.go

@@ -81,6 +81,14 @@ var _ = Describe("Cat file pit vow repository", func() {
 			Expect(err.Error()).To(ContainSubstring("pq: duplicate key value violates unique constraint"))
 		})
 
+		It("allows for multiple cat file pit events in one transaction if they have different log indexes", func() {
+			catFilePitVow := test_data.CatFilePitVowModel
+			catFilePitVow.LogIndex = catFilePitVow.LogIndex + 1
+			err = catFileRepository.Create(headerID, []pit_vow.CatFilePitVowModel{catFilePitVow})
+
+			Expect(err).NotTo(HaveOccurred())
+		})
+
 		It("removes cat file pit vow if corresponding header is deleted", func() {
 			_, err = db.Exec(`DELETE FROM headers WHERE id = $1`, headerID)
 

pkg/transformers/flip_kick/converter.go

@@ -51,6 +51,7 @@ func (FlipKickConverter) ToEntities(contractAbi string, ethLogs []types.Log) (re
 		}
 		entity.Raw = ethLog
 		entity.TransactionIndex = ethLog.TxIndex
+		entity.LogIndex = ethLog.Index
 		results = append(results, *entity)
 	}
 
@@ -86,6 +87,7 @@ func (FlipKickConverter) ToModels(flipKicks []FlipKickEntity) (results []FlipKic
 			Urn:              urn,
 			Tab:              tab,
 			TransactionIndex: flipKick.TransactionIndex,
+			LogIndex:         flipKick.LogIndex,
 			Raw:              rawLogString,
 		}
 		results = append(results, model)

pkg/transformers/flip_kick/converter_test.go

@@ -38,14 +38,7 @@ var _ = Describe("FlipKick Converter", func() {
 			Expect(err).NotTo(HaveOccurred())
 			Expect(len(entities)).To(Equal(1))
 			entity := entities[0]
-			Expect(entity.Id).To(Equal(test_data.FlipKickEntity.Id))
+			Expect(entity).To(Equal(test_data.FlipKickEntity))
-			Expect(entity.Lot).To(Equal(test_data.FlipKickEntity.Lot))
-			Expect(entity.Bid).To(Equal(test_data.FlipKickEntity.Bid))
-			Expect(entity.Gal).To(Equal(test_data.FlipKickEntity.Gal))
-			Expect(entity.End).To(Equal(test_data.FlipKickEntity.End))
-			Expect(entity.Urn).To(Equal(test_data.FlipKickEntity.Urn))
-			Expect(entity.Tab).To(Equal(test_data.FlipKickEntity.Tab))
-			Expect(entity.Raw).To(Equal(test_data.FlipKickEntity.Raw))
 		})
 
 		It("returns an error if converting log to entity fails", func() {

pkg/transformers/flip_kick/entity.go

@@ -30,5 +30,6 @@ type FlipKickEntity struct {
 	Urn              [32]byte
 	Tab              *big.Int
 	TransactionIndex uint
+	LogIndex         uint
 	Raw              types.Log
 }

pkg/transformers/flip_kick/model.go

@@ -25,5 +25,6 @@ type FlipKickModel struct {
 	Urn              string
 	Tab              string
 	TransactionIndex uint   `db:"tx_idx"`
+	LogIndex         uint   `db:"log_idx"`
 	Raw              string `db:"raw_log"`
 }

pkg/transformers/flip_kick/repository.go

@@ -41,9 +41,9 @@ func (fkr FlipKickRepository) Create(headerId int64, flipKicks []FlipKickModel)
 	}
 	for _, flipKick := range flipKicks {
 		_, err := tx.Exec(
-			`INSERT into maker.flip_kick (header_id, bid_id, lot, bid, gal, "end", urn, tab, tx_idx, raw_log)
+			`INSERT into maker.flip_kick (header_id, bid_id, lot, bid, gal, "end", urn, tab, tx_idx, log_idx, raw_log)
-        VALUES($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`,
+        VALUES($1, $2::NUMERIC, $3::NUMERIC, $4::NUMERIC, $5, $6, $7, $8::NUMERIC, $9, $10, $11)`,
-			headerId, flipKick.BidId, flipKick.Lot, flipKick.Bid, flipKick.Gal, flipKick.End, flipKick.Urn, flipKick.Tab, flipKick.TransactionIndex, flipKick.Raw,
+			headerId, flipKick.BidId, flipKick.Lot, flipKick.Bid, flipKick.Gal, flipKick.End, flipKick.Urn, flipKick.Tab, flipKick.TransactionIndex, flipKick.LogIndex, flipKick.Raw,
 		)
 		if err != nil {
 			tx.Rollback()

pkg/transformers/flip_kick/repository_test.go

@@ -69,6 +69,7 @@ var _ = Describe("FlipKick Repository", func() {
 			Expect(dbResult.Urn).To(Equal(flipKick.Urn))
 			Expect(dbResult.Tab).To(Equal(flipKick.Tab))
 			Expect(dbResult.TransactionIndex).To(Equal(flipKick.TransactionIndex))
+			Expect(dbResult.LogIndex).To(Equal(flipKick.LogIndex))
 			Expect(dbResult.Raw).To(MatchJSON(flipKick.Raw))
 		})
 
@@ -82,6 +83,18 @@ var _ = Describe("FlipKick Repository", func() {
 			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 = flipKickRepository.Create(headerId, []flip_kick.FlipKickModel{flipKick})
+			Expect(err).NotTo(HaveOccurred())
+
+			var headerChecked bool
+			err = db.Get(&headerChecked, `SELECT flip_kick_checked FROM public.checked_headers WHERE header_id = $1`, headerId)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(headerChecked).To(BeTrue())
+		})
+
 		It("returns an error if inserting the flip_kick record fails", func() {
 			err := flipKickRepository.Create(headerId, []flip_kick.FlipKickModel{flipKick})
 			Expect(err).NotTo(HaveOccurred())
@@ -91,6 +104,14 @@ var _ = Describe("FlipKick Repository", func() {
 			Expect(err.Error()).To(ContainSubstring("pq: duplicate key value violates unique constraint"))
 		})
 
+		It("allows for multiple flip kick events in one transaction if they have different log indexes", func() {
+			newFlipKick := test_data.FlipKickModel
+			newFlipKick.LogIndex = newFlipKick.LogIndex + 1
+			err := flipKickRepository.Create(headerId, []flip_kick.FlipKickModel{newFlipKick})
+
+			Expect(err).NotTo(HaveOccurred())
+		})
+
 		It("deletes the flip_kick records if its corresponding header record is deleted", func() {
 			err := flipKickRepository.Create(headerId, []flip_kick.FlipKickModel{flipKick})
 			Expect(err).NotTo(HaveOccurred())

pkg/transformers/flop_kick/converter.go

@@ -50,6 +50,7 @@ func (FlopKickConverter) ToEntities(contractAbi string, ethLogs []types.Log) ([]
 		}
 		entity.Raw = ethLog
 		entity.TransactionIndex = ethLog.TxIndex
+		entity.LogIndex = ethLog.Index
 		results = append(results, entity)
 	}
 	return results, nil
@@ -71,6 +72,7 @@ func (FlopKickConverter) ToModels(entities []Entity) ([]Model, error) {
 			Gal:              entity.Gal.String(),
 			End:              time.Unix(endValue, 0),
 			TransactionIndex: entity.TransactionIndex,
+			LogIndex:         entity.LogIndex,
 			Raw:              rawLogJson,
 		}
 		results = append(results, model)

pkg/transformers/flop_kick/converter_test.go

@@ -34,13 +34,7 @@ var _ = Describe("FlopKick Converter", func() {
 
 			Expect(err).NotTo(HaveOccurred())
 			entity := entities[0]
-			Expect(entity.Id).To(Equal(test_data.FlopKickEntity.Id))
+			Expect(entity).To(Equal(test_data.FlopKickEntity))
-			Expect(entity.Lot).To(Equal(test_data.FlopKickEntity.Lot))
-			Expect(entity.Bid).To(Equal(test_data.FlopKickEntity.Bid))
-			Expect(entity.Gal).To(Equal(test_data.FlopKickEntity.Gal))
-			Expect(entity.End).To(Equal(test_data.FlopKickEntity.End))
-			Expect(entity.TransactionIndex).To(Equal(test_data.FlopKickEntity.TransactionIndex))
-			Expect(entity.Raw).To(Equal(test_data.FlopKickEntity.Raw))
 		})
 
 		It("returns an error if converting the log to an entity fails", func() {

pkg/transformers/flop_kick/entity.go

@@ -28,5 +28,6 @@ type Entity struct {
 	Gal              common.Address
 	End              *big.Int
 	TransactionIndex uint
+	LogIndex         uint
 	Raw              types.Log
 }

pkg/transformers/flop_kick/model.go

@@ -23,5 +23,6 @@ type Model struct {
 	Gal              string
 	End              time.Time
 	TransactionIndex uint   `db:"tx_idx"`
+	LogIndex         uint   `db:"log_idx"`
 	Raw              []byte `db:"raw_log"`
 }

pkg/transformers/flop_kick/repository.go

@@ -40,9 +40,9 @@ func (r FlopKickRepository) Create(headerId int64, flopKicks []Model) error {
 	}
 	for _, flopKick := range flopKicks {
 		_, err = tx.Exec(
-			`INSERT into maker.flop_kick (header_id, bid_id, lot, bid, gal, "end", tx_idx, raw_log)
+			`INSERT into maker.flop_kick (header_id, bid_id, lot, bid, gal, "end", tx_idx, log_idx, raw_log)
-        VALUES($1, $2, $3, $4, $5, $6, $7, $8)`,
+        VALUES($1, $2::NUMERIC, $3::NUMERIC, $4::NUMERIC, $5, $6, $7, $8, $9)`,
-			headerId, flopKick.BidId, flopKick.Lot, flopKick.Bid, flopKick.Gal, flopKick.End, flopKick.TransactionIndex, flopKick.Raw,
+			headerId, flopKick.BidId, flopKick.Lot, flopKick.Bid, flopKick.Gal, flopKick.End, flopKick.TransactionIndex, flopKick.LogIndex, flopKick.Raw,
 		)
 		if err != nil {
 			tx.Rollback()

pkg/transformers/flop_kick/repository_test.go

@@ -50,12 +50,12 @@ var _ = Describe("FlopRepository", func() {
 		BeforeEach(func() {
 			headerId, err = headerRepository.CreateOrUpdateHeader(fakes.FakeHeader)
 			Expect(err).NotTo(HaveOccurred())
-
-			err := repository.Create(headerId, []flop_kick.Model{test_data.FlopKickModel})
-			Expect(err).NotTo(HaveOccurred())
 		})
 
 		It("creates FlopKick records", func() {
+			err := repository.Create(headerId, []flop_kick.Model{test_data.FlopKickModel})
+			Expect(err).NotTo(HaveOccurred())
+
 			err = db.QueryRowx(`SELECT * FROM maker.flop_kick WHERE header_id = $1`, headerId).StructScan(&dbResult)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(dbResult.HeaderId).To(Equal(headerId))
@@ -65,10 +65,26 @@ var _ = Describe("FlopRepository", func() {
 			Expect(dbResult.Gal).To(Equal(test_data.FlopKickModel.Gal))
 			Expect(dbResult.End.Equal(test_data.FlopKickModel.End)).To(BeTrue())
 			Expect(dbResult.TransactionIndex).To(Equal(test_data.FlopKickModel.TransactionIndex))
+			Expect(dbResult.LogIndex).To(Equal(test_data.FlopKickModel.LogIndex))
 			Expect(dbResult.Raw).To(MatchJSON(test_data.FlopKickModel.Raw))
 		})
 
 		It("marks headerId as checked for flop kick logs", func() {
+			err := repository.Create(headerId, []flop_kick.Model{test_data.FlopKickModel})
+			Expect(err).NotTo(HaveOccurred())
+
+			var headerChecked bool
+			err = db.Get(&headerChecked, `SELECT flop_kick_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 = repository.Create(headerId, []flop_kick.Model{test_data.FlopKickModel})
+			Expect(err).NotTo(HaveOccurred())
+
 			var headerChecked bool
 			err = db.Get(&headerChecked, `SELECT flop_kick_checked FROM public.checked_headers WHERE header_id = $1`, headerId)
 			Expect(err).NotTo(HaveOccurred())
@@ -76,12 +92,29 @@ var _ = Describe("FlopRepository", func() {
 		})
 
 		It("returns an error if inserting the flop_kick record fails", func() {
+			err := repository.Create(headerId, []flop_kick.Model{test_data.FlopKickModel})
+			Expect(err).NotTo(HaveOccurred())
+
 			err = repository.Create(headerId, []flop_kick.Model{test_data.FlopKickModel})
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("pq: duplicate key value violates unique constraint"))
 		})
 
+		It("allows for multiple flop kick events in one transaction if they have different log indexes", func() {
+			err := repository.Create(headerId, []flop_kick.Model{test_data.FlopKickModel})
+			Expect(err).NotTo(HaveOccurred())
+
+			newFlopKick := test_data.FlopKickModel
+			newFlopKick.LogIndex = newFlopKick.LogIndex + 1
+			err = repository.Create(headerId, []flop_kick.Model{newFlopKick})
+
+			Expect(err).NotTo(HaveOccurred())
+		})
+
 		It("deletes the flop_kick records if its corresponding header record is deleted", func() {
+			err := repository.Create(headerId, []flop_kick.Model{test_data.FlopKickModel})
+			Expect(err).NotTo(HaveOccurred())
+
 			var flopKickCount int
 			err = db.QueryRow(`SELECT count(*) FROM maker.flop_kick`).Scan(&flopKickCount)
 			Expect(err).NotTo(HaveOccurred())

pkg/transformers/test_data/flip_kick.go

@@ -63,7 +63,7 @@ var EthFlipKickLog = types.Log{
 	TxHash:      common.HexToHash(flipKickTransactionHash),
 	TxIndex:     999,
 	BlockHash:   common.HexToHash(flipKickBlockHash),
-	Index:       0,
+	Index:       1,
 	Removed:     false,
 }
 
@@ -76,6 +76,7 @@ var FlipKickEntity = flip_kick.FlipKickEntity{
 	Urn:              urn,
 	Tab:              tab,
 	TransactionIndex: EthFlipKickLog.TxIndex,
+	LogIndex:         EthFlipKickLog.Index,
 	Raw:              EthFlipKickLog,
 }
 
@@ -88,6 +89,7 @@ var FlipKickModel = flip_kick.FlipKickModel{
 	Urn:              urnString,
 	Tab:              tabString,
 	TransactionIndex: EthFlipKickLog.TxIndex,
+	LogIndex:         EthFlipKickLog.Index,
 	Raw:              rawLogString,
 }
 

pkg/transformers/test_data/flop_kick.go

@@ -57,6 +57,7 @@ var (
 		Gal:              common.HexToAddress(flopGal),
 		End:              big.NewInt(flopEnd),
 		TransactionIndex: flopTxIndex,
+		LogIndex:         FlopKickLog.Index,
 		Raw:              FlopKickLog,
 	}
 
@@ -67,6 +68,7 @@ var (
 		Gal:              flopGal,
 		End:              time.Unix(flopEnd, 0),
 		TransactionIndex: flopTxIndex,
+		LogIndex:         FlopKickLog.Index,
 		Raw:              rawFlopLogJson,
 	}
 )

pkg/transformers/test_data/vat_flux.go

@@ -37,7 +37,7 @@ var VatFluxLog = types.Log{
 	TxHash:      common.HexToHash("0xf98681bab9b8c75bd8aa4a7d0a8142ff527c5ea8fa54f3c2835d4533838b2e6f"),
 	TxIndex:     0,
 	BlockHash:   common.HexToHash("0xc3fe212ad4f81ade1265af6de2b4bb50d962b1a4db06aabc982e7f9cb0972c2d"),
-	Index:       0,
+	Index:       3,
 	Removed:     false,
 }
 
@@ -47,6 +47,7 @@ var VatFluxModel = vat_flux.VatFluxModel{
 	Src:              "0x7FA9EF6609Ca7921112231f8f195138ebba29770",
 	Dst:              "0x93086347c52a8878af71bB818509d484c6a2e1bF",
 	Rad:              "123",
-	TransactionIndex: 0,
+	TransactionIndex: VatFluxLog.TxIndex,
+	LogIndex:         VatFluxLog.Index,
 	Raw:              rawFluxLog,
 }

pkg/transformers/test_data/vat_slip.go

@@ -38,7 +38,7 @@ var EthVatSlipLog = types.Log{
 	TxHash:      common.HexToHash("0xb114ba306c80c86d51bdbf4a5ac8ed151020cd81b70cfa1dc9822f4a1f73930b"),
 	TxIndex:     3,
 	BlockHash:   common.HexToHash("0x34b7e5ddb3be73257a5a0087f10b8bf68d4df5c8831ec04c63ecae4094de72ad"),
-	Index:       0,
+	Index:       2,
 	Removed:     false,
 }
 
@@ -48,5 +48,6 @@ var VatSlipModel = vat_slip.VatSlipModel{
 	Guy:              common.HexToAddress("0x7d7bEe5fCfD8028cf7b00876C5b1421c800561A6").String(),
 	Rad:              "987654321",
 	TransactionIndex: EthVatSlipLog.TxIndex,
+	LogIndex:         EthVatSlipLog.Index,
 	Raw:              rawVatSlipLog,
 }

pkg/transformers/test_data/vat_toll.go

@@ -32,5 +32,6 @@ var VatTollModel = vat_toll.VatTollModel{
 	Urn:              "0xA3E37186E017747DbA34042e83e3F76Ad3CcE9b0",
 	Take:             big.NewInt(123456789).String(),
 	TransactionIndex: EthVatTollLog.TxIndex,
+	LogIndex:         EthVatTollLog.Index,
 	Raw:              rawVatTollLog,
 }

pkg/transformers/test_data/vat_tune.go

@@ -40,5 +40,6 @@ var VatTuneModel = vat_tune.VatTuneModel{
 	Dink:             big.NewInt(0).String(),
 	Dart:             vatTuneDart.String(),
 	TransactionIndex: EthVatTuneLog.TxIndex,
+	LogIndex:         EthVatTuneLog.Index,
 	Raw:              rawVatTuneLog,
 }

pkg/transformers/vat_flux/converter.go

@@ -59,6 +59,7 @@ func (VatFluxConverter) ToModels(ethLogs []types.Log) ([]VatFluxModel, error) {
 			Dst:              dst.String(),
 			Rad:              rad,
 			TransactionIndex: ethLog.TxIndex,
+			LogIndex:         ethLog.Index,
 			Raw:              rawLogJson,
 		}
 

pkg/transformers/vat_flux/converter_test.go

@@ -31,12 +31,7 @@ var _ = Describe("VatFlux converter", func() {
 
 		Expect(err).NotTo(HaveOccurred())
 		Expect(len(models)).To(Equal(1))
-		Expect(models[0].Ilk).To(Equal(test_data.VatFluxModel.Ilk))
+		Expect(models[0]).To(Equal(test_data.VatFluxModel))
-		Expect(models[0].Src).To(Equal(test_data.VatFluxModel.Src))
-		Expect(models[0].Dst).To(Equal(test_data.VatFluxModel.Dst))
-		Expect(models[0].Rad).To(Equal(test_data.VatFluxModel.Rad))
-		Expect(models[0].TransactionIndex).To(Equal(test_data.VatFluxModel.TransactionIndex))
-		Expect(models[0].Raw).To(Equal(test_data.VatFluxModel.Raw))
 	})
 
 	It("Returns an error there are missing topics", func() {

pkg/transformers/vat_flux/model.go

@@ -20,5 +20,6 @@ type VatFluxModel struct {
 	Dst              string
 	Rad              string
 	TransactionIndex uint   `db:"tx_idx"`
+	LogIndex         uint   `db:"log_idx"`
 	Raw              []byte `db:"raw_log"`
 }

pkg/transformers/vat_flux/repository.go

@@ -40,9 +40,9 @@ func (repository VatFluxRepository) Create(headerId int64, models []VatFluxModel
 	}
 
 	for _, model := range models {
-		_, err := tx.Exec(`INSERT INTO maker.vat_flux (header_id, ilk, dst, src, rad, tx_idx, raw_log)
+		_, err := tx.Exec(`INSERT INTO maker.vat_flux (header_id, ilk, dst, src, rad, tx_idx, log_idx, raw_log)
-		VALUES($1, $2, $3, $4, $5::numeric, $6, $7)`,
+		VALUES($1, $2, $3, $4, $5::NUMERIC, $6, $7, $8)`,
-			headerId, model.Ilk, model.Dst, model.Src, model.Rad, model.TransactionIndex, model.Raw)
+			headerId, model.Ilk, model.Dst, model.Src, model.Rad, model.TransactionIndex, model.LogIndex, model.Raw)
 		if err != nil {
 			tx.Rollback()
 			return err

pkg/transformers/vat_flux/repository_test.go

@@ -69,6 +69,7 @@ var _ = Describe("VatFlux Repository", func() {
 			Expect(dbResult[0].Rad).To(Equal(test_data.VatFluxModel.Rad))
 			Expect(dbResult[0].TransactionIndex).To(Equal(test_data.VatFluxModel.TransactionIndex))
 			Expect(dbResult[1].TransactionIndex).To(Equal(test_data.VatFluxModel.TransactionIndex + 1))
+			Expect(dbResult[0].LogIndex).To(Equal(test_data.VatFluxModel.LogIndex))
 			Expect(dbResult[0].Raw).To(MatchJSON(test_data.VatFluxModel.Raw))
 			Expect(dbResult[0].HeaderId).To(Equal(headerId))
 		})
@@ -81,6 +82,17 @@ var _ = Describe("VatFlux Repository", func() {
 			Expect(err.Error()).To(ContainSubstring("pq: duplicate key value violates unique constraint"))
 		})
 
+		It("allows for multiple vat flux events in one transaction if they have different log indexes", func() {
+			err = repository.Create(headerId, []vat_flux.VatFluxModel{test_data.VatFluxModel})
+			Expect(err).NotTo(HaveOccurred())
+
+			anotherVatFlux := test_data.VatFluxModel
+			anotherVatFlux.LogIndex = anotherVatFlux.LogIndex + 1
+			err = repository.Create(headerId, []vat_flux.VatFluxModel{anotherVatFlux})
+
+			Expect(err).NotTo(HaveOccurred())
+		})
+
 		It("marks the header as checked for vat flux logs", func() {
 			err = repository.Create(headerId, []vat_flux.VatFluxModel{test_data.VatFluxModel})
 			Expect(err).NotTo(HaveOccurred())
@@ -91,6 +103,18 @@ var _ = Describe("VatFlux Repository", func() {
 			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 = repository.Create(headerId, []vat_flux.VatFluxModel{test_data.VatFluxModel})
+			Expect(err).NotTo(HaveOccurred())
+
+			var headerChecked bool
+			err = db.Get(&headerChecked, `SELECT vat_flux_checked FROM public.checked_headers WHERE header_id = $1`, headerId)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(headerChecked).To(BeTrue())
+		})
+
 		It("removes vat flux if corresponding header is deleted", func() {
 			err = repository.Create(headerId, []vat_flux.VatFluxModel{test_data.VatFluxModel})
 			Expect(err).NotTo(HaveOccurred())

pkg/transformers/vat_slip/converter.go

@@ -48,6 +48,7 @@ func (VatSlipConverter) ToModels(ethLogs []types.Log) ([]VatSlipModel, error) {
 			Guy:              guy.String(),
 			Rad:              rad.String(),
 			TransactionIndex: ethLog.TxIndex,
+			LogIndex:         ethLog.Index,
 			Raw:              raw,
 		}
 		models = append(models, model)

pkg/transformers/vat_slip/model.go

@@ -19,5 +19,6 @@ type VatSlipModel struct {
 	Guy              string
 	Rad              string
 	TransactionIndex uint   `db:"tx_idx"`
+	LogIndex         uint   `db:"log_idx"`
 	Raw              []byte `db:"raw_log"`
 }

pkg/transformers/vat_slip/repository.go

@@ -28,9 +28,9 @@ func (repository VatSlipRepository) Create(headerID int64, models []VatSlipModel
 	}
 	for _, model := range models {
 		_, err = tx.Exec(
-			`INSERT into maker.vat_slip (header_id, ilk, guy, rad, tx_idx, raw_log)
+			`INSERT into maker.vat_slip (header_id, ilk, guy, rad, tx_idx, log_idx, raw_log)
-        VALUES($1, $2, $3, $4::NUMERIC, $5, $6)`,
+        VALUES($1, $2, $3, $4::NUMERIC, $5, $6, $7)`,
-			headerID, model.Ilk, model.Guy, model.Rad, model.TransactionIndex, model.Raw,
+			headerID, model.Ilk, model.Guy, model.Rad, model.TransactionIndex, model.LogIndex, model.Raw,
 		)
 		if err != nil {
 			tx.Rollback()

pkg/transformers/vat_slip/repository_test.go

@@ -37,23 +37,39 @@ var _ = Describe("Vat slip repository", func() {
 		BeforeEach(func() {
 			headerID, err = headerRepository.CreateOrUpdateHeader(fakes.FakeHeader)
 			Expect(err).NotTo(HaveOccurred())
-
-			err = vatSlipRepository.Create(headerID, []vat_slip.VatSlipModel{test_data.VatSlipModel})
-			Expect(err).NotTo(HaveOccurred())
 		})
 
 		It("adds a vat slip event", func() {
+			err = vatSlipRepository.Create(headerID, []vat_slip.VatSlipModel{test_data.VatSlipModel})
+			Expect(err).NotTo(HaveOccurred())
+
 			var dbVatSlip vat_slip.VatSlipModel
-			err = db.Get(&dbVatSlip, `SELECT ilk, guy, rad, tx_idx, raw_log FROM maker.vat_slip WHERE header_id = $1`, headerID)
+			err = db.Get(&dbVatSlip, `SELECT ilk, guy, rad, tx_idx, log_idx, raw_log FROM maker.vat_slip WHERE header_id = $1`, headerID)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(dbVatSlip.Ilk).To(Equal(test_data.VatSlipModel.Ilk))
 			Expect(dbVatSlip.Guy).To(Equal(test_data.VatSlipModel.Guy))
 			Expect(dbVatSlip.Rad).To(Equal(test_data.VatSlipModel.Rad))
 			Expect(dbVatSlip.TransactionIndex).To(Equal(test_data.VatSlipModel.TransactionIndex))
+			Expect(dbVatSlip.LogIndex).To(Equal(test_data.VatSlipModel.LogIndex))
 			Expect(dbVatSlip.Raw).To(MatchJSON(test_data.VatSlipModel.Raw))
 		})
 
 		It("marks header as checked for logs", func() {
+			err = vatSlipRepository.Create(headerID, []vat_slip.VatSlipModel{test_data.VatSlipModel})
+			Expect(err).NotTo(HaveOccurred())
+
+			var headerChecked bool
+			err = db.Get(&headerChecked, `SELECT vat_slip_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 = vatSlipRepository.Create(headerID, []vat_slip.VatSlipModel{test_data.VatSlipModel})
+			Expect(err).NotTo(HaveOccurred())
+
 			var headerChecked bool
 			err = db.Get(&headerChecked, `SELECT vat_slip_checked FROM public.checked_headers WHERE header_id = $1`, headerID)
 			Expect(err).NotTo(HaveOccurred())
@@ -61,13 +77,30 @@ var _ = Describe("Vat slip repository", func() {
 		})
 
 		It("does not duplicate vat slip events", func() {
+			err = vatSlipRepository.Create(headerID, []vat_slip.VatSlipModel{test_data.VatSlipModel})
+			Expect(err).NotTo(HaveOccurred())
+
 			err = vatSlipRepository.Create(headerID, []vat_slip.VatSlipModel{test_data.VatSlipModel})
 
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("pq: duplicate key value violates unique constraint"))
 		})
 
+		It("allows for multiple vat slip events in one transaction if they have different log indexes", func() {
+			err = vatSlipRepository.Create(headerID, []vat_slip.VatSlipModel{test_data.VatSlipModel})
+			Expect(err).NotTo(HaveOccurred())
+
+			newVatSlip := test_data.VatSlipModel
+			newVatSlip.LogIndex = newVatSlip.LogIndex + 1
+			err := vatSlipRepository.Create(headerID, []vat_slip.VatSlipModel{newVatSlip})
+
+			Expect(err).NotTo(HaveOccurred())
+		})
+
 		It("removes vat slip if corresponding header is deleted", func() {
+			err = vatSlipRepository.Create(headerID, []vat_slip.VatSlipModel{test_data.VatSlipModel})
+			Expect(err).NotTo(HaveOccurred())
+
 			_, err = db.Exec(`DELETE FROM headers WHERE id = $1`, headerID)
 
 			Expect(err).NotTo(HaveOccurred())

pkg/transformers/vat_toll/converter.go

@@ -35,6 +35,7 @@ func (VatTollConverter) ToModels(ethLogs []types.Log) ([]VatTollModel, error) {
 			Urn:              urn.String(),
 			Take:             take.String(),
 			TransactionIndex: ethLog.TxIndex,
+			LogIndex:         ethLog.Index,
 			Raw:              raw,
 		}
 		models = append(models, model)

pkg/transformers/vat_toll/model.go

@@ -5,5 +5,6 @@ type VatTollModel struct {
 	Urn              string
 	Take             string
 	TransactionIndex uint   `db:"tx_idx"`
+	LogIndex         uint   `db:"log_idx"`
 	Raw              []byte `db:"raw_log"`
 }

pkg/transformers/vat_toll/repository.go

@@ -28,9 +28,9 @@ func (repository VatTollRepository) Create(headerID int64, models []VatTollModel
 	}
 	for _, model := range models {
 		_, err = tx.Exec(
-			`INSERT into maker.vat_toll (header_id, ilk, urn, take, tx_idx, raw_log)
+			`INSERT into maker.vat_toll (header_id, ilk, urn, take, tx_idx, log_idx, raw_log)
-        VALUES($1, $2, $3, $4::NUMERIC, $5, $6)`,
+        VALUES($1, $2, $3, $4::NUMERIC, $5, $6, $7)`,
-			headerID, model.Ilk, model.Urn, model.Take, model.TransactionIndex, model.Raw,
+			headerID, model.Ilk, model.Urn, model.Take, model.TransactionIndex, model.LogIndex, model.Raw,
 		)
 		if err != nil {
 			tx.Rollback()

pkg/transformers/vat_toll/repository_test.go

@@ -37,23 +37,39 @@ var _ = Describe("Vat toll repository", func() {
 		BeforeEach(func() {
 			headerID, err = headerRepository.CreateOrUpdateHeader(fakes.FakeHeader)
 			Expect(err).NotTo(HaveOccurred())
-
-			err = vatTollRepository.Create(headerID, []vat_toll.VatTollModel{test_data.VatTollModel})
-			Expect(err).NotTo(HaveOccurred())
 		})
 
 		It("adds a vat toll event", func() {
+			err = vatTollRepository.Create(headerID, []vat_toll.VatTollModel{test_data.VatTollModel})
+			Expect(err).NotTo(HaveOccurred())
+
 			var dbVatToll vat_toll.VatTollModel
-			err = db.Get(&dbVatToll, `SELECT ilk, urn, take, tx_idx, raw_log FROM maker.vat_toll WHERE header_id = $1`, headerID)
+			err = db.Get(&dbVatToll, `SELECT ilk, urn, take, tx_idx, log_idx, raw_log FROM maker.vat_toll WHERE header_id = $1`, headerID)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(dbVatToll.Ilk).To(Equal(test_data.VatTollModel.Ilk))
 			Expect(dbVatToll.Urn).To(Equal(test_data.VatTollModel.Urn))
 			Expect(dbVatToll.Take).To(Equal(test_data.VatTollModel.Take))
 			Expect(dbVatToll.TransactionIndex).To(Equal(test_data.VatTollModel.TransactionIndex))
+			Expect(dbVatToll.LogIndex).To(Equal(test_data.VatTollModel.LogIndex))
 			Expect(dbVatToll.Raw).To(MatchJSON(test_data.VatTollModel.Raw))
 		})
 
 		It("marks header as checked for logs", func() {
+			err = vatTollRepository.Create(headerID, []vat_toll.VatTollModel{test_data.VatTollModel})
+			Expect(err).NotTo(HaveOccurred())
+
+			var headerChecked bool
+			err = db.Get(&headerChecked, `SELECT vat_toll_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 = vatTollRepository.Create(headerID, []vat_toll.VatTollModel{test_data.VatTollModel})
+			Expect(err).NotTo(HaveOccurred())
+
 			var headerChecked bool
 			err = db.Get(&headerChecked, `SELECT vat_toll_checked FROM public.checked_headers WHERE header_id = $1`, headerID)
 			Expect(err).NotTo(HaveOccurred())
@@ -62,11 +78,24 @@ var _ = Describe("Vat toll repository", func() {
 
 		It("does not duplicate vat toll events", func() {
 			err = vatTollRepository.Create(headerID, []vat_toll.VatTollModel{test_data.VatTollModel})
+			Expect(err).NotTo(HaveOccurred())
 
+			err = vatTollRepository.Create(headerID, []vat_toll.VatTollModel{test_data.VatTollModel})
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("pq: duplicate key value violates unique constraint"))
 		})
 
+		It("allows for multiple vat toll events in one transaction if they have different log indexes", func() {
+			err = vatTollRepository.Create(headerID, []vat_toll.VatTollModel{test_data.VatTollModel})
+			Expect(err).NotTo(HaveOccurred())
+
+			newVatToll := test_data.VatTollModel
+			newVatToll.LogIndex = newVatToll.LogIndex + 1
+			err := vatTollRepository.Create(headerID, []vat_toll.VatTollModel{newVatToll})
+
+			Expect(err).NotTo(HaveOccurred())
+		})
+
 		It("removes vat toll if corresponding header is deleted", func() {
 			_, err = db.Exec(`DELETE FROM headers WHERE id = $1`, headerID)
 

pkg/transformers/vat_tune/converter.go

@@ -47,6 +47,7 @@ func (VatTuneConverter) ToModels(ethLogs []types.Log) ([]VatTuneModel, error) {
 			Dink:             dink,
 			Dart:             dart,
 			TransactionIndex: ethLog.TxIndex,
+			LogIndex:         ethLog.Index,
 			Raw:              raw,
 		}
 		models = append(models, model)

pkg/transformers/vat_tune/model.go

@@ -8,5 +8,6 @@ type VatTuneModel struct {
 	Dink             string
 	Dart             string
 	TransactionIndex uint   `db:"tx_idx"`
+	LogIndex         uint   `db:"log_idx"`
 	Raw              []byte `db:"raw_log"`
 }

pkg/transformers/vat_tune/repository.go

@@ -28,9 +28,9 @@ func (repository VatTuneRepository) Create(headerID int64, models []VatTuneModel
 	}
 	for _, model := range models {
 		_, err = tx.Exec(
-			`INSERT into maker.vat_tune (header_id, ilk, urn, v, w, dink, dart, tx_idx, raw_log)
+			`INSERT into maker.vat_tune (header_id, ilk, urn, v, w, dink, dart, tx_idx, log_idx, raw_log)
-	   VALUES($1, $2, $3, $4, $5, $6::NUMERIC, $7::NUMERIC, $8, $9)`,
+	   VALUES($1, $2, $3, $4, $5, $6::NUMERIC, $7::NUMERIC, $8, $9, $10)`,
-			headerID, model.Ilk, model.Urn, model.V, model.W, model.Dink, model.Dart, model.TransactionIndex, model.Raw,
+			headerID, model.Ilk, model.Urn, model.V, model.W, model.Dink, model.Dart, model.TransactionIndex, model.LogIndex, model.Raw,
 		)
 		if err != nil {
 			tx.Rollback()

pkg/transformers/vat_tune/repository_test.go

@@ -37,14 +37,14 @@ var _ = Describe("Vat tune repository", func() {
 		BeforeEach(func() {
 			headerID, err = headerRepository.CreateOrUpdateHeader(fakes.FakeHeader)
 			Expect(err).NotTo(HaveOccurred())
-
-			err = vatTuneRepository.Create(headerID, []vat_tune.VatTuneModel{test_data.VatTuneModel})
-			Expect(err).NotTo(HaveOccurred())
 		})
 
 		It("adds a vat tune event", func() {
+			err = vatTuneRepository.Create(headerID, []vat_tune.VatTuneModel{test_data.VatTuneModel})
+			Expect(err).NotTo(HaveOccurred())
+
 			var dbVatTune vat_tune.VatTuneModel
-			err = db.Get(&dbVatTune, `SELECT ilk, urn, v, w, dink, dart, tx_idx, raw_log FROM maker.vat_tune WHERE header_id = $1`, headerID)
+			err = db.Get(&dbVatTune, `SELECT ilk, urn, v, w, dink, dart, tx_idx, log_idx, raw_log FROM maker.vat_tune WHERE header_id = $1`, headerID)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(dbVatTune.Ilk).To(Equal(test_data.VatTuneModel.Ilk))
 			Expect(dbVatTune.Urn).To(Equal(test_data.VatTuneModel.Urn))
@@ -53,10 +53,26 @@ var _ = Describe("Vat tune repository", func() {
 			Expect(dbVatTune.Dink).To(Equal(test_data.VatTuneModel.Dink))
 			Expect(dbVatTune.Dart).To(Equal(test_data.VatTuneModel.Dart))
 			Expect(dbVatTune.TransactionIndex).To(Equal(test_data.VatTuneModel.TransactionIndex))
+			Expect(dbVatTune.LogIndex).To(Equal(test_data.VatTuneModel.LogIndex))
 			Expect(dbVatTune.Raw).To(MatchJSON(test_data.VatTuneModel.Raw))
 		})
 
 		It("marks header as checked for logs", func() {
+			err = vatTuneRepository.Create(headerID, []vat_tune.VatTuneModel{test_data.VatTuneModel})
+			Expect(err).NotTo(HaveOccurred())
+
+			var headerChecked bool
+			err = db.Get(&headerChecked, `SELECT vat_tune_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 = vatTuneRepository.Create(headerID, []vat_tune.VatTuneModel{test_data.VatTuneModel})
+			Expect(err).NotTo(HaveOccurred())
+
 			var headerChecked bool
 			err = db.Get(&headerChecked, `SELECT vat_tune_checked FROM public.checked_headers WHERE header_id = $1`, headerID)
 			Expect(err).NotTo(HaveOccurred())
@@ -64,13 +80,30 @@ var _ = Describe("Vat tune repository", func() {
 		})
 
 		It("does not duplicate pit file vat_tune events", func() {
+			err = vatTuneRepository.Create(headerID, []vat_tune.VatTuneModel{test_data.VatTuneModel})
+			Expect(err).NotTo(HaveOccurred())
+
 			err = vatTuneRepository.Create(headerID, []vat_tune.VatTuneModel{test_data.VatTuneModel})
 
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("pq: duplicate key value violates unique constraint"))
 		})
 
+		It("allows for multiple flop kick events in one transaction if they have different log indexes", func() {
+			err = vatTuneRepository.Create(headerID, []vat_tune.VatTuneModel{test_data.VatTuneModel})
+			Expect(err).NotTo(HaveOccurred())
+
+			newVatTune := test_data.VatTuneModel
+			newVatTune.LogIndex = newVatTune.LogIndex + 1
+			err := vatTuneRepository.Create(headerID, []vat_tune.VatTuneModel{newVatTune})
+
+			Expect(err).NotTo(HaveOccurred())
+		})
+
 		It("removes pit file vat_tune if corresponding header is deleted", func() {
+			err = vatTuneRepository.Create(headerID, []vat_tune.VatTuneModel{test_data.VatTuneModel})
+			Expect(err).NotTo(HaveOccurred())
+
 			_, err = db.Exec(`DELETE FROM headers WHERE id = $1`, headerID)
 
 			Expect(err).NotTo(HaveOccurred())