feat(backend): 地理围栏提醒闭环 + 提醒确认置 done + 音频下发有界等待

backend/src/timeflow/business/reminders/geofence_trigger.py

@@ -0,0 +1,69 @@
+"""日程地理围栏提醒的判定逻辑。"""
+
+import math
+from collections.abc import Iterable
+from enum import Enum
+from typing import Protocol
+
+from timeflow.business.schedules import ScheduleRecord
+
+_EARTH_RADIUS_METERS = 6_371_000.0
+
+
+class GeofenceQueryPort(Protocol):
+    """business.reminders 对外声明的 Query Port,由 data 层实现。"""
+
+    def list_geofence_schedules(self, user_id: str) -> Iterable[ScheduleRecord]:
+        """返回该用户名下状态为 scheduled、绑定了经纬度且尚未触发过地理提醒的日程。"""
+        ...
+
+
+class GeofenceTransition(Enum):
+    """一次位置上报相对某条日程围栏状态的变化。"""
+
+    NO_CHANGE = "no_change"
+    ARMED = "armed"
+    TRIGGERED = "triggered"
+
+
+def evaluate_geofence(schedule: ScheduleRecord, latitude: float, longitude: float) -> GeofenceTransition:
+    """判定这次上报的位置相对该日程围栏的状态变化。
+
+    `geofence_armed=False` 表示"当前在围栏内、还没离开过"(比如日程创建时人已经在围栏内);
+    只有先侦测到离开(布防),之后再次进入才算命中,避免创建时立刻误触发。
+    """
+    is_inside = _distance_meters(schedule.latitude, schedule.longitude, latitude, longitude) <= (
+        schedule.geofence_radius_meters
+    )
+    if schedule.geofence_armed:
+        return GeofenceTransition.TRIGGERED if is_inside else GeofenceTransition.NO_CHANGE
+    return GeofenceTransition.NO_CHANGE if is_inside else GeofenceTransition.ARMED
+
+
+class GeofenceTriggerService:
+    """判定一次位置上报命中了哪些日程的围栏变化。"""
+
+    def __init__(self, query_port: GeofenceQueryPort) -> None:
+        self._query_port = query_port
+
+    def find_geofence_transitions(
+        self, user_id: str, latitude: float, longitude: float
+    ) -> list[tuple[ScheduleRecord, GeofenceTransition]]:
+        """返回该用户名下所有发生了状态变化(非 NO_CHANGE)的日程及其变化类型。"""
+        results: list[tuple[ScheduleRecord, GeofenceTransition]] = []
+        for schedule in self._query_port.list_geofence_schedules(user_id):
+            transition = evaluate_geofence(schedule, latitude, longitude)
+            if transition is not GeofenceTransition.NO_CHANGE:
+                results.append((schedule, transition))
+        return results
+
+
+def _distance_meters(lat1: float | None, lon1: float | None, lat2: float, lon2: float) -> float:
+    """Haversine 公式计算两点间的球面距离(米)。"""
+    if lat1 is None or lon1 is None:
+        return math.inf
+    phi1, phi2 = math.radians(lat1), math.radians(lat2)
+    delta_phi = math.radians(lat2 - lat1)
+    delta_lambda = math.radians(lon2 - lon1)
+    a = math.sin(delta_phi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(delta_lambda / 2) ** 2
+    return 2 * _EARTH_RADIUS_METERS * math.asin(math.sqrt(a))

backend/src/timeflow/business/reminders/reminder_dispatch.py

@@ -24,3 +24,11 @@ class ScheduleDispatchCommandPort(Protocol):
     def mark_time_triggered(self, schedule_id: str, triggered_at: datetime) -> bool:
         """记录该日程本轮时间提醒已经命中;仅在此前未记录时真正写入,返回是否发生了写入。"""
         ...
+
+    def mark_done(self, schedule_id: str, updated_at: datetime) -> bool:
+        """提醒已被客户端确认后,把日程置为 `done`,结束后续监听。
+
+        仅对仍处于 `scheduled` 的日程生效,返回是否发生了写入——已经是 `done`
+        或已被删除的日程不应该被这条路径改回去。
+        """
+        ...

backend/src/timeflow/data/schedule_dispatch.py

@@ -16,7 +16,8 @@
 
 
 class SqlAlchemyScheduleDispatchAdapter:
-    """Implements `ScheduleQueryPort` and `ScheduleDispatchCommandPort` against `schedules`."""
+    """Implements the reminders module's Query/Command ports against `schedules`
+    (time-window trigger, geofence trigger)."""
 
     def __init__(self, session: Session) -> None:
         self._session = session
@@ -43,6 +44,57 @@ def mark_time_triggered(self, schedule_id: str, triggered_at: datetime) -> bool:
         )
         return result.rowcount > 0
 
