diff --git a/.changeset/mighty-flies-breathe.md b/.changeset/mighty-flies-breathe.md new file mode 100644 index 00000000000..d983aad7086 --- /dev/null +++ b/.changeset/mighty-flies-breathe.md @@ -0,0 +1,5 @@ +--- +"chainlink": patch +--- + +#added ORM and corresponding tables for CCIP gas prices and token prices diff --git a/core/services/ccip/mocks/orm.go b/core/services/ccip/mocks/orm.go new file mode 100644 index 00000000000..b9afc6c8695 --- /dev/null +++ b/core/services/ccip/mocks/orm.go @@ -0,0 +1,164 @@ +// Code generated by mockery v2.42.2. DO NOT EDIT. + +package mocks + +import ( + context "context" + + ccip "github.com/smartcontractkit/chainlink/v2/core/services/ccip" + + mock "github.com/stretchr/testify/mock" + + time "time" +) + +// ORM is an autogenerated mock type for the ORM type +type ORM struct { + mock.Mock +} + +// ClearGasPricesByDestChain provides a mock function with given fields: ctx, destChainSelector, to +func (_m *ORM) ClearGasPricesByDestChain(ctx context.Context, destChainSelector uint64, to time.Time) error { + ret := _m.Called(ctx, destChainSelector, to) + + if len(ret) == 0 { + panic("no return value specified for ClearGasPricesByDestChain") + } + + var r0 error + if rf, ok := ret.Get(0).(func(context.Context, uint64, time.Time) error); ok { + r0 = rf(ctx, destChainSelector, to) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// ClearTokenPricesByDestChain provides a mock function with given fields: ctx, destChainSelector, to +func (_m *ORM) ClearTokenPricesByDestChain(ctx context.Context, destChainSelector uint64, to time.Time) error { + ret := _m.Called(ctx, destChainSelector, to) + + if len(ret) == 0 { + panic("no return value specified for ClearTokenPricesByDestChain") + } + + var r0 error + if rf, ok := ret.Get(0).(func(context.Context, uint64, time.Time) error); ok { + r0 = rf(ctx, destChainSelector, to) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// GetGasPricesByDestChain provides a mock function with given fields: ctx, destChainSelector +func (_m *ORM) GetGasPricesByDestChain(ctx context.Context, destChainSelector uint64) ([]ccip.GasPrice, error) { + ret := _m.Called(ctx, destChainSelector) + + if len(ret) == 0 { + panic("no return value specified for GetGasPricesByDestChain") + } + + var r0 []ccip.GasPrice + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, uint64) ([]ccip.GasPrice, error)); ok { + return rf(ctx, destChainSelector) + } + if rf, ok := ret.Get(0).(func(context.Context, uint64) []ccip.GasPrice); ok { + r0 = rf(ctx, destChainSelector) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([]ccip.GasPrice) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, uint64) error); ok { + r1 = rf(ctx, destChainSelector) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// GetTokenPricesByDestChain provides a mock function with given fields: ctx, destChainSelector +func (_m *ORM) GetTokenPricesByDestChain(ctx context.Context, destChainSelector uint64) ([]ccip.TokenPrice, error) { + ret := _m.Called(ctx, destChainSelector) + + if len(ret) == 0 { + panic("no return value specified for GetTokenPricesByDestChain") + } + + var r0 []ccip.TokenPrice + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, uint64) ([]ccip.TokenPrice, error)); ok { + return rf(ctx, destChainSelector) + } + if rf, ok := ret.Get(0).(func(context.Context, uint64) []ccip.TokenPrice); ok { + r0 = rf(ctx, destChainSelector) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([]ccip.TokenPrice) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, uint64) error); ok { + r1 = rf(ctx, destChainSelector) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// InsertGasPricesForDestChain provides a mock function with given fields: ctx, destChainSelector, jobId, gasPrices +func (_m *ORM) InsertGasPricesForDestChain(ctx context.Context, destChainSelector uint64, jobId int32, gasPrices []ccip.GasPriceUpdate) error { + ret := _m.Called(ctx, destChainSelector, jobId, gasPrices) + + if len(ret) == 0 { + panic("no return value specified for InsertGasPricesForDestChain") + } + + var r0 error + if rf, ok := ret.Get(0).(func(context.Context, uint64, int32, []ccip.GasPriceUpdate) error); ok { + r0 = rf(ctx, destChainSelector, jobId, gasPrices) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// InsertTokenPricesForDestChain provides a mock function with given fields: ctx, destChainSelector, jobId, tokenPrices +func (_m *ORM) InsertTokenPricesForDestChain(ctx context.Context, destChainSelector uint64, jobId int32, tokenPrices []ccip.TokenPriceUpdate) error { + ret := _m.Called(ctx, destChainSelector, jobId, tokenPrices) + + if len(ret) == 0 { + panic("no return value specified for InsertTokenPricesForDestChain") + } + + var r0 error + if rf, ok := ret.Get(0).(func(context.Context, uint64, int32, []ccip.TokenPriceUpdate) error); ok { + r0 = rf(ctx, destChainSelector, jobId, tokenPrices) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// NewORM creates a new instance of ORM. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. +// The first argument is typically a *testing.T value. +func NewORM(t interface { + mock.TestingT + Cleanup(func()) +}) *ORM { + mock := &ORM{} + mock.Mock.Test(t) + + t.Cleanup(func() { mock.AssertExpectations(t) }) + + return mock +} diff --git a/core/services/ccip/orm.go b/core/services/ccip/orm.go new file mode 100644 index 00000000000..8af7762b18d --- /dev/null +++ b/core/services/ccip/orm.go @@ -0,0 +1,163 @@ +package ccip + +import ( + "context" + "fmt" + "time" + + "github.com/smartcontractkit/chainlink-common/pkg/sqlutil" + + "github.com/smartcontractkit/chainlink/v2/core/chains/evm/assets" +) + +type GasPrice struct { + SourceChainSelector uint64 + GasPrice *assets.Wei + CreatedAt time.Time +} + +type GasPriceUpdate struct { + SourceChainSelector uint64 + GasPrice *assets.Wei +} + +type TokenPrice struct { + TokenAddr string + TokenPrice *assets.Wei + CreatedAt time.Time +} + +type TokenPriceUpdate struct { + TokenAddr string + TokenPrice *assets.Wei +} + +//go:generate mockery --quiet --name ORM --output ./mocks/ --case=underscore +type ORM interface { + GetGasPricesByDestChain(ctx context.Context, destChainSelector uint64) ([]GasPrice, error) + GetTokenPricesByDestChain(ctx context.Context, destChainSelector uint64) ([]TokenPrice, error) + + InsertGasPricesForDestChain(ctx context.Context, destChainSelector uint64, jobId int32, gasPrices []GasPriceUpdate) error + InsertTokenPricesForDestChain(ctx context.Context, destChainSelector uint64, jobId int32, tokenPrices []TokenPriceUpdate) error + + ClearGasPricesByDestChain(ctx context.Context, destChainSelector uint64, to time.Time) error + ClearTokenPricesByDestChain(ctx context.Context, destChainSelector uint64, to time.Time) error +} + +type orm struct { + ds sqlutil.DataSource +} + +var _ ORM = (*orm)(nil) + +func NewORM(ds sqlutil.DataSource) (ORM, error) { + if ds == nil { + return nil, fmt.Errorf("datasource to CCIP NewORM cannot be nil") + } + + return &orm{ + ds: ds, + }, nil +} + +func (o *orm) GetGasPricesByDestChain(ctx context.Context, destChainSelector uint64) ([]GasPrice, error) { + var gasPrices []GasPrice + stmt := ` + SELECT DISTINCT ON (source_chain_selector) + source_chain_selector, gas_price, created_at + FROM ccip.observed_gas_prices + WHERE chain_selector = $1 + ORDER BY source_chain_selector, created_at DESC; + ` + err := o.ds.SelectContext(ctx, &gasPrices, stmt, destChainSelector) + if err != nil { + return nil, err + } + + return gasPrices, nil +} + +func (o *orm) GetTokenPricesByDestChain(ctx context.Context, destChainSelector uint64) ([]TokenPrice, error) { + var tokenPrices []TokenPrice + stmt := ` + SELECT DISTINCT ON (token_addr) + token_addr, token_price, created_at + FROM ccip.observed_token_prices + WHERE chain_selector = $1 + ORDER BY token_addr, created_at DESC; + ` + err := o.ds.SelectContext(ctx, &tokenPrices, stmt, destChainSelector) + if err != nil { + return nil, err + } + + return tokenPrices, nil +} + +func (o *orm) InsertGasPricesForDestChain(ctx context.Context, destChainSelector uint64, jobId int32, gasPrices []GasPriceUpdate) error { + if len(gasPrices) == 0 { + return nil + } + + now := time.Now() + insertData := make([]map[string]interface{}, 0, len(gasPrices)) + for _, price := range gasPrices { + insertData = append(insertData, map[string]interface{}{ + "chain_selector": destChainSelector, + "job_id": jobId, + "source_chain_selector": price.SourceChainSelector, + "gas_price": price.GasPrice, + "created_at": now, + }) + } + + stmt := `INSERT INTO ccip.observed_gas_prices (chain_selector, job_id, source_chain_selector, gas_price, created_at) + VALUES (:chain_selector, :job_id, :source_chain_selector, :gas_price, :created_at);` + _, err := o.ds.NamedExecContext(ctx, stmt, insertData) + if err != nil { + err = fmt.Errorf("error inserting gas prices for job %d: %w", jobId, err) + } + + return err +} + +func (o *orm) InsertTokenPricesForDestChain(ctx context.Context, destChainSelector uint64, jobId int32, tokenPrices []TokenPriceUpdate) error { + if len(tokenPrices) == 0 { + return nil + } + + now := time.Now() + insertData := make([]map[string]interface{}, 0, len(tokenPrices)) + for _, price := range tokenPrices { + insertData = append(insertData, map[string]interface{}{ + "chain_selector": destChainSelector, + "job_id": jobId, + "token_addr": price.TokenAddr, + "token_price": price.TokenPrice, + "created_at": now, + }) + } + + stmt := `INSERT INTO ccip.observed_token_prices (chain_selector, job_id, token_addr, token_price, created_at) + VALUES (:chain_selector, :job_id, :token_addr, :token_price, :created_at);` + _, err := o.ds.NamedExecContext(ctx, stmt, insertData) + if err != nil { + err = fmt.Errorf("error inserting token prices for job %d: %w", jobId, err) + } + + return err +} + +func (o *orm) ClearGasPricesByDestChain(ctx context.Context, destChainSelector uint64, to time.Time) error { + stmt := `DELETE FROM ccip.observed_gas_prices WHERE chain_selector = $1 AND created_at < $2` + + _, err := o.ds.ExecContext(ctx, stmt, destChainSelector, to) + return err +} + +func (o *orm) ClearTokenPricesByDestChain(ctx context.Context, destChainSelector uint64, to time.Time) error { + stmt := `DELETE FROM ccip.observed_token_prices WHERE chain_selector = $1 AND created_at < $2` + + _, err := o.ds.ExecContext(ctx, stmt, destChainSelector, to) + return err +} diff --git a/core/services/ccip/orm_test.go b/core/services/ccip/orm_test.go new file mode 100644 index 00000000000..741cf4b5b38 --- /dev/null +++ b/core/services/ccip/orm_test.go @@ -0,0 +1,346 @@ +package ccip + +import ( + "math/big" + "math/rand" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/chainlink-common/pkg/sqlutil" + + "github.com/smartcontractkit/chainlink/v2/core/chains/evm/assets" + "github.com/smartcontractkit/chainlink/v2/core/chains/evm/utils" + "github.com/smartcontractkit/chainlink/v2/core/internal/testutils" + "github.com/smartcontractkit/chainlink/v2/core/internal/testutils/pgtest" +) + +func setupORM(t *testing.T) (ORM, sqlutil.DataSource) { + t.Helper() + + db := pgtest.NewSqlxDB(t) + orm, err := NewORM(db) + + require.NoError(t, err) + + return orm, db +} + +func generateChainSelectors(n int) []uint64 { + selectors := make([]uint64, n) + for i := 0; i < n; i++ { + selectors[i] = rand.Uint64() + } + + return selectors +} + +func generateGasPriceUpdates(chainSelector uint64, n int) []GasPriceUpdate { + updates := make([]GasPriceUpdate, n) + for i := 0; i < n; i++ { + // gas prices can take up whole range of uint256 + uint256Max := new(big.Int).Sub(new(big.Int).Exp(big.NewInt(2), big.NewInt(256), nil), big.NewInt(1)) + row := GasPriceUpdate{ + SourceChainSelector: chainSelector, + GasPrice: assets.NewWei(new(big.Int).Sub(uint256Max, big.NewInt(int64(i)))), + } + updates[i] = row + } + + return updates +} + +func generateTokenAddresses(n int) []string { + addrs := make([]string, n) + for i := 0; i < n; i++ { + addrs[i] = utils.RandomAddress().Hex() + } + + return addrs +} + +func generateTokenPriceUpdates(tokenAddr string, n int) []TokenPriceUpdate { + updates := make([]TokenPriceUpdate, n) + for i := 0; i < n; i++ { + row := TokenPriceUpdate{ + TokenAddr: tokenAddr, + TokenPrice: assets.NewWei(new(big.Int).Mul(big.NewInt(1e18), big.NewInt(int64(i)))), + } + updates[i] = row + } + + return updates +} + +func getGasTableRowCount(t *testing.T, ds sqlutil.DataSource) int { + var count int + stmt := `SELECT COUNT(*) FROM ccip.observed_gas_prices;` + err := ds.QueryRowxContext(testutils.Context(t), stmt).Scan(&count) + require.NoError(t, err) + + return count +} + +func getTokenTableRowCount(t *testing.T, ds sqlutil.DataSource) int { + var count int + stmt := `SELECT COUNT(*) FROM ccip.observed_token_prices;` + err := ds.QueryRowxContext(testutils.Context(t), stmt).Scan(&count) + require.NoError(t, err) + + return count +} + +func TestInitORM(t *testing.T) { + t.Parallel() + + orm, _ := setupORM(t) + assert.NotNil(t, orm) +} + +func TestORM_EmptyGasPrices(t *testing.T) { + t.Parallel() + ctx := testutils.Context(t) + + orm, _ := setupORM(t) + + prices, err := orm.GetGasPricesByDestChain(ctx, 1) + assert.Equal(t, 0, len(prices)) + assert.NoError(t, err) +} + +func TestORM_EmptyTokenPrices(t *testing.T) { + t.Parallel() + ctx := testutils.Context(t) + + orm, _ := setupORM(t) + + prices, err := orm.GetTokenPricesByDestChain(ctx, 1) + assert.Equal(t, 0, len(prices)) + assert.NoError(t, err) +} + +func TestORM_InsertAndGetGasPrices(t *testing.T) { + t.Parallel() + ctx := testutils.Context(t) + + orm, db := setupORM(t) + + numJobs := 5 + numSourceChainSelectors := 10 + numUpdatesPerSourceSelector := 20 + destSelector := uint64(1) + + sourceSelectors := generateChainSelectors(numSourceChainSelectors) + + updates := make(map[uint64][]GasPriceUpdate) + for _, selector := range sourceSelectors { + updates[selector] = generateGasPriceUpdates(selector, numUpdatesPerSourceSelector) + } + + // 5 jobs, each inserting prices for 10 chains, with 20 updates per chain. + expectedPrices := make(map[uint64]GasPriceUpdate) + for i := 0; i < numJobs; i++ { + for selector, updatesPerSelector := range updates { + lastIndex := len(updatesPerSelector) - 1 + + err := orm.InsertGasPricesForDestChain(ctx, destSelector, int32(i), updatesPerSelector[:lastIndex]) + assert.NoError(t, err) + err = orm.InsertGasPricesForDestChain(ctx, destSelector, int32(i), updatesPerSelector[lastIndex:]) + assert.NoError(t, err) + + expectedPrices[selector] = updatesPerSelector[lastIndex] + } + } + + // verify number of rows inserted + numRows := getGasTableRowCount(t, db) + assert.Equal(t, numJobs*numSourceChainSelectors*numUpdatesPerSourceSelector, numRows) + + prices, err := orm.GetGasPricesByDestChain(ctx, destSelector) + assert.NoError(t, err) + // should return 1 price per source chain selector + assert.Equal(t, numSourceChainSelectors, len(prices)) + + // verify getGasPrices returns prices of latest timestamp + for _, price := range prices { + selector := price.SourceChainSelector + assert.Equal(t, expectedPrices[selector].GasPrice, price.GasPrice) + } + + // after the initial inserts, insert new round of prices, 1 price per selector this time + var combinedUpdates []GasPriceUpdate + for selector, updatesPerSelector := range updates { + combinedUpdates = append(combinedUpdates, updatesPerSelector[0]) + expectedPrices[selector] = updatesPerSelector[0] + } + + err = orm.InsertGasPricesForDestChain(ctx, destSelector, 1, combinedUpdates) + assert.NoError(t, err) + assert.Equal(t, numJobs*numSourceChainSelectors*numUpdatesPerSourceSelector+numSourceChainSelectors, getGasTableRowCount(t, db)) + + prices, err = orm.GetGasPricesByDestChain(ctx, destSelector) + assert.NoError(t, err) + assert.Equal(t, numSourceChainSelectors, len(prices)) + + for _, price := range prices { + selector := price.SourceChainSelector + assert.Equal(t, expectedPrices[selector].GasPrice, price.GasPrice) + } +} + +func TestORM_InsertAndDeleteGasPrices(t *testing.T) { + t.Parallel() + ctx := testutils.Context(t) + + orm, db := setupORM(t) + + numSourceChainSelectors := 10 + numUpdatesPerSourceSelector := 20 + destSelector := uint64(1) + + sourceSelectors := generateChainSelectors(numSourceChainSelectors) + + updates := make(map[uint64][]GasPriceUpdate) + for _, selector := range sourceSelectors { + updates[selector] = generateGasPriceUpdates(selector, numUpdatesPerSourceSelector) + } + + for _, updatesPerSelector := range updates { + err := orm.InsertGasPricesForDestChain(ctx, destSelector, 1, updatesPerSelector) + assert.NoError(t, err) + } + + interimTimeStamp := time.Now() + + // insert for the 2nd time after interimTimeStamp + for _, updatesPerSelector := range updates { + err := orm.InsertGasPricesForDestChain(ctx, destSelector, 1, updatesPerSelector) + assert.NoError(t, err) + } + + assert.Equal(t, 2*numSourceChainSelectors*numUpdatesPerSourceSelector, getGasTableRowCount(t, db)) + + // clear by interimTimeStamp should delete rows inserted before it + err := orm.ClearGasPricesByDestChain(ctx, destSelector, interimTimeStamp) + assert.NoError(t, err) + assert.Equal(t, numSourceChainSelectors*numUpdatesPerSourceSelector, getGasTableRowCount(t, db)) + + // clear by Now() should delete all rows + err = orm.ClearGasPricesByDestChain(ctx, destSelector, time.Now()) + assert.NoError(t, err) + assert.Equal(t, 0, getGasTableRowCount(t, db)) +} + +func TestORM_InsertAndGetTokenPrices(t *testing.T) { + t.Parallel() + ctx := testutils.Context(t) + + orm, db := setupORM(t) + + numJobs := 5 + numAddresses := 10 + numUpdatesPerAddress := 20 + destSelector := uint64(1) + + addrs := generateTokenAddresses(numAddresses) + + updates := make(map[string][]TokenPriceUpdate) + for _, addr := range addrs { + updates[addr] = generateTokenPriceUpdates(addr, numUpdatesPerAddress) + } + + // 5 jobs, each inserting prices for 10 chains, with 20 updates per chain. + expectedPrices := make(map[string]TokenPriceUpdate) + for i := 0; i < numJobs; i++ { + for addr, updatesPerAddr := range updates { + lastIndex := len(updatesPerAddr) - 1 + + err := orm.InsertTokenPricesForDestChain(ctx, destSelector, int32(i), updatesPerAddr[:lastIndex]) + assert.NoError(t, err) + err = orm.InsertTokenPricesForDestChain(ctx, destSelector, int32(i), updatesPerAddr[lastIndex:]) + assert.NoError(t, err) + + expectedPrices[addr] = updatesPerAddr[lastIndex] + } + } + + // verify number of rows inserted + numRows := getTokenTableRowCount(t, db) + assert.Equal(t, numJobs*numAddresses*numUpdatesPerAddress, numRows) + + prices, err := orm.GetTokenPricesByDestChain(ctx, destSelector) + assert.NoError(t, err) + // should return 1 price per source chain selector + assert.Equal(t, numAddresses, len(prices)) + + // verify getTokenPrices returns prices of latest timestamp + for _, price := range prices { + addr := price.TokenAddr + assert.Equal(t, expectedPrices[addr].TokenPrice, price.TokenPrice) + } + + // after the initial inserts, insert new round of prices, 1 price per selector this time + var combinedUpdates []TokenPriceUpdate + for addr, updatesPerAddr := range updates { + combinedUpdates = append(combinedUpdates, updatesPerAddr[0]) + expectedPrices[addr] = updatesPerAddr[0] + } + + err = orm.InsertTokenPricesForDestChain(ctx, destSelector, 1, combinedUpdates) + assert.NoError(t, err) + assert.Equal(t, numJobs*numAddresses*numUpdatesPerAddress+numAddresses, getTokenTableRowCount(t, db)) + + prices, err = orm.GetTokenPricesByDestChain(ctx, destSelector) + assert.NoError(t, err) + assert.Equal(t, numAddresses, len(prices)) + + for _, price := range prices { + addr := price.TokenAddr + assert.Equal(t, expectedPrices[addr].TokenPrice, price.TokenPrice) + } +} + +func TestORM_InsertAndDeleteTokenPrices(t *testing.T) { + t.Parallel() + ctx := testutils.Context(t) + + orm, db := setupORM(t) + + numAddresses := 10 + numUpdatesPerAddress := 20 + destSelector := uint64(1) + + addrs := generateTokenAddresses(numAddresses) + + updates := make(map[string][]TokenPriceUpdate) + for _, addr := range addrs { + updates[addr] = generateTokenPriceUpdates(addr, numUpdatesPerAddress) + } + + for _, updatesPerAddr := range updates { + err := orm.InsertTokenPricesForDestChain(ctx, destSelector, 1, updatesPerAddr) + assert.NoError(t, err) + } + + interimTimeStamp := time.Now() + + // insert for the 2nd time after interimTimeStamp + for _, updatesPerAddr := range updates { + err := orm.InsertTokenPricesForDestChain(ctx, destSelector, 1, updatesPerAddr) + assert.NoError(t, err) + } + + assert.Equal(t, 2*numAddresses*numUpdatesPerAddress, getTokenTableRowCount(t, db)) + + // clear by interimTimeStamp should delete rows inserted before it + err := orm.ClearTokenPricesByDestChain(ctx, destSelector, interimTimeStamp) + assert.NoError(t, err) + assert.Equal(t, numAddresses*numUpdatesPerAddress, getTokenTableRowCount(t, db)) + + // clear by Now() should delete all rows + err = orm.ClearTokenPricesByDestChain(ctx, destSelector, time.Now()) + assert.NoError(t, err) + assert.Equal(t, 0, getTokenTableRowCount(t, db)) +} diff --git a/core/store/migrate/migrations/0236_ccip_prices_cache.sql b/core/store/migrate/migrations/0236_ccip_prices_cache.sql new file mode 100644 index 00000000000..e88b68e5575 --- /dev/null +++ b/core/store/migrate/migrations/0236_ccip_prices_cache.sql @@ -0,0 +1,36 @@ +-- +goose Up +-- +goose StatementBegin +CREATE SCHEMA ccip; + +CREATE TABLE ccip.observed_gas_prices( + chain_selector NUMERIC(20,0) NOT NULL, + job_id INTEGER NOT NULL, + source_chain_selector NUMERIC(20,0) NOT NULL, + gas_price NUMERIC(78,0) NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE TABLE ccip.observed_token_prices( + chain_selector NUMERIC(20,0) NOT NULL, + job_id INTEGER NOT NULL, + token_addr BYTEA NOT NULL, + token_price NUMERIC(78,0) NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX idx_ccip_gas_prices_chain_gas_price_timestamp ON ccip.observed_gas_prices (chain_selector, source_chain_selector, created_at DESC); +CREATE INDEX idx_ccip_token_prices_token_price_timestamp ON ccip.observed_token_prices (chain_selector, token_addr, created_at DESC); + +-- +goose StatementEnd + + +-- +goose Down +-- +goose StatementBegin +DROP INDEX IF EXISTS idx_ccip_token_prices_token_value; +DROP INDEX IF EXISTS idx_ccip_gas_prices_chain_value; + +DROP TABLE ccip.observed_token_prices; +DROP TABLE ccip.observed_gas_prices; + +DROP SCHEMA ccip; +-- +goose StatementEnd