Add cast error checking to refactored transformers

pkg/transformers/drip_file/ilk/repository.go

@@ -15,6 +15,7 @@
 package ilk
 
 import (
+	"fmt"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 )
@@ -29,9 +30,13 @@ func (repository DripFileIlkRepository) Create(headerID int64, models []interfac
 		return err
 	}
 
-	var ilk DripFileIlkModel
 	for _, model := range models {
-		ilk = model.(DripFileIlkModel)
+		ilk, ok := model.(DripFileIlkModel)
+		if !ok {
+			tx.Rollback()
+			return fmt.Errorf("model of type %T, not %T", model, DripFileIlkModel{})
+		}
+
 		_, err = tx.Exec(
 			`INSERT into maker.drip_file_ilk (header_id, ilk, vow, tax, log_idx, tx_idx, raw_log)
         	VALUES($1, $2, $3, $4::NUMERIC, $5, $6, $7)`,

pkg/transformers/drip_file/ilk/repository_test.go

@@ -89,6 +89,12 @@ var _ = Describe("Drip file ilk repository", func() {
 			Expect(err).To(HaveOccurred())
 			Expect(err).To(MatchError(sql.ErrNoRows))
 		})
+
+		It("Returns an error if model is of wrong type", func() {
+			err = dripFileIlkRepository.Create(headerID, []interface{}{test_data.WrongModel{}})
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("model of type"))
+		})
 	})
 
 	Describe("MarkHeaderChecked", func() {

pkg/transformers/drip_file/repo/repository.go

@@ -15,6 +15,7 @@
 package repo
 
 import (
+	"fmt"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 )
@@ -29,9 +30,13 @@ func (repository DripFileRepoRepository) Create(headerID int64, models []interfa
 		return err
 	}
 
-	var repo DripFileRepoModel
 	for _, model := range models {
-		repo = model.(DripFileRepoModel)
+		repo, ok := model.(DripFileRepoModel)
+		if !ok {
+			tx.Rollback()
+			return fmt.Errorf("model of type %T, not %T", model, DripFileRepoModel{})
+		}
+
 		_, err = tx.Exec(
 			`INSERT into maker.drip_file_repo (header_id, what, data, log_idx, tx_idx, raw_log)
         	VALUES($1, $2, $3::NUMERIC, $4, $5, $6)`,

pkg/transformers/drip_file/repo/repository_test.go

@@ -90,6 +90,12 @@ var _ = Describe("Drip file repo repository", func() {
 			Expect(err).To(HaveOccurred())
 			Expect(err).To(MatchError(sql.ErrNoRows))
 		})
+
+		It("Returns an error if model is of wrong type", func() {
+			err = dripFileRepoRepository.Create(headerID, []interface{}{test_data.WrongModel{}})
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("model of type"))
+		})
 	})
 
 	Describe("MarkHeaderChecked", func() {

pkg/transformers/drip_file/vow/repository.go

@@ -15,6 +15,7 @@
 package vow
 
 import (
+	"fmt"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 )
@@ -29,9 +30,13 @@ func (repository DripFileVowRepository) Create(headerID int64, models []interfac
 		return err
 	}
 
-	var vow DripFileVowModel
 	for _, model := range models {
-		vow = model.(DripFileVowModel)
+		vow, ok := model.(DripFileVowModel)
+		if !ok {
+			tx.Rollback()
+			return fmt.Errorf("model of type %T, not %T", model, DripFileVowModel{})
+		}
+
 		_, err = tx.Exec(
 			`INSERT into maker.drip_file_vow (header_id, what, data, log_idx, tx_idx, raw_log)
         	VALUES($1, $2, $3, $4, $5, $6)`,

pkg/transformers/drip_file/vow/repository_test.go

@@ -90,6 +90,12 @@ var _ = Describe("Drip file vow repository", func() {
 			Expect(err).To(HaveOccurred())
 			Expect(err).To(MatchError(sql.ErrNoRows))
 		})
+
+		It("Returns an error if model is of wrong type", func() {
+			err = dripFileVowRepository.Create(headerID, []interface{}{test_data.WrongModel{}})
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("model of type"))
+		})
 	})
 
 	Describe("MarkHeaderChecked", func() {

pkg/transformers/integration_tests/drip_file_vow.go

@@ -17,8 +17,9 @@ package integration_tests
 import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
 
-	"github.com/vulcanize/vulcanizedb/pkg/transformers/drip_file"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/drip_file/vow"
 	"github.com/vulcanize/vulcanizedb/test_config"
 )
@@ -26,7 +27,7 @@ import (
 var _ = Describe("Drip File Vow Transformer", func() {
 	It("transforms DripFileVow log events", func() {
 		blockNumber := int64(8762197)
-		config := drip_file.DripFileConfig
+		config := vow.DripFileVowConfig
 		config.StartingBlockNumber = blockNumber
 		config.EndingBlockNumber = blockNumber
 
@@ -41,8 +42,13 @@ var _ = Describe("Drip File Vow Transformer", func() {
 		err = persistHeader(rpcClient, db, blockNumber)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := vow.DripFileVowTransformerInitializer{Config: config}
-		transformer := initializer.NewDripFileVowTransformer(db, blockchain)
+		initializer := factories.Transformer{
+			Config:     config,
+			Fetcher:    &shared.Fetcher{},
+			Converter:  &vow.DripFileVowConverter{},
+			Repository: &vow.DripFileVowRepository{},
+		}
+		transformer := initializer.NewTransformer(db, blockchain)
 		err = transformer.Execute()
 		Expect(err).NotTo(HaveOccurred())
 

pkg/transformers/integration_tests/pit_file_debt_ceiling.go

@@ -17,8 +17,9 @@ package integration_tests
 import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
 
-	"github.com/vulcanize/vulcanizedb/pkg/transformers/pit_file"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/pit_file/debt_ceiling"
 	"github.com/vulcanize/vulcanizedb/test_config"
 )
@@ -26,7 +27,7 @@ import (
 var _ = Describe("PitFileDebtCeiling Transformer", func() {
 	It("fetches and transforms a PitFileDebtCeiling event from Kovan chain", func() {
 		blockNumber := int64(8535578)
-		config := pit_file.PitFileConfig
+		config := debt_ceiling.DebtCeilingFileConfig
 		config.StartingBlockNumber = blockNumber
 		config.EndingBlockNumber = blockNumber
 
@@ -41,8 +42,13 @@ var _ = Describe("PitFileDebtCeiling Transformer", func() {
 		err = persistHeader(rpcClient, db, blockNumber)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := debt_ceiling.PitFileDebtCeilingTransformerInitializer{Config: config}
-		transformer := initializer.NewPitFileDebtCeilingTransformer(db, blockchain)
+		initializer := factories.Transformer{
+			Config:     config,
+			Fetcher:    &shared.Fetcher{},
+			Converter:  &debt_ceiling.PitFileDebtCeilingConverter{},
+			Repository: &debt_ceiling.PitFileDebtCeilingRepository{},
+		}
+		transformer := initializer.NewTransformer(db, blockchain)
 		err = transformer.Execute()
 		Expect(err).NotTo(HaveOccurred())
 

pkg/transformers/integration_tests/pit_file_ilk.go

@@ -17,8 +17,9 @@ package integration_tests
 import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
 
-	"github.com/vulcanize/vulcanizedb/pkg/transformers/pit_file"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/pit_file/ilk"
 	"github.com/vulcanize/vulcanizedb/test_config"
 )
@@ -26,7 +27,7 @@ import (
 var _ = Describe("PitFileIlk Transformer", func() {
 	It("fetches and transforms a PitFileIlk event from Kovan chain", func() {
 		blockNumber := int64(9103223)
-		config := pit_file.PitFileConfig
+		config := ilk.IlkFileConfig
 		config.StartingBlockNumber = blockNumber
 		config.EndingBlockNumber = blockNumber
 
@@ -41,8 +42,13 @@ var _ = Describe("PitFileIlk Transformer", func() {
 		err = persistHeader(rpcClient, db, blockNumber)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := ilk.PitFileIlkTransformerInitializer{Config: config}
-		transformer := initializer.NewPitFileIlkTransformer(db, blockchain)
+		initializer := factories.Transformer{
+			Config:     config,
+			Fetcher:    &shared.Fetcher{},
+			Converter:  &ilk.PitFileIlkConverter{},
+			Repository: &ilk.PitFileIlkRepository{},
+		}
+		transformer := initializer.NewTransformer(db, blockchain)
 		err = transformer.Execute()
 		Expect(err).NotTo(HaveOccurred())
 

pkg/transformers/integration_tests/pit_file_stability_fee.go

@@ -17,8 +17,9 @@ package integration_tests
 import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
 
-	"github.com/vulcanize/vulcanizedb/pkg/transformers/pit_file"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/pit_file/stability_fee"
 	"github.com/vulcanize/vulcanizedb/test_config"
 )
@@ -26,7 +27,7 @@ import (
 var _ = Describe("PitFileStabilityFee Transformer", func() {
 	It("fetches and transforms a PitFileStabilityFee event from Kovan chain", func() {
 		blockNumber := int64(8535544)
-		config := pit_file.PitFileConfig
+		config := stability_fee.StabilityFeeFileConfig
 		config.StartingBlockNumber = blockNumber
 		config.EndingBlockNumber = blockNumber
 
@@ -41,8 +42,13 @@ var _ = Describe("PitFileStabilityFee Transformer", func() {
 		err = persistHeader(rpcClient, db, blockNumber)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := stability_fee.PitFileStabilityFeeTransformerInitializer{Config: config}
-		transformer := initializer.NewPitFileStabilityFeeTransformer(db, blockchain)
+		initializer := factories.Transformer{
+			Config:     config,
+			Fetcher:    &shared.Fetcher{},
+			Converter:  &stability_fee.PitFileStabilityFeeConverter{},
+			Repository: &stability_fee.PitFileStabilityFeeRepository{},
+		}
+		transformer := initializer.NewTransformer(db, blockchain)
 		err = transformer.Execute()
 		Expect(err).NotTo(HaveOccurred())
 

pkg/transformers/integration_tests/tend.go

@@ -17,6 +17,8 @@ package integration_tests
 import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
 
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/tend"
 	"github.com/vulcanize/vulcanizedb/test_config"
@@ -40,8 +42,13 @@ var _ = Describe("Tend Transformer", func() {
 		err = persistHeader(rpcClient, db, blockNumber)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := tend.TendTransformerInitializer{Config: config}
-		transformer := initializer.NewTendTransformer(db, blockchain)
+		initializer := factories.Transformer{
+			Config:     config,
+			Fetcher:    &shared.Fetcher{},
+			Converter:  &tend.TendConverter{},
+			Repository: &tend.TendRepository{},
+		}
+		transformer := initializer.NewTransformer(db, blockchain)
 		err = transformer.Execute()
 		Expect(err).NotTo(HaveOccurred())
 
@@ -74,8 +81,13 @@ var _ = Describe("Tend Transformer", func() {
 		err = persistHeader(rpcClient, db, blockNumber)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := tend.TendTransformerInitializer{Config: config}
-		transformer := initializer.NewTendTransformer(db, blockchain)
+		initializer := factories.Transformer{
+			Config:     config,
+			Fetcher:    &shared.Fetcher{},
+			Converter:  &tend.TendConverter{},
+			Repository: &tend.TendRepository{},
+		}
+		transformer := initializer.NewTransformer(db, blockchain)
 		err = transformer.Execute()
 		Expect(err).NotTo(HaveOccurred())
 

pkg/transformers/integration_tests/vat_init.go

@@ -17,6 +17,8 @@ package integration_tests
 import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
 
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/vat_init"
 	"github.com/vulcanize/vulcanizedb/test_config"
@@ -40,8 +42,13 @@ var _ = Describe("VatInit Transformer", func() {
 		err = persistHeader(rpcClient, db, blockNumber)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := vat_init.VatInitTransformerInitializer{Config: config}
-		transformer := initializer.NewVatInitTransformer(db, blockchain)
+		initializer := factories.Transformer{
+			Config:     config,
+			Fetcher:    &shared.Fetcher{},
+			Converter:  &vat_init.VatInitConverter{},
+			Repository: &vat_init.VatInitRepository{},
+		}
+		transformer := initializer.NewTransformer(db, blockchain)
 		err = transformer.Execute()
 		Expect(err).NotTo(HaveOccurred())
 

pkg/transformers/integration_tests/vat_move.go

@@ -18,6 +18,8 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
 
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/vat_move"
 	"github.com/vulcanize/vulcanizedb/test_config"
@@ -41,8 +43,13 @@ var _ = Describe("VatMove Transformer", func() {
 		err = persistHeader(rpcClient, db, blockNumber)
 		Expect(err).NotTo(HaveOccurred())
 
-		initializer := vat_move.VatMoveTransformerInitializer{Config: config}
-		transformer := initializer.NewVatMoveTransformer(db, blockchain)
+		initializer := factories.Transformer{
+			Config:     config,
+			Fetcher:    &shared.Fetcher{},
+			Converter:  &vat_move.VatMoveConverter{},
+			Repository: &vat_move.VatMoveRepository{},
+		}
+		transformer := initializer.NewTransformer(db, blockchain)
 		err = transformer.Execute()
 		Expect(err).NotTo(HaveOccurred())
 

pkg/transformers/pit_file/debt_ceiling/repository.go

@@ -15,6 +15,7 @@
 package debt_ceiling
 
 import (
+	"fmt"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 )
@@ -29,9 +30,13 @@ func (repository PitFileDebtCeilingRepository) Create(headerID int64, models []i
 		return err
 	}
 
-	var pitFileDC PitFileDebtCeilingModel
 	for _, model := range models {
-		pitFileDC = model.(PitFileDebtCeilingModel)
+		pitFileDC, ok := model.(PitFileDebtCeilingModel)
+		if !ok {
+			tx.Rollback()
+			return fmt.Errorf("model of type %T, not %T", model, PitFileDebtCeilingModel{})
+		}
+
 		_, err = tx.Exec(
 			`INSERT into maker.pit_file_debt_ceiling (header_id, what, data, tx_idx, raw_log)
         VALUES($1, $2, $3::NUMERIC, $4, $5)`,

pkg/transformers/pit_file/debt_ceiling/repository_test.go

@@ -89,6 +89,12 @@ var _ = Describe("Pit file debt ceiling repository", func() {
 			Expect(err).To(HaveOccurred())
 			Expect(err).To(MatchError(sql.ErrNoRows))
 		})
+
+		It("Returns an error if model is of wrong type", func() {
+			err = pitFileDebtCeilingRepository.Create(headerID, []interface{}{test_data.WrongModel{}})
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("model of type"))
+		})
 	})
 
 	Describe("MarkHeaderChecked", func() {

pkg/transformers/pit_file/ilk/repository.go

@@ -15,6 +15,7 @@
 package ilk
 
 import (
+	"fmt"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 )
@@ -29,9 +30,13 @@ func (repository PitFileIlkRepository) Create(headerID int64, models []interface
 		return err
 	}
 
-	var pitFileIlk PitFileIlkModel
 	for _, model := range models {
-		pitFileIlk = model.(PitFileIlkModel)
+		pitFileIlk, ok := model.(PitFileIlkModel)
+		if !ok {
+			tx.Rollback()
+			return fmt.Errorf("model of type %T, not %T", model, PitFileIlkModel{})
+		}
+
 		_, err = tx.Exec(
 			`INSERT into maker.pit_file_ilk (header_id, ilk, what, data, tx_idx, raw_log)
         VALUES($1, $2, $3, $4::NUMERIC, $5, $6)`,

pkg/transformers/pit_file/ilk/repository_test.go

@@ -90,6 +90,12 @@ var _ = Describe("Pit file ilk repository", func() {
 			Expect(err).To(HaveOccurred())
 			Expect(err).To(MatchError(sql.ErrNoRows))
 		})
+
+		It("Returns an error if model is of wrong type", func() {
+			err = pitFileRepository.Create(headerID, []interface{}{test_data.WrongModel{}})
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("model of type"))
+		})
 	})
 
 	Describe("MarkHeaderChecked", func() {

pkg/transformers/pit_file/stability_fee/repository.go

@@ -15,6 +15,7 @@
 package stability_fee
 
 import (
+	"fmt"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 )
@@ -29,9 +30,13 @@ func (repository PitFileStabilityFeeRepository) Create(headerID int64, models []
 		return err
 	}
 
-	var pitFileSF PitFileStabilityFeeModel
 	for _, model := range models {
-		pitFileSF = model.(PitFileStabilityFeeModel)
+		pitFileSF, ok := model.(PitFileStabilityFeeModel)
+		if !ok {
+			tx.Rollback()
+			return fmt.Errorf("model of type %T, not %T", model, PitFileStabilityFeeModel{})
+		}
+
 		_, err = tx.Exec(
 			`INSERT into maker.pit_file_stability_fee (header_id, what, data, tx_idx, raw_log)
         VALUES($1, $2, $3, $4, $5)`,

pkg/transformers/pit_file/stability_fee/repository_test.go

@@ -89,6 +89,12 @@ var _ = Describe("Pit file stability fee repository", func() {
 			Expect(err).To(HaveOccurred())
 			Expect(err).To(MatchError(sql.ErrNoRows))
 		})
+
+		It("Returns an error if model is of wrong type", func() {
+			err = pitFileStabilityFeeRepository.Create(headerID, []interface{}{test_data.WrongModel{}})
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("model of type"))
+		})
 	})
 
 	Describe("MarkHeaderChecked", func() {

pkg/transformers/tend/repository.go

@@ -15,6 +15,7 @@
 package tend
 
 import (
+	"fmt"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 )
@@ -28,9 +29,14 @@ func (repository TendRepository) Create(headerId int64, models []interface{}) er
 	if err != nil {
 		return err
 	}
-	var tend TendModel
+
 	for _, model := range models {
-		tend = model.(TendModel)
+		tend, ok := model.(TendModel)
+		if !ok {
+			tx.Rollback()
+			return fmt.Errorf("model of type %T, not %T", model, TendModel{})
+		}
+
 		_, err = tx.Exec(
 			`INSERT into maker.tend (header_id, bid_id, lot, bid, guy, tic, tx_idx, raw_log)
         	VALUES($1, $2, $3, $4, $5, $6, $7, $8)`,

pkg/transformers/tend/repository_test.go

@@ -98,6 +98,12 @@ var _ = Describe("TendRepository", func() {
 			Expect(err).NotTo(HaveOccurred())
 			Expect(count).To(Equal(0))
 		})
+
+		It("Returns an error if model is of wrong type", func() {
+			err = tendRepository.Create(headerId, []interface{}{test_data.WrongModel{}})
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("model of type"))
+		})
 	})
 
 	Describe("MarkHeaderChecked", func() {

pkg/transformers/test_data/wrong_model.go

@@ -0,0 +1,17 @@
+// Copyright 2018 Vulcanize
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package test_data
+
+type WrongModel struct{}

pkg/transformers/vat_init/repository.go

@@ -15,6 +15,7 @@
 package vat_init
 
 import (
+	"fmt"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 	"log"
@@ -29,9 +30,14 @@ func (repository VatInitRepository) Create(headerID int64, models []interface{})
 	if err != nil {
 		return err
 	}
-	var vatInit VatInitModel
+
 	for _, model := range models {
-		vatInit = model.(VatInitModel)
+		vatInit, ok := model.(VatInitModel)
+		if !ok {
+			tx.Rollback()
+			return fmt.Errorf("model of type %T, not %T", model, VatInitModel{})
+		}
+
 		log.Printf("VatInit model: %v", vatInit)
 		_, err = tx.Exec(
 			`INSERT INTO maker.vat_init (header_id, ilk, tx_idx, raw_log)

pkg/transformers/vat_init/repository_test.go

@@ -87,6 +87,12 @@ var _ = Describe("Vat init repository", func() {
 			Expect(err).NotTo(HaveOccurred())
 			Expect(headerChecked).To(BeTrue())
 		})
+
+		It("Returns an error if model is of wrong type", func() {
+			err = vatInitRepository.Create(headerID, []interface{}{test_data.WrongModel{}})
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("model of type"))
+		})
 	})
 
 	Describe("MarkHeaderChecked", func() {

pkg/transformers/vat_move/repository.go

@@ -15,6 +15,7 @@
 package vat_move
 
 import (
+	"fmt"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 )
@@ -29,9 +30,13 @@ func (repository VatMoveRepository) Create(headerID int64, models []interface{})
 		return err
 	}
 
-	var vatMove VatMoveModel
 	for _, model := range models {
-		vatMove = model.(VatMoveModel)
+		vatMove, ok := model.(VatMoveModel)
+		if !ok {
+			tx.Rollback()
+			return fmt.Errorf("model of type %T, not %T", model, VatMoveModel{})
+		}
+
 		_, err = tx.Exec(
 			`INSERT INTO maker.vat_move (header_id, src, dst, rad, tx_idx, raw_log)
         	VALUES ($1, $2, $3, $4::NUMERIC, $5, $6)`,

pkg/transformers/vat_move/repository_test.go

@@ -16,7 +16,6 @@ package vat_move_test
 
 import (
 	"database/sql"
-
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
@@ -85,6 +84,12 @@ var _ = Describe("Vat Move", func() {
 			Expect(err).To(HaveOccurred())
 			Expect(err).To(MatchError(sql.ErrNoRows))
 		})
+
+		It("Returns an error if model is of wrong type", func() {
+			err = vatMoveRepository.Create(headerID, []interface{}{test_data.WrongModel{}})
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("model of type"))
+		})
 	})
 
 	Describe("MissingHeaders", func() {