fix(bug): replication timestamp

fixes bug on last_ts date when changeset is upserted

Author: kshitijrajsharma

api/queries.py

@@ -137,8 +137,17 @@ def _user_stats_sql(*, filter_dates: bool, filter_hashtags: bool, include_tags:
 
 
 async def fetch_state() -> dict[str, Any] | None:
+    # last_ts/last_seq come from the worst-lagging source (slowest source bounds real freshness);
+    # updated_at is the most recent heartbeat across all sources (any tick proves the worker is alive).
     async with get_pool().acquire() as conn:
-        row = await conn.fetchrow("SELECT last_seq, last_ts, updated_at FROM state ORDER BY updated_at DESC LIMIT 1")
+        row = await conn.fetchrow(
+            """
+            SELECT last_seq, last_ts, (SELECT MAX(updated_at) FROM state) AS updated_at
+            FROM state
+            ORDER BY last_ts ASC
+            LIMIT 1
+            """
+        )
     if row is None:
         return None
     return dict(row)

osmsg/pipeline.py

@@ -408,7 +408,7 @@ def run(cfg: RunConfig) -> dict[str, Any]:
                 conn,
                 source_url=CHANGESETS_REPLICATION,
                 last_seq=cs_end,
-                last_ts=cfg.end_date.astimezone(UTC),
+                last_ts=cs_repl.sequence_to_timestamp(cs_end),
                 updated_at=dt.datetime.now(UTC),
             )
             info("Changeset processing complete.")
@@ -439,6 +439,15 @@ def run(cfg: RunConfig) -> dict[str, Any]:
 
         if not urls:
             info(f"  {url}: already up-to-date")
+            if resume_seq is not None:
+                # Heartbeat: bump updated_at so /health can tell "alive, idle" apart from "stuck".
+                upsert_state(
+                    conn,
+                    source_url=url,
+                    last_seq=resume_seq - 1,
+                    last_ts=url_start,
+                    updated_at=dt.datetime.now(UTC),
+                )
             continue
 
         cf_dir.mkdir(parents=True, exist_ok=True)