feat: add Google Health meal backfill

docs/architecture.md

@@ -51,8 +51,9 @@ server-side identity, and stores only an encrypted refresh token plus safe
 connection metadata. A bounded background job refreshes and reads recent data,
 normalizes it into daily SQLite records, and the Telegram commands and
 `get_health_summary` tool read those normalized records. When both nutrition
-scopes are granted, meal mutations from eligible private chats also enqueue
-durable, account-bound Google `nutrition-log` revisions. The export worker
+scopes are granted, a durable coordinator queues existing meals once per Google
+account. New meal mutations from eligible private chats also enqueue durable,
+account-bound Google `nutrition-log` revisions. The export worker
 retries pending work independently of health imports, preserves operation
 ordering per meal, and exposes safe pending, synced, failed, and
 authorization-required counts. Tokens, raw provider payloads, meal
@@ -152,8 +153,9 @@ current read-only activity/fitness, measurements, and sleep scopes plus
 `googlehealth.nutrition.readonly` and `googlehealth.nutrition.writeonly` for
 optional meal export. It handles missing or partially imported categories as
 unavailable. Health commands reject group chats, and the summary tool requires
-private Telegram session state. Meal export has no historical backfill and
-keeps local save status separate from remote sync status. `/disconnect_health`
+private Telegram session state. Meal export performs one durable historical
+backfill per Google account and keeps local save status separate from remote
+sync status. `/disconnect_health`
 requires an explicit inline-button confirmation, cancels future meal sync,
 retains local calorie logs, and does not purge records already sent to Google;
 requests already submitted may still complete.

docs/telegram-setup.md

@@ -88,8 +88,9 @@ retains the model's tool call and arguments.
 
 Blacki can read normalized health summaries after a user completes Google OAuth
 from a private Telegram chat. If the user grants both nutrition permissions,
-Blacki also exports future meal logs, edits, and deletions from that private
-chat. This is intentionally named **Connect Google Health**: Blacki does not
+Blacki also queues existing meals from that private chat once for the connected
+Google account, then exports future meal logs, edits, and deletions. This is
+intentionally named **Connect Google Health**: Blacki does not
 request Apple ID credentials, access HealthKit, scrape Fitbit, or receive
 arbitrary Apple Health records. The user must first configure an Apple
 Health-to-Google Health/Fitbit-compatible import path if their account and app
@@ -108,11 +109,11 @@ to the exact public HTTPS URL. In Telegram:
    ID; it does not import unrelated food logs.
 3. Return to Telegram and use `/health_refresh` for an on-demand sync or
    `/health_summary` for the latest stored records.
-4. Log meals normally. Eligible new meals show a `google_health_sync` status;
-   `pending` is retried in the background, `synced` confirms the remote write,
-   `authorization_required` asks you to reconnect, and `failed` remains visible
-   for follow-up. A local Blacki save is still successful when remote sync is
-   pending or fails, and the meal must not be logged again.
+4. Log meals normally. Existing eligible meals are queued once after the
+   connection is saved. New meals continue through the background export
+   worker. A local Blacki save remains successful when remote sync is pending
+   or fails, and the meal must not be logged again. Ask Blacki for meal sync
+   status when you want to check the queue, or ask it to retry failed exports.
 5. Use `/disconnect_health`, then confirm the button, to revoke the token
    best-effort, cancel pending meal sync, and remove Blacki's stored token and
    normalized health summaries. Local calorie logs remain. Blacki does not
@@ -124,10 +125,11 @@ window so late device imports can replace earlier daily records. Missing values
 are omitted rather than guessed. Stored data is limited to normalized daily
 activity, workout, sleep, heart-rate, weight, and body-fat summaries; raw
 Google payloads and provider IDs are not persisted in the summary table. Meal
-exports are persisted separately with retry state and opaque data point IDs;
-there is no historical backfill. The meal export worker runs every minute
-independently of health imports. Run only one active scheduler process per
-`tools.db` so a deployment does not dispatch duplicate work.
+exports are persisted separately with retry state and opaque data point IDs. A
+one-time per-account backfill queues existing local meals with a durable cursor;
+the meal export worker runs every minute independently of health imports. Run
+only one active scheduler process per `tools.db` so a deployment does not
+dispatch duplicate work.
 
 Google's v4 discovery document currently lists `nutrition-log` as a supported
 data type. Blacki writes only the local meal description, kcal, available

src/blacki/calories/__init__.py

@@ -2,14 +2,18 @@
     delete_meal,
     edit_meal,
     get_calorie_summary,
+    get_meal_sync_status,
     log_meal,
+    retry_meal_sync,
     set_calorie_goal,
 )
 
 __all__ = [
     "delete_meal",
     "edit_meal",
     "get_calorie_summary",
+    "get_meal_sync_status",
     "log_meal",
+    "retry_meal_sync",
     "set_calorie_goal",
 ]

src/blacki/calories/service.py

@@ -250,8 +250,9 @@ async def _enqueue_export(
             await health.get_connection(canonical) if canonical is not None else None
         )
         previous = await nutrition.meal(entry_id)
