FabledScribe/src/fabledassistant/services/briefing_scheduler.py

"""
APScheduler-based briefing scheduler — per-user, timezone-aware.

Each enabled user gets 4 individual CronTrigger jobs keyed to their IANA
timezone (stored in briefing_config.timezone).  Changing the config via the
settings UI calls update_user_schedule() which live-patches the scheduler
without a restart.

Uses a background thread scheduler (not async) because APScheduler 3.x's
AsyncIOScheduler has known issues with Quart/hypercorn.  Jobs are async
functions wrapped with asyncio.run_coroutine_threadsafe().
"""

import asyncio
import logging
from datetime import date, datetime, time, timedelta
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from sqlalchemy import select

from fabledassistant.models import async_session
from fabledassistant.models.setting import Setting

logger = logging.getLogger(__name__)

_scheduler: BackgroundScheduler | None = None
_loop: asyncio.AbstractEventLoop | None = None

# Slot definitions: (name, hour, minute) — local time in the user's timezone
SLOTS = [
    ("compilation", 4, 0),
    ("morning", 8, 0),
    ("midday", 12, 0),
    ("afternoon", 16, 0),
]


# ── Helpers ───────────────────────────────────────────────────────────────────

def _resolve_timezone(tz_str: str) -> str:
    """Validate and return an IANA timezone string, falling back to UTC."""
    if not tz_str:
        return "UTC"
    try:
        ZoneInfo(tz_str)
        return tz_str
    except (ZoneInfoNotFoundError, KeyError):
        logger.warning("Invalid timezone %r in briefing config, falling back to UTC", tz_str)
        return "UTC"


async def _get_briefing_enabled_users() -> list[tuple[int, str]]:
    """Return [(user_id, iana_timezone)] for all users with briefing enabled."""
    import json
    async with async_session() as session:
        result = await session.execute(
            select(Setting).where(Setting.key.in_(["briefing_config", "user_timezone"]))
        )
        rows = list(result.scalars().all())

    by_user: dict[int, dict[str, str]] = {}
    for row in rows:
        by_user.setdefault(row.user_id, {})[row.key] = row.value or ""

    enabled = []
    for user_id, settings in by_user.items():
        try:
            config = json.loads(settings.get("briefing_config", "{}") or "{}")
            if config.get("enabled"):
                tz_str = settings.get("user_timezone") or config.get("timezone", "UTC")
                tz = _resolve_timezone(tz_str)
                enabled.append((user_id, tz))
        except Exception:
            pass
    return enabled


def _job_id(user_id: int, slot: str) -> str:
    return f"briefing_{slot}_user_{user_id}"


def _add_user_jobs(user_id: int, tz: str) -> None:
    """Add (or replace) all 4 slot jobs for a user in their timezone."""
    if _scheduler is None or _loop is None:
        return
    for slot_name, hour, minute in SLOTS:
        _scheduler.add_job(
            _run_user_slot_sync,
            CronTrigger(hour=hour, minute=minute, timezone=tz),
            args=[user_id, slot_name],
            id=_job_id(user_id, slot_name),
            replace_existing=True,
            misfire_grace_time=3600,
        )
    logger.info("Scheduled briefing jobs for user %d in timezone %s", user_id, tz)


def _remove_user_jobs(user_id: int) -> None:
    """Remove all slot jobs for a user."""
    if _scheduler is None:
        return
    for slot_name, _, _ in SLOTS:
        jid = _job_id(user_id, slot_name)
        if _scheduler.get_job(jid):
            _scheduler.remove_job(jid)
    logger.info("Removed briefing jobs for user %d", user_id)


# ── Public API ────────────────────────────────────────────────────────────────

def update_user_schedule(user_id: int, config: dict, tz_override: str | None = None) -> None:
    """
    Called when a user saves their briefing config via the settings UI.
    Live-patches the scheduler — no restart required.
    tz_override takes priority over any timezone in config.
    """
    if config.get("enabled"):
        tz_str = tz_override or config.get("timezone", "UTC")
        tz = _resolve_timezone(tz_str)
        _add_user_jobs(user_id, tz)
    else:
        _remove_user_jobs(user_id)


# ── Job execution ─────────────────────────────────────────────────────────────

