fix(ingestion): expand EPCIS export KDE joins (Phase B)

services/ingestion/app/epcis_export.py

@@ -98,18 +98,32 @@ def _query_tenant_events(
                     k_to.kde_value AS ship_to,
                     k_to_gln.kde_value AS ship_to_gln,
                     k_carrier.kde_value AS carrier,
-                    k_temp.kde_value AS temperature_c
+                    k_temp.kde_value AS temperature_c,
+                    k_ref.kde_value AS reference_document,
+                    k_tlc_src.kde_value AS tlc_source_reference,
+                    k_prev.kde_value AS immediate_previous_source,
+                    k_field.kde_value AS field_id,
+                    k_harvester.kde_value AS harvester_name,
+                    k_harvester_biz.kde_value AS harvester_business_name,
+                    k_harvester_phone.kde_value AS harvester_business_phone
                 FROM fsma.cte_events e
-                LEFT JOIN tenant_kdes k_from ON k_from.cte_event_id = e.id AND k_from.kde_key = 'ship_from_location'
+                LEFT JOIN tenant_kdes k_from ON k_from.cte_event_id = e.id AND k_from.tenant_id = e.tenant_id AND k_from.kde_key = 'ship_from_location'
                 LEFT JOIN tenant_kdes k_epcis ON k_epcis.cte_event_id = e.id AND k_epcis.tenant_id = e.tenant_id AND k_epcis.kde_key = 'epcis_document_json'
                 LEFT JOIN tenant_kdes k_gtin ON k_gtin.cte_event_id = e.id AND k_gtin.tenant_id = e.tenant_id AND k_gtin.kde_key = 'gtin'
                 LEFT JOIN tenant_kdes k_product ON k_product.cte_event_id = e.id AND k_product.tenant_id = e.tenant_id AND k_product.kde_key = 'product_id'
                 LEFT JOIN tenant_kdes k_epc_class ON k_epc_class.cte_event_id = e.id AND k_epc_class.tenant_id = e.tenant_id AND k_epc_class.kde_key = 'epc_class'
-                LEFT JOIN tenant_kdes k_from_gln ON k_from_gln.cte_event_id = e.id AND k_from_gln.kde_key = 'ship_from_gln'
-                LEFT JOIN tenant_kdes k_to ON k_to.cte_event_id = e.id AND k_to.kde_key = 'ship_to_location'
-                LEFT JOIN tenant_kdes k_to_gln ON k_to_gln.cte_event_id = e.id AND k_to_gln.kde_key = 'ship_to_gln'
-                LEFT JOIN tenant_kdes k_carrier ON k_carrier.cte_event_id = e.id AND k_carrier.kde_key = 'carrier_name'
-                LEFT JOIN tenant_kdes k_temp ON k_temp.cte_event_id = e.id AND k_temp.kde_key = 'temperature_celsius'
+                LEFT JOIN tenant_kdes k_from_gln ON k_from_gln.cte_event_id = e.id AND k_from_gln.tenant_id = e.tenant_id AND k_from_gln.kde_key = 'ship_from_gln'
+                LEFT JOIN tenant_kdes k_to ON k_to.cte_event_id = e.id AND k_to.tenant_id = e.tenant_id AND k_to.kde_key = 'ship_to_location'
+                LEFT JOIN tenant_kdes k_to_gln ON k_to_gln.cte_event_id = e.id AND k_to_gln.tenant_id = e.tenant_id AND k_to_gln.kde_key = 'ship_to_gln'
+                LEFT JOIN tenant_kdes k_carrier ON k_carrier.cte_event_id = e.id AND k_carrier.tenant_id = e.tenant_id AND k_carrier.kde_key = 'carrier_name'
+                LEFT JOIN tenant_kdes k_temp ON k_temp.cte_event_id = e.id AND k_temp.tenant_id = e.tenant_id AND k_temp.kde_key = 'temperature_celsius'
+                LEFT JOIN tenant_kdes k_ref ON k_ref.cte_event_id = e.id AND k_ref.tenant_id = e.tenant_id AND k_ref.kde_key = 'reference_document'
+                LEFT JOIN tenant_kdes k_tlc_src ON k_tlc_src.cte_event_id = e.id AND k_tlc_src.tenant_id = e.tenant_id AND k_tlc_src.kde_key = 'tlc_source_reference'
+                LEFT JOIN tenant_kdes k_prev ON k_prev.cte_event_id = e.id AND k_prev.tenant_id = e.tenant_id AND k_prev.kde_key = 'immediate_previous_source'
+                LEFT JOIN tenant_kdes k_field ON k_field.cte_event_id = e.id AND k_field.tenant_id = e.tenant_id AND k_field.kde_key = 'field_id'
+                LEFT JOIN tenant_kdes k_harvester ON k_harvester.cte_event_id = e.id AND k_harvester.tenant_id = e.tenant_id AND k_harvester.kde_key = 'harvester_name'
+                LEFT JOIN tenant_kdes k_harvester_biz ON k_harvester_biz.cte_event_id = e.id AND k_harvester_biz.tenant_id = e.tenant_id AND k_harvester_biz.kde_key = 'harvester_business_name'
+                LEFT JOIN tenant_kdes k_harvester_phone ON k_harvester_phone.cte_event_id = e.id AND k_harvester_phone.tenant_id = e.tenant_id AND k_harvester_phone.kde_key = 'harvester_business_phone'
                 WHERE e.tenant_id = :tenant_id
                 {"AND " + " AND ".join(where) if where else ""}
                 ORDER BY e.event_timestamp DESC

services/ingestion/tests/test_epcis_export_kde_joins.py

@@ -0,0 +1,129 @@
+"""Shared _query_tenant_events KDE join expansion (Phase B prerequisite).
+
+Synthetic fixtures only. Ensures reference_document, tlc_source_reference,
+immediate_previous_source, and harvester KDEs are selected with tenant_id on
+every cte_kdes join.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from uuid import uuid4
+
+import pytest
+
+_ingestion_dir = Path(__file__).resolve().parent.parent
+if str(_ingestion_dir) not in sys.path:
+    sys.path.insert(0, str(_ingestion_dir))
+_services_dir = _ingestion_dir.parent
+if str(_services_dir) not in sys.path:
+    sys.path.insert(0, str(_services_dir))
+
+pytest.importorskip("fastapi")
+
+import app.epcis_export as epcis_export
+from app.epcis_export import _query_tenant_events
+
+REQUIRED_SELECT_ALIASES = (
+    "reference_document",
+    "tlc_source_reference",
+    "immediate_previous_source",
+    "field_id",
+    "harvester_name",
+    "harvester_business_name",
+    "harvester_business_phone",
+    "ship_from",
+    "ship_to",
+    "ship_from_gln",
+    "ship_to_gln",
+)
+
+
+class _FakeResult:
+    def __init__(self, rows):
+        self._rows = rows
+
+    def fetchall(self):
+        return list(self._rows)
+
+
+class _FakeSession:
+    def __init__(self):
+        self.sql = ""
+        self.params = None
+        self.closed = False
+
+    def execute(self, stmt, params=None):
+        self.sql = str(stmt)
+        self.params = params
+        return _FakeResult([])
+
+    def close(self):
+        self.closed = True
+
+
+def test_query_selects_shipping_and_harvest_kde_aliases(monkeypatch: pytest.MonkeyPatch) -> None:
+    session = _FakeSession()
+    monkeypatch.setattr(epcis_export, "get_db_safe", lambda: session)
+    assert _query_tenant_events(str(uuid4()), None, None, None) == []
+    sql = " ".join(session.sql.split())
+    for alias in REQUIRED_SELECT_ALIASES:
+        assert f"AS {alias}" in sql or f"as {alias}" in sql
+    assert "kde_key = 'reference_document'" in sql
+    assert "kde_key = 'tlc_source_reference'" in sql
+    assert "kde_key = 'immediate_previous_source'" in sql
+    assert "kde_key = 'harvester_name'" in sql
+    assert "kde_key = 'field_id'" in sql
+
+
+def test_every_tenant_kdes_join_has_tenant_predicate(monkeypatch: pytest.MonkeyPatch) -> None:
+    session = _FakeSession()
+    monkeypatch.setattr(epcis_export, "get_db_safe", lambda: session)
+    _query_tenant_events(str(uuid4()), None, None, None)
+    sql = session.sql
+    # Each LEFT JOIN tenant_kdes alias must also filter tenant_id = e.tenant_id
+    import re
+
+    joins = re.findall(r"LEFT JOIN tenant_kdes (\w+) ON ([^\n]+)", sql)
+    assert len(joins) >= 17
+    for alias, cond in joins:
+        assert f"{alias}.tenant_id = e.tenant_id" in cond, alias
+    assert session.closed
+
+
+def test_joined_row_mapping_exposes_new_keys(monkeypatch: pytest.MonkeyPatch) -> None:
+    class _Row:
+        def __init__(self, mapping):
+            self._mapping = mapping
+
+    class _Session:
+        def execute(self, *_a, **_k):
+            return _FakeResult(
+                [
+                    _Row(
+                        {
+                            "event_type": "shipping",
+                            "traceability_lot_code": "LOT-ROM-2026-0312A",
+                            "reference_document": "BM:BOL-9001",
+                            "tlc_source_reference": "https://id.gs1.org/414/6141410000014",
+                            "immediate_previous_source": None,
+                            "field_id": None,
+                            "harvester_name": None,
+                            "harvester_business_name": None,
+                            "harvester_business_phone": None,
+                            "ship_from_gln": "6141410000014",
+                            "ship_to_gln": "6141410000069",
+                        }
+                    )
+                ]
+            )
+
+        def close(self):
+            pass
+
+    monkeypatch.setattr(epcis_export, "get_db_safe", lambda: _Session())
+    rows = _query_tenant_events(str(uuid4()), None, None, None)
+    assert rows[0]["reference_document"] == "BM:BOL-9001"
+    assert rows[0]["tlc_source_reference"].startswith("https://id.gs1.org/414/")
+    assert rows[0]["ship_from_gln"] == "6141410000014"

services/ingestion/tests/test_second_format_kde_map.py

@@ -0,0 +1,187 @@
+"""Second format shipping KDE map cases T1 to T5 (synthetic fixtures only).
+
+Locks the map in /workspace/rnd/second-format-kde-map.md against
+services/ingestion/tests/fixtures. No partner file. SEL-549 hold stays.
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+_ingestion_dir = Path(__file__).resolve().parent.parent
+if str(_ingestion_dir) not in sys.path:
+    sys.path.insert(0, str(_ingestion_dir))
+_services_dir = _ingestion_dir.parent
+if str(_services_dir) not in sys.path:
+    sys.path.insert(0, str(_services_dir))
+
+from app.webhook_models import REQUIRED_KDES_BY_CTE, WebhookCTEType
+from shared.fsma_validation import validate_tlc
+
+FIXTURES = Path(__file__).resolve().parent / "fixtures"
+ASN_EXPECTED = FIXTURES / "edi_corpus" / "asn_856.expected.json"
+HARVEST_CSV = FIXTURES / "01_harvesting.csv"
+
+SHIPPING_REQUIRED = list(REQUIRED_KDES_BY_CTE[WebhookCTEType.SHIPPING])
+
+
+def _harvest_lot() -> str:
+    with HARVEST_CSV.open(newline="") as fh:
+        row = next(csv.DictReader(fh))
+    return row["traceability_lot_code"]
+
+
+def _asn_to_shipping_kdes(asn: dict, *, lot=None, tlc_source=None, extras=None) -> dict:
+    """Map asn_856.expected.json shape onto REQUIRED shipping KDEs (map B)."""
+    item = asn["line_items"][0]
+    raw_date = asn.get("ship_date_raw") or ""
+    ship_date = None
+    if len(raw_date) == 8 and raw_date.isdigit():
+        ship_date = f"{raw_date[0:4]}-{raw_date[4:6]}-{raw_date[6:8]}"
+    kdes = {
+        "traceability_lot_code": lot if lot is not None else item.get("lot_number"),
+        "product_description": item.get("product_description"),
+        "quantity": item.get("quantity"),
+        "unit_of_measure": item.get("unit_of_measure"),
+        "ship_date": ship_date,
+        "ship_from_location": asn.get("ship_from_name"),
+        "ship_to_location": asn.get("ship_to_name"),
+        "reference_document": (
+            f"BM:{asn['reference_document_number']}"
+            if asn.get("reference_document_number")
+            else None
+        ),
+        "tlc_source_reference": tlc_source,
+    }
+    unmapped = {
+        "ship_from_gln": asn.get("ship_from_gln"),
+        "ship_to_gln": asn.get("ship_to_gln"),
+        "asn_number": asn.get("asn_number"),
+        "carrier": asn.get("carrier"),
+    }
+    if extras:
+        unmapped.update(extras)
+    return {"kdes": kdes, "unmapped_fields": unmapped}
+
+
+def _gaps(kdes: dict) -> list[str]:
+    missing = []
+    for key in SHIPPING_REQUIRED:
+        val = kdes.get(key)
+        if val is None or val == "":
+            missing.append(key)
+    return missing
+
+
+def test_t1_asn_856_expected_gaps() -> None:
+    """T1: current asn_856 fixture is gap (missing TLC and TLC source)."""
+    asn = json.loads(ASN_EXPECTED.read_text())
+    assert asn["line_items"][0]["lot_number"] is None
+    mapped = _asn_to_shipping_kdes(asn)
+    gaps = _gaps(mapped["kdes"])
+    assert "traceability_lot_code" in gaps
+    assert "tlc_source_reference" in gaps
+    # Present fields from fixture must not be dropped
+    assert mapped["kdes"]["product_description"] == "ROMAINE-12CT"
+    assert mapped["kdes"]["quantity"] == 200.0
+    assert mapped["kdes"]["reference_document"] == "BM:BOL-9001"
+    assert mapped["unmapped_fields"]["ship_from_gln"] == "6141410000014"
+
+
+def test_t2_asn_plus_synthetic_lot_and_tlc_source_can_pass() -> None:
+    """T2: asn_856 + harvest lot + synthetic TLC source fills required nine."""
+    asn = json.loads(ASN_EXPECTED.read_text())
+    lot = _harvest_lot()
+    mapped = _asn_to_shipping_kdes(
+        asn,
+        lot=lot,
+        tlc_source="https://id.gs1.org/414/6141410000014",
+    )
+    assert _gaps(mapped["kdes"]) == []
+    lenient = validate_tlc(lot, strict=False)
+    assert lenient.is_valid
+    strict = validate_tlc(lot, strict=True)
+    # Harvest style lot: a GS1 pattern mismatch is a warning, not a failure (#3719)
+    assert strict.is_valid
+    assert any("production pattern" in w for w in strict.warnings)
+
+
+def test_t3_strict_style_tlc_no_format_warning() -> None:
+    """T3: production style TLC passes lenient and strict format checks."""
+    asn = json.loads(ASN_EXPECTED.read_text())
+    lot = "00012345678901-LotA"
+    mapped = _asn_to_shipping_kdes(
+        asn,
+        lot=lot,
+        tlc_source="https://id.gs1.org/414/6141410000014",
+    )
+    assert _gaps(mapped["kdes"]) == []
+    assert validate_tlc(lot, strict=False).is_valid
+    assert validate_tlc(lot, strict=True).is_valid
+
+
+def test_t4_fda_spreadsheet_shaped_shipping_row() -> None:
+    """T4: FDA Shipping sheet shaped synthetic row maps to the same nine KDEs."""
+    lot = _harvest_lot()
+    row = {
+        "Traceability Lot Code (assigned by Initial Packer / First Land Based Receiver / person who Transformed the Food)": lot,
+        "Quantity": "200",
+        "Unit of Measure": "cases",
+        "Product Description - Product Name": "Organic Romaine Hearts",
+        "Product Description - Brand Name (if applicable)": "Valley Fresh",
+        "Shipping Date (mm/dd/yyyy)": "03/12/2026",
+        "Location Description for the Immediate Subsequent Recipient - Business Name": "Metro Distribution Center",
+        "Location Description for the Location from which you Shipped the Food - Business Name": "Valley Fresh Farms",
+        "Traceability Lot Code Source Reference (may be provided instead of the TLC Source location description)": "https://id.gs1.org/414/6141410000014",
+        "Reference Document Type": "BOL",
+        "Reference Document Number": "BOL-9001",
+        "PartnerInternalScore": "42",
+    }
+    kdes = {
+        "traceability_lot_code": row[
+            "Traceability Lot Code (assigned by Initial Packer / First Land Based Receiver / person who Transformed the Food)"
+        ],
+        "product_description": row["Product Description - Product Name"],
+        "quantity": float(row["Quantity"]),
+        "unit_of_measure": row["Unit of Measure"],
+        "ship_date": "2026-03-12",
+        "ship_from_location": row[
+            "Location Description for the Location from which you Shipped the Food - Business Name"
+        ],
+        "ship_to_location": row[
+            "Location Description for the Immediate Subsequent Recipient - Business Name"
+        ],
+        "reference_document": f"{row['Reference Document Type']}:{row['Reference Document Number']}",
+        "tlc_source_reference": row[
+            "Traceability Lot Code Source Reference (may be provided instead of the TLC Source location description)"
+        ],
+    }
+    extras = {
+        "Product Description - Brand Name (if applicable)": row[
+            "Product Description - Brand Name (if applicable)"
+        ],
+        "PartnerInternalScore": row["PartnerInternalScore"],
+    }
+    assert _gaps(kdes) == []
+    assert extras["PartnerInternalScore"] == "42"
+    assert "Valley Fresh" in extras["Product Description - Brand Name (if applicable)"]
+
+
+def test_t5_unknown_column_preserved() -> None:
+    """T5: deliberate extra column is flagged and retained, never dropped."""
+    asn = json.loads(ASN_EXPECTED.read_text())
+    mapped = _asn_to_shipping_kdes(
+        asn,
+        lot=_harvest_lot(),
+        tlc_source="https://id.gs1.org/414/6141410000014",
+        extras={"PartnerInternalScore": "42", "REF*ZZ*CUSTOM": "KEEP-ME"},
+    )
+    assert "PartnerInternalScore" in mapped["unmapped_fields"]
+    assert mapped["unmapped_fields"]["REF*ZZ*CUSTOM"] == "KEEP-ME"
+    assert mapped["unmapped_fields"]["PartnerInternalScore"] == "42"
+    assert _gaps(mapped["kdes"]) == []