0pt/src/connectors/postgres-query-impl.ts

/**
 * src/connectors/postgres-query-impl.ts
 * ========================================
 * The REAL `QueryImpl` (see postgres-watermark-connector.ts) — a thin,
 * read-only wrapper over `postgres` (postgres.js, already a repo dependency
 * — see src/repositories/db.ts for the sibling pattern this mirrors).
 *
 * Used ONLY by scripts/vehiq-oltp-ingest/run.ts and the opt-in realinfra
 * test. Every connector-level test in tests/connectors/*.test.ts injects a
 * fixture QueryImpl instead — no real Postgres required to prove the query
 * shape, the watermark math, or the narrow-scan property.
 *
 * DOUBLE-ENFORCED READ-ONLY. The connectors above only ever build SELECT
 * statements — there is no code path that builds anything else. This
 * wrapper adds a second, independent check anyway: it refuses to execute
 * any statement whose first keyword is not SELECT. Belt-and-suspenders is
 * appropriate here specifically because the thing on the other end is a
 * separate product's live production OLTP primary, not 0pt's own store.
 */
import postgres from "postgres";
import type { QueryImpl, SourceRow } from "./postgres-watermark-connector";

export interface PostgresQueryImplOptions {
  connectionString: string;
  ssl?: "require" | false;
  /** Statement timeout (ms), applied per-connection so a slow query cannot pin the source's OLTP primary. Default 5000. */
  statementTimeoutMs?: number;
}

class NonSelectStatementError extends Error {
  constructor(sql: string) {
    super(`refusing to execute a non-SELECT statement against a read-only source connector: ${sql.slice(0, 120)}`);
    this.name = "NonSelectStatementError";
  }
}

export function assertReadOnlyStatement(sql: string): void {
  const firstWord = sql.trim().split(/\s+/, 1)[0]?.toUpperCase();
  if (firstWord !== "SELECT") throw new NonSelectStatementError(sql);
}

/**
 * Builds a `QueryImpl` bound to one `postgres` client. The caller owns the
 * client's lifecycle (open once per orchestration run, `sql.end()` when
 * done) — this function does not memoize or reuse connections across calls,
 * unlike `src/repositories/db.ts`'s deliberate memoization for 0pt's own
 * store, because this one instance is scoped to a single ingestion run
 * against an EXTERNAL production database and should not outlive it.
 */
export function makePostgresQueryImpl(opts: PostgresQueryImplOptions): { query: QueryImpl; sql: ReturnType<typeof postgres>; close: () => Promise<void> } {
  const sql = postgres(opts.connectionString, {
    prepare: false,
    idle_timeout: 20,
    max: 2,
    ssl: opts.ssl ?? false,
    connection: {
      statement_timeout: opts.statementTimeoutMs ?? 5_000,
      // A narrow, self-identifying application_name — same posture as
      // http-json-connector.ts's fixed User-Agent: no impersonation, easy
      // to spot in the source's own pg_stat_activity.
      application_name: "0pt-vehiq-oltp-extractor",
    },
  });
  const query: QueryImpl = async (text: string, params: unknown[]): Promise<SourceRow[]> => {
    assertReadOnlyStatement(text);
    const rows = await sql.unsafe(text, params as never[]);
    return rows.map((r) => ({ ...r }));
  };
  return { query, sql, close: () => sql.end({ timeout: 5 }) };
}