feat(storage): cut over first-file uploads to partner-uploads

.github/workflows/backend-ci.yml

@@ -1409,6 +1409,10 @@ jobs:
         env:
           DATABASE_URL: postgresql://postgres:postgres@localhost:5432/live_suite  # pragma: allowlist secret
           ADMIN_DATABASE_URL: postgresql://regengine_sysadmin:admin-runtime-test-password@localhost:5432/live_suite  # pragma: allowlist secret
+          # No object storage in this job. Partner uploads fail closed unless
+          # the explicit non-production bytea fallback is on, so opt in here
+          # and the CSV live tests keep exercising the runtime role.
+          REGENGINE_ALLOW_MANIFEST_BYTEA_FALLBACK: "1"
         run: |
           repo_root="$(pwd)"
           export PYTHONPATH="${repo_root}:${repo_root}/services:${PYTHONPATH}"

services/ingestion/app/partner_file_storage.py

@@ -0,0 +1,246 @@
+"""First-file partner upload persistence (SEL-549 / partner-uploads).
+
+Uploads land in the private ``partner-uploads`` bucket under a tenant-scoped
+path. Metadata is recorded in ``public.partner_files``; ``fsma.manifest_blobs``
+keeps the SHA-256 + ``storage_url`` pointer and leaves ``raw_bytes`` NULL when
+object storage is configured.
+
+Synthetic / test fixtures only until SEL-590 clears. Do not invent partner
+payloads; callers pass bytes they already hold.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import logging
+import re
+from typing import Any, Dict, Optional
+from uuid import uuid4
+
+from fastapi import HTTPException
+from sqlalchemy import text
+
+logger = logging.getLogger("partner_file_storage")
+
+PARTNER_UPLOADS_BUCKET = "partner-uploads"
+
+def allow_manifest_bytea_fallback() -> bool:
+    """Permit raw_bytes fallback only in non-production with an explicit flag.
+
+    Production always fail-closes when object storage is not configured so
+    partner CSV payloads cannot silently land in Postgres. Local/test must set
+    ``REGENGINE_ALLOW_MANIFEST_BYTEA_FALLBACK=1`` to opt into the legacy path.
+    """
+    from shared.env import is_production
+
+    if is_production():
+        return False
+    flag = os.getenv("REGENGINE_ALLOW_MANIFEST_BYTEA_FALLBACK", "").strip().lower()
+    return flag in {"1", "true", "yes", "on"}
+
+
+_SAFE_NAME = re.compile(r"[^A-Za-z0-9._-]+")
+
+
+def tenant_object_key(
+    *,
+    tenant_id: str,
+    engagement_id: str,
+    filename: str,
+    object_id: Optional[str] = None,
+) -> str:
+    """Build ``tenants/{tenant}/engagements/{engagement}/{uuid}-{filename}``."""
+    from .s3_utils import safe_s3_key_component, _validate_s3_key
+
+    oid = object_id or str(uuid4())
+    safe_name = safe_s3_key_component(filename, fallback="upload.csv")
+    key = (
+        f"tenants/{safe_s3_key_component(tenant_id, fallback='tenant')}"
+        f"/engagements/{safe_s3_key_component(engagement_id, fallback='engagement')}"
+        f"/{safe_s3_key_component(oid, fallback='object')}-{safe_name}"
+    )
+    _validate_s3_key(key)
+    return key
+
+
+def persist_partner_upload(
+    db,
+    *,
+    tenant_id: str,
+    engagement_id: str,
+    content: bytes,
+    filename: str,
+    content_type: Optional[str],
+    uploaded_by: Optional[str],
+    ingestion_run_id: str,
+) -> Dict[str, Any]:
+    """Store bytes in partner-uploads and write metadata + blob pointer.
+
+    Production fail-closes with HTTP 503 when object storage is not configured.
+    Local/test may set ``REGENGINE_ALLOW_MANIFEST_BYTEA_FALLBACK=1`` to keep the
+    legacy ``manifest_blobs.raw_bytes`` path.
+
+    Returns a dict with ``partner_file_id``, ``storage_url``, ``storage_path``,
+    ``sha256``, and ``used_object_storage``.
+    """
+    from .remediation.audit_utils import audit_event
+    from .s3_utils import is_object_storage_configured, put_bytes
+
+    if not content:
+        raise HTTPException(status_code=400, detail="Empty upload")
+
+    sha256 = hashlib.sha256(content).hexdigest()
+    ctype = content_type or "text/csv"
+    partner_file_id = str(uuid4())
+    storage_path = tenant_object_key(
+        tenant_id=tenant_id,
+        engagement_id=engagement_id,
+        filename=filename,
+        object_id=partner_file_id,
+    )
+    used_object_storage = False
+    storage_url: Optional[str] = None
+    raw_bytes_param: Optional[bytes] = None
+
+    if is_object_storage_configured():
+        storage_url = put_bytes(
+            PARTNER_UPLOADS_BUCKET,
+            storage_path,
+            content,
+            ctype,
+        )
+        used_object_storage = True
+        raw_bytes_param = None  # pointer-only row; bytes live in the bucket
+        db.execute(
+            # nosemgrep: avoid-sqlalchemy-text — parameterized with :param
+            text(
+                """
+                INSERT INTO public.partner_files (
+                    id, tenant_id, engagement_id, storage_bucket, storage_path,
+                    content_sha256, byte_size, content_type, original_filename,
+                    status
+                ) VALUES (
+                    CAST(:id AS uuid),
+                    CAST(:tenant_id AS uuid),
+                    CAST(:engagement_id AS uuid),
+                    :bucket,
+                    :storage_path,
+                    :sha256,
+                    :byte_size,
+                    :content_type,
+                    :filename,
+                    'uploaded'
+                )
+                """
+            ),
+            {
+                "id": partner_file_id,
+                "tenant_id": tenant_id,
+                "engagement_id": engagement_id,
+                "bucket": PARTNER_UPLOADS_BUCKET,
+                "storage_path": storage_path,
+                "sha256": sha256,
+                "byte_size": len(content),
+                "content_type": ctype,
+                "filename": filename,
+            },
+        )
+    else:
+        if not allow_manifest_bytea_fallback():
+            logger.error(
+                "partner_upload_storage_unconfigured tenant=%s run=%s",
+                tenant_id,
+                ingestion_run_id,
+            )
+            raise HTTPException(
+                status_code=503,
+                detail=(
+                    "Partner object storage is not configured. "
+                    "Refusing to store upload bytes in the database."
+                ),
+            )
+        # Explicit local/test fallback only.
+        raw_bytes_param = content
+        storage_url = None
+
+    db.execute(
+        # nosemgrep: avoid-sqlalchemy-text — parameterized with :param
+        text(
+            """
+            INSERT INTO fsma.manifest_blobs
+                (ingestion_run_id, tenant_id, raw_bytes, sha256,
+                 original_filename, content_type, uploaded_by, storage_url)
+            VALUES
+                (CAST(:run_id AS uuid), CAST(:tenant_id AS uuid), :raw_bytes,
+                 :sha256, :filename, :content_type, :uploaded_by, :storage_url)
+            """
+        ),
+        {
+            "run_id": ingestion_run_id,
+            "tenant_id": tenant_id,
+            "raw_bytes": raw_bytes_param,
+            "sha256": sha256,
+            "filename": filename,
+            "content_type": ctype,
+            "uploaded_by": uploaded_by,
+            "storage_url": storage_url,
+        },
+    )
+
+    audit_event(
+        db,
+        tenant_id=tenant_id,
+        event_type="partner_file.upload",
+        action="partner_file.upload",
+        severity="info",
+        actor_id=uploaded_by,
+        resource_type="partner_file" if used_object_storage else "manifest_blob",
+        resource_id=partner_file_id if used_object_storage else ingestion_run_id,
+        metadata={
+            "ingestion_run_id": ingestion_run_id,
+            "engagement_id": engagement_id,
+            "storage_bucket": PARTNER_UPLOADS_BUCKET if used_object_storage else None,
+            "storage_path": storage_path if used_object_storage else None,
+            "storage_url": storage_url,
+            "content_sha256": sha256,
+            "byte_size": len(content),
+            "original_filename": filename,
+            "used_object_storage": used_object_storage,
+        },
+    )
+
+    return {
+        "partner_file_id": partner_file_id if used_object_storage else None,
+        "storage_url": storage_url,
+        "storage_path": storage_path if used_object_storage else None,
+        "sha256": sha256,
+        "used_object_storage": used_object_storage,
+    }
+
+
+def load_bytes_from_storage_url(storage_url: str) -> bytes:
+    """Fetch object bytes for an ``s3://bucket/key`` storage_url."""
+    from .s3_utils import get_raw_document
+
+    if not storage_url.startswith("s3://"):
+        raise HTTPException(status_code=500, detail="Unsupported storage_url scheme")
+    without = storage_url[len("s3://") :]
+    bucket, _, key = without.partition("/")
+    if not bucket or not key:
+        raise HTTPException(status_code=500, detail="Malformed storage_url")
+    doc = get_raw_document(key=key, bucket=bucket)
+    return doc["content"]
+
+
+def resolve_manifest_raw_bytes(
+    *,
+    raw_bytes: Any,
+    storage_url: Optional[str],
+) -> bytes:
+    """Return manifest bytes from bytea or, when null, from storage_url."""
+    if raw_bytes is not None:
+        return bytes(raw_bytes)
+    if storage_url:
+        return load_bytes_from_storage_url(storage_url)
+    raise HTTPException(status_code=404, detail="Manifest payload not found")

