Deouple ETL & Bridge Processors

indexer/cli/cli.go

@@ -40,7 +40,7 @@ func runIndexer(ctx *cli.Context) error {
 		return err
 	}
 
-	indexer, err := indexer.NewIndexer(cfg.Chain, cfg.RPCs, db, logger)
+	indexer, err := indexer.NewIndexer(logger, cfg.Chain, cfg.RPCs, db)
 	if err != nil {
 		return err
 	}

indexer/config/config.go

@@ -3,7 +3,6 @@ package config
 import (
 	"fmt"
 	"os"
-	"reflect"
 
 	"github.com/BurntSushi/toml"
 	"github.com/ethereum/go-ethereum/common"
@@ -23,31 +22,18 @@ type Config struct {
 
 // fetch this via onchain config from RPCsConfig and remove from config in future
 type L1Contracts struct {
-	OptimismPortal         common.Address `toml:"optimism-portal"`
-	L2OutputOracle         common.Address `toml:"l2-output-oracle"`
-	L1CrossDomainMessenger common.Address `toml:"l1-cross-domain-messenger"`
-	L1StandardBridge       common.Address `toml:"l1-standard-bridge"`
-	L1ERC721Bridge         common.Address `toml:"l1-erc721-bridge"`
+	OptimismPortalProxy         common.Address `toml:"optimism-portal"`
+	L2OutputOracleProxy         common.Address `toml:"l2-output-oracle"`
+	L1CrossDomainMessengerProxy common.Address `toml:"l1-cross-domain-messenger"`
+	L1StandardBridgeProxy       common.Address `toml:"l1-standard-bridge"`
 
-	// Some more contracts -- ProxyAdmin, SystemConfig, etcc
+	// Some more contracts -- L1ERC721Bridge, ProxyAdmin, SystemConfig, etc
 	// Ignore the auxiliary contracts?
 
 	// Legacy contracts? We'll add this in to index the legacy chain.
 	// Remove afterwards?
 }
 
-func (c L1Contracts) ToSlice() []common.Address {
-	fields := reflect.VisibleFields(reflect.TypeOf(c))
-	v := reflect.ValueOf(c)
-
-	contracts := make([]common.Address, len(fields))
-	for i, field := range fields {
-		contracts[i] = (v.FieldByName(field.Name).Interface()).(common.Address)
-	}
-
-	return contracts
-}
-
 // ChainConfig configures of the chain being indexed
 type ChainConfig struct {
 	// Configure known chains with the l2 chain id

indexer/config/config_test.go

@@ -54,11 +54,10 @@ func TestLoadConfig(t *testing.T) {
 	require.NoError(t, err)
 
 	require.Equal(t, conf.Chain.Preset, 420)
-	require.Equal(t, conf.Chain.L1Contracts.OptimismPortal.String(), presetL1Contracts[420].OptimismPortal.String())
-	require.Equal(t, conf.Chain.L1Contracts.L1CrossDomainMessenger.String(), presetL1Contracts[420].L1CrossDomainMessenger.String())
-	require.Equal(t, conf.Chain.L1Contracts.L1ERC721Bridge.String(), presetL1Contracts[420].L1ERC721Bridge.String())
-	require.Equal(t, conf.Chain.L1Contracts.L1StandardBridge.String(), presetL1Contracts[420].L1StandardBridge.String())
-	require.Equal(t, conf.Chain.L1Contracts.L2OutputOracle.String(), presetL1Contracts[420].L2OutputOracle.String())
+	require.Equal(t, conf.Chain.L1Contracts.OptimismPortalProxy.String(), presetL1Contracts[420].OptimismPortalProxy.String())
+	require.Equal(t, conf.Chain.L1Contracts.L1CrossDomainMessengerProxy.String(), presetL1Contracts[420].L1CrossDomainMessengerProxy.String())
+	require.Equal(t, conf.Chain.L1Contracts.L1StandardBridgeProxy.String(), presetL1Contracts[420].L1StandardBridgeProxy.String())
+	require.Equal(t, conf.Chain.L1Contracts.L2OutputOracleProxy.String(), presetL1Contracts[420].L2OutputOracleProxy.String())
 	require.Equal(t, conf.RPCs.L1RPC, "https://l1.example.com")
 	require.Equal(t, conf.RPCs.L2RPC, "https://l2.example.com")
 	require.Equal(t, conf.DB.Host, "127.0.0.1")
@@ -104,11 +103,10 @@ func TestLoadConfig_WithoutPreset(t *testing.T) {
 	conf, err := LoadConfig(logger, tmpfile.Name())
 	require.NoError(t, err)
 
-	require.Equal(t, conf.Chain.L1Contracts.OptimismPortal.String(), common.HexToAddress("0x4205Fc579115071764c7423A4f12eDde41f106Ed").String())
-	require.Equal(t, conf.Chain.L1Contracts.L2OutputOracle.String(), common.HexToAddress("0x42097868233d1aa22e815a266982f2cf17685a27").String())
-	require.Equal(t, conf.Chain.L1Contracts.L1CrossDomainMessenger.String(), common.HexToAddress("0x420ce71c97B33Cc4729CF772ae268934F7ab5fA1").String())
-	require.Equal(t, conf.Chain.L1Contracts.L1StandardBridge.String(), common.HexToAddress("0x4209fc46f92E8a1c0deC1b1747d010903E884bE1").String())
-	require.Equal(t, conf.Chain.L1Contracts.L1ERC721Bridge.String(), common.HexToAddress("0x420749f83b81B301cAb5f48EB8516B986DAef23D").String())
+	require.Equal(t, conf.Chain.L1Contracts.OptimismPortalProxy.String(), common.HexToAddress("0x4205Fc579115071764c7423A4f12eDde41f106Ed").String())
+	require.Equal(t, conf.Chain.L1Contracts.L2OutputOracleProxy.String(), common.HexToAddress("0x42097868233d1aa22e815a266982f2cf17685a27").String())
+	require.Equal(t, conf.Chain.L1Contracts.L1CrossDomainMessengerProxy.String(), common.HexToAddress("0x420ce71c97B33Cc4729CF772ae268934F7ab5fA1").String())
+	require.Equal(t, conf.Chain.L1Contracts.L1StandardBridgeProxy.String(), common.HexToAddress("0x4209fc46f92E8a1c0deC1b1747d010903E884bE1").String())
 	require.Equal(t, conf.Chain.Preset, 0)
 }
 

indexer/config/presets.go

@@ -10,54 +10,44 @@ import (
 var presetL1Contracts = map[int]L1Contracts{
 	// OP Mainnet
 	10: {
-		OptimismPortal:         common.HexToAddress("0xbEb5Fc579115071764c7423A4f12eDde41f106Ed"),
-		L2OutputOracle:         common.HexToAddress("0xdfe97868233d1aa22e815a266982f2cf17685a27"),
-		L1CrossDomainMessenger: common.HexToAddress("0x25ace71c97B33Cc4729CF772ae268934F7ab5fA1"),
-		L1StandardBridge:       common.HexToAddress("0x99C9fc46f92E8a1c0deC1b1747d010903E884bE1"),
-		L1ERC721Bridge:         common.HexToAddress("0x5a7749f83b81B301cAb5f48EB8516B986DAef23D"),
+		OptimismPortalProxy:         common.HexToAddress("0xbEb5Fc579115071764c7423A4f12eDde41f106Ed"),
+		L2OutputOracleProxy:         common.HexToAddress("0xdfe97868233d1aa22e815a266982f2cf17685a27"),
+		L1CrossDomainMessengerProxy: common.HexToAddress("0x25ace71c97B33Cc4729CF772ae268934F7ab5fA1"),
+		L1StandardBridgeProxy:       common.HexToAddress("0x99C9fc46f92E8a1c0deC1b1747d010903E884bE1"),
 	},
 	// OP Goerli
 	420: {
-		OptimismPortal:         common.HexToAddress("0x5b47E1A08Ea6d985D6649300584e6722Ec4B1383"),
-		L2OutputOracle:         common.HexToAddress("0xE6Dfba0953616Bacab0c9A8ecb3a9BBa77FC15c0"),
-		L1CrossDomainMessenger: common.HexToAddress("0x5086d1eEF304eb5284A0f6720f79403b4e9bE294"),
-		L1StandardBridge:       common.HexToAddress("0x636Af16bf2f682dD3109e60102b8E1A089FedAa8"),
-		L1ERC721Bridge:         common.HexToAddress("0x8DD330DdE8D9898d43b4dc840Da27A07dF91b3c9"),
+		OptimismPortalProxy:         common.HexToAddress("0x5b47E1A08Ea6d985D6649300584e6722Ec4B1383"),
+		L2OutputOracleProxy:         common.HexToAddress("0xE6Dfba0953616Bacab0c9A8ecb3a9BBa77FC15c0"),
+		L1CrossDomainMessengerProxy: common.HexToAddress("0x5086d1eEF304eb5284A0f6720f79403b4e9bE294"),
+		L1StandardBridgeProxy:       common.HexToAddress("0x636Af16bf2f682dD3109e60102b8E1A089FedAa8"),
 	},
 	// Base Mainnet
 	8453: {
-		OptimismPortal:         common.HexToAddress("0x49048044D57e1C92A77f79988d21Fa8fAF74E97e"),
-		L2OutputOracle:         common.HexToAddress("0x56315b90c40730925ec5485cf004d835058518A0"),
-		L1CrossDomainMessenger: common.HexToAddress("0x866E82a600A1414e583f7F13623F1aC5d58b0Afa"),
-		L1StandardBridge:       common.HexToAddress("0x3154Cf16ccdb4C6d922629664174b904d80F2C35"),
-		// FIXME update this to the correct address
-		L1ERC721Bridge: common.HexToAddress("0x0000000000000000000000000000000000000000"),
+		OptimismPortalProxy:         common.HexToAddress("0x49048044D57e1C92A77f79988d21Fa8fAF74E97e"),
+		L2OutputOracleProxy:         common.HexToAddress("0x56315b90c40730925ec5485cf004d835058518A0"),
+		L1CrossDomainMessengerProxy: common.HexToAddress("0x866E82a600A1414e583f7F13623F1aC5d58b0Afa"),
+		L1StandardBridgeProxy:       common.HexToAddress("0x3154Cf16ccdb4C6d922629664174b904d80F2C35"),
 	},
 	// Base Goerli
 	84531: {
-		OptimismPortal:         common.HexToAddress("0xe93c8cD0D409341205A592f8c4Ac1A5fe5585cfA"),
-		L2OutputOracle:         common.HexToAddress("0x2A35891ff30313CcFa6CE88dcf3858bb075A2298"),
-		L1CrossDomainMessenger: common.HexToAddress("0x8e5693140eA606bcEB98761d9beB1BC87383706D"),
-		L1StandardBridge:       common.HexToAddress("0xfA6D8Ee5BE770F84FC001D098C4bD604Fe01284a"),
-		// FIXME update this to the correct address
-		L1ERC721Bridge: common.HexToAddress("0x0000000000000000000000000000000000000000"),
+		OptimismPortalProxy:         common.HexToAddress("0xe93c8cD0D409341205A592f8c4Ac1A5fe5585cfA"),
+		L2OutputOracleProxy:         common.HexToAddress("0x2A35891ff30313CcFa6CE88dcf3858bb075A2298"),
+		L1CrossDomainMessengerProxy: common.HexToAddress("0x8e5693140eA606bcEB98761d9beB1BC87383706D"),
+		L1StandardBridgeProxy:       common.HexToAddress("0xfA6D8Ee5BE770F84FC001D098C4bD604Fe01284a"),
 	},
 	// Zora mainnet
 	7777777: {
-		OptimismPortal:         common.HexToAddress("0x1a0ad011913A150f69f6A19DF447A0CfD9551054"),
-		L2OutputOracle:         common.HexToAddress("0x9E6204F750cD866b299594e2aC9eA824E2e5f95c"),
-		L1CrossDomainMessenger: common.HexToAddress("0xdC40a14d9abd6F410226f1E6de71aE03441ca506"),
-		L1StandardBridge:       common.HexToAddress("0x3e2Ea9B92B7E48A52296fD261dc26fd995284631"),
-		// FIXME update this to the correct address
-		L1ERC721Bridge: common.HexToAddress("0x0000000000000000000000000000000000000000"),
+		OptimismPortalProxy:         common.HexToAddress("0x1a0ad011913A150f69f6A19DF447A0CfD9551054"),
+		L2OutputOracleProxy:         common.HexToAddress("0x9E6204F750cD866b299594e2aC9eA824E2e5f95c"),
+		L1CrossDomainMessengerProxy: common.HexToAddress("0xdC40a14d9abd6F410226f1E6de71aE03441ca506"),
+		L1StandardBridgeProxy:       common.HexToAddress("0x3e2Ea9B92B7E48A52296fD261dc26fd995284631"),
 	},
 	// Zora goerli
 	999: {
-		OptimismPortal:         common.HexToAddress("0xDb9F51790365e7dc196e7D072728df39Be958ACe"),
-		L2OutputOracle:         common.HexToAddress("0xdD292C9eEd00f6A32Ff5245d0BCd7f2a15f24e00"),
-		L1CrossDomainMessenger: common.HexToAddress("0xD87342e16352D33170557A7dA1e5fB966a60FafC"),
-		L1StandardBridge:       common.HexToAddress("0x7CC09AC2452D6555d5e0C213Ab9E2d44eFbFc956"),
-		// FIXME update this to the correct address
-		L1ERC721Bridge: common.HexToAddress("0x0000000000000000000000000000000000000000"),
+		OptimismPortalProxy:         common.HexToAddress("0xDb9F51790365e7dc196e7D072728df39Be958ACe"),
+		L2OutputOracleProxy:         common.HexToAddress("0xdD292C9eEd00f6A32Ff5245d0BCd7f2a15f24e00"),
+		L1CrossDomainMessengerProxy: common.HexToAddress("0xD87342e16352D33170557A7dA1e5fB966a60FafC"),
+		L1StandardBridgeProxy:       common.HexToAddress("0x7CC09AC2452D6555d5e0C213Ab9E2d44eFbFc956"),
 	},
 }

indexer/database/blocks.go

@@ -38,11 +38,11 @@ func BlockHeaderFromHeader(header *types.Header) BlockHeader {
 }
 
 type L1BlockHeader struct {
-	BlockHeader
+	BlockHeader `gorm:"embedded"`
 }
 
 type L2BlockHeader struct {
-	BlockHeader
+	BlockHeader `gorm:"embedded"`
 }
 
 type LegacyStateBatch struct {
@@ -65,14 +65,18 @@ type OutputProposal struct {
 }
 
 type BlocksView interface {
-	L1BlockHeader(*big.Int) (*L1BlockHeader, error)
-	LatestL1BlockHeader() (*L1BlockHeader, error)
+	L1BlockHeader(common.Hash) (*L1BlockHeader, error)
+	L1BlockHeaderWithFilter(BlockHeader) (*L1BlockHeader, error)
+	L1LatestBlockHeader() (*L1BlockHeader, error)
+
+	L2BlockHeader(common.Hash) (*L2BlockHeader, error)
+	L2BlockHeaderWithFilter(BlockHeader) (*L2BlockHeader, error)
+	L2LatestBlockHeader() (*L2BlockHeader, error)
 
 	LatestCheckpointedOutput() (*OutputProposal, error)
 	OutputProposal(index *big.Int) (*OutputProposal, error)
 
-	L2BlockHeader(*big.Int) (*L2BlockHeader, error)
-	LatestL2BlockHeader() (*L2BlockHeader, error)
+	LatestEpoch() (*Epoch, error)
 }
 
 type BlocksDB interface {
@@ -114,21 +118,24 @@ func (db *blocksDB) StoreOutputProposals(outputs []*OutputProposal) error {
 	return result.Error
 }
 
-func (db *blocksDB) L1BlockHeader(height *big.Int) (*L1BlockHeader, error) {
+func (db *blocksDB) L1BlockHeader(hash common.Hash) (*L1BlockHeader, error) {
+	return db.L1BlockHeaderWithFilter(BlockHeader{Hash: hash})
+}
+
+func (db *blocksDB) L1BlockHeaderWithFilter(filter BlockHeader) (*L1BlockHeader, error) {
 	var l1Header L1BlockHeader
-	result := db.gorm.Where(&BlockHeader{Number: U256{Int: height}}).Take(&l1Header)
+	result := db.gorm.Where(&filter).Take(&l1Header)
 	if result.Error != nil {
 		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
 			return nil, nil
 		}
-
 		return nil, result.Error
 	}
 
 	return &l1Header, nil
 }
 
-func (db *blocksDB) LatestL1BlockHeader() (*L1BlockHeader, error) {
+func (db *blocksDB) L1LatestBlockHeader() (*L1BlockHeader, error) {
 	var l1Header L1BlockHeader
 	result := db.gorm.Order("number DESC").Take(&l1Header)
 	if result.Error != nil {
@@ -177,31 +184,66 @@ func (db *blocksDB) StoreL2BlockHeaders(headers []*L2BlockHeader) error {
 	return result.Error
 }
 
-func (db *blocksDB) L2BlockHeader(height *big.Int) (*L2BlockHeader, error) {
+func (db *blocksDB) L2BlockHeader(hash common.Hash) (*L2BlockHeader, error) {
+	return db.L2BlockHeaderWithFilter(BlockHeader{Hash: hash})
+}
+
+func (db *blocksDB) L2BlockHeaderWithFilter(filter BlockHeader) (*L2BlockHeader, error) {
 	var l2Header L2BlockHeader
-	result := db.gorm.Where(&BlockHeader{Number: U256{Int: height}}).Take(&l2Header)
+	result := db.gorm.Where(&filter).Take(&l2Header)
 	if result.Error != nil {
 		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
 			return nil, nil
 		}
-
 		return nil, result.Error
 	}
 
 	return &l2Header, nil
 }
 
-func (db *blocksDB) LatestL2BlockHeader() (*L2BlockHeader, error) {
+func (db *blocksDB) L2LatestBlockHeader() (*L2BlockHeader, error) {
 	var l2Header L2BlockHeader
 	result := db.gorm.Order("number DESC").Take(&l2Header)
 	if result.Error != nil {
 		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
 			return nil, nil
 		}
-
 		return nil, result.Error
 	}
 
 	result.Logger.Info(context.Background(), "number ", l2Header.Number)
 	return &l2Header, nil
 }
+
+// Auxiliary Methods on both L1 & L2
+
+type Epoch struct {
+	L1BlockHeader L1BlockHeader `gorm:"embedded"`
+	L2BlockHeader L2BlockHeader `gorm:"embedded"`
+}
+
+// LatestEpoch return the latest epoch, seen on  L1 & L2. In other words
+// this returns the latest indexed L1 block that has a corresponding
+// indexed L2 block with a matching L1Origin (equal timestamps).
+//
+// For more, see the protocol spec:
+//   - https://github.com/ethereum-optimism/optimism/blob/develop/specs/derivation.md
+func (db *blocksDB) LatestEpoch() (*Epoch, error) {
+	// Since L1 blocks occur less frequently than L2, we do a INNER JOIN from L1 on
+	// L2 for a faster query. Per the protocol, the L2 block that starts a new epoch
+	// will have a matching timestamp with the L1 origin.
+	query := db.gorm.Table("l1_block_headers").Order("l1_block_headers.timestamp DESC")
+	query = query.Joins("INNER JOIN l2_block_headers ON l1_block_headers.timestamp = l2_block_headers.timestamp")
+	query = query.Select("*")
+
+	var epoch Epoch
+	result := query.Take(&epoch)
+	if result.Error != nil {
+		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+			return nil, nil
+		}
+		return nil, result.Error
+	}
+
+	return &epoch, nil
+}

indexer/database/bridge_transfers.go

@@ -32,7 +32,6 @@ type BridgeTransfer struct {
 
 type L1BridgeDeposit struct {
 	BridgeTransfer        `gorm:"embedded"`
-
 	TransactionSourceHash common.Hash `gorm:"primaryKey;serializer:json"`
 }
 
@@ -45,7 +44,6 @@ type L1BridgeDepositWithTransactionHashes struct {
 
 type L2BridgeWithdrawal struct {
 	BridgeTransfer            `gorm:"embedded"`
-
 	TransactionWithdrawalHash common.Hash `gorm:"primaryKey;serializer:json"`
 }
 

indexer/database/contract_events.go

@@ -2,6 +2,7 @@ package database
 
 import (
 	"errors"
+	"fmt"
 	"math/big"
 
 	"gorm.io/gorm"
@@ -74,12 +75,16 @@ type L2ContractEvent struct {
 
 type ContractEventsView interface {
 	L1ContractEvent(uuid.UUID) (*L1ContractEvent, error)
-	L1ContractEventByTxLogIndex(common.Hash, uint64) (*L1ContractEvent, error)
+	L1ContractEventWithFilter(ContractEvent) (*L1ContractEvent, error)
 	L1ContractEventsWithFilter(ContractEvent, *big.Int, *big.Int) ([]L1ContractEvent, error)
+	L1LatestContractEventWithFilter(ContractEvent) (*L1ContractEvent, error)
 
 	L2ContractEvent(uuid.UUID) (*L2ContractEvent, error)
-	L2ContractEventByTxLogIndex(common.Hash, uint64) (*L2ContractEvent, error)
+	L2ContractEventWithFilter(ContractEvent) (*L2ContractEvent, error)
 	L2ContractEventsWithFilter(ContractEvent, *big.Int, *big.Int) ([]L2ContractEvent, error)
+	L2LatestContractEventWithFilter(ContractEvent) (*L2ContractEvent, error)
+
+	ContractEventsWithFilter(ContractEvent, string, *big.Int, *big.Int) ([]ContractEvent, error)
 }
 
 type ContractEventsDB interface {
@@ -109,27 +114,16 @@ func (db *contractEventsDB) StoreL1ContractEvents(events []*L1ContractEvent) err
 }
 
 func (db *contractEventsDB) L1ContractEvent(uuid uuid.UUID) (*L1ContractEvent, error) {
-	var l1ContractEvent L1ContractEvent
-	result := db.gorm.Where(&ContractEvent{GUID: uuid}).Take(&l1ContractEvent)
-	if result.Error != nil {
-		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
-			return nil, nil
-		}
-
-		return nil, result.Error
-	}
-
-	return &l1ContractEvent, nil
+	return db.L1ContractEventWithFilter(ContractEvent{GUID: uuid})
 }
 
-func (db *contractEventsDB) L1ContractEventByTxLogIndex(txHash common.Hash, logIndex uint64) (*L1ContractEvent, error) {
+func (db *contractEventsDB) L1ContractEventWithFilter(filter ContractEvent) (*L1ContractEvent, error) {
 	var l1ContractEvent L1ContractEvent
-	result := db.gorm.Where(&ContractEvent{TransactionHash: txHash, LogIndex: logIndex}).Take(&l1ContractEvent)
+	result := db.gorm.Where(&filter).Take(&l1ContractEvent)
 	if result.Error != nil {
 		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
 			return nil, nil
 		}
-
 		return nil, result.Error
 	}
 
@@ -140,6 +134,12 @@ func (db *contractEventsDB) L1ContractEventsWithFilter(filter ContractEvent, fro
 	if fromHeight == nil {
 		fromHeight = big.NewInt(0)
 	}
+	if toHeight == nil {
+		return nil, errors.New("end height unspecified")
+	}
+	if fromHeight.Cmp(toHeight) > 0 {
+		return nil, fmt.Errorf("fromHeight %d is greater than toHeight %d", fromHeight, toHeight)
+	}
 
 	query := db.gorm.Table("l1_contract_events").Where(&filter)
 	query = query.Joins("INNER JOIN l1_block_headers ON l1_contract_events.block_hash = l1_block_headers.hash")
@@ -160,6 +160,19 @@ func (db *contractEventsDB) L1ContractEventsWithFilter(filter ContractEvent, fro
 	return events, nil
 }
 
+func (db *contractEventsDB) L1LatestContractEventWithFilter(filter ContractEvent) (*L1ContractEvent, error) {
+	var l1ContractEvent L1ContractEvent
+	result := db.gorm.Where(&filter).Order("timestamp DESC").Take(&l1ContractEvent)
+	if result.Error != nil {
+		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+			return nil, nil
+		}
+		return nil, result.Error
+	}
+
+	return &l1ContractEvent, nil
+}
+
 // L2
 
 func (db *contractEventsDB) StoreL2ContractEvents(events []*L2ContractEvent) error {
@@ -168,27 +181,16 @@ func (db *contractEventsDB) StoreL2ContractEvents(events []*L2ContractEvent) err
 }
 
 func (db *contractEventsDB) L2ContractEvent(uuid uuid.UUID) (*L2ContractEvent, error) {
-	var l2ContractEvent L2ContractEvent
-	result := db.gorm.Where(&ContractEvent{GUID: uuid}).Take(&l2ContractEvent)
-	if result.Error != nil {
-		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
-			return nil, nil
-		}
-
-		return nil, result.Error
-	}
-
-	return &l2ContractEvent, nil
+	return db.L2ContractEventWithFilter(ContractEvent{GUID: uuid})
 }
 
-func (db *contractEventsDB) L2ContractEventByTxLogIndex(txHash common.Hash, logIndex uint64) (*L2ContractEvent, error) {
+func (db *contractEventsDB) L2ContractEventWithFilter(filter ContractEvent) (*L2ContractEvent, error) {
 	var l2ContractEvent L2ContractEvent
-	result := db.gorm.Where(&ContractEvent{TransactionHash: txHash, LogIndex: logIndex}).Take(&l2ContractEvent)
+	result := db.gorm.Where(&filter).Take(&l2ContractEvent)
 	if result.Error != nil {
 		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
 			return nil, nil
 		}
-
 		return nil, result.Error
 	}
 
@@ -199,6 +201,12 @@ func (db *contractEventsDB) L2ContractEventsWithFilter(filter ContractEvent, fro
 	if fromHeight == nil {
 		fromHeight = big.NewInt(0)
 	}
+	if toHeight == nil {
+		return nil, errors.New("end height unspecified")
+	}
+	if fromHeight.Cmp(toHeight) > 0 {
+		return nil, fmt.Errorf("fromHeight %d is greater than toHeight %d", fromHeight, toHeight)
+	}
 
 	query := db.gorm.Table("l2_contract_events").Where(&filter)
 	query = query.Joins("INNER JOIN l2_block_headers ON l2_contract_events.block_hash = l2_block_headers.hash")
@@ -218,3 +226,46 @@ func (db *contractEventsDB) L2ContractEventsWithFilter(filter ContractEvent, fro
 
 	return events, nil
 }
+
+func (db *contractEventsDB) L2LatestContractEventWithFilter(filter ContractEvent) (*L2ContractEvent, error) {
+	var l2ContractEvent L2ContractEvent
+	result := db.gorm.Where(&filter).Order("timestamp DESC").Take(&l2ContractEvent)
+	if result.Error != nil {
+		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+			return nil, nil
+		}
+		return nil, result.Error
+	}
+
+	return &l2ContractEvent, nil
+}
+
+// Auxiliary methods for both L1 and L2
+
+// ContractEventsWithFilter will retrieve contract events within the specified range according to the `chainSelector`.
+func (db *contractEventsDB) ContractEventsWithFilter(filter ContractEvent, chainSelector string, fromHeight, toHeight *big.Int) ([]ContractEvent, error) {
+	switch chainSelector {
+	case "l1":
+		l1Events, err := db.L1ContractEventsWithFilter(filter, fromHeight, toHeight)
+		if err != nil {
+			return nil, err
+		}
+		events := make([]ContractEvent, len(l1Events))
+		for i := range l1Events {
+			events[i] = l1Events[i].ContractEvent
+		}
+		return events, nil
+	case "l2":
+		l2Events, err := db.L2ContractEventsWithFilter(filter, fromHeight, toHeight)
+		if err != nil {
+			return nil, err
+		}
+		events := make([]ContractEvent, len(l2Events))
+		for i := range l2Events {
+			events[i] = l2Events[i].ContractEvent
+		}
+		return events, nil
+	default:
+		return nil, errors.New("expected 'l1' or 'l2' for chain selection")
+	}
+}

indexer/e2e_tests/bridge_messages_e2e_test.go

@@ -32,9 +32,6 @@ func TestE2EBridgeL1CrossDomainMessenger(t *testing.T) {
 	require.NoError(t, err)
 	l1Opts.Value = big.NewInt(params.Ether)
 
-	// Pause the processor to track relayed event
-	testSuite.Indexer.L2Processor.PauseForTest()
-
 	// (1) Send the Message
 	sentMsgTx, err := l1CrossDomainMessenger.SendMessage(l1Opts, aliceAddr, calldata, 100_000)
 	require.NoError(t, err)
@@ -46,7 +43,7 @@ func TestE2EBridgeL1CrossDomainMessenger(t *testing.T) {
 
 	// wait for processor catchup
 	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
-		l1Header := testSuite.Indexer.L1Processor.LatestProcessedHeader()
+		l1Header := testSuite.Indexer.BridgeProcessor.LatestL1Header
 		return l1Header != nil && l1Header.Number.Uint64() >= sentMsgReceipt.BlockNumber.Uint64(), nil
 	}))
 
@@ -70,17 +67,18 @@ func TestE2EBridgeL1CrossDomainMessenger(t *testing.T) {
 	require.ElementsMatch(t, calldata, sentMessage.Tx.Data)
 
 	// (2) Process RelayedMesssage on inclusion
-	require.Nil(t, sentMessage.RelayedMessageEventGUID)
-	testSuite.Indexer.L2Processor.ResumeForTest()
+	//   - We dont assert that `RelayedMessageEventGUID` is nil prior to inclusion since there isn't a
+	//   a straightforward way of pausing/resuming the processors at the right time. The codepath is the
+	//   same for L2->L1 messages which does check for this so we are still covered
 	transaction, err := testSuite.DB.BridgeTransactions.L1TransactionDeposit(sentMessage.TransactionSourceHash)
 	require.NoError(t, err)
 
 	// wait for processor catchup
-	depositReceipt, err := wait.ForReceiptOK(context.Background(), testSuite.L2Client, transaction.L2TransactionHash)
+	l2DepositReceipt, err := wait.ForReceiptOK(context.Background(), testSuite.L2Client, transaction.L2TransactionHash)
 	require.NoError(t, err)
 	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
-		l2Header := testSuite.Indexer.L2Processor.LatestProcessedHeader()
-		return l2Header != nil && l2Header.Number.Uint64() >= depositReceipt.BlockNumber.Uint64(), nil
+		l2Header := testSuite.Indexer.BridgeProcessor.LatestL2Header
+		return l2Header != nil && l2Header.Number.Uint64() >= l2DepositReceipt.BlockNumber.Uint64(), nil
 	}))
 
 	sentMessage, err = testSuite.DB.BridgeMessages.L1BridgeMessage(parsedMessage.MessageHash)
@@ -132,7 +130,7 @@ func TestE2EBridgeL2CrossDomainMessenger(t *testing.T) {
 
 	// wait for processor catchup
 	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
-		l2Header := testSuite.Indexer.L2Processor.LatestProcessedHeader()
+		l2Header := testSuite.Indexer.BridgeProcessor.LatestL2Header
 		return l2Header != nil && l2Header.Number.Uint64() >= sentMsgReceipt.BlockNumber.Uint64(), nil
 	}))
 
@@ -161,7 +159,7 @@ func TestE2EBridgeL2CrossDomainMessenger(t *testing.T) {
 
 	// wait for processor catchup
 	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
-		l1Header := testSuite.Indexer.L1Processor.LatestProcessedHeader()
+		l1Header := testSuite.Indexer.BridgeProcessor.LatestL1Header
 		return l1Header != nil && l1Header.Number.Uint64() >= finalizedReceipt.BlockNumber.Uint64(), nil
 	}))
 

indexer/e2e_tests/bridge_transactions_e2e_test.go

@@ -46,12 +46,13 @@ func TestE2EBridgeTransactionsOptimismPortalDeposits(t *testing.T) {
 
 	// wait for processor catchup
 	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
-		l1Header := testSuite.Indexer.L1Processor.LatestProcessedHeader()
+		l1Header := testSuite.Indexer.BridgeProcessor.LatestL1Header
 		return l1Header != nil && l1Header.Number.Uint64() >= depositReceipt.BlockNumber.Uint64(), nil
 	}))
 
 	deposit, err := testSuite.DB.BridgeTransactions.L1TransactionDeposit(depositInfo.DepositTx.SourceHash)
 	require.NoError(t, err)
+	require.NotNil(t, deposit)
 	require.Equal(t, depositL2TxHash, deposit.L2TransactionHash)
 	require.Equal(t, big.NewInt(100_000), deposit.GasLimit.Int)
 	require.Equal(t, big.NewInt(params.Ether), deposit.Tx.Amount.Int)
@@ -100,7 +101,7 @@ func TestE2EBridgeTransactionsL2ToL1MessagePasserWithdrawal(t *testing.T) {
 
 	// wait for processor catchup
 	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
-		l2Header := testSuite.Indexer.L2Processor.LatestProcessedHeader()
+		l2Header := testSuite.Indexer.BridgeProcessor.LatestL2Header
 		return l2Header != nil && l2Header.Number.Uint64() >= withdrawReceipt.BlockNumber.Uint64(), nil
 	}))
 
@@ -111,6 +112,7 @@ func TestE2EBridgeTransactionsL2ToL1MessagePasserWithdrawal(t *testing.T) {
 
 	withdraw, err := testSuite.DB.BridgeTransactions.L2TransactionWithdrawal(withdrawalHash)
 	require.NoError(t, err)
+	require.NotNil(t, withdraw)
 	require.Equal(t, msgPassed.Nonce.Uint64(), withdraw.Nonce.Int.Uint64())
 	require.Equal(t, big.NewInt(100_000), withdraw.GasLimit.Int)
 	require.Equal(t, big.NewInt(params.Ether), withdraw.Tx.Amount.Int)
@@ -129,7 +131,7 @@ func TestE2EBridgeTransactionsL2ToL1MessagePasserWithdrawal(t *testing.T) {
 
 	withdrawParams, proveReceipt := op_e2e.ProveWithdrawal(t, *testSuite.OpCfg, testSuite.L1Client, testSuite.OpSys.Nodes["sequencer"], testSuite.OpCfg.Secrets.Alice, withdrawReceipt)
 	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
-		l1Header := testSuite.Indexer.L1Processor.LatestProcessedHeader()
+		l1Header := testSuite.Indexer.BridgeProcessor.LatestL1Header
 		return l1Header != nil && l1Header.Number.Uint64() >= proveReceipt.BlockNumber.Uint64(), nil
 	}))
 
@@ -147,7 +149,7 @@ func TestE2EBridgeTransactionsL2ToL1MessagePasserWithdrawal(t *testing.T) {
 
 	finalizeReceipt := op_e2e.FinalizeWithdrawal(t, *testSuite.OpCfg, testSuite.L1Client, testSuite.OpCfg.Secrets.Alice, proveReceipt, withdrawParams)
 	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
-		l1Header := testSuite.Indexer.L1Processor.LatestProcessedHeader()
+		l1Header := testSuite.Indexer.BridgeProcessor.LatestL1Header
 		return l1Header != nil && l1Header.Number.Uint64() >= finalizeReceipt.BlockNumber.Uint64(), nil
 	}))
 
