Merge pull request #6608 from ethereum-optimism/08-03-bridge_messages._schema_models

feat(indexer): Index CrossDomainMessenger messages

indexer/database/bridge_messages.go

+package database
+
+import (
+	"errors"
+	"fmt"
+	"math/big"
+
+	"gorm.io/gorm"
+
+	"github.com/ethereum/go-ethereum/common"
+
+	"github.com/google/uuid"
+)
+
+/**
+ * Types
+ */
+
+type BridgeMessage struct {
+	Nonce       U256        `gorm:"primaryKey"`
+	MessageHash common.Hash `gorm:"serializer:json"`
+
+	SentMessageEventGUID    uuid.UUID
+	RelayedMessageEventGUID *uuid.UUID
+
+	Tx       Transaction `gorm:"embedded"`
+	GasLimit U256
+}
+
+type L1BridgeMessage struct {
+	BridgeMessage         `gorm:"embedded"`
+	TransactionSourceHash common.Hash `gorm:"serializer:json"`
+}
+
+type L2BridgeMessage struct {
+	BridgeMessage             `gorm:"embedded"`
+	TransactionWithdrawalHash common.Hash `gorm:"serializer:json"`
+}
+
+type BridgeMessagesView interface {
+	L1BridgeMessage(*big.Int) (*L1BridgeMessage, error)
+	L1BridgeMessageByHash(common.Hash) (*L1BridgeMessage, error)
+	LatestL1BridgeMessageNonce() (*big.Int, error)
+
+	L2BridgeMessage(*big.Int) (*L2BridgeMessage, error)
+	L2BridgeMessageByHash(common.Hash) (*L2BridgeMessage, error)
+	LatestL2BridgeMessageNonce() (*big.Int, error)
+}
+
+type BridgeMessagesDB interface {
+	BridgeMessagesView
+
+	StoreL1BridgeMessages([]*L1BridgeMessage) error
+	MarkRelayedL1BridgeMessage(common.Hash, uuid.UUID) error
+
+	StoreL2BridgeMessages([]*L2BridgeMessage) error
+	MarkRelayedL2BridgeMessage(common.Hash, uuid.UUID) error
+}
+
+/**
+ * Implementation
+ */
+
+type bridgeMessagesDB struct {
+	gorm *gorm.DB
+}
+
+func newBridgeMessagesDB(db *gorm.DB) BridgeMessagesDB {
+	return &bridgeMessagesDB{gorm: db}
+}
+
+/**
+ * Arbitrary Messages Sent from L1
+ */
+
+func (db bridgeMessagesDB) StoreL1BridgeMessages(messages []*L1BridgeMessage) error {
+	result := db.gorm.Create(&messages)
+	return result.Error
+}
+
+func (db bridgeMessagesDB) L1BridgeMessage(nonce *big.Int) (*L1BridgeMessage, error) {
+	var sentMessage L1BridgeMessage
+	result := db.gorm.Where(&BridgeMessage{Nonce: U256{Int: nonce}}).Take(&sentMessage)
+	if result.Error != nil {
+		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+			return nil, nil
+		}
+		return nil, result.Error
+	}
+
+	return &sentMessage, nil
+}
+
+func (db bridgeMessagesDB) L1BridgeMessageByHash(messageHash common.Hash) (*L1BridgeMessage, error) {
+	var sentMessage L1BridgeMessage
+	result := db.gorm.Where(&BridgeMessage{MessageHash: messageHash}).Take(&sentMessage)
+	if result.Error != nil {
+		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+			return nil, nil
+		}
+		return nil, result.Error
+	}
+
+	return &sentMessage, nil
+}
+
+func (db bridgeMessagesDB) LatestL1BridgeMessageNonce() (*big.Int, error) {
+	var sentMessage L1BridgeMessage
+	result := db.gorm.Order("nonce DESC").Take(&sentMessage)
+	if result.Error != nil {
+		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+			return nil, nil
+		}
+		return nil, result.Error
+	}
+
+	return sentMessage.Nonce.Int, nil
+}
+
+/**
+ * Arbitrary Messages Sent from L2
+ */
+
+func (db bridgeMessagesDB) MarkRelayedL1BridgeMessage(messageHash common.Hash, relayEvent uuid.UUID) error {
+	message, err := db.L1BridgeMessageByHash(messageHash)
+	if err != nil {
+		return err
+	} else if message == nil {
+		return fmt.Errorf("L1BridgeMessage with message hash %s not found", messageHash)
+	}
+
+	message.RelayedMessageEventGUID = &relayEvent
+	result := db.gorm.Save(message)
+	return result.Error
+}
+
+func (db bridgeMessagesDB) StoreL2BridgeMessages(messages []*L2BridgeMessage) error {
+	result := db.gorm.Create(&messages)
+	return result.Error
+}
+
+func (db bridgeMessagesDB) L2BridgeMessage(nonce *big.Int) (*L2BridgeMessage, error) {
+	var sentMessage L2BridgeMessage
+	result := db.gorm.Where(&BridgeMessage{Nonce: U256{Int: nonce}}).Take(&sentMessage)
+	if result.Error != nil {
+		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+			return nil, nil
+		}
+		return nil, result.Error
+	}
+
+	return &sentMessage, nil
+}
+
+func (db bridgeMessagesDB) L2BridgeMessageByHash(messageHash common.Hash) (*L2BridgeMessage, error) {
+	var sentMessage L2BridgeMessage
+	result := db.gorm.Where(&BridgeMessage{MessageHash: messageHash}).Take(&sentMessage)
+	if result.Error != nil {
+		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+			return nil, nil
+		}
+		return nil, result.Error
+	}
+
+	return &sentMessage, nil
+}
+
+func (db bridgeMessagesDB) LatestL2BridgeMessageNonce() (*big.Int, error) {
+	var sentMessage L2BridgeMessage
+	result := db.gorm.Order("nonce DESC").Take(&sentMessage)
+	if result.Error != nil {
+		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+			return nil, nil
+		}
+		return nil, result.Error
+	}
+
+	return sentMessage.Nonce.Int, nil
+}
+
+func (db bridgeMessagesDB) MarkRelayedL2BridgeMessage(messageHash common.Hash, relayEvent uuid.UUID) error {
+	message, err := db.L2BridgeMessageByHash(messageHash)
+	if err != nil {
+		return err
+	} else if message == nil {
+		return fmt.Errorf("L2BridgeMessage with message hash %s not found", messageHash)
+	}
+
+	message.RelayedMessageEventGUID = &relayEvent
+	result := db.gorm.Save(message)
+	return result.Error
+}

