Remove live reserve state polling

ocp/currency/reserve.go

@@ -14,26 +14,26 @@ import (
 
 // GetLaunchpadCurrencyCirculatingSupply gets the current circulating supply in
 // quarks for a launchpad currency directly from the blockchain
-func GetLaunchpadCurrencyCirculatingSupply(ctx context.Context, data ocp_data.Provider, mint *common.Account) (uint64, time.Time, error) {
+func GetLaunchpadCurrencyCirculatingSupply(ctx context.Context, data ocp_data.Provider, mint *common.Account) (uint64, uint64, time.Time, error) {
 	metadataRecord, err := data.GetCurrencyMetadata(ctx, mint.PublicKey().ToBase58())
 	if err != nil {
-		return 0, time.Time{}, err
+		return 0, 0, time.Time{}, err
 	}
 
 	accounts, err := common.GetLaunchpadCurrencyAccounts(metadataRecord)
 	if err != nil {
-		return 0, time.Time{}, err
+		return 0, 0, time.Time{}, err
 	}
 
-	ai, _, err := data.GetBlockchainAccountInfo(ctx, accounts.VaultMint.PublicKey().ToBase58(), solana.CommitmentFinalized)
+	ai, slot, err := data.GetBlockchainAccountInfo(ctx, accounts.VaultMint.PublicKey().ToBase58(), solana.CommitmentFinalized)
 	if err != nil {
-		return 0, time.Time{}, err
+		return 0, 0, time.Time{}, err
 	}
 
 	var tokenAccount token.Account
 	if !tokenAccount.Unmarshal(ai.Data) {
-		return 0, time.Time{}, errors.New("invalid token account state")
+		return 0, 0, time.Time{}, errors.New("invalid token account state")
 	}
 
-	return currencycreator.DefaultMintMaxQuarkSupply - tokenAccount.Amount, time.Now(), nil
+	return currencycreator.DefaultMintMaxQuarkSupply - tokenAccount.Amount, slot, time.Now(), nil
 }

ocp/data/currency/memory/store.go

@@ -16,13 +16,15 @@ const (
 )
 
 type store struct {
-	mu                    sync.Mutex
-	exchangeRateRecords   []*currency.ExchangeRateRecord
-	lastExchangeRateIndex uint64
-	metadataRecords       []*currency.MetadataRecord
-	lastMetadataIndex     uint64
-	reserveRecords        []*currency.ReserveRecord
-	lastReserveIndex      uint64
+	mu                         sync.Mutex
+	exchangeRateRecords        []*currency.ExchangeRateRecord
+	lastExchangeRateIndex      uint64
+	metadataRecords            []*currency.MetadataRecord
+	lastMetadataIndex          uint64
+	historicalReserveRecords   []*currency.ReserveRecord
+	lastHistoricalReserveIndex uint64
+	liveReserveRecords         map[string]*currency.ReserveRecord
+	lastLiveReserveIndex       uint64
 }
 
 type RateByTime []*currency.ExchangeRateRecord
@@ -47,6 +49,8 @@ func New() currency.Store {
 	return &store{
 		exchangeRateRecords:   make([]*currency.ExchangeRateRecord, 0),
 		lastExchangeRateIndex: 1,
+		liveReserveRecords:    make(map[string]*currency.ReserveRecord),
+		lastLiveReserveIndex:  1,
 	}
 }
 
@@ -56,8 +60,10 @@ func (s *store) reset() {
 	s.lastExchangeRateIndex = 1
 	s.metadataRecords = make([]*currency.MetadataRecord, 0)
 	s.lastMetadataIndex = 1
-	s.reserveRecords = make([]*currency.ReserveRecord, 0)
-	s.lastReserveIndex = 1
+	s.historicalReserveRecords = make([]*currency.ReserveRecord, 0)
+	s.lastHistoricalReserveIndex = 1
+	s.liveReserveRecords = make(map[string]*currency.ReserveRecord)
+	s.lastLiveReserveIndex = 1
 	s.mu.Unlock()
 }
 
@@ -306,7 +312,7 @@ func (s *store) CountMints(ctx context.Context) (uint64, error) {
 	return uint64(len(s.metadataRecords)), nil
 }
 
-func (s *store) PutReserveRecord(ctx context.Context, data *currency.ReserveRecord) error {
+func (s *store) PutHistoricalReserveRecord(ctx context.Context, data *currency.ReserveRecord) error {
 	if err := data.Validate(); err != nil {
 		return err
 	}
@@ -315,15 +321,15 @@ func (s *store) PutReserveRecord(ctx context.Context, data *currency.ReserveReco
 	defer s.mu.Unlock()
 
 	// Not ideal but fine for testing the currency store
-	for _, item := range s.reserveRecords {
+	for _, item := range s.historicalReserveRecords {
 		if item.Mint == data.Mint && item.Time.Unix() == data.Time.Unix() {
 			return currency.ErrExists
 		}
 	}
 
-	data.Id = s.lastReserveIndex
-	s.reserveRecords = append(s.reserveRecords, data.Clone())
-	s.lastReserveIndex = s.lastReserveIndex + 1
+	data.Id = s.lastHistoricalReserveIndex
+	s.historicalReserveRecords = append(s.historicalReserveRecords, data.Clone())
+	s.lastHistoricalReserveIndex = s.lastHistoricalReserveIndex + 1
 
 	return nil
 }
@@ -334,7 +340,7 @@ func (s *store) GetReserveAtTime(ctx context.Context, mint string, t time.Time)
 
 	// Not ideal but fine for testing the currency store
 	var results []*currency.ReserveRecord
-	for _, item := range s.reserveRecords {
+	for _, item := range s.historicalReserveRecords {
 		if item.Mint == mint && item.Time.Unix() <= t.Unix() && item.Time.Format(dateFormat) == t.Format(dateFormat) {
 			results = append(results, item)
 		}
@@ -353,11 +359,11 @@ func (s *store) GetReservesInRange(ctx context.Context, mint string, interval qu
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	sort.Sort(ReserveByTime(s.reserveRecords))
+	sort.Sort(ReserveByTime(s.historicalReserveRecords))
 
 	// Not ideal but fine for testing the currency store
 	var all []*currency.ReserveRecord
-	for _, item := range s.reserveRecords {
+	for _, item := range s.historicalReserveRecords {
 		if item.Mint == mint && item.Time.Unix() >= start.Unix() && item.Time.Unix() <= end.Unix() {
 			all = append(all, item.Clone())
 		}
@@ -377,3 +383,57 @@ func (s *store) GetReservesInRange(ctx context.Context, mint string, interval qu
 
 	return all, nil
 }
+
+func (s *store) PutLiveReserveRecord(ctx context.Context, data *currency.ReserveRecord) error {
+	if err := data.Validate(); err != nil {
+		return err
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if existing, ok := s.liveReserveRecords[data.Mint]; ok {
+		if data.Slot <= existing.Slot {
+			return currency.ErrStaleReserveState
+		}
+
+		cloned := data.Clone()
+		cloned.Id = existing.Id
+		s.liveReserveRecords[data.Mint] = cloned
+		cloned.CopyTo(data)
+		return nil
+	}
+
+	data.Id = s.lastLiveReserveIndex
+	s.liveReserveRecords[data.Mint] = data.Clone()
+	s.lastLiveReserveIndex++
+
+	return nil
+}
+
+func (s *store) GetLiveReserve(ctx context.Context, mint string) (*currency.ReserveRecord, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	record, ok := s.liveReserveRecords[mint]
+	if !ok {
+		return nil, currency.ErrNotFound
+	}
+
+	return record.Clone(), nil
+}
+
+func (s *store) GetAllLiveReserves(ctx context.Context) (map[string]*currency.ReserveRecord, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if len(s.liveReserveRecords) == 0 {
+		return nil, currency.ErrNotFound
+	}
+
+	res := make(map[string]*currency.ReserveRecord, len(s.liveReserveRecords))
+	for mint, record := range s.liveReserveRecords {
+		res[mint] = record.Clone()
+	}
+	return res, nil
+}

ocp/data/currency/model.go

@@ -266,6 +266,7 @@ type ReserveRecord struct {
 	Id                uint64
 	Mint              string
 	SupplyFromBonding uint64
+	Slot              uint64 // Not available for historical records
 	Time              time.Time
 }
 
@@ -286,6 +287,7 @@ func (m *ReserveRecord) Clone() *ReserveRecord {
 		Id:                m.Id,
 		Mint:              m.Mint,
 		SupplyFromBonding: m.SupplyFromBonding,
+		Slot:              m.Slot,
 		Time:              m.Time,
 	}
 }
@@ -294,6 +296,7 @@ func (m *ReserveRecord) CopyTo(dst *ReserveRecord) {
 	dst.Id = m.Id
 	dst.Mint = m.Mint
 	dst.SupplyFromBonding = m.SupplyFromBonding
+	dst.Slot = m.Slot
 	dst.Time = m.Time
 }
 

ocp/data/currency/postgres/model.go

@@ -19,6 +19,7 @@ const (
 	exchangeRateTableName = "ocp__core_exchangerate"
 	metadataTableName     = "ocp__core_currencymetadata"
 	reserveTableName      = "ocp__core_currencyreserve"
+	liveReserveTableName  = "ocp__core_currencyreserve2"
 
 	dateFormat = "2006-01-02"
 )
@@ -186,20 +187,20 @@ func fromMetadataModel(obj *metadataModel) *currency.MetadataRecord {
 	}
 }
 
-type reserveModel struct {
+type historicalReserveModel struct {
 	Id                sql.NullInt64 `db:"id"`
 	ForDate           string        `db:"for_date"`
 	ForTimestamp      time.Time     `db:"for_timestamp"`
 	Mint              string        `db:"mint"`
 	SupplyFromBonding uint64        `db:"supply_from_bonding"`
 }
 
-func toReserveModel(obj *currency.ReserveRecord) (*reserveModel, error) {
+func toHistoricalReserveModel(obj *currency.ReserveRecord) (*historicalReserveModel, error) {
 	if err := obj.Validate(); err != nil {
 		return nil, err
 	}
 
-	return &reserveModel{
+	return &historicalReserveModel{
 		Id:                sql.NullInt64{Int64: int64(obj.Id), Valid: obj.Id > 0},
 		ForDate:           obj.Time.UTC().Format(dateFormat),
 		ForTimestamp:      obj.Time.UTC(),
@@ -208,7 +209,7 @@ func toReserveModel(obj *currency.ReserveRecord) (*reserveModel, error) {
 	}, nil
 }
 
-func fromReserveModel(obj *reserveModel) *currency.ReserveRecord {
+func fromHistoricalReserveModel(obj *historicalReserveModel) *currency.ReserveRecord {
 	return &currency.ReserveRecord{
 		Id:                uint64(obj.Id.Int64),
 		Time:              obj.ForTimestamp.UTC(),
@@ -217,6 +218,34 @@ func fromReserveModel(obj *reserveModel) *currency.ReserveRecord {
 	}
 }
 
+type liveReserveModel struct {
+	Id                sql.NullInt64 `db:"id"`
+	Mint              string        `db:"mint"`
+	SupplyFromBonding uint64        `db:"supply_from_bonding"`
+	Slot              uint64        `db:"slot"`
+	LastUpdatedAt     time.Time     `db:"last_updated_at"`
+}
+
+func toLiveReserveModel(obj *currency.ReserveRecord) *liveReserveModel {
+	return &liveReserveModel{
+		Id:                sql.NullInt64{Int64: int64(obj.Id), Valid: obj.Id > 0},
+		Mint:              obj.Mint,
+		SupplyFromBonding: obj.SupplyFromBonding,
+		Slot:              obj.Slot,
+		LastUpdatedAt:     obj.Time.UTC(),
+	}
+}
+
+func fromLiveReserveModel(obj *liveReserveModel) *currency.ReserveRecord {
+	return &currency.ReserveRecord{
+		Id:                uint64(obj.Id.Int64),
+		Mint:              obj.Mint,
+		SupplyFromBonding: obj.SupplyFromBonding,
+		Slot:              obj.Slot,
+		Time:              obj.LastUpdatedAt.UTC(),
+	}
+}
+
 func marshalSocialLinks(links []currency.SocialLink) string {
 	if len(links) == 0 {
 		return "[]"
@@ -327,7 +356,7 @@ func (m *metadataModel) dbSave(ctx context.Context, db *sqlx.DB) error {
 	})
 }
 
-func (m *reserveModel) dbSave(ctx context.Context, db *sqlx.DB) error {
+func (m *historicalReserveModel) dbSave(ctx context.Context, db *sqlx.DB) error {
 	return pgutil.ExecuteInTx(ctx, db, sql.LevelDefault, func(tx *sqlx.Tx) error {
 		err := tx.QueryRowxContext(ctx,
 			`INSERT INTO `+reserveTableName+`
@@ -462,8 +491,8 @@ func dbCountMetadataByState(ctx context.Context, db *sqlx.DB, state currency.Met
 	return res, nil
 }
 
-func dbGetReserveByMintAndTime(ctx context.Context, db *sqlx.DB, mint string, t time.Time, ordering q.Ordering) (*reserveModel, error) {
-	res := &reserveModel{}
+func dbGetReserveByMintAndTime(ctx context.Context, db *sqlx.DB, mint string, t time.Time, ordering q.Ordering) (*historicalReserveModel, error) {
+	res := &historicalReserveModel{}
 	err := db.GetContext(ctx, res,
 		makeTimeBasedGetQuery(reserveTableName, "mint = $1 AND for_date = $2 AND for_timestamp <= $3", ordering),
 		mint,
@@ -473,8 +502,8 @@ func dbGetReserveByMintAndTime(ctx context.Context, db *sqlx.DB, mint string, t
 	return res, pgutil.CheckNoRows(err, currency.ErrNotFound)
 }
 
-func dbGetAllReservesForRange(ctx context.Context, db *sqlx.DB, mint string, interval q.Interval, start time.Time, end time.Time, ordering q.Ordering) ([]*reserveModel, error) {
-	res := []*reserveModel{}
+func dbGetAllReservesForRange(ctx context.Context, db *sqlx.DB, mint string, interval q.Interval, start time.Time, end time.Time, ordering q.Ordering) ([]*historicalReserveModel, error) {
+	res := []*historicalReserveModel{}
 	err := db.SelectContext(ctx, &res,
 		makeTimeBasedRangeQuery(reserveTableName, "mint = $1 AND for_timestamp >= $2 AND for_timestamp <= $3", ordering, interval),
 		mint, start.UTC(), end.UTC(),
@@ -489,3 +518,51 @@ func dbGetAllReservesForRange(ctx context.Context, db *sqlx.DB, mint string, int
 
 	return res, nil
 }
+
+func (m *liveReserveModel) dbSave(ctx context.Context, db *sqlx.DB) error {
+	return pgutil.ExecuteInTx(ctx, db, sql.LevelDefault, func(tx *sqlx.Tx) error {
+		err := tx.QueryRowxContext(ctx,
+			`INSERT INTO `+liveReserveTableName+`
+			(mint, supply_from_bonding, slot, last_updated_at)
+			VALUES ($1, $2, $3, $4)
+
+			ON CONFLICT (mint)
+			DO UPDATE SET supply_from_bonding = $2, slot = $3, last_updated_at = $4
+				WHERE `+liveReserveTableName+`.slot < $3
+
+			RETURNING id, mint, supply_from_bonding, slot, last_updated_at`,
+			m.Mint,
+			m.SupplyFromBonding,
+			m.Slot,
+			m.LastUpdatedAt,
+		).StructScan(m)
+
+		return pgutil.CheckNoRows(err, currency.ErrStaleReserveState)
+	})
+}
+
+func dbGetLiveReserveByMint(ctx context.Context, db *sqlx.DB, mint string) (*liveReserveModel, error) {
+	res := &liveReserveModel{}
+	err := db.GetContext(ctx, res,
+		`SELECT id, mint, supply_from_bonding, slot, last_updated_at
+		FROM `+liveReserveTableName+`
+		WHERE mint = $1`,
+		mint,
+	)
+	return res, pgutil.CheckNoRows(err, currency.ErrNotFound)
+}
+
+func dbGetAllLiveReserves(ctx context.Context, db *sqlx.DB) ([]*liveReserveModel, error) {
+	var res []*liveReserveModel
+	err := db.SelectContext(ctx, &res,
+		`SELECT id, mint, supply_from_bonding, slot, last_updated_at
+		FROM `+liveReserveTableName,
+	)
+	if err != nil {
+		return nil, pgutil.CheckNoRows(err, currency.ErrNotFound)
+	}
+	if len(res) == 0 {
+		return nil, currency.ErrNotFound
+	}
+	return res, nil
+}