-- Adds the Redis-free Supabase outbox worker contract to databases created
-- before queue leases were introduced. Safe to apply after the current 0001.

alter table public.ai_jobs add column if not exists claimed_by text;
alter table public.ai_jobs add column if not exists lease_expires_at timestamptz;
alter table public.ai_jobs add column if not exists available_at timestamptz not null default now();
alter table public.ai_jobs add column if not exists max_attempts integer not null default 3;

-- A worker from the old implementation cannot own a database lease. Return
-- such unfinished jobs to the queue before enforcing lease consistency.
update public.ai_jobs
set status = 'queued',
    claimed_by = null,
    lease_expires_at = null,
    available_at = now(),
    updated_at = now()
where status = 'running'
  and (claimed_by is null or lease_expires_at is null);

update public.ai_jobs set available_at = now() where available_at is null;
update public.ai_jobs set max_attempts = 3 where max_attempts is null;
alter table public.ai_jobs alter column available_at set default now();
alter table public.ai_jobs alter column available_at set not null;
alter table public.ai_jobs alter column max_attempts set default 3;
alter table public.ai_jobs alter column max_attempts set not null;
alter table public.ai_jobs enable row level security;

do $$
begin
  if not exists (
    select 1 from pg_constraint
    where conrelid = 'public.ai_jobs'::regclass
      and conname = 'ai_jobs_max_attempts_check'
  ) then
    alter table public.ai_jobs
      add constraint ai_jobs_max_attempts_check check (max_attempts between 1 and 10);
  end if;

  if not exists (
    select 1 from pg_constraint
    where conrelid = 'public.ai_jobs'::regclass
      and conname = 'ai_jobs_lease_consistency_check'
  ) then
    alter table public.ai_jobs
      add constraint ai_jobs_lease_consistency_check check (
        (status = 'running' and claimed_by is not null and lease_expires_at is not null)
        or (status <> 'running' and claimed_by is null and lease_expires_at is null)
      );
  end if;
end;
$$;

create index if not exists ai_jobs_poll_idx
  on public.ai_jobs(available_at, created_at) where status in ('queued', 'running');
create index if not exists ai_jobs_expired_lease_idx
  on public.ai_jobs(lease_expires_at) where status = 'running';

create or replace function public.claim_ai_job(
  p_worker_id text
) returns table(id uuid, job_type text, entity_id uuid)
language plpgsql security definer set search_path = '' as $$
declare
  v_job public.ai_jobs%rowtype;
  v_exhausted public.ai_jobs%rowtype;
begin
  if p_worker_id is null or btrim(p_worker_id) = '' then
    raise exception 'worker id is required';
  end if;

  for v_exhausted in
    select * from public.ai_jobs
    where (
        (status = 'running' and lease_expires_at <= now())
        or (status = 'queued' and available_at <= now())
      )
      and attempts >= max_attempts
    order by created_at
    for update skip locked
    limit 25
  loop
    update public.ai_jobs
    set status = 'failed', claimed_by = null, lease_expires_at = null,
        error = coalesce(error, 'Worker lease expired after the final attempt'), updated_at = now()
    where public.ai_jobs.id = v_exhausted.id;
    if v_exhausted.job_type = 'resolve-round' then
      update public.rounds
      set status = 'failed', error = 'Worker lease expired after the final attempt'
      where public.rounds.id = v_exhausted.entity_id and status <> 'resolved';
    elsif v_exhausted.job_type = 'generate-world' then
      update public.coauthor_sessions
      set status = 'failed', updated_at = now()
      where public.coauthor_sessions.id = v_exhausted.entity_id and status <> 'ready';
    end if;
  end loop;

  select * into v_job
  from public.ai_jobs
  where attempts < max_attempts
    and available_at <= now()
    and (
      status = 'queued'
      or (status = 'running' and lease_expires_at <= now())
    )
  order by created_at
  for update skip locked
  limit 1;

  if not found then return; end if;

  update public.ai_jobs
  set status = 'running',
      attempts = attempts + 1,
      claimed_by = p_worker_id,
      lease_expires_at = now() + interval '3 minutes',
      error = null,
      updated_at = now()
  where public.ai_jobs.id = v_job.id;

  if v_job.job_type = 'resolve-round' then
    update public.rounds set status = 'resolving', error = null where public.rounds.id = v_job.entity_id;
  elsif v_job.job_type = 'generate-world' then
    update public.coauthor_sessions set status = 'generating', updated_at = now() where public.coauthor_sessions.id = v_job.entity_id;
  end if;

  return query select v_job.id, v_job.job_type, v_job.entity_id;
end;
$$;

