feat(eventindexer): introduce disperser (#16694)

packages/eventindexer/abigen.sh

@@ -5,9 +5,9 @@ if [ ! -d "../protocol/out" ]; then
     exit 1
 fi
 
-paths=("TaikoL1.sol" "Bridge.sol" "AssignmentHook.sol")
+paths=("TaikoL1.sol" "Bridge.sol" "AssignmentHook.sol" "SgxVerifier.sol" "TaikoToken.sol")
 
-names=("TaikoL1" "Bridge" "AssignmentHook")
+names=("TaikoL1" "Bridge" "AssignmentHook" "SgxVerifier" "TaikoToken")
 
 
 for (( i = 0; i < ${#paths[@]}; ++i ));

packages/eventindexer/cmd/flags/common.go

@@ -8,6 +8,7 @@ var (
 	commonCategory    = "COMMON"
 	indexerCategory   = "INDEXER"
 	generatorCategory = "GENERATOR"
+	disperserCategory = "DISPERSER"
 )
 
 var (
@@ -71,7 +72,6 @@ var (
 		Value:    6061,
 		EnvVars:  []string{"METRICS_HTTP_PORT"},
 	}
-
 	Layer = &cli.StringFlag{
 		Name:     "layer",
 		Usage:    "Which layer indexing is occurring on",

packages/eventindexer/cmd/flags/disperser.go

@@ -0,0 +1,41 @@
+package flags
+
+import "github.com/urfave/cli/v2"
+
+var (
+	DisperserPrivateKey = &cli.StringFlag{
+		Name:     "disperserPrivateKey",
+		Usage:    "Disperser private key which contains TTKO",
+		Required: true,
+		Category: disperserCategory,
+		EnvVars:  []string{"DISPERSER_PRIVATE_KEY"},
+	}
+	TaikoTokenAddress = &cli.StringFlag{
+		Name:     "taikoTokenAddress",
+		Usage:    "Address of the TaikoToken contract",
+		Required: true,
+		Category: disperserCategory,
+		EnvVars:  []string{"TAIKO_TOKEN_ADDRESS"},
+	}
+	DispersalAmount = &cli.StringFlag{
+		Name:     "taikoTokenAddress",
+		Usage:    "Dispersal amount in wei",
+		Required: true,
+		Category: disperserCategory,
+		EnvVars:  []string{"DISPERSAL_AMOUNT"},
+	}
+	RPCUrl = &cli.StringFlag{
+		Name:     "rpcUrl",
+		Usage:    "RPC URL for the source chain",
+		Required: true,
+		Category: commonCategory,
+		EnvVars:  []string{"RPC_URL"},
+	}
+)
+
+var DisperserFlags = MergeFlags(CommonFlags, []cli.Flag{
+	DisperserPrivateKey,
+	TaikoTokenAddress,
+	DispersalAmount,
+	RPCUrl,
+})

packages/eventindexer/cmd/flags/indexer.go

@@ -51,6 +51,13 @@ var (
 		Category: indexerCategory,
 		EnvVars:  []string{"ASSIGNMENT_HOOK_ADDRESS"},
 	}
+	SgxVerifierAddress = &cli.StringFlag{
+		Name:     "sgxVerifierAddress",
+		Usage:    "Address of the SGXVerifier contract",
+		Required: false,
+		Category: indexerCategory,
+		EnvVars:  []string{"SGX_VERIFIER_ADDRESS"},
+	}
 	BlockBatchSize = &cli.Uint64Flag{
 		Name:     "blockBatchSize",
 		Usage:    "Block batch size when iterating through blocks",
@@ -90,6 +97,7 @@ var IndexerFlags = MergeFlags(CommonFlags, []cli.Flag{
 	L1TaikoAddress,
 	BridgeAddress,
 	SwapAddresses,
+	SgxVerifierAddress,
 	AssignmentHookAddress,
 	BlockBatchSize,
 	SubscriptionBackoff,

packages/eventindexer/cmd/main.go

@@ -9,6 +9,7 @@ import (
 	"github.com/taikoxyz/taiko-mono/packages/eventindexer/api"
 	"github.com/taikoxyz/taiko-mono/packages/eventindexer/cmd/flags"
 	"github.com/taikoxyz/taiko-mono/packages/eventindexer/cmd/utils"
+	"github.com/taikoxyz/taiko-mono/packages/eventindexer/disperser"
 	"github.com/taikoxyz/taiko-mono/packages/eventindexer/generator"
 	"github.com/taikoxyz/taiko-mono/packages/eventindexer/indexer"
 	"github.com/urfave/cli/v2"
@@ -58,6 +59,13 @@ func main() {
 			Description: "Taiko time-series data generator",
 			Action:      utils.SubcommandAction(new(generator.Generator)),
 		},
+		{
+			Name:        "disperser",
+			Flags:       flags.DisperserFlags,
+			Usage:       "Starts the disperser software",
+			Description: "Taiko TTKO disperser",
+			Action:      utils.SubcommandAction(new(disperser.Disperser)),
+		},
 	}
 
 	if err := app.Run(os.Args); err != nil {

packages/eventindexer/disperser/config.go

@@ -0,0 +1,98 @@
+package disperser
+
+import (
+	"crypto/ecdsa"
+	"database/sql"
+	"fmt"
+	"math/big"
+
+	"github.com/ethereum-optimism/optimism/op-service/txmgr"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/taikoxyz/taiko-mono/packages/eventindexer/cmd/flags"
+	"github.com/taikoxyz/taiko-mono/packages/eventindexer/pkg/db"
+	pkgFlags "github.com/taikoxyz/taiko-mono/packages/relayer/pkg/flags"
+	"github.com/urfave/cli/v2"
+	"gorm.io/driver/mysql"
+	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
+)
+
+type DB interface {
+	DB() (*sql.DB, error)
+	GormDB() *gorm.DB
+}
+
+type Config struct {
+	// db configs
+	DatabaseUsername        string
+	DatabasePassword        string
+	DatabaseName            string
+	DatabaseHost            string
+	DatabaseMaxIdleConns    uint64
+	DatabaseMaxOpenConns    uint64
+	DatabaseMaxConnLifetime uint64
+	MetricsHTTPPort         uint64
+	DisperserPrivateKey     *ecdsa.PrivateKey
+	DispersalAmount         *big.Int
+	TaikoTokenAddress       common.Address
+	TxmgrConfigs            *txmgr.CLIConfig
+	RPCURL                  string
+	OpenDBFunc              func() (DB, error)
+}
+
+// NewConfigFromCliContext creates a new config instance from command line flags.
+func NewConfigFromCliContext(c *cli.Context) (*Config, error) {
+	disperserPrivateKey, err := crypto.ToECDSA(
+		common.Hex2Bytes(c.String(flags.DisperserPrivateKey.Name)),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("invalid disperserPrivateKey: %w", err)
+	}
+
+	dispersalAmount, ok := new(big.Int).SetString(c.String(flags.DispersalAmount.Name), 10)
+	if !ok {
+		return nil, fmt.Errorf("Invalid dispersal amount")
+	}
+
+	return &Config{
+		DatabaseUsername:        c.String(flags.DatabaseUsername.Name),
+		DatabasePassword:        c.String(flags.DatabasePassword.Name),
+		DatabaseName:            c.String(flags.DatabaseName.Name),
+		DatabaseHost:            c.String(flags.DatabaseHost.Name),
+		DatabaseMaxIdleConns:    c.Uint64(flags.DatabaseMaxIdleConns.Name),
+		DatabaseMaxOpenConns:    c.Uint64(flags.DatabaseMaxOpenConns.Name),
+		DatabaseMaxConnLifetime: c.Uint64(flags.DatabaseConnMaxLifetime.Name),
+		MetricsHTTPPort:         c.Uint64(flags.MetricsHTTPPort.Name),
+		DisperserPrivateKey:     disperserPrivateKey,
+		RPCURL:                  c.String(flags.RPCUrl.Name),
+		DispersalAmount:         dispersalAmount,
+		TaikoTokenAddress:       common.HexToAddress(c.String(flags.TaikoTokenAddress.Name)),
+		TxmgrConfigs: pkgFlags.InitTxmgrConfigsFromCli(
+			c.String(flags.RPCUrl.Name),
+			disperserPrivateKey,
+			c,
+		),
+		OpenDBFunc: func() (DB, error) {
+			return db.OpenDBConnection(db.DBConnectionOpts{
+				Name:            c.String(flags.DatabaseUsername.Name),
+				Password:        c.String(flags.DatabasePassword.Name),
+				Database:        c.String(flags.DatabaseName.Name),
+				Host:            c.String(flags.DatabaseHost.Name),
+				MaxIdleConns:    c.Uint64(flags.DatabaseMaxIdleConns.Name),
+				MaxOpenConns:    c.Uint64(flags.DatabaseMaxOpenConns.Name),
+				MaxConnLifetime: c.Uint64(flags.DatabaseConnMaxLifetime.Name),
+				OpenFunc: func(dsn string) (*db.DB, error) {
+					gormDB, err := gorm.Open(mysql.Open(dsn), &gorm.Config{
+						Logger: logger.Default.LogMode(logger.Silent),
+					})
+					if err != nil {
+						return nil, err
+					}
+
+					return db.New(gormDB), nil
+				},
+			})
+		},
+	}, nil
+}

packages/eventindexer/disperser/disperser.go

@@ -0,0 +1,120 @@
+package disperser
+
+import (
+	"context"
+	"log/slog"
+	"math/big"
+
+	txmgrMetrics "github.com/ethereum-optimism/optimism/op-service/txmgr/metrics"
+	"github.com/taikoxyz/taiko-mono/packages/eventindexer/encoding"
+
+	"github.com/ethereum-optimism/optimism/op-service/txmgr"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/urfave/cli/v2"
+)
+
+var (
+	ZeroAddress = common.HexToAddress("0x0000000000000000000000000000000000000000")
+)
+
+// Disperser is a subcommand which is intended to be run on an interval, like
+// a cronjob, to parse the indexed data from the database, and generate
+// time series data that can easily be displayed via charting libraries.
+type Disperser struct {
+	db                DB
+	dispersalAmount   *big.Int
+	taikoTokenAddress common.Address
+	txmgr             txmgr.TxManager
+}
+
+func (d *Disperser) InitFromCli(ctx context.Context, c *cli.Context) error {
+	config, err := NewConfigFromCliContext(c)
+	if err != nil {
+		return err
+	}
+
+	return InitFromConfig(ctx, d, config)
+}
+
+func InitFromConfig(ctx context.Context, d *Disperser, cfg *Config) error {
+	db, err := cfg.OpenDBFunc()
+	if err != nil {
+		return err
+	}
+
+	if d.txmgr, err = txmgr.NewSimpleTxManager(
+		"disperser",
+		log.Root(),
+		new(txmgrMetrics.NoopTxMetrics),
+		*cfg.TxmgrConfigs,
+	); err != nil {
+		return err
+	}
+
+	d.db = db
+
+	d.dispersalAmount = cfg.DispersalAmount
+	d.taikoTokenAddress = cfg.TaikoTokenAddress
+
+	return nil
+}
+
+func (d *Disperser) Name() string {
+	return "disperser"
+}
+
+func (d *Disperser) Start() error {
+	addresses, err := d.findAllAddresses()
+	if err != nil {
+		return err
+	}
+
+	for _, address := range addresses {
+		slog.Info("dispersing to", "address", address)
+
+		data, err := encoding.TaikoTokenABI.Pack("transfer", common.HexToAddress(address), d.dispersalAmount)
+		if err != nil {
+			return err
+		}
+
+		candidate := txmgr.TxCandidate{
+			TxData: data,
+			Blobs:  nil,
+			To:     &d.taikoTokenAddress,
+		}
+
+		receipt, err := d.txmgr.Send(context.Background(), candidate)
+		if err != nil {
+			slog.Warn("Failed to send transfer transaction", "error", err.Error())
+			return err
+		}
+
+		slog.Info("sent tx", "tx", receipt.TxHash.Hex())
+	}
+
+	return nil
+}
+
+func (d *Disperser) findAllAddresses() ([]string, error) {
+	var addresses []string
+	// Execute raw SQL query to find distinct addresses where event is 'InstanceAdded'
+	err := d.db.GormDB().Raw("SELECT DISTINCT address FROM events WHERE event = ?", "InstanceAdded").Scan(&addresses).Error
+
+	if err != nil {
+		return nil, err
+	}
+
+	return addresses, nil
+}
+
+func (d *Disperser) Close(ctx context.Context) {
+	sqlDB, err := d.db.DB()
+	if err != nil {
+		slog.Error("error getting sqldb when closing Disperser", "err", err.Error())
+	}
+
+	if err := sqlDB.Close(); err != nil {
+		slog.Error("error closing sqlbd connection", "err", err.Error())
+	}
+}

packages/eventindexer/encoding/types.go

@@ -0,0 +1,18 @@
+package encoding
+
+import (
+	"github.com/ethereum/go-ethereum/accounts/abi"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/taikoxyz/taiko-mono/packages/eventindexer/contracts/taikotoken"
+)
+
+var TaikoTokenABI *abi.ABI
+
+var err error
+
+func init() {
+	if TaikoTokenABI, err = taikotoken.TaikoTokenMetaData.GetAbi(); err != nil {
+		log.Crit("Get TaikoTokenABI ABI error", "error", err)
+	}
+}

packages/eventindexer/event.go

@@ -22,6 +22,7 @@ var (
 	EventNameSwap                = "Swap"
 	EventNameMint                = "Mint"
 	EventNameNFTTransfer         = "Transfer"
+	EventNameInstanceAdded       = "InstanceAdded"
 )
 
 // Event represents a stored EVM event. The fields will be serialized

packages/eventindexer/indexer/config.go

@@ -33,6 +33,7 @@ type Config struct {
 	L1TaikoAddress          common.Address
 	BridgeAddress           common.Address
 	AssignmentHookAddress   common.Address
+	SgxVerifierAddress      common.Address
 	SwapAddresses           []common.Address
 	BlockBatchSize          uint64
 	SubscriptionBackoff     uint64
@@ -67,6 +68,7 @@ func NewConfigFromCliContext(c *cli.Context) (*Config, error) {
 		L1TaikoAddress:          common.HexToAddress(c.String(flags.L1TaikoAddress.Name)),
 		BridgeAddress:           common.HexToAddress(c.String(flags.BridgeAddress.Name)),
 		AssignmentHookAddress:   common.HexToAddress(c.String(flags.AssignmentHookAddress.Name)),
+		SgxVerifierAddress:      common.HexToAddress(flags.SgxVerifierAddress.Name),
 		SwapAddresses:           swaps,
 		BlockBatchSize:          c.Uint64(flags.BlockBatchSize.Name),
 		SubscriptionBackoff:     c.Uint64(flags.SubscriptionBackoff.Name),

packages/eventindexer/indexer/filter.go

@@ -156,6 +156,22 @@ func filterFunc(
 		}
 	}
 
+	if i.sgxVerifier != nil {
+		wg.Go(func() error {
+			instancesAdded, err := i.sgxVerifier.FilterInstanceAdded(filterOpts, nil, nil)
+			if err != nil {
+				return errors.Wrap(err, "i.sgxVerifier.FilterInstanceAdded")
+			}
+
+			err = i.saveInstanceAddedEvents(ctx, chainID, instancesAdded)
+			if err != nil {
+				return errors.Wrap(err, "i.saveInstanceAddedEvents")
+			}
+
+			return nil
+		})
+	}
+
 	wg.Go(func() error {
 		if err := i.indexRawBlockData(ctx, chainID, filterOpts.Start, *filterOpts.End); err != nil {
 			return errors.Wrap(err, "i.indexRawBlockData")