Merge pull request #5861 from ethereum-optimism/indexer.database

feat(indexer) schemas and database module

indexer/database/blocks.go

+package database
+
+import (
+	"context"
+	"errors"
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/common"
+
+	"gorm.io/gorm"
+)
+
+/**
+ * Types
+ */
+
+type BlockHeader struct {
+	Hash       common.Hash `gorm:"primaryKey;serializer:json"`
+	ParentHash common.Hash `gorm:"serializer:json"`
+	Number     U256
+	Timestamp  uint64
+}
+
+type L1BlockHeader struct {
+	BlockHeader
+}
+
+type L2BlockHeader struct {
+	BlockHeader
+
+	// Marked when the proposed output is finalized on L1.
+	// All bedrock blocks will have `LegacyStateBatchIndex ^== NULL`
+	L1BlockHash           *common.Hash `gorm:"serializer:json"`
+	LegacyStateBatchIndex *uint64
+}
+
+type LegacyStateBatch struct {
+	// `default:0` is added since gorm would interepret 0 as NULL
+	// violating the primary key constraint.
+	Index uint64 `gorm:"primaryKey;default:0"`
+
+	Root        common.Hash `gorm:"serializer:json"`
+	Size        uint64
+	PrevTotal   uint64
+	L1BlockHash common.Hash `gorm:"serializer:json"`
+}
+
+type BlocksView interface {
+	FinalizedL1BlockHeight() (*big.Int, error)
+	FinalizedL2BlockHeight() (*big.Int, error)
+}
+
+type BlocksDB interface {
+	BlocksView
+
+	StoreL1BlockHeaders([]*L1BlockHeader) error
+	StoreLegacyStateBatch(*LegacyStateBatch) error
+
+	StoreL2BlockHeaders([]*L2BlockHeader) error
+	MarkFinalizedL1RootForL2Block(common.Hash, common.Hash) error
+}
+
+/**
+ * Implementation
+ */
+
+type blocksDB struct {
+	gorm *gorm.DB
+}
+
+func newBlocksDB(db *gorm.DB) BlocksDB {
+	return &blocksDB{gorm: db}
+}
+
+// L1
+
+func (db *blocksDB) StoreL1BlockHeaders(headers []*L1BlockHeader) error {
+	result := db.gorm.Create(&headers)
+	return result.Error
+}
+
+func (db *blocksDB) StoreLegacyStateBatch(stateBatch *LegacyStateBatch) error {
+	// Even though transaction control flow is managed, could we benefit
+	// from a nested transaction here?
+
+	result := db.gorm.Create(stateBatch)
+	if result.Error != nil {
+		return result.Error
+	}
+
+	// Mark this state batch index & l1 block hash for all applicable l2 blocks
+	l2Headers := make([]*L2BlockHeader, stateBatch.Size)
+
+	// [start, end] range is inclusive. Since `PrevTotal` is the index of the prior batch, no
+	// need to subtract one when adding the size
+	startHeight := U256{Int: big.NewInt(int64(stateBatch.PrevTotal + 1))}
+	endHeight := U256{Int: big.NewInt(int64(stateBatch.PrevTotal + stateBatch.Size))}
+	result = db.gorm.Where("number BETWEEN ? AND ?", &startHeight, &endHeight).Find(&l2Headers)
+	if result.Error != nil {
+		return result.Error
+	} else if result.RowsAffected != int64(stateBatch.Size) {
+		return errors.New("state batch size exceeds number of indexed l2 blocks")
+	}
+
+	for _, header := range l2Headers {
+		header.LegacyStateBatchIndex = &stateBatch.Index
+		header.L1BlockHash = &stateBatch.L1BlockHash
+	}
+
+	result = db.gorm.Save(&l2Headers)
+	return result.Error
+}
+
+func (db *blocksDB) FinalizedL1BlockHeight() (*big.Int, error) {
+	var l1Header L1BlockHeader
+	result := db.gorm.Order("number DESC").Take(&l1Header)
+	if result.Error != nil {
+		return nil, result.Error
+	}
+
+	return l1Header.Number.Int, nil
+}
+
+// L2
+
+func (db *blocksDB) StoreL2BlockHeaders(headers []*L2BlockHeader) error {
+	result := db.gorm.Create(&headers)
+	return result.Error
+}
+
+func (db *blocksDB) FinalizedL2BlockHeight() (*big.Int, error) {
+	var l2Header L2BlockHeader
+	result := db.gorm.Order("number DESC").Take(&l2Header)
+	if result.Error != nil {
+		return nil, result.Error
+	}
+
+	result.Logger.Info(context.Background(), "number ", l2Header.Number)
+	return l2Header.Number.Int, nil
+}
+
+func (db *blocksDB) MarkFinalizedL1RootForL2Block(l2Root, l1Root common.Hash) error {
+	var l2Header L2BlockHeader
+	l2Header.Hash = l2Root // set the primary key
+
+	result := db.gorm.First(&l2Header)
+	if result.Error != nil {
+		return result.Error
+	}
+
+	l2Header.L1BlockHash = &l1Root
+	result = db.gorm.Save(&l2Header)
+	return result.Error
+}

