feat: implement state expiry for chores and rewards, adding scheduler and event handling

backend/utils/state_expiry_scheduler.py

@@ -0,0 +1,131 @@
+import logging
+import os
+from datetime import datetime, timezone
+
+from apscheduler.schedulers.background import BackgroundScheduler
+from tinydb import Query
+
+from events.types.child_chore_confirmation import ChildChoreConfirmation
+from events.types.child_reward_request import ChildRewardRequest
+from events.types.event import Event
+from events.types.event_types import EventType
+from events.sse import send_event_to_user
+
+logger = logging.getLogger(__name__)
+
+
+def _get_today_start_timestamp(tz_str: str | None) -> float:
+    """Return a Unix timestamp for the start of today (midnight) in the user's timezone."""
+    try:
+        if tz_str:
+            from zoneinfo import ZoneInfo
+            now_local = datetime.now(ZoneInfo(tz_str))
+            today_local = now_local.replace(hour=0, minute=0, second=0, microsecond=0)
+            return today_local.timestamp()
+    except Exception:
+        pass
+    now_utc = datetime.now(timezone.utc)
+    today_utc = now_utc.replace(hour=0, minute=0, second=0, microsecond=0)
+    return today_utc.timestamp()
+
+
+def expire_stale_pending_for_user(user_id: str, tz_str: str | None) -> int:
+    """Delete stale pending confirmations for a user and fire SSE events.
+
+    A record is stale if status='pending' and created_at is before the start of
+    today in the user's timezone. Returns the number of records expired.
+    Must be called within a Flask app context.
+    """
+    from db.db import pending_confirmations_db
+
+    today_start = _get_today_start_timestamp(tz_str)
+    PendingQ = Query()
+
+    stale_items = pending_confirmations_db.search(
+        (PendingQ.user_id == user_id) &
+        (PendingQ.status == 'pending') &
+        (PendingQ.created_at < today_start)
+    )
+
+    if not stale_items:
+        return 0
+
+    for item in stale_items:
+        item_id = item.get('id')
+        child_id = item.get('child_id')
+        entity_id = item.get('entity_id')
+        entity_type = item.get('entity_type')
+
+        pending_confirmations_db.remove(PendingQ.id == item_id)
+
+        if entity_type == 'chore':
+            event = Event(
+                EventType.CHILD_CHORE_CONFIRMATION.value,
+                ChildChoreConfirmation(
+                    child_id=child_id,
+                    task_id=entity_id,
+                    operation=ChildChoreConfirmation.OPERATION_RESET,
+                )
+            )
+        else:
+            event = Event(
+                EventType.CHILD_REWARD_REQUEST.value,
+                ChildRewardRequest(
+                    child_id=child_id,
+                    reward_id=entity_id,
+                    operation=ChildRewardRequest.REQUEST_CANCELLED,
+                )
+            )
+
+        send_event_to_user(user_id, event)
+        logger.info(
+            f'State expiry: expired {entity_type} {entity_id} for child {child_id} (user {user_id})'
+        )
+
+    return len(stale_items)
+
+
+def run_state_expiry_check(app) -> None:
+    """Check all verified users and expire stale pending records."""
+    if os.environ.get('DB_ENV') == 'e2e':
+        return
+
+    with app.app_context():
+        try:
+            from db.db import users_db
+
+            UserQ = Query()
+            users = users_db.search(UserQ.verified == True)
+
+            for user_dict in users:
+                user_id = user_dict.get('id')
+                tz_str = user_dict.get('timezone')
+                try:
+                    expired = expire_stale_pending_for_user(user_id, tz_str)
+                    if expired:
+                        logger.info(f'State expiry: expired {expired} record(s) for user {user_id}')
+                except Exception as e:
+                    logger.error(f'State expiry: failed for user {user_id}: {e}')
+
+        except Exception as e:
+            logger.error(f'Error in run_state_expiry_check: {e}')
+
+
+def start_state_expiry_scheduler(app) -> BackgroundScheduler:
+    """Start the hourly state expiry scheduler. Returns the scheduler instance."""
+    scheduler = BackgroundScheduler()
+    scheduler.add_job(
+        run_state_expiry_check,
+        'cron',
+        minute=0,
+        args=[app],
+        id='state_expiry_scheduler',
+        replace_existing=True,
+    )
+    scheduler.start()
+    logger.info('State expiry scheduler started (hourly)')
+
+    # Run immediately on startup to handle any records that expired while server was down
+    run_state_expiry_check(app)
+
+    return scheduler