indexer/database/db.go

@@ -13,6 +13,7 @@ type DB struct {
 	Blocks             BlocksDB
 	ContractEvents     ContractEventsDB
 	BridgeTransfers    BridgeTransfersDB
+	BridgeMessages     BridgeMessagesDB
 	BridgeTransactions BridgeTransactionsDB
 }
 
@@ -35,8 +36,9 @@ func NewDB(dsn string) (*DB, error) {
 		gorm:               gorm,
 		Blocks:             newBlocksDB(gorm),
 		ContractEvents:     newContractEventsDB(gorm),
-		BridgeTransactions: newBridgeTransactionsDB(gorm),
 		BridgeTransfers:    newBridgeTransfersDB(gorm),
+		BridgeMessages:     newBridgeMessagesDB(gorm),
+		BridgeTransactions: newBridgeTransactionsDB(gorm),
 	}
 
 	return db, nil
@@ -64,7 +66,8 @@ func dbFromGormTx(tx *gorm.DB) *DB {
 		gorm:               tx,
 		Blocks:             newBlocksDB(tx),
 		ContractEvents:     newContractEventsDB(tx),
-		BridgeTransactions: newBridgeTransactionsDB(tx),
 		BridgeTransfers:    newBridgeTransfersDB(tx),
+		BridgeMessages:     newBridgeMessagesDB(tx),
+		BridgeTransactions: newBridgeTransactionsDB(tx),
 	}
 }

indexer/e2e_tests/bridge_messages_e2e_test.go

