gogo-telefon/gogo/jobs.py

"""Background jobs: proposal expiry/reminders, partner polling fallback, retention.

APScheduler (in-process) — no extra infra needed for MVP scale (§4).
"""

from __future__ import annotations

import logging

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from sqlalchemy import select

from gogo.db import get_sessionmaker
from gogo.domain import BookingStatus, Slot
from gogo.models import BookingRequest, Tenant
from gogo.proposals import engine as proposal_engine
from gogo.scheduling.base import get_provider

log = logging.getLogger("gogo.jobs")


async def run_expiry_job() -> None:
    async with get_sessionmaker()() as session:
        expired = await proposal_engine.process_expirations(session)
        await session.commit()
        if expired:
            log.info("expired %d proposals", expired)


async def run_partner_polling_job() -> None:
    """Polling fallback (§B.3) for partner tenants that cannot call webhooks."""
    async with get_sessionmaker()() as session:
        pending = (
            (
                await session.execute(
                    select(BookingRequest).where(
                        BookingRequest.status == BookingStatus.pending.value,
                        BookingRequest.partner_request_id.is_not(None),
                    )
                )
            )
            .scalars()
            .all()
        )
        for req in pending:
            tenant = (
                await session.execute(select(Tenant).where(Tenant.id == req.tenant_id))
            ).scalar_one()
            if tenant.scheduling_provider != "partner_api":
                continue
            provider = await get_provider(session, tenant)
            if not provider.config.get("polling_fallback"):
                continue
            try:
                data = await provider.poll_status(req.partner_request_id)
            except Exception:  # noqa: BLE001
                log.exception("polling failed for request %s", req.id)
                continue
            status = data.get("status")
            if status == "confirmed" and data.get("confirmed_slot"):
                slot = Slot.model_validate(data["confirmed_slot"])
                await proposal_engine.confirm_request(session, tenant, req, slot, recheck=False)
            elif status == "resolved":
                await proposal_engine.resolve_request(session, tenant, req)
            elif status == "rejected":
                await proposal_engine.reject_request(session, tenant, req)
        await session.commit()


async def run_retention_job() -> None:
    """Delete call audio older than retention window; keep transcripts (§5.4)."""
    import os
    from datetime import UTC, datetime, timedelta

    from gogo.config import get_settings
    from gogo.models import Call

    cutoff = datetime.now(UTC) - timedelta(days=get_settings().audio_retention_days)
    async with get_sessionmaker()() as session:
        rows = (
            (
                await session.execute(
                    select(Call).where(
                        Call.recording_path.is_not(None), Call.started_at < cutoff
                    )
                )
            )
            .scalars()
            .all()
        )
        for call in rows:
            path = call.recording_path
            if path and os.path.exists(path):
                try:
                    os.remove(path)
                except OSError:
                    log.exception("failed to delete recording %s", path)
                    continue
            call.recording_path = None

        # drop stale dialplan handoff rows (they are only needed for minutes)
        from sqlalchemy import delete

        from gogo.models import CallRegistration

        reg_cutoff = datetime.now(UTC) - timedelta(days=7)
        await session.execute(
            delete(CallRegistration).where(CallRegistration.created_at < reg_cutoff)
        )
        await session.commit()
        if rows:
            log.info("retention: cleared %d recordings", len(rows))


def build_scheduler() -> AsyncIOScheduler:
    scheduler = AsyncIOScheduler()
    scheduler.add_job(run_expiry_job, "interval", minutes=5, id="proposal_expiry")
    scheduler.add_job(run_partner_polling_job, "interval", minutes=5, id="partner_polling")
    scheduler.add_job(run_retention_job, "cron", hour=4, minute=0, id="retention")
    return scheduler