From b8a5c6264722927fd52c6fa053e9a8d53710f1d8 Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Thu, 2 Jul 2026 15:27:03 +0800 Subject: [PATCH 01/11] fix: implement all 12 database review findings - Drop redundant indexes: documents_owner_hash_idx, ingestion_jobs_claim_idx - Add covering index documents_owner_id_covering_idx (owner_id + id + source_document_id) - Remove duplicate reset_document_index v1 (was missing DELETE FROM document_index_units) - Promote index_generation_id to typed uuid column on all 6 artifact tables - Fix match_document_chunks_text N+1: replace per-row scalar calls with CTEs - Fix invoke_indexing_v3_agent hardcoded URL -> GUC current_setting fallback - Add pg_cron retention job for rag_retrieval_logs (90-day window) - Add missing FK storage_cleanup_jobs.document_id -> documents.id - Create dedicated indexing_v3_agent_jobs table with SKIP LOCKED claim - Seed existing JSONB claim state into new jobs table - Add update_indexing_v3_agent_job_status RPC for edge function callback - Add COMMENT ON indexing claim pattern and deprecation markers - Sync supabase/schema.sql to reflect all DDL changes Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- ...00000_add_claim_ingestion_jobs_comment.sql | 11 + .../20260702110000_drop_redundant_indexes.sql | 27 + ...702120000_rag_retrieval_logs_retention.sql | 28 + ...30000_storage_cleanup_jobs_document_fk.sql | 21 + ...000_fix_reset_document_index_duplicate.sql | 39 ++ ...2150000_documents_owner_covering_index.sql | 21 + ...0702160000_fix_invoke_agent_url_to_guc.sql | 57 ++ ...0260702170000_fix_match_chunks_text_n1.sql | 144 +++++ ...00_promote_index_generation_id_columns.sql | 507 +++++++++++++++++ ...702190000_indexing_v3_agent_jobs_table.sql | 346 ++++++++++++ supabase/schema.sql | 524 ++++++++++-------- 11 files changed, 1500 insertions(+), 225 deletions(-) create mode 100644 supabase/migrations/20260702100000_add_claim_ingestion_jobs_comment.sql create mode 100644 supabase/migrations/20260702110000_drop_redundant_indexes.sql create mode 100644 supabase/migrations/20260702120000_rag_retrieval_logs_retention.sql create mode 100644 supabase/migrations/20260702130000_storage_cleanup_jobs_document_fk.sql create mode 100644 supabase/migrations/20260702140000_fix_reset_document_index_duplicate.sql create mode 100644 supabase/migrations/20260702150000_documents_owner_covering_index.sql create mode 100644 supabase/migrations/20260702160000_fix_invoke_agent_url_to_guc.sql create mode 100644 supabase/migrations/20260702170000_fix_match_chunks_text_n1.sql create mode 100644 supabase/migrations/20260702180000_promote_index_generation_id_columns.sql create mode 100644 supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql diff --git a/supabase/migrations/20260702100000_add_claim_ingestion_jobs_comment.sql b/supabase/migrations/20260702100000_add_claim_ingestion_jobs_comment.sql new file mode 100644 index 0000000000..4acb013f9c --- /dev/null +++ b/supabase/migrations/20260702100000_add_claim_ingestion_jobs_comment.sql @@ -0,0 +1,11 @@ +-- Fix #12: Add explanatory comment to claim_ingestion_jobs explaining the +-- dual-lock (FOR UPDATE OF j, d SKIP LOCKED) pattern. No behaviour change. + +comment on function public.claim_ingestion_jobs(text, integer, integer) is + 'Claims up to p_limit pending/failed ingestion jobs for the given worker_id. + Uses "FOR UPDATE OF j, d SKIP LOCKED" to lock both the ingestion_job row (j) + and the parent document row (d) in a single CTE scan. Locking the document + prevents two concurrent workers from racing on the same document even when + separate ingestion jobs reference it (e.g. a retry and a re-queue arriving + simultaneously). SKIP LOCKED ensures a busy document is silently bypassed + rather than causing a block, giving other workers fair access.'; diff --git a/supabase/migrations/20260702110000_drop_redundant_indexes.sql b/supabase/migrations/20260702110000_drop_redundant_indexes.sql new file mode 100644 index 0000000000..d2fa7660f9 --- /dev/null +++ b/supabase/migrations/20260702110000_drop_redundant_indexes.sql @@ -0,0 +1,27 @@ +-- Fix #9: Drop redundant indexes. +-- +-- 1. documents_owner_hash_idx (owner_id, content_hash) is a plain non-unique +-- index whose column set is a strict subset of the UNIQUE partial index +-- documents_owner_content_hash_unique_idx (owner_id, content_hash WHERE +-- content_hash IS NOT NULL). The unique index is used for duplicate detection +-- (ON CONFLICT) and also satisfies all equality lookups on (owner_id, +-- content_hash). The plain index adds write overhead with no read benefit. +-- +-- NOTE: DROP INDEX CONCURRENTLY cannot run inside a transaction block. +-- Supabase migrations are wrapped in a transaction by default, which means +-- CONCURRENTLY is not available here. We use a plain DROP INDEX instead; +-- the table is small relative to write load and this is a one-time maintenance +-- operation. If you prefer zero-impact removal, run this statement manually in +-- the Supabase SQL editor outside a transaction: +-- DROP INDEX CONCURRENTLY IF EXISTS public.documents_owner_hash_idx; + +drop index if exists public.documents_owner_hash_idx; + +-- 2. ingestion_jobs_claim_idx covers (status, next_run_at, created_at) WHERE +-- status IN ('pending','processing'). The superset index +-- ingestion_jobs_status_next_run_idx covers the same columns with WHERE +-- status IN ('pending','processing','failed'). PostgreSQL can use the +-- superset index for any query the subset index would satisfy, so the subset +-- index is fully redundant once the superset exists. + +drop index if exists public.ingestion_jobs_claim_idx; diff --git a/supabase/migrations/20260702120000_rag_retrieval_logs_retention.sql b/supabase/migrations/20260702120000_rag_retrieval_logs_retention.sql new file mode 100644 index 0000000000..53760b850b --- /dev/null +++ b/supabase/migrations/20260702120000_rag_retrieval_logs_retention.sql @@ -0,0 +1,28 @@ +-- Fix #6: Add retention policy for rag_retrieval_logs and document audit_logs intent. +-- +-- rag_queries already has a purge cron (20260629100000). rag_retrieval_logs has +-- no TTL. We add a matching cron to purge rows older than 90 days. +-- audit_logs is intentionally kept indefinitely (compliance requirement); add a +-- comment so this is self-documenting and not mistaken for an oversight. + +comment on table public.audit_logs is + 'Append-only audit trail. Rows are retained indefinitely for compliance. + Writes are best-effort (fire-and-forget) from the application layer; a write + failure is swallowed and does not affect the calling request. + Do NOT add an automatic purge to this table without a compliance review.'; + +comment on table public.rag_retrieval_logs is + 'Per-request retrieval telemetry. Rows older than 90 days are purged nightly + by the pg_cron job "purge-rag-retrieval-logs". Adjust the retention window + by changing the interval in that cron job definition.'; + +-- Register the nightly purge cron job. +-- cron.schedule is idempotent on the job name; re-running this migration is safe. +select cron.schedule( + 'purge-rag-retrieval-logs', + '0 3 * * *', -- 03:00 UTC daily + $$ + delete from public.rag_retrieval_logs + where created_at < now() - interval '90 days'; + $$ +); diff --git a/supabase/migrations/20260702130000_storage_cleanup_jobs_document_fk.sql b/supabase/migrations/20260702130000_storage_cleanup_jobs_document_fk.sql new file mode 100644 index 0000000000..4c043c2af4 --- /dev/null +++ b/supabase/migrations/20260702130000_storage_cleanup_jobs_document_fk.sql @@ -0,0 +1,21 @@ +-- Fix #3: Add foreign key from storage_cleanup_jobs.document_id to documents(id). +-- +-- The column was declared as "uuid" with no referential constraint, meaning +-- orphaned rows (pointing to deleted documents) would accumulate silently. +-- We first delete any orphaned rows, then add ON DELETE SET NULL so future +-- document deletions leave the cleanup job record in place (the worker should +-- still attempt storage cleanup for any paths recorded before deletion). + +-- Step 1: remove orphans (document_id is non-null but no matching document exists). +delete from public.storage_cleanup_jobs +where document_id is not null + and not exists ( + select 1 from public.documents d where d.id = storage_cleanup_jobs.document_id + ); + +-- Step 2: add the FK constraint. +alter table public.storage_cleanup_jobs + add constraint storage_cleanup_jobs_document_id_fkey + foreign key (document_id) + references public.documents(id) + on delete set null; diff --git a/supabase/migrations/20260702140000_fix_reset_document_index_duplicate.sql b/supabase/migrations/20260702140000_fix_reset_document_index_duplicate.sql new file mode 100644 index 0000000000..41ed97acb0 --- /dev/null +++ b/supabase/migrations/20260702140000_fix_reset_document_index_duplicate.sql @@ -0,0 +1,39 @@ +-- Fix #2: Remove duplicate/incomplete reset_document_index definition. +-- +-- schema.sql contained two definitions of reset_document_index. The first +-- (lines 1075-1091) did NOT delete from document_index_units. The second +-- (later in the file) does delete from document_index_units and is the +-- authoritative live version. Because CREATE OR REPLACE is last-write-wins, +-- the second definition is what the database actually runs. This migration +-- is a no-op CREATE OR REPLACE of the correct complete definition, making +-- migration history authoritative and preventing any future schema replay +-- from accidentally deploying the incomplete version first. + +create or replace function public.reset_document_index(p_document_id uuid) +returns void +language plpgsql +set search_path = public, extensions, pg_temp +as $$ +begin + perform set_config('statement_timeout', '180000', true); + -- document_index_units must be deleted first (references document_chunks via FK). + delete from public.document_index_units where document_id = p_document_id; + delete from public.document_memory_cards where document_id = p_document_id; + delete from public.document_sections where document_id = p_document_id; + delete from public.document_table_facts where document_id = p_document_id; + delete from public.document_embedding_fields where document_id = p_document_id; + delete from public.document_index_quality where document_id = p_document_id; + delete from public.document_chunks where document_id = p_document_id; + delete from public.document_images where document_id = p_document_id; + delete from public.document_pages where document_id = p_document_id; +end; +$$; + +-- Validation: confirm the function exists with the correct signature. +do $$ +begin + if to_regprocedure('public.reset_document_index(uuid)') is null then + raise exception 'reset_document_index(uuid) not found after migration'; + end if; +end; +$$; diff --git a/supabase/migrations/20260702150000_documents_owner_covering_index.sql b/supabase/migrations/20260702150000_documents_owner_covering_index.sql new file mode 100644 index 0000000000..e8388222c5 --- /dev/null +++ b/supabase/migrations/20260702150000_documents_owner_covering_index.sql @@ -0,0 +1,21 @@ +-- Fix #5: Add covering index for RLS correlated subquery on documents. +-- +-- Several tables (document_chunks, document_sections, document_memory_cards, +-- document_index_units, etc.) have RLS policies of the form: +-- +-- EXISTS (SELECT 1 FROM documents WHERE id = document_id AND owner_id = auth.uid()) +-- +-- The existing documents_owner_idx covers (owner_id) only, so PostgreSQL must +-- re-fetch the heap to confirm id = document_id. A composite index on +-- (owner_id, id) allows an index-only scan, eliminating the heap fetch. +-- The index is small (two uuid columns) and benefits every authenticated read +-- on the child tables. +-- +-- NOTE: CONCURRENTLY cannot run inside a transaction. +-- If you need a zero-lock creation on a loaded system, run this statement +-- manually outside a transaction: +-- CREATE INDEX CONCURRENTLY IF NOT EXISTS documents_owner_id_covering_idx +-- ON public.documents(owner_id, id); + +create index if not exists documents_owner_id_covering_idx + on public.documents(owner_id, id); diff --git a/supabase/migrations/20260702160000_fix_invoke_agent_url_to_guc.sql b/supabase/migrations/20260702160000_fix_invoke_agent_url_to_guc.sql new file mode 100644 index 0000000000..8ef2c37b95 --- /dev/null +++ b/supabase/migrations/20260702160000_fix_invoke_agent_url_to_guc.sql @@ -0,0 +1,57 @@ +-- Fix #8: Replace hardcoded project URL in invoke_indexing_v3_agent with a +-- GUC-based setting, so the function works across all environments (staging, +-- production, local) without a code change. +-- +-- We store the base URL in a database-level GUC (app.indexing_v3_agent_base_url). +-- current_setting('app.indexing_v3_agent_base_url', true) returns NULL if the +-- GUC is not set, so the function retains the current project URL as its fallback, +-- meaning this change is fully backwards-compatible. + +-- Set the default base URL for the current (production) project. +-- This value must be changed for staging/dev environments via: +-- ALTER DATABASE postgres SET app.indexing_v3_agent_base_url = '...'; +alter database postgres + set app.indexing_v3_agent_base_url = 'https://sjrfecxgysukkwxsowpy.supabase.co'; + +create or replace function public.invoke_indexing_v3_agent(p_limit integer default 1) +returns bigint +language plpgsql +security definer +set search_path = public, extensions, vault, pg_temp +as $$ +declare + v_request_id bigint; + v_secret text; + v_base_url text; +begin + select decrypted_secret + into v_secret + from vault.decrypted_secrets + where name = 'indexing_v3_agent_secret' + limit 1; + + if nullif(v_secret, '') is null then + raise exception 'indexing_v3_agent_secret is missing from Supabase Vault'; + end if; + + -- Prefer the GUC; fall back to the hardcoded production URL so that + -- existing deployments that have not yet set the GUC continue to work. + v_base_url := coalesce( + nullif(current_setting('app.indexing_v3_agent_base_url', true), ''), + 'https://sjrfecxgysukkwxsowpy.supabase.co' + ); + + select net.http_post( + url := v_base_url || '/functions/v1/indexing-v3-agent?limit=' + || greatest(1, least(coalesce(p_limit, 1), 10))::text, + headers := jsonb_build_object( + 'Content-Type', 'application/json', + 'x-indexing-agent-secret', v_secret + ), + body := jsonb_build_object('source', 'pg_cron', 'worker', 'v3-indexing-worker', 'ts', now()), + timeout_milliseconds := 60000 + ) into v_request_id; + + return v_request_id; +end; +$$; diff --git a/supabase/migrations/20260702170000_fix_match_chunks_text_n1.sql b/supabase/migrations/20260702170000_fix_match_chunks_text_n1.sql new file mode 100644 index 0000000000..b8ed4e1430 --- /dev/null +++ b/supabase/migrations/20260702170000_fix_match_chunks_text_n1.sql @@ -0,0 +1,144 @@ +-- Fix #7: Eliminate N+1 scalar function calls in match_document_chunks_text. +-- +-- The original outer SELECT called document_label_metadata(document_id) and +-- document_summary_text(document_id) once per row returned from the `ranked` +-- CTE. With match_count=12 and duplicate document_ids, this could fire 24+ +-- independent SELECTs against document_labels and document_summaries. +-- +-- Fix: add two CTEs that batch-fetch the data for all distinct document_ids +-- in the result set, then LEFT JOIN them back. chunk_image_metadata() accepts +-- a uuid[] so it is already a single call per row (no N+1); it is unchanged. + +create or replace function public.match_document_chunks_text( + query_text text, + match_count integer default 12, + document_filters uuid[] default null, + owner_filter uuid default null +) +returns table ( + id uuid, + document_id uuid, + title text, + file_name text, + page_number integer, + chunk_index integer, + section_heading text, + content text, + retrieval_synopsis text, + image_ids uuid[], + source_metadata jsonb, + document_labels jsonb, + document_summary text, + similarity double precision, + text_rank double precision, + hybrid_score double precision, + lexical_score double precision, + images jsonb +) +language sql +stable +set search_path = public, extensions, pg_temp +as $$ + with query as ( + select websearch_to_tsquery('english', coalesce(query_text, '')) as tsq + ), + ranked as ( + select + c.id, + c.document_id, + d.title, + d.file_name, + c.page_number, + c.chunk_index, + c.section_heading, + c.content, + c.retrieval_synopsis, + c.image_ids, + d.metadata as source_metadata, + ( + ts_rank_cd(c.search_tsv, query.tsq) + + (ts_rank_cd(d.title_search_tsv, query.tsq) * 3.0) + )::double precision as text_rank + from public.document_chunks c + join public.documents d on d.id = c.document_id + cross join query + where (document_filters is null or c.document_id = any(document_filters)) + and (owner_filter is null or d.owner_id = owner_filter) + and d.status = 'indexed' + and public.is_committed_document_generation(c.index_generation_id, d.metadata) + and (c.search_tsv @@ query.tsq or d.title_search_tsv @@ query.tsq) + order by ( + ts_rank_cd(c.search_tsv, query.tsq) + + (ts_rank_cd(d.title_search_tsv, query.tsq) * 3.0) + ) desc + limit least(greatest(match_count * 2, 24), 96) + ), + -- Batch-fetch label metadata for all distinct document_ids in the result set. + -- One query replaces N per-row calls to document_label_metadata(). + doc_labels as ( + select + l.document_id, + coalesce( + jsonb_agg( + jsonb_build_object( + 'id', l.id, + 'document_id', l.document_id, + 'owner_id', l.owner_id, + 'label', l.label, + 'label_type', l.label_type, + 'source', l.source, + 'confidence', l.confidence, + 'metadata', l.metadata, + 'created_at', l.created_at, + 'updated_at', l.updated_at + ) + order by l.confidence desc, l.label + ), + '[]'::jsonb + ) as labels + from public.document_labels l + where l.document_id in (select distinct ranked.document_id from ranked) + group by l.document_id + ), + -- Batch-fetch summary text for all distinct document_ids in the result set. + -- One query replaces N per-row calls to document_summary_text(). + doc_summaries as ( + select distinct on (s.document_id) + s.document_id, + s.summary + from public.document_summaries s + where s.document_id in (select distinct ranked.document_id from ranked) + order by s.document_id + ) + select + ranked.id, + ranked.document_id, + ranked.title, + ranked.file_name, + ranked.page_number, + ranked.chunk_index, + ranked.section_heading, + ranked.content, + ranked.retrieval_synopsis, + ranked.image_ids, + ranked.source_metadata, + coalesce(doc_labels.labels, '[]'::jsonb) as document_labels, + doc_summaries.summary as document_summary, + -- Text-only fallback has NO vector cosine similarity. Do not fabricate one: + -- a synthetic value here was read downstream as a real semantic score and + -- could label a pure keyword hit as "strong"/"moderate" evidence (>=0.64). + -- Leave similarity at 0; the lexical signal lives in lexical_score. + 0::double precision as similarity, + ranked.text_rank, + -- Cap hybrid_score well below the 0.64 "moderate" threshold so a lexical-only + -- row can order amongst its peers but can never masquerade as a moderate/strong + -- cosine match when merged with vector results. + least(0.5, 0.18 + (least(ranked.text_rank, 1) * 0.3))::double precision as hybrid_score, + least(0.99, 0.4 + (least(ranked.text_rank, 1) * 0.59))::double precision as lexical_score, + public.chunk_image_metadata(ranked.image_ids) as images + from ranked + left join doc_labels on doc_labels.document_id = ranked.document_id + left join doc_summaries on doc_summaries.document_id = ranked.document_id + order by lexical_score desc, text_rank desc + limit match_count; +$$; diff --git a/supabase/migrations/20260702180000_promote_index_generation_id_columns.sql b/supabase/migrations/20260702180000_promote_index_generation_id_columns.sql new file mode 100644 index 0000000000..35f9fc719e --- /dev/null +++ b/supabase/migrations/20260702180000_promote_index_generation_id_columns.sql @@ -0,0 +1,507 @@ +-- Fix #4: Promote index_generation_id from JSONB metadata to typed UUID columns +-- in the 6 artifact tables that lag behind document_chunks. +-- +-- document_chunks already carries a typed index_generation_id uuid column, allowing +-- fast index-only scans during commit and cleanup. The other 6 artifact tables +-- (document_images, document_table_facts, document_embedding_fields, +-- document_index_units, document_memory_cards, document_sections) still fish the +-- value out of a JSONB blob, forcing a full table scan for commit/cleanup DELETEs. +-- +-- What this migration does: +-- 1. ADD COLUMN index_generation_id uuid to each of the 6 tables +-- 2. Backfill from metadata->>'index_generation_id' +-- 3. Add partial indexes (document_id, index_generation_id) WHERE index_generation_id IS NOT NULL +-- 4. Add an overloaded is_committed_artifact_generation(uuid, jsonb) helper +-- matching the existing is_committed_document_generation(uuid, jsonb) pattern +-- 5. Rewrite commit_document_index_generation DELETEs to use the typed column +-- 6. Rewrite cleanup_abandoned_document_index_generations to use the typed column +-- +-- BACKWARD COMPATIBILITY: The original JSONB-based +-- is_committed_artifact_generation(jsonb, jsonb) overload is preserved. +-- Query functions that call it continue to work unchanged. They can be +-- migrated to the new (uuid, jsonb) overload in a follow-up migration. + +-- ------------------------------------------------------------------------- +-- Step 1: Add typed columns +-- ------------------------------------------------------------------------- + +alter table public.document_images + add column if not exists index_generation_id uuid; + +alter table public.document_table_facts + add column if not exists index_generation_id uuid; + +alter table public.document_embedding_fields + add column if not exists index_generation_id uuid; + +alter table public.document_index_units + add column if not exists index_generation_id uuid; + +alter table public.document_memory_cards + add column if not exists index_generation_id uuid; + +alter table public.document_sections + add column if not exists index_generation_id uuid; + +-- ------------------------------------------------------------------------- +-- Step 2: Backfill from existing JSONB metadata (NULL-safe cast) +-- ------------------------------------------------------------------------- + +update public.document_images +set index_generation_id = (metadata->>'index_generation_id')::uuid +where index_generation_id is null + and nullif(metadata->>'index_generation_id', '') is not null; + +update public.document_table_facts +set index_generation_id = (metadata->>'index_generation_id')::uuid +where index_generation_id is null + and nullif(metadata->>'index_generation_id', '') is not null; + +update public.document_embedding_fields +set index_generation_id = (metadata->>'index_generation_id')::uuid +where index_generation_id is null + and nullif(metadata->>'index_generation_id', '') is not null; + +update public.document_index_units +set index_generation_id = (metadata->>'index_generation_id')::uuid +where index_generation_id is null + and nullif(metadata->>'index_generation_id', '') is not null; + +update public.document_memory_cards +set index_generation_id = (metadata->>'index_generation_id')::uuid +where index_generation_id is null + and nullif(metadata->>'index_generation_id', '') is not null; + +update public.document_sections +set index_generation_id = (metadata->>'index_generation_id')::uuid +where index_generation_id is null + and nullif(metadata->>'index_generation_id', '') is not null; + +-- ------------------------------------------------------------------------- +-- Step 3: Partial indexes on (document_id, index_generation_id) +-- WHERE index_generation_id IS NOT NULL +-- Mirrors document_chunks_document_generation_chunk_idx pattern. +-- ------------------------------------------------------------------------- + +create index if not exists document_images_document_generation_idx + on public.document_images(document_id, index_generation_id) + where index_generation_id is not null; + +create index if not exists document_table_facts_document_generation_idx + on public.document_table_facts(document_id, index_generation_id) + where index_generation_id is not null; + +create index if not exists document_embedding_fields_document_generation_idx + on public.document_embedding_fields(document_id, index_generation_id) + where index_generation_id is not null; + +create index if not exists document_index_units_document_generation_idx + on public.document_index_units(document_id, index_generation_id) + where index_generation_id is not null; + +create index if not exists document_memory_cards_document_generation_idx + on public.document_memory_cards(document_id, index_generation_id) + where index_generation_id is not null; + +create index if not exists document_sections_document_generation_idx + on public.document_sections(document_id, index_generation_id) + where index_generation_id is not null; + +-- ------------------------------------------------------------------------- +-- Step 4: New overload is_committed_artifact_generation(uuid, jsonb) +-- Mirrors is_committed_document_generation(uuid, jsonb). +-- Returns true if artifact_generation_id is null (uncommitted/legacy) +-- OR if it matches the document's committed generation. +-- ------------------------------------------------------------------------- + +create or replace function public.is_committed_artifact_generation( + artifact_generation_id uuid, + document_metadata jsonb +) +returns boolean +language sql +stable +set search_path = public, extensions, pg_temp +as $$ + select artifact_generation_id is null + or artifact_generation_id::text = + nullif(coalesce(document_metadata, '{}'::jsonb)->>'index_generation_id', ''); +$$; + +revoke execute on function public.is_committed_artifact_generation(uuid, jsonb) from public, anon, authenticated; +grant execute on function public.is_committed_artifact_generation(uuid, jsonb) to service_role; + +-- ------------------------------------------------------------------------- +-- Step 5: Update commit_document_index_generation to use typed columns +-- ------------------------------------------------------------------------- + +create or replace function public.commit_document_index_generation( + p_document_id uuid, + p_index_generation_id uuid, + p_status text default 'indexed', + p_page_count integer default 0, + p_chunk_count integer default 0, + p_image_count integer default 0, + p_metadata jsonb default '{}'::jsonb, + p_pages jsonb default null, + p_quality jsonb default null +) +returns jsonb +language plpgsql +set search_path = public, extensions, pg_temp +as $$ +begin + perform set_config('statement_timeout', '180000', true); + + update public.documents + set + status = p_status, + page_count = p_page_count, + chunk_count = p_chunk_count, + image_count = p_image_count, + error_message = null, + metadata = coalesce(p_metadata, '{}'::jsonb) || jsonb_build_object('index_generation_id', p_index_generation_id), + updated_at = now() + where id = p_document_id; + + if p_pages is not null then + delete from public.document_pages + where document_id = p_document_id; + + insert into public.document_pages (document_id, page_number, text, ocr_used, metadata) + select + p_document_id, + page_row.page_number, + coalesce(page_row.text, ''), + coalesce(page_row.ocr_used, false), + coalesce(page_row.metadata, '{}'::jsonb) + from jsonb_to_recordset(coalesce(p_pages, '[]'::jsonb)) as page_row( + page_number integer, + text text, + ocr_used boolean, + metadata jsonb + ) + where page_row.page_number is not null; + end if; + + if p_quality is not null then + insert into public.document_index_quality ( + document_id, + owner_id, + quality_score, + extraction_quality, + metrics, + issues, + updated_at + ) + values ( + p_document_id, + nullif(p_quality->>'owner_id', '')::uuid, + coalesce((p_quality->>'quality_score')::real, 0), + coalesce(nullif(p_quality->>'extraction_quality', ''), 'unknown'), + coalesce(p_quality->'metrics', '{}'::jsonb), + coalesce( + array(select jsonb_array_elements_text(coalesce(p_quality->'issues', '[]'::jsonb))), + '{}'::text[] + ), + now() + ) + on conflict on constraint document_index_quality_pkey + do update set + owner_id = excluded.owner_id, + quality_score = excluded.quality_score, + extraction_quality = excluded.extraction_quality, + metrics = excluded.metrics, + issues = excluded.issues, + updated_at = excluded.updated_at; + end if; + + -- document_chunks: typed column (unchanged) + delete from public.document_chunks + where document_id = p_document_id + and (index_generation_id is null or index_generation_id <> p_index_generation_id); + + -- artifact tables: now use typed index_generation_id column + delete from public.document_images + where document_id = p_document_id + and (index_generation_id is null or index_generation_id <> p_index_generation_id); + + delete from public.document_table_facts + where document_id = p_document_id + and (index_generation_id is null or index_generation_id <> p_index_generation_id); + + delete from public.document_embedding_fields + where document_id = p_document_id + and (index_generation_id is null or index_generation_id <> p_index_generation_id); + + delete from public.document_index_units + where document_id = p_document_id + and (index_generation_id is null or index_generation_id <> p_index_generation_id); + + delete from public.document_memory_cards + where document_id = p_document_id + and (index_generation_id is null or index_generation_id <> p_index_generation_id); + + delete from public.document_sections + where document_id = p_document_id + and (index_generation_id is null or index_generation_id <> p_index_generation_id); + + return jsonb_build_object( + 'ok', true, + 'document_id', p_document_id, + 'index_generation_id', p_index_generation_id + ); +end; +$$; + +revoke execute on function public.commit_document_index_generation(uuid, uuid, text, integer, integer, integer, jsonb, jsonb, jsonb) from public, anon, authenticated; +grant execute on function public.commit_document_index_generation(uuid, uuid, text, integer, integer, integer, jsonb, jsonb, jsonb) to service_role; + +-- ------------------------------------------------------------------------- +-- Step 6: Update cleanup_abandoned_document_index_generations +-- ------------------------------------------------------------------------- + +create or replace function public.cleanup_abandoned_document_index_generations( + p_document_id uuid default null, + p_limit integer default 100, + p_dry_run boolean default true +) +returns jsonb +language plpgsql +set search_path = public, extensions, pg_temp +as $$ +declare + target_document_ids uuid[] := '{}'::uuid[]; + chunk_count integer := 0; + image_count integer := 0; + table_fact_count integer := 0; + embedding_field_count integer := 0; + index_unit_count integer := 0; + memory_card_count integer := 0; + section_count integer := 0; +begin + perform set_config('statement_timeout', '180000', true); + + -- Collect distinct document_ids that have stale (non-committed) artifact rows. + -- document_chunks uses its typed column; artifact tables use their new typed columns. + with candidate_documents as ( + select distinct document_id + from ( + -- document_chunks (typed index_generation_id) + select c.document_id + from public.document_chunks c + join public.documents d on d.id = c.document_id + where (p_document_id is null or c.document_id = p_document_id) + and c.index_generation_id is not null + and c.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = c.document_id + and j.status in ('pending', 'processing') + ) + union all + -- document_images (typed index_generation_id) + select a.document_id + from public.document_images a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + union all + -- document_table_facts (typed index_generation_id) + select a.document_id + from public.document_table_facts a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + union all + -- document_embedding_fields (typed index_generation_id) + select a.document_id + from public.document_embedding_fields a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + union all + -- document_index_units (typed index_generation_id) + select a.document_id + from public.document_index_units a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + union all + -- document_memory_cards (typed index_generation_id) + select a.document_id + from public.document_memory_cards a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + union all + -- document_sections (typed index_generation_id) + select a.document_id + from public.document_sections a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + ) candidates + limit least(greatest(coalesce(p_limit, 100), 1), 1000) + ) + select coalesce(array_agg(document_id), '{}'::uuid[]) + into target_document_ids + from candidate_documents; + + -- Count stale rows (typed column comparisons) + select count(*) into chunk_count + from public.document_chunks c + join public.documents d on d.id = c.document_id + where c.document_id = any(target_document_ids) + and c.index_generation_id is not null + and c.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into image_count + from public.document_images a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into table_fact_count + from public.document_table_facts a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into embedding_field_count + from public.document_embedding_fields a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into index_unit_count + from public.document_index_units a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into memory_card_count + from public.document_memory_cards a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into section_count + from public.document_sections a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + if not coalesce(p_dry_run, true) then + delete from public.document_chunks c + using public.documents d + where d.id = c.document_id + and c.document_id = any(target_document_ids) + and c.index_generation_id is not null + and c.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_images a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_table_facts a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_embedding_fields a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_index_units a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_memory_cards a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_sections a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + end if; + + return jsonb_build_object( + 'ok', true, + 'dry_run', coalesce(p_dry_run, true), + 'document_count', coalesce(array_length(target_document_ids, 1), 0), + 'document_ids', to_jsonb(target_document_ids), + 'counts', jsonb_build_object( + 'document_chunks', chunk_count, + 'document_images', image_count, + 'document_table_facts', table_fact_count, + 'document_embedding_fields', embedding_field_count, + 'document_index_units', index_unit_count, + 'document_memory_cards', memory_card_count, + 'document_sections', section_count + ) + ); +end; +$$; + +revoke execute on function public.cleanup_abandoned_document_index_generations(uuid, integer, boolean) from public, anon, authenticated; +grant execute on function public.cleanup_abandoned_document_index_generations(uuid, integer, boolean) to service_role; diff --git a/supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql b/supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql new file mode 100644 index 0000000000..851d4d06c2 --- /dev/null +++ b/supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql @@ -0,0 +1,346 @@ +-- Fix #1: Replace documents.metadata JSONB worker state with a dedicated +-- indexing_v3_agent_jobs table for the v3 enrichment pipeline. +-- +-- PROBLEM: +-- claim_indexing_v3_agent_jobs does FOR UPDATE SKIP LOCKED on the documents +-- table — the largest table in the schema — and must parse multiple JSONB +-- fields for every candidate row. At moderate scale this is a multi-second +-- sequential scan. Partial indexes on JSONB expressions help for +-- documents_indexing_v3_agent_claim_idx, but the state is still scattered +-- across ~9 JSONB keys in metadata. +-- +-- FIX: +-- 1. Create indexing_v3_agent_jobs with proper typed columns and a compound +-- index suited for SKIP LOCKED claiming. +-- 2. Seed the table from existing JSONB state for all documents that are not +-- yet completed. +-- 3. Rewrite claim_indexing_v3_agent_jobs to SELECT FOR UPDATE SKIP LOCKED +-- on the small jobs table rather than the documents table. +-- The RPC also patches documents.metadata to maintain backward +-- compatibility with any edge function code that still reads from JSONB. +-- +-- *** CRITICAL EDGE FUNCTION NOTE *** +-- The supabase/functions/indexing-v3-agent/ edge function currently writes +-- completion/failure state back to documents.metadata directly. +-- Once this migration is applied, those JSONB writes are still safe (they +-- do not break anything), but the jobs table row will NOT be updated by +-- them and will remain stuck in 'processing' until it becomes stale and is +-- re-claimed or manually updated. +-- +-- You MUST update the edge function to also call a completion/failure RPC +-- (or UPDATE indexing_v3_agent_jobs directly) after applying this migration. +-- Until then, completed jobs will be picked up again after the stale timeout +-- (p_stale_after_minutes, default 45), wasting agent cycles but not +-- corrupting data (commit_document_index_generation is idempotent). +-- +-- Recommended follow-up: +-- - Add update_indexing_v3_agent_job_status(document_id, status, error) +-- RPC (service_role-only) and call it from the edge function on +-- success/failure/backoff. +-- - Remove the documents.metadata JSONB sync from +-- claim_indexing_v3_agent_jobs once the edge function is updated. + +-- ------------------------------------------------------------------------- +-- Step 1: Create the dedicated jobs table +-- ------------------------------------------------------------------------- + +create table if not exists public.indexing_v3_agent_jobs ( + id uuid primary key default gen_random_uuid(), + document_id uuid not null references public.documents(id) on delete cascade, + -- v3 agent processing status (mirrors metadata->>'indexing_v3_agent_status') + status text not null default 'pending' + check (status in ('pending', 'processing', 'completed', 'failed', 'needs_enrichment_artifacts')), + -- enrichment pipeline status (mirrors metadata->>'enrichment_status') + enrichment_status text not null default 'pending' + check (enrichment_status in ('pending', 'processing', 'completed', 'failed', 'needs_enrichment_artifacts')), + attempt_count integer not null default 0, + max_attempts integer not null default 3, + locked_by text, + locked_at timestamptz, + next_run_at timestamptz, + version text not null default 'visual-core-v3', + last_error text, + metadata jsonb not null default '{}'::jsonb, + created_at timestamptz not null default now(), + updated_at timestamptz not null default now() +); + +-- One row per document; re-running resets the row in-place +create unique index if not exists indexing_v3_agent_jobs_document_id_idx + on public.indexing_v3_agent_jobs(document_id); + +-- Hot path for claim: eligible candidates ordered by next_run_at +create index if not exists indexing_v3_agent_jobs_claim_idx + on public.indexing_v3_agent_jobs(status, enrichment_status, next_run_at, id) + where status not in ('completed', 'needs_enrichment_artifacts'); + +-- Operational: find stale processing jobs +create index if not exists indexing_v3_agent_jobs_locked_at_idx + on public.indexing_v3_agent_jobs(locked_at) + where status = 'processing'; + +-- RLS + grants (service_role only, same as ingestion_jobs) +alter table public.indexing_v3_agent_jobs enable row level security; + +drop policy if exists "indexing v3 agent jobs service role all" on public.indexing_v3_agent_jobs; +create policy "indexing v3 agent jobs service role all" + on public.indexing_v3_agent_jobs + for all to service_role + using (true) + with check (true); + +grant select, insert, update, delete + on table public.indexing_v3_agent_jobs to service_role; + +-- ------------------------------------------------------------------------- +-- Step 2: Seed from existing JSONB state +-- Insert one row per document that has ever been touched by the +-- v3 agent (i.e., has indexing_v3_agent_status in metadata) and +-- hasn't completed. Documents with no JSONB keys are not yet +-- eligible and will get a row on their first claim. +-- ------------------------------------------------------------------------- + +insert into public.indexing_v3_agent_jobs ( + document_id, + status, + enrichment_status, + attempt_count, + max_attempts, + locked_by, + locked_at, + next_run_at, + version, + last_error, + metadata, + created_at, + updated_at +) +select + d.id, + case + when coalesce(d.metadata->>'indexing_v3_agent_status', 'pending') in + ('completed', 'needs_enrichment_artifacts', 'failed') + then coalesce(d.metadata->>'indexing_v3_agent_status', 'pending') + when coalesce(d.metadata->>'indexing_v3_agent_status', '') = 'processing' + and ( + nullif(d.metadata->>'indexing_v3_agent_locked_at', '') is null + or (d.metadata->>'indexing_v3_agent_locked_at')::timestamptz < now() - interval '2 hours' + ) + then 'pending' -- stale processing → reset to pending + else coalesce(d.metadata->>'indexing_v3_agent_status', 'pending') + end as status, + coalesce(d.metadata->>'enrichment_status', 'pending') as enrichment_status, + case + when coalesce(d.metadata->>'indexing_v3_agent_attempt_count', '') ~ '^[0-9]+$' + then (d.metadata->>'indexing_v3_agent_attempt_count')::integer + else 0 + end as attempt_count, + greatest( + case + when coalesce(d.metadata->>'indexing_v3_agent_max_attempts', '') ~ '^[0-9]+$' + then (d.metadata->>'indexing_v3_agent_max_attempts')::integer + else 3 + end, + 1 + ) as max_attempts, + nullif(d.metadata->>'indexing_v3_agent_locked_by', '') as locked_by, + case + when coalesce(d.metadata->>'indexing_v3_agent_locked_at', '') ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}' + then (d.metadata->>'indexing_v3_agent_locked_at')::timestamptz + else null + end as locked_at, + case + when coalesce(d.metadata->>'indexing_v3_agent_next_run_at', '') ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}' + then (d.metadata->>'indexing_v3_agent_next_run_at')::timestamptz + else null + end as next_run_at, + coalesce(nullif(d.metadata->>'indexing_v3_agent_version', ''), 'visual-core-v3') as version, + nullif(d.metadata->>'indexing_v3_agent_last_error', '') as last_error, + '{}'::jsonb as metadata, + coalesce(d.created_at, now()) as created_at, + coalesce(d.updated_at, now()) as updated_at +from public.documents d +where d.metadata ? 'indexing_v3_agent_status' +on conflict (document_id) do nothing; + +-- ------------------------------------------------------------------------- +-- Step 3: Rewrite claim_indexing_v3_agent_jobs +-- Uses SKIP LOCKED on the small jobs table. +-- Also patches documents.metadata for backward compatibility with +-- the existing edge function (see CRITICAL NOTE above). +-- ------------------------------------------------------------------------- + +create or replace function public.claim_indexing_v3_agent_jobs( + p_worker_id text, + p_claim_limit integer default 1, + p_stale_after_minutes integer default 45 +) +returns table ( + id uuid, + document_id uuid, + batch_id uuid, + status text, + stage text, + progress integer, + error_message text, + attempt_count integer, + max_attempts integer, + locked_at timestamptz, + locked_by text, + documents jsonb +) +language plpgsql +set search_path = public, extensions, pg_temp +as $$ +-- Dual-write compatibility note: +-- This RPC claims via the jobs table (SKIP LOCKED) and also patches +-- documents.metadata so the edge function continues to read correct +-- state. Once the edge function writes completions to this table, +-- the documents.metadata patch below should be removed. +begin + return query + with eligible_jobs as ( + select j.id, j.document_id, j.attempt_count, j.max_attempts + from public.indexing_v3_agent_jobs j + -- must join documents to confirm document.status = 'indexed' + -- and to gate on enrichment_status (also stored in the job row) + where j.status not in ('completed', 'needs_enrichment_artifacts') + and j.enrichment_status in ('pending', 'failed', 'processing') + and j.attempt_count < j.max_attempts + and coalesce(j.next_run_at, now()) <= now() + and ( + j.status <> 'processing' + or j.locked_at is null + or j.locked_at < now() - make_interval(mins => p_stale_after_minutes) + ) + order by coalesce(j.next_run_at, j.updated_at), j.id + limit greatest(p_claim_limit, 1) + for update of j skip locked + ), + claimed_jobs as ( + update public.indexing_v3_agent_jobs j + set + status = 'processing', + enrichment_status = 'processing', + locked_by = p_worker_id, + locked_at = now(), + attempt_count = e.attempt_count + 1, + last_error = null, + next_run_at = null, + updated_at = now() + from eligible_jobs e + where j.id = e.id + returning j.* + ), + -- Patch documents.metadata for backward compatibility with edge function + patched_documents as ( + update public.documents d + set + metadata = jsonb_strip_nulls( + (coalesce(d.metadata, '{}'::jsonb) + - 'indexing_v3_agent_next_run_at' + - 'indexing_v3_agent_last_error') + || jsonb_build_object( + 'indexing_v3_agent_status', 'processing', + 'indexing_v3_agent_version', cj.version, + 'indexing_v3_agent_locked_by', p_worker_id, + 'indexing_v3_agent_locked_at', cj.locked_at, + 'indexing_v3_agent_attempt_count', cj.attempt_count, + 'indexing_v3_agent_max_attempts', cj.max_attempts, + 'indexing_v3_agent_updated_at', now(), + 'enrichment_status', 'processing' + ) + ), + updated_at = now() + from claimed_jobs cj + where d.id = cj.document_id + and d.status = 'indexed' -- safety: only touch documents still eligible + returning d.*, cj.id as job_id, cj.attempt_count as job_attempt_count, + cj.max_attempts as job_max_attempts, cj.locked_at as job_locked_at + ) + select + pd.job_id as id, + pd.id as document_id, + pd.import_batch_id as batch_id, + 'processing'::text as status, + 'v3 enrichment claimed'::text as stage, + 95::integer as progress, + null::text as error_message, + pd.job_attempt_count, + pd.job_max_attempts, + pd.job_locked_at as locked_at, + p_worker_id as locked_by, + to_jsonb(pd.*) - 'job_id' - 'job_attempt_count' - 'job_max_attempts' - 'job_locked_at' as documents + from patched_documents pd; +end; +$$; + +revoke execute on function public.claim_indexing_v3_agent_jobs(text, integer, integer) from public, anon, authenticated; +grant execute on function public.claim_indexing_v3_agent_jobs(text, integer, integer) to service_role; + +-- ------------------------------------------------------------------------- +-- Step 4: Helper RPC for edge function to complete/fail a job +-- This unblocks the jobs table from being permanently stuck in +-- 'processing'. Pair with edge function update. +-- ------------------------------------------------------------------------- + +create or replace function public.update_indexing_v3_agent_job_status( + p_document_id uuid, + p_status text, -- 'completed', 'failed', 'needs_enrichment_artifacts', 'pending' + p_error text default null, + p_next_run_at timestamptz default null +) +returns jsonb +language plpgsql +set search_path = public, extensions, pg_temp +as $$ +declare + v_job_id uuid; +begin + if p_status not in ('pending', 'completed', 'failed', 'needs_enrichment_artifacts') then + raise exception 'invalid status %', p_status; + end if; + + update public.indexing_v3_agent_jobs + set + status = p_status, + enrichment_status = case + when p_status = 'completed' then 'completed' + when p_status = 'failed' then 'failed' + when p_status = 'needs_enrichment_artifacts' then 'needs_enrichment_artifacts' + else enrichment_status + end, + last_error = p_error, + next_run_at = case + when p_status = 'pending' then coalesce(p_next_run_at, now()) + else null + end, + locked_by = null, + locked_at = null, + updated_at = now() + where document_id = p_document_id + returning id into v_job_id; + + return jsonb_build_object( + 'ok', v_job_id is not null, + 'job_id', v_job_id, + 'document_id', p_document_id, + 'status', p_status + ); +end; +$$; + +revoke execute on function public.update_indexing_v3_agent_job_status(uuid, text, text, timestamptz) from public, anon, authenticated; +grant execute on function public.update_indexing_v3_agent_job_status(uuid, text, text, timestamptz) to service_role; + +-- ------------------------------------------------------------------------- +-- Step 5: Keep documents_indexing_v3_agent_claim_idx in place for now +-- since the backward-compat documents.metadata patch still writes +-- to that JSONB path. It can be dropped after the edge function +-- migration removes JSONB claim reads entirely. +-- ------------------------------------------------------------------------- +comment on index public.documents_indexing_v3_agent_claim_idx is + 'Retained for backward compatibility while edge function still writes enrichment_status / indexing_v3_agent_status to documents.metadata. Drop after edge function migration.'; + +comment on table public.indexing_v3_agent_jobs is + 'Dedicated worker-state table for the v3 indexing / enrichment agent. Replaces JSONB state in documents.metadata. claim_indexing_v3_agent_jobs uses SKIP LOCKED here; update_indexing_v3_agent_job_status completes/fails a job. See migration 20260702190000 for transition notes.'; diff --git a/supabase/schema.sql b/supabase/schema.sql index ec9b58a638..802105ca6b 100644 --- a/supabase/schema.sql +++ b/supabase/schema.sql @@ -137,6 +137,7 @@ create table if not exists public.document_images ( image_hash text, perceptual_hash text, labels text[] not null default '{}', + index_generation_id uuid, metadata jsonb not null default '{}'::jsonb, created_at timestamptz not null default now() ); @@ -214,6 +215,7 @@ create table if not exists public.document_sections ( tags text[] not null default '{}', extraction_quality text not null default 'unknown' check (extraction_quality in ('good', 'partial', 'poor', 'unknown')), + index_generation_id uuid, metadata jsonb not null default '{}'::jsonb, created_at timestamptz not null default now(), updated_at timestamptz not null default now(), @@ -244,6 +246,7 @@ create table if not exists public.document_memory_cards ( source_chunk_ids uuid[] not null default '{}', source_image_ids uuid[] not null default '{}', confidence real not null default 0.5 check (confidence >= 0 and confidence <= 1), + index_generation_id uuid, metadata jsonb not null default '{}'::jsonb, embedding extensions.vector(1536) not null, search_tsv tsvector generated always as ( @@ -293,6 +296,7 @@ create table if not exists public.document_table_facts ( threshold_value text, action text, normalized_terms text[] not null default '{}', + index_generation_id uuid, metadata jsonb not null default '{}'::jsonb, search_tsv tsvector generated always as ( to_tsvector( @@ -328,6 +332,7 @@ create table if not exists public.document_embedding_fields ( content text not null, content_hash text, embedding extensions.vector(1536) not null, + index_generation_id uuid, metadata jsonb not null default '{}'::jsonb, search_tsv tsvector generated always as (to_tsvector('english', content)) stored, created_at timestamptz not null default now() @@ -530,9 +535,9 @@ create index if not exists import_batches_owner_status_idx on public.import_batc create index if not exists documents_status_idx on public.documents(status); create index if not exists documents_owner_status_idx on public.documents(owner_id, status, created_at desc); create index if not exists documents_import_batch_idx on public.documents(import_batch_id); -create index if not exists documents_owner_hash_idx on public.documents(owner_id, content_hash); create index if not exists documents_search_idx on public.documents using gin(search_tsv); create index if not exists documents_title_search_idx on public.documents using gin(title_search_tsv); +create index if not exists documents_owner_id_covering_idx on public.documents(owner_id, id); create index if not exists documents_indexed_owner_title_idx on public.documents(owner_id, title, file_name) where status = 'indexed'; @@ -660,9 +665,6 @@ create index if not exists document_index_quality_owner_score_idx on public.document_index_quality(owner_id, quality_score, updated_at desc); create index if not exists ingestion_jobs_document_idx on public.ingestion_jobs(document_id); create index if not exists ingestion_jobs_batch_idx on public.ingestion_jobs(batch_id, status); -create index if not exists ingestion_jobs_claim_idx - on public.ingestion_jobs(status, next_run_at, created_at) - where status in ('pending', 'processing'); create index if not exists ingestion_jobs_status_next_run_idx on public.ingestion_jobs(status, next_run_at, created_at) where status in ('pending', 'processing', 'failed'); @@ -981,58 +983,48 @@ returns table ( language plpgsql set search_path = public, extensions, pg_temp as $$ +-- Dual-write compatibility note: +-- This RPC claims via the jobs table (SKIP LOCKED) and also patches +-- documents.metadata so the edge function continues to read correct +-- state. Once the edge function writes completions to this table, +-- the documents.metadata patch below should be removed. begin return query - with eligible as ( - select - d.id, - d.import_batch_id, - state.attempt_count, - state.max_attempts - from public.documents d - cross join lateral ( - select - coalesce(d.metadata->>'enrichment_status', 'pending') as enrichment_status, - coalesce(d.metadata->>'indexing_v3_agent_status', 'pending') as agent_status, - case - when coalesce(d.metadata->>'indexing_v3_agent_attempt_count', '') ~ '^[0-9]+$' - then (d.metadata->>'indexing_v3_agent_attempt_count')::integer - else 0 - end as attempt_count, - greatest( - case - when coalesce(d.metadata->>'indexing_v3_agent_max_attempts', '') ~ '^[0-9]+$' - then (d.metadata->>'indexing_v3_agent_max_attempts')::integer - else 3 - end, - 1 - ) as max_attempts, - case - when coalesce(d.metadata->>'indexing_v3_agent_locked_at', '') ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}' - then (d.metadata->>'indexing_v3_agent_locked_at')::timestamptz - else null - end as locked_at, - case - when coalesce(d.metadata->>'indexing_v3_agent_next_run_at', '') ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}' - then (d.metadata->>'indexing_v3_agent_next_run_at')::timestamptz - else null - end as next_run_at - ) state - where d.status = 'indexed' - and state.enrichment_status in ('pending', 'failed', 'processing') - and state.agent_status not in ('completed', 'needs_enrichment_artifacts') - and state.attempt_count < state.max_attempts - and coalesce(state.next_run_at, now()) <= now() + with eligible_jobs as ( + select j.id, j.document_id, j.attempt_count, j.max_attempts + from public.indexing_v3_agent_jobs j + -- must join documents to confirm document.status = 'indexed' + -- and to gate on enrichment_status (also stored in the job row) + where j.status not in ('completed', 'needs_enrichment_artifacts') + and j.enrichment_status in ('pending', 'failed', 'processing') + and j.attempt_count < j.max_attempts + and coalesce(j.next_run_at, now()) <= now() and ( - state.agent_status <> 'processing' - or state.locked_at is null - or state.locked_at < now() - make_interval(mins => p_stale_after_minutes) + j.status <> 'processing' + or j.locked_at is null + or j.locked_at < now() - make_interval(mins => p_stale_after_minutes) ) - order by coalesce(state.next_run_at, d.updated_at), d.id + order by coalesce(j.next_run_at, j.updated_at), j.id limit greatest(p_claim_limit, 1) - for update of d skip locked + for update of j skip locked ), - claimed as ( + claimed_jobs as ( + update public.indexing_v3_agent_jobs j + set + status = 'processing', + enrichment_status = 'processing', + locked_by = p_worker_id, + locked_at = now(), + attempt_count = e.attempt_count + 1, + last_error = null, + next_run_at = null, + updated_at = now() + from eligible_jobs e + where j.id = e.id + returning j.* + ), + -- Patch documents.metadata for backward compatibility with edge function + patched_documents as ( update public.documents d set metadata = jsonb_strip_nulls( @@ -1041,52 +1033,36 @@ begin - 'indexing_v3_agent_last_error') || jsonb_build_object( 'indexing_v3_agent_status', 'processing', - 'indexing_v3_agent_version', 'visual-core-v3', + 'indexing_v3_agent_version', cj.version, 'indexing_v3_agent_locked_by', p_worker_id, - 'indexing_v3_agent_locked_at', now(), - 'indexing_v3_agent_attempt_count', e.attempt_count + 1, - 'indexing_v3_agent_max_attempts', e.max_attempts, + 'indexing_v3_agent_locked_at', cj.locked_at, + 'indexing_v3_agent_attempt_count', cj.attempt_count, + 'indexing_v3_agent_max_attempts', cj.max_attempts, 'indexing_v3_agent_updated_at', now(), 'enrichment_status', 'processing' ) ), updated_at = now() - from eligible e - where d.id = e.id - returning d.*, e.attempt_count + 1 as claimed_attempt_count, e.max_attempts as claimed_max_attempts + from claimed_jobs cj + where d.id = cj.document_id + and d.status = 'indexed' -- safety: only touch documents still eligible + returning d.*, cj.id as job_id, cj.attempt_count as job_attempt_count, + cj.max_attempts as job_max_attempts, cj.locked_at as job_locked_at ) select - c.id, - c.id as document_id, - c.import_batch_id as batch_id, + pd.job_id as id, + pd.id as document_id, + pd.import_batch_id as batch_id, 'processing'::text as status, 'v3 enrichment claimed'::text as stage, 95::integer as progress, null::text as error_message, - c.claimed_attempt_count, - c.claimed_max_attempts, - (c.metadata->>'indexing_v3_agent_locked_at')::timestamptz as locked_at, - c.metadata->>'indexing_v3_agent_locked_by' as locked_by, - to_jsonb(c.*) - 'claimed_attempt_count' - 'claimed_max_attempts' as documents - from claimed c; -end; -$$; - -create or replace function public.reset_document_index(p_document_id uuid) -returns void -language plpgsql -set search_path = public, extensions, pg_temp -as $$ -begin - perform set_config('statement_timeout', '180000', true); - delete from public.document_memory_cards where document_id = p_document_id; - delete from public.document_sections where document_id = p_document_id; - delete from public.document_table_facts where document_id = p_document_id; - delete from public.document_embedding_fields where document_id = p_document_id; - delete from public.document_index_quality where document_id = p_document_id; - delete from public.document_chunks where document_id = p_document_id; - delete from public.document_images where document_id = p_document_id; - delete from public.document_pages where document_id = p_document_id; + pd.job_attempt_count, + pd.job_max_attempts, + pd.job_locked_at as locked_at, + p_worker_id as locked_by, + to_jsonb(pd.*) - 'job_id' - 'job_attempt_count' - 'job_max_attempts' - 'job_locked_at' as documents + from patched_documents pd; end; $$; @@ -1171,123 +1147,35 @@ begin updated_at = excluded.updated_at; end if; - -- M13 (audit 2026-07-01): superseded-generation rows always go; legacy - -- NULL-generation rows go only when this generation wrote replacement rows - -- into the same table (see 20260702000000_commit_generation_preserve_legacy_artifacts). - -- Guarantee scope: fully protects document_images/document_memory_cards/ - -- document_sections; chunk-anchored artifacts (table facts, embedding - -- fields, index units) cascade with their legacy chunks via - -- source_chunk_id ON DELETE CASCADE when chunks are replaced. + -- document_chunks: typed column (unchanged) delete from public.document_chunks where document_id = p_document_id - and ( - (index_generation_id is not null and index_generation_id <> p_index_generation_id) - or ( - index_generation_id is null - and exists ( - select 1 - from public.document_chunks replacement - where replacement.document_id = p_document_id - and replacement.index_generation_id = p_index_generation_id - ) - ) - ); + and (index_generation_id is null or index_generation_id <> p_index_generation_id); + -- artifact tables: now use typed index_generation_id column delete from public.document_images where document_id = p_document_id - and ( - (nullif(metadata->>'index_generation_id', '') is not null - and metadata->>'index_generation_id' <> p_index_generation_id::text) - or ( - nullif(metadata->>'index_generation_id', '') is null - and exists ( - select 1 - from public.document_images replacement - where replacement.document_id = p_document_id - and replacement.metadata->>'index_generation_id' = p_index_generation_id::text - ) - ) - ); + and (index_generation_id is null or index_generation_id <> p_index_generation_id); delete from public.document_table_facts where document_id = p_document_id - and ( - (nullif(metadata->>'index_generation_id', '') is not null - and metadata->>'index_generation_id' <> p_index_generation_id::text) - or ( - nullif(metadata->>'index_generation_id', '') is null - and exists ( - select 1 - from public.document_table_facts replacement - where replacement.document_id = p_document_id - and replacement.metadata->>'index_generation_id' = p_index_generation_id::text - ) - ) - ); + and (index_generation_id is null or index_generation_id <> p_index_generation_id); delete from public.document_embedding_fields where document_id = p_document_id - and ( - (nullif(metadata->>'index_generation_id', '') is not null - and metadata->>'index_generation_id' <> p_index_generation_id::text) - or ( - nullif(metadata->>'index_generation_id', '') is null - and exists ( - select 1 - from public.document_embedding_fields replacement - where replacement.document_id = p_document_id - and replacement.metadata->>'index_generation_id' = p_index_generation_id::text - ) - ) - ); + and (index_generation_id is null or index_generation_id <> p_index_generation_id); delete from public.document_index_units where document_id = p_document_id - and ( - (nullif(metadata->>'index_generation_id', '') is not null - and metadata->>'index_generation_id' <> p_index_generation_id::text) - or ( - nullif(metadata->>'index_generation_id', '') is null - and exists ( - select 1 - from public.document_index_units replacement - where replacement.document_id = p_document_id - and replacement.metadata->>'index_generation_id' = p_index_generation_id::text - ) - ) - ); + and (index_generation_id is null or index_generation_id <> p_index_generation_id); delete from public.document_memory_cards where document_id = p_document_id - and ( - (nullif(metadata->>'index_generation_id', '') is not null - and metadata->>'index_generation_id' <> p_index_generation_id::text) - or ( - nullif(metadata->>'index_generation_id', '') is null - and exists ( - select 1 - from public.document_memory_cards replacement - where replacement.document_id = p_document_id - and replacement.metadata->>'index_generation_id' = p_index_generation_id::text - ) - ) - ); + and (index_generation_id is null or index_generation_id <> p_index_generation_id); delete from public.document_sections where document_id = p_document_id - and ( - (nullif(metadata->>'index_generation_id', '') is not null - and metadata->>'index_generation_id' <> p_index_generation_id::text) - or ( - nullif(metadata->>'index_generation_id', '') is null - and exists ( - select 1 - from public.document_sections replacement - where replacement.document_id = p_document_id - and replacement.metadata->>'index_generation_id' = p_index_generation_id::text - ) - ) - ); + and (index_generation_id is null or index_generation_id <> p_index_generation_id); return jsonb_build_object( 'ok', true, @@ -1318,9 +1206,12 @@ declare begin perform set_config('statement_timeout', '180000', true); + -- Collect distinct document_ids that have stale (non-committed) artifact rows. + -- document_chunks uses its typed column; artifact tables use their new typed columns. with candidate_documents as ( select distinct document_id from ( + -- document_chunks (typed index_generation_id) select c.document_id from public.document_chunks c join public.documents d on d.id = c.document_id @@ -1333,72 +1224,78 @@ begin and j.status in ('pending', 'processing') ) union all + -- document_images (typed index_generation_id) select a.document_id from public.document_images a join public.documents d on d.id = a.document_id where (p_document_id is null or a.document_id = p_document_id) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') and not exists ( select 1 from public.ingestion_jobs j where j.document_id = a.document_id and j.status in ('pending', 'processing') ) union all + -- document_table_facts (typed index_generation_id) select a.document_id from public.document_table_facts a join public.documents d on d.id = a.document_id where (p_document_id is null or a.document_id = p_document_id) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') and not exists ( select 1 from public.ingestion_jobs j where j.document_id = a.document_id and j.status in ('pending', 'processing') ) union all + -- document_embedding_fields (typed index_generation_id) select a.document_id from public.document_embedding_fields a join public.documents d on d.id = a.document_id where (p_document_id is null or a.document_id = p_document_id) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') and not exists ( select 1 from public.ingestion_jobs j where j.document_id = a.document_id and j.status in ('pending', 'processing') ) union all + -- document_index_units (typed index_generation_id) select a.document_id from public.document_index_units a join public.documents d on d.id = a.document_id where (p_document_id is null or a.document_id = p_document_id) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') and not exists ( select 1 from public.ingestion_jobs j where j.document_id = a.document_id and j.status in ('pending', 'processing') ) union all + -- document_memory_cards (typed index_generation_id) select a.document_id from public.document_memory_cards a join public.documents d on d.id = a.document_id where (p_document_id is null or a.document_id = p_document_id) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') and not exists ( select 1 from public.ingestion_jobs j where j.document_id = a.document_id and j.status in ('pending', 'processing') ) union all + -- document_sections (typed index_generation_id) select a.document_id from public.document_sections a join public.documents d on d.id = a.document_id where (p_document_id is null or a.document_id = p_document_id) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') and not exists ( select 1 from public.ingestion_jobs j where j.document_id = a.document_id @@ -1411,6 +1308,7 @@ begin into target_document_ids from candidate_documents; + -- Count stale rows (typed column comparisons) select count(*) into chunk_count from public.document_chunks c join public.documents d on d.id = c.document_id @@ -1422,43 +1320,43 @@ begin from public.document_images a join public.documents d on d.id = a.document_id where a.document_id = any(target_document_ids) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); select count(*) into table_fact_count from public.document_table_facts a join public.documents d on d.id = a.document_id where a.document_id = any(target_document_ids) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); select count(*) into embedding_field_count from public.document_embedding_fields a join public.documents d on d.id = a.document_id where a.document_id = any(target_document_ids) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); select count(*) into index_unit_count from public.document_index_units a join public.documents d on d.id = a.document_id where a.document_id = any(target_document_ids) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); select count(*) into memory_card_count from public.document_memory_cards a join public.documents d on d.id = a.document_id where a.document_id = any(target_document_ids) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); select count(*) into section_count from public.document_sections a join public.documents d on d.id = a.document_id where a.document_id = any(target_document_ids) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); if not coalesce(p_dry_run, true) then delete from public.document_chunks c @@ -1472,43 +1370,43 @@ begin using public.documents d where d.id = a.document_id and a.document_id = any(target_document_ids) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); delete from public.document_table_facts a using public.documents d where d.id = a.document_id and a.document_id = any(target_document_ids) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); delete from public.document_embedding_fields a using public.documents d where d.id = a.document_id and a.document_id = any(target_document_ids) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); delete from public.document_index_units a using public.documents d where d.id = a.document_id and a.document_id = any(target_document_ids) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); delete from public.document_memory_cards a using public.documents d where d.id = a.document_id and a.document_id = any(target_document_ids) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); delete from public.document_sections a using public.documents d where d.id = a.document_id and a.document_id = any(target_document_ids) - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null - and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + and a.index_generation_id is not null + and a.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); end if; return jsonb_build_object( @@ -2579,6 +2477,43 @@ as $$ (ts_rank_cd(d.title_search_tsv, query.tsq) * 3.0) ) desc limit least(greatest(match_count * 2, 24), 96) + ), + -- Batch-fetch label metadata for all distinct document_ids in the result set. + -- One query replaces N per-row calls to document_label_metadata(). + doc_labels as ( + select + l.document_id, + coalesce( + jsonb_agg( + jsonb_build_object( + 'id', l.id, + 'document_id', l.document_id, + 'owner_id', l.owner_id, + 'label', l.label, + 'label_type', l.label_type, + 'source', l.source, + 'confidence', l.confidence, + 'metadata', l.metadata, + 'created_at', l.created_at, + 'updated_at', l.updated_at + ) + order by l.confidence desc, l.label + ), + '[]'::jsonb + ) as labels + from public.document_labels l + where l.document_id in (select distinct ranked.document_id from ranked) + group by l.document_id + ), + -- Batch-fetch summary text for all distinct document_ids in the result set. + -- One query replaces N per-row calls to document_summary_text(). + doc_summaries as ( + select distinct on (s.document_id) + s.document_id, + s.summary + from public.document_summaries s + where s.document_id in (select distinct ranked.document_id from ranked) + order by s.document_id ) select ranked.id, @@ -2592,21 +2527,23 @@ as $$ ranked.retrieval_synopsis, ranked.image_ids, ranked.source_metadata, - coalesce(public.document_label_metadata(ranked.document_id), '[]'::jsonb) as document_labels, - public.document_summary_text(ranked.document_id) as document_summary, + coalesce(doc_labels.labels, '[]'::jsonb) as document_labels, + doc_summaries.summary as document_summary, -- Text-only fallback has NO vector cosine similarity. Do not fabricate one: -- a synthetic value here was read downstream as a real semantic score and -- could label a pure keyword hit as "strong"/"moderate" evidence (>=0.64). -- Leave similarity at 0; the lexical signal lives in lexical_score. - 0::double precision as similarity, + 0::double precision as similarity, ranked.text_rank, -- Cap hybrid_score well below the 0.64 "moderate" threshold so a lexical-only -- row can order amongst its peers but can never masquerade as a moderate/strong -- cosine match when merged with vector results. - least(0.5, 0.18 + (least(ranked.text_rank, 1) * 0.3))::double precision as hybrid_score, - least(0.99, 0.4 + (least(ranked.text_rank, 1) * 0.59))::double precision as lexical_score, - public.chunk_image_metadata(ranked.image_ids) as images + least(0.5, 0.18 + (least(ranked.text_rank, 1) * 0.3))::double precision as hybrid_score, + least(0.99, 0.4 + (least(ranked.text_rank, 1) * 0.59))::double precision as lexical_score, + public.chunk_image_metadata(ranked.image_ids) as images from ranked + left join doc_labels on doc_labels.document_id = ranked.document_id + left join doc_summaries on doc_summaries.document_id = ranked.document_id order by lexical_score desc, text_rank desc limit match_count; $$; @@ -3360,6 +3297,9 @@ $$; revoke execute on function public.complete_strict_enrichment_job(uuid, uuid, text, text, text) from public, anon, authenticated; grant execute on function public.complete_strict_enrichment_job(uuid, uuid, text, text, text) to service_role; +alter database postgres + set app.indexing_v3_agent_base_url = 'https://sjrfecxgysukkwxsowpy.supabase.co'; + create or replace function public.invoke_indexing_v3_agent(p_limit integer default 1) returns bigint language plpgsql @@ -3368,7 +3308,8 @@ set search_path = public, extensions, vault, pg_temp as $$ declare v_request_id bigint; - v_secret text; + v_secret text; + v_base_url text; begin select decrypted_secret into v_secret @@ -3380,8 +3321,16 @@ begin raise exception 'indexing_v3_agent_secret is missing from Supabase Vault'; end if; + -- Prefer the GUC; fall back to the hardcoded production URL so that + -- existing deployments that have not yet set the GUC continue to work. + v_base_url := coalesce( + nullif(current_setting('app.indexing_v3_agent_base_url', true), ''), + 'https://sjrfecxgysukkwxsowpy.supabase.co' + ); + select net.http_post( - url := 'https://sjrfecxgysukkwxsowpy.supabase.co/functions/v1/indexing-v3-agent?limit=' || greatest(1, least(coalesce(p_limit, 1), 10))::text, + url := v_base_url || '/functions/v1/indexing-v3-agent?limit=' + || greatest(1, least(coalesce(p_limit, 1), 10))::text, headers := jsonb_build_object( 'Content-Type', 'application/json', 'x-indexing-agent-secret', v_secret @@ -3449,6 +3398,8 @@ revoke execute on function public.claim_indexing_v3_agent_jobs(text, integer, in grant execute on function public.claim_indexing_v3_agent_jobs(text, integer, integer) to service_role; revoke execute on function public.invoke_indexing_v3_agent(integer) from public, anon, authenticated; grant execute on function public.invoke_indexing_v3_agent(integer) to service_role; +revoke execute on function public.update_indexing_v3_agent_job_status(uuid, text, text, timestamptz) from public, anon, authenticated; +grant execute on function public.update_indexing_v3_agent_job_status(uuid, text, text, timestamptz) to service_role; grant select on table public.import_batches, @@ -3622,6 +3573,7 @@ create table if not exists public.document_index_units ( source_span jsonb, quality_score real not null default 0.7 check (quality_score >= 0 and quality_score <= 1), extraction_mode text not null default 'deterministic' check (extraction_mode in ('deterministic', 'model_heavy', 'hybrid')), + index_generation_id uuid, embedding extensions.vector(1536) not null, metadata jsonb not null default '{}'::jsonb, search_tsv tsvector generated always as (to_tsvector('english', coalesce(unit_type, '') || ' ' || coalesce(title, '') || ' ' || coalesce(content, ''))) stored, @@ -3775,7 +3727,129 @@ grant execute on function public.is_committed_document_generation(uuid, jsonb) t revoke execute on function public.is_committed_artifact_generation(jsonb, jsonb) from public, anon, authenticated; grant execute on function public.is_committed_artifact_generation(jsonb, jsonb) to service_role; +-- Typed overload: compares uuid directly against document committed id stored in metadata +create or replace function public.is_committed_artifact_generation(p_artifact_gen_id uuid, p_document_metadata jsonb) +returns boolean +language sql +stable +set search_path = public, extensions, pg_temp +as $$ + select p_artifact_gen_id is not null + and p_artifact_gen_id::text = nullif(coalesce(p_document_metadata, '{}'::jsonb)->>'index_generation_id', ''); +$$; + +revoke execute on function public.is_committed_artifact_generation(uuid, jsonb) from public, anon, authenticated; +grant execute on function public.is_committed_artifact_generation(uuid, jsonb) to service_role; + create policy "document index units owner read" on public.document_index_units for select to authenticated using ( exists (select 1 from public.documents d where d.id = document_id and d.owner_id = (select auth.uid())) ); + +-- ------------------------------------------------------------------------- +-- indexing_v3_agent_jobs: dedicated worker-state table (Finding #1) +-- Replaces JSONB claim state in documents.metadata with typed rows that +-- support SKIP LOCKED on a small, hot table instead of a full-table scan. +-- ------------------------------------------------------------------------- + +create table if not exists public.indexing_v3_agent_jobs ( + id uuid primary key default gen_random_uuid(), + document_id uuid not null references public.documents(id) on delete cascade, + -- v3 agent processing status (mirrors metadata->>'indexing_v3_agent_status') + status text not null default 'pending' + check (status in ('pending', 'processing', 'completed', 'failed', 'needs_enrichment_artifacts')), + -- enrichment pipeline status (mirrors metadata->>'enrichment_status') + enrichment_status text not null default 'pending' + check (enrichment_status in ('pending', 'processing', 'completed', 'failed', 'needs_enrichment_artifacts')), + attempt_count integer not null default 0, + max_attempts integer not null default 3, + locked_by text, + locked_at timestamptz, + next_run_at timestamptz, + version text not null default 'visual-core-v3', + last_error text, + metadata jsonb not null default '{}'::jsonb, + created_at timestamptz not null default now(), + updated_at timestamptz not null default now() +); + +-- One row per document; re-running resets the row in-place +create unique index if not exists indexing_v3_agent_jobs_document_id_idx + on public.indexing_v3_agent_jobs(document_id); + +-- Hot path for claim: eligible candidates ordered by next_run_at +create index if not exists indexing_v3_agent_jobs_claim_idx + on public.indexing_v3_agent_jobs(status, enrichment_status, next_run_at, id) + where status not in ('completed', 'needs_enrichment_artifacts'); + +-- Operational: find stale processing jobs +create index if not exists indexing_v3_agent_jobs_locked_at_idx + on public.indexing_v3_agent_jobs(locked_at) + where status = 'processing'; + +-- RLS + grants (service_role only, same as ingestion_jobs) +alter table public.indexing_v3_agent_jobs enable row level security; + +create policy "indexing v3 agent jobs service role all" + on public.indexing_v3_agent_jobs + for all to service_role + using (true) + with check (true); + +grant select, insert, update, delete + on table public.indexing_v3_agent_jobs to service_role; + +create or replace function public.update_indexing_v3_agent_job_status( + p_document_id uuid, + p_status text, -- 'completed', 'failed', 'needs_enrichment_artifacts', 'pending' + p_error text default null, + p_next_run_at timestamptz default null +) +returns jsonb +language plpgsql +set search_path = public, extensions, pg_temp +as $$ +declare + v_job_id uuid; +begin + if p_status not in ('pending', 'completed', 'failed', 'needs_enrichment_artifacts') then + raise exception 'invalid status %', p_status; + end if; + + update public.indexing_v3_agent_jobs + set + status = p_status, + enrichment_status = case + when p_status = 'completed' then 'completed' + when p_status = 'failed' then 'failed' + when p_status = 'needs_enrichment_artifacts' then 'needs_enrichment_artifacts' + else enrichment_status + end, + last_error = p_error, + next_run_at = case + when p_status = 'pending' then coalesce(p_next_run_at, now()) + else null + end, + locked_by = null, + locked_at = null, + updated_at = now() + where document_id = p_document_id + returning id into v_job_id; + + return jsonb_build_object( + 'ok', v_job_id is not null, + 'job_id', v_job_id, + 'document_id', p_document_id, + 'status', p_status + ); +end; +$$; + +revoke execute on function public.update_indexing_v3_agent_job_status(uuid, text, text, timestamptz) from public, anon, authenticated; +grant execute on function public.update_indexing_v3_agent_job_status(uuid, text, text, timestamptz) to service_role; + +comment on index public.documents_indexing_v3_agent_claim_idx is + 'Retained for backward compatibility while edge function still writes enrichment_status / indexing_v3_agent_status to documents.metadata. Drop after edge function migration.'; + +comment on table public.indexing_v3_agent_jobs is + 'Dedicated worker-state table for the v3 indexing / enrichment agent. Replaces JSONB state in documents.metadata. claim_indexing_v3_agent_jobs uses SKIP LOCKED here; update_indexing_v3_agent_job_status completes/fails a job. See migration 20260702190000 for transition notes.'; From 8dbe257bde86811f7320188190c0cf4dac7b5297 Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Thu, 2 Jul 2026 15:48:37 +0800 Subject: [PATCH 02/11] fix: sync indexing v3 agent job status updates Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- supabase/functions/indexing-v3-agent/index.ts | 30 +++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/supabase/functions/indexing-v3-agent/index.ts b/supabase/functions/indexing-v3-agent/index.ts index 254ea19a32..0b3a70df3c 100644 --- a/supabase/functions/indexing-v3-agent/index.ts +++ b/supabase/functions/indexing-v3-agent/index.ts @@ -121,6 +121,8 @@ type ChunkSectionSource = { content: string; }; +type AgentJobStatus = "pending" | "completed" | "failed" | "needs_enrichment_artifacts"; + const GENERATED_BY = "indexing-v3-agent"; const AGENT_SECRET = Deno.env.get("INDEXING_V3_AGENT_SECRET") ?? Deno.env.get("CRON_SECRET") ?? ""; const EXPECTED_EMBED_DIM = 1536; @@ -1872,6 +1874,26 @@ async function needsVisualArtifacts(job: ClaimedJob): Promise { return shouldRunVisualArtifacts(row); } +async function updateAgentJobStatus( + job: ClaimedJob, + status: AgentJobStatus, + error: string | null = null, + nextRunAt: string | null = null, +): Promise { + const rows = await sql>` + select * + from public.update_indexing_v3_agent_job_status( + ${job.document_id}::uuid, + ${status}::text, + ${error}::text, + ${nextRunAt}::timestamptz + ) + `; + if (!rows[0]?.ok) { + throw new Error(`Failed to update indexing_v3_agent_jobs status to ${status} for document ${job.document_id}`); + } +} + function logCompletionGate(job: ClaimedJob, gate: CompletionGate): void { console.log( JSON.stringify({ @@ -1919,6 +1941,12 @@ async function deferJob(job: ClaimedJob, gate: CompletionGate): Promise { updated_at = now() where id = ${job.document_id}::uuid `; + await updateAgentJobStatus( + job, + decision.status === "needs_enrichment_artifacts" ? "needs_enrichment_artifacts" : "pending", + null, + decision.status === "needs_enrichment_artifacts" ? null : decision.next_run_at, + ); } async function completeJob(job: ClaimedJob): Promise { @@ -1948,6 +1976,7 @@ async function completeJob(job: ClaimedJob): Promise { })}`, ); } + await updateAgentJobStatus(job, "completed"); } async function markJobFailure(job: ClaimedJob, message: string): Promise { @@ -1976,6 +2005,7 @@ async function markJobFailure(job: ClaimedJob, message: string): Promise Date: Thu, 2 Jul 2026 16:35:43 +0800 Subject: [PATCH 03/11] fix(schema): keep legacy NULL artifact generations visible Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- supabase/schema.sql | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/supabase/schema.sql b/supabase/schema.sql index 802105ca6b..597a26a8c3 100644 --- a/supabase/schema.sql +++ b/supabase/schema.sql @@ -3727,15 +3727,15 @@ grant execute on function public.is_committed_document_generation(uuid, jsonb) t revoke execute on function public.is_committed_artifact_generation(jsonb, jsonb) from public, anon, authenticated; grant execute on function public.is_committed_artifact_generation(jsonb, jsonb) to service_role; --- Typed overload: compares uuid directly against document committed id stored in metadata +-- Typed overload: NULL keeps legacy artifacts visible; otherwise compare to committed id create or replace function public.is_committed_artifact_generation(p_artifact_gen_id uuid, p_document_metadata jsonb) returns boolean language sql stable set search_path = public, extensions, pg_temp as $$ - select p_artifact_gen_id is not null - and p_artifact_gen_id::text = nullif(coalesce(p_document_metadata, '{}'::jsonb)->>'index_generation_id', ''); + select p_artifact_gen_id is null + or p_artifact_gen_id::text = nullif(coalesce(p_document_metadata, '{}'::jsonb)->>'index_generation_id', ''); $$; revoke execute on function public.is_committed_artifact_generation(uuid, jsonb) from public, anon, authenticated; From e8b7e04f0a3a6504d2ef2d586f3e3848179d84d3 Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Thu, 2 Jul 2026 17:23:02 +0800 Subject: [PATCH 04/11] Sync storage cleanup FK in schema snapshot Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- supabase/schema.sql | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/supabase/schema.sql b/supabase/schema.sql index 597a26a8c3..d12ff27fca 100644 --- a/supabase/schema.sql +++ b/supabase/schema.sql @@ -525,7 +525,11 @@ create table if not exists public.storage_cleanup_jobs ( metadata jsonb not null default '{}'::jsonb, completed_at timestamptz, created_at timestamptz not null default now(), - updated_at timestamptz not null default now() + updated_at timestamptz not null default now(), + constraint storage_cleanup_jobs_document_id_fkey + foreign key (document_id) + references public.documents(id) + on delete set null ); create unique index if not exists documents_owner_content_hash_unique_idx From 07fea452121aa9646dfc6c766859c4e351579ba8 Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Thu, 2 Jul 2026 17:26:08 +0800 Subject: [PATCH 05/11] Fix schema grant order for indexing helper Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- supabase/schema.sql | 2 -- 1 file changed, 2 deletions(-) diff --git a/supabase/schema.sql b/supabase/schema.sql index d12ff27fca..c9a23881b0 100644 --- a/supabase/schema.sql +++ b/supabase/schema.sql @@ -3402,8 +3402,6 @@ revoke execute on function public.claim_indexing_v3_agent_jobs(text, integer, in grant execute on function public.claim_indexing_v3_agent_jobs(text, integer, integer) to service_role; revoke execute on function public.invoke_indexing_v3_agent(integer) from public, anon, authenticated; grant execute on function public.invoke_indexing_v3_agent(integer) to service_role; -revoke execute on function public.update_indexing_v3_agent_job_status(uuid, text, text, timestamptz) from public, anon, authenticated; -grant execute on function public.update_indexing_v3_agent_job_status(uuid, text, text, timestamptz) to service_role; grant select on table public.import_batches, From 9dbda7ef20676c3d20162a1d517c584b6469badf Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Thu, 2 Jul 2026 17:30:18 +0800 Subject: [PATCH 06/11] fix(migration): gate job claims by document status Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../20260702190000_indexing_v3_agent_jobs_table.sql | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql b/supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql index 851d4d06c2..4c288efe96 100644 --- a/supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql +++ b/supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql @@ -202,8 +202,9 @@ begin with eligible_jobs as ( select j.id, j.document_id, j.attempt_count, j.max_attempts from public.indexing_v3_agent_jobs j - -- must join documents to confirm document.status = 'indexed' - -- and to gate on enrichment_status (also stored in the job row) + join public.documents d + on d.id = j.document_id + and d.status = 'indexed' where j.status not in ('completed', 'needs_enrichment_artifacts') and j.enrichment_status in ('pending', 'failed', 'processing') and j.attempt_count < j.max_attempts From fcba4693ec604fd6a2ffb2a1a107b95612d4dbbb Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Thu, 2 Jul 2026 17:41:05 +0800 Subject: [PATCH 07/11] fix(migration): preserve legacy generation rows on commit Restore replacement-exists guards for NULL generation artifacts so last good chunks/artifacts remain if a generation commits without replacement rows. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- ...00_promote_index_generation_id_columns.sql | 93 +++++++++++++++++-- 1 file changed, 85 insertions(+), 8 deletions(-) diff --git a/supabase/migrations/20260702180000_promote_index_generation_id_columns.sql b/supabase/migrations/20260702180000_promote_index_generation_id_columns.sql index 35f9fc719e..d29109f852 100644 --- a/supabase/migrations/20260702180000_promote_index_generation_id_columns.sql +++ b/supabase/migrations/20260702180000_promote_index_generation_id_columns.sql @@ -216,35 +216,112 @@ begin updated_at = excluded.updated_at; end if; - -- document_chunks: typed column (unchanged) + -- Preserve legacy NULL-generation rows unless this generation wrote replacements. delete from public.document_chunks where document_id = p_document_id - and (index_generation_id is null or index_generation_id <> p_index_generation_id); + and ( + (index_generation_id is not null and index_generation_id <> p_index_generation_id) + or ( + index_generation_id is null + and exists ( + select 1 + from public.document_chunks replacement + where replacement.document_id = p_document_id + and replacement.index_generation_id = p_index_generation_id + ) + ) + ); -- artifact tables: now use typed index_generation_id column delete from public.document_images where document_id = p_document_id - and (index_generation_id is null or index_generation_id <> p_index_generation_id); + and ( + (index_generation_id is not null and index_generation_id <> p_index_generation_id) + or ( + index_generation_id is null + and exists ( + select 1 + from public.document_images replacement + where replacement.document_id = p_document_id + and replacement.index_generation_id = p_index_generation_id + ) + ) + ); delete from public.document_table_facts where document_id = p_document_id - and (index_generation_id is null or index_generation_id <> p_index_generation_id); + and ( + (index_generation_id is not null and index_generation_id <> p_index_generation_id) + or ( + index_generation_id is null + and exists ( + select 1 + from public.document_table_facts replacement + where replacement.document_id = p_document_id + and replacement.index_generation_id = p_index_generation_id + ) + ) + ); delete from public.document_embedding_fields where document_id = p_document_id - and (index_generation_id is null or index_generation_id <> p_index_generation_id); + and ( + (index_generation_id is not null and index_generation_id <> p_index_generation_id) + or ( + index_generation_id is null + and exists ( + select 1 + from public.document_embedding_fields replacement + where replacement.document_id = p_document_id + and replacement.index_generation_id = p_index_generation_id + ) + ) + ); delete from public.document_index_units where document_id = p_document_id - and (index_generation_id is null or index_generation_id <> p_index_generation_id); + and ( + (index_generation_id is not null and index_generation_id <> p_index_generation_id) + or ( + index_generation_id is null + and exists ( + select 1 + from public.document_index_units replacement + where replacement.document_id = p_document_id + and replacement.index_generation_id = p_index_generation_id + ) + ) + ); delete from public.document_memory_cards where document_id = p_document_id - and (index_generation_id is null or index_generation_id <> p_index_generation_id); + and ( + (index_generation_id is not null and index_generation_id <> p_index_generation_id) + or ( + index_generation_id is null + and exists ( + select 1 + from public.document_memory_cards replacement + where replacement.document_id = p_document_id + and replacement.index_generation_id = p_index_generation_id + ) + ) + ); delete from public.document_sections where document_id = p_document_id - and (index_generation_id is null or index_generation_id <> p_index_generation_id); + and ( + (index_generation_id is not null and index_generation_id <> p_index_generation_id) + or ( + index_generation_id is null + and exists ( + select 1 + from public.document_sections replacement + where replacement.document_id = p_document_id + and replacement.index_generation_id = p_index_generation_id + ) + ) + ); return jsonb_build_object( 'ok', true, From d7b5c2fc08548389e11d98e7e1f797a667f78317 Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Thu, 2 Jul 2026 17:45:12 +0800 Subject: [PATCH 08/11] fix(migration): normalize legacy v3 agent statuses Map legacy JSONB statuses (deferred/retry_pending and unknown values) to pending during jobs-table seeding so inserts always satisfy the new status check constraint. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../20260702190000_indexing_v3_agent_jobs_table.sql | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql b/supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql index 4c288efe96..51bcf32a98 100644 --- a/supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql +++ b/supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql @@ -121,13 +121,16 @@ select when coalesce(d.metadata->>'indexing_v3_agent_status', 'pending') in ('completed', 'needs_enrichment_artifacts', 'failed') then coalesce(d.metadata->>'indexing_v3_agent_status', 'pending') + when coalesce(d.metadata->>'indexing_v3_agent_status', 'pending') in + ('deferred', 'retry_pending') + then 'pending' when coalesce(d.metadata->>'indexing_v3_agent_status', '') = 'processing' and ( nullif(d.metadata->>'indexing_v3_agent_locked_at', '') is null or (d.metadata->>'indexing_v3_agent_locked_at')::timestamptz < now() - interval '2 hours' ) then 'pending' -- stale processing → reset to pending - else coalesce(d.metadata->>'indexing_v3_agent_status', 'pending') + else 'pending' end as status, coalesce(d.metadata->>'enrichment_status', 'pending') as enrichment_status, case From 091a3aca7f881a32214450d5e69d787b30edfb65 Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Thu, 2 Jul 2026 17:56:48 +0800 Subject: [PATCH 09/11] Ensure v3 jobs are enqueued during claim Backfill missing indexing_v3_agent_jobs rows inside claim_indexing_v3_agent_jobs for indexed documents that are still queued via metadata status, so newly enqueued v3 work is claimable after migration. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- ...702190000_indexing_v3_agent_jobs_table.sql | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql b/supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql index 51bcf32a98..7d3f156ba0 100644 --- a/supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql +++ b/supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql @@ -201,6 +201,39 @@ as $$ -- state. Once the edge function writes completions to this table, -- the documents.metadata patch below should be removed. begin + -- Backward compatibility: old ingestion path still enqueues by writing + -- documents.metadata.indexing_v3_agent_status = 'pending'. Ensure those + -- documents get a jobs-table row before claiming. + insert into public.indexing_v3_agent_jobs ( + document_id, + status, + enrichment_status, + next_run_at, + version, + metadata, + created_at, + updated_at + ) + select + d.id, + 'pending', + coalesce(d.metadata->>'enrichment_status', 'pending') as enrichment_status, + case + when coalesce(d.metadata->>'indexing_v3_agent_next_run_at', '') ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}' + then (d.metadata->>'indexing_v3_agent_next_run_at')::timestamptz + else null + end as next_run_at, + coalesce(nullif(d.metadata->>'indexing_v3_agent_version', ''), 'visual-core-v3') as version, + '{}'::jsonb as metadata, + coalesce(d.created_at, now()) as created_at, + now() as updated_at + from public.documents d + where d.status = 'indexed' + and d.metadata ? 'indexing_v3_agent_status' + and coalesce(d.metadata->>'indexing_v3_agent_status', 'pending') + not in ('completed', 'needs_enrichment_artifacts') + on conflict (document_id) do nothing; + return query with eligible_jobs as ( select j.id, j.document_id, j.attempt_count, j.max_attempts From 6843db872d72d273e9df28fd9063db56b1566d2d Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 2 Jul 2026 10:04:30 +0000 Subject: [PATCH 10/11] Fix outdated indexing schema assertions --- tests/supabase-schema.test.ts | 32 ++++++++++++++++++++++++++------ 1 file changed, 26 insertions(+), 6 deletions(-) diff --git a/tests/supabase-schema.test.ts b/tests/supabase-schema.test.ts index 226b0852b8..64483aed04 100644 --- a/tests/supabase-schema.test.ts +++ b/tests/supabase-schema.test.ts @@ -50,6 +50,10 @@ const auditLogsServiceRolePolicyMigration = readFileSync( new URL("../supabase/migrations/20260630090000_audit_logs_service_role_policy.sql", import.meta.url), "utf8", ).replace(/\s+/g, " "); +const indexingV3AgentJobsMigration = readFileSync( + new URL("../supabase/migrations/20260702190000_indexing_v3_agent_jobs_table.sql", import.meta.url), + "utf8", +).replace(/\s+/g, " "); function extractTextChunkFunction(sql: string) { const start = sql.indexOf("function public.match_document_chunks_text"); @@ -196,18 +200,31 @@ describe("Supabase schema Data API grants", () => { ); expect(schema).toContain("drop index if exists public.ingestion_job_stages_doc_idx"); expect(schema).toContain("create index if not exists ingestion_job_stages_document_started_idx"); - expect(schema).toContain("create or replace function public.claim_indexing_v3_agent_jobs"); - expect(schema).toContain("where d.status = 'indexed'"); - expect(schema).toContain("state.enrichment_status in ('pending', 'failed', 'processing')"); - expect(schema).toContain("'indexing_v3_agent_locked_by', p_worker_id"); - expect(schema).toContain("'indexing_v3_agent_attempt_count', e.attempt_count + 1"); + for (const sql of [schema, indexingV3AgentJobsMigration]) { + expect(sql).toContain("create table if not exists public.indexing_v3_agent_jobs"); + expect(sql).toContain("document_id uuid not null references public.documents(id) on delete cascade"); + expect(sql).toContain("create index if not exists indexing_v3_agent_jobs_claim_idx"); + expect(sql).toContain("create or replace function public.claim_indexing_v3_agent_jobs"); + expect(sql).toContain("from public.indexing_v3_agent_jobs j"); + expect(sql).toContain("j.enrichment_status in ('pending', 'failed', 'processing')"); + expect(sql).toContain("update public.indexing_v3_agent_jobs j"); + expect(sql).toContain("'indexing_v3_agent_locked_by', p_worker_id"); + expect(sql).toContain("'indexing_v3_agent_attempt_count', cj.attempt_count"); + expect(sql).toContain("create or replace function public.update_indexing_v3_agent_job_status"); + expect(sql).toContain( + "grant execute on function public.update_indexing_v3_agent_job_status(uuid, text, text, timestamptz) to service_role", + ); + } expect(schema).toContain( "grant execute on function public.claim_indexing_v3_agent_jobs(text, integer, integer) to service_role", ); expect(schema).toContain("alter table public.ingestion_job_stages enable row level security"); expect(schema).toContain('create policy "ingestion job stages service role all" on public.ingestion_job_stages'); + expect(schema).toContain("alter table public.indexing_v3_agent_jobs enable row level security"); + expect(schema).toContain('create policy "indexing v3 agent jobs service role all" on public.indexing_v3_agent_jobs'); const authenticatedSelectGrant = schema.match(/grant select on table ([^;]+) to authenticated;/)?.[1] ?? ""; expect(authenticatedSelectGrant).not.toContain("public.ingestion_job_stages"); + expect(authenticatedSelectGrant).not.toContain("public.indexing_v3_agent_jobs"); }); it("keeps the cron indexing-v3 invoker in the schema snapshot with service-role-only execute grants", () => { @@ -217,8 +234,11 @@ describe("Supabase schema Data API grants", () => { expect(schema).toContain("set search_path = public, extensions, vault, pg_temp"); expect(schema).toContain("from vault.decrypted_secrets"); expect(schema).toContain("where name = 'indexing_v3_agent_secret'"); + expect(schema).toContain("set app.indexing_v3_agent_base_url = 'https://sjrfecxgysukkwxsowpy.supabase.co';"); + expect(schema).toContain("nullif(current_setting('app.indexing_v3_agent_base_url', true), '')"); expect(schema).toContain("select net.http_post("); - expect(schema).toContain("https://sjrfecxgysukkwxsowpy.supabase.co/functions/v1/indexing-v3-agent?limit="); + expect(schema).toContain("v_base_url || '/functions/v1/indexing-v3-agent?limit='"); + expect(schema).toContain("'https://sjrfecxgysukkwxsowpy.supabase.co'"); expect(schema).toContain( "revoke execute on function public.invoke_indexing_v3_agent(integer) from public, anon, authenticated", ); From 9ae167406ccefa35b4fd98c0b2af9f2982a02602 Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Thu, 2 Jul 2026 18:12:57 +0800 Subject: [PATCH 11/11] fix(migration): fall back to metadata gen for null-typed artifact rows in commit The artifact writer still writes index_generation_id only into JSONB metadata, leaving the new typed column NULL. The original EXISTS guard only checked for typed-column replacements, so it always returned false for artifact tables, meaning stale null-typed rows from a previous run could never be purged on re-index (they accumulate silently). For the null-typed-column branch of the six artifact DELETE statements in commit_document_index_generation: - Add (metadata->>''index_generation_id'')::uuid IS DISTINCT FROM p_index_generation_id so that rows belonging to the *current* generation (metadata gen = p_index_generation_id) are excluded from deletion. - Expand the EXISTS replacement check to accept both typed-column matches and metadata-based matches, so the safety guard fires correctly even when the writer only populates metadata. document_chunks is unchanged: the worker already writes its typed column. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .impeccable/hook.cache.json | 1 + ...00_promote_index_generation_id_columns.sql | 48 ++++++++++++++++--- 2 files changed, 42 insertions(+), 7 deletions(-) create mode 100644 .impeccable/hook.cache.json diff --git a/.impeccable/hook.cache.json b/.impeccable/hook.cache.json new file mode 100644 index 0000000000..a2d4e16581 --- /dev/null +++ b/.impeccable/hook.cache.json @@ -0,0 +1 @@ +{"version":1,"sessions":{"70c96a73-49a6-4422-a310-d0294a45dc49":{"updatedAt":1782977742215,"files":{"C:\\Users\\joshs\\.copilot\\repos\\copilot-worktrees\\Database\\bigsimmo-bookish-barnacle\\supabase\\functions\\indexing-v3-agent\\index.ts":{"editCount":1,"findings":[]}}}}} \ No newline at end of file diff --git a/supabase/migrations/20260702180000_promote_index_generation_id_columns.sql b/supabase/migrations/20260702180000_promote_index_generation_id_columns.sql index d29109f852..bef99a6d74 100644 --- a/supabase/migrations/20260702180000_promote_index_generation_id_columns.sql +++ b/supabase/migrations/20260702180000_promote_index_generation_id_columns.sql @@ -232,18 +232,27 @@ begin ) ); - -- artifact tables: now use typed index_generation_id column + -- artifact tables: use typed column where set; fall back to metadata when typed is NULL + -- because the writer still populates metadata.index_generation_id rather than the typed + -- column. Without the metadata fallback, stale null-typed rows from a prior run would + -- never be cleaned up (the typed-column EXISTS guard would always be false), allowing + -- artifact rows to accumulate across re-indexes. delete from public.document_images where document_id = p_document_id and ( (index_generation_id is not null and index_generation_id <> p_index_generation_id) or ( index_generation_id is null + and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id and exists ( select 1 from public.document_images replacement where replacement.document_id = p_document_id - and replacement.index_generation_id = p_index_generation_id + and ( + replacement.index_generation_id = p_index_generation_id + or (replacement.index_generation_id is null + and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id) + ) ) ) ); @@ -254,11 +263,16 @@ begin (index_generation_id is not null and index_generation_id <> p_index_generation_id) or ( index_generation_id is null + and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id and exists ( select 1 from public.document_table_facts replacement where replacement.document_id = p_document_id - and replacement.index_generation_id = p_index_generation_id + and ( + replacement.index_generation_id = p_index_generation_id + or (replacement.index_generation_id is null + and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id) + ) ) ) ); @@ -269,11 +283,16 @@ begin (index_generation_id is not null and index_generation_id <> p_index_generation_id) or ( index_generation_id is null + and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id and exists ( select 1 from public.document_embedding_fields replacement where replacement.document_id = p_document_id - and replacement.index_generation_id = p_index_generation_id + and ( + replacement.index_generation_id = p_index_generation_id + or (replacement.index_generation_id is null + and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id) + ) ) ) ); @@ -284,11 +303,16 @@ begin (index_generation_id is not null and index_generation_id <> p_index_generation_id) or ( index_generation_id is null + and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id and exists ( select 1 from public.document_index_units replacement where replacement.document_id = p_document_id - and replacement.index_generation_id = p_index_generation_id + and ( + replacement.index_generation_id = p_index_generation_id + or (replacement.index_generation_id is null + and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id) + ) ) ) ); @@ -299,11 +323,16 @@ begin (index_generation_id is not null and index_generation_id <> p_index_generation_id) or ( index_generation_id is null + and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id and exists ( select 1 from public.document_memory_cards replacement where replacement.document_id = p_document_id - and replacement.index_generation_id = p_index_generation_id + and ( + replacement.index_generation_id = p_index_generation_id + or (replacement.index_generation_id is null + and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id) + ) ) ) ); @@ -314,11 +343,16 @@ begin (index_generation_id is not null and index_generation_id <> p_index_generation_id) or ( index_generation_id is null + and (metadata->>'index_generation_id')::uuid is distinct from p_index_generation_id and exists ( select 1 from public.document_sections replacement where replacement.document_id = p_document_id - and replacement.index_generation_id = p_index_generation_id + and ( + replacement.index_generation_id = p_index_generation_id + or (replacement.index_generation_id is null + and (replacement.metadata->>'index_generation_id')::uuid = p_index_generation_id) + ) ) ) );