@@ -189,7 +191,7 @@ func TestE2EBridgeTransactionsL2ToL1MessagePasserFailedWithdrawal(t *testing.T)
 	// Prove&Finalize withdrawal
 	_, finalizeReceipt := op_e2e.ProveAndFinalizeWithdrawal(t, *testSuite.OpCfg, testSuite.L1Client, testSuite.OpSys.Nodes["sequencer"], testSuite.OpCfg.Secrets.Alice, withdrawReceipt)
 	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
-		l1Header := testSuite.Indexer.L1Processor.LatestProcessedHeader()
+		l1Header := testSuite.Indexer.BridgeProcessor.LatestL1Header
 		return l1Header != nil && l1Header.Number.Uint64() >= finalizeReceipt.BlockNumber.Uint64(), nil
 	}))
 

indexer/e2e_tests/bridge_transfers_e2e_test.go

@@ -43,7 +43,7 @@ func TestE2EBridgeTransfersStandardBridgeETHDeposit(t *testing.T) {
 
 	// wait for processor catchup
 	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
-		l1Header := testSuite.Indexer.L1Processor.LatestProcessedHeader()
+		l1Header := testSuite.Indexer.BridgeProcessor.LatestL1Header
 		return l1Header != nil && l1Header.Number.Uint64() >= depositReceipt.BlockNumber.Uint64(), nil
 	}))
 
