edr-platform/integration/src/client.ts

/**
 * Plumbing for the integration suite: HTTP against this worker's IN-PROCESS
 * freight API and its containerized payment API, SQL against the worker's own
 * throwaway database, and its own gateway mock's control plane.
 *
 * Everything here is shard-scoped — see app.ts for the topology. supertest takes
 * an `http.Server` exactly where it takes a base URL, so pointing the suite at
 * the in-process app is a one-token change at each call site.
 *
 * This is the Cypress-free port of e2e/freight/cypress/e2e/flows/import-utils.ts —
 * same request sequences, same SQL, `pg.Pool` instead of `cy.task`.
 */
import request from "supertest";
import { Pool, type QueryResultRow } from "pg";

import { DB_URL, GATEWAY, PAYMENT_API, SHARD, freightServer, tag } from "./app";

export { GATEWAY, PAYMENT_API, SHARD, freightServer };

export const SERVICE_TOKEN = process.env.IT_SERVICE_TOKEN ?? "e2e-service-token";

// ---------------------------------------------------------------------------
// users (e2e/freight/cypress/fixtures/seed-users.sql + users.json)
// ---------------------------------------------------------------------------

export const customerA = "user@gmail.com";
export const customerB = "user2@gmail.com";
export const opsStaff = "operation@edr.local";
export const chief = "chief@edr.local";
/** isSuperAdmin bypasses assertFreightPermission — used for the GL/clearance steps. */
export const superAdmin = "superadmin@tria.com";

const STAFF_PASSWORD = "password@tria";
const CUSTOMER_PASSWORD = "12345678";

// ---------------------------------------------------------------------------
// database
// ---------------------------------------------------------------------------

const pool = new Pool({ connectionString: DB_URL, max: 12 });

export async function db<T extends QueryResultRow = Record<string, unknown>>(
  sql: string,
  params: unknown[] = [],
): Promise<T[]> {
  const res = await pool.query<T>(sql, params);
  return res.rows;
}

/**
 * No-op. The pool is per-WORKER, not per-file: with `isolate: false` the module
 * registry survives across every spec a worker runs, so the first `afterAll` to
 * call this would break every later file on that shard. The pool dies with the
 * worker process.
 */
export async function closeDb(): Promise<void> {}

/**
 * Poll a query until `check` passes. Async settlement here is broker-driven and
 * tick-driven, so the interval is pure overshoot: at the old 2000ms every wait
 * in the suite finished up to two seconds after the state it wanted was already
 * in the database, several hundred times over. The ceiling
 * (`attempts × intervalMs`) is unchanged.
 */
const POLL_INTERVAL_MS = Number(process.env.IT_POLL_INTERVAL_MS ?? 250);

export async function poll<T extends QueryResultRow = Record<string, unknown>>(
  label: string,
  sql: string,
  params: unknown[],
  check: (row: T | undefined) => boolean,
  { attempts = 40, intervalMs = 2000 } = {},
): Promise<T> {
  // Callers express patience as attempts × their own interval; keep that
  // deadline and just sample it finely.
  const deadlineMs = attempts * intervalMs;
  const tries = Math.ceil(deadlineMs / POLL_INTERVAL_MS);

  let last: T | undefined;
  for (let i = 0; i < tries; i++) {
    last = (await db<T>(sql, params))[0];
    if (check(last)) return last as T;
    await sleep(POLL_INTERVAL_MS);
  }
  throw new Error(
    tag(
      `timed out waiting for ${label} after ${Math.round(deadlineMs / 1000)}s\n` +
        `  sql: ${sql.replace(/\s+/g, " ").trim()}\n` +
        `  params: ${JSON.stringify(params)}\n` +
        `  last row: ${JSON.stringify(last)}`,
    ),
  );
}

export const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

// ---------------------------------------------------------------------------
// auth
// ---------------------------------------------------------------------------

const tokens = new Map<string, string>();

/**
 * Bearer token for a seeded account. The login audience is not cosmetic:
 * `user_type='individual'` accounts are rejected on the backoffice audience and
 * vice versa (EDRFREIGHT-415), so it is derived from the address.
 */
