synapsecns/sanguine

package base

import (
    "context"
    "fmt"
    "golang.org/x/sync/errgroup"

    "github.com/ethereum/go-ethereum/common"
    "github.com/synapsecns/sanguine/core/metrics"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/trace"
    "gorm.io/gorm/clause"
)

const lastIndexedLivefillKey = "0x0000000000000000000000000000000000000000"

// StoreLastIndexed stores the last indexed block number for a contract.
// It updates the value if there is a previous last indexed value, and creates a new
// entry if there's no previous value.
func (s Store) StoreLastIndexed(parentCtx context.Context, contractAddress common.Address, chainID uint32, blockNumber uint64, livefillAtHead bool) (err error) {
    ctx, span := s.metrics.Tracer().Start(parentCtx, "StoreLastIndexed", trace.WithAttributes(
        attribute.String("contractAddress", contractAddress.String()),
        attribute.Int("chainID", int(chainID)),
        attribute.Int("blockNumber", int(blockNumber)),
    ))

    defer func() {
        metrics.EndSpanWithErr(span, err)
    }()

    address := contractAddress.String()
    if livefillAtHead {
        address = lastIndexedLivefillKey
    }

    dbTx := s.DB().WithContext(ctx).
        Clauses(clause.OnConflict{
            Columns:   []clause.Column{{Name: ContractAddressFieldName}, {Name: ChainIDFieldName}},
            DoUpdates: clause.AssignmentColumns([]string{BlockNumberFieldName}),
            Where: clause.Where{
                Exprs: []clause.Expression{
                    clause.And(
                        clause.Where{
                            Exprs: []clause.Expression{
                                clause.Eq{
                                    Column: clause.Column{Name: ContractAddressFieldName},
                                    Value:  address,
                                },
                                clause.Eq{
                                    Column: clause.Column{Name: ChainIDFieldName},
                                    Value:  chainID,
                                },
                            },
                        },
                        clause.Lt{
                            Column: clause.Column{Name: BlockNumberFieldName},
                            Value:  blockNumber,
                        },
                    ),
                },
            },
        }).
        Create(&LastIndexedInfo{
            ContractAddress: address,
            ChainID:         chainID,
            BlockNumber:     blockNumber,
        })
    if dbTx.Error != nil {
        return fmt.Errorf("could not update last indexed info: %w", dbTx.Error)
    }
    return nil
}

// RetrieveLastIndexed retrieves the last indexed block number for a contract.
func (s Store) RetrieveLastIndexed(ctx context.Context, contractAddress common.Address, chainID uint32, livefill bool) (uint64, error) {
    entry := LastIndexedInfo{}
    address := contractAddress.String()
    if livefill {
        address = lastIndexedLivefillKey
    }

    dbTx := s.DB().WithContext(ctx).
        Model(&LastIndexedInfo{}).
        Where(&LastIndexedInfo{
            ContractAddress: address,
            ChainID:         chainID,
        }).
        First(&entry)
    if dbTx.RowsAffected == 0 {
        return 0, nil
    }
    if dbTx.Error != nil {
        return 0, fmt.Errorf("could not retrieve last indexed info: %w", dbTx.Error)
    }
    return entry.BlockNumber, nil
}

// StoreLastIndexedMultiple stores the last indexed block numbers for numerous contracts.
func (s Store) StoreLastIndexedMultiple(parentCtx context.Context, contractAddresses []common.Address, chainID uint32, blockNumber uint64) error {
    g, groupCtx := errgroup.WithContext(parentCtx)

    for i := range contractAddresses {
        index := i
        g.Go(func() error {
            err := s.StoreLastIndexed(groupCtx, contractAddresses[index], chainID, blockNumber, false)
            if err != nil {
                return fmt.Errorf("could not backfill: %w", err)
            }
            return nil
        })
    }
    if err := g.Wait(); err != nil {
        return fmt.Errorf("could not store last indexed: %w", err)
    }

    return nil
}

// RetrieveLastIndexedMultiple retrieves the last indexed block numbers for numerous contracts.
func (s Store) RetrieveLastIndexedMultiple(ctx context.Context, contractAddresses []common.Address, chainID uint32) (map[common.Address]uint64, error) {
    var entries []LastIndexedInfo
    addrStrings := make([]string, len(contractAddresses))
    for i, addr := range contractAddresses {
        addrStrings[i] = addr.String()
    }

    dbTx := s.DB().WithContext(ctx).
        Model(&LastIndexedInfo{}).
        Where("contract_address in ? AND chain_id = ?", addrStrings, chainID).
        Find(&entries)

    if dbTx.Error != nil {
        return nil, fmt.Errorf("could not retrieve last indexed info: %w", dbTx.Error)
    }

    result := make(map[common.Address]uint64)
    for _, addr := range contractAddresses {
        result[addr] = 0
    }

    for _, entry := range entries {
        addr := common.HexToAddress(entry.ContractAddress)
        result[addr] = entry.BlockNumber
    }

    return result, nil
}