-        if previous is not None and str(previous["status"]) == "cancelled":
-            return "not_enabled"
+        previous_cancelled = (
+            previous is not None and str(previous["status"]) == "cancelled"
+        )
 
         eligible = _nutrition_authorized(connection)
         existing_account = (
@@ -267,27 +268,35 @@ async def _enqueue_export(
             or connection_account is None
             or existing_account == connection_account
         )
+        can_use_account = account_matches or (
+            previous_cancelled and connection is not None
+        )
 
-        # Only a newly created private meal with both nutrition scopes enrolls.
-        # A connection added later must not backfill meals that predate consent.
+        # New private meals enroll immediately. Historical meals are enrolled by
+        # NutritionBackfillCoordinator after the user grants both scopes.
         should_enqueue = (
             private
-            and account_matches
+            and can_use_account
             and ((created and eligible) or (not created and previous is not None))
         )
         if not should_enqueue:
             if created and connection is not None and not eligible:
                 return "authorization_required"
             return "not_enabled"
 
-        health_user_id = existing_account or (
-            connection.health_user_id if connection is not None else ""
+        health_user_id = (
+            connection.health_user_id
+            if previous_cancelled and connection is not None
+            else existing_account
+            or (connection.health_user_id if connection is not None else "")
         )
         if not health_user_id:
             return "authorization_required"
 
         if entry is None:
-            target = await self._latest_remote_resource(nutrition, entry_id)
+            target = await self._latest_remote_resource(
+                nutrition, entry_id, health_user_id
+            )
             await nutrition.enqueue(
                 meal_id=entry_id,
                 owner_id=user_id,
@@ -313,7 +322,9 @@ async def _enqueue_export(
                 previous is not None
                 and str(previous.get("desired_operation")) == "upsert"
             ):
-                target = await self._latest_remote_resource(nutrition, entry_id)
+                target = await self._latest_remote_resource(
+                    nutrition, entry_id, health_user_id
+                )
                 if target is not None:
                     await nutrition.enqueue(
                         meal_id=entry_id,
@@ -343,16 +354,30 @@ async def _enqueue_export(
         return "authorization_required"
 
     async def _latest_remote_resource(
-        self, nutrition: NutritionStorage | Any, meal_id: int
+        self,
+        nutrition: NutritionStorage | Any,
+        meal_id: int,
+        health_user_id: str | None = None,
     ) -> str | None:
         revisions = await nutrition.revisions(meal_id)
         for revision in reversed(revisions):
             if revision.get("operation", "upsert") != "upsert":
                 continue
+            resource_name = revision.get("resource_name")
+            if resource_name is None:
+                continue
+            if health_user_id is not None and not str(resource_name).startswith(
+                f"users/{health_user_id}/dataTypes/nutrition-log/dataPoints/"
+            ):
+                continue
             state = str(revision.get("state", "queued"))
             if state in {"synced", "in_flight", "uncertain"}:
-                return str(revision["resource_name"])
-        return None
+                return str(resource_name)
+        latest_remote_resource = getattr(nutrition, "latest_remote_resource", None)
+        if latest_remote_resource is None:
+            return None
+        resource_name = await latest_remote_resource(meal_id, health_user_id)
+        return str(resource_name) if resource_name is not None else None
 
 
 async def container_execute(conn: Any, query: str, params: tuple[Any, ...]) -> None:

src/blacki/calories/storage.py

@@ -7,6 +7,10 @@
 
 from pydantic import BaseModel
 
+from blacki.health.config import (
+    health_user_id_for_telegram_user,
+    telegram_chat_id_for_health_user,
+)
 from blacki.storage.base import SqlStorage
 
 if TYPE_CHECKING:
@@ -108,6 +112,67 @@ async def add_entry(self, entry: CalorieEntry) -> int:
         )
         return rid
 
+    async def health_backfill_high_water(self, health_user_id: str) -> int:
+        """Return the current meal ID high-water mark for one private chat."""
+        if not _is_private_health_user_id(health_user_id):
+            return 0
+        row = await self._fetch_one(
+            """
+            SELECT MAX(id) AS high_water_meal_id
+            FROM calorie_logs
+            WHERE user_id = ? OR user_id LIKE ?
+            """,
+            (health_user_id, f"{health_user_id}-thread-%"),
+        )
+        return (
+            int(row["high_water_meal_id"])
+            if row is not None and row["high_water_meal_id"] is not None
+            else 0
+        )
+
+    async def health_backfill_batch(
+        self,
+        health_user_id: str,
+        *,
+        after_id: int,
+        through_id: int,
+        limit: int,
+    ) -> tuple[list[CalorieEntry], int | None]:
+        """Return valid private-chat meals and the raw cursor for a batch.
+
+        The SQL prefix is deliberately narrow. The full identity helper is
+        still applied to every candidate so malformed topics cannot enter the
+        export queue, and the raw candidate cursor lets the caller advance
+        past skipped rows without looping forever.
+        """
+        if not _is_private_health_user_id(health_user_id):
+            return [], None
+        rows = await self._fetch_all(
+            """
+            SELECT * FROM calorie_logs
+            WHERE (user_id = ? OR user_id LIKE ?)
+              AND id > ? AND id <= ?
+            ORDER BY id ASC
+            LIMIT ?
+            """,
+            (
+                health_user_id,
+                f"{health_user_id}-thread-%",
+                after_id,
+                through_id,
+                limit,
+            ),
+        )
+        if not rows:
+            return [], None
+        cursor = max(int(row["id"]) for row in rows)
+        entries = [
+            self._row_to_entry(row)
+            for row in rows
+            if health_user_id_for_telegram_user(str(row["user_id"])) == health_user_id
+        ]
+        return entries, cursor
+
     async def get_daily_summary(self, user_id: str, date_str: str) -> DailySummary:
         """Get summary and up to 50 entries for a specific day."""
         rows = await self._fetch_all(
@@ -259,3 +324,9 @@ def get_storage() -> SqliteCalorieStorage:
             "Calorie storage not initialized. Call storage.initialize() first."
         )
     return storage
+
+
+def _is_private_health_user_id(health_user_id: str) -> bool:
+    """Reject group-chat identities before any historical meal query."""
+    chat_id = telegram_chat_id_for_health_user(health_user_id)
+    return chat_id is not None and chat_id > 0