create or replace function public.claim_ai_job_by_id(
  p_job_id uuid,
  p_worker_id text
) returns table(id uuid, job_type text, entity_id uuid)
language plpgsql security definer set search_path = '' as $$
declare
  v_job public.ai_jobs%rowtype;
begin
  if p_job_id is null then raise exception 'job id is required'; end if;
  if p_worker_id is null or btrim(p_worker_id) = '' then
    raise exception 'worker id is required';
  end if;

  select * into v_job from public.ai_jobs where public.ai_jobs.id = p_job_id for update;

  if not found then return; end if;

  if (
      (v_job.status = 'running' and v_job.lease_expires_at <= now())
      or (v_job.status = 'queued' and v_job.available_at <= now())
    )
    and v_job.attempts >= v_job.max_attempts then
    update public.ai_jobs
    set status = 'failed', claimed_by = null, lease_expires_at = null,
        error = coalesce(error, 'Worker lease expired after the final attempt'), updated_at = now()
    where public.ai_jobs.id = v_job.id;
    if v_job.job_type = 'resolve-round' then
      update public.rounds
      set status = 'failed', error = 'Worker lease expired after the final attempt'
      where public.rounds.id = v_job.entity_id and status <> 'resolved';
    elsif v_job.job_type = 'generate-world' then
      update public.coauthor_sessions
      set status = 'failed', updated_at = now()
      where public.coauthor_sessions.id = v_job.entity_id and status <> 'ready';
    end if;
    return;
  end if;

  if v_job.attempts >= v_job.max_attempts
    or v_job.available_at > now()
    or not (
      v_job.status = 'queued'
      or (v_job.status = 'running' and v_job.lease_expires_at <= now())
    ) then
    return;
  end if;

  update public.ai_jobs
  set status = 'running',
      attempts = attempts + 1,
      claimed_by = p_worker_id,
      lease_expires_at = now() + interval '3 minutes',
      error = null,
      updated_at = now()
  where public.ai_jobs.id = v_job.id;

  if v_job.job_type = 'resolve-round' then
    update public.rounds set status = 'resolving', error = null where public.rounds.id = v_job.entity_id;
  elsif v_job.job_type = 'generate-world' then
    update public.coauthor_sessions set status = 'generating', updated_at = now() where public.coauthor_sessions.id = v_job.entity_id;
  end if;

  return query select v_job.id, v_job.job_type, v_job.entity_id;
end;
$$;

create or replace function public.complete_ai_job(
  p_job_id uuid,
  p_worker_id text
) returns void language plpgsql security definer set search_path = '' as $$
begin
  if p_worker_id is null or btrim(p_worker_id) = '' then
    raise exception 'worker id is required';
  end if;
  update public.ai_jobs
  set status = 'complete', claimed_by = null, lease_expires_at = null, updated_at = now()
  where id = p_job_id
    and status = 'running'
    and claimed_by = p_worker_id
    and lease_expires_at > now();
  if not found then raise exception 'job is not owned by this worker'; end if;
end;
$$;

create or replace function public.retry_ai_job(
  p_job_id uuid,
  p_worker_id text,
  p_error text
) returns void language plpgsql security definer set search_path = '' as $$
declare
  v_job public.ai_jobs%rowtype;
  v_retry boolean;
begin
  if p_worker_id is null or btrim(p_worker_id) = '' then
    raise exception 'worker id is required';
  end if;
  select * into v_job from public.ai_jobs where id = p_job_id for update;
  if not found or v_job.status <> 'running' or v_job.claimed_by <> p_worker_id then
    raise exception 'job is not owned by this worker';
  end if;
  if v_job.lease_expires_at <= now() then
    raise exception 'job lease has expired';
  end if;

  v_retry := v_job.attempts < v_job.max_attempts;
  update public.ai_jobs
  set status = case when v_retry then 'queued' else 'failed' end,
      claimed_by = null,
      lease_expires_at = null,
      available_at = case
        when v_retry then now() + make_interval(secs => 5 * power(2, v_job.attempts - 1)::integer)
        else available_at
      end,
      error = left(p_error, 2000),
      updated_at = now()
  where id = p_job_id;

  if v_job.job_type = 'resolve-round' then
    update public.rounds
    set status = case
          when v_retry then 'queued'::public.round_status
          else 'failed'::public.round_status
        end,
        error = left(p_error, 2000)
    where id = v_job.entity_id and status <> 'resolved';
  elsif v_job.job_type = 'generate-world' then
    update public.coauthor_sessions
    set status = case when v_retry then 'generating' else 'failed' end,
        updated_at = now()
    where id = v_job.entity_id and status <> 'ready';
  end if;
end;
$$;

