Syncing ArcGIS Online Edits to Local GeoPackage

import time
import json
import sqlite3
import logging
from datetime import datetime, timezone, timedelta
from typing import Any, Callable, Optional

from arcgis.gis import GIS
from arcgis.features import FeatureLayer
from requests.exceptions import RequestException, Timeout, HTTPError

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger("agol_gpkg_sync")

# ICS authority tiers: higher wins a same-second last_edited_date tie.
ICS_PRECEDENCE: dict[str, int] = {"FED": 3, "STATE": 2, "LOCAL": 1}


class AGOLToGeoPackageSync:
    """Deterministic, audited AGOL feature-service -> local GeoPackage sync."""

    def __init__(
        self,
        agol_url: str,
        gpkg_path: str,
        max_retries: int = 3,
        backoff_factor: float = 2.0,
    ) -> None:
        self.gis = GIS(agol_url)
        self.gpkg_path = gpkg_path
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor

    def _with_backoff(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> Optional[Any]:
        """Run a network call with exponential backoff; queue offline on exhaustion."""
        for attempt in range(self.max_retries):
            try:
                return func(*args, **kwargs)
            except HTTPError as exc:
                status = getattr(exc.response, "status_code", None)
                if status != 429:  # only retry rate-limit; re-raise hard errors
                    logger.error("Non-retryable HTTP %s from feature service: %s", status, exc)
                    raise
                wait = self.backoff_factor ** attempt
                logger.warning("Throttled (429); backing off %.1fs", wait)
                time.sleep(wait)
            except (Timeout, RequestException) as exc:
                wait = self.backoff_factor ** attempt
                logger.warning("Network degradation (attempt %d): %s; waiting %.1fs",
                               attempt + 1, exc, wait)
                time.sleep(wait)
        logger.critical("Retries exhausted; routing pull to offline delta queue.")
        self._queue_offline()
        return None

    def _queue_offline(self) -> None:
        """Persist intent to re-pull on the next connected window (DLQ stub)."""
        logger.info("Offline fallback engaged; this window will be retried.")

    def sync(self, layer_url: str, window_hours: int = 2) -> None:
        """Pull a bounded delta and merge it into the GeoPackage transactionally."""
        layer = FeatureLayer(layer_url, self.gis)
        cutoff = (datetime.now(timezone.utc) - timedelta(hours=window_hours)).strftime(
            "%Y-%m-%d %H:%M:%S"
        )
        params: dict[str, Any] = {
            "where": f"last_edited_date >= TIMESTAMP '{cutoff}'",
            "out_fields": "globalid,last_edited_date,agency_type",
            "return_geometry": True,
            "f": "geojson",
        }
        result = self._with_backoff(layer.query, **params)
        if not result:
            return

        # --- Tier 2: stage in an isolated workspace, never the live file ---
        stage = sqlite3.connect(":memory:")
        stage.execute(
            "CREATE TABLE delta (globalid TEXT, agency_type TEXT, "
            "last_edited_date TEXT, geometry TEXT)"
        )
        for feat in result.features:
            stage.execute(
                "INSERT INTO delta VALUES (?, ?, ?, ?)",
                (
                    feat.attributes.get("globalid"),
                    feat.attributes.get("agency_type"),
                    feat.attributes.get("last_edited_date"),
                    json.dumps(feat.geometry),
                ),
            )
        stage.commit()
        rows = stage.execute("SELECT * FROM delta").fetchall()
        stage.close()
        logger.info("Staged %d delta records for window starting %s", len(rows), cutoff)

        self._commit(rows)

    def _resolve(self, incoming: tuple, existing: Optional[tuple]) -> tuple[str, dict]:
        """Decide the winner for one merge key; return (action, audit_payload)."""
        gid, agency, edited, geom = incoming
        if existing is None:
            return "insert", {"rule": "new_feature", "globalid": gid}

        ex_gid, ex_agency, ex_edited, _ = existing

        # Tier 3: same key, different physical feature -> re-key local, keep lineage.
        if gid == ex_gid and geom_signature(geom) != geom_signature(existing[3]):
            return "rekey", {"rule": "globalid_collision", "globalid": gid,
                             "relabelled": f"LOCAL-{ex_gid}", "parent_globalid": ex_gid}

        # Tier 4: timestamp race -> ICS precedence, then last-writer-wins.
        if edited == ex_edited:
            inc_rank = ICS_PRECEDENCE.get(str(agency).split("-")[0], 0)
            ex_rank = ICS_PRECEDENCE.get(str(ex_agency).split("-")[0], 0)
            if inc_rank != ex_rank:
                winner = "incoming" if inc_rank > ex_rank else "existing"
                return ("update" if winner == "incoming" else "keep",
                        {"rule": "ics_precedence", "globalid": gid, "winner": winner})
            return "update", {"rule": "tie_last_writer_wins", "globalid": gid}

        return ("update" if edited > ex_edited else "keep",
                {"rule": "recency", "globalid": gid})

    def _commit(self, rows: list[tuple]) -> None:
        """Apply reconciled rows + audit in a single WAL transaction."""
        conn = sqlite3.connect(self.gpkg_path)
        # Tier 5: WAL + busy_timeout so a lock contender waits, never corrupts.
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute("PRAGMA busy_timeout=5000")
        try:
            conn.execute("BEGIN")
            for row in rows:
                gid = row[0]
                existing = conn.execute(
                    "SELECT globalid, agency_type, last_edited_date, geometry "
                    "FROM features WHERE globalid = ?",
                    (gid,),
                ).fetchone()
                action, audit = self._resolve(row, existing)
                self._apply(conn, action, row)
                # Tier 6: immutable audit row per decision, before commit.
                conn.execute(
                    "INSERT INTO sync_audit (ts, action, payload) VALUES (?, ?, ?)",
                    (datetime.now(timezone.utc).isoformat(), action, json.dumps(audit)),
                )
                logger.info("Resolved %s -> %s (%s)", gid, action, audit["rule"])
            conn.commit()
            logger.info("GeoPackage sync committed: %d records.", len(rows))
        except sqlite3.OperationalError as exc:
            conn.rollback()
            logger.error("Lock/IO failure; rolled back, replica untouched: %s", exc)
            self._queue_offline()
        except Exception as exc:  # noqa: BLE001 - never leave a half-applied merge
            conn.rollback()
            logger.exception("Merge failed; rolled back to prevent corruption: %s", exc)
            raise
        finally:
            conn.close()

    def _apply(self, conn: sqlite3.Connection, action: str, row: tuple) -> None:
        """Translate a resolution decision into a write (geometry write via pyogrio in prod)."""
        # 'keep' is a no-op by design; 'rekey'/'insert'/'update' write here.
        ...


def geom_signature(geom_json: str) -> str:
    """Stable hash of geometry used to tell two same-key features apart."""
    return str(hash(geom_json))