+package e2e_tests
+
+import (
+	"context"
+	"math/big"
+	"testing"
+	"time"
+
+	e2etest_utils "github.com/ethereum-optimism/optimism/indexer/e2e_tests/utils"
+	"github.com/ethereum-optimism/optimism/op-bindings/bindings"
+	"github.com/ethereum-optimism/optimism/op-bindings/predeploys"
+	op_e2e "github.com/ethereum-optimism/optimism/op-e2e"
+	"github.com/ethereum-optimism/optimism/op-node/withdrawals"
+	"github.com/ethereum-optimism/optimism/op-service/client/utils"
+
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+	"github.com/ethereum/go-ethereum/params"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestE2EBridgeL1CrossDomainMessenger(t *testing.T) {
+	testSuite := createE2ETestSuite(t)
+
+	l1CrossDomainMessenger, err := bindings.NewL1CrossDomainMessenger(testSuite.OpCfg.L1Deployments.L1CrossDomainMessengerProxy, testSuite.L1Client)
+	require.NoError(t, err)
+
+	aliceAddr := testSuite.OpCfg.Secrets.Addresses().Alice
+
+	// Attach 1ETH and random calldata to the sent messages
+	calldata := []byte{byte(1), byte(2), byte(3)}
+	l1Opts, err := bind.NewKeyedTransactorWithChainID(testSuite.OpCfg.Secrets.Alice, testSuite.OpCfg.L1ChainIDBig())
+	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)
+	sentMsgReceipt, err := utils.WaitReceiptOK(context.Background(), testSuite.L1Client, sentMsgTx.Hash())
+	require.NoError(t, err)
+
+	depositInfo, err := e2etest_utils.ParseDepositInfo(sentMsgReceipt)
+	require.NoError(t, err)
+
+	// wait for processor catchup
+	require.NoError(t, utils.WaitFor(context.Background(), 500*time.Millisecond, func() (bool, error) {
+		l1Header := testSuite.Indexer.L1Processor.LatestProcessedHeader()
+		return l1Header != nil && l1Header.Number.Uint64() >= sentMsgReceipt.BlockNumber.Uint64(), nil
+	}))
+
+	parsedMessage, err := e2etest_utils.ParseCrossDomainMessage(sentMsgReceipt)
+	require.NoError(t, err)
+
+	// nonce for this message is zero but the current cross domain message version is 1.
+	nonceBytes := [31]byte{0: byte(1)}
+	nonce := new(big.Int).SetBytes(nonceBytes[:])
+
+	sentMessage, err := testSuite.DB.BridgeMessages.L1BridgeMessage(nonce)
+	require.NoError(t, err)
+	require.NotNil(t, sentMessage)
+	require.NotNil(t, sentMessage.SentMessageEventGUID)
+	require.Equal(t, depositInfo.DepositTx.SourceHash, sentMessage.TransactionSourceHash)
+	require.Equal(t, parsedMessage.MessageHash, sentMessage.MessageHash)
+	require.Equal(t, uint64(100_000), sentMessage.GasLimit.Int.Uint64())
+	require.Equal(t, big.NewInt(params.Ether), sentMessage.Tx.Amount.Int)
+	require.Equal(t, aliceAddr, sentMessage.Tx.FromAddress)
+	require.Equal(t, aliceAddr, sentMessage.Tx.ToAddress)
+	require.ElementsMatch(t, calldata, sentMessage.Tx.Data)
+
+	// (2) Process RelayedMesssage on inclusion
+	require.Nil(t, sentMessage.RelayedMessageEventGUID)
+	testSuite.Indexer.L2Processor.ResumeForTest()
+	transaction, err := testSuite.DB.BridgeTransactions.L1TransactionDeposit(sentMessage.TransactionSourceHash)
+	require.NoError(t, err)
+
+	// wait for processor catchup
+	depositReceipt, err := utils.WaitReceiptOK(context.Background(), testSuite.L2Client, transaction.L2TransactionHash)
+	require.NoError(t, err)
+	require.NoError(t, utils.WaitFor(context.Background(), 500*time.Millisecond, func() (bool, error) {
+		l2Header := testSuite.Indexer.L2Processor.LatestProcessedHeader()
+		return l2Header != nil && l2Header.Number.Uint64() >= depositReceipt.BlockNumber.Uint64(), nil
+	}))
+
+	sentMessage, err = testSuite.DB.BridgeMessages.L1BridgeMessage(nonce)
+	require.NoError(t, err)
+	require.NotNil(t, sentMessage)
+	require.NotNil(t, sentMessage.RelayedMessageEventGUID)
+
+	event, err := testSuite.DB.ContractEvents.L2ContractEvent(*sentMessage.RelayedMessageEventGUID)
+	require.NoError(t, err)
+	require.NotNil(t, event)
+	require.Equal(t, event.TransactionHash, transaction.L2TransactionHash)
+}
+
+func TestE2EBridgeL2CrossDomainMessenger(t *testing.T) {
+	testSuite := createE2ETestSuite(t)
+
+	optimismPortal, err := bindings.NewOptimismPortal(testSuite.OpCfg.L1Deployments.OptimismPortalProxy, testSuite.L1Client)
+	require.NoError(t, err)
+	l2CrossDomainMessenger, err := bindings.NewL2CrossDomainMessenger(predeploys.L2CrossDomainMessengerAddr, testSuite.L2Client)
+	require.NoError(t, err)
+
+	aliceAddr := testSuite.OpCfg.Secrets.Addresses().Alice
+
+	// Attach 1ETH and random calldata to the sent messages
+	calldata := []byte{byte(1), byte(2), byte(3)}
+	l2Opts, err := bind.NewKeyedTransactorWithChainID(testSuite.OpCfg.Secrets.Alice, testSuite.OpCfg.L2ChainIDBig())
+	require.NoError(t, err)
+	l2Opts.Value = big.NewInt(params.Ether)
+
+	// Ensure L1 has enough funds for the withdrawal by depositing an equal amount into the OptimismPortal
+	l1Opts, err := bind.NewKeyedTransactorWithChainID(testSuite.OpCfg.Secrets.Alice, testSuite.OpCfg.L1ChainIDBig())
+	require.NoError(t, err)
+	l1Opts.Value = l2Opts.Value
+	depositTx, err := optimismPortal.Receive(l1Opts)
+	require.NoError(t, err)
+	_, err = utils.WaitReceiptOK(context.Background(), testSuite.L1Client, depositTx.Hash())
+	require.NoError(t, err)
+
+	// (1) Send the Message
+	sentMsgTx, err := l2CrossDomainMessenger.SendMessage(l2Opts, aliceAddr, calldata, 100_000)
+	require.NoError(t, err)
+	sentMsgReceipt, err := utils.WaitReceiptOK(context.Background(), testSuite.L2Client, sentMsgTx.Hash())
+	require.NoError(t, err)
+
+	msgPassed, err := withdrawals.ParseMessagePassed(sentMsgReceipt)
+	require.NoError(t, err)
+	withdrawalHash, err := withdrawals.WithdrawalHash(msgPassed)
+	require.NoError(t, err)
+
+	// wait for processor catchup
+	require.NoError(t, utils.WaitFor(context.Background(), 500*time.Millisecond, func() (bool, error) {
+		l2Header := testSuite.Indexer.L2Processor.LatestProcessedHeader()
+		return l2Header != nil && l2Header.Number.Uint64() >= sentMsgReceipt.BlockNumber.Uint64(), nil
+	}))
+
+	parsedMessage, err := e2etest_utils.ParseCrossDomainMessage(sentMsgReceipt)
+	require.NoError(t, err)
+
+	// nonce for this message is zero but the current message version is 1.
+	nonceBytes := [31]byte{0: byte(1)}
+	nonce := new(big.Int).SetBytes(nonceBytes[:])
+
+	sentMessage, err := testSuite.DB.BridgeMessages.L2BridgeMessage(nonce)
+	require.NoError(t, err)
+	require.NotNil(t, sentMessage)
+	require.NotNil(t, sentMessage.SentMessageEventGUID)
+	require.Equal(t, withdrawalHash, sentMessage.TransactionWithdrawalHash)
+	require.Equal(t, parsedMessage.MessageHash, sentMessage.MessageHash)
+	require.Equal(t, uint64(100_000), sentMessage.GasLimit.Int.Uint64())
+	require.Equal(t, big.NewInt(params.Ether), sentMessage.Tx.Amount.Int)
+	require.Equal(t, aliceAddr, sentMessage.Tx.FromAddress)
+	require.Equal(t, aliceAddr, sentMessage.Tx.ToAddress)
+	require.ElementsMatch(t, calldata, sentMessage.Tx.Data)
+
+	// (2) Process RelayedMessage on withdrawal finalization
+	require.Nil(t, sentMessage.RelayedMessageEventGUID)
+	_, finalizedReceipt := op_e2e.ProveAndFinalizeWithdrawal(t, *testSuite.OpCfg, testSuite.L1Client, testSuite.OpSys.Nodes["sequencer"], testSuite.OpCfg.Secrets.Alice, sentMsgReceipt)
+
+	// wait for processor catchup
+	require.NoError(t, utils.WaitFor(context.Background(), 500*time.Millisecond, func() (bool, error) {
+		l1Header := testSuite.Indexer.L1Processor.LatestProcessedHeader()
+		return l1Header != nil && l1Header.Number.Uint64() >= finalizedReceipt.BlockNumber.Uint64(), nil
+	}))
+
+	// message is marked as relayed
+	sentMessage, err = testSuite.DB.BridgeMessages.L2BridgeMessage(nonce)
+	require.NoError(t, err)
+	require.NotNil(t, sentMessage)
+	require.NotNil(t, sentMessage.RelayedMessageEventGUID)
+
+	event, err := testSuite.DB.ContractEvents.L1ContractEvent(*sentMessage.RelayedMessageEventGUID)
+	require.NoError(t, err)
+	require.NotNil(t, event)
+	require.Equal(t, event.TransactionHash, finalizedReceipt.TxHash)
+}

