thermograph/backend/accounts/api_accounts.py

"""Authenticated API for subscriptions and notifications.

All routes require a logged-in user (the fastapi-users cookie session) and are
scoped to that user — a row that isn't theirs reads as 404, never 403, so the API
doesn't leak which ids exist. Mounted under {BASE}/api/v2 alongside the rest of v2.
"""
import os
import time

from fastapi import APIRouter, Depends, HTTPException, Query, Request
from fastapi.concurrency import run_in_threadpool
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from core import audit
from data import climate
from data import grid
from notifications import push
from accounts.db import get_async_session, get_read_session
from accounts.models import Notification, PushSubscription, Subscription
from accounts.schemas import (
    NotificationList,
    NotificationOut,
    PushSubscriptionIn,
    PushUnsubscribeIn,
    SubscriptionIn,
    SubscriptionOut,
    SubscriptionPatch,
)
from accounts.users import current_active_user

_BASE = os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
BASE = f"/{_BASE}" if _BASE else ""

router = APIRouter(tags=["accounts"])


async def _owned_subscription(session: AsyncSession, sub_id: int, user) -> Subscription:
    sub = (
        await session.execute(
            select(Subscription).where(
                Subscription.id == sub_id, Subscription.user_id == user.id
            )
        )
    ).scalar_one_or_none()
    if sub is None:
        raise HTTPException(status_code=404, detail="Subscription not found.")
    return sub


# --- subscriptions -----------------------------------------------------------
@router.get("/subscriptions", response_model=list[SubscriptionOut])
async def list_subscriptions(
    user=Depends(current_active_user),
    session: AsyncSession = Depends(get_read_session),   # pure read → read-only engine
):
    rows = (
        await session.execute(
            select(Subscription)
            .where(Subscription.user_id == user.id)
            .order_by(Subscription.created_at)
        )
    ).scalars().all()
    return rows


@router.post("/subscriptions", response_model=SubscriptionOut, status_code=201)
async def create_subscription(
    body: SubscriptionIn,
    user=Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session),
):
    cell = grid.snap(body.lat, body.lon)
    cell_id = cell["id"]

    dup = (
        await session.execute(
            select(Subscription).where(
                Subscription.user_id == user.id,
                Subscription.cell_id == cell_id,
                Subscription.kind == body.kind,
            )
        )
    ).scalar_one_or_none()
    if dup is not None:
        raise HTTPException(
            status_code=409,
            detail=f"You already have a {body.kind} alert for this location.",
        )

    label = body.label
    if not label:
        # Cache-only lookup — never block the event loop on Nominatim. The frontend
        # normally supplies the label from /api/v2/place; this is just a fallback.
        # Still a sync DB read under the hood, so keep it off the event loop too.
        _, label = await run_in_threadpool(
            climate.reverse_geocode_cached, cell["center_lat"], cell["center_lon"]
        )

    sub = Subscription(
        user_id=user.id,
        cell_id=cell_id,
        label=label,
        lat=body.lat,
        lon=body.lon,
        threshold=body.threshold,
        metrics=body.metrics,
        kind=body.kind,
        two_sided=body.two_sided,
    )
    session.add(sub)
    await session.commit()
    await session.refresh(sub)
    audit.log_activity("sub.create", {"user_id": str(user.id), "sub_id": sub.id,
                                      "cell_id": cell_id, "kind": body.kind})
    return sub


@router.patch("/subscriptions/{sub_id}", response_model=SubscriptionOut)
async def update_subscription(
    sub_id: int,
    body: SubscriptionPatch,
    user=Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session),
):
    sub = await _owned_subscription(session, sub_id, user)
    changed = body.model_dump(exclude_unset=True)
    for key, value in changed.items():
        setattr(sub, key, value)
    await session.commit()
    await session.refresh(sub)
    audit.log_activity("sub.update", {"user_id": str(user.id), "sub_id": sub.id,
                                      "changed": list(changed.keys())})
    return sub


@router.delete("/subscriptions/{sub_id}", status_code=204)
async def delete_subscription(
    sub_id: int,
    user=Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session),
):
    sub = await _owned_subscription(session, sub_id, user)
    kind = sub.kind
    await session.delete(sub)
    await session.commit()
    audit.log_activity("sub.delete", {"user_id": str(user.id), "sub_id": sub_id, "kind": kind})