create or replace function public.renew_ai_job_lease(
  p_job_id uuid,
  p_worker_id text
) returns timestamptz language plpgsql security definer set search_path = '' as $$
declare
  v_lease_expires_at timestamptz;
begin
  if p_worker_id is null or btrim(p_worker_id) = '' then
    raise exception 'worker id is required';
  end if;
  update public.ai_jobs
  set lease_expires_at = now() + interval '3 minutes', updated_at = now()
  where id = p_job_id
    and status = 'running'
    and claimed_by = p_worker_id
    and lease_expires_at > now()
  returning lease_expires_at into v_lease_expires_at;
  if not found then raise exception 'job is not owned by this worker or lease expired'; end if;
  return v_lease_expires_at;
end;
$$;

-- Replace the pre-lease commit function as well: old databases completed a
-- job without clearing its lease, which conflicts with the new invariant.
create or replace function public.commit_round_resolution(
  p_round_id uuid,
  p_narration text,
  p_next_prompt text,
  p_rolls jsonb,
  p_events jsonb,
  p_character_states jsonb,
  p_memory jsonb,
  p_idempotency_key text
) returns void language plpgsql security definer set search_path = '' as $$
declare
  v_round public.rounds%rowtype;
  v_job public.ai_jobs%rowtype;
  v_roll jsonb;
  v_event jsonb;
  v_state jsonb;
  v_character_id uuid;
  v_seen_character_ids uuid[] := '{}';
  v_row_count integer;
begin
  if p_idempotency_key is null or btrim(p_idempotency_key) = '' then
    raise exception 'idempotency key is required';
  end if;
  if p_narration is null or btrim(p_narration) = '' or p_next_prompt is null or btrim(p_next_prompt) = '' then
    raise exception 'narration and next prompt are required';
  end if;
  if coalesce(jsonb_typeof(p_rolls), '') <> 'array'
    or coalesce(jsonb_typeof(p_events), '') <> 'array'
    or coalesce(jsonb_typeof(p_character_states), '') <> 'array' then
    raise exception 'rolls, events, and character states must be arrays';
  end if;

  select * into v_round from public.rounds where id = p_round_id for update;
  if not found then raise exception 'round not found'; end if;
  if v_round.status = 'resolved' then return; end if;
  if v_round.status not in ('queued', 'resolving') then
    raise exception 'round is not queued for resolution';
  end if;

  select * into v_job
  from public.ai_jobs
  where idempotency_key = p_idempotency_key
     or id::text = p_idempotency_key
  for update;
  if not found then raise exception 'resolution job not found'; end if;
  if v_job.job_type <> 'resolve-round' or v_job.entity_id <> p_round_id then
    raise exception 'idempotency key belongs to another job';
  end if;
  if v_job.status = 'complete' then return; end if;
  if v_job.status <> 'running' then
    raise exception 'resolution job must be claimed before commit';
  end if;

  update public.ai_jobs set error = null, updated_at = now() where id = v_job.id;
  update public.rounds set status = 'resolving', error = null where id = p_round_id;

  for v_roll in select * from jsonb_array_elements(p_rolls) loop
    if not exists (
      select 1 from public.characters
      where id = (v_roll->>'actorId')::uuid and campaign_id = v_round.campaign_id
    ) or (
      nullif(v_roll->>'targetId', '') is not null
      and not exists (
        select 1 from public.characters
        where id = nullif(v_roll->>'targetId', '')::uuid and campaign_id = v_round.campaign_id
      )
    ) then
      raise exception 'roll actor or target is outside the round campaign';
    end if;
    insert into public.dice_rolls(
      id, round_id, actor_id, target_id, check_kind, formula, rolls, kept,
      modifier, total, difficulty, success, created_at
    ) values (
      (v_roll->>'id')::uuid, p_round_id, (v_roll->>'actorId')::uuid,
      nullif(v_roll->>'targetId', '')::uuid, v_roll->>'checkKind', v_roll->>'formula',
      array(select jsonb_array_elements_text(v_roll->'rolls')::integer),
      array(select jsonb_array_elements_text(v_roll->'kept')::integer),
      (v_roll->>'modifier')::integer, (v_roll->>'total')::integer,
      nullif(v_roll->>'difficulty', '')::integer, (v_roll->>'success')::boolean,
      (v_roll->>'createdAt')::timestamptz
    );
  end loop;

  for v_event in select * from jsonb_array_elements(p_events) loop
    if coalesce(btrim(v_event->>'type'), '') = '' then
      raise exception 'event type is required';
    end if;
    insert into public.game_events(campaign_id, round_id, event_type, payload)
    values (v_round.campaign_id, p_round_id, v_event->>'type', v_event);
  end loop;

  for v_state in select * from jsonb_array_elements(p_character_states) loop
    v_character_id := (v_state->>'id')::uuid;
    if v_character_id = any(v_seen_character_ids) then
      raise exception 'duplicate character state for %', v_character_id;
    end if;
    v_seen_character_ids := array_append(v_seen_character_ids, v_character_id);
    if coalesce(jsonb_typeof(v_state->'inventory'), '') <> 'array'
      or coalesce(jsonb_typeof(v_state->'statuses'), '') <> 'array' then
      raise exception 'inventory and statuses must be arrays';
    end if;
    update public.characters
    set hp = (v_state->>'hp')::integer,
        inventory = v_state->'inventory',
        statuses = v_state->'statuses'
    where id = v_character_id and campaign_id = v_round.campaign_id;
    get diagnostics v_row_count = row_count;
    if v_row_count <> 1 then
      raise exception 'character % is outside the round campaign', v_character_id;
    end if;
  end loop;

  if p_memory is not null and p_memory <> 'null'::jsonb then
    if coalesce(jsonb_typeof(p_memory), '') <> 'object' then
      raise exception 'memory must be an object';
    end if;
    insert into public.memories(campaign_id, round_id, summary, importance, tags, entity_ids)
    values (
      v_round.campaign_id,
      p_round_id,
      p_memory->>'summary',
      (p_memory->>'importance')::integer,
      array(select jsonb_array_elements_text(coalesce(p_memory->'tags', '[]'::jsonb))),
      array(select jsonb_array_elements_text(coalesce(p_memory->'entityIds', '[]'::jsonb))::uuid)
    );
  end if;

  update public.rounds
  set status = 'resolved', narration = p_narration, next_prompt = p_next_prompt, resolved_at = now()
  where id = p_round_id;
  update public.campaigns
  set current_scene = p_narration, next_prompt = p_next_prompt, updated_at = now()
  where id = v_round.campaign_id;
  update public.ai_jobs
  set status = 'complete', claimed_by = null, lease_expires_at = null, updated_at = now()
  where id = v_job.id;