indexer/database/bridge.go

+package database
+
+import (
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/hexutil"
+	"gorm.io/gorm"
+)
+
+/**
+ * Types
+ */
+
+type Transaction struct {
+	FromAddress common.Address `gorm:"serializer:json"`
+	ToAddress   common.Address `gorm:"serializer:json"`
+	Amount      U256
+	Data        hexutil.Bytes `gorm:"serializer:json"`
+	Timestamp   uint64
+}
+
+type TokenPair struct {
+	L1TokenAddress common.Address `gorm:"serializer:json"`
+	L2TokenAddress common.Address `gorm:"serializer:json"`
+}
+
+type Deposit struct {
+	GUID                 string `gorm:"primaryKey"`
+	InitiatedL1EventGUID string
+
+	Tx        Transaction `gorm:"embedded"`
+	TokenPair TokenPair   `gorm:"embedded"`
+}
+
+type DepositWithTransactionHash struct {
+	Deposit           Deposit     `gorm:"embedded"`
+	L1TransactionHash common.Hash `gorm:"serializer:json"`
+}
+
+type Withdrawal struct {
+	GUID                 string `gorm:"primaryKey"`
+	InitiatedL2EventGUID string
+
+	WithdrawalHash       common.Hash `gorm:"serializer:json"`
+	ProvenL1EventGUID    *string
+	FinalizedL1EventGUID *string
+
+	Tx        Transaction `gorm:"embedded"`
+	TokenPair TokenPair   `gorm:"embedded"`
+}
+
+type WithdrawalWithTransactionHashes struct {
+	Withdrawal        Withdrawal  `gorm:"embedded"`
+	L2TransactionHash common.Hash `gorm:"serializer:json"`
+
+	ProvenL1TransactionHash    *common.Hash `gorm:"serializer:json"`
+	FinalizedL1TransactionHash *common.Hash `gorm:"serializer:json"`
+}
+
+type BridgeView interface {
+	DepositsByAddress(address common.Address) ([]*DepositWithTransactionHash, error)
+	WithdrawalsByAddress(address common.Address) ([]*WithdrawalWithTransactionHashes, error)
+}
+
+type BridgeDB interface {
+	BridgeView
+
+	StoreDeposits([]*Deposit) error
+	StoreWithdrawals([]*Withdrawal) error
+	MarkProvenWithdrawalEvent(string, string) error
+	MarkFinalizedWithdrawalEvent(string, string) error
+}
+
+/**
+ * Implementation
+ */
+
+type bridgeDB struct {
+	gorm *gorm.DB
+}
+
+func newBridgeDB(db *gorm.DB) BridgeDB {
+	return &bridgeDB{gorm: db}
+}
+
+// Deposits
+
+func (db *bridgeDB) StoreDeposits(deposits []*Deposit) error {
+	result := db.gorm.Create(&deposits)
+	return result.Error
+}
+
+func (db *bridgeDB) DepositsByAddress(address common.Address) ([]*DepositWithTransactionHash, error) {
+	depositsQuery := db.gorm.Table("deposits").Select("deposits.*, l1_contract_events.transaction_hash AS l1_transaction_hash")
+	eventsJoinQuery := depositsQuery.Joins("LEFT JOIN l1_contract_events ON deposits.initiated_l1_event_guid = l1_contract_events.guid")
+
+	// add in cursoring options
+	filteredQuery := eventsJoinQuery.Where(&Transaction{FromAddress: address}).Order("deposits.timestamp DESC").Limit(100)
+
+	deposits := make([]*DepositWithTransactionHash, 100)
+	result := filteredQuery.Scan(&deposits)
+	if result.Error != nil {
+		return nil, result.Error
+	}
+
+	return deposits, nil
+}
+
+// Withdrawals
+
+func (db *bridgeDB) StoreWithdrawals(withdrawals []*Withdrawal) error {
+	result := db.gorm.Create(&withdrawals)
+	return result.Error
+}
+
+func (db *bridgeDB) MarkProvenWithdrawalEvent(guid, provenL1EventGuid string) error {
+	var withdrawal Withdrawal
+	result := db.gorm.First(&withdrawal, "guid = ?", guid)
+	if result.Error == nil {
+		withdrawal.ProvenL1EventGUID = &provenL1EventGuid
+		db.gorm.Save(&withdrawal)
+	}
+
+	return result.Error
+}
+
+func (db *bridgeDB) MarkFinalizedWithdrawalEvent(guid, finalizedL1EventGuid string) error {
+	var withdrawal Withdrawal
+	result := db.gorm.First(&withdrawal, "guid = ?", guid)
+	if result.Error == nil {
+		withdrawal.FinalizedL1EventGUID = &finalizedL1EventGuid
+		db.gorm.Save(&withdrawal)
+	}
+
+	return result.Error
+}
+
+func (db *bridgeDB) WithdrawalsByAddress(address common.Address) ([]*WithdrawalWithTransactionHashes, error) {
+	withdrawalsQuery := db.gorm.Debug().Table("withdrawals").Select("withdrawals.*, l2_contract_events.transaction_hash AS l2_transaction_hash, proven_l1_contract_events.transaction_hash AS proven_l1_transaction_hash, finalized_l1_contract_events.transaction_hash AS finalized_l1_transaction_hash")
+
+	eventsJoinQuery := withdrawalsQuery.Joins("LEFT JOIN l2_contract_events ON withdrawals.initiated_l2_event_guid = l2_contract_events.guid")
+	provenJoinQuery := eventsJoinQuery.Joins("LEFT JOIN l1_contract_events AS proven_l1_contract_events ON withdrawals.proven_l1_event_guid = proven_l1_contract_events.guid")
+	finalizedJoinQuery := provenJoinQuery.Joins("LEFT JOIN l1_contract_events AS finalized_l1_contract_events ON withdrawals.finalized_l1_event_guid = finalized_l1_contract_events.guid")
+
+	// add in cursoring options
+	filteredQuery := finalizedJoinQuery.Where(&Transaction{FromAddress: address}).Order("withdrawals.timestamp DESC").Limit(100)
+
+	withdrawals := make([]*WithdrawalWithTransactionHashes, 100)
+	result := filteredQuery.Scan(&withdrawals)
+	if result.Error != nil {
+		return nil, result.Error
+	}
+
+	return withdrawals, nil
+}