# --- notifications -----------------------------------------------------------
@router.get("/notifications", response_model=NotificationList)
async def list_notifications(
    unread: bool = Query(False, description="only return unread notifications"),
    limit: int = Query(50, ge=1, le=200),
    user=Depends(current_active_user),
    session: AsyncSession = Depends(get_read_session),   # pure read → read-only engine
):
    q = select(Notification).where(Notification.user_id == user.id)
    if unread:
        q = q.where(Notification.read_at.is_(None))
    q = q.order_by(Notification.created_at.desc()).limit(limit)
    rows = (await session.execute(q)).scalars().all()

    unread_count = (
        await session.execute(
            select(func.count())
            .select_from(Notification)
            .where(Notification.user_id == user.id, Notification.read_at.is_(None))
        )
    ).scalar_one()
    return NotificationList(
        notifications=[NotificationOut.model_validate(r) for r in rows],
        unread_count=unread_count,
    )


@router.post("/notifications/{notif_id}/read", status_code=204)
async def mark_notification_read(
    notif_id: int,
    user=Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session),
):
    import time

    notif = (
        await session.execute(
            select(Notification).where(
                Notification.id == notif_id, Notification.user_id == user.id
            )
        )
    ).scalar_one_or_none()
    if notif is None:
        raise HTTPException(status_code=404, detail="Notification not found.")
    if notif.read_at is None:
        notif.read_at = time.time()
        await session.commit()


@router.post("/notifications/read-all", status_code=204)
async def mark_all_read(
    user=Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session),
):
    rows = (
        await session.execute(
            select(Notification).where(
                Notification.user_id == user.id, Notification.read_at.is_(None)
            )
        )
    ).scalars().all()
    now = time.time()
    for n in rows:
        n.read_at = now
    if rows:
        await session.commit()


# --- web push -----------------------------------------------------------------
# The VAPID public key is not a secret and the browser needs it before it can
# subscribe, so this one route is open (still under BASE). Everything else is
# user-scoped like the subscription routes above.
@router.get("/push/vapid-key")
async def push_vapid_key():
    return {"key": push.public_key()}


@router.post("/push/subscribe", status_code=201)
async def push_subscribe(
    body: PushSubscriptionIn,
    request: Request,
    user=Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session),
):
    """Register (or refresh) this device's Web Push endpoint for the user.

    Keyed on `endpoint` — re-subscribing from the same device updates the keys in
    place rather than creating a duplicate, and a device that was registered to a
    different account is re-homed to the current user.
    """
    now = time.time()
    existing = (
        await session.execute(
            select(PushSubscription).where(PushSubscription.endpoint == body.endpoint)
        )
    ).scalar_one_or_none()
    if existing is not None:
        existing.user_id = user.id
        existing.p256dh = body.keys.p256dh
        existing.auth = body.keys.auth
        existing.user_agent = request.headers.get("user-agent")
        existing.last_used_at = now
    else:
        session.add(
            PushSubscription(
                user_id=user.id,
                endpoint=body.endpoint,
                p256dh=body.keys.p256dh,
                auth=body.keys.auth,
                user_agent=request.headers.get("user-agent"),
                created_at=now,
                last_used_at=now,
            )
        )
    await session.commit()
    audit.log_activity("push.subscribe", {"user_id": str(user.id),
                                          "refreshed": existing is not None})
    return {"ok": True}


@router.delete("/push/subscribe", status_code=204)
async def push_unsubscribe(
    body: PushUnsubscribeIn,
    user=Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session),
):
    """Drop this device's endpoint (called on toggle-off / permission revoked)."""
    row = (
        await session.execute(
            select(PushSubscription).where(
                PushSubscription.endpoint == body.endpoint,
                PushSubscription.user_id == user.id,
            )
        )
    ).scalar_one_or_none()
    if row is not None:
        await session.delete(row)
        await session.commit()
        audit.log_activity("push.unsubscribe", {"user_id": str(user.id)})


@router.post("/push/test", status_code=202)
async def push_test(
    user=Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session),
):
    """Send a canned push to every device registered to the user — the on-device
    delivery check. Prunes any endpoint the push service reports as gone."""
    rows = (
        await session.execute(
            select(PushSubscription).where(PushSubscription.user_id == user.id)
        )
    ).scalars().all()
    payload = {
        "title": "Thermograph",
        "body": "Push notifications are working on this device.",
        "url": f"{BASE}/alerts",
        "tag": "thermograph-test",
    }
    sent, pruned, failed = 0, 0, 0
    for row in rows:
        info = {"endpoint": row.endpoint, "keys": {"p256dh": row.p256dh, "auth": row.auth}}
        # pywebpush blocks on network I/O — keep it off the event loop.
        result = await run_in_threadpool(push.send, info, payload)
        if result == "ok":
            sent += 1
        elif result == "gone":
            await session.delete(row)
            pruned += 1
        else:
            failed += 1
    if pruned:
        await session.commit()
    # `failed` (delivery rejected — e.g. VAPID key mismatch) is surfaced so the client
    # can tell "request accepted" apart from "actually delivered".
    return {"devices": len(rows), "sent": sent, "pruned": pruned, "failed": failed}