+    def mark_done(self, schedule_id: str, updated_at: datetime) -> bool:
+        """Mark an acknowledged schedule as done; only affects still-scheduled rows."""
+        result = cast(
+            CursorResult[object],
+            self._session.execute(
+                update(Schedule)
+                .where(Schedule.id == schedule_id, Schedule.status == "scheduled")
+                .values(status="done", updated_at=updated_at.isoformat())
+            ),
+        )
+        return result.rowcount > 0
+
+    def list_geofence_schedules(self, user_id: str) -> Iterable[ScheduleRecord]:
+        """Return this user's scheduled, not-yet-geo-triggered, location-only rows.
+
+        带 `start_time` 的日程(既有时间又有地点)一律走时间维度,不参与地理围栏判定——
+        否则同一条日程会被两条链路各推一次提醒(`time_triggered_at`/`geo_triggered_at`
+        是两个独立字段,各自的幂等保护互不干涉)。跟 `time_window_trigger` 里
+        "没有 start_time 的纯地点日程不参与时间维度判定" 正好对称。
+        """
+        statement = select(Schedule).where(
+            Schedule.user_id == user_id,
+            Schedule.status == "scheduled",
+            Schedule.start_time.is_(None),
+            Schedule.latitude.is_not(None),
+            Schedule.longitude.is_not(None),
+            Schedule.geo_triggered_at.is_(None),
+        )
+        rows = self._session.execute(statement).scalars().all()
+        return [_to_record(row) for row in rows]
+
+    def set_geofence_armed(self, schedule_id: str, armed: bool) -> None:
+        """Update the armed flag after detecting the user has left the geofence."""
+        self._session.execute(
+            update(Schedule)
+            .where(Schedule.id == schedule_id)
+            .values(geofence_armed=1 if armed else 0)
+        )
+
+    def mark_geo_triggered(self, schedule_id: str, triggered_at: datetime) -> bool:
+        """Record the geofence-reminder hit; only the first call for a schedule writes."""
+        result = cast(
+            CursorResult[object],
+            self._session.execute(
+                update(Schedule)
+                .where(Schedule.id == schedule_id, Schedule.geo_triggered_at.is_(None))
+                .values(geo_triggered_at=triggered_at.isoformat())
+            ),
+        )
+        return result.rowcount > 0
+
 
 def _to_record(row: Schedule) -> ScheduleRecord:
     return ScheduleRecord(

backend/src/timeflow/infrastructure/websocket/handlers/location.py

@@ -0,0 +1,91 @@
+"""WebSocket handlers for location reports (geofence trigger)."""
+
+import asyncio
+from collections.abc import Callable
+from datetime import UTC, datetime
+from typing import Any, Protocol
+
+from pydantic import ValidationError
+
+from timeflow.business.reminders.reminder_dispatch import TriggeredSchedule
+from timeflow.infrastructure.websocket.envelope import build_error_envelope
+from timeflow.infrastructure.websocket.messages.location import LocationReport, LocationReportAck
+
+
+class _Dispatcher(Protocol):
+    async def dispatch(self, schedules: list[TriggeredSchedule], now: datetime) -> None: ...
+
+
+class _Sender(Protocol):
+    async def send(self, device_id: str, message: dict[str, Any]) -> bool: ...
+
+
+class LocationWebSocketHandlers:
+    """Adapt `location.report` to the geofence trigger use case + reminder dispatch."""
+
+    def __init__(
+        self,
+        run_report: Callable[[str, float, float, datetime], list[TriggeredSchedule]],
+        dispatcher: _Dispatcher,
+        connections: _Sender,
+    ) -> None:
+        self._run_report = run_report
+        self._dispatcher = dispatcher
+        self._connections = connections
+
+    async def handle_report(self, raw_message: dict[str, Any], device_id: str) -> None:
+        """Handle `location.report`: validate, judge geofence transitions, dispatch hits.
+
+        自己发 `location.report.ack` 而不是 return 给路由层代发,是为了保证顺序:
+        先把这次上报的 ack 回出去,再推可能命中的 `reminder.control`。如果 return 给
+        路由层,dispatch 就必须发生在 return 之前,客户端会先收到提醒再收到 ack。
+        """
+        error = self._validate(raw_message)
+        if error is not None:
+            await self._connections.send(device_id, error)
+            return None
+
+        message = LocationReport.model_validate(raw_message)
+        now = datetime.now(UTC)
+        triggered = await asyncio.to_thread(
+            self._run_report, device_id, message.latitude, message.longitude, now
+        )
+        await self._connections.send(
+            device_id, LocationReportAck(ok=True).model_dump(exclude_none=True)
+        )
+        await self._dispatcher.dispatch(triggered, now)
+        return None
+
+    @staticmethod
+    def _validate(raw_message: dict[str, Any]) -> dict[str, Any] | None:
+        """返回错误信封;`None` 表示这条上报合法。"""
+        try:
+            message = LocationReport.model_validate(raw_message)
+        except ValidationError as exc:
+            return build_error_envelope(
+                "location.report.ack",
+                None,
+                "INVALID_LOCATION",
+                "位置信息不合法",
+                {"errors": exc.errors()},
+            )
+        if not -90 <= message.latitude <= 90:
+            return build_error_envelope(
+                "location.report.ack",
+                None,
+                "INVALID_LOCATION",
+                "位置信息不合法",
+                {"field": "latitude"},
+            )
+        if not -180 <= message.longitude <= 180:
+            return build_error_envelope(
+                "location.report.ack",
+                None,
+                "INVALID_LOCATION",
+                "位置信息不合法",
+                {"field": "longitude"},
+            )
+        return None
+
+
+__all__ = ["LocationWebSocketHandlers"]

backend/src/timeflow/infrastructure/websocket/handlers/schedules.py

@@ -2,9 +2,7 @@
 
 from __future__ import annotations
 
-import asyncio
 import logging
-from collections.abc import Callable
 from typing import Any
 
 from pydantic import ValidationError
@@ -36,6 +34,7 @@
     ScheduleUpsertResult,
     ScheduleUpsertResultPayload,
 )
