feat: add creator key trading operations

prisma/schema/creator.prisma

@@ -9,6 +9,8 @@ model CreatorProfile {
   avatarUrl       String?
   perkSummary     String?
   isVerified      Boolean  @default(false)
+  tradingPaused   Boolean  @default(false)
+  circulatingSupply Decimal @default(0)
   perks           Json?
   followersCount  Int      @default(0)
   createdAt       DateTime @default(now())
@@ -17,10 +19,29 @@ model CreatorProfile {
   user           User                   @relation(fields: [userId], references: [id], onDelete: Cascade)
   priceSnapshot  CreatorPriceSnapshot?
   priceHistory   CreatorPriceHistory[]
+  pendingPurchases PendingKeyPurchase[]
   posts          CreatorPost[]
   followers      Follow[]
 }
 
+model PendingKeyPurchase {
+  id             String   @id @default(cuid())
+  creatorId      String
+  buyerAddress   String
+  memo           String   @unique
+  quantity       Decimal
+  status         String   @default("PENDING")
+  transactionHash String?
+  settledAt      DateTime?
+  createdAt      DateTime @default(now())
+  updatedAt      DateTime @updatedAt
+
+  creator CreatorProfile @relation(fields: [creatorId], references: [id], onDelete: Cascade)
+
+  @@index([creatorId, status])
+  @@index([buyerAddress, status])
+}
+
 model CreatorPost {
   id        String         @id @default(cuid())
   content   String

prisma/schema/migrations/20260826000000_add_creator_key_operations/migration.sql

@@ -0,0 +1,26 @@
+ALTER TABLE "CreatorProfile"
+ADD COLUMN "tradingPaused" BOOLEAN NOT NULL DEFAULT false,
+ADD COLUMN "circulatingSupply" DECIMAL NOT NULL DEFAULT 0;
+
+CREATE TABLE "PendingKeyPurchase" (
+    "id" TEXT NOT NULL,
+    "creatorId" TEXT NOT NULL,
+    "buyerAddress" TEXT NOT NULL,
+    "memo" TEXT NOT NULL,
+    "quantity" DECIMAL NOT NULL,
+    "status" TEXT NOT NULL DEFAULT 'PENDING',
+    "transactionHash" TEXT,
+    "settledAt" TIMESTAMP(3),
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "PendingKeyPurchase_pkey" PRIMARY KEY ("id")
+);
+
+CREATE UNIQUE INDEX "PendingKeyPurchase_memo_key" ON "PendingKeyPurchase"("memo");
+CREATE INDEX "PendingKeyPurchase_creatorId_status_idx" ON "PendingKeyPurchase"("creatorId", "status");
+CREATE INDEX "PendingKeyPurchase_buyerAddress_status_idx" ON "PendingKeyPurchase"("buyerAddress", "status");
+
+ALTER TABLE "PendingKeyPurchase"
+ADD CONSTRAINT "PendingKeyPurchase_creatorId_fkey"
+FOREIGN KEY ("creatorId") REFERENCES "CreatorProfile"("id") ON DELETE CASCADE ON UPDATE CASCADE;

src/config.schema.ts

@@ -167,6 +167,7 @@ export const envSchema = z
          )
          .default('https://soroban-testnet.stellar.org'),
       STELLAR_AUTH_SECRET: optionalNonEmptyString,
+      HORIZON_WEBHOOK_SECRET: optionalNonEmptyString,
 
       // Shared secret that lets trusted internal services bypass per-wallet
       // rate limits (e.g. the buy endpoint's sliding window limiter).

src/middlewares/admin-guard.middleware.ts

@@ -1,4 +1,7 @@
 import { Request, Response, NextFunction } from 'express';
+import jwt from 'jsonwebtoken';
+import { envConfig } from '../config';
+import { sendForbidden, sendUnauthorized } from '../utils/api-response.utils';
 
 export interface AdminRequest extends Request {
    adminId?: string;
@@ -9,23 +12,26 @@ export function adminGuard(
    res: Response,
    next: NextFunction
 ): void {
-   const adminIdHeader = req.headers['x-admin-id'];
-   const adminId =
-      typeof adminIdHeader === 'string'
-         ? adminIdHeader
-         : Array.isArray(adminIdHeader)
-           ? adminIdHeader[0]
-           : undefined;
-
-   if (!adminId) {
-      res.status(403).json({
-         type: 'FORBIDDEN',
-         message: 'Admin authorization required.',
-         timestamp: new Date().toISOString(),
-      });
+   const authHeader = req.headers.authorization;
+   if (!authHeader?.startsWith('Bearer ')) {
+      sendUnauthorized(res, 'Missing or invalid authorization header');
       return;
    }
 
-   req.adminId = adminId;
-   next();
+   try {
+      const payload = jwt.verify(authHeader.slice(7), envConfig.JWT_SECRET) as {
+         sub?: string;
+         adminId?: string;
+         role?: string;
+      };
+      const adminId = payload.adminId ?? payload.sub;
+      if (payload.role !== 'admin' || !adminId) {
+         sendForbidden(res, 'Admin authorization required');
+         return;
+      }
+      req.adminId = adminId;
+      next();
+   } catch {
+      sendUnauthorized(res, 'Invalid or expired token');
+   }
 }

src/modules/admin/admin.controllers.ts

@@ -202,3 +202,37 @@ export const httpReplayIndexerEvents: AsyncController = async (
       next(error);
    }
 };
+
+export const httpSetKeyTradingPaused = async (
+   req: AdminRequest,
+   res: Response,
+   next: (error: unknown) => void
+): Promise<void> => {
+   try {
+      const creatorId = String(req.params.keyId);
+      const tradingPaused = req.path.endsWith('/pause');
+      const creator = await prisma.creatorProfile.findUnique({
+         where: { id: creatorId },
+         select: { id: true, tradingPaused: true },
+      });
+      if (!creator) {
+         sendCreatorParamNotFound(res);
+         return;
+      }
+      const updated = await prisma.creatorProfile.update({
+         where: { id: creatorId },
+         data: { tradingPaused },
+      });
+      if (creator.tradingPaused !== tradingPaused) {
+         await emitAuditEvent({
+            actor: req.adminId!,
+            action: tradingPaused ? 'key_trading_paused' : 'key_trading_resumed',
+            target: 'CreatorKey',
+            targetId: creatorId,
+         });
+      }
+      sendSuccess(res, updated);
+   } catch (error) {
+      next(error);
+   }
+};

src/modules/admin/admin.routes.ts

@@ -2,12 +2,15 @@ import { Router } from 'express';
 import {
    httpUpdateCreatorMetadata,
    httpReplayIndexerEvents,
+   httpSetKeyTradingPaused,
 } from './admin.controllers';
 import { adminGuard } from '../../middlewares/admin-guard.middleware';
 
 const adminRouter = Router();
 
 adminRouter.patch('/creators/:id/metadata', httpUpdateCreatorMetadata);
 adminRouter.post('/indexer/replay', adminGuard, httpReplayIndexerEvents);
+adminRouter.post('/keys/:keyId/pause', adminGuard, httpSetKeyTradingPaused);
+adminRouter.post('/keys/:keyId/resume', adminGuard, httpSetKeyTradingPaused);
 
 export default adminRouter;

src/modules/creator/buy.controller.ts

@@ -4,6 +4,7 @@ import type { StellarSignedRequest } from '../../middlewares/stellar-signature.m
 import { ErrorCode } from '../../constants/error.constants';
 import { sendError, sendSuccess } from '../../utils/api-response.utils';
 import { buyGateway } from './buy.service';
+import { assertTradingActive, TradingPausedError } from '../keys/key-trading.service';
 
 export const buySchema = z.object({
    quantity: z.number().int().positive(),
@@ -22,6 +23,15 @@ export async function httpBuyCreatorKey(
    const body = req.body as BuyRequestBody;
 
    const walletAddress = req.walletAddress!;
+   try {
+      await assertTradingActive(String(req.params.id));
+   } catch (error) {
+      if (error instanceof TradingPausedError) {
+         sendError(res, 503, ErrorCode.INTERNAL_ERROR, error.message);
+         return;
+      }
+      throw error;
+   }
    const required = body.key_cost_xlm * body.quantity + body.fee_xlm;
    const balance = await buyGateway.getXlmBalance(walletAddress);
    if (balance < required) {

src/modules/index.ts

@@ -15,6 +15,8 @@ import walletsRouter from './wallets/wallets.routes';
 import alertsRouter from './alerts/alert.router';
 import tradingRouter from './trading/multi-buy.routes';
 import sequencerRouter from './admin/sequencer.routes';
+import keyPriceHistoryRouter from './keys/key-price-history.routes';
+import horizonWebhookRouter from './webhooks/horizon-webhook.routes';
 import { BASE as CREATORS_BASE } from '../constants/creator.constants';
 import { routeBodySizeLimit } from '../middlewares/body-size-limit.middleware';
 
@@ -40,5 +42,7 @@ router.use('/wallets', routeBodySizeLimit('default'), walletsRouter);
 router.use('/alerts', routeBodySizeLimit('default'), alertsRouter);
 router.use('/trading', routeBodySizeLimit('default'), tradingRouter);
 router.use('/internal', routeBodySizeLimit('default'), sequencerRouter);
+router.use('/keys', routeBodySizeLimit('default'), keyPriceHistoryRouter);
+router.use('/webhooks', routeBodySizeLimit('default'), horizonWebhookRouter);
 
 export default router;

src/modules/indexer/indexer-pipeline.service.ts

@@ -7,6 +7,7 @@ import { logger } from '../../utils/logger.utils';
 import { processIndexerChainEvents, IndexerChainEvent } from '../../utils/indexer-event-processor.utils';
 import { dedupeChainEvents } from '../../utils/indexer-dedupe.utils';
 import { logSellTransactionConfirmed } from '../../utils/sell-transaction-logger.utils';
+import { persistCirculatingSupply } from './persist-circulating-supply.service';
 import { invalidateVolumeLeaderboardCache } from '../creators/creator-leaderboard-volume.service';
 
 /**
@@ -75,6 +76,8 @@ export async function processTradeEvents(events: IndexerChainEvent[]): Promise<v
          ledger: Number(ledger),
       });
 
+      await persistCirculatingSupply(creatorId);
+
       // 4. Emit a structured log for confirmed sells, mirroring buy-side logging.
       if (event.eventType === 'KEY_SOLD') {
          const [creatorProfile, supplyAggregate] = await Promise.all([

src/modules/indexer/persist-circulating-supply.service.ts

@@ -0,0 +1,36 @@
+import { prisma } from '../../utils/prisma.utils';
+
+const MAX_ATTEMPTS = 3;
+
+function delay(milliseconds: number): Promise<void> {
+   return new Promise(resolve => setTimeout(resolve, milliseconds));
+}
+
+export async function persistCirculatingSupply(creatorId: string): Promise<void> {
+   let lastError: unknown;
+   for (let attempt = 0; attempt < MAX_ATTEMPTS; attempt += 1) {
+      try {
+         await prisma.$transaction(async transaction => {
+            const activities = await transaction.activity.findMany({
+               where: { creatorId, type: { in: ['KEY_BOUGHT', 'KEY_SOLD'] } },
+               select: { type: true, payload: true },
+            });
+            const supply = activities.reduce((total, activity) => {
+               const amount = Number((activity.payload as { amount?: number }).amount ?? 0);
+               return activity.type === 'KEY_BOUGHT' ? total + amount : total - amount;
+            }, 0);
+            await transaction.creatorProfile.update({
+               where: { id: creatorId },
+               data: { circulatingSupply: supply },
+            });
+         });
+         return;
+      } catch (error) {
+         lastError = error;
+         if (attempt < MAX_ATTEMPTS - 1) {
+            await delay(100 * 2 ** attempt);
+         }
+      }
+   }
+   throw lastError;
+}

src/modules/keys/key-price-history.routes.ts

@@ -0,0 +1,33 @@
+import { Router } from 'express';
+import { z } from 'zod';
+import { sendError, sendSuccess, zodIssuesToDetails } from '../../utils/api-response.utils';
+import { ErrorCode } from '../../constants/error.constants';
+import { getKeyPriceHistory, PRICE_HISTORY_INTERVALS } from './key-price-history.service';
+
+const querySchema = z.object({
+   from: z.string().datetime(),
+   to: z.string().datetime(),
+   interval: z.enum(PRICE_HISTORY_INTERVALS),
+});
+const router = Router();
+
+router.get('/:keyId/price-history', async (req, res, next) => {
+   const parsed = querySchema.safeParse(req.query);
+   if (!parsed.success) {
+      sendError(res, 400, ErrorCode.VALIDATION_ERROR, 'Invalid price-history query', zodIssuesToDetails(parsed.error.issues));
+      return;
+   }
+   const from = new Date(parsed.data.from);
+   const to = new Date(parsed.data.to);
+   if (from > to) {
+      sendError(res, 400, ErrorCode.BAD_REQUEST, 'from must be before or equal to to');
+      return;
+   }
+   try {
+      sendSuccess(res, await getKeyPriceHistory(req.params.keyId, from, to, parsed.data.interval));
+   } catch (error) {
+      next(error);
+   }
+});
+
+export default router;

src/modules/keys/key-price-history.service.ts

@@ -0,0 +1,30 @@
+import { prisma } from '../../utils/prisma.utils';
+
+export const PRICE_HISTORY_INTERVALS = ['1h', '24h', '7d'] as const;
+export type PriceHistoryInterval = (typeof PRICE_HISTORY_INTERVALS)[number];
+
+const intervalMs: Record<PriceHistoryInterval, number> = {
+   '1h': 60 * 60 * 1000,
+   '24h': 24 * 60 * 60 * 1000,
+   '7d': 7 * 24 * 60 * 60 * 1000,
+};
+
+export async function getKeyPriceHistory(
+   creatorId: string,
+   from: Date,
+   to: Date,
+   interval: PriceHistoryInterval
+) {
+   const snapshots = await prisma.creatorPriceHistory.findMany({
+      where: { creatorId, recordedAt: { gte: from, lte: to } },
+      orderBy: { recordedAt: 'asc' },
+   });
+   const buckets = new Map<number, (typeof snapshots)[number]>();
+   for (const snapshot of snapshots) {
+      const bucket = Math.floor(snapshot.recordedAt.getTime() / intervalMs[interval]) * intervalMs[interval];
+      buckets.set(bucket, snapshot);
+   }
+   return Array.from(buckets.entries())
+      .slice(0, 500)
+      .map(([timestamp, snapshot]) => ({ timestamp: new Date(timestamp), price: snapshot.price }));
+}

src/modules/keys/key-trading.service.ts

@@ -0,0 +1,17 @@
+import { prisma } from '../../utils/prisma.utils';
+
+export class TradingPausedError extends Error {
+   constructor() {
+      super('Trading paused for this key');
+   }
+}
+
+export async function assertTradingActive(creatorId: string): Promise<void> {
+   const creator = await prisma.creatorProfile.findUnique({
+      where: { id: creatorId },
+      select: { tradingPaused: true },
+   });
+   if (creator?.tradingPaused) {
+      throw new TradingPausedError();
+   }
+}

src/modules/trading/multi-buy.controllers.ts

@@ -9,6 +9,7 @@ import {
    ErrorCode,
 } from '../../utils/api-response.utils';
 import { horizonGet } from '../../clients/horizon.client';
+import { assertTradingActive, TradingPausedError } from '../keys/key-trading.service';
 
 async function getCurrentLedger(): Promise<number> {
    const res = await horizonGet('/');
@@ -67,6 +68,8 @@ export const httpMultiBuy: AsyncController = async (req, res, next) => {
 
       const { buyer, legs, global_deadline_ledger } = parsed.data;
 
+      await Promise.all(legs.map(leg => assertTradingActive(leg.creator)));
+
       const results = await executeMultiBuy(
          buyer,
          legs,
@@ -80,6 +83,10 @@ export const httpMultiBuy: AsyncController = async (req, res, next) => {
 
       sendSuccess(res, results);
    } catch (err) {
+      if (err instanceof TradingPausedError) {
+         sendError(res, 503, ErrorCode.INTERNAL_ERROR, err.message);
+         return;
+      }
       if (err instanceof MultiBuyError) {
          const statusMap: Record<string, number> = {
             legs_empty: 400,