DataOps-platform/app/runner/rule_evidence.py

"""Transactional, server-attested evidence for governed rule execution."""

from __future__ import annotations

import hashlib
import json
import os
import re
import tempfile
from contextlib import suppress
from typing import Any

import polars as pl
from sqlalchemy import text

from app.core.common.identifiers import (
    ensure_governance_uid,
    new_governance_uid,
)

_DIGEST = re.compile(r"^[0-9a-f]{64}$")
_FINAL_STATUSES = {"success", "failed", "unknown", "cancelled"}
_COMMIT_OUTCOMES = {
    "not_applicable",
    "not_committed",
    "committed",
    "unknown",
}
_FINISH_KEYS = {
    "status",
    "rows_in",
    "rows_out",
    "rows_rejected",
    "rows_quarantined",
    "commit_outcome",
    "timings",
    "public_result",
    "violation_sample",
    "sample_count",
    "redaction_policy",
}


def _uid(value: Any, label: str) -> str:
    try:
        return ensure_governance_uid({"uid": str(value)})
    except ValueError as exc:
        raise ValueError(f"{label} is invalid") from exc


def _canonical_digest(value: Any) -> str:
    encoded = json.dumps(
        value,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def _bounded_count(value: Any, label: str) -> int:
    if isinstance(value, bool):
        raise ValueError(f"{label} is invalid")
    try:
        normalized = int(value or 0)
    except (TypeError, ValueError) as exc:
        raise ValueError(f"{label} is invalid") from exc
    if normalized < 0 or normalized > 10_000_000_000:
        raise ValueError(f"{label} is outside the evidence limit")
    return normalized


def _sample_fields(sample: list[dict[str, Any]]) -> list[dict[str, Any]]:
    names = sorted({str(key) for row in sample for key in row})
    if not names:
        raise ValueError("violation sample has no fields")
    return [
        {"name": name, "type": "string", "nullable": True}
        for name in names
    ]


class PostgresRuleEvidenceWriter:
    """Persist one immutable run and at most one expiring violation sample."""

    def __init__(
        self,
        engine,
        artifact_store,
        *,
        sample_ttl_seconds: int = 3600,
    ):
        self.engine = engine
        self.artifact_store = artifact_store
        self.sample_ttl_seconds = int(sample_ttl_seconds)
        if (
            self.sample_ttl_seconds < 1
            or self.sample_ttl_seconds
            > self.artifact_store.max_ttl_seconds
        ):
            raise ValueError("violation sample TTL is invalid")

    def start(
        self,
        *,
        component_binding_id: str,
        rule_version_id: str,
        plan_hash: str,
        correlation_id: str,
        dataflow_uid: str,
        workflow_version: int,
        node_id: str,
    ) -> str:
        component = _uid(component_binding_id, "component binding id")
        rule = _uid(rule_version_id, "rule version id")
        correlation = _uid(correlation_id, "correlation id")
        dataflow = _uid(dataflow_uid, "dataflow id")
        if _DIGEST.fullmatch(str(plan_hash or "")) is None:
            raise ValueError("plan hash is invalid")
        if (
            isinstance(workflow_version, bool)
            or not isinstance(workflow_version, int)
            or workflow_version < 1
        ):
            raise ValueError("workflow version is invalid")
        if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_-]{0,99}", str(node_id)):
            raise ValueError("node id is invalid")
        evidence_key = _canonical_digest(
            {
                "component_binding_id": component,
                "correlation_id": correlation,
                "dataflow_uid": dataflow,
                "node_id": node_id,
                "plan_hash": plan_hash,
                "rule_version_id": rule,
                "workflow_version": workflow_version,
            }
        )
        with self.engine.begin() as connection:
            deployments = connection.execute(
                text(
                    """
                    SELECT d.id::text AS deployment_id
                    FROM public.dataflow_component_bindings b
                    JOIN public.dataflow_versions v
                      ON v.id = b.dataflow_version_id
                    JOIN public.rule_execution_plans p
                      ON p.component_binding_id = b.id
                    JOIN public.dataflow_deployments d
                      ON d.dataflow_version_id = v.id
                    WHERE b.id = CAST(:component_binding_id AS uuid)
                      AND b.rule_version_id =
                          CAST(:rule_version_id AS uuid)
                      AND p.plan_hash = :plan_hash
                      AND p.status = 'published'
                      AND v.dataflow_uid = CAST(:dataflow_uid AS uuid)
                      AND v.version_no = :workflow_version
                      AND d.status IN ('canary','active')
                    ORDER BY
                      CASE d.status WHEN 'active' THEN 0 ELSE 1 END,
                      d.created_at DESC
                    LIMIT 2
                    """
                ),
                {
                    "component_binding_id": component,
                    "rule_version_id": rule,
                    "plan_hash": plan_hash,
                    "dataflow_uid": dataflow,
                    "workflow_version": workflow_version,
                },
            ).mappings().all()
            if len(deployments) != 1:
                raise ValueError(
                    "canonical rule deployment is missing or ambiguous"
                )
            rule_run_id = new_governance_uid()
            selected = connection.execute(
                text(
                    """
                    INSERT INTO public.rule_runs (
                        id, deployment_id, component_binding_id,
                        rule_version_id, plan_hash, status,
                        correlation_id, evidence_key, started_at
                    ) VALUES (
                        CAST(:id AS uuid), CAST(:deployment_id AS uuid),
                        CAST(:component_binding_id AS uuid),
                        CAST(:rule_version_id AS uuid), :plan_hash,
                        'running', CAST(:correlation_id AS uuid),
                        :evidence_key, CURRENT_TIMESTAMP
                    )
                    ON CONFLICT (evidence_key) DO NOTHING
                    RETURNING id::text
                    """
                ),
                {
                    "id": rule_run_id,
                    "deployment_id": deployments[0]["deployment_id"],
                    "component_binding_id": component,
                    "rule_version_id": rule,
                    "plan_hash": plan_hash,
                    "correlation_id": correlation,
                    "evidence_key": evidence_key,
                },
            ).scalar_one_or_none()
            if selected is None:
                selected = connection.execute(
                    text(
                        """
                        SELECT id::text
                        FROM public.rule_runs
                        WHERE evidence_key = :evidence_key
                        """
                    ),
                    {"evidence_key": evidence_key},
                ).scalar_one()
        return str(selected)

    def replay(self, rule_run_id: str) -> dict[str, Any] | None:
        run_id = _uid(rule_run_id, "rule run id")
        with self.engine.connect() as connection:
            row = connection.execute(
                text(
                    """
                    SELECT status, commit_outcome, public_result
                    FROM public.rule_runs
                    WHERE id = CAST(:id AS uuid)
                    """
                ),
                {"id": run_id},
            ).mappings().one_or_none()
        if row is None:
            raise ValueError("rule run was not found")
        if row["status"] in {"queued", "running"}:
            return None
        result = row["public_result"]
        if isinstance(result, str):
            result = json.loads(result)
        return {
            **(dict(result) if isinstance(result, dict) else {}),
            "status": str(row["status"]),
            "commit_outcome": str(row["commit_outcome"]),
        }

    @staticmethod
    def _validate_finish(result: Any) -> dict[str, Any]:
        if not isinstance(result, dict) or set(result) - _FINISH_KEYS:
            raise ValueError("rule evidence result has unsupported fields")
        status = result.get("status")
        commit_outcome = result.get("commit_outcome", "not_applicable")
        if status not in _FINAL_STATUSES:
            raise ValueError("rule evidence status is invalid")
        if commit_outcome not in _COMMIT_OUTCOMES:
            raise ValueError("rule evidence commit outcome is invalid")
        timings = result.get("timings", {})
        if (
            not isinstance(timings, dict)
            or set(timings) != {"duration_ms"}
            or isinstance(timings.get("duration_ms"), bool)
            or not isinstance(timings.get("duration_ms"), int)
            or timings["duration_ms"] < 0
            or timings["duration_ms"] > 86_400_000
        ):
            raise ValueError("rule evidence timings are invalid")
        public_result = result.get("public_result")
        if public_result is not None and (
            not isinstance(public_result, dict)
            or any(str(key).startswith("_") for key in public_result)
            or "rows" in public_result
            or len(
                json.dumps(
                    public_result,
                    sort_keys=True,
                    separators=(",", ":"),
                    ensure_ascii=False,
                ).encode("utf-8")
            )
            > 32_768
        ):
            raise ValueError("public rule result is not evidence safe")
        sample = result.get("violation_sample")
        if sample is not None:
            if (
                status != "success"
                or not isinstance(sample, list)
                or not 1 <= len(sample) <= 100
                or result.get("sample_count") != len(sample)
                or result.get("redaction_policy")
                != "rule-violation-default-v1"
            ):
                raise ValueError("violation sample is invalid")
            for row in sample:
                if not isinstance(row, dict):
                    raise ValueError("violation sample row is invalid")
                for value in row.values():
                    if value not in {None, "[REDACTED]"}:
                        raise ValueError(
                            "violation sample contains unredacted values"
                        )
        return {
            **result,
            "rows_in": _bounded_count(result.get("rows_in"), "rows_in"),
            "rows_out": _bounded_count(result.get("rows_out"), "rows_out"),
            "rows_rejected": _bounded_count(
                result.get("rows_rejected"), "rows_rejected"
            ),
            "rows_quarantined": _bounded_count(
                result.get("rows_quarantined"), "rows_quarantined"
            ),
            "commit_outcome": commit_outcome,
        }

    def _prepare_sample(
        self,
        run_id: str,
        correlation_id: str,
        sample: list[dict[str, Any]],
        redaction_policy: str,
    ) -> tuple[str, dict[str, Any], str]:
        fields = _sample_fields(sample)
        frame = pl.DataFrame(
            {
                field["name"]: [
                    row.get(field["name"]) for row in sample
                ]
                for field in fields
            },
            schema={field["name"]: pl.String for field in fields},
        )
        with tempfile.NamedTemporaryFile(
            prefix="dataops-rule-violation-",
            suffix=".parquet",
            delete=False,
        ) as handle:
            path = handle.name
        sample_id = None
        prepared = None
        try:
            frame.write_parquet(path)
            prepared = self.artifact_store.prepare_path(
                path,
                correlation_id,
                self.sample_ttl_seconds,
                schema_fields=fields,
                limits={
                    "max_rows": min(100, self.artifact_store.max_rows),
                    "max_artifact_bytes": min(
                        4 * 1024 * 1024,
                        self.artifact_store.max_artifact_bytes,
                    ),
                    "memory_limit_bytes": min(
                        16 * 1024 * 1024,
                        self.artifact_store.memory_limit_bytes,
                    ),
                },
            )
            sample_id = new_governance_uid()
            with self.engine.begin() as connection:
                row = connection.execute(
                    text(
                        """
                        SELECT id::text, artifact_ref, artifact_digest,
                               schema_hash, sample_count,
                               redaction_policy, expires_at, handoff_status
                        FROM public.rule_violation_samples
                        WHERE rule_run_id = CAST(:rule_run_id AS uuid)
                        FOR UPDATE
                        """
                    ),
                    {"rule_run_id": run_id},
                ).mappings().one_or_none()
                if row is None:
                    connection.execute(
                        text(
                            """
                            INSERT INTO public.rule_violation_samples (
                                id, rule_run_id, artifact_ref,
                                artifact_digest, schema_hash, sample_count,
                                redaction_policy, expires_at,
                                handoff_status
                            ) VALUES (
                                CAST(:id AS uuid),
                                CAST(:rule_run_id AS uuid), :artifact_ref,
                                :artifact_digest, :schema_hash,
                                :sample_count, :redaction_policy,
                                CAST(:expires_at AS timestamptz), 'pending'
                            )
                            """
                        ),
                        {
                            "id": sample_id,
                            "rule_run_id": run_id,
                            "artifact_ref": prepared["artifact_ref"],
                            "artifact_digest": prepared["digest"],
                            "schema_hash": prepared["schema_hash"],
                            "sample_count": len(sample),
                            "redaction_policy": redaction_policy,
                            "expires_at": prepared["expires_at"],
                        },
                    )
                else:
                    if (
                        str(row["artifact_digest"]) != prepared["digest"]
                        or int(row["sample_count"]) != len(sample)
                        or str(row["redaction_policy"])
                        != redaction_policy
                    ):
                        raise ValueError(
                            "violation sample evidence is immutable"
                        )
                    sample_id = str(row["id"])
                    prepared.update(
                        {
                            "artifact_ref": str(row["artifact_ref"]),
                            "digest": str(row["artifact_digest"]),
                            "schema_hash": str(row["schema_hash"]),
                            "expires_at": str(row["expires_at"]),
                        }
                    )
                    if row["handoff_status"] == "ready":
                        stored = self.artifact_store.describe(
                            prepared["artifact_ref"]
                        )
                        if (
                            stored["digest"] != prepared["digest"]
                            or stored["schema_hash"]
                            != prepared["schema_hash"]
                            or stored["row_count"] != len(sample)
                        ):
                            raise ValueError(
                                "ready violation sample does not match storage"
                            )
                        return sample_id, prepared, path
                    if row["handoff_status"] == "failed":
                        raise ValueError(
                            "violation sample handoff already failed"
                        )
            self.artifact_store.upload_path(
                path,
                prepared,
                limits={
                    "max_rows": min(100, self.artifact_store.max_rows),
                    "max_artifact_bytes": min(
                        4 * 1024 * 1024,
                        self.artifact_store.max_artifact_bytes,
                    ),
                    "memory_limit_bytes": min(
                        16 * 1024 * 1024,
                        self.artifact_store.memory_limit_bytes,
                    ),
                },
            )
            with self.engine.begin() as connection:
                updated = connection.execute(
                    text(
                        """
                        UPDATE public.rule_violation_samples
                        SET handoff_status = 'ready',
                            updated_at = CURRENT_TIMESTAMP
                        WHERE id = CAST(:id AS uuid)
                          AND handoff_status = 'pending'
                          AND artifact_digest = :artifact_digest
                        """
                    ),
                    {
                        "id": sample_id,
                        "artifact_digest": prepared["digest"],
                    },
                )
                if updated.rowcount != 1:
                    state = connection.execute(
                        text(
                            """
                            SELECT handoff_status
                            FROM public.rule_violation_samples
                            WHERE id = CAST(:id AS uuid)
                            """
                        ),
                        {"id": sample_id},
                    ).scalar_one_or_none()
                    if state != "ready":
                        raise RuntimeError(
                            "violation sample finalize outcome is unknown"
                        )
            return sample_id, prepared, path
        except Exception:
            if sample_id is not None and prepared is not None:
                try:
                    with self.engine.connect() as connection:
                        committed = connection.execute(
                            text(
                                """
                                SELECT handoff_status, artifact_digest
                                FROM public.rule_violation_samples
                                WHERE id = CAST(:id AS uuid)
                                """
                            ),
                            {"id": sample_id},
                        ).mappings().one_or_none()
                    if (
                        committed is not None
                        and committed["handoff_status"] == "ready"
                        and str(committed["artifact_digest"])
                        == prepared["digest"]
                    ):
                        return sample_id, prepared, path
                except Exception:
                    pass
            with self.engine.begin() as connection:
                connection.execute(
                    text(
                        """
                        UPDATE public.rule_violation_samples
                        SET handoff_status = 'failed',
                            failure_code = 'sample_handoff_failed',
                            updated_at = CURRENT_TIMESTAMP
                        WHERE rule_run_id = CAST(:rule_run_id AS uuid)
                          AND handoff_status = 'pending'
                        """
                    ),
                    {"rule_run_id": run_id},
                )
            with suppress(FileNotFoundError):
                os.unlink(path)
            raise

    def finish(self, rule_run_id: str, result: Any) -> None:
        run_id = _uid(rule_run_id, "rule run id")
        normalized = self._validate_finish(result)
        sample_path = None
        try:
            with self.engine.connect() as connection:
                current = connection.execute(
                    text(
                        """
                        SELECT status, correlation_id::text
                        FROM public.rule_runs
                        WHERE id = CAST(:id AS uuid)
                        """
                    ),
                    {"id": run_id},
                ).mappings().one_or_none()
            if current is None:
                raise ValueError("rule run was not found")
            if current["status"] not in {"queued", "running"}:
                replay = self.replay(run_id)
                if replay and replay["status"] == normalized["status"]:
                    return
                raise ValueError("rule run evidence is immutable")
            if normalized.get("violation_sample"):
                _sample_id, _prepared, sample_path = self._prepare_sample(
                    run_id,
                    str(current["correlation_id"]),
                    normalized["violation_sample"],
                    normalized["redaction_policy"],
                )
            try:
                with self.engine.begin() as connection:
                    updated = connection.execute(
                        text(
                            """
                            UPDATE public.rule_runs
                            SET rows_in = :rows_in,
                                rows_out = :rows_out,
                                rows_rejected = :rows_rejected,
                                rows_quarantined = :rows_quarantined,
                                status = :status,
                                timings = CAST(:timings AS jsonb),
                                commit_outcome = :commit_outcome,
                                public_result = CAST(:public_result AS jsonb),
                                failure_code = :failure_code,
                                finished_at = CURRENT_TIMESTAMP,
                                updated_at = CURRENT_TIMESTAMP
                            WHERE id = CAST(:id AS uuid)
                              AND status IN ('queued','running')
                            """
                        ),
                        {
                            "id": run_id,
                            "rows_in": normalized["rows_in"],
                            "rows_out": normalized["rows_out"],
                            "rows_rejected": normalized[
                                "rows_rejected"
                            ],
                            "rows_quarantined": normalized[
                                "rows_quarantined"
                            ],
                            "status": normalized["status"],
                            "timings": json.dumps(
                                normalized["timings"]
                            ),
                            "commit_outcome": normalized[
                                "commit_outcome"
                            ],
                            "public_result": (
                                json.dumps(
                                    normalized.get("public_result")
                                )
                                if normalized.get("public_result")
                                is not None
                                else None
                            ),
                            "failure_code": (
                                None
                                if normalized["status"] == "success"
                                else (
                                    f"execution_{normalized['status']}"
                                )
                            ),
                        },
                    )
                    if updated.rowcount != 1:
                        state = connection.execute(
                            text(
                                """
                                SELECT status, commit_outcome
                                FROM public.rule_runs
                                WHERE id = CAST(:id AS uuid)
                                """
                            ),
                            {"id": run_id},
                        ).mappings().one_or_none()
                        if (
                            state is None
                            or state["status"] != normalized["status"]
                            or state["commit_outcome"]
                            != normalized["commit_outcome"]
                        ):
                            raise RuntimeError(
                                "rule run finalize outcome is unknown"
                            )
            except Exception as exc:
                try:
                    replay = self.replay(run_id)
                except Exception as recheck_exc:
                    raise RuntimeError(
                        "rule run finalize outcome is unknown"
                    ) from recheck_exc
                if (
                    replay is None
                    or replay["status"] != normalized["status"]
                    or replay["commit_outcome"]
                    != normalized["commit_outcome"]
                ):
                    raise RuntimeError(
                        "rule run finalize outcome is unknown"
                    ) from exc
        finally:
            if sample_path is not None:
                with suppress(FileNotFoundError):
                    os.unlink(sample_path)

    def cleanup_expired(self, *, limit: int = 100) -> int:
        if isinstance(limit, bool) or not isinstance(limit, int):
            raise ValueError("cleanup limit is invalid")
        if limit < 1 or limit > 1000:
            raise ValueError("cleanup limit is invalid")
        removed = 0
        with self.engine.begin() as connection:
            rows = connection.execute(
                text(
                    """
                    SELECT id::text, artifact_ref
                    FROM public.rule_violation_samples
                    WHERE expires_at <= CURRENT_TIMESTAMP
                      AND handoff_status IN ('legacy','ready','failed')
                    ORDER BY expires_at, id
                    FOR UPDATE SKIP LOCKED
                    LIMIT :limit
                    """
                ),
                {"limit": limit},
            ).mappings().all()
            for row in rows:
                try:
                    self.artifact_store.delete(str(row["artifact_ref"]))
                except Exception:
                    continue
                connection.execute(
                    text(
                        """
                        DELETE FROM public.rule_violation_samples
                        WHERE id = CAST(:id AS uuid)
                        """
                    ),
                    {"id": str(row["id"])},
                )
                removed += 1
        return removed


__all__ = ["PostgresRuleEvidenceWriter"]