indexer/e2e_tests/bridge_transfers_e2e_test.go

@@ -70,8 +70,18 @@ func TestE2EBridgeTransfersStandardBridgeETHDeposit(t *testing.T) {
 	_, nonce := processor.DecodeVersionedNonce(deposit.CrossDomainMessengerNonce.Int)
 	require.Zero(t, nonce.Uint64())
 
-	// (2) Test Deposit Finalization
-	// Nothing to do as we rely on the derivation process to include the deposit
+	// (2) Test Deposit Finalization via CrossDomainMessenger relayed message
+	depositReceipt, err = utils.WaitReceiptOK(context.Background(), testSuite.L2Client, types.NewTx(depositInfo.DepositTx).Hash())
+	require.NoError(t, err)
+	require.NoError(t, utils.WaitFor(context.Background(), 500*time.Millisecond, func() (bool, error) {
+		l2Header := testSuite.Indexer.L2Processor.LatestProcessedHeader()
+		return l2Header != nil && l2Header.Number.Uint64() >= depositReceipt.BlockNumber.Uint64(), nil
+	}))
+
+	crossDomainBridgeMessage, err := testSuite.DB.BridgeMessages.L1BridgeMessage(deposit.CrossDomainMessengerNonce.Int)
+	require.NoError(t, err)
+	require.NotNil(t, crossDomainBridgeMessage)
+	require.NotNil(t, crossDomainBridgeMessage.RelayedMessageEventGUID)
 }
 
 func TestE2EBridgeTransfersOptimismPortalETHReceive(t *testing.T) {

indexer/e2e_tests/utils/cross_domain_messages.go

+package utils
+
+import (
+	"errors"
+	"math/big"
+
+	"github.com/ethereum-optimism/optimism/indexer/processor"
+	"github.com/ethereum-optimism/optimism/op-bindings/bindings"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+)
+
+type CrossDomainMessengerSentMessage struct {
+	*bindings.CrossDomainMessengerSentMessage
+	Value       *big.Int
+	MessageHash common.Hash
+}
+
+func ParseCrossDomainMessage(sentMessageReceipt *types.Receipt) (CrossDomainMessengerSentMessage, error) {
+	abi, err := bindings.CrossDomainMessengerMetaData.GetAbi()
+	if err != nil {
+		return CrossDomainMessengerSentMessage{}, err
+	}
+
+	sentMessageEventAbi := abi.Events["SentMessage"]
+	messenger, err := bindings.NewCrossDomainMessenger(common.Address{}, nil)
+	if err != nil {
+		return CrossDomainMessengerSentMessage{}, err
+	}
+
+	for i, log := range sentMessageReceipt.Logs {
+		if len(log.Topics) > 0 && log.Topics[0] == sentMessageEventAbi.ID {
+			sentMessage, err := messenger.ParseSentMessage(*log)
+			if err != nil {
+				return CrossDomainMessengerSentMessage{}, err
+			}
+			sentMessageExtension, err := messenger.ParseSentMessageExtension1(*sentMessageReceipt.Logs[i+1])
+			if err != nil {
+				return CrossDomainMessengerSentMessage{}, err
+			}
+			msgHash, err := CrossDomainMessengerSentMessageHash(sentMessage, sentMessageExtension.Value)
+			if err != nil {
+				return CrossDomainMessengerSentMessage{}, err
+			}
+
+			return CrossDomainMessengerSentMessage{sentMessage, sentMessageExtension.Value, msgHash}, nil
+		}
+	}
+
+	return CrossDomainMessengerSentMessage{}, errors.New("missing SentMessage receipts")
+}
+
+func CrossDomainMessengerSentMessageHash(sentMessage *bindings.CrossDomainMessengerSentMessage, value *big.Int) (common.Hash, error) {
+	abi, err := bindings.CrossDomainMessengerMetaData.GetAbi()
+	if err != nil {
+		return common.Hash{}, err
+	}
+
+	return processor.CrossDomainMessageHash(abi, sentMessage, value)
+}

indexer/migrations/20230523_create_schema.sql

@@ -108,13 +108,47 @@ CREATE TABLE IF NOT EXISTS l2_transaction_withdrawals (
     timestamp    INTEGER NOT NULL CHECK (timestamp > 0)
 );
 
+-- CrossDomainMessenger
+CREATE TABLE IF NOT EXISTS l1_bridge_messages(
+    nonce                   UINT256 NOT NULL PRIMARY KEY,
+    message_hash            VARCHAR NOT NULL,
+    transaction_source_hash VARCHAR NOT NULL REFERENCES l1_transaction_deposits(source_hash),
+
+    sent_message_event_guid    VARCHAR NOT NULL UNIQUE REFERENCES l1_contract_events(guid),
+    relayed_message_event_guid VARCHAR UNIQUE REFERENCES l2_contract_events(guid),
+
+    -- sent message
+    from_address VARCHAR NOT NULL,
+    to_address   VARCHAR NOT NULL,
+    amount       UINT256 NOT NULL,
+    gas_limit    UINT256 NOT NULL,
+    data         VARCHAR NOT NULL,
+    timestamp    INTEGER NOT NULL CHECK (timestamp > 0)
+);
+CREATE TABLE IF NOT EXISTS l2_bridge_messages(
+    nonce                       UINT256 NOT NULL PRIMARY KEY,
+    message_hash                VARCHAR NOT NULL,
+    transaction_withdrawal_hash VARCHAR NOT NULL REFERENCES l2_transaction_withdrawals(withdrawal_hash),
+
+    sent_message_event_guid    VARCHAR NOT NULL UNIQUE REFERENCES l2_contract_events(guid),
+    relayed_message_event_guid VARCHAR UNIQUE REFERENCES l1_contract_events(guid),
+
+    -- sent message
+    from_address VARCHAR NOT NULL,
+    to_address   VARCHAR NOT NULL,
+    amount       UINT256 NOT NULL,
+    gas_limit    UINT256 NOT NULL,
+    data         VARCHAR NOT NULL,
+    timestamp    INTEGER NOT NULL CHECK (timestamp > 0)
+);
+
 -- StandardBridge
 CREATE TABLE IF NOT EXISTS l1_bridge_deposits (
     transaction_source_hash VARCHAR PRIMARY KEY REFERENCES l1_transaction_deposits(source_hash),
 
     -- We allow the cross_domain_messenger_nonce to be NULL-able to account
     -- for scenarios where ETH is simply sent to the OptimismPortal contract
-    cross_domain_messenger_nonce UINT256 UNIQUE,
+    cross_domain_messenger_nonce UINT256 UNIQUE REFERENCES l1_bridge_messages(nonce),
 
     -- Deposit information
 	from_address     VARCHAR NOT NULL,
@@ -130,7 +164,7 @@ CREATE TABLE IF NOT EXISTS l2_bridge_withdrawals (
 
     -- We allow the cross_domain_messenger_nonce to be NULL-able to account for
     -- scenarios where ETH is simply sent to the L2ToL1MessagePasser contract
-    cross_domain_messenger_nonce UINT256 UNIQUE,
+    cross_domain_messenger_nonce UINT256 UNIQUE REFERENCES l2_bridge_messages(nonce),
 
     -- Withdrawal information
 	from_address     VARCHAR NOT NULL,

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
+	RawEvent    *database.ContractEvent
+}
+
+type CrossDomainMessengerRelayedMessageEvent struct {
+	*bindings.CrossDomainMessengerRelayedMessage
+	RawEvent *database.ContractEvent
+}
+
+func CrossDomainMessengerSentMessageEvents(events *ProcessedContractEvents) ([]CrossDomainMessengerSentMessageEvent, error) {
+	crossDomainMessengerABI, err := bindings.L1CrossDomainMessengerMetaData.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 := events.eventLog[sentMessageEvent.GUID]
+
+		var sentMsgData bindings.CrossDomainMessengerSentMessage
+		err = UnpackLog(&sentMsgData, log, sentMessageEventAbi.Name, crossDomainMessengerABI)
+		if err != nil {
+			return nil, err
+		}
+
+		var sentMsgExtensionData bindings.CrossDomainMessengerSentMessageExtension1
+		extensionLog := events.eventLog[events.eventByLogIndex[ProcessedContractEventLogIndexKey{log.BlockHash, log.Index + 1}].GUID]
+		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,
+			RawEvent:                        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 := events.eventLog[relayedMessageEvent.GUID]
+
+		var relayedMsgData bindings.CrossDomainMessengerRelayedMessage
+		err = UnpackLog(&relayedMsgData, log, relayedMessageEventAbi.Name, crossDomainMessengerABI)
+		if err != nil {
+			return nil, err
+		}
+
+		crossDomainMessageEvents[i] = CrossDomainMessengerRelayedMessageEvent{&relayedMsgData, 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/l1_processor.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"fmt"
 	"math/big"
 	"reflect"
 
@@ -240,6 +241,12 @@ func l1ProcessFn(processLog log.Logger, ethClient node.EthClient, l1Contracts L1
 				return err
 			}
 
+			// forward along contract events to bridge messages processor
+			err = l1ProcessContractEventsBridgeCrossDomainMessages(processLog, db, processedContractEvents)
+			if err != nil {
+				return err
+			}
+
 			// forward along contract events to standard bridge processor
 			err = l1ProcessContractEventsStandardBridge(processLog, db, ethClient, processedContractEvents)
 			if err != nil {
@@ -383,6 +390,84 @@ func l1ProcessContractEventsBridgeTransactions(processLog log.Logger, db *databa
 	return nil
 }
 
+func l1ProcessContractEventsBridgeCrossDomainMessages(processLog log.Logger, db *database.DB, events *ProcessedContractEvents) error {
+	// (1) Process New Messages
+	sentMessageEvents, err := CrossDomainMessengerSentMessageEvents(events)
+	if err != nil {
+		return err
+	}
+
+	sentMessages := make([]*database.L1BridgeMessage, len(sentMessageEvents))
+	for i, sentMessageEvent := range sentMessageEvents {
+		log := events.eventLog[sentMessageEvent.RawEvent.GUID]
+
+		// extract the deposit hash from the previous TransactionDepositedEvent
+		transactionDepositedLog := events.eventLog[events.eventByLogIndex[ProcessedContractEventLogIndexKey{log.BlockHash, log.Index - 1}].GUID]
+		depositTx, err := derive.UnmarshalDepositLogEvent(transactionDepositedLog)
+		if err != nil {
+			return err
+		}
+
+		sentMessages[i] = &database.L1BridgeMessage{
+			TransactionSourceHash: depositTx.SourceHash,
+			BridgeMessage: database.BridgeMessage{
+				Nonce:                database.U256{Int: sentMessageEvent.MessageNonce},
+				MessageHash:          sentMessageEvent.MessageHash,
+				SentMessageEventGUID: sentMessageEvent.RawEvent.GUID,
+				GasLimit:             database.U256{Int: sentMessageEvent.GasLimit},
+				Tx: database.Transaction{
+					FromAddress: sentMessageEvent.Sender,
+					ToAddress:   sentMessageEvent.Target,
+					Amount:      database.U256{Int: sentMessageEvent.Value},
+					Data:        sentMessageEvent.Message,
+					Timestamp:   sentMessageEvent.RawEvent.Timestamp,
+				},
+			},
+		}
+	}
+
+	if len(sentMessages) > 0 {
+		processLog.Info("detected L1CrossDomainMessenger messages", "size", len(sentMessages))
+		err := db.BridgeMessages.StoreL1BridgeMessages(sentMessages)
+		if err != nil {
+			return err
+		}
+	}
+
+	// (2) Process Relayed Messages.
+	//
+	// NOTE: Should we care about failed messages? A failed message can be
+	// inferred via a finalized withdrawal that has not been marked as relayed.
+	relayedMessageEvents, err := CrossDomainMessengerRelayedMessageEvents(events)
+	if err != nil {
+		return err
+	}
+
+	for _, relayedMessage := range relayedMessageEvents {
+		message, err := db.BridgeMessages.L2BridgeMessageByHash(relayedMessage.MsgHash)
+		if err != nil {
+			return err
+		} else if message == nil {
+			// Since L2 withdrawals must be proven before being relayed, the transaction processor
+			// ensures that we are in indexed state on L2 if we've seen this finalization event
+			processLog.Crit("missing indexed L2CrossDomainMessenger sent message", "message_hash", relayedMessage.MsgHash)
+			return fmt.Errorf("missing indexed L2CrossDomainMessager mesesage: 0x%x", relayedMessage.MsgHash)
+		}
+
+		err = db.BridgeMessages.MarkRelayedL2BridgeMessage(relayedMessage.MsgHash, relayedMessage.RawEvent.GUID)
+		if err != nil {
+			return err
+		}
+	}
+
+	if len(relayedMessageEvents) > 0 {
+		processLog.Info("relayed L2CrossDomainMessenger messages", "size", len(relayedMessageEvents))
+	}
+
+	// a-ok!
+	return nil
+}
+
 func l1ProcessContractEventsStandardBridge(processLog log.Logger, db *database.DB, ethClient node.EthClient, events *ProcessedContractEvents) error {
 	rawEthClient := ethclient.NewClient(ethClient.RawRpcClient())
 

indexer/processor/l2_processor.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"fmt"
 	"reflect"
 
 	"github.com/ethereum-optimism/optimism/indexer/database"
@@ -162,6 +163,11 @@ func l2ProcessFn(processLog log.Logger, ethClient node.EthClient, l2Contracts L2
 				return err
 			}
 
+			err = l2ProcessContractEventsBridgeCrossDomainMessages(processLog, db, processedContractEvents)
+			if err != nil {
+				return err
+			}
+
 			// forward along contract events to standard bridge processor
 			err = l2ProcessContractEventsStandardBridge(processLog, db, ethClient, processedContractEvents)
 			if err != nil {
@@ -235,6 +241,104 @@ func l2ProcessContractEventsBridgeTransactions(processLog log.Logger, db *databa
 	return nil
 }
 
+func l2ProcessContractEventsBridgeCrossDomainMessages(processLog log.Logger, db *database.DB, events *ProcessedContractEvents) error {
+	l2ToL1MessagePasserABI, err := bindings.NewL2ToL1MessagePasser(common.Address{}, nil)
+	if err != nil {
+		return err
+	}
+
+	// (2) Process New Messages
+	sentMessageEvents, err := CrossDomainMessengerSentMessageEvents(events)
+	if err != nil {
+		return err
+	}
+
+	sentMessages := make([]*database.L2BridgeMessage, len(sentMessageEvents))
+	for i, sentMessageEvent := range sentMessageEvents {
+		log := events.eventLog[sentMessageEvent.RawEvent.GUID]
+
+		// extract the withdrawal hash from the previous MessagePassed event
+		msgPassedLog := events.eventLog[events.eventByLogIndex[ProcessedContractEventLogIndexKey{log.BlockHash, log.Index - 1}].GUID]
+		msgPassedEvent, err := l2ToL1MessagePasserABI.ParseMessagePassed(*msgPassedLog)
+		if err != nil {
+			return err
+		}
+
+		sentMessages[i] = &database.L2BridgeMessage{
+			TransactionWithdrawalHash: msgPassedEvent.WithdrawalHash,
+			BridgeMessage: database.BridgeMessage{
+				Nonce:                database.U256{Int: sentMessageEvent.MessageNonce},
+				MessageHash:          sentMessageEvent.MessageHash,
+				SentMessageEventGUID: sentMessageEvent.RawEvent.GUID,
+				GasLimit:             database.U256{Int: sentMessageEvent.GasLimit},
+				Tx: database.Transaction{
+					FromAddress: sentMessageEvent.Sender,
+					ToAddress:   sentMessageEvent.Target,
+					Amount:      database.U256{Int: sentMessageEvent.Value},
+					Data:        sentMessageEvent.Message,
+					Timestamp:   sentMessageEvent.RawEvent.Timestamp,
+				},
+			},
+		}
+	}
+
+	if len(sentMessages) > 0 {
+		processLog.Info("detected L2CrossDomainMessenger messages", "size", len(sentMessages))
+		err := db.BridgeMessages.StoreL2BridgeMessages(sentMessages)
+		if err != nil {
+			return err
+		}
+	}
+
+	// (2) Process Relayed Messages.
+	//
+	// NOTE: Should we care about failed messages? A failed message can be
+	// inferred via an included deposit on L2 that has not been marked as relayed.
+	relayedMessageEvents, err := CrossDomainMessengerRelayedMessageEvents(events)
+	if err != nil {
+		return err
+	}
+
+	latestL1Header, err := db.Blocks.LatestL1BlockHeader()
+	if err != nil {
+		return err
+	} else if len(relayedMessageEvents) > 0 && latestL1Header == nil {
+		return errors.New("no indexed L1 headers to relay messages. waiting for L1Processor to catch up")
+	}
+
+	for _, relayedMessage := range relayedMessageEvents {
+		message, err := db.BridgeMessages.L1BridgeMessageByHash(relayedMessage.MsgHash)
+		if err != nil {
+			return err
+		}
+
+		if message == nil {
+			// Since the transaction processor running prior does not ensure the deposit inclusion, we need to
+			// ensure we are in a caught up state before claiming a missing event. Since L2 timestamps are derived
+			// from L1, we can simply compare the timestamp of this event with the latest L1 header.
+			if latestL1Header == nil || relayedMessage.RawEvent.Timestamp > latestL1Header.Timestamp {
+				processLog.Warn("waiting for L1Processor to catch up on L1CrossDomainMessages")
+				return errors.New("waiting for L1Processor to catch up")
+			} else {
+				processLog.Crit("missing indexed L1CrossDomainMessenger message", "message_hash", relayedMessage.MsgHash)
+				return fmt.Errorf("missing indexed L1CrossDomainMessager mesesage: 0x%x", relayedMessage.MsgHash)
+			}
+		}
+
+		err = db.BridgeMessages.MarkRelayedL1BridgeMessage(relayedMessage.MsgHash, relayedMessage.RawEvent.GUID)
+		if err != nil {
+			return err
+		}
+	}
+
+	if len(relayedMessageEvents) > 0 {
+		processLog.Info("relayed L1CrossDomainMessenger messages", "size", len(relayedMessageEvents))
+	}
+
+	// a-ok!
+	return nil
+}
+
 func l2ProcessContractEventsStandardBridge(processLog log.Logger, db *database.DB, ethClient node.EthClient, events *ProcessedContractEvents) error {
 	rawEthClient := ethclient.NewClient(ethClient.RawRpcClient())
 
@@ -297,7 +401,7 @@ func l2ProcessContractEventsStandardBridge(processLog log.Logger, db *database.D
 		if err != nil {
 			return err
 		} else if deposit == nil {
-			// NOTE: We'll be indexing CrossDomainMessenger messages that'll ensure we're in a caught up state here
+			// Indexed CrossDomainMessenger messages ensure we're in a caught up state here
 			processLog.Error("missing indexed L1StandardBridge deposit on finalization", "cross_domain_messenger_nonce", finalizedDepositEvent.CrossDomainMessengerNonce)
 			return errors.New("missing indexed L1StandardBridge deposit on finalization")
 		}