github/scroll
1
1
Fork 0
mirror of https://github.com/scroll-tech/scroll.git synced 2026-01-10 06:28:04 -05:00
Code Issues Packages Projects Releases Wiki Activity

refactor(bridge-history-api): use Gorm (#656)

Browse Source 
Co-authored-by: vincent <419436363@qq.com>
Co-authored-by: HAOYUatHZ <37070449+HAOYUatHZ@users.noreply.github.com>
Co-authored-by: colinlyguo <colinlyguo@scroll.io>
Co-authored-by: colin <102356659+colinlyguo@users.noreply.github.com>
Co-authored-by: georgehao <haohongfan@gmail.com>
This commit is contained in:
ChuhanJin
2023-07-24 17:35:43 +08:00
committed by GitHub
parent e8c66e4597
commit 5c4f7c33fd
31 changed files with 1113 additions and 1224 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.github/workflows/bridge_history_api.yml vendored
View File

@@ -51,7 +51,6 @@ jobs:
uses: actions/checkout@v2
- name: Test
run: |
go get ./...
make test
- name: Upload coverage reports to Codecov
uses: codecov/codecov-action@v3

35
bridge-history-api/cmd/backend_server/app/app.go
View File

@@ -1,6 +1,7 @@
package app
import (
"context"
"fmt"
"os"
@@ -9,20 +10,22 @@ import (
"github.com/kataras/iris/v12"
"github.com/kataras/iris/v12/mvc"
"github.com/urfave/cli/v2"
"gorm.io/gorm"
"bridge-history-api/config"
"bridge-history-api/controller"
"bridge-history-api/db"
"bridge-history-api/service"
cutils "bridge-history-api/utils"
"scroll-tech/common/database"
)
var (
app *cli.App
app *cli.App
db *gorm.DB
subCtx context.Context
)
var database db.OrmFactory
func pong(ctx iris.Context) {
_, err := ctx.WriteString("pong")
if err != nil {
@@ -33,7 +36,8 @@ func pong(ctx iris.Context) {
func setupQueryByAddressHandler(backendApp *mvc.Application) {
// Register Dependencies.
backendApp.Register(
database,
subCtx,
db,
service.NewHistoryService,
)
@@ -44,7 +48,8 @@ func setupQueryByAddressHandler(backendApp *mvc.Application) {
func setupQueryClaimableHandler(backendApp *mvc.Application) {
// Register Dependencies.
backendApp.Register(
database,
subCtx,
db,
service.NewHistoryService,
)
@@ -54,7 +59,8 @@ func setupQueryClaimableHandler(backendApp *mvc.Application) {
func setupQueryByHashHandler(backendApp *mvc.Application) {
backendApp.Register(
database,
subCtx,
db,
service.NewHistoryService,
)
backendApp.Handle(new(controller.QueryHashController))
@@ -86,15 +92,22 @@ func action(ctx *cli.Context) error {
if err != nil {
log.Crit("failed to load config file", "config file", cfgFile, "error", err)
}
database, err = db.NewOrmFactory(cfg)
dbCfg := &database.Config{
DriverName: cfg.DB.DriverName,
DSN: cfg.DB.DSN,
MaxOpenNum: cfg.DB.MaxOpenNum,
MaxIdleNum: cfg.DB.MaxIdleNum,
}
db, err = database.InitDB(dbCfg)
if err != nil {
log.Crit("can not connect to database", "err", err)
log.Crit("failed to init db", "err", err)
}
defer func() {
if err = database.Close(); err != nil {
log.Error("failed to close database", "err", err)
if deferErr := database.CloseDB(db); deferErr != nil {
log.Error("failed to close db", "err", err)
}
}()
subCtx = ctx.Context
bridgeApp := iris.New()
bridgeApp.UseRouter(corsOptions)
bridgeApp.Get("/ping", pong).Describe("healthcheck")

23
bridge-history-api/cmd/cross_msg_fetcher/app/app.go
View File

@@ -14,8 +14,10 @@ import (
"bridge-history-api/config"
"bridge-history-api/crossmsg"
"bridge-history-api/crossmsg/messageproof"
"bridge-history-api/db"
"bridge-history-api/orm"
cutils "bridge-history-api/utils"
"scroll-tech/common/database"
)
var (
@@ -55,9 +57,18 @@ func action(ctx *cli.Context) error {
log.Crit("failed to connect l2 geth", "config file", cfgFile, "error", err)
}
db, err := db.NewOrmFactory(cfg)
dbCfg := &database.Config{
DriverName: cfg.DB.DriverName,
DSN: cfg.DB.DSN,
MaxOpenNum: cfg.DB.MaxOpenNum,
MaxIdleNum: cfg.DB.MaxIdleNum,
}
db, err := database.InitDB(dbCfg)
if err != nil {
log.Crit("failed to init db", "err", err)
}
defer func() {
if deferErr := db.Close(); deferErr != nil {
if deferErr := database.CloseDB(db); deferErr != nil {
log.Error("failed to close db", "err", err)
}
}()
@@ -105,12 +116,14 @@ func action(ctx *cli.Context) error {
go l2crossMsgFetcher.Start()
defer l2crossMsgFetcher.Stop()
CrossMsgOrm := orm.NewCrossMsg(db)
// BlockTimestamp fetcher for l1 and l2
l1BlockTimeFetcher := crossmsg.NewBlockTimestampFetcher(subCtx, cfg.L1.Confirmation, int(cfg.L1.BlockTime), l1client, db.UpdateL1BlockTimestamp, db.GetL1EarliestNoBlockTimestampHeight)
l1BlockTimeFetcher := crossmsg.NewBlockTimestampFetcher(subCtx, cfg.L1.Confirmation, int(cfg.L1.BlockTime), l1client, CrossMsgOrm.UpdateL1BlockTimestamp, CrossMsgOrm.GetL1EarliestNoBlockTimestampHeight)
go l1BlockTimeFetcher.Start()
defer l1BlockTimeFetcher.Stop()
l2BlockTimeFetcher := crossmsg.NewBlockTimestampFetcher(subCtx, cfg.L2.Confirmation, int(cfg.L2.BlockTime), l2client, db.UpdateL2BlockTimestamp, db.GetL2EarliestNoBlockTimestampHeight)
l2BlockTimeFetcher := crossmsg.NewBlockTimestampFetcher(subCtx, cfg.L2.Confirmation, int(cfg.L2.BlockTime), l2client, CrossMsgOrm.UpdateL2BlockTimestamp, CrossMsgOrm.GetL2EarliestNoBlockTimestampHeight)
go l2BlockTimeFetcher.Start()
defer l2BlockTimeFetcher.Stop()

64
bridge-history-api/cmd/db_cli/app/client.go
View File

@@ -2,13 +2,14 @@ package app
import (
"github.com/ethereum/go-ethereum/log"
"github.com/jmoiron/sqlx"
"github.com/urfave/cli/v2"
"gorm.io/gorm"
"bridge-history-api/config"
"bridge-history-api/db"
"bridge-history-api/db/migrate"
"bridge-history-api/orm/migrate"
"bridge-history-api/utils"
"scroll-tech/common/database"
)
func getConfig(ctx *cli.Context) (*config.Config, error) {
@@ -20,14 +21,14 @@ func getConfig(ctx *cli.Context) (*config.Config, error) {
return dbCfg, nil
}
func initDB(dbCfg *config.Config) (*sqlx.DB, error) {
factory, err := db.NewOrmFactory(dbCfg)
if err != nil {
return nil, err
func initDB(dbCfg *config.DBConfig) (*gorm.DB, error) {
cfg := &database.Config{
DriverName: dbCfg.DriverName,
DSN: dbCfg.DSN,
MaxOpenNum: dbCfg.MaxOpenNum,
MaxIdleNum: dbCfg.MaxIdleNum,
}
log.Debug("Got db config from env", "driver name", dbCfg.DB.DriverName, "dsn", dbCfg.DB.DSN)
return factory.GetDB(), nil
return database.InitDB(cfg)
}
// resetDB clean or reset database.
@@ -36,11 +37,15 @@ func resetDB(ctx *cli.Context) error {
if err != nil {
return err
}
db, err := initDB(cfg)
gormDB, err := initDB(cfg.DB)
if err != nil {
return err
}
err = migrate.ResetDB(db.DB)
db, err := gormDB.DB()
if err != nil {
return err
}
err = migrate.ResetDB(db)
if err != nil {
return err
}
@@ -54,12 +59,15 @@ func checkDBStatus(ctx *cli.Context) error {
if err != nil {
return err
}
db, err := initDB(cfg)
gormDB, err := initDB(cfg.DB)
if err != nil {
return err
}
return migrate.Status(db.DB)
db, err := gormDB.DB()
if err != nil {
return err
}
return migrate.Status(db)
}
// dbVersion return the latest version
@@ -68,12 +76,15 @@ func dbVersion(ctx *cli.Context) error {
if err != nil {
return err
}
db, err := initDB(cfg)
gormDB, err := initDB(cfg.DB)
if err != nil {
return err
}
version, err := migrate.Current(db.DB)
db, err := gormDB.DB()
if err != nil {
return err
}
version, err := migrate.Current(db)
log.Info("show database version", "db version", version)
return err
@@ -85,12 +96,15 @@ func migrateDB(ctx *cli.Context) error {
if err != nil {
return err
}
db, err := initDB(cfg)
gormDB, err := initDB(cfg.DB)
if err != nil {
return err
}
return migrate.Migrate(db.DB)
db, err := gormDB.DB()
if err != nil {
return err
}
return migrate.Migrate(db)
}
// rollbackDB rollback db by version
@@ -99,10 +113,14 @@ func rollbackDB(ctx *cli.Context) error {
if err != nil {
return err
}
db, err := initDB(cfg)
gormDB, err := initDB(cfg.DB)
if err != nil {
return err
}
db, err := gormDB.DB()
if err != nil {
return err
}
version := ctx.Int64("version")
return migrate.Rollback(db.DB, &version)
return migrate.Rollback(db, &version)
}

4
bridge-history-api/controller/controller.go
View File

@@ -24,7 +24,7 @@ type QueryClaimableController struct {
// Get defines the http get method behavior for QueryClaimableController
func (c *QueryClaimableController) Get(req model.QueryByAddressRequest) (*model.QueryByAddressResponse, error) {
txs, total, err := c.Service.GetClaimableTxsByAddress(common.HexToAddress(req.Address), int64(req.Offset), int64(req.Limit))
txs, total, err := c.Service.GetClaimableTxsByAddress(common.HexToAddress(req.Address), req.Offset, req.Limit)
if err != nil {
return &model.QueryByAddressResponse{Message: "500", Data: &model.Data{}}, err
}
@@ -38,7 +38,7 @@ func (c *QueryClaimableController) Get(req model.QueryByAddressRequest) (*model.
// Get defines the http get method behavior for QueryAddressController
func (c *QueryAddressController) Get(req model.QueryByAddressRequest) (*model.QueryByAddressResponse, error) {
message, total, err := c.Service.GetTxsByAddress(common.HexToAddress(req.Address), int64(req.Offset), int64(req.Limit))
message, total, err := c.Service.GetTxsByAddress(common.HexToAddress(req.Address), req.Offset, req.Limit)
if err != nil {
return &model.QueryByAddressResponse{Message: "500", Data: &model.Data{}}, err
}

20
bridge-history-api/crossmsg/batch_info_fetcher.go
View File

@@ -7,9 +7,10 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/ethclient"
"github.com/ethereum/go-ethereum/log"
"gorm.io/gorm"
"bridge-history-api/crossmsg/messageproof"
"bridge-history-api/db"
"bridge-history-api/orm"
"bridge-history-api/utils"
)
@@ -21,12 +22,13 @@ type BatchInfoFetcher struct {
confirmation uint64
blockTimeInSec int
client *ethclient.Client
db db.OrmFactory
db *gorm.DB
rollupOrm *orm.RollupBatch
msgProofUpdater *messageproof.MsgProofUpdater
}
// NewBatchInfoFetcher creates a new BatchInfoFetcher instance
func NewBatchInfoFetcher(ctx context.Context, scrollChainAddr common.Address, batchInfoStartNumber uint64, confirmation uint64, blockTimeInSec int, client *ethclient.Client, db db.OrmFactory, msgProofUpdater *messageproof.MsgProofUpdater) *BatchInfoFetcher {
func NewBatchInfoFetcher(ctx context.Context, scrollChainAddr common.Address, batchInfoStartNumber uint64, confirmation uint64, blockTimeInSec int, client *ethclient.Client, db *gorm.DB, msgProofUpdater *messageproof.MsgProofUpdater) *BatchInfoFetcher {
return &BatchInfoFetcher{
ctx: ctx,
scrollChainAddr: scrollChainAddr,
@@ -35,6 +37,7 @@ func NewBatchInfoFetcher(ctx context.Context, scrollChainAddr common.Address, ba
blockTimeInSec: blockTimeInSec,
client: client,
db: db,
rollupOrm: orm.NewRollupBatch(db),
msgProofUpdater: msgProofUpdater,
}
}
@@ -80,19 +83,20 @@ func (b *BatchInfoFetcher) fetchBatchInfo() error {
log.Error("Can not get latest block number: ", "err", err)
return err
}
latestBatch, err := b.db.GetLatestRollupBatch()
latestBatchHeight, err := b.rollupOrm.GetLatestRollupBatchProcessedHeight(b.ctx)
if err != nil {
log.Error("Can not get latest BatchInfo: ", "err", err)
return err
}
var startHeight uint64
if latestBatch == nil {
if latestBatchHeight == 0 {
log.Info("no batch record in database, start from batchInfoStartNumber", "batchInfoStartNumber", b.batchInfoStartNumber)
startHeight = b.batchInfoStartNumber
} else {
startHeight = latestBatch.CommitHeight + 1
startHeight = latestBatchHeight + 1
}
for from := startHeight; number >= from; from += uint64(fetchLimit) {
to := from + uint64(fetchLimit) - 1
for from := startHeight; number >= from; from += fetchLimit {
to := from + fetchLimit - 1
// number - confirmation can never less than 0 since the for loop condition
// but watch out the overflow
if to > number {

10
bridge-history-api/crossmsg/block_timestamp_fetcher.go
View File

@@ -10,10 +10,10 @@ import (
)
// GetEarliestNoBlockTimestampHeightFunc is a function type that gets the earliest record without block timestamp from database
type GetEarliestNoBlockTimestampHeightFunc func() (uint64, error)
type GetEarliestNoBlockTimestampHeightFunc func(ctx context.Context) (uint64, error)
// UpdateBlockTimestampFunc is a function type that updates block timestamp into database
type UpdateBlockTimestampFunc func(height uint64, timestamp time.Time) error
type UpdateBlockTimestampFunc func(ctx context.Context, height uint64, timestamp time.Time) error
// BlockTimestampFetcher fetches block timestamp from blockchain and saves them to database
type BlockTimestampFetcher struct {
@@ -52,7 +52,7 @@ func (b *BlockTimestampFetcher) Start() {
log.Error("Can not get latest block number", "err", err)
continue
}
startHeight, err := b.getEarliestNoBlockTimestampHeightFunc()
startHeight, err := b.getEarliestNoBlockTimestampHeightFunc(b.ctx)
if err != nil {
log.Error("Can not get latest record without block timestamp", "err", err)
continue
@@ -63,12 +63,12 @@ func (b *BlockTimestampFetcher) Start() {
log.Error("Can not get block by number", "err", err)
break
}
err = b.updateBlockTimestampFunc(height, time.Unix(int64(block.Time), 0))
err = b.updateBlockTimestampFunc(b.ctx, height, time.Unix(int64(block.Time), 0))
if err != nil {
log.Error("Can not update blockTimestamp into DB ", "err", err)
break
}
height, err = b.getEarliestNoBlockTimestampHeightFunc()
height, err = b.getEarliestNoBlockTimestampHeightFunc(b.ctx)
if err != nil {
log.Error("Can not get latest record without block timestamp", "err", err)
break

25
bridge-history-api/crossmsg/cross_msg_fetcher.go
View File

@@ -12,9 +12,9 @@ import (
"github.com/ethereum/go-ethereum/ethclient"
"github.com/ethereum/go-ethereum/log"
"github.com/modern-go/reflect2"
"gorm.io/gorm"
"bridge-history-api/config"
"bridge-history-api/db"
"bridge-history-api/utils"
)
@@ -22,7 +22,7 @@ import (
type MsgFetcher struct {
ctx context.Context
config *config.LayerConfig
db db.OrmFactory
db *gorm.DB
client *ethclient.Client
worker *FetchEventWorker
reorgHandling ReorgHandling
@@ -34,7 +34,7 @@ type MsgFetcher struct {
}
// NewMsgFetcher creates a new MsgFetcher instance
func NewMsgFetcher(ctx context.Context, config *config.LayerConfig, db db.OrmFactory, client *ethclient.Client, worker *FetchEventWorker, addressList []common.Address, reorg ReorgHandling) (*MsgFetcher, error) {
func NewMsgFetcher(ctx context.Context, config *config.LayerConfig, db *gorm.DB, client *ethclient.Client, worker *FetchEventWorker, addressList []common.Address, reorg ReorgHandling) (*MsgFetcher, error) {
msgFetcher := &MsgFetcher{
ctx: ctx,
config: config,
@@ -120,22 +120,23 @@ func (c *MsgFetcher) forwardFetchAndSaveMissingEvents(confirmation uint64) {
log.Error(fmt.Sprintf("%s: invalid get/fetch function", c.worker.Name))
return
}
processedHeight, err := c.worker.G(c.db)
processedHeight, err := c.worker.G(c.ctx, c.db)
if err != nil {
log.Error(fmt.Sprintf("%s: can not get latest processed block height", c.worker.Name))
}
log.Info(fmt.Sprintf("%s: ", c.worker.Name), "height", processedHeight)
if processedHeight <= 0 || processedHeight < int64(c.config.StartHeight) {
processedHeight = int64(c.config.StartHeight)
if processedHeight <= 0 || processedHeight < c.config.StartHeight {
processedHeight = c.config.StartHeight
} else {
processedHeight++
}
for from := processedHeight; from <= int64(number); from += fetchLimit {
for from := processedHeight; from <= number; from += fetchLimit {
to := from + fetchLimit - 1
if to > int64(number) {
to = int64(number)
if to > number {
to = number
}
err := c.worker.F(c.ctx, c.client, c.db, from, to, c.addressList)
// watch for overflow here, tho its unlikely to happen
err := c.worker.F(c.ctx, c.client, c.db, int64(from), int64(to), c.addressList)
if err != nil {
log.Error(fmt.Sprintf("%s: failed!", c.worker.Name), "err", err)
break
@@ -191,7 +192,7 @@ func (c *MsgFetcher) fetchMissingLatestHeaders() {
log.Crit("Can not get safe number during reorg, quit the process", "err", err)
}
// clear all our saved data, because no data is safe now
err = c.reorgHandling(c.ctx, int64(num), c.db)
err = c.reorgHandling(c.ctx, num, c.db)
// if handling success then we can update the cachedHeaders
if err == nil {
c.cachedHeaders = c.cachedHeaders[:0]
@@ -200,7 +201,7 @@ func (c *MsgFetcher) fetchMissingLatestHeaders() {
c.reorgEndCh <- struct{}{}
return
}
err = c.reorgHandling(c.ctx, c.cachedHeaders[index].Number.Int64(), c.db)
err = c.reorgHandling(c.ctx, c.cachedHeaders[index].Number.Uint64(), c.db)
// if handling success then we can update the cachedHeaders
if err == nil {
c.cachedHeaders = c.cachedHeaders[:index+1]

148
bridge-history-api/crossmsg/fetch_missing_event.go
View File

@@ -8,23 +8,24 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/ethclient"
"github.com/ethereum/go-ethereum/log"
"gorm.io/gorm"
backendabi "bridge-history-api/abi"
"bridge-history-api/db"
"bridge-history-api/orm"
"bridge-history-api/utils"
)
// Todo : read from config
var (
// the number of blocks fetch per round
fetchLimit = int64(3000)
fetchLimit = uint64(3000)
)
// FetchAndSave is a function type that fetches events from blockchain and saves them to database
type FetchAndSave func(ctx context.Context, client *ethclient.Client, database db.OrmFactory, from int64, to int64, addressList []common.Address) error
type FetchAndSave func(ctx context.Context, client *ethclient.Client, database *gorm.DB, from int64, to int64, addressList []common.Address) error
// GetLatestProcessed is a function type that gets the latest processed block height from database
type GetLatestProcessed func(db db.OrmFactory) (int64, error)
type GetLatestProcessed func(ctx context.Context, db *gorm.DB) (uint64, error)
// FetchEventWorker defines worker with fetch and save function, processed number getter, and name
type FetchEventWorker struct {
@@ -34,13 +35,15 @@ type FetchEventWorker struct {
}
// GetLatestL1ProcessedHeight get L1 the latest processed height
func GetLatestL1ProcessedHeight(db db.OrmFactory) (int64, error) {
crossHeight, err := db.GetLatestL1ProcessedHeight()
func GetLatestL1ProcessedHeight(ctx context.Context, db *gorm.DB) (uint64, error) {
l1CrossMsgOrm := orm.NewCrossMsg(db)
relayedOrm := orm.NewRelayedMsg(db)
crossHeight, err := l1CrossMsgOrm.GetLatestL1ProcessedHeight(ctx)
if err != nil {
log.Error("failed to get L1 cross message processed height: ", "err", err)
return 0, err
}
relayedHeight, err := db.GetLatestRelayedHeightOnL1()
relayedHeight, err := relayedOrm.GetLatestRelayedHeightOnL1(ctx)
if err != nil {
log.Error("failed to get L1 relayed message processed height: ", "err", err)
return 0, err
@@ -52,18 +55,21 @@ func GetLatestL1ProcessedHeight(db db.OrmFactory) (int64, error) {
}
// GetLatestL2ProcessedHeight get L2 latest processed height
func GetLatestL2ProcessedHeight(db db.OrmFactory) (int64, error) {
crossHeight, err := db.GetLatestL2ProcessedHeight()
func GetLatestL2ProcessedHeight(ctx context.Context, db *gorm.DB) (uint64, error) {
l2CrossMsgOrm := orm.NewCrossMsg(db)
relayedOrm := orm.NewRelayedMsg(db)
l2SentMsgOrm := orm.NewL2SentMsg(db)
crossHeight, err := l2CrossMsgOrm.GetLatestL2ProcessedHeight(ctx)
if err != nil {
log.Error("failed to get L2 cross message processed height", "err", err)
return 0, err
}
relayedHeight, err := db.GetLatestRelayedHeightOnL2()
relayedHeight, err := relayedOrm.GetLatestRelayedHeightOnL2(ctx)
if err != nil {
log.Error("failed to get L2 relayed message processed height", "err", err)
return 0, err
}
l2SentHeight, err := db.GetLatestSentMsgHeightOnL2()
l2SentHeight, err := l2SentMsgOrm.GetLatestSentMsgHeightOnL2(ctx)
if err != nil {
log.Error("failed to get L2 sent message processed height", "err", err)
return 0, err
@@ -79,7 +85,9 @@ func GetLatestL2ProcessedHeight(db db.OrmFactory) (int64, error) {
}
// L1FetchAndSaveEvents fetch and save events on L1
func L1FetchAndSaveEvents(ctx context.Context, client *ethclient.Client, database db.OrmFactory, from int64, to int64, addrList []common.Address) error {
func L1FetchAndSaveEvents(ctx context.Context, client *ethclient.Client, db *gorm.DB, from int64, to int64, addrList []common.Address) error {
l1CrossMsgOrm := orm.NewCrossMsg(db)
relayedOrm := orm.NewRelayedMsg(db)
query := geth.FilterQuery{
FromBlock: big.NewInt(from), // inclusive
ToBlock: big.NewInt(to), // inclusive
@@ -105,41 +113,28 @@ func L1FetchAndSaveEvents(ctx context.Context, client *ethclient.Client, databas
log.Error("l1FetchAndSaveEvents: Failed to parse cross msg event logs", "err", err)
return err
}
dbTx, err := database.Beginx()
if err != nil {
log.Error("l2FetchAndSaveEvents: Failed to begin db transaction", "err", err)
return err
}
err = database.BatchInsertL1CrossMsgDBTx(dbTx, depositL1CrossMsgs)
if err != nil {
if rollBackErr := dbTx.Rollback(); rollBackErr != nil {
log.Error("dbTx Rollback failed", "err", rollBackErr)
err = db.Transaction(func(tx *gorm.DB) error {
if txErr := l1CrossMsgOrm.InsertL1CrossMsg(ctx, depositL1CrossMsgs, tx); txErr != nil {
log.Error("l1FetchAndSaveEvents: Failed to insert cross msg event logs", "err", txErr)
return txErr
}
log.Crit("l1FetchAndSaveEvents: Failed to insert cross msg event logs", "err", err)
}
err = database.BatchInsertRelayedMsgDBTx(dbTx, relayedMsg)
if err != nil {
if rollBackErr := dbTx.Rollback(); rollBackErr != nil {
log.Error("dbTx Rollback failed", "err", rollBackErr)
if txErr := relayedOrm.InsertRelayedMsg(ctx, relayedMsg, tx); txErr != nil {
log.Error("l1FetchAndSaveEvents: Failed to insert relayed msg event logs", "err", txErr)
return txErr
}
log.Crit("l1FetchAndSaveEvents: Failed to insert relayed message event logs", "err", err)
}
err = dbTx.Commit()
return nil
})
if err != nil {
// if we can not insert into DB, there must something wrong, need a on-call member handle the dababase manually
if rollBackErr := dbTx.Rollback(); rollBackErr != nil {
log.Error("dbTx Rollback failed", "err", rollBackErr)
}
log.Error("l1FetchAndSaveEvents: Failed to commit db transaction", "err", err)
return err
log.Crit("l2FetchAndSaveEvents: Failed to finish transaction", "err", err)
}
return nil
return err
}
// L2FetchAndSaveEvents fetche and save events on L2
func L2FetchAndSaveEvents(ctx context.Context, client *ethclient.Client, database db.OrmFactory, from int64, to int64, addrList []common.Address) error {
func L2FetchAndSaveEvents(ctx context.Context, client *ethclient.Client, db *gorm.DB, from int64, to int64, addrList []common.Address) error {
l2CrossMsgOrm := orm.NewCrossMsg(db)
relayedOrm := orm.NewRelayedMsg(db)
l2SentMsgOrm := orm.NewL2SentMsg(db)
query := geth.FilterQuery{
FromBlock: big.NewInt(from), // inclusive
ToBlock: big.NewInt(to), // inclusive
@@ -166,50 +161,32 @@ func L2FetchAndSaveEvents(ctx context.Context, client *ethclient.Client, databas
return err
}
dbTx, err := database.Beginx()
if err != nil {
log.Error("l2FetchAndSaveEvents: Failed to begin db transaction", "err", err)
return err
}
err = database.BatchInsertL2CrossMsgDBTx(dbTx, depositL2CrossMsgs)
if err != nil {
if rollBackErr := dbTx.Rollback(); rollBackErr != nil {
log.Error("dbTx Rollback failed", "err", rollBackErr)
err = db.Transaction(func(tx *gorm.DB) error {
if txErr := l2CrossMsgOrm.InsertL2CrossMsg(ctx, depositL2CrossMsgs, tx); txErr != nil {
log.Error("l2FetchAndSaveEvents: Failed to insert cross msg event logs", "err", txErr)
return txErr
}
log.Crit("l2FetchAndSaveEvents: Failed to insert cross msg event logs", "err", err)
}
err = database.BatchInsertRelayedMsgDBTx(dbTx, relayedMsg)
if err != nil {
if rollBackErr := dbTx.Rollback(); rollBackErr != nil {
log.Error("dbTx Rollback failed", "err", rollBackErr)
if txErr := relayedOrm.InsertRelayedMsg(ctx, relayedMsg, tx); txErr != nil {
log.Error("l2FetchAndSaveEvents: Failed to insert relayed message event logs", "err", txErr)
return txErr
}
log.Crit("l2FetchAndSaveEvents: Failed to insert relayed message event logs", "err", err)
}
err = database.BatchInsertL2SentMsgDBTx(dbTx, l2SentMsgs)
if err != nil {
if rollBackErr := dbTx.Rollback(); rollBackErr != nil {
log.Error("dbTx Rollback failed", "err", rollBackErr)
if txErr := l2SentMsgOrm.InsertL2SentMsg(ctx, l2SentMsgs, tx); txErr != nil {
log.Error("l2FetchAndSaveEvents: Failed to insert l2 sent message", "err", txErr)
return txErr
}
log.Crit("l2FetchAndSaveEvents: Failed to insert l2 sent message", "err", err)
}
err = dbTx.Commit()
return nil
})
if err != nil {
// if we can not insert into DB, there must something wrong, need a on-call member handle the dababase manually
if rollBackErr := dbTx.Rollback(); rollBackErr != nil {
log.Error("dbTx Rollback failed", "err", rollBackErr)
}
log.Error("l2FetchAndSaveEvents: Failed to commit db transaction", "err", err)
return err
log.Crit("l2FetchAndSaveEvents: Failed to begin db transaction", "err", err)
}
return nil
return err
}
// FetchAndSaveBatchIndex fetche and save batch index
func FetchAndSaveBatchIndex(ctx context.Context, client *ethclient.Client, database db.OrmFactory, from int64, to int64, scrollChainAddr common.Address) error {
func FetchAndSaveBatchIndex(ctx context.Context, client *ethclient.Client, db *gorm.DB, from int64, to int64, scrollChainAddr common.Address) error {
rollupBatchOrm := orm.NewRollupBatch(db)
query := geth.FilterQuery{
FromBlock: big.NewInt(from), // inclusive
ToBlock: big.NewInt(to), // inclusive
@@ -228,26 +205,9 @@ func FetchAndSaveBatchIndex(ctx context.Context, client *ethclient.Client, datab
log.Error("FetchAndSaveBatchIndex: Failed to parse batch commit msg event logs", "err", err)
return err
}
dbTx, err := database.Beginx()
if err != nil {
log.Error("FetchAndSaveBatchIndex: Failed to begin db transaction", "err", err)
return err
}
err = database.BatchInsertRollupBatchDBTx(dbTx, rollupBatches)
if err != nil {
if rollBackErr := dbTx.Rollback(); rollBackErr != nil {
log.Error("dbTx Rollback failed", "err", rollBackErr)
}
log.Crit("FetchAndSaveBatchIndex: Failed to insert batch commit msg event logs", "err", err)
}
err = dbTx.Commit()
if err != nil {
// if we can not insert into DB, there must something wrong, need a on-call member handle the dababase manually
if rollBackErr := dbTx.Rollback(); rollBackErr != nil {
log.Error("dbTx Rollback failed", "err", rollBackErr)
}
log.Error("FetchAndSaveBatchIndex: Failed to commit db transaction", "err", err)
return err
if txErr := rollupBatchOrm.InsertRollupBatch(ctx, rollupBatches); txErr != nil {
log.Crit("FetchAndSaveBatchIndex: Failed to insert batch commit msg event logs", "err", txErr)
return txErr
}
return nil
}

64
bridge-history-api/crossmsg/messageproof/msg_proof_updater.go
View File

@@ -2,30 +2,32 @@ package messageproof
import (
"context"
"database/sql"
"errors"
"fmt"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"gorm.io/gorm"
"bridge-history-api/db"
"bridge-history-api/db/orm"
"bridge-history-api/orm"
)
// MsgProofUpdater is used to update message proof in db
type MsgProofUpdater struct {
ctx context.Context
db db.OrmFactory
db *gorm.DB
l2SentMsgOrm *orm.L2SentMsg
rollupOrm *orm.RollupBatch
withdrawTrie *WithdrawTrie
}
// NewMsgProofUpdater new MsgProofUpdater instance
func NewMsgProofUpdater(ctx context.Context, confirmations uint64, startBlock uint64, db db.OrmFactory) *MsgProofUpdater {
func NewMsgProofUpdater(ctx context.Context, confirmations uint64, startBlock uint64, db *gorm.DB) *MsgProofUpdater {
return &MsgProofUpdater{
ctx: ctx,
db: db,
l2SentMsgOrm: orm.NewL2SentMsg(db),
rollupOrm: orm.NewRollupBatch(db),
withdrawTrie: NewWithdrawTrie(),
}
}
@@ -42,7 +44,7 @@ func (m *MsgProofUpdater) Start() {
tick.Stop()
return
case <-tick.C:
latestBatch, err := m.db.GetLatestRollupBatch()
latestBatch, err := m.rollupOrm.GetLatestRollupBatch(m.ctx)
if err != nil {
log.Warn("MsgProofUpdater: Can not get latest RollupBatch: ", "err", err)
continue
@@ -50,7 +52,7 @@ func (m *MsgProofUpdater) Start() {
if latestBatch == nil {
continue
}
latestBatchIndexWithProof, err := m.db.GetLatestL2SentMsgBatchIndex()
latestBatchIndexWithProof, err := m.l2SentMsgOrm.GetLatestL2SentMsgBatchIndex(m.ctx)
if err != nil {
log.Error("MsgProofUpdater: Can not get latest L2SentMsgBatchIndex: ", "err", err)
continue
@@ -62,7 +64,7 @@ func (m *MsgProofUpdater) Start() {
start = uint64(latestBatchIndexWithProof) + 1
}
for i := start; i <= latestBatch.BatchIndex; i++ {
batch, err := m.db.GetRollupBatchByIndex(i)
batch, err := m.rollupOrm.GetRollupBatchByIndex(m.ctx, i)
if err != nil {
log.Error("MsgProofUpdater: Can not get RollupBatch: ", "err", err, "index", i)
break
@@ -111,8 +113,8 @@ func (m *MsgProofUpdater) initialize(ctx context.Context) {
func (m *MsgProofUpdater) initializeWithdrawTrie() error {
var batch *orm.RollupBatch
firstMsg, err := m.db.GetL2SentMessageByNonce(0)
if err != nil && !errors.Is(err, sql.ErrNoRows) {
firstMsg, err := m.l2SentMsgOrm.GetL2SentMessageByNonce(m.ctx, 0)
if err != nil {
return fmt.Errorf("failed to get first l2 message: %v", err)
}
// no l2 message
@@ -123,7 +125,7 @@ func (m *MsgProofUpdater) initializeWithdrawTrie() error {
}
// if no batch, return and wait for next try round
batch, err = m.db.GetLatestRollupBatch()
batch, err = m.rollupOrm.GetLatestRollupBatch(m.ctx)
if err != nil {
return fmt.Errorf("failed to get latest batch: %v", err)
}
@@ -135,7 +137,7 @@ func (m *MsgProofUpdater) initializeWithdrawTrie() error {
batchIndex := batch.BatchIndex
for {
var msg *orm.L2SentMsg
msg, err = m.db.GetLatestL2SentMsgLEHeight(batch.EndBlockNumber)
msg, err = m.l2SentMsgOrm.GetLatestL2SentMsgLEHeight(m.ctx, batch.EndBlockNumber)
if err != nil {
log.Warn("failed to get l2 sent message less than height", "endBlocknum", batch.EndBlockNumber, "err", err)
}
@@ -159,8 +161,8 @@ func (m *MsgProofUpdater) initializeWithdrawTrie() error {
// iterate for next batch
batchIndex--
batch, err = m.db.GetRollupBatchByIndex(batchIndex)
if err != nil {
batch, err = m.rollupOrm.GetRollupBatchByIndex(m.ctx, batchIndex)
if err != nil || batch == nil {
return fmt.Errorf("failed to get block batch %v: %v", batchIndex, err)
}
}
@@ -191,36 +193,26 @@ func (m *MsgProofUpdater) updateMsgProof(msgs []*orm.L2SentMsg, proofs [][]byte,
if len(msgs) != len(proofs) {
return fmt.Errorf("illegal state: len(msgs) != len(proofs)")
}
dbTx, err := m.db.Beginx()
err := m.db.Transaction(func(tx *gorm.DB) error {
for i, msg := range msgs {
proofHex := common.Bytes2Hex(proofs[i])
log.Debug("updateMsgProof", "msgHash", msg.MsgHash, "batchIndex", batchIndex, "proof", proofHex)
if err := m.l2SentMsgOrm.UpdateL2MessageProof(m.ctx, msg.MsgHash, proofHex, batchIndex, tx); err != nil {
return err
}
}
return nil
})
if err != nil {
return err
}
for i, msg := range msgs {
proofHex := common.Bytes2Hex(proofs[i])
log.Debug("updateMsgProof", "msgHash", msg.MsgHash, "batchIndex", batchIndex, "proof", proofHex)
if dbTxErr := m.db.UpdateL2MessageProofInDBTx(m.ctx, dbTx, msg.MsgHash, proofHex, batchIndex); dbTxErr != nil {
if err := dbTx.Rollback(); err != nil {
log.Error("dbTx.Rollback()", "err", err)
}
return dbTxErr
}
}
if dbTxErr := dbTx.Commit(); dbTxErr != nil {
if err := dbTx.Rollback(); err != nil {
log.Error("dbTx.Rollback()", "err", err)
}
return dbTxErr
}
return nil
}
// appendL2Messages will append all messages between firstBlock and lastBlock (both inclusive) to withdrawTrie and compute corresponding merkle proof of each message.
func (m *MsgProofUpdater) appendL2Messages(firstBlock, lastBlock uint64) ([]*orm.L2SentMsg, [][]byte, error) {
var msgProofs [][]byte
messages, err := m.db.GetL2SentMsgMsgHashByHeightRange(firstBlock, lastBlock)
messages, err := m.l2SentMsgOrm.GetL2SentMsgMsgHashByHeightRange(m.ctx, firstBlock, lastBlock)
if err != nil {
log.Error("GetL2SentMsgMsgHashByHeightRange failed", "error", err, "firstBlock", firstBlock, "lastBlock", lastBlock)
return messages, msgProofs, err

94
bridge-history-api/crossmsg/reorg_handle.go
View File

@@ -6,12 +6,13 @@ import (
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/ethclient"
"github.com/ethereum/go-ethereum/log"
"gorm.io/gorm"
"bridge-history-api/db"
"bridge-history-api/orm"
)
// ReorgHandling handles reorg function type
type ReorgHandling func(ctx context.Context, reorgHeight int64, db db.OrmFactory) error
type ReorgHandling func(ctx context.Context, reorgHeight uint64, db *gorm.DB) error
func reverseArray(arr []*types.Header) []*types.Header {
for i := 0; i < len(arr)/2; i++ {
@@ -60,73 +61,48 @@ func BackwardFindReorgBlock(ctx context.Context, headers []*types.Header, client
}
// L1ReorgHandling handles l1 reorg
func L1ReorgHandling(ctx context.Context, reorgHeight int64, db db.OrmFactory) error {
dbTx, err := db.Beginx()
if err != nil {
log.Crit("begin db tx failed", "err", err)
}
err = db.DeleteL1CrossMsgAfterHeightDBTx(dbTx, reorgHeight)
if err != nil {
if rollBackErr := dbTx.Rollback(); rollBackErr != nil {
log.Error("dbTx Rollback failed", "err", rollBackErr)
func L1ReorgHandling(ctx context.Context, reorgHeight uint64, db *gorm.DB) error {
l1CrossMsgOrm := orm.NewCrossMsg(db)
relayedOrm := orm.NewRelayedMsg(db)
err := db.Transaction(func(tx *gorm.DB) error {
if err := l1CrossMsgOrm.DeleteL1CrossMsgAfterHeight(ctx, reorgHeight, tx); err != nil {
log.Error("delete l1 cross msg from height", "height", reorgHeight, "err", err)
return err
}
log.Crit("delete l1 cross msg from height", "height", reorgHeight, "err", err)
}
err = db.DeleteL1RelayedHashAfterHeightDBTx(dbTx, reorgHeight)
if err != nil {
if rollBackErr := dbTx.Rollback(); rollBackErr != nil {
log.Error("dbTx Rollback failed", "err", rollBackErr)
if err := relayedOrm.DeleteL1RelayedHashAfterHeight(ctx, reorgHeight, tx); err != nil {
log.Error("delete l1 relayed msg from height", "height", reorgHeight, "err", err)
return err
}
log.Crit("delete l1 relayed hash from height", "height", reorgHeight, "err", err)
}
err = dbTx.Commit()
return nil
})
if err != nil {
if rollBackErr := dbTx.Rollback(); rollBackErr != nil {
log.Error("dbTx Rollback failed", "err", rollBackErr)
}
log.Error("commit tx failed", "err", err)
return err
log.Crit("l1 reorg handling failed", "err", err)
}
return nil
return err
}
// L2ReorgHandling handles l2 reorg
func L2ReorgHandling(ctx context.Context, reorgHeight int64, db db.OrmFactory) error {
dbTx, err := db.Beginx()
if err != nil {
if rollBackErr := dbTx.Rollback(); rollBackErr != nil {
log.Error("dbTx Rollback failed", "err", rollBackErr)
func L2ReorgHandling(ctx context.Context, reorgHeight uint64, db *gorm.DB) error {
l2CrossMsgOrm := orm.NewCrossMsg(db)
relayedOrm := orm.NewRelayedMsg(db)
l2SentMsgOrm := orm.NewL2SentMsg(db)
err := db.Transaction(func(tx *gorm.DB) error {
if err := l2CrossMsgOrm.DeleteL2CrossMsgFromHeight(ctx, reorgHeight, tx); err != nil {
log.Error("delete l2 cross msg from height", "height", reorgHeight, "err", err)
return err
}
log.Crit("begin db tx failed", "err", err)
}
err = db.DeleteL2CrossMsgFromHeightDBTx(dbTx, reorgHeight)
if err != nil {
if rollBackErr := dbTx.Rollback(); rollBackErr != nil {
log.Error("dbTx Rollback failed", "err", rollBackErr)
if err := relayedOrm.DeleteL2RelayedHashAfterHeight(ctx, reorgHeight, tx); err != nil {
log.Error("delete l2 relayed msg from height", "height", reorgHeight, "err", err)
return err
}
log.Crit("delete l2 cross msg from height", "height", reorgHeight, "err", err)
}
err = db.DeleteL2RelayedHashAfterHeightDBTx(dbTx, reorgHeight)
if err != nil {
if rollBackErr := dbTx.Rollback(); rollBackErr != nil {
log.Error("dbTx Rollback failed", "err", rollBackErr)
if err := l2SentMsgOrm.DeleteL2SentMsgAfterHeight(ctx, reorgHeight, tx); err != nil {
log.Error("delete l2 sent msg from height", "height", reorgHeight, "err", err)
return err
}
log.Crit("delete l2 relayed hash from height", "height", reorgHeight, "err", err)
}
err = db.DeleteL2SentMsgAfterHeightDBTx(dbTx, reorgHeight)
return nil
})
if err != nil {
if rollBackErr := dbTx.Rollback(); rollBackErr != nil {
log.Error("dbTx Rollback failed", "err", rollBackErr)
}
log.Crit("delete l2 sent msg from height", "height", reorgHeight, "err", err)
log.Crit("l2 reorg handling failed", "err", err)
}
err = dbTx.Commit()
if err != nil {
if rollBackErr := dbTx.Rollback(); rollBackErr != nil {
log.Error("dbTx Rollback failed", "err", rollBackErr)
}
log.Error("commit tx failed", "err", err)
return err
}
return nil
return err
}

71
bridge-history-api/db/orm/batch.go
View File

@@ -1,71 +0,0 @@
package orm
import (
"database/sql"
"github.com/ethereum/go-ethereum/log"
"github.com/jmoiron/sqlx"
)
type rollupBatchOrm struct {
db *sqlx.DB
}
// RollupBatch is the struct for rollup_batch table
type RollupBatch struct {
ID uint64 `json:"id" db:"id"`
BatchIndex uint64 `json:"batch_index" db:"batch_index"`
BatchHash string `json:"batch_hash" db:"batch_hash"`
CommitHeight uint64 `json:"commit_height" db:"commit_height"`
StartBlockNumber uint64 `json:"start_block_number" db:"start_block_number"`
EndBlockNumber uint64 `json:"end_block_number" db:"end_block_number"`
}
// NewRollupBatchOrm create an NewRollupBatchOrm instance
func NewRollupBatchOrm(db *sqlx.DB) RollupBatchOrm {
return &rollupBatchOrm{db: db}
}
func (b *rollupBatchOrm) BatchInsertRollupBatchDBTx(dbTx *sqlx.Tx, batches []*RollupBatch) error {
if len(batches) == 0 {
return nil
}
var err error
batchMaps := make([]map[string]interface{}, len(batches))
for i, batch := range batches {
batchMaps[i] = map[string]interface{}{
"commit_height": batch.CommitHeight,
"batch_index": batch.BatchIndex,
"batch_hash": batch.BatchHash,
"start_block_number": batch.StartBlockNumber,
"end_block_number": batch.EndBlockNumber,
}
}
_, err = dbTx.NamedExec(`insert into rollup_batch(commit_height, batch_index, batch_hash, start_block_number, end_block_number) values(:commit_height, :batch_index, :batch_hash, :start_block_number, :end_block_number);`, batchMaps)
if err != nil {
log.Error("BatchInsertRollupBatchDBTx: failed to insert batch event msgs", "err", err)
return err
}
return nil
}
func (b *rollupBatchOrm) GetLatestRollupBatch() (*RollupBatch, error) {
result := &RollupBatch{}
row := b.db.QueryRowx(`SELECT id, batch_index, commit_height, batch_hash, start_block_number, end_block_number FROM rollup_batch ORDER BY batch_index DESC LIMIT 1;`)
if err := row.StructScan(result); err != nil {
if err == sql.ErrNoRows {
return nil, nil
}
return nil, err
}
return result, nil
}
func (b *rollupBatchOrm) GetRollupBatchByIndex(index uint64) (*RollupBatch, error) {
result := &RollupBatch{}
row := b.db.QueryRowx(`SELECT id, batch_index, batch_hash, commit_height, start_block_number, end_block_number FROM rollup_batch WHERE batch_index = $1;`, index)
if err := row.StructScan(result); err != nil {
return nil, err
}
return result, nil
}

132
bridge-history-api/db/orm/interface.go
View File

@@ -1,132 +0,0 @@
package orm
import (
"context"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/jmoiron/sqlx"
)
// AssetType can be ETH/ERC20/ERC1155/ERC721
type AssetType int
// MsgType can be layer1/layer2 msg
type MsgType int
func (a AssetType) String() string {
switch a {
case ETH:
return "ETH"
case ERC20:
return "ERC20"
case ERC1155:
return "ERC1155"
case ERC721:
return "ERC721"
}
return "Unknown Asset Type"
}
const (
// ETH = 0
ETH AssetType = iota
// ERC20 = 1
ERC20
// ERC721 = 2
ERC721
// ERC1155 = 3
ERC1155
)
const (
// UnknownMsg = 0
UnknownMsg MsgType = iota
// Layer1Msg = 1
Layer1Msg
// Layer2Msg = 2
Layer2Msg
)
// CrossMsg represents a cross message from layer 1 to layer 2
type CrossMsg struct {
ID uint64 `json:"id" db:"id"`
MsgHash string `json:"msg_hash" db:"msg_hash"`
Height uint64 `json:"height" db:"height"`
Sender string `json:"sender" db:"sender"`
Target string `json:"target" db:"target"`
Amount string `json:"amount" db:"amount"`
Layer1Hash string `json:"layer1_hash" db:"layer1_hash"`
Layer2Hash string `json:"layer2_hash" db:"layer2_hash"`
Layer1Token string `json:"layer1_token" db:"layer1_token"`
Layer2Token string `json:"layer2_token" db:"layer2_token"`
TokenIDs string `json:"token_ids" db:"token_ids"`
TokenAmounts string `json:"token_amounts" db:"token_amounts"`
Asset int `json:"asset" db:"asset"`
MsgType int `json:"msg_type" db:"msg_type"`
Timestamp *time.Time `json:"timestamp" db:"block_timestamp"`
CreatedAt *time.Time `json:"created_at" db:"created_at"`
UpdatedAt *time.Time `json:"updated_at" db:"updated_at"`
DeletedAt *time.Time `json:"deleted_at" db:"deleted_at"`
}
// L1CrossMsgOrm provides operations on l1_cross_message table
type L1CrossMsgOrm interface {
GetL1CrossMsgByHash(l1Hash common.Hash) (*CrossMsg, error)
GetL1CrossMsgsByAddress(sender common.Address) ([]*CrossMsg, error)
BatchInsertL1CrossMsgDBTx(dbTx *sqlx.Tx, messages []*CrossMsg) error
// UpdateL1CrossMsgHashDBTx invoked when SentMessage event is received
UpdateL1CrossMsgHashDBTx(ctx context.Context, dbTx *sqlx.Tx, l1Hash, msgHash common.Hash) error
UpdateL1CrossMsgHash(ctx context.Context, l1Hash, msgHash common.Hash) error
GetLatestL1ProcessedHeight() (int64, error)
DeleteL1CrossMsgAfterHeightDBTx(dbTx *sqlx.Tx, height int64) error
UpdateL1BlockTimestamp(height uint64, timestamp time.Time) error
GetL1EarliestNoBlockTimestampHeight() (uint64, error)
}
// L2CrossMsgOrm provides operations on cross_message table
type L2CrossMsgOrm interface {
GetL2CrossMsgByHash(l2Hash common.Hash) (*CrossMsg, error)
GetL2CrossMsgByAddress(sender common.Address) ([]*CrossMsg, error)
BatchInsertL2CrossMsgDBTx(dbTx *sqlx.Tx, messages []*CrossMsg) error
// UpdateL2CrossMsgHashDBTx invoked when SentMessage event is received
UpdateL2CrossMsgHashDBTx(ctx context.Context, dbTx *sqlx.Tx, l2Hash, msgHash common.Hash) error
UpdateL2CrossMsgHash(ctx context.Context, l2Hash, msgHash common.Hash) error
GetLatestL2ProcessedHeight() (int64, error)
DeleteL2CrossMsgFromHeightDBTx(dbTx *sqlx.Tx, height int64) error
UpdateL2BlockTimestamp(height uint64, timestamp time.Time) error
GetL2EarliestNoBlockTimestampHeight() (uint64, error)
GetL2CrossMsgByMsgHashList(msgHashList []string) ([]*CrossMsg, error)
}
// RelayedMsgOrm provides operations on relayed_msg table
type RelayedMsgOrm interface {
BatchInsertRelayedMsgDBTx(dbTx *sqlx.Tx, messages []*RelayedMsg) error
GetRelayedMsgByHash(msgHash string) (*RelayedMsg, error)
GetLatestRelayedHeightOnL1() (int64, error)
GetLatestRelayedHeightOnL2() (int64, error)
DeleteL1RelayedHashAfterHeightDBTx(dbTx *sqlx.Tx, height int64) error
DeleteL2RelayedHashAfterHeightDBTx(dbTx *sqlx.Tx, height int64) error
}
// L2SentMsgOrm provides operations on l2_sent_msg table
type L2SentMsgOrm interface {
BatchInsertL2SentMsgDBTx(dbTx *sqlx.Tx, messages []*L2SentMsg) error
GetL2SentMsgByHash(l2Hash string) (*L2SentMsg, error)
GetLatestSentMsgHeightOnL2() (int64, error)
GetL2SentMessageByNonce(nonce uint64) (*L2SentMsg, error)
GetLatestL2SentMsgLEHeight(endBlockNumber uint64) (*L2SentMsg, error)
GetL2SentMsgMsgHashByHeightRange(startHeight, endHeight uint64) ([]*L2SentMsg, error)
UpdateL2MessageProofInDBTx(ctx context.Context, dbTx *sqlx.Tx, msgHash string, proof string, batchIndex uint64) error
GetLatestL2SentMsgBatchIndex() (int64, error)
GetClaimableL2SentMsgByAddressWithOffset(address string, offset int64, limit int64) ([]*L2SentMsg, error)
GetClaimableL2SentMsgByAddressTotalNum(address string) (uint64, error)
DeleteL2SentMsgAfterHeightDBTx(dbTx *sqlx.Tx, height int64) error
}
// RollupBatchOrm provides operations on rollup_batch table
type RollupBatchOrm interface {
GetLatestRollupBatch() (*RollupBatch, error)
GetRollupBatchByIndex(index uint64) (*RollupBatch, error)
BatchInsertRollupBatchDBTx(dbTx *sqlx.Tx, messages []*RollupBatch) error
}

148
bridge-history-api/db/orm/l1_cross_msg.go
View File

@@ -1,148 +0,0 @@
package orm
import (
"context"
"database/sql"
"errors"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"github.com/jmoiron/sqlx"
)
type l1CrossMsgOrm struct {
db *sqlx.DB
}
// NewL1CrossMsgOrm create an NewL1CrossMsgOrm instance
func NewL1CrossMsgOrm(db *sqlx.DB) L1CrossMsgOrm {
return &l1CrossMsgOrm{db: db}
}
func (l *l1CrossMsgOrm) GetL1CrossMsgByHash(l1Hash common.Hash) (*CrossMsg, error) {
result := &CrossMsg{}
row := l.db.QueryRowx(`SELECT * FROM cross_message WHERE layer1_hash = $1 AND msg_type = $2 AND deleted_at IS NULL;`, l1Hash.String(), Layer1Msg)
if err := row.StructScan(result); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
}
return nil, err
}
return result, nil
}
// GetL1CrossMsgsByAddress returns all layer1 cross messages under given address
// Warning: return empty slice if no data found
func (l *l1CrossMsgOrm) GetL1CrossMsgsByAddress(sender common.Address) ([]*CrossMsg, error) {
var results []*CrossMsg
rows, err := l.db.Queryx(`SELECT * FROM cross_message WHERE sender = $1 AND msg_type = 1 AND deleted_at IS NULL;`, sender.String(), Layer1Msg)
if err != nil {
return nil, err
}
defer func() {
if err = rows.Close(); err != nil {
log.Error("failed to close rows", "err", err)
}
}()
for rows.Next() {
msg := &CrossMsg{}
if err = rows.StructScan(msg); err != nil {
break
}
results = append(results, msg)
}
if len(results) == 0 && errors.Is(err, sql.ErrNoRows) {
} else if err != nil {
return nil, err
}
return results, nil
}
func (l *l1CrossMsgOrm) BatchInsertL1CrossMsgDBTx(dbTx *sqlx.Tx, messages []*CrossMsg) error {
if len(messages) == 0 {
return nil
}
var err error
messageMaps := make([]map[string]interface{}, len(messages))
for i, msg := range messages {
messageMaps[i] = map[string]interface{}{
"height": msg.Height,
"sender": msg.Sender,
"target": msg.Target,
"amount": msg.Amount,
"asset": msg.Asset,
"msg_hash": msg.MsgHash,
"layer1_hash": msg.Layer1Hash,
"layer1_token": msg.Layer1Token,
"layer2_token": msg.Layer2Token,
"token_ids": msg.TokenIDs,
"msg_type": Layer1Msg,
}
}
_, err = dbTx.NamedExec(`insert into cross_message(height, sender, target, amount, asset, msg_hash, layer1_hash, layer1_token, layer2_token, token_ids, msg_type) values(:height, :sender, :target, :amount, :asset, :msg_hash, :layer1_hash, :layer1_token, :layer2_token, :token_ids, :msg_type);`, messageMaps)
if err != nil {
log.Error("BatchInsertL1CrossMsgDBTx: failed to insert l1 cross msgs", "err", err)
return err
}
return nil
}
// UpdateL1CrossMsgHashDBTx update l1 cross msg hash in db, no need to check msg_type since layer1_hash wont be empty if its layer1 msg
func (l *l1CrossMsgOrm) UpdateL1CrossMsgHashDBTx(ctx context.Context, dbTx *sqlx.Tx, l1Hash, msgHash common.Hash) error {
if _, err := dbTx.ExecContext(ctx, l.db.Rebind("update public.cross_message set msg_hash = ? where layer1_hash = ? AND deleted_at IS NULL;"), msgHash.String(), l1Hash.String()); err != nil {
return err
}
return nil
}
func (l *l1CrossMsgOrm) UpdateL1CrossMsgHash(ctx context.Context, l1Hash, msgHash common.Hash) error {
if _, err := l.db.ExecContext(ctx, l.db.Rebind("update public.l1_cross_message set msg_hash = ? where layer1_hash = ? AND deleted_at IS NULL;"), msgHash.String(), l1Hash.String()); err != nil {
return err
}
return nil
}
func (l *l1CrossMsgOrm) GetLatestL1ProcessedHeight() (int64, error) {
row := l.db.QueryRowx(`SELECT height FROM cross_message WHERE msg_type = $1 AND deleted_at IS NULL ORDER BY id DESC LIMIT 1;`, Layer1Msg)
var result sql.NullInt64
if err := row.Scan(&result); err != nil {
if err == sql.ErrNoRows || !result.Valid {
return -1, nil
}
return 0, err
}
if result.Valid {
return result.Int64, nil
}
return 0, nil
}
func (l *l1CrossMsgOrm) DeleteL1CrossMsgAfterHeightDBTx(dbTx *sqlx.Tx, height int64) error {
if _, err := l.db.Exec(`UPDATE cross_message SET deleted_at = current_timestamp WHERE height > $1 AND msg_type = $2;`, height, Layer1Msg); err != nil {
return err
}
return nil
}
func (l *l1CrossMsgOrm) UpdateL1BlockTimestamp(height uint64, timestamp time.Time) error {
if _, err := l.db.Exec(`UPDATE cross_message SET block_timestamp = $1 where height = $2 AND msg_type = $3 AND deleted_at IS NULL`, timestamp, height, Layer1Msg); err != nil {
return err
}
return nil
}
func (l *l1CrossMsgOrm) GetL1EarliestNoBlockTimestampHeight() (uint64, error) {
row := l.db.QueryRowx(`SELECT height FROM cross_message WHERE block_timestamp IS NULL AND msg_type = $1 AND deleted_at IS NULL ORDER BY height ASC LIMIT 1;`, Layer1Msg)
var result uint64
if err := row.Scan(&result); err != nil {
if err == sql.ErrNoRows {
return 0, nil
}
return 0, err
}
return result, nil
}

177
bridge-history-api/db/orm/l2_cross_msg.go
View File

@@ -1,177 +0,0 @@
package orm
import (
"context"
"database/sql"
"errors"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"github.com/jmoiron/sqlx"
"github.com/lib/pq"
)
type l2CrossMsgOrm struct {
db *sqlx.DB
}
// NewL2CrossMsgOrm create an NewL2CrossMsgOrm instance
func NewL2CrossMsgOrm(db *sqlx.DB) L2CrossMsgOrm {
return &l2CrossMsgOrm{db: db}
}
func (l *l2CrossMsgOrm) GetL2CrossMsgByHash(l2Hash common.Hash) (*CrossMsg, error) {
result := &CrossMsg{}
row := l.db.QueryRowx(`SELECT * FROM cross_message WHERE layer2_hash = $1 AND deleted_at IS NULL;`, l2Hash.String())
if err := row.StructScan(result); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
}
return nil, err
}
return result, nil
}
// GetL2CrossMsgByAddress returns all layer2 cross messages under given address
// Warning: return empty slice if no data found
func (l *l2CrossMsgOrm) GetL2CrossMsgByAddress(sender common.Address) ([]*CrossMsg, error) {
var results []*CrossMsg
rows, err := l.db.Queryx(`SELECT * FROM cross_message WHERE sender = $1 AND msg_type = $2 AND deleted_at IS NULL;`, sender.String(), Layer2Msg)
if err != nil {
return nil, err
}
defer func() {
if err = rows.Close(); err != nil {
log.Error("failed to close rows", "err", err)
}
}()
for rows.Next() {
msg := &CrossMsg{}
if err = rows.StructScan(msg); err != nil {
break
}
results = append(results, msg)
}
if len(results) == 0 && errors.Is(err, sql.ErrNoRows) {
// log.Warn("no unprocessed layer1 messages in db", "err", err)
} else if err != nil {
return nil, err
}
return results, nil
}
func (l *l2CrossMsgOrm) DeleteL2CrossMsgFromHeightDBTx(dbTx *sqlx.Tx, height int64) error {
_, err := dbTx.Exec(`UPDATE cross_message SET deleted_at = current_timestamp where height > $1 AND msg_type = $2 ;`, height, Layer2Msg)
if err != nil {
log.Error("DeleteL1CrossMsgAfterHeightDBTx: failed to delete", "height", height, "err", err)
return err
}
return nil
}
func (l *l2CrossMsgOrm) BatchInsertL2CrossMsgDBTx(dbTx *sqlx.Tx, messages []*CrossMsg) error {
if len(messages) == 0 {
return nil
}
var err error
messageMaps := make([]map[string]interface{}, len(messages))
for i, msg := range messages {
messageMaps[i] = map[string]interface{}{
"height": msg.Height,
"sender": msg.Sender,
"target": msg.Target,
"asset": msg.Asset,
"msg_hash": msg.MsgHash,
"layer2_hash": msg.Layer2Hash,
"layer1_token": msg.Layer1Token,
"layer2_token": msg.Layer2Token,
"token_ids": msg.TokenIDs,
"amount": msg.Amount,
"msg_type": Layer2Msg,
}
}
_, err = dbTx.NamedExec(`insert into cross_message(height, sender, target, asset, msg_hash, layer2_hash, layer1_token, layer2_token, token_ids, amount, msg_type) values(:height, :sender, :target, :asset, :msg_hash, :layer2_hash, :layer1_token, :layer2_token, :token_ids, :amount, :msg_type);`, messageMaps)
if err != nil {
log.Error("BatchInsertL2CrossMsgDBTx: failed to insert l2 cross msgs", "err", err)
return err
}
return nil
}
func (l *l2CrossMsgOrm) UpdateL2CrossMsgHashDBTx(ctx context.Context, dbTx *sqlx.Tx, l2Hash, msgHash common.Hash) error {
if _, err := dbTx.ExecContext(ctx, l.db.Rebind("update cross_message set msg_hash = ? where layer2_hash = ? AND deleted_at IS NULL;"), msgHash.String(), l2Hash.String()); err != nil {
return err
}
return nil
}
func (l *l2CrossMsgOrm) UpdateL2CrossMsgHash(ctx context.Context, l2Hash, msgHash common.Hash) error {
if _, err := l.db.ExecContext(ctx, l.db.Rebind("update cross_message set msg_hash = ? where layer2_hash = ? AND deleted_at IS NULL;"), msgHash.String(), l2Hash.String()); err != nil {
return err
}
return nil
}
func (l *l2CrossMsgOrm) GetLatestL2ProcessedHeight() (int64, error) {
row := l.db.QueryRowx(`SELECT height FROM cross_message WHERE msg_type = $1 AND deleted_at IS NULL ORDER BY id DESC LIMIT 1;`, Layer2Msg)
var result sql.NullInt64
if err := row.Scan(&result); err != nil {
if err == sql.ErrNoRows || !result.Valid {
return -1, nil
}
return 0, err
}
if result.Valid {
return result.Int64, nil
}
return 0, nil
}
func (l *l2CrossMsgOrm) UpdateL2BlockTimestamp(height uint64, timestamp time.Time) error {
if _, err := l.db.Exec(`UPDATE cross_message SET block_timestamp = $1 where height = $2 AND msg_type = $3 AND deleted_at IS NULL`, timestamp, height, Layer2Msg); err != nil {
return err
}
return nil
}
func (l *l2CrossMsgOrm) GetL2EarliestNoBlockTimestampHeight() (uint64, error) {
row := l.db.QueryRowx(`SELECT height FROM cross_message WHERE block_timestamp IS NULL AND msg_type = $1 AND deleted_at IS NULL ORDER BY height ASC LIMIT 1;`, Layer2Msg)
var result uint64
if err := row.Scan(&result); err != nil {
if err == sql.ErrNoRows {
return 0, nil
}
return 0, err
}
return result, nil
}
func (l *l2CrossMsgOrm) GetL2CrossMsgByMsgHashList(msgHashList []string) ([]*CrossMsg, error) {
var results []*CrossMsg
rows, err := l.db.Queryx(`SELECT * FROM cross_message WHERE msg_hash = ANY($1) AND msg_type = $2 AND deleted_at IS NULL;`, pq.Array(msgHashList), Layer2Msg)
if err != nil {
return nil, err
}
defer func() {
if err = rows.Close(); err != nil {
log.Error("failed to close rows", "err", err)
}
}()
for rows.Next() {
msg := &CrossMsg{}
if err = rows.StructScan(msg); err != nil {
break
}
results = append(results, msg)
}
if err != nil && !errors.Is(err, sql.ErrNoRows) {
return nil, err
}
if len(results) == 0 {
log.Debug("no L2CrossMsg under given msg hashes", "msg hash list", msgHashList)
}
return results, nil
}

189
bridge-history-api/db/orm/l2_sent_msg.go
View File

@@ -1,189 +0,0 @@
package orm
import (
"context"
"database/sql"
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/jmoiron/sqlx"
)
// L2SentMsg defines the struct for l2_sent_msg table record
type L2SentMsg struct {
ID uint64 `json:"id" db:"id"`
OriginalSender string `json:"original_sender" db:"original_sender"`
TxHash string `json:"tx_hash" db:"tx_hash"`
MsgHash string `json:"msg_hash" db:"msg_hash"`
Sender string `json:"sender" db:"sender"`
Target string `json:"target" db:"target"`
Value string `json:"value" db:"value"`
Height uint64 `json:"height" db:"height"`
Nonce uint64 `json:"nonce" db:"nonce"`
BatchIndex uint64 `json:"batch_index" db:"batch_index"`
MsgProof string `json:"msg_proof" db:"msg_proof"`
MsgData string `json:"msg_data" db:"msg_data"`
CreatedAt *time.Time `json:"created_at" db:"created_at"`
UpdatedAt *time.Time `json:"updated_at" db:"updated_at"`
DeletedAt *time.Time `json:"deleted_at" db:"deleted_at"`
}
type l2SentMsgOrm struct {
db *sqlx.DB
}
// NewL2SentMsgOrm create an NewRollupBatchOrm instance
func NewL2SentMsgOrm(db *sqlx.DB) L2SentMsgOrm {
return &l2SentMsgOrm{db: db}
}
func (l *l2SentMsgOrm) GetL2SentMsgByHash(msgHash string) (*L2SentMsg, error) {
result := &L2SentMsg{}
row := l.db.QueryRowx(`SELECT * FROM l2_sent_msg WHERE msg_hash = $1 AND deleted_at IS NULL;`, msgHash)
if err := row.StructScan(result); err != nil {
return nil, err
}
return result, nil
}
func (l *l2SentMsgOrm) BatchInsertL2SentMsgDBTx(dbTx *sqlx.Tx, messages []*L2SentMsg) error {
if len(messages) == 0 {
return nil
}
var err error
messageMaps := make([]map[string]interface{}, len(messages))
for i, msg := range messages {
messageMaps[i] = map[string]interface{}{
"original_sender": msg.OriginalSender,
"tx_hash": msg.TxHash,
"sender": msg.Sender,
"target": msg.Target,
"value": msg.Value,
"msg_hash": msg.MsgHash,
"height": msg.Height,
"nonce": msg.Nonce,
"batch_index": msg.BatchIndex,
"msg_proof": msg.MsgProof,
"msg_data": msg.MsgData,
}
}
_, err = dbTx.NamedExec(`insert into l2_sent_msg(original_sender, tx_hash, sender, target, value, msg_hash, height, nonce, batch_index, msg_proof, msg_data) values(:original_sender, :tx_hash, :sender, :target, :value, :msg_hash, :height, :nonce, :batch_index, :msg_proof, :msg_data);`, messageMaps)
if err != nil {
log.Error("BatchInsertL2SentMsgDBTx: failed to insert l2 sent msgs", "err", err)
return err
}
return err
}
func (l *l2SentMsgOrm) GetLatestSentMsgHeightOnL2() (int64, error) {
row := l.db.QueryRow(`SELECT height FROM l2_sent_msg WHERE deleted_at IS NULL ORDER BY nonce DESC LIMIT 1;`)
var result sql.NullInt64
if err := row.Scan(&result); err != nil {
if err == sql.ErrNoRows || !result.Valid {
return -1, nil
}
return 0, err
}
if result.Valid {
return result.Int64, nil
}
return 0, nil
}
func (l *l2SentMsgOrm) UpdateL2MessageProofInDBTx(ctx context.Context, dbTx *sqlx.Tx, msgHash string, proof string, batchIndex uint64) error {
if _, err := dbTx.ExecContext(ctx, l.db.Rebind("update l2_sent_msg set msg_proof = ?, batch_index = ? where msg_hash = ? AND deleted_at IS NULL;"), proof, batchIndex, msgHash); err != nil {
return err
}
return nil
}
func (l *l2SentMsgOrm) GetLatestL2SentMsgBatchIndex() (int64, error) {
row := l.db.QueryRow(`SELECT batch_index FROM l2_sent_msg WHERE batch_index != 0 AND deleted_at IS NULL ORDER BY batch_index DESC LIMIT 1;`)
var result sql.NullInt64
if err := row.Scan(&result); err != nil {
if err == sql.ErrNoRows || !result.Valid {
return -1, nil
}
return -1, err
}
if result.Valid {
return result.Int64, nil
}
return -1, nil
}
func (l *l2SentMsgOrm) GetL2SentMsgMsgHashByHeightRange(startHeight, endHeight uint64) ([]*L2SentMsg, error) {
var results []*L2SentMsg
rows, err := l.db.Queryx(`SELECT * FROM l2_sent_msg WHERE height >= $1 AND height <= $2 AND deleted_at IS NULL ORDER BY nonce ASC;`, startHeight, endHeight)
if err != nil {
return nil, err
}
defer func() {
if err = rows.Close(); err != nil {
log.Error("failed to close rows", "err", err)
}
}()
for rows.Next() {
msg := &L2SentMsg{}
if err = rows.StructScan(msg); err != nil {
break
}
results = append(results, msg)
}
return results, err
}
func (l *l2SentMsgOrm) GetL2SentMessageByNonce(nonce uint64) (*L2SentMsg, error) {
result := &L2SentMsg{}
row := l.db.QueryRowx(`SELECT * FROM l2_sent_msg WHERE nonce = $1 AND deleted_at IS NULL;`, nonce)
err := row.StructScan(result)
if err != nil {
return nil, err
}
return result, nil
}
func (l *l2SentMsgOrm) GetLatestL2SentMsgLEHeight(endBlockNumber uint64) (*L2SentMsg, error) {
result := &L2SentMsg{}
row := l.db.QueryRowx(`select * from l2_sent_msg where height <= $1 AND deleted_at IS NULL order by nonce desc limit 1`, endBlockNumber)
err := row.StructScan(result)
if err != nil {
return nil, err
}
return result, nil
}
func (l *l2SentMsgOrm) DeleteL2SentMsgAfterHeightDBTx(dbTx *sqlx.Tx, height int64) error {
_, err := dbTx.Exec(`UPDATE l2_sent_msg SET deleted_at = current_timestamp WHERE height > $1;`, height)
return err
}
func (l *l2SentMsgOrm) GetClaimableL2SentMsgByAddressWithOffset(address string, offset int64, limit int64) ([]*L2SentMsg, error) {
var results []*L2SentMsg
rows, err := l.db.Queryx(`SELECT * FROM l2_sent_msg WHERE id NOT IN (SELECT l2_sent_msg.id FROM l2_sent_msg INNER JOIN relayed_msg ON l2_sent_msg.msg_hash = relayed_msg.msg_hash WHERE l2_sent_msg.deleted_at IS NULL AND relayed_msg.deleted_at IS NULL) AND (original_sender=$1 OR sender = $1) ORDER BY id DESC LIMIT $2 OFFSET $3;`, address, limit, offset)
if err != nil {
return nil, err
}
defer func() {
if err = rows.Close(); err != nil {
log.Error("failed to close rows", "err", err)
}
}()
for rows.Next() {
msg := &L2SentMsg{}
if err = rows.StructScan(msg); err != nil {
break
}
results = append(results, msg)
}
return results, err
}
func (l *l2SentMsgOrm) GetClaimableL2SentMsgByAddressTotalNum(address string) (uint64, error) {
var count uint64
row := l.db.QueryRowx(`SELECT COUNT(*) FROM l2_sent_msg WHERE id NOT IN (SELECT l2_sent_msg.id FROM l2_sent_msg INNER JOIN relayed_msg ON l2_sent_msg.msg_hash = relayed_msg.msg_hash WHERE l2_sent_msg.deleted_at IS NULL AND relayed_msg.deleted_at IS NULL) AND (original_sender=$1 OR sender = $1);`, address)
if err := row.Scan(&count); err != nil {
return 0, err
}
return count, nil
}

100
bridge-history-api/db/orm/relayed_msg.go
View File

@@ -1,100 +0,0 @@
package orm
import (
"database/sql"
"errors"
"github.com/ethereum/go-ethereum/log"
"github.com/jmoiron/sqlx"
)
// RelayedMsg is the struct for relayed_msg table
type RelayedMsg struct {
MsgHash string `json:"msg_hash" db:"msg_hash"`
Height uint64 `json:"height" db:"height"`
Layer1Hash string `json:"layer1_hash" db:"layer1_hash"`
Layer2Hash string `json:"layer2_hash" db:"layer2_hash"`
}
type relayedMsgOrm struct {
db *sqlx.DB
}
// NewRelayedMsgOrm create an NewRelayedMsgOrm instance
func NewRelayedMsgOrm(db *sqlx.DB) RelayedMsgOrm {
return &relayedMsgOrm{db: db}
}
func (l *relayedMsgOrm) BatchInsertRelayedMsgDBTx(dbTx *sqlx.Tx, messages []*RelayedMsg) error {
if len(messages) == 0 {
return nil
}
var err error
messageMaps := make([]map[string]interface{}, len(messages))
for i, msg := range messages {
messageMaps[i] = map[string]interface{}{
"msg_hash": msg.MsgHash,
"height": msg.Height,
"layer1_hash": msg.Layer1Hash,
"layer2_hash": msg.Layer2Hash,
}
}
_, err = dbTx.NamedExec(`insert into relayed_msg(msg_hash, height, layer1_hash, layer2_hash) values(:msg_hash, :height, :layer1_hash, :layer2_hash);`, messageMaps)
if err != nil {
log.Error("BatchInsertRelayedMsgDBTx: failed to insert relayed msgs", "err", err)
return err
}
return nil
}
func (l *relayedMsgOrm) GetRelayedMsgByHash(msgHash string) (*RelayedMsg, error) {
result := &RelayedMsg{}
row := l.db.QueryRowx(`SELECT msg_hash, height, layer1_hash, layer2_hash FROM relayed_msg WHERE msg_hash = $1 AND deleted_at IS NULL;`, msgHash)
if err := row.StructScan(result); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
}
return nil, err
}
return result, nil
}
func (l *relayedMsgOrm) GetLatestRelayedHeightOnL1() (int64, error) {
row := l.db.QueryRow(`SELECT height FROM relayed_msg WHERE layer1_hash != '' AND deleted_at IS NULL ORDER BY height DESC LIMIT 1;`)
var result sql.NullInt64
if err := row.Scan(&result); err != nil {
if err == sql.ErrNoRows || !result.Valid {
return -1, nil
}
return 0, err
}
if result.Valid {
return result.Int64, nil
}
return 0, nil
}
func (l *relayedMsgOrm) GetLatestRelayedHeightOnL2() (int64, error) {
row := l.db.QueryRow(`SELECT height FROM relayed_msg WHERE layer2_hash != '' AND deleted_at IS NULL ORDER BY height DESC LIMIT 1;`)
var result sql.NullInt64
if err := row.Scan(&result); err != nil {
if err == sql.ErrNoRows || !result.Valid {
return -1, nil
}
return 0, err
}
if result.Valid {
return result.Int64, nil
}
return 0, nil
}
func (l *relayedMsgOrm) DeleteL1RelayedHashAfterHeightDBTx(dbTx *sqlx.Tx, height int64) error {
_, err := dbTx.Exec(`UPDATE relayed_msg SET deleted_at = current_timestamp WHERE height > $1 AND layer1_hash != '';`, height)
return err
}
func (l *relayedMsgOrm) DeleteL2RelayedHashAfterHeightDBTx(dbTx *sqlx.Tx, height int64) error {
_, err := dbTx.Exec(`UPDATE relayed_msg SET deleted_at = current_timestamp WHERE height > $1 AND layer2_hash != '';`, height)
return err
}

103
bridge-history-api/db/orm_factory.go
View File

@@ -1,103 +0,0 @@
package db
import (
"database/sql"
"errors"
"github.com/ethereum/go-ethereum/log"
"github.com/jmoiron/sqlx"
_ "github.com/lib/pq" //nolint:golint
"bridge-history-api/config"
"bridge-history-api/db/orm"
)
// OrmFactory include all ormFactory interface
type OrmFactory interface {
orm.L1CrossMsgOrm
orm.L2CrossMsgOrm
orm.RelayedMsgOrm
orm.L2SentMsgOrm
orm.RollupBatchOrm
GetTotalCrossMsgCountByAddress(sender string) (uint64, error)
GetCrossMsgsByAddressWithOffset(sender string, offset int64, limit int64) ([]*orm.CrossMsg, error)
GetDB() *sqlx.DB
Beginx() (*sqlx.Tx, error)
Close() error
}
type ormFactory struct {
orm.L1CrossMsgOrm
orm.L2CrossMsgOrm
orm.RelayedMsgOrm
orm.L2SentMsgOrm
orm.RollupBatchOrm
*sqlx.DB
}
// NewOrmFactory create an ormFactory factory include all ormFactory interface
func NewOrmFactory(cfg *config.Config) (OrmFactory, error) {
// Initialize sql/sqlx
db, err := sqlx.Open(cfg.DB.DriverName, cfg.DB.DSN)
if err != nil {
return nil, err
}
db.SetMaxOpenConns(cfg.DB.MaxOpenNum)
db.SetMaxIdleConns(cfg.DB.MaxIdleNum)
if err = db.Ping(); err != nil {
return nil, err
}
return &ormFactory{
L1CrossMsgOrm: orm.NewL1CrossMsgOrm(db),
L2CrossMsgOrm: orm.NewL2CrossMsgOrm(db),
RelayedMsgOrm: orm.NewRelayedMsgOrm(db),
L2SentMsgOrm: orm.NewL2SentMsgOrm(db),
RollupBatchOrm: orm.NewRollupBatchOrm(db),
DB: db,
}, nil
}
func (o *ormFactory) GetDB() *sqlx.DB {
return o.DB
}
func (o *ormFactory) Beginx() (*sqlx.Tx, error) {
return o.DB.Beginx()
}
func (o *ormFactory) GetTotalCrossMsgCountByAddress(sender string) (uint64, error) {
var count uint64
row := o.DB.QueryRowx(`SELECT COUNT(*) FROM cross_message WHERE sender = $1 AND deleted_at IS NULL;`, sender)
if err := row.Scan(&count); err != nil {
return 0, err
}
return count, nil
}
func (o *ormFactory) GetCrossMsgsByAddressWithOffset(sender string, offset int64, limit int64) ([]*orm.CrossMsg, error) {
para := sender
var results []*orm.CrossMsg
rows, err := o.DB.Queryx(`SELECT * FROM cross_message WHERE sender = $1 AND deleted_at IS NULL ORDER BY block_timestamp DESC NULLS FIRST, id DESC LIMIT $2 OFFSET $3;`, para, limit, offset)
if err != nil || rows == nil {
return nil, err
}
defer func() {
if err = rows.Close(); err != nil {
log.Error("failed to close rows", "err", err)
}
}()
for rows.Next() {
msg := &orm.CrossMsg{}
if err = rows.StructScan(msg); err != nil {
break
}
results = append(results, msg)
}
if len(results) == 0 && errors.Is(err, sql.ErrNoRows) {
} else if err != nil {
return nil, err
}
return results, nil
}

9
bridge-history-api/go.mod
View File

@@ -5,15 +5,14 @@ go 1.19
require (
github.com/ethereum/go-ethereum v1.12.0
github.com/iris-contrib/middleware/cors v0.0.0-20230531125531-980d3a09a458
github.com/jmoiron/sqlx v1.3.5
github.com/kataras/iris/v12 v12.2.0
github.com/lib/pq v1.10.9
github.com/mattn/go-colorable v0.1.13
github.com/mattn/go-isatty v0.0.19
github.com/modern-go/reflect2 v1.0.2
github.com/pressly/goose/v3 v3.7.0
github.com/stretchr/testify v1.8.3
github.com/urfave/cli/v2 v2.25.7
gorm.io/gorm v1.25.2
)
require (
@@ -54,7 +53,6 @@ require (
github.com/gballet/go-libpcsclite v0.0.0-20190607065134-2772fd86a8ff // indirect
github.com/getsentry/sentry-go v0.18.0 // indirect
github.com/go-ole/go-ole v1.2.6 // indirect
github.com/go-sql-driver/mysql v1.7.1 // indirect
github.com/go-stack/stack v1.8.1 // indirect
github.com/gobwas/httphead v0.1.0 // indirect
github.com/gobwas/pool v0.2.1 // indirect
@@ -78,6 +76,8 @@ require (
github.com/iris-contrib/go.uuid v2.0.0+incompatible // indirect
github.com/iris-contrib/schema v0.0.6 // indirect
github.com/jackpal/go-nat-pmp v1.0.2 // indirect
github.com/jinzhu/inflection v1.0.0 // indirect
github.com/jinzhu/now v1.1.5 // indirect
github.com/josharian/intern v1.0.0 // indirect
github.com/kataras/blocks v0.0.7 // indirect
github.com/kataras/golog v0.1.8 // indirect
@@ -91,10 +91,9 @@ require (
github.com/mailgun/raymond/v2 v2.0.48 // indirect
github.com/mailru/easyjson v0.7.7 // indirect
github.com/mattn/go-runewidth v0.0.14 // indirect
github.com/mattn/go-sqlite3 v1.14.16 // indirect
github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
github.com/mediocregopher/radix/v3 v3.8.1 // indirect
github.com/microcosm-cc/bluemonday v1.0.23 // indirect
github.com/microcosm-cc/bluemonday v1.0.25 // indirect
github.com/mitchellh/mapstructure v1.5.0 // indirect
github.com/mitchellh/pointerstructure v1.2.0 // indirect
github.com/mmcloughlin/addchain v0.4.0 // indirect

21
bridge-history-api/go.sum
View File

@@ -143,9 +143,6 @@ github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY=
github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
github.com/go-openapi/jsonpointer v0.19.5/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
github.com/go-openapi/swag v0.19.5/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
github.com/go-sql-driver/mysql v1.6.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
github.com/go-sql-driver/mysql v1.7.1 h1:lUIinVbN1DY0xBg0eMOzmmtGoHwWBbvnWubQUrtU8EI=
github.com/go-sql-driver/mysql v1.7.1/go.mod h1:OXbVy3sEdcQ2Doequ6Z5BW6fXNQTmx+9S1MCJN5yJMI=
github.com/go-stack/stack v1.8.1 h1:ntEHSVwIt7PNXNpgPmVfMrNhLtgjlmnZha2kOpuRiDw=
github.com/go-stack/stack v1.8.1/go.mod h1:dcoOX6HbPZSZptuspn9bctJ+N/CnF5gGygcUP3XYfe4=
github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0/go.mod h1:fyg7847qk6SyHyPtNmDHnmrv/HOrqktSC+C9fM+CJOE=
@@ -251,8 +248,10 @@ github.com/iris-contrib/schema v0.0.6/go.mod h1:iYszG0IOsuIsfzjymw1kMzTL8YQcCWlm
github.com/jackpal/go-nat-pmp v1.0.2 h1:KzKSgb7qkJvOUTqYl9/Hg/me3pWgBmERKrTGD7BdWus=
github.com/jackpal/go-nat-pmp v1.0.2/go.mod h1:QPH045xvCAeXUZOxsnwmrtiCoxIr9eob+4orBN1SBKc=
github.com/jessevdk/go-flags v0.0.0-20141203071132-1679536dcc89/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
github.com/jmoiron/sqlx v1.3.5 h1:vFFPA71p1o5gAeqtEAwLU4dnX2napprKtHr7PYIcN3g=
github.com/jmoiron/sqlx v1.3.5/go.mod h1:nRVWtLre0KfCLJvgxzCsLVMogSvQ1zNJtpYr2Ccp0mQ=
github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc=
github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ=
github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
github.com/jrick/logrotate v1.0.0/go.mod h1:LNinyqDIJnpAur+b8yyulnQw/wDuN1+BYKlTRt3OuAQ=
@@ -302,9 +301,6 @@ github.com/labstack/echo/v4 v4.2.1/go.mod h1:AA49e0DZ8kk5jTOOCKNuPR6oTnBS0dYiM4F
github.com/labstack/echo/v4 v4.5.0/go.mod h1:czIriw4a0C1dFun+ObrXp7ok03xON0N1awStJ6ArI7Y=
github.com/labstack/gommon v0.3.0/go.mod h1:MULnywXg0yavhxWKc+lOruYdAhDwPK9wf0OL7NoOu+k=
github.com/leanovate/gopter v0.2.9 h1:fQjYxZaynp97ozCzfOyOuAGOU4aU/z37zf/tOujFk7c=
github.com/lib/pq v1.2.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
github.com/mailgun/raymond/v2 v2.0.48 h1:5dmlB680ZkFG2RN/0lvTAghrSxIESeu9/2aeDqACtjw=
github.com/mailgun/raymond/v2 v2.0.48/go.mod h1:lsgvL50kgt1ylcFJYZiULi5fjPBkkhNfj4KA0W54Z18=
@@ -332,9 +328,6 @@ github.com/mattn/go-runewidth v0.0.3/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzp
github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
github.com/mattn/go-runewidth v0.0.14 h1:+xnbZSEeDbOIg5/mE6JF0w6n9duR1l3/WmbinWVwUuU=
github.com/mattn/go-runewidth v0.0.14/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
github.com/mattn/go-sqlite3 v1.14.6/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
github.com/mattn/go-sqlite3 v1.14.16 h1:yOQRA0RpS5PFz/oikGwBEqvAWhWg5ufRz4ETLjwpU1Y=
github.com/mattn/go-sqlite3 v1.14.16/go.mod h1:2eHXhiwb8IkHr+BDWZGa96P6+rkvnG63S2DGjv9HUNg=
github.com/mattn/goveralls v0.0.2/go.mod h1:8d1ZMHsd7fW6IRPKQh46F2WRpyib5/X4FOpevwGNQEw=
github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo=
github.com/matttproud/golang_protobuf_extensions v1.0.4/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
@@ -342,8 +335,8 @@ github.com/mediocregopher/radix/v3 v3.4.2/go.mod h1:8FL3F6UQRXHXIBSPUs5h0RybMF8i
github.com/mediocregopher/radix/v3 v3.8.1 h1:rOkHflVuulFKlwsLY01/M2cM2tWCjDoETcMqKbAWu1M=
github.com/mediocregopher/radix/v3 v3.8.1/go.mod h1:8FL3F6UQRXHXIBSPUs5h0RybMF8i4n7wVopoX3x7Bv8=
github.com/microcosm-cc/bluemonday v1.0.2/go.mod h1:iVP4YcDBq+n/5fb23BhYFvIMq/leAFZyRl6bYmGDlGc=
github.com/microcosm-cc/bluemonday v1.0.23 h1:SMZe2IGa0NuHvnVNAZ+6B38gsTbi5e4sViiWJyDDqFY=
github.com/microcosm-cc/bluemonday v1.0.23/go.mod h1:mN70sk7UkkF8TUr2IGBpNN0jAgStuPzlK76QuruE/z4=
github.com/microcosm-cc/bluemonday v1.0.25 h1:4NEwSfiJ+Wva0VxN5B8OwMicaJvD8r9tlJWm9rtloEg=
github.com/microcosm-cc/bluemonday v1.0.25/go.mod h1:ZIOjCQp1OrzBBPIJmfX4qDYFuhU02nx4bn030ixfHLE=
github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA/g=
github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
github.com/mitchellh/go-wordwrap v1.0.1 h1:TLuKupo69TCn6TQSyGxwI1EblZZEsQ0vMlAFQflz0v0=
@@ -720,6 +713,8 @@ gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gorm.io/gorm v1.25.2 h1:gs1o6Vsa+oVKG/a9ElL3XgyGfghFfkKA2SInQaCyMho=
gorm.io/gorm v1.25.2/go.mod h1:L4uxeKpfBml98NYqVqwAdmV1a2nBtAec/cf3fpucW/k=
honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
lukechampine.com/uint128 v1.2.0 h1:mBi/5l91vocEN8otkC5bDLhi2KdCticRiwbdB0O+rjI=

91
bridge-history-api/orm/batch.go Normal file
View File

@@ -0,0 +1,91 @@
package orm
import (
"context"
"fmt"
"time"
"github.com/ethereum/go-ethereum/log"
"gorm.io/gorm"
)
// RollupBatch is the struct for rollup_batch table
type RollupBatch struct {
db *gorm.DB `gorm:"column:-"`
ID uint64 `json:"id" gorm:"column:id"`
BatchIndex uint64 `json:"batch_index" gorm:"column:batch_index"`
BatchHash string `json:"batch_hash" gorm:"column:batch_hash"`
CommitHeight uint64 `json:"commit_height" gorm:"column:commit_height"`
StartBlockNumber uint64 `json:"start_block_number" gorm:"column:start_block_number"`
EndBlockNumber uint64 `json:"end_block_number" gorm:"column:end_block_number"`
CreatedAt *time.Time `json:"created_at" gorm:"column:created_at"`
UpdatedAt *time.Time `json:"updated_at" gorm:"column:updated_at"`
DeletedAt gorm.DeletedAt `json:"deleted_at" gorm:"column:deleted_at;default:NULL"`
}
// NewRollupBatch create an RollupBatch instance
func NewRollupBatch(db *gorm.DB) *RollupBatch {
return &RollupBatch{db: db}
}
// TableName returns the table name for the Batch model.
func (*RollupBatch) TableName() string {
return "rollup_batch"
}
// GetLatestRollupBatchProcessedHeight return latest processed height from rollup_batch table
func (r *RollupBatch) GetLatestRollupBatchProcessedHeight(ctx context.Context) (uint64, error) {
var result RollupBatch
err := r.db.WithContext(ctx).Unscoped().Select("commit_height").Order("id desc").First(&result).Error
if err != nil {
return 0, fmt.Errorf("RollupBatch.GetLatestRollupBatchProcessedHeight error: %w", err)
}
return result.CommitHeight, nil
}
// GetLatestRollupBatch return the latest rollup batch in db
func (r *RollupBatch) GetLatestRollupBatch(ctx context.Context) (*RollupBatch, error) {
var result RollupBatch
err := r.db.WithContext(ctx).Model(&RollupBatch{}).Where("batch_hash is not NULL").Order("batch_index desc").First(&result).Error
if err != nil {
if err == gorm.ErrRecordNotFound {
return nil, nil
}
return nil, fmt.Errorf("RollupBatch.GetLatestRollupBatch error: %w", err)
}
return &result, nil
}
// GetRollupBatchByIndex return the rollup batch by index
func (r *RollupBatch) GetRollupBatchByIndex(ctx context.Context, index uint64) (*RollupBatch, error) {
var result RollupBatch
err := r.db.WithContext(ctx).Model(&RollupBatch{}).Where("batch_index = ?", index).First(&result).Error
if err != nil {
return nil, fmt.Errorf("RollupBatch.GetRollupBatchByIndex error: %w", err)
}
return &result, nil
}
// InsertRollupBatch batch insert rollup batch into db and return the transaction
func (r *RollupBatch) InsertRollupBatch(ctx context.Context, batches []*RollupBatch, dbTx ...*gorm.DB) error {
if len(batches) == 0 {
return nil
}
db := r.db
if len(dbTx) > 0 && dbTx[0] != nil {
db = dbTx[0]
}
err := db.WithContext(ctx).Model(&RollupBatch{}).Create(&batches).Error
if err != nil {
batchIndexes := make([]uint64, 0, len(batches))
heights := make([]uint64, 0, len(batches))
for _, batch := range batches {
batchIndexes = append(batchIndexes, batch.BatchIndex)
heights = append(heights, batch.CommitHeight)
}
log.Error("failed to insert rollup batch", "batchIndexes", batchIndexes, "heights", heights)
return fmt.Errorf("RollupBatch.InsertRollupBatch error: %w", err)
}
return nil
}

370
bridge-history-api/orm/cross_msg.go Normal file
View File

@@ -0,0 +1,370 @@
package orm
import (
"context"
"errors"
"fmt"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"gorm.io/gorm"
)
// AssetType can be ETH/ERC20/ERC1155/ERC721
type AssetType int
// MsgType can be layer1/layer2 msg
type MsgType int
func (a AssetType) String() string {
switch a {
case ETH:
return "ETH"
case ERC20:
return "ERC20"
case ERC1155:
return "ERC1155"
case ERC721:
return "ERC721"
}
return "Unknown Asset Type"
}
const (
// ETH = 0
ETH AssetType = iota
// ERC20 = 1
ERC20
// ERC721 = 2
ERC721
// ERC1155 = 3
ERC1155
)
const (
// UnknownMsg = 0
UnknownMsg MsgType = iota
// Layer1Msg = 1
Layer1Msg
// Layer2Msg = 2
Layer2Msg
)
// CrossMsg represents a cross message from layer 1 to layer 2
type CrossMsg struct {
db *gorm.DB `gorm:"column:-"`
ID uint64 `json:"id" gorm:"column:id"`
MsgHash string `json:"msg_hash" gorm:"column:msg_hash"`
Height uint64 `json:"height" gorm:"column:height"`
Sender string `json:"sender" gorm:"column:sender"`
Target string `json:"target" gorm:"column:target"`
Amount string `json:"amount" gorm:"column:amount"`
Layer1Hash string `json:"layer1_hash" gorm:"column:layer1_hash;default:''"`
Layer2Hash string `json:"layer2_hash" gorm:"column:layer2_hash;default:''"`
Layer1Token string `json:"layer1_token" gorm:"column:layer1_token;default:''"`
Layer2Token string `json:"layer2_token" gorm:"column:layer2_token;default:''"`
TokenIDs string `json:"token_ids" gorm:"column:token_ids;default:''"`
TokenAmounts string `json:"token_amounts" gorm:"column:token_amounts;default:''"`
Asset int `json:"asset" gorm:"column:asset"`
MsgType int `json:"msg_type" gorm:"column:msg_type"`
Timestamp *time.Time `json:"timestamp" gorm:"column:block_timestamp;default;NULL"`
CreatedAt *time.Time `json:"created_at" gorm:"column:created_at"`
UpdatedAt *time.Time `json:"updated_at" gorm:"column:updated_at"`
DeletedAt gorm.DeletedAt `json:"deleted_at" gorm:"column:deleted_at;default:NULL"`
}
// TableName returns the table name for the CrossMsg model.
func (*CrossMsg) TableName() string {
return "cross_message"
}
// NewCrossMsg returns a new instance of CrossMsg.
func NewCrossMsg(db *gorm.DB) *CrossMsg {
return &CrossMsg{db: db}
}
// L1 Cross Msgs Operations
// GetL1CrossMsgByHash returns layer1 cross message by given hash
func (c *CrossMsg) GetL1CrossMsgByHash(ctx context.Context, l1Hash common.Hash) (*CrossMsg, error) {
var result CrossMsg
err := c.db.WithContext(ctx).Model(&CrossMsg{}).Where("layer1_hash = ? AND msg_type = ?", l1Hash.String(), Layer1Msg).First(&result).Error
if err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, nil
}
return nil, fmt.Errorf("CrossMsg.GetL1CrossMsgByHash error: %w", err)
}
return &result, nil
}
// GetLatestL1ProcessedHeight returns the latest processed height of layer1 cross messages
func (c *CrossMsg) GetLatestL1ProcessedHeight(ctx context.Context) (uint64, error) {
var result CrossMsg
err := c.db.WithContext(ctx).Model(&CrossMsg{}).Where("msg_type = ?", Layer1Msg).
Select("height").
Order("id DESC").
First(&result).
Error
if err != nil {
if err == gorm.ErrRecordNotFound {
return 0, nil
}
return 0, fmt.Errorf("CrossMsg.GetLatestL1ProcessedHeight error: %w", err)
}
return result.Height, nil
}
// GetL1EarliestNoBlockTimestampHeight returns the earliest layer1 cross message height which has no block timestamp
func (c *CrossMsg) GetL1EarliestNoBlockTimestampHeight(ctx context.Context) (uint64, error) {
var result CrossMsg
err := c.db.WithContext(ctx).Model(&CrossMsg{}).
Where("block_timestamp IS NULL AND msg_type = ?", Layer1Msg).
Select("height").
Order("height ASC").
First(&result).
Error
if err != nil {
if err == gorm.ErrRecordNotFound {
return 0, nil
}
return 0, fmt.Errorf("CrossMsg.GetL1EarliestNoBlockTimestampHeight error: %w", err)
}
return result.Height, nil
}
// InsertL1CrossMsg batch insert layer1 cross messages into db
func (c *CrossMsg) InsertL1CrossMsg(ctx context.Context, messages []*CrossMsg, dbTx ...*gorm.DB) error {
if len(messages) == 0 {
return nil
}
db := c.db
if len(dbTx) > 0 && dbTx[0] != nil {
db = dbTx[0]
}
db.WithContext(ctx)
err := db.Model(&CrossMsg{}).Create(&messages).Error
if err != nil {
l1hashes := make([]string, 0, len(messages))
heights := make([]uint64, 0, len(messages))
for _, msg := range messages {
l1hashes = append(l1hashes, msg.Layer1Hash)
heights = append(heights, msg.Height)
}
log.Error("failed to insert l1 cross messages", "l1hashes", l1hashes, "heights", heights, "err", err)
return fmt.Errorf("CrossMsg.InsertL1CrossMsg error: %w", err)
}
return nil
}
// UpdateL1CrossMsgHash update l1 cross msg hash in db, no need to check msg_type since layer1_hash wont be empty if its layer1 msg
func (c *CrossMsg) UpdateL1CrossMsgHash(ctx context.Context, l1Hash, msgHash common.Hash, dbTx ...*gorm.DB) error {
db := c.db
if len(dbTx) > 0 && dbTx[0] != nil {
db = dbTx[0]
}
db.WithContext(ctx)
err := c.db.Model(&CrossMsg{}).Where("layer1_hash = ?", l1Hash.Hex()).Update("msg_hash", msgHash.Hex()).Error
if err != nil {
return fmt.Errorf("CrossMsg.UpdateL1CrossMsgHash error: %w", err)
}
return nil
}
// UpdateL1BlockTimestamp update layer1 block timestamp
func (c *CrossMsg) UpdateL1BlockTimestamp(ctx context.Context, height uint64, timestamp time.Time) error {
err := c.db.WithContext(ctx).Model(&CrossMsg{}).
Where("height = ? AND msg_type = ?", height, Layer1Msg).
Update("block_timestamp", timestamp).Error
if err != nil {
return fmt.Errorf("CrossMsg.UpdateL1BlockTimestamp error: %w", err)
}
return err
}
// DeleteL1CrossMsgAfterHeight soft delete layer1 cross messages after given height
func (c *CrossMsg) DeleteL1CrossMsgAfterHeight(ctx context.Context, height uint64, dbTx ...*gorm.DB) error {
db := c.db
if len(dbTx) > 0 && dbTx[0] != nil {
db = dbTx[0]
}
db.WithContext(ctx)
err := db.Delete(&CrossMsg{}, "height > ? AND msg_type = ?", height, Layer1Msg).Error
if err != nil {
return fmt.Errorf("CrossMsg.DeleteL1CrossMsgAfterHeight error: %w", err)
}
return nil
}
// L2 Cross Msgs Operations
// GetL2CrossMsgByHash returns layer2 cross message by given hash
func (c *CrossMsg) GetL2CrossMsgByHash(ctx context.Context, l2Hash common.Hash) (*CrossMsg, error) {
var result CrossMsg
err := c.db.WithContext(ctx).Model(&CrossMsg{}).Where("layer2_hash = ? AND msg_type = ?", l2Hash.String(), Layer1Msg).First(&result).Error
if err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, nil
}
return nil, fmt.Errorf("CrossMsg.GetL2CrossMsgByHash error: %w", err)
}
return &result, nil
}
// GetLatestL2ProcessedHeight returns the latest processed height of layer2 cross messages
func (c *CrossMsg) GetLatestL2ProcessedHeight(ctx context.Context) (uint64, error) {
var result CrossMsg
err := c.db.WithContext(ctx).Model(&CrossMsg{}).
Select("height").
Where("msg_type = ?", Layer2Msg).
Order("id DESC").
First(&result).
Error
if err != nil {
if err == gorm.ErrRecordNotFound {
return 0, nil
}
return 0, fmt.Errorf("CrossMsg.GetLatestL2ProcessedHeight error: %w", err)
}
return result.Height, nil
}
// GetL2CrossMsgByMsgHashList returns layer2 cross messages under given msg hashes
func (c *CrossMsg) GetL2CrossMsgByMsgHashList(ctx context.Context, msgHashList []string) ([]*CrossMsg, error) {
var results []*CrossMsg
err := c.db.WithContext(ctx).Model(&CrossMsg{}).
Where("msg_hash IN (?) AND msg_type = ?", msgHashList, Layer2Msg).
Find(&results).
Error
if err != nil {
return nil, fmt.Errorf("CrossMsg.GetL2CrossMsgByMsgHashList error: %w", err)
}
if len(results) == 0 {
log.Debug("no CrossMsg under given msg hashes", "msg hash list", msgHashList)
}
return results, nil
}
// GetL2EarliestNoBlockTimestampHeight returns the earliest layer2 cross message height which has no block timestamp
func (c *CrossMsg) GetL2EarliestNoBlockTimestampHeight(ctx context.Context) (uint64, error) {
var result CrossMsg
err := c.db.WithContext(ctx).Model(&CrossMsg{}).
Where("block_timestamp IS NULL AND msg_type = ?", Layer2Msg).
Select("height").
Order("height ASC").
First(&result).
Error
if err != nil {
if err == gorm.ErrRecordNotFound {
return 0, nil
}
return 0, fmt.Errorf("CrossMsg.GetL2EarliestNoBlockTimestampHeight error: %w", err)
}
return result.Height, nil
}
// InsertL2CrossMsg batch insert layer2 cross messages
func (c *CrossMsg) InsertL2CrossMsg(ctx context.Context, messages []*CrossMsg, dbTx ...*gorm.DB) error {
if len(messages) == 0 {
return nil
}
db := c.db
if len(dbTx) > 0 && dbTx[0] != nil {
db = dbTx[0]
}
db.WithContext(ctx)
err := db.Model(&CrossMsg{}).Create(&messages).Error
if err != nil {
l2hashes := make([]string, 0, len(messages))
heights := make([]uint64, 0, len(messages))
for _, msg := range messages {
l2hashes = append(l2hashes, msg.Layer2Hash)
heights = append(heights, msg.Height)
}
log.Error("failed to insert l2 cross messages", "l2hashes", l2hashes, "heights", heights, "err", err)
return fmt.Errorf("CrossMsg.InsertL2CrossMsg error: %w", err)
}
return nil
}
// UpdateL2CrossMsgHash update layer2 cross message hash
func (c *CrossMsg) UpdateL2CrossMsgHash(ctx context.Context, l2Hash, msgHash common.Hash, dbTx ...*gorm.DB) error {
db := c.db
if len(dbTx) > 0 && dbTx[0] != nil {
db = dbTx[0]
}
db.WithContext(ctx)
err := db.Model(&CrossMsg{}).
Where("layer2_hash = ?", l2Hash.String()).
Update("msg_hash", msgHash.String()).
Error
if err != nil {
return fmt.Errorf("CrossMsg.UpdateL2CrossMsgHash error: %w", err)
}
return nil
}
// UpdateL2BlockTimestamp update layer2 cross message block timestamp
func (c *CrossMsg) UpdateL2BlockTimestamp(ctx context.Context, height uint64, timestamp time.Time) error {
err := c.db.WithContext(ctx).Model(&CrossMsg{}).
Where("height = ? AND msg_type = ?", height, Layer2Msg).
Update("block_timestamp", timestamp).Error
if err != nil {
return fmt.Errorf("CrossMsg.UpdateL2BlockTimestamp error: %w", err)
}
return nil
}
// DeleteL2CrossMsgFromHeight delete layer2 cross messages from given height
func (c *CrossMsg) DeleteL2CrossMsgFromHeight(ctx context.Context, height uint64, dbTx ...*gorm.DB) error {
db := c.db
if len(dbTx) > 0 && dbTx[0] != nil {
db = dbTx[0]
}
db.WithContext(ctx)
err := db.Model(&CrossMsg{}).Delete("height > ? AND msg_type = ?", height, Layer2Msg).Error
if err != nil {
return fmt.Errorf("CrossMsg.DeleteL2CrossMsgFromHeight error: %w", err)
}
return nil
}
// General Operations
// GetTotalCrossMsgCountByAddress get total cross msg count by address
func (c *CrossMsg) GetTotalCrossMsgCountByAddress(ctx context.Context, sender string) (uint64, error) {
var count int64
err := c.db.WithContext(ctx).Model(&CrossMsg{}).
Where("sender = ?", sender).
Count(&count).
Error
if err != nil {
if err == gorm.ErrRecordNotFound {
return 0, nil
}
return 0, fmt.Errorf("CrossMsg.GetTotalCrossMsgCountByAddress error: %w", err)
}
return uint64(count), nil
}
// GetCrossMsgsByAddressWithOffset get cross msgs by address with offset
func (c *CrossMsg) GetCrossMsgsByAddressWithOffset(ctx context.Context, sender string, offset int, limit int) ([]CrossMsg, error) {
var messages []CrossMsg
err := c.db.WithContext(ctx).Model(&CrossMsg{}).
Where("sender = ?", sender).
Order("block_timestamp DESC NULLS FIRST, id DESC").
Limit(limit).
Offset(offset).
Find(&messages).
Error
if err != nil {
return nil, fmt.Errorf("CrossMsg.GetCrossMsgsByAddressWithOffset error: %w", err)
}
return messages, nil
}

215
bridge-history-api/orm/l2_sent_msg.go Normal file
View File

@@ -0,0 +1,215 @@
package orm
import (
"context"
"fmt"
"time"
"github.com/ethereum/go-ethereum/log"
"gorm.io/gorm"
)
// L2SentMsg defines the struct for l2_sent_msg table record
type L2SentMsg struct {
db *gorm.DB `gorm:"column:-"`
ID uint64 `json:"id" gorm:"column:id"`
OriginalSender string `json:"original_sender" gorm:"column:original_sender;default:''"`
TxHash string `json:"tx_hash" gorm:"column:tx_hash"`
MsgHash string `json:"msg_hash" gorm:"column:msg_hash"`
Sender string `json:"sender" gorm:"column:sender"`
Target string `json:"target" gorm:"column:target"`
Value string `json:"value" gorm:"column:value"`
Height uint64 `json:"height" gorm:"column:height"`
Nonce uint64 `json:"nonce" gorm:"column:nonce"`
BatchIndex uint64 `json:"batch_index" gorm:"column:batch_index;default:0"`
MsgProof string `json:"msg_proof" gorm:"column:msg_proof;default:''"`
MsgData string `json:"msg_data" gorm:"column:msg_data;default:''"`
CreatedAt *time.Time `json:"created_at" gorm:"column:created_at"`
UpdatedAt *time.Time `json:"updated_at" gorm:"column:updated_at"`
DeletedAt gorm.DeletedAt `json:"deleted_at" gorm:"column:deleted_at;default:NULL"`
}
// NewL2SentMsg create an NewL2SentMsg instance
func NewL2SentMsg(db *gorm.DB) *L2SentMsg {
return &L2SentMsg{db: db}
}
// TableName returns the table name for the L2SentMsg model.
func (*L2SentMsg) TableName() string {
return "l2_sent_msg"
}
// GetL2SentMsgByHash get l2 sent msg by hash
func (l *L2SentMsg) GetL2SentMsgByHash(ctx context.Context, msgHash string) (*L2SentMsg, error) {
var result L2SentMsg
err := l.db.WithContext(ctx).Model(&L2SentMsg{}).
Where("msg_hash = ?", msgHash).
First(&result).
Error
if err != nil {
return nil, fmt.Errorf("L2SentMsg.GetL2SentMsgByHash error: %w", err)
}
return &result, nil
}
// GetLatestSentMsgHeightOnL2 get latest sent msg height on l2
func (l *L2SentMsg) GetLatestSentMsgHeightOnL2(ctx context.Context) (uint64, error) {
var result L2SentMsg
err := l.db.WithContext(ctx).Model(&L2SentMsg{}).
Select("height").
Order("nonce DESC").
First(&result).Error
if err != nil {
if err == gorm.ErrRecordNotFound {
return 0, nil
}
return 0, fmt.Errorf("L2SentMsg.GetLatestSentMsgHeightOnL2 error: %w", err)
}
return result.Height, nil
}
// GetClaimableL2SentMsgByAddressWithOffset get claimable l2 sent msg by address with offset
func (l *L2SentMsg) GetClaimableL2SentMsgByAddressWithOffset(ctx context.Context, address string, offset int, limit int) ([]*L2SentMsg, error) {
var results []*L2SentMsg
err := l.db.WithContext(ctx).Raw(`SELECT * FROM l2_sent_msg WHERE id NOT IN (SELECT l2_sent_msg.id FROM l2_sent_msg INNER JOIN relayed_msg ON l2_sent_msg.msg_hash = relayed_msg.msg_hash WHERE l2_sent_msg.deleted_at IS NULL AND relayed_msg.deleted_at IS NULL) AND (original_sender=$1 OR sender = $1) AND msg_proof !='' ORDER BY id DESC LIMIT $2 OFFSET $3;`, address, limit, offset).
Scan(&results).Error
if err != nil {
return nil, fmt.Errorf("L2SentMsg.GetClaimableL2SentMsgByAddressWithOffset error: %w", err)
}
return results, nil
}
// GetClaimableL2SentMsgByAddressTotalNum get claimable l2 sent msg by address total num
func (l *L2SentMsg) GetClaimableL2SentMsgByAddressTotalNum(ctx context.Context, address string) (uint64, error) {
var count uint64
err := l.db.WithContext(ctx).Raw(`SELECT COUNT(*) FROM l2_sent_msg WHERE id NOT IN (SELECT l2_sent_msg.id FROM l2_sent_msg INNER JOIN relayed_msg ON l2_sent_msg.msg_hash = relayed_msg.msg_hash WHERE l2_sent_msg.deleted_at IS NULL AND relayed_msg.deleted_at IS NULL) AND (original_sender=$1 OR sender = $1) AND msg_proof !='';`, address).
Scan(&count).Error
if err != nil {
return 0, fmt.Errorf("L2SentMsg.GetClaimableL2SentMsgByAddressTotalNum error: %w", err)
}
return count, nil
}
// GetLatestL2SentMsgBatchIndex get latest l2 sent msg batch index
func (l *L2SentMsg) GetLatestL2SentMsgBatchIndex(ctx context.Context) (int64, error) {
var result L2SentMsg
err := l.db.WithContext(ctx).Model(&L2SentMsg{}).
Where("batch_index != 0").
Order("batch_index DESC").
Select("batch_index").
First(&result).
Error
if err != nil {
return -1, fmt.Errorf("L2SentMsg.GetLatestL2SentMsgBatchIndex error: %w", err)
}
// Watch for overflow, tho its not likely to happen
return int64(result.Height), nil
}
// GetL2SentMsgMsgHashByHeightRange get l2 sent msg msg hash by height range
func (l *L2SentMsg) GetL2SentMsgMsgHashByHeightRange(ctx context.Context, startHeight, endHeight uint64) ([]*L2SentMsg, error) {
var results []*L2SentMsg
err := l.db.WithContext(ctx).Model(&L2SentMsg{}).
Where("height >= ? AND height <= ?", startHeight, endHeight).
Order("nonce ASC").
Find(&results).
Error
if err != nil {
return nil, fmt.Errorf("L2SentMsg.GetL2SentMsgMsgHashByHeightRange error: %w", err)
}
return results, nil
}
// GetL2SentMessageByNonce get l2 sent message by nonce
func (l *L2SentMsg) GetL2SentMessageByNonce(ctx context.Context, nonce uint64) (*L2SentMsg, error) {
var result L2SentMsg
err := l.db.WithContext(ctx).Model(&L2SentMsg{}).
Where("nonce = ?", nonce).
First(&result).
Error
if err != nil {
if err == gorm.ErrRecordNotFound {
return nil, nil
}
return nil, fmt.Errorf("L2SentMsg.GetL2SentMessageByNonce error: %w", err)
}
return &result, nil
}
// GetLatestL2SentMsgLEHeight get latest l2 sent msg less than or equal to end block number
func (l *L2SentMsg) GetLatestL2SentMsgLEHeight(ctx context.Context, endBlockNumber uint64) (*L2SentMsg, error) {
var result L2SentMsg
err := l.db.WithContext(ctx).Model(&L2SentMsg{}).
Where("height <= ?", endBlockNumber).
Order("nonce DESC").
First(&result).
Error
if err != nil {
if err == gorm.ErrRecordNotFound {
return nil, nil
}
return nil, fmt.Errorf("L2SentMsg.GetLatestL2SentMsgLEHeight error: %w", err)
}
return &result, nil
}
// InsertL2SentMsg batch insert l2 sent msg
func (l *L2SentMsg) InsertL2SentMsg(ctx context.Context, messages []*L2SentMsg, dbTx ...*gorm.DB) error {
if len(messages) == 0 {
return nil
}
db := l.db
if len(dbTx) > 0 && dbTx[0] != nil {
db = dbTx[0]
}
db.WithContext(ctx)
err := db.Model(&L2SentMsg{}).Create(&messages).Error
if err != nil {
l2hashes := make([]string, 0, len(messages))
heights := make([]uint64, 0, len(messages))
for _, msg := range messages {
l2hashes = append(l2hashes, msg.TxHash)
heights = append(heights, msg.Height)
}
log.Error("failed to insert l2 sent messages", "l2hashes", l2hashes, "heights", heights, "err", err)
return fmt.Errorf("L2SentMsg.InsertL2SentMsg error: %w", err)
}
return nil
}
// UpdateL2MessageProof update l2 message proof in db tx
func (l *L2SentMsg) UpdateL2MessageProof(ctx context.Context, msgHash string, proof string, batchIndex uint64, dbTx ...*gorm.DB) error {
db := l.db
if len(dbTx) > 0 && dbTx[0] != nil {
db = dbTx[0]
}
db.WithContext(ctx)
err := db.Model(&L2SentMsg{}).
Where("msg_hash = ?", msgHash).
Updates(map[string]interface{}{
"msg_proof": proof,
"batch_index": batchIndex,
}).Error
if err != nil {
return fmt.Errorf("L2SentMsg.UpdateL2MessageProof error: %w", err)
}
return nil
}
// DeleteL2SentMsgAfterHeight delete l2 sent msg after height
func (l *L2SentMsg) DeleteL2SentMsgAfterHeight(ctx context.Context, height uint64, dbTx ...*gorm.DB) error {
db := l.db
if len(dbTx) > 0 && dbTx[0] != nil {
db = dbTx[0]
}
err := db.WithContext(ctx).Model(&L2SentMsg{}).Delete("height > ?", height).Error
if err != nil {
return fmt.Errorf("L2SentMsg.DeleteL2SentMsgAfterHeight error: %w", err)
}
return nil
}

0
bridge-history-api/db/migrate/migrate.go → bridge-history-api/orm/migrate/migrate.go
View File

0
bridge-history-api/db/migrate/migrations/00001_cross_msg.sql → bridge-history-api/orm/migrate/migrations/00001_cross_msg.sql
View File

0
bridge-history-api/db/migrate/migrations/00002_relayed_msg.sql → bridge-history-api/orm/migrate/migrations/00002_relayed_msg.sql
View File

0
bridge-history-api/db/migrate/migrations/00003_l2_sent_msg.sql → bridge-history-api/orm/migrate/migrations/00003_l2_sent_msg.sql
View File

0
bridge-history-api/db/migrate/migrations/00004_rollup_batch.sql → bridge-history-api/orm/migrate/migrations/00004_rollup_batch.sql
View File

142
bridge-history-api/orm/relayed_msg.go Normal file
View File

@@ -0,0 +1,142 @@
package orm
import (
"context"
"fmt"
"time"
"github.com/ethereum/go-ethereum/log"
"gorm.io/gorm"
)
// RelayedMsg is the struct for relayed_msg table
type RelayedMsg struct {
db *gorm.DB `gorm:"column:-"`
ID uint64 `json:"id" gorm:"column:id"`
MsgHash string `json:"msg_hash" gorm:"column:msg_hash"`
Height uint64 `json:"height" gorm:"column:height"`
Layer1Hash string `json:"layer1_hash" gorm:"column:layer1_hash;default:''"`
Layer2Hash string `json:"layer2_hash" gorm:"column:layer2_hash;default:''"`
CreatedAt *time.Time `json:"created_at" gorm:"column:created_at"`
UpdatedAt *time.Time `json:"updated_at" gorm:"column:updated_at"`
DeletedAt gorm.DeletedAt `json:"deleted_at" gorm:"column:deleted_at;default:NULL"`
}
// NewRelayedMsg create an NewRelayedMsg instance
func NewRelayedMsg(db *gorm.DB) *RelayedMsg {
return &RelayedMsg{db: db}
}
// TableName returns the table name for the RelayedMsg model.
func (*RelayedMsg) TableName() string {
return "relayed_msg"
}
// GetRelayedMsgByHash get relayed msg by hash
func (r *RelayedMsg) GetRelayedMsgByHash(ctx context.Context, msgHash string) (*RelayedMsg, error) {
var result RelayedMsg
err := r.db.WithContext(ctx).Model(&RelayedMsg{}).
Where("msg_hash = ?", msgHash).
First(&result).
Error
if err != nil {
if err == gorm.ErrRecordNotFound {
return nil, nil
}
return nil, fmt.Errorf("RelayedMsg.GetRelayedMsgByHash error: %w", err)
}
return &result, nil
}
// GetLatestRelayedHeightOnL1 get latest relayed height on l1
func (r *RelayedMsg) GetLatestRelayedHeightOnL1(ctx context.Context) (uint64, error) {
var result RelayedMsg
err := r.db.WithContext(ctx).Model(&RelayedMsg{}).
Select("height").
Where("layer1_hash != ''").
Order("height DESC").
First(&result).
Error
if err != nil {
if err == gorm.ErrRecordNotFound {
return 0, nil
}
return 0, fmt.Errorf("RelayedMsg.GetLatestRelayedHeightOnL1 error: %w", err)
}
return result.Height, err
}
// GetLatestRelayedHeightOnL2 get latest relayed height on l2
func (r *RelayedMsg) GetLatestRelayedHeightOnL2(ctx context.Context) (uint64, error) {
var result RelayedMsg
err := r.db.WithContext(ctx).Model(&RelayedMsg{}).
Select("height").
Where("layer2_hash != ''").
Order("height DESC").
First(&result).
Error
if err != nil {
if err == gorm.ErrRecordNotFound {
return 0, nil
}
return 0, fmt.Errorf("RelayedMsg.GetLatestRelayedHeightOnL2 error: %w", err)
}
return result.Height, nil
}
// InsertRelayedMsg batch insert relayed msg into db and return the transaction
func (r *RelayedMsg) InsertRelayedMsg(ctx context.Context, messages []*RelayedMsg, dbTx ...*gorm.DB) error {
if len(messages) == 0 {
return nil
}
db := r.db
if len(dbTx) > 0 && dbTx[0] != nil {
db = dbTx[0]
}
db.WithContext(ctx)
err := db.Model(&RelayedMsg{}).Create(&messages).Error
if err != nil {
l2hashes := make([]string, 0, len(messages))
l1hashes := make([]string, 0, len(messages))
heights := make([]uint64, 0, len(messages))
for _, msg := range messages {
l2hashes = append(l2hashes, msg.Layer2Hash)
l1hashes = append(l1hashes, msg.Layer1Hash)
heights = append(heights, msg.Height)
}
log.Error("failed to insert l2 sent messages", "l2hashes", l2hashes, "l1hashes", l1hashes, "heights", heights, "err", err)
return fmt.Errorf("RelayedMsg.InsertRelayedMsg error: %w", err)
}
return nil
}
// DeleteL1RelayedHashAfterHeight delete l1 relayed hash after height
func (r *RelayedMsg) DeleteL1RelayedHashAfterHeight(ctx context.Context, height uint64, dbTx ...*gorm.DB) error {
db := r.db
if len(dbTx) > 0 && dbTx[0] != nil {
db = dbTx[0]
}
db.WithContext(ctx)
err := db.Model(&RelayedMsg{}).
Delete("height > ? AND layer1_hash != ''", height).Error
if err != nil {
return fmt.Errorf("RelayedMsg.DeleteL1RelayedHashAfterHeight error: %w", err)
}
return nil
}
// DeleteL2RelayedHashAfterHeight delete l2 relayed hash after heights
func (r *RelayedMsg) DeleteL2RelayedHashAfterHeight(ctx context.Context, height uint64, dbTx ...*gorm.DB) error {
db := r.db
if len(dbTx) > 0 && dbTx[0] != nil {
db = dbTx[0]
}
db.WithContext(ctx)
err := db.Model(&RelayedMsg{}).
Delete("height > ? AND layer2_hash != ''", height).Error
if err != nil {
return fmt.Errorf("RelayedMsg.DeleteL2RelayedHashAfterHeight error: %w", err)
}
return nil
}

67
bridge-history-api/service/service.go
View File

@@ -1,14 +1,15 @@
package service
import (
"context"
"strconv"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"gorm.io/gorm"
"bridge-history-api/db"
"bridge-history-api/db/orm"
"bridge-history-api/orm"
)
// Finalized the schema of tx finalized infos
@@ -48,30 +49,33 @@ type TxHistoryInfo struct {
// HistoryService example service.
type HistoryService interface {
GetTxsByAddress(address common.Address, offset int64, limit int64) ([]*TxHistoryInfo, uint64, error)
GetTxsByAddress(address common.Address, offset int, limit int) ([]*TxHistoryInfo, uint64, error)
GetTxsByHashes(hashes []string) ([]*TxHistoryInfo, error)
GetClaimableTxsByAddress(address common.Address, offset int64, limit int64) ([]*TxHistoryInfo, uint64, error)
GetClaimableTxsByAddress(address common.Address, offset int, limit int) ([]*TxHistoryInfo, uint64, error)
}
// NewHistoryService returns a service backed with a "db"
func NewHistoryService(db db.OrmFactory) HistoryService {
service := &historyBackend{db: db, prefix: "Scroll-Bridge-History-Server"}
func NewHistoryService(ctx context.Context, db *gorm.DB) HistoryService {
service := &historyBackend{ctx: ctx, db: db, prefix: "Scroll-Bridge-History-Server"}
return service
}
type historyBackend struct {
prefix string
db db.OrmFactory
ctx context.Context
db *gorm.DB
}
// GetCrossTxClaimInfo get UserClaimInfos by address
func GetCrossTxClaimInfo(msgHash string, db db.OrmFactory) *UserClaimInfo {
l2sentMsg, err := db.GetL2SentMsgByHash(msgHash)
if err != nil {
func GetCrossTxClaimInfo(ctx context.Context, msgHash string, db *gorm.DB) *UserClaimInfo {
l2SentMsgOrm := orm.NewL2SentMsg(db)
rollupOrm := orm.NewRollupBatch(db)
l2sentMsg, err := l2SentMsgOrm.GetL2SentMsgByHash(ctx, msgHash)
if err != nil || l2sentMsg == nil {
log.Debug("GetCrossTxClaimInfo failed", "error", err)
return &UserClaimInfo{}
}
batch, err := db.GetRollupBatchByIndex(l2sentMsg.BatchIndex)
batch, err := rollupOrm.GetRollupBatchByIndex(ctx, l2sentMsg.BatchIndex)
if err != nil {
log.Debug("GetCrossTxClaimInfo failed", "error", err)
return &UserClaimInfo{}
@@ -89,10 +93,11 @@ func GetCrossTxClaimInfo(msgHash string, db db.OrmFactory) *UserClaimInfo {
}
func updateCrossTxHash(msgHash string, txInfo *TxHistoryInfo, db db.OrmFactory) {
relayed, err := db.GetRelayedMsgByHash(msgHash)
func updateCrossTxHash(ctx context.Context, msgHash string, txInfo *TxHistoryInfo, db *gorm.DB) {
relayed := orm.NewRelayedMsg(db)
relayed, err := relayed.GetRelayedMsgByHash(ctx, msgHash)
if err != nil {
log.Error("updateCrossTxHash failed", "error", err)
log.Debug("updateCrossTxHash failed", "error", err)
return
}
if relayed == nil {
@@ -112,13 +117,15 @@ func updateCrossTxHash(msgHash string, txInfo *TxHistoryInfo, db db.OrmFactory)
}
// GetClaimableTxsByAddress get all claimable txs under given address
func (h *historyBackend) GetClaimableTxsByAddress(address common.Address, offset int64, limit int64) ([]*TxHistoryInfo, uint64, error) {
func (h *historyBackend) GetClaimableTxsByAddress(address common.Address, offset int, limit int) ([]*TxHistoryInfo, uint64, error) {
var txHistories []*TxHistoryInfo
total, err := h.db.GetClaimableL2SentMsgByAddressTotalNum(address.Hex())
l2SentMsgOrm := orm.NewL2SentMsg(h.db)
l2CrossMsgOrm := orm.NewCrossMsg(h.db)
total, err := l2SentMsgOrm.GetClaimableL2SentMsgByAddressTotalNum(h.ctx, address.Hex())
if err != nil || total == 0 {
return txHistories, 0, err
}
results, err := h.db.GetClaimableL2SentMsgByAddressWithOffset(address.Hex(), offset, limit)
results, err := l2SentMsgOrm.GetClaimableL2SentMsgByAddressWithOffset(h.ctx, address.Hex(), offset, limit)
if err != nil || len(results) == 0 {
return txHistories, 0, err
}
@@ -126,7 +133,7 @@ func (h *historyBackend) GetClaimableTxsByAddress(address common.Address, offset
for _, result := range results {
msgHashList = append(msgHashList, result.MsgHash)
}
crossMsgs, err := h.db.GetL2CrossMsgByMsgHashList(msgHashList)
crossMsgs, err := l2CrossMsgOrm.GetL2CrossMsgByMsgHashList(h.ctx, msgHashList)
// crossMsgs can be empty, because they can be emitted by user directly call contract
if err != nil {
return txHistories, 0, err
@@ -141,7 +148,7 @@ func (h *historyBackend) GetClaimableTxsByAddress(address common.Address, offset
IsL1: false,
BlockNumber: result.Height,
FinalizeTx: &Finalized{},
ClaimInfo: GetCrossTxClaimInfo(result.MsgHash, h.db),
ClaimInfo: GetCrossTxClaimInfo(h.ctx, result.MsgHash, h.db),
}
if crossMsg, exist := crossMsgMap[result.MsgHash]; exist {
txInfo.Amount = crossMsg.Amount
@@ -155,13 +162,14 @@ func (h *historyBackend) GetClaimableTxsByAddress(address common.Address, offset
}
// GetTxsByAddress get all txs under given address
func (h *historyBackend) GetTxsByAddress(address common.Address, offset int64, limit int64) ([]*TxHistoryInfo, uint64, error) {
func (h *historyBackend) GetTxsByAddress(address common.Address, offset int, limit int) ([]*TxHistoryInfo, uint64, error) {
var txHistories []*TxHistoryInfo
total, err := h.db.GetTotalCrossMsgCountByAddress(address.String())
utilOrm := orm.NewCrossMsg(h.db)
total, err := utilOrm.GetTotalCrossMsgCountByAddress(h.ctx, address.String())
if err != nil || total == 0 {
return txHistories, 0, err
}
result, err := h.db.GetCrossMsgsByAddressWithOffset(address.String(), offset, limit)
result, err := utilOrm.GetCrossMsgsByAddressWithOffset(h.ctx, address.String(), offset, limit)
if err != nil {
return nil, 0, err
@@ -178,9 +186,9 @@ func (h *historyBackend) GetTxsByAddress(address common.Address, offset int64, l
FinalizeTx: &Finalized{
Hash: "",
},
ClaimInfo: GetCrossTxClaimInfo(msg.MsgHash, h.db),
ClaimInfo: GetCrossTxClaimInfo(h.ctx, msg.MsgHash, h.db),
}
updateCrossTxHash(msg.MsgHash, txHistory, h.db)
updateCrossTxHash(h.ctx, msg.MsgHash, txHistory, h.db)
txHistories = append(txHistories, txHistory)
}
return txHistories, total, nil
@@ -189,8 +197,9 @@ func (h *historyBackend) GetTxsByAddress(address common.Address, offset int64, l
// GetTxsByHashes get tx infos under given tx hashes
func (h *historyBackend) GetTxsByHashes(hashes []string) ([]*TxHistoryInfo, error) {
txHistories := make([]*TxHistoryInfo, 0)
CrossMsgOrm := orm.NewCrossMsg(h.db)
for _, hash := range hashes {
l1result, err := h.db.GetL1CrossMsgByHash(common.HexToHash(hash))
l1result, err := CrossMsgOrm.GetL1CrossMsgByHash(h.ctx, common.HexToHash(hash))
if err != nil {
return nil, err
}
@@ -207,11 +216,11 @@ func (h *historyBackend) GetTxsByHashes(hashes []string) ([]*TxHistoryInfo, erro
Hash: "",
},
}
updateCrossTxHash(l1result.MsgHash, txHistory, h.db)
updateCrossTxHash(h.ctx, l1result.MsgHash, txHistory, h.db)
txHistories = append(txHistories, txHistory)
continue
}
l2result, err := h.db.GetL2CrossMsgByHash(common.HexToHash(hash))
l2result, err := CrossMsgOrm.GetL2CrossMsgByHash(h.ctx, common.HexToHash(hash))
if err != nil {
return nil, err
}
@@ -227,9 +236,9 @@ func (h *historyBackend) GetTxsByHashes(hashes []string) ([]*TxHistoryInfo, erro
FinalizeTx: &Finalized{
Hash: "",
},
ClaimInfo: GetCrossTxClaimInfo(l2result.MsgHash, h.db),
ClaimInfo: GetCrossTxClaimInfo(h.ctx, l2result.MsgHash, h.db),
}
updateCrossTxHash(l2result.MsgHash, txHistory, h.db)
updateCrossTxHash(h.ctx, l2result.MsgHash, txHistory, h.db)
txHistories = append(txHistories, txHistory)
continue
}

14
bridge-history-api/utils/parse_event.go
View File

@@ -11,7 +11,7 @@ import (
"github.com/ethereum/go-ethereum/log"
backendabi "bridge-history-api/abi"
"bridge-history-api/db/orm"
"bridge-history-api/orm"
)
// CachedParsedTxCalldata store parsed batch infos
@@ -46,6 +46,7 @@ func ParseBackendL1EventLogs(logs []types.Log) ([]*orm.CrossMsg, []*orm.RelayedM
Amount: event.Amount.String(),
Asset: int(orm.ETH),
Layer1Hash: vlog.TxHash.Hex(),
MsgType: int(orm.Layer1Msg),
MsgHash: msgHash,
})
case backendabi.L1DepositERC20Sig:
@@ -64,6 +65,7 @@ func ParseBackendL1EventLogs(logs []types.Log) ([]*orm.CrossMsg, []*orm.RelayedM
Layer1Hash: vlog.TxHash.Hex(),
Layer1Token: event.L1Token.Hex(),
Layer2Token: event.L2Token.Hex(),
MsgType: int(orm.Layer1Msg),
MsgHash: msgHash,
})
case backendabi.L1DepositERC721Sig:
@@ -82,6 +84,7 @@ func ParseBackendL1EventLogs(logs []types.Log) ([]*orm.CrossMsg, []*orm.RelayedM
Layer1Token: event.L1Token.Hex(),
Layer2Token: event.L2Token.Hex(),
TokenIDs: event.TokenID.String(),
MsgType: int(orm.Layer1Msg),
MsgHash: msgHash,
})
case backendabi.L1DepositERC1155Sig:
@@ -101,6 +104,7 @@ func ParseBackendL1EventLogs(logs []types.Log) ([]*orm.CrossMsg, []*orm.RelayedM
Layer2Token: event.L2Token.Hex(),
TokenIDs: event.TokenID.String(),
Amount: event.Amount.String(),
MsgType: int(orm.Layer1Msg),
MsgHash: msgHash,
})
case backendabi.L1SentMessageEventSignature:
@@ -128,6 +132,7 @@ func ParseBackendL1EventLogs(logs []types.Log) ([]*orm.CrossMsg, []*orm.RelayedM
Layer1Token: event.L1Token.Hex(),
Layer2Token: event.L2Token.Hex(),
TokenIDs: convertBigIntArrayToString(event.TokenIDs),
MsgType: int(orm.Layer1Msg),
MsgHash: msgHash,
})
case backendabi.L1BatchDepositERC1155Sig:
@@ -147,6 +152,7 @@ func ParseBackendL1EventLogs(logs []types.Log) ([]*orm.CrossMsg, []*orm.RelayedM
Layer2Token: event.L2Token.Hex(),
TokenIDs: convertBigIntArrayToString(event.TokenIDs),
TokenAmounts: convertBigIntArrayToString(event.TokenAmounts),
MsgType: int(orm.Layer1Msg),
MsgHash: msgHash,
})
case backendabi.L1RelayedMessageEventSignature:
@@ -194,6 +200,7 @@ func ParseBackendL2EventLogs(logs []types.Log) ([]*orm.CrossMsg, []*orm.RelayedM
Amount: event.Amount.String(),
Asset: int(orm.ETH),
Layer2Hash: vlog.TxHash.Hex(),
MsgType: int(orm.Layer2Msg),
MsgHash: l2SentMsgs[len(l2SentMsgs)-1].MsgHash,
})
case backendabi.L2WithdrawERC20Sig:
@@ -213,6 +220,7 @@ func ParseBackendL2EventLogs(logs []types.Log) ([]*orm.CrossMsg, []*orm.RelayedM
Layer2Hash: vlog.TxHash.Hex(),
Layer1Token: event.L1Token.Hex(),
Layer2Token: event.L2Token.Hex(),
MsgType: int(orm.Layer2Msg),
MsgHash: l2SentMsgs[len(l2SentMsgs)-1].MsgHash,
})
case backendabi.L2WithdrawERC721Sig:
@@ -232,6 +240,7 @@ func ParseBackendL2EventLogs(logs []types.Log) ([]*orm.CrossMsg, []*orm.RelayedM
Layer1Token: event.L1Token.Hex(),
Layer2Token: event.L2Token.Hex(),
TokenIDs: event.TokenID.String(),
MsgType: int(orm.Layer2Msg),
MsgHash: l2SentMsgs[len(l2SentMsgs)-1].MsgHash,
})
case backendabi.L2WithdrawERC1155Sig:
@@ -252,6 +261,7 @@ func ParseBackendL2EventLogs(logs []types.Log) ([]*orm.CrossMsg, []*orm.RelayedM
Layer2Token: event.L2Token.Hex(),
TokenIDs: event.TokenID.String(),
Amount: event.Amount.String(),
MsgType: int(orm.Layer2Msg),
MsgHash: l2SentMsgs[len(l2SentMsgs)-1].MsgHash,
})
case backendabi.L2BatchWithdrawERC721Sig:
@@ -270,6 +280,7 @@ func ParseBackendL2EventLogs(logs []types.Log) ([]*orm.CrossMsg, []*orm.RelayedM
Layer1Hash: vlog.TxHash.Hex(),
Layer1Token: event.L1Token.Hex(),
Layer2Token: event.L2Token.Hex(),
MsgType: int(orm.Layer2Msg),
TokenIDs: convertBigIntArrayToString(event.TokenIDs),
MsgHash: l2SentMsgs[len(l2SentMsgs)-1].MsgHash,
})
@@ -289,6 +300,7 @@ func ParseBackendL2EventLogs(logs []types.Log) ([]*orm.CrossMsg, []*orm.RelayedM
Layer1Hash: vlog.TxHash.Hex(),
Layer1Token: event.L1Token.Hex(),
Layer2Token: event.L2Token.Hex(),
MsgType: int(orm.Layer2Msg),
TokenIDs: convertBigIntArrayToString(event.TokenIDs),
TokenAmounts: convertBigIntArrayToString(event.TokenAmounts),
MsgHash: l2SentMsgs[len(l2SentMsgs)-1].MsgHash,