indexer/database/contract_events.go

+package database
+
+import (
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/hexutil"
+	"gorm.io/gorm"
+)
+
+/**
+ * Types
+ */
+
+type ContractEvent struct {
+	GUID            string      `gorm:"primaryKey"`
+	BlockHash       common.Hash `gorm:"serializer:json"`
+	TransactionHash common.Hash `gorm:"serializer:json"`
+
+	EventSignature hexutil.Bytes `gorm:"serializer:json"`
+	LogIndex       uint64
+	Timestamp      uint64
+}
+
+type L1ContractEvent struct {
+	ContractEvent `gorm:"embedded"`
+}
+
+type L2ContractEvent struct {
+	ContractEvent `gorm:"embedded"`
+}
+
+type ContractEventsView interface {
+	L1ContractEventByGUID(string) (*L1ContractEvent, error)
+	L2ContractEventByGUID(string) (*L2ContractEvent, error)
+}
+
+type ContractEventsDB interface {
+	ContractEventsView
+
+	StoreL1ContractEvents([]*L1ContractEvent) error
+	StoreL2ContractEvents([]*L2ContractEvent) error
+}
+
+/**
+ * Implementation
+ */
+
+type contractEventsDB struct {
+	gorm *gorm.DB
+}
+
+func newContractEventsDB(db *gorm.DB) ContractEventsDB {
+	return &contractEventsDB{gorm: db}
+}
+
+// L1
+
+func (db *contractEventsDB) StoreL1ContractEvents(events []*L1ContractEvent) error {
+	result := db.gorm.Create(&events)
+	return result.Error
+}
+
+func (db *contractEventsDB) L1ContractEventByGUID(guid string) (*L1ContractEvent, error) {
+	var event L1ContractEvent
+	result := db.gorm.First(&event, "guid = ?", guid)
+	if result.Error != nil {
+		return nil, result.Error
+	}
+
+	return &event, nil
+}
+
+// L2
+
+func (db *contractEventsDB) StoreL2ContractEvents(events []*L2ContractEvent) error {
+	result := db.gorm.Create(&events)
+	return result.Error
+}
+
+func (db *contractEventsDB) L2ContractEventByGUID(guid string) (*L2ContractEvent, error) {
+	var event L2ContractEvent
+	result := db.gorm.First(&event, "guid = ?", guid)
+	if result.Error != nil {
+		return nil, result.Error
+	}
+
+	return &event, nil
+}