end;
$$;

create or replace function public.commit_claimed_round_resolution(
  p_round_id uuid,
  p_narration text,
  p_next_prompt text,
  p_rolls jsonb,
  p_events jsonb,
  p_character_states jsonb,
  p_memory jsonb,
  p_idempotency_key text,
  p_worker_id text
) returns void language plpgsql security definer set search_path = '' as $$
declare
  v_job public.ai_jobs%rowtype;
begin
  if p_worker_id is null or btrim(p_worker_id) = '' then
    raise exception 'worker id is required';
  end if;
  select * into v_job
    from public.ai_jobs
    where (idempotency_key = p_idempotency_key or id::text = p_idempotency_key)
      and job_type = 'resolve-round'
      and entity_id = p_round_id
    for update;
  if not found
    or v_job.status <> 'running'
    or v_job.claimed_by <> p_worker_id
    or v_job.lease_expires_at <= now() then
    raise exception 'resolution job is not owned by this worker';
  end if;

  perform public.commit_round_resolution(
    p_round_id,
    p_narration,
    p_next_prompt,
    p_rolls,
    p_events,
    p_character_states,
    p_memory,
    p_idempotency_key
  );
end;
$$;

revoke all on function public.claim_ai_job(text) from public, anon, authenticated;
revoke all on function public.claim_ai_job_by_id(uuid, text) from public, anon, authenticated;
revoke all on function public.complete_ai_job(uuid, text) from public, anon, authenticated;
revoke all on function public.retry_ai_job(uuid, text, text) from public, anon, authenticated;
revoke all on function public.renew_ai_job_lease(uuid, text) from public, anon, authenticated;
revoke all on function public.commit_round_resolution(uuid, text, text, jsonb, jsonb, jsonb, jsonb, text) from public, anon, authenticated;
revoke all on function public.commit_claimed_round_resolution(uuid, text, text, jsonb, jsonb, jsonb, jsonb, text, text) from public, anon, authenticated;

grant execute on function public.claim_ai_job(text) to service_role;
grant execute on function public.claim_ai_job_by_id(uuid, text) to service_role;
grant execute on function public.complete_ai_job(uuid, text) to service_role;
grant execute on function public.retry_ai_job(uuid, text, text) to service_role;
grant execute on function public.renew_ai_job_lease(uuid, text) to service_role;
grant execute on function public.commit_round_resolution(uuid, text, text, jsonb, jsonb, jsonb, jsonb, text) to service_role;
grant execute on function public.commit_claimed_round_resolution(uuid, text, text, jsonb, jsonb, jsonb, jsonb, text, text) to service_role;

-- Make newly-created RPCs visible to PostgREST immediately.
notify pgrst, 'reload schema';