services/ingestion/app/remediation/service.py

@@ -109,53 +109,21 @@ def ingest_manifest(
     )
     run_id = store.create_ingestion_run(run)
 
-    db.execute(
-        # nosemgrep: avoid-sqlalchemy-text — parameterized with :param
-        text(
-            """
-            INSERT INTO fsma.manifest_blobs
-                (ingestion_run_id, tenant_id, raw_bytes, sha256,
-                 original_filename, content_type, uploaded_by)
-            VALUES
-                (CAST(:run_id AS uuid), CAST(:tenant_id AS uuid), :raw_bytes,
-                 :sha256, :filename, :content_type, :uploaded_by)
-            """
-        ),
-        {
-            "run_id": run_id,
-            "tenant_id": tenant_id,
-            "raw_bytes": raw_bytes,
-            "sha256": sha256,
-            "filename": filename,
-            "content_type": content_type,
-            "uploaded_by": uploaded_by,
-        },
-    )
+    # First-file cutover: prefer private partner-uploads (pointer-only
+    # manifest_blobs + partner_files metadata + audit). Falls back to
+    # bytea when object storage is not configured (local/dev).
+    from ..partner_file_storage import persist_partner_upload
 
-    # Optional off-box mirror; the bytea row remains the authoritative copy.
-    try:
-        from ..s3_utils import RAW_INGEST_BUCKET, is_object_storage_configured, put_bytes
-
-        if is_object_storage_configured():
-            storage_url = put_bytes(
-                RAW_INGEST_BUCKET,
-                f"remediation/{tenant_id}/{run_id}/{filename}",
-                raw_bytes,
-                content_type or "text/csv",
-            )
-            db.execute(
-                # nosemgrep: avoid-sqlalchemy-text — parameterized with :param
-                text(
-                    """
-                    UPDATE fsma.manifest_blobs SET storage_url = :storage_url
-                    WHERE ingestion_run_id = CAST(:run_id AS uuid)
-                      AND tenant_id = CAST(:tenant_id AS uuid)
-                    """
-                ),
-                {"storage_url": storage_url, "run_id": run_id, "tenant_id": tenant_id},
-            )
-    except Exception as exc:  # pragma: no cover — mirror is best-effort
-        logger.warning("manifest_blob_mirror_failed run=%s error=%s", run_id, exc)
+    persist_partner_upload(
+        db,
+        tenant_id=tenant_id,
+        engagement_id=run_id,
+        content=raw_bytes,
+        filename=filename,
+        content_type=content_type,
+        uploaded_by=uploaded_by,
+        ingestion_run_id=run_id,
+    )
 
     seen_fingerprints: Dict[str, int] = {}
     records: List[Dict[str, Any]] = []
