From cf5a5d1164d9e0ceb53749749fbdc289ea0e18d5 Mon Sep 17 00:00:00 2001 From: Akanimoh12 Date: Wed, 26 Aug 2026 01:07:33 +0100 Subject: [PATCH 1/2] feat: add creator key trading operations --- prisma/schema/creator.prisma | 21 ++++++ .../migration.sql | 26 +++++++ src/config.schema.ts | 1 + src/middlewares/admin-guard.middleware.ts | 38 ++++++----- src/modules/admin/admin.controllers.ts | 34 ++++++++++ src/modules/admin/admin.routes.ts | 3 + src/modules/creator/buy.controller.ts | 10 +++ src/modules/index.ts | 4 ++ .../indexer/indexer-pipeline.service.ts | 3 + .../persist-circulating-supply.service.ts | 36 ++++++++++ src/modules/keys/key-price-history.routes.ts | 33 +++++++++ src/modules/keys/key-price-history.service.ts | 30 +++++++++ src/modules/keys/key-trading.service.ts | 17 +++++ src/modules/trading/multi-buy.controllers.ts | 7 ++ .../webhooks/horizon-webhook.routes.ts | 67 +++++++++++++++++++ 15 files changed, 314 insertions(+), 16 deletions(-) create mode 100644 prisma/schema/migrations/20260826000000_add_creator_key_operations/migration.sql create mode 100644 src/modules/indexer/persist-circulating-supply.service.ts create mode 100644 src/modules/keys/key-price-history.routes.ts create mode 100644 src/modules/keys/key-price-history.service.ts create mode 100644 src/modules/keys/key-trading.service.ts create mode 100644 src/modules/webhooks/horizon-webhook.routes.ts diff --git a/prisma/schema/creator.prisma b/prisma/schema/creator.prisma index 2142251..df0ce5f 100644 --- a/prisma/schema/creator.prisma +++ b/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 diff --git a/prisma/schema/migrations/20260826000000_add_creator_key_operations/migration.sql b/prisma/schema/migrations/20260826000000_add_creator_key_operations/migration.sql new file mode 100644 index 0000000..fb3773f --- /dev/null +++ b/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; diff --git a/src/config.schema.ts b/src/config.schema.ts index 2b4781b..cdf9ec5 100644 --- a/src/config.schema.ts +++ b/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, // Ownership snapshot cleanup job OWNERSHIP_SNAPSHOT_TABLE_NAME: z diff --git a/src/middlewares/admin-guard.middleware.ts b/src/middlewares/admin-guard.middleware.ts index f6ea8ab..bcb43f2 100644 --- a/src/middlewares/admin-guard.middleware.ts +++ b/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'); + } } diff --git a/src/modules/admin/admin.controllers.ts b/src/modules/admin/admin.controllers.ts index 55a8b86..13f9d4b 100644 --- a/src/modules/admin/admin.controllers.ts +++ b/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 => { + 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); + } +}; diff --git a/src/modules/admin/admin.routes.ts b/src/modules/admin/admin.routes.ts index 0f2b01d..65b097d 100644 --- a/src/modules/admin/admin.routes.ts +++ b/src/modules/admin/admin.routes.ts @@ -2,6 +2,7 @@ import { Router } from 'express'; import { httpUpdateCreatorMetadata, httpReplayIndexerEvents, + httpSetKeyTradingPaused, } from './admin.controllers'; import { adminGuard } from '../../middlewares/admin-guard.middleware'; @@ -9,5 +10,7 @@ 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; diff --git a/src/modules/creator/buy.controller.ts b/src/modules/creator/buy.controller.ts index 1709101..fa53715 100644 --- a/src/modules/creator/buy.controller.ts +++ b/src/modules/creator/buy.controller.ts @@ -8,6 +8,7 @@ import { zodIssuesToDetails, } from '../../utils/api-response.utils'; import { buyGateway } from './buy.service'; +import { assertTradingActive, TradingPausedError } from '../keys/key-trading.service'; const buySchema = z.object({ quantity: z.number().int().positive(), @@ -32,6 +33,15 @@ export async function httpBuyCreatorKey( } 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 = parsed.data.key_cost_xlm * parsed.data.quantity + parsed.data.fee_xlm; const balance = await buyGateway.getXlmBalance(walletAddress); diff --git a/src/modules/index.ts b/src/modules/index.ts index d7b6b04..4eca658 100644 --- a/src/modules/index.ts +++ b/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; diff --git a/src/modules/indexer/indexer-pipeline.service.ts b/src/modules/indexer/indexer-pipeline.service.ts index 06f3027..6986b42 100644 --- a/src/modules/indexer/indexer-pipeline.service.ts +++ b/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'; /** * Processes a batch of on-chain trade events (KEY_BOUGHT or KEY_SOLD). @@ -70,6 +71,8 @@ export async function processTradeEvents(events: IndexerChainEvent[]): Promise { + return new Promise(resolve => setTimeout(resolve, milliseconds)); +} + +export async function persistCirculatingSupply(creatorId: string): Promise { + 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; +} \ No newline at end of file diff --git a/src/modules/keys/key-price-history.routes.ts b/src/modules/keys/key-price-history.routes.ts new file mode 100644 index 0000000..12522cb --- /dev/null +++ b/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; \ No newline at end of file diff --git a/src/modules/keys/key-price-history.service.ts b/src/modules/keys/key-price-history.service.ts new file mode 100644 index 0000000..67dab90 --- /dev/null +++ b/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 = { + '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(); + 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 })); +} \ No newline at end of file diff --git a/src/modules/keys/key-trading.service.ts b/src/modules/keys/key-trading.service.ts new file mode 100644 index 0000000..5744d49 --- /dev/null +++ b/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 { + const creator = await prisma.creatorProfile.findUnique({ + where: { id: creatorId }, + select: { tradingPaused: true }, + }); + if (creator?.tradingPaused) { + throw new TradingPausedError(); + } +} \ No newline at end of file diff --git a/src/modules/trading/multi-buy.controllers.ts b/src/modules/trading/multi-buy.controllers.ts index 7b11f09..3bbccbb 100644 --- a/src/modules/trading/multi-buy.controllers.ts +++ b/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 { 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 = { legs_empty: 400, diff --git a/src/modules/webhooks/horizon-webhook.routes.ts b/src/modules/webhooks/horizon-webhook.routes.ts new file mode 100644 index 0000000..c93c913 --- /dev/null +++ b/src/modules/webhooks/horizon-webhook.routes.ts @@ -0,0 +1,67 @@ +import { Router } from 'express'; +import { createHmac, timingSafeEqual } from 'crypto'; +import { envConfig } from '../../config'; +import { sendError, sendSuccess } from '../../utils/api-response.utils'; +import { ErrorCode } from '../../constants/error.constants'; +import { prisma } from '../../utils/prisma.utils'; +import { updateOwnership } from '../ownership/ownership.service'; + +const router = Router(); + +function hasValidSignature(payload: unknown, signature: string | undefined): boolean { + if (!envConfig.HORIZON_WEBHOOK_SECRET || !signature) return false; + const expected = createHmac('sha256', envConfig.HORIZON_WEBHOOK_SECRET) + .update(JSON.stringify(payload)) + .digest('hex'); + if (signature.length !== expected.length) return false; + return timingSafeEqual(Buffer.from(signature), Buffer.from(expected)); +} + +router.post('/horizon', async (req, res, next) => { + const signature = req.header('x-horizon-signature'); + if (!hasValidSignature(req.body, signature)) { + sendError(res, 401, ErrorCode.UNAUTHORIZED, 'Invalid Horizon webhook signature'); + return; + } + const event = req.body as { + type?: string; + memo?: string; + transaction_hash?: string; + }; + if (event.type !== 'payment_received' && event.type !== 'transaction_successful') { + sendSuccess(res, { ignored: true }); + return; + } + if (!event.memo) { + sendError(res, 400, ErrorCode.BAD_REQUEST, 'Horizon event memo is required'); + return; + } + try { + const order = await prisma.pendingKeyPurchase.findUnique({ where: { memo: event.memo } }); + if (!order || order.status !== 'PENDING') { + sendSuccess(res, { ignored: true }); + return; + } + await prisma.$transaction(async transaction => { + await transaction.pendingKeyPurchase.update({ + where: { id: order.id }, + data: { status: 'SETTLED', transactionHash: event.transaction_hash, settledAt: new Date() }, + }); + await transaction.auditEvent.create({ + data: { + actor: 'stellar-horizon', + action: 'ownership_transferred', + target: 'CreatorKey', + targetId: order.creatorId, + metadata: { purchaseId: order.id, transactionHash: event.transaction_hash }, + }, + }); + }); + await updateOwnership(order.buyerAddress, order.creatorId, Number(order.quantity)); + sendSuccess(res, { settled: true }); + } catch (error) { + next(error); + } +}); + +export default router; \ No newline at end of file From 2f93268df7f8a36a216f2334dd33f63dd39b092e Mon Sep 17 00:00:00 2001 From: Akanimoh12 Date: Wed, 26 Aug 2026 02:34:43 +0100 Subject: [PATCH 2/2] fix: remove stale buy cost declaration --- src/modules/creator/buy.controller.ts | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/modules/creator/buy.controller.ts b/src/modules/creator/buy.controller.ts index 5276f7c..87543eb 100644 --- a/src/modules/creator/buy.controller.ts +++ b/src/modules/creator/buy.controller.ts @@ -32,8 +32,6 @@ export async function httpBuyCreatorKey( } throw error; } - const required = - parsed.data.key_cost_xlm * parsed.data.quantity + parsed.data.fee_xlm; const required = body.key_cost_xlm * body.quantity + body.fee_xlm; const balance = await buyGateway.getXlmBalance(walletAddress); if (balance < required) {