feat: metadata drift detector between backend and on-chain records (#…

apps/data-processing/METADATA_DRIFT_DETECTOR.md

@@ -0,0 +1,85 @@
+# Metadata Drift Detector (#882)
+
+## Goal
+Detect when off-chain backend metadata (`ProjectView`, `ProjectMilestone`)
+falls out of sync with on-chain identifiers or status.
+
+## How it works
+`ContractEvent` rows are the raw, immutable record of events ingested from
+the Soroban RPC event stream — the closest thing this service has to
+ground-truth on-chain state. `ProjectView` and `ProjectMilestone` are
+mutable "materialized view" tables that get incrementally updated as events
+arrive (see `PostgresService.save_project_view` / `save_project_milestone`).
+
+`src/metadata_drift_detector.py` recomputes canonical project/milestone
+state directly from the `ContractEvent` log (independent of however the
+materialized views were last updated) and diffs it against what's currently
+persisted. Any mismatch indicates drift — e.g. a missed event, a bug in
+incremental update logic, or a partial write.
+
+Compared fields:
+- **Project**: `status`, `total_contributions`, `unique_contributors`,
+  `last_event_ledger` (staleness only), and whether a `ProjectView` row
+  exists at all.
+- **Milestone**: `status`, and whether a `ProjectMilestone` row exists for
+  every milestone observed in the event log.
+
+## Acceptance criteria mapping
+- **Compares selected backend records against chain-derived state** —
+  `MetadataDriftDetector.run_for_project` / `run_all`.
+- **Produces actionable drift reports** — findings are returned as a
+  `DriftReport` and persisted to the `metadata_drift_findings` table (with
+  `reviewed` / `reviewed_by` / `review_notes` fields, consistent with the
+  existing `RoundAnomalySignal` review workflow).
+- **Supports scheduled and manual execution** — registered as a 6-hourly
+  APScheduler job (`metadata_drift_detection` in `src/scheduler.py`) and
+  exposed as a CLI script (`scripts/detect_metadata_drift.py`).
+- **Does not mutate source data automatically** — the detector only reads
+  `ContractEvent`, `ProjectView`, and `ProjectMilestone`; it never writes to
+  them. Findings are written exclusively to the separate
+  `metadata_drift_findings` table. See
+  `tests/test_metadata_drift_detector.py::test_does_not_mutate_source_records`.
+
+## Usage
+
+```bash
+# Check every project that has at least one ingested contract event
+python scripts/detect_metadata_drift.py
+
+# Check a single project
+python scripts/detect_metadata_drift.py --project-id 101
+
+# Preview findings without persisting them
+python scripts/detect_metadata_drift.py --no-persist --json
+```
+
+Programmatically:
+
+```python
+from src.metadata_drift_detector import MetadataDriftDetector
+
+detector = MetadataDriftDetector()
+report = detector.run_and_persist()  # all known projects
+print(report.to_dict())
+```
+
+## Reviewing findings
+
+```python
+from src.db.postgres_service import PostgresService
+
+service = PostgresService()
+findings = service.get_metadata_drift_findings(reviewed=False, severity="critical")
+service.mark_metadata_drift_finding_reviewed(findings[0].id, reviewed_by="maintainer")
+```
+
+## Limitations / follow-ups
+- Drift is derived from the locally ingested `ContractEvent` log, not a live
+  Soroban RPC read. If event ingestion itself has gaps, this detector won't
+  catch them (that's covered by the separate ingestion quality checks in
+  `src/ingestion/stellar_ingestion_checks.py`). A future iteration could add
+  an optional live-RPC spot-check mode for higher assurance.
+- `last_event_ledger` drift is only flagged when the backend is *behind* the
+  chain-derived ledger (staleness), not when it's ahead, since a backend
+  legitimately may have ingested events not yet reflected in a partial
+  ContractEvent backfill window.

apps/data-processing/scripts/detect_metadata_drift.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+"""
+Metadata Drift Detector — manual execution entrypoint (#882)
+
+Compares backend ProjectView/ProjectMilestone records against state
+recomputed from the raw ContractEvent log and reports any drift.
+Read-only: never mutates ProjectView/ProjectMilestone.
+
+Usage:
+    python scripts/detect_metadata_drift.py
+    python scripts/detect_metadata_drift.py --project-id 101
+    python scripts/detect_metadata_drift.py --limit 1000 --no-persist
+"""
+
+import sys
+import os
+import json
+import argparse
+import logging
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from src.metadata_drift_detector import MetadataDriftDetector
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="Detect drift between backend records and on-chain-derived state."
+    )
+    parser.add_argument(
+        "--project-id",
+        type=int,
+        default=None,
+        help="Check a single project instead of all known projects.",
+    )
+    parser.add_argument(
+        "--limit",
+        type=int,
+        default=500,
+        help="Max number of projects to check when --project-id is not given.",
+    )
+    parser.add_argument(
+        "--no-persist",
+        action="store_true",
+        help="Print findings without writing them to metadata_drift_findings.",
+    )
+    parser.add_argument(
+        "--json",
+        action="store_true",
+        help="Print the report as JSON instead of a human-readable summary.",
+    )
+    args = parser.parse_args()
+
+    detector = MetadataDriftDetector()
+
+    if args.no_persist:
+        report = (
+            detector.run_for_project(args.project_id)
+            if args.project_id is not None
+            else detector.run_all(limit=args.limit)
+        )
+    else:
+        report = detector.run_and_persist(project_id=args.project_id, limit=args.limit)
+
+    if args.json:
+        print(json.dumps(report.to_dict(), indent=2))
+    else:
+        logger.info("=" * 60)
+        logger.info("Metadata Drift Detection Report")
+        logger.info("=" * 60)
+        logger.info(f"Run ID:             {report.run_id}")
+        logger.info(f"Projects checked:   {report.projects_checked}")
+        logger.info(f"Projects with drift: {report.projects_with_drift}")
+        logger.info(f"Total findings:     {len(report.findings)}")
+        for finding in report.findings:
+            logger.info(
+                "  [%s] project=%s %s%s: backend=%r chain_derived=%r",
+                finding.severity.upper(),
+                finding.project_id,
+                finding.scope,
+                f" milestone={finding.milestone_id}" if finding.milestone_id is not None else "",
+                finding.backend_value,
+                finding.chain_derived_value,
+            )
+
+    return 1 if report.drift_detected else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

