From dbf6eb4d629f0955e2498aabf018e0a4f18b24c9 Mon Sep 17 00:00:00 2001 From: Sidney Swift <158200036+sidneyswift@users.noreply.github.com> Date: Mon, 9 Nov 2026 18:34:26 -0500 Subject: [PATCH 1/4] feat: store organization ingestion datasets in shared database --- .github/workflows/ingestion-catalog.yml | 37 ++++ README.md | 4 + .../20261110010000_ingestion_catalog.sql | 196 ++++++++++++++++++ tests/INGESTION-CATALOG.md | 34 +++ tests/test_ingestion_catalog.py | 188 +++++++++++++++++ 5 files changed, 459 insertions(+) create mode 100644 .github/workflows/ingestion-catalog.yml create mode 100644 supabase/migrations/20261110010000_ingestion_catalog.sql create mode 100644 tests/INGESTION-CATALOG.md create mode 100644 tests/test_ingestion_catalog.py diff --git a/.github/workflows/ingestion-catalog.yml b/.github/workflows/ingestion-catalog.yml new file mode 100644 index 0000000..9788542 --- /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 + 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..9547a28 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,9 @@ # Recoup Supabase +## 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. + Supabase migration scripts for the Recoup project. ## Scheduled posts migration 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/tests/INGESTION-CATALOG.md b/tests/INGESTION-CATALOG.md new file mode 100644 index 0000000..c4ad4c6 --- /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. 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. Repeated completion must match the entire original completion payload. Lists and row pages are bounded and scoped; 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 ten 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..e66ae0b --- /dev/null +++ b/tests/test_ingestion_catalog.py @@ -0,0 +1,188 @@ +"""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] +MIGRATION = '20261110010000_ingestion_catalog.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(account_id uuid, organization_id uuid); + GRANT USAGE ON SCHEMA public TO service_role, anon, authenticated;''') + 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 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.assertNotEqual(self.sql(self.statement('register', {'manifest': value}), succeeds=False).returncode, 0) + 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_browser_and_service_direct_table_access_denied(self): + for role in ['anon', 'authenticated']: + self.assertNotEqual(self.sql(self.statement('datasets', {}, role=role), succeeds=False).returncode, 0) + for table in ['ingestion_datasets', 'ingestion_imports', 'ingestion_records']: + for role in ['anon', 'authenticated', 'service_role']: + self.assertNotEqual(self.sql(f'SET ROLE {role}; SELECT * FROM public.{table}', succeeds=False).returncode, 0) + 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() From ee76c22d30c69ec1f236202240e385b66ed5316a Mon Sep 17 00:00:00 2001 From: Sidney Swift <158200036+sidneyswift@users.noreply.github.com> Date: Mon, 9 Nov 2026 18:40:14 -0500 Subject: [PATCH 2/4] fix: bound shared ingestion record pages by bytes --- .github/workflows/ingestion-catalog.yml | 4 +- ...10010100_ingestion_catalog_page_budget.sql | 141 ++++++++++++++++++ tests/INGESTION-CATALOG.md | 6 +- tests/test_ingestion_catalog.py | 22 ++- 4 files changed, 166 insertions(+), 7 deletions(-) create mode 100644 supabase/migrations/20261110010100_ingestion_catalog_page_budget.sql diff --git a/.github/workflows/ingestion-catalog.yml b/.github/workflows/ingestion-catalog.yml index 9788542..b066243 100644 --- a/.github/workflows/ingestion-catalog.yml +++ b/.github/workflows/ingestion-catalog.yml @@ -3,13 +3,13 @@ on: pull_request: branches: [main] paths: - - "supabase/migrations/*_ingestion_catalog.sql" + - "supabase/migrations/*_ingestion_catalog*.sql" - "tests/test_ingestion_catalog.py" - ".github/workflows/ingestion-catalog.yml" push: branches: [main] paths: - - "supabase/migrations/*_ingestion_catalog.sql" + - "supabase/migrations/*_ingestion_catalog*.sql" - "tests/test_ingestion_catalog.py" - ".github/workflows/ingestion-catalog.yml" permissions: 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 index c4ad4c6..274eeb3 100644 --- a/tests/INGESTION-CATALOG.md +++ b/tests/INGESTION-CATALOG.md @@ -1,10 +1,10 @@ # Organization ingestion catalog -`20261110010000_ingestion_catalog.sql` adds organization-owned datasets, immutable import versions and queryable gold rows to the existing Recoup database. No database or schema is created per client. Original files and processing artifacts stay in private Supabase Storage. +`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. Repeated completion must match the entire original completion payload. Lists and row pages are bounded and scoped; reading rows requires both dataset and import version. New imports do not replace earlier records or imply semantic deduplication across statements. +`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. Repeated completion must match the entire original completion payload. 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: @@ -31,4 +31,4 @@ If rolling back, revert dependent API behavior first and retain datasets/import ## Status — 2026-11-09 -Local PostgreSQL 17 verification passes ten 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. +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 index e66ae0b..bb4eb21 100644 --- a/tests/test_ingestion_catalog.py +++ b/tests/test_ingestion_catalog.py @@ -9,7 +9,7 @@ from concurrent.futures import ThreadPoolExecutor ROOT = pathlib.Path(__file__).resolve().parents[1] -MIGRATION = '20261110010000_ingestion_catalog.sql' +MIGRATIONS = ['20261110010000_ingestion_catalog.sql', '20261110010100_ingestion_catalog_page_budget.sql'] class IngestionCatalog(unittest.TestCase): @@ -27,7 +27,8 @@ def setUpClass(cls): CREATE TABLE public.accounts(id uuid PRIMARY KEY); CREATE TABLE public.account_organization_ids(account_id uuid, organization_id uuid); GRANT USAGE ON SCHEMA public TO service_role, anon, authenticated;''') - cls.sql((ROOT / 'supabase/migrations' / MIGRATION).read_text()) + for migration in MIGRATIONS: + cls.sql((ROOT / 'supabase/migrations' / migration).read_text()) @classmethod def stop(cls): @@ -157,6 +158,23 @@ def test_records_pagination_is_bounded_and_version_specific(self): 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.assertNotEqual(self.sql(self.statement('datasets', {}, role=role), succeeds=False).returncode, 0) From 7cf77484d4b2b6b382985542e22b96e599cc1cde Mon Sep 17 00:00:00 2001 From: Sidney Swift <158200036+sidneyswift@users.noreply.github.com> Date: Mon, 9 Nov 2026 18:43:28 -0500 Subject: [PATCH 3/4] test: verify specific ingestion permission failures --- .github/workflows/ingestion-catalog.yml | 2 +- README.md | 4 ++-- tests/INGESTION-CATALOG.md | 2 +- tests/test_ingestion_catalog.py | 6 +++--- 4 files changed, 7 insertions(+), 7 deletions(-) diff --git a/.github/workflows/ingestion-catalog.yml b/.github/workflows/ingestion-catalog.yml index b066243..51e2453 100644 --- a/.github/workflows/ingestion-catalog.yml +++ b/.github/workflows/ingestion-catalog.yml @@ -28,7 +28,7 @@ jobs: run: | sudo apt-get update sudo apt-get install -y postgresql-common - sudo /usr/share/postgresql-common/pgdg/apt.postgresql.org.sh -y + 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 diff --git a/README.md b/README.md index 9547a28..743af6a 100644 --- a/README.md +++ b/README.md @@ -1,11 +1,11 @@ # Recoup Supabase +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. -Supabase migration scripts for the Recoup project. - ## 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/tests/INGESTION-CATALOG.md b/tests/INGESTION-CATALOG.md index 274eeb3..8d3a707 100644 --- a/tests/INGESTION-CATALOG.md +++ b/tests/INGESTION-CATALOG.md @@ -4,7 +4,7 @@ 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. Repeated completion must match the entire original completion payload. 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. +`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: diff --git a/tests/test_ingestion_catalog.py b/tests/test_ingestion_catalog.py index bb4eb21..1c585e6 100644 --- a/tests/test_ingestion_catalog.py +++ b/tests/test_ingestion_catalog.py @@ -145,7 +145,7 @@ def test_bad_gold_is_atomic(self): def test_private_original_scope_and_unknown_dataset(self): value = self.manifest() value['bronze_key'] = f'organization/{self.other}/originals/' + 'c' * 64 - self.assertNotEqual(self.sql(self.statement('register', {'manifest': value}), succeeds=False).returncode, 0) + 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): @@ -177,10 +177,10 @@ def test_an_oversized_single_record_is_not_silently_skipped(self): def test_browser_and_service_direct_table_access_denied(self): for role in ['anon', 'authenticated']: - self.assertNotEqual(self.sql(self.statement('datasets', {}, role=role), succeeds=False).returncode, 0) + 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.assertNotEqual(self.sql(f'SET ROLE {role}; SELECT * FROM public.{table}', succeeds=False).returncode, 0) + 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') From d0262c7cb975e3d045273effa18c60c04a31b996 Mon Sep 17 00:00:00 2001 From: Sidney Swift <158200036+sidneyswift@users.noreply.github.com> Date: Mon, 9 Nov 2026 18:56:44 -0500 Subject: [PATCH 4/4] test: mirror organization membership column defaults --- tests/test_ingestion_catalog.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_ingestion_catalog.py b/tests/test_ingestion_catalog.py index 1c585e6..c170f56 100644 --- a/tests/test_ingestion_catalog.py +++ b/tests/test_ingestion_catalog.py @@ -25,7 +25,7 @@ def setUpClass(cls): 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(account_id uuid, organization_id uuid); + 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()) @@ -45,7 +45,7 @@ def sql(cls, statement, succeeds=True): 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 VALUES ('{self.actor}','{self.org}')") + 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