import json import logging import traceback from datetime import datetime, timezone from pathlib import Path from apscheduler.schedulers.asyncio import AsyncIOScheduler from app.config import Settings from app.database import ( get_db, get_pending_brief_items, mark_brief_items_sent, run_retention_cleanup, ) from app.lib.utils.brief_html import build_brief_html from app.lib.utils.notifier import _build_sendmail_payload, send_developer_alert logger = logging.getLogger(__name__) class RenewalScheduler: def __init__(self, subscription_manager, settings: Settings) -> None: self._scheduler = AsyncIOScheduler() self._manager = subscription_manager self._settings = settings def _renewal_interval_seconds(self) -> int: # Renew at 80% of the subscription lifetime minus 15 minutes. # With the default 4230-minute expiry that fires roughly every 53.5 hours, # well before Graph's expiry and the 60-minute cutoff in ensure_subscription. return int(self._settings.subscription_expiry_minutes * 60 * 0.80) - 900 def start(self, subscription_id: str) -> None: interval_seconds = self._renewal_interval_seconds() self._scheduler.add_job( self._renew_job, trigger="interval", seconds=interval_seconds, id="subscription_renewal", replace_existing=True, args=[subscription_id], ) self._scheduler.add_job( self._send_daily_brief, trigger="cron", hour=18, minute=0, timezone="America/Chicago", id="daily_brief", replace_existing=True, ) self._scheduler.add_job( self._run_retention, trigger="cron", hour=2, minute=0, id="retention_cleanup", replace_existing=True, ) self._scheduler.start() logger.info( "Renewal scheduler started; will renew subscription every %d seconds", interval_seconds, ) async def _renew_job(self, subscription_id: str) -> None: logger.info("Renewing subscription %s", subscription_id) try: await self._manager.renew_subscription(subscription_id) except Exception: logger.warning( "Renewal failed for %s; attempting to recreate subscription", subscription_id ) try: folder_id = await self._get_folder_id() new_id = await self._manager.ensure_subscription(folder_id) self._reschedule(new_id) except Exception: tb = traceback.format_exc() logger.exception("Failed to recreate subscription after renewal failure") await send_developer_alert( self._settings, subject="[RCM] Subscription renewal and recreation both failed", body=( f"Subscription ID: {subscription_id}\n\n" f"Renewal failed and the attempt to recreate the subscription " f"also failed. Email notifications are NOT being received.\n\n" f"{tb}" ), ) async def _get_folder_id(self) -> str: async with get_db(self._manager._db_path) as conn: from app.database import get_active_subscription sub = await get_active_subscription(conn) if sub: return sub["folder_id"] raise RuntimeError("No active subscription found to determine folder_id for recreation") def _reschedule(self, new_subscription_id: str) -> None: interval_seconds = self._renewal_interval_seconds() self._scheduler.reschedule_job( "subscription_renewal", trigger="interval", seconds=interval_seconds, args=[new_subscription_id], ) logger.info("Rescheduled renewal for new subscription %s", new_subscription_id) async def _send_daily_brief(self) -> None: db_path = self._settings.database_path async with get_db(db_path) as conn: items = await get_pending_brief_items(conn) if not items: logger.info("Daily brief: no pending items, skipping send") return recipients = self._settings.brief_recipient_list() if not recipients: logger.warning("Daily brief: BRIEF_RECIPIENTS not configured, skipping send") return html = build_brief_html(items) path, payload = _build_sendmail_payload( recipients=recipients, subject=f"RCM Email Automation — Daily Brief {datetime.now(timezone.utc).strftime('%Y-%m-%d')}", body_html=html, mailbox_user=self._settings.mailbox_user, ) try: await self._manager._client.post(path, payload) ids = [item["id"] for item in items] async with get_db(db_path) as conn: await mark_brief_items_sent(conn, ids) logger.info("Daily brief sent to %d recipient(s), %d item(s)", len(recipients), len(items)) except Exception: tb = traceback.format_exc() logger.exception("Daily brief send failed; items will retry tomorrow") await send_developer_alert( self._settings, subject="[RCM] Daily brief send failed", body=( f"The daily brief could not be sent. " f"{len(items)} item(s) remain pending and will retry tomorrow.\n\n" f"{tb}" ), ) async def _run_retention(self) -> None: db_path = self._settings.database_path try: async with get_db(db_path) as conn: await run_retention_cleanup(conn) logger.info("Retention cleanup complete") except Exception: tb = traceback.format_exc() logger.exception("Retention cleanup failed") await send_developer_alert( self._settings, subject="[RCM] Retention cleanup failed", body=f"The scheduled retention cleanup job raised an exception.\n\n{tb}", ) try: _cleanup_log_files(self._settings.log_path) except Exception: logger.exception("Log file cleanup failed") def shutdown(self) -> None: if self._scheduler.running: self._scheduler.shutdown(wait=False) def _cleanup_log_files(log_path: Path) -> None: cutoff = datetime.now(timezone.utc).replace(tzinfo=None) for log_file in log_path.glob("processor.log.*"): suffix = log_file.suffix.lstrip(".") try: file_date = datetime.strptime(suffix, "%Y-%m-%d") age_days = (cutoff - file_date).days if age_days > 365: log_file.unlink() logger.info("Deleted old log file %s", log_file) except ValueError: logger.debug("Skipping log file with unrecognised date suffix: %s", log_file)