async def _run_slot_for_user(user_id: int, slot: str) -> None:
    """Execute one slot job for one user."""
    from fabledassistant.services.briefing_conversations import (
        get_or_create_today_conversation, post_message
    )
    from fabledassistant.services.briefing_pipeline import run_compilation, run_slot_injection
    from fabledassistant.services.settings import get_setting
    from fabledassistant.config import Config

    model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL)

    if slot == "compilation":
        # Refresh external data first
        try:
            import json
            from fabledassistant.services.rss import refresh_all_feeds
            config_raw = await get_setting(user_id, "briefing_config", "{}")
            config = json.loads(config_raw) if isinstance(config_raw, str) else {}
            await refresh_all_feeds(user_id)
            from fabledassistant.services import weather as wx
            for key, loc in config.get("locations", {}).items():
                if loc.get("lat") and loc.get("lon"):
                    await wx.refresh_location_cache(
                        user_id=user_id,
                        location_key=key,
                        location_label=loc.get("label", key),
                        lat=loc["lat"],
                        lon=loc["lon"],
                    )
        except Exception:
            logger.warning("Pre-compilation refresh failed for user %d", user_id, exc_info=True)

        await _run_profile_closeout(user_id, model)
        conv = await get_or_create_today_conversation(user_id, model)
        text, metadata = await run_compilation(user_id, slot, model)
        if text:
            await post_message(conv.id, "assistant", text, metadata=metadata)

    else:
        conv = await get_or_create_today_conversation(user_id, model)
        text = await run_slot_injection(user_id, slot, model)
        if text:
            await post_message(conv.id, "user", f"[{slot.title()} briefing update]")
            await post_message(conv.id, "assistant", text)

    try:
        from fabledassistant.services.push import send_push_notification
        slot_labels = {
            "compilation": "Morning briefing ready",
            "morning": "Office briefing ready",
            "midday": "Midday check-in",
            "afternoon": "End of day wrap-up",
        }
        await send_push_notification(
            user_id=user_id,
            title="Briefing",
            body=slot_labels.get(slot, "Briefing update"),
            url="/briefing",
        )
    except Exception:
        logger.debug("Push notification failed for briefing slot %s", slot, exc_info=True)

    logger.info("Briefing slot '%s' completed for user %d", slot, user_id)


def _run_user_slot_sync(user_id: int, slot: str) -> None:
    """Synchronous wrapper called by APScheduler's background thread."""
    if _loop is None:
        logger.error("No event loop available for briefing slot %s user %d", slot, user_id)
        return
    future = asyncio.run_coroutine_threadsafe(_run_slot_for_user(user_id, slot), _loop)
    try:
        future.result(timeout=600)
    except Exception:
        logger.exception("Briefing slot '%s' failed for user %d", slot, user_id)


async def _run_profile_closeout(user_id: int, model: str) -> None:
    """
    Read yesterday's briefing conversation, extract preference observations,
    and append them to the briefing profile note.
    """
    from fabledassistant.services.user_profile import append_observations
    from fabledassistant.services.briefing_pipeline import _llm_synthesise
    from fabledassistant.models.conversation import Conversation, Message

    yesterday = (date.today() - timedelta(days=1))
    async with async_session() as session:
        result = await session.execute(
            select(Conversation).where(
                Conversation.user_id == user_id,
                Conversation.conversation_type == "briefing",
                Conversation.briefing_date == yesterday,
            )
        )
        conv = result.scalars().first()
        if not conv:
            return
        msgs_result = await session.execute(
            select(Message).where(Message.conversation_id == conv.id).order_by(Message.created_at)
        )
        messages = list(msgs_result.scalars().all())

    if len(messages) < 2:
        return

    transcript = "\n".join(
        f"{m.role.upper()}: {m.content[:500]}" for m in messages[-20:]
    )
    system = (
        "You are reviewing a day's briefing conversation to extract preference observations. "
        "Identify any patterns, preferences, or schedule facts the user revealed. "
        "Write 2-5 short bullet points. Be specific and factual. "
        "Example: '- User skipped news about finance', '- Prefers weather for home location first'. "
        "If nothing notable, output only: (nothing to note)"
    )
    observations = await _llm_synthesise(system, transcript, model)
    if observations and "(nothing to note)" not in observations.lower():
        await append_observations(user_id, observations)


