487 lines
18 KiB
PL/PgSQL
487 lines
18 KiB
PL/PgSQL
-- 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';
|