@@ -73,11 +73,11 @@ func TestE2EBridgeTransfersStandardBridgeETHDeposit(t *testing.T) {
 	require.NotNil(t, deposit.CrossDomainMessageHash)
 
 	// (2) Test Deposit Finalization via CrossDomainMessenger relayed message
-	depositReceipt, err = wait.ForReceiptOK(context.Background(), testSuite.L2Client, types.NewTx(depositInfo.DepositTx).Hash())
+	l2DepositReceipt, err := wait.ForReceiptOK(context.Background(), testSuite.L2Client, types.NewTx(depositInfo.DepositTx).Hash())
 	require.NoError(t, err)
 	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
-		l2Header := testSuite.Indexer.L2Processor.LatestProcessedHeader()
-		return l2Header != nil && l2Header.Number.Uint64() >= depositReceipt.BlockNumber.Uint64(), nil
+		l2Header := testSuite.Indexer.BridgeProcessor.LatestL2Header
+		return l2Header != nil && l2Header.Number.Uint64() >= l2DepositReceipt.BlockNumber.Uint64(), nil
 	}))
 
 	crossDomainBridgeMessage, err := testSuite.DB.BridgeMessages.L1BridgeMessage(*deposit.CrossDomainMessageHash)
@@ -221,7 +221,7 @@ func TestE2EBridgeTransfersOptimismPortalETHReceive(t *testing.T) {
 
 	// wait for processor catchup
 	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
-		l1Header := testSuite.Indexer.L1Processor.LatestProcessedHeader()
+		l1Header := testSuite.Indexer.BridgeProcessor.LatestL1Header
 		return l1Header != nil && l1Header.Number.Uint64() >= portalDepositReceipt.BlockNumber.Uint64(), nil
 	}))
 
@@ -242,11 +242,11 @@ func TestE2EBridgeTransfersOptimismPortalETHReceive(t *testing.T) {
 	require.Nil(t, deposit.CrossDomainMessageHash)
 
 	// (2) Test Deposit Finalization
-	depositReceipt, err := wait.ForReceiptOK(context.Background(), testSuite.L2Client, types.NewTx(depositInfo.DepositTx).Hash())
+	l2DepositReceipt, err := wait.ForReceiptOK(context.Background(), testSuite.L2Client, types.NewTx(depositInfo.DepositTx).Hash())
 	require.NoError(t, err)
 	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
-		l2Header := testSuite.Indexer.L2Processor.LatestProcessedHeader()
-		return l2Header != nil && l2Header.Number.Uint64() >= depositReceipt.BlockNumber.Uint64(), nil
+		l2Header := testSuite.Indexer.BridgeProcessor.LatestL2Header
+		return l2Header != nil && l2Header.Number.Uint64() >= l2DepositReceipt.BlockNumber.Uint64(), nil
 	}))
 
 	// Still nil as the withdrawal did not occur through the standard bridge
@@ -286,7 +286,7 @@ func TestE2EBridgeTransfersStandardBridgeETHWithdrawal(t *testing.T) {
 
 	// wait for processor catchup
 	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
-		l2Header := testSuite.Indexer.L2Processor.LatestProcessedHeader()
+		l2Header := testSuite.Indexer.BridgeProcessor.LatestL2Header
 		return l2Header != nil && l2Header.Number.Uint64() >= withdrawReceipt.BlockNumber.Uint64(), nil
 	}))
 
@@ -324,7 +324,7 @@ func TestE2EBridgeTransfersStandardBridgeETHWithdrawal(t *testing.T) {
 	// wait for processor catchup
 	proveReceipt, finalizeReceipt := op_e2e.ProveAndFinalizeWithdrawal(t, *testSuite.OpCfg, testSuite.L1Client, testSuite.OpSys.Nodes["sequencer"], testSuite.OpCfg.Secrets.Alice, withdrawReceipt)
 	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
-		l1Header := testSuite.Indexer.L1Processor.LatestProcessedHeader()
+		l1Header := testSuite.Indexer.BridgeProcessor.LatestL1Header
 		return l1Header != nil && l1Header.Number.Uint64() >= finalizeReceipt.BlockNumber.Uint64(), nil
 	}))
 
@@ -370,7 +370,7 @@ func TestE2EBridgeTransfersL2ToL1MessagePasserReceive(t *testing.T) {
 
 	// wait for processor catchup
 	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
-		l2Header := testSuite.Indexer.L2Processor.LatestProcessedHeader()
+		l2Header := testSuite.Indexer.BridgeProcessor.LatestL2Header
 		return l2Header != nil && l2Header.Number.Uint64() >= l2ToL1WithdrawReceipt.BlockNumber.Uint64(), nil
 	}))
 