export async function tokenFor(email: string): Promise<string> {
  const cached = tokens.get(email);
  if (cached) return cached;

  const portal = email.endsWith("@gmail.com");
  const res = await request(await freightServer())
    .post("/api/auth/login")
    .set("x-client-app", portal ? "portal" : "backoffice")
    .send({ email, password: portal ? CUSTOMER_PASSWORD : STAFF_PASSWORD });

  // The login response is flattened by the app (no `.data` envelope) and 201s.
  const token = res.body?.token ?? res.body?.data?.token;
  if (!token) {
    throw new Error(`login failed for ${email}: ${res.status} ${JSON.stringify(res.body)}`);
  }
  tokens.set(email, token);
  return token;
}

/** Login without caching, so audience-rejection can be asserted. */
export async function login(email: string, password: string, app: "portal" | "backoffice") {
  return request(await freightServer())
    .post("/api/auth/login")
    .set("x-client-app", app)
    .send({ email, password });
}

// ---------------------------------------------------------------------------
// freight API
// ---------------------------------------------------------------------------

export type Method = "get" | "post" | "patch" | "delete";

/** Authenticated call to the freight API as `email`. Never throws on 4xx/5xx. */
export async function api(
  email: string,
  method: Method,
  path: string,
  body?: unknown,
): Promise<request.Response> {
  const token = await tokenFor(email);
  const req = request(await freightServer())
    [method](path)
    .set("Authorization", `Bearer ${token}`);
  return method === "get" || method === "delete" ? req.send() : req.send(body ?? {});
}

/** Same, but fails loudly on a non-2xx — for arrange steps that must succeed. */
export async function apiOk(
  email: string,
  method: Method,
  path: string,
  body?: unknown,
): Promise<request.Response> {
  const res = await api(email, method, path, body);
  if (res.status < 200 || res.status > 201) {
    throw new Error(
      `${method.toUpperCase()} ${path} as ${email} → ${res.status}: ${JSON.stringify(res.body)}`,
    );
  }
  return res;
}

/** Multipart upload (clearance documents). supertest handles the encoding. */
export async function upload(
  email: string,
  path: string,
  filePath: string,
  field = "files",
  fields: Record<string, string> = {},
): Promise<request.Response> {
  const token = await tokenFor(email);
  const req = request(await freightServer())
    .post(path)
    .set("Authorization", `Bearer ${token}`);
  for (const [k, v] of Object.entries(fields)) req.field(k, v);
  return req.attach(field, filePath);
}

// ---------------------------------------------------------------------------
// payment API (service-to-service surface)
// ---------------------------------------------------------------------------

export function payment(method: Method, path: string, body?: unknown) {
  const req = request(PAYMENT_API)[method](path).set("x-service-token", SERVICE_TOKEN);
  return method === "get" || method === "delete" ? req.send() : req.send(body ?? {});
}

/** Payment-side rows. The payment service owns its own schema in the same DB. */
export function paymentDb<T extends QueryResultRow = Record<string, unknown>>(
  sql: string,
  params: unknown[] = [],
) {
  return db<T>(sql, params);
}

export interface IntentRow extends QueryResultRow {
  id: string;
  status: string;
  provider: string;
  merchant_order_id: string;
  amount_minor: string;
  currency: string;
  reference_id: string;
  provider_txn_id: string | null;
  expires_at: string | null;
}

export function intentByMerchantOrderId(merchantOrderId: string) {
  return db<IntentRow>(
    `SELECT * FROM edr_payment.payment_intent WHERE merchant_order_id = $1`,
    [merchantOrderId],
  );
}

// ---------------------------------------------------------------------------
// gateway mock control plane
// ---------------------------------------------------------------------------

export const gateway = {
  reset: () => request(GATEWAY).post("/__control/reset").send({}),

  /** Force a provider's next `times` calls (or all of them) into a mode. */
  mode: (provider: string, mode: "ok" | "fail" | "timeout" | "pending" | "paid", times?: number) =>
    request(GATEWAY).post(`/__control/provider/${provider}`).send({ mode, times }),

  /** Mark the order settled at the gateway WITHOUT a callback (polling path). */
  settle: (merchantOrderId: string) =>
    request(GATEWAY).post("/__control/settle").send({ merchantOrderId }),

  /** Fire a signed provider callback at the payment API. */
  webhook: (opts: {
    merchantOrderId: string;
    provider?: string;
    status?: string;
    eventId?: string;
    transactionId?: string;
    signature?: "bad";
  }) => request(GATEWAY).post("/__control/webhook").send(opts),

  calls: () => request(GATEWAY).get("/__control/calls").send(),
};