indexer/database/db.go

+// Database module defines the data DB struct which wraps specific DB interfaces for L1/L2 block headers, contract events, bridging schemas.
+package database
+
+import (
+	"gorm.io/driver/postgres"
+	"gorm.io/gorm"
+)
+
+type DB struct {
+	gorm *gorm.DB
+
+	Blocks         BlocksDB
+	ContractEvents ContractEventsDB
+	Bridge         BridgeDB
+}
+
+func NewDB(dsn string) (*DB, error) {
+	gorm, err := gorm.Open(postgres.Open(dsn), &gorm.Config{
+		// The indexer will explicitly manage the transaction
+		// flow processing blocks
+		SkipDefaultTransaction: true,
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	db := &DB{
+		gorm:           gorm,
+		Blocks:         newBlocksDB(gorm),
+		ContractEvents: newContractEventsDB(gorm),
+		Bridge:         newBridgeDB(gorm),
+	}
+
+	return db, nil
+}

indexer/database/types.go

+package database
+
+import (
+	"database/sql/driver"
+	"errors"
+	"math/big"
+
+	"github.com/jackc/pgtype"
+)
+
+var u256BigIntOverflow = new(big.Int).Exp(big.NewInt(2), big.NewInt(256), nil)
+var big10 = big.NewInt(10)
+
+var ErrU256Overflow = errors.New("number exceeds u256")
+var ErrU256ContainsDecimal = errors.New("number contains fractional digits")
+var ErrU256NotNull = errors.New("number cannot be null")
+
+// U256 is a wrapper over big.Int that conforms to the database U256 numeric domain type
+type U256 struct {
+	Int *big.Int
+}
+
+// Scan implements the database/sql Scanner interface.
+func (u256 *U256) Scan(src interface{}) error {
+	// deserialize as a numeric
+	var numeric pgtype.Numeric
+	err := numeric.Scan(src)
+	if err != nil {
+		return err
+	} else if numeric.Exp < 0 {
+		return ErrU256ContainsDecimal
+	} else if numeric.Status == pgtype.Null {
+		return ErrU256NotNull
+	}
+
+	// factor in the powers of 10
+	num := numeric.Int
+	if numeric.Exp > 0 {
+		factor := new(big.Int).Exp(big10, big.NewInt(int64(numeric.Exp)), nil)
+		num.Mul(num, factor)
+	}
+
+	// check bounds before setting the u256
+	if num.Cmp(u256BigIntOverflow) >= 0 {
+		return ErrU256Overflow
+	} else {
+		u256.Int = num
+	}
+
+	return nil
+}
+
+// Value implements the database/sql/driver Valuer interface.
+func (u256 U256) Value() (driver.Value, error) {
+	// check bounds
+	if u256.Int == nil {
+		return nil, ErrU256NotNull
+	} else if u256.Int.Cmp(u256BigIntOverflow) >= 0 {
+		return nil, ErrU256Overflow
+	}
+
+	// simply encode as a numeric with no Exp set (non-decimal)
+	numeric := pgtype.Numeric{Int: u256.Int, Status: pgtype.Present}
+	return numeric.Value()
+}

indexer/go.mod

@@ -80,9 +80,16 @@ require (
 	github.com/ipfs/go-datastore v0.6.0 // indirect
 	github.com/ipfs/go-log v1.0.5 // indirect
 	github.com/ipfs/go-log/v2 v2.5.1 // indirect
+	github.com/jackc/pgio v1.0.0 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
+	github.com/jackc/pgtype v1.14.0 // indirect
+	github.com/jackc/pgx/v5 v5.3.1 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
 	github.com/jbenet/goprocess v0.1.4 // indirect
+	github.com/jinzhu/inflection v1.0.0 // indirect
+	github.com/jinzhu/now v1.1.5 // indirect
 	github.com/klauspost/compress v1.15.15 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.3 // indirect
 	github.com/koron/go-ssdp v0.0.3 // indirect
@@ -159,20 +166,22 @@ require (
 	go.uber.org/fx v1.19.1 // indirect
 	go.uber.org/multierr v1.9.0 // indirect
 	go.uber.org/zap v1.24.0 // indirect
-	golang.org/x/crypto v0.6.0 // indirect
+	golang.org/x/crypto v0.8.0 // indirect
 	golang.org/x/exp v0.0.0-20230213192124-5e25df0256eb // indirect
 	golang.org/x/mod v0.8.0 // indirect
-	golang.org/x/net v0.7.0 // indirect
+	golang.org/x/net v0.9.0 // indirect
 	golang.org/x/sync v0.1.0 // indirect
-	golang.org/x/sys v0.5.0 // indirect
-	golang.org/x/term v0.5.0 // indirect
-	golang.org/x/text v0.7.0 // indirect
+	golang.org/x/sys v0.7.0 // indirect
+	golang.org/x/term v0.7.0 // indirect
+	golang.org/x/text v0.9.0 // indirect
 	golang.org/x/time v0.0.0-20220922220347-f3bd1da661af // indirect
 	golang.org/x/tools v0.6.0 // indirect
 	google.golang.org/protobuf v1.28.1 // indirect
 	gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
+	gorm.io/driver/postgres v1.5.2 // indirect
+	gorm.io/gorm v1.25.1 // indirect
 	lukechampine.com/blake3 v1.1.7 // indirect
 	nhooyr.io/websocket v1.8.7 // indirect
 )

indexer/migrations/20230523_create_schema.sql

+
+CREATE DOMAIN UINT256 AS NUMERIC NOT NULL
+    CHECK (VALUE >= 0 AND VALUE < 2^256 and SCALE(VALUE) = 0);
+
+/**
+ * BLOCK DATA
+ */
+
+CREATE TABLE IF NOT EXISTS l1_block_headers (
+	hash        VARCHAR NOT NULL PRIMARY KEY,
+	parent_hash VARCHAR NOT NULL,
+	number      UINT256,
+	timestamp   INTEGER NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS legacy_state_batches (
+	index         INTEGER NOT NULL PRIMARY KEY,
+	root          VARCHAR NOT NULL,
+	size          INTEGER NOT NULL,
+	prev_total    INTEGER NOT NULL,
+
+    -- Finalization information. Unlike `l2_block_headers` the NOT NULL
+    -- constraint is added since the l1 block hash will be known when
+    -- when reading the output event
+	l1_block_hash VARCHAR NOT NULL REFERENCES l1_block_headers(hash)
+);
+
+CREATE TABLE IF NOT EXISTS l2_block_headers (
+    -- Block header
+	hash                     VARCHAR NOT NULL PRIMARY KEY,
+	parent_hash              VARCHAR NOT NULL,
+	number                   UINT256,
+	timestamp                INTEGER NOT NULL,
+
+    -- Finalization information
+    l1_block_hash            VARCHAR REFERENCES l1_block_headers(hash),
+    legacy_state_batch_index INTEGER REFERENCES legacy_state_batches(index)
+);
+
+/** 
+ * EVENT DATA
+ */
+
+CREATE TABLE IF NOT EXISTS l1_contract_events (
+    guid             VARCHAR NOT NULL PRIMARY KEY,
+	block_hash       VARCHAR NOT NULL REFERENCES l1_block_headers(hash),
+    transaction_hash VARCHAR NOT NULL,
+    event_signature  VARCHAR NOT NULL,
+    log_index        INTEGER NOT NULL,
+    timestamp        INTEGER NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS l2_contract_events (
+    guid             VARCHAR NOT NULL PRIMARY KEY,
+	block_hash       VARCHAR NOT NULL REFERENCES l2_block_headers(hash),
+    transaction_hash VARCHAR NOT NULL,
+    event_signature  VARCHAR NOT NULL,
+    log_index        INTEGER NOT NULL,
+    timestamp        INTEGER NOT NULL
+);
+
+/**
+ * BRIDGING DATA
+ */
+
+CREATE TABLE IF NOT EXISTS deposits (
+	guid                 VARCHAR PRIMARY KEY NOT NULL,
+
+    -- Event causing the deposit
+    initiated_l1_event_guid VARCHAR NOT NULL REFERENCES l1_contract_events(guid),
+
+    -- Deposit information (do we need indexes on from/to?)
+	from_address     VARCHAR NOT NULL,
+	to_address       VARCHAR NOT NULL,
+	l1_token_address VARCHAR NOT NULL,
+	l2_token_address VARCHAR NOT NULL,
+	amount           UINT256,
+	data             VARCHAR NOT NULL,
+    timestamp        INTEGER NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS withdrawals (
+	guid                VARCHAR PRIMARY KEY NOT NULL,
+
+    -- Event causing this withdrawal
+    initiated_l2_event_guid VARCHAR NOT NULL REFERENCES l2_contract_events(guid),
+
+    -- Multistep (bedrock) process of a withdrawal
+    withdrawal_hash      VARCHAR NOT NULL,
+    proven_l1_event_guid VARCHAR REFERENCES l1_contract_events(guid),
+
+    -- Finalization marker (legacy & bedrock)
+    finalized_l1_event_guid VARCHAR REFERENCES l1_contract_events(guid),
+
+    -- Withdrawal information (do we need indexes on from/to?)
+	from_address     VARCHAR NOT NULL,
+	to_address       VARCHAR NOT NULL,
+	l1_token_address VARCHAR NOT NULL,
+	l2_token_address VARCHAR NOT NULL,
+	amount           UINT256,
+	data             VARCHAR NOT NULL,
+    timestamp        INTEGER NOT NULL
+);