/**
 * Postgres produce job store: the poll/claim/complete half of the produce worker.
 *
 * - Claim: `SELECT * FROM produce_job_claim(lane, stale_after)` — a SECURITY
 *   DEFINER function (040, the 035 submission_box_lookup / 017 invite_link_for_claim
 *   pattern) because the worker polls with NO tenant context while
 *   video_processing_jobs is FORCE ROW LEVEL SECURITY (009 tenant_isolation). The
 *   definer claim is the one statement that may cross tenants: it bumps
 *   `queued → transcoding` inside FOR UPDATE SKIP LOCKED as the lease (no separate
 *   lease column) and requeues rows whose lease has gone stale, then returns the
 *   claimed row WITH its workspace_id.
 * - Every write after the claim happens inside the job's tenant transaction
 *   (`SET LOCAL app.workspace_id`), so those statements stay RLS-bound as sp_app.
 * - A row whose `input_json` or source asset fails re-validation is FAILED by THE
 *   CLAIM ITSELF (never thrown out of `claim`), so an undecodable row can never sit
 *   in `transcoding` forever holding its lease.
 */
import type { Pool, PoolClient } from "pg";
import type { Lane, ProduceJobKind } from "./command-builder";
import { parseProduceRequest } from "./command-builder";
import type { ClaimedProduceJob, ProduceJobStore, ProduceOutput, ProduceSourceAsset } from "./store";

const KINDS: readonly ProduceJobKind[] = ["compose", "mix"];

/**
 * A claimed `transcoding` row older than this is assumed ownerless (the worker
 * died mid-encode) and is requeued by the claim itself. Long enough that no real
 * encode loses its lease to the sweep: the runner reports progress at every stage
 * (claimed/downloaded/encoded/published) and each write bumps `updated_at`, so a
 * live lease is touched far more often than this window.
 */
export const PRODUCE_LEASE_STALE_AFTER = "45 minutes";

interface ClaimRow {
  id: string;
  workspace_id: string;
  kind: string;
  input_json: unknown;
  video_asset_id: string;
}

function uuid(value: unknown, field: string): string {
  const text = typeof value === "string" ? value : undefined;
  if (text === undefined || !/^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/.test(text)) {
    throw new Error(`produce job store: row has an invalid ${field}`);
  }
  return text;
}

function nullableNumber(value: unknown): number | null {
  if (value === null || value === undefined) return null;
  const n = typeof value === "number" ? value : Number(value);
  return Number.isFinite(n) ? n : null;
}

function sourceFromRow(row: Record<string, unknown>): ProduceSourceAsset {
  return {
    id: uuid(row.id, "video_asset id"),
    storageKey: String(row.storage_key ?? ""),
    mimeType: String(row.mime_type ?? ""),
    sizeBytes: Number(row.size_bytes ?? 0),
    width: nullableNumber(row.width),
    height: nullableNumber(row.height),
    durationMs: row.duration_ms === null || row.duration_ms === undefined ? null : Math.round(Number(row.duration_ms)),
    frameRate: nullableNumber(row.frame_rate),
    // Probed "no audio track" (009 stores an empty codec) → a silent bed, so the
    // output always has an audio stream.
    hasAudio: row.audio_codec === null || row.audio_codec === undefined ? null : String(row.audio_codec) !== "",
  };
}

/** The store's own reads are RLS-scoped to the job's workspace (009 tenant_isolation). */
export class PostgresProduceJobStore implements ProduceJobStore {
  constructor(private readonly pool: Pool, private readonly now: () => Date = () => new Date()) {}

  private async tenant<T>(workspaceId: string, fn: (client: PoolClient) => Promise<T>): Promise<T> {
    const client = await this.pool.connect();
    try {
      await client.query("BEGIN");
      // SET LOCAL: the tenant context dies with this transaction, so a pooled
      // client can never leak one workspace's scope into the next borrower.
      await client.query("SELECT set_config('app.workspace_id', $1, true)", [workspaceId]);
      const result = await fn(client);
      await client.query("COMMIT");
      return result;
    } catch (error) {
      await client.query("ROLLBACK").catch(() => undefined);
      throw error;
    } finally {
      client.release();
    }
  }