apps/data-processing/src/db/models.py

@@ -487,6 +487,56 @@ def __repr__(self):
         return f"<AssetTrend(asset={self.asset}, metric={self.metric_name}, trend={self.trend_direction})>"
 
 
+class MetadataDriftFinding(Base):
+    """
+    Stores individual drift findings produced by the metadata drift detector
+    (backend ProjectView/ProjectMilestone records vs. chain-derived state
+    recomputed from the immutable ContractEvent log).
+    """
+
+    __tablename__ = "metadata_drift_findings"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    run_id = Column(String(64), nullable=False, index=True)
+    project_id = Column(BigInteger, nullable=False, index=True)
+    scope = Column(String(20), nullable=False, index=True)  # "project" or "milestone"
+    milestone_id = Column(Integer, nullable=True, index=True)
+    field = Column(String(100), nullable=False, index=True)
+    backend_value = Column(Text, nullable=True)
+    chain_derived_value = Column(Text, nullable=True)
+    severity = Column(String(20), nullable=False, default="warning", index=True)
+    detected_at = Column(DateTime(timezone=True), nullable=False, index=True)
+    created_at = Column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+
+    # Review status (consistent with RoundAnomalySignal review workflow)
+    reviewed = Column(Boolean, nullable=False, default=False)
+    review_notes = Column(Text, nullable=True)
+    reviewed_at = Column(DateTime(timezone=True), nullable=True)
+    reviewed_by = Column(String(255), nullable=True)
+
+    __table_args__ = (
+        Index("idx_metadata_drift_findings_run_id", "run_id"),
+        Index("idx_metadata_drift_findings_project_id", "project_id"),
+        Index("idx_metadata_drift_findings_scope", "scope"),
+        Index("idx_metadata_drift_findings_field", "field"),
+        Index("idx_metadata_drift_findings_severity", "severity"),
+        Index("idx_metadata_drift_findings_reviewed", "reviewed"),
+        Index(
+            "idx_metadata_drift_findings_project_field",
+            "project_id",
+            "field",
+        ),
+    )
+
+    def __repr__(self):
+        return (
+            f"<MetadataDriftFinding(project_id={self.project_id}, scope={self.scope}, "
+            f"field={self.field}, severity={self.severity}, reviewed={self.reviewed})>"
+        )
+
+
 class RoundAnomalySignal(Base):
     """
     Stores anomaly signals detected in quadratic funding rounds for maintainer review.

apps/data-processing/src/db/postgres_service.py

@@ -29,6 +29,7 @@
     NewsInsight,
     AssetTrend,
     RoundAnomalySignal,
+    MetadataDriftFinding,
 )
 from src.analytics.ner_service import NERService
 from src.analytics.onchain_entity_linker import (
@@ -2067,3 +2068,127 @@ def cleanup_old_data(self, days: int = 30) -> Dict[str, int]:
                 "news_insights": 0,
                 "asset_trends": 0,
             }
+
+    # Metadata Drift Finding Methods (#882)
+
+    def save_metadata_drift_finding(
+        self, finding_data: Dict[str, Any]
+    ) -> Optional[MetadataDriftFinding]:
+        """
+        Persist a single metadata drift finding produced by the drift detector.
+
+        Args:
+            finding_data: Dictionary matching MetadataDriftFinding fields
+                (run_id, project_id, scope, milestone_id, field,
+                backend_value, chain_derived_value, severity, detected_at)
+
+        Returns:
+            MetadataDriftFinding object if successful, None otherwise
+        """
+        def _save():
+            with self.get_session() as session:
+                finding = MetadataDriftFinding(
+                    run_id=finding_data["run_id"],
+                    project_id=finding_data["project_id"],
+                    scope=finding_data["scope"],
+                    milestone_id=finding_data.get("milestone_id"),
+                    field=finding_data["field"],
+                    backend_value=finding_data.get("backend_value"),
+                    chain_derived_value=finding_data.get("chain_derived_value"),
+                    severity=finding_data.get("severity", "warning"),
+                    detected_at=finding_data.get("detected_at", datetime.utcnow()),
+                )
+                session.add(finding)
+                session.flush()
+                session.refresh(finding)
+                return finding
+
+        try:
+            return self._retry_operation(_save)
+        except SQLAlchemyError as e:
+            logger.error(f"Failed to save metadata drift finding: {e}")
+            return None
+
+    def save_metadata_drift_findings(
+        self, findings: List[Dict[str, Any]]
+    ) -> int:
+        """
+        Persist multiple metadata drift findings in a batch.
+
+        Args:
+            findings: List of finding data dictionaries
+
+        Returns:
+            Number of findings saved successfully
+        """
+        saved_count = 0
+        for finding_data in findings:
+            if self.save_metadata_drift_finding(finding_data):
+                saved_count += 1
+        logger.info(f"Saved {saved_count}/{len(findings)} metadata drift findings")
+        return saved_count
+
+    def get_metadata_drift_findings(
+        self,
+        run_id: Optional[str] = None,
+        project_id: Optional[int] = None,
+        scope: Optional[str] = None,
+        severity: Optional[str] = None,
+        reviewed: Optional[bool] = None,
+        limit: int = 200,
+    ) -> List[MetadataDriftFinding]:
+        """
+        Retrieve metadata drift findings with optional filters.
+        """
+        try:
+            with self.get_session() as session:
+                query = select(MetadataDriftFinding)
+
+                if run_id is not None:
+                    query = query.where(MetadataDriftFinding.run_id == run_id)
+                if project_id is not None:
+                    query = query.where(MetadataDriftFinding.project_id == project_id)
+                if scope is not None:
+                    query = query.where(MetadataDriftFinding.scope == scope)
+                if severity is not None:
+                    query = query.where(MetadataDriftFinding.severity == severity)
+                if reviewed is not None:
+                    query = query.where(MetadataDriftFinding.reviewed == reviewed)
+
+                query = query.order_by(desc(MetadataDriftFinding.detected_at)).limit(limit)
+                return session.execute(query).scalars().all()
+        except SQLAlchemyError as e:
+            logger.error(f"Failed to get metadata drift findings: {e}")
+            return []
+
+    def mark_metadata_drift_finding_reviewed(
+        self,
+        finding_id: int,
+        reviewed_by: str,
+        review_notes: Optional[str] = None,
+    ) -> bool:
+        """
+        Mark a metadata drift finding as reviewed.
+        """
+        try:
+            with self.get_session() as session:
+                finding = session.execute(
+                    select(MetadataDriftFinding).where(MetadataDriftFinding.id == finding_id)
+                ).scalar_one_or_none()
+
+                if not finding:
+                    logger.warning(f"Metadata drift finding {finding_id} not found")
+                    return False
+
+                finding.reviewed = True
+                finding.reviewed_at = datetime.utcnow()
+                finding.reviewed_by = reviewed_by
+                finding.review_notes = review_notes
+
+                logger.info(
+                    f"Marked metadata drift finding {finding_id} as reviewed by {reviewed_by}"
+                )
+                return True
+        except SQLAlchemyError as e:
+            logger.error(f"Failed to mark metadata drift finding as reviewed: {e}")
+            return False