/**
 * A1 — Postgres adapter for the repository ports (store/ports.ts). Each store maps its port
 * methods to PARAMETERIZED SQL over the injected `query` wrapper (db/pool.ts) — values ALWAYS
 * travel as bound params, never string-interpolated (SQLi structurally impossible, §7).
 *
 * This adapter MUST reproduce store/memory.ts observable semantics EXACTLY:
 *   - INV8 versioning: `swapStatus` appends a `*_status_versions` row AND bumps the single-row
 *     `status`/`status_version` pointer ATOMICALLY. Postgres has real transactions, but we get
 *     the same all-or-nothing with a single writable-CTE statement (no explicit tx / PoolClient).
 *   - `insert` duplicates throw (PK / UNIQUE violation, pg code 23505 → `throw new Error`).
 *   - `casRedeem` is a single-winner CAS of `redeemed_at` from null.
 *   - `RouteStore` fails closed on TTL: `expires_at <= now()` ⇒ absent (INV7), lazily DELETEd.
 *   - metering/audit are append-only with inclusive `[from,to]` time-window queries.
 *
 * Timestamps: DB columns are `timestamptz` (node-pg returns them as JS `Date`); records use ISO
 * strings. `toIso` normalizes Date|string → ISO so returned records match memory.ts string shapes.
 */
import type {
  AccountRecord,
  AccountStatus,
  AccountStatusVersionRow,
  HostRecord,
  HostStatus,
  HostStatusVersionRow,
  MeteringSampleRow,
  PairingCodeRecord,
  PlanTier,
  RouteEntry,
  SessionRecord,
} from '../model/records.js'
import type { QueryFn } from '../db/pool.js'
import type {
  AccountStore,
  AuditRow,
  AuditStore,
  CasOutcome,
  HostStore,
  MeteringStore,
  NodeRow,
  NodeStatus,
  NodeStore,
  PairingRow,
  PairingStore,
  RouteStore,
  SessionStore,
  Stores,
  SubdomainStore,
} from './ports.js'

// ---- helpers ----------------------------------------------------------------------------------

const toIso = (v: Date | string): string => (v instanceof Date ? v : new Date(v)).toISOString()
const toIsoN = (v: Date | string | null): string | null => (v === null ? null : toIso(v))

/** pg unique/PK violation. */
function isUniqueViolation(err: unknown): err is { code: string; constraint?: string } {
  return typeof err === 'object' && err !== null && (err as { code?: unknown }).code === '23505'
}

// ---- raw DB row shapes ------------------------------------------------------------------------

interface AccountRow {
  account_id: string
  plan: PlanTier
  created_at: Date | string
  status: AccountStatus
}
interface HostRow {
  host_id: string
  account_id: string
  subdomain: string
  agent_pubkey: Buffer
  enroll_fpr: string
  status: HostStatus
  last_seen: Date | string
  created_at: Date | string
  revoked_at: Date | string | null
}
interface SessionRow {
  session_id: string
  host_id: string
  account_id: string
  created_at: Date | string
  last_attach_at: Date | string
}
interface AccountVersionRow {
  account_id: string
  version: number
  status: AccountStatus
  changed_at: Date | string
  changed_by: string
}
interface HostVersionRow {
  host_id: string
  version: number
  status: HostStatus
  revoked_at: Date | string | null
  changed_at: Date | string
  changed_by: string
}

function mapAccount(r: AccountRow): AccountRecord {
  return { accountId: r.account_id, plan: r.plan, createdAt: toIso(r.created_at), status: r.status }
}
function mapHost(r: HostRow): HostRecord {
  return {
    hostId: r.host_id,
    accountId: r.account_id,
    subdomain: r.subdomain,
    agentPubkey: new Uint8Array(r.agent_pubkey),
    enrollFpr: r.enroll_fpr,
    status: r.status,
    lastSeen: toIso(r.last_seen),
    createdAt: toIso(r.created_at),
    revokedAt: toIsoN(r.revoked_at),
  }
}

// ---- AccountStore -----------------------------------------------------------------------------