  async claim(lane: Lane): Promise<ClaimedProduceJob | null> {
    // Lane and the stale-lease window are typed bind parameters of the definer
    // function (040): never interpolated into the SQL text, so the parameter list
    // and the statement can no longer disagree (the 08P01 bind-message bug).
    const claimed = await this.pool.query<ClaimRow>(
      "SELECT id, workspace_id, kind, input_json, video_asset_id FROM produce_job_claim($1, $2::interval)",
      [lane, PRODUCE_LEASE_STALE_AFTER],
    );
    if (claimed.rowCount === 0) return null;
    const row = claimed.rows[0]!;
    const workspaceId = uuid(row.workspace_id, "workspace id");
    const kind = row.kind === "mix" ? "mix" : row.kind === "compose" ? "compose" : null;
    let request = null;
    if (kind !== null) {
      try {
        request = parseProduceRequest(row.input_json, kind);
      } catch {
        request = null;
      }
    }
    if (kind === null || request === null) {
      // A row the worker cannot honour must FAIL here — never strand in `transcoding`.
      // A null return (not a throw) commits the failure instead of rolling it back.
      await this.tenant(workspaceId, (client) =>
        client.query(
          `UPDATE video_processing_jobs
              SET state = 'failed', progress_percent = 100, error = $2, updated_at = now()
            WHERE id = $1 AND state = 'transcoding'`,
          [row.id, kind === null ? "produce job store: unknown kind" : "produce invalid request"],
        ));
      return null;
    }

    return this.tenant<ClaimedProduceJob | null>(workspaceId, async (client) => {
      const assets = await client.query(
        `SELECT id, storage_key, mime_type, size_bytes, width, height, duration_ms, frame_rate, audio_codec
           FROM video_assets WHERE id = $1`,
        [row.video_asset_id],
      );
      if (assets.rowCount === 0) {
        // Same rule as an undecodable payload: fail the row, release the lease.
        await client.query(
          `UPDATE video_processing_jobs
              SET state = 'failed', progress_percent = 100, error = 'produce asset unavailable', updated_at = now()
            WHERE id = $1 AND state = 'transcoding'`,
          [row.id],
        );
        return null;
      }
      return {
        jobId: uuid(row.id, "job id"),
        workspaceId,
        kind,
        request,
        source: sourceFromRow(assets.rows[0] as Record<string, unknown>),
        claimedAt: this.now().toISOString(),
      };
    });
  }

  async progress(jobId: string, percent: number): Promise<void> {
    await this.tenant(await this.workspaceOf(jobId), async (client) => {
      await client.query(
        `UPDATE video_processing_jobs SET progress_percent = $2, updated_at = now()
          WHERE id = $1 AND state = 'transcoding'`,
        [jobId, percent],
      );
    });
  }

  async complete(jobId: string, output: ProduceOutput, derivedFrom: string): Promise<string> {
    const workspaceId = await this.workspaceOf(jobId);
    return this.tenant(workspaceId, async (client) => {
      const inserted = await client.query<{ id: string }>(
        `INSERT INTO video_assets
             (workspace_id, storage_key, mime_type, size_bytes, width, height, duration_ms, frame_rate, codec, audio_codec, derived_from)
         VALUES ($1, $2, 'video/mp4', $3, $4, $5, $6, $7, 'h264', 'aac', $8)
         RETURNING id`,
        [workspaceId, output.storageKey, output.sizeBytes, output.width, output.height, output.durationMs, output.frameRate, derivedFrom],
      );
      const producedAssetId = inserted.rows[0]!.id;
      const updated = await client.query(
        `UPDATE video_processing_jobs
            SET state = 'complete', progress_percent = 100, output_key = $2,
                produced_asset_id = $3, error = NULL, updated_at = now()
          WHERE id = $1 AND state = 'transcoding'
          RETURNING id`,
        [jobId, output.storageKey, producedAssetId],
      );
      if (updated.rowCount === 0) throw new Error(`produce job store: job ${jobId} lost its claim`);
      return producedAssetId;
    });
  }

  async fail(jobId: string, reason: string): Promise<void> {
    await this.tenant(await this.workspaceOf(jobId), async (client) => {
      await client.query(
        `UPDATE video_processing_jobs
            SET state = 'failed', error = $2, updated_at = now()
          WHERE id = $1 AND state IN ('queued', 'transcoding')`,
        [jobId, reason],
      );
    });
  }

  async asset(workspaceId: string, assetId: string): Promise<{ storageKey: string; mimeType: string } | null> {
    return this.tenant(workspaceId, async (client) => {
      const rows = await client.query<{ storage_key: string; mime_type: string }>(
        `SELECT storage_key, mime_type FROM video_assets WHERE id = $1`,
        [assetId],
      );
      if (rows.rowCount === 0) return null;
      return { storageKey: rows.rows[0]!.storage_key, mimeType: rows.rows[0]!.mime_type };
    });
  }

  /**
   * A claimed job's workspace, for the tenant-scoped follow-up writes: the 040
   * definer lookup (012's live_monitor_source_workspace shape), because an
   * ordinary SELECT as sp_app with no tenant context sees no rows at all under
   * FORCE RLS. `job not found` here is exactly what it means: the row is gone
   * (tenant-deleted) or the lease was reclaimed and the job finished elsewhere.
   */
  private async workspaceOf(jobId: string): Promise<string> {
    const rows = await this.pool.query<{ workspace_id: string }>(
      "SELECT produce_job_workspace($1) AS workspace_id",
      [jobId],
    );
    const workspaceId = rows.rows[0]?.workspace_id;
    if (workspaceId === undefined || workspaceId === null) throw new Error(`produce job store: job ${jobId} not found`);
    return uuid(workspaceId, "workspace id");
  }
}

export const PRODUCE_JOB_KINDS = KINDS;