inkwell/src/thoughtsync/app.py

from __future__ import annotations

import asyncio
import mimetypes
import os
import secrets
from contextlib import suppress
from datetime import timedelta

from quart import Quart, has_request_context, jsonify, request, send_from_directory
from quart.sessions import SecureCookieSessionInterface

from . import __version__
from .auth import bp as auth_bp
from .client_dist import advertisement as client_advertisement, bp as client_bp
from .config import Config
from .db import session_scope
from .graph import bp as graph_bp
from .labels import bp as labels_bp
from .notes import bp as notes_bp
from .retention import run_sweeper
from .saved_filters import bp as saved_filters_bp
from .settings import get_public_config, get_setting, load_or_create_secret_key
from .settings_api import bp as settings_bp
from .sync import bp as sync_bp, protocol_advertisement

STATIC_DIR = os.path.join(os.path.dirname(__file__), "static")

# `.webmanifest` isn't in every base image's mime map; register it so the PWA
# manifest is served as application/manifest+json instead of octet-stream.
mimetypes.add_type("application/manifest+json", ".webmanifest")


class _AutoSecureSessionInterface(SecureCookieSessionInterface):
    """Mark the session cookie `Secure` whenever the request arrived over HTTPS —
    directly, or via a TLS-terminating reverse proxy that sets X-Forwarded-Proto.

    Auto-detecting per request (rather than a fixed SESSION_COOKIE_SECURE flag)
    hardens the cookie on HTTPS deployments without breaking a plain-HTTP install on
    a trusted LAN, where a hard-forced Secure flag would stop the browser from ever
    sending the cookie back — i.e. silently break login. No configuration required.
    """

    def get_cookie_secure(self, app: Quart) -> bool:
        if not has_request_context():
            return False
        forwarded = request.headers.get("X-Forwarded-Proto", "").split(",")[0].strip().lower()
        return forwarded == "https" or request.is_secure


def create_app() -> Quart:
    # static_folder=None: the SPA catch-all below owns static serving.
    app = Quart(__name__, static_folder=None)
    # Ephemeral/env secret so the app (and DB-free unit tests) construct without a
    # database. before_serving swaps in the real, DB-persisted key before serving.
    app.secret_key = Config.secret_key_env() or secrets.token_urlsafe(48)
    app.config["APP_VERSION"] = os.environ.get("APP_VERSION", __version__)
    app.config["SESSION_COOKIE_HTTPONLY"] = True
    app.config["SESSION_COOKIE_SAMESITE"] = "Lax"
    # Auto-mark the session cookie Secure on HTTPS requests (see the interface above).
    app.session_interface = _AutoSecureSessionInterface()
    app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(days=30)
    # Hard request-body ceiling (any-file attachments, import zips, sync push). The
    # per-file attachment limit is the DB-backed `max_attachment_mb` setting, enforced
    # in the upload handler; this must stay >= the largest value that allows.
    app.config["MAX_CONTENT_LENGTH"] = 64 * 1024 * 1024

    app.register_blueprint(auth_bp)
    app.register_blueprint(notes_bp)
    app.register_blueprint(labels_bp)
    app.register_blueprint(graph_bp)
    app.register_blueprint(settings_bp)
    app.register_blueprint(sync_bp)
    app.register_blueprint(saved_filters_bp)
    app.register_blueprint(client_bp)

    @app.before_serving
    async def _bootstrap() -> None:
        # Load (or generate + persist) the real signing secret and the live session
        # lifetime from the DB, before any request is served.
        async with session_scope() as db:
            app.secret_key = await load_or_create_secret_key(db)
            try:
                days = int(await get_setting(db, "session_ttl_days"))
                app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(days=days)
            except (ValueError, TypeError, KeyError):
                pass
        # Expire old trash in the background (retention.py). One task per process is
        # correct because the image serves with a single hypercorn worker (Dockerfile);
        # if that ever gains `--workers`, this needs a lock so N workers don't each
        # sweep. Duplicate sweeps would be harmless but wasteful — a purged row is
        # skipped by `purged_at IS NULL` — so this is about load, not correctness.
        app.config["TRASH_SWEEPER"] = asyncio.create_task(run_sweeper())

    @app.after_serving
    async def _shutdown() -> None:
        task = app.config.get("TRASH_SWEEPER")
        if task is not None:
            task.cancel()
            # Await the cancellation so shutdown doesn't race a sweep mid-transaction.
            with suppress(asyncio.CancelledError):
                await task

    @app.get("/api/health")
    async def health():
        return jsonify({"status": "ok", "version": app.config["APP_VERSION"]})

    @app.get("/api/config")
    async def public_config():
        # Public: the login/register screen reads site name + whether signups are open.
        async with session_scope() as db:
            data = await get_public_config(db)
        data["version"] = app.config["APP_VERSION"]
        # The sync-protocol handshake (M10.6). A native client reads this BEFORE
        # linking — while it still has no token and possibly no account — to decide
        # whether it can talk to this server, and which optional features to offer.
        data.update(protocol_advertisement())
        # Which Android client this server can hand out, if any. Absent rather than
        # null when it has none, so the web UI hides the download instead of
        # offering a button that 404s.
        data.update(client_advertisement())
        return jsonify(data)

    @app.get("/", defaults={"path": ""})
    @app.get("/<path:path>")
    async def spa(path: str):
        if path.startswith("api/"):
            return jsonify({"error": "not found"}), 404
        candidate = os.path.join(STATIC_DIR, path)
        if path and os.path.isfile(candidate):
            return await send_from_directory(STATIC_DIR, path)
        index = os.path.join(STATIC_DIR, "index.html")
        if os.path.isfile(index):
            return await send_from_directory(STATIC_DIR, "index.html")
        return jsonify({"error": "frontend not built"}), 404

    return app