function pgAccountStore(query: QueryFn): AccountStore {
  return {
    async insert(rec) {
      // Writable CTE: create the row AND its INV8 version-1 companion in ONE statement.
      // version-1 changed_at = created_at, changed_by = 'system' (matches memory.ts).
      try {
        await query(
          `WITH ins AS (
             INSERT INTO accounts (account_id, plan, created_at, status, status_version)
             VALUES ($1, $2, $3, $4, 1)
             RETURNING account_id, created_at, status
           )
           INSERT INTO account_status_versions (account_id, version, status, changed_at, changed_by)
           SELECT account_id, 1, status, created_at, 'system' FROM ins`,
          [rec.accountId, rec.plan, rec.createdAt, rec.status],
        )
      } catch (err) {
        if (isUniqueViolation(err)) throw new Error('duplicate accountId')
        throw err
      }
    },
    async get(accountId) {
      const rows = await query<AccountRow>(
        `SELECT account_id, plan, created_at, status FROM accounts WHERE account_id = $1`,
        [accountId],
      )
      const row = rows[0]
      return row === undefined ? null : mapAccount(row)
    },
    async swapStatus(accountId, status, changedBy) {
      // Atomic INV8: bump pointer + append version row in one writable-CTE statement.
      const rows = await query<AccountRow>(
        `WITH upd AS (
           UPDATE accounts SET status = $2, status_version = status_version + 1
           WHERE account_id = $1
           RETURNING account_id, plan, created_at, status, status_version
         ),
         ins AS (
           INSERT INTO account_status_versions (account_id, version, status, changed_by)
           SELECT account_id, status_version, status, $3 FROM upd
         )
         SELECT account_id, plan, created_at, status FROM upd`,
        [accountId, status, changedBy],
      )
      const row = rows[0]
      if (row === undefined) throw new Error('account not found')
      return mapAccount(row)
    },
    async versions(accountId) {
      const rows = await query<AccountVersionRow>(
        `SELECT account_id, version, status, changed_at, changed_by
         FROM account_status_versions WHERE account_id = $1 ORDER BY version ASC`,
        [accountId],
      )
      return rows.map(
        (r): AccountStatusVersionRow => ({
          accountId: r.account_id,
          version: r.version,
          status: r.status,
          changedAt: toIso(r.changed_at),
          changedBy: r.changed_by,
        }),
      )
    },
  }
}

// ---- HostStore --------------------------------------------------------------------------------

function pgHostStore(query: QueryFn): HostStore {
  return {
    async insert(rec) {
      try {
        await query(
          `WITH ins AS (
             INSERT INTO hosts
               (host_id, account_id, subdomain, agent_pubkey, enroll_fpr, status, last_seen, created_at, revoked_at, status_version)
             VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 1)
             RETURNING host_id, created_at, status, revoked_at
           )
           INSERT INTO host_status_versions (host_id, version, status, revoked_at, changed_at, changed_by)
           SELECT host_id, 1, status, revoked_at, created_at, 'system' FROM ins`,
          [
            rec.hostId,
            rec.accountId,
            rec.subdomain,
            Buffer.from(rec.agentPubkey),
            rec.enrollFpr,
            rec.status,
            rec.lastSeen,
            rec.createdAt,
            rec.revokedAt,
          ],
        )
      } catch (err) {
        if (isUniqueViolation(err)) {
          // subdomain UNIQUE is the subdomain single-winner substrate; PK is host_id.
          throw new Error(err.constraint?.includes('subdomain') ? 'duplicate subdomain' : 'duplicate hostId')
        }
        throw err
      }
    },
    async get(hostId) {
      const rows = await query<HostRow>(`SELECT * FROM hosts WHERE host_id = $1`, [hostId])
      const row = rows[0]
      return row === undefined ? null : mapHost(row)
    },
    async getBySubdomain(subdomain) {
      const rows = await query<HostRow>(`SELECT * FROM hosts WHERE subdomain = $1`, [subdomain])
      const row = rows[0]
      return row === undefined ? null : mapHost(row)
    },
    async listByAccount(accountId) {
      const rows = await query<HostRow>(`SELECT * FROM hosts WHERE account_id = $1 ORDER BY created_at ASC`, [
        accountId,
      ])
      return rows.map(mapHost)
    },
    async swapStatus(hostId, status, revokedAt, changedBy) {
      const rows = await query<HostRow>(
        `WITH upd AS (
           UPDATE hosts SET status = $2, revoked_at = $3, status_version = status_version + 1
           WHERE host_id = $1
           RETURNING host_id, account_id, subdomain, agent_pubkey, enroll_fpr, status, last_seen, created_at, revoked_at, status_version
         ),
         ins AS (
           INSERT INTO host_status_versions (host_id, version, status, revoked_at, changed_by)
           SELECT host_id, status_version, status, revoked_at, $4 FROM upd
         )
         SELECT host_id, account_id, subdomain, agent_pubkey, enroll_fpr, status, last_seen, created_at, revoked_at FROM upd`,
        [hostId, status, revokedAt, changedBy],
      )
      const row = rows[0]
      if (row === undefined) throw new Error('host not found')
      return mapHost(row)
    },
    async touchLastSeen(hostId, ts) {
      // Advisory cache — NO version row (INV7). Silent no-op if absent (matches memory.ts).
      await query(`UPDATE hosts SET last_seen = $2 WHERE host_id = $1`, [hostId, ts])
    },
    async versions(hostId) {
      const rows = await query<HostVersionRow>(
        `SELECT host_id, version, status, revoked_at, changed_at, changed_by
         FROM host_status_versions WHERE host_id = $1 ORDER BY version ASC`,
        [hostId],
      )
      return rows.map(
        (r): HostStatusVersionRow => ({
          hostId: r.host_id,
          version: r.version,
          status: r.status,
          revokedAt: toIsoN(r.revoked_at),
          changedAt: toIso(r.changed_at),
          changedBy: r.changed_by,
        }),
      )
    },
  }
}

