feat(epcis): document receipts and live EPCIS failure metric

alembic/versions/20261024_a3f7ce361b0e_v137_epcis_document_receipts.py

@@ -0,0 +1,110 @@
+"""EPCIS document receipts: fsma.epcis_document_receipts.
+
+Mirrors supabase/migrations/20261024090000_epcis_document_receipts.sql exactly:
+same table, columns, constraints, index, RLS policy and grants. One row per
+EPCIS document the ingest routes receive, accepted or rejected, keyed by the
+SHA-256 of the raw request body. error_summary is a short sanitized summary,
+never payload content.
+
+Additive and idempotent (IF NOT EXISTS, drop-if-exists-then-create policy,
+role-guarded grants), so the plain SQL file can run ahead of it. Downgrade
+drops only what this revision created.
+
+Revision ID: a3f7ce361b0e
+Revises: b005ddbf1dba
+Create Date: 2026-10-24
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "a3f7ce361b0e"  # pragma: allowlist secret
+down_revision: Union[str, Sequence[str], None] = "b005ddbf1dba"  # pragma: allowlist secret
+branch_labels = None
+depends_on = None
+
+_UPGRADE_STATEMENTS: tuple[str, ...] = (
+    """
+CREATE TABLE IF NOT EXISTS fsma.epcis_document_receipts (
+    id              uuid primary key default gen_random_uuid(),
+    tenant_id       uuid not null,
+    document_sha256 text not null,
+    received_at     timestamptz not null default now(),
+    status          text not null,
+    error_summary   text,
+    constraint epcis_document_receipts_status_check
+        check (status in ('accepted', 'rejected')),
+    constraint epcis_document_receipts_sha256_check
+        check (document_sha256 ~ '^[0-9a-f]{64}$'),
+    constraint epcis_document_receipts_error_summary_length_check
+        check (error_summary is null or char_length(error_summary) <= 200)
+);
+""",
+    """
+-- Read path: rejected share per tenant over a rolling window.
+create index if not exists idx_epcis_document_receipts_tenant_received
+    on fsma.epcis_document_receipts (tenant_id, received_at);
+""",
+    """
+ALTER TABLE fsma.epcis_document_receipts ENABLE ROW LEVEL SECURITY;
+""",
+    """
+ALTER TABLE fsma.epcis_document_receipts FORCE ROW LEVEL SECURITY;
+""",
+    """
+-- CASE form, not OR: public.get_tenant_context() raises when app.tenant_id is
+-- unset and Postgres does not order OR operands, so the sysadmin arm is tested
+-- first (same predicate shape as alembic v096, v134 and v135).
+drop policy if exists tenant_isolation_epcis_document_receipts on fsma.epcis_document_receipts;
+""",
+    """
+create policy tenant_isolation_epcis_document_receipts on fsma.epcis_document_receipts
+    for all
+    using (
+        case
+            when (select current_setting('regengine.is_sysadmin', true)) = 'true'
+                 and current_user = 'regengine_sysadmin' then true
+            else tenant_id = (select public.get_tenant_context())
+        end
+    )
+    with check (
+        case
+            when (select current_setting('regengine.is_sysadmin', true)) = 'true'
+                 and current_user = 'regengine_sysadmin' then true
+            else tenant_id = (select public.get_tenant_context())
+        end
+    );
+""",
+    """
+-- Runtime role allowlist (alembic v111): new tables are inaccessible to the
+-- app role until a migration names them. Insert and select only: a receipt is
+-- evidence and is never updated or deleted by the app.
+do $$
+begin
+    if exists (select 1 from pg_roles where rolname = 'anon') then
+        revoke all on fsma.epcis_document_receipts from anon;
+    end if;
+    if exists (select 1 from pg_roles where rolname = 'authenticated') then
+        revoke all on fsma.epcis_document_receipts from authenticated;
+    end if;
+    if exists (select 1 from pg_roles where rolname = 'regengine') then
+        grant select, insert on fsma.epcis_document_receipts to regengine;
+    end if;
+    if exists (select 1 from pg_roles where rolname = 'regengine_sysadmin') then
+        grant select on fsma.epcis_document_receipts to regengine_sysadmin;
+    end if;
+end $$;
+""",
+)
+
+
+def upgrade() -> None:
+    for statement in _UPGRADE_STATEMENTS:
+        op.execute(statement)
+
+
+def downgrade() -> None:
+    op.execute("DROP POLICY IF EXISTS tenant_isolation_epcis_document_receipts ON fsma.epcis_document_receipts")
+    op.execute("DROP INDEX IF EXISTS fsma.idx_epcis_document_receipts_tenant_received")
+    op.execute("DROP TABLE IF EXISTS fsma.epcis_document_receipts")

docs/schema/expected_schema.json

@@ -1474,6 +1474,35 @@
       }
     ]
   },
+  "fsma.epcis_document_receipts": {
+    "columns": {
+      "document_sha256": {
+        "nullable": false,
+        "type": "text"
+      },
+      "error_summary": {
+        "nullable": true,
+        "type": "text"
+      },
+      "id": {
+        "nullable": false,
+        "type": "uuid"
+      },
+      "received_at": {
+        "nullable": false,
+        "type": "timestamp"
+      },
+      "status": {
+        "nullable": false,
+        "type": "text"
+      },
+      "tenant_id": {
+        "nullable": false,
+        "type": "uuid"
+      }
+    },
+    "foreign_keys": []
+  },
   "fsma.evidence_attachments": {
     "columns": {
       "created_at": {

frontend/src/__tests__/components/supplier-data-quality.test.tsx

@@ -119,6 +119,29 @@ describe('DataQuality', () => {
         expect(screen.getByText(LATE_LABEL)).toBeInTheDocument();
     });
 
+    it('shows a computed EPCIS share with its tenant wide reason and no drill down', () => {
+        const receipts = uuids(MIN_RECORDS);
+        const rejected = receipts.filter((_, index) => index % 2 === 0);
+        const share = rejected.length / receipts.length;
+        const tenantReason = crypto.randomUUID();
+        renderQuality(
+            summary([
+                metric({
+                    key: 'epcis_failed',
+                    status: 'ok',
+                    numerator: rejected.length,
+                    denominator: receipts.length,
+                    share,
+                    reason: tenantReason,
+                }),
+            ]),
+        );
+        expect(screen.getByText('EPCIS (Preview) documents rejected across your whole account:')).toBeInTheDocument();
+        expect(screen.getByText(`${rejected.length} of ${receipts.length} documents (${formatShare(share)})`)).toBeInTheDocument();
+        expect(screen.getByText(tenantReason)).toBeInTheDocument();
+        expect(screen.queryByRole('button', { name: 'View records' })).not.toBeInTheDocument();
+    });
+
     it('offers drill down only for computable metrics with flagged records', () => {
         renderQuality();
         expect(screen.getAllByRole('button', { name: 'View records' })).toHaveLength(2);

frontend/src/app/dashboard/suppliers/data-quality.tsx

@@ -16,7 +16,7 @@ const METRIC_COPY: Record<SupplierQualityMetricKey, { label: string; unit: strin
     missing_kde: { label: 'Records missing a KDE', unit: 'records' },
     unlinked_lot: { label: 'Lot codes not linked back through the chain', unit: 'lot codes' },
     late_cte: { label: 'Late CTE records', unit: 'records' },
-    epcis_failed: { label: 'EPCIS (Preview) files failing validation', unit: 'files' },
+    epcis_failed: { label: 'EPCIS (Preview) documents rejected across your whole account', unit: 'documents' },
 };
 
 function isDrilldown(key: SupplierQualityMetricKey): key is SupplierQualityDrilldownMetric {
@@ -170,6 +170,7 @@ function MetricRow({
                     <SendingHoursNote hours={agreedSendingHours} />
                 </>
             )}
+            {metric.status !== 'unavailable' && metric.reason && <p className="text-re-text-tertiary">{metric.reason}</p>}
             {open && isDrilldown(metric.key) && (
                 <div className="space-y-1 pl-3">
                     <DrilldownList metric={metric.key} state={drilldown} />

frontend/src/types/api.ts

@@ -524,7 +524,7 @@ export interface SupplierQualityMetric {
     numerator: number | null;
     denominator: number | null;
     share: number | null;
-    /** Set when status is unavailable, or late_cte has no agreed sending time. */
+    /** Set when status is unavailable, when late_cte has no agreed sending time, and on epcis_failed to say the count is tenant wide. */
     reason: string | null;
 }
 

services/admin/app/supplier_compliance_routes.py

@@ -268,7 +268,9 @@ async def get_supplier_data_quality(
 ) -> SupplierDataQualityResponse:
     """Four separate data quality measures per facility over a rolling window.
 
