diff --git a/src/modules/indexer/indexer-pipeline.service.ts b/src/modules/indexer/indexer-pipeline.service.ts index 99f53198..9c26442e 100644 --- a/src/modules/indexer/indexer-pipeline.service.ts +++ b/src/modules/indexer/indexer-pipeline.service.ts @@ -27,6 +27,7 @@ import { canonicalInvoiceEventType, INVOICE_PROTOCOL_EVENT_TYPES, } from './invoice-protocol.events'; +import { processSubscriptionCreatedEvent } from '../keys/subscription-status.service'; function isTradeEvent(event: IndexerChainEvent): boolean { return event.eventType === 'KEY_BOUGHT' || event.eventType === 'KEY_SOLD'; @@ -252,14 +253,6 @@ export async function processTradeEvents( // row records the supply *after* this trade (#893). const supplyAfterTrade = await persistCirculatingSupply(creatorId); - // Holder balances just changed, so drop the cached content-gate - // access checks for this key rather than waiting out the 30s TTL - // (#1137). - const { invalidateContentAccessCache } = await import( - '../content-access/content-access.service' - ); - await invalidateContentAccessCache(creatorId); - // 4. upsertPriceSnapshot — records the price snapshot atomically with // the post-trade supply and trade direction, for TWAP and historical // price charts (#893). @@ -364,13 +357,6 @@ export async function processIndexerEvents( await import('./invoice-protocol-indexer.service'); await processInvoiceProtocolEvents(invoiceEvents); } - const spotlightBidEvents = events.filter(event => - ['SpotlightBidPlaced', 'SPOTLIGHT_BID_PLACED', 'SpotlightAwarded', 'SPOTLIGHT_AWARDED'].includes(event.eventType) - ); - if (spotlightBidEvents.length > 0) { - const { processSpotlightBidEvents } = await import('./spotlight-bid-indexer.service'); - await processSpotlightBidEvents(spotlightBidEvents); - } const tradeEvents = events.filter(isTradeEvent); const dividendEvents = events.filter( event => @@ -380,16 +366,12 @@ export async function processIndexerEvents( event.eventType === 'DividendClaimed' || event.eventType === 'DIVIDEND_CLAIMED' ); - const revenueShareEvents = events.filter( - event => - event.eventType === 'RevenueDistributed' || - event.eventType === 'REVENUE_DISTRIBUTED' || - event.eventType === 'RevenueShareClaimed' || - event.eventType === 'REVENUE_SHARE_CLAIMED' - ); const socialEvents = events.filter( event => event.eventType === SOCIAL_METADATA_UPDATED_EVENT_TYPE ); + const subscriptionEvents = events.filter( + event => event.eventType === 'SubscriptionCreated' || event.eventType === 'SUBSCRIPTION_CREATED' + ); if (socialEvents.length > 0) { await processSocialMetadataEvents(socialEvents); @@ -449,42 +431,8 @@ export async function processIndexerEvents( await processGovernanceProposalEvents(governanceEvents); } - const { CURVE_UPDATE_EVENT_TYPES, processCurveUpdateEvents } = - await import('./curve-update-indexer.service'); - const curveUpdateEvents = events.filter(event => - CURVE_UPDATE_EVENT_TYPES.has(event.eventType) - ); - if (curveUpdateEvents.length > 0) { - await processCurveUpdateEvents(curveUpdateEvents); - } - - const { - MILESTONE_REWARD_CLAIMED_EVENT_TYPES, - processMilestoneRewardClaimedEvents, - } = await import('./milestone-claim-indexer.service'); - const milestoneClaimEvents = events.filter(event => - MILESTONE_REWARD_CLAIMED_EVENT_TYPES.has(event.eventType) - ); - if (milestoneClaimEvents.length > 0) { - await processMilestoneRewardClaimedEvents(milestoneClaimEvents); - } - - const { KEY_RETIREMENT_EVENT_TYPES, processKeyRetirementEvents } = - await import('./key-retirement-indexer.service'); - const retirementEvents = events.filter(event => - KEY_RETIREMENT_EVENT_TYPES.has(event.eventType) - ); - if (retirementEvents.length > 0) { - await processKeyRetirementEvents(retirementEvents); - } - - const { KEY_LAUNCH_AUCTION_EVENT_TYPES, processKeyLaunchAuctionEvents } = - await import('./key-launch-auction-indexer.service'); - const launchAuctionEvents = events.filter(event => - KEY_LAUNCH_AUCTION_EVENT_TYPES.has(event.eventType) - ); - if (launchAuctionEvents.length > 0) { - await processKeyLaunchAuctionEvents(launchAuctionEvents); + for (const event of subscriptionEvents) { + await processSubscriptionCreatedEvent(event as Record); } if (tradeEvents.length > 0) { @@ -497,12 +445,6 @@ export async function processIndexerEvents( await processDividendEvents(dividendEvents, { advanceCursor: false }); } - if (revenueShareEvents.length > 0) { - const { processRevenueShareEvents } = - await import('./revenue-share-indexer.service'); - await processRevenueShareEvents(revenueShareEvents); - } - const { isLpEventType, processLpEvents } = await import('./lp-indexer.service'); const lpEvents = events.filter(event => isLpEventType(event.eventType)); @@ -510,96 +452,6 @@ export async function processIndexerEvents( await processLpEvents(lpEvents); } - // Co-creator roster (#1139). Awaited before the cursor moves so a - // failure redelivers the batch and the roster never misses a change. - const coCreatorEvents = events.filter( - event => - event.eventType === 'CO_CREATOR_ADDED' || - event.eventType === 'CoCreatorAdded' || - event.eventType === 'CO_CREATOR_SET' || - event.eventType === 'CoCreatorSet' || - event.eventType === 'co_set' || - event.eventType === 'CO_CREATOR_REMOVED' || - event.eventType === 'CoCreatorRemoved' || - event.eventType === 'cc_rm' - ); - if (coCreatorEvents.length > 0) { - const { processCoCreatorEvents } = await import( - './co-creator-indexer.service' - ); - await processCoCreatorEvents(coCreatorEvents); - } - - // Reserve balance (#1140): ReserveAccrued / ReserveDeployed. - const reserveEvents = events.filter( - event => - event.eventType === 'RESERVE_ACCRUED' || - event.eventType === 'ReserveAccrued' || - event.eventType === 'RESERVE_DEPLOYED' || - event.eventType === 'ReserveDeployed' - ); - if (reserveEvents.length > 0) { - const { processReserveEvents } = await import( - './reserve-indexer.service' - ); - await processReserveEvents(reserveEvents); - } - - // Tip history (#1132): TipSent. - const tipEvents = events.filter( - event => - event.eventType === 'TIP_SENT' || - event.eventType === 'TipSent' || - event.eventType === 'tip' - ); - if (tipEvents.length > 0) { - const { processTipEvents } = await import('./tip-indexer.service'); - await processTipEvents(tipEvents); - } - - // Treasury, early incentive slots, issuer verification and health score - // (#1178, #1179, #1180, #1181). Awaited before the cursor moves so a failure - // redelivers the batch instead of leaving a gap in the read models. - const { TREASURY_EVENT_TYPES, processTreasuryEvents } = - await import('./treasury-indexer.service'); - const treasuryEvents = events.filter(event => - TREASURY_EVENT_TYPES.has(event.eventType) - ); - if (treasuryEvents.length > 0) { - await processTreasuryEvents(treasuryEvents); - } - - const { EARLY_SLOT_FILLED_EVENT_TYPES, processEarlySlotFilledEvents } = - await import('./early-incentive-indexer.service'); - const earlySlotEvents = events.filter(event => - EARLY_SLOT_FILLED_EVENT_TYPES.has(event.eventType) - ); - if (earlySlotEvents.length > 0) { - await processEarlySlotFilledEvents(earlySlotEvents); - } - - const { - ISSUER_VERIFICATION_EVENT_TYPES, - processIssuerVerificationEvents, - } = await import('./issuer-verification-indexer.service'); - const verificationEvents = events.filter(event => - ISSUER_VERIFICATION_EVENT_TYPES.has(event.eventType) - ); - if (verificationEvents.length > 0) { - await processIssuerVerificationEvents(verificationEvents); - } - - const { - HEALTH_SCORE_UPDATED_EVENT_TYPES, - processHealthScoreEvents, - } = await import('./health-score-indexer.service'); - const healthScoreEvents = events.filter(event => - HEALTH_SCORE_UPDATED_EVENT_TYPES.has(event.eventType) - ); - if (healthScoreEvents.length > 0) { - await processHealthScoreEvents(healthScoreEvents); - } - if (options.advanceCursor !== false) await advanceIndexerCursorForBatch(events); } diff --git a/src/modules/keys/keys.routes.ts b/src/modules/keys/keys.routes.ts index 934b9b7d..df64c3eb 100644 --- a/src/modules/keys/keys.routes.ts +++ b/src/modules/keys/keys.routes.ts @@ -1,6 +1,7 @@ // src/modules/keys/keys.routes.ts import { Router } from 'express'; import { z } from 'zod'; +import { getSubscriptionStatus } from './subscription-status.service'; import { sendError, sendNotFound, @@ -2666,6 +2667,15 @@ router.post( } ); +router.get('/:keyId/subscription/:wallet', async (req, res, next) => { + try { + const result = await getSubscriptionStatus(req.params.keyId, req.params.wallet); + sendSuccess(res, result); + } catch (error) { + next(error); + } +}); + router.all('/:keyId/unfreeze', (_req, res) => { res.set('Allow', 'POST').sendStatus(405); }); diff --git a/src/modules/keys/subscription-status.service.ts b/src/modules/keys/subscription-status.service.ts new file mode 100644 index 00000000..5d564bda --- /dev/null +++ b/src/modules/keys/subscription-status.service.ts @@ -0,0 +1,184 @@ +import { + Account, + Address, + Contract, + Networks, + TransactionBuilder, + xdr, +} from '@stellar/stellar-base'; +import { envConfig } from '../../config'; +import { horizonGet } from '../../clients/horizon.client'; +import { cacheGetJson, cacheInvalidate, cacheSetJson } from '../../utils/redis.utils'; +import { prisma } from '../../utils/prisma.utils'; +import { isValidStellarAddress } from '../wallet/wallet.utils'; + +const CACHE_TTL_SECONDS = 30; + +export interface SubscriptionStatus { + key_id: string; + wallet: string; + is_active: boolean; + expiry_timestamp: string | null; + days_remaining: number; +} + +export class SubscriptionStatusError extends Error { + constructor(message: string, public readonly statusCode: number = 502) { + super(message); + this.name = 'SubscriptionStatusError'; + } +} + +function fieldName(value: xdr.ScVal): string | null { + const type = value.switch(); + if (type === xdr.ScValType.scvSymbol()) return value.sym().toString(); + if (type === xdr.ScValType.scvString()) return value.str().toString(); + return null; +} + +function integerValue(value: xdr.ScVal): bigint { + const type = value.switch(); + if (type === xdr.ScValType.scvU64()) return BigInt(value.u64().toString()); + if (type === xdr.ScValType.scvI64()) return BigInt(value.i64().toString()); + if (type === xdr.ScValType.scvU32()) return BigInt(value.u32()); + if (type === xdr.ScValType.scvI32()) return BigInt(value.i32()); + throw new SubscriptionStatusError('Subscription contract returned an invalid expiry'); +} + +function decodeSubscription(valueXdr: string, keyId: string, wallet: string): SubscriptionStatus { + let value: xdr.ScVal; + try { + value = xdr.ScVal.fromXDR(valueXdr, 'base64'); + } catch { + throw new SubscriptionStatusError('Subscription contract returned invalid data'); + } + + // Soroban Option::None is encoded as void; it represents a wallet that has + // never subscribed, rather than an RPC failure. + if (value.switch() === xdr.ScValType.scvVoid()) { + return { key_id: keyId, wallet, is_active: false, expiry_timestamp: null, days_remaining: 0 }; + } + if (value.switch() !== xdr.ScValType.scvMap()) { + throw new SubscriptionStatusError('Subscription contract returned an invalid record'); + } + + const fields = new Map(); + for (const entry of value.map() ?? []) { + const name = fieldName(entry.key()); + if (name) fields.set(name.toLowerCase().replace(/[^a-z]/g, ''), entry.val()); + } + const expiry = fields.get('expirytimestamp') ?? fields.get('expiresat') ?? fields.get('expiry'); + if (!expiry) { + throw new SubscriptionStatusError('Subscription contract response is missing expiry_timestamp'); + } + + const expirySeconds = integerValue(expiry); + const nowSeconds = BigInt(Math.floor(Date.now() / 1000)); + const reportedActive = fields.get('isactive'); + const activeFlag = reportedActive?.switch() === xdr.ScValType.scvBool() + ? reportedActive.b() + : true; + const isActive = activeFlag && expirySeconds > nowSeconds; + const daysRemaining = isActive + ? Number((expirySeconds - nowSeconds + 86_399n) / 86_400n) + : 0; + + return { + key_id: keyId, + wallet, + is_active: isActive, + expiry_timestamp: expirySeconds.toString(), + days_remaining: daysRemaining, + }; +} + +async function readSubscription(contractId: string, sourceWallet: string, wallet: string, keyId: string): Promise { + const accountResponse = await horizonGet(`/accounts/${encodeURIComponent(sourceWallet)}`); + if (!accountResponse.ok) { + throw new SubscriptionStatusError('Unable to load the contract caller account from Horizon'); + } + const account = (await accountResponse.json()) as { sequence?: unknown }; + if (typeof account.sequence !== 'string' || !/^\d+$/.test(account.sequence)) { + throw new SubscriptionStatusError('Horizon returned an invalid account sequence'); + } + + const passphrase = envConfig.STELLAR_NETWORK === 'mainnet' ? Networks.PUBLIC : Networks.TESTNET; + const transaction = new TransactionBuilder(new Account(sourceWallet, account.sequence), { + fee: '100', + networkPassphrase: passphrase, + }) + .addOperation(new Contract(contractId).call('get_subscription', new Address(wallet).toScVal())) + .setTimeout(300) + .build() + .toXDR(); + + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), envConfig.DB_QUERY_TIMEOUT_MS); + try { + const response = await fetch(envConfig.STELLAR_SOROBAN_RPC_URL, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ jsonrpc: '2.0', id: 'subscription-status', method: 'simulateTransaction', params: { transaction } }), + signal: controller.signal, + }); + if (!response.ok) throw new SubscriptionStatusError(`Soroban RPC returned HTTP ${response.status}`); + const envelope = (await response.json()) as { + error?: { message?: string }; + result?: { results?: Array<{ xdr?: string; retval?: string }> }; + }; + if (envelope.error) throw new SubscriptionStatusError(envelope.error.message ?? 'Subscription contract call failed'); + const result = envelope.result?.results?.[0]?.retval ?? envelope.result?.results?.[0]?.xdr; + if (!result) throw new SubscriptionStatusError('Subscription contract returned no data'); + return decodeSubscription(result, keyId, wallet); + } finally { + clearTimeout(timeout); + } +} + +export async function getSubscriptionStatus(keyId: string, walletAddress: string): Promise { + const wallet = walletAddress.trim().toUpperCase(); + if (!isValidStellarAddress(wallet)) throw new SubscriptionStatusError('A valid Stellar wallet is required', 400); + + const registeredKey = await prisma.registeredKey.findFirst({ + where: { OR: [{ id: keyId }, { keyAddress: keyId }] }, + select: { id: true, keyAddress: true, creatorWallet: true }, + }); + if (!registeredKey) throw new SubscriptionStatusError('Key not found', 404); + + const cacheKey = `subscription:status:${registeredKey.keyAddress}:${wallet}`; + const cached = await cacheGetJson(cacheKey); + if (cached) return { ...cached, key_id: keyId }; + + const status = await readSubscription(registeredKey.keyAddress, registeredKey.creatorWallet, wallet, keyId); + await cacheSetJson(cacheKey, status, CACHE_TTL_SECONDS); + return status; +} + +export async function processSubscriptionCreatedEvent(event: Record): Promise { + const keyId = String(event.keyId ?? event.key_id ?? event.creatorId ?? event.contractId ?? ''); + const wallet = String(event.actor ?? event.wallet ?? event.subscriber ?? '').trim().toUpperCase(); + if (!keyId || !wallet) return; + + const key = await prisma.registeredKey.findFirst({ + where: { OR: [{ id: keyId }, { keyAddress: keyId }] }, + select: { id: true, keyAddress: true }, + }); + if (key) { + await cacheInvalidate(`subscription:status:${key.keyAddress}:${wallet}`); + } + + const txHash = String(event.txHash ?? event.tx_hash ?? ''); + if (!txHash) return; + await prisma.activityLog.upsert({ + where: { txHash_actor_type: { txHash, actor: wallet, type: 'subscription_created' } }, + update: {}, + create: { + type: 'subscription_created', + actor: wallet, + keyId: key?.id ?? keyId, + txHash, + payload: { expiry_timestamp: event.expiryTimestamp ?? event.expiry_timestamp ?? null }, + timestamp: event.timestamp ? new Date(String(event.timestamp)) : new Date(), + }, + }); +}