package inventory import ( "context" "fmt" "math/big" "github.com/synapsecns/sanguine/services/cctp-relayer/contracts/cctp" "github.com/ethereum/go-ethereum/accounts/abi/bind" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/core/types" "github.com/synapsecns/sanguine/core/metrics" "github.com/synapsecns/sanguine/ethergo/listener" "github.com/synapsecns/sanguine/ethergo/submitter" "github.com/synapsecns/sanguine/services/rfq/relayer/relconfig" "github.com/synapsecns/sanguine/services/rfq/relayer/reldb" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" "golang.org/x/sync/errgroup" ) type rebalanceManagerSynapseCCTP struct { // cfg is the config cfg relconfig.Config // handler is the metrics handler handler metrics.Handler // chainClient is an omnirpc client chainClient submitter.ClientFetcher // txSubmitter is the transaction submitter txSubmitter submitter.TransactionSubmitter // cctpContracts is the map of cctp contracts (used for rebalancing) cctpContracts map[int]*cctp.SynapseCCTP // relayerAddress contains the relayer address relayerAddress common.Address // chainListeners is the map of chain listeners for CCTP events chainListeners map[int]listener.ContractListener // db is the database db reldb.Service } func newRebalanceManagerSynapseCCTP(cfg relconfig.Config, handler metrics.Handler, chainClient submitter.ClientFetcher, txSubmitter submitter.TransactionSubmitter, relayerAddress common.Address, db reldb.Service) *rebalanceManagerSynapseCCTP { return &rebalanceManagerSynapseCCTP{ cfg: cfg, handler: handler, chainClient: chainClient, txSubmitter: txSubmitter, cctpContracts: make(map[int]*cctp.SynapseCCTP), relayerAddress: relayerAddress, chainListeners: make(map[int]listener.ContractListener), db: db, } } func (c *rebalanceManagerSynapseCCTP) Start(ctx context.Context) (err error) { err = c.initContracts(ctx) if err != nil { return fmt.Errorf("could not initialize contracts: %w", err) } err = c.initListeners(ctx) if err != nil { return fmt.Errorf("could not initialize listeners: %w", err) } g, _ := errgroup.WithContext(ctx) for cid := range c.cfg.Chains { // capture func literal chainID := cid g.Go(func() error { return c.listen(ctx, chainID) }) } err = g.Wait() if err != nil { return fmt.Errorf("error listening to contract: %w", err) } return nil } func (c *rebalanceManagerSynapseCCTP) initContracts(parentCtx context.Context) (err error) { ctx, span := c.handler.Tracer().Start(parentCtx, "initContracts") defer func(err error) { metrics.EndSpanWithErr(span, err) }(err) for chainID := range c.cfg.Chains { contractAddr, err := c.cfg.GetSynapseCCTPAddress(chainID) if err != nil { return fmt.Errorf("could not get cctp address: %w", err) } chainClient, err := c.chainClient.GetClient(ctx, big.NewInt(int64(chainID))) if err != nil { return fmt.Errorf("could not get chain client: %w", err) } contract, err := cctp.NewSynapseCCTP(contractAddr, chainClient) if err != nil { return fmt.Errorf("could not get cctp: %w", err) } c.cctpContracts[chainID] = contract } return nil } func (c *rebalanceManagerSynapseCCTP) initListeners(ctx context.Context) (err error) { for chainID := range c.cfg.GetChains() { cctpAddr, err := c.cfg.GetSynapseCCTPAddress(chainID) if err != nil { return fmt.Errorf("could not get cctp address: %w", err) } chainClient, err := c.chainClient.GetClient(ctx, big.NewInt(int64(chainID))) if err != nil { return fmt.Errorf("could not get chain client: %w", err) } initialBlock, err := c.cfg.GetRebalanceStartBlock(chainID) if err != nil { return fmt.Errorf("could not get cctp start block: %w", err) } chainListener, err := listener.NewChainListener(chainClient, c.db, cctpAddr, initialBlock, c.handler) if err != nil { return fmt.Errorf("could not get chain listener: %w", err) } c.chainListeners[chainID] = chainListener } return nil } func (c *rebalanceManagerSynapseCCTP) Execute(parentCtx context.Context, rebalance *RebalanceData) (err error) { contract, ok := c.cctpContracts[rebalance.OriginMetadata.ChainID] if !ok { return fmt.Errorf("could not find cctp contract for chain %d", rebalance.OriginMetadata.ChainID) } ctx, span := c.handler.Tracer().Start(parentCtx, "rebalance.Execute", trace.WithAttributes( attribute.Int("rebalance_origin", rebalance.OriginMetadata.ChainID), attribute.Int("rebalance_dest", rebalance.DestMetadata.ChainID), attribute.String("rebalance_amount", rebalance.Amount.String()), )) defer func(err error) { metrics.EndSpanWithErr(span, err) }(err) // perform rebalance by calling sendCircleToken() _, err = c.txSubmitter.SubmitTransaction(ctx, big.NewInt(int64(rebalance.OriginMetadata.ChainID)), func(transactor *bind.TransactOpts) (tx *types.Transaction, err error) { tx, err = contract.SendCircleToken( transactor, c.relayerAddress, big.NewInt(int64(rebalance.DestMetadata.ChainID)), rebalance.OriginMetadata.Addr, rebalance.Amount, 0, // TODO: inspect []byte{}, // TODO: inspect ) if err != nil { return nil, fmt.Errorf("could not send circle token: %w", err) } return tx, nil }) if err != nil { return fmt.Errorf("could not submit CCTP rebalance: %w", err) } // store the rebalance in the db model := reldb.Rebalance{ Origin: uint64(rebalance.OriginMetadata.ChainID), //nolint:gosec // ChainID is always positive and within uint64 range Destination: uint64(rebalance.DestMetadata.ChainID), //nolint:gosec // ChainID is always positive and within uint64 range OriginAmount: rebalance.Amount, Status: reldb.RebalanceInitiated, OriginTokenAddr: rebalance.OriginMetadata.Addr, TokenName: rebalance.OriginMetadata.Name, } err = c.db.StoreRebalance(ctx, model) if err != nil { return fmt.Errorf("could not store rebalance: %w", err) } return nil } // nolint:cyclop func (c *rebalanceManagerSynapseCCTP) listen(parentCtx context.Context, chainID int) (err error) { listener, ok := c.chainListeners[chainID] if !ok { return fmt.Errorf("could not find listener for chain %d", chainID) } ethClient, err := c.chainClient.GetClient(parentCtx, big.NewInt(int64(chainID))) if err != nil { return fmt.Errorf("could not get chain client: %w", err) } cctpAddr, err := c.cfg.GetSynapseCCTPAddress(chainID) if err != nil { return fmt.Errorf("could not get cctp address: %w", err) } parser, err := cctp.NewSynapseCCTPEvents(cctpAddr, ethClient) if err != nil { return fmt.Errorf("could not get cctp events: %w", err) } err = listener.Listen(parentCtx, func(parentCtx context.Context, log types.Log) (err error) { ctx, span := c.handler.Tracer().Start(parentCtx, "rebalance.Listen", trace.WithAttributes( attribute.Int(metrics.ChainID, chainID), )) defer func(err error) { metrics.EndSpanWithErr(span, err) }(err) switch log.Topics[0] { case cctp.CircleRequestSentTopic: parsedEvent, err := parser.ParseCircleRequestSent(log) if err != nil { logger.Warnf("could not parse circle request sent: %v", err) return nil } if parsedEvent.Sender != c.relayerAddress { return nil } span.SetAttributes( attribute.String("log_type", "CircleRequestSent"), attribute.String("request_id", hexutil.Encode(parsedEvent.RequestID[:])), ) // update rebalance model in db requestIDHex := hexutil.Encode(parsedEvent.RequestID[:]) rebalanceModel := reldb.Rebalance{ RebalanceID: &requestIDHex, Origin: uint64(chainID), OriginTxHash: log.TxHash, OriginTokenAddr: parsedEvent.Token, Status: reldb.RebalancePending, } err = c.db.UpdateRebalance(ctx, rebalanceModel, true) if err != nil { logger.Warnf("could not update rebalance status: %v", err) return nil } case cctp.CircleRequestFulfilledTopic: parsedEvent, err := parser.ParseCircleRequestFulfilled(log) if err != nil { logger.Warnf("could not parse circle request fulfilled: %v", err) return nil } if parsedEvent.Recipient != c.relayerAddress { return nil } span.SetAttributes( attribute.String("log_type", "CircleRequestFulfilled"), attribute.String("request_id", hexutil.Encode(parsedEvent.RequestID[:])), ) // update rebalance model in db requestIDHex := hexutil.Encode(parsedEvent.RequestID[:]) rebalanceModel := reldb.Rebalance{ RebalanceID: &requestIDHex, DestTxHash: log.TxHash, Status: reldb.RebalanceCompleted, } err = c.db.UpdateRebalance(parentCtx, rebalanceModel, false) if err != nil { logger.Warnf("could not update rebalance status: %v", err) return nil } default: logger.Warnf("unknown event %s", log.Topics[0]) } return nil }) if err != nil { return fmt.Errorf("could not listen to contract: %w", err) } return nil }