# ── Startup / catchup ─────────────────────────────────────────────────────────

async def _catchup_missed_slots(loop: asyncio.AbstractEventLoop) -> None:
    """
    On startup, fire any slot that was missed in the last 24 hours
    (one catch-up per slot per user, evaluated in the user's local timezone).
    """
    users = await _get_briefing_enabled_users()
    for user_id, tz in users:
        user_tz = ZoneInfo(tz)
        now_local = datetime.now(user_tz)
        today_local = now_local.date()

        for slot_name, hour, minute in SLOTS:
            slot_local = datetime.combine(today_local, time(hour, minute), tzinfo=user_tz)
            if slot_local > now_local:
                continue  # Not yet due
            age = (now_local - slot_local).total_seconds()
            if age > 86400:
                continue  # More than 24h ago — skip

            # Check if today's conversation already has a message from after slot time
            async with async_session() as session:
                from fabledassistant.models.conversation import Conversation, Message
                result = await session.execute(
                    select(Conversation).where(
                        Conversation.user_id == user_id,
                        Conversation.conversation_type == "briefing",
                        Conversation.briefing_date == today_local,
                    )
                )
                conv = result.scalars().first()
                if conv:
                    # Convert slot_local to UTC for DB comparison (stored as UTC)
                    slot_utc = slot_local.astimezone(ZoneInfo("UTC"))
                    msgs = await session.execute(
                        select(Message).where(
                            Message.conversation_id == conv.id,
                            Message.created_at >= slot_utc,
                        ).limit(1)
                    )
                    if msgs.scalars().first():
                        continue  # Already covered

            logger.info(
                "Catching up missed briefing slot '%s' for user %d (tz: %s)",
                slot_name, user_id, tz,
            )
            try:
                await _run_slot_for_user(user_id, slot_name)
            except Exception:
                logger.exception(
                    "Catch-up for slot '%s' user %d failed", slot_name, user_id
                )


async def start_briefing_scheduler(loop: asyncio.AbstractEventLoop) -> None:
    """
    Start the APScheduler background scheduler with per-user timezone-aware jobs.
    Must be awaited from the app's before_serving hook (async context).
    """
    global _scheduler, _loop
    if _scheduler is not None:
        return

    _loop = loop
    _scheduler = BackgroundScheduler(timezone="UTC")

    # Await directly — we're already on the event loop, so run_coroutine_threadsafe
    # would deadlock (it blocks the calling thread, which IS the event loop thread).
    try:
        users = await _get_briefing_enabled_users()
    except Exception:
        logger.exception("Failed to load briefing users at startup")
        users = []

    for user_id, tz in users:
        _add_user_jobs(user_id, tz)

    from fabledassistant.services.recurrence import spawn_recurring_tasks as _spawn_recurring

    def _run_recurrence_spawn() -> None:
        future = asyncio.run_coroutine_threadsafe(_spawn_recurring(), _loop)
        try:
            count = future.result(timeout=300)
            logger.info("Recurrence spawn: %d task(s) created", count)
        except Exception as exc:
            logger.error("Recurrence spawn failed: %s", exc)

    _scheduler.add_job(
        _run_recurrence_spawn,
        CronTrigger(hour=0, minute=0, timezone="UTC"),
        id="recurrence_daily",
        replace_existing=True,
    )

    def _run_kokoro_update_check() -> None:
        from fabledassistant.services.tts import check_for_kokoro_updates
        future = asyncio.run_coroutine_threadsafe(check_for_kokoro_updates(), _loop)
        try:
            future.result(timeout=300)
        except Exception as exc:
            logger.error("Kokoro update check failed: %s", exc)

    _scheduler.add_job(
        _run_kokoro_update_check,
        CronTrigger(hour=3, minute=0, timezone="UTC"),
        id="kokoro_update_check_daily",
        replace_existing=True,
    )

    _scheduler.start()
    logger.info(
        "Briefing scheduler started with %d user(s) across %d job(s)",
        len(users), len(users) * len(SLOTS),
    )

    asyncio.create_task(_catchup_missed_slots(loop))


def stop_briefing_scheduler() -> None:
    global _scheduler, _loop
    if _scheduler:
        _scheduler.shutdown(wait=False)
        _scheduler = None
    _loop = None