+from timeflow.infrastructure.websocket.reminder_audio import ReminderAudioGenerationTracker
 
 logger = logging.getLogger(__name__)
 
@@ -47,10 +46,15 @@ def __init__(
         self,
         service: ScheduleService,
         reminder_audio_service: ReminderAudioGenerationPort | None = None,
+        generation_tracker: ReminderAudioGenerationTracker | None = None,
     ) -> None:
         self._service = service
         self._reminder_audio_service = reminder_audio_service
-        self._audio_generation_tasks: dict[str, asyncio.Task[bool]] = {}
+        # 生成任务的注册表交给 tracker 统一持有,`ReminderAudioSender` 下发提醒前
+        # 要靠它判断"这条日程的音频是不是还在生成中",所以不能是本类私有状态。
+        self._generation_tracker = generation_tracker or ReminderAudioGenerationTracker(
+            reminder_audio_service
+        )
 
     async def handle_upsert(
         self,
@@ -118,53 +122,16 @@ async def handle_upsert(
 
     async def aclose(self) -> None:
         """Cancel pending TTS generation tasks during application shutdown."""
-        tasks = tuple(self._audio_generation_tasks.values())
-        self._audio_generation_tasks.clear()
-        for task in tasks:
-            task.cancel()
-        if tasks:
-            await asyncio.gather(*tasks, return_exceptions=True)
+        await self._generation_tracker.aclose()
 
     async def _delete_audio(self, schedule_id: str) -> None:
         if self._reminder_audio_service is None:
             return
-        task = self._audio_generation_tasks.pop(schedule_id, None)
-        if task is not None and not task.done():
-            task.cancel()
-            await asyncio.gather(task, return_exceptions=True)
+        await self._generation_tracker.discard(schedule_id)
         await self._reminder_audio_service.delete(schedule_id)
 
     def _submit_audio_generation(self, schedule: ScheduleRecord) -> None:
-        if self._reminder_audio_service is None:
-            return
-        previous = self._audio_generation_tasks.get(schedule.id)
-        if previous is not None and not previous.done():
-            previous.cancel()
-        task = asyncio.create_task(
-            self._reminder_audio_service.generate(schedule),
-            name=f"tts-schedule-{schedule.id}",
-        )
-        self._audio_generation_tasks[schedule.id] = task
-        task.add_done_callback(self._audio_generation_callback(schedule.id))
-
-    def _audio_generation_callback(
-        self,
-        schedule_id: str,
-    ) -> Callable[[asyncio.Task[bool]], None]:
-        def callback(completed: asyncio.Task[bool]) -> None:
-            self._audio_generation_done(schedule_id, completed)
-
-        return callback
-
-    def _audio_generation_done(self, schedule_id: str, task: asyncio.Task[bool]) -> None:
-        if self._audio_generation_tasks.get(schedule_id) is task:
-            self._audio_generation_tasks.pop(schedule_id, None)
-        if task.cancelled():
-            return
-        try:
-            task.result()
-        except Exception:  # noqa: BLE001 - background generation must not break the WS loop
-            logger.exception("Reminder audio generation failed", extra={"schedule_id": schedule_id})
+        self._generation_tracker.submit(schedule)
 
     async def handle_list(
         self,