// ---- SessionStore -----------------------------------------------------------------------------

function pgSessionStore(query: QueryFn): SessionStore {
  return {
    async insert(rec) {
      try {
        await query(
          `INSERT INTO sessions (session_id, host_id, account_id, created_at, last_attach_at)
           VALUES ($1, $2, $3, $4, $5)`,
          [rec.sessionId, rec.hostId, rec.accountId, rec.createdAt, rec.lastAttachAt],
        )
      } catch (err) {
        if (isUniqueViolation(err)) throw new Error('duplicate sessionId')
        throw err
      }
    },
    async get(sessionId) {
      const rows = await query<SessionRow>(`SELECT * FROM sessions WHERE session_id = $1`, [sessionId])
      const row = rows[0]
      if (row === undefined) return null
      return {
        sessionId: row.session_id,
        hostId: row.host_id,
        accountId: row.account_id,
        createdAt: toIso(row.created_at),
        lastAttachAt: toIso(row.last_attach_at),
      }
    },
  }
}

// ---- SubdomainStore ---------------------------------------------------------------------------

/**
 * No separate reservations table exists in the schema — the `hosts.subdomain UNIQUE` constraint
 * is the single-winner substrate (the true winner is decided at host insert). So `reserve`/`isTaken`
 * just report whether a host already owns the label (matches memory.ts intent). Two concurrent
 * `reserve()`s can both see "free" before any host row exists; that is fine, because the actual
 * single-winner is enforced when the losing insert hits the UNIQUE violation (INV1).
 */
function pgSubdomainStore(query: QueryFn): SubdomainStore {
  const taken = async (subdomain: string): Promise<boolean> => {
    const rows = await query<{ one: number }>(`SELECT 1 AS one FROM hosts WHERE subdomain = $1 LIMIT 1`, [
      subdomain,
    ])
    return rows.length > 0
  }
  return {
    async reserve(subdomain) {
      return !(await taken(subdomain))
    },
    async isTaken(subdomain) {
      return taken(subdomain)
    },
  }
}

// ---- PairingStore -----------------------------------------------------------------------------

interface PairingCodeRow {
  code_hash: string
  account_id: string
  expires_at: Date | string
  redeemed_at: Date | string | null
  redeem_attempts: number
}

