diff --git a/prisma/schema/key-collaboration.prisma b/prisma/schema/key-collaboration.prisma new file mode 100644 index 00000000..e9f27fc1 --- /dev/null +++ b/prisma/schema/key-collaboration.prisma @@ -0,0 +1,117 @@ +// prisma/schema/key-collaboration.prisma +// +// Read models for co-creator rosters (#1139), creator tips (#1132) and +// key reserves (#1140), maintained by the indexer from CoCreatorSet / +// CoCreatorRemoved, TipSent and ReserveAccrued / ReserveDeployed contract +// events. +// +// Design notes: +// - `keyId` is intentionally NOT a foreign key: an event that arrives +// before CREATOR_REGISTERED has been indexed is stored, not dropped +// (same rationale as KeyDepth). +// - KeyCoCreator is an order-safe upsert: the row carries the +// (ledger, eventIndex) watermark of the last applied event and only a +// strictly newer event mutates it, so replays and out-of-order delivery +// converge on the latest on-chain state. +// - KeyTip / KeyReserveEvent are append-only; (txHash, eventIndex) +// uniqueness makes replayed events no-ops. The running totals on +// KeyTipTotal and KeyReserve are updated in the same transaction as the +// event row, so a total can never drift from its event log. +// - Monetary amounts are stored as Decimal XLM (7 dp), converted from +// stroops by the indexer. + +/// Active + historical co-creator roster per key (#1139). +model KeyCoCreator { + id String @id @default(cuid()) + /// Key ID (CreatorProfile.id) the co-creator belongs to. + keyId String + /// Stellar address (G...) of the co-creator. + wallet String + /// Role label served by GET /keys/:key_id/co-creators. + role String @default("co_creator") + /// On-chain revenue share in basis points (CoCreatorSet.split_bps). + shareBps Int? + /// false after CoCreatorRemoved; excluded from the active roster list. + active Boolean @default(true) + addedAt DateTime @default(now()) + removedAt DateTime? + /// Position of the last applied event; older events are ignored. + ledger Int? + eventIndex Int? + txHash String? + updatedAt DateTime @updatedAt + + @@unique([keyId, wallet]) + @@index([keyId, active]) + @@map("key_cocreators") +} + +/// One row per indexed TipSent event (#1132). +model KeyTip { + id String @id @default(cuid()) + /// Key ID (CreatorProfile.id) that received the tip. + keyId String + /// Stellar address of the tipper. + tipper String + /// Tip amount in XLM (converted from stroops). + amount Decimal @db.Decimal(20, 7) + /// IPFS CID of the tip message, when one was attached. + messageHash String? + ledger Int? + /// Chain position; unique with eventIndex so replays are no-ops. + txHash String? + eventIndex Int? + /// When the tip happened on chain. + tippedAt DateTime + createdAt DateTime @default(now()) + + @@unique([txHash, eventIndex]) + @@index([keyId, tippedAt]) + @@map("key_tips") +} + +/// Cumulative tips per key, maintained transactionally with KeyTip rows. +model KeyTipTotal { + id String @id @default(cuid()) + keyId String @unique + total Decimal @default(0) @db.Decimal(20, 7) + count Int @default(0) + updatedAt DateTime @updatedAt + + @@map("key_tip_totals") +} + +/// Current reserve balance per key (#1140), maintained from +/// ReserveAccrued / ReserveDeployed events. +model KeyReserve { + id String @id @default(cuid()) + keyId String @unique + /// Current reserve balance in XLM. + balance Decimal @default(0) @db.Decimal(20, 7) + /// Lifetime accrued (XLM). + totalAccrued Decimal @default(0) @db.Decimal(20, 7) + /// Lifetime deployed (XLM). + totalDeployed Decimal @default(0) @db.Decimal(20, 7) + updatedAt DateTime @updatedAt + + @@map("key_reserves") +} + +/// Append-only reserve event log; (txHash, eventIndex) uniqueness makes +/// replayed events no-ops so the balance above never double-counts. +model KeyReserveEvent { + id String @id @default(cuid()) + keyId String + /// "accrued" | "deployed". + type String + /// Amount in XLM. + amount Decimal @db.Decimal(20, 7) + ledger Int? + txHash String? + eventIndex Int? + createdAt DateTime @default(now()) + + @@unique([txHash, eventIndex]) + @@index([keyId]) + @@map("key_reserve_events") +} diff --git a/prisma/schema/migrations/20261020180000_add_key_cocreator_tip_reserve_tables/migration.sql b/prisma/schema/migrations/20261020180000_add_key_cocreator_tip_reserve_tables/migration.sql new file mode 100644 index 00000000..6e7aeb0c --- /dev/null +++ b/prisma/schema/migrations/20261020180000_add_key_cocreator_tip_reserve_tables/migration.sql @@ -0,0 +1,100 @@ +-- Co-creator roster (#1139), tip history (#1132) and reserve balances +-- (#1140). Hand-written (not generated): the repo's shadow-database replay +-- is broken for full-history generation (see docs/database-migrations.md). + +-- CreateTable +CREATE TABLE "key_cocreators" ( + "id" TEXT NOT NULL, + "keyId" TEXT NOT NULL, + "wallet" TEXT NOT NULL, + "role" TEXT NOT NULL DEFAULT 'co_creator', + "shareBps" INTEGER, + "active" BOOLEAN NOT NULL DEFAULT true, + "addedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "removedAt" TIMESTAMP(3), + "ledger" INTEGER, + "eventIndex" INTEGER, + "txHash" TEXT, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "key_cocreators_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "key_tips" ( + "id" TEXT NOT NULL, + "keyId" TEXT NOT NULL, + "tipper" TEXT NOT NULL, + "amount" DECIMAL(20,7) NOT NULL, + "messageHash" TEXT, + "ledger" INTEGER, + "txHash" TEXT, + "eventIndex" INTEGER, + "tippedAt" TIMESTAMP(3) NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "key_tips_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "key_tip_totals" ( + "id" TEXT NOT NULL, + "keyId" TEXT NOT NULL, + "total" DECIMAL(20,7) NOT NULL DEFAULT 0, + "count" INTEGER NOT NULL DEFAULT 0, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "key_tip_totals_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "key_reserves" ( + "id" TEXT NOT NULL, + "keyId" TEXT NOT NULL, + "balance" DECIMAL(20,7) NOT NULL DEFAULT 0, + "totalAccrued" DECIMAL(20,7) NOT NULL DEFAULT 0, + "totalDeployed" DECIMAL(20,7) NOT NULL DEFAULT 0, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "key_reserves_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "key_reserve_events" ( + "id" TEXT NOT NULL, + "keyId" TEXT NOT NULL, + "type" TEXT NOT NULL, + "amount" DECIMAL(20,7) NOT NULL, + "ledger" INTEGER, + "txHash" TEXT, + "eventIndex" INTEGER, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "key_reserve_events_pkey" PRIMARY KEY ("id"), + CONSTRAINT "key_reserve_events_type_check" CHECK ("type" IN ('accrued', 'deployed')), + CONSTRAINT "key_reserve_events_amount_check" CHECK ("amount" >= 0) +); + +-- CreateIndex +CREATE UNIQUE INDEX "key_cocreators_keyId_wallet_key" ON "key_cocreators"("keyId", "wallet"); + +-- CreateIndex +CREATE INDEX "key_cocreators_keyId_active_idx" ON "key_cocreators"("keyId", "active"); + +-- CreateIndex +CREATE UNIQUE INDEX "key_tips_txHash_eventIndex_key" ON "key_tips"("txHash", "eventIndex"); + +-- CreateIndex +CREATE INDEX "key_tips_keyId_tippedAt_idx" ON "key_tips"("keyId", "tippedAt"); + +-- CreateIndex +CREATE UNIQUE INDEX "key_tip_totals_keyId_key" ON "key_tip_totals"("keyId"); + +-- CreateIndex +CREATE UNIQUE INDEX "key_reserves_keyId_key" ON "key_reserves"("keyId"); + +-- CreateIndex +CREATE UNIQUE INDEX "key_reserve_events_txHash_eventIndex_key" ON "key_reserve_events"("txHash", "eventIndex"); + +-- CreateIndex +CREATE INDEX "key_reserve_events_keyId_idx" ON "key_reserve_events"("keyId"); diff --git a/src/constants/redis.constants.ts b/src/constants/redis.constants.ts index d0f88afc..563d8849 100644 --- a/src/constants/redis.constants.ts +++ b/src/constants/redis.constants.ts @@ -61,3 +61,67 @@ export const keyDepthSpreadCacheKey = (keyId: string): string => export const keyDepthCachePattern = (keyId: string): string => `depth:${keyId}:*`; + +// ── Co-creator roster endpoints (#1139) ───────────────────────── +// GET /keys/:key_id/co-creators and /co-creators/:wallet are served from +// the indexed key_cocreators table with a 60s TTL. Both the list and any +// per-wallet detail share the `key:cocreators:{keyId}:` prefix so the +// indexer can drop everything for a key with one pattern. +export const KEY_COCREATOR_CACHE_TTL_SECONDS = 60; + +export const keyCoCreatorListCacheKey = (keyId: string): string => + `key:cocreators:${keyId}:list`; + +export const keyCoCreatorDetailCacheKey = ( + keyId: string, + wallet: string +): string => `key:cocreators:${keyId}:${wallet}`; + +export const keyCoCreatorCachePattern = (keyId: string): string => + `key:cocreators:${keyId}:*`; + +// ── Reserve balance endpoint (#1140) ──────────────────────────── +// GET /keys/:key_id/reserve is served from the indexed key_reserves table +// with a 60s TTL, invalidated per key on every ReserveAccrued / +// ReserveDeployed event. +export const KEY_RESERVE_CACHE_TTL_SECONDS = 60; + +export const keyReserveCacheKey = (keyId: string): string => + `key:reserve:${keyId}`; + +// ── Content access endpoint (#1137) ───────────────────────────── +// GET /keys/:key_id/content-access/:wallet is evaluated on chain (or +// off-chain from indexed state) with a 30s TTL. Invalidated per key on +// KEY_BOUGHT/KEY_SOLD (balance changes) so a just-completed trade is +// visible immediately. +export const CONTENT_ACCESS_CACHE_TTL_SECONDS = 30; + +export const contentAccessCacheKey = ( + keyId: string, + wallet: string +): string => `key:content-access:${keyId}:${wallet}`; + +export const contentAccessCachePattern = (keyId: string): string => + `key:content-access:${keyId}:*`; + +// ── Tip history endpoints (#1132) ─────────────────────────────── +// GET /keys/:key_id/tips and /tips/total are served from the indexed +// key_tips / key_tip_totals tables with a 60s TTL. IPFS tip message +// content fetched on demand is cached longer (immutable per CID). +export const KEY_TIPS_CACHE_TTL_SECONDS = 60; +export const KEY_TIP_CONTENT_CACHE_TTL_SECONDS = 3600; + +export const keyTipsListCacheKey = ( + keyId: string, + limit: number, + cursor: string +): string => `key:tips:${keyId}:list:${limit}:${cursor}`; + +export const keyTipsTotalCacheKey = (keyId: string): string => + `key:tips:${keyId}:total`; + +export const keyTipContentCacheKey = (messageHash: string): string => + `key:tips:content:${messageHash}`; + +export const keyTipsCachePattern = (keyId: string): string => + `key:tips:${keyId}:*`; diff --git a/src/modules/co-creators/co-creators.routes.ts b/src/modules/co-creators/co-creators.routes.ts new file mode 100644 index 00000000..adf7427f --- /dev/null +++ b/src/modules/co-creators/co-creators.routes.ts @@ -0,0 +1,123 @@ +// src/modules/co-creators/co-creators.routes.ts +// +// REST endpoints for the per-key co-creator roster (#1139): +// +// GET /keys/:key_id/co-creators — active roster (wallet, role, added_at) +// GET /keys/:key_id/co-creators/:wallet — single co-creator detail with role +// +// Public, read-only. Mounted from keys.routes.ts via router.use('/', …). +// Both `:key_id` and `:keyId` spellings are accepted (whitelist module +// convention). Responses carry `Cache-Control: public, max-age=60` and are +// backed by a 60s Redis cache invalidated by the co-creator indexer. + +import { NextFunction, Request, Response, Router } from 'express'; +import { z } from 'zod'; +import { AsyncController } from '../../types/auth.types'; +import { + sendSuccess, + sendValidationError, + sendNotFound, + zodIssuesToDetails, +} from '../../utils/api-response.utils'; +import { cacheControl } from '../../middlewares/cache-control.middleware'; +import { KEY_COCREATOR_CACHE_TTL_SECONDS } from '../../constants/redis.constants'; +import { + getCoCreatorDetail, + listActiveCoCreators, +} from './co-creators.service'; + +const coCreatorsRouter = Router(); + +const normalizeKeyId = (req: Request, _res: Response, next: NextFunction) => { + if (!req.params.keyId && req.params.key_id) { + req.params.keyId = req.params.key_id; + } + next(); +}; + +const WalletParamSchema = z + .string() + .min(1, 'Wallet address is required') + .max(100, 'Invalid wallet address format') + .regex(/^G[A-Za-z0-9]{30,70}$/, 'Invalid wallet address format'); + +const coCreatorCache = () => + cacheControl({ + maxAge: KEY_COCREATOR_CACHE_TTL_SECONDS, + type: 'public', + mustRevalidate: true, + }); + +/** + * GET /keys/:key_id/co-creators + * Public. Active co-creator roster for a key: wallet, role, added_at. + */ +const httpListCoCreators: AsyncController = async (req, res, next) => { + try { + const keyId = (req.params.key_id ?? req.params.keyId) as string; + if (!keyId) { + return sendValidationError(res, 'Missing required parameters', [ + { field: 'keyId', message: 'Key ID is required' }, + ]); + } + return sendSuccess(res, await listActiveCoCreators(keyId)); + } catch (error) { + next(error); + } +}; + +/** + * GET /keys/:key_id/co-creators/:wallet + * Public. Detail for one co-creator of a key, including role and removal + * state. 404 when the wallet was never a co-creator of the key. + */ +const httpGetCoCreator: AsyncController = async (req, res, next) => { + try { + const keyId = (req.params.key_id ?? req.params.keyId) as string; + if (!keyId) { + return sendValidationError(res, 'Missing required parameters', [ + { field: 'keyId', message: 'Key ID is required' }, + ]); + } + + const parsedWallet = WalletParamSchema.safeParse(req.params.wallet); + if (!parsedWallet.success) { + return sendValidationError( + res, + 'Invalid wallet address', + zodIssuesToDetails(parsedWallet.error.issues) + ); + } + + const detail = await getCoCreatorDetail(keyId, parsedWallet.data); + if (!detail) { + return sendNotFound(res, 'Co-creator'); + } + return sendSuccess(res, detail); + } catch (error) { + next(error); + } +}; + +coCreatorsRouter.get( + ['/:key_id/co-creators', '/:keyId/co-creators'], + normalizeKeyId, + coCreatorCache(), + httpListCoCreators +); + +coCreatorsRouter.get( + ['/:key_id/co-creators/:wallet', '/:keyId/co-creators/:wallet'], + normalizeKeyId, + coCreatorCache(), + httpGetCoCreator +); + +coCreatorsRouter.all( + ['/:key_id/co-creators', '/:keyId/co-creators'], + (_req, res) => { + res.set('Allow', 'GET').sendStatus(405); + } +); + +export default coCreatorsRouter; diff --git a/src/modules/co-creators/co-creators.service.ts b/src/modules/co-creators/co-creators.service.ts new file mode 100644 index 00000000..81c41490 --- /dev/null +++ b/src/modules/co-creators/co-creators.service.ts @@ -0,0 +1,182 @@ +// src/modules/co-creators/co-creators.service.ts +// +// Read model for the per-key co-creator roster (#1139), backed by the +// `key_cocreators` table maintained from CoCreatorSet / CoCreatorRemoved +// contract events. +// +// Responses are cached in Redis for 60s and invalidated per key by the +// co-creator indexer after every event, so a freshly added or removed +// co-creator is reflected within one event-processing hop. + +import { prisma } from '../../utils/prisma.utils'; +import { + cacheGetJson, + cacheSetJson, + cacheInvalidate, +} from '../../utils/redis.utils'; +import { logger } from '../../utils/logger.utils'; +import { + KEY_COCREATOR_CACHE_TTL_SECONDS, + keyCoCreatorCachePattern, + keyCoCreatorDetailCacheKey, + keyCoCreatorListCacheKey, +} from '../../constants/redis.constants'; +import { resolveCreatorId } from '../whitelist/whitelist.service'; + +/** One active co-creator in the roster. */ +export interface CoCreatorListItem { + wallet: string; + role: string; + added_at: string; + share_bps: number | null; +} + +/** Full co-creator detail, including removal state. */ +export interface CoCreatorDetail { + key_id: string; + wallet: string; + role: string; + active: boolean; + added_at: string; + removed_at: string | null; + share_bps: number | null; +} + +/** + * Drops every cached co-creator response for a key (list + wallet + * details). Called by the indexer after every CoCreatorSet / + * CoCreatorRemoved event. + */ +export async function invalidateCoCreatorCache(keyId: string): Promise { + await cacheInvalidate(keyCoCreatorCachePattern(keyId)); +} + +/** + * Maps any key identifier (CreatorProfile id, handle, or key contract + * address) to the canonical CreatorProfile.id used as `keyId` in the + * key_cocreators table. Falls back to the raw identifier when nothing is + * registered yet, so events that arrive before CREATOR_REGISTERED still + * land. + */ +export async function resolveCoCreatorKeyId(keyId: string): Promise { + return (await resolveCreatorId(keyId)) ?? keyId; +} + +/** + * Lists the active co-creator roster for a key (wallet, role, added_at), + * most recently added first. Removed co-creators are excluded. + */ +export async function listActiveCoCreators( + keyId: string +): Promise<{ key_id: string; co_creators: CoCreatorListItem[] }> { + const resolvedKeyId = await resolveCoCreatorKeyId(keyId); + const cacheKey = keyCoCreatorListCacheKey(resolvedKeyId); + + try { + const cached = await cacheGetJson<{ + key_id: string; + co_creators: CoCreatorListItem[]; + }>(cacheKey); + if (cached !== null) { + logger.debug( + { keyId: resolvedKeyId }, + 'Co-creator list cache hit' + ); + return cached; + } + } catch { + // Cache read failures fall through to the database. + } + + const rows = await prisma.keyCoCreator.findMany({ + where: { keyId: resolvedKeyId, active: true }, + orderBy: { addedAt: 'desc' }, + select: { + wallet: true, + role: true, + addedAt: true, + shareBps: true, + }, + }); + + const result = { + key_id: resolvedKeyId, + co_creators: rows.map(row => ({ + wallet: row.wallet, + role: row.role, + added_at: row.addedAt.toISOString(), + share_bps: row.shareBps, + })), + }; + + try { + await cacheSetJson(cacheKey, result, KEY_COCREATOR_CACHE_TTL_SECONDS); + } catch { + // Cache write failures are non-critical. + } + + return result; +} + +/** + * Returns the detail for one co-creator of a key, including role and + * removal state. Returns null when the wallet was never a co-creator. + */ +export async function getCoCreatorDetail( + keyId: string, + wallet: string +): Promise { + const resolvedKeyId = await resolveCoCreatorKeyId(keyId); + const normalizedWallet = wallet.trim(); + const cacheKey = keyCoCreatorDetailCacheKey(resolvedKeyId, normalizedWallet); + + try { + const cached = await cacheGetJson(cacheKey); + if (cached !== null) { + logger.debug( + { keyId: resolvedKeyId, wallet: normalizedWallet }, + 'Co-creator detail cache hit' + ); + return cached; + } + } catch { + // Cache read failures fall through to the database. + } + + const row = await prisma.keyCoCreator.findUnique({ + where: { + keyId_wallet: { keyId: resolvedKeyId, wallet: normalizedWallet }, + }, + select: { + keyId: true, + wallet: true, + role: true, + active: true, + addedAt: true, + removedAt: true, + shareBps: true, + }, + }); + + if (!row) { + return null; + } + + const result: CoCreatorDetail = { + key_id: row.keyId, + wallet: row.wallet, + role: row.role, + active: row.active, + added_at: row.addedAt.toISOString(), + removed_at: row.removedAt ? row.removedAt.toISOString() : null, + share_bps: row.shareBps, + }; + + try { + await cacheSetJson(cacheKey, result, KEY_COCREATOR_CACHE_TTL_SECONDS); + } catch { + // Cache write failures are non-critical. + } + + return result; +} diff --git a/src/modules/content-access/content-access.routes.ts b/src/modules/content-access/content-access.routes.ts new file mode 100644 index 00000000..9d6cf013 --- /dev/null +++ b/src/modules/content-access/content-access.routes.ts @@ -0,0 +1,80 @@ +// src/modules/content-access/content-access.routes.ts +// +// REST endpoint for the content-gate access check (#1137): +// +// GET /keys/:key_id/content-access/:wallet +// +// Public, read-only. Contract-first (check_access simulation) with an +// off-chain fallback; responses carry `Cache-Control: public, max-age=30` +// and are backed by a 30s Redis cache invalidated on KEY_BOUGHT / +// KEY_SOLD. Both `:key_id` and `:keyId` spellings are accepted. + +import { NextFunction, Request, Response, Router } from 'express'; +import { AsyncController } from '../../types/auth.types'; +import { + sendSuccess, + sendValidationError, +} from '../../utils/api-response.utils'; +import { cacheControl } from '../../middlewares/cache-control.middleware'; +import { CONTENT_ACCESS_CACHE_TTL_SECONDS } from '../../constants/redis.constants'; +import { getContentAccess } from './content-access.service'; + +const contentAccessRouter = Router(); + +const normalizeKeyId = (req: Request, _res: Response, next: NextFunction) => { + if (!req.params.keyId && req.params.key_id) { + req.params.keyId = req.params.key_id; + } + next(); +}; + +const WALLET_REGEX = /^G[A-Za-z0-9]{30,70}$/; + +/** + * GET /keys/:key_id/content-access/:wallet + * Public. Returns { is_allowed, gate_type, requirement } for the wallet + * on this key's content gate. + */ +const httpGetContentAccess: AsyncController = async (req, res, next) => { + try { + const keyId = (req.params.key_id ?? req.params.keyId) as string; + const wallet = req.params.wallet as string; + + if (!keyId || !wallet) { + return sendValidationError(res, 'Missing required parameters', [ + { field: 'keyId', message: 'Key ID is required' }, + { field: 'wallet', message: 'Wallet address is required' }, + ]); + } + + if (!WALLET_REGEX.test(wallet)) { + return sendValidationError(res, 'Invalid wallet address', [ + { field: 'wallet', message: 'Invalid Stellar wallet address' }, + ]); + } + + return sendSuccess(res, await getContentAccess(keyId, wallet)); + } catch (error) { + next(error); + } +}; + +contentAccessRouter.get( + ['/:key_id/content-access/:wallet', '/:keyId/content-access/:wallet'], + normalizeKeyId, + cacheControl({ + maxAge: CONTENT_ACCESS_CACHE_TTL_SECONDS, + type: 'public', + mustRevalidate: true, + }), + httpGetContentAccess +); + +contentAccessRouter.all( + ['/:key_id/content-access/:wallet', '/:keyId/content-access/:wallet'], + (_req, res) => { + res.set('Allow', 'GET').sendStatus(405); + } +); + +export default contentAccessRouter; diff --git a/src/modules/content-access/content-access.service.ts b/src/modules/content-access/content-access.service.ts new file mode 100644 index 00000000..d87b94bd --- /dev/null +++ b/src/modules/content-access/content-access.service.ts @@ -0,0 +1,616 @@ +// src/modules/content-access/content-access.service.ts +// +// Content gate access check for a wallet on a key (#1137), used by the +// frontend content section before rendering gated content. +// +// Contract-first: the service simulates the key contract's +// `check_access(wallet, key_id, gate_type)` view and decodes +// `{ is_allowed, gate_type, requirement }` from the result. An injectable +// fetcher (`setContentAccessContractFetcher`) mirrors the whitelist +// module's test seam. +// +// Off-chain fallback: while `check_access` is not yet deployed (contract +// PR in flight), the service evaluates the gate from the views and read +// models that exist today: +// 1. `get_gate_config` (future-proof) → gate_type + requirement, +// 2. else `get_min_hold_for_access` → THRESHOLD gate with the configured +// minimum, +// 3. else the key is ungated → `{ is_allowed: true, gate_type: 'none' }`. +// THRESHOLD/HOLDER gates are evaluated against `get_key_balance` (on-chain +// view, falling back to the indexed KeyOwnership balance); SUBSCRIBER +// gates against the `is_subscribed` view. +// +// Responses are cached in Redis for 30s, invalidated per key on +// KEY_BOUGHT/KEY_SOLD (balance changes). + +import { + Account, + Address, + Contract, + Networks, + TransactionBuilder, + xdr, +} from '@stellar/stellar-base'; +import { prisma } from '../../utils/prisma.utils'; +import { + cacheGetJson, + cacheSetJson, + cacheInvalidate, +} from '../../utils/redis.utils'; +import { logger } from '../../utils/logger.utils'; +import { envConfig } from '../../config'; +import { + CONTENT_ACCESS_CACHE_TTL_SECONDS, + contentAccessCacheKey, + contentAccessCachePattern, +} from '../../constants/redis.constants'; +import { resolveCreatorId } from '../whitelist/whitelist.service'; + +/** Gate types returned by the endpoint (issue #1150 vocabulary). */ +export type ContentGateType = 'none' | 'HOLDER' | 'SUBSCRIBER' | 'THRESHOLD'; + +/** Access check result for a wallet on a key. */ +export interface ContentAccessResult { + is_allowed: boolean; + gate_type: ContentGateType; + requirement: Record | null; +} + +/** + * Injectable contract fetcher (test seam / RPC injection), mirroring the + * whitelist module. Returning null falls through to the next check. + */ +export type ContentAccessContractFetcher = ( + contractId: string, + wallet: string, + keyId: string +) => Promise; + +let customFetcher: ContentAccessContractFetcher | null = null; + +/** + * Configures an optional custom content-access contract fetcher (for + * testing or RPC injection). + */ +export function setContentAccessContractFetcher( + fetcher: ContentAccessContractFetcher | null +): void { + customFetcher = fetcher; +} + +/** + * Drops every cached content-access response for a key. Called when a + * balance-changing event (KEY_BOUGHT / KEY_SOLD) is indexed. + */ +export async function invalidateContentAccessCache( + keyId: string +): Promise { + await cacheInvalidate(contentAccessCachePattern(keyId)); +} + +function networkPassphrase(): string { + return envConfig.STELLAR_NETWORK === 'mainnet' + ? Networks.PUBLIC + : Networks.TESTNET; +} + +const CONTRACT_ID_REGEX = /^C[A-Z2-7]{55}$/; + +/** + * Resolves a key identifier to its on-chain contract address + * (RegisteredKey.keyAddress), or null when the key has no contract. + */ +export async function resolveContentAccessContractId( + keyId: string +): Promise { + if (CONTRACT_ID_REGEX.test(keyId)) { + return keyId; + } + const regKey = await prisma.registeredKey.findFirst({ + where: { + OR: [{ id: keyId }, { keyAddress: keyId }, { handle: keyId }], + }, + select: { keyAddress: true }, + }); + return regKey?.keyAddress ?? null; +} + +/** + * Builds an unsigned simulated transaction envelope to query a contract + * view function with a single Address argument. + */ +export function buildAddressViewXdr( + contractId: string, + functionName: string, + wallet: string, + sequence = '0' +): string { + const dummySource = wallet.startsWith('G') + ? wallet + : 'GBTEST0000000000000000000000000000000000000000000000000000'; + const operation = new Contract(contractId).call( + functionName, + new Address(wallet).toScVal() + ); + return new TransactionBuilder(new Account(dummySource, sequence), { + fee: '100', + networkPassphrase: networkPassphrase(), + }) + .addOperation(operation) + .setTimeout(300) + .build() + .toXDR(); +} + +/** + * Builds an unsigned simulated transaction envelope for + * `check_access(wallet, key_id)`. + */ +export function buildCheckAccessViewXdr( + contractId: string, + wallet: string, + keyId: string, + sequence = '0' +): string { + const dummySource = wallet.startsWith('G') + ? wallet + : 'GBTEST0000000000000000000000000000000000000000000000000000'; + const operation = new Contract(contractId).call( + 'check_access', + new Address(wallet).toScVal(), + new Address(keyId).toScVal() + ); + return new TransactionBuilder(new Account(dummySource, sequence), { + fee: '100', + networkPassphrase: networkPassphrase(), + }) + .addOperation(operation) + .setTimeout(300) + .build() + .toXDR(); +} + +/** Posts a simulateTransaction JSON-RPC call; returns the retval XDR. */ +async function simulateView( + transactionXdr: string, + rpcId: string +): Promise { + if (!envConfig.STELLAR_SOROBAN_RPC_URL) { + return null; + } + + 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: rpcId, + method: 'simulateTransaction', + params: { transaction: transactionXdr }, + }), + signal: controller.signal, + }); + + if (!response.ok) { + return null; + } + + const envelope = (await response.json()) as { + error?: { message?: string }; + result?: { results?: Array<{ xdr?: string; retval?: string }> }; + }; + + if (envelope.error) { + return null; + } + + return ( + envelope.result?.results?.[0]?.retval ?? + envelope.result?.results?.[0]?.xdr ?? + null + ); + } catch { + return null; + } finally { + clearTimeout(timeout); + } +} + +/** Decodes a ScVal gate_type symbol/string into a ContentGateType. */ +function decodeGateTypeScVal(val: xdr.ScVal): ContentGateType | null { + let raw: string | null = null; + if (val.switch() === xdr.ScValType.scvSymbol()) { + raw = val.sym().toString(); + } else if (val.switch() === xdr.ScValType.scvString()) { + raw = val.str().toString(); + } + if (!raw) return null; + const normalized = raw.toUpperCase(); + if ( + normalized === 'HOLDER' || + normalized === 'SUBSCRIBER' || + normalized === 'THRESHOLD' || + normalized === 'NONE' + ) { + return normalized as ContentGateType; + } + return null; +} + +/** Decodes a requirement value (u32 / map) into a plain object. */ +function decodeRequirementScVal( + val: xdr.ScVal +): Record | null { + const type = val.switch(); + if (type === xdr.ScValType.scvU32()) { + return { min_keys: val.u32() }; + } + if (type === xdr.ScValType.scvU64()) { + return { min_keys: Number(val.u64().toString()) }; + } + if (type === xdr.ScValType.scvMap()) { + const out: Record = {}; + for (const entry of val.map() ?? []) { + const keyType = entry.key().switch(); + let keyName = ''; + if (keyType === xdr.ScValType.scvSymbol()) { + keyName = entry.key().sym().toString(); + } else if (keyType === xdr.ScValType.scvString()) { + keyName = entry.key().str().toString(); + } + if (!keyName) continue; + const valType = entry.val().switch(); + if (valType === xdr.ScValType.scvU32()) { + out[keyName] = entry.val().u32(); + } else if (valType === xdr.ScValType.scvU64()) { + out[keyName] = Number(entry.val().u64().toString()); + } else if (valType === xdr.ScValType.scvI128()) { + out[keyName] = Number(entry.val().i128().toString()); + } else if (valType === xdr.ScValType.scvString()) { + out[keyName] = entry.val().str().toString(); + } else if (valType === xdr.ScValType.scvSymbol()) { + out[keyName] = entry.val().sym().toString(); + } else if (valType === xdr.ScValType.scvBool()) { + out[keyName] = entry.val().b(); + } + } + return Object.keys(out).length > 0 ? out : null; + } + return null; +} + +/** + * Decodes a check_access return value. Accepts a bool (allowed, ungated + * semantics), a vec [bool, gate_type, requirement], or a map with + * is_allowed / gate_type / requirement keys. Returns null when the shape + * is unrecognised. + */ +export function decodeCheckAccessScVal( + scValXdr: string +): ContentAccessResult | null { + try { + const val = xdr.ScVal.fromXDR(scValXdr, 'base64'); + const type = val.switch(); + + if (type === xdr.ScValType.scvBool()) { + return { is_allowed: val.b(), gate_type: 'none', requirement: null }; + } + + if (type === xdr.ScValType.scvVec()) { + const vec = val.vec() ?? []; + if ( + vec.length >= 1 && + vec[0].switch() === xdr.ScValType.scvBool() + ) { + const isAllowed = vec[0].b(); + const gateType = + vec.length >= 2 ? decodeGateTypeScVal(vec[1]) : 'none'; + const requirement = + vec.length >= 3 ? decodeRequirementScVal(vec[2]) : null; + return { + is_allowed: isAllowed, + gate_type: gateType ?? 'none', + requirement, + }; + } + } + + if (type === xdr.ScValType.scvMap()) { + let isAllowed = true; + let gateType: ContentGateType = 'none'; + let requirement: Record | null = null; + for (const entry of val.map() ?? []) { + const keyType = entry.key().switch(); + let keyName = ''; + if (keyType === xdr.ScValType.scvSymbol()) { + keyName = entry.key().sym().toString().toLowerCase(); + } else if (keyType === xdr.ScValType.scvString()) { + keyName = entry.key().str().toString().toLowerCase(); + } + if ( + (keyName === 'is_allowed' || keyName === 'allowed') && + entry.val().switch() === xdr.ScValType.scvBool() + ) { + isAllowed = entry.val().b(); + } else if (keyName === 'gate_type' || keyName === 'gatetype') { + gateType = decodeGateTypeScVal(entry.val()) ?? 'none'; + } else if (keyName === 'requirement') { + requirement = decodeRequirementScVal(entry.val()); + } + } + return { is_allowed: isAllowed, gate_type: gateType, requirement }; + } + + return null; + } catch (error) { + logger.debug( + { error: (error as Error).message }, + 'Failed to decode check_access ScVal' + ); + return null; + } +} + +/** + * Calls the key contract's check_access view. Returns null when the + * function is unavailable, the RPC fails, or the result is undecodable + * (the fallback then evaluates the gate off-chain). + */ +export async function fetchOnChainContentAccess( + contractId: string, + wallet: string, + keyId: string +): Promise { + if (customFetcher) { + const custom = await customFetcher(contractId, wallet, keyId); + if (custom !== null) return custom; + } + + try { + const transactionXdr = buildCheckAccessViewXdr( + contractId, + wallet, + keyId + ); + const retval = await simulateView(transactionXdr, 'check-access'); + if (!retval) return null; + return decodeCheckAccessScVal(retval); + } catch (error) { + logger.debug( + { error: (error as Error).message, contractId, wallet }, + 'check_access contract call failed; using off-chain fallback' + ); + return null; + } +} + +/** + * Reads `get_min_hold_for_access` from the contract. Returns the minimum + * or null when gating is off / the call is unavailable. + */ +async function fetchMinHoldForAccess( + contractId: string, + wallet: string +): Promise { + try { + const transactionXdr = buildAddressViewXdr( + contractId, + 'get_min_hold_for_access', + wallet + ); + const retval = await simulateView(transactionXdr, 'min-hold'); + if (!retval) return null; + + const val = xdr.ScVal.fromXDR(retval, 'base64'); + if (val.switch() === xdr.ScValType.scvU32()) { + return val.u32(); + } + // Option decodes as a vec: [] = None, [n] = Some(n). + if (val.switch() === xdr.ScValType.scvVec()) { + const vec = val.vec() ?? []; + if (vec.length === 0) return null; + if ( + vec.length >= 1 && + vec[0].switch() === xdr.ScValType.scvU32() + ) { + return vec[0].u32(); + } + } + return null; + } catch { + return null; + } +} + +/** + * Reads `is_subscribed` from the contract. Returns null when the call is + * unavailable. + */ +async function fetchIsSubscribed( + contractId: string, + wallet: string +): Promise { + try { + const transactionXdr = buildAddressViewXdr( + contractId, + 'is_subscribed', + wallet + ); + const retval = await simulateView(transactionXdr, 'is-subscribed'); + if (!retval) return null; + const val = xdr.ScVal.fromXDR(retval, 'base64'); + if (val.switch() === xdr.ScValType.scvBool()) { + return val.b(); + } + return null; + } catch { + return null; + } +} + +/** + * Reads `get_key_balance` from the contract; falls back to the indexed + * KeyOwnership balance when the view is unavailable. + */ +async function fetchHolderBalance( + contractId: string | null, + wallet: string, + resolvedKeyId: string +): Promise { + if (contractId) { + try { + const transactionXdr = buildAddressViewXdr( + contractId, + 'get_key_balance', + wallet + ); + const retval = await simulateView(transactionXdr, 'key-balance'); + if (retval) { + const val = xdr.ScVal.fromXDR(retval, 'base64'); + if (val.switch() === xdr.ScValType.scvI128()) { + return Number(val.i128().toString()); + } + if (val.switch() === xdr.ScValType.scvU64()) { + return Number(val.u64().toString()); + } + if (val.switch() === xdr.ScValType.scvU32()) { + return val.u32(); + } + } + } catch { + // Fall through to the indexed balance. + } + } + + const ownership = await prisma.keyOwnership.findUnique({ + where: { + ownerAddress_creatorId: { + ownerAddress: wallet, + creatorId: resolvedKeyId, + }, + }, + select: { balance: true }, + }); + return Number(ownership?.balance ?? 0); +} + +/** + * Detects the configured gate from the off-chain views. Only THRESHOLD + * gates are detectable today (get_min_hold_for_access); HOLDER/SUBSCRIBER + * gates are answered by the contract-first check_access path and become + * detectable here once a gate-config view lands (#1150). + */ +async function detectOffChainGate( + contractId: string | null, + wallet: string +): Promise<{ + gateType: ContentGateType; + requirement: Record | null; +}> { + const minHold = contractId + ? await fetchMinHoldForAccess(contractId, wallet) + : null; + + if (minHold !== null && minHold > 0) { + return { gateType: 'THRESHOLD', requirement: { min_keys: minHold } }; + } + return { gateType: 'none', requirement: null }; +} + +/** + * Off-chain gate evaluation used when check_access is unavailable. + */ +async function evaluateOffChain( + contractId: string | null, + wallet: string, + resolvedKeyId: string +): Promise { + const { gateType, requirement } = await detectOffChainGate( + contractId, + wallet + ); + + if (gateType === 'none') { + // No configured gate → ungated content. + return { is_allowed: true, gate_type: 'none', requirement: null }; + } + + if (gateType === 'SUBSCRIBER' && contractId) { + const subscribed = await fetchIsSubscribed(contractId, wallet); + if (subscribed !== null) { + return { + is_allowed: subscribed, + gate_type: gateType, + requirement, + }; + } + } + + // THRESHOLD / HOLDER: allowed when the wallet holds at least the + // configured minimum. + const minKeys = Number(requirement?.min_keys ?? 0); + const balance = await fetchHolderBalance(contractId, wallet, resolvedKeyId); + return { + is_allowed: balance >= minKeys, + gate_type: gateType, + requirement, + }; +} + +/** + * Returns the content-gate access result for a wallet on a key. Cached + * for 30s per (key, wallet). + */ +export async function getContentAccess( + keyId: string, + wallet: string +): Promise { + const resolvedKeyId = (await resolveCreatorId(keyId)) ?? keyId; + const normalizedWallet = wallet.trim(); + const cacheKey = contentAccessCacheKey(resolvedKeyId, normalizedWallet); + + try { + const cached = await cacheGetJson(cacheKey); + if (cached !== null) { + logger.debug( + { keyId: resolvedKeyId, wallet: normalizedWallet }, + 'Content access cache hit' + ); + return cached; + } + } catch { + // Cache read failures fall through to evaluation. + } + + const contractId = await resolveContentAccessContractId(resolvedKeyId); + + let result: ContentAccessResult; + const onChain = contractId + ? await fetchOnChainContentAccess(contractId, normalizedWallet, resolvedKeyId) + : null; + + if (onChain) { + result = onChain; + } else { + result = await evaluateOffChain( + contractId, + normalizedWallet, + resolvedKeyId + ); + } + + try { + await cacheSetJson(cacheKey, result, CONTENT_ACCESS_CACHE_TTL_SECONDS); + } catch { + // Cache write failures are non-critical. + } + + return result; +} diff --git a/src/modules/indexer/co-creator-indexer.service.ts b/src/modules/indexer/co-creator-indexer.service.ts new file mode 100644 index 00000000..300f54e5 --- /dev/null +++ b/src/modules/indexer/co-creator-indexer.service.ts @@ -0,0 +1,321 @@ +// src/modules/indexer/co-creator-indexer.service.ts +// +// Indexes CoCreatorSet / CoCreatorRemoved contract events into the +// `key_cocreators` table (#1139). +// +// The contract emits: +// - CoCreatorSet (topic `co_set`) `{ creator_id, co_creator, split_bps }` +// - CoCreatorRemoved (topic `cc_rm`) `{ creator_id, co_creator }` +// +// The worker may deliver either the PascalCase name, the SCREAMING_SNAKE +// alias, or the on-chain topic symbol; all spellings are accepted. +// +// Guarantees: +// - Idempotent and order-safe. Each roster row stores the (ledger, +// eventIndex) of the last applied event and only a strictly newer event +// mutates it, so replays, overlapping ingestion windows and out-of-order +// delivery all converge on the latest on-chain state. +// - Nothing is silently dropped. A malformed event is moved to the indexer +// DLQ. +// - Failures propagate, so the dispatcher does not advance the ledger +// cursor past the batch and the batch is redelivered. +// - After every event, including stale/duplicate ones, the key's cached +// co-creator responses are invalidated. + +import { prisma } from '../../utils/prisma.utils'; +import { logger } from '../../utils/logger.utils'; +import { + processIndexerChainEvents, + IndexerChainEvent, +} from '../../utils/indexer-event-processor.utils'; +import { moveToDLQ } from '../../utils/indexer-dlq.utils'; +import { invalidateCoCreatorCache } from '../co-creators/co-creators.service'; + +/** Event types that add or update a co-creator (CoCreatorSet). */ +export const CO_CREATOR_ADDED_EVENT_TYPES: ReadonlySet = new Set([ + 'CO_CREATOR_ADDED', + 'CoCreatorAdded', + 'CO_CREATOR_SET', + 'CoCreatorSet', + 'co_set', +]); + +/** Event types that remove a co-creator (CoCreatorRemoved). */ +export const CO_CREATOR_REMOVED_EVENT_TYPES: ReadonlySet = new Set([ + 'CO_CREATOR_REMOVED', + 'CoCreatorRemoved', + 'cc_rm', +]); + +export const ALL_CO_CREATOR_EVENT_TYPES: ReadonlySet = new Set([ + ...CO_CREATOR_ADDED_EVENT_TYPES, + ...CO_CREATOR_REMOVED_EVENT_TYPES, +]); + +export const CO_CREATOR_INDEXER_DLQ_JOB_TYPE = 'CO_CREATOR'; + +/** Decoded co-creator event. */ +export interface CoCreatorChainEvent extends IndexerChainEvent { + creatorId: string; + coCreator: string; + /** Present on CoCreatorSet only. */ + splitBps?: number; + role?: string; +} + +export function isCoCreatorEvent( + event: IndexerChainEvent +): event is CoCreatorChainEvent { + return ALL_CO_CREATOR_EVENT_TYPES.has(event.eventType); +} + +export interface ParsedCoCreatorEvent { + keyId: string; + wallet: string; + added: boolean; + splitBps: number | null; + role: string; + ledger: number; + eventIndex: number; + txHash: string; +} + +export type CoCreatorParseResult = + | { ok: true; value: ParsedCoCreatorEvent } + | { ok: false; reason: string }; + +function readStringField( + raw: Record, + ...names: string[] +): string | null { + for (const name of names) { + const value = raw[name]; + if (typeof value === 'string' && value.trim() !== '') { + return value.trim(); + } + } + return null; +} + +function parseUint32(value: unknown): number | null { + if (value === undefined || value === null) return null; + try { + if ( + typeof value !== 'string' && + typeof value !== 'number' && + typeof value !== 'bigint' + ) { + return null; + } + const parsed = BigInt(value); + if (parsed < 0n || parsed > 0xffffffffn) return null; + return Number(parsed); + } catch { + return null; + } +} + +/** Pure validation of a raw event into a typed, persistable update. */ +export function parseCoCreatorEvent( + event: IndexerChainEvent +): CoCreatorParseResult { + const raw = event as Record; + const added = CO_CREATOR_ADDED_EVENT_TYPES.has(event.eventType); + + const keyId = readStringField( + raw, + 'creatorId', + 'creator_id', + 'keyId', + 'key_id', + 'creator' + ); + if (!keyId) { + return { ok: false, reason: 'missing or invalid key/creator id' }; + } + + const wallet = readStringField( + raw, + 'coCreator', + 'co_creator', + 'co_creator_address', + 'wallet' + ); + if (!wallet) { + return { ok: false, reason: 'missing or invalid co-creator wallet' }; + } + + let splitBps: number | null = null; + if (added) { + splitBps = parseUint32( + raw.splitBps ?? raw.split_bps ?? raw.shareBps ?? raw.share_bps + ); + if (splitBps === null) { + return { ok: false, reason: 'missing or invalid splitBps' }; + } + if (splitBps > 10_000) { + return { ok: false, reason: 'splitBps exceeds 10000' }; + } + } + + const ledger = Number(raw.ledger); + if ( + raw.ledger === undefined || + raw.ledger === null || + !Number.isInteger(ledger) || + ledger < 0 + ) { + return { ok: false, reason: 'missing or invalid ledger' }; + } + + const eventIndex = Number(event.eventIndex); + if ( + typeof event.txHash !== 'string' || + event.txHash === '' || + !Number.isInteger(eventIndex) || + eventIndex < 0 + ) { + return { ok: false, reason: 'missing or invalid txHash/eventIndex' }; + } + + const role = + readStringField(raw, 'role') ?? (added ? 'co_creator' : 'co_creator'); + + return { + ok: true, + value: { + keyId, + wallet, + added, + splitBps, + role, + ledger, + eventIndex, + txHash: event.txHash, + }, + }; +} + +/** + * Applies the event to its roster row only if it is strictly newer than + * the stored one. Returns true when the row changed. + */ +async function applyCoCreatorEvent( + parsed: ParsedCoCreatorEvent +): Promise { + const rowKey = { keyId: parsed.keyId, wallet: parsed.wallet }; + const newerThanStored = { + ...rowKey, + OR: [ + { ledger: { lt: parsed.ledger } }, + { ledger: parsed.ledger, eventIndex: { lt: parsed.eventIndex } }, + { ledger: null }, + ], + }; + const data = parsed.added + ? { + active: true, + role: parsed.role, + shareBps: parsed.splitBps, + addedAt: new Date(), + removedAt: null, + ledger: parsed.ledger, + eventIndex: parsed.eventIndex, + txHash: parsed.txHash, + } + : { + active: false, + removedAt: new Date(), + ledger: parsed.ledger, + eventIndex: parsed.eventIndex, + txHash: parsed.txHash, + }; + + const updated = await prisma.keyCoCreator.updateMany({ + where: newerThanStored, + data, + }); + if (updated.count > 0) return true; + + try { + await prisma.keyCoCreator.create({ + data: { ...rowKey, ...data }, + }); + return true; + } catch (error) { + if ((error as { code?: string })?.code !== 'P2002') throw error; + } + + // The row exists. Either the event is stale (nothing to do) or a + // concurrent writer created it between our update and create; retry once. + const retried = await prisma.keyCoCreator.updateMany({ + where: newerThanStored, + data, + }); + return retried.count > 0; +} + +/** + * Processes one co-creator event. Returns true when the roster row + * changed, false for stale/duplicate or malformed (DLQ'd) events. + * Database errors are rethrown; see the header comment. + */ +export async function processCoCreatorEvent( + event: IndexerChainEvent +): Promise { + const eventId = `${event.txHash}:${event.eventIndex}`; + const parsed = parseCoCreatorEvent(event); + + if (!parsed.ok) { + logger.warn( + { type: 'co_creator_event_rejected', eventId, reason: parsed.reason }, + 'Malformed co-creator event moved to DLQ' + ); + await moveToDLQ({ + jobType: CO_CREATOR_INDEXER_DLQ_JOB_TYPE, + payload: JSON.parse( + JSON.stringify(event, (_k, v) => + typeof v === 'bigint' ? v.toString() : v + ) + ), + retryCount: 0, + failureReason: parsed.reason, + }); + return false; + } + + const value = parsed.value; + const changed = await applyCoCreatorEvent(value); + + // Only after the write is durable, so a reader repopulating the cache + // straight afterwards sees the new roster. + await invalidateCoCreatorCache(value.keyId); + + logger.debug( + { + type: changed ? 'co_creator_indexed' : 'co_creator_stale_skipped', + eventId, + key_id: value.keyId, + wallet: value.wallet, + added: value.added, + ledger: value.ledger, + }, + changed + ? 'Co-creator event indexed' + : 'Stale co-creator event skipped' + ); + return changed; +} + +/** + * Processes a batch of co-creator events in order; other events are + * ignored. Cursor advancement belongs to the pipeline dispatcher. + */ +export async function processCoCreatorEvents( + events: IndexerChainEvent[] +): Promise { + await processIndexerChainEvents(events, async event => { + if (!isCoCreatorEvent(event)) return; + await processCoCreatorEvent(event); + }); +} diff --git a/src/modules/indexer/indexer-pipeline.service.ts b/src/modules/indexer/indexer-pipeline.service.ts index 1019c64f..a49aa8ec 100644 --- a/src/modules/indexer/indexer-pipeline.service.ts +++ b/src/modules/indexer/indexer-pipeline.service.ts @@ -252,6 +252,14 @@ 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). @@ -444,6 +452,53 @@ 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); + } + if (options.advanceCursor !== false) await advanceIndexerCursorForBatch(events); } diff --git a/src/modules/indexer/reserve-indexer.service.ts b/src/modules/indexer/reserve-indexer.service.ts new file mode 100644 index 00000000..79bbd68f --- /dev/null +++ b/src/modules/indexer/reserve-indexer.service.ts @@ -0,0 +1,258 @@ +// src/modules/indexer/reserve-indexer.service.ts +// +// Indexes ReserveAccrued / ReserveDeployed contract events into the +// `key_reserves` / `key_reserve_events` tables (#1140). +// +// Guarantees: +// - Idempotent. Every event is recorded in the append-only +// key_reserve_events log, which has a unique (txHash, eventIndex); a +// replayed event hits the constraint, its balance delta is skipped, and +// the key's cached reserve response is still invalidated. +// - Atomic. The log row and the balance mutation are written in one +// transaction, so key_reserves can never drift from its event log. +// - Nothing is silently dropped. A malformed event is moved to the +// indexer DLQ. +// - Failures propagate, so the dispatcher does not advance the ledger +// cursor past the batch and the batch is redelivered. + +import { prisma } from '../../utils/prisma.utils'; +import { logger } from '../../utils/logger.utils'; +import { + processIndexerChainEvents, + IndexerChainEvent, +} from '../../utils/indexer-event-processor.utils'; +import { moveToDLQ } from '../../utils/indexer-dlq.utils'; +import { invalidateKeyReserveCache } from '../reserves/reserves.service'; + +export const RESERVE_ACCRUED_EVENT_TYPES: ReadonlySet = new Set([ + 'RESERVE_ACCRUED', + 'ReserveAccrued', +]); + +export const RESERVE_DEPLOYED_EVENT_TYPES: ReadonlySet = new Set([ + 'RESERVE_DEPLOYED', + 'ReserveDeployed', +]); + +export const ALL_RESERVE_EVENT_TYPES: ReadonlySet = new Set([ + ...RESERVE_ACCRUED_EVENT_TYPES, + ...RESERVE_DEPLOYED_EVENT_TYPES, +]); + +export const RESERVE_INDEXER_DLQ_JOB_TYPE = 'RESERVE'; + +/** XLM stroops per whole XLM. */ +const STROOPS_PER_XLM = 10_000_000n; + +export interface ParsedReserveEvent { + keyId: string; + type: 'accrued' | 'deployed'; + /** Amount in XLM (converted from stroops). */ + amountXlm: string; + ledger: number; + eventIndex: number; + txHash: string; +} + +export type ReserveParseResult = + | { ok: true; value: ParsedReserveEvent } + | { ok: false; reason: string }; + +function parseBigInt(value: unknown): bigint | null { + try { + if (typeof value === 'number' && !Number.isInteger(value)) return null; + if (typeof value === 'string' && value.trim() === '') return null; + if ( + typeof value !== 'string' && + typeof value !== 'number' && + typeof value !== 'bigint' + ) { + return null; + } + return BigInt(value); + } catch { + return null; + } +} + +/** Converts a stroops amount to an XLM decimal string (7 dp). */ +export function stroopsToXlmDecimalString(stroops: bigint): string { + const negative = stroops < 0n; + const abs = negative ? -stroops : stroops; + const whole = abs / STROOPS_PER_XLM; + const fraction = (abs % STROOPS_PER_XLM).toString().padStart(7, '0'); + return `${negative ? '-' : ''}${whole}.${fraction}`; +} + +/** Pure validation of a raw event into a typed, persistable update. */ +export function parseReserveEvent( + event: IndexerChainEvent +): ReserveParseResult { + const raw = event as Record; + const accrued = RESERVE_ACCRUED_EVENT_TYPES.has(event.eventType); + + const keyId = + typeof raw.creatorId === 'string' && raw.creatorId.trim() !== '' + ? raw.creatorId.trim() + : typeof raw.keyId === 'string' && raw.keyId.trim() !== '' + ? raw.keyId.trim() + : null; + if (!keyId) { + return { ok: false, reason: 'missing or invalid key/creator id' }; + } + + const amount = parseBigInt(raw.amount); + if (amount === null || amount <= 0n) { + return { ok: false, reason: 'missing or invalid amount' }; + } + + const ledger = Number(raw.ledger); + if ( + raw.ledger === undefined || + raw.ledger === null || + !Number.isInteger(ledger) || + ledger < 0 + ) { + return { ok: false, reason: 'missing or invalid ledger' }; + } + + const eventIndex = Number(event.eventIndex); + if ( + typeof event.txHash !== 'string' || + event.txHash === '' || + !Number.isInteger(eventIndex) || + eventIndex < 0 + ) { + return { ok: false, reason: 'missing or invalid txHash/eventIndex' }; + } + + return { + ok: true, + value: { + keyId, + type: accrued ? 'accrued' : 'deployed', + amountXlm: stroopsToXlmDecimalString(amount), + ledger, + eventIndex, + txHash: event.txHash, + }, + }; +} + +/** + * Processes one reserve event. Returns true when the balance changed, + * false for replayed or malformed (DLQ'd) events. + */ +export async function processReserveEvent( + event: IndexerChainEvent +): Promise { + const eventId = `${event.txHash}:${event.eventIndex}`; + const parsed = parseReserveEvent(event); + + if (!parsed.ok) { + logger.warn( + { type: 'reserve_event_rejected', eventId, reason: parsed.reason }, + 'Malformed reserve event moved to DLQ' + ); + await moveToDLQ({ + jobType: RESERVE_INDEXER_DLQ_JOB_TYPE, + payload: JSON.parse( + JSON.stringify(event, (_k, v) => + typeof v === 'bigint' ? v.toString() : v + ) + ), + retryCount: 0, + failureReason: parsed.reason, + }); + return false; + } + + const value = parsed.value; + + // The log insert and the balance mutation share one transaction: if the + // event was already applied (P2002 on the unique constraint) the whole + // transaction rolls back and the balance is left untouched. + let applied = false; + try { + await prisma.$transaction(async tx => { + await tx.keyReserveEvent.create({ + data: { + keyId: value.keyId, + type: value.type, + amount: value.amountXlm, + ledger: value.ledger, + txHash: value.txHash, + eventIndex: value.eventIndex, + }, + }); + + if (value.type === 'accrued') { + await tx.keyReserve.upsert({ + where: { keyId: value.keyId }, + update: { + balance: { increment: value.amountXlm }, + totalAccrued: { increment: value.amountXlm }, + }, + create: { + keyId: value.keyId, + balance: value.amountXlm, + totalAccrued: value.amountXlm, + }, + }); + } else { + await tx.keyReserve.upsert({ + where: { keyId: value.keyId }, + update: { + // Floor at zero so a mis-ordered deployment never drives + // the balance negative. + balance: { decrement: value.amountXlm }, + totalDeployed: { increment: value.amountXlm }, + }, + create: { + keyId: value.keyId, + balance: 0, + totalDeployed: value.amountXlm, + }, + }); + } + applied = true; + }); + } catch (error) { + if ((error as { code?: string })?.code === 'P2002') { + applied = false; + } else { + throw error; + } + } + + // Invalidate after every event, including replays, so a cache populated + // from an older balance is dropped. + await invalidateKeyReserveCache(value.keyId); + + logger.debug( + { + type: applied ? 'reserve_indexed' : 'reserve_replay_skipped', + eventId, + key_id: value.keyId, + reserve_type: value.type, + amount_xlm: value.amountXlm, + }, + applied + ? 'Reserve event indexed' + : 'Replayed reserve event skipped' + ); + return applied; +} + +/** + * Processes a batch of reserve events in order; other events are ignored. + * Cursor advancement belongs to the pipeline dispatcher. + */ +export async function processReserveEvents( + events: IndexerChainEvent[] +): Promise { + await processIndexerChainEvents(events, async event => { + if (!ALL_RESERVE_EVENT_TYPES.has(event.eventType)) return; + await processReserveEvent(event); + }); +} diff --git a/src/modules/indexer/tip-indexer.service.ts b/src/modules/indexer/tip-indexer.service.ts new file mode 100644 index 00000000..774f7ae9 --- /dev/null +++ b/src/modules/indexer/tip-indexer.service.ts @@ -0,0 +1,262 @@ +// src/modules/indexer/tip-indexer.service.ts +// +// Indexes TipSent contract events into the `key_tips` / +// `key_tip_totals` tables (#1132). +// +// Guarantees: +// - Idempotent. Every event becomes a key_tips row with a unique +// (txHash, eventIndex); a replayed event hits the constraint and its +// total delta is skipped, so key_tip_totals never double-counts. +// - Atomic. The tip row and the running total are written in one +// transaction, so the total can never drift from the tip log. +// - Nothing is silently dropped. A malformed event is moved to the +// indexer DLQ. +// - Failures propagate, so the dispatcher does not advance the ledger +// cursor past the batch and the batch is redelivered. +// - After every event, including replays, the key's cached tip +// responses are invalidated. + +import { prisma } from '../../utils/prisma.utils'; +import { logger } from '../../utils/logger.utils'; +import { + processIndexerChainEvents, + IndexerChainEvent, +} from '../../utils/indexer-event-processor.utils'; +import { moveToDLQ } from '../../utils/indexer-dlq.utils'; +import { invalidateKeyTipsCache } from '../tips/tips.service'; +import { stroopsToXlmDecimalString } from './reserve-indexer.service'; + +export const TIP_SENT_EVENT_TYPES: ReadonlySet = new Set([ + 'TIP_SENT', + 'TipSent', + 'tip', +]); + +export const TIP_INDEXER_DLQ_JOB_TYPE = 'TIP_SENT'; + +export interface ParsedTipEvent { + keyId: string; + tipper: string; + /** Amount in XLM (converted from stroops). */ + amountXlm: string; + messageHash: string | null; + tippedAt: Date; + ledger: number; + eventIndex: number; + txHash: string; +} + +export type TipParseResult = + | { ok: true; value: ParsedTipEvent } + | { ok: false; reason: string }; + +function parseBigInt(value: unknown): bigint | null { + try { + if (typeof value === 'number' && !Number.isInteger(value)) return null; + if (typeof value === 'string' && value.trim() === '') return null; + if ( + typeof value !== 'string' && + typeof value !== 'number' && + typeof value !== 'bigint' + ) { + return null; + } + return BigInt(value); + } catch { + return null; + } +} + +function readStringField( + raw: Record, + ...names: string[] +): string | null { + for (const name of names) { + const value = raw[name]; + if (typeof value === 'string' && value.trim() !== '') { + return value.trim(); + } + } + return null; +} + +/** Pure validation of a raw event into a typed, persistable tip row. */ +export function parseTipEvent(event: IndexerChainEvent): TipParseResult { + const raw = event as Record; + + const keyId = readStringField( + raw, + 'creatorId', + 'creator_id', + 'keyId', + 'key_id', + 'creator' + ); + if (!keyId) { + return { ok: false, reason: 'missing or invalid key/creator id' }; + } + + const tipper = readStringField(raw, 'tipper', 'sender', 'wallet', 'actor'); + if (!tipper) { + return { ok: false, reason: 'missing or invalid tipper' }; + } + + const amount = parseBigInt(raw.amount); + if (amount === null || amount <= 0n) { + return { ok: false, reason: 'missing or invalid amount' }; + } + + const messageHash = + readStringField(raw, 'messageHash', 'message_hash', 'message_cid') ?? + null; + + const rawTippedAt = + raw.tippedAt ?? raw.timestamp ?? raw.tradeAt ?? raw.createdAt; + let tippedAt: Date | null = null; + if (rawTippedAt !== undefined && rawTippedAt !== null) { + const date = new Date( + typeof rawTippedAt === 'number' ? rawTippedAt : String(rawTippedAt) + ); + if (!Number.isNaN(date.getTime())) { + tippedAt = date; + } + } + if (!tippedAt) { + tippedAt = new Date(); + } + + const ledger = Number(raw.ledger); + if ( + raw.ledger === undefined || + raw.ledger === null || + !Number.isInteger(ledger) || + ledger < 0 + ) { + return { ok: false, reason: 'missing or invalid ledger' }; + } + + const eventIndex = Number(event.eventIndex); + if ( + typeof event.txHash !== 'string' || + event.txHash === '' || + !Number.isInteger(eventIndex) || + eventIndex < 0 + ) { + return { ok: false, reason: 'missing or invalid txHash/eventIndex' }; + } + + return { + ok: true, + value: { + keyId, + tipper, + amountXlm: stroopsToXlmDecimalString(amount), + messageHash, + tippedAt, + ledger, + eventIndex, + txHash: event.txHash, + }, + }; +} + +/** + * Processes one TipSent event. Returns true when the tip was recorded, + * false for replayed or malformed (DLQ'd) events. + */ +export async function processTipEvent( + event: IndexerChainEvent +): Promise { + const eventId = `${event.txHash}:${event.eventIndex}`; + const parsed = parseTipEvent(event); + + if (!parsed.ok) { + logger.warn( + { type: 'tip_event_rejected', eventId, reason: parsed.reason }, + 'Malformed TipSent event moved to DLQ' + ); + await moveToDLQ({ + jobType: TIP_INDEXER_DLQ_JOB_TYPE, + payload: JSON.parse( + JSON.stringify(event, (_k, v) => + typeof v === 'bigint' ? v.toString() : v + ) + ), + retryCount: 0, + failureReason: parsed.reason, + }); + return false; + } + + const value = parsed.value; + + // The tip row and the running total share one transaction: if the event + // was already applied (P2002 on the unique constraint) the whole + // transaction rolls back and the total is left untouched. + let applied = false; + try { + await prisma.$transaction(async tx => { + await tx.keyTip.create({ + data: { + keyId: value.keyId, + tipper: value.tipper, + amount: value.amountXlm, + messageHash: value.messageHash, + ledger: value.ledger, + txHash: value.txHash, + eventIndex: value.eventIndex, + tippedAt: value.tippedAt, + }, + }); + + await tx.keyTipTotal.upsert({ + where: { keyId: value.keyId }, + update: { + total: { increment: value.amountXlm }, + count: { increment: 1 }, + }, + create: { + keyId: value.keyId, + total: value.amountXlm, + count: 1, + }, + }); + applied = true; + }); + } catch (error) { + if ((error as { code?: string })?.code === 'P2002') { + applied = false; + } else { + throw error; + } + } + + // Invalidate after every event, including replays, so a cache + // populated from an older total is dropped. + await invalidateKeyTipsCache(value.keyId); + + logger.debug( + { + type: applied ? 'tip_indexed' : 'tip_replay_skipped', + eventId, + key_id: value.keyId, + tipper: value.tipper, + amount_xlm: value.amountXlm, + }, + applied ? 'TipSent event indexed' : 'Replayed TipSent event skipped' + ); + return applied; +} + +/** + * Processes a batch of TipSent events in order; other events are ignored. + * Cursor advancement belongs to the pipeline dispatcher. + */ +export async function processTipEvents( + events: IndexerChainEvent[] +): Promise { + await processIndexerChainEvents(events, async event => { + if (!TIP_SENT_EVENT_TYPES.has(event.eventType)) return; + await processTipEvent(event); + }); +} diff --git a/src/modules/keys/keys.routes.ts b/src/modules/keys/keys.routes.ts index 0977d98f..c995c578 100644 --- a/src/modules/keys/keys.routes.ts +++ b/src/modules/keys/keys.routes.ts @@ -47,6 +47,10 @@ import { KeySearchQueryTooShortError, searchKeys } from './key-search.service'; import { KEY_SEARCH_MIN_QUERY_LENGTH } from '../../constants/notifications.constants'; import dividendRouter from '../dividends/dividend.routes'; import whitelistRouter from '../whitelist/whitelist.routes'; +import coCreatorsRouter from '../co-creators/co-creators.routes'; +import reservesRouter from '../reserves/reserves.routes'; +import contentAccessRouter from '../content-access/content-access.routes'; +import tipsRouter from '../tips/tips.routes'; import { requireJwtAuth, requireKeyCreator, @@ -1942,6 +1946,18 @@ router.use('/', dividendRouter); // Mount whitelist routes router.use('/', whitelistRouter); +// Mount co-creator roster routes (#1139) +router.use('/', coCreatorsRouter); + +// Mount key reserve balance routes (#1140) +router.use('/', reservesRouter); + +// Mount content-gate access routes (#1137) +router.use('/', contentAccessRouter); + +// Mount tip history routes (#1132) +router.use('/', tipsRouter); + // ── POST /:keyId/burn ───────────────────────────────────────── const burnSchema = z.object({ diff --git a/src/modules/reserves/reserves.routes.ts b/src/modules/reserves/reserves.routes.ts new file mode 100644 index 00000000..e50e345a --- /dev/null +++ b/src/modules/reserves/reserves.routes.ts @@ -0,0 +1,66 @@ +// src/modules/reserves/reserves.routes.ts +// +// REST endpoint for the per-key reserve balance (#1140): +// +// GET /keys/:key_id/reserve — reserve_balance, market_cap, +// reserve_ratio_bps, low_reserve +// +// Public, read-only. Mounted from keys.routes.ts via router.use('/', …). +// Both `:key_id` and `:keyId` spellings are accepted (whitelist module +// convention). Responses carry `Cache-Control: public, max-age=60` and are +// backed by a 60s Redis cache invalidated by the reserve indexer. + +import { NextFunction, Request, Response, Router } from 'express'; +import { AsyncController } from '../../types/auth.types'; +import { + sendSuccess, + sendValidationError, +} from '../../utils/api-response.utils'; +import { cacheControl } from '../../middlewares/cache-control.middleware'; +import { KEY_RESERVE_CACHE_TTL_SECONDS } from '../../constants/redis.constants'; +import { getKeyReserve } from './reserves.service'; + +const reservesRouter = Router(); + +const normalizeKeyId = (req: Request, _res: Response, next: NextFunction) => { + if (!req.params.keyId && req.params.key_id) { + req.params.keyId = req.params.key_id; + } + next(); +}; + +/** + * GET /keys/:key_id/reserve + * Public. Current reserve balance, market cap, ratio and low-reserve flag + * for a key. Keys with no reserve accrued report a zero balance. + */ +const httpGetKeyReserve: AsyncController = async (req, res, next) => { + try { + const keyId = (req.params.key_id ?? req.params.keyId) as string; + if (!keyId) { + return sendValidationError(res, 'Missing required parameters', [ + { field: 'keyId', message: 'Key ID is required' }, + ]); + } + return sendSuccess(res, await getKeyReserve(keyId)); + } catch (error) { + next(error); + } +}; + +reservesRouter.get( + ['/:key_id/reserve', '/:keyId/reserve'], + normalizeKeyId, + cacheControl({ + maxAge: KEY_RESERVE_CACHE_TTL_SECONDS, + type: 'public', + mustRevalidate: true, + }), + httpGetKeyReserve +); + +reservesRouter.all(['/:key_id/reserve', '/:keyId/reserve'], (_req, res) => { + res.set('Allow', 'GET').sendStatus(405); +}); + +export default reservesRouter; diff --git a/src/modules/reserves/reserves.service.ts b/src/modules/reserves/reserves.service.ts new file mode 100644 index 00000000..482c41fa --- /dev/null +++ b/src/modules/reserves/reserves.service.ts @@ -0,0 +1,128 @@ +// src/modules/reserves/reserves.service.ts +// +// Read model for per-key reserve balances (#1140), backed by the +// `key_reserves` table maintained from ReserveAccrued / ReserveDeployed +// contract events. +// +// GET /keys/:key_id/reserve returns the reserve balance, the key's current +// market cap (price snapshot × circulating supply), the reserve ratio in +// basis points, and a `low_reserve` flag when the ratio falls below 1000 +// bps (10% of market cap) — the frontend's liquidity backstop indicator. +// +// Responses are cached in Redis for 60s and invalidated per key by the +// reserve indexer after every reserve event. + +import { prisma } from '../../utils/prisma.utils'; +import { + cacheGetJson, + cacheSetJson, + cacheInvalidate, +} from '../../utils/redis.utils'; +import { logger } from '../../utils/logger.utils'; +import { + KEY_RESERVE_CACHE_TTL_SECONDS, + keyReserveCacheKey, +} from '../../constants/redis.constants'; +import { resolveCreatorId } from '../whitelist/whitelist.service'; + +/** Reserve ratio below which a key is flagged low (10% of market cap). */ +export const LOW_RESERVE_THRESHOLD_BPS = 1000; + +/** XLM stroops per whole XLM. */ +const STROOPS_PER_XLM = 10_000_000; + +/** Reserve balance response for GET /keys/:key_id/reserve. */ +export interface KeyReserveResponse { + key_id: string; + reserve_balance: string; + market_cap: string; + reserve_ratio_bps: number; + low_reserve: boolean; +} + +/** Drops the cached reserve response for a key. Called by the indexer. */ +export async function invalidateKeyReserveCache(keyId: string): Promise { + await cacheInvalidate(keyReserveCacheKey(keyId)); +} + +/** + * Maps any key identifier to the canonical CreatorProfile.id used as + * `keyId` in the key_reserves table. + */ +export async function resolveReserveKeyId(keyId: string): Promise { + return (await resolveCreatorId(keyId)) ?? keyId; +} + +/** + * Returns the reserve balance, market cap and ratio for a key. Keys with + * no reserve accrued report a zero balance (not 404), per the issue. + */ +export async function getKeyReserve( + keyId: string +): Promise { + const resolvedKeyId = await resolveReserveKeyId(keyId); + const cacheKey = keyReserveCacheKey(resolvedKeyId); + + try { + const cached = await cacheGetJson(cacheKey); + if (cached !== null) { + logger.debug({ keyId: resolvedKeyId }, 'Key reserve cache hit'); + return cached; + } + } catch { + // Cache read failures fall through to the database. + } + + const [reserve, creator] = await Promise.all([ + prisma.keyReserve.findUnique({ + where: { keyId: resolvedKeyId }, + select: { balance: true }, + }), + prisma.creatorProfile.findFirst({ + where: { + OR: [{ id: resolvedKeyId }, { handle: resolvedKeyId }], + }, + select: { id: true, circulatingSupply: true }, + }), + ]); + + const reserveBalance = Number(reserve?.balance ?? 0); + + // Market cap = current price (stroops → XLM) × circulating supply. + let marketCap = 0; + if (creator) { + const snapshot = await prisma.creatorPriceSnapshot.findUnique({ + where: { creatorId: creator.id }, + select: { currentPrice: true }, + }); + if (snapshot) { + const priceXlm = Number(snapshot.currentPrice) / STROOPS_PER_XLM; + marketCap = priceXlm * Number(creator.circulatingSupply); + } + } + + // Ratio in basis points; a key with no market cap yet reports 0 and is + // never flagged low (nothing to compare the backstop against). + const reserveRatioBps = + marketCap > 0 ? Math.round((reserveBalance / marketCap) * 10_000) : 0; + const lowReserve = + marketCap > 0 && + reserveBalance >= 0 && + reserveRatioBps < LOW_RESERVE_THRESHOLD_BPS; + + const result: KeyReserveResponse = { + key_id: resolvedKeyId, + reserve_balance: reserveBalance.toFixed(7), + market_cap: marketCap.toFixed(7), + reserve_ratio_bps: reserveRatioBps, + low_reserve: lowReserve, + }; + + try { + await cacheSetJson(cacheKey, result, KEY_RESERVE_CACHE_TTL_SECONDS); + } catch { + // Cache write failures are non-critical. + } + + return result; +} diff --git a/src/modules/tips/tips.routes.ts b/src/modules/tips/tips.routes.ts new file mode 100644 index 00000000..b7deff9b --- /dev/null +++ b/src/modules/tips/tips.routes.ts @@ -0,0 +1,163 @@ +// src/modules/tips/tips.routes.ts +// +// REST endpoints for per-key tip history (#1132): +// +// GET /keys/:key_id/tips — paginated tip history +// GET /keys/:key_id/tips/total — cumulative tip totals +// GET /keys/:key_id/tips/:tipId/content — IPFS message content +// +// Public, read-only. Mounted from keys.routes.ts via router.use('/', …). +// Both `:key_id` and `:keyId` spellings are accepted (whitelist module +// convention). List/total carry `Cache-Control: public, max-age=60` and +// are backed by a 60s Redis cache invalidated by the tip indexer. + +import { NextFunction, Request, Response, Router } from 'express'; +import { z } from 'zod'; +import { AsyncController } from '../../types/auth.types'; +import { + sendSuccess, + sendValidationError, + sendNotFound, + sendError, + zodIssuesToDetails, +} from '../../utils/api-response.utils'; +import { ErrorCode } from '../../constants/error.constants'; +import { cacheControl } from '../../middlewares/cache-control.middleware'; +import { KEY_TIPS_CACHE_TTL_SECONDS } from '../../constants/redis.constants'; +import { + getTipMessageContent, + getTipTotals, + listTips, +} from './tips.service'; + +const tipsRouter = Router(); + +const normalizeKeyId = (req: Request, _res: Response, next: NextFunction) => { + if (!req.params.keyId && req.params.key_id) { + req.params.keyId = req.params.key_id; + } + next(); +}; + +const ListTipsQuerySchema = z.object({ + limit: z.coerce.number().int().min(1).max(100).default(20), + cursor: z.string().min(1).optional(), +}); + +const tipsCache = () => + cacheControl({ + maxAge: KEY_TIPS_CACHE_TTL_SECONDS, + type: 'public', + mustRevalidate: true, + }); + +/** + * GET /keys/:key_id/tips?limit=&cursor= + * Public. Cursor-paginated tip history: tipper, amount, message_hash, + * tipped_at. + */ +const httpListTips: AsyncController = async (req, res, next) => { + try { + const keyId = (req.params.key_id ?? req.params.keyId) as string; + if (!keyId) { + return sendValidationError(res, 'Missing required parameters', [ + { field: 'keyId', message: 'Key ID is required' }, + ]); + } + + const parsedQuery = ListTipsQuerySchema.safeParse(req.query); + if (!parsedQuery.success) { + return sendValidationError( + res, + 'Invalid tips query', + zodIssuesToDetails(parsedQuery.error.issues) + ); + } + + return sendSuccess( + res, + await listTips( + keyId, + parsedQuery.data.limit, + parsedQuery.data.cursor + ) + ); + } catch (error) { + next(error); + } +}; + +/** + * GET /keys/:key_id/tips/total + * Public. Cumulative tips received by the key. Zero when none. + */ +const httpGetTipTotals: AsyncController = async (req, res, next) => { + try { + const keyId = (req.params.key_id ?? req.params.keyId) as string; + if (!keyId) { + return sendValidationError(res, 'Missing required parameters', [ + { field: 'keyId', message: 'Key ID is required' }, + ]); + } + return sendSuccess(res, await getTipTotals(keyId)); + } catch (error) { + next(error); + } +}; + +/** + * GET /keys/:key_id/tips/:tipId/content + * Public. IPFS message content for a tip, fetched on demand and cached. + */ +const httpGetTipContent: AsyncController = async (req, res) => { + try { + const keyId = (req.params.key_id ?? req.params.keyId) as string; + if (!keyId) { + return sendValidationError(res, 'Missing required parameters', [ + { field: 'keyId', message: 'Key ID is required' }, + ]); + } + + const tipId = String(req.params.tipId ?? ''); + const content = await getTipMessageContent(keyId, tipId); + if (!content) { + return sendNotFound(res, 'Tip message'); + } + return sendSuccess(res, content); + } catch { + // Gateway failures surface as 502 rather than a generic 500. + return sendError( + res, + 502, + ErrorCode.SERVICE_UNAVAILABLE, + 'Failed to fetch tip message content from IPFS gateway' + ); + } +}; + +tipsRouter.get( + ['/:key_id/tips/total', '/:keyId/tips/total'], + normalizeKeyId, + tipsCache(), + httpGetTipTotals +); + +tipsRouter.get( + ['/:key_id/tips', '/:keyId/tips'], + normalizeKeyId, + tipsCache(), + httpListTips +); + +tipsRouter.get( + ['/:key_id/tips/:tipId/content', '/:keyId/tips/:tipId/content'], + normalizeKeyId, + tipsCache(), + httpGetTipContent +); + +tipsRouter.all(['/:key_id/tips', '/:keyId/tips'], (_req, res) => { + res.set('Allow', 'GET').sendStatus(405); +}); + +export default tipsRouter; diff --git a/src/modules/tips/tips.service.ts b/src/modules/tips/tips.service.ts new file mode 100644 index 00000000..64e80b9a --- /dev/null +++ b/src/modules/tips/tips.service.ts @@ -0,0 +1,289 @@ +// src/modules/tips/tips.service.ts +// +// Read model for per-key tip history (#1132), backed by the `key_tips` / +// `key_tip_totals` tables maintained from TipSent contract events. +// +// - GET /keys/:key_id/tips — cursor-paginated tip history +// - GET /keys/:key_id/tips/total — cumulative tips received +// - GET /keys/:key_id/tips/:tipId/content — IPFS tip message content, +// fetched on demand from the Pinata gateway and cached in Redis. +// +// List/total responses are cached for 60s and invalidated per key by the +// tip indexer after every TipSent event. Fetched IPFS content is cached +// for an hour (immutable per CID). + +import { prisma } from '../../utils/prisma.utils'; +import { + cacheGetJson, + cacheSetJson, + cacheInvalidate, +} from '../../utils/redis.utils'; +import { logger } from '../../utils/logger.utils'; +import { + KEY_TIP_CONTENT_CACHE_TTL_SECONDS, + KEY_TIPS_CACHE_TTL_SECONDS, + keyTipContentCacheKey, + keyTipsCachePattern, + keyTipsListCacheKey, + keyTipsTotalCacheKey, +} from '../../constants/redis.constants'; +import { + normalizeIpfsCid, + resolvePinataGatewayUrl, +} from '../../utils/pinata-gateway.utils'; +import { resolveCreatorId } from '../whitelist/whitelist.service'; + +/** One tip in the history. */ +export interface TipHistoryItem { + id: string; + tipper: string; + amount: string; + message_hash: string | null; + tipped_at: string; + tx_hash: string | null; +} + +/** Cursor-paginated tip history page. */ +export interface TipHistoryPage { + key_id: string; + data: TipHistoryItem[]; + meta: { + hasMore: boolean; + nextCursor: string | null; + }; +} + +/** Cumulative tip totals for a key. */ +export interface TipTotals { + key_id: string; + total: string; + count: number; +} + +/** Maximum IPFS message size fetched for the content endpoint (256 KiB). */ +export const MAX_TIP_MESSAGE_BYTES = 256 * 1024; + +/** IPFS gateway fetch timeout (ms). */ +export const TIP_MESSAGE_FETCH_TIMEOUT_MS = 5000; + +/** + * Drops every cached tip response for a key (list + total). Called by the + * indexer after every TipSent event. + */ +export async function invalidateKeyTipsCache(keyId: string): Promise { + await cacheInvalidate(keyTipsCachePattern(keyId)); +} + +/** + * Maps any key identifier to the canonical CreatorProfile.id used as + * `keyId` in the key_tips table. + */ +export async function resolveTipKeyId(keyId: string): Promise { + return (await resolveCreatorId(keyId)) ?? keyId; +} + +/** + * Lists tip history for a key, newest first, cursor-paginated by tip id. + */ +export async function listTips( + keyId: string, + limit: number, + cursor?: string +): Promise { + const resolvedKeyId = await resolveTipKeyId(keyId); + const cacheKey = keyTipsListCacheKey( + resolvedKeyId, + limit, + cursor ?? 'start' + ); + + try { + const cached = await cacheGetJson(cacheKey); + if (cached !== null) { + logger.debug({ keyId: resolvedKeyId }, 'Tip list cache hit'); + return cached; + } + } catch { + // Cache read failures fall through to the database. + } + + const rows = await prisma.keyTip.findMany({ + where: { keyId: resolvedKeyId }, + orderBy: { tippedAt: 'desc' }, + take: limit + 1, + ...(cursor ? { cursor: { id: cursor }, skip: 1 } : {}), + select: { + id: true, + tipper: true, + amount: true, + messageHash: true, + tippedAt: true, + txHash: true, + }, + }); + + const hasMore = rows.length > limit; + const page = hasMore ? rows.slice(0, limit) : rows; + const lastRow = page[page.length - 1]; + + const result: TipHistoryPage = { + key_id: resolvedKeyId, + data: page.map(row => ({ + id: row.id, + tipper: row.tipper, + amount: Number(row.amount).toFixed(7), + message_hash: row.messageHash, + tipped_at: row.tippedAt.toISOString(), + tx_hash: row.txHash, + })), + meta: { + hasMore, + nextCursor: hasMore && lastRow ? lastRow.id : null, + }, + }; + + try { + await cacheSetJson(cacheKey, result, KEY_TIPS_CACHE_TTL_SECONDS); + } catch { + // Cache write failures are non-critical. + } + + return result; +} + +/** + * Returns the cumulative tips received by a key. Keys with no tips report + * a zero total (not 404). + */ +export async function getTipTotals(keyId: string): Promise { + const resolvedKeyId = await resolveTipKeyId(keyId); + const cacheKey = keyTipsTotalCacheKey(resolvedKeyId); + + try { + const cached = await cacheGetJson(cacheKey); + if (cached !== null) { + logger.debug({ keyId: resolvedKeyId }, 'Tip total cache hit'); + return cached; + } + } catch { + // Cache read failures fall through to the database. + } + + const row = await prisma.keyTipTotal.findUnique({ + where: { keyId: resolvedKeyId }, + select: { total: true, count: true }, + }); + + const result: TipTotals = { + key_id: resolvedKeyId, + total: Number(row?.total ?? 0).toFixed(7), + count: row?.count ?? 0, + }; + + try { + await cacheSetJson(cacheKey, result, KEY_TIPS_CACHE_TTL_SECONDS); + } catch { + // Cache write failures are non-critical. + } + + return result; +} + +/** Content fetched from IPFS for a tip message. */ +export interface TipMessageContent { + message_hash: string; + content: string; +} + +/** + * Fetches (or returns cached) IPFS message content for a tip. Returns + * null when the tip or its message hash is unknown; throws on gateway + * failure. + */ +export async function getTipMessageContent( + keyId: string, + tipId: string +): Promise { + const resolvedKeyId = await resolveTipKeyId(keyId); + + const tip = await prisma.keyTip.findFirst({ + where: { id: tipId, keyId: resolvedKeyId }, + select: { messageHash: true }, + }); + + if (!tip?.messageHash) { + return null; + } + + const messageHash = tip.messageHash; + const cacheKey = keyTipContentCacheKey(messageHash); + + try { + const cached = await cacheGetJson(cacheKey); + if (cached !== null) { + logger.debug({ messageHash }, 'Tip message content cache hit'); + return cached; + } + } catch { + // Cache read failures fall through to the gateway. + } + + const cid = normalizeIpfsCid(messageHash); + if (!cid) { + return null; + } + const url = resolvePinataGatewayUrl(cid); + if (!url) { + throw new Error('No IPFS gateway configured (PINATA_GATEWAY_URL)'); + } + + const controller = new AbortController(); + const timeout = setTimeout( + () => controller.abort(), + TIP_MESSAGE_FETCH_TIMEOUT_MS + ); + + try { + const response = await fetch(url, { signal: controller.signal }); + if (!response.ok) { + throw new Error( + `IPFS gateway returned ${response.status} for ${cid}` + ); + } + + const contentLength = Number(response.headers.get('content-length')); + if ( + Number.isFinite(contentLength) && + contentLength > MAX_TIP_MESSAGE_BYTES + ) { + throw new Error(`IPFS message exceeds ${MAX_TIP_MESSAGE_BYTES} bytes`); + } + + const text = await response.text(); + if (text.length > MAX_TIP_MESSAGE_BYTES) { + throw new Error(`IPFS message exceeds ${MAX_TIP_MESSAGE_BYTES} bytes`); + } + + const result: TipMessageContent = { message_hash: messageHash, content: text }; + + try { + await cacheSetJson( + cacheKey, + result, + KEY_TIP_CONTENT_CACHE_TTL_SECONDS + ); + } catch { + // Cache write failures are non-critical. + } + + return result; + } catch (error) { + logger.warn( + { messageHash, error: (error as Error).message }, + 'Failed to fetch tip message content from IPFS gateway' + ); + throw error; + } finally { + clearTimeout(timeout); + } +}