diff --git a/CHANGELOG.md b/CHANGELOG.md index cc1cb67b..ed16de2e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ### Added +- Backend: asynchronous campaign export worker (`POST /campaigns/:id/exports`) on the durable job + queue (#1260), Horizon deposit watcher for operator reward-token deposits (#1261), IP-cluster + fraud detection with an admin review API (#1258), and a generated OpenAPI reference with Swagger + UI at `/docs/api` (#1259). See `docs/BACKEND_ANALYTICS_AND_GROWTH.md`. - Production deployment guide with environment matrix - Error codes documentation with frontend-friendly message mapping - Release process documentation with semantic versioning policy diff --git a/backend/.env.example b/backend/.env.example index 21703469..d8451908 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -151,3 +151,16 @@ VAPID_SUBJECT=mailto:notifications@trivela.app # retweet and follow tasks before points are awarded. When unset, # POST /api/v1/social-tasks/twitter/verify responds 503 TWITTER_NOT_CONFIGURED. # TWITTER_BEARER_TOKEN= + +# ── Fraud detection (#1258) ─────────────────────────────────────────────── +# Distinct accounts registering from one IP subnet (/24 IPv4, /64 IPv6) within the +# window before a fraud flag is raised for review. +# FRAUD_SUBNET_THRESHOLD=3 +# FRAUD_WINDOW_HOURS=24 +# Known proxy / VPN exit nodes (comma-separated IPs or IPv4 CIDR ranges). +# FRAUD_PROXY_EXIT_LIST= + +# ── Operator deposit watcher (#1261) ────────────────────────────────────── +# Contract-custody accounts to watch on Horizon for incoming reward-token deposits. +# The reward asset comes from REWARD_TOKEN_CODE / REWARD_TOKEN_ISSUER (default: XLM). +# OPERATOR_DEPOSIT_ACCOUNTS= diff --git a/backend/src/dal/sqliteExportJobRepository.js b/backend/src/dal/sqliteExportJobRepository.js new file mode 100644 index 00000000..c1fb2503 --- /dev/null +++ b/backend/src/dal/sqliteExportJobRepository.js @@ -0,0 +1,119 @@ +// @ts-check +import { randomUUID } from 'node:crypto'; + +/** + * @typedef {{ + * id: string, + * campaignId: string, + * format: 'csv' | 'json', + * fromDate: string | null, + * toDate: string | null, + * status: 'queued' | 'running' | 'completed' | 'failed', + * rowCount: number | null, + * fileUrl: string | null, + * fileKey: string | null, + * error: string | null, + * attempts: number, + * createdAt: string, + * startedAt: string | null, + * completedAt: string | null, + * }} ExportJob + */ + +/** @returns {ExportJob | null} */ +function rowToJob(row) { + if (!row) return null; + return { + id: row.id, + campaignId: row.campaign_id, + format: row.format, + fromDate: row.from_date ?? null, + toDate: row.to_date ?? null, + status: row.status, + rowCount: row.row_count ?? null, + fileUrl: row.file_url ?? null, + fileKey: row.file_key ?? null, + error: row.error ?? null, + attempts: row.attempts, + createdAt: row.created_at, + startedAt: row.started_at ?? null, + completedAt: row.completed_at ?? null, + }; +} + +/** + * Persistence for asynchronous campaign export jobs (#1260). + * + * @param {{ db: InstanceType }} params + */ +export function createSqliteExportJobRepository({ db }) { + const insertStmt = db.prepare(` + INSERT INTO export_jobs (id, campaign_id, format, from_date, to_date, status, created_at) + VALUES (@id, @campaignId, @format, @fromDate, @toDate, 'queued', @createdAt) + `); + const byIdStmt = db.prepare('SELECT * FROM export_jobs WHERE id = ?'); + const listStmt = db.prepare( + 'SELECT * FROM export_jobs WHERE campaign_id = ? ORDER BY created_at DESC, rowid DESC LIMIT ?', + ); + const runningStmt = db.prepare(` + UPDATE export_jobs + SET status = 'running', started_at = ?, attempts = attempts + 1, error = NULL + WHERE id = ? + `); + const completedStmt = db.prepare(` + UPDATE export_jobs + SET status = 'completed', row_count = ?, file_url = ?, file_key = ?, + completed_at = ?, error = NULL + WHERE id = ? + `); + const failedStmt = db.prepare("UPDATE export_jobs SET status = 'failed', error = ? WHERE id = ?"); + + return { + /** + * @param {{ campaignId: string, format: 'csv' | 'json', fromDate?: string | null, toDate?: string | null }} input + * @returns {ExportJob} + */ + create({ campaignId, format, fromDate = null, toDate = null }) { + const id = randomUUID(); + insertStmt.run({ + id, + campaignId: String(campaignId), + format, + fromDate, + toDate, + createdAt: new Date().toISOString(), + }); + return /** @type {ExportJob} */ (rowToJob(byIdStmt.get(id))); + }, + + /** @param {string} id */ + getById(id) { + return rowToJob(byIdStmt.get(id)); + }, + + /** @param {string | number} campaignId @param {number} [limit] */ + listByCampaign(campaignId, limit = 20) { + return listStmt + .all(String(campaignId), Math.min(Math.max(limit, 1), 100)) + .map((row) => /** @type {ExportJob} */ (rowToJob(row))); + }, + + /** @param {string} id */ + markRunning(id) { + runningStmt.run(new Date().toISOString(), id); + }, + + /** + * @param {string} id + * @param {{ rowCount: number, fileUrl: string, fileKey?: string | null }} result + */ + markCompleted(id, { rowCount, fileUrl, fileKey = null }) { + completedStmt.run(rowCount, fileUrl, fileKey, new Date().toISOString(), id); + }, + + /** @param {string} id @param {string} error */ + markFailed(id, error) { + failedStmt.run(error.slice(0, 500), id); + }, + }; +} diff --git a/backend/src/db/migrations/041_export_jobs.js b/backend/src/db/migrations/041_export_jobs.js new file mode 100644 index 00000000..df0389e3 --- /dev/null +++ b/backend/src/db/migrations/041_export_jobs.js @@ -0,0 +1,32 @@ +export const version = 41; +export const description = 'Add export_jobs table for asynchronous campaign export processing'; + +/** + * Tracks campaign export requests that are processed off the request path by + * the durable job queue (#1260). Status moves queued -> running -> completed, + * or failed when the worker throws (the queue retries; a later attempt moves + * the row back to running). + */ +export function up(db) { + db.exec(` + CREATE TABLE IF NOT EXISTS export_jobs ( + id TEXT PRIMARY KEY, + campaign_id TEXT NOT NULL, + format TEXT NOT NULL CHECK (format IN ('csv', 'json')), + from_date TEXT, + to_date TEXT, + status TEXT NOT NULL DEFAULT 'queued' + CHECK (status IN ('queued', 'running', 'completed', 'failed')), + row_count INTEGER, + file_url TEXT, + file_key TEXT, + error TEXT, + attempts INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + started_at TEXT, + completed_at TEXT + ); + CREATE INDEX IF NOT EXISTS idx_export_jobs_campaign + ON export_jobs(campaign_id, created_at DESC); + `); +} diff --git a/backend/src/db/migrations/042_operator_deposits.js b/backend/src/db/migrations/042_operator_deposits.js new file mode 100644 index 00000000..31a5ed54 --- /dev/null +++ b/backend/src/db/migrations/042_operator_deposits.js @@ -0,0 +1,34 @@ +export const version = 42; +export const description = 'Add operator_deposits and horizon_watcher_cursors for deposit watching'; + +/** + * Deposits observed by the Horizon account watcher (#1261). `paging_token` is + * unique so replayed stream events (after a reconnect) are idempotent. + */ +export function up(db) { + db.exec(` + CREATE TABLE IF NOT EXISTS operator_deposits ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + paging_token TEXT NOT NULL UNIQUE, + account TEXT NOT NULL, + from_address TEXT, + asset_code TEXT NOT NULL, + asset_issuer TEXT, + amount TEXT NOT NULL, + tx_hash TEXT NOT NULL, + verified INTEGER NOT NULL DEFAULT 1, + observed_at TEXT NOT NULL, + recorded_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_operator_deposits_account + ON operator_deposits(account, id DESC); + CREATE INDEX IF NOT EXISTS idx_operator_deposits_tx + ON operator_deposits(tx_hash); + + CREATE TABLE IF NOT EXISTS horizon_watcher_cursors ( + account TEXT PRIMARY KEY, + cursor TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + `); +} diff --git a/backend/src/db/migrations/043_fraud_detection.js b/backend/src/db/migrations/043_fraud_detection.js new file mode 100644 index 00000000..30b9563a --- /dev/null +++ b/backend/src/db/migrations/043_fraud_detection.js @@ -0,0 +1,40 @@ +export const version = 43; +export const description = 'Add signup_events and fraud_flags for IP-cluster fraud detection'; + +/** + * Only the /24 (IPv4) or /64 (IPv6) network prefix of a signup's IP is stored + * (#1258), never the full address — except for known public proxy exit nodes, + * whose address is a published list entry rather than personal data. + */ +export function up(db) { + db.exec(` + CREATE TABLE IF NOT EXISTS signup_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + campaign_id TEXT NOT NULL, + account TEXT NOT NULL, + ip_subnet TEXT NOT NULL, + proxy_exit TEXT, + created_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_signup_events_subnet + ON signup_events(campaign_id, ip_subnet, created_at); + CREATE INDEX IF NOT EXISTS idx_signup_events_proxy + ON signup_events(campaign_id, proxy_exit, created_at); + + CREATE TABLE IF NOT EXISTS fraud_flags ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + campaign_id TEXT NOT NULL, + rule TEXT NOT NULL CHECK (rule IN ('ip_subnet_cluster', 'proxy_exit_node')), + subject TEXT NOT NULL, + account_count INTEGER NOT NULL, + accounts TEXT NOT NULL DEFAULT '[]', + status TEXT NOT NULL DEFAULT 'open' + CHECK (status IN ('open', 'dismissed', 'confirmed')), + first_seen_at TEXT NOT NULL, + last_seen_at TEXT NOT NULL, + UNIQUE (campaign_id, rule, subject) + ); + CREATE INDEX IF NOT EXISTS idx_fraud_flags_status + ON fraud_flags(status, last_seen_at DESC); + `); +} diff --git a/backend/src/index.js b/backend/src/index.js index 1b0c13c7..d04cc5e0 100644 --- a/backend/src/index.js +++ b/backend/src/index.js @@ -112,6 +112,17 @@ import { createNotificationPreferencesRoutes, } from './routes/notifications.js'; import { createOperatorBalanceJob } from './jobs/operatorBalanceJob.js'; +import { createSqliteExportJobRepository } from './dal/sqliteExportJobRepository.js'; +import { + CAMPAIGN_EXPORT_JOB_TYPE, + createCampaignExportWorker, +} from './jobs/campaignExportWorker.js'; +import { createCampaignExportJobRoutes } from './routes/campaignExportJobs.js'; +import { createFraudDetector } from './services/fraudDetection.js'; +import { createFraudFlagRoutes } from './routes/fraudFlags.js'; +import { createHorizonDepositWatcher } from './services/horizonDepositWatcher.js'; +import { createOperatorDepositRoutes } from './routes/operatorDeposits.js'; +import { createApiDocsRouter } from './routes/apiDocs.js'; import { createPruningJob } from './jobs/pruningJob.js'; import { purgePiiForUser, @@ -845,10 +856,21 @@ export async function createApp(options = {}) { setInterval(doExport, 24 * 60 * 60 * 1_000).unref?.(); } + // #1260 — asynchronous campaign exports, processed off the request path + const campaignExportJobRepository = createSqliteExportJobRepository({ db: dal.db }); + const campaignExportWorker = createCampaignExportWorker({ + db: dal.db, + repository: campaignExportJobRepository, + campaignRepository, + storage: storageAdapter, + logger: log, + }); + // Durable job queue — starts poll loop and recovers stale jobs from prior crashes (#565) const durableJobQueue = createDurableJobQueue({ store: jobQueueStore, handlers: { + [CAMPAIGN_EXPORT_JOB_TYPE]: campaignExportWorker.handle, // #922 — end-of-campaign claimable balance creation, enqueued from // POST /campaigns/:id/claimable-balances instead of running inline. [CLAIMABLE_BALANCES_JOB_TYPE]: createClaimableBalancesJobHandler({ @@ -897,6 +919,49 @@ export async function createApp(options = {}) { operatorBalanceJob.start(); } + // #1258 — IP-cluster fraud detection for multi-account signups + const parseList = (/** @type {unknown} */ value) => + String(value ?? '') + .split(',') + .map((v) => v.trim()) + .filter(Boolean); + const fraudDetector = createFraudDetector({ + db: dal.db, + subnetThreshold: normalizePositiveInteger( + /** @type {any} */ (options.fraudSubnetThreshold) ?? process.env.FRAUD_SUBNET_THRESHOLD, + 3, + ), + windowMs: + normalizePositiveInteger( + /** @type {any} */ (options.fraudWindowHours) ?? process.env.FRAUD_WINDOW_HOURS, + 24, + ) * + 60 * + 60 * + 1000, + proxyList: parseList(options.fraudProxyExitList ?? process.env.FRAUD_PROXY_EXIT_LIST), + logger: log, + }); + + // #1261 — Horizon watcher for operator reward-token deposits + const depositWatchAccounts = parseList( + options.operatorDepositAccounts ?? process.env.OPERATOR_DEPOSIT_ACCOUNTS, + ); + const depositWatcher = createHorizonDepositWatcher({ + db: dal.db, + horizonUrl: stellarConfig.horizonUrl, + accounts: depositWatchAccounts, + asset: { + code: process.env.REWARD_TOKEN_CODE || 'XLM', + issuer: process.env.REWARD_TOKEN_ISSUER || null, + }, + allowHttp: !String(stellarConfig.horizonUrl ?? '').startsWith('https'), + logger: log, + }); + if (!options.disableJobs && depositWatchAccounts.length > 0 && stellarConfig.horizonUrl) { + depositWatcher.start(); + } + async function buildHealthPayload() { const rpcUrl = rpcPool.getHealthyRpcUrl(); const rpc = rpcHealthCache.payload ?? (await checkSorobanRpcHealth({ rpcUrl, fetchImpl })); @@ -1034,6 +1099,8 @@ export async function createApp(options = {}) { paths: {}, }; } + // #1259 — generated OpenAPI 3.0 document + Swagger UI at /docs/api + app.use('/docs/api', createApiDocsRouter({ baseSpec: swaggerSpec })); app.use('/docs', swaggerUi.serve); app.get( '/docs', @@ -2889,6 +2956,18 @@ export async function createApp(options = {}) { }); } + // #1258 — flag clusters of signups from one IP subnet / proxy exit node. + // Detection is advisory: a failure here must never fail the signup. + try { + fraudDetector.recordSignup({ + campaignId: String(campaign.id), + account: refereeAddress.trim(), + ip: req.ip, + }); + } catch (err) { + log.warn({ err }, 'fraud detection failed'); + } + // Live-update anyone watching this campaign's referral leaderboard stream. broadcastCampaignEvent(`${req.params.id}:leaderboard`, 'referral', { campaignId: String(campaign.id), @@ -3126,6 +3205,22 @@ export async function createApp(options = {}) { }); app.use(prefix, rateLimiter, auditRouter); + // #1258 — fraud flag review (master key) + app.use( + `${prefix}/admin/fraud`, + rateLimiter, + requireMasterKey, + createFraudFlagRoutes({ fraudDetector, recordAudit: recordAuditEntry }), + ); + + // #1261 — operator deposit verification (master key) + app.use( + `${prefix}/operator`, + rateLimiter, + requireMasterKey, + createOperatorDepositRoutes({ watcher: depositWatcher }), + ); + // Variant routes for A/B testing (Issue #624) const variantRouter = createVariantRoutes({ variantRepo: variantRepository, @@ -3184,6 +3279,20 @@ export async function createApp(options = {}) { }); app.use(`${prefix}/sponsored-accounts`, rateLimiter, ...guard, sponsoredAccountRouter); + // #1260 — asynchronous campaign exports (queued, polled by job id) + app.use( + prefix, + rateLimiter, + createCampaignExportJobRoutes({ + repository: campaignExportJobRepository, + campaignRepository, + jobQueue: durableJobQueue, + guard, + recordAudit: recordAuditEntry, + logger: log, + }), + ); + // #548 — Claimable balances for unclaimed/expired rewards // #922 — submission runs via durableJobQueue; idempotencyMiddleware // guards the POST route against duplicate enqueues on request retry. @@ -3298,6 +3407,7 @@ export async function createApp(options = {}) { app._close = () => { isShuttingDown = true; + depositWatcher.stop(); try { dal.db.close(); } catch (_) { diff --git a/backend/src/jobs/campaignExportWorker.js b/backend/src/jobs/campaignExportWorker.js new file mode 100644 index 00000000..e57d5b5b --- /dev/null +++ b/backend/src/jobs/campaignExportWorker.js @@ -0,0 +1,89 @@ +// @ts-check +import { + PARTICIPANT_COLUMNS, + buildCsv, + collectParticipants, +} from '../services/campaignExportData.js'; + +/** Durable-queue job type handled by {@link createCampaignExportWorker}. */ +export const CAMPAIGN_EXPORT_JOB_TYPE = 'campaign_export'; + +const MIME_TYPES = { csv: 'text/csv; charset=utf-8', json: 'application/json; charset=utf-8' }; + +/** + * Worker for asynchronous campaign exports (#1260). + * + * Heavy CSV/JSON generation runs off the HTTP request path via the durable job + * queue, so large campaigns no longer hit request timeouts. The result is + * uploaded through the configured storage adapter and the `export_jobs` row is + * updated so clients can poll for a download URL. + * + * Redelivery-safe: a job that already completed is a no-op, and a failed + * attempt records the error and rethrows so the queue applies its backoff. + * + * @param {{ + * db: InstanceType, + * repository: ReturnType, + * campaignRepository: { getById: (id: string) => { id: string | number, name: string } | null | undefined }, + * storage: import('../storage/storageAdapter.js').StorageAdapter, + * logger?: { info?: Function, warn?: Function, error?: Function }, + * }} options + */ +export function createCampaignExportWorker({ + db, + repository, + campaignRepository, + storage, + logger = console, +}) { + /** @param {{ exportJobId: string }} payload */ + async function handle({ exportJobId }) { + const job = repository.getById(exportJobId); + if (!job) { + logger.warn?.({ exportJobId }, 'campaignExport:drop reason=unknown_job'); + return; + } + if (job.status === 'completed') return; + + repository.markRunning(job.id); + try { + const campaign = campaignRepository.getById(job.campaignId); + if (!campaign) throw new Error(`Campaign ${job.campaignId} no longer exists`); + + const participants = collectParticipants(db, job.campaignId, { + fromDate: job.fromDate, + toDate: job.toDate, + }); + + const content = + job.format === 'csv' + ? buildCsv(PARTICIPANT_COLUMNS, participants) + : JSON.stringify( + { campaign: { id: campaign.id, name: campaign.name }, participants }, + null, + 2, + ); + + const filename = `exports/campaigns/${job.campaignId}/${job.id}.${job.format}`; + const { url, key } = await storage.upload({ + buffer: Buffer.from(content, 'utf8'), + filename, + mimeType: MIME_TYPES[job.format], + }); + + repository.markCompleted(job.id, { + rowCount: participants.length, + fileUrl: url, + fileKey: key ?? filename, + }); + logger.info?.({ exportJobId: job.id, rows: participants.length }, 'campaignExport:completed'); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + repository.markFailed(job.id, message); + logger.warn?.({ exportJobId: job.id, err: message }, 'campaignExport:failed'); + throw err; + } + } + + return { handle }; +} diff --git a/backend/src/jobs/campaignExportWorker.test.js b/backend/src/jobs/campaignExportWorker.test.js new file mode 100644 index 00000000..5c6b9c18 --- /dev/null +++ b/backend/src/jobs/campaignExportWorker.test.js @@ -0,0 +1,146 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import Database from 'better-sqlite3'; +import { runMigrations } from '../db/migrate.js'; +import { createSqliteExportJobRepository } from '../dal/sqliteExportJobRepository.js'; +import { CAMPAIGN_EXPORT_JOB_TYPE, createCampaignExportWorker } from './campaignExportWorker.js'; + +const silent = { info() {}, warn() {}, error() {} }; +const campaignRepository = { + getById: (id) => (id === 'c1' ? { id: 'c1', name: 'Spring Drop' } : null), +}; + +async function setup(storageOverrides = {}) { + const db = new Database(':memory:'); + await runMigrations(db); + db.exec(` + INSERT INTO referrals (campaign_id, referrer_address, referee_address, created_at) + VALUES ('c1', 'GREFERRER', 'GALICE', '2026-09-01T00:00:00Z'), + ('c1', 'GREFERRER', 'GBOB', '2026-09-10T00:00:00Z'); + INSERT INTO credit_events (user, amount) VALUES ('GALICE', '10'), ('GBOB', '4'); + INSERT INTO claim_events (user, amount) VALUES ('GALICE', '3'); + `); + const uploads = []; + const storage = { + backendName: 'test', + async upload(params) { + uploads.push(params); + return { url: `https://files.test/${params.filename}`, key: params.filename }; + }, + ...storageOverrides, + }; + const repository = createSqliteExportJobRepository({ db }); + const worker = createCampaignExportWorker({ + db, + repository, + campaignRepository, + storage, + logger: silent, + }); + return { db, repository, worker, uploads }; +} + +test('exposes a stable job type', () => { + assert.equal(CAMPAIGN_EXPORT_JOB_TYPE, 'campaign_export'); +}); + +test('builds a CSV, uploads it, and marks the job completed with a download URL', async () => { + const { repository, worker, uploads } = await setup(); + const job = repository.create({ campaignId: 'c1', format: 'csv' }); + assert.equal(job.status, 'queued'); + + await worker.handle({ exportJobId: job.id }); + + const done = repository.getById(job.id); + assert.equal(done.status, 'completed'); + assert.equal(done.rowCount, 2); + assert.equal(done.fileUrl, `https://files.test/exports/campaigns/c1/${job.id}.csv`); + assert.ok(done.completedAt); + + assert.equal(uploads.length, 1); + assert.equal(uploads[0].mimeType, 'text/csv; charset=utf-8'); + const csv = uploads[0].buffer.toString('utf8'); + assert.match( + csv, + /^participantAddress,registeredAt,pointsCredited,pointsClaimed,netPoints,referredBy\n/, + ); + assert.match(csv, /GALICE/); + assert.match(csv, /GBOB/); +}); + +test('supports JSON output and reports per-participant point totals', async () => { + const { repository, worker, uploads } = await setup(); + const job = repository.create({ + campaignId: 'c1', + format: 'json', + fromDate: '2026-09-05T00:00:00Z', + }); + assert.equal(repository.getById(job.id).fromDate, '2026-09-05T00:00:00Z'); + + await worker.handle({ exportJobId: job.id }); + + const body = JSON.parse(uploads[0].buffer.toString('utf8')); + assert.deepEqual(body.campaign, { id: 'c1', name: 'Spring Drop' }); + const alice = body.participants.find((p) => p.participantAddress === 'GALICE'); + assert.equal(alice.pointsCredited, 10); + assert.equal(alice.pointsClaimed, 3); + assert.equal(alice.netPoints, 7); + assert.equal(uploads[0].mimeType, 'application/json; charset=utf-8'); + assert.equal(repository.getById(job.id).rowCount, 2); +}); + +test('a redelivered, already-completed job is a no-op', async () => { + const { repository, worker, uploads } = await setup(); + const job = repository.create({ campaignId: 'c1', format: 'csv' }); + await worker.handle({ exportJobId: job.id }); + await worker.handle({ exportJobId: job.id }); + assert.equal(uploads.length, 1); +}); + +test('an unknown job id is dropped without throwing', async () => { + const { worker } = await setup(); + await worker.handle({ exportJobId: 'nope' }); +}); + +test('upload failure marks the job failed and rethrows so the queue can retry', async () => { + const { repository, worker } = await setup({ + async upload() { + throw new Error('storage down'); + }, + }); + const job = repository.create({ campaignId: 'c1', format: 'csv' }); + + await assert.rejects(() => worker.handle({ exportJobId: job.id }), /storage down/); + + const failed = repository.getById(job.id); + assert.equal(failed.status, 'failed'); + assert.equal(failed.error, 'storage down'); + assert.equal(failed.attempts, 1); +}); + +test('a retry after failure moves the job back to running and can complete', async () => { + let fail = true; + const { repository, worker } = await setup({ + async upload(params) { + if (fail) throw new Error('flaky'); + return { url: `https://files.test/${params.filename}`, key: params.filename }; + }, + }); + const job = repository.create({ campaignId: 'c1', format: 'csv' }); + await assert.rejects(() => worker.handle({ exportJobId: job.id })); + + fail = false; + await worker.handle({ exportJobId: job.id }); + + const done = repository.getById(job.id); + assert.equal(done.status, 'completed'); + assert.equal(done.error, null); + assert.equal(done.attempts, 2); +}); + +test('a campaign deleted after queueing fails the job', async () => { + const { repository, worker } = await setup(); + const job = repository.create({ campaignId: 'gone', format: 'csv' }); + await assert.rejects(() => worker.handle({ exportJobId: job.id }), /no longer exists/); + assert.equal(repository.getById(job.id).status, 'failed'); +}); diff --git a/backend/src/routes/apiDocs.js b/backend/src/routes/apiDocs.js new file mode 100644 index 00000000..9cd61941 --- /dev/null +++ b/backend/src/routes/apiDocs.js @@ -0,0 +1,37 @@ +// @ts-check +import { Router } from 'express'; +import swaggerUi from 'swagger-ui-express'; +import { buildOpenApiDocument, loadOpenApiSpec } from '../services/openapiGenerator.js'; + +/** + * Interactive REST API documentation (#1259), mounted at `/docs/api`. + * + * GET /docs/api Swagger UI + * GET /docs/api/openapi.json the generated OpenAPI 3.0 document + * + * The document is generated once at startup from `openapi.yaml` plus the Zod + * request schemas, and cached. + * + * @param {{ baseSpec?: Record }} [options] defaults to `backend/openapi.yaml` + */ +export function createApiDocsRouter({ baseSpec } = {}) { + const router = Router(); + const hasPaths = baseSpec?.paths && Object.keys(baseSpec.paths).length > 0; + const document = buildOpenApiDocument({ baseSpec: hasPaths ? baseSpec : loadOpenApiSpec() }); + + router.get('/openapi.json', (_req, res) => { + res.setHeader('Cache-Control', 'public, max-age=300'); + res.json(document); + }); + + router.use('/', swaggerUi.serve); + router.get( + '/', + swaggerUi.setup(document, { + customSiteTitle: 'Trivela REST API', + swaggerOptions: { persistAuthorization: true, docExpansion: 'list', filter: true }, + }), + ); + + return router; +} diff --git a/backend/src/routes/backendGrowthFeatures.test.js b/backend/src/routes/backendGrowthFeatures.test.js new file mode 100644 index 00000000..cc0e0a75 --- /dev/null +++ b/backend/src/routes/backendGrowthFeatures.test.js @@ -0,0 +1,160 @@ +// @ts-check +/** + * Integration tests for the fraud review API (#1258), operator deposit API + * (#1261) and the generated API docs (#1259). + * Run with: node --test src/routes/backendGrowthFeatures.test.js + */ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import request from 'supertest'; +import { createApp } from '../index.js'; + +const API_KEY = 'test-key-123'; +const MASTER = 'test-master-key'; + +function makeApp(options = {}) { + return createApp({ + dbPath: ':memory:', + campaigns: [], + disableJobs: true, + skipEnvValidation: true, + rateLimit: { windowMs: 60_000, maxRequests: 10_000 }, + apiKeys: API_KEY, + masterKey: MASTER, + fraudSubnetThreshold: 3, + ...options, + }); +} + +async function createCampaign(app) { + const res = await request(app) + .post('/api/v1/campaigns') + .set('X-API-Key', API_KEY) + .send({ name: 'Growth', rewardPerAction: 1 }) + .expect(201); + return res.body; +} + +// ── #1258 fraud detection ──────────────────────────────────────────────────── + +test('referrals from one IP subnet raise a fraud flag that admins can review', async () => { + const app = await makeApp(); + const campaign = await createCampaign(app); + + // Every supertest request comes from the same loopback address, i.e. one subnet. + for (const referee of ['GREF1', 'GREF2', 'GREF3']) { + await request(app) + .post(`/api/v1/campaigns/${campaign.id}/referrals`) + .send({ referrerAddress: 'GREFERRER', refereeAddress: referee }) + .expect(201); + } + + const flags = await request(app) + .get('/api/v1/admin/fraud/flags?status=open') + .set('X-API-Key', MASTER) + .expect(200); + assert.equal(flags.body.total, 1); + const [flag] = flags.body.data; + assert.equal(flag.rule, 'ip_subnet_cluster'); + assert.equal(flag.accountCount, 3); + assert.deepEqual(flag.accounts.sort(), ['GREF1', 'GREF2', 'GREF3']); + + const dismissed = await request(app) + .patch(`/api/v1/admin/fraud/flags/${flag.id}`) + .set('X-API-Key', MASTER) + .send({ status: 'dismissed' }) + .expect(200); + assert.equal(dismissed.body.status, 'dismissed'); + + const open = await request(app) + .get('/api/v1/admin/fraud/flags?status=open') + .set('X-API-Key', MASTER); + assert.equal(open.body.total, 0); +}); + +test('flagging never blocks a signup and stays quiet below the threshold', async () => { + const app = await makeApp(); + const campaign = await createCampaign(app); + + for (const referee of ['GA', 'GB']) { + await request(app) + .post(`/api/v1/campaigns/${campaign.id}/referrals`) + .send({ referrerAddress: 'GREFERRER', refereeAddress: referee }) + .expect(201); + } + const flags = await request(app) + .get('/api/v1/admin/fraud/flags') + .set('X-API-Key', MASTER) + .expect(200); + assert.equal(flags.body.total, 0); +}); + +test('fraud admin API validates input, 404s unknown flags, and requires the master key', async () => { + const app = await makeApp(); + + await request(app).get('/api/v1/admin/fraud/flags').set('X-API-Key', API_KEY).expect(401); + await request(app) + .get('/api/v1/admin/fraud/flags?status=bogus') + .set('X-API-Key', MASTER) + .expect(400); + await request(app) + .patch('/api/v1/admin/fraud/flags/999') + .set('X-API-Key', MASTER) + .send({ status: 'confirmed' }) + .expect(404); + await request(app) + .patch('/api/v1/admin/fraud/flags/1') + .set('X-API-Key', MASTER) + .send({ status: 'nonsense' }) + .expect(400); +}); + +// ── #1261 Horizon deposit watcher ──────────────────────────────────────────── + +test('operator deposit API exposes watcher status and deposit lookups behind the master key', async () => { + const app = await makeApp({ operatorDepositAccounts: 'GCUSTODY' }); + + await request(app).get('/api/v1/operator/deposits').set('X-API-Key', API_KEY).expect(401); + + const status = await request(app) + .get('/api/v1/operator/watcher/status') + .set('X-API-Key', MASTER) + .expect(200); + assert.equal(status.body.running, false, 'the stream is not opened when jobs are disabled'); + assert.equal(status.body.depositsRecorded, 0); + + const list = await request(app) + .get('/api/v1/operator/deposits') + .set('X-API-Key', MASTER) + .expect(200); + assert.deepEqual(list.body, { data: [], total: 0 }); + + await request(app) + .get('/api/v1/operator/deposits/unknown-tx') + .set('X-API-Key', MASTER) + .expect(404); +}); + +// ── #1259 API docs ─────────────────────────────────────────────────────────── + +test('GET /docs/api serves Swagger UI', async () => { + const app = await makeApp(); + const res = await request(app).get('/docs/api/').expect(200); + assert.match(res.headers['content-type'], /text\/html/); + assert.match(res.text, /swagger-ui/i); +}); + +test('GET /docs/api/openapi.json serves the generated OpenAPI 3.0 document', async () => { + const app = await makeApp(); + const res = await request(app).get('/docs/api/openapi.json').expect(200); + + assert.match(res.body.openapi, /^3\.[01]\./); + assert.ok(Object.keys(res.body.paths).length > 0, 'includes the documented paths'); + assert.ok(res.body.components.schemas.CampaignCreate, 'includes schemas generated from Zod'); + assert.ok(res.body.components.schemas.Claim); +}); + +test('the existing /docs Swagger UI keeps working', async () => { + const app = await makeApp(); + await request(app).get('/docs/').expect(200); +}); diff --git a/backend/src/routes/campaignExport.js b/backend/src/routes/campaignExport.js index 573776e5..93adbfc0 100644 --- a/backend/src/routes/campaignExport.js +++ b/backend/src/routes/campaignExport.js @@ -2,6 +2,11 @@ import { Router } from 'express'; import { Readable } from 'node:stream'; import { pipeline } from 'node:stream/promises'; +import { + PARTICIPANT_COLUMNS, + buildCsv, + collectParticipants, +} from '../services/campaignExportData.js'; const EXPORT_RATE_LIMIT_WINDOW_MS = 60 * 60 * 1000; const EXPORT_RATE_LIMIT_MAX = 5; @@ -33,23 +38,6 @@ function checkExportRateLimit(campaignId, actorKey) { }; } -/** - * RFC 4180-compliant CSV serializer. - * @param {string[]} columns - * @param {Record[]} rows - */ -function buildCsv(columns, rows) { - const escape = (v) => { - const s = v === null || v === undefined ? '' : String(v); - return s.includes(',') || s.includes('"') || s.includes('\n') || s.includes('\r') - ? '"' + s.replace(/"/g, '""') + '"' - : s; - }; - const lines = [columns.join(',')]; - for (const row of rows) lines.push(columns.map((c) => escape(row[c])).join(',')); - return lines.join('\n') + '\n'; -} - /** * @param {{ * db: InstanceType, @@ -96,91 +84,7 @@ export function createCampaignExportRoute({ }); } - const hasCreditEvents = db - .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='credit_events'") - .get(); - const hasClaimEvents = db - .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='claim_events'") - .get(); - - let participants = []; - - if (hasCreditEvents) { - const dateFilters = []; - const vals = [String(id)]; - if (fromDate) { - dateFilters.push('r.created_at >= ?'); - vals.push(fromDate); - } - if (toDate) { - dateFilters.push('r.created_at <= ?'); - vals.push(toDate); - } - const dateWhere = dateFilters.length ? `AND ${dateFilters.join(' AND ')}` : ''; - - participants = db - .prepare( - ` - WITH participants AS ( - SELECT DISTINCT user FROM credit_events - ), - credited AS ( - SELECT user, SUM(CAST(amount AS INTEGER)) AS total FROM credit_events GROUP BY user - ), - claimed AS ( - SELECT user, SUM(CAST(amount AS INTEGER)) AS total FROM claim_events GROUP BY user - ) - SELECT - p.user AS participantAddress, - r.created_at AS registeredAt, - COALESCE(cr.total, 0) AS pointsCredited, - COALESCE(cl.total, 0) AS pointsClaimed, - COALESCE(cr.total, 0) - COALESCE(cl.total, 0) AS netPoints, - ref.referrer_address AS referredBy - FROM participants p - LEFT JOIN referrals r - ON r.referee_address = p.user AND r.campaign_id = ? ${dateWhere} - LEFT JOIN credited cr ON cr.user = p.user - LEFT JOIN ${hasClaimEvents ? 'claimed' : '(SELECT NULL AS user, 0 AS total) dummy_cl'} cl ON cl.user = p.user - LEFT JOIN referrals ref - ON ref.referee_address = p.user AND ref.campaign_id = ? - `, - ) - .all(...vals, String(id)); - } else { - // Fall back to referrals-only when event tables haven't been created yet - const dateFilters = []; - const vals = [String(id)]; - if (fromDate) { - dateFilters.push('created_at >= ?'); - vals.push(fromDate); - } - if (toDate) { - dateFilters.push('created_at <= ?'); - vals.push(toDate); - } - const dateWhere = dateFilters.length ? `AND ${dateFilters.join(' AND ')}` : ''; - - const rows = db - .prepare( - ` - SELECT referee_address, referrer_address, created_at - FROM referrals - WHERE campaign_id = ? ${dateWhere} - ORDER BY created_at ASC - `, - ) - .all(...vals); - - participants = rows.map((row) => ({ - participantAddress: row.referee_address, - registeredAt: row.created_at, - pointsCredited: 0, - pointsClaimed: 0, - netPoints: 0, - referredBy: row.referrer_address ?? null, - })); - } + const participants = collectParticipants(db, id, { fromDate, toDate }); try { auditLogRepository.create({ @@ -197,17 +101,9 @@ export function createCampaignExportRoute({ const filename = `campaign-${id}-export.${format}`; if (format === 'csv') { - const columns = [ - 'participantAddress', - 'registeredAt', - 'pointsCredited', - 'pointsClaimed', - 'netPoints', - 'referredBy', - ]; res.setHeader('Content-Type', 'text/csv; charset=utf-8'); res.setHeader('Content-Disposition', `attachment; filename="${filename}"`); - await pipeline(Readable.from([buildCsv(columns, participants)]), res); + await pipeline(Readable.from([buildCsv(PARTICIPANT_COLUMNS, participants)]), res); } else { const payload = JSON.stringify( { campaign: { id: campaign.id, name: campaign.name }, participants }, diff --git a/backend/src/routes/campaignExportJobs.js b/backend/src/routes/campaignExportJobs.js new file mode 100644 index 00000000..8b709859 --- /dev/null +++ b/backend/src/routes/campaignExportJobs.js @@ -0,0 +1,122 @@ +// @ts-check +import { Router } from 'express'; +import { CAMPAIGN_EXPORT_JOB_TYPE } from '../jobs/campaignExportWorker.js'; + +const FORMATS = new Set(['csv', 'json']); + +/** @param {unknown} value */ +function isValidDate(value) { + return typeof value === 'string' && value.length > 0 && !Number.isNaN(Date.parse(value)); +} + +/** + * Asynchronous campaign exports (#1260). + * + * POST /campaigns/:id/exports -> 202, queues the export + * GET /campaigns/:id/exports -> recent export jobs + * GET /campaigns/:id/exports/:jobId -> status (+ downloadUrl once completed) + * + * The synchronous `GET /campaigns/:id/export` endpoint is unchanged. + * + * @param {{ + * repository: ReturnType, + * campaignRepository: { getById: (id: string) => unknown }, + * jobQueue: { enqueue: (type: string, payload: unknown, opts?: object) => void }, + * guard?: import('express').RequestHandler[], + * recordAudit?: (req: import('express').Request, entry: object) => void, + * logger?: { info?: Function }, + * }} options + */ +export function createCampaignExportJobRoutes({ + repository, + campaignRepository, + jobQueue, + guard = [], + recordAudit, + logger = console, +}) { + const router = Router(); + + /** @param {import('../dal/sqliteExportJobRepository.js').ExportJob} job @param {string} base */ + function present(job, base) { + return { + jobId: job.id, + campaignId: job.campaignId, + format: job.format, + status: job.status, + from: job.fromDate, + to: job.toDate, + rowCount: job.rowCount, + downloadUrl: job.status === 'completed' ? job.fileUrl : null, + error: job.status === 'failed' ? job.error : null, + createdAt: job.createdAt, + startedAt: job.startedAt, + completedAt: job.completedAt, + statusUrl: `${base}/campaigns/${job.campaignId}/exports/${job.id}`, + }; + } + + router.post('/campaigns/:id/exports', ...guard, (req, res) => { + const campaign = campaignRepository.getById(req.params.id); + if (!campaign) { + return res.status(404).json({ error: 'Campaign not found', code: 'NOT_FOUND' }); + } + + const format = String(req.body?.format ?? 'csv').toLowerCase(); + if (!FORMATS.has(format)) { + return res + .status(400) + .json({ error: 'Invalid format. Use "csv" or "json"', code: 'INVALID_FORMAT' }); + } + + const { from, to } = req.body ?? {}; + for (const [name, value] of [ + ['from', from], + ['to', to], + ]) { + if (value !== undefined && value !== null && !isValidDate(value)) { + return res + .status(400) + .json({ error: `"${name}" must be an ISO 8601 date`, code: 'VALIDATION_ERROR' }); + } + } + + const job = repository.create({ + campaignId: req.params.id, + format: /** @type {'csv' | 'json'} */ (format), + fromDate: from ?? null, + toDate: to ?? null, + }); + jobQueue.enqueue(CAMPAIGN_EXPORT_JOB_TYPE, { exportJobId: job.id }, { maxAttempts: 3 }); + recordAudit?.(req, { + action: 'campaign.export.queued', + entity: 'campaign', + entityId: req.params.id, + diff: { jobId: job.id, format, from: job.fromDate, to: job.toDate }, + }); + logger.info?.({ exportJobId: job.id, campaignId: req.params.id }, 'campaignExport:queued'); + + return res.status(202).json(present(job, req.baseUrl)); + }); + + router.get('/campaigns/:id/exports', ...guard, (req, res) => { + if (!campaignRepository.getById(req.params.id)) { + return res.status(404).json({ error: 'Campaign not found', code: 'NOT_FOUND' }); + } + const limit = Number.parseInt(String(req.query.limit ?? ''), 10) || 20; + const data = repository + .listByCampaign(req.params.id, limit) + .map((j) => present(j, req.baseUrl)); + return res.json({ data, total: data.length }); + }); + + router.get('/campaigns/:id/exports/:jobId', ...guard, (req, res) => { + const job = repository.getById(req.params.jobId); + if (!job || job.campaignId !== String(req.params.id)) { + return res.status(404).json({ error: 'Export job not found', code: 'NOT_FOUND' }); + } + return res.json(present(job, req.baseUrl)); + }); + + return router; +} diff --git a/backend/src/routes/campaignExportJobs.test.js b/backend/src/routes/campaignExportJobs.test.js new file mode 100644 index 00000000..5b44b124 --- /dev/null +++ b/backend/src/routes/campaignExportJobs.test.js @@ -0,0 +1,180 @@ +// @ts-check +/** + * Integration tests for #1260 — asynchronous campaign exports. + * Run with: node --test src/routes/campaignExportJobs.test.js + */ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import request from 'supertest'; +import { createApp } from '../index.js'; + +const API_KEY = 'test-key-123'; + +function makeStorage() { + /** @type {Array<{ filename: string, buffer: Buffer }>} */ + const uploads = []; + return { + backendName: 'test', + uploads, + async upload({ buffer, filename }) { + uploads.push({ filename, buffer }); + return { url: `https://files.test/${filename}`, key: filename }; + }, + }; +} + +async function makeApp(options = {}) { + const storage = makeStorage(); + const app = await createApp({ + dbPath: ':memory:', + campaigns: [], + disableJobs: true, + skipEnvValidation: true, + rateLimit: { windowMs: 60_000, maxRequests: 10_000 }, + apiKeys: API_KEY, + storageAdapter: storage, + ...options, + }); + return { app, storage }; +} + +let campaignSeq = 0; +async function createCampaign(app) { + const res = await request(app) + .post('/api/v1/campaigns') + .set('X-API-Key', API_KEY) + .send({ name: `Export Me ${++campaignSeq}`, rewardPerAction: 1 }) + .expect(201); + return res.body; +} + +async function waitForStatus(app, url, wanted, attempts = 50) { + for (let i = 0; i < attempts; i++) { + const res = await request(app).get(url).set('X-API-Key', API_KEY); + if (res.body.status === wanted) return res.body; + await new Promise((resolve) => setTimeout(resolve, 20)); + } + throw new Error(`export job never reached status "${wanted}"`); +} + +test('POST /campaigns/:id/exports returns 202 immediately and the job completes in the background', async () => { + const { app, storage } = await makeApp(); + const campaign = await createCampaign(app); + + const res = await request(app) + .post(`/api/v1/campaigns/${campaign.id}/exports`) + .set('X-API-Key', API_KEY) + .send({ format: 'csv' }) + .expect(202); + + assert.equal(res.body.status, 'queued'); + assert.equal(res.body.format, 'csv'); + assert.equal(res.body.downloadUrl, null); + assert.ok(res.body.jobId); + assert.equal(res.body.statusUrl, `/api/v1/campaigns/${campaign.id}/exports/${res.body.jobId}`); + + const done = await waitForStatus(app, res.body.statusUrl, 'completed'); + assert.equal( + done.downloadUrl, + `https://files.test/exports/campaigns/${campaign.id}/${res.body.jobId}.csv`, + ); + assert.equal(done.rowCount, 0); + assert.equal(storage.uploads.length, 1); + assert.match(storage.uploads[0].buffer.toString('utf8'), /^participantAddress,/); +}); + +test('exports default to CSV and list newest first', async () => { + const { app } = await makeApp(); + const campaign = await createCampaign(app); + + const first = await request(app) + .post(`/api/v1/campaigns/${campaign.id}/exports`) + .set('X-API-Key', API_KEY) + .send({}) + .expect(202); + assert.equal(first.body.format, 'csv'); + const second = await request(app) + .post(`/api/v1/campaigns/${campaign.id}/exports`) + .set('X-API-Key', API_KEY) + .send({ format: 'JSON' }) + .expect(202); + + const list = await request(app) + .get(`/api/v1/campaigns/${campaign.id}/exports`) + .set('X-API-Key', API_KEY) + .expect(200); + assert.equal(list.body.total, 2); + assert.equal(list.body.data[0].jobId, second.body.jobId); + assert.equal(list.body.data[0].format, 'json'); +}); + +test('validates format and date inputs', async () => { + const { app } = await makeApp(); + const campaign = await createCampaign(app); + const post = (body) => + request(app) + .post(`/api/v1/campaigns/${campaign.id}/exports`) + .set('X-API-Key', API_KEY) + .send(body); + + assert.equal((await post({ format: 'xml' }).expect(400)).body.code, 'INVALID_FORMAT'); + assert.equal((await post({ from: 'yesterday-ish' }).expect(400)).body.code, 'VALIDATION_ERROR'); + assert.equal((await post({ to: 123 }).expect(400)).body.code, 'VALIDATION_ERROR'); + await post({ from: '2026-01-01', to: '2026-12-31' }).expect(202); +}); + +test('unknown campaigns and jobs return 404; jobs are scoped to their campaign', async () => { + const { app } = await makeApp(); + const campaign = await createCampaign(app); + const other = await createCampaign(app); + + await request(app) + .post('/api/v1/campaigns/missing/exports') + .set('X-API-Key', API_KEY) + .send({}) + .expect(404); + await request(app).get('/api/v1/campaigns/missing/exports').set('X-API-Key', API_KEY).expect(404); + + const job = await request(app) + .post(`/api/v1/campaigns/${campaign.id}/exports`) + .set('X-API-Key', API_KEY) + .send({}) + .expect(202); + + await request(app) + .get(`/api/v1/campaigns/${campaign.id}/exports/does-not-exist`) + .set('X-API-Key', API_KEY) + .expect(404); + await request(app) + .get(`/api/v1/campaigns/${other.id}/exports/${job.body.jobId}`) + .set('X-API-Key', API_KEY) + .expect(404); +}); + +test('requires an API key', async () => { + const { app } = await makeApp(); + const campaign = await createCampaign(app); + await request(app).post(`/api/v1/campaigns/${campaign.id}/exports`).send({}).expect(401); + await request(app).get(`/api/v1/campaigns/${campaign.id}/exports`).expect(401); +}); + +test('a failing upload surfaces as a failed job with the error message', async () => { + const { app } = await makeApp({ + storageAdapter: { + backendName: 'broken', + async upload() { + throw new Error('bucket unavailable'); + }, + }, + }); + const campaign = await createCampaign(app); + const res = await request(app) + .post(`/api/v1/campaigns/${campaign.id}/exports`) + .set('X-API-Key', API_KEY) + .send({}) + .expect(202); + + const failed = await waitForStatus(app, res.body.statusUrl, 'failed'); + assert.equal(failed.error, 'bucket unavailable'); + assert.equal(failed.downloadUrl, null); +}); diff --git a/backend/src/routes/fraudFlags.js b/backend/src/routes/fraudFlags.js new file mode 100644 index 00000000..88f41367 --- /dev/null +++ b/backend/src/routes/fraudFlags.js @@ -0,0 +1,57 @@ +// @ts-check +import { Router } from 'express'; +import { FLAG_STATUSES } from '../services/fraudDetection.js'; + +/** + * Admin review API for fraud flags (#1258). Mount behind the master key. + * + * GET /flags?status=&campaignId=&limit=&offset= + * PATCH /flags/:id { status: 'open' | 'dismissed' | 'confirmed' } + * + * @param {{ + * fraudDetector: ReturnType, + * recordAudit?: (req: import('express').Request, entry: object) => void, + * }} options + */ +export function createFraudFlagRoutes({ fraudDetector, recordAudit }) { + const router = Router(); + + router.get('/flags', (req, res) => { + const status = typeof req.query.status === 'string' ? req.query.status : undefined; + if (status && !FLAG_STATUSES.includes(status)) { + return res.status(400).json({ + error: `status must be one of: ${FLAG_STATUSES.join(', ')}`, + code: 'VALIDATION_ERROR', + }); + } + const result = fraudDetector.listFlags({ + status, + campaignId: typeof req.query.campaignId === 'string' ? req.query.campaignId : undefined, + limit: Number.parseInt(String(req.query.limit ?? ''), 10) || 50, + offset: Number.parseInt(String(req.query.offset ?? ''), 10) || 0, + }); + return res.json(result); + }); + + router.patch('/flags/:id', (req, res) => { + const status = req.body?.status; + if (!FLAG_STATUSES.includes(status)) { + return res.status(400).json({ + error: `status must be one of: ${FLAG_STATUSES.join(', ')}`, + code: 'VALIDATION_ERROR', + }); + } + const flag = fraudDetector.updateFlagStatus(Number(req.params.id), status); + if (!flag) return res.status(404).json({ error: 'Flag not found', code: 'NOT_FOUND' }); + + recordAudit?.(req, { + action: 'fraud.flag.updated', + entity: 'fraud_flag', + entityId: String(flag.id), + diff: { status }, + }); + return res.json(flag); + }); + + return router; +} diff --git a/backend/src/routes/operatorDeposits.js b/backend/src/routes/operatorDeposits.js new file mode 100644 index 00000000..b386fac9 --- /dev/null +++ b/backend/src/routes/operatorDeposits.js @@ -0,0 +1,43 @@ +// @ts-check +import { Router } from 'express'; + +/** + * Read API over deposits seen by the Horizon watcher (#1261). Mount behind the + * master key: it exposes custody-account activity. + * + * GET /deposits?account=&limit=&beforeId= recent deposits + * GET /deposits/:txHash verify a deposit by transaction hash + * GET /watcher/status stream health, cursors and counters + * + * @param {{ watcher: ReturnType }} options + */ +export function createOperatorDepositRoutes({ watcher }) { + const router = Router(); + + router.get('/deposits', (req, res) => { + const data = watcher.listDeposits({ + account: typeof req.query.account === 'string' ? req.query.account : undefined, + limit: Number.parseInt(String(req.query.limit ?? ''), 10) || 50, + beforeId: Number.parseInt(String(req.query.beforeId ?? ''), 10) || undefined, + }); + res.json({ data, total: data.length }); + }); + + router.get('/deposits/:txHash', (req, res) => { + const deposits = watcher.getByTxHash(req.params.txHash); + if (deposits.length === 0) { + return res.status(404).json({ error: 'Deposit not found', code: 'NOT_FOUND' }); + } + return res.json({ + txHash: req.params.txHash, + verified: deposits.every((d) => d.verified), + deposits, + }); + }); + + router.get('/watcher/status', (_req, res) => { + res.json(watcher.getStatus()); + }); + + return router; +} diff --git a/backend/src/services/campaignExportData.js b/backend/src/services/campaignExportData.js new file mode 100644 index 00000000..69031fcc --- /dev/null +++ b/backend/src/services/campaignExportData.js @@ -0,0 +1,128 @@ +// @ts-check +// Shared campaign-export data access, used by both the synchronous export +// endpoint (routes/campaignExport.js) and the async export worker +// (jobs/campaignExportWorker.js) so the two can never drift apart. + +export const PARTICIPANT_COLUMNS = [ + 'participantAddress', + 'registeredAt', + 'pointsCredited', + 'pointsClaimed', + 'netPoints', + 'referredBy', +]; + +/** + * RFC 4180-compliant CSV serializer. + * @param {string[]} columns + * @param {Record[]} rows + */ +export function buildCsv(columns, rows) { + const escape = (v) => { + const s = v === null || v === undefined ? '' : String(v); + return s.includes(',') || s.includes('"') || s.includes('\n') || s.includes('\r') + ? '"' + s.replace(/"/g, '""') + '"' + : s; + }; + const lines = [columns.join(',')]; + for (const row of rows) lines.push(columns.map((c) => escape(row[c])).join(',')); + return lines.join('\n') + '\n'; +} + +/** + * Collect the participant rows for a campaign export. + * + * @param {InstanceType} db + * @param {string | number} campaignId + * @param {{ fromDate?: string | null, toDate?: string | null }} [range] + * @returns {Record[]} + */ +export function collectParticipants(db, campaignId, { fromDate = null, toDate = null } = {}) { + const hasCreditEvents = db + .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='credit_events'") + .get(); + const hasClaimEvents = db + .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='claim_events'") + .get(); + + let participants = []; + + if (hasCreditEvents) { + const dateFilters = []; + const vals = [String(campaignId)]; + if (fromDate) { + dateFilters.push('r.created_at >= ?'); + vals.push(fromDate); + } + if (toDate) { + dateFilters.push('r.created_at <= ?'); + vals.push(toDate); + } + const dateWhere = dateFilters.length ? `AND ${dateFilters.join(' AND ')}` : ''; + + participants = db + .prepare( + ` + WITH participants AS ( + SELECT DISTINCT user FROM credit_events + ), + credited AS ( + SELECT user, SUM(CAST(amount AS INTEGER)) AS total FROM credit_events GROUP BY user + ), + claimed AS ( + SELECT user, SUM(CAST(amount AS INTEGER)) AS total FROM claim_events GROUP BY user + ) + SELECT + p.user AS participantAddress, + r.created_at AS registeredAt, + COALESCE(cr.total, 0) AS pointsCredited, + COALESCE(cl.total, 0) AS pointsClaimed, + COALESCE(cr.total, 0) - COALESCE(cl.total, 0) AS netPoints, + ref.referrer_address AS referredBy + FROM participants p + LEFT JOIN referrals r + ON r.referee_address = p.user AND r.campaign_id = ? ${dateWhere} + LEFT JOIN credited cr ON cr.user = p.user + LEFT JOIN ${hasClaimEvents ? 'claimed' : '(SELECT NULL AS user, 0 AS total) dummy_cl'} cl ON cl.user = p.user + LEFT JOIN referrals ref + ON ref.referee_address = p.user AND ref.campaign_id = ? + `, + ) + .all(...vals, String(campaignId)); + } else { + // Fall back to referrals-only when event tables haven't been created yet + const dateFilters = []; + const vals = [String(campaignId)]; + if (fromDate) { + dateFilters.push('created_at >= ?'); + vals.push(fromDate); + } + if (toDate) { + dateFilters.push('created_at <= ?'); + vals.push(toDate); + } + const dateWhere = dateFilters.length ? `AND ${dateFilters.join(' AND ')}` : ''; + + const rows = db + .prepare( + ` + SELECT referee_address, referrer_address, created_at + FROM referrals + WHERE campaign_id = ? ${dateWhere} + ORDER BY created_at ASC + `, + ) + .all(...vals); + + participants = rows.map((row) => ({ + participantAddress: row.referee_address, + registeredAt: row.created_at, + pointsCredited: 0, + pointsClaimed: 0, + netPoints: 0, + referredBy: row.referrer_address ?? null, + })); + } + + return participants; +} diff --git a/backend/src/services/fraudDetection.js b/backend/src/services/fraudDetection.js new file mode 100644 index 00000000..4f1b0857 --- /dev/null +++ b/backend/src/services/fraudDetection.js @@ -0,0 +1,278 @@ +// @ts-check +import { isIPv4, isIPv6 } from 'node:net'; + +/** + * Automated fraud detection for multi-account signups (#1258). + * + * Signups are grouped by network prefix (/24 for IPv4, /64 for IPv6) and by + * known proxy exit nodes. When too many distinct accounts register from one + * group inside the sliding window, a flag is raised for manual review. Flags + * never block a signup; they only surface anomalies to operators. + * + * Privacy: only the network prefix of an address is persisted, except for + * addresses that match the configured (public) proxy exit list. + */ + +export const RULE_SUBNET_CLUSTER = 'ip_subnet_cluster'; +export const RULE_PROXY_EXIT = 'proxy_exit_node'; +export const FLAG_STATUSES = ['open', 'dismissed', 'confirmed']; + +/** Strip brackets, zone ids and the IPv4-mapped IPv6 prefix. @param {unknown} raw */ +export function normalizeIp(raw) { + if (typeof raw !== 'string') return null; + let ip = raw.trim(); + if (ip.startsWith('[') && ip.includes(']')) ip = ip.slice(1, ip.indexOf(']')); + ip = ip.split('%')[0]; + const mapped = ip.match(/^::ffff:(\d{1,3}(?:\.\d{1,3}){3})$/i); + if (mapped) ip = mapped[1]; + return isIPv4(ip) || isIPv6(ip) ? ip.toLowerCase() : null; +} + +/** @param {string} ip @returns {number} */ +function ipv4ToInt(ip) { + return ip.split('.').reduce((acc, octet) => acc * 256 + Number(octet), 0); +} + +/** @param {string} ip IPv6 address @returns {string[]} eight hextets */ +function expandIpv6(ip) { + const [head, tail] = ip.split('::'); + const headParts = head ? head.split(':') : []; + const tailParts = tail ? tail.split(':') : []; + const fill = ip.includes('::') ? 8 - headParts.length - tailParts.length : 0; + return [...headParts, ...Array(Math.max(fill, 0)).fill('0'), ...tailParts].map((h) => + parseInt(h, 16).toString(16), + ); +} + +/** + * Network prefix used to cluster signups: /24 for IPv4, /64 for IPv6. + * + * @param {unknown} raw + * @returns {string | null} e.g. `203.0.113.0/24` or `2001:db8:1:2::/64` + */ +export function getSubnetKey(raw) { + const ip = normalizeIp(raw); + if (!ip) return null; + if (isIPv4(ip)) { + const [a, b, c] = ip.split('.'); + return `${a}.${b}.${c}.0/24`; + } + const hextets = expandIpv6(ip); + if (hextets.length !== 8 || hextets.some((h) => h === 'NaN')) return null; + return `${hextets.slice(0, 4).join(':')}::/64`; +} + +/** + * Build a matcher for a list of proxy exit nodes. Entries may be exact IPs or + * IPv4 CIDR ranges (e.g. `198.51.100.0/24`). + * + * @param {string[]} entries + * @returns {(ip: unknown) => string | null} the matching list entry's base address, or null + */ +export function createProxyMatcher(entries = []) { + /** @type {Set} */ + const exact = new Set(); + /** @type {Array<{ base: number, mask: number }>} */ + const ranges = []; + + for (const entry of entries.map((e) => e.trim()).filter(Boolean)) { + const [addr, bitsRaw] = entry.split('/'); + const ip = normalizeIp(addr); + if (!ip) continue; + if (bitsRaw === undefined) { + exact.add(ip); + } else if (isIPv4(ip)) { + const bits = Number(bitsRaw); + if (!Number.isInteger(bits) || bits < 0 || bits > 32) continue; + const mask = bits === 0 ? 0 : (0xffffffff << (32 - bits)) >>> 0; + ranges.push({ base: (ipv4ToInt(ip) & mask) >>> 0, mask }); + } + } + + return (raw) => { + const ip = normalizeIp(raw); + if (!ip) return null; + if (exact.has(ip)) return ip; + if (isIPv4(ip)) { + const value = ipv4ToInt(ip); + if (ranges.some((r) => (value & r.mask) >>> 0 === r.base)) return ip; + } + return null; + }; +} + +/** + * @typedef {{ + * id: number, campaignId: string, rule: string, subject: string, accountCount: number, + * accounts: string[], status: string, firstSeenAt: string, lastSeenAt: string, + * }} FraudFlag + */ + +/** @returns {FraudFlag} */ +function rowToFlag(row) { + let accounts = []; + try { + accounts = JSON.parse(row.accounts); + } catch { + /* keep [] for corrupt rows */ + } + return { + id: row.id, + campaignId: row.campaign_id, + rule: row.rule, + subject: row.subject, + accountCount: row.account_count, + accounts, + status: row.status, + firstSeenAt: row.first_seen_at, + lastSeenAt: row.last_seen_at, + }; +} + +/** + * @param {{ + * db: InstanceType, + * subnetThreshold?: number, + * proxyThreshold?: number, + * windowMs?: number, + * proxyList?: string[], + * logger?: { warn?: Function }, + * now?: () => number, + * }} options + */ +export function createFraudDetector({ + db, + subnetThreshold = 3, + proxyThreshold = 2, + windowMs = 24 * 60 * 60 * 1000, + proxyList = [], + logger = console, + now = () => Date.now(), +}) { + const matchProxy = createProxyMatcher(proxyList); + + const insertEvent = db.prepare( + `INSERT INTO signup_events (campaign_id, account, ip_subnet, proxy_exit, created_at) + VALUES (?, ?, ?, ?, ?)`, + ); + const subnetAccounts = db.prepare( + `SELECT DISTINCT account FROM signup_events + WHERE campaign_id = ? AND ip_subnet = ? AND created_at >= ?`, + ); + const proxyAccounts = db.prepare( + `SELECT DISTINCT account FROM signup_events + WHERE campaign_id = ? AND proxy_exit = ? AND created_at >= ?`, + ); + const upsertFlag = db.prepare( + `INSERT INTO fraud_flags + (campaign_id, rule, subject, account_count, accounts, status, first_seen_at, last_seen_at) + VALUES (@campaignId, @rule, @subject, @count, @accounts, 'open', @at, @at) + ON CONFLICT (campaign_id, rule, subject) DO UPDATE SET + account_count = excluded.account_count, + accounts = excluded.accounts, + last_seen_at = excluded.last_seen_at, + status = CASE + WHEN fraud_flags.status = 'dismissed' AND excluded.account_count > fraud_flags.account_count + THEN 'open' + ELSE fraud_flags.status + END`, + ); + const flagBySubject = db.prepare( + 'SELECT * FROM fraud_flags WHERE campaign_id = ? AND rule = ? AND subject = ?', + ); + + /** + * @param {string} campaignId + * @param {string} rule + * @param {string} subject + * @param {string[]} accounts + * @param {string} at + */ + function raise(campaignId, rule, subject, accounts, at) { + upsertFlag.run({ + campaignId, + rule, + subject, + count: accounts.length, + accounts: JSON.stringify(accounts), + at, + }); + logger.warn?.({ campaignId, rule, subject, accounts: accounts.length }, 'fraud:flag'); + return rowToFlag(flagBySubject.get(campaignId, rule, subject)); + } + + return { + /** + * Record a signup and evaluate the clustering rules. + * + * @param {{ campaignId: string | number, account: string, ip: unknown }} signup + * @returns {{ flags: FraudFlag[] }} flags raised or refreshed by this signup + */ + recordSignup({ campaignId, account, ip }) { + const subnet = getSubnetKey(ip); + if (!subnet) return { flags: [] }; + + const campaign = String(campaignId); + const at = new Date(now()).toISOString(); + const since = new Date(now() - windowMs).toISOString(); + const proxyExit = matchProxy(ip); + + insertEvent.run(campaign, account, subnet, proxyExit, at); + + /** @type {FraudFlag[]} */ + const flags = []; + + const inSubnet = subnetAccounts.all(campaign, subnet, since).map((r) => r.account); + if (inSubnet.length >= subnetThreshold) { + flags.push(raise(campaign, RULE_SUBNET_CLUSTER, subnet, inSubnet, at)); + } + + if (proxyExit) { + const viaProxy = proxyAccounts.all(campaign, proxyExit, since).map((r) => r.account); + if (viaProxy.length >= proxyThreshold) { + flags.push(raise(campaign, RULE_PROXY_EXIT, proxyExit, viaProxy, at)); + } + } + + return { flags }; + }, + + /** + * @param {{ status?: string, campaignId?: string, limit?: number, offset?: number }} [filter] + */ + listFlags({ status, campaignId, limit = 50, offset = 0 } = {}) { + const where = []; + const params = []; + if (status) { + where.push('status = ?'); + params.push(status); + } + if (campaignId) { + where.push('campaign_id = ?'); + params.push(String(campaignId)); + } + const clause = where.length ? `WHERE ${where.join(' AND ')}` : ''; + const total = db.prepare(`SELECT COUNT(*) AS n FROM fraud_flags ${clause}`).get(...params).n; + const rows = db + .prepare( + `SELECT * FROM fraud_flags ${clause} ORDER BY last_seen_at DESC, id DESC LIMIT ? OFFSET ?`, + ) + .all(...params, Math.min(Math.max(limit, 1), 200), Math.max(offset, 0)); + return { data: rows.map(rowToFlag), total }; + }, + + /** @param {number} id @param {string} status @returns {FraudFlag | null} */ + updateFlagStatus(id, status) { + if (!FLAG_STATUSES.includes(status)) throw new Error(`Invalid status: ${status}`); + const result = db.prepare('UPDATE fraud_flags SET status = ? WHERE id = ?').run(status, id); + if (result.changes === 0) return null; + return rowToFlag(db.prepare('SELECT * FROM fraud_flags WHERE id = ?').get(id)); + }, + + /** Drop signup events that have aged out of the detection window. */ + pruneEvents() { + const cutoff = new Date(now() - windowMs).toISOString(); + return db.prepare('DELETE FROM signup_events WHERE created_at < ?').run(cutoff).changes; + }, + }; +} diff --git a/backend/src/services/fraudDetection.test.js b/backend/src/services/fraudDetection.test.js new file mode 100644 index 00000000..875c3538 --- /dev/null +++ b/backend/src/services/fraudDetection.test.js @@ -0,0 +1,176 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import Database from 'better-sqlite3'; +import { runMigrations } from '../db/migrate.js'; +import { + RULE_PROXY_EXIT, + RULE_SUBNET_CLUSTER, + createFraudDetector, + createProxyMatcher, + getSubnetKey, + normalizeIp, +} from './fraudDetection.js'; + +async function makeDb() { + const db = new Database(':memory:'); + await runMigrations(db); + return db; +} + +test('getSubnetKey: IPv4 uses /24, IPv6 uses /64, mapped IPv4 is unwrapped', () => { + assert.equal(getSubnetKey('203.0.113.77'), '203.0.113.0/24'); + assert.equal(getSubnetKey('::ffff:203.0.113.9'), '203.0.113.0/24'); + assert.equal(getSubnetKey('2001:db8:1:2:aaaa:bbbb:cccc:dddd'), '2001:db8:1:2::/64'); + assert.equal(getSubnetKey('2001:db8::1'), '2001:db8:0:0::/64'); + assert.equal(getSubnetKey('[2001:DB8::1]'), '2001:db8:0:0::/64'); +}); + +test('getSubnetKey / normalizeIp: reject junk', () => { + for (const bad of [undefined, null, '', 'not-an-ip', '999.1.1.1', 42]) { + assert.equal(getSubnetKey(bad), null); + assert.equal(normalizeIp(bad), null); + } +}); + +test('createProxyMatcher: exact IPs and IPv4 CIDR ranges', () => { + const match = createProxyMatcher(['192.0.2.10', '198.51.100.0/24', 'bad-entry', '10.0.0.0/99']); + assert.equal(match('192.0.2.10'), '192.0.2.10'); + assert.equal(match('198.51.100.200'), '198.51.100.200'); + assert.equal(match('198.51.101.1'), null); + assert.equal(match('10.0.0.1'), null); + assert.equal(match('garbage'), null); +}); + +test('recordSignup: flags once distinct accounts from one /24 reach the threshold', async () => { + const db = await makeDb(); + const detector = createFraudDetector({ db, subnetThreshold: 3, logger: {} }); + + assert.deepEqual( + detector.recordSignup({ campaignId: 'c1', account: 'GA', ip: '203.0.113.1' }).flags, + [], + ); + assert.deepEqual( + detector.recordSignup({ campaignId: 'c1', account: 'GB', ip: '203.0.113.2' }).flags, + [], + ); + // A repeated account must not inflate the distinct count. + assert.deepEqual( + detector.recordSignup({ campaignId: 'c1', account: 'GB', ip: '203.0.113.2' }).flags, + [], + ); + + const { flags } = detector.recordSignup({ campaignId: 'c1', account: 'GC', ip: '203.0.113.3' }); + assert.equal(flags.length, 1); + assert.equal(flags[0].rule, RULE_SUBNET_CLUSTER); + assert.equal(flags[0].subject, '203.0.113.0/24'); + assert.equal(flags[0].accountCount, 3); + assert.deepEqual(flags[0].accounts.sort(), ['GA', 'GB', 'GC']); + assert.equal(flags[0].status, 'open'); +}); + +test('recordSignup: different subnets and different campaigns are not clustered', async () => { + const db = await makeDb(); + const detector = createFraudDetector({ db, subnetThreshold: 2, logger: {} }); + + detector.recordSignup({ campaignId: 'c1', account: 'GA', ip: '203.0.113.1' }); + assert.equal( + detector.recordSignup({ campaignId: 'c1', account: 'GB', ip: '203.0.114.1' }).flags.length, + 0, + ); + assert.equal( + detector.recordSignup({ campaignId: 'c2', account: 'GC', ip: '203.0.113.9' }).flags.length, + 0, + ); + assert.equal(detector.listFlags().total, 0); +}); + +test('recordSignup: signups outside the sliding window do not count', async () => { + const db = await makeDb(); + let now = Date.parse('2026-01-01T00:00:00Z'); + const detector = createFraudDetector({ + db, + subnetThreshold: 2, + windowMs: 60 * 60 * 1000, + now: () => now, + logger: {}, + }); + + detector.recordSignup({ campaignId: 'c1', account: 'GA', ip: '203.0.113.1' }); + now += 2 * 60 * 60 * 1000; // 2h later — the first signup has aged out + assert.equal( + detector.recordSignup({ campaignId: 'c1', account: 'GB', ip: '203.0.113.2' }).flags.length, + 0, + ); + + assert.equal(detector.pruneEvents(), 1); +}); + +test('recordSignup: flags accounts sharing a proxy exit node across subnets', async () => { + const db = await makeDb(); + const detector = createFraudDetector({ + db, + subnetThreshold: 10, + proxyThreshold: 2, + proxyList: ['198.51.100.0/24', '192.0.2.50'], + logger: {}, + }); + + assert.equal( + detector.recordSignup({ campaignId: 'c1', account: 'GA', ip: '192.0.2.50' }).flags.length, + 0, + ); + assert.equal( + detector.recordSignup({ campaignId: 'c1', account: 'GA', ip: '192.0.2.50' }).flags.length, + 0, + 'the same account twice is still one account', + ); + + const { flags } = detector.recordSignup({ campaignId: 'c1', account: 'GB', ip: '192.0.2.50' }); + assert.equal(flags.length, 1); + assert.equal(flags[0].rule, RULE_PROXY_EXIT); + assert.equal(flags[0].subject, '192.0.2.50'); + assert.equal(flags[0].accountCount, 2); + + // A non-proxy address in the same range of accounts never raises the proxy rule. + assert.equal( + detector.recordSignup({ campaignId: 'c1', account: 'GC', ip: '203.0.113.5' }).flags.length, + 0, + ); +}); + +test('flags: list/filter, status updates, and dismissed flags reopen when the cluster grows', async () => { + const db = await makeDb(); + const detector = createFraudDetector({ db, subnetThreshold: 2, logger: {} }); + detector.recordSignup({ campaignId: 'c1', account: 'GA', ip: '203.0.113.1' }); + const [flag] = detector.recordSignup({ + campaignId: 'c1', + account: 'GB', + ip: '203.0.113.2', + }).flags; + + assert.equal(detector.listFlags({ status: 'open' }).total, 1); + assert.equal(detector.listFlags({ campaignId: 'other' }).total, 0); + + assert.equal(detector.updateFlagStatus(flag.id, 'dismissed').status, 'dismissed'); + assert.equal(detector.updateFlagStatus(9999, 'confirmed'), null); + assert.throws(() => detector.updateFlagStatus(flag.id, 'bogus'), /Invalid status/); + + const [grown] = detector.recordSignup({ + campaignId: 'c1', + account: 'GC', + ip: '203.0.113.3', + }).flags; + assert.equal(grown.id, flag.id, 'the same flag row is refreshed, not duplicated'); + assert.equal(grown.accountCount, 3); + assert.equal(grown.status, 'open', 'a dismissed flag reopens when more accounts join'); +}); + +test('recordSignup: ignores signups without a usable IP', async () => { + const db = await makeDb(); + const detector = createFraudDetector({ db, logger: {} }); + assert.deepEqual( + detector.recordSignup({ campaignId: 'c1', account: 'GA', ip: undefined }).flags, + [], + ); + assert.equal(db.prepare('SELECT COUNT(*) AS n FROM signup_events').get().n, 0); +}); diff --git a/backend/src/services/horizonDepositWatcher.js b/backend/src/services/horizonDepositWatcher.js new file mode 100644 index 00000000..bf2e4c87 --- /dev/null +++ b/backend/src/services/horizonDepositWatcher.js @@ -0,0 +1,301 @@ +// @ts-check +import { Horizon } from '@stellar/stellar-sdk'; + +/** + * Horizon account watcher for operator reward-token deposits (#1261). + * + * Streams payments for each watched contract-custody account, keeps only + * incoming payments of the configured reward asset, and records them + * idempotently so operators (and downstream checks) can verify that a deposit + * actually landed. Stream cursors are persisted so a restart resumes where the + * previous run stopped instead of missing deposits. + */ + +const INCOMING_TYPES = new Set([ + 'payment', + 'path_payment_strict_send', + 'path_payment_strict_receive', + 'create_account', +]); + +/** + * @typedef {{ + * pagingToken: string, account: string, from: string | null, assetCode: string, + * assetIssuer: string | null, amount: string, txHash: string, verified: boolean, + * observedAt: string, + * }} Deposit + */ + +/** + * Turn a Horizon operation record into a deposit, or null when it is not an + * incoming deposit of the watched asset into a watched account. + * + * @param {Record} record + * @param {{ watched: Set, asset: { code: string, issuer?: string | null } }} options + * @returns {Deposit | null} + */ +export function parseDeposit(record, { watched, asset }) { + if (!record || !INCOMING_TYPES.has(record.type)) return null; + + const isCreate = record.type === 'create_account'; + const to = isCreate ? record.account : record.to; + const from = isCreate ? record.funder : record.from; + if (!watched.has(to)) return null; + // Movement between two watched custody accounts is not an external deposit. + if (from && watched.has(from)) return null; + + const isNative = isCreate || record.asset_type === 'native'; + const assetCode = isNative ? 'XLM' : record.asset_code; + const assetIssuer = isNative ? null : (record.asset_issuer ?? null); + + const wantsNative = asset.code === 'XLM' && !asset.issuer; + if ( + wantsNative ? !isNative : assetCode !== asset.code || assetIssuer !== (asset.issuer ?? null) + ) { + return null; + } + + const amount = String(isCreate ? record.starting_balance : record.amount); + if (!(Number(amount) > 0)) return null; + + return { + pagingToken: String(record.paging_token), + account: to, + from: from ?? null, + assetCode, + assetIssuer, + amount, + txHash: record.transaction_hash, + // Horizon streams only successful operations unless include_failed is set. + verified: record.transaction_successful !== false, + observedAt: record.created_at ?? new Date().toISOString(), + }; +} + +/** + * @param {{ + * db: InstanceType, + * horizonUrl: string, + * accounts: string[], + * asset?: { code: string, issuer?: string | null }, + * allowHttp?: boolean, + * serverFactory?: (url: string, opts: { allowHttp: boolean }) => any, + * onDeposit?: (deposit: Deposit) => void, + * logger?: { info?: Function, warn?: Function, error?: Function }, + * reconnectBaseMs?: number, + * reconnectMaxMs?: number, + * }} options + */ +export function createHorizonDepositWatcher({ + db, + horizonUrl, + accounts, + asset = { code: 'XLM', issuer: null }, + allowHttp = false, + serverFactory = (url, opts) => new Horizon.Server(url, opts), + onDeposit, + logger = console, + reconnectBaseMs = 5_000, + reconnectMaxMs = 60_000, +}) { + const watched = new Set(accounts); + const insertStmt = db.prepare(` + INSERT OR IGNORE INTO operator_deposits + (paging_token, account, from_address, asset_code, asset_issuer, amount, tx_hash, + verified, observed_at, recorded_at) + VALUES (@pagingToken, @account, @from, @assetCode, @assetIssuer, @amount, @txHash, + @verified, @observedAt, @recordedAt) + `); + const getCursorStmt = db.prepare('SELECT cursor FROM horizon_watcher_cursors WHERE account = ?'); + const setCursorStmt = db.prepare(` + INSERT INTO horizon_watcher_cursors (account, cursor, updated_at) VALUES (?, ?, ?) + ON CONFLICT (account) DO UPDATE SET cursor = excluded.cursor, updated_at = excluded.updated_at + `); + + /** @type {Map void) | null, timer: any, attempts: number, connected: boolean, lastEventAt: string | null }>} */ + const streams = new Map(); + let running = false; + const metrics = { depositsRecorded: 0, duplicates: 0, errors: 0 }; + + /** + * Process one Horizon operation record. + * @param {Record} record + * @param {string} [streamAccount] the account whose stream delivered the record + * @returns {{ deposit: Deposit, duplicate: boolean } | null} + */ + function handleRecord(record, streamAccount = record?.to ?? record?.account) { + const deposit = parseDeposit(record, { watched, asset }); + // Advance the stream's cursor for every record (even ignored ones, e.g. + // outgoing payments) so a restart does not re-scan history. + if (record?.paging_token && watched.has(streamAccount)) { + setCursorStmt.run(streamAccount, String(record.paging_token), new Date().toISOString()); + } + if (!deposit) return null; + + const result = insertStmt.run({ + ...deposit, + verified: deposit.verified ? 1 : 0, + recordedAt: new Date().toISOString(), + }); + const duplicate = result.changes === 0; + if (duplicate) { + metrics.duplicates++; + } else { + metrics.depositsRecorded++; + logger.info?.( + { + account: deposit.account, + amount: deposit.amount, + asset: deposit.assetCode, + tx: deposit.txHash, + }, + 'depositWatcher:deposit', + ); + try { + onDeposit?.(deposit); + } catch (err) { + logger.warn?.({ err }, 'depositWatcher:onDeposit_failed'); + } + } + return { deposit, duplicate }; + } + + /** @param {string} account */ + function connect(account) { + if (!running) return; + const state = /** @type {NonNullable>} */ (streams.get(account)); + const saved = /** @type {{ cursor: string } | undefined} */ (getCursorStmt.get(account)); + + try { + const server = serverFactory(horizonUrl, { allowHttp }); + state.close = server + .payments() + .forAccount(account) + .cursor(saved?.cursor ?? 'now') + .stream({ + onmessage: (/** @type {Record} */ record) => { + state.connected = true; + state.attempts = 0; + state.lastEventAt = new Date().toISOString(); + try { + handleRecord(record, account); + } catch (err) { + metrics.errors++; + logger.error?.({ err, account }, 'depositWatcher:handle_error'); + } + }, + onerror: (/** @type {unknown} */ err) => { + metrics.errors++; + state.connected = false; + logger.warn?.({ err, account }, 'depositWatcher:stream_error'); + state.close?.(); + state.close = null; + scheduleReconnect(account); + }, + }); + } catch (err) { + metrics.errors++; + logger.error?.({ err, account }, 'depositWatcher:connect_failed'); + scheduleReconnect(account); + } + } + + /** @param {string} account */ + function scheduleReconnect(account) { + const state = streams.get(account); + if (!running || !state) return; + state.attempts += 1; + const delay = Math.min(reconnectBaseMs * 2 ** (state.attempts - 1), reconnectMaxMs); + state.timer = setTimeout(() => connect(account), delay); + state.timer.unref?.(); + } + + function start() { + if (running) return; + running = true; + for (const account of watched) { + streams.set(account, { + close: null, + timer: null, + attempts: 0, + connected: false, + lastEventAt: null, + }); + connect(account); + } + logger.info?.({ accounts: watched.size, horizonUrl }, 'depositWatcher:started'); + } + + function stop() { + running = false; + for (const state of streams.values()) { + if (state.timer) clearTimeout(state.timer); + state.close?.(); + state.close = null; + state.connected = false; + } + } + + function getStatus() { + return { + running, + asset: { code: asset.code, issuer: asset.issuer ?? null }, + accounts: [...streams.entries()].map(([account, s]) => ({ + account, + connected: s.connected, + reconnectAttempts: s.attempts, + lastEventAt: s.lastEventAt, + cursor: + /** @type {{ cursor: string } | undefined} */ (getCursorStmt.get(account))?.cursor ?? + null, + })), + ...metrics, + }; + } + + /** + * @param {{ account?: string, limit?: number, beforeId?: number }} [filter] + */ + function listDeposits({ account, limit = 50, beforeId } = {}) { + const where = []; + const params = []; + if (account) { + where.push('account = ?'); + params.push(account); + } + if (beforeId) { + where.push('id < ?'); + params.push(beforeId); + } + const clause = where.length ? `WHERE ${where.join(' AND ')}` : ''; + return db + .prepare(`SELECT * FROM operator_deposits ${clause} ORDER BY id DESC LIMIT ?`) + .all(...params, Math.min(Math.max(limit, 1), 200)) + .map(rowToDeposit); + } + + /** @param {string} txHash */ + function getByTxHash(txHash) { + return db + .prepare('SELECT * FROM operator_deposits WHERE tx_hash = ? ORDER BY id') + .all(txHash) + .map(rowToDeposit); + } + + return { start, stop, getStatus, handleRecord, listDeposits, getByTxHash }; +} + +function rowToDeposit(row) { + return { + id: row.id, + account: row.account, + from: row.from_address, + assetCode: row.asset_code, + assetIssuer: row.asset_issuer, + amount: row.amount, + txHash: row.tx_hash, + verified: Boolean(row.verified), + observedAt: row.observed_at, + recordedAt: row.recorded_at, + }; +} diff --git a/backend/src/services/horizonDepositWatcher.test.js b/backend/src/services/horizonDepositWatcher.test.js new file mode 100644 index 00000000..f5218b6e --- /dev/null +++ b/backend/src/services/horizonDepositWatcher.test.js @@ -0,0 +1,240 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import Database from 'better-sqlite3'; +import { runMigrations } from '../db/migrate.js'; +import { createHorizonDepositWatcher, parseDeposit } from './horizonDepositWatcher.js'; + +const CUSTODY = 'GCUSTODY1111111111111111111111111111111111111111111111'; +const OTHER_CUSTODY = 'GCUSTODY2222222222222222222222222222222222222222222222'; +const OPERATOR = 'GOPERATOR111111111111111111111111111111111111111111111'; +const ISSUER = 'GISSUER1111111111111111111111111111111111111111111111111'; +const REWARD = { code: 'TRV', issuer: ISSUER }; + +function payment(overrides = {}) { + return { + type: 'payment', + paging_token: '1001', + from: OPERATOR, + to: CUSTODY, + asset_type: 'credit_alphanum4', + asset_code: 'TRV', + asset_issuer: ISSUER, + amount: '250.0000000', + transaction_hash: 'tx-1', + transaction_successful: true, + created_at: '2026-09-26T00:00:00Z', + ...overrides, + }; +} + +const watched = new Set([CUSTODY, OTHER_CUSTODY]); + +test('parseDeposit: accepts an incoming reward-token payment', () => { + const d = parseDeposit(payment(), { watched, asset: REWARD }); + assert.deepEqual(d, { + pagingToken: '1001', + account: CUSTODY, + from: OPERATOR, + assetCode: 'TRV', + assetIssuer: ISSUER, + amount: '250.0000000', + txHash: 'tx-1', + verified: true, + observedAt: '2026-09-26T00:00:00Z', + }); +}); + +test('parseDeposit: ignores other assets, wrong issuers, outgoing and internal transfers', () => { + const opts = { watched, asset: REWARD }; + assert.equal(parseDeposit(payment({ asset_code: 'USDC' }), opts), null); + assert.equal( + parseDeposit(payment({ asset_issuer: 'GFAKE' }), opts), + null, + 'same code, different issuer', + ); + assert.equal(parseDeposit(payment({ asset_type: 'native', asset_code: undefined }), opts), null); + assert.equal(parseDeposit(payment({ from: CUSTODY, to: OPERATOR }), opts), null, 'outgoing'); + assert.equal( + parseDeposit(payment({ from: OTHER_CUSTODY }), opts), + null, + 'between custody accounts', + ); + assert.equal(parseDeposit(payment({ to: 'GSOMEONEELSE' }), opts), null); + assert.equal(parseDeposit(payment({ amount: '0.0000000' }), opts), null); + assert.equal(parseDeposit({ type: 'manage_offer', paging_token: '1' }, opts), null); + assert.equal(parseDeposit(null, opts), null); +}); + +test('parseDeposit: native XLM deposits, create_account and path payments', () => { + const native = { watched, asset: { code: 'XLM' } }; + assert.equal( + parseDeposit( + payment({ asset_type: 'native', asset_code: undefined, asset_issuer: undefined }), + native, + )?.assetCode, + 'XLM', + ); + const created = parseDeposit( + { + type: 'create_account', + paging_token: '7', + funder: OPERATOR, + account: CUSTODY, + starting_balance: '5.0', + transaction_hash: 'tx-c', + }, + native, + ); + assert.equal(created?.amount, '5.0'); + const path = parseDeposit(payment({ type: 'path_payment_strict_receive' }), { + watched, + asset: REWARD, + }); + assert.equal(path?.amount, '250.0000000'); +}); + +test('parseDeposit: marks unsuccessful transactions as unverified', () => { + const d = parseDeposit(payment({ transaction_successful: false }), { watched, asset: REWARD }); + assert.equal(d?.verified, false); +}); + +async function makeWatcher(extra = {}) { + const db = new Database(':memory:'); + await runMigrations(db); + const watcher = createHorizonDepositWatcher({ + db, + horizonUrl: 'https://horizon.test', + accounts: [CUSTODY, OTHER_CUSTODY], + asset: REWARD, + logger: { info() {}, warn() {}, error() {} }, + ...extra, + }); + return { db, watcher }; +} + +test('handleRecord: records a deposit once and de-duplicates replays', async () => { + const seen = []; + const { watcher } = await makeWatcher({ onDeposit: (d) => seen.push(d) }); + + const first = watcher.handleRecord(payment(), CUSTODY); + assert.equal(first.duplicate, false); + const replay = watcher.handleRecord(payment(), CUSTODY); + assert.equal(replay.duplicate, true); + + assert.equal(seen.length, 1, 'onDeposit fires only for new deposits'); + const list = watcher.listDeposits({ account: CUSTODY }); + assert.equal(list.length, 1); + assert.equal(list[0].amount, '250.0000000'); + assert.equal(list[0].verified, true); + assert.equal(watcher.getStatus().depositsRecorded, 1); + assert.equal(watcher.getStatus().duplicates, 1); +}); + +test('handleRecord: ignored records still advance the stream cursor', async () => { + const { db, watcher } = await makeWatcher(); + const outgoing = payment({ paging_token: '2002', from: CUSTODY, to: OPERATOR }); + assert.equal(watcher.handleRecord(outgoing, CUSTODY), null); + + assert.equal(watcher.listDeposits().length, 0); + const row = db + .prepare('SELECT cursor FROM horizon_watcher_cursors WHERE account = ?') + .get(CUSTODY); + assert.equal(row.cursor, '2002'); +}); + +test('getByTxHash: verifies a deposit by transaction hash', async () => { + const { watcher } = await makeWatcher(); + watcher.handleRecord(payment({ transaction_hash: 'abc' }), CUSTODY); + assert.equal(watcher.getByTxHash('abc').length, 1); + assert.equal(watcher.getByTxHash('missing').length, 0); +}); + +function fakeHorizon() { + const streams = []; + const factory = () => ({ + payments: () => ({ + forAccount: (account) => ({ + cursor: (cursor) => ({ + stream: (handlers) => { + const entry = { account, cursor, handlers, closed: false }; + streams.push(entry); + return () => { + entry.closed = true; + }; + }, + }), + }), + }), + }); + return { streams, factory }; +} + +test('start/stop: streams each account from "now", persists cursors and resumes from them', async () => { + const fake = fakeHorizon(); + const { watcher, db } = await makeWatcher({ serverFactory: fake.factory }); + + watcher.start(); + assert.equal(fake.streams.length, 2); + assert.deepEqual( + fake.streams.map((s) => s.cursor), + ['now', 'now'], + ); + + const custody = fake.streams.find((s) => s.account === CUSTODY); + custody.handlers.onmessage(payment({ paging_token: '3003' })); + assert.equal(watcher.listDeposits().length, 1); + assert.equal(watcher.getStatus().accounts.find((a) => a.account === CUSTODY).cursor, '3003'); + + watcher.stop(); + assert.ok(fake.streams.every((s) => s.closed)); + assert.equal(watcher.getStatus().running, false); + + // A fresh watcher on the same database resumes from the saved cursor. + const second = fakeHorizon(); + const resumed = createHorizonDepositWatcher({ + db, + horizonUrl: 'https://horizon.test', + accounts: [CUSTODY], + asset: REWARD, + serverFactory: second.factory, + logger: {}, + }); + resumed.start(); + assert.equal(second.streams[0].cursor, '3003'); + resumed.stop(); +}); + +test('stream errors close the stream and reconnect with backoff', async () => { + const fake = fakeHorizon(); + const { watcher } = await makeWatcher({ + serverFactory: fake.factory, + reconnectBaseMs: 5, + reconnectMaxMs: 20, + }); + + watcher.start(); + const first = fake.streams[0]; + first.handlers.onerror(new Error('boom')); + assert.equal(first.closed, true); + assert.equal(watcher.getStatus().errors, 1); + + await new Promise((resolve) => setTimeout(resolve, 60)); + assert.ok(fake.streams.length > 2, 'a new stream was opened for the failed account'); + watcher.stop(); +}); + +test('a throwing serverFactory is retried instead of crashing the process', async () => { + let calls = 0; + const { watcher } = await makeWatcher({ + serverFactory: () => { + calls++; + throw new Error('bad horizon url'); + }, + reconnectBaseMs: 5, + reconnectMaxMs: 10, + }); + watcher.start(); + await new Promise((resolve) => setTimeout(resolve, 40)); + watcher.stop(); + assert.ok(calls > 2, 'connect was retried'); +}); diff --git a/backend/src/services/openapiGenerator.js b/backend/src/services/openapiGenerator.js new file mode 100644 index 00000000..cfbb599a --- /dev/null +++ b/backend/src/services/openapiGenerator.js @@ -0,0 +1,201 @@ +// @ts-check +import { readFileSync } from 'node:fs'; +import { z } from 'zod'; +import { load as yamlLoad } from 'js-yaml'; +import * as routeSchemas from '../schemas.js'; +import { ADDITIONAL_PATHS } from './openapiRoutes.js'; + +/** + * OpenAPI 3.0 document generator (#1259). + * + * The hand-written `openapi.yaml` stays the source of truth for paths. On top + * of it this module derives component schemas from the Zod request schemas the + * routes actually validate with (`schemas.js`), so documented request bodies + * cannot drift from runtime validation. Hand-written components always win. + */ + +/** @param {string} name e.g. `campaignCreateSchema` -> `CampaignCreate` */ +export function schemaComponentName(name) { + const base = name.replace(/Schema$/, ''); + return base.charAt(0).toUpperCase() + base.slice(1); +} + +/** + * Convert a Zod (v3) schema into an OpenAPI 3.0 schema object. + * Unsupported constructs degrade to an unconstrained schema (`{}`) rather than + * throwing, so documentation generation never breaks the server. + * + * @param {any} schema + * @returns {Record} + */ +export function zodToOpenApi(schema) { + const def = schema?._def; + if (!def) return {}; + + switch (def.typeName) { + case 'ZodString': { + /** @type {Record} */ + const out = { type: 'string' }; + for (const check of def.checks ?? []) { + if (check.kind === 'min') out.minLength = check.value; + else if (check.kind === 'max') out.maxLength = check.value; + else if (check.kind === 'url') out.format = 'uri'; + else if (check.kind === 'email') out.format = 'email'; + else if (check.kind === 'uuid') out.format = 'uuid'; + else if (check.kind === 'datetime') out.format = 'date-time'; + else if (check.kind === 'regex') out.pattern = check.regex.source; + } + return out; + } + case 'ZodNumber': { + /** @type {Record} */ + const out = { type: 'number' }; + for (const check of def.checks ?? []) { + if (check.kind === 'int') out.type = 'integer'; + else if (check.kind === 'min') out.minimum = check.value; + else if (check.kind === 'max') out.maximum = check.value; + } + return out; + } + case 'ZodBoolean': + return { type: 'boolean' }; + case 'ZodLiteral': + return { type: typeof def.value, enum: [def.value] }; + case 'ZodEnum': + return { type: 'string', enum: [...def.values] }; + case 'ZodNativeEnum': + return { enum: Object.values(def.values) }; + case 'ZodArray': { + /** @type {Record} */ + const out = { type: 'array', items: zodToOpenApi(def.type) }; + if (def.minLength) out.minItems = def.minLength.value; + if (def.maxLength) out.maxItems = def.maxLength.value; + return out; + } + case 'ZodObject': { + const shape = typeof def.shape === 'function' ? def.shape() : def.shape; + /** @type {Record} */ + const properties = {}; + /** @type {string[]} */ + const required = []; + for (const [key, value] of Object.entries(shape)) { + properties[key] = zodToOpenApi(value); + if (!isOptional(value)) required.push(key); + } + /** @type {Record} */ + const out = { type: 'object', properties }; + if (required.length) out.required = required; + return out; + } + case 'ZodRecord': + return { type: 'object', additionalProperties: zodToOpenApi(def.valueType) }; + case 'ZodUnion': + return { oneOf: def.options.map(zodToOpenApi) }; + case 'ZodNullable': + return { ...zodToOpenApi(def.innerType), nullable: true }; + case 'ZodOptional': + case 'ZodDefault': + case 'ZodReadonly': + return zodToOpenApi(def.innerType); + case 'ZodEffects': // .refine() / .transform() / .preprocess() + return zodToOpenApi(def.schema); + default: + return {}; + } +} + +/** @param {any} schema */ +function isOptional(schema) { + return typeof schema?.isOptional === 'function' ? schema.isOptional() : false; +} + +/** + * Component schemas derived from every exported Zod schema in `schemas.js`. + * @returns {Record>} + */ +export function generateComponentSchemas(source = routeSchemas) { + /** @type {Record>} */ + const out = {}; + for (const [exportName, value] of Object.entries(source)) { + if (!(value instanceof z.ZodType)) continue; + const converted = zodToOpenApi(value); + if (converted.type !== 'object') continue; // only document request bodies/queries + out[schemaComponentName(exportName)] = { + ...converted, + 'x-generated-from': `schemas.js#${exportName}`, + }; + } + return out; +} + +/** + * OpenAPI 3.1 dropped the `nullable` keyword in favour of JSON Schema type + * unions (`type: ['string', 'null']`). Rewrite in place for 3.1 documents. + * + * @param {any} node + * @returns {any} the same node, normalised + */ +function toNullableTypeUnions(node) { + if (Array.isArray(node)) { + node.forEach(toNullableTypeUnions); + } else if (node && typeof node === 'object') { + if (node.nullable === true && typeof node.type === 'string') node.type = [node.type, 'null']; + if ('nullable' in node && typeof node.nullable === 'boolean') delete node.nullable; + Object.values(node).forEach(toNullableTypeUnions); + } + return node; +} + +/** + * Merge the base spec with generated component schemas. + * + * @param {{ baseSpec: Record, serverUrl?: string, generated?: Record, additionalPaths?: Record }} options + * @returns {Record} a new document; `baseSpec` is not mutated + */ +export function buildOpenApiDocument({ + baseSpec, + serverUrl, + generated = generateComponentSchemas(), + additionalPaths = ADDITIONAL_PATHS, +}) { + const doc = structuredClone(baseSpec ?? {}); + doc.openapi ??= '3.0.3'; + doc.info ??= { title: 'Trivela API', version: '0.0.0' }; + doc.paths ??= {}; + doc.components ??= {}; + doc.components.schemas ??= {}; + + const is31 = String(doc.openapi).startsWith('3.1'); + const adapt = (/** @type {any} */ value) => { + const copy = structuredClone(value); + return is31 ? toNullableTypeUnions(copy) : copy; + }; + + for (const [name, schema] of Object.entries(generated)) { + if (!(name in doc.components.schemas)) doc.components.schemas[name] = adapt(schema); + } + + for (const [path, item] of Object.entries(additionalPaths)) { + doc.paths[path] = { ...adapt(item), ...doc.paths[path] }; // hand-written operations win + } + + if (serverUrl && !(Array.isArray(doc.servers) && doc.servers.length)) { + doc.servers = [{ url: serverUrl }]; + } + return doc; +} + +/** + * Load `backend/openapi.yaml` relative to this module, so the result does not + * depend on the process working directory (repo root vs. `backend/`). + * + * @returns {Record} the parsed spec, or a minimal empty document + */ +export function loadOpenApiSpec() { + try { + const path = new URL('../../openapi.yaml', import.meta.url); + return /** @type {Record} */ (yamlLoad(readFileSync(path, 'utf8'))); + } catch { + return { openapi: '3.0.3', info: { title: 'Trivela API', version: '0.0.0' }, paths: {} }; + } +} diff --git a/backend/src/services/openapiGenerator.test.js b/backend/src/services/openapiGenerator.test.js new file mode 100644 index 00000000..c76d015a --- /dev/null +++ b/backend/src/services/openapiGenerator.test.js @@ -0,0 +1,138 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { z } from 'zod'; +import { + buildOpenApiDocument, + generateComponentSchemas, + schemaComponentName, + zodToOpenApi, +} from './openapiGenerator.js'; + +test('schemaComponentName: strips the Schema suffix and capitalises', () => { + assert.equal(schemaComponentName('campaignCreateSchema'), 'CampaignCreate'); + assert.equal(schemaComponentName('claim'), 'Claim'); +}); + +test('zodToOpenApi: strings, numbers, booleans, enums and constraints', () => { + assert.deepEqual(zodToOpenApi(z.string().min(1).max(5)), { + type: 'string', + minLength: 1, + maxLength: 5, + }); + assert.deepEqual(zodToOpenApi(z.string().url()), { type: 'string', format: 'uri' }); + assert.deepEqual(zodToOpenApi(z.number().int().min(0).max(9)), { + type: 'integer', + minimum: 0, + maximum: 9, + }); + assert.deepEqual(zodToOpenApi(z.boolean()), { type: 'boolean' }); + assert.deepEqual(zodToOpenApi(z.enum(['a', 'b'])), { type: 'string', enum: ['a', 'b'] }); +}); + +test('zodToOpenApi: objects report required vs optional fields', () => { + const schema = z.object({ + name: z.string(), + note: z.string().optional(), + tags: z.array(z.string()).max(3).optional(), + parent: z.string().nullable(), + }); + const out = zodToOpenApi(schema); + assert.equal(out.type, 'object'); + assert.deepEqual(out.required, ['name', 'parent']); + assert.deepEqual(out.properties.tags, { type: 'array', items: { type: 'string' }, maxItems: 3 }); + assert.deepEqual(out.properties.parent, { type: 'string', nullable: true }); +}); + +test('zodToOpenApi: unwraps refinements/defaults and degrades unknown types to {}', () => { + assert.deepEqual(zodToOpenApi(z.string().refine(() => true)), { type: 'string' }); + assert.deepEqual(zodToOpenApi(z.number().default(3)), { type: 'number' }); + assert.deepEqual(zodToOpenApi(z.union([z.string(), z.number()])), { + oneOf: [{ type: 'string' }, { type: 'number' }], + }); + assert.deepEqual(zodToOpenApi(z.date()), {}); + assert.deepEqual(zodToOpenApi(undefined), {}); +}); + +test('generateComponentSchemas: derives request schemas from schemas.js', () => { + const schemas = generateComponentSchemas(); + assert.ok(schemas.CampaignCreate, 'CampaignCreate is generated'); + assert.equal(schemas.CampaignCreate.type, 'object'); + assert.ok(schemas.CampaignCreate.required.includes('name')); + assert.match(schemas.CampaignCreate['x-generated-from'], /campaignCreateSchema/); +}); + +test('buildOpenApiDocument: hand-written components win, base spec is not mutated', () => { + const base = { + openapi: '3.0.3', + info: { title: 'T', version: '1' }, + paths: { '/x': {} }, + components: { schemas: { CampaignCreate: { type: 'object', description: 'hand-written' } } }, + }; + const snapshot = JSON.stringify(base); + + const doc = buildOpenApiDocument({ baseSpec: base, serverUrl: 'https://api.example.test' }); + + assert.equal(doc.components.schemas.CampaignCreate.description, 'hand-written'); + assert.ok(doc.components.schemas.Claim, 'missing components are generated'); + assert.deepEqual(doc.servers, [{ url: 'https://api.example.test' }]); + assert.equal(JSON.stringify(base), snapshot); +}); + +test('buildOpenApiDocument: tolerates an empty base spec', () => { + const doc = buildOpenApiDocument({ baseSpec: undefined }); + assert.equal(doc.openapi, '3.0.3'); + assert.ok( + doc.paths['/api/v1/campaigns/{id}/exports'], + 'still documents the additional endpoints', + ); + assert.ok(Object.keys(doc.components.schemas).length > 0); +}); + +test('buildOpenApiDocument: adds documentation for router-mounted endpoints without overriding the spec', () => { + const doc = buildOpenApiDocument({ + baseSpec: { + openapi: '3.1.0', + info: { title: 'T', version: '1' }, + paths: { '/api/v1/campaigns/{id}/exports': { get: { summary: 'hand-written' } } }, + }, + }); + + assert.ok(doc.paths['/api/v1/admin/fraud/flags'].get, 'fraud flags are documented'); + assert.ok(doc.paths['/api/v1/operator/watcher/status'].get); + const exports = doc.paths['/api/v1/campaigns/{id}/exports']; + assert.equal(exports.get.summary, 'hand-written', 'existing operations are preserved'); + assert.ok(exports.post, 'missing operations on the same path are added'); +}); + +test('every documented additional operation has a summary, tags and responses', async () => { + const { ADDITIONAL_PATHS } = await import('./openapiRoutes.js'); + for (const [path, item] of Object.entries(ADDITIONAL_PATHS)) { + for (const [method, op] of Object.entries(item)) { + assert.ok(op.summary, `${method} ${path} has a summary`); + assert.ok(op.tags?.length, `${method} ${path} has tags`); + assert.ok(op.operationId, `${method} ${path} has an operationId`); + assert.ok(Object.keys(op.responses).length, `${method} ${path} has responses`); + } + } +}); + +test('buildOpenApiDocument: uses type unions instead of `nullable` for OpenAPI 3.1 documents', () => { + const doc31 = buildOpenApiDocument({ + baseSpec: { openapi: '3.1.0', info: { title: 'T', version: '1' }, paths: {} }, + generated: { + Thing: { type: 'object', properties: { note: { type: 'string', nullable: true } } }, + }, + additionalPaths: {}, + }); + assert.deepEqual(doc31.components.schemas.Thing.properties.note.type, ['string', 'null']); + assert.equal('nullable' in doc31.components.schemas.Thing.properties.note, false); + + const doc30 = buildOpenApiDocument({ + baseSpec: { openapi: '3.0.3', info: { title: 'T', version: '1' }, paths: {} }, + generated: { + Thing: { type: 'object', properties: { note: { type: 'string', nullable: true } } }, + }, + additionalPaths: {}, + }); + assert.equal(doc30.components.schemas.Thing.properties.note.nullable, true); +}); diff --git a/backend/src/services/openapiRoutes.js b/backend/src/services/openapiRoutes.js new file mode 100644 index 00000000..63e3f198 --- /dev/null +++ b/backend/src/services/openapiRoutes.js @@ -0,0 +1,301 @@ +// @ts-check +/** + * OpenAPI path items for router-mounted endpoints that `openapi.yaml` does not + * describe yet (#1259). `buildOpenApiDocument` merges these into the served + * document at `/docs/api` without overriding anything hand-written, so the + * checked-in spec (and the SDK generated from it) is left untouched. + */ + +const idParam = { + name: 'id', + in: 'path', + required: true, + description: 'Campaign ID', + schema: { type: 'string' }, +}; +const jobIdParam = { + name: 'jobId', + in: 'path', + required: true, + description: 'Export job ID', + schema: { type: 'string', format: 'uuid' }, +}; + +const errorBody = { + content: { 'application/json': { schema: { $ref: '#/components/schemas/Error' } } }, +}; +const notFound = { description: 'Not found', ...errorBody }; +const badRequest = { description: 'Invalid request', ...errorBody }; +const unauthorized = { description: 'Missing or invalid API key', ...errorBody }; + +const apiKey = [{ ApiKeyAuth: [] }]; +const masterKey = [{ ApiKeyAuth: [] }]; + +const exportJob = { + type: 'object', + properties: { + jobId: { type: 'string', format: 'uuid' }, + campaignId: { type: 'string' }, + format: { type: 'string', enum: ['csv', 'json'] }, + status: { type: 'string', enum: ['queued', 'running', 'completed', 'failed'] }, + from: { type: 'string', nullable: true }, + to: { type: 'string', nullable: true }, + rowCount: { type: 'integer', nullable: true }, + downloadUrl: { type: 'string', format: 'uri', nullable: true }, + error: { type: 'string', nullable: true }, + createdAt: { type: 'string', format: 'date-time' }, + startedAt: { type: 'string', format: 'date-time', nullable: true }, + completedAt: { type: 'string', format: 'date-time', nullable: true }, + statusUrl: { type: 'string' }, + }, +}; + +const fraudFlag = { + type: 'object', + properties: { + id: { type: 'integer' }, + campaignId: { type: 'string' }, + rule: { type: 'string', enum: ['ip_subnet_cluster', 'proxy_exit_node'] }, + subject: { type: 'string', description: 'The IP subnet (/24, /64) or proxy exit node' }, + accountCount: { type: 'integer' }, + accounts: { type: 'array', items: { type: 'string' } }, + status: { type: 'string', enum: ['open', 'dismissed', 'confirmed'] }, + firstSeenAt: { type: 'string', format: 'date-time' }, + lastSeenAt: { type: 'string', format: 'date-time' }, + }, +}; + +const deposit = { + type: 'object', + properties: { + id: { type: 'integer' }, + account: { type: 'string' }, + from: { type: 'string', nullable: true }, + assetCode: { type: 'string' }, + assetIssuer: { type: 'string', nullable: true }, + amount: { type: 'string' }, + txHash: { type: 'string' }, + verified: { type: 'boolean' }, + observedAt: { type: 'string', format: 'date-time' }, + recordedAt: { type: 'string', format: 'date-time' }, + }, +}; + +/** @param {Record} schema @param {string} description */ +const json = (schema, description) => ({ + description, + content: { 'application/json': { schema } }, +}); + +/** @type {Record>} */ +export const ADDITIONAL_PATHS = { + '/api/v1/campaigns/{id}/exports': { + post: { + summary: 'Queue an asynchronous campaign export', + description: + 'Queues a CSV/JSON export that is generated by a background worker, so large campaigns do not time out. Poll the returned `statusUrl` until `status` is `completed`, then download from `downloadUrl`.', + tags: ['Campaigns'], + operationId: 'createCampaignExport', + security: apiKey, + parameters: [idParam], + requestBody: { + required: false, + content: { + 'application/json': { + schema: { + type: 'object', + properties: { + format: { type: 'string', enum: ['csv', 'json'], default: 'csv' }, + from: { type: 'string', format: 'date-time' }, + to: { type: 'string', format: 'date-time' }, + }, + }, + }, + }, + }, + responses: { + 202: json(exportJob, 'Export queued'), + 400: badRequest, + 401: unauthorized, + 404: notFound, + }, + }, + get: { + summary: 'List recent exports for a campaign', + tags: ['Campaigns'], + operationId: 'listCampaignExports', + security: apiKey, + parameters: [ + idParam, + { name: 'limit', in: 'query', schema: { type: 'integer', minimum: 1, maximum: 100 } }, + ], + responses: { + 200: json( + { + type: 'object', + properties: { + data: { type: 'array', items: exportJob }, + total: { type: 'integer' }, + }, + }, + 'Most recent exports, newest first', + ), + 401: unauthorized, + 404: notFound, + }, + }, + }, + '/api/v1/campaigns/{id}/exports/{jobId}': { + get: { + summary: 'Get the status of a campaign export', + tags: ['Campaigns'], + operationId: 'getCampaignExport', + security: apiKey, + parameters: [idParam, jobIdParam], + responses: { 200: json(exportJob, 'Export job'), 401: unauthorized, 404: notFound }, + }, + }, + '/api/v1/admin/fraud/flags': { + get: { + summary: 'List fraud flags', + description: + 'Flags raised when several distinct accounts sign up from one IP subnet or proxy exit node. Requires the master key.', + tags: ['Admin'], + operationId: 'listFraudFlags', + security: masterKey, + parameters: [ + { + name: 'status', + in: 'query', + schema: { type: 'string', enum: ['open', 'dismissed', 'confirmed'] }, + }, + { name: 'campaignId', in: 'query', schema: { type: 'string' } }, + { name: 'limit', in: 'query', schema: { type: 'integer', minimum: 1, maximum: 200 } }, + { name: 'offset', in: 'query', schema: { type: 'integer', minimum: 0 } }, + ], + responses: { + 200: json( + { + type: 'object', + properties: { data: { type: 'array', items: fraudFlag }, total: { type: 'integer' } }, + }, + 'Fraud flags, newest activity first', + ), + 400: badRequest, + 401: unauthorized, + }, + }, + }, + '/api/v1/admin/fraud/flags/{id}': { + patch: { + summary: 'Review a fraud flag', + tags: ['Admin'], + operationId: 'updateFraudFlag', + security: masterKey, + parameters: [{ name: 'id', in: 'path', required: true, schema: { type: 'integer' } }], + requestBody: { + required: true, + content: { + 'application/json': { + schema: { + type: 'object', + required: ['status'], + properties: { status: { type: 'string', enum: ['open', 'dismissed', 'confirmed'] } }, + }, + }, + }, + }, + responses: { + 200: json(fraudFlag, 'Updated flag'), + 400: badRequest, + 401: unauthorized, + 404: notFound, + }, + }, + }, + '/api/v1/operator/deposits': { + get: { + summary: 'List observed operator deposits', + description: + 'Reward-token deposits seen on the watched custody accounts. Requires the master key.', + tags: ['Admin'], + operationId: 'listOperatorDeposits', + security: masterKey, + parameters: [ + { name: 'account', in: 'query', schema: { type: 'string' } }, + { name: 'limit', in: 'query', schema: { type: 'integer', minimum: 1, maximum: 200 } }, + { name: 'beforeId', in: 'query', schema: { type: 'integer' } }, + ], + responses: { + 200: json( + { + type: 'object', + properties: { data: { type: 'array', items: deposit }, total: { type: 'integer' } }, + }, + 'Deposits, newest first', + ), + 401: unauthorized, + }, + }, + }, + '/api/v1/operator/deposits/{txHash}': { + get: { + summary: 'Verify a deposit by transaction hash', + tags: ['Admin'], + operationId: 'getOperatorDeposit', + security: masterKey, + parameters: [{ name: 'txHash', in: 'path', required: true, schema: { type: 'string' } }], + responses: { + 200: json( + { + type: 'object', + properties: { + txHash: { type: 'string' }, + verified: { type: 'boolean' }, + deposits: { type: 'array', items: deposit }, + }, + }, + 'Deposits recorded for the transaction', + ), + 401: unauthorized, + 404: notFound, + }, + }, + }, + '/api/v1/operator/watcher/status': { + get: { + summary: 'Horizon deposit watcher status', + tags: ['Admin'], + operationId: 'getDepositWatcherStatus', + security: masterKey, + responses: { + 200: json( + { + type: 'object', + properties: { + running: { type: 'boolean' }, + depositsRecorded: { type: 'integer' }, + duplicates: { type: 'integer' }, + errors: { type: 'integer' }, + accounts: { + type: 'array', + items: { + type: 'object', + properties: { + account: { type: 'string' }, + connected: { type: 'boolean' }, + reconnectAttempts: { type: 'integer' }, + lastEventAt: { type: 'string', nullable: true }, + cursor: { type: 'string', nullable: true }, + }, + }, + }, + }, + }, + 'Watcher health', + ), + 401: unauthorized, + }, + }, + }, +}; diff --git a/docs/BACKEND_ANALYTICS_AND_GROWTH.md b/docs/BACKEND_ANALYTICS_AND_GROWTH.md new file mode 100644 index 00000000..a7ddfa5d --- /dev/null +++ b/docs/BACKEND_ANALYTICS_AND_GROWTH.md @@ -0,0 +1,74 @@ +# Analytics exports, fraud detection and deposit watching + +Backend features for the growth engine and analytics area. All of them ship with unit and +integration tests under `backend/src`. + +## Interactive API reference — `/docs/api` + +`GET /docs/api` serves Swagger UI and `GET /docs/api/openapi.json` serves the underlying OpenAPI +document. The document is generated at startup from three sources (`services/openapiGenerator.js`): + +1. `backend/openapi.yaml` — the hand-written spec. It always wins. +2. Component schemas derived from the Zod request schemas in `backend/src/schemas.js`, so the + documented request bodies match runtime validation. Generated schemas carry an `x-generated-from` + marker. +3. Path items for router-mounted endpoints not yet in `openapi.yaml` (`services/openapiRoutes.js`). + +The checked-in `openapi.yaml` (and the SDK generated from it) is not modified by this. + +## Asynchronous campaign exports + +Large exports no longer have to finish inside one HTTP request. The synchronous +`GET /api/v1/campaigns/:id/export` endpoint is unchanged; these endpoints queue the work on the +durable job queue (SQLite-backed, retried with backoff, dead-lettered when exhausted): + +| Method | Path | Purpose | +| ------ | -------------------------------------- | -------------------------------- | +| POST | `/api/v1/campaigns/:id/exports` | Queue an export (`202 Accepted`) | +| GET | `/api/v1/campaigns/:id/exports` | Recent exports for the campaign | +| GET | `/api/v1/campaigns/:id/exports/:jobId` | Status and, once done, the URL | + +```bash +curl -X POST "$API/api/v1/campaigns/42/exports" \ + -H "X-API-Key: $KEY" -H "Content-Type: application/json" \ + -d '{"format":"csv","from":"2026-01-01"}' +# -> 202 { "jobId": "…", "status": "queued", "statusUrl": "/api/v1/campaigns/42/exports/…" } + +curl "$API/api/v1/campaigns/42/exports/$JOB_ID" -H "X-API-Key: $KEY" +# -> { "status": "completed", "rowCount": 1280, "downloadUrl": "https://…/exports/campaigns/42/….csv" } +``` + +Status moves `queued → running → completed`, or `failed` (with `error`) when an attempt throws; the +queue retries up to three times and a retry moves the job back to `running`. The file is written +through the configured storage adapter (local, S3 or IPFS). + +## Fraud detection for multi-account signups + +Each referral signup records the network prefix of the request IP (`/24` IPv4, `/64` IPv6; only the +prefix is stored). When the number of **distinct accounts** from one prefix inside the window +reaches `FRAUD_SUBNET_THRESHOLD`, or accounts share an address on the `FRAUD_PROXY_EXIT_LIST`, a +flag is raised. Detection is advisory: it never blocks a signup. + +| Method | Path | Purpose | +| ------ | ------------------------------- | ------------------------------------------- | +| GET | `/api/v1/admin/fraud/flags` | List flags (`status`, `campaignId`, paging) | +| PATCH | `/api/v1/admin/fraud/flags/:id` | Set `open`, `dismissed` or `confirmed` | + +Both require the master key. A dismissed flag reopens automatically if the cluster grows. The +address seen by the server is Express's `req.ip`, so set `trust proxy` appropriately when running +behind a load balancer. + +## Horizon deposit watcher + +When `OPERATOR_DEPOSIT_ACCOUNTS` is set, the server streams payments for each account from Horizon +and records incoming deposits of the reward asset (`REWARD_TOKEN_CODE` / `REWARD_TOKEN_ISSUER`, +default XLM). Deposits are idempotent (keyed by Horizon paging token), stream cursors are persisted +so restarts resume where they stopped, and dropped streams reconnect with exponential backoff. + +| Method | Path | Purpose | +| ------ | ----------------------------------- | ------------------------------------------ | +| GET | `/api/v1/operator/deposits` | Recent deposits (`account`, `limit`) | +| GET | `/api/v1/operator/deposits/:txHash` | Verify that a transaction's deposit landed | +| GET | `/api/v1/operator/watcher/status` | Stream health, cursors and counters | + +These require the master key.