function pgPairingStore(query: QueryFn): PairingStore {
  return {
    async insert(rec) {
      try {
        await query(
          `INSERT INTO pairing_codes (code_hash, account_id, expires_at, redeemed_at, redeem_attempts)
           VALUES ($1, $2, $3, $4, 0)`,
          [rec.codeHash, rec.accountId, rec.expiresAt, rec.redeemedAt],
        )
      } catch (err) {
        if (isUniqueViolation(err)) throw new Error('duplicate code_hash')
        throw err
      }
    },
    async get(codeHash): Promise<PairingRow | null> {
      const rows = await query<PairingCodeRow>(`SELECT * FROM pairing_codes WHERE code_hash = $1`, [codeHash])
      const row = rows[0]
      if (row === undefined) return null
      const record: PairingCodeRecord = {
        codeHash: row.code_hash,
        accountId: row.account_id,
        expiresAt: toIso(row.expires_at),
        redeemedAt: toIsoN(row.redeemed_at),
      }
      return { record, redeemAttempts: row.redeem_attempts }
    },
    async casRedeem(codeHash, nowIso): Promise<CasOutcome> {
      // Single-winner CAS: only the row with redeemed_at IS NULL is updated; exactly one wins.
      const won = await query<{ code_hash: string }>(
        `UPDATE pairing_codes SET redeemed_at = $2
         WHERE code_hash = $1 AND redeemed_at IS NULL
         RETURNING code_hash`,
        [codeHash, nowIso],
      )
      if (won.length === 1) return 'ok'
      // 0 rows: either already redeemed or unknown — distinguish by existence.
      const exists = await query<{ one: number }>(`SELECT 1 AS one FROM pairing_codes WHERE code_hash = $1`, [
        codeHash,
      ])
      return exists.length > 0 ? 'already_redeemed' : 'unknown'
    },
    async registerFailure(codeHash) {
      const rows = await query<{ redeem_attempts: number }>(
        `UPDATE pairing_codes SET redeem_attempts = redeem_attempts + 1
         WHERE code_hash = $1
         RETURNING redeem_attempts`,
        [codeHash],
      )
      const row = rows[0]
      return row === undefined ? 0 : row.redeem_attempts // 0 for unknown code (matches memory.ts)
    },
  }
}

// ---- RouteStore -------------------------------------------------------------------------------

interface RouteRow {
  host_id: string
  relay_node_id: string
  updated_at: Date | string
}

function pgRouteStore(query: QueryFn): RouteStore {
  const mapEntry = (r: RouteRow): RouteEntry => ({ relayNodeId: r.relay_node_id, updatedAt: toIso(r.updated_at) })
  return {
    async set(hostId, entry, ttlSec) {
      // Postgres has no per-key TTL: store an explicit expires_at = now() + ttl (server clock).
      await query(
        `INSERT INTO routes (host_id, relay_node_id, updated_at, expires_at)
         VALUES ($1, $2, $3, now() + make_interval(secs => $4))
         ON CONFLICT (host_id) DO UPDATE SET
           relay_node_id = EXCLUDED.relay_node_id,
           updated_at    = EXCLUDED.updated_at,
           expires_at    = EXCLUDED.expires_at`,
        [hostId, entry.relayNodeId, entry.updatedAt, ttlSec],
      )
    },
    async refreshTtl(hostId, ttlSec) {
      // Refresh only a still-live key (fail closed on already-expired/absent, INV7).
      const rows = await query<{ host_id: string }>(
        `UPDATE routes SET expires_at = now() + make_interval(secs => $2)
         WHERE host_id = $1 AND expires_at > now()
         RETURNING host_id`,
        [hostId, ttlSec],
      )
      return rows.length === 1
    },
    async get(hostId) {
      // Lazily reap the expired row, then a surviving row is guaranteed live (INV7 fail-closed).
      await query(`DELETE FROM routes WHERE host_id = $1 AND expires_at <= now()`, [hostId])
      const rows = await query<RouteRow>(
        `SELECT host_id, relay_node_id, updated_at FROM routes WHERE host_id = $1`,
        [hostId],
      )
      const row = rows[0]
      return row === undefined ? null : mapEntry(row)
    },
    async delete(hostId) {
      await query(`DELETE FROM routes WHERE host_id = $1`, [hostId])
    },
    async listByNode(nodeId) {
      await query(`DELETE FROM routes WHERE expires_at <= now()`, [])
      const rows = await query<RouteRow>(
        `SELECT host_id, relay_node_id, updated_at FROM routes
         WHERE relay_node_id = $1 AND expires_at > now() ORDER BY host_id ASC`,
        [nodeId],
      )
      return rows.map((r) => ({ hostId: r.host_id, entry: mapEntry(r) }))
    },
  }
}

// ---- NodeStore --------------------------------------------------------------------------------

interface NodeDbRow {
  node_id: string
  addr: string
  status: NodeStatus
  last_seen: Date | string
}

