diff --git a/docs/architecture.md b/docs/architecture.md index 4eeb1bf..e1d5cd3 100644 --- a/docs/architecture.md +++ b/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. diff --git a/docs/telegram-setup.md b/docs/telegram-setup.md index b691917..58007b2 100644 --- a/docs/telegram-setup.md +++ b/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 diff --git a/src/blacki/calories/__init__.py b/src/blacki/calories/__init__.py index fea43fd..5be19e3 100644 --- a/src/blacki/calories/__init__.py +++ b/src/blacki/calories/__init__.py @@ -2,7 +2,9 @@ delete_meal, edit_meal, get_calorie_summary, + get_meal_sync_status, log_meal, + retry_meal_sync, set_calorie_goal, ) @@ -10,6 +12,8 @@ "delete_meal", "edit_meal", "get_calorie_summary", + "get_meal_sync_status", "log_meal", + "retry_meal_sync", "set_calorie_goal", ] diff --git a/src/blacki/calories/service.py b/src/blacki/calories/service.py index a70a47a..efd956d 100644 --- a/src/blacki/calories/service.py +++ b/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,12 +268,15 @@ 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: @@ -280,14 +284,19 @@ async def _enqueue_export( 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: diff --git a/src/blacki/calories/storage.py b/src/blacki/calories/storage.py index 179fb8d..42e6756 100644 --- a/src/blacki/calories/storage.py +++ b/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 diff --git a/src/blacki/calories/tools.py b/src/blacki/calories/tools.py index fcea257..31c700a 100644 --- a/src/blacki/calories/tools.py +++ b/src/blacki/calories/tools.py @@ -7,6 +7,11 @@ from google.adk.tools import ToolContext from pydantic import ValidationError +from blacki.container import get_container +from blacki.health.config import ( + GOOGLE_HEALTH_NUTRITION_SCOPES, + health_user_id_for_telegram_user, +) from blacki.utils.dates import parse_date from blacki.utils.preferences import get_preferences_storage from blacki.utils.timezone import get_app_timezone, now_utc @@ -337,6 +342,102 @@ async def set_calorie_goal( } +async def get_meal_sync_status(tool_context: ToolContext) -> dict[str, Any]: + """Read the current Google Health meal-export status for this private chat.""" + user_id = tool_context.user_id + if not user_id: # pragma: no cover + return {"status": "error", "message": "Missing user_id in tool_context"} + if not _is_private_tool_context(tool_context): + return { + "status": "error", + "message": "Google Health meal export is available only in private chats", + } + + health_user_id = health_user_id_for_telegram_user(user_id) + if health_user_id is None: + return {"status": "error", "message": "Invalid private Telegram identity"} + + try: + health = get_container().google_health_storage + await health.initialize() + connection = await health.get_connection(health_user_id) + counts = await health.nutrition.counts(health_user_id) + except RuntimeError: + return {"status": "error", "message": "Health storage is not initialized"} + except Exception: + logger.exception("Failed to read Google Health meal-export status") + return {"status": "error", "message": "Could not read meal export status"} + + return { + "status": "success", + "google_health_connection": ( + connection.status if connection is not None else "not_connected" + ), + "nutrition_permissions": _has_nutrition_scopes(connection), + "google_health_sync": counts, + } + + +async def retry_meal_sync(tool_context: ToolContext) -> dict[str, Any]: + """Retry failed Google Health meal exports for this private chat.""" + user_id = tool_context.user_id + if not user_id: # pragma: no cover + return {"status": "error", "message": "Missing user_id in tool_context"} + if not _is_private_tool_context(tool_context): + return { + "status": "error", + "message": "Google Health meal export is available only in private chats", + } + + health_user_id = health_user_id_for_telegram_user(user_id) + if health_user_id is None: + return {"status": "error", "message": "Invalid private Telegram identity"} + + try: + container = get_container() + health = container.google_health_storage + await health.initialize() + connection = await health.get_connection(health_user_id) + if connection is None: + return { + "status": "not_connected", + "requeued": 0, + "message": "Connect Google Health before retrying meal exports.", + } + if not _health_connection_can_export(connection): + return { + "status": "authorization_required", + "requeued": 0, + "message": ( + "Both Google Health nutrition permissions are required before " + "failed meal exports can be retried." + ), + } + + requeued = await health.nutrition.retry_failed( + health_user_id, connection.health_user_id + ) + if requeued and container.nutrition_export_worker is not None: + container.nutrition_export_worker.wake() + counts = await health.nutrition.counts(health_user_id) + message = ( + f"Queued {requeued} failed meal export(s) for retry." + if requeued + else "There are no failed meal exports to retry." + ) + return { + "status": "success", + "requeued": requeued, + "message": message, + "google_health_sync": counts, + } + except RuntimeError: + return {"status": "error", "message": "Health storage is not initialized"} + except Exception: + logger.exception("Failed to retry Google Health meal exports") + return {"status": "error", "message": "Could not retry meal exports"} + + def _try_get_meal_service() -> Any | None: """Use the atomic service when the application container is available. @@ -359,9 +460,30 @@ def _is_private_tool_context(tool_context: ToolContext) -> bool: def _meal_saved_message(message: str, sync_status: str) -> str: if sync_status == "pending": - return f"{message} Saved in Blacki; Google Health sync is pending." + return f"{message} Saved in Blacki." if sync_status == "authorization_required": return f"{message} Saved in Blacki; reconnect Google Health to sync it." if sync_status == "failed": - return f"{message} Saved in Blacki; Google Health sync failed and will retry." + return ( + f"{message} Saved in Blacki; Google Health export failed. " + "Ask me to retry failed meal exports." + ) return f"{message} Saved in Blacki." + + +def _has_nutrition_scopes(connection: Any | None) -> bool: + """Return whether a connection grants both nutrition permissions.""" + return bool( + connection is not None + and set(GOOGLE_HEALTH_NUTRITION_SCOPES) <= set(connection.scopes) + ) + + +def _health_connection_can_export(connection: Any | None) -> bool: + """Return whether a connected account can retry provider writes.""" + return bool( + connection is not None + and connection.status == "connected" + and connection.encrypted_refresh_token is not None + and _has_nutrition_scopes(connection) + ) diff --git a/src/blacki/health/nutrition_backfill.py b/src/blacki/health/nutrition_backfill.py new file mode 100644 index 0000000..628c9d8 --- /dev/null +++ b/src/blacki/health/nutrition_backfill.py @@ -0,0 +1,237 @@ +"""One-time historical Google Health nutrition export coordination.""" + +from __future__ import annotations + +import asyncio +import logging +import time +from collections.abc import Callable +from dataclasses import dataclass + +from blacki.calories.service import nutrition_payload +from blacki.calories.storage import SqliteCalorieStorage + +from .config import ( + GOOGLE_HEALTH_NUTRITION_SCOPES, + health_user_id_for_telegram_user, + telegram_chat_id_for_health_user, +) +from .nutrition_storage import ( + BACKFILL_BATCH_SIZE, + BACKFILL_LEASE_SECONDS, + BACKFILL_VERSION, +) +from .storage import HealthConnection, SqliteGoogleHealthStorage + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True, slots=True) +class NutritionBackfillResult: + """Safe local outcome for one backfill coordination attempt.""" + + status: str + telegram_user_id: str + health_user_id: str | None = None + queued_count: int = 0 + skipped_count: int = 0 + + +class NutritionBackfillCoordinator: + """Queue historical meals without making provider calls.""" + + def __init__( + self, + health_storage: SqliteGoogleHealthStorage, + calorie_storage: SqliteCalorieStorage, + *, + wake: Callable[[], None] | None = None, + ) -> None: + self.health_storage = health_storage + self.calorie_storage = calorie_storage + self.wake = wake + + async def run_all_eligible(self) -> list[NutritionBackfillResult]: + """Sweep connected accounts that granted both nutrition scopes.""" + results: list[NutritionBackfillResult] = [] + for connection in await self.health_storage.list_active_connections(): + if not _nutrition_authorized(connection): + continue + try: + results.append(await self.run_user(connection.telegram_user_id)) + except Exception: + logger.exception( + "Google Health nutrition backfill failed for one account" + ) + results.append( + NutritionBackfillResult( + status="failed", + telegram_user_id=connection.telegram_user_id, + health_user_id=connection.health_user_id, + ) + ) + return results + + async def run_user(self, telegram_user_id: str) -> NutritionBackfillResult: + """Claim and advance one private user's account-bound backfill.""" + health_user_id = health_user_id_for_telegram_user(telegram_user_id) + if health_user_id is None or not _is_private_health_user_id(health_user_id): + return NutritionBackfillResult( + status="skipped", telegram_user_id=telegram_user_id + ) + + connection = await self.health_storage.get_connection(health_user_id) + if connection is None: + return NutritionBackfillResult( + status="not_connected", telegram_user_id=health_user_id + ) + if not _nutrition_authorized(connection): + return NutritionBackfillResult( + status="not_eligible", + telegram_user_id=health_user_id, + health_user_id=connection.health_user_id, + ) + + nutrition = self.health_storage.nutrition + high_water = await self.calorie_storage.health_backfill_high_water( + health_user_id + ) + claimed = await nutrition.claim_backfill( + health_user_id, + connection.health_user_id, + high_water, + version=BACKFILL_VERSION, + lease_seconds=BACKFILL_LEASE_SECONDS, + ) + if claimed is None: + existing = await nutrition.get_backfill( + health_user_id, connection.health_user_id, BACKFILL_VERSION + ) + return NutritionBackfillResult( + status=str(existing["status"]) if existing is not None else "running", + telegram_user_id=health_user_id, + health_user_id=connection.health_user_id, + queued_count=( + int(existing["queued_count"]) if existing is not None else 0 + ), + skipped_count=( + int(existing["skipped_count"]) if existing is not None else 0 + ), + ) + + queued_total = 0 + skipped_total = 0 + cursor = int(claimed["cursor_meal_id"]) + high_water = int(claimed["high_water_meal_id"]) + try: + if high_water == 0: + await nutrition.advance_backfill( + health_user_id, + connection.health_user_id, + 0, + queued_count=0, + skipped_count=0, + version=BACKFILL_VERSION, + lease_seconds=BACKFILL_LEASE_SECONDS, + ) + return NutritionBackfillResult( + status="completed", + telegram_user_id=health_user_id, + health_user_id=connection.health_user_id, + ) + + while cursor < high_water: + ( + entries, + batch_cursor, + ) = await self.calorie_storage.health_backfill_batch( + health_user_id, + after_id=cursor, + through_id=high_water, + limit=BACKFILL_BATCH_SIZE, + ) + next_cursor = batch_cursor if batch_cursor is not None else high_water + batch_queued = 0 + batch_skipped = 0 + + async with nutrition._lock: + await nutrition.conn.execute("BEGIN") + try: + for entry in entries: + result = await nutrition.ensure_backfill_export( + meal_id=int(entry.id or 0), + owner_id=entry.user_id, + telegram_user_id=health_user_id, + health_user_id=connection.health_user_id, + payload=nutrition_payload(entry), + ) + if result == "queued": + batch_queued += 1 + else: + batch_skipped += 1 + updated = await nutrition._advance_backfill_unlocked( + health_user_id, + connection.health_user_id, + next_cursor, + queued_count=batch_queued, + skipped_count=batch_skipped, + version=BACKFILL_VERSION, + now=time.time(), + lease_seconds=BACKFILL_LEASE_SECONDS, + ) + if not updated: + raise RuntimeError("nutrition backfill lease was lost") + await nutrition.conn.execute("COMMIT") + except BaseException: + await nutrition._rollback() + raise + + cursor = next_cursor + queued_total += batch_queued + skipped_total += batch_skipped + if batch_queued and self.wake is not None: + self.wake() + if batch_cursor is None or cursor >= high_water: + break + + return NutritionBackfillResult( + status="completed", + telegram_user_id=health_user_id, + health_user_id=connection.health_user_id, + queued_count=queued_total, + skipped_count=skipped_total, + ) + except asyncio.CancelledError: + raise + except Exception: + await nutrition.fail_backfill( + health_user_id, + connection.health_user_id, + "backfill_queue_error", + version=BACKFILL_VERSION, + ) + logger.exception( + "Google Health nutrition backfill could not queue local meals" + ) + return NutritionBackfillResult( + status="failed", + telegram_user_id=health_user_id, + health_user_id=connection.health_user_id, + queued_count=queued_total, + skipped_count=skipped_total, + ) + + +def _nutrition_authorized(connection: HealthConnection) -> bool: + """Return whether the connection can export nutrition data.""" + return bool( + connection.status == "connected" + and connection.encrypted_refresh_token is not None + and set(GOOGLE_HEALTH_NUTRITION_SCOPES) <= set(connection.scopes) + ) + + +def _is_private_health_user_id(health_user_id: str) -> bool: + """Reject negative Telegram group IDs from historical export.""" + chat_id = telegram_chat_id_for_health_user(health_user_id) + return chat_id is not None and chat_id > 0 diff --git a/src/blacki/health/nutrition_storage.py b/src/blacki/health/nutrition_storage.py index 6cfc47f..756c5ea 100644 --- a/src/blacki/health/nutrition_storage.py +++ b/src/blacki/health/nutrition_storage.py @@ -1,24 +1,31 @@ """Durable, account-bound meal export state. -Calorie rows remain the local source of truth. This module stores the latest -desired revision plus immutable create revisions for Google Health. A delete -sets ``desired_revision`` to NULL while retaining the revisions the worker must -remove remotely, so local deletion never loses reconciliation intent. +Calorie rows remain the local source of truth. This module stores the latest +desired revision plus immutable revisions for Google Health. A separate +history table records terminal provider results so disconnecting and +reconnecting the same Google account cannot create duplicate data points. """ from __future__ import annotations import json +import time +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager, suppress from typing import Any from uuid import uuid4 from blacki.storage.base import SqlStorage +BACKFILL_VERSION = 1 +BACKFILL_BATCH_SIZE = 50 +BACKFILL_LEASE_SECONDS = 300 + _MISSING = object() class NutritionStorage(SqlStorage): - """Store desired meals and immutable remote create revisions.""" + """Store desired meals, immutable revisions, and export history.""" async def _create_tables(self) -> None: await self._conn.executescript( @@ -43,10 +50,42 @@ async def _create_tables(self) -> None: payload_json TEXT, state TEXT NOT NULL DEFAULT 'queued' ); + CREATE TABLE IF NOT EXISTS nutrition_export_history ( + meal_id INTEGER NOT NULL, + telegram_user_id TEXT NOT NULL, + health_user_id TEXT NOT NULL, + resource_name TEXT NOT NULL, + operation TEXT NOT NULL, + payload_json TEXT, + state TEXT NOT NULL, + backfill_version INTEGER NOT NULL DEFAULT 1, + updated_at REAL NOT NULL, + PRIMARY KEY (meal_id, health_user_id, resource_name, operation) + ); + CREATE TABLE IF NOT EXISTS google_health_nutrition_backfills ( + telegram_user_id TEXT NOT NULL, + health_user_id TEXT NOT NULL, + backfill_version INTEGER NOT NULL, + high_water_meal_id INTEGER NOT NULL DEFAULT 0, + cursor_meal_id INTEGER NOT NULL DEFAULT 0, + status TEXT NOT NULL DEFAULT 'pending', + lease_expires_at REAL, + queued_count INTEGER NOT NULL DEFAULT 0, + skipped_count INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + started_at REAL, + updated_at REAL NOT NULL, + completed_at REAL, + PRIMARY KEY (telegram_user_id, health_user_id, backfill_version) + ); CREATE INDEX IF NOT EXISTS nutrition_due ON nutrition_exports(status, next_attempt); CREATE INDEX IF NOT EXISTS nutrition_meal_revisions ON nutrition_revisions(meal_id, sequence); + CREATE INDEX IF NOT EXISTS nutrition_history_account + ON nutrition_export_history(telegram_user_id, health_user_id); + CREATE INDEX IF NOT EXISTS nutrition_backfills_status + ON google_health_nutrition_backfills(status, lease_expires_at); """ ) @@ -69,11 +108,33 @@ async def enqueue( dispatched remotely). The worker resolves prior in-flight/uncertain revisions before allowing a replacement create. """ + await self._enqueue_unlocked( + meal_id=meal_id, + owner_id=owner_id, + telegram_user_id=telegram_user_id, + health_user_id=health_user_id, + payload=payload, + operation=operation, + target_resource_name=target_resource_name, + ) + + async def _enqueue_unlocked( + self, + *, + meal_id: int, + owner_id: str, + telegram_user_id: str, + health_user_id: str, + payload: dict[str, Any] | None, + operation: str, + target_resource_name: str | None = None, + ) -> None: + """Persist an export while the caller owns the transaction or lock.""" if operation not in {"upsert", "delete"}: raise ValueError(f"unsupported nutrition export operation: {operation}") existing = await self._fetch_one( - "SELECT owner_id, telegram_user_id, health_user_id " + "SELECT owner_id, telegram_user_id, health_user_id, status " "FROM nutrition_exports WHERE meal_id = ?", (meal_id,), ) @@ -83,7 +144,11 @@ async def enqueue( if str(existing["telegram_user_id"]) != telegram_user_id: raise ValueError("nutrition export identity cannot change") old_health_user_id = str(existing["health_user_id"]) - if old_health_user_id and old_health_user_id != health_user_id: + if ( + old_health_user_id + and old_health_user_id != health_user_id + and str(existing["status"]) != "cancelled" + ): raise ValueError("nutrition export account cannot change") if operation == "upsert": @@ -106,6 +171,7 @@ async def enqueue( ON CONFLICT(meal_id) DO UPDATE SET desired_revision = excluded.desired_revision, desired_operation = excluded.desired_operation, + health_user_id = excluded.health_user_id, status = 'pending', attempts = 0, next_attempt = 0, error_code = NULL """, @@ -174,22 +240,70 @@ async def latest_payload(self, meal_id: int) -> dict[str, Any] | None: """, (meal_id,), ) - for row in rows: - try: - payload = json.loads(row["payload_json"]) - except (TypeError, json.JSONDecodeError): + payload = _first_dict_payload(rows) + if payload is not None: + return payload + + history = await self._fetch_all( + """ + SELECT payload_json FROM nutrition_export_history + WHERE meal_id = ? AND operation = 'upsert' AND state = 'synced' + ORDER BY updated_at DESC + """, + (meal_id,), + ) + return _first_dict_payload(history) + + async def latest_remote_resource( + self, meal_id: int, health_user_id: str | None = None + ) -> str | None: + """Return the latest known remote create resource for one meal.""" + revisions = await self.revisions(meal_id) + for revision in reversed(revisions): + if revision.get("operation", "upsert") != "upsert": continue - if isinstance(payload, dict): - return payload - return None + resource_name = revision.get("resource_name") + if resource_name is None: + continue + if health_user_id is not None and not _resource_belongs_to_health_user( + str(resource_name), health_user_id + ): + continue + if str(revision.get("state", "queued")) in { + "synced", + "in_flight", + "uncertain", + }: + return str(resource_name) + + if health_user_id is not None: + row = await self._fetch_one( + """ + SELECT resource_name FROM nutrition_export_history + WHERE meal_id = ? AND health_user_id = ? + AND operation = 'upsert' AND state = 'synced' + ORDER BY updated_at DESC LIMIT 1 + """, + (meal_id, health_user_id), + ) + else: + row = await self._fetch_one( + """ + SELECT resource_name FROM nutrition_export_history + WHERE meal_id = ? AND operation = 'upsert' AND state = 'synced' + ORDER BY updated_at DESC LIMIT 1 + """, + (meal_id,), + ) + return str(row["resource_name"]) if row is not None else None async def revision_state(self, sequence: int, state: str) -> None: - """Record provider progress for one immutable revision. + """Record provider progress for one immutable revision.""" + async with self._write_transaction(): + await self._revision_state_unlocked(sequence, state) - Keyed by ``sequence`` rather than ``resource_name``: a delete revision - deliberately reuses the resource name of the create it targets, so - the name alone cannot identify a single row. - """ + async def _revision_state_unlocked(self, sequence: int, state: str) -> None: + """Update revision state while the caller owns the lock or tx.""" await self._conn.execute( "UPDATE nutrition_revisions SET state = ? WHERE sequence = ?", (state, sequence), @@ -204,12 +318,26 @@ async def result( next_attempt: float = 0, expected_revision: str | None | object = _MISSING, ) -> bool: - """Record a result without clobbering a newer desired edit. + """Record a result without clobbering a newer desired edit.""" + async with self._write_transaction(): + return await self._result_unlocked( + meal_id, + status, + error=error, + next_attempt=next_attempt, + expected_revision=expected_revision, + ) - ``IS`` is used for the guard so deletion rows with a NULL desired - revision can be guarded as well. Omitting the guard preserves the - pre-feature calling convention. - """ + async def _result_unlocked( + self, + meal_id: int, + status: str, + *, + error: str | None = None, + next_attempt: float = 0, + expected_revision: str | None | object = _MISSING, + ) -> bool: + """Update export status while the caller owns the lock or tx.""" params: list[Any] = [status, error, next_attempt, meal_id] where = "meal_id = ? AND status != 'cancelled'" if expected_revision is not _MISSING: @@ -229,6 +357,300 @@ async def result( ) return cursor.rowcount > 0 + async def record_remote_result(self, sequence: int) -> None: + """Persist a terminal provider result for reconnect idempotency.""" + async with self._write_transaction(): + row = await self._fetch_one( + """ + SELECT r.meal_id, r.resource_name, r.operation, r.payload_json, + r.state, e.telegram_user_id + FROM nutrition_revisions AS r + JOIN nutrition_exports AS e ON e.meal_id = r.meal_id + WHERE r.sequence = ? + """, + (sequence,), + ) + if row is None or str(row["state"]) not in {"synced", "deleted"}: + return + resource_name = row["resource_name"] + if resource_name is None: + return + health_user_id = _health_user_id_from_resource(str(resource_name)) + if health_user_id is None: + return + row["health_user_id"] = health_user_id + await self._record_remote_result_unlocked(row) + + async def _record_remote_result_unlocked(self, row: dict[str, Any]) -> None: + """Record a terminal revision while the caller owns the lock or tx.""" + await self._conn.execute( + """ + INSERT INTO nutrition_export_history + (meal_id, telegram_user_id, health_user_id, resource_name, + operation, payload_json, state, backfill_version, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (meal_id, health_user_id, resource_name, operation) + DO UPDATE SET + telegram_user_id = excluded.telegram_user_id, + operation = excluded.operation, + payload_json = excluded.payload_json, + state = excluded.state, + backfill_version = excluded.backfill_version, + updated_at = excluded.updated_at + """, + ( + int(row["meal_id"]), + str(row["telegram_user_id"]), + str(row["health_user_id"]), + str(row["resource_name"]), + str(row["operation"]), + row["payload_json"], + str(row["state"]), + BACKFILL_VERSION, + time.time(), + ), + ) + + async def ensure_backfill_export( + self, + *, + meal_id: int, + owner_id: str, + telegram_user_id: str, + health_user_id: str, + payload: dict[str, Any], + ) -> str: + """Ensure one historical meal has an idempotent export intent. + + The caller must hold the shared lock and an open transaction. Existing + pending, failed, or terminal current-account rows are left untouched. + A previously synced resource with an unchanged payload is also left + untouched. A changed payload gets a delete for the old resource and a + new upsert in order. + """ + existing = await self._fetch_one( + "SELECT * FROM nutrition_exports WHERE meal_id = ?", (meal_id,) + ) + history: dict[str, Any] | None = None + if existing is not None: + if str(existing["owner_id"]) != owner_id: + raise ValueError("nutrition export owner cannot change") + if str(existing["telegram_user_id"]) != telegram_user_id: + raise ValueError("nutrition export identity cannot change") + old_health_user_id = str(existing["health_user_id"]) + if ( + old_health_user_id + and old_health_user_id != health_user_id + and str(existing["status"]) != "cancelled" + ): + raise ValueError("nutrition export account cannot change") + if str(existing["status"]) != "cancelled": + return "existing" + + if not old_health_user_id or old_health_user_id == health_user_id: + history = await self._latest_synced_history_unlocked( + meal_id, health_user_id + ) + if history is not None and _payload_json_matches( + history["payload_json"], payload + ): + await self._restore_synced_history_unlocked( + existing, + history, + owner_id=owner_id, + telegram_user_id=telegram_user_id, + health_user_id=health_user_id, + ) + return "history" + restored = await self._restore_cancelled_revision_unlocked( + existing, health_user_id=health_user_id, payload=payload + ) + if restored is not None: + return restored + + if old_health_user_id and old_health_user_id != health_user_id: + await self._cancel_unresolved_revisions_unlocked( + meal_id, old_health_user_id + ) + + if history is None: + history = await self._latest_synced_history_unlocked( + meal_id, health_user_id + ) + if history is not None and _payload_json_matches( + history["payload_json"], payload + ): + await self._restore_synced_history_unlocked( + existing, + history, + owner_id=owner_id, + telegram_user_id=telegram_user_id, + health_user_id=health_user_id, + ) + return "history" + + if history is not None: + await self._enqueue_unlocked( + meal_id=meal_id, + owner_id=owner_id, + telegram_user_id=telegram_user_id, + health_user_id=health_user_id, + payload=None, + operation="delete", + target_resource_name=str(history["resource_name"]), + ) + await self._enqueue_unlocked( + meal_id=meal_id, + owner_id=owner_id, + telegram_user_id=telegram_user_id, + health_user_id=health_user_id, + payload=payload, + operation="upsert", + ) + return "queued" + + async def _restore_synced_history_unlocked( + self, + existing: dict[str, Any] | None, + history: dict[str, Any], + *, + owner_id: str, + telegram_user_id: str, + health_user_id: str, + ) -> None: + """Restore a cancelled meal from a known synced remote point.""" + resource_name = str(history["resource_name"]) + if existing is None: + await self._conn.execute( + """ + INSERT INTO nutrition_exports + (meal_id, owner_id, telegram_user_id, health_user_id, + desired_revision, desired_operation, status, attempts, + next_attempt, error_code) + VALUES (?, ?, ?, ?, ?, 'upsert', 'synced', 0, 0, NULL) + """, + ( + int(history["meal_id"]), + owner_id, + telegram_user_id, + health_user_id, + resource_name, + ), + ) + return + await self._conn.execute( + """ + UPDATE nutrition_exports + SET health_user_id = ?, desired_revision = ?, + desired_operation = 'upsert', status = 'synced', attempts = 0, + next_attempt = 0, error_code = NULL + WHERE meal_id = ? AND status = 'cancelled' + """, + (health_user_id, resource_name, int(existing["meal_id"])), + ) + + async def _restore_cancelled_revision_unlocked( + self, + existing: dict[str, Any], + *, + health_user_id: str, + payload: dict[str, Any], + ) -> str | None: + """Restore one unresolved same-account upsert after disconnect.""" + revisions = await self._fetch_all( + """ + SELECT * FROM nutrition_revisions + WHERE meal_id = ? + ORDER BY sequence DESC + """, + (int(existing["meal_id"]),), + ) + for revision in revisions: + resource_name = revision["resource_name"] + state = str(revision["state"]) + if ( + str(revision["operation"]) != "upsert" + or resource_name is None + or not _resource_belongs_to_health_user( + str(resource_name), health_user_id + ) + or state == "cancelled" + or not _payload_json_matches(revision["payload_json"], payload) + ): + continue + if state == "synced": + row = dict(existing) + row.update( + { + "resource_name": resource_name, + "operation": "upsert", + "payload_json": revision["payload_json"], + "state": "synced", + "health_user_id": health_user_id, + } + ) + await self._record_remote_result_unlocked(row) + await self._restore_synced_history_unlocked( + existing, + { + "meal_id": int(existing["meal_id"]), + "resource_name": resource_name, + }, + owner_id=str(existing["owner_id"]), + telegram_user_id=str(existing["telegram_user_id"]), + health_user_id=health_user_id, + ) + return "history" + status = "failed" if state == "failed" else "pending" + await self._conn.execute( + """ + UPDATE nutrition_exports + SET health_user_id = ?, desired_revision = ?, + desired_operation = 'upsert', status = ?, attempts = 0, + next_attempt = 0, error_code = CASE + WHEN ? = 'failed' THEN error_code ELSE NULL END + WHERE meal_id = ? AND status = 'cancelled' + """, + ( + health_user_id, + str(resource_name), + status, + status, + int(existing["meal_id"]), + ), + ) + return "existing" + return None + + async def _cancel_unresolved_revisions_unlocked( + self, meal_id: int, health_user_id: str + ) -> None: + """Prevent old-account work from running after an account switch.""" + await self._conn.execute( + """ + UPDATE nutrition_revisions + SET state = 'cancelled' + WHERE meal_id = ? AND state IN ('queued', 'in_flight', 'uncertain') + AND resource_name LIKE ? + """, + (meal_id, f"users/{health_user_id}/dataTypes/nutrition-log/dataPoints/%"), + ) + + async def _latest_synced_history_unlocked( + self, meal_id: int, health_user_id: str + ) -> dict[str, Any] | None: + """Return the newest known synced resource for one account.""" + return await self._fetch_one( + """ + SELECT * FROM nutrition_export_history + WHERE meal_id = ? AND health_user_id = ? + AND operation = 'upsert' AND state = 'synced' + ORDER BY updated_at DESC + LIMIT 1 + """, + (meal_id, health_user_id), + ) + async def counts(self, user_id: str) -> dict[str, int]: """Return durable non-cancelled export counts for one identity.""" rows = await self._fetch_all( @@ -242,26 +664,50 @@ async def counts(self, user_id: str) -> dict[str, int]: ) return {str(row["status"]): int(row["count"]) for row in rows} - async def cancel(self, user_id: str) -> None: - """Cancel dispatch and purge locally stored provider payloads/IDs.""" + async def has_other_account(self, user_id: str, health_user_id: str) -> bool: + """Detect retained meal state bound to a different Health account.""" + row = await self._fetch_one( + """ + SELECT 1 FROM nutrition_exports + WHERE telegram_user_id = ? AND health_user_id != ? + LIMIT 1 + """, + (user_id, health_user_id), + ) + return row is not None + + async def cancel(self, user_id: str, *, cancel_revisions: bool = False) -> None: + """Cancel dispatch while retaining revisions for safe reconnects.""" await self._conn.execute( """ UPDATE nutrition_exports - SET status = 'cancelled', health_user_id = '', - desired_revision = NULL, error_code = NULL + SET status = 'cancelled', desired_revision = NULL WHERE telegram_user_id = ? """, (user_id,), ) await self._conn.execute( """ - DELETE FROM nutrition_revisions - WHERE meal_id IN ( - SELECT meal_id FROM nutrition_exports WHERE telegram_user_id = ? - ) + UPDATE google_health_nutrition_backfills + SET status = 'pending', high_water_meal_id = 0, cursor_meal_id = 0, + lease_expires_at = NULL, + queued_count = 0, skipped_count = 0, last_error = NULL, + started_at = NULL, updated_at = ?, completed_at = NULL + WHERE telegram_user_id = ? """, - (user_id,), + (time.time(), user_id), ) + if cancel_revisions: + await self._conn.execute( + """ + UPDATE nutrition_revisions + SET state = 'cancelled' + WHERE meal_id IN ( + SELECT meal_id FROM nutrition_exports WHERE telegram_user_id = ? + ) AND state IN ('queued', 'in_flight', 'uncertain') + """, + (user_id,), + ) async def resume(self, user_id: str, health_user_id: str) -> None: """Resume only authorization-paused work for the same account.""" @@ -274,3 +720,317 @@ async def resume(self, user_id: str, health_user_id: str) -> None: """, (user_id, health_user_id), ) + + async def retry_failed(self, user_id: str, health_user_id: str) -> int: + """Requeue only current-account failures with a matching failed revision.""" + async with self._lock: + await self._conn.execute("BEGIN") + try: + rows = await self._fetch_all( + """ + SELECT meal_id, desired_revision + FROM nutrition_exports + WHERE telegram_user_id = ? AND health_user_id = ? + AND status = 'failed' AND desired_revision IS NOT NULL + """, + (user_id, health_user_id), + ) + requeued = 0 + for row in rows: + revision = await self._fetch_one( + """ + SELECT sequence + FROM nutrition_revisions + WHERE meal_id = ? AND resource_name = ? AND state = 'failed' + ORDER BY sequence DESC + LIMIT 1 + """, + (int(row["meal_id"]), row["desired_revision"]), + ) + if revision is None: + continue + await self._conn.execute( + "UPDATE nutrition_revisions SET state = 'queued' " + "WHERE sequence = ?", + (int(revision["sequence"]),), + ) + cursor = await self._conn.execute( + """ + UPDATE nutrition_exports + SET status = 'pending', attempts = 0, next_attempt = 0, + error_code = NULL + WHERE meal_id = ? AND status = 'failed' + AND desired_revision = ? + """, + (int(row["meal_id"]), row["desired_revision"]), + ) + requeued += cursor.rowcount + await self._conn.execute("COMMIT") + return requeued + except BaseException: + await self._rollback() + raise + + async def get_backfill( + self, user_id: str, health_user_id: str, version: int = BACKFILL_VERSION + ) -> dict[str, Any] | None: + """Return one durable backfill ledger row.""" + return await self._fetch_one( + """ + SELECT * FROM google_health_nutrition_backfills + WHERE telegram_user_id = ? AND health_user_id = ? + AND backfill_version = ? + """, + (user_id, health_user_id, version), + ) + + async def claim_backfill( + self, + user_id: str, + health_user_id: str, + high_water_meal_id: int, + *, + version: int = BACKFILL_VERSION, + now: float | None = None, + lease_seconds: int = BACKFILL_LEASE_SECONDS, + ) -> dict[str, Any] | None: + """Claim a pending or expired backfill lease under the write lock.""" + reference_time = time.time() if now is None else now + async with self._lock: + await self._conn.execute("BEGIN") + try: + row = await self._fetch_one( + """ + SELECT * FROM google_health_nutrition_backfills + WHERE telegram_user_id = ? AND health_user_id = ? + AND backfill_version = ? + """, + (user_id, health_user_id, version), + ) + if row is not None: + lease_expires_at = row["lease_expires_at"] + if str(row["status"]) == "completed" or ( + str(row["status"]) == "running" + and lease_expires_at is not None + and float(lease_expires_at) > reference_time + ): + await self._conn.execute("COMMIT") + return None + stored_high_water = int(row["high_water_meal_id"]) + if stored_high_water > 0: + high_water_meal_id = stored_high_water + else: + high_water_meal_id = max(0, high_water_meal_id) + await self._conn.execute( + """ + UPDATE google_health_nutrition_backfills + SET high_water_meal_id = ?, status = 'running', + lease_expires_at = ?, + last_error = NULL, updated_at = ?, + started_at = COALESCE(started_at, ?) + WHERE telegram_user_id = ? AND health_user_id = ? + AND backfill_version = ? + """, + ( + high_water_meal_id, + reference_time + lease_seconds, + reference_time, + reference_time, + user_id, + health_user_id, + version, + ), + ) + else: + await self._conn.execute( + """ + INSERT INTO google_health_nutrition_backfills + (telegram_user_id, health_user_id, backfill_version, + high_water_meal_id, status, lease_expires_at, + updated_at, started_at) + VALUES (?, ?, ?, ?, 'running', ?, ?, ?) + """, + ( + user_id, + health_user_id, + version, + max(0, high_water_meal_id), + reference_time + lease_seconds, + reference_time, + reference_time, + ), + ) + await self._conn.execute("COMMIT") + return await self.get_backfill(user_id, health_user_id, version) + except BaseException: + await self._rollback() + raise + + async def advance_backfill( + self, + user_id: str, + health_user_id: str, + cursor_meal_id: int, + *, + queued_count: int, + skipped_count: int, + version: int = BACKFILL_VERSION, + now: float | None = None, + lease_seconds: int = BACKFILL_LEASE_SECONDS, + ) -> bool: + """Advance a claimed backfill in a short transaction.""" + reference_time = time.time() if now is None else now + async with self._lock: + await self._conn.execute("BEGIN") + try: + updated = await self._advance_backfill_unlocked( + user_id, + health_user_id, + cursor_meal_id, + queued_count=queued_count, + skipped_count=skipped_count, + version=version, + now=reference_time, + lease_seconds=lease_seconds, + ) + await self._conn.execute("COMMIT") + return updated + except BaseException: + await self._rollback() + raise + + async def _advance_backfill_unlocked( + self, + user_id: str, + health_user_id: str, + cursor_meal_id: int, + *, + queued_count: int, + skipped_count: int, + version: int, + now: float, + lease_seconds: int, + ) -> bool: + """Advance a claimed backfill while the caller owns the lock or tx.""" + row = await self._fetch_one( + """ + SELECT high_water_meal_id FROM google_health_nutrition_backfills + WHERE telegram_user_id = ? AND health_user_id = ? + AND backfill_version = ? AND status = 'running' + """, + (user_id, health_user_id, version), + ) + if row is None: + return False + completed = cursor_meal_id >= int(row["high_water_meal_id"]) + await self._conn.execute( + """ + UPDATE google_health_nutrition_backfills + SET cursor_meal_id = ?, status = ?, lease_expires_at = ?, + queued_count = queued_count + ?, skipped_count = skipped_count + ?, + updated_at = ?, completed_at = CASE WHEN ? THEN ? ELSE completed_at END + WHERE telegram_user_id = ? AND health_user_id = ? + AND backfill_version = ? AND status = 'running' + """, + ( + cursor_meal_id, + "completed" if completed else "running", + None if completed else now + lease_seconds, + queued_count, + skipped_count, + now, + completed, + now if completed else None, + user_id, + health_user_id, + version, + ), + ) + return True + + async def fail_backfill( + self, + user_id: str, + health_user_id: str, + error_code: str, + *, + version: int = BACKFILL_VERSION, + now: float | None = None, + ) -> None: + """Release a backfill lease with a safe resumable error code.""" + safe_error = _safe_error_code(error_code) + reference_time = time.time() if now is None else now + async with self._write_transaction(): + await self._conn.execute( + """ + UPDATE google_health_nutrition_backfills + SET status = 'pending', lease_expires_at = NULL, + last_error = ?, updated_at = ? + WHERE telegram_user_id = ? AND health_user_id = ? + AND backfill_version = ? AND status = 'running' + """, + (safe_error, reference_time, user_id, health_user_id, version), + ) + + @asynccontextmanager + async def _write_transaction(self) -> AsyncIterator[None]: + """Run one short nutrition state update transaction under the lock.""" + async with self._lock: + await self._conn.execute("BEGIN") + try: + yield + except BaseException: + await self._rollback() + raise + else: + await self._conn.execute("COMMIT") + + async def _rollback(self) -> None: + """Roll back a transaction without hiding the original exception.""" + with suppress(Exception): + await self._conn.execute("ROLLBACK") + + +def _first_dict_payload(rows: list[dict[str, Any]]) -> dict[str, Any] | None: + """Return the first JSON object from newest-first storage rows.""" + for row in rows: + try: + payload = json.loads(row["payload_json"]) + except (TypeError, json.JSONDecodeError): + continue + if isinstance(payload, dict): + return payload + return None + + +def _payload_json_matches(payload_json: Any, payload: dict[str, Any]) -> bool: + """Compare stored and intended payloads without trusting malformed JSON.""" + try: + stored = json.loads(payload_json) + except (TypeError, json.JSONDecodeError): + return False + return isinstance(stored, dict) and stored == payload + + +def _safe_error_code(error_code: str) -> str: + """Keep ledger errors bounded and free of provider payloads.""" + if error_code.isascii() and error_code.isprintable(): + return error_code[:80] + return "backfill_error" + + +def _resource_belongs_to_health_user(resource_name: str, health_user_id: str) -> bool: + """Match a resource to one exact Google Health account.""" + return resource_name.startswith( + f"users/{health_user_id}/dataTypes/nutrition-log/dataPoints/" + ) + + +def _health_user_id_from_resource(resource_name: str) -> str | None: + """Extract an account ID from a canonical nutrition resource name.""" + prefix = "users/" + suffix = "/dataTypes/nutrition-log/dataPoints/" + if not resource_name.startswith(prefix) or suffix not in resource_name: + return None + health_user_id = resource_name[len(prefix) :].split(suffix, 1)[0] + return health_user_id or None diff --git a/src/blacki/health/nutrition_worker.py b/src/blacki/health/nutrition_worker.py index 07aef70..2cc4f34 100644 --- a/src/blacki/health/nutrition_worker.py +++ b/src/blacki/health/nutrition_worker.py @@ -124,6 +124,8 @@ async def _process_meal(self, row: dict[str, Any]) -> None: await nutrition.revision_state(int(revision["sequence"]), "cancelled") if pending is None: + if _desired_revision_state(revisions, desired_revision) == "failed": + return await nutrition.result( meal_id, _final_status(desired_operation), @@ -136,10 +138,20 @@ async def _process_meal(self, row: dict[str, Any]) -> None: connection is None or connection.status != "connected" or connection.encrypted_refresh_token is None - or connection.health_user_id != health_user_id ): await self.storage.mark_reauthorization_required(telegram_user_id) return + if connection.health_user_id != health_user_id: + # A stale due-row snapshot can outlive a Google account switch. + # Do not mark the replacement account as unauthorized; retire only + # the old revision, guarded by the snapshot's desired resource. + await nutrition.result( + meal_id, + "cancelled", + error="account_replaced", + expected_revision=desired_revision, + ) + return if not set(GOOGLE_HEALTH_NUTRITION_SCOPES) <= set(connection.scopes): # The connection itself is fine, only nutrition write scope is @@ -193,6 +205,7 @@ async def _process_meal(self, row: dict[str, Any]) -> None: return if outcome == "resolved": + await nutrition.record_remote_result(int(pending["sequence"])) remaining = await nutrition.revisions(meal_id) still_pending, _ = _select_pending_revision(remaining, desired_revision) if still_pending is None: @@ -420,6 +433,20 @@ def _select_pending_revision( return None, stale +def _desired_revision_state( + revisions: list[dict[str, Any]], desired_revision: str | None +) -> str | None: + """Return the latest state for the parent's current desired resource.""" + matching = [ + revision + for revision in revisions + if revision.get("resource_name") == desired_revision + ] + if not matching: + return None + return str(matching[-1]["state"]) + + def _is_transient(exc: GoogleHealthApiError) -> bool: if exc.transport or exc.status_code is None: return True diff --git a/src/blacki/health/storage.py b/src/blacki/health/storage.py index 9b79f20..ca5cfab 100644 --- a/src/blacki/health/storage.py +++ b/src/blacki/health/storage.py @@ -168,6 +168,8 @@ async def upsert_connection( identity_changed = ( existing is not None and str(existing["health_user_id"]) != health_user_id + ) or await self._nutrition.has_other_account( + telegram_user_id, health_user_id ) if identity_changed: await self._conn.execute( @@ -175,7 +177,9 @@ async def upsert_connection( "WHERE telegram_user_id = ?", (telegram_user_id,), ) - await self._nutrition.cancel(telegram_user_id) + await self._nutrition.cancel( + telegram_user_id, cancel_revisions=True + ) await self._conn.execute( """ diff --git a/src/blacki/privacy.py b/src/blacki/privacy.py index 91bc9bb..6331aff 100644 --- a/src/blacki/privacy.py +++ b/src/blacki/privacy.py @@ -18,6 +18,8 @@ "edit_meal", "delete_meal", "get_calorie_summary", + "get_meal_sync_status", + "retry_meal_sync", "set_calorie_goal", "get_health_summary", "send_text_to_speech", diff --git a/src/blacki/prompt.py b/src/blacki/prompt.py index 385f649..8beafc5 100644 --- a/src/blacki/prompt.py +++ b/src/blacki/prompt.py @@ -69,16 +69,22 @@ meal date in the tool call; never replace an invalid date with today. Use only breakfast, lunch, dinner, or snack as meal types. -After log_meal, edit_meal, or delete_meal succeeds locally, report that local -result and the returned google_health_sync status separately. A successful -log or edit remains saved in Blacki even when that status is pending, failed, -not_enabled, or authorization_required; a successful delete remains deleted -locally in those states. Never claim Google Health accepted a change unless the -status says synced. Do not repeat a meal mutation just because remote export -failed or is still pending, because that can create a duplicate local meal. -Only eligible future meals from a private Telegram chat are exported after the -user grants both nutrition permissions. There is no historical backfill, and -missing nutrition values are omitted rather than invented. +After log_meal, edit_meal, or delete_meal succeeds locally, report the local +result. Do not mention a pending background export in the ordinary confirmation. +A successful log or edit remains saved in Blacki even when remote export is +pending, failed, not_enabled, or authorization_required; a successful delete +remains deleted locally in those states. If export failed, tell the user they +can ask you to retry failed meal exports. If authorization is required, tell +the user to reconnect Google Health. Never claim Google Health accepted a +change unless the status says synced. Do not repeat a meal mutation because +remote export failed or is still pending. Use get_meal_sync_status only when +the user asks about export state, and use retry_meal_sync only for an explicit +retry request. + +After both nutrition permissions are granted in a private Telegram chat, Blacki +queues eligible existing meals once for that Google account, then exports new +meal logs, edits, and deletions. Missing nutrition values are omitted rather +than invented. """ @@ -134,7 +140,8 @@ DOMAIN_PATTERNS = { "nutrition": re.compile( r"\b(?:ate|eaten|eating|drank|drink|food|meal|breakfast|lunch|dinner|" - r"snack|calorie|calories|kcal|macro|macros|nutrition|protein|carbs?|fat)\b", + r"snack|calorie|calories|kcal|macro|macros|nutrition|protein|carbs?|fat|" + r"google\s+health\s+(?:meal\s+)?(?:sync|export)|backfill|retry)\b", re.IGNORECASE, ), "workout": re.compile( @@ -167,6 +174,8 @@ "edit_meal", "delete_meal", "set_calorie_goal", + "get_meal_sync_status", + "retry_meal_sync", } ), "workout": frozenset( diff --git a/src/blacki/registry.py b/src/blacki/registry.py index 2b99829..3e3cbcc 100644 --- a/src/blacki/registry.py +++ b/src/blacki/registry.py @@ -153,11 +153,21 @@ def _build_calorie_tools() -> list[Any]: delete_meal, edit_meal, get_calorie_summary, + get_meal_sync_status, log_meal, + retry_meal_sync, set_calorie_goal, ) - return [log_meal, get_calorie_summary, edit_meal, delete_meal, set_calorie_goal] + return [ + log_meal, + get_calorie_summary, + edit_meal, + delete_meal, + set_calorie_goal, + get_meal_sync_status, + retry_meal_sync, + ] except ImportError as e: # pragma: no cover logger.warning("Failed to load Calorie tools: %s", e) return [] diff --git a/src/blacki/server.py b/src/blacki/server.py index fc84f05..71265c1 100644 --- a/src/blacki/server.py +++ b/src/blacki/server.py @@ -5,11 +5,13 @@ interactive agent testing. """ +import asyncio import logging import os from collections.abc import AsyncIterator from contextlib import asynccontextmanager from pathlib import Path +from typing import Any import uvicorn from fastapi import FastAPI @@ -50,7 +52,46 @@ _container: AppContainer | None = None _google_health_service = None _google_health_scheduler = None -_google_health_export_worker = None +_google_health_export_worker: Any = None +_google_health_backfill_tasks: set[asyncio.Task[None]] = set() + + +def _schedule_google_health_backfill(telegram_user_id: str | None = None) -> None: + """Queue historical meal enrollment without delaying request handling.""" + if not isinstance(_container, AppContainer): + return + if _google_health_export_worker is None: + return + + from .health.nutrition_backfill import NutritionBackfillCoordinator + + coordinator = NutritionBackfillCoordinator( + _container.google_health_storage, + _container.calorie_storage, + wake=_google_health_export_worker.wake, + ) + + async def run() -> None: + try: + if telegram_user_id is None: + await coordinator.run_all_eligible() + else: + await coordinator.run_user(telegram_user_id) + except asyncio.CancelledError: + raise + except Exception: + logger.exception("Google Health nutrition backfill task failed") + + task = asyncio.create_task( + run(), + name=( + "google_health_nutrition_backfill" + if telegram_user_id is None + else "google_health_nutrition_backfill_user" + ), + ) + _google_health_backfill_tasks.add(task) + task.add_done_callback(_google_health_backfill_tasks.discard) async def _start_google_health() -> None: @@ -101,6 +142,7 @@ async def _start_google_health() -> None: _google_health_scheduler = scheduler _google_health_export_worker = export_worker _container.nutrition_export_worker = export_worker + _schedule_google_health_backfill() logger.info("Google Health connector initialized") @@ -200,6 +242,13 @@ async def _stop_google_health() -> None: _google_health_service, \ _google_health_export_worker + backfill_tasks = list(_google_health_backfill_tasks) + for task in backfill_tasks: + task.cancel() + if backfill_tasks: + await asyncio.gather(*backfill_tasks, return_exceptions=True) + _google_health_backfill_tasks.clear() + if _google_health_export_worker is not None: if _container is not None: _container.nutrition_export_worker = None @@ -354,6 +403,9 @@ async def google_health_callback( except Exception: logger.exception("Failed to notify Telegram after Google Health OAuth") + if completion.connected: + _schedule_google_health_backfill(completion.telegram_user_id) + title = ( "Google Health connected" if completion.connected diff --git a/src/blacki/telegram/bot.py b/src/blacki/telegram/bot.py index 8442c25..803e2f9 100644 --- a/src/blacki/telegram/bot.py +++ b/src/blacki/telegram/bot.py @@ -1164,10 +1164,11 @@ async def _connect_health(self, message: Message) -> None: "Google Health can provide read-only wellness summaries from " "data that reaches Google Health from Fitbit-compatible sources. " "If you grant both Google Health nutrition permissions, Blacki " - "will automatically export future meal logs, edits, and " - "deletions from this private chat. The nutrition read permission " + "will queue the meals already saved in this private chat once " + "for this Google account, then automatically export future meal " + "logs, edits, and deletions. The nutrition read permission " "also lets Blacki verify records it created; it does not import " - "unrelated food logs. Older meals are not backfilled. " + "unrelated food logs. " "Read-only summaries remain available without nutrition " "permissions. Existing connections must reconnect to add them. " "Blacki does not receive Apple ID credentials or raw Apple Health " @@ -1343,8 +1344,9 @@ async def notify_health_connection( text = ( "Google Health is connected. Use /health_refresh for a fresh sync or " "/health_summary to read the latest stored records. If both nutrition " - "permissions were granted, future private meal logs, edits, and " - "deletions will sync automatically; older meals are not backfilled." + "permissions were granted, meals already saved in this private chat " + "will be queued once for this Google account, and future private meal " + "logs, edits, and deletions will sync automatically." if connected else ( "Google Health authorization was cancelled. No credentials were stored." diff --git a/tests/calories/test_service.py b/tests/calories/test_service.py index a0a246d..76f3988 100644 --- a/tests/calories/test_service.py +++ b/tests/calories/test_service.py @@ -339,7 +339,7 @@ async def test_mutate_edit_with_new_date_does_not_preserve_interval( async def test_mutate_edit_of_unenrolled_meal_stays_not_enabled( container: AppContainer, ) -> None: - """A meal logged before Google Health was connected must not backfill.""" + """The backfill coordinator, not an edit, enrolls older meals.""" service = MealService(container) entry_id, sync_status = await service.mutate(USER_ID, entry=_entry()) assert sync_status == "not_enabled" @@ -415,17 +415,45 @@ async def test_mutate_delete_never_dispatched_has_no_target( assert revisions[-1]["state"] == "queued" -async def test_mutate_delete_of_cancelled_export_stays_not_enabled( +async def test_mutate_delete_of_cancelled_export_reuses_active_connection( container: AppContainer, ) -> None: await _connect(container) service = MealService(container) entry_id, _ = await service.mutate(USER_ID, private=True, entry=_entry()) + nutrition = container.google_health_storage.nutrition + revision = (await nutrition.revisions(entry_id))[0] + await nutrition.revision_state(int(revision["sequence"]), "synced") - await container.google_health_storage.nutrition.cancel(USER_ID) + await nutrition.cancel(USER_ID) _, sync_status = await service.mutate(USER_ID, private=True, entry_id=entry_id) - assert sync_status == "not_enabled" + assert sync_status == "pending" + revisions = await nutrition.revisions(entry_id) + assert revisions[-1]["operation"] == "delete" + assert revisions[-1]["resource_name"] == revision["resource_name"] + + +async def test_mutate_edit_after_account_replacement_uses_new_account( + container: AppContainer, +) -> None: + await _connect(container, health_user_id="old-health-account") + service = MealService(container) + entry_id, _ = await service.mutate(USER_ID, private=True, entry=_entry()) + + await _connect(container, health_user_id="new-health-account") + _, sync_status = await service.mutate( + USER_ID, + private=True, + entry_id=entry_id, + updates={"calories": 450}, + ) + + assert sync_status == "pending" + revisions = await container.google_health_storage.nutrition.revisions(entry_id) + assert len(revisions) == 2 + assert revisions[-1]["operation"] == "upsert" + assert str(revisions[-1]["resource_name"]).startswith("users/new-health-account/") # --- MealService.mutate: transaction integrity ------------------------------ @@ -591,6 +619,29 @@ async def test_latest_remote_resource_skips_past_a_pending_delete( assert target == upsert_revision["resource_name"] +async def test_latest_remote_resource_filters_invalid_and_other_accounts( + container: AppContainer, +) -> None: + class _NutritionWithoutHistory: + async def revisions(self, meal_id: int) -> list[dict[str, object]]: + return [ + {"operation": "upsert", "resource_name": None, "state": "synced"}, + { + "operation": "upsert", + "resource_name": "users/other/dataTypes/nutrition-log/dataPoints/x", + "state": "synced", + }, + {"operation": "delete", "resource_name": "users/current/x"}, + ] + + service = MealService(container) + + assert ( + await service._latest_remote_resource(_NutritionWithoutHistory(), 1, "current") + is None + ) + + async def test_get_meal_service_binds_the_process_container( container: AppContainer, ) -> None: diff --git a/tests/calories/test_storage.py b/tests/calories/test_storage.py index cd96f26..154f24e 100644 --- a/tests/calories/test_storage.py +++ b/tests/calories/test_storage.py @@ -68,6 +68,18 @@ async def test_add_entry(self, storage) -> None: assert entry_id == 1 + @pytest.mark.asyncio + async def test_health_backfill_rejects_nonprivate_and_empty_ranges( + self, storage + ) -> None: + assert await storage.health_backfill_high_water("telegram-chat--100") == 0 + assert await storage.health_backfill_batch( + "telegram-chat--100", after_id=0, through_id=10, limit=50 + ) == ([], None) + assert await storage.health_backfill_batch( + "telegram-chat-42", after_id=0, through_id=10, limit=50 + ) == ([], None) + @pytest.mark.asyncio async def test_add_entry_with_macros(self, storage) -> None: """Should add an entry with macro nutrients.""" diff --git a/tests/calories/test_tools.py b/tests/calories/test_tools.py index 917576f..427842d 100644 --- a/tests/calories/test_tools.py +++ b/tests/calories/test_tools.py @@ -370,7 +370,8 @@ async def test_log_meal_uses_meal_service_when_available( assert result["status"] == "success" assert result["entry_id"] == 7 assert result["google_health_sync"] == "pending" - assert "sync is pending" in result["message"] + assert result["message"].endswith("Saved in Blacki.") + assert "sync is pending" not in result["message"] mock_service.mutate.assert_called_once() assert mock_service.mutate.call_args.kwargs["private"] is True @@ -531,9 +532,9 @@ def test_is_private_tool_context_false_when_state_has_no_getter() -> None: assert _is_private_tool_context(cast(ToolContext, ctx)) is False -def test_meal_saved_message_pending() -> None: +def test_meal_saved_message_pending_is_local_only() -> None: message = _meal_saved_message("Logged", "pending") - assert message == "Logged Saved in Blacki; Google Health sync is pending." + assert message == "Logged Saved in Blacki." def test_meal_saved_message_authorization_required() -> None: @@ -544,7 +545,8 @@ def test_meal_saved_message_authorization_required() -> None: def test_meal_saved_message_failed() -> None: message = _meal_saved_message("Logged", "failed") assert message == ( - "Logged Saved in Blacki; Google Health sync failed and will retry." + "Logged Saved in Blacki; Google Health export failed. " + "Ask me to retry failed meal exports." ) diff --git a/tests/test_google_health.py b/tests/test_google_health.py index 015f625..c4b17df 100644 --- a/tests/test_google_health.py +++ b/tests/test_google_health.py @@ -1169,7 +1169,7 @@ async def test_delete_connection_rolls_back_on_failure( ) original_cancel = health_storage.nutrition.cancel - async def _boom(user_id: str) -> None: + async def _boom(user_id: str, *, cancel_revisions: bool = False) -> None: raise RuntimeError("simulated failure") health_storage.nutrition.cancel = _boom # type: ignore[method-assign] @@ -1423,6 +1423,45 @@ async def test_health_storage_replaces_identity_and_window_atomically( ) == [{"date": "2026-08-21", "steps": 300}] +@pytest.mark.asyncio +async def test_health_storage_replacement_after_disconnect_cancels_retained_work( + health_storage: SqliteGoogleHealthStorage, +) -> None: + await health_storage.upsert_connection( + telegram_user_id="telegram-chat-42", + encrypted_refresh_token="old-token", + health_user_id="old-health-id", + legacy_fitbit_user_id=None, + scopes=GOOGLE_HEALTH_SCOPES, + ) + await health_storage.nutrition.enqueue( + meal_id=900, + owner_id="telegram-chat-42", + telegram_user_id="telegram-chat-42", + health_user_id="old-health-id", + payload={"nutritionLog": {"energy": {"kcal": 1}}}, + operation="upsert", + ) + await health_storage.delete_connection("telegram-chat-42") + await health_storage.conn.execute( + "UPDATE nutrition_exports SET status = 'authorization_required' " + "WHERE meal_id = 900" + ) + + await health_storage.upsert_connection( + telegram_user_id="telegram-chat-42", + encrypted_refresh_token="new-token", + health_user_id="new-health-id", + legacy_fitbit_user_id=None, + scopes=GOOGLE_HEALTH_SCOPES, + ) + + row = await health_storage.nutrition.meal(900) + assert row is not None + assert row["status"] == "cancelled" + assert (await health_storage.nutrition.revisions(900))[0]["state"] == "cancelled" + + @pytest.mark.asyncio async def test_health_storage_replacement_rolls_back_on_failure( health_storage: SqliteGoogleHealthStorage, diff --git a/tests/test_nutrition_backfill.py b/tests/test_nutrition_backfill.py new file mode 100644 index 0000000..80dc430 --- /dev/null +++ b/tests/test_nutrition_backfill.py @@ -0,0 +1,660 @@ +"""Tests for one-time Google Health nutrition backfill coordination.""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncGenerator +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import aiosqlite +import pytest +from cryptography.fernet import Fernet + +from blacki.calories.service import MealService +from blacki.calories.storage import CalorieEntry +from blacki.container import AppContainer +from blacki.health.config import ( + GOOGLE_HEALTH_NUTRITION_SCOPES, + GOOGLE_HEALTH_READ_SCOPES, + GoogleHealthConfig, +) +from blacki.health.nutrition_backfill import NutritionBackfillCoordinator +from blacki.health.nutrition_storage import BACKFILL_VERSION + +USER_ID = "telegram-chat-42" +HEALTH_USER_ID = "google-account-42" + + +@pytest.fixture +async def container() -> AsyncGenerator[AppContainer, None]: + conn = await aiosqlite.connect(":memory:", isolation_level=None) + conn.row_factory = aiosqlite.Row + app_container = AppContainer(conn=conn) + await app_container.initialize_all_storages() + yield app_container + await app_container.close() + + +def _entry(user_id: str, number: int) -> CalorieEntry: + return CalorieEntry( + user_id=user_id, + description=f"Meal {number}", + calories=300 + number, + logged_at="2026-08-27T08:00:00+00:00", + logged_date="2026-08-27", + ) + + +def _config() -> GoogleHealthConfig: + return GoogleHealthConfig( + client_id="client-id", + client_secret="client-secret", + redirect_uri="https://example.test/callback", + token_encryption_key=Fernet.generate_key().decode(), + ) + + +async def _connect( + container: AppContainer, + *, + health_user_id: str = HEALTH_USER_ID, + scopes: tuple[str, ...] = GOOGLE_HEALTH_NUTRITION_SCOPES, +) -> None: + config = _config() + await container.google_health_storage.upsert_connection( + telegram_user_id=USER_ID, + encrypted_refresh_token=config.cipher.encrypt("refresh-token"), + health_user_id=health_user_id, + legacy_fitbit_user_id=None, + scopes=scopes, + ) + + +async def _add_meals(container: AppContainer, user_ids: list[str]) -> list[int]: + ids: list[int] = [] + for number, user_id in enumerate(user_ids, start=1): + ids.append(await container.calorie_storage.add_entry(_entry(user_id, number))) + return ids + + +@pytest.mark.asyncio +async def test_backfill_queues_valid_direct_and_topic_meals_once( + container: AppContainer, +) -> None: + ids = await _add_meals( + container, + [ + USER_ID, + "telegram-chat-42-thread-7", + "telegram-chat-420", + "telegram-chat-42-thread-not-a-number", + "telegram-chat-42-thread-8-extra", + "telegram-chat--100", + ], + ) + await _connect(container) + wake = MagicMock() + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, + container.calorie_storage, + wake=wake, + ) + + result = await coordinator.run_user(USER_ID) + + assert result.status == "completed" + assert result.queued_count == 2 + assert wake.call_count == 1 + nutrition = container.google_health_storage.nutrition + for entry_id in ids[:2]: + row = await nutrition.meal(entry_id) + assert row is not None + assert row["telegram_user_id"] == USER_ID + assert row["status"] == "pending" + assert len(await nutrition.revisions(entry_id)) == 1 + for entry_id in ids[2:]: + assert await nutrition.meal(entry_id) is None + + second = await coordinator.run_user(USER_ID) + + assert second.status == "completed" + assert second.queued_count == 2 + revision_count = 0 + for entry_id in ids[:2]: + revision_count += len(await nutrition.revisions(entry_id)) + assert revision_count == 2 + ledger = await nutrition.get_backfill(USER_ID, HEALTH_USER_ID, BACKFILL_VERSION) + assert ledger is not None + assert ledger["status"] == "completed" + assert ledger["cursor_meal_id"] == ledger["high_water_meal_id"] + + +@pytest.mark.asyncio +async def test_backfill_skips_read_only_connection_until_scopes_are_granted( + container: AppContainer, +) -> None: + entry_id = (await _add_meals(container, [USER_ID]))[0] + await _connect(container, scopes=GOOGLE_HEALTH_READ_SCOPES) + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + + skipped = await coordinator.run_user(USER_ID) + + assert skipped.status == "not_eligible" + assert await container.google_health_storage.nutrition.meal(entry_id) is None + assert ( + await container.google_health_storage.nutrition.get_backfill( + USER_ID, HEALTH_USER_ID, BACKFILL_VERSION + ) + is None + ) + + await _connect(container) + completed = await coordinator.run_user(USER_ID) + assert completed.status == "completed" + assert await container.google_health_storage.nutrition.meal(entry_id) is not None + + +@pytest.mark.asyncio +async def test_run_all_sweeps_only_active_nutrition_authorized_connections( + container: AppContainer, +) -> None: + await _connect(container) + config = _config() + await container.google_health_storage.upsert_connection( + telegram_user_id="telegram-chat-43", + encrypted_refresh_token=config.cipher.encrypt("refresh-token"), + health_user_id="google-account-43", + legacy_fitbit_user_id=None, + scopes=GOOGLE_HEALTH_READ_SCOPES, + ) + + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + results = await coordinator.run_all_eligible() + + assert [result.telegram_user_id for result in results] == [USER_ID] + assert results[0].status == "completed" + assert ( + await container.google_health_storage.nutrition.get_backfill( + "telegram-chat-43", "google-account-43", BACKFILL_VERSION + ) + is None + ) + + +@pytest.mark.asyncio +async def test_run_all_reports_one_account_failure_without_stopping_sweep( + container: AppContainer, +) -> None: + await _connect(container) + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + run_user = AsyncMock(side_effect=RuntimeError("queue failed")) + coordinator.run_user = run_user # type: ignore[method-assign] + + results = await coordinator.run_all_eligible() + + assert len(results) == 1 + assert results[0].status == "failed" + assert results[0].telegram_user_id == USER_ID + assert results[0].health_user_id == HEALTH_USER_ID + assert results[0].queued_count == 0 + assert results[0].skipped_count == 0 + + +@pytest.mark.asyncio +async def test_run_user_rejects_groups_missing_connections_and_empty_backfills( + container: AppContainer, +) -> None: + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + + assert (await coordinator.run_user("telegram-chat--100")).status == "skipped" + assert (await coordinator.run_user("not-a-telegram-user")).status == "skipped" + assert (await coordinator.run_user(USER_ID)).status == "not_connected" + + await _connect(container) + completed = await coordinator.run_user(USER_ID) + + assert completed.status == "completed" + ledger = await container.google_health_storage.nutrition.get_backfill( + USER_ID, HEALTH_USER_ID, BACKFILL_VERSION + ) + assert ledger is not None + assert ledger["high_water_meal_id"] == 0 + + +@pytest.mark.asyncio +async def test_run_user_handles_an_existing_cursor_at_high_water( + container: AppContainer, +) -> None: + await _add_meals(container, [USER_ID]) + await _connect(container) + nutrition = container.google_health_storage.nutrition + await nutrition._conn.execute( + """ + INSERT INTO google_health_nutrition_backfills + (telegram_user_id, health_user_id, backfill_version, + high_water_meal_id, cursor_meal_id, status, updated_at) + VALUES (?, ?, ?, 1, 1, 'pending', 0) + """, + (USER_ID, HEALTH_USER_ID, BACKFILL_VERSION), + ) + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + + result = await coordinator.run_user(USER_ID) + + assert result.status == "completed" + assert await nutrition.meal(1) is None + + +@pytest.mark.asyncio +async def test_run_user_releases_lease_when_cursor_update_is_lost( + container: AppContainer, +) -> None: + await _add_meals(container, [USER_ID]) + await _connect(container) + nutrition = container.google_health_storage.nutrition + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + with patch.object( + nutrition, "_advance_backfill_unlocked", new=AsyncMock(return_value=False) + ): + result = await coordinator.run_user(USER_ID) + + assert result.status == "failed" + ledger = await nutrition.get_backfill(USER_ID, HEALTH_USER_ID, BACKFILL_VERSION) + assert ledger is not None + assert ledger["status"] == "pending" + assert ledger["last_error"] == "backfill_queue_error" + + +@pytest.mark.asyncio +async def test_run_user_propagates_cancellation_for_lease_recovery( + container: AppContainer, +) -> None: + await _add_meals(container, [USER_ID]) + await _connect(container) + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + with ( + patch.object( + container.calorie_storage, + "health_backfill_batch", + new=AsyncMock(side_effect=asyncio.CancelledError), + ), + pytest.raises(asyncio.CancelledError), + ): + await coordinator.run_user(USER_ID) + + ledger = await container.google_health_storage.nutrition.get_backfill( + USER_ID, HEALTH_USER_ID, BACKFILL_VERSION + ) + assert ledger is not None + assert ledger["status"] == "running" + + +@pytest.mark.asyncio +async def test_backfill_resumes_after_batch_failure_without_duplicates( + container: AppContainer, +) -> None: + ids = await _add_meals(container, [USER_ID] * 55) + await _connect(container) + nutrition = container.google_health_storage.nutrition + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + original = nutrition.ensure_backfill_export + calls = 0 + + async def fail_on_second_batch(**kwargs: Any) -> str: + nonlocal calls + calls += 1 + if calls == 51: + raise RuntimeError("simulated queue failure") + return await original(**kwargs) + + nutrition.ensure_backfill_export = fail_on_second_batch # type: ignore[method-assign] + failed = await coordinator.run_user(USER_ID) + nutrition.ensure_backfill_export = original # type: ignore[method-assign] + + assert failed.status == "failed" + ledger = await nutrition.get_backfill(USER_ID, HEALTH_USER_ID, BACKFILL_VERSION) + assert ledger is not None + assert ledger["status"] == "pending" + assert ledger["cursor_meal_id"] == 50 + first_revision_count = 0 + for entry_id in ids[:50]: + first_revision_count += len(await nutrition.revisions(entry_id)) + assert first_revision_count == 50 + second_revision_count = 0 + for entry_id in ids[50:]: + second_revision_count += len(await nutrition.revisions(entry_id)) + assert second_revision_count == 0 + + resumed = await coordinator.run_user(USER_ID) + + assert resumed.status == "completed" + revision_count = 0 + for entry_id in ids: + revision_count += len(await nutrition.revisions(entry_id)) + assert revision_count == 55 + + +@pytest.mark.asyncio +async def test_concurrent_backfill_runs_share_one_ledger( + container: AppContainer, +) -> None: + ids = await _add_meals(container, [USER_ID, USER_ID]) + await _connect(container) + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + + first, second = await asyncio.gather( + coordinator.run_user(USER_ID), coordinator.run_user(USER_ID) + ) + + assert {first.status, second.status} <= {"completed", "running"} + revision_count = 0 + for entry_id in ids: + revision_count += len( + await container.google_health_storage.nutrition.revisions(entry_id) + ) + assert revision_count == 2 + + +@pytest.mark.asyncio +async def test_reconnect_requeues_unsent_meals_and_preserves_synced_history( + container: AppContainer, +) -> None: + entry_id = (await _add_meals(container, [USER_ID]))[0] + await _connect(container) + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + await coordinator.run_user(USER_ID) + nutrition = container.google_health_storage.nutrition + first_revision = (await nutrition.revisions(entry_id))[0] + await nutrition.revision_state(int(first_revision["sequence"]), "synced") + await nutrition.record_remote_result(int(first_revision["sequence"])) + + await container.google_health_storage.delete_connection(USER_ID) + await _connect(container) + result = await coordinator.run_user(USER_ID) + + assert result.status == "completed" + assert result.queued_count == 0 + assert result.skipped_count == 1 + revisions = await nutrition.revisions(entry_id) + assert len(revisions) == 1 + assert revisions[0]["state"] == "synced" + history = await nutrition._fetch_all( + "SELECT * FROM nutrition_export_history WHERE meal_id = ?", (entry_id,) + ) + assert len(history) == 1 + + +@pytest.mark.asyncio +async def test_same_account_reconnect_restores_future_edit_export_path( + container: AppContainer, +) -> None: + entry_id = (await _add_meals(container, [USER_ID]))[0] + await _connect(container) + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + await coordinator.run_user(USER_ID) + nutrition = container.google_health_storage.nutrition + first_revision = (await nutrition.revisions(entry_id))[0] + await nutrition.revision_state(int(first_revision["sequence"]), "synced") + await nutrition.record_remote_result(int(first_revision["sequence"])) + + await container.google_health_storage.delete_connection(USER_ID) + await _connect(container) + await coordinator.run_user(USER_ID) + + service = MealService(container) + _, status = await service.mutate( + USER_ID, + private=True, + entry_id=entry_id, + updates={"calories": 999}, + ) + + assert status == "pending" + revisions = await nutrition.revisions(entry_id) + assert revisions[-2]["operation"] == "delete" + assert revisions[-2]["resource_name"] == first_revision["resource_name"] + assert revisions[-1]["operation"] == "upsert" + + +@pytest.mark.asyncio +async def test_same_account_reconnect_allows_delete_before_backfill( + container: AppContainer, +) -> None: + entry_id = (await _add_meals(container, [USER_ID]))[0] + await _connect(container) + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + await coordinator.run_user(USER_ID) + nutrition = container.google_health_storage.nutrition + first_revision = (await nutrition.revisions(entry_id))[0] + await nutrition.revision_state(int(first_revision["sequence"]), "synced") + await nutrition.record_remote_result(int(first_revision["sequence"])) + + await container.google_health_storage.delete_connection(USER_ID) + await _connect(container) + + service = MealService(container) + _, status = await service.mutate(USER_ID, private=True, entry_id=entry_id) + + assert status == "pending" + revisions = await nutrition.revisions(entry_id) + assert revisions[-1]["operation"] == "delete" + assert revisions[-1]["resource_name"] == first_revision["resource_name"] + + +@pytest.mark.asyncio +async def test_same_account_reconnect_backfills_meals_logged_while_disconnected( + container: AppContainer, +) -> None: + first_id = (await _add_meals(container, [USER_ID]))[0] + await _connect(container) + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + await coordinator.run_user(USER_ID) + await container.google_health_storage.delete_connection(USER_ID) + late_id = (await _add_meals(container, [USER_ID]))[0] + await _connect(container) + + result = await coordinator.run_user(USER_ID) + + assert result.queued_count == 1 + assert len(await container.google_health_storage.nutrition.revisions(first_id)) == 1 + assert len(await container.google_health_storage.nutrition.revisions(late_id)) == 1 + + +@pytest.mark.asyncio +async def test_reconnect_requeues_an_unsent_historical_intent( + container: AppContainer, +) -> None: + entry_id = (await _add_meals(container, [USER_ID]))[0] + await _connect(container) + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + await coordinator.run_user(USER_ID) + await container.google_health_storage.delete_connection(USER_ID) + await _connect(container) + + original_revision = ( + await container.google_health_storage.nutrition.revisions(entry_id) + )[0] + result = await coordinator.run_user(USER_ID) + + assert result.queued_count == 0 + assert result.skipped_count == 1 + restored = await container.google_health_storage.nutrition.revisions(entry_id) + assert len(restored) == 1 + assert restored[0]["resource_name"] == original_revision["resource_name"] + row = await container.google_health_storage.nutrition.meal(entry_id) + assert row is not None + assert row["status"] == "pending" + + +@pytest.mark.asyncio +async def test_reconnect_preserves_in_flight_resource_for_worker_reconciliation( + container: AppContainer, +) -> None: + entry_id = (await _add_meals(container, [USER_ID]))[0] + await _connect(container) + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + await coordinator.run_user(USER_ID) + nutrition = container.google_health_storage.nutrition + revision = (await nutrition.revisions(entry_id))[0] + await nutrition.revision_state(int(revision["sequence"]), "in_flight") + resource_name = revision["resource_name"] + + await container.google_health_storage.delete_connection(USER_ID) + await _connect(container) + result = await coordinator.run_user(USER_ID) + + assert result.queued_count == 0 + assert result.skipped_count == 1 + restored = (await nutrition.revisions(entry_id))[0] + assert restored["resource_name"] == resource_name + assert restored["state"] == "in_flight" + row = await nutrition.meal(entry_id) + assert row is not None + assert row["status"] == "pending" + + +@pytest.mark.asyncio +async def test_replacement_google_account_gets_a_new_backfill_generation( + container: AppContainer, +) -> None: + entry_id = (await _add_meals(container, [USER_ID]))[0] + await _connect(container) + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + await coordinator.run_user(USER_ID) + + await _connect(container, health_user_id="google-account-new") + result = await coordinator.run_user(USER_ID) + + assert result.status == "completed" + assert result.queued_count == 1 + row = await container.google_health_storage.nutrition.meal(entry_id) + assert row is not None + assert row["health_user_id"] == "google-account-new" + revisions = await container.google_health_storage.nutrition.revisions(entry_id) + assert len(revisions) == 2 + assert revisions[0]["state"] == "cancelled" + assert str(revisions[1]["resource_name"]).startswith("users/google-account-new/") + await container.google_health_storage.nutrition.revision_state( + int(revisions[1]["sequence"]), "synced" + ) + await container.google_health_storage.nutrition.record_remote_result( + int(revisions[1]["sequence"]) + ) + + service = MealService(container) + _, status = await service.mutate( + USER_ID, + private=True, + entry_id=entry_id, + updates={"calories": 999}, + ) + assert status == "pending" + revisions = await container.google_health_storage.nutrition.revisions(entry_id) + assert revisions[-2]["operation"] == "delete" + assert revisions[-2]["resource_name"] == revisions[1]["resource_name"] + + +@pytest.mark.asyncio +async def test_backfill_reconciles_an_edited_meal_after_reconnect( + container: AppContainer, +) -> None: + entry_id = (await _add_meals(container, [USER_ID]))[0] + await _connect(container) + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + await coordinator.run_user(USER_ID) + nutrition = container.google_health_storage.nutrition + first_revision = (await nutrition.revisions(entry_id))[0] + await nutrition.revision_state(int(first_revision["sequence"]), "synced") + await nutrition.record_remote_result(int(first_revision["sequence"])) + + await container.google_health_storage.delete_connection(USER_ID) + await container.calorie_storage.update_entry(entry_id, USER_ID, calories=999) + await _connect(container) + + result = await coordinator.run_user(USER_ID) + + assert result.queued_count == 1 + revisions = await nutrition.revisions(entry_id) + assert [revision["operation"] for revision in revisions] == [ + "upsert", + "delete", + "upsert", + ] + assert revisions[1]["resource_name"] == first_revision["resource_name"] + + +@pytest.mark.asyncio +async def test_meals_added_after_completed_backfill_are_not_scanned_again( + container: AppContainer, +) -> None: + await _add_meals(container, [USER_ID]) + await _connect(container) + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + await coordinator.run_user(USER_ID) + late_id = (await _add_meals(container, [USER_ID]))[0] + + await coordinator.run_user(USER_ID) + + assert await container.google_health_storage.nutrition.meal(late_id) is None + + +@pytest.mark.asyncio +async def test_new_meals_after_completed_backfill_use_normal_mutation_path( + container: AppContainer, +) -> None: + await _add_meals(container, [USER_ID]) + await _connect(container) + coordinator = NutritionBackfillCoordinator( + container.google_health_storage, container.calorie_storage + ) + await coordinator.run_user(USER_ID) + worker = MagicMock() + container.nutrition_export_worker = worker + + service = MealService(container) + entry_id, status = await service.mutate( + USER_ID, + private=True, + entry=_entry(USER_ID, 99), + ) + + assert status == "pending" + assert await container.google_health_storage.nutrition.meal(entry_id) is not None + worker.wake.assert_called_once() diff --git a/tests/test_nutrition_storage.py b/tests/test_nutrition_storage.py index e4513e5..5c59d73 100644 --- a/tests/test_nutrition_storage.py +++ b/tests/test_nutrition_storage.py @@ -3,8 +3,10 @@ from __future__ import annotations import asyncio +import json from collections.abc import AsyncGenerator from typing import Any +from unittest.mock import AsyncMock, patch import aiosqlite import pytest @@ -312,6 +314,100 @@ async def test_result_blocks_stale_write_when_revision_changed( assert row["status"] == "pending" +async def test_standalone_state_transaction_rolls_back_on_failure( + storage: SqliteGoogleHealthStorage, +) -> None: + with ( + patch.object( + storage.nutrition, + "_revision_state_unlocked", + new=AsyncMock(side_effect=RuntimeError("state write failed")), + ), + pytest.raises(RuntimeError, match="state write failed"), + ): + await storage.nutrition.revision_state(1, "in_flight") + + assert await storage.nutrition.meal(1) is None + + +async def test_retry_failed_requeues_matching_revision_and_resets_backoff( + storage: SqliteGoogleHealthStorage, +) -> None: + await storage.nutrition.enqueue( + meal_id=33, + owner_id=OWNER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + operation="upsert", + ) + row = await storage.nutrition.meal(33) + assert row is not None + desired_revision = str(row["desired_revision"]) + revision = (await storage.nutrition.revisions(33))[0] + await storage.nutrition.revision_state(int(revision["sequence"]), "failed") + await storage.nutrition.result( + 33, + "failed", + error="invalid_argument", + next_attempt=999, + expected_revision=desired_revision, + ) + + requeued = await storage.nutrition.retry_failed(USER_ID, HEALTH_USER_ID) + + assert requeued == 1 + row = await storage.nutrition.meal(33) + assert row is not None + assert row["status"] == "pending" + assert row["attempts"] == 0 + assert row["next_attempt"] == 0 + assert row["error_code"] is None + assert (await storage.nutrition.revisions(33))[0]["state"] == "queued" + + +async def test_retry_failed_does_not_touch_non_failed_or_mismatched_work( + storage: SqliteGoogleHealthStorage, +) -> None: + await storage.nutrition.enqueue( + meal_id=34, + owner_id=OWNER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + operation="upsert", + ) + await storage.nutrition.enqueue( + meal_id=35, + owner_id=OWNER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + operation="upsert", + ) + await storage.nutrition.enqueue( + meal_id=36, + owner_id=OWNER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + operation="upsert", + ) + for meal_id, state in ((34, "queued"), (35, "in_flight"), (36, "failed")): + revision = (await storage.nutrition.revisions(meal_id))[0] + await storage.nutrition.revision_state(int(revision["sequence"]), state) + await storage.nutrition.result( + meal_id, "pending" if meal_id != 36 else "failed" + ) + await storage.nutrition._conn.execute( + "UPDATE nutrition_exports SET status = 'authorization_required' " + "WHERE meal_id = 36" + ) + + assert await storage.nutrition.retry_failed(USER_ID, HEALTH_USER_ID) == 0 + assert (await storage.nutrition.revisions(36))[0]["state"] == "failed" + + # --- counts() -------------------------------------------------------------- @@ -360,3 +456,407 @@ async def test_counts_excludes_cancelled_rows( assert counts == {"pending": 1, "synced": 1} assert "cancelled" not in counts + + +async def test_has_other_account_detects_retained_account_state( + storage: SqliteGoogleHealthStorage, +) -> None: + assert not await storage.nutrition.has_other_account(USER_ID, HEALTH_USER_ID) + await storage.nutrition.enqueue( + meal_id=42, + owner_id=OWNER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + operation="upsert", + ) + + assert not await storage.nutrition.has_other_account(USER_ID, HEALTH_USER_ID) + assert await storage.nutrition.has_other_account(USER_ID, "other-health") + + +async def test_latest_remote_resource_filters_revision_accounts_and_null_names( + storage: SqliteGoogleHealthStorage, +) -> None: + await storage.nutrition.enqueue( + meal_id=50, + owner_id=OWNER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + operation="upsert", + ) + resource_name = (await storage.nutrition.revisions(50))[0]["resource_name"] + await storage.nutrition.revision_state(1, "synced") + await storage.nutrition.enqueue( + meal_id=50, + owner_id=OWNER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=None, + operation="delete", + target_resource_name=resource_name, + ) + await storage.nutrition._conn.execute( + """ + INSERT INTO nutrition_revisions + (meal_id, resource_name, operation, payload_json, state) + VALUES (?, NULL, 'upsert', ?, 'synced') + """, + (50, "{}"), + ) + + assert await storage.nutrition.latest_remote_resource(50, "other-account") is None + assert ( + await storage.nutrition.latest_remote_resource(50, HEALTH_USER_ID) + == resource_name + ) + + await storage.nutrition._conn.execute( + """ + INSERT INTO nutrition_export_history + (meal_id, telegram_user_id, health_user_id, resource_name, + operation, payload_json, state, backfill_version, updated_at) + VALUES (?, ?, ?, ?, 'upsert', ?, 'synced', 1, 1) + """, + ( + 51, + USER_ID, + HEALTH_USER_ID, + "users/health-user-1/dataTypes/nutrition-log/dataPoints/point-51", + "{}", + ), + ) + assert ( + await storage.nutrition.latest_remote_resource(51) + == "users/health-user-1/dataTypes/nutrition-log/dataPoints/point-51" + ) + + +async def test_record_remote_result_ignores_nonterminal_and_invalid_revisions( + storage: SqliteGoogleHealthStorage, +) -> None: + await storage.nutrition.enqueue( + meal_id=52, + owner_id=OWNER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + operation="upsert", + ) + queued_sequence = int((await storage.nutrition.revisions(52))[0]["sequence"]) + await storage.nutrition.record_remote_result(queued_sequence) + + await storage.nutrition._conn.execute( + """ + INSERT INTO nutrition_revisions + (meal_id, resource_name, operation, payload_json, state) + VALUES (?, NULL, 'upsert', ?, 'synced') + """, + (52, "{}"), + ) + null_sequence = int((await storage.nutrition.revisions(52))[-1]["sequence"]) + await storage.nutrition.record_remote_result(null_sequence) + + await storage.nutrition._conn.execute( + """ + INSERT INTO nutrition_revisions + (meal_id, resource_name, operation, payload_json, state) + VALUES (?, 'not-a-resource', 'upsert', ?, 'synced') + """, + (52, "{}"), + ) + invalid_sequence = int((await storage.nutrition.revisions(52))[-1]["sequence"]) + await storage.nutrition.record_remote_result(invalid_sequence) + + assert ( + await storage.nutrition._fetch_all( + "SELECT * FROM nutrition_export_history WHERE meal_id = ?", (52,) + ) + == [] + ) + + +async def test_ensure_backfill_validates_identity_and_reuses_current_work( + storage: SqliteGoogleHealthStorage, +) -> None: + await storage.nutrition.enqueue( + meal_id=60, + owner_id=OWNER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + operation="upsert", + ) + with pytest.raises(ValueError, match="owner cannot change"): + await storage.nutrition.ensure_backfill_export( + meal_id=60, + owner_id="other-owner", + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + ) + with pytest.raises(ValueError, match="identity cannot change"): + await storage.nutrition.ensure_backfill_export( + meal_id=60, + owner_id=OWNER_ID, + telegram_user_id="other-user", + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + ) + with pytest.raises(ValueError, match="account cannot change"): + await storage.nutrition.ensure_backfill_export( + meal_id=60, + owner_id=OWNER_ID, + telegram_user_id=USER_ID, + health_user_id="other-account", + payload=PAYLOAD, + ) + assert ( + await storage.nutrition.ensure_backfill_export( + meal_id=60, + owner_id=OWNER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + ) + == "existing" + ) + + +async def test_ensure_backfill_restores_history_without_current_export_row( + storage: SqliteGoogleHealthStorage, +) -> None: + resource_name = "users/health-user-1/dataTypes/nutrition-log/dataPoints/point-61" + await storage.nutrition._conn.execute( + """ + INSERT INTO nutrition_export_history + (meal_id, telegram_user_id, health_user_id, resource_name, + operation, payload_json, state, backfill_version, updated_at) + VALUES (?, ?, ?, ?, 'upsert', ?, 'synced', 1, 1) + """, + (61, USER_ID, HEALTH_USER_ID, resource_name, json.dumps(PAYLOAD)), + ) + + result = await storage.nutrition.ensure_backfill_export( + meal_id=61, + owner_id=OWNER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + ) + + assert result == "history" + row = await storage.nutrition.meal(61) + assert row is not None + assert row["status"] == "synced" + assert row["desired_revision"] == resource_name + + +async def test_ensure_backfill_restores_synced_revision_when_history_was_not_recorded( + storage: SqliteGoogleHealthStorage, +) -> None: + await storage.nutrition.enqueue( + meal_id=62, + owner_id=OWNER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + operation="upsert", + ) + revision = (await storage.nutrition.revisions(62))[0] + await storage.nutrition.revision_state(int(revision["sequence"]), "synced") + await storage.nutrition.cancel(USER_ID) + + result = await storage.nutrition.ensure_backfill_export( + meal_id=62, + owner_id=OWNER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + ) + + assert result == "history" + history = await storage.nutrition._fetch_one( + "SELECT state FROM nutrition_export_history WHERE meal_id = ?", (62,) + ) + assert history is not None + assert history["state"] == "synced" + + +async def test_ensure_backfill_handles_malformed_history_payload( + storage: SqliteGoogleHealthStorage, +) -> None: + await storage.nutrition._conn.execute( + """ + INSERT INTO nutrition_export_history + (meal_id, telegram_user_id, health_user_id, resource_name, + operation, payload_json, state, backfill_version, updated_at) + VALUES (?, ?, ?, ?, 'upsert', '{bad', 'synced', 1, 1) + """, + ( + 63, + USER_ID, + HEALTH_USER_ID, + "users/health-user-1/dataTypes/nutrition-log/dataPoints/point-63", + ), + ) + + result = await storage.nutrition.ensure_backfill_export( + meal_id=63, + owner_id=OWNER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + ) + + assert result == "queued" + assert [ + revision["operation"] for revision in await storage.nutrition.revisions(63) + ] == ["delete", "upsert"] + + +async def test_retry_failed_skips_missing_revision_and_rolls_back_failures( + storage: SqliteGoogleHealthStorage, +) -> None: + await storage.nutrition._conn.execute( + """ + INSERT INTO nutrition_exports + (meal_id, owner_id, telegram_user_id, health_user_id, + desired_revision, desired_operation, status) + VALUES (?, ?, ?, ?, 'missing-resource', 'upsert', 'failed') + """, + (70, OWNER_ID, USER_ID, HEALTH_USER_ID), + ) + assert await storage.nutrition.retry_failed(USER_ID, HEALTH_USER_ID) == 0 + + with ( + patch.object( + storage.nutrition, + "_fetch_all", + new=AsyncMock(side_effect=RuntimeError("database failure")), + ), + pytest.raises(RuntimeError, match="database failure"), + ): + await storage.nutrition.retry_failed(USER_ID, HEALTH_USER_ID) + + +async def test_backfill_ledger_claim_resume_and_expiry( + storage: SqliteGoogleHealthStorage, +) -> None: + user_id = "telegram-chat-7" + health_user_id = "health-account-7" + claimed = await storage.nutrition.claim_backfill( + user_id, health_user_id, 10, now=100, lease_seconds=10 + ) + assert claimed is not None + assert claimed["status"] == "running" + assert claimed["high_water_meal_id"] == 10 + assert ( + await storage.nutrition.claim_backfill( + user_id, health_user_id, 99, now=105, lease_seconds=10 + ) + is None + ) + resumed = await storage.nutrition.claim_backfill( + user_id, health_user_id, 99, now=110, lease_seconds=10 + ) + assert resumed is not None + assert resumed["high_water_meal_id"] == 10 + + assert await storage.nutrition.advance_backfill( + user_id, + health_user_id, + 5, + queued_count=2, + skipped_count=1, + now=200, + lease_seconds=10, + ) + assert await storage.nutrition.advance_backfill( + user_id, + health_user_id, + 10, + queued_count=1, + skipped_count=0, + now=300, + lease_seconds=10, + ) + completed = await storage.nutrition.get_backfill(user_id, health_user_id) + assert completed is not None + assert completed["status"] == "completed" + assert completed["queued_count"] == 3 + assert completed["skipped_count"] == 1 + assert ( + await storage.nutrition.claim_backfill(user_id, health_user_id, 10, now=400) + is None + ) + + +async def test_cancelled_backfill_recaptures_new_high_water_mark( + storage: SqliteGoogleHealthStorage, +) -> None: + user_id = "telegram-chat-11" + health_user_id = "health-account-11" + await storage.nutrition.claim_backfill(user_id, health_user_id, 10) + await storage.nutrition.advance_backfill( + user_id, + health_user_id, + 10, + queued_count=1, + skipped_count=0, + ) + + await storage.nutrition.cancel(user_id) + reclaimed = await storage.nutrition.claim_backfill(user_id, health_user_id, 20) + + assert reclaimed is not None + assert reclaimed["high_water_meal_id"] == 20 + assert reclaimed["cursor_meal_id"] == 0 + + +async def test_backfill_ledger_handles_unknown_rows_and_transaction_errors( + storage: SqliteGoogleHealthStorage, +) -> None: + assert not await storage.nutrition.advance_backfill( + "telegram-chat-8", + "health-account-8", + 1, + queued_count=0, + skipped_count=0, + ) + with ( + patch.object( + storage.nutrition, + "_fetch_one", + new=AsyncMock(side_effect=RuntimeError("claim failure")), + ), + pytest.raises(RuntimeError, match="claim failure"), + ): + await storage.nutrition.claim_backfill("telegram-chat-9", "health-account-9", 1) + + user_id = "telegram-chat-10" + health_user_id = "health-account-10" + await storage.nutrition.claim_backfill(user_id, health_user_id, 1) + with ( + patch.object( + storage.nutrition, + "_advance_backfill_unlocked", + new=AsyncMock(side_effect=RuntimeError("advance failure")), + ), + pytest.raises(RuntimeError, match="advance failure"), + ): + await storage.nutrition.advance_backfill( + user_id, + health_user_id, + 1, + queued_count=0, + skipped_count=0, + ) + await storage.nutrition.fail_backfill( + user_id, health_user_id, "bad\nprovider error", now=500 + ) + failed = await storage.nutrition.get_backfill(user_id, health_user_id) + assert failed is not None + assert failed["status"] == "pending" + assert failed["last_error"] == "backfill_error" diff --git a/tests/test_nutrition_tools.py b/tests/test_nutrition_tools.py new file mode 100644 index 0000000..13e093c --- /dev/null +++ b/tests/test_nutrition_tools.py @@ -0,0 +1,222 @@ +"""Tests for private meal-export status and retry tools.""" + +from __future__ import annotations + +from collections.abc import AsyncGenerator +from types import SimpleNamespace +from typing import cast +from unittest.mock import MagicMock, patch + +import aiosqlite +import pytest +from cryptography.fernet import Fernet +from google.adk.tools import ToolContext + +from blacki.calories.tools import get_meal_sync_status, retry_meal_sync +from blacki.container import AppContainer, reset_container_for_tests, set_container +from blacki.health.config import GOOGLE_HEALTH_NUTRITION_SCOPES, GoogleHealthConfig + +USER_ID = "telegram-chat-42" +HEALTH_USER_ID = "google-account-42" +PAYLOAD = { + "nutritionLog": { + "foodDisplayName": "Oatmeal", + "energy": {"kcal": 300}, + } +} + + +@pytest.fixture +async def container() -> AsyncGenerator[AppContainer, None]: + conn = await aiosqlite.connect(":memory:", isolation_level=None) + conn.row_factory = aiosqlite.Row + app_container = AppContainer(conn=conn) + await app_container.initialize_all_storages() + set_container(app_container) + yield app_container + reset_container_for_tests() + await app_container.close() + + +def _context(*, private: bool = True, user_id: str = USER_ID) -> ToolContext: + return cast( + ToolContext, + SimpleNamespace( + user_id=user_id, + state={"telegram_chat_type": "private" if private else "group"}, + ), + ) + + +async def _connect(container: AppContainer, scopes: tuple[str, ...]) -> None: + config = GoogleHealthConfig( + client_id="client-id", + client_secret="client-secret", + redirect_uri="https://example.test/callback", + token_encryption_key=Fernet.generate_key().decode(), + ) + await container.google_health_storage.upsert_connection( + telegram_user_id=USER_ID, + encrypted_refresh_token=config.cipher.encrypt("refresh-token"), + health_user_id=HEALTH_USER_ID, + legacy_fitbit_user_id=None, + scopes=scopes, + ) + + +@pytest.mark.asyncio +async def test_get_meal_sync_status_returns_connection_and_counts( + container: AppContainer, +) -> None: + await _connect(container, GOOGLE_HEALTH_NUTRITION_SCOPES) + await container.google_health_storage.nutrition.enqueue( + meal_id=1, + owner_id=USER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + operation="upsert", + ) + + result = await get_meal_sync_status(_context()) + + assert result == { + "status": "success", + "google_health_connection": "connected", + "nutrition_permissions": True, + "google_health_sync": {"pending": 1}, + } + + +@pytest.mark.asyncio +async def test_retry_meal_sync_requeues_failed_rows_and_wakes_worker( + container: AppContainer, +) -> None: + await _connect(container, GOOGLE_HEALTH_NUTRITION_SCOPES) + nutrition = container.google_health_storage.nutrition + await nutrition.enqueue( + meal_id=2, + owner_id=USER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + operation="upsert", + ) + revision = (await nutrition.revisions(2))[0] + await nutrition.revision_state(int(revision["sequence"]), "failed") + await nutrition.result(2, "failed", error="bad_request") + worker = MagicMock() + container.nutrition_export_worker = worker + + result = await retry_meal_sync(_context()) + + assert result["status"] == "success" + assert result["requeued"] == 1 + assert result["google_health_sync"] == {"pending": 1} + worker.wake.assert_called_once() + assert (await nutrition.revisions(2))[0]["state"] == "queued" + + +@pytest.mark.asyncio +async def test_retry_meal_sync_requires_private_chat_and_nutrition_scopes( + container: AppContainer, +) -> None: + private_error = await retry_meal_sync(_context(private=False)) + assert private_error["status"] == "error" + + await _connect(container, ()) + missing_scopes = await retry_meal_sync(_context()) + assert missing_scopes["status"] == "authorization_required" + assert missing_scopes["requeued"] == 0 + + +@pytest.mark.asyncio +async def test_status_tools_return_safe_no_connection_and_invalid_identity( + container: AppContainer, +) -> None: + no_connection = await get_meal_sync_status(_context()) + assert no_connection["google_health_connection"] == "not_connected" + assert no_connection["google_health_sync"] == {} + + invalid = await get_meal_sync_status(_context(user_id="not-a-private-user")) + assert invalid["status"] == "error" + private_error = await get_meal_sync_status(_context(private=False)) + assert private_error["status"] == "error" + retry_invalid = await retry_meal_sync(_context(user_id="not-a-private-user")) + assert retry_invalid["status"] == "error" + retry_no_connection = await retry_meal_sync(_context()) + assert retry_no_connection["status"] == "not_connected" + + +@pytest.mark.asyncio +async def test_status_tools_handle_uninitialized_storage() -> None: + reset_container_for_tests() + context = _context() + assert (await get_meal_sync_status(context))["status"] == "error" + assert (await retry_meal_sync(context))["status"] == "error" + + +@pytest.mark.asyncio +async def test_status_tools_handle_storage_failure( + container: AppContainer, +) -> None: + with patch.object( + container.google_health_storage, + "get_connection", + side_effect=RuntimeError("storage unavailable"), + ): + result = await get_meal_sync_status(_context()) + assert result["status"] == "error" + assert result["message"] == "Health storage is not initialized" + + +@pytest.mark.asyncio +async def test_status_tools_handle_unexpected_failures( + container: AppContainer, +) -> None: + with patch.object( + container.google_health_storage, + "get_connection", + side_effect=ValueError("unexpected"), + ): + status = await get_meal_sync_status(_context()) + assert status == { + "status": "error", + "message": "Could not read meal export status", + } + + await _connect(container, GOOGLE_HEALTH_NUTRITION_SCOPES) + with patch.object( + container.google_health_storage.nutrition, + "retry_failed", + side_effect=ValueError("unexpected"), + ): + retry = await retry_meal_sync(_context()) + assert retry == { + "status": "error", + "message": "Could not retry meal exports", + } + + +@pytest.mark.asyncio +async def test_retry_without_running_worker_returns_counts( + container: AppContainer, +) -> None: + await _connect(container, GOOGLE_HEALTH_NUTRITION_SCOPES) + nutrition = container.google_health_storage.nutrition + await nutrition.enqueue( + meal_id=3, + owner_id=USER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + operation="upsert", + ) + revision = (await nutrition.revisions(3))[0] + await nutrition.revision_state(int(revision["sequence"]), "failed") + await nutrition.result(3, "failed", error="bad_request") + + result = await retry_meal_sync(_context()) + + assert result["status"] == "success" + assert result["requeued"] == 1 diff --git a/tests/test_nutrition_worker.py b/tests/test_nutrition_worker.py index 4c64fbd..6d6fcc4 100644 --- a/tests/test_nutrition_worker.py +++ b/tests/test_nutrition_worker.py @@ -192,6 +192,77 @@ async def test_worker_marks_permanent_failure( assert revisions[0]["state"] == "failed" +async def test_worker_does_not_mark_failed_revision_as_synced( + storage: SqliteGoogleHealthStorage, +) -> None: + config = _config() + await _connect(storage, config) + await storage.nutrition.enqueue( + meal_id=12, + owner_id=USER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + operation="upsert", + ) + revision = (await storage.nutrition.revisions(12))[0] + await storage.nutrition.revision_state(int(revision["sequence"]), "failed") + await storage.nutrition._conn.execute( + "UPDATE nutrition_exports SET status = 'pending' WHERE meal_id = 12" + ) + worker, client = _worker(storage, config) + + await worker._dispatch_due() + + row = await storage.nutrition.meal(12) + assert row is not None + assert row["status"] == "pending" + client.refresh_access_token.assert_not_awaited() + + +def test_desired_revision_state_returns_none_for_missing_revision() -> None: + from blacki.health.nutrition_worker import _desired_revision_state + + assert _desired_revision_state([], "missing-resource") is None + + +async def test_worker_retries_explicitly_requeued_failure( + storage: SqliteGoogleHealthStorage, +) -> None: + config = _config() + await _connect(storage, config) + await storage.nutrition.enqueue( + meal_id=13, + owner_id=USER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + operation="upsert", + ) + worker, client = _worker(storage, config) + client.refresh_access_token.return_value = _token() + client.create_nutrition_log.side_effect = [ + GoogleHealthApiError("bad request", status_code=400, error_code="bad_request"), + GoogleHealthOperation(done=True, name="op/13", response={}), + ] + + await worker._dispatch_due() + failed_revision = (await storage.nutrition.revisions(13))[0] + failed_resource = failed_revision["resource_name"] + assert await storage.nutrition.retry_failed(USER_ID, HEALTH_USER_ID) == 1 + + await worker._dispatch_due() + + row = await storage.nutrition.meal(13) + assert row is not None + assert row["status"] == "synced" + revisions = await storage.nutrition.revisions(13) + assert len(revisions) == 1 + assert revisions[0]["resource_name"] == failed_resource + assert revisions[0]["state"] == "synced" + assert client.create_nutrition_log.await_count == 2 + + async def test_worker_pauses_on_auth_error( storage: SqliteGoogleHealthStorage, ) -> None: @@ -503,6 +574,38 @@ async def test_worker_disconnected_marks_authorization_required( client.refresh_access_token.assert_not_awaited() +async def test_worker_retires_stale_revision_after_account_switch( + storage: SqliteGoogleHealthStorage, +) -> None: + config = _config() + await _connect(storage, config) + await storage.nutrition.enqueue( + meal_id=11, + owner_id=USER_ID, + telegram_user_id=USER_ID, + health_user_id=HEALTH_USER_ID, + payload=PAYLOAD, + operation="upsert", + ) + await storage.conn.execute( + "UPDATE google_health_connections SET health_user_id = ? " + "WHERE telegram_user_id = ?", + ("replacement-health-id", USER_ID), + ) + worker, client = _worker(storage, config) + + await worker._dispatch_due() + + row = await storage.nutrition.meal(11) + assert row is not None + assert row["status"] == "cancelled" + connection = await storage.get_connection(USER_ID) + assert connection is not None + assert connection.health_user_id == "replacement-health-id" + assert connection.status == "connected" + client.refresh_access_token.assert_not_awaited() + + async def test_worker_start_stop_is_idempotent( storage: SqliteGoogleHealthStorage, ) -> None: diff --git a/tests/test_privacy.py b/tests/test_privacy.py index 4f55696..6a1b232 100644 --- a/tests/test_privacy.py +++ b/tests/test_privacy.py @@ -63,6 +63,8 @@ def test_private_tool_identification_uses_zepto_prefix() -> None: "delete_meal", "get_calorie_summary", "set_calorie_goal", + "get_meal_sync_status", + "retry_meal_sync", ], ) def test_calorie_tools_are_private(tool_name: str) -> None: diff --git a/tests/test_prompt.py b/tests/test_prompt.py index 866de27..50e1203 100644 --- a/tests/test_prompt.py +++ b/tests/test_prompt.py @@ -136,11 +136,11 @@ def test_nutrition_policy_separates_local_save_and_google_sync() -> None: "Log my lunch", {"log_meal", "edit_meal", "delete_meal"} ) - assert "google_health_sync status separately" in instruction + assert "Do not mention a pending background export" in instruction assert "saved in Blacki" in instruction assert "Never claim Google Health accepted" in instruction assert "Do not repeat a meal mutation" in instruction - assert "no historical backfill" in instruction + assert "queues eligible existing meals once" in instruction class TestDomainPolicyAssembly: diff --git a/tests/test_registry.py b/tests/test_registry.py index 630a483..2b7c8c4 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -538,7 +538,7 @@ def test_returns_tools_when_available(self) -> None: tools = _build_calorie_tools() - assert len(tools) == 5 + assert len(tools) == 7 class TestBuildWorkoutTools: diff --git a/tests/test_server_config.py b/tests/test_server_config.py index a099dcb..1bc02db 100644 --- a/tests/test_server_config.py +++ b/tests/test_server_config.py @@ -1,12 +1,14 @@ # mypy: disable-error-code="no-untyped-def" """Tests for server configuration.""" +import asyncio import json import sys from collections.abc import Generator from typing import Any, cast from unittest.mock import AsyncMock, MagicMock, patch +import aiosqlite import pytest from fastapi import FastAPI @@ -601,11 +603,13 @@ async def test_google_health_callback_handles_cancel_and_safe_errors( service.complete_authorization = AsyncMock( return_value=OAuthCompletion("telegram-chat-42", connected=False) ) - response = await server.google_health_callback( - state="state", code=None, error="access_denied" - ) + with patch.object(server, "_schedule_google_health_backfill") as schedule: + response = await server.google_health_callback( + state="state", code=None, error="access_denied" + ) assert response.status_code == 200 assert b"cancelled" in response.body + schedule.assert_not_called() service.complete_authorization = AsyncMock( side_effect=GoogleHealthOAuthError("state secret") @@ -640,8 +644,10 @@ async def test_google_health_callback_handles_cancel_and_safe_errors( bot = MagicMock() bot.notify_health_connection = AsyncMock(side_effect=RuntimeError("notify")) server._telegram_bot = bot - response = await server.google_health_callback(state="state", code="code") + with patch.object(server, "_schedule_google_health_backfill") as schedule: + response = await server.google_health_callback(state="state", code="code") assert response.status_code == 200 + schedule.assert_called_once_with("telegram-chat-42") finally: server._google_health_service = None server._telegram_bot = None @@ -718,6 +724,137 @@ async def test_google_health_start_and_stop_are_optional( server._container = None +@pytest.mark.asyncio +async def test_google_health_backfill_schedule_runs_global_and_user_tasks( + mock_dependencies: MagicMock, +) -> None: + if "blacki.server" in sys.modules: + del sys.modules["blacki.server"] + + import blacki.server as server + from blacki.container import AppContainer + + conn = await aiosqlite.connect(":memory:", isolation_level=None) + conn.row_factory = aiosqlite.Row + container = AppContainer(conn=conn) + await container.initialize_all_storages() + worker = MagicMock() + server._container = container + server._google_health_export_worker = worker + run_all = AsyncMock(return_value=[]) + run_user = AsyncMock(return_value=None) + try: + with ( + patch( + "blacki.health.nutrition_backfill.NutritionBackfillCoordinator.run_all_eligible", + new=run_all, + ), + patch( + "blacki.health.nutrition_backfill.NutritionBackfillCoordinator.run_user", + new=run_user, + ), + ): + server._schedule_google_health_backfill() + server._schedule_google_health_backfill("telegram-chat-42") + tasks = list(server._google_health_backfill_tasks) + await asyncio.gather(*tasks) + + run_all.assert_awaited_once() + run_user.assert_awaited_once_with("telegram-chat-42") + server._google_health_export_worker = None + server._schedule_google_health_backfill() + assert not server._google_health_backfill_tasks + finally: + server._google_health_backfill_tasks.clear() + server._container = None + server._google_health_export_worker = None + await container.close() + + +@pytest.mark.asyncio +async def test_google_health_backfill_task_swallows_unexpected_errors( + mock_dependencies: MagicMock, + caplog: pytest.LogCaptureFixture, +) -> None: + if "blacki.server" in sys.modules: + del sys.modules["blacki.server"] + + import blacki.server as server + from blacki.container import AppContainer + + conn = await aiosqlite.connect(":memory:", isolation_level=None) + conn.row_factory = aiosqlite.Row + container = AppContainer(conn=conn) + await container.initialize_all_storages() + server._container = container + server._google_health_export_worker = MagicMock() + try: + with ( + patch( + "blacki.health.nutrition_backfill.NutritionBackfillCoordinator.run_all_eligible", + new=AsyncMock(side_effect=RuntimeError("backfill failure")), + ), + caplog.at_level("ERROR", logger="blacki.server"), + ): + server._schedule_google_health_backfill() + await asyncio.gather(*list(server._google_health_backfill_tasks)) + assert "Google Health nutrition backfill task failed" in caplog.text + finally: + server._google_health_backfill_tasks.clear() + server._container = None + server._google_health_export_worker = None + await container.close() + + +@pytest.mark.asyncio +async def test_google_health_backfill_task_preserves_cancellation( + mock_dependencies: MagicMock, +) -> None: + if "blacki.server" in sys.modules: + del sys.modules["blacki.server"] + + import blacki.server as server + from blacki.container import AppContainer + + conn = await aiosqlite.connect(":memory:", isolation_level=None) + conn.row_factory = aiosqlite.Row + container = AppContainer(conn=conn) + await container.initialize_all_storages() + server._container = container + server._google_health_export_worker = MagicMock() + try: + with patch( + "blacki.health.nutrition_backfill.NutritionBackfillCoordinator.run_all_eligible", + new=AsyncMock(side_effect=asyncio.CancelledError), + ): + server._schedule_google_health_backfill() + task = next(iter(server._google_health_backfill_tasks)) + await asyncio.gather(task, return_exceptions=True) + assert task.cancelled() + finally: + server._google_health_backfill_tasks.clear() + server._container = None + server._google_health_export_worker = None + await container.close() + + +@pytest.mark.asyncio +async def test_google_health_stop_cancels_backfill_tasks( + mock_dependencies: MagicMock, +) -> None: + if "blacki.server" in sys.modules: + del sys.modules["blacki.server"] + + import blacki.server as server + + task = asyncio.create_task(asyncio.sleep(60)) + server._google_health_backfill_tasks.add(task) + server._container = None + await server._stop_google_health() + + assert task.cancelled() + + @pytest.mark.asyncio async def test_google_health_stop_suppresses_scheduler_and_client_errors( mock_dependencies: MagicMock, diff --git a/tests/test_telegram_health.py b/tests/test_telegram_health.py index e96295f..de6ed0e 100644 --- a/tests/test_telegram_health.py +++ b/tests/test_telegram_health.py @@ -83,7 +83,7 @@ async def test_connect_health_sends_protected_authorization_link() -> None: .url.startswith("https://accounts.google.com/") ) assert "future meal logs, edits, and deletions" in kwargs["text"] - assert "not backfilled" in kwargs["text"] + assert "meals already saved" in kwargs["text"] assert "verify records it created" in kwargs["text"] assert "Read-only summaries remain available" in kwargs["text"] assert "Apple ID" in kwargs["text"]