diff --git a/go.mod b/go.mod index cfb0919..6556a3a 100644 --- a/go.mod +++ b/go.mod @@ -11,7 +11,7 @@ require ( github.com/aws/aws-sdk-go-v2/service/dynamodb v1.59.0 github.com/aws/aws-sdk-go-v2/service/s3 v1.102.2 github.com/code-payments/code-vm-indexer v1.2.0 - github.com/code-payments/ocp-protobuf-api v1.14.1-0.20260814155826-8088d9d58830 + github.com/code-payments/ocp-protobuf-api v1.14.1-0.20260818131607-042819236352 github.com/emirpasic/gods v1.12.0 github.com/envoyproxy/protoc-gen-validate v1.3.3 github.com/golang/protobuf v1.5.4 diff --git a/go.sum b/go.sum index 1a771c1..b05e79d 100644 --- a/go.sum +++ b/go.sum @@ -80,8 +80,8 @@ github.com/cockroachdb/apd v1.1.0 h1:3LFP3629v+1aKXU5Q37mxmRxX/pIu1nijXydLShEq5I github.com/cockroachdb/apd v1.1.0/go.mod h1:8Sl8LxpKi29FqWXR16WEFZRNSz3SoPzUzeMeY4+DwBQ= github.com/code-payments/code-vm-indexer v1.2.0 h1:rSHpBMiT9BKgmKcXg/VIoi/h0t7jNxGx07Qz59m+6Q0= github.com/code-payments/code-vm-indexer v1.2.0/go.mod h1:vn91YN2qNqb+gGJeZe2+l+TNxVmEEiRHXXnIn2Y40h8= -github.com/code-payments/ocp-protobuf-api v1.14.1-0.20260814155826-8088d9d58830 h1:PVX61XNEbm8iBKAUw6i+NN04qOX9bbenkpRWs0R6CCc= -github.com/code-payments/ocp-protobuf-api v1.14.1-0.20260814155826-8088d9d58830/go.mod h1:tw6BooY5a8l6CtSZnKOruyKII0W04n89pcM4BizrgG8= +github.com/code-payments/ocp-protobuf-api v1.14.1-0.20260818131607-042819236352 h1:RsDDYM1VXBNcEYSZ+GvnMXCua9pMckYdIwdJnmiX+FI= +github.com/code-payments/ocp-protobuf-api v1.14.1-0.20260818131607-042819236352/go.mod h1:tw6BooY5a8l6CtSZnKOruyKII0W04n89pcM4BizrgG8= github.com/containerd/continuity v0.0.0-20190827140505-75bee3e2ccb6 h1:NmTXa/uVnDyp0TY5MKi197+3HWcnYWfnHGyaFthlnGw= github.com/containerd/continuity v0.0.0-20190827140505-75bee3e2ccb6/go.mod h1:GL3xCUCBDV3CZiTSEKksMWbLE66hEyuu9qyDOOqM47Y= github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk= diff --git a/ocp/currency/data_provider.go b/ocp/currency/data_provider.go index ce9dad2..2ac4898 100644 --- a/ocp/currency/data_provider.go +++ b/ocp/currency/data_provider.go @@ -50,6 +50,13 @@ type LiveHolderCountData struct { LastWeekDelta int64 } +// LiveMarketCapData represents live market cap data for a currency +type LiveMarketCapData struct { + Mint *common.Account + CurrentMarketCap float64 + LastWeekDelta float64 +} + type cachedProtoMint struct { mint *currencypb.Mint lastUpdatedAt time.Time @@ -77,6 +84,7 @@ type MintDataProvider struct { exchangeRates *LiveExchangeRateData launchpadReserves map[string]*LiveReserveStateData holderCounts map[string]*LiveHolderCountData + marketCaps map[string]*LiveMarketCapData streamsMu sync.RWMutex streams map[string]*LiveMintDataStream @@ -131,6 +139,7 @@ func NewMintDataProvider( currencyMetadata: make(map[string]*currency.MetadataRecord), launchpadReserves: make(map[string]*LiveReserveStateData), holderCounts: make(map[string]*LiveHolderCountData), + marketCaps: make(map[string]*LiveMarketCapData), streams: make(map[string]*LiveMintDataStream), exchangeRatesReady: make(chan struct{}), reserveStatesReady: make(chan struct{}), @@ -176,7 +185,7 @@ func (m *MintDataProvider) Stop() { func (m *MintDataProvider) ToProtoMint( ctx context.Context, metadataRecord *currency.MetadataRecord, - includeLiveReserveState, includeLiveHolderMetrics bool, + includeLiveReserveState, includeLiveHolderMetrics, includeLiveMarketCapMetrics bool, ) (*currencypb.Mint, error) { mint, err := common.NewAccountFromPublicKeyString(metadataRecord.Mint) if err != nil { @@ -278,6 +287,13 @@ func (m *MintDataProvider) ToProtoMint( } } + if includeLiveMarketCapMetrics { + err = m.InjectLiveMarketCapMetrics(ctx, protoMint) + if err != nil { + return nil, err + } + } + return protoMint, nil } @@ -346,6 +362,39 @@ func SetHolderMetrics(protoMint *currencypb.Mint, holderData *LiveHolderCountDat } } +func (m *MintDataProvider) InjectLiveMarketCapMetrics(ctx context.Context, protoMint *currencypb.Mint) error { + if protoMint.LaunchpadMetadata == nil { + return errors.New("only launchpad currencies supported for market cap") + } + + mint, err := common.NewAccountFromProto(protoMint.Address) + if err != nil { + return errors.New("invalid proto mint") + } + + marketCapData, err := m.GetLiveMarketCap(ctx, mint) + if err != nil { + return err + } + + SetMarketCapMetrics(protoMint, marketCapData) + return nil +} + +// SetMarketCapMetrics applies market cap data to a proto Mint without fetching +// from the provider. +func SetMarketCapMetrics(protoMint *currencypb.Mint, marketCapData *LiveMarketCapData) { + protoMint.MarketCapMetrics = ¤cypb.MarketCapMetrics{ + CurrentMarketCap: marketCapData.CurrentMarketCap, + MarketCapDeltas: []*currencypb.MarketCapMetrics_DeltaMarketCap{ + { + Range: currencypb.PredefinedRange_LAST_WEEK, + Delta: marketCapData.LastWeekDelta, + }, + }, + } +} + // GetProtoMint gets a proto Mint object. Static and infrequently updated metadata is // heavily cached. func (m *MintDataProvider) GetProtoMint(ctx context.Context, mint *common.Account) (*currencypb.Mint, error) { @@ -411,7 +460,7 @@ func (m *MintDataProvider) GetProtoMint(ctx context.Context, mint *common.Accoun return nil, currency.ErrNotFound } - protoMetadata, err = m.ToProtoMint(ctx, metadataRecord, true, false) + protoMetadata, err = m.ToProtoMint(ctx, metadataRecord, true, false, false) if err != nil { return nil, err } @@ -538,6 +587,24 @@ func (m *MintDataProvider) GetAllCachedHolderCounts(ctx context.Context) (map[st return out, nil } +// GetAllCachedMarketCaps returns a snapshot of all currently cached market cap +// data keyed by mint address. It blocks until the first successful poll has +// completed. +func (m *MintDataProvider) GetAllCachedMarketCaps(ctx context.Context) (map[string]*LiveMarketCapData, error) { + if err := m.waitForReserveStates(ctx); err != nil { + return nil, err + } + + m.stateMu.RLock() + defer m.stateMu.RUnlock() + + out := make(map[string]*LiveMarketCapData, len(m.marketCaps)) + for k, v := range m.marketCaps { + out[k] = v + } + return out, nil +} + // GetLiveReserveState returns a current pre-signed live launchpad currency reserve state for a mint func (m *MintDataProvider) GetLiveReserveState(ctx context.Context, mint *common.Account) (*LiveReserveStateData, error) { m.stateMu.RLock() @@ -571,6 +638,40 @@ func (m *MintDataProvider) GetLiveReserveState(ctx context.Context, mint *common return data, nil } +// GetLiveMarketCap returns live market cap data for a mint, blocking until the +// reserve poller has retrieved the state it's derived from. +func (m *MintDataProvider) GetLiveMarketCap(ctx context.Context, mint *common.Account) (*LiveMarketCapData, error) { + m.stateMu.RLock() + data, ok := m.marketCaps[mint.PublicKey().ToBase58()] + m.stateMu.RUnlock() + + if !ok { + isSupported, err := common.IsSupportedMint(ctx, m.data, mint) + if err != nil { + return nil, err + } + if !isSupported { + return nil, common.ErrUnsupportedMint + } + } else { + return data, nil + } + + err := m.waitForReserveState(ctx, mint) + if err != nil { + return nil, err + } + + m.stateMu.RLock() + defer m.stateMu.RUnlock() + + data, ok = m.marketCaps[mint.PublicKey().ToBase58()] + if !ok { + return nil, errors.New("not found") + } + return data, nil +} + // waitForExchangeRates blocks until exchange rate data is available or context is cancelled func (m *MintDataProvider) waitForExchangeRates(ctx context.Context) error { select { @@ -937,6 +1038,23 @@ func (m *MintDataProvider) fetchAndUpdateReserveStates(ctx context.Context) { return } + mints := make([]string, 0, len(liveReserves)) + for mintAddr := range liveReserves { + mints = append(mints, mintAddr) + } + + // Market cap is derived from supply, so the weekly delta comes from the supply + // held a week ago. + var includeWeeklyDeltas bool + oneWeekAgo := time.Now().Add(-7 * 24 * time.Hour) + endOfWeekAgoDay := time.Date(oneWeekAgo.Year(), oneWeekAgo.Month(), oneWeekAgo.Day(), 23, 59, 59, 0, time.UTC) + historicalReserves, err := m.reserveStore.GetReservesForDay(ctx, mints, endOfWeekAgoDay) + if err != nil && err != currency.ErrNotFound { + m.log.With(zap.Error(err)).Warn("failed to fetch historical reserves for weekly market cap delta") + } else { + includeWeeklyDeltas = true + } + var updatedStates []*LiveReserveStateData for mintAddr, record := range liveReserves { mint, err := common.NewAccountFromPublicKeyString(mintAddr) @@ -963,8 +1081,25 @@ func (m *MintDataProvider) fetchAndUpdateReserveStates(ctx context.Context) { SignedState: signedState, } + var pastSupplyFromBonding uint64 + if historicalReserves != nil { + if pastRecord, ok := historicalReserves[mintAddr]; ok { + pastSupplyFromBonding = pastRecord.SupplyFromBonding + } + } + + currentMarketCap := CalculateMarketCap(record.SupplyFromBonding, 1.0) + marketCapData := &LiveMarketCapData{ + Mint: mint, + CurrentMarketCap: currentMarketCap, + } + if includeWeeklyDeltas { + marketCapData.LastWeekDelta = currentMarketCap - CalculateMarketCap(pastSupplyFromBonding, 1.0) + } + m.stateMu.Lock() m.launchpadReserves[mintAddr] = stateData + m.marketCaps[mintAddr] = marketCapData m.stateMu.Unlock() m.markReserveStateReady(mint) diff --git a/ocp/data/currency/reserve/cache/store.go b/ocp/data/currency/reserve/cache/store.go index 5b64844..5428627 100644 --- a/ocp/data/currency/reserve/cache/store.go +++ b/ocp/data/currency/reserve/cache/store.go @@ -2,7 +2,7 @@ // reserve lookups in front of a wrapped store. // // Point-in-time reads are keyed by mint and a coarse time bucket that doubles as -// the freshness window. Range and live reads pass straight through. Live writes +// the freshness window. Range, day and live reads pass straight through. Live writes // are guarded against the last successfully saved slot per mint: a write whose // slot is not greater is rejected with currency.ErrStaleReserveState without a // round-trip to the backing store. Everything else passes straight through. @@ -75,6 +75,10 @@ func (s *store) GetReserveAtTime(ctx context.Context, mint string, t time.Time) return record, nil } +func (s *store) GetReservesForDay(ctx context.Context, mints []string, t time.Time) (map[string]*currency.ReserveRecord, error) { + return s.backing.GetReservesForDay(ctx, mints, t) +} + func (s *store) GetReservesInRange(ctx context.Context, mint string, interval query.Interval, start time.Time, end time.Time, ordering query.Ordering) ([]*currency.ReserveRecord, error) { return s.backing.GetReservesInRange(ctx, mint, interval, start, end, ordering) } diff --git a/ocp/data/currency/reserve/dynamodb/store.go b/ocp/data/currency/reserve/dynamodb/store.go index d9c9c70..3328f46 100644 --- a/ocp/data/currency/reserve/dynamodb/store.go +++ b/ocp/data/currency/reserve/dynamodb/store.go @@ -33,6 +33,7 @@ import ( "errors" "fmt" "strconv" + "strings" "time" "github.com/aws/aws-sdk-go-v2/aws" @@ -60,6 +61,9 @@ const ( // codeConditionalCheckFailed is the DynamoDB cancellation reason code for a // transaction item whose ConditionExpression evaluated false. codeConditionalCheckFailed = "ConditionalCheckFailed" + + // maxBatchGetItems is DynamoDB's per-call BatchGetItem key limit. + maxBatchGetItems = 100 ) // rollupResolutions are the coarse resolutions maintained alongside the raw @@ -163,6 +167,64 @@ func (s *store) GetReserveAtTime(ctx context.Context, mint string, t time.Time) return historyRecord(mint, out.Items[0]) } +// GetReservesForDay returns each mint's reserve state as of the UTC day of t — +// the close of that mint's day rollup bucket. Because a rollup bucket has a +// deterministic key (mint#day, bucketStart), this is a single batched key get +// rather than a per-mint query. Mints with no record on that day are omitted. +func (s *store) GetReservesForDay(ctx context.Context, mints []string, t time.Time) (map[string]*currency.ReserveRecord, error) { + bucket := skN(bucketStart(t, resDay)) + + // Dedup so a repeated mint isn't fetched twice. + seen := make(map[string]struct{}, len(mints)) + keys := make([]map[string]types.AttributeValue, 0, len(mints)) + for _, mint := range mints { + if _, ok := seen[mint]; ok { + continue + } + seen[mint] = struct{}{} + keys = append(keys, map[string]types.AttributeValue{ + attrPK: avS(historyPK(mint, resDay)), + attrSK: bucket, + }) + } + + res := make(map[string]*currency.ReserveRecord, len(keys)) + for start := 0; start < len(keys); start += maxBatchGetItems { + end := start + maxBatchGetItems + if end > len(keys) { + end = len(keys) + } + + req := map[string]types.KeysAndAttributes{ + s.historyTable: {Keys: keys[start:end]}, + } + // Drain UnprocessedKeys (DynamoDB may return a partial batch under load). + for len(req[s.historyTable].Keys) > 0 { + out, err := s.client.BatchGetItem(ctx, &dynamodb.BatchGetItemInput{RequestItems: req}) + if err != nil { + return nil, err + } + for _, item := range out.Responses[s.historyTable] { + mint, err := mintFromDayPK(item) + if err != nil { + return nil, err + } + rec, err := historyRecord(mint, item) + if err != nil { + return nil, err + } + res[mint] = rec + } + if unprocessed, ok := out.UnprocessedKeys[s.historyTable]; ok && len(unprocessed.Keys) > 0 { + req = map[string]types.KeysAndAttributes{s.historyTable: unprocessed} + } else { + break + } + } + } + return res, nil +} + func (s *store) GetReservesInRange(ctx context.Context, mint string, interval query.Interval, start time.Time, end time.Time, ordering query.Ordering) ([]*currency.ReserveRecord, error) { if interval > query.IntervalMonth { return nil, currency.ErrInvalidInterval @@ -412,6 +474,18 @@ func bucketStart(t time.Time, res string) time.Time { func historyPK(mint, res string) string { return mint + "#" + res } +// mintFromDayPK recovers the mint from a day-rollup item's pk ("#day"). +// Mints are base58, which never contains '#', so the resolution suffix is +// unambiguous. +func mintFromDayPK(item map[string]types.AttributeValue) (string, error) { + pk := asS(item[attrPK]) + suffix := "#" + resDay + if !strings.HasSuffix(pk, suffix) { + return "", fmt.Errorf("unexpected day-rollup pk %q", pk) + } + return strings.TrimSuffix(pk, suffix), nil +} + // skN encodes t's unix-nanos as the numeric sort key, which sorts in // chronological order. func skN(t time.Time) types.AttributeValue { return avN(t.UTC().UnixNano()) } diff --git a/ocp/data/currency/reserve/memory/store.go b/ocp/data/currency/reserve/memory/store.go index 719840a..60e6a29 100644 --- a/ocp/data/currency/reserve/memory/store.go +++ b/ocp/data/currency/reserve/memory/store.go @@ -87,6 +87,35 @@ func (s *store) GetReserveAtTime(ctx context.Context, mint string, t time.Time) return results[0].Clone(), nil } +func (s *store) GetReservesForDay(ctx context.Context, mints []string, t time.Time) (map[string]*currency.ReserveRecord, error) { + s.mu.Lock() + defer s.mu.Unlock() + + res := make(map[string]*currency.ReserveRecord, len(mints)) + for _, mint := range mints { + // The close of t's UTC day for this mint: its most recent record that day. + var latest *currency.ReserveRecord + for _, item := range s.historical { + if item.Mint != mint || !sameUTCDay(item.Time, t) { + continue + } + if latest == nil || item.Time.After(latest.Time) { + latest = item + } + } + if latest != nil { + res[mint] = latest.Clone() + } + } + return res, nil +} + +func sameUTCDay(a, b time.Time) bool { + ay, am, ad := a.UTC().Date() + by, bm, bd := b.UTC().Date() + return ay == by && am == bm && ad == bd +} + func (s *store) GetReservesInRange(ctx context.Context, mint string, interval query.Interval, start time.Time, end time.Time, ordering query.Ordering) ([]*currency.ReserveRecord, error) { if interval > query.IntervalMonth { return nil, currency.ErrInvalidInterval diff --git a/ocp/data/currency/reserve/store.go b/ocp/data/currency/reserve/store.go index 33fb821..f1e003c 100644 --- a/ocp/data/currency/reserve/store.go +++ b/ocp/data/currency/reserve/store.go @@ -29,6 +29,16 @@ type Store interface { // provided time. GetReserveAtTime(ctx context.Context, mint string, t time.Time) (*currency.ReserveRecord, error) + // GetReservesForDay gets the reserve state for each of the given currency + // creator mints as of the UTC day of t — the close of that day (the mint's most + // recent record within the day), keyed by mint. Mints with no record on that day + // are omitted from the result rather than reported as an error. + // + // Unlike GetReserveAtTime this is day-granularity: it does not fall back to an + // earlier day, and for a mid-day t the returned record may be later than t (the + // day's close). It is served as a single batched key get against the day rollups. + GetReservesForDay(ctx context.Context, mints []string, t time.Time) (map[string]*currency.ReserveRecord, error) + // GetReservesInRange gets the reserve records for a range of time given a // currency creator mint and interval. // diff --git a/ocp/data/currency/reserve/tests/tests.go b/ocp/data/currency/reserve/tests/tests.go index 99fcc9a..44b8fba 100644 --- a/ocp/data/currency/reserve/tests/tests.go +++ b/ocp/data/currency/reserve/tests/tests.go @@ -18,6 +18,7 @@ import ( func RunStoreTests(t *testing.T, s reserve.Store, teardown func()) { for _, tf := range []func(t *testing.T, s reserve.Store){ testReserveRoundTrip, + testGetReservesForDay, testGetReservesInRange, testLiveReserveRoundTrip, testGetAllLiveReserves, @@ -73,6 +74,43 @@ func testReserveRoundTrip(t *testing.T, s reserve.Store) { assert.Equal(t, currency.ErrNotFound, err) } +func testGetReservesForDay(t *testing.T, s reserve.Store) { + ctx := context.Background() + day := time.Date(2022, 05, 10, 0, 0, 0, 0, time.UTC) + + // mintA: a prior-day record plus two on `day` (close = 800 at 20:00). + require.NoError(t, s.PutHistoricalReserve(ctx, ¤cy.ReserveRecord{Mint: "mintA", SupplyFromBonding: 100, Time: day.Add(-15 * time.Hour)})) + require.NoError(t, s.PutHistoricalReserve(ctx, ¤cy.ReserveRecord{Mint: "mintA", SupplyFromBonding: 500, Time: day.Add(10 * time.Hour)})) + require.NoError(t, s.PutHistoricalReserve(ctx, ¤cy.ReserveRecord{Mint: "mintA", SupplyFromBonding: 800, Time: day.Add(20 * time.Hour)})) + // mintB: a single record on `day`. + require.NoError(t, s.PutHistoricalReserve(ctx, ¤cy.ReserveRecord{Mint: "mintB", SupplyFromBonding: 10000, Time: day.Add(12 * time.Hour)})) + // mintC: a record only on the next day — should be omitted for a `day` query. + require.NoError(t, s.PutHistoricalReserve(ctx, ¤cy.ReserveRecord{Mint: "mintC", SupplyFromBonding: 5000, Time: day.AddDate(0, 0, 1).Add(8 * time.Hour)})) + + queryT := time.Date(2022, 05, 10, 23, 59, 59, 0, time.UTC) + res, err := s.GetReservesForDay(ctx, []string{"mintA", "mintB", "mintC", "mintD"}, queryT) + require.NoError(t, err) + require.Len(t, res, 2) + + // mintA: close of the day = 800 at 20:00. + require.Contains(t, res, "mintA") + assert.EqualValues(t, 800, res["mintA"].SupplyFromBonding) + assert.Equal(t, day.Add(20*time.Hour).Unix(), res["mintA"].Time.Unix()) + + // mintB: its single same-day record. + require.Contains(t, res, "mintB") + assert.EqualValues(t, 10000, res["mintB"].SupplyFromBonding) + + // mintC has no record on `day`; mintD has none at all — both omitted. + assert.NotContains(t, res, "mintC") + assert.NotContains(t, res, "mintD") + + // Empty input yields an empty map, not an error. + empty, err := s.GetReservesForDay(ctx, nil, queryT) + require.NoError(t, err) + assert.Empty(t, empty) +} + func testGetReservesInRange(t *testing.T, s reserve.Store) { var reserves []currency.ReserveRecord diff --git a/ocp/rpc/currency/discovery.go b/ocp/rpc/currency/discovery.go index 9b61fbf..94a5c7f 100644 --- a/ocp/rpc/currency/discovery.go +++ b/ocp/rpc/currency/discovery.go @@ -72,11 +72,17 @@ func (s *currencyServer) Discover(req *currencypb.DiscoverRequest, stream curren log.With(zap.Error(err)).Warn("failure getting cached holder counts") return status.Error(codes.Internal, "") } + cachedMarketCaps, err := s.mintDataProvider.GetAllCachedMarketCaps(ctx) + if err != nil { + log.With(zap.Error(err)).Warn("failure getting cached market caps") + return status.Error(codes.Internal, "") + } type discoveredMint struct { - record *currency.MetadataRecord - reserveState *ocp_currency.LiveReserveStateData - holderData *ocp_currency.LiveHolderCountData + record *currency.MetadataRecord + reserveState *ocp_currency.LiveReserveStateData + holderData *ocp_currency.LiveHolderCountData + marketCapData *ocp_currency.LiveMarketCapData } var candidates []*discoveredMint @@ -99,14 +105,20 @@ func (s *currencyServer) Discover(req *currencypb.DiscoverRequest, stream curren continue } + marketCapData, ok := cachedMarketCaps[record.Mint] + if !ok { + continue + } + if !categoryFilterFunc(record) { continue } candidates = append(candidates, &discoveredMint{ - record: record, - reserveState: reserveState, - holderData: holderData, + record: record, + reserveState: reserveState, + holderData: holderData, + marketCapData: marketCapData, }) } @@ -128,7 +140,7 @@ func (s *currencyServer) Discover(req *currencypb.DiscoverRequest, stream curren for _, candidate := range candidates { log := log.With(zap.String("mint", candidate.record.Mint)) - protoMint, err := s.mintDataProvider.ToProtoMint(ctx, candidate.record, false, false) + protoMint, err := s.mintDataProvider.ToProtoMint(ctx, candidate.record, false, false, false) if err != nil { log.With(zap.Error(err)).Warn("failure converting metadata to proto mint") continue @@ -136,6 +148,7 @@ func (s *currencyServer) Discover(req *currencypb.DiscoverRequest, stream curren ocp_currency.SetLaunchpadReserveData(protoMint, candidate.reserveState) ocp_currency.SetHolderMetrics(protoMint, candidate.holderData) + ocp_currency.SetMarketCapMetrics(protoMint, candidate.marketCapData) protoMints = append(protoMints, protoMint) }