function pgNodeStore(query: QueryFn): NodeStore {
  return {
    async upsert(row) {
      await query(
        `INSERT INTO relay_nodes (node_id, addr, status, last_seen)
         VALUES ($1, $2, $3, $4)
         ON CONFLICT (node_id) DO UPDATE SET
           addr = EXCLUDED.addr, status = EXCLUDED.status, last_seen = EXCLUDED.last_seen`,
        [row.nodeId, row.addr, row.status, row.lastSeen],
      )
    },
    async get(nodeId) {
      const rows = await query<NodeDbRow>(`SELECT * FROM relay_nodes WHERE node_id = $1`, [nodeId])
      const row = rows[0]
      if (row === undefined) return null
      return { nodeId: row.node_id, addr: row.addr, status: row.status, lastSeen: toIso(row.last_seen) }
    },
    async setStatus(nodeId, status) {
      const rows = await query<{ node_id: string }>(
        `UPDATE relay_nodes SET status = $2 WHERE node_id = $1 RETURNING node_id`,
        [nodeId, status],
      )
      if (rows.length === 0) throw new Error('node not found')
    },
    async touch(nodeId, ts) {
      // Silent no-op if absent (matches memory.ts).
      await query(`UPDATE relay_nodes SET last_seen = $2 WHERE node_id = $1`, [nodeId, ts])
    },
    async delete(nodeId) {
      await query(`DELETE FROM relay_nodes WHERE node_id = $1`, [nodeId])
    },
  }
}

// ---- MeteringStore ----------------------------------------------------------------------------

interface MeteringDbRow {
  host_id: string
  account_id: string
  node_id: string
  concurrent_viewers: number
  sampled_at: Date | string
}

function pgMeteringStore(query: QueryFn): MeteringStore {
  return {
    async append(row) {
      await query(
        `INSERT INTO metering_samples (host_id, account_id, node_id, concurrent_viewers, sampled_at)
         VALUES ($1, $2, $3, $4, $5)`,
        [row.hostId, row.accountId, row.nodeId, row.concurrentViewers, row.sampledAt],
      )
    },
    async query(accountId, fromIso, toIso2) {
      const rows = await query<MeteringDbRow>(
        `SELECT host_id, account_id, node_id, concurrent_viewers, sampled_at
         FROM metering_samples
         WHERE account_id = $1 AND sampled_at >= $2 AND sampled_at <= $3
         ORDER BY id ASC`,
        [accountId, fromIso, toIso2],
      )
      return rows.map(
        (r): MeteringSampleRow => ({
          hostId: r.host_id,
          accountId: r.account_id,
          nodeId: r.node_id,
          concurrentViewers: r.concurrent_viewers,
          sampledAt: toIso(r.sampled_at),
        }),
      )
    },
  }
}

// ---- AuditStore -------------------------------------------------------------------------------

interface AuditDbRow {
  action: string
  principal_id: string
  account_id: string
  host_id: string | null
  ts: Date | string
  meta: Record<string, string>
}

function pgAuditStore(query: QueryFn): AuditStore {
  return {
    async append(row) {
      await query(
        `INSERT INTO audit_log (action, principal_id, account_id, host_id, ts, meta)
         VALUES ($1, $2, $3, $4, $5, $6::jsonb)`,
        [row.action, row.principalId, row.accountId, row.hostId, row.ts, JSON.stringify(row.meta)],
      )
    },
    async query(accountId, fromIso, toIso2) {
      const rows = await query<AuditDbRow>(
        `SELECT action, principal_id, account_id, host_id, ts, meta
         FROM audit_log
         WHERE account_id = $1 AND ts >= $2 AND ts <= $3
         ORDER BY id ASC`,
        [accountId, fromIso, toIso2],
      )
      return rows.map(
        (r): AuditRow => ({
          action: r.action,
          principalId: r.principal_id,
          accountId: r.account_id,
          hostId: r.host_id,
          ts: toIso(r.ts),
          meta: r.meta,
        }),
      )
    },
  }
}

// ---- assembly ---------------------------------------------------------------------------------

/** Build a full Postgres-backed store set (all ports mapped to parameterized SQL over `query`). */
export function createPgStores(query: QueryFn): Stores {
  return {
    accounts: pgAccountStore(query),
    hosts: pgHostStore(query),
    sessions: pgSessionStore(query),
    subdomains: pgSubdomainStore(query),
    pairing: pgPairingStore(query),
    routes: pgRouteStore(query),
    nodes: pgNodeStore(query),
    metering: pgMeteringStore(query),
    audit: pgAuditStore(query),
  }
}