espresso_frame/server/app/main.py

"""ESPresso Frame server: pulls photos from Immich, pre-processes them
for the panel, and serves ESP32 frames ready-to-display images.

This module is assembly only -- routes live in app/routers/:
  device.py       the firmware-facing /frame/* protocol (paths frozen)
  api_frames.py   the web UI's JSON API, /api/frames/{id}/...
  api_widgets.py  widget CRUD + grid placement, /api/frames/{id}/widgets
  api_layouts.py  named, user-owned saved layouts, /api/layouts,
                  /api/frames/{id}/layouts
  frame_pages.py  the per-frame Photos/Configuration/Layout/Stats pages
  pages.py        setup/login/claim/settings/admin
  manage.py       the limited manage-QR surface (/m/, /api/m/)
Storage is SQLite via models.py/db.py; migration.py imports a
pre-database config.json deployment on first boot."""

from __future__ import annotations

import logging
import time
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from sqlalchemy import select

from . import html_render, logging_setup, migration
from .auth import (
    browser_token_valid,
    current_user,
    management_token,
    user_frames,
    users_exist,
)
from .db import SessionLocal
from .models import Frame
from .routers import api_frames, api_layouts, api_widgets, device, frame_pages, manage, pages
from .routers.common import shell_context

logger = logging.getLogger(__name__)

# Before anything else logs: a handler exists to catch it, and it lands in
# the same persistent volume the admin log viewer reads from.
logging_setup.configure_logging()

# Schema + legacy-config import, before the first request is served.
migration.run_migrations()

@asynccontextmanager
async def _lifespan(app: FastAPI):
    """Startup does nothing browser-related -- html_render.start() is
    lazy (only the weather widget's opt-in "modern" render style ever
    triggers it, see that module's docstring), so a deployment that
    never uses it never launches Chromium or needs Playwright's browser
    binaries installed. Shutdown calls html_render.stop() unconditionally
    (a no-op if it was never started) so a server restart never leaves
    an orphaned Chromium process running."""
    yield
    html_render.stop()


app = FastAPI(title="ESPresso Frame Server", lifespan=_lifespan)
templates = Jinja2Templates(directory="app/templates")


@app.middleware("http")
async def log_device_requests(request: Request, call_next):
    """Access log for the firmware-facing /frame/* protocol -- the admin
    log viewer otherwise only ever shows exceptions (device.py logs
    those, not successful requests), so a slow-but-200 request or a
    device hammering a stale/wrong token leaves no trace at all. Logs
    the device id (query param, not the token -- never log credentials)
    and wall time, which is exactly what's needed to spot a request that
    blew past the firmware's fixed HTTP timeout without technically
    failing server-side."""
    if not request.url.path.startswith("/frame/"):
        return await call_next(request)
    start = time.monotonic()
    device_id = request.query_params.get("id", "") or "-"
    response = await call_next(request)
    elapsed_ms = (time.monotonic() - start) * 1000
    logger.info(
        "%s %s id=%s -> %d (%.0fms)",
        request.method, request.url.path, device_id, response.status_code, elapsed_ms,
    )
    return response


app.mount("/static", StaticFiles(directory="app/static"), name="static")

app.include_router(device.router)
app.include_router(api_frames.router)
app.include_router(api_widgets.router)
app.include_router(api_layouts.router)
app.include_router(frame_pages.router)
app.include_router(pages.router)
app.include_router(manage.router)


@app.get("/health")
def health() -> dict:
    return {"status": "ok"}


@app.get("/sw.js")
def service_worker() -> FileResponse:
    # Served from / rather than /static/sw.js so its default scope is the
    # whole app -- a SW can only ever control paths at or below its own URL.
    return FileResponse("app/static/sw.js", media_type="application/javascript")


def _device_credential_redirect(request: Request, db, allow_legacy: bool) -> str | None:
    """The on-frame manage QR points at the server root with the device's
    own credentials (new firmware: ?id=&token=; deployed firmware:
    ?token=<legacy shared token>). Those scans get the frame's limited
    manage page -- never the full UI, which requires a login.
    allow_legacy is False before /setup has run: at that point a bare
    ?token= hit is the admin coming through the token prompt to do
    first-run setup, not a QR scan."""
    device_id = request.query_params.get("id", "").strip().lower()
    token = request.query_params.get("token", "")
    if device_id and token:
        frame = db.scalars(select(Frame).where(Frame.device_id == device_id)).first()
        if frame is not None and token == frame.device_token:
            return f"/m/{frame.manage_token}"
    if allow_legacy and token and management_token() and token == management_token():
        frame = db.scalars(
            select(Frame).where(Frame.legacy_token_enabled == True)  # noqa: E712
        ).first()
        if frame is not None:
            return f"/m/{frame.manage_token}"
    return None


@app.get("/", response_class=HTMLResponse)
def index(request: Request):
    """Routing hub: manage-QR scans go to the limited manage page, users
    land on their first frame (or an empty-state page), and everyone
    else is walked through setup/login."""
    with SessionLocal() as db:
        have_users = users_exist(db)
        manage_redirect = _device_credential_redirect(request, db, allow_legacy=have_users)
        if manage_redirect is not None:
            return RedirectResponse(manage_redirect, status_code=303)

        user = current_user(request, db)
        if user is None:
            if not have_users:
                if management_token() and not browser_token_valid(request):
                    supplied = request.query_params.get("token")
                    return templates.TemplateResponse(
                        "token_prompt.html", {"request": request, "wrong": supplied is not None}
                    )
                # Pre-setup: reachable (optionally token-gated), nudge setup.
                return RedirectResponse("/setup", status_code=303)
            return RedirectResponse("/login", status_code=303)

        frames = user_frames(db, user)
        if frames:
            return RedirectResponse(f"/frames/{frames[0].id}", status_code=303)
        return templates.TemplateResponse("frames_empty.html", shell_context(request, db, user))