diff --git a/.github/workflows/ingestion-catalog.yml b/.github/workflows/ingestion-catalog.yml new file mode 100644 index 0000000..51e2453 --- /dev/null +++ b/.github/workflows/ingestion-catalog.yml @@ -0,0 +1,37 @@ +name: Organization ingestion catalog +on: + pull_request: + branches: [main] + paths: + - "supabase/migrations/*_ingestion_catalog*.sql" + - "tests/test_ingestion_catalog.py" + - ".github/workflows/ingestion-catalog.yml" + push: + branches: [main] + paths: + - "supabase/migrations/*_ingestion_catalog*.sql" + - "tests/test_ingestion_catalog.py" + - ".github/workflows/ingestion-catalog.yml" +permissions: + contents: read +jobs: + ingestion-catalog: + runs-on: ubuntu-24.04 + timeout-minutes: 10 + strategy: + fail-fast: false + matrix: + postgres: [15, 17] + steps: + - uses: actions/checkout@v4 + - name: Install PostgreSQL + run: | + sudo apt-get update + sudo apt-get install -y postgresql-common + sudo /usr/share/postgresql-common/pgdg/apt.postgresql.org.sh -y -v ${{ matrix.postgres }} + sudo apt-get update + sudo apt-get install -y postgresql-${{ matrix.postgres }} + - name: Verify shared organization data and access + run: | + export PATH="/usr/lib/postgresql/${{ matrix.postgres }}/bin:$PATH" + python3 -m unittest discover -s tests -p 'test_ingestion_catalog.py' diff --git a/README.md b/README.md index 0007f8f..743af6a 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,10 @@ Supabase migration scripts for the Recoup project. +## Organization ingestion datasets + +The additive ingestion catalog stores organization-owned datasets, import history and validated gold rows in the existing database. See [schema, tests and rollout](tests/INGESTION-CATALOG.md). Original files remain in private Storage; no per-client database is provisioned. + ## Scheduled posts migration `20261015230000_create_scheduled_posts.sql` adds `scheduled_posts` for `POST /api/posts`; see [tests/SCHEDULED-POSTS.md](tests/SCHEDULED-POSTS.md). diff --git a/supabase/migrations/20261110010000_ingestion_catalog.sql b/supabase/migrations/20261110010000_ingestion_catalog.sql new file mode 100644 index 0000000..d91db0b --- /dev/null +++ b/supabase/migrations/20261110010000_ingestion_catalog.sql @@ -0,0 +1,196 @@ +-- Shared organization datasets. Originals and processing artifacts remain in private Storage. +BEGIN; +CREATE TABLE public.ingestion_datasets ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + organization_id uuid NOT NULL REFERENCES public.accounts(id), + name text NOT NULL CHECK (length(name) BETWEEN 1 AND 200), + created_by uuid NOT NULL REFERENCES public.accounts(id), + created_at timestamptz NOT NULL DEFAULT clock_timestamp(), + UNIQUE (id, organization_id) +); +CREATE INDEX ingestion_datasets_org_created ON public.ingestion_datasets(organization_id, created_at DESC, id DESC); +CREATE INDEX ingestion_datasets_creator ON public.ingestion_datasets(created_by); +CREATE TABLE public.ingestion_imports ( + organization_id uuid NOT NULL, + job_id text NOT NULL CHECK (job_id ~ '^[a-f0-9]{64}$'), + dataset_id uuid NOT NULL, + actor_id uuid NOT NULL REFERENCES public.accounts(id), + manifest jsonb NOT NULL CHECK (jsonb_typeof(manifest)='object' AND octet_length(manifest::text)<=1048576), + status text NOT NULL DEFAULT 'queued' CHECK (status IN ('queued','ready','held','rejected')), + result jsonb, + field_types jsonb, + completion_sha256 text, + created_at timestamptz NOT NULL DEFAULT clock_timestamp(), + completed_at timestamptz, + PRIMARY KEY (organization_id,job_id), + UNIQUE (organization_id,job_id,dataset_id), + FOREIGN KEY (dataset_id,organization_id) REFERENCES public.ingestion_datasets(id,organization_id), + CHECK ((status='queued' AND result IS NULL AND completed_at IS NULL AND completion_sha256 IS NULL) + OR (status<>'queued' AND result IS NOT NULL AND completed_at IS NOT NULL AND completion_sha256 IS NOT NULL)) +); +CREATE INDEX ingestion_imports_dataset_created ON public.ingestion_imports(organization_id,dataset_id,created_at DESC,job_id DESC); +CREATE INDEX ingestion_imports_actor ON public.ingestion_imports(actor_id); +CREATE TABLE public.ingestion_records ( + organization_id uuid NOT NULL, + job_id text NOT NULL, + dataset_id uuid NOT NULL, + row_number integer NOT NULL CHECK (row_number BETWEEN 1 AND 10000), + record jsonb NOT NULL CHECK (jsonb_typeof(record)='object'), + PRIMARY KEY (organization_id,job_id,row_number), + FOREIGN KEY (organization_id,job_id,dataset_id) REFERENCES public.ingestion_imports(organization_id,job_id,dataset_id) +); +-- The database gateway is the only access path, including for service_role. +ALTER TABLE public.ingestion_datasets ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.ingestion_imports ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.ingestion_records ENABLE ROW LEVEL SECURITY; +REVOKE ALL ON public.ingestion_datasets,public.ingestion_imports,public.ingestion_records FROM PUBLIC,anon,authenticated,service_role; + +CREATE FUNCTION public.authorize_ingestion_organization(p_actor uuid,p_org uuid) +RETURNS void LANGUAGE plpgsql SECURITY DEFINER SET search_path='' AS $$ +BEGIN + IF p_actor IS NULL OR p_org IS NULL THEN RAISE EXCEPTION 'Ingestion access denied' USING ERRCODE='42501'; END IF; + IF NOT EXISTS(SELECT 1 FROM public.accounts WHERE id=p_actor) OR NOT EXISTS(SELECT 1 FROM public.accounts WHERE id=p_org) THEN + RAISE EXCEPTION 'Ingestion access denied' USING ERRCODE='42501'; + END IF; + IF p_actor<>p_org THEN + PERFORM 1 FROM public.account_organization_ids WHERE account_id=p_actor AND organization_id=p_org FOR SHARE; + IF NOT FOUND THEN RAISE EXCEPTION 'Ingestion access denied' USING ERRCODE='42501'; END IF; + END IF; +END $$; +REVOKE ALL ON FUNCTION public.authorize_ingestion_organization(uuid,uuid) FROM PUBLIC,anon,authenticated,service_role; + +CREATE FUNCTION public.ingestion_catalog_command(p_actor uuid,p_org uuid,p_action text,p_input jsonb DEFAULT '{}') +RETURNS jsonb LANGUAGE plpgsql SECURITY DEFINER SET search_path='' AS $$ +DECLARE + v_manifest jsonb; v_job text; v_dataset uuid; v_import public.ingestion_imports; + v_result jsonb; v_gold jsonb; v_status text; v_fingerprint text; v_count integer; + v_limit integer; v_after integer; v_items jsonb; v_cursor_time timestamptz; v_cursor_id uuid; v_cursor_job text; +BEGIN + PERFORM public.authorize_ingestion_organization(p_actor,p_org); + IF p_input IS NULL OR jsonb_typeof(p_input)<>'object' OR octet_length(p_input::text)>25165824 THEN + RAISE EXCEPTION 'Invalid ingestion catalog input' USING ERRCODE='22023'; + END IF; + IF p_action='register' THEN + v_manifest:=p_input->'manifest'; v_job:=v_manifest->>'job_id'; + IF v_manifest IS NULL OR jsonb_typeof(v_manifest)<>'object' OR octet_length(v_manifest::text)>1048576 + OR v_job IS NULL OR v_job !~ '^[a-f0-9]{64}$' + OR (v_manifest->>'scope') IS DISTINCT FROM ('organization/'||p_org::text) + OR coalesce(v_manifest->>'input_sha256','') !~ '^[a-f0-9]{64}$' + OR coalesce(v_manifest->>'bronze_sha256','') !~ '^[a-f0-9]{64}$' + OR (v_manifest->>'bronze_key') IS DISTINCT FROM ('organization/'||p_org::text||'/originals/'||(v_manifest->>'bronze_sha256')) + OR coalesce(v_manifest->>'format','') NOT IN ('csv','tsv','json','jsonl','xlsx','pdf') + OR coalesce(length(v_manifest->>'filename'),0) NOT BETWEEN 1 AND 200 + OR coalesce(jsonb_typeof(v_manifest->'contract'),'')<>'object' + THEN RAISE EXCEPTION 'Invalid ingestion catalog input' USING ERRCODE='22023'; END IF; + PERFORM pg_catalog.pg_advisory_xact_lock(pg_catalog.hashtextextended('ingestion:'||p_org::text||':'||v_job,0)); + SELECT * INTO v_import FROM public.ingestion_imports WHERE organization_id=p_org AND job_id=v_job FOR UPDATE; + IF FOUND THEN + IF v_import.manifest<>v_manifest THEN RAISE EXCEPTION 'Ingestion catalog conflict'; END IF; + RETURN jsonb_build_object('dataset_id',v_import.dataset_id,'job_id',v_job); + END IF; + v_dataset:=(v_manifest->>'dataset_id')::uuid; + IF v_dataset IS NULL THEN + INSERT INTO public.ingestion_datasets(organization_id,name,created_by) VALUES(p_org,v_manifest->>'filename',p_actor) RETURNING id INTO v_dataset; + ELSE + PERFORM 1 FROM public.ingestion_datasets WHERE id=v_dataset AND organization_id=p_org FOR SHARE; + IF NOT FOUND THEN RAISE EXCEPTION 'Ingestion access denied' USING ERRCODE='42501'; END IF; + END IF; + INSERT INTO public.ingestion_imports(organization_id,job_id,dataset_id,actor_id,manifest) VALUES(p_org,v_job,v_dataset,p_actor,v_manifest); + RETURN jsonb_build_object('dataset_id',v_dataset,'job_id',v_job); + ELSIF p_action='complete' THEN + v_job:=p_input->>'job_id'; v_result:=p_input->'result'; v_gold:=nullif(p_input->'gold','null'::jsonb); v_status:=v_result->>'status'; + SELECT * INTO v_import FROM public.ingestion_imports WHERE organization_id=p_org AND job_id=v_job FOR UPDATE; + IF NOT FOUND THEN RAISE EXCEPTION 'Ingestion access denied' USING ERRCODE='42501'; END IF; + IF v_result IS NULL OR jsonb_typeof(v_result)<>'object' OR v_status IS NULL OR v_status NOT IN ('ready','held','rejected') + OR (v_result->>'job_id') IS DISTINCT FROM v_job + OR (v_result->>'source_sha256') IS DISTINCT FROM (v_import.manifest->>'bronze_sha256') + OR (v_result->>'engine_version') IS DISTINCT FROM (v_import.manifest->>'engine_version') + OR coalesce(v_result->>'silver_sha256','') !~ '^[a-f0-9]{64}$' + OR coalesce(v_result->>'rows_read','') !~ '^\d{1,5}$' OR (v_result->>'rows_read')::integer>10000 + OR coalesce(v_result->>'rows_output','') !~ '^\d{1,5}$' OR (v_result->>'rows_output')::integer>10000 + THEN RAISE EXCEPTION 'Invalid ingestion catalog input' USING ERRCODE='22023'; END IF; + IF v_status='ready' THEN + IF v_gold IS NULL OR jsonb_typeof(v_gold)<>'object' OR coalesce(jsonb_typeof(v_gold->'records'),'')<>'array' + OR coalesce(jsonb_typeof(v_gold->'field_types'),'')<>'object' + OR (v_gold->>'engine_version') IS DISTINCT FROM (v_import.manifest->>'engine_version') + OR (v_gold->>'contract_version') IS DISTINCT FROM (v_import.manifest->'contract'->>'version') + OR coalesce(v_result->>'gold_sha256','') !~ '^[a-f0-9]{64}$' + THEN RAISE EXCEPTION 'Invalid ingestion gold' USING ERRCODE='22023'; END IF; + v_count:=jsonb_array_length(v_gold->'records'); + IF v_count NOT BETWEEN 1 AND 10000 OR v_count<>(v_result->>'rows_output')::integer + OR (SELECT count(*) FROM jsonb_object_keys(v_gold->'field_types')) NOT BETWEEN 1 AND 100 + OR EXISTS(SELECT 1 FROM jsonb_each(v_gold->'field_types') f WHERE jsonb_typeof(f.value)<>'string' OR f.value#>>'{}' NOT IN ('string','decimal','integer','date','boolean')) + THEN RAISE EXCEPTION 'Invalid ingestion gold' USING ERRCODE='22023'; END IF; + IF EXISTS(SELECT 1 FROM jsonb_array_elements(v_gold->'records') r WHERE jsonb_typeof(r)<>'object') THEN + RAISE EXCEPTION 'Invalid ingestion gold' USING ERRCODE='22023'; + END IF; + IF EXISTS(SELECT 1 FROM jsonb_array_elements(v_gold->'records') r WHERE + (SELECT count(*) FROM jsonb_object_keys(r))<>(SELECT count(*) FROM jsonb_object_keys(v_gold->'field_types')) + OR EXISTS(SELECT 1 FROM jsonb_each(r) f WHERE NOT (v_gold->'field_types' ? f.key) OR jsonb_typeof(f.value) NOT IN ('string','null'))) + THEN RAISE EXCEPTION 'Invalid ingestion gold' USING ERRCODE='22023'; END IF; + ELSE + v_count:=0; + IF v_gold IS NOT NULL OR (v_result->>'gold_sha256') IS NOT NULL OR (v_result->>'rows_output')::integer<>0 THEN + RAISE EXCEPTION 'Invalid ingestion gold' USING ERRCODE='22023'; + END IF; + END IF; + v_fingerprint:=pg_catalog.encode(pg_catalog.sha256(pg_catalog.convert_to(jsonb_build_object('result',v_result,'gold',v_gold)::text,'UTF8')),'hex'); + IF v_import.status<>'queued' THEN + IF v_import.completion_sha256<>v_fingerprint THEN RAISE EXCEPTION 'Ingestion catalog conflict'; END IF; + RETURN jsonb_build_object('dataset_id',v_import.dataset_id,'job_id',v_job,'status',v_import.status,'rows',v_count); + END IF; + INSERT INTO public.ingestion_records(organization_id,job_id,dataset_id,row_number,record) + SELECT p_org,v_job,v_import.dataset_id,n::integer,r FROM jsonb_array_elements(v_gold->'records') WITH ORDINALITY AS rows(r,n); + UPDATE public.ingestion_imports SET status=v_status,result=v_result,field_types=v_gold->'field_types',completion_sha256=v_fingerprint,completed_at=clock_timestamp() + WHERE organization_id=p_org AND job_id=v_job; + RETURN jsonb_build_object('dataset_id',v_import.dataset_id,'job_id',v_job,'status',v_status,'rows',v_count); + ELSIF p_action IN ('datasets','imports','records') THEN + v_limit:=coalesce((p_input->>'limit')::integer,25); + IF v_limit NOT BETWEEN 1 AND (CASE WHEN p_action='records' THEN 100 ELSE 50 END) THEN RAISE EXCEPTION 'Invalid ingestion page' USING ERRCODE='22023'; END IF; + IF p_action='datasets' THEN + v_cursor_id:=(p_input->>'after')::uuid; + IF v_cursor_id IS NOT NULL THEN + SELECT created_at INTO v_cursor_time FROM public.ingestion_datasets WHERE id=v_cursor_id AND organization_id=p_org; + IF NOT FOUND THEN RAISE EXCEPTION 'Ingestion access denied' USING ERRCODE='42501'; END IF; + END IF; + SELECT coalesce(jsonb_agg(item ORDER BY created_at DESC,id DESC),'[]'::jsonb) INTO v_items FROM ( + SELECT d.id,d.created_at,jsonb_build_object('dataset_id',d.id,'name',d.name,'created_at',d.created_at) item + FROM public.ingestion_datasets d WHERE d.organization_id=p_org AND (v_cursor_id IS NULL OR (d.created_at,d.id)<(v_cursor_time,v_cursor_id)) + ORDER BY d.created_at DESC,d.id DESC LIMIT v_limit+1 + ) q; + RETURN jsonb_build_object('datasets',v_items- v_limit,'next_after',CASE WHEN jsonb_array_length(v_items)>v_limit THEN v_items->(v_limit-1)->>'dataset_id' ELSE NULL END); + END IF; + v_dataset:=(p_input->>'dataset_id')::uuid; + PERFORM 1 FROM public.ingestion_datasets WHERE id=v_dataset AND organization_id=p_org; + IF NOT FOUND THEN RAISE EXCEPTION 'Ingestion access denied' USING ERRCODE='42501'; END IF; + IF p_action='imports' THEN + v_cursor_job:=p_input->>'after'; + IF v_cursor_job IS NOT NULL THEN + SELECT created_at INTO v_cursor_time FROM public.ingestion_imports WHERE organization_id=p_org AND dataset_id=v_dataset AND job_id=v_cursor_job; + IF NOT FOUND THEN RAISE EXCEPTION 'Ingestion access denied' USING ERRCODE='42501'; END IF; + END IF; + SELECT coalesce(jsonb_agg(item ORDER BY created_at DESC,job_id DESC),'[]'::jsonb) INTO v_items FROM ( + SELECT i.created_at,i.job_id,jsonb_build_object('job_id',i.job_id,'dataset_id',i.dataset_id,'status',i.status,'filename',i.manifest->>'filename', + 'format',i.manifest->>'format','source_sha256',i.manifest->>'bronze_sha256','created_at',i.created_at,'completed_at',i.completed_at, + 'rows_read',i.result->'rows_read','rows_output',i.result->'rows_output','gold_sha256',i.result->'gold_sha256') item + FROM public.ingestion_imports i WHERE i.organization_id=p_org AND i.dataset_id=v_dataset AND (v_cursor_job IS NULL OR (i.created_at,i.job_id)<(v_cursor_time,v_cursor_job)) + ORDER BY i.created_at DESC,i.job_id DESC LIMIT v_limit+1 + ) q; + RETURN jsonb_build_object('imports',v_items-v_limit,'next_after',CASE WHEN jsonb_array_length(v_items)>v_limit THEN v_items->(v_limit-1)->>'job_id' ELSE NULL END); + END IF; + v_job:=p_input->>'job_id'; v_after:=coalesce((p_input->>'after_row')::integer,0); + IF v_after NOT BETWEEN 0 AND 10000 THEN RAISE EXCEPTION 'Invalid ingestion page' USING ERRCODE='22023'; END IF; + SELECT * INTO v_import FROM public.ingestion_imports WHERE organization_id=p_org AND dataset_id=v_dataset AND job_id=v_job; + IF NOT FOUND THEN RAISE EXCEPTION 'Ingestion access denied' USING ERRCODE='42501'; END IF; + IF v_import.status<>'ready' THEN RAISE EXCEPTION 'Artifact not ready'; END IF; + SELECT coalesce(jsonb_agg(jsonb_build_object('row_number',row_number,'record',record) ORDER BY row_number),'[]'::jsonb) INTO v_items FROM ( + SELECT row_number,record FROM public.ingestion_records WHERE organization_id=p_org AND job_id=v_job AND row_number>v_after ORDER BY row_number LIMIT v_limit+1 + ) q; + RETURN jsonb_build_object('dataset_id',v_dataset,'job_id',v_job,'field_types',v_import.field_types,'gold_sha256',v_import.result->>'gold_sha256', + 'records',v_items-v_limit,'next_after',CASE WHEN jsonb_array_length(v_items)>v_limit THEN (v_items->(v_limit-1)->>'row_number')::integer ELSE NULL END); + END IF; + RAISE EXCEPTION 'Invalid ingestion catalog action' USING ERRCODE='22023'; +END $$; +REVOKE ALL ON FUNCTION public.ingestion_catalog_command(uuid,uuid,text,jsonb) FROM PUBLIC,anon,authenticated,service_role; +GRANT EXECUTE ON FUNCTION public.ingestion_catalog_command(uuid,uuid,text,jsonb) TO service_role; +COMMIT; diff --git a/supabase/migrations/20261110010100_ingestion_catalog_page_budget.sql b/supabase/migrations/20261110010100_ingestion_catalog_page_budget.sql new file mode 100644 index 0000000..78fa2ec --- /dev/null +++ b/supabase/migrations/20261110010100_ingestion_catalog_page_budget.sql @@ -0,0 +1,141 @@ +-- Forward-only correction: keep shared record pages within an inline response budget. +BEGIN; +CREATE OR REPLACE FUNCTION public.ingestion_catalog_command(p_actor uuid,p_org uuid,p_action text,p_input jsonb DEFAULT '{}') +RETURNS jsonb LANGUAGE plpgsql SECURITY DEFINER SET search_path='' AS $$ +DECLARE + v_manifest jsonb; v_job text; v_dataset uuid; v_import public.ingestion_imports; + v_result jsonb; v_gold jsonb; v_status text; v_fingerprint text; v_count integer; + v_limit integer; v_after integer; v_last integer; v_items jsonb; v_cursor_time timestamptz; v_cursor_id uuid; v_cursor_job text; +BEGIN + PERFORM public.authorize_ingestion_organization(p_actor,p_org); + IF p_input IS NULL OR jsonb_typeof(p_input)<>'object' OR octet_length(p_input::text)>25165824 THEN + RAISE EXCEPTION 'Invalid ingestion catalog input' USING ERRCODE='22023'; + END IF; + IF p_action='register' THEN + v_manifest:=p_input->'manifest'; v_job:=v_manifest->>'job_id'; + IF v_manifest IS NULL OR jsonb_typeof(v_manifest)<>'object' OR octet_length(v_manifest::text)>1048576 + OR v_job IS NULL OR v_job !~ '^[a-f0-9]{64}$' + OR (v_manifest->>'scope') IS DISTINCT FROM ('organization/'||p_org::text) + OR coalesce(v_manifest->>'input_sha256','') !~ '^[a-f0-9]{64}$' + OR coalesce(v_manifest->>'bronze_sha256','') !~ '^[a-f0-9]{64}$' + OR (v_manifest->>'bronze_key') IS DISTINCT FROM ('organization/'||p_org::text||'/originals/'||(v_manifest->>'bronze_sha256')) + OR coalesce(v_manifest->>'format','') NOT IN ('csv','tsv','json','jsonl','xlsx','pdf') + OR coalesce(length(v_manifest->>'filename'),0) NOT BETWEEN 1 AND 200 + OR coalesce(jsonb_typeof(v_manifest->'contract'),'')<>'object' + THEN RAISE EXCEPTION 'Invalid ingestion catalog input' USING ERRCODE='22023'; END IF; + PERFORM pg_catalog.pg_advisory_xact_lock(pg_catalog.hashtextextended('ingestion:'||p_org::text||':'||v_job,0)); + SELECT * INTO v_import FROM public.ingestion_imports WHERE organization_id=p_org AND job_id=v_job FOR UPDATE; + IF FOUND THEN + IF v_import.manifest<>v_manifest THEN RAISE EXCEPTION 'Ingestion catalog conflict'; END IF; + RETURN jsonb_build_object('dataset_id',v_import.dataset_id,'job_id',v_job); + END IF; + v_dataset:=(v_manifest->>'dataset_id')::uuid; + IF v_dataset IS NULL THEN + INSERT INTO public.ingestion_datasets(organization_id,name,created_by) VALUES(p_org,v_manifest->>'filename',p_actor) RETURNING id INTO v_dataset; + ELSE + PERFORM 1 FROM public.ingestion_datasets WHERE id=v_dataset AND organization_id=p_org FOR SHARE; + IF NOT FOUND THEN RAISE EXCEPTION 'Ingestion access denied' USING ERRCODE='42501'; END IF; + END IF; + INSERT INTO public.ingestion_imports(organization_id,job_id,dataset_id,actor_id,manifest) VALUES(p_org,v_job,v_dataset,p_actor,v_manifest); + RETURN jsonb_build_object('dataset_id',v_dataset,'job_id',v_job); + ELSIF p_action='complete' THEN + v_job:=p_input->>'job_id'; v_result:=p_input->'result'; v_gold:=nullif(p_input->'gold','null'::jsonb); v_status:=v_result->>'status'; + SELECT * INTO v_import FROM public.ingestion_imports WHERE organization_id=p_org AND job_id=v_job FOR UPDATE; + IF NOT FOUND THEN RAISE EXCEPTION 'Ingestion access denied' USING ERRCODE='42501'; END IF; + IF v_result IS NULL OR jsonb_typeof(v_result)<>'object' OR v_status IS NULL OR v_status NOT IN ('ready','held','rejected') + OR (v_result->>'job_id') IS DISTINCT FROM v_job + OR (v_result->>'source_sha256') IS DISTINCT FROM (v_import.manifest->>'bronze_sha256') + OR (v_result->>'engine_version') IS DISTINCT FROM (v_import.manifest->>'engine_version') + OR coalesce(v_result->>'silver_sha256','') !~ '^[a-f0-9]{64}$' + OR coalesce(v_result->>'rows_read','') !~ '^\d{1,5}$' OR (v_result->>'rows_read')::integer>10000 + OR coalesce(v_result->>'rows_output','') !~ '^\d{1,5}$' OR (v_result->>'rows_output')::integer>10000 + THEN RAISE EXCEPTION 'Invalid ingestion catalog input' USING ERRCODE='22023'; END IF; + IF v_status='ready' THEN + IF v_gold IS NULL OR jsonb_typeof(v_gold)<>'object' OR coalesce(jsonb_typeof(v_gold->'records'),'')<>'array' + OR coalesce(jsonb_typeof(v_gold->'field_types'),'')<>'object' + OR (v_gold->>'engine_version') IS DISTINCT FROM (v_import.manifest->>'engine_version') + OR (v_gold->>'contract_version') IS DISTINCT FROM (v_import.manifest->'contract'->>'version') + OR coalesce(v_result->>'gold_sha256','') !~ '^[a-f0-9]{64}$' + THEN RAISE EXCEPTION 'Invalid ingestion gold' USING ERRCODE='22023'; END IF; + v_count:=jsonb_array_length(v_gold->'records'); + IF v_count NOT BETWEEN 1 AND 10000 OR v_count<>(v_result->>'rows_output')::integer + OR (SELECT count(*) FROM jsonb_object_keys(v_gold->'field_types')) NOT BETWEEN 1 AND 100 + OR EXISTS(SELECT 1 FROM jsonb_each(v_gold->'field_types') f WHERE jsonb_typeof(f.value)<>'string' OR f.value#>>'{}' NOT IN ('string','decimal','integer','date','boolean')) + THEN RAISE EXCEPTION 'Invalid ingestion gold' USING ERRCODE='22023'; END IF; + IF EXISTS(SELECT 1 FROM jsonb_array_elements(v_gold->'records') r WHERE jsonb_typeof(r)<>'object') THEN + RAISE EXCEPTION 'Invalid ingestion gold' USING ERRCODE='22023'; + END IF; + IF EXISTS(SELECT 1 FROM jsonb_array_elements(v_gold->'records') r WHERE + (SELECT count(*) FROM jsonb_object_keys(r))<>(SELECT count(*) FROM jsonb_object_keys(v_gold->'field_types')) + OR EXISTS(SELECT 1 FROM jsonb_each(r) f WHERE NOT (v_gold->'field_types' ? f.key) OR jsonb_typeof(f.value) NOT IN ('string','null'))) + THEN RAISE EXCEPTION 'Invalid ingestion gold' USING ERRCODE='22023'; END IF; + ELSE + v_count:=0; + IF v_gold IS NOT NULL OR (v_result->>'gold_sha256') IS NOT NULL OR (v_result->>'rows_output')::integer<>0 THEN + RAISE EXCEPTION 'Invalid ingestion gold' USING ERRCODE='22023'; + END IF; + END IF; + v_fingerprint:=pg_catalog.encode(pg_catalog.sha256(pg_catalog.convert_to(jsonb_build_object('result',v_result,'gold',v_gold)::text,'UTF8')),'hex'); + IF v_import.status<>'queued' THEN + IF v_import.completion_sha256<>v_fingerprint THEN RAISE EXCEPTION 'Ingestion catalog conflict'; END IF; + RETURN jsonb_build_object('dataset_id',v_import.dataset_id,'job_id',v_job,'status',v_import.status,'rows',v_count); + END IF; + INSERT INTO public.ingestion_records(organization_id,job_id,dataset_id,row_number,record) + SELECT p_org,v_job,v_import.dataset_id,n::integer,r FROM jsonb_array_elements(v_gold->'records') WITH ORDINALITY AS rows(r,n); + UPDATE public.ingestion_imports SET status=v_status,result=v_result,field_types=v_gold->'field_types',completion_sha256=v_fingerprint,completed_at=clock_timestamp() + WHERE organization_id=p_org AND job_id=v_job; + RETURN jsonb_build_object('dataset_id',v_import.dataset_id,'job_id',v_job,'status',v_status,'rows',v_count); + ELSIF p_action IN ('datasets','imports','records') THEN + v_limit:=coalesce((p_input->>'limit')::integer,25); + IF v_limit NOT BETWEEN 1 AND (CASE WHEN p_action='records' THEN 100 ELSE 50 END) THEN RAISE EXCEPTION 'Invalid ingestion page' USING ERRCODE='22023'; END IF; + IF p_action='datasets' THEN + v_cursor_id:=(p_input->>'after')::uuid; + IF v_cursor_id IS NOT NULL THEN + SELECT created_at INTO v_cursor_time FROM public.ingestion_datasets WHERE id=v_cursor_id AND organization_id=p_org; + IF NOT FOUND THEN RAISE EXCEPTION 'Ingestion access denied' USING ERRCODE='42501'; END IF; + END IF; + SELECT coalesce(jsonb_agg(item ORDER BY created_at DESC,id DESC),'[]'::jsonb) INTO v_items FROM ( + SELECT d.id,d.created_at,jsonb_build_object('dataset_id',d.id,'name',d.name,'created_at',d.created_at) item + FROM public.ingestion_datasets d WHERE d.organization_id=p_org AND (v_cursor_id IS NULL OR (d.created_at,d.id)<(v_cursor_time,v_cursor_id)) + ORDER BY d.created_at DESC,d.id DESC LIMIT v_limit+1 + ) q; + RETURN jsonb_build_object('datasets',v_items- v_limit,'next_after',CASE WHEN jsonb_array_length(v_items)>v_limit THEN v_items->(v_limit-1)->>'dataset_id' ELSE NULL END); + END IF; + v_dataset:=(p_input->>'dataset_id')::uuid; + PERFORM 1 FROM public.ingestion_datasets WHERE id=v_dataset AND organization_id=p_org; + IF NOT FOUND THEN RAISE EXCEPTION 'Ingestion access denied' USING ERRCODE='42501'; END IF; + IF p_action='imports' THEN + v_cursor_job:=p_input->>'after'; + IF v_cursor_job IS NOT NULL THEN + SELECT created_at INTO v_cursor_time FROM public.ingestion_imports WHERE organization_id=p_org AND dataset_id=v_dataset AND job_id=v_cursor_job; + IF NOT FOUND THEN RAISE EXCEPTION 'Ingestion access denied' USING ERRCODE='42501'; END IF; + END IF; + SELECT coalesce(jsonb_agg(item ORDER BY created_at DESC,job_id DESC),'[]'::jsonb) INTO v_items FROM ( + SELECT i.created_at,i.job_id,jsonb_build_object('job_id',i.job_id,'dataset_id',i.dataset_id,'status',i.status,'filename',i.manifest->>'filename', + 'format',i.manifest->>'format','source_sha256',i.manifest->>'bronze_sha256','created_at',i.created_at,'completed_at',i.completed_at, + 'rows_read',i.result->'rows_read','rows_output',i.result->'rows_output','gold_sha256',i.result->'gold_sha256') item + FROM public.ingestion_imports i WHERE i.organization_id=p_org AND i.dataset_id=v_dataset AND (v_cursor_job IS NULL OR (i.created_at,i.job_id)<(v_cursor_time,v_cursor_job)) + ORDER BY i.created_at DESC,i.job_id DESC LIMIT v_limit+1 + ) q; + RETURN jsonb_build_object('imports',v_items-v_limit,'next_after',CASE WHEN jsonb_array_length(v_items)>v_limit THEN v_items->(v_limit-1)->>'job_id' ELSE NULL END); + END IF; + v_job:=p_input->>'job_id'; v_after:=coalesce((p_input->>'after_row')::integer,0); + IF v_after NOT BETWEEN 0 AND 10000 THEN RAISE EXCEPTION 'Invalid ingestion page' USING ERRCODE='22023'; END IF; + SELECT * INTO v_import FROM public.ingestion_imports WHERE organization_id=p_org AND dataset_id=v_dataset AND job_id=v_job; + IF NOT FOUND THEN RAISE EXCEPTION 'Ingestion access denied' USING ERRCODE='42501'; END IF; + IF v_import.status<>'ready' THEN RAISE EXCEPTION 'Artifact not ready'; END IF; + SELECT coalesce(jsonb_agg(jsonb_build_object('row_number',row_number,'record',record) ORDER BY row_number),'[]'::jsonb) INTO v_items FROM ( + SELECT row_number,record,sum(octet_length(record::text)+64) OVER (ORDER BY row_number) page_bytes FROM ( + SELECT row_number,record FROM public.ingestion_records WHERE organization_id=p_org AND job_id=v_job AND row_number>v_after ORDER BY row_number LIMIT v_limit + ) bounded + ) q WHERE page_bytes<=2097152; + v_last:=(v_items->-1->>'row_number')::integer; + IF v_last IS NULL AND EXISTS(SELECT 1 FROM public.ingestion_records WHERE organization_id=p_org AND job_id=v_job AND row_number>v_after) THEN + RAISE EXCEPTION 'Ingestion record size limit'; + END IF; + RETURN jsonb_build_object('dataset_id',v_dataset,'job_id',v_job,'field_types',v_import.field_types,'gold_sha256',v_import.result->>'gold_sha256', + 'records',v_items,'next_after',CASE WHEN EXISTS(SELECT 1 FROM public.ingestion_records WHERE organization_id=p_org AND job_id=v_job AND row_number>v_last) THEN v_last ELSE NULL END); + END IF; + RAISE EXCEPTION 'Invalid ingestion catalog action' USING ERRCODE='22023'; +END $$; +COMMIT; diff --git a/tests/INGESTION-CATALOG.md b/tests/INGESTION-CATALOG.md new file mode 100644 index 0000000..8d3a707 --- /dev/null +++ b/tests/INGESTION-CATALOG.md @@ -0,0 +1,34 @@ +# Organization ingestion catalog + +`20261110010000_ingestion_catalog.sql` adds organization-owned datasets, immutable import versions and queryable gold rows to the existing Recoup database. Apply `20261110010100_ingestion_catalog_page_budget.sql` immediately afterward for bounded inline record pages. The initial migration is preserved; the correction is forward-only. No database or schema is created per client. Original files and processing artifacts stay in private Supabase Storage. + +The only runtime database entry point is `ingestion_catalog_command(actor, organization, action, input)`. It is executable by the service role only. The authenticated API supplies the actor; every command checks current organization membership and holds its membership lock through commit. Tables have RLS and no direct browser or service-role privileges. The security-definer functions use an empty search path and fully qualified relations. The internal authorization helper cannot be called by browser or service roles. + +`register` assigns a new dataset or attaches an import to an existing dataset in the same organization. Immutable manifests bind the scoped source object, contract and selected dataset. `complete` publishes ready gold rows and the import receipt atomically, or records held/rejected metadata without gold. A receipt compares the canonical received result and gold objects for the selected immutable import. The trusted worker is the authority for `gold_sha256`: it hashes the exact private artifact bytes before sending their content to SQL. The database independently fingerprints the received result/gold JSONB for retry consistency. Lists and row pages are bounded and scoped; record payloads have a 2 MiB page budget and oversized individual rows return a controlled error rather than disappearing. Reading rows requires both dataset and import version. New imports do not replace earlier records or imply semantic deduplication across statements. + +Run isolated tests with PostgreSQL binaries from one installation on PATH: + +```sh +python3 -m unittest discover -s tests -p 'test_ingestion_catalog.py' +``` + +The harness starts and removes a disposable Unix-socket-only cluster. It bootstraps minimal account/membership dependencies, applies the actual migration and exercises the real service role, cross-org denial, revocation, concurrency, exact strings, immutable versions, failed-write rollback and pagination. This is not a full historical migration replay or a production deployment. + +Optionally exercise the matching API processor against that same temporary database: + +```sh +INGESTION_API_WORKTREE=/absolute/path/to/api-worktree \ + python3 -m unittest discover -s tests -p 'test_ingestion_catalog.py' +``` + +The API checkout needs dependencies installed and `lib/ingestion/__tests__/catalog.integration.test.ts`. The test uses synthetic CSV bytes, the real parsing pipeline, the guarded SQL gateway and private in-memory artifact storage. It verifies a committed completion losing its acknowledgement, recovery without duplicate rows, and database-enforced organization/revocation checks. It does not exercise hosted Storage, PostgREST, HTTP credentials or Workflow infrastructure. + +## Rollout + +Apply the additive migration to the existing Recoup project before deploying the dependent API change. Keep ingestion disabled until hosted access/storage/workflow qualification. There are no new client connection strings, external database identities, scheduled jobs or production writes in the test runner. + +If rolling back, revert dependent API behavior first and retain datasets/import history. Do not drop or rewrite accepted records as routine rollback. Existing private ingestion artifacts are not deleted or automatically combined across versions. + +## Status — 2026-11-09 + +Local PostgreSQL 17 verification passes twelve database tests plus the optional joined API lifecycle test. CI covers PostgreSQL 15 and 17. Migration deployment, hosted org-scoped ingestion and production activation remain unverified and are separate rollout steps. diff --git a/tests/test_ingestion_catalog.py b/tests/test_ingestion_catalog.py new file mode 100644 index 0000000..c170f56 --- /dev/null +++ b/tests/test_ingestion_catalog.py @@ -0,0 +1,206 @@ +"""Shared organization ingestion on disposable PostgreSQL; no hosted credentials.""" +import os +import json +import pathlib +import subprocess +import tempfile +import unittest +import uuid +from concurrent.futures import ThreadPoolExecutor + +ROOT = pathlib.Path(__file__).resolve().parents[1] +MIGRATIONS = ['20261110010000_ingestion_catalog.sql', '20261110010100_ingestion_catalog_page_budget.sql'] + + +class IngestionCatalog(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.tmp = tempfile.TemporaryDirectory(prefix='recoup-ingestion-') + cls.addClassCleanup(cls.tmp.cleanup) + cls.cluster = pathlib.Path(cls.tmp.name) / 'pg' + subprocess.run(['initdb', '-D', str(cls.cluster), '-A', 'trust'], check=True, capture_output=True) + subprocess.run(['pg_ctl', '-D', str(cls.cluster), '-l', f'{cls.tmp.name}/pg.log', + '-o', f"-k {cls.tmp.name} -p 55489 -c listen_addresses=''", '-w', 'start'], + check=True, capture_output=True) + cls.addClassCleanup(cls.stop) + cls.sql('''CREATE ROLE service_role BYPASSRLS; CREATE ROLE anon; CREATE ROLE authenticated; + CREATE TABLE public.accounts(id uuid PRIMARY KEY); + CREATE TABLE public.account_organization_ids(id uuid PRIMARY KEY DEFAULT gen_random_uuid(), account_id uuid REFERENCES public.accounts(id), organization_id uuid REFERENCES public.accounts(id), updated_at timestamptz DEFAULT now()); + GRANT USAGE ON SCHEMA public TO service_role, anon, authenticated;''') + for migration in MIGRATIONS: + cls.sql((ROOT / 'supabase/migrations' / migration).read_text()) + + @classmethod + def stop(cls): + subprocess.run(['pg_ctl', '-D', str(cls.cluster), '-m', 'immediate', '-w', 'stop'], check=True, capture_output=True) + + @classmethod + def sql(cls, statement, succeeds=True): + result = subprocess.run(['psql', '-h', cls.tmp.name, '-p', '55489', '-d', 'postgres', '-XAtq', '-v', 'ON_ERROR_STOP=1'], + input=statement, text=True, capture_output=True) + if succeeds and result.returncode: + raise AssertionError(result.stderr) + return result + + def setUp(self): + self.actor, self.org, self.other = [str(uuid.uuid4()) for _ in range(3)] + self.job = 'a' * 64 + self.sql(f"INSERT INTO accounts VALUES ('{self.actor}'),('{self.org}'),('{self.other}'); INSERT INTO account_organization_ids(account_id,organization_id) VALUES ('{self.actor}','{self.org}')") + + def manifest(self, job=None, dataset=None, org=None): + job, org = job or self.job, org or self.org + value = {'job_id': job, 'input_sha256': 'b' * 64, 'engine_version': '1', + 'scope': f'organization/{org}', 'format': 'csv', 'filename': 'synthetic.csv', + 'bronze_key': f'organization/{org}/originals/' + 'c' * 64, 'bronze_sha256': 'c' * 64, + 'source': {'kind': 'inline'}, 'contract': {'version': '1'}, 'created_at': '2026-11-10T00:00:00.000Z'} + if dataset: + value['dataset_id'] = dataset + return value + + def statement(self, action, value, actor=None, org=None, role='service_role'): + payload = json.dumps(value).replace("'", "''") + return f"SET ROLE {role}; SELECT public.ingestion_catalog_command('{actor or self.actor}','{org or self.org}','{action}','{payload}'::jsonb)" + + def call(self, action, value, **kwargs): + return json.loads(self.sql(self.statement(action, value, **kwargs)).stdout.strip()) + + def register(self, **kwargs): + return self.call('register', {'manifest': self.manifest(**kwargs)}) + + def completion(self, status='ready', job=None, records=None): + records = records or [{'id': '001', 'amount': '9007199254740993.00000001'}] + gold = {'engine_version': '1', 'contract_version': '1', 'records': records, + 'field_types': {'id': 'string', 'amount': 'decimal'}} if status == 'ready' else None + return {'job_id': job or self.job, 'gold': gold, + 'result': {'job_id': job or self.job, 'status': status, 'engine_version': '1', + 'source_sha256': 'c' * 64, 'rows_read': len(records), 'rows_output': len(records) if gold else 0, + 'issues': [], 'issue_count': 0, 'issues_truncated': 0, 'controls': [], + 'silver_sha256': 'e' * 64, 'gold_sha256': 'f' * 64 if gold else None}} + + def denied(self, action, value, **kwargs): + result = self.sql(self.statement(action, value, **kwargs), succeeds=False) + self.assertNotEqual(result.returncode, 0) + self.assertIn('Ingestion access denied', result.stderr) + + def test_register_retry_and_list(self): + first = self.register() + self.assertEqual(self.register(), first) + listed = self.call('datasets', {'limit': 25}) + self.assertEqual([x['dataset_id'] for x in listed['datasets']], [first['dataset_id']]) + imports = self.call('imports', {'dataset_id': first['dataset_id'], 'limit': 25}) + self.assertEqual(imports['imports'][0]['status'], 'queued') + changed = self.manifest() + changed['filename'] = 'changed.csv' + result = self.sql(self.statement('register', {'manifest': changed}), succeeds=False) + self.assertIn('Ingestion catalog conflict', result.stderr) + + def test_exact_gold_and_idempotent_completion(self): + dataset = self.register()['dataset_id'] + receipt = self.call('complete', self.completion()) + self.assertEqual(self.call('complete', self.completion()), receipt) + page = self.call('records', {'dataset_id': dataset, 'job_id': self.job, 'limit': 25, 'after_row': 0}) + self.assertEqual(page['records'], [{'row_number': 1, 'record': {'id': '001', 'amount': '9007199254740993.00000001'}}]) + changed = self.completion() + changed['gold']['records'][0]['amount'] = '0' + result = self.sql(self.statement('complete', changed), succeeds=False) + self.assertIn('Ingestion catalog conflict', result.stderr) + self.assertEqual(self.call('records', {'dataset_id': dataset, 'job_id': self.job})['records'], page['records']) + + def test_versions_share_dataset_without_overwriting(self): + dataset = self.register()['dataset_id'] + self.call('complete', self.completion()) + self.register(job='d' * 64, dataset=dataset) + self.call('complete', self.completion(job='d' * 64, records=[{'id': '002', 'amount': '1'}])) + versions = self.call('imports', {'dataset_id': dataset})['imports'] + self.assertEqual(len(versions), 2) + self.assertEqual(self.call('records', {'dataset_id': dataset, 'job_id': self.job})['records'][0]['record']['id'], '001') + + def test_cross_organization_and_revoked_membership(self): + dataset = self.register()['dataset_id'] + self.call('complete', self.completion()) + self.denied('datasets', {}, org=self.other) + self.denied('records', {'dataset_id': dataset, 'job_id': self.job}, org=self.other, actor=self.other) + self.denied('register', {'manifest': self.manifest(job='d' * 64, dataset=dataset, org=self.other)}, org=self.other, actor=self.other) + self.sql(f"DELETE FROM account_organization_ids WHERE account_id='{self.actor}'") + self.denied('records', {'dataset_id': dataset, 'job_id': self.job}) + self.denied('complete', self.completion()) + + def test_held_imports_have_no_queryable_gold(self): + dataset = self.register()['dataset_id'] + self.call('complete', self.completion('held')) + result = self.sql(self.statement('records', {'dataset_id': dataset, 'job_id': self.job}), succeeds=False) + self.assertIn('Artifact not ready', result.stderr) + self.assertEqual(self.sql(f"SELECT count(*) FROM ingestion_records WHERE organization_id='{self.org}'").stdout.strip(), '0') + + def test_bad_gold_is_atomic(self): + self.register() + value = self.completion() + value['gold']['records'].append({'id': '002', 'amount': {'nested': 'invalid'}}) + value['result']['rows_output'] = 2 + result = self.sql(self.statement('complete', value), succeeds=False) + self.assertNotEqual(result.returncode, 0) + self.assertEqual(self.sql(f"SELECT status FROM ingestion_imports WHERE organization_id='{self.org}'").stdout.strip(), 'queued') + self.assertEqual(self.sql(f"SELECT count(*) FROM ingestion_records WHERE organization_id='{self.org}'").stdout.strip(), '0') + + def test_private_original_scope_and_unknown_dataset(self): + value = self.manifest() + value['bronze_key'] = f'organization/{self.other}/originals/' + 'c' * 64 + self.assertIn('Invalid ingestion catalog input', self.sql(self.statement('register', {'manifest': value}), succeeds=False).stderr) + self.denied('register', {'manifest': self.manifest(dataset=str(uuid.uuid4()))}) + + def test_records_pagination_is_bounded_and_version_specific(self): + dataset = self.register()['dataset_id'] + self.call('complete', self.completion(records=[{'id': '001', 'amount': '1'}, {'id': '002', 'amount': '2'}])) + first = self.call('records', {'dataset_id': dataset, 'job_id': self.job, 'limit': 1}) + second = self.call('records', {'dataset_id': dataset, 'job_id': self.job, 'limit': 1, 'after_row': first['next_after']}) + self.assertEqual(first['next_after'], 1) + self.assertIsNone(second['next_after']) + self.assertEqual(second['records'][0]['record']['id'], '002') + self.assertNotEqual(self.sql(self.statement('records', {'dataset_id': dataset, 'job_id': self.job, 'limit': 101}), succeeds=False).returncode, 0) + + def test_record_pages_have_a_byte_budget(self): + dataset = self.register()['dataset_id'] + records = [{'id': 'x' * 150000 + str(i), 'amount': '1'} for i in range(20)] + self.call('complete', self.completion(records=records)) + first = self.call('records', {'dataset_id': dataset, 'job_id': self.job, 'limit': 25}) + self.assertLess(len(first['records']), 20) + self.assertGreater(len(first['records']), 0) + second = self.call('records', {'dataset_id': dataset, 'job_id': self.job, 'limit': 25, 'after_row': first['next_after']}) + self.assertEqual(len(first['records']) + len(second['records']), 20) + self.assertIsNone(second['next_after']) + + def test_an_oversized_single_record_is_not_silently_skipped(self): + dataset = self.register()['dataset_id'] + self.call('complete', self.completion(records=[{'id': 'x' * (3 * 1024 * 1024), 'amount': '1'}])) + result = self.sql(self.statement('records', {'dataset_id': dataset, 'job_id': self.job}), succeeds=False) + self.assertIn('Ingestion record size limit', result.stderr) + + def test_browser_and_service_direct_table_access_denied(self): + for role in ['anon', 'authenticated']: + self.assertIn('permission denied', self.sql(self.statement('datasets', {}, role=role), succeeds=False).stderr) + for table in ['ingestion_datasets', 'ingestion_imports', 'ingestion_records']: + for role in ['anon', 'authenticated', 'service_role']: + self.assertIn('permission denied', self.sql(f'SET ROLE {role}; SELECT * FROM public.{table}', succeeds=False).stderr) + self.assertEqual(self.sql("SELECT count(*) FROM pg_class WHERE relname IN ('ingestion_datasets','ingestion_imports','ingestion_records') AND relrowsecurity").stdout.strip(), '3') + + @unittest.skipUnless(os.environ.get('INGESTION_API_WORKTREE'), 'Optional API worktree integration') + def test_actual_api_processing_and_recovery(self): + result = subprocess.run(['pnpm', 'dlx', 'node@22', 'node_modules/vitest/vitest.mjs', 'run', + 'lib/ingestion/__tests__/catalog.integration.test.ts'], + cwd=os.environ['INGESTION_API_WORKTREE'], + env={**os.environ, 'INGESTION_CATALOG_TEST_SOCKET': self.tmp.name}, + text=True, capture_output=True, timeout=120) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + def test_concurrent_registration_and_completion_converge(self): + with ThreadPoolExecutor(max_workers=4) as pool: + ids = list(pool.map(lambda _: self.register()['dataset_id'], range(4))) + self.assertEqual(len(set(ids)), 1) + with ThreadPoolExecutor(max_workers=4) as pool: + receipts = list(pool.map(lambda _: self.call('complete', self.completion()), range(4))) + self.assertTrue(all(x == receipts[0] for x in receipts)) + self.assertEqual(self.sql(f"SELECT count(*) FROM ingestion_records WHERE organization_id='{self.org}'").stdout.strip(), '1') + + +if __name__ == '__main__': + unittest.main()