-    Scoped to the tenant and to facilities owned by the calling user.
+    Scoped to the tenant and to facilities owned by the calling user, except
+    epcis_failed: EPCIS receipts carry no facility, so that share is tenant
+    wide and repeated on every facility.
     """
     tenant_id = TenantContext.get_tenant_context(db)
     if not tenant_id:

services/admin/tests/test_supplier_data_quality_routes.py

@@ -15,6 +15,8 @@
 from fastapi.testclient import TestClient
 from shared.supplier_data_quality import (
     DEFAULT_WINDOW_DAYS,
+    EPCIS_TENANT_SCOPE_REASON,
+    EPCIS_UNAVAILABLE_REASON,
     LATE_CTE_LABEL,
     LATE_CTE_NO_HOURS_REASON,
     MAX_WINDOW_DAYS,
@@ -60,6 +62,11 @@
 ]
 
 
+# Tenant wide EPCIS receipts: enough for a share, every other one rejected.
+RECEIPTS = [uuid4() for _ in range(MIN_RECORDS)]
+REJECTED_RECEIPTS = RECEIPTS[::2]
+
+
 class FakeSession:
     def __init__(self, own_hours: int | None = None) -> None:
         self.commits = 0
@@ -98,14 +105,25 @@ def calls():
     return []
 
 
-def _client(monkeypatch, calls, tenant=TENANT, session=None, role=None):
+def _client(
+    monkeypatch,
+    calls,
+    tenant=TENANT,
+    session=None,
+    role=None,
+    receipt_counts=(len(REJECTED_RECEIPTS), len(RECEIPTS)),
+):
     import app.supplier_compliance_routes as routes
 
     # Membership role of the caller in the bound tenant; None means no membership.
     monkeypatch.setattr(routes, "_get_user_role_name", lambda _tenant, _user, _db: role)
 
     session = session if session is not None else FakeSession()
 
+    def fake_receipt_counts(_db, tenant_id, since):
+        calls.append(("receipts", tenant_id))
+        return receipt_counts
+
     def fake_records(_db, tenant_id, user_id, facility_ids, since):
         calls.append(("records", tenant_id, user_id, list(facility_ids)))
         return [dict(row) for row in RECORD_ROWS]
@@ -124,6 +142,7 @@ def fake_hours(db, tenant_id, user_id, facility_ids):
     monkeypatch.setattr(routes, "fetch_lot_rows", fake_lots)
     monkeypatch.setattr("shared.supplier_data_quality.fetch_record_rows", fake_records)
     monkeypatch.setattr("shared.supplier_data_quality.fetch_lot_rows", fake_lots)
+    monkeypatch.setattr("shared.supplier_data_quality.fetch_receipt_counts", fake_receipt_counts)
     monkeypatch.setattr(routes.TenantContext, "get_tenant_context", lambda _db: tenant)
 
     app = FastAPI()
@@ -152,9 +171,39 @@ def test_summary_returns_four_separate_metrics_for_owned_facilities(monkeypatch,
     assert supplier["agreed_sending_hours"] is None
     late = supplier["metrics"][2]
     assert (late["status"], late["numerator"], late["reason"]) == ("not_enough_data", None, LATE_CTE_NO_HOURS_REASON)
-    assert supplier["metrics"][3]["status"] == "unavailable"
-    assert all(call[1:3] == (str(TENANT), str(USER)) for call in calls)
-    assert all(call[3] == [str(OWN_FACILITY)] for call in calls)
+    epcis = supplier["metrics"][3]
+    assert (epcis["status"], epcis["numerator"], epcis["denominator"], epcis["share"], epcis["reason"]) == (
+        "ok",
+        len(REJECTED_RECEIPTS),
+        len(RECEIPTS),
+        len(REJECTED_RECEIPTS) / len(RECEIPTS),
+        EPCIS_TENANT_SCOPE_REASON,
+    )
+    scoped = [call for call in calls if call[0] != "receipts"]
+    assert all(call[1:3] == (str(TENANT), str(USER)) for call in scoped)
+    assert all(call[3] == [str(OWN_FACILITY)] for call in scoped)
+    # Receipts carry no facility: one tenant scoped count, no user or facility filter.
+    assert [call for call in calls if call[0] == "receipts"] == [("receipts", str(TENANT))]
+
+
+def test_epcis_under_min_records_is_not_enough_data(monkeypatch, calls):
+    receipts = RECEIPTS[: MIN_RECORDS - 1]
+    rejected = receipts[::2]
+    client = _client(monkeypatch, calls, receipt_counts=(len(rejected), len(receipts)))
+    [supplier] = client.get("/v1/supplier/data-quality").json()["suppliers"]
+    epcis = supplier["metrics"][3]
+    assert (epcis["status"], epcis["numerator"], epcis["denominator"], epcis["share"]) == (
+        "not_enough_data",
+        len(rejected),
+        len(receipts),
+        None,
+    )
+
+
+def test_epcis_unreadable_receipts_is_unavailable(monkeypatch, calls):
+    [supplier] = _client(monkeypatch, calls, receipt_counts=None).get("/v1/supplier/data-quality").json()["suppliers"]
+    epcis = supplier["metrics"][3]
+    assert (epcis["status"], epcis["reason"]) == ("unavailable", EPCIS_UNAVAILABLE_REASON)
 
 
 @pytest.mark.parametrize("window", [0, MAX_WINDOW_DAYS + 1, -1])