@@ -787,12 +755,14 @@ def release_record(db, *, tenant_id: str, record: Dict[str, Any], source_meta: D
 
 
 def get_run_meta(db, *, tenant_id: str, run_id: str) -> Dict[str, Any]:
+    from ..partner_file_storage import resolve_manifest_raw_bytes
+
     row = db.execute(
         # nosemgrep: avoid-sqlalchemy-text — parameterized with :param
         text(
             """
-            SELECT b.sha256, b.original_filename, b.raw_bytes, r.status,
-                   r.record_count, r.accepted_count, r.rejected_count
+            SELECT b.sha256, b.original_filename, b.raw_bytes, b.storage_url,
+                   r.status, r.record_count, r.accepted_count, r.rejected_count
             FROM fsma.manifest_blobs b
             JOIN fsma.ingestion_runs r ON r.id = b.ingestion_run_id
             WHERE b.ingestion_run_id = CAST(:run_id AS uuid)
@@ -807,11 +777,11 @@ def get_run_meta(db, *, tenant_id: str, run_id: str) -> Dict[str, Any]:
     return {
         "sha256": row[0],
         "filename": row[1],
-        "raw_bytes": bytes(row[2]),
-        "run_status": row[3],
-        "record_count": row[4],
-        "accepted_count": row[5],
-        "rejected_count": row[6],
+        "raw_bytes": resolve_manifest_raw_bytes(raw_bytes=row[2], storage_url=row[3]),
+        "run_status": row[4],
+        "record_count": row[5],
+        "accepted_count": row[6],
+        "rejected_count": row[7],
     }
 
 

services/ingestion/app/webhook_compat.py

@@ -173,27 +173,18 @@ def _open_ingestion_run(db_session, tenant_id: str, source: str, receipt: Source
         initiated_by=receipt.initiated_by,
     )
     run_id = store.create_ingestion_run(run)
-    db_session.execute(
-        # nosemgrep: avoid-sqlalchemy-text — parameterized with :param
-        text(
-            """
-            INSERT INTO fsma.manifest_blobs
-                (ingestion_run_id, tenant_id, raw_bytes, sha256,
-                 original_filename, content_type, uploaded_by)
-            VALUES
-                (CAST(:run_id AS uuid), CAST(:tenant_id AS uuid), :raw_bytes,
-                 :sha256, :filename, :content_type, :uploaded_by)
-            """
-        ),
-        {
-            "run_id": run_id,
-            "tenant_id": tenant_id,
-            "raw_bytes": receipt.raw_bytes,
-            "sha256": receipt.sha256,
-            "filename": receipt.filename,
-            "content_type": receipt.content_type,
-            "uploaded_by": receipt.initiated_by,
-        },
+    # First-file cutover: partner-uploads + pointer when object storage is on.
+    from .partner_file_storage import persist_partner_upload
+
+    persist_partner_upload(
+        db_session,
+        tenant_id=tenant_id,
+        engagement_id=run_id,
+        content=receipt.raw_bytes,
+        filename=receipt.filename,
+        content_type=receipt.content_type,
+        uploaded_by=receipt.initiated_by,
+        ingestion_run_id=run_id,
     )
     return run_id