"""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 {})),
    )