M4: ingestion management (incremental, GS1 supplement, dedup/conflict, quality, scheduler)

- OFF incremental fetch via search API + persistent watermark (ingest_state, migration 0004)
- GS1 barcode supplement adapter (offline mapping + GS1-style API) filling only gaps with field-level provenance
- Non-GTIN dedup with canonical selection + merge_log; field-level conflict resolution (source trust > recency)
- Quality scoring (0.4 completeness + 0.3 source trust + 0.2 multi-source + 0.1 freshness) wired into load/merge
- Jobs: update_off, dedup, schedule; docs/ingestion-management.md
- 19 new tests (pure + DB-integration), ruff clean

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

ingestion/opengoods/etl/state.py

@@ -0,0 +1,45 @@
+"""Persistent ingestion watermark stored in the ``ingest_state`` table.
+
+The incremental updater uses this to remember how far it got for each source
+(e.g. Open Food Facts exposes a ``last_modified_t`` unix timestamp on every
+product) so repeated runs only fetch what changed.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import psycopg
+from psycopg.types.json import Jsonb
+
+
+def get_watermark(conn: psycopg.Connection, source: str) -> int:
+    """Return the last processed ``last_modified_t`` for *source* (0 if none)."""
+    row = conn.execute(
+        "SELECT last_modified_t FROM ingest_state WHERE source = %s", (source,)
+    ).fetchone()
+    return int(row[0]) if row else 0
+
+
+def set_watermark(
+    conn: psycopg.Connection,
+    source: str,
+    last_modified_t: int,
+    stats: dict[str, Any] | None = None,
+) -> None:
+    """Upsert the watermark and run metadata for *source*.
+
+    The watermark only ever moves forward: a lower ``last_modified_t`` is
+    ignored so an out-of-order or partial run cannot rewind progress.
+    """
+    conn.execute(
+        """
+        INSERT INTO ingest_state (source, last_modified_t, last_run_at, stats)
+        VALUES (%s, %s, now(), %s)
+        ON CONFLICT (source) DO UPDATE SET
+            last_modified_t = GREATEST(ingest_state.last_modified_t, EXCLUDED.last_modified_t),
+            last_run_at = now(),
+            stats = EXCLUDED.stats
+        """,
+        (source, int(last_modified_t), Jsonb(stats or {})),
+    )