-- 040: produce-worker job claim — the queue half of 039's producer feature.
--
-- WHY A SECURITY DEFINER CLAIM: video_processing_jobs is FORCE ROW LEVEL SECURITY
-- with a tenant_isolation policy (009), but the produce worker is a lane-global
-- poller with NO tenant context: it must see queued rows in EVERY workspace, then
-- work each claimed row inside that workspace's scope. 006/012/017/035 established
-- the pattern for exactly this shape — a narrow definer-rights claim that
-- discovers the tenant and nothing else; the caller enters the tenant context
-- afterwards and every subsequent statement stays RLS-bound as sp_app.
--
-- The same migration also fixes the production bug this claim surfaced
-- (2026-10 produce worker logging a store error every ~2 s): the store's inline
-- claim interpolated its lane predicate into the SQL text while still passing a
-- bind parameter for it, so node-postgres sent `bind message supplies 1
-- parameters, but prepared statement "" requires 0` (08P01) on EVERY poll. A
-- stored function with a real $1 takes the driver out of the hand-built-SQL
-- business entirely; lane becomes a typed parameter that cannot desynchronise.
--
-- Leasing: the `queued -> transcoding` bump inside FOR UPDATE SKIP LOCKED is the
-- lease (no separate lease column). A worker that dies mid-encode leaves its row
-- in `transcoding` forever, so the claim also requeues rows whose lease has gone
-- stale (transcoding for longer than p_stale_after with no progress): the row
-- returns to `queued` and any worker may retake it. Requeue happens BEFORE the
-- claim sub-select in the same statement snapshot, so a row released by the sweep
-- is claimable by the very same call.

CREATE FUNCTION produce_job_claim(p_lane text, p_stale_after interval)
RETURNS TABLE (
  id uuid,
  workspace_id uuid,
  kind text,
  input_json jsonb,
  video_asset_id uuid
)
LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$
BEGIN
  -- Lease reclaim: a crashed/killed worker's rows go back to the queue. The
  -- progress write (the runner's per-stage checkpoint) is the liveness signal;
  -- `updated_at` is bumped by every progress/claim write, so `updated_at` older
  -- than the stale window means the owner is gone, not merely slow.
  UPDATE video_processing_jobs stale
     SET state = 'queued', progress_percent = 0, error = NULL, updated_at = now()
   WHERE stale.state = 'transcoding'
     AND stale.updated_at < now() - p_stale_after;

  -- The claim: one lane-eligible queued row, locked so a concurrent worker cannot
  -- take the same one. The kind check + lane check live in the sub-select, so a
  -- worker never locks a row it would only have to release.
  RETURN QUERY
  UPDATE video_processing_jobs j
     SET state = 'transcoding', progress_percent = 5, error = NULL, updated_at = now()
   WHERE j.id IN (
     SELECT c.id
       FROM video_processing_jobs c
      WHERE c.kind IN ('compose', 'mix')
        AND c.state = 'queued'
        AND (
              (p_lane = 'remote' AND c.input_json->>'lane' = 'remote')
           OR (p_lane <> 'remote' AND c.input_json->>'lane' IS DISTINCT FROM 'remote')
            )
      ORDER BY c.created_at, c.id
        FOR UPDATE SKIP LOCKED
        LIMIT 1
   )
  RETURNING j.id, j.workspace_id, j.kind, j.input_json, j.video_asset_id;
END
$$;

REVOKE ALL ON FUNCTION produce_job_claim(text, interval) FROM PUBLIC;
GRANT EXECUTE ON FUNCTION produce_job_claim(text, interval) TO sp_app;

-- The worker's follow-up writes are workspace-scoped already (SET LOCAL
-- app.workspace_id inside a transaction), so no definer rights are needed there:
-- FORCE RLS binds those statements as sp_app, which is the point.

-- The claim's follow-up writes (progress / complete / fail / the derived asset
-- insert) run inside the job's tenant context, so they stay RLS-bound as sp_app —
-- but the worker only holds a job id at that point, and an ordinary SELECT cannot
-- discover which workspace owns it (FORCE RLS hides every row without a tenant
-- context). This is the 012 live_monitor_source_workspace shape verbatim: a narrow
-- definer lookup that discovers the owning tenant and nothing else.
CREATE FUNCTION produce_job_workspace(p_job_id uuid) RETURNS uuid
LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$
  SELECT workspace_id FROM video_processing_jobs WHERE id = p_job_id
$$;

REVOKE ALL ON FUNCTION produce_job_workspace(uuid) FROM PUBLIC;
GRANT EXECUTE ON FUNCTION produce_job_workspace(uuid) TO sp_app;