@@ -402,7 +402,7 @@ func TestE2EBridgeTransfersL2ToL1MessagePasserReceive(t *testing.T) {
 	// wait for processor catchup
 	proveReceipt, finalizeReceipt := op_e2e.ProveAndFinalizeWithdrawal(t, *testSuite.OpCfg, testSuite.L1Client, testSuite.OpSys.Nodes["sequencer"], testSuite.OpCfg.Secrets.Alice, l2ToL1WithdrawReceipt)
 	require.NoError(t, wait.For(context.Background(), 500*time.Millisecond, func() (bool, error) {
-		l1Header := testSuite.Indexer.L1Processor.LatestProcessedHeader()
+		l1Header := testSuite.Indexer.BridgeProcessor.LatestL1Header
 		return l1Header != nil && l1Header.Number.Uint64() >= finalizeReceipt.BlockNumber.Uint64(), nil
 	}))
 

indexer/e2e_tests/blocks_e2e_test.go → indexer/e2e_tests/etl_e2e_test.go

@@ -6,43 +6,45 @@ import (
 	"testing"
 	"time"
 
-	"github.com/ethereum-optimism/optimism/indexer/node"
+	"github.com/ethereum-optimism/optimism/indexer/database"
 	"github.com/ethereum-optimism/optimism/op-bindings/bindings"
-	"github.com/ethereum-optimism/optimism/op-bindings/predeploys"
 	"github.com/ethereum-optimism/optimism/op-e2e/e2eutils/wait"
 
 	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/rlp"
 
 	"github.com/stretchr/testify/require"
 )
 
-func TestE2EBlockHeaders(t *testing.T) {
+func TestE2EETL(t *testing.T) {
 	testSuite := createE2ETestSuite(t)
 
 	l2OutputOracle, err := bindings.NewL2OutputOracle(testSuite.OpCfg.L1Deployments.L2OutputOracleProxy, testSuite.L1Client)
 	require.NoError(t, err)
 
-	// wait for at least 10 L2 blocks to be created & posted on L1
+	// wait for at least 10 L2 blocks posted on L1
 	require.NoError(t, wait.For(context.Background(), time.Second, func() (bool, error) {
 		l2Height, err := l2OutputOracle.LatestBlockNumber(&bind.CallOpts{Context: context.Background()})
 		return l2Height != nil && l2Height.Uint64() >= 9, err
 	}))
 
-	// ensure the processors are caught up to this state
+	// ensure we've indexed up to this state
 	l1Height, err := testSuite.L1Client.BlockNumber(context.Background())
 	require.NoError(t, err)
-	require.NoError(t, wait.For(context.Background(), time.Second, func() (bool, error) {
-		l1Header := testSuite.Indexer.L1Processor.LatestProcessedHeader()
-		l2Header := testSuite.Indexer.L2Processor.LatestProcessedHeader()
-		return (l1Header != nil && l1Header.Number.Uint64() >= l1Height) && (l2Header != nil && l2Header.Number.Uint64() >= 9), nil
+	require.NoError(t, wait.For(context.Background(), 100*time.Millisecond, func() (bool, error) {
+		l1Header, err := testSuite.DB.Blocks.L1LatestBlockHeader()
+		require.NoError(t, err)
+
+		l2Header, err := testSuite.DB.Blocks.L2LatestBlockHeader()
+		require.NoError(t, err)
+
+		return (l1Header != nil && l1Header.Number.Int.Uint64() >= l1Height) && (l2Header != nil && l2Header.Number.Int.Uint64() >= 9), nil
 	}))
 
-	t.Run("indexes L2 blocks", func(t *testing.T) {
-		latestL2Header, err := testSuite.DB.Blocks.LatestL2BlockHeader()
+	t.Run("indexes all L2 blocks", func(t *testing.T) {
+		latestL2Header, err := testSuite.DB.Blocks.L2LatestBlockHeader()
 		require.NoError(t, err)
 		require.NotNil(t, latestL2Header)
 		require.True(t, latestL2Header.Number.Int.Uint64() >= 9)
@@ -50,7 +52,7 @@ func TestE2EBlockHeaders(t *testing.T) {
 		for i := int64(0); i < 10; i++ {
 			height := big.NewInt(i)
 
-			indexedHeader, err := testSuite.DB.Blocks.L2BlockHeader(height)
+			indexedHeader, err := testSuite.DB.Blocks.L2BlockHeaderWithFilter(database.BlockHeader{Number: database.U256{Int: height}})
 			require.NoError(t, err)
 			require.NotNil(t, indexedHeader)
 
@@ -63,11 +65,14 @@ func TestE2EBlockHeaders(t *testing.T) {
 			require.Equal(t, header.ParentHash, indexedHeader.ParentHash)
 			require.Equal(t, header.Time, indexedHeader.Timestamp)
 
-			// ensure the right rlp encoding is stored. checking the hashes sufficies
+			// ensure the right rlp encoding is stored. checking the hashes
+			// suffices as it is based on the rlp bytes of the header
 			require.Equal(t, header.Hash(), indexedHeader.RLPHeader.Hash())
 		}
 	})
 
+	/*
+		TODO: ADD THIS BACK IN WHEN THESE MARKERS ARE INDEXED
 			t.Run("indexes L2 checkpoints", func(t *testing.T) {
 				latestOutput, err := testSuite.DB.Blocks.LatestCheckpointedOutput()
 				require.NoError(t, err)
@@ -103,19 +108,18 @@ func TestE2EBlockHeaders(t *testing.T) {
 					require.Equal(t, crypto.Keccak256Hash(outputRootPreImage[:]), output.OutputRoot)
 				}
 			})
+	*/
 
-	t.Run("indexes L1 logs and associated blocks", func(t *testing.T) {
-		testCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-		defer cancel()
-
+	t.Run("indexes L1 blocks with accompanying contract event", func(t *testing.T) {
 		l1Contracts := []common.Address{}
 		testSuite.OpCfg.L1Deployments.ForEach(func(name string, addr common.Address) { l1Contracts = append(l1Contracts, addr) })
 		logFilter := ethereum.FilterQuery{FromBlock: big.NewInt(0), ToBlock: big.NewInt(int64(l1Height)), Addresses: l1Contracts}
-		logs, err := testSuite.L1Client.FilterLogs(testCtx, logFilter) // []types.Log
+		logs, err := testSuite.L1Client.FilterLogs(context.Background(), logFilter) // []types.Log
 		require.NoError(t, err)
 
-		for _, log := range logs {
-			contractEvent, err := testSuite.DB.ContractEvents.L1ContractEventByTxLogIndex(log.TxHash, uint64(log.Index))
+		for i := range logs {
+			log := logs[i]
+			contractEvent, err := testSuite.DB.ContractEvents.L1ContractEventWithFilter(database.ContractEvent{TransactionHash: log.TxHash, LogIndex: uint64(log.Index)})
 			require.NoError(t, err)
 			require.Equal(t, log.Topics[0], contractEvent.EventSignature)
 			require.Equal(t, log.BlockHash, contractEvent.BlockHash)
@@ -131,12 +135,12 @@ func TestE2EBlockHeaders(t *testing.T) {
 			require.ElementsMatch(t, logRlp, contractEventRlp)
 
 			// ensure the block is also indexed
-			block, err := testSuite.L1Client.BlockByNumber(testCtx, big.NewInt(int64(log.BlockNumber)))
+			block, err := testSuite.L1Client.BlockByNumber(context.Background(), big.NewInt(int64(log.BlockNumber)))
 			require.NoError(t, err)
-
 			require.Equal(t, block.Time(), contractEvent.Timestamp)
+			require.Equal(t, block.Hash(), contractEvent.BlockHash)
 
-			l1BlockHeader, err := testSuite.DB.Blocks.L1BlockHeader(block.Number())
+			l1BlockHeader, err := testSuite.DB.Blocks.L1BlockHeader(block.Hash())
 			require.NoError(t, err)
 			require.Equal(t, block.Hash(), l1BlockHeader.Hash)
 			require.Equal(t, block.ParentHash(), l1BlockHeader.ParentHash)

indexer/e2e_tests/setup.go

@@ -71,23 +71,17 @@ func createE2ETestSuite(t *testing.T) E2ETestSuite {
 		},
 		Chain: config.ChainConfig{
 			L1Contracts: config.L1Contracts{
-				OptimismPortal:         opCfg.L1Deployments.OptimismPortalProxy,
-				L2OutputOracle:         opCfg.L1Deployments.L2OutputOracleProxy,
-				L1CrossDomainMessenger: opCfg.L1Deployments.L1CrossDomainMessengerProxy,
-				L1StandardBridge:       opCfg.L1Deployments.L1StandardBridgeProxy,
-				L1ERC721Bridge:         opCfg.L1Deployments.L1ERC721BridgeProxy,
+				OptimismPortalProxy:         opCfg.L1Deployments.OptimismPortalProxy,
+				L2OutputOracleProxy:         opCfg.L1Deployments.L2OutputOracleProxy,
+				L1CrossDomainMessengerProxy: opCfg.L1Deployments.L1CrossDomainMessengerProxy,
+				L1StandardBridgeProxy:       opCfg.L1Deployments.L1StandardBridgeProxy,
 			},
 		},
 	}
 
 	db, err := database.NewDB(indexerCfg.DB)
 	require.NoError(t, err)
-	indexer, err := indexer.NewIndexer(
-		indexerCfg.Chain,
-		indexerCfg.RPCs,
-		db,
-		logger,
-	)
+	indexer, err := indexer.NewIndexer(logger, indexerCfg.Chain, indexerCfg.RPCs, db)
 	require.NoError(t, err)
 
 	indexerStoppedCh := make(chan interface{}, 1)

indexer/e2e_tests/utils/cross_domain_messages.go

@@ -4,7 +4,7 @@ import (
 	"errors"
 	"math/big"
 
-	"github.com/ethereum-optimism/optimism/indexer/processor"
+	contracts "github.com/ethereum-optimism/optimism/indexer/processors/contracts_bedrock"
 	"github.com/ethereum-optimism/optimism/op-bindings/bindings"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -57,5 +57,5 @@ func CrossDomainMessengerSentMessageHash(sentMessage *bindings.CrossDomainMessen
 		return common.Hash{}, err
 	}
 
-	return processor.CrossDomainMessageHash(abi, sentMessage, value)
+	return contracts.CrossDomainMessageHash(abi, sentMessage, value)
 }

indexer/etl/etl.go

+package etl
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/ethereum-optimism/optimism/indexer/node"
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/ethclient"
+	"github.com/ethereum/go-ethereum/log"
+)
+
+const (
+	defaultLoopInterval     = 5 * time.Second
+	defaultHeaderBufferSize = 500
+)
+
+type ETL struct {
+	log log.Logger
+
+	headerTraversal *node.HeaderTraversal
+	ethClient       *ethclient.Client
+	contracts       []common.Address
+
+	etlBatches chan ETLBatch
+}
+
+type ETLBatch struct {
+	Logger log.Logger
+
+	Headers   []types.Header
+	HeaderMap map[common.Hash]*types.Header
+
+	Logs           []types.Log
+	HeadersWithLog map[common.Hash]bool
+}
+
+func (etl *ETL) Start(ctx context.Context) error {
+	done := ctx.Done()
+	pollTicker := time.NewTicker(defaultLoopInterval)
+	defer pollTicker.Stop()
+
+	etl.log.Info("starting etl...")
+	var headers []types.Header
+	for {
+		select {
+		case <-done:
+			etl.log.Info("stopping etl")
+			return nil
+
+		case <-pollTicker.C:
+			if len(headers) == 0 {
+				newHeaders, err := etl.headerTraversal.NextFinalizedHeaders(defaultHeaderBufferSize)
+				if err != nil {
+					etl.log.Error("error querying for headers", "err", err)
+					continue
+				}
+				if len(newHeaders) == 0 {
+					// Logged as an error since this loop should be operating at a longer interval than the provider
+					etl.log.Error("no new headers. processor unexpectedly at head...")
+					continue
+				}
+
+				headers = newHeaders
+			} else {
+				etl.log.Info("retrying previous batch")
+			}
+
+			firstHeader := headers[0]
+			lastHeader := headers[len(headers)-1]
+			batchLog := etl.log.New("batch_start_block_number", firstHeader.Number, "batch_end_block_number", lastHeader.Number)
+			batchLog.Info("extracting batch", "size", len(headers))
+
+			headerMap := make(map[common.Hash]*types.Header, len(headers))
+			for i := range headers {
+				headerMap[headers[i].Hash()] = &headers[i]
+			}
+
+			headersWithLog := make(map[common.Hash]bool, len(headers))
+			logFilter := ethereum.FilterQuery{FromBlock: firstHeader.Number, ToBlock: lastHeader.Number, Addresses: etl.contracts}
+			logs, err := etl.ethClient.FilterLogs(context.Background(), logFilter)
+			if err != nil {
+				batchLog.Info("unable to extract logs within batch", "err", err)
+				continue // spin and try again
+			}
+
+			for i := range logs {
+				if _, ok := headerMap[logs[i].BlockHash]; !ok {
+					// NOTE. Definitely an error state if the none of the headers were re-orged out in between
+					// the blocks and logs retreival operations. However, we need to gracefully handle reorgs
+					batchLog.Error("log found with block hash not in the batch", "block_hash", logs[i].BlockHash, "log_index", logs[i].Index)
+					return errors.New("parsed log with a block hash not in the fetched batch")
+				}
+				headersWithLog[logs[i].BlockHash] = true
+			}
+
+			if len(logs) > 0 {
+				batchLog.Info("detected logs", "size", len(logs))
+			}
+
+			// create a new reference such that subsequent changes to `headers` does not affect the reference
+			headersRef := headers
+			batch := ETLBatch{Logger: batchLog, Headers: headersRef, HeaderMap: headerMap, Logs: logs, HeadersWithLog: headersWithLog}
+
+			headers = nil
+			etl.etlBatches <- batch
+		}
+	}
+}

indexer/etl/l1_etl.go

+package etl
+
+import (
+	"context"
+	"errors"
+	"reflect"
+
+	"github.com/ethereum-optimism/optimism/indexer/config"
+	"github.com/ethereum-optimism/optimism/indexer/database"
+	"github.com/ethereum-optimism/optimism/indexer/node"
+	"github.com/ethereum-optimism/optimism/op-service/backoff"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
+)
+
+type L1ETL struct {
+	ETL
+
+	db *database.DB
+}
+
+func NewL1ETL(log log.Logger, db *database.DB, client node.EthClient, contracts config.L1Contracts) (*L1ETL, error) {
+	log = log.New("etl", "l1")
+
+	contractValue := reflect.ValueOf(contracts)
+	fields := reflect.VisibleFields(reflect.TypeOf(contracts))
+	l1Contracts := make([]common.Address, len(fields))
+	for i, field := range fields {
+		// ruleid: unsafe-reflect-by-name
+		addr, ok := (contractValue.FieldByName(field.Name).Interface()).(common.Address)
+		if !ok {
+			log.Error("non-address found in L1Contracts", "name", field.Name)
+			return nil, errors.New("non-address found in L1Contracts")
+		}
+
+		log.Info("configured contract", "name", field.Name, "addr", addr)
+		l1Contracts[i] = addr
+	}
+
+	latestHeader, err := db.Blocks.L1LatestBlockHeader()
+	if err != nil {
+		return nil, err
+	}
+
+	var fromHeader *types.Header
+	if latestHeader != nil {
+		log.Info("detected last indexed block", "number", latestHeader.Number.Int, "hash", latestHeader.Hash)
+		fromHeader = latestHeader.RLPHeader.Header()
+	} else {
+		log.Info("no indexed state, starting from genesis")
+	}
+
+	etlBatches := make(chan ETLBatch)
+	etl := ETL{
+		log:             log,
+		headerTraversal: node.NewHeaderTraversal(client, fromHeader),
+		ethClient:       client.GethEthClient(),
+		contracts:       l1Contracts,
+		etlBatches:      etlBatches,
+	}
+
+	return &L1ETL{ETL: etl, db: db}, nil
+}
+
+func (l1Etl *L1ETL) Start(ctx context.Context) error {
+	errCh := make(chan error, 1)
+	go func() {
+		errCh <- l1Etl.ETL.Start(ctx)
+	}()
+
+	for {
+		select {
+		case err := <-errCh:
+			return err
+
+		// Index incoming batches
+		case batch := <-l1Etl.etlBatches:
+			// Pull out only L1 blocks that have emitted a log ( <= batch.Headers )
+			l1BlockHeaders := make([]*database.L1BlockHeader, 0, len(batch.Headers))
+			for i := range batch.Headers {
+				if _, ok := batch.HeadersWithLog[batch.Headers[i].Hash()]; ok {
+					l1BlockHeaders = append(l1BlockHeaders, &database.L1BlockHeader{BlockHeader: database.BlockHeaderFromHeader(&batch.Headers[i])})
+				}
+			}
+
+			if len(l1BlockHeaders) == 0 {
+				batch.Logger.Info("no l1 blocks with logs in batch")
+				continue
+			}
+
+			l1ContractEvents := make([]*database.L1ContractEvent, len(batch.Logs))
+			for i := range batch.Logs {
+				timestamp := batch.HeaderMap[batch.Logs[i].BlockHash].Time
+				l1ContractEvents[i] = &database.L1ContractEvent{ContractEvent: database.ContractEventFromLog(&batch.Logs[i], timestamp)}
+			}
+
+			// Continually try to persist this batch. If it fails after 10 attempts, we simply error out
+			backoffStrategy := &backoff.ExponentialStrategy{Min: 1000, Max: 20_000, MaxJitter: 250}
+			_, err := backoff.Do[interface{}](ctx, 10, backoffStrategy, func() (interface{}, error) {
+				err := l1Etl.db.Transaction(func(tx *database.DB) error {
+					if err := tx.Blocks.StoreL1BlockHeaders(l1BlockHeaders); err != nil {
+						return err
+					}
+
+					// we must have logs if we have l1 blocks
+					if err := tx.ContractEvents.StoreL1ContractEvents(l1ContractEvents); err != nil {
+						return err
+					}
+					return nil
+				})
+
+				if err != nil {
+					batch.Logger.Error("unable to persist batch", "err", err)
+					return nil, err
+				}
+
+				// a-ok! Can merge with the above block but being explicit
+				return nil, nil
+			})
+
+			if err != nil {
+				return err
+			}
+
+			batch.Logger.Info("indexed batch")
+		}
+	}
+}

indexer/etl/l2_etl.go

+package etl
+
+import (
+	"context"
+
+	"github.com/ethereum-optimism/optimism/indexer/database"
+	"github.com/ethereum-optimism/optimism/indexer/node"
+	"github.com/ethereum-optimism/optimism/op-bindings/predeploys"
+	"github.com/ethereum-optimism/optimism/op-service/backoff"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
+)
+
+type L2ETL struct {
+	ETL
+
+	db *database.DB
+}
+
+func NewL2ETL(log log.Logger, db *database.DB, client node.EthClient) (*L2ETL, error) {
+	log = log.New("etl", "l2")
+
+	// allow predeploys to be overridable
+	l2Contracts := []common.Address{}
+	for name, addr := range predeploys.Predeploys {
+		log.Info("configured contract", "name", name, "addr", addr)
+		l2Contracts = append(l2Contracts, *addr)
+	}
+
+	latestHeader, err := db.Blocks.L2LatestBlockHeader()
+	if err != nil {
+		return nil, err
+	}
+
+	var fromHeader *types.Header
+	if latestHeader != nil {
+		log.Info("detected last indexed block", "number", latestHeader.Number.Int, "hash", latestHeader.Hash)
+		fromHeader = latestHeader.RLPHeader.Header()
+	} else {
+		log.Info("no indexed state, starting from genesis")
+	}
+
+	etlBatches := make(chan ETLBatch)
+	etl := ETL{
+		log:             log,
+		headerTraversal: node.NewHeaderTraversal(client, fromHeader),
+		ethClient:       client.GethEthClient(),
+		contracts:       l2Contracts,
+		etlBatches:      etlBatches,
+	}
+
+	return &L2ETL{ETL: etl, db: db}, nil
+}
+
+func (l2Etl *L2ETL) Start(ctx context.Context) error {
+	errCh := make(chan error, 1)
+	go func() {
+		errCh <- l2Etl.ETL.Start(ctx)
+	}()
+
+	for {
+		select {
+		case err := <-errCh:
+			return err
+
+		// Index incoming batches
+		case batch := <-l2Etl.etlBatches:
+			// We're indexing every L2 block.
+			l2BlockHeaders := make([]*database.L2BlockHeader, len(batch.Headers))
+			for i := range batch.Headers {
+				l2BlockHeaders[i] = &database.L2BlockHeader{BlockHeader: database.BlockHeaderFromHeader(&batch.Headers[i])}
+			}
+
+			l2ContractEvents := make([]*database.L2ContractEvent, len(batch.Logs))
+			for i := range batch.Logs {
+				timestamp := batch.HeaderMap[batch.Logs[i].BlockHash].Time
+				l2ContractEvents[i] = &database.L2ContractEvent{ContractEvent: database.ContractEventFromLog(&batch.Logs[i], timestamp)}
+			}
+
+			// Continually try to persist this batch. If it fails after 5 attempts, we simply error out
+			backoffStrategy := &backoff.ExponentialStrategy{Min: 1000, Max: 20_000, MaxJitter: 250}
+			_, err := backoff.Do[interface{}](ctx, 10, backoffStrategy, func() (interface{}, error) {
+				err := l2Etl.db.Transaction(func(tx *database.DB) error {
+					if err := tx.Blocks.StoreL2BlockHeaders(l2BlockHeaders); err != nil {
+						return err
+					}
+					if len(l2ContractEvents) > 0 {
+						if err := tx.ContractEvents.StoreL2ContractEvents(l2ContractEvents); err != nil {
+							return err
+						}
+					}
+					return nil
+				})
+
+				if err != nil {
+					batch.Logger.Error("unable to persist batch", "err", err)
+					return nil, err
+				}
+
+				// a-ok! Can merge with the above block but being explicit
+				return nil, nil
+			})
+
+			if err != nil {
+				return err
+			}
+
+			batch.Logger.Info("indexed batch")
+		}
+	}
+}

indexer/indexer.go

@@ -3,14 +3,16 @@ package indexer
 import (
 	"context"
 	"fmt"
+	"runtime/debug"
 	"sync"
 
 	"github.com/ethereum/go-ethereum/log"
 
 	"github.com/ethereum-optimism/optimism/indexer/config"
 	"github.com/ethereum-optimism/optimism/indexer/database"
+	"github.com/ethereum-optimism/optimism/indexer/etl"
 	"github.com/ethereum-optimism/optimism/indexer/node"
-	"github.com/ethereum-optimism/optimism/indexer/processor"
+	"github.com/ethereum-optimism/optimism/indexer/processors"
 )
 
 // Indexer contains the necessary resources for
@@ -19,29 +21,36 @@ type Indexer struct {
 	db  *database.DB
 	log log.Logger
 
-	L1Processor *processor.L1Processor
-	L2Processor *processor.L2Processor
+	L1ETL *etl.L1ETL
+	L2ETL *etl.L2ETL
+
+	BridgeProcessor *processors.BridgeProcessor
 }
 
 // NewIndexer initializes an instance of the Indexer
-func NewIndexer(chainConfig config.ChainConfig, rpcsConfig config.RPCsConfig, db *database.DB, logger log.Logger) (*Indexer, error) {
-	l1Contracts := chainConfig.L1Contracts
+func NewIndexer(logger log.Logger, chainConfig config.ChainConfig, rpcsConfig config.RPCsConfig, db *database.DB) (*Indexer, error) {
 	l1EthClient, err := node.DialEthClient(rpcsConfig.L1RPC)
 	if err != nil {
 		return nil, err
 	}
-	l1Processor, err := processor.NewL1Processor(logger, l1EthClient, db, l1Contracts)
+
+	l1Etl, err := etl.NewL1ETL(logger, db, l1EthClient, chainConfig.L1Contracts)
 	if err != nil {
 		return nil, err
 	}
 
-	// L2Processor (predeploys). Although most likely the right setting, make this configurable?
-	l2Contracts := processor.L2ContractPredeploys()
 	l2EthClient, err := node.DialEthClient(rpcsConfig.L2RPC)
 	if err != nil {
 		return nil, err
 	}
-	l2Processor, err := processor.NewL2Processor(logger, l2EthClient, db, l2Contracts)
+
+	// Currently defaults to the predeploys
+	l2Etl, err := etl.NewL2ETL(logger, db, l2EthClient)
+	if err != nil {
+		return nil, err
+	}
+
+	bridgeProcessor, err := processors.NewBridgeProcessor(logger, db, chainConfig)
 	if err != nil {
 		return nil, err
 	}
@@ -49,8 +58,10 @@ func NewIndexer(chainConfig config.ChainConfig, rpcsConfig config.RPCsConfig, db
 	indexer := &Indexer{
 		db:  db,
 		log: logger,
-		L1Processor: l1Processor,
-		L2Processor: l2Processor,
+
+		L1ETL:           l1Etl,
+		L2ETL:           l2Etl,
+		BridgeProcessor: bridgeProcessor,
 	}
 
 	return indexer, nil
@@ -59,39 +70,41 @@ func NewIndexer(chainConfig config.ChainConfig, rpcsConfig config.RPCsConfig, db
 // Start starts the indexing service on L1 and L2 chains
 func (i *Indexer) Run(ctx context.Context) error {
 	var wg sync.WaitGroup
-	errCh := make(chan error, 1)
+	errCh := make(chan error, 3)
 
 	// If either processor errors out, we stop
-	processorCtx, cancel := context.WithCancel(ctx)
+	subCtx, cancel := context.WithCancel(ctx)
 	run := func(start func(ctx context.Context) error) {
 		wg.Add(1)
 		defer func() {
 			if err := recover(); err != nil {
 				i.log.Error("halting indexer on panic", "err", err)
+				debug.PrintStack()
 				errCh <- fmt.Errorf("panic: %v", err)
 			}
 
+			cancel()
 			wg.Done()
 		}()
 
-		err := start(processorCtx)
+		err := start(subCtx)
 		if err != nil {
 			i.log.Error("halting indexer on error", "err", err)
-			cancel()
 		}
 
-		// Send a value down regardless if we've received an error or halted
-		// via cancellation where err == nil
+		// Send a value down regardless if we've received an error
+		// or halted via cancellation where err == nil
 		errCh <- err
 	}
 
-	// Kick off the processors
-	go run(i.L1Processor.Start)
-	go run(i.L2Processor.Start)
+	// Kick off all the dependent routines
+	go run(i.L1ETL.Start)
+	go run(i.L2ETL.Start)
+	go run(i.BridgeProcessor.Start)
 	err := <-errCh
 
-	// ensure both processors have halted before returning
 	wg.Wait()
+	i.log.Info("indexer stopped")
 	return err
 }
 

indexer/node/client.go

@@ -27,12 +27,13 @@ const (
 type EthClient interface {
 	FinalizedBlockHeight() (*big.Int, error)
 
-	BlockHeadersByRange(*big.Int, *big.Int) ([]*types.Header, error)
+	BlockHeadersByRange(*big.Int, *big.Int) ([]types.Header, error)
 	BlockHeaderByHash(common.Hash) (*types.Header, error)
 
 	StorageHash(common.Address, *big.Int) (common.Hash, error)
 
-	RawRpcClient() *rpc.Client
+	GethRpcClient() *rpc.Client
+	GethEthClient() *ethclient.Client
 }
 
 type client struct {
@@ -56,10 +57,14 @@ func NewEthClient(rpcClient *rpc.Client) EthClient {
 	return &client{rpcClient}
 }
 
-func (c *client) RawRpcClient() *rpc.Client {
+func (c *client) GethRpcClient() *rpc.Client {
 	return c.rpcClient
 }
 
+func (c *client) GethEthClient() *ethclient.Client {
+	return ethclient.NewClient(c.GethRpcClient())
+}
+
 // FinalizedBlockHeight retrieves the latest block height in a finalized state
 func (c *client) FinalizedBlockHeight() (*big.Int, error) {
 	ctxwt, cancel := context.WithTimeout(context.Background(), defaultRequestTimeout)
@@ -97,7 +102,7 @@ func (c *client) BlockHeaderByHash(hash common.Hash) (*types.Header, error) {
 // BlockHeadersByRange will retrieve block headers within the specified range -- includsive. No restrictions
 // are placed on the range such as blocks in the "latest", "safe" or "finalized" states. If the specified
 // range is too large, `endHeight > latest`, the resulting list is truncated to the available headers
-func (c *client) BlockHeadersByRange(startHeight, endHeight *big.Int) ([]*types.Header, error) {
+func (c *client) BlockHeadersByRange(startHeight, endHeight *big.Int) ([]types.Header, error) {
 	count := new(big.Int).Sub(endHeight, startHeight).Uint64() + 1
 	batchElems := make([]rpc.BatchElem, count)
 	for i := uint64(0); i < count; i++ {
@@ -121,7 +126,7 @@ func (c *client) BlockHeadersByRange(startHeight, endHeight *big.Int) ([]*types.
 	//  - Ensure integrity that they build on top of each other
 	//  - Truncate out headers that do not exist (endHeight > "latest")
 	size := 0
-	headers := make([]*types.Header, count)
+	headers := make([]types.Header, count)
 	for i, batchElem := range batchElems {
 		if batchElem.Error != nil {
 			return nil, batchElem.Error
@@ -129,17 +134,19 @@ func (c *client) BlockHeadersByRange(startHeight, endHeight *big.Int) ([]*types.
 			break
 		}
 
-		header := batchElem.Result.(*types.Header)
+		header, ok := batchElem.Result.(*types.Header)
+		if !ok {
+			return nil, fmt.Errorf("unable to transform rpc response %v into types.Header", batchElem.Result)
+		}
 		if i > 0 && header.ParentHash != headers[i-1].Hash() {
-			// Warn here that we got a bad (malicious?) response
-			break
+			return nil, fmt.Errorf("queried header %s does not follow parent %s", header.Hash(), headers[i-1].Hash())
 		}
 
-		headers[i] = header
+		headers[i] = *header
 		size = size + 1
 	}
-	headers = headers[:size]
 
+	headers = headers[:size]
 	return headers, nil
 }
 

indexer/node/client_test.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/ethereum/go-ethereum/rpc"
 )
 
@@ -21,9 +22,9 @@ func (m *MockEthClient) FinalizedBlockHeight() (*big.Int, error) {
 	return args.Get(0).(*big.Int), args.Error(1)
 }
 
-func (m *MockEthClient) BlockHeadersByRange(from, to *big.Int) ([]*types.Header, error) {
+func (m *MockEthClient) BlockHeadersByRange(from, to *big.Int) ([]types.Header, error) {
 	args := m.Called(from, to)
-	return args.Get(0).([]*types.Header), args.Error(1)
+	return args.Get(0).([]types.Header), args.Error(1)
 }
 
 func (m *MockEthClient) BlockHeaderByHash(hash common.Hash) (*types.Header, error) {
@@ -36,7 +37,12 @@ func (m *MockEthClient) StorageHash(address common.Address, blockNumber *big.Int
 	return args.Get(0).(common.Hash), args.Error(1)
 }
 
-func (m *MockEthClient) RawRpcClient() *rpc.Client {
+func (m *MockEthClient) GethRpcClient() *rpc.Client {
 	args := m.Called()
 	return args.Get(0).(*rpc.Client)
 }
+
+func (m *MockEthClient) GethEthClient() *ethclient.Client {
+	args := m.Called()
+	return args.Get(0).(*ethclient.Client)
+}

indexer/node/header_traversal.go

@@ -26,7 +26,7 @@ func NewHeaderTraversal(ethClient EthClient, fromHeader *types.Header) *HeaderTr
 
 // NextFinalizedHeaders retrives the next set of headers that have been
 // marked as finalized by the connected client, bounded by the supplied size
-func (f *HeaderTraversal) NextFinalizedHeaders(maxSize uint64) ([]*types.Header, error) {
+func (f *HeaderTraversal) NextFinalizedHeaders(maxSize uint64) ([]types.Header, error) {
 	finalizedBlockHeight, err := f.ethClient.FinalizedBlockHeight()
 	if err != nil {
 		return nil, err
@@ -61,6 +61,6 @@ func (f *HeaderTraversal) NextFinalizedHeaders(maxSize uint64) ([]*types.Header,
 		return nil, ErrHeaderTraversalAndProviderMismatchedState
 	}
 
-	f.lastHeader = headers[numHeaders-1]
+	f.lastHeader = &headers[numHeaders-1]
 	return headers, nil
 }

indexer/node/header_traversal_test.go

@@ -11,22 +11,21 @@ import (
 )
 
 // make a set of headers which chain correctly
-func makeHeaders(numHeaders uint64, prevHeader *types.Header) []*types.Header {
-	headers := make([]*types.Header, numHeaders)
+func makeHeaders(numHeaders uint64, prevHeader *types.Header) []types.Header {
+	headers := make([]types.Header, numHeaders)
 	for i := range headers {
 		if i == 0 {
 			if prevHeader == nil {
 				// genesis
-				headers[i] = &types.Header{Number: big.NewInt(0)}
+				headers[i] = types.Header{Number: big.NewInt(0)}
 			} else {
 				// chain onto the previous header
-				headers[i] = &types.Header{Number: big.NewInt(prevHeader.Number.Int64() + 1)}
+				headers[i] = types.Header{Number: big.NewInt(prevHeader.Number.Int64() + 1)}
 				headers[i].ParentHash = prevHeader.Hash()
 			}
 		} else {
-			prevHeader = headers[i-1]
-			headers[i] = &types.Header{Number: big.NewInt(prevHeader.Number.Int64() + 1)}
-			headers[i].ParentHash = prevHeader.Hash()
+			headers[i] = types.Header{Number: big.NewInt(headers[i-1].Number.Int64() + 1)}
+			headers[i].ParentHash = headers[i-1].Hash()
 		}
 	}
 
@@ -62,7 +61,7 @@ func TestHeaderTraversalNextFinalizedHeadersCursored(t *testing.T) {
 	require.Len(t, headers, 5)
 
 	// blocks [5..9]
-	headers = makeHeaders(5, headers[len(headers)-1])
+	headers = makeHeaders(5, &headers[len(headers)-1])
 	client.On("FinalizedBlockHeight").Return(big.NewInt(9), nil)
 	client.On("BlockHeadersByRange", mock.MatchedBy(bigIntMatcher(5)), mock.MatchedBy(bigIntMatcher(9))).Return(headers, nil)
 	headers, err = headerTraversal.NextFinalizedHeaders(5)
@@ -87,7 +86,7 @@ func TestHeaderTraversalNextFinalizedHeadersMaxSize(t *testing.T) {
 	require.Len(t, headers, 5)
 
 	// clamped by the supplied size. FinalizedHeight == 100
-	headers = makeHeaders(10, headers[len(headers)-1])
+	headers = makeHeaders(10, &headers[len(headers)-1])
 	client.On("BlockHeadersByRange", mock.MatchedBy(bigIntMatcher(5)), mock.MatchedBy(bigIntMatcher(14))).Return(headers, nil)
 	headers, err = headerTraversal.NextFinalizedHeaders(10)
 	require.NoError(t, err)

indexer/processor/cross_domain_messenger.go

-package processor
-
-import (
-	"fmt"
-	"math/big"
-
-	"github.com/ethereum/go-ethereum/accounts/abi"
-	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/crypto"
-
-	"github.com/ethereum-optimism/optimism/indexer/database"
-	"github.com/ethereum-optimism/optimism/op-bindings/bindings"
-)
-
-var (
-	// Standard ABI types copied from golang ABI tests
-	Uint256Type, _ = abi.NewType("uint256", "", nil)
-	BytesType, _   = abi.NewType("bytes", "", nil)
-	AddressType, _ = abi.NewType("address", "", nil)
-
-	LegacyCrossDomainMessengerRelayMessageMethod = abi.NewMethod(
-		"relayMessage",
-		"relayMessage",
-		abi.Function,
-		"external", // mutability
-		false,      // isConst
-		true,       // payable
-		abi.Arguments{ // inputs
-			{Name: "sender", Type: AddressType},
-			{Name: "target", Type: AddressType},
-			{Name: "data", Type: BytesType},
-			{Name: "nonce", Type: Uint256Type},
-		},
-		abi.Arguments{}, // outputs
-	)
-)
-
-type CrossDomainMessengerSentMessageEvent struct {
-	*bindings.CrossDomainMessengerSentMessage
-
-	Value       *big.Int
-	MessageHash common.Hash
-	Event       *database.ContractEvent
-}
-
-type CrossDomainMessengerRelayedMessageEvent struct {
-	*bindings.CrossDomainMessengerRelayedMessage
-	Event *database.ContractEvent
-}
-
-func CrossDomainMessengerSentMessageEvents(events *ProcessedContractEvents) ([]CrossDomainMessengerSentMessageEvent, error) {
-	crossDomainMessengerABI, err := bindings.CrossDomainMessengerMetaData.GetAbi()
-	if err != nil {
-		return nil, err
-	}
-
-	sentMessageEventAbi := crossDomainMessengerABI.Events["SentMessage"]
-	sentMessageEventExtensionAbi := crossDomainMessengerABI.Events["SentMessageExtension1"]
-
-	processedSentMessageEvents := events.eventsBySignature[sentMessageEventAbi.ID]
-	crossDomainMessageEvents := make([]CrossDomainMessengerSentMessageEvent, len(processedSentMessageEvents))
-	for i, sentMessageEvent := range processedSentMessageEvents {
-		log := sentMessageEvent.RLPLog
-
-		var sentMsgData bindings.CrossDomainMessengerSentMessage
-		sentMsgData.Raw = *log
-		err = UnpackLog(&sentMsgData, log, sentMessageEventAbi.Name, crossDomainMessengerABI)
-		if err != nil {
-			return nil, err
-		}
-
-		var sentMsgExtensionData bindings.CrossDomainMessengerSentMessageExtension1
-		extensionLog := events.eventByLogIndex[ProcessedContractEventLogIndexKey{log.BlockHash, log.Index + 1}].RLPLog
-		sentMsgExtensionData.Raw = *extensionLog
-		err = UnpackLog(&sentMsgExtensionData, extensionLog, sentMessageEventExtensionAbi.Name, crossDomainMessengerABI)
-		if err != nil {
-			return nil, err
-		}
-
-		msgHash, err := CrossDomainMessageHash(crossDomainMessengerABI, &sentMsgData, sentMsgExtensionData.Value)
-		if err != nil {
-			return nil, err
-		}
-
-		crossDomainMessageEvents[i] = CrossDomainMessengerSentMessageEvent{
-			CrossDomainMessengerSentMessage: &sentMsgData,
-			Value:                           sentMsgExtensionData.Value,
-			MessageHash:                     msgHash,
-			Event:                           sentMessageEvent,
-		}
-	}
-
-	return crossDomainMessageEvents, nil
-}
-
-func CrossDomainMessengerRelayedMessageEvents(events *ProcessedContractEvents) ([]CrossDomainMessengerRelayedMessageEvent, error) {
-	crossDomainMessengerABI, err := bindings.L1CrossDomainMessengerMetaData.GetAbi()
-	if err != nil {
-		return nil, err
-	}
-
-	relayedMessageEventAbi := crossDomainMessengerABI.Events["RelayedMessage"]
-	processedRelayedMessageEvents := events.eventsBySignature[relayedMessageEventAbi.ID]
-	crossDomainMessageEvents := make([]CrossDomainMessengerRelayedMessageEvent, len(processedRelayedMessageEvents))
-	for i, relayedMessageEvent := range processedRelayedMessageEvents {
-		log := relayedMessageEvent.RLPLog
-
-		var relayedMsgData bindings.CrossDomainMessengerRelayedMessage
-		relayedMsgData.Raw = *log
-		err = UnpackLog(&relayedMsgData, log, relayedMessageEventAbi.Name, crossDomainMessengerABI)
-		if err != nil {
-			return nil, err
-		}
-
-		crossDomainMessageEvents[i] = CrossDomainMessengerRelayedMessageEvent{
-			CrossDomainMessengerRelayedMessage: &relayedMsgData,
-			Event:                              relayedMessageEvent,
-		}
-	}
-
-	return crossDomainMessageEvents, nil
-}
-
-// Replica of `Hashing.sol#hashCrossDomainMessage` solidity implementation
-func CrossDomainMessageHash(abi *abi.ABI, sentMsg *bindings.CrossDomainMessengerSentMessage, value *big.Int) (common.Hash, error) {
-	version, _ := DecodeVersionedNonce(sentMsg.MessageNonce)
-	switch version {
-	case 0:
-		// Legacy Message
-		inputBytes, err := LegacyCrossDomainMessengerRelayMessageMethod.Inputs.Pack(sentMsg.Sender, sentMsg.Target, sentMsg.Message, sentMsg.MessageNonce)
-		if err != nil {
-			return common.Hash{}, err
-		}
-		msgBytes := append(LegacyCrossDomainMessengerRelayMessageMethod.ID, inputBytes...)
-		return crypto.Keccak256Hash(msgBytes), nil
-	case 1:
-		// Current Message
-		msgBytes, err := abi.Pack("relayMessage", sentMsg.MessageNonce, sentMsg.Sender, sentMsg.Target, value, sentMsg.GasLimit, sentMsg.Message)
-		if err != nil {
-			return common.Hash{}, err
-		}
-		return crypto.Keccak256Hash(msgBytes), nil
-	}
-
-	return common.Hash{}, fmt.Errorf("unsupported cross domain messenger version: %d", version)
-}

indexer/processor/encoding.go

-package processor
-
-import (
-	"encoding/binary"
-	"math/big"
-)
-
-// DecodeVersionNonce is an re-implementation of Encoding.sol#decodeVersionedNonce.
-// If the nonce is greater than 32 bytes (solidity uint256), bytes [32:] are ignored
-func DecodeVersionedNonce(nonce *big.Int) (uint16, *big.Int) {
-	nonceBytes := nonce.Bytes()
-	nonceByteLen := len(nonceBytes)
-	if nonceByteLen < 30 {
-		// version is 0x0000
-		return 0, nonce
-	} else if nonceByteLen == 31 {
-		// version is 0x00[01..ff]
-		return uint16(nonceBytes[0]), new(big.Int).SetBytes(nonceBytes[1:])
-	} else {
-		// fully specified
-		version := binary.BigEndian.Uint16(nonceBytes[:2])
-		return version, new(big.Int).SetBytes(nonceBytes[2:])
-	}
-}

indexer/processor/l2_to_l1_message_passer.go

-package processor
-
-import (
-	"github.com/ethereum-optimism/optimism/indexer/database"
-	"github.com/ethereum-optimism/optimism/op-bindings/bindings"
-)
-
-type L2ToL1MessagePasserMessagePassed struct {
-	*bindings.L2ToL1MessagePasserMessagePassed
-	Event *database.ContractEvent
-}
-
-func L2ToL1MessagePasserMessagesPassed(events *ProcessedContractEvents) ([]L2ToL1MessagePasserMessagePassed, error) {
-	l2ToL1MessagePasserAbi, err := bindings.L2ToL1MessagePasserMetaData.GetAbi()
-	if err != nil {
-		return nil, err
-	}
-
-	eventName := "MessagePassed"
-	processedMessagePassedEvents := events.eventsBySignature[l2ToL1MessagePasserAbi.Events[eventName].ID]
-	messagesPassed := make([]L2ToL1MessagePasserMessagePassed, len(processedMessagePassedEvents))
-	for i, messagePassedEvent := range processedMessagePassedEvents {
-		log := messagePassedEvent.RLPLog
-
-		var messagePassed bindings.L2ToL1MessagePasserMessagePassed
-		messagePassed.Raw = *log
-		err := UnpackLog(&messagePassed, log, eventName, l2ToL1MessagePasserAbi)
-		if err != nil {
-			return nil, err
-		}
-
-		messagesPassed[i] = L2ToL1MessagePasserMessagePassed{
-			L2ToL1MessagePasserMessagePassed: &messagePassed,
-			Event:                            messagePassedEvent,
-		}
-	}
-
-	return messagesPassed, nil
-}

indexer/processor/optimism_portal.go

-package processor
-
-import (
-	"errors"
-	"math/big"
-
-	"github.com/ethereum-optimism/optimism/indexer/database"
-	"github.com/ethereum-optimism/optimism/op-bindings/bindings"
-	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
-
-	"github.com/ethereum/go-ethereum/core/types"
-)
-
-type OptimismPortalTransactionDepositEvent struct {
-	*bindings.OptimismPortalTransactionDeposited
-	DepositTx *types.DepositTx
-	Event     *database.ContractEvent
-}
-
-type OptimismPortalWithdrawalProvenEvent struct {
-	*bindings.OptimismPortalWithdrawalProven
-	Event *database.ContractEvent
-}
-
-type OptimismPortalWithdrawalFinalizedEvent struct {
-	*bindings.OptimismPortalWithdrawalFinalized
-	Event *database.ContractEvent
-}
-
-type OptimismPortalProvenWithdrawal struct {
-	OutputRoot    [32]byte
-	Timestamp     *big.Int
-	L2OutputIndex *big.Int
-}
-
-func OptimismPortalTransactionDepositEvents(events *ProcessedContractEvents) ([]OptimismPortalTransactionDepositEvent, error) {
-	optimismPortalAbi, err := bindings.OptimismPortalMetaData.GetAbi()
-	if err != nil {
-		return nil, err
-	}
-
-	eventName := "TransactionDeposited"
-	if optimismPortalAbi.Events[eventName].ID != derive.DepositEventABIHash {
-		return nil, errors.New("op-node deposit event abi hash & optimism portal tx deposit mismatch")
-	}
-
-	processedTxDepositedEvents := events.eventsBySignature[derive.DepositEventABIHash]
-	txDeposits := make([]OptimismPortalTransactionDepositEvent, len(processedTxDepositedEvents))
-	for i, txDepositEvent := range processedTxDepositedEvents {
-		log := txDepositEvent.RLPLog
-
-		depositTx, err := derive.UnmarshalDepositLogEvent(log)
-		if err != nil {
-			return nil, err
-		}
-
-		var txDeposit bindings.OptimismPortalTransactionDeposited
-		txDeposit.Raw = *log
-		err = UnpackLog(&txDeposit, log, eventName, optimismPortalAbi)
-		if err != nil {
-			return nil, err
-		}
-
-		txDeposits[i] = OptimismPortalTransactionDepositEvent{
-			OptimismPortalTransactionDeposited: &txDeposit,
-			DepositTx:                          depositTx,
-			Event:                              txDepositEvent,
-		}
-	}
-
-	return txDeposits, nil
-}
-
-func OptimismPortalWithdrawalProvenEvents(events *ProcessedContractEvents) ([]OptimismPortalWithdrawalProvenEvent, error) {
-	optimismPortalAbi, err := bindings.OptimismPortalMetaData.GetAbi()
-	if err != nil {
-		return nil, err
-	}
-
-	eventName := "WithdrawalProven"
-	processedWithdrawalProvenEvents := events.eventsBySignature[optimismPortalAbi.Events[eventName].ID]
-	provenEvents := make([]OptimismPortalWithdrawalProvenEvent, len(processedWithdrawalProvenEvents))
-	for i, provenEvent := range processedWithdrawalProvenEvents {
-		log := provenEvent.RLPLog
-
-		var withdrawalProven bindings.OptimismPortalWithdrawalProven
-		withdrawalProven.Raw = *log
-		err := UnpackLog(&withdrawalProven, log, eventName, optimismPortalAbi)
-		if err != nil {
-			return nil, err
-		}
-
-		provenEvents[i] = OptimismPortalWithdrawalProvenEvent{
-			OptimismPortalWithdrawalProven: &withdrawalProven,
-			Event:                          provenEvent,
-		}
-	}
-
-	return provenEvents, nil
-}
-
-func OptimismPortalWithdrawalFinalizedEvents(events *ProcessedContractEvents) ([]OptimismPortalWithdrawalFinalizedEvent, error) {
-	optimismPortalAbi, err := bindings.OptimismPortalMetaData.GetAbi()
-	if err != nil {
-		return nil, err
-	}
-
-	eventName := "WithdrawalFinalized"
-	processedWithdrawalFinalizedEvents := events.eventsBySignature[optimismPortalAbi.Events[eventName].ID]
-	finalizedEvents := make([]OptimismPortalWithdrawalFinalizedEvent, len(processedWithdrawalFinalizedEvents))
-	for i, finalizedEvent := range processedWithdrawalFinalizedEvents {
-		log := finalizedEvent.RLPLog
-
-		var withdrawalFinalized bindings.OptimismPortalWithdrawalFinalized
-		err := UnpackLog(&withdrawalFinalized, log, eventName, optimismPortalAbi)
-		if err != nil {
-			return nil, err
-		}
-
-		finalizedEvents[i] = OptimismPortalWithdrawalFinalizedEvent{
-			OptimismPortalWithdrawalFinalized: &withdrawalFinalized,
-			Event:                             finalizedEvent,
-		}
-	}
-
-	return finalizedEvents, nil
-}

indexer/processor/processor.go

-package processor
-
-import (
-	"context"
-	"time"
-
-	"github.com/ethereum-optimism/optimism/indexer/database"
-	"github.com/ethereum-optimism/optimism/indexer/node"
-
-	"github.com/ethereum/go-ethereum/core/types"
-	"github.com/ethereum/go-ethereum/log"
-)
-
-const (
-	defaultLoopInterval     = 5 * time.Second
-	defaultHeaderBufferSize = 500
-)
-
-// ProcessFn is the the entrypoint for processing a batch of headers.
-// In the event of failure, database operations are rolled back
-type ProcessFn func(*database.DB, []*types.Header) error
-
-type processor struct {
-	headerTraversal *node.HeaderTraversal
-
-	db         *database.DB
-	processFn  ProcessFn
-	processLog log.Logger
-
-	paused                bool
-	latestProcessedHeader *types.Header
-}
-
-// Start kicks off the processing loop. This is a block operation
-// unless the processor encountering an error, abrupting the loop,
-// or the supplied context is cancelled.
-func (p *processor) Start(ctx context.Context) error {
-	done := ctx.Done()
-	pollTicker := time.NewTicker(defaultLoopInterval)
-	defer pollTicker.Stop()
-
-	p.processLog.Info("starting processor...")
-	var unprocessedHeaders []*types.Header
-	for {
-		select {
-		case <-done:
-			p.processLog.Info("stopping processor")
-			return nil
-
-		case <-pollTicker.C:
-			if p.paused {
-				p.processLog.Warn("processor is paused...")
-				continue
-			}
-
-			if len(unprocessedHeaders) == 0 {
-				newHeaders, err := p.headerTraversal.NextFinalizedHeaders(defaultHeaderBufferSize)
-				if err != nil {
-					p.processLog.Error("error querying for headers", "err", err)
-					continue
-				} else if len(newHeaders) == 0 {
-					// Logged as an error since this loop should be operating at a longer interval than the provider
-					p.processLog.Error("no new headers. processor unexpectedly at head...")
-					continue
-				}
-
-				unprocessedHeaders = newHeaders
-			} else {
-				p.processLog.Info("retrying previous batch")
-			}
-
-			firstHeader := unprocessedHeaders[0]
-			lastHeader := unprocessedHeaders[len(unprocessedHeaders)-1]
-			batchLog := p.processLog.New("batch_start_block_number", firstHeader.Number, "batch_end_block_number", lastHeader.Number)
-			err := p.db.Transaction(func(db *database.DB) error {
-				batchLog.Info("processing batch")
-				return p.processFn(db, unprocessedHeaders)
-			})
-
-			// Eventually, we want to halt the processor on any error rather than rely
-			// on this loop for retry functionality.
-			if err != nil {
-				batchLog.Warn("error processing batch. no operations committed", "err", err)
-			} else {
-				batchLog.Info("fully committed batch")
-
-				unprocessedHeaders = nil
-				p.latestProcessedHeader = lastHeader
-			}
-		}
-	}
-}
-
-func (p processor) LatestProcessedHeader() *types.Header {
-	return p.latestProcessedHeader
-}
-
-// Useful ONLY for tests!
-
-func (p *processor) PauseForTest() {
-	p.paused = true
-}
-
-func (p *processor) ResumeForTest() {
-	p.paused = false
-}

indexer/processor/standard_bridge.go

-package processor
-
-import (
-	"bytes"
-	"errors"
-
-	"github.com/ethereum-optimism/optimism/indexer/database"
-	"github.com/ethereum-optimism/optimism/op-bindings/bindings"
-	"github.com/ethereum-optimism/optimism/op-bindings/predeploys"
-
-	"github.com/ethereum/go-ethereum/common"
-)
-
-type StandardBridgeInitiatedEvent struct {
-	// We hardcode to ERC20 since ETH can be pseudo-represented as an ERC20 utilizing
-	// the hardcoded ETH address
-	*bindings.StandardBridgeERC20BridgeInitiated
-
-	CrossDomainMessageHash common.Hash
-	Event                  *database.ContractEvent
-}
-
-type StandardBridgeFinalizedEvent struct {
-	// We hardcode to ERC20 since ETH can be pseudo-represented as an ERC20 utilizing
-	// the hardcoded ETH address
-	*bindings.StandardBridgeERC20BridgeFinalized
-
-	CrossDomainMessageHash common.Hash
-	Event                  *database.ContractEvent
-}
-
-// StandardBridgeInitiatedEvents extracts all initiated bridge events from the contracts that follow the StandardBridge ABI. The
-// correlated CrossDomainMessenger nonce is also parsed from the associated messenger events.
-func StandardBridgeInitiatedEvents(events *ProcessedContractEvents) ([]StandardBridgeInitiatedEvent, error) {
-	ethBridgeInitiatedEvents, err := _standardBridgeInitiatedEvents[bindings.StandardBridgeETHBridgeInitiated](events)
-	if err != nil {
-		return nil, err
-	}
-
-	erc20BridgeInitiatedEvents, err := _standardBridgeInitiatedEvents[bindings.StandardBridgeERC20BridgeInitiated](events)
-	if err != nil {
-		return nil, err
-	}
-
-	return append(ethBridgeInitiatedEvents, erc20BridgeInitiatedEvents...), nil
-}
-
-// StandardBridgeFinalizedEvents extracts all finalization bridge events from the contracts that follow the StandardBridge ABI. The
-// correlated CrossDomainMessenger nonce is also parsed by looking at the parameters of the corresponding relayMessage transaction data.
-func StandardBridgeFinalizedEvents(events *ProcessedContractEvents) ([]StandardBridgeFinalizedEvent, error) {
-	ethBridgeFinalizedEvents, err := _standardBridgeFinalizedEvents[bindings.StandardBridgeETHBridgeFinalized](events)
-	if err != nil {
-		return nil, err
-	}
-
-	erc20BridgeFinalizedEvents, err := _standardBridgeFinalizedEvents[bindings.StandardBridgeERC20BridgeFinalized](events)
-	if err != nil {
-		return nil, err
-	}
-
-	return append(ethBridgeFinalizedEvents, erc20BridgeFinalizedEvents...), nil
-}
-
-// parse out eth or erc20 bridge initiated events
-func _standardBridgeInitiatedEvents[BridgeEvent bindings.StandardBridgeETHBridgeInitiated | bindings.StandardBridgeERC20BridgeInitiated](
-	events *ProcessedContractEvents,
-) ([]StandardBridgeInitiatedEvent, error) {
-	standardBridgeABI, err := bindings.StandardBridgeMetaData.GetAbi()
-	if err != nil {
-		return nil, err
-	}
-
-	crossDomainMessengerABI, err := bindings.CrossDomainMessengerMetaData.GetAbi()
-	if err != nil {
-		return nil, err
-	}
-
-	sentMessageEventAbi := crossDomainMessengerABI.Events["SentMessage"]
-	sentMessageExtensionEventAbi := crossDomainMessengerABI.Events["SentMessageExtension1"]
-
-	var tmp BridgeEvent
-	var eventName string
-	var finalizeMethodName string
-	switch any(tmp).(type) {
-	case bindings.StandardBridgeETHBridgeInitiated:
-		eventName = "ETHBridgeInitiated"
-		finalizeMethodName = "finalizeBridgeETH"
-	case bindings.StandardBridgeERC20BridgeInitiated:
-		eventName = "ERC20BridgeInitiated"
-		finalizeMethodName = "finalizeBridgeERC20"
-	default:
-		panic("should not be here")
-	}
-
-	processedInitiatedBridgeEvents := events.eventsBySignature[standardBridgeABI.Events[eventName].ID]
-	initiatedBridgeEvents := make([]StandardBridgeInitiatedEvent, len(processedInitiatedBridgeEvents))
-	for i, bridgeInitiatedEvent := range processedInitiatedBridgeEvents {
-		log := bridgeInitiatedEvent.RLPLog
-
-		var bridgeData BridgeEvent
-		err := UnpackLog(&bridgeData, log, eventName, standardBridgeABI)
-		if err != nil {
-			return nil, err
-		}
-
-		// Look for the sent message event to compute the message hash of the relayed tx
-		//   - L1: BridgeInitiated -> Portal#DepositTransaction -> SentMessage ...
-		//   - L1: BridgeInitiated -> L2ToL1MessagePasser#MessagePassed -> SentMessage ...
-		var sentMsgData bindings.CrossDomainMessengerSentMessage
-		sentMsgLog := events.eventByLogIndex[ProcessedContractEventLogIndexKey{log.BlockHash, log.Index + 2}].RLPLog
-		if sentMsgLog.Topics[0] != sentMessageEventAbi.ID {
-			return nil, errors.New("unexpected bridge event ordering")
-		}
-		sentMsgData.Raw = *sentMsgLog
-		err = UnpackLog(&sentMsgData, sentMsgLog, sentMessageEventAbi.Name, crossDomainMessengerABI)
-		if err != nil {
-			return nil, err
-		}
-
-		var sentMsgExtensionData bindings.CrossDomainMessengerSentMessageExtension1
-		sentMsgExtensionLog := events.eventByLogIndex[ProcessedContractEventLogIndexKey{log.BlockHash, log.Index + 3}].RLPLog
-		if sentMsgExtensionLog.Topics[0] != sentMessageExtensionEventAbi.ID {
-			return nil, errors.New("unexpected bridge event ordering")
-		}
-		sentMsgData.Raw = *sentMsgLog
-		err = UnpackLog(&sentMsgExtensionData, sentMsgExtensionLog, sentMessageExtensionEventAbi.Name, crossDomainMessengerABI)
-		if err != nil {
-			return nil, err
-		}
-
-		msgHash, err := CrossDomainMessageHash(crossDomainMessengerABI, &sentMsgData, sentMsgExtensionData.Value)
-		if err != nil {
-			return nil, err
-		}
-
-		var erc20BridgeData *bindings.StandardBridgeERC20BridgeInitiated
-		var expectedCrossDomainMessage []byte
-		switch any(bridgeData).(type) {
-		case bindings.StandardBridgeETHBridgeInitiated:
-			ethBridgeData := any(bridgeData).(bindings.StandardBridgeETHBridgeInitiated)
-			expectedCrossDomainMessage, err = standardBridgeABI.Pack(finalizeMethodName, ethBridgeData.From, ethBridgeData.To, ethBridgeData.Amount, ethBridgeData.ExtraData)
-			if err != nil {
-				return nil, err
-			}
-
-			// represent eth bridge as an erc20
-			erc20BridgeData = &bindings.StandardBridgeERC20BridgeInitiated{
-				Raw: *log,
-				// Represent ETH using the hardcoded address
-				LocalToken: predeploys.LegacyERC20ETHAddr, RemoteToken: predeploys.LegacyERC20ETHAddr,
-				// Bridge data
-				From: ethBridgeData.From, To: ethBridgeData.To, Amount: ethBridgeData.Amount, ExtraData: ethBridgeData.ExtraData,
-			}
-
-		case bindings.StandardBridgeERC20BridgeInitiated:
-			_temp := any(bridgeData).(bindings.StandardBridgeERC20BridgeInitiated)
-			erc20BridgeData = &_temp
-			erc20BridgeData.Raw = *log
-			expectedCrossDomainMessage, err = standardBridgeABI.Pack(finalizeMethodName, erc20BridgeData.RemoteToken, erc20BridgeData.LocalToken, erc20BridgeData.From, erc20BridgeData.To, erc20BridgeData.Amount, erc20BridgeData.ExtraData)
-			if err != nil {
-				return nil, err
-			}
-		}
-
-		if !bytes.Equal(sentMsgData.Message, expectedCrossDomainMessage) {
-			return nil, errors.New("bridge cross domain message mismatch")
-		}
-
-		initiatedBridgeEvents[i] = StandardBridgeInitiatedEvent{
-			StandardBridgeERC20BridgeInitiated: erc20BridgeData,
-			CrossDomainMessageHash:             msgHash,
-			Event:                              bridgeInitiatedEvent,
-		}
-	}
-
-	return initiatedBridgeEvents, nil
-}
-
-// parse out eth or erc20 bridge finalization events
-func _standardBridgeFinalizedEvents[BridgeEvent bindings.StandardBridgeETHBridgeFinalized | bindings.StandardBridgeERC20BridgeFinalized](
-	events *ProcessedContractEvents,
-) ([]StandardBridgeFinalizedEvent, error) {
-	standardBridgeABI, err := bindings.StandardBridgeMetaData.GetAbi()
-	if err != nil {
-		return nil, err
-	}
-
-	crossDomainMessengerABI, err := bindings.CrossDomainMessengerMetaData.GetAbi()
-	if err != nil {
-		return nil, err
-	}
-
-	relayedMessageEventAbi := crossDomainMessengerABI.Events["RelayedMessage"]
-
-	var bridgeData BridgeEvent
-	var eventName string
-	switch any(bridgeData).(type) {
-	case bindings.StandardBridgeETHBridgeFinalized:
-		eventName = "ETHBridgeFinalized"
-	case bindings.StandardBridgeERC20BridgeFinalized:
-		eventName = "ERC20BridgeFinalized"
-	default:
-		panic("should not be here")
-	}
-
-	processedFinalizedBridgeEvents := events.eventsBySignature[standardBridgeABI.Events[eventName].ID]
-	finalizedBridgeEvents := make([]StandardBridgeFinalizedEvent, len(processedFinalizedBridgeEvents))
-	for i, bridgeFinalizedEvent := range processedFinalizedBridgeEvents {
-		log := bridgeFinalizedEvent.RLPLog
-
-		var bridgeData BridgeEvent
-		err := UnpackLog(&bridgeData, log, eventName, standardBridgeABI)
-		if err != nil {
-			return nil, err
-		}
-
-		// Look for the RelayedMessage event that follows right after the BridgeFinalized Event
-		var relayedMsgData bindings.CrossDomainMessengerRelayedMessage
-		relayedMsgLog := events.eventByLogIndex[ProcessedContractEventLogIndexKey{log.BlockHash, log.Index + 1}].RLPLog
-		if relayedMsgLog.Topics[0] != relayedMessageEventAbi.ID {
-			return nil, errors.New("unexpected bridge event ordering")
-		}
-		err = UnpackLog(&relayedMsgData, relayedMsgLog, relayedMessageEventAbi.Name, crossDomainMessengerABI)
-		if err != nil {
-			return nil, err
-		}
-
-		var erc20BridgeData *bindings.StandardBridgeERC20BridgeFinalized
-		switch any(bridgeData).(type) {
-		case bindings.StandardBridgeETHBridgeFinalized:
-			ethBridgeData := any(bridgeData).(bindings.StandardBridgeETHBridgeFinalized)
-			erc20BridgeData = &bindings.StandardBridgeERC20BridgeFinalized{
-				Raw: *log,
-				// Represent ETH using the hardcoded address
-				LocalToken: predeploys.LegacyERC20ETHAddr, RemoteToken: predeploys.LegacyERC20ETHAddr,
-				// Bridge data
-				From: ethBridgeData.From, To: ethBridgeData.To, Amount: ethBridgeData.Amount, ExtraData: ethBridgeData.ExtraData,
-			}
-
-		case bindings.StandardBridgeERC20BridgeFinalized:
-			_temp := any(bridgeData).(bindings.StandardBridgeERC20BridgeFinalized)
-			erc20BridgeData = &_temp
-			erc20BridgeData.Raw = *log
-		}
-
-		finalizedBridgeEvents[i] = StandardBridgeFinalizedEvent{
-			StandardBridgeERC20BridgeFinalized: erc20BridgeData,
-			CrossDomainMessageHash:             relayedMsgData.MsgHash,
-			Event:                              bridgeFinalizedEvent,
-		}
-	}
-
-	return finalizedBridgeEvents, nil
-}

indexer/processors/bridge.go

+package processors
+
+import (
+	"context"
+	"errors"
+	"math/big"
+	"time"
+
+	"github.com/ethereum-optimism/optimism/indexer/config"
+	"github.com/ethereum-optimism/optimism/indexer/database"
+	"github.com/ethereum-optimism/optimism/indexer/processors/bridge"
+
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
+)
+
+type BridgeProcessor struct {
+	log         log.Logger
+	db          *database.DB
+	chainConfig config.ChainConfig
+
+	// NOTE: We'll need this processor to handle for reorgs events.
+
+	LatestL1Header *types.Header
+	LatestL2Header *types.Header
+}
+
+func NewBridgeProcessor(log log.Logger, db *database.DB, chainConfig config.ChainConfig) (*BridgeProcessor, error) {
+	log = log.New("processor", "bridge")
+
+	latestL1Header, err := bridge.L1LatestBridgeEventHeader(db, chainConfig)
+	if err != nil {
+		return nil, err
+	}
+	latestL2Header, err := bridge.L2LatestBridgeEventHeader(db)
+	if err != nil {
+		return nil, err
+	}
+
+	// Since the bridge processor indexes events based on epochs, there's
+	// no scenario in which we have indexed L2 data with no L1 data.
+	//
+	// NOTE: Technically there is an exception if our bridging contracts are
+	// used to bridges native from L2 and an op-chain happens to launch where
+	// only L2 native bridge events have occurred. This is a rare situation now
+	// and it's worth the assertion as an integrity check. We can revisit this
+	// as more chains launch with primarily L2-native activity.
+	if latestL1Header == nil && latestL2Header != nil {
+		log.Error("detected indexed L2 bridge activity with no indexed L1 state", "l2_block_number", latestL2Header.Number)
+		return nil, errors.New("detected indexed L2 bridge activity with no indexed L1 state")
+	}
+
+	if latestL1Header == nil && latestL2Header == nil {
+		log.Info("no indexed state, starting from genesis")
+	} else {
+		log.Info("detected the latest indexed state", "l1_block_number", latestL1Header.Number, "l2_block_number", latestL2Header.Number)
+	}
+
+	return &BridgeProcessor{log, db, chainConfig, latestL1Header, latestL2Header}, nil
+}
+
+func (b *BridgeProcessor) Start(ctx context.Context) error {
+	done := ctx.Done()
+
+	// NOTE: This should run on same iterval as L1 ETL rather than as finding the
+	// lasted epoch is constrained to how much L1 data we've indexed.
+	pollTicker := time.NewTicker(5 * time.Second)
+	defer pollTicker.Stop()
+
+	// In order to ensure all seen bridge finalization events correspond with seen
+	// bridge initiated events, we establish a shared marker between L1 and L2 when
+	// processing events.
+	//
+	// As L1 and L2 blocks are indexed, the highest indexed L2 block starting a new
+	// sequencing epoch and corresponding L1 origin that has also been indexed
+	// serves as this shared marker.
+
+	// TODOs:
+	// 	  1. Fix Logging. Should be clear if we're looking at L1 or L2 side of things
+
+	b.log.Info("starting bridge processor...")
+	for {
+		select {
+		case <-done:
+			b.log.Info("stopping bridge processor")
+			return nil
+
+		case <-pollTicker.C:
+			latestEpoch, err := b.db.Blocks.LatestEpoch()
+			if err != nil {
+				return err
+			}
+			if latestEpoch == nil {
+				if b.LatestL1Header != nil {
+					// Once we have some satte `latestEpoch` should never return nil.
+					b.log.Error("started with indexed bridge state, but no blocks epochs returned", "latest_bridge_l1_block_number", b.LatestL1Header.Number)
+					return errors.New("started with indexed bridge state, but no blocks epochs returned")
+				} else {
+					b.log.Warn("no indexed block state. waiting...")
+					continue
+				}
+			}
+
+			if b.LatestL1Header != nil && latestEpoch.L1BlockHeader.Hash == b.LatestL1Header.Hash() {
+				// Marked as a warning since the bridge should always be processing at least 1 new epoch
+				b.log.Warn("all available epochs indexed by the bridge", "latest_epoch_number", b.LatestL1Header.Number)
+				continue
+			}
+
+			toL1Height, toL2Height := latestEpoch.L1BlockHeader.Number.Int, latestEpoch.L2BlockHeader.Number.Int
+			fromL1Height, fromL2Height := big.NewInt(0), big.NewInt(0)
+			if b.LatestL1Header != nil {
+				// `NewBridgeProcessor` ensures that LatestL2Header must not be nil if LatestL1Header is set
+				fromL1Height = new(big.Int).Add(b.LatestL1Header.Number, big.NewInt(1))
+				fromL2Height = new(big.Int).Add(b.LatestL2Header.Number, big.NewInt(1))
+			}
+
+			batchLog := b.log.New("epoch_start_number", fromL1Height, "epoch_end_number", toL1Height)
+			batchLog.Info("scanning bridge events")
+			err = b.db.Transaction(func(tx *database.DB) error {
+				l1BridgeLog := b.log.New("from_l1_block_number", fromL1Height, "to_l1_block_number", toL1Height)
+				l2BridgeLog := b.log.New("from_l2_block_number", fromL2Height, "to_l2_block_number", toL2Height)
+
+				// First, find all possible initiated bridge events
+				if err := bridge.L1ProcessInitiatedBridgeEvents(l1BridgeLog, tx, b.chainConfig, fromL1Height, toL1Height); err != nil {
+					return err
+				}
+				if err := bridge.L2ProcessInitiatedBridgeEvents(l2BridgeLog, tx, fromL2Height, toL2Height); err != nil {
+					return err
+				}
+
+				// Now that all initiated events have been indexed, it is ensured that all finalization can find their counterpart.
+				if err := bridge.L1ProcessFinalizedBridgeEvents(l1BridgeLog, tx, b.chainConfig, fromL1Height, toL1Height); err != nil {
+					return err
+				}
+				if err := bridge.L2ProcessFinalizedBridgeEvents(l2BridgeLog, tx, fromL2Height, toL2Height); err != nil {
+					return err
+				}
+
+				// a-ok
+				return nil
+			})
+
+			if err != nil {
+				// Try again on a subsequent interval
+				batchLog.Error("unable to index new bridge events", "err", err)
+			} else {
+				batchLog.Info("done indexing new bridge events", "latest_l1_block_number", toL1Height, "latest_l2_block_number", toL2Height)
+				b.LatestL1Header = latestEpoch.L1BlockHeader.RLPHeader.Header()
+				b.LatestL2Header = latestEpoch.L2BlockHeader.RLPHeader.Header()
+			}
+		}
+	}
+}

indexer/processors/bridge/types.go

+package bridge
+
+import "github.com/ethereum/go-ethereum/common"
+
+type logKey struct {
+	BlockHash common.Hash
+	LogIndex  uint64
+}

indexer/processors/contracts_bedrock/cross_domain_messenger.go

+package contracts_bedrock
+
+import (
+	"fmt"
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/accounts/abi"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
+
+	"github.com/ethereum-optimism/optimism/indexer/database"
+	"github.com/ethereum-optimism/optimism/op-bindings/bindings"
+)
+
+var (
+	// Standard ABI types copied from golang ABI tests
+	uint256Type, _ = abi.NewType("uint256", "", nil)
+	bytesType, _   = abi.NewType("bytes", "", nil)
+	addressType, _ = abi.NewType("address", "", nil)
+
+	legacyCrossDomainMessengerRelayMessageMethod = abi.NewMethod(
+		"relayMessage",
+		"relayMessage",
+		abi.Function,
+		"external", // mutability
+		false,      // isConst
+		true,       // payable
+		abi.Arguments{ // inputs
+			{Name: "sender", Type: addressType},
+			{Name: "target", Type: addressType},
+			{Name: "data", Type: bytesType},
+			{Name: "nonce", Type: uint256Type},
+		},
+		abi.Arguments{}, // outputs
+	)
+)
+
+type CrossDomainMessengerSentMessageEvent struct {
+	Event         *database.ContractEvent
+	BridgeMessage database.BridgeMessage
+}
+
+type CrossDomainMessengerRelayedMessageEvent struct {
+	Event       *database.ContractEvent
+	MessageHash common.Hash
+}
+
+func CrossDomainMessengerSentMessageEvents(chainSelector string, contractAddress common.Address, db *database.DB, fromHeight, toHeight *big.Int) ([]CrossDomainMessengerSentMessageEvent, error) {
+	crossDomainMessengerAbi, err := bindings.CrossDomainMessengerMetaData.GetAbi()
+	if err != nil {
+		return nil, err
+	}
+
+	sentMessageEventAbi := crossDomainMessengerAbi.Events["SentMessage"]
+	contractEventFilter := database.ContractEvent{ContractAddress: contractAddress, EventSignature: sentMessageEventAbi.ID}
+	sentMessageEvents, err := db.ContractEvents.ContractEventsWithFilter(contractEventFilter, chainSelector, fromHeight, toHeight)
+	if err != nil {
+		return nil, err
+	}
+	if len(sentMessageEvents) == 0 {
+		// prevent the following db queries if we dont need them
+		return nil, nil
+	}
+
+	sentMessageExtensionEventAbi := crossDomainMessengerAbi.Events["SentMessageExtension1"]
+	contractEventFilter = database.ContractEvent{ContractAddress: contractAddress, EventSignature: sentMessageExtensionEventAbi.ID}
+	sentMessageExtensionEvents, err := db.ContractEvents.ContractEventsWithFilter(contractEventFilter, chainSelector, fromHeight, toHeight)
+	if err != nil {
+		return nil, err
+	}
+	if len(sentMessageEvents) != len(sentMessageExtensionEvents) {
+		return nil, fmt.Errorf("mismatch in SentMessage events. %d sent messages & %d sent message extensions", len(sentMessageEvents), len(sentMessageExtensionEvents))
+	}
+
+	crossDomainSentMessages := make([]CrossDomainMessengerSentMessageEvent, len(sentMessageEvents))
+	for i := range sentMessageEvents {
+		sentMessage := bindings.CrossDomainMessengerSentMessage{Raw: *sentMessageEvents[i].RLPLog}
+		err = UnpackLog(&sentMessage, sentMessageEvents[i].RLPLog, sentMessageEventAbi.Name, crossDomainMessengerAbi)
+		if err != nil {
+			return nil, err
+		}
+		sentMessageExtension := bindings.CrossDomainMessengerSentMessageExtension1{Raw: *sentMessageExtensionEvents[i].RLPLog}
+		err = UnpackLog(&sentMessageExtension, sentMessageExtensionEvents[i].RLPLog, sentMessageExtensionEventAbi.Name, crossDomainMessengerAbi)
+		if err != nil {
+			return nil, err
+		}
+
+		messageHash, err := CrossDomainMessageHash(crossDomainMessengerAbi, &sentMessage, sentMessageExtension.Value)
+		if err != nil {
+			return nil, err
+		}
+
+		crossDomainSentMessages[i] = CrossDomainMessengerSentMessageEvent{
+			Event: &sentMessageEvents[i],
+			BridgeMessage: database.BridgeMessage{
+				MessageHash:          messageHash,
+				Nonce:                database.U256{Int: sentMessage.MessageNonce},
+				SentMessageEventGUID: sentMessageEvents[i].GUID,
+				GasLimit:             database.U256{Int: sentMessage.GasLimit},
+				Tx: database.Transaction{
+					FromAddress: sentMessage.Sender,
+					ToAddress:   sentMessage.Target,
+					Amount:      database.U256{Int: sentMessageExtension.Value},
+					Data:        sentMessage.Message,
+					Timestamp:   sentMessageEvents[i].Timestamp,
+				},
+			},
+		}
+
+	}
+
+	return crossDomainSentMessages, nil
+}
+
+func CrossDomainMessengerRelayedMessageEvents(chainSelector string, contractAddress common.Address, db *database.DB, fromHeight, toHeight *big.Int) ([]CrossDomainMessengerRelayedMessageEvent, error) {
+	crossDomainMessengerAbi, err := bindings.CrossDomainMessengerMetaData.GetAbi()
+	if err != nil {
+		return nil, err
+	}
+
+	relayedMessageEventAbi := crossDomainMessengerAbi.Events["RelayedMessage"]
+	contractEventFilter := database.ContractEvent{ContractAddress: contractAddress, EventSignature: relayedMessageEventAbi.ID}
+	relayedMessageEvents, err := db.ContractEvents.ContractEventsWithFilter(contractEventFilter, chainSelector, fromHeight, toHeight)
+	if err != nil {
+		return nil, err
+	}
+
+	crossDomainRelayedMessages := make([]CrossDomainMessengerRelayedMessageEvent, len(relayedMessageEvents))
+	for i := range relayedMessageEvents {
+		relayedMessage := bindings.CrossDomainMessengerRelayedMessage{Raw: *relayedMessageEvents[i].RLPLog}
+		err = UnpackLog(&relayedMessage, relayedMessageEvents[i].RLPLog, relayedMessageEventAbi.Name, crossDomainMessengerAbi)
+		if err != nil {
+			return nil, err
+		}
+
+		crossDomainRelayedMessages[i] = CrossDomainMessengerRelayedMessageEvent{
+			Event:       &relayedMessageEvents[i],
+			MessageHash: relayedMessage.MsgHash,
+		}
+	}
+
+	return crossDomainRelayedMessages, nil
+}
+
+// Replica of `Hashing.sol#hashCrossDomainMessage` solidity implementation
+func CrossDomainMessageHash(abi *abi.ABI, sentMsg *bindings.CrossDomainMessengerSentMessage, value *big.Int) (common.Hash, error) {
+	version, _ := DecodeVersionedNonce(sentMsg.MessageNonce)
+	switch version {
+	case 0:
+		// Legacy Message
+		inputBytes, err := legacyCrossDomainMessengerRelayMessageMethod.Inputs.Pack(sentMsg.Sender, sentMsg.Target, sentMsg.Message, sentMsg.MessageNonce)
+		if err != nil {
+			return common.Hash{}, err
+		}
+		msgBytes := append(legacyCrossDomainMessengerRelayMessageMethod.ID, inputBytes...)
+		return crypto.Keccak256Hash(msgBytes), nil
+	case 1:
+		// Current Message
+		msgBytes, err := abi.Pack("relayMessage", sentMsg.MessageNonce, sentMsg.Sender, sentMsg.Target, value, sentMsg.GasLimit, sentMsg.Message)
+		if err != nil {
+			return common.Hash{}, err
+		}
+		return crypto.Keccak256Hash(msgBytes), nil
+	}
+
+	return common.Hash{}, fmt.Errorf("unsupported cross domain messenger version: %d", version)
+}

indexer/processors/contracts_bedrock/l2_to_l1_message_passer.go

+package contracts_bedrock
+
+import (
+	"math/big"
+
+	"github.com/ethereum-optimism/optimism/indexer/database"
+	"github.com/ethereum-optimism/optimism/op-bindings/bindings"
+	"github.com/ethereum/go-ethereum/common"
+)
+
+type L2ToL1MessagePasserMessagePassed struct {
+	Event          *database.ContractEvent
+	WithdrawalHash common.Hash
+	GasLimit       database.U256
+	Nonce          database.U256
+	Tx             database.Transaction
+}
+
+func L2ToL1MessagePasserMessagePassedEvents(contractAddress common.Address, db *database.DB, fromHeight, toHeight *big.Int) ([]L2ToL1MessagePasserMessagePassed, error) {
+	l2ToL1MessagePasserAbi, err := bindings.L2ToL1MessagePasserMetaData.GetAbi()
+	if err != nil {
+		return nil, err
+	}
+
+	messagePassedAbi := l2ToL1MessagePasserAbi.Events["MessagePassed"]
+	contractEventFilter := database.ContractEvent{ContractAddress: contractAddress, EventSignature: messagePassedAbi.ID}
+	messagePassedEvents, err := db.ContractEvents.L2ContractEventsWithFilter(contractEventFilter, fromHeight, toHeight)
+	if err != nil {
+		return nil, err
+	}
+
+	messagesPassed := make([]L2ToL1MessagePasserMessagePassed, len(messagePassedEvents))
+	for i := range messagePassedEvents {
+		messagePassed := bindings.L2ToL1MessagePasserMessagePassed{Raw: *messagePassedEvents[i].RLPLog}
+		err := UnpackLog(&messagePassed, messagePassedEvents[i].RLPLog, messagePassedAbi.Name, l2ToL1MessagePasserAbi)
+		if err != nil {
+			return nil, err
+		}
+
+		messagesPassed[i] = L2ToL1MessagePasserMessagePassed{
+			Event:          &messagePassedEvents[i].ContractEvent,
+			WithdrawalHash: messagePassed.WithdrawalHash,
+			Nonce:          database.U256{Int: messagePassed.Nonce},
+			GasLimit:       database.U256{Int: messagePassed.GasLimit},
+			Tx: database.Transaction{
+				FromAddress: messagePassed.Sender,
+				ToAddress:   messagePassed.Target,
+				Amount:      database.U256{Int: messagePassed.Value},
+				Data:        messagePassed.Data,
+				Timestamp:   messagePassedEvents[i].Timestamp,
+			},
+		}
+	}
+
+	return messagesPassed, nil
+}

indexer/processors/contracts_bedrock/optimism_portal.go

+package contracts_bedrock
+
+import (
+	"errors"
+	"math/big"
+
+	"github.com/ethereum-optimism/optimism/indexer/database"
+	"github.com/ethereum-optimism/optimism/op-bindings/bindings"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+)
+
+type OptimismPortalTransactionDepositEvent struct {
+	Event     *database.ContractEvent
+	DepositTx *types.DepositTx
+	Tx        database.Transaction
+	GasLimit  database.U256
+}
+
+type OptimismPortalWithdrawalProvenEvent struct {
+	*bindings.OptimismPortalWithdrawalProven
+	Event *database.ContractEvent
+}
+
+type OptimismPortalWithdrawalFinalizedEvent struct {
+	*bindings.OptimismPortalWithdrawalFinalized
+	Event *database.ContractEvent
+}
+
+type OptimismPortalProvenWithdrawal struct {
+	OutputRoot    [32]byte
+	Timestamp     *big.Int
+	L2OutputIndex *big.Int
+}
+
+func OptimismPortalTransactionDepositEvents(contractAddress common.Address, db *database.DB, fromHeight, toHeight *big.Int) ([]OptimismPortalTransactionDepositEvent, error) {
+	optimismPortalAbi, err := bindings.OptimismPortalMetaData.GetAbi()
+	if err != nil {
+		return nil, err
+	}
+	transactionDepositedEventAbi := optimismPortalAbi.Events["TransactionDeposited"]
+	if transactionDepositedEventAbi.ID != derive.DepositEventABIHash {
+		return nil, errors.New("op-node DepositEventABIHash & optimism portal TransactionDeposited ID mismatch")
+	}
+
+	contractEventFilter := database.ContractEvent{ContractAddress: contractAddress, EventSignature: transactionDepositedEventAbi.ID}
+	transactionDepositEvents, err := db.ContractEvents.L1ContractEventsWithFilter(contractEventFilter, fromHeight, toHeight)
+	if err != nil {
+		return nil, err
+	}
+
+	optimismPortalTxDeposits := make([]OptimismPortalTransactionDepositEvent, len(transactionDepositEvents))
+	for i := range transactionDepositEvents {
+		depositTx, err := derive.UnmarshalDepositLogEvent(transactionDepositEvents[i].RLPLog)
+		if err != nil {
+			return nil, err
+		}
+
+		txDeposit := bindings.OptimismPortalTransactionDeposited{Raw: *transactionDepositEvents[i].RLPLog}
+		err = UnpackLog(&txDeposit, transactionDepositEvents[i].RLPLog, transactionDepositedEventAbi.Name, optimismPortalAbi)
+		if err != nil {
+			return nil, err
+		}
+
+		optimismPortalTxDeposits[i] = OptimismPortalTransactionDepositEvent{
+			Event:     &transactionDepositEvents[i].ContractEvent,
+			DepositTx: depositTx,
+			GasLimit:  database.U256{Int: new(big.Int).SetUint64(depositTx.Gas)},
+			Tx: database.Transaction{
+				FromAddress: txDeposit.From,
+				ToAddress:   txDeposit.To,
+				Amount:      database.U256{Int: depositTx.Value},
+				Data:        depositTx.Data,
+				Timestamp:   transactionDepositEvents[i].Timestamp,
+			},
+		}
+	}
+
+	return optimismPortalTxDeposits, nil
+}
+
+func OptimismPortalWithdrawalProvenEvents(contractAddress common.Address, db *database.DB, fromHeight, toHeight *big.Int) ([]OptimismPortalWithdrawalProvenEvent, error) {
+	optimismPortalAbi, err := bindings.OptimismPortalMetaData.GetAbi()
+	if err != nil {
+		return nil, err
+	}
+
+	withdrawalProvenEventAbi := optimismPortalAbi.Events["WithdrawalProven"]
+	contractEventFilter := database.ContractEvent{ContractAddress: contractAddress, EventSignature: withdrawalProvenEventAbi.ID}
+	withdrawalProvenEvents, err := db.ContractEvents.L1ContractEventsWithFilter(contractEventFilter, fromHeight, toHeight)
+	if err != nil {
+		return nil, err
+	}
+
+	provenWithdrawals := make([]OptimismPortalWithdrawalProvenEvent, len(withdrawalProvenEvents))
+	for i := range withdrawalProvenEvents {
+		withdrawalProven := bindings.OptimismPortalWithdrawalProven{Raw: *withdrawalProvenEvents[i].RLPLog}
+		err := UnpackLog(&withdrawalProven, withdrawalProvenEvents[i].RLPLog, withdrawalProvenEventAbi.Name, optimismPortalAbi)
+		if err != nil {
+			return nil, err
+		}
+
+		provenWithdrawals[i] = OptimismPortalWithdrawalProvenEvent{
+			OptimismPortalWithdrawalProven: &withdrawalProven,
+			Event:                          &withdrawalProvenEvents[i].ContractEvent,
+		}
+	}
+
+	return provenWithdrawals, nil
+}
+
+func OptimismPortalWithdrawalFinalizedEvents(contractAddress common.Address, db *database.DB, fromHeight, toHeight *big.Int) ([]OptimismPortalWithdrawalFinalizedEvent, error) {
+	optimismPortalAbi, err := bindings.OptimismPortalMetaData.GetAbi()
+	if err != nil {
+		return nil, err
+	}
+
+	withdrawalFinalizedEventAbi := optimismPortalAbi.Events["WithdrawalFinalized"]
+	contractEventFilter := database.ContractEvent{ContractAddress: contractAddress, EventSignature: withdrawalFinalizedEventAbi.ID}
+	withdrawalFinalizedEvents, err := db.ContractEvents.L1ContractEventsWithFilter(contractEventFilter, fromHeight, toHeight)
+	if err != nil {
+		return nil, err
+	}
+
+	finalizedWithdrawals := make([]OptimismPortalWithdrawalFinalizedEvent, len(withdrawalFinalizedEvents))
+	for i := range withdrawalFinalizedEvents {
+		withdrawalFinalized := bindings.OptimismPortalWithdrawalFinalized{Raw: *withdrawalFinalizedEvents[i].RLPLog}
+		err := UnpackLog(&withdrawalFinalized, withdrawalFinalizedEvents[i].RLPLog, withdrawalFinalizedEventAbi.Name, optimismPortalAbi)
+		if err != nil {
+			return nil, err
+		}
+
+		finalizedWithdrawals[i] = OptimismPortalWithdrawalFinalizedEvent{
+			OptimismPortalWithdrawalFinalized: &withdrawalFinalized,
+			Event:                             &withdrawalFinalizedEvents[i].ContractEvent,
+		}
+	}
+
+	return finalizedWithdrawals, nil
+}

indexer/processors/contracts_bedrock/standard_bridge.go

+package contracts_bedrock
+
+import (
+	"math/big"
+
+	"github.com/ethereum-optimism/optimism/indexer/database"
+	"github.com/ethereum-optimism/optimism/op-bindings/bindings"
+	"github.com/ethereum-optimism/optimism/op-bindings/predeploys"
+
+	"github.com/ethereum/go-ethereum/common"
+)
+
+type StandardBridgeInitiatedEvent struct {
+	Event          *database.ContractEvent
+	BridgeTransfer database.BridgeTransfer
+}
+
+type StandardBridgeFinalizedEvent struct {
+	Event          *database.ContractEvent
+	BridgeTransfer database.BridgeTransfer
+}
+
+// StandardBridgeInitiatedEvents extracts all initiated bridge events from the contracts that follow the StandardBridge ABI. The
+// correlated CrossDomainMessenger nonce is also parsed from the associated messenger events.
+func StandardBridgeInitiatedEvents(chainSelector string, contractAddress common.Address, db *database.DB, fromHeight, toHeight *big.Int) ([]StandardBridgeInitiatedEvent, error) {
+	ethBridgeInitiatedEvents, err := _standardBridgeInitiatedEvents[bindings.StandardBridgeETHBridgeInitiated](contractAddress, chainSelector, db, fromHeight, toHeight)
+	if err != nil {
+		return nil, err
+	}
+
+	erc20BridgeInitiatedEvents, err := _standardBridgeInitiatedEvents[bindings.StandardBridgeERC20BridgeInitiated](contractAddress, chainSelector, db, fromHeight, toHeight)
+	if err != nil {
+		return nil, err
+	}
+
+	return append(ethBridgeInitiatedEvents, erc20BridgeInitiatedEvents...), nil
+}
+
+// StandardBridgeFinalizedEvents extracts all finalization bridge events from the contracts that follow the StandardBridge ABI. The
+// correlated CrossDomainMessenger nonce is also parsed by looking at the parameters of the corresponding relayMessage transaction data.
+func StandardBridgeFinalizedEvents(chainSelector string, contractAddress common.Address, db *database.DB, fromHeight, toHeight *big.Int) ([]StandardBridgeFinalizedEvent, error) {
+	ethBridgeFinalizedEvents, err := _standardBridgeFinalizedEvents[bindings.StandardBridgeETHBridgeFinalized](contractAddress, chainSelector, db, fromHeight, toHeight)
+	if err != nil {
+		return nil, err
+	}
+
+	erc20BridgeFinalizedEvents, err := _standardBridgeFinalizedEvents[bindings.StandardBridgeERC20BridgeFinalized](contractAddress, chainSelector, db, fromHeight, toHeight)
+	if err != nil {
+		return nil, err
+	}
+
+	return append(ethBridgeFinalizedEvents, erc20BridgeFinalizedEvents...), nil
+}
+
+// parse out eth or erc20 bridge initiated events
+func _standardBridgeInitiatedEvents[BridgeEventType bindings.StandardBridgeETHBridgeInitiated | bindings.StandardBridgeERC20BridgeInitiated](
+	contractAddress common.Address, chainSelector string, db *database.DB, fromHeight, toHeight *big.Int,
+) ([]StandardBridgeInitiatedEvent, error) {
+	standardBridgeAbi, err := bindings.StandardBridgeMetaData.GetAbi()
+	if err != nil {
+		return nil, err
+	}
+
+	var eventType BridgeEventType
+	var eventName string
+	switch any(eventType).(type) {
+	case bindings.StandardBridgeETHBridgeInitiated:
+		eventName = "ETHBridgeInitiated"
+	case bindings.StandardBridgeERC20BridgeInitiated:
+		eventName = "ERC20BridgeInitiated"
+	default:
+		panic("should not be here")
+	}
+
+	initiatedBridgeEventAbi := standardBridgeAbi.Events[eventName]
+	contractEventFilter := database.ContractEvent{ContractAddress: contractAddress, EventSignature: initiatedBridgeEventAbi.ID}
+	initiatedBridgeEvents, err := db.ContractEvents.ContractEventsWithFilter(contractEventFilter, chainSelector, fromHeight, toHeight)
+	if err != nil {
+		return nil, err
+	}
+
+	standardBridgeInitiatedEvents := make([]StandardBridgeInitiatedEvent, len(initiatedBridgeEvents))
+	for i := range initiatedBridgeEvents {
+		erc20Bridge := bindings.StandardBridgeERC20BridgeInitiated{Raw: *initiatedBridgeEvents[i].RLPLog}
+		err := UnpackLog(&erc20Bridge, initiatedBridgeEvents[i].RLPLog, eventName, standardBridgeAbi)
+		if err != nil {
+			return nil, err
+		}
+
+		// If an ETH bridge, lets fill in the needed fields
+		switch any(eventType).(type) {
+		case bindings.StandardBridgeETHBridgeInitiated:
+			erc20Bridge.LocalToken = predeploys.LegacyERC20ETHAddr
+			erc20Bridge.RemoteToken = predeploys.LegacyERC20ETHAddr
+		}
+
+		standardBridgeInitiatedEvents[i] = StandardBridgeInitiatedEvent{
+			Event: &initiatedBridgeEvents[i],
+			BridgeTransfer: database.BridgeTransfer{
+				TokenPair: database.TokenPair{LocalTokenAddress: erc20Bridge.LocalToken, RemoteTokenAddress: erc20Bridge.RemoteToken},
+				Tx: database.Transaction{
+					FromAddress: erc20Bridge.From,
+					ToAddress:   erc20Bridge.To,
+					Amount:      database.U256{Int: erc20Bridge.Amount},
+					Data:        erc20Bridge.ExtraData,
+					Timestamp:   initiatedBridgeEvents[i].Timestamp,
+				},
+			},
+		}
+	}
+
+	return standardBridgeInitiatedEvents, nil
+}
+
+// parse out eth or erc20 bridge finalization events
+func _standardBridgeFinalizedEvents[BridgeEventType bindings.StandardBridgeETHBridgeFinalized | bindings.StandardBridgeERC20BridgeFinalized](
+	contractAddress common.Address, chainSelector string, db *database.DB, fromHeight, toHeight *big.Int,
+) ([]StandardBridgeFinalizedEvent, error) {
+	standardBridgeAbi, err := bindings.StandardBridgeMetaData.GetAbi()
+	if err != nil {
+		return nil, err
+	}
+
+	var eventType BridgeEventType
+	var eventName string
+	switch any(eventType).(type) {
+	case bindings.StandardBridgeETHBridgeFinalized:
+		eventName = "ETHBridgeFinalized"
+	case bindings.StandardBridgeERC20BridgeFinalized:
+		eventName = "ERC20BridgeFinalized"
+	default:
+		panic("should not be here")
+	}
+
+	bridgeFinalizedEventAbi := standardBridgeAbi.Events[eventName]
+	contractEventFilter := database.ContractEvent{ContractAddress: contractAddress, EventSignature: bridgeFinalizedEventAbi.ID}
+	bridgeFinalizedEvents, err := db.ContractEvents.ContractEventsWithFilter(contractEventFilter, chainSelector, fromHeight, toHeight)
+	if err != nil {
+		return nil, err
+	}
+
+	standardBridgeFinalizedEvents := make([]StandardBridgeFinalizedEvent, len(bridgeFinalizedEvents))
+	for i := range bridgeFinalizedEvents {
+		erc20Bridge := bindings.StandardBridgeERC20BridgeFinalized{Raw: *bridgeFinalizedEvents[i].RLPLog}
+		err := UnpackLog(&erc20Bridge, bridgeFinalizedEvents[i].RLPLog, eventName, standardBridgeAbi)
+		if err != nil {
+			return nil, err
+		}
+
+		// If an ETH bridge, lets fill in the needed fields
+		switch any(eventType).(type) {
+		case bindings.StandardBridgeETHBridgeFinalized:
+			erc20Bridge.LocalToken = predeploys.LegacyERC20ETHAddr
+			erc20Bridge.RemoteToken = predeploys.LegacyERC20ETHAddr
+		}
+
+		standardBridgeFinalizedEvents[i] = StandardBridgeFinalizedEvent{
+			Event: &bridgeFinalizedEvents[i],
+			BridgeTransfer: database.BridgeTransfer{
+				TokenPair: database.TokenPair{LocalTokenAddress: erc20Bridge.LocalToken, RemoteTokenAddress: erc20Bridge.RemoteToken},
+				Tx: database.Transaction{
+					FromAddress: erc20Bridge.From,
+					ToAddress:   erc20Bridge.To,
+					Amount:      database.U256{Int: erc20Bridge.Amount},
+					Data:        erc20Bridge.ExtraData,
+					Timestamp:   bridgeFinalizedEvents[i].Timestamp,
+				},
+			},
+		}
+	}
+
+	return standardBridgeFinalizedEvents, nil
+}

indexer/processor/contract_events.go → indexer/processors/contracts_bedrock/utils.go

-package processor
+package contracts_bedrock
 
 import (
+	"encoding/binary"
 	"errors"
 	"fmt"
-
-	"github.com/ethereum-optimism/optimism/indexer/database"
+	"math/big"
 
 	"github.com/ethereum/go-ethereum/accounts/abi"
-	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 )
 
-type ProcessedContractEventLogIndexKey struct {
-	blockHash common.Hash
-	index     uint
-}
-
-type ProcessedContractEvents struct {
-	events            []*database.ContractEvent
-	eventsBySignature map[common.Hash][]*database.ContractEvent
-	eventByLogIndex   map[ProcessedContractEventLogIndexKey]*database.ContractEvent
-}
-
-func NewProcessedContractEvents() *ProcessedContractEvents {
-	return &ProcessedContractEvents{
-		events:            []*database.ContractEvent{},
-		eventsBySignature: make(map[common.Hash][]*database.ContractEvent),
-		eventByLogIndex:   make(map[ProcessedContractEventLogIndexKey]*database.ContractEvent),
+// DecodeVersionNonce is an re-implementation of Encoding.sol#decodeVersionedNonce.
+// If the nonce is greater than 32 bytes (solidity uint256), bytes [32:] are ignored
+func DecodeVersionedNonce(nonce *big.Int) (uint16, *big.Int) {
+	nonceBytes := nonce.Bytes()
+	nonceByteLen := len(nonceBytes)
+	if nonceByteLen < 30 {
+		// version is 0x0000
+		return 0, nonce
+	} else if nonceByteLen == 31 {
+		// version is 0x00[01..ff]
+		return uint16(nonceBytes[0]), new(big.Int).SetBytes(nonceBytes[1:])
+	} else {
+		// fully specified
+		version := binary.BigEndian.Uint16(nonceBytes[:2])
+		return version, new(big.Int).SetBytes(nonceBytes[2:])
 	}
 }
 
-func (p *ProcessedContractEvents) AddLog(log *types.Log, time uint64) *database.ContractEvent {
-	event := database.ContractEventFromLog(log, time)
-	emptyHash := common.Hash{}
-
-	p.events = append(p.events, &event)
-	p.eventByLogIndex[ProcessedContractEventLogIndexKey{log.BlockHash, log.Index}] = &event
-	if event.EventSignature != emptyHash {
-		p.eventsBySignature[event.EventSignature] = append(p.eventsBySignature[event.EventSignature], &event)
-	}
-
-	return &event
-}
-
 func UnpackLog(out interface{}, log *types.Log, name string, contractAbi *abi.ABI) error {
 	eventAbi, ok := contractAbi.Events[name]
 	if !ok {