from __future__ import annotations

import hashlib
import json
import math
from pathlib import Path
import shutil
import uuid

from backend.chatapp import db as db_mod
from backend.chatapp.http import HttpError, Request, Response, Router, json_response
from backend.chatapp.security import (
    PasswordHash,
    clamp_push_preview,
    generate_otp32,
    hash_password,
    new_session_token,
    sha256_hex,
    verify_password,
)
from backend.chatapp.state import AppState
from backend.chatapp.util import now_ts, safe_mkdir


def register_routes(router: Router, state: AppState) -> None:
    # Admin bootstrap + admin APIs
    router.add("POST", "/api/v1/admin/bootstrap", lambda req: _admin_bootstrap(req, state))
    router.add("GET", "/api/v1/admin/users", lambda req: _admin_users_list(req, state))
    router.add("POST", "/api/v1/admin/users/{user_id}/ban", lambda req: _admin_user_ban(req, state))
    router.add("POST", "/api/v1/admin/users/{user_id}/warn", lambda req: _admin_user_warn(req, state))
    router.add(
        "POST",
        "/api/v1/admin/users/{user_id}/otp-reset",
        lambda req: _admin_user_otp_reset(req, state),
    )
    router.add("GET", "/api/v1/admin/groups", lambda req: _admin_groups_list(req, state))
    router.add("GET", "/api/v1/admin/logs", lambda req: _admin_logs(req, state))
    router.add("GET", "/api/v1/admin/metrics", lambda req: _admin_metrics(req, state))
    router.add("POST", "/api/v1/admin/bots", lambda req: _admin_bot_create(req, state))
    router.add("POST", "/api/v1/admin/bots/{bot_id}/verify", lambda req: _admin_bot_verify(req, state))
    router.add("GET", "/api/v1/admin/integrations", lambda req: _admin_integrations_list(req, state))
    router.add("POST", "/api/v1/admin/integrations", lambda req: _admin_integrations_create(req, state))
    router.add(
        "PATCH",
        "/api/v1/admin/integrations/{integration_id}",
        lambda req: _admin_integrations_update(req, state),
    )
    router.add(
        "POST",
        "/api/v1/admin/integrations/{integration_id}/test-fire",
        lambda req: _admin_integrations_test_fire(req, state),
    )

    # Auth + account
    router.add("POST", "/api/v1/auth/register", lambda req: _auth_register(req, state))
    router.add("POST", "/api/v1/auth/login", lambda req: _auth_login(req, state))
    router.add("POST", "/api/v1/auth/logout", lambda req: _auth_logout(req, state))
    router.add("POST", "/api/v1/auth/reset-password/complete", lambda req: _auth_reset_complete(req, state))
    router.add("GET", "/api/v1/me", lambda req: _me(req, state))
    router.add("PATCH", "/api/v1/me/settings", lambda req: _me_settings(req, state))

    # Conversations + messages
    router.add("GET", "/api/v1/conversations", lambda req: _conversations_list(req, state))
    router.add("POST", "/api/v1/conversations/dm", lambda req: _conversations_create_dm(req, state))
    router.add("POST", "/api/v1/conversations/group", lambda req: _conversations_create_group(req, state))
    router.add("GET", "/api/v1/conversations/{conversation_id}", lambda req: _conversation_get(req, state))
    router.add(
        "POST",
        "/api/v1/conversations/{conversation_id}/members",
        lambda req: _conversation_members_add(req, state),
    )
    router.add(
        "DELETE",
        "/api/v1/conversations/{conversation_id}/members/{user_id}",
        lambda req: _conversation_members_remove(req, state),
    )
    router.add(
        "GET",
        "/api/v1/conversations/{conversation_id}/messages",
        lambda req: _messages_list(req, state),
    )
    router.add(
        "POST",
        "/api/v1/conversations/{conversation_id}/messages",
        lambda req: _messages_send(req, state),
    )
    router.add("PATCH", "/api/v1/messages/{message_id}", lambda req: _message_edit(req, state))
    router.add("DELETE", "/api/v1/messages/{message_id}", lambda req: _message_delete(req, state))
    router.add("POST", "/api/v1/messages/{message_id}/reactions", lambda req: _message_react(req, state))

    # Uploads + attachments
    router.add("POST", "/api/v1/uploads", lambda req: _upload_create(req, state))
    router.add("GET", "/api/v1/uploads/{upload_id}", lambda req: _upload_get(req, state))
    router.add(
        "PUT",
        "/api/v1/uploads/{upload_id}/chunks/{chunk_index}",
        lambda req: _upload_chunk_put(req, state),
    )
    router.add("POST", "/api/v1/uploads/{upload_id}/finalize", lambda req: _upload_finalize(req, state))
    router.add("GET", "/api/v1/attachments/{attachment_id}", lambda req: _attachment_get(req, state))

    # Push registration (delivery worker is out-of-scope here; events are queued)
    router.add(
        "POST",
        "/api/v1/push/unifiedpush/register",
        lambda req: _push_unifiedpush_register(req, state),
    )
    router.add(
        "POST",
        "/api/v1/push/webpush/subscribe",
        lambda req: _push_webpush_subscribe(req, state),
    )


def _new_id() -> str:
    return uuid.uuid4().hex


def _require_json_fields(payload: dict, fields: list[str]) -> None:
    for f in fields:
        if f not in payload:
            raise HttpError(400, f"Missing field: {f}", code="missing_field")


def _normalize_login(value: str) -> str:
    return value.strip()


def _token_from_auth(req: Request) -> str:
    auth = req.header("authorization")
    if not auth:
        raise HttpError(401, "Missing Authorization header", code="unauthorized")
    parts = auth.split(" ", 1)
    if len(parts) != 2 or parts[0].lower() != "bearer" or parts[1].strip() == "":
        raise HttpError(401, "Invalid Authorization header", code="unauthorized")
    return parts[1].strip()


def _auth_user(conn, req: Request, state: AppState, *, allow_otp_reset: bool = False) -> dict:
    token = _token_from_auth(req)
    token_hash = sha256_hex(token)
    now = now_ts()
    row = db_mod.fetch_one(
        conn,
        """
        SELECT
          s.id AS session_id,
          s.kind AS session_kind,
          s.expires_at AS session_expires_at,
          u.*
        FROM sessions s
        JOIN users u ON u.id = s.user_id
        WHERE s.token_hash_hex = ?
          AND s.revoked_at IS NULL
          AND s.expires_at > ?
        """,
        (token_hash, now),
    )
    if row is None:
        raise HttpError(401, "Invalid or expired session", code="unauthorized")
    if int(row["is_banned"]) == 1:
        raise HttpError(403, "Account banned", code="banned")
    if row["session_kind"] == "otp_reset" and not allow_otp_reset:
        raise HttpError(423, "Password reset required", code="password_reset_required")

    conn.execute("UPDATE sessions SET last_seen_at = ? WHERE id = ?", (now, row["session_id"]))
    user = {k: row[k] for k in row.keys() if k not in ("password_salt_b64", "password_iterations", "password_hash_b64")}
    req.user = user
    req.session = {"id": row["session_id"], "kind": row["session_kind"]}
    return user


def _require_admin(conn, req: Request, state: AppState) -> dict:
    user = _auth_user(conn, req, state)
    if int(user["is_admin"]) != 1:
        raise HttpError(403, "Admin only", code="forbidden")
    return user


def _log(conn, *, level: str, event: str, details: dict | None = None) -> None:
    conn.execute(
        "INSERT INTO logs (id, level, event, details_json, created_at) VALUES (?,?,?,?,?)",
        (_new_id(), level, event, json.dumps(details) if details is not None else None, now_ts()),
    )


def _inc_metric(conn, key: str, amount: int = 1) -> None:
    now = now_ts()
    conn.execute(
        """
        INSERT INTO metrics_counters (key, value, updated_at) VALUES (?,?,?)
        ON CONFLICT(key) DO UPDATE SET value = value + excluded.value, updated_at = excluded.updated_at
        """,
        (key, int(amount), now),
    )


def _admin_bootstrap(req: Request, state: AppState) -> Response:
    payload = req.json()
    _require_json_fields(payload, ["email", "username", "password"])
    email = str(payload["email"]).strip().lower()
    username = str(payload["username"]).strip()
    password = str(payload["password"])
    if len(password) > 256:
        raise HttpError(400, "Password too long", code="password_too_long")

    conn = db_mod.connect(state.db_path)
    try:
        existing_admin = db_mod.fetch_one(conn, "SELECT 1 FROM users WHERE is_admin = 1 LIMIT 1")
        if existing_admin is not None:
            raise HttpError(403, "Admin already exists", code="bootstrap_closed")

        pw = hash_password(password)
        now = now_ts()
        user_id = _new_id()
        with db_mod.transaction(conn):
            conn.execute(
                """
                INSERT INTO users (
                  id, email, username,
                  password_salt_b64, password_iterations, password_hash_b64,
                  is_admin, created_at
                ) VALUES (?,?,?,?,?,?,?,?)
                """,
                (user_id, email, username, pw.salt_b64, pw.iterations, pw.hash_b64, 1, now),
            )
            conn.execute(
                "INSERT INTO user_settings (user_id, receive_everyone_mentions) VALUES (?,?)",
                (user_id, 0),
            )
            _log(conn, level="INFO", event="admin_bootstrap", details={"user_id": user_id})
    finally:
        conn.close()

    return json_response(201, {"ok": True})


def _admin_users_list(req: Request, state: AppState) -> Response:
    conn = db_mod.connect(state.db_path)
    try:
        _require_admin(conn, req, state)
        rows = db_mod.fetch_all(
            conn,
            """
            SELECT id, email, username, is_admin, is_bot, bot_verified, is_banned, ban_reason, warned_reason, warned_at, created_at
            FROM users
            ORDER BY created_at DESC
            """,
        )
        return json_response(200, {"users": [dict(r) for r in rows]})
    finally:
        conn.close()


def _admin_user_ban(req: Request, state: AppState) -> Response:
    payload = req.json()
    banned = bool(payload.get("banned", True))
    reason = payload.get("reason")
    if reason is not None:
        reason = str(reason)[:500]

    user_id = req.path_params["user_id"]
    conn = db_mod.connect(state.db_path)
    try:
        admin = _require_admin(conn, req, state)
        conn.execute(
            "UPDATE users SET is_banned = ?, ban_reason = ? WHERE id = ?",
            (1 if banned else 0, reason if banned else None, user_id),
        )
        if banned:
            now = now_ts()
            conn.execute("UPDATE sessions SET revoked_at = ? WHERE user_id = ? AND revoked_at IS NULL", (now, user_id))
        _log(
            conn,
            level="WARN" if banned else "INFO",
            event="user_ban" if banned else "user_unban",
            details={"admin_id": admin["id"], "user_id": user_id, "reason": reason},
        )
        return json_response(200, {"ok": True})
    finally:
        conn.close()


def _admin_user_warn(req: Request, state: AppState) -> Response:
    payload = req.json()
    reason = str(payload.get("reason", "")).strip()
    if reason == "":
        raise HttpError(400, "Missing reason", code="missing_field")
    reason = reason[:500]

    user_id = req.path_params["user_id"]
    conn = db_mod.connect(state.db_path)
    try:
        admin = _require_admin(conn, req, state)
        now = now_ts()
        conn.execute(
            "UPDATE users SET warned_reason = ?, warned_at = ? WHERE id = ?",
            (reason, now, user_id),
        )
        _log(conn, level="WARN", event="user_warn", details={"admin_id": admin["id"], "user_id": user_id, "reason": reason})
        return json_response(200, {"ok": True})
    finally:
        conn.close()


def _admin_user_otp_reset(req: Request, state: AppState) -> Response:
    user_id = req.path_params["user_id"]
    conn = db_mod.connect(state.db_path)
    try:
        admin = _require_admin(conn, req, state)
        otp = generate_otp32()
        now = now_ts()
        expires_at = now + int(state.config.otp_ttl_seconds)
        with db_mod.transaction(conn):
            conn.execute("UPDATE users SET password_reset_required = 1 WHERE id = ?", (user_id,))
            conn.execute(
                """
                INSERT INTO password_reset_otps (
                  user_id, otp_hash_hex, expires_at, used_at, created_by_admin_id, created_at
                ) VALUES (?,?,?,?,?,?)
                ON CONFLICT(user_id) DO UPDATE SET
                  otp_hash_hex = excluded.otp_hash_hex,
                  expires_at = excluded.expires_at,
                  used_at = NULL,
                  created_by_admin_id = excluded.created_by_admin_id,
                  created_at = excluded.created_at
                """,
                (user_id, sha256_hex(otp), expires_at, None, admin["id"], now),
            )
            _log(
                conn,
                level="INFO",
                event="otp_reset_created",
                details={"admin_id": admin["id"], "user_id": user_id, "expires_at": expires_at},
            )
        return json_response(200, {"otp": otp, "expires_at": expires_at})
    finally:
        conn.close()


def _admin_groups_list(req: Request, state: AppState) -> Response:
    conn = db_mod.connect(state.db_path)
    try:
        _require_admin(conn, req, state)
        rows = db_mod.fetch_all(
            conn,
            """
            SELECT
              c.id, c.kind, c.name, c.created_at,
              (SELECT COUNT(*) FROM conversation_members m WHERE m.conversation_id = c.id) AS member_count
            FROM conversations c
            WHERE c.kind = 'group'
            ORDER BY c.created_at DESC
            """,
        )
        return json_response(200, {"groups": [dict(r) for r in rows]})
    finally:
        conn.close()


def _admin_logs(req: Request, state: AppState) -> Response:
    conn = db_mod.connect(state.db_path)
    try:
        _require_admin(conn, req, state)
        limit = int((req.query.get("limit") or ["200"])[0])
        limit = max(1, min(500, limit))
        rows = db_mod.fetch_all(
            conn,
            "SELECT id, level, event, details_json, created_at FROM logs ORDER BY created_at DESC LIMIT ?",
            (limit,),
        )
        out = []
        for r in rows:
            d = dict(r)
            if d.get("details_json"):
                try:
                    d["details"] = json.loads(d["details_json"])
                except Exception:  # noqa: BLE001
                    d["details"] = None
            d.pop("details_json", None)
            out.append(d)
        return json_response(200, {"logs": out})
    finally:
        conn.close()


def _admin_metrics(req: Request, state: AppState) -> Response:
    conn = db_mod.connect(state.db_path)
    try:
        _require_admin(conn, req, state)
        now = now_ts()
        msg_last_min = db_mod.fetch_one(conn, "SELECT COUNT(*) AS c FROM messages WHERE created_at >= ?", (now - 60,))
        upload_last_min = db_mod.fetch_one(
            conn, "SELECT COALESCE(SUM(size),0) AS b FROM upload_chunks WHERE received_at >= ?", (now - 60,)
        )
        active_users = db_mod.fetch_one(conn, "SELECT COUNT(*) AS c FROM users WHERE is_banned = 0")
        online_devices = db_mod.fetch_one(
            conn,
            """
            SELECT (
              (SELECT COUNT(*) FROM unifiedpush_devices WHERE last_seen_at >= ?)
              + (SELECT COUNT(*) FROM webpush_subscriptions WHERE last_seen_at >= ?)
            ) AS c
            """,
            (now - 600, now - 600),
        )
        counters = db_mod.fetch_all(conn, "SELECT key, value, updated_at FROM metrics_counters ORDER BY key ASC")
        return json_response(
            200,
            {
                "msg_per_min": int(msg_last_min["c"]),
                "upload_bytes_per_min": int(upload_last_min["b"]),
                "active_users": int(active_users["c"]),
                "online_devices": int(online_devices["c"]),
                "counters": [dict(r) for r in counters],
            },
        )
    finally:
        conn.close()


def _admin_bot_create(req: Request, state: AppState) -> Response:
    payload = req.json()
    _require_json_fields(payload, ["email", "username", "password"])
    email = str(payload["email"]).strip().lower()
    username = str(payload["username"]).strip()
    password = str(payload["password"])
    if len(password) > 256:
        raise HttpError(400, "Password too long", code="password_too_long")

    conn = db_mod.connect(state.db_path)
    try:
        admin = _require_admin(conn, req, state)
        pw = hash_password(password)
        now = now_ts()
        bot_id = _new_id()
        with db_mod.transaction(conn):
            conn.execute(
                """
                INSERT INTO users (
                  id, email, username,
                  password_salt_b64, password_iterations, password_hash_b64,
                  is_bot, bot_verified, is_admin, created_at
                ) VALUES (?,?,?,?,?,?,?,?,?,?)
                """,
                (bot_id, email, username, pw.salt_b64, pw.iterations, pw.hash_b64, 1, 0, 0, now),
            )
            conn.execute("INSERT INTO user_settings (user_id, receive_everyone_mentions) VALUES (?,?)", (bot_id, 0))
            _log(conn, level="INFO", event="bot_created", details={"admin_id": admin["id"], "bot_id": bot_id})
        return json_response(201, {"bot_id": bot_id})
    finally:
        conn.close()


def _admin_bot_verify(req: Request, state: AppState) -> Response:
    bot_id = req.path_params["bot_id"]
    conn = db_mod.connect(state.db_path)
    try:
        admin = _require_admin(conn, req, state)
        conn.execute("UPDATE users SET bot_verified = 1 WHERE id = ? AND is_bot = 1", (bot_id,))
        _log(conn, level="INFO", event="bot_verified", details={"admin_id": admin["id"], "bot_id": bot_id})
        return json_response(200, {"ok": True})
    finally:
        conn.close()


def _admin_integrations_list(req: Request, state: AppState) -> Response:
    conn = db_mod.connect(state.db_path)
    try:
        _require_admin(conn, req, state)
        rows = db_mod.fetch_all(
            conn,
            """
            SELECT id, kind, name, bot_user_id, target_conversation_id, enabled, config_json, created_at, updated_at
            FROM integrations
            ORDER BY updated_at DESC
            """,
        )
        out = []
        for r in rows:
            d = dict(r)
            try:
                d["config"] = json.loads(d["config_json"])
            except Exception:  # noqa: BLE001
                d["config"] = None
            d.pop("config_json", None)
            out.append(d)
        return json_response(200, {"integrations": out})
    finally:
        conn.close()


def _admin_integrations_create(req: Request, state: AppState) -> Response:
    payload = req.json()
    _require_json_fields(payload, ["kind", "name", "bot_user_id", "target_conversation_id", "config"])
    kind = str(payload["kind"]).strip().lower()
    if kind not in ("twitch", "modrinth"):
        raise HttpError(400, "Invalid integration kind", code="invalid_input")
    name = str(payload["name"]).strip()[:100]
    bot_user_id = str(payload["bot_user_id"]).strip()
    target_conversation_id = str(payload["target_conversation_id"]).strip()
    enabled = bool(payload.get("enabled", True))
    config = payload["config"]

    conn = db_mod.connect(state.db_path)
    try:
        admin = _require_admin(conn, req, state)
        bot = db_mod.fetch_one(
            conn,
            "SELECT id, is_bot, bot_verified FROM users WHERE id = ?",
            (bot_user_id,),
        )
        if bot is None or int(bot["is_bot"]) != 1 or int(bot["bot_verified"]) != 1:
            raise HttpError(400, "bot_user_id must be a verified bot", code="invalid_input")
        convo = db_mod.fetch_one(conn, "SELECT id, kind FROM conversations WHERE id = ?", (target_conversation_id,))
        if convo is None or convo["kind"] != "group":
            raise HttpError(400, "target_conversation_id must be a group chat", code="invalid_input")

        now = now_ts()
        integration_id = _new_id()
        with db_mod.transaction(conn):
            conn.execute(
                """
                INSERT INTO integrations (
                  id, kind, name, bot_user_id, target_conversation_id, enabled, config_json, created_at, updated_at
                ) VALUES (?,?,?,?,?,?,?,?,?)
                """,
                (
                    integration_id,
                    kind,
                    name,
                    bot_user_id,
                    target_conversation_id,
                    1 if enabled else 0,
                    json.dumps(config),
                    now,
                    now,
                ),
            )
            # Ensure bot is member of target group (required to post).
            conn.execute(
                """
                INSERT INTO conversation_members (conversation_id, user_id, is_admin, is_banned, joined_at)
                VALUES (?,?,?,?,?)
                ON CONFLICT(conversation_id, user_id) DO UPDATE SET is_banned = 0
                """,
                (target_conversation_id, bot_user_id, 0, 0, now),
            )
            _log(conn, level="INFO", event="integration_created", details={"admin_id": admin["id"], "integration_id": integration_id, "kind": kind})
        return json_response(201, {"integration_id": integration_id})
    finally:
        conn.close()


def _admin_integrations_update(req: Request, state: AppState) -> Response:
    integration_id = req.path_params["integration_id"]
    payload = req.json()
    enabled = payload.get("enabled")
    config = payload.get("config")

    if enabled is None and config is None:
        raise HttpError(400, "Nothing to update", code="invalid_input")

    conn = db_mod.connect(state.db_path)
    try:
        admin = _require_admin(conn, req, state)
        now = now_ts()
        with db_mod.transaction(conn):
            if enabled is not None:
                conn.execute("UPDATE integrations SET enabled = ?, updated_at = ? WHERE id = ?", (1 if bool(enabled) else 0, now, integration_id))
            if config is not None:
                conn.execute("UPDATE integrations SET config_json = ?, updated_at = ? WHERE id = ?", (json.dumps(config), now, integration_id))
            _log(conn, level="INFO", event="integration_updated", details={"admin_id": admin["id"], "integration_id": integration_id})
        return json_response(200, {"ok": True})
    finally:
        conn.close()


def _admin_integrations_test_fire(req: Request, state: AppState) -> Response:
    integration_id = req.path_params["integration_id"]
    payload = req.json(max_bytes=200_000)
    text = str(payload.get("text", "Test notification")).strip()
    if text == "":
        text = "Test notification"

    conn = db_mod.connect(state.db_path)
    try:
        admin = _require_admin(conn, req, state)
        integ = db_mod.fetch_one(conn, "SELECT * FROM integrations WHERE id = ?", (integration_id,))
        if integ is None:
            raise HttpError(404, "Integration not found", code="not_found")
        now = now_ts()
        msg_id = _new_id()
        with db_mod.transaction(conn):
            conn.execute(
                """
                INSERT INTO messages (
                  id, conversation_id, sender_id, kind, text, attachment_id,
                  is_disappearing, created_at, contains_everyone_mention
                ) VALUES (?,?,?,?,?,?,?,?,?)
                """,
                (
                    msg_id,
                    integ["target_conversation_id"],
                    integ["bot_user_id"],
                    "text",
                    text,
                    None,
                    0,
                    now,
                    0,
                ),
            )
            _inc_metric(conn, "messages_total", 1)
            _log(conn, level="INFO", event="integration_test_fire", details={"admin_id": admin["id"], "integration_id": integration_id, "message_id": msg_id})
            _queue_push_for_message(conn, integ["target_conversation_id"], msg_id)
        return json_response(201, {"message_id": msg_id})
    finally:
        conn.close()


def _auth_register(req: Request, state: AppState) -> Response:
    payload = req.json()
    _require_json_fields(payload, ["email", "username", "password"])
    email = str(payload["email"]).strip().lower()
    username = str(payload["username"]).strip()
    password = str(payload["password"])
    if len(password) > 256:
        raise HttpError(400, "Password too long", code="password_too_long")
    if email == "" or username == "":
        raise HttpError(400, "Email/username required", code="invalid_input")

    pw = hash_password(password)
    now = now_ts()
    user_id = _new_id()

    conn = db_mod.connect(state.db_path)
    try:
        with db_mod.transaction(conn):
            conn.execute(
                """
                INSERT INTO users (
                  id, email, username,
                  password_salt_b64, password_iterations, password_hash_b64,
                  created_at
                ) VALUES (?,?,?,?,?,?,?)
                """,
                (user_id, email, username, pw.salt_b64, pw.iterations, pw.hash_b64, now),
            )
            conn.execute("INSERT INTO user_settings (user_id, receive_everyone_mentions) VALUES (?,?)", (user_id, 0))
            _log(conn, level="INFO", event="user_registered", details={"user_id": user_id})
    except Exception as exc:  # noqa: BLE001
        if "UNIQUE constraint failed: users.email" in str(exc):
            raise HttpError(409, "Email already exists", code="email_exists") from exc
        if "UNIQUE constraint failed: users.username" in str(exc):
            raise HttpError(409, "Username already exists", code="username_exists") from exc
        raise
    finally:
        conn.close()

    return json_response(201, {"user_id": user_id})


def _auth_login(req: Request, state: AppState) -> Response:
    payload = req.json()
    _require_json_fields(payload, ["login", "password"])
    login = _normalize_login(str(payload["login"]))
    password = str(payload["password"])
    if len(password) > 256:
        raise HttpError(400, "Password too long", code="password_too_long")

    conn = db_mod.connect(state.db_path)
    try:
        row = db_mod.fetch_one(conn, "SELECT * FROM users WHERE email = ? OR username = ?", (login.lower(), login))
        if row is None:
            raise HttpError(401, "Invalid credentials", code="invalid_credentials")
        if int(row["is_banned"]) == 1:
            raise HttpError(403, "Account banned", code="banned")

        now = now_ts()
        if int(row["password_reset_required"]) == 1:
            otp_row = db_mod.fetch_one(conn, "SELECT * FROM password_reset_otps WHERE user_id = ?", (row["id"],))
            if otp_row is None:
                raise HttpError(423, "Password reset required", code="password_reset_required")
            if otp_row["used_at"] is not None:
                raise HttpError(423, "Password reset required", code="password_reset_required")
            if int(otp_row["expires_at"]) <= now:
                raise HttpError(423, "OTP expired", code="otp_expired")
            if sha256_hex(password) != otp_row["otp_hash_hex"]:
                raise HttpError(401, "Invalid credentials", code="invalid_credentials")

            token = new_session_token()
            token_hash = sha256_hex(token)
            session_id = _new_id()
            with db_mod.transaction(conn):
                conn.execute(
                    "UPDATE password_reset_otps SET used_at = ? WHERE user_id = ?",
                    (now, row["id"]),
                )
                conn.execute(
                    """
                    INSERT INTO sessions (id, user_id, token_hash_hex, kind, created_at, expires_at)
                    VALUES (?,?,?,?,?,?)
                    """,
                    (session_id, row["id"], token_hash, "otp_reset", now, now + int(state.config.session_ttl_seconds)),
                )
                _log(conn, level="INFO", event="otp_login", details={"user_id": row["id"]})
            return json_response(
                200,
                {
                    "token": token,
                    "kind": "otp_reset",
                    "must_renew_password": True,
                },
            )

        pw = PasswordHash(
            salt_b64=row["password_salt_b64"],
            iterations=int(row["password_iterations"]),
            hash_b64=row["password_hash_b64"],
        )
        if not verify_password(password, pw):
            raise HttpError(401, "Invalid credentials", code="invalid_credentials")

        token = new_session_token()
        token_hash = sha256_hex(token)
        session_id = _new_id()
        conn.execute(
            """
            INSERT INTO sessions (id, user_id, token_hash_hex, kind, created_at, expires_at)
            VALUES (?,?,?,?,?,?)
            """,
            (session_id, row["id"], token_hash, "normal", now, now + int(state.config.session_ttl_seconds)),
        )
        _log(conn, level="INFO", event="login", details={"user_id": row["id"]})
        return json_response(200, {"token": token, "kind": "normal"})
    finally:
        conn.close()


def _auth_logout(req: Request, state: AppState) -> Response:
    conn = db_mod.connect(state.db_path)
    try:
        _auth_user(conn, req, state, allow_otp_reset=True)
        token = _token_from_auth(req)
        token_hash = sha256_hex(token)
        now = now_ts()
        conn.execute("UPDATE sessions SET revoked_at = ? WHERE token_hash_hex = ? AND revoked_at IS NULL", (now, token_hash))
        return json_response(200, {"ok": True})
    finally:
        conn.close()


def _auth_reset_complete(req: Request, state: AppState) -> Response:
    payload = req.json()
    _require_json_fields(payload, ["new_password"])
    new_password = str(payload["new_password"])
    if len(new_password) > 256:
        raise HttpError(400, "Password too long", code="password_too_long")

    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state, allow_otp_reset=True)
        if req.session is None or req.session.get("kind") != "otp_reset":
            raise HttpError(403, "Not an OTP reset session", code="forbidden")

        row = db_mod.fetch_one(conn, "SELECT password_reset_required FROM users WHERE id = ?", (user["id"],))
        if row is None or int(row["password_reset_required"]) != 1:
            raise HttpError(400, "No password reset pending", code="no_reset_pending")

        pw = hash_password(new_password)
        now = now_ts()
        with db_mod.transaction(conn):
            conn.execute(
                """
                UPDATE users
                SET password_salt_b64 = ?, password_iterations = ?, password_hash_b64 = ?,
                    password_reset_required = 0
                WHERE id = ?
                """,
                (pw.salt_b64, pw.iterations, pw.hash_b64, user["id"]),
            )
            conn.execute("DELETE FROM password_reset_otps WHERE user_id = ?", (user["id"],))
            # Revoke all sessions and create a fresh normal one.
            conn.execute("UPDATE sessions SET revoked_at = ? WHERE user_id = ? AND revoked_at IS NULL", (now, user["id"]))
            token = new_session_token()
            conn.execute(
                "INSERT INTO sessions (id, user_id, token_hash_hex, kind, created_at, expires_at) VALUES (?,?,?,?,?,?)",
                (_new_id(), user["id"], sha256_hex(token), "normal", now, now + int(state.config.session_ttl_seconds)),
            )
            _log(conn, level="INFO", event="password_reset_completed", details={"user_id": user["id"]})
        return json_response(200, {"token": token, "kind": "normal"})
    finally:
        conn.close()


def _me(req: Request, state: AppState) -> Response:
    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state, allow_otp_reset=True)
        settings = db_mod.fetch_one(conn, "SELECT receive_everyone_mentions FROM user_settings WHERE user_id = ?", (user["id"],))
        user_out = {
            "id": user["id"],
            "email": user["email"],
            "username": user["username"],
            "is_admin": int(user["is_admin"]),
            "is_bot": int(user["is_bot"]),
            "bot_verified": int(user["bot_verified"]),
            "password_reset_required": int(user["password_reset_required"]),
            "settings": {"receive_everyone_mentions": int(settings["receive_everyone_mentions"]) if settings else 0},
        }
        return json_response(200, {"user": user_out, "session": req.session})
    finally:
        conn.close()


def _me_settings(req: Request, state: AppState) -> Response:
    payload = req.json()
    if "receive_everyone_mentions" not in payload:
        raise HttpError(400, "Missing field: receive_everyone_mentions", code="missing_field")
    value = bool(payload["receive_everyone_mentions"])
    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state, allow_otp_reset=True)
        conn.execute(
            "UPDATE user_settings SET receive_everyone_mentions = ? WHERE user_id = ?",
            (1 if value else 0, user["id"]),
        )
        _log(conn, level="INFO", event="settings_updated", details={"user_id": user["id"]})
        return json_response(200, {"ok": True})
    finally:
        conn.close()


def _require_member(conn, conversation_id: str, user_id: str) -> dict:
    row = db_mod.fetch_one(
        conn,
        """
        SELECT c.id, c.kind, c.name, m.is_admin, m.is_banned
        FROM conversations c
        JOIN conversation_members m ON m.conversation_id = c.id
        WHERE c.id = ? AND m.user_id = ?
        """,
        (conversation_id, user_id),
    )
    if row is None:
        raise HttpError(404, "Conversation not found", code="not_found")
    if int(row["is_banned"]) == 1:
        raise HttpError(403, "Banned from conversation", code="forbidden")
    return dict(row)


def _conversations_list(req: Request, state: AppState) -> Response:
    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        rows = db_mod.fetch_all(
            conn,
            """
            SELECT
              c.id, c.kind, c.name, c.created_at,
              (SELECT COUNT(*) FROM conversation_members m WHERE m.conversation_id = c.id) AS member_count,
              (SELECT MAX(created_at) FROM messages mm WHERE mm.conversation_id = c.id) AS last_message_at
            FROM conversations c
            JOIN conversation_members m ON m.conversation_id = c.id
            WHERE m.user_id = ? AND m.is_banned = 0
            ORDER BY COALESCE(last_message_at, c.created_at) DESC
            """,
            (user["id"],),
        )
        return json_response(200, {"conversations": [dict(r) for r in rows]})
    finally:
        conn.close()


def _conversations_create_dm(req: Request, state: AppState) -> Response:
    payload = req.json()
    _require_json_fields(payload, ["other_user_id"])
    other_user_id = str(payload["other_user_id"]).strip()

    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        if other_user_id == user["id"]:
            raise HttpError(400, "Cannot DM yourself", code="invalid_input")
        other = db_mod.fetch_one(conn, "SELECT id FROM users WHERE id = ? AND is_banned = 0", (other_user_id,))
        if other is None:
            raise HttpError(404, "User not found", code="not_found")

        existing = db_mod.fetch_one(
            conn,
            """
            SELECT c.id
            FROM conversations c
            JOIN conversation_members m1 ON m1.conversation_id = c.id AND m1.user_id = ?
            JOIN conversation_members m2 ON m2.conversation_id = c.id AND m2.user_id = ?
            WHERE c.kind = 'dm'
            LIMIT 1
            """,
            (user["id"], other_user_id),
        )
        if existing is not None:
            return json_response(200, {"conversation_id": existing["id"]})

        now = now_ts()
        conv_id = _new_id()
        with db_mod.transaction(conn):
            conn.execute(
                "INSERT INTO conversations (id, kind, name, created_by, created_at) VALUES (?,?,?,?,?)",
                (conv_id, "dm", None, user["id"], now),
            )
            conn.execute(
                "INSERT INTO conversation_members (conversation_id, user_id, is_admin, is_banned, joined_at) VALUES (?,?,?,?,?)",
                (conv_id, user["id"], 0, 0, now),
            )
            conn.execute(
                "INSERT INTO conversation_members (conversation_id, user_id, is_admin, is_banned, joined_at) VALUES (?,?,?,?,?)",
                (conv_id, other_user_id, 0, 0, now),
            )
            _log(
                conn,
                level="INFO",
                event="dm_created",
                details={"conversation_id": conv_id, "a": user["id"], "b": other_user_id},
            )
        return json_response(201, {"conversation_id": conv_id})
    finally:
        conn.close()


def _conversations_create_group(req: Request, state: AppState) -> Response:
    payload = req.json()
    _require_json_fields(payload, ["name", "member_ids"])
    name = str(payload["name"]).strip()[:100]
    if name == "":
        raise HttpError(400, "Missing group name", code="missing_field")
    member_ids = payload["member_ids"]
    if not isinstance(member_ids, list):
        raise HttpError(400, "member_ids must be a list", code="invalid_input")

    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        ids = {user["id"]}
        for mid in member_ids:
            ids.add(str(mid))
        ids = {i for i in ids if i}
        if len(ids) > 25:
            raise HttpError(400, "Group too large (max 25)", code="group_too_large")

        existing_users = db_mod.fetch_all(
            conn, f"SELECT id FROM users WHERE id IN ({','.join('?' for _ in ids)}) AND is_banned = 0", tuple(ids)
        )
        if len(existing_users) != len(ids):
            raise HttpError(400, "Unknown or banned user in member_ids", code="invalid_input")

        now = now_ts()
        conv_id = _new_id()
        with db_mod.transaction(conn):
            conn.execute(
                "INSERT INTO conversations (id, kind, name, created_by, created_at) VALUES (?,?,?,?,?)",
                (conv_id, "group", name, user["id"], now),
            )
            for uid in ids:
                conn.execute(
                    "INSERT INTO conversation_members (conversation_id, user_id, is_admin, is_banned, joined_at) VALUES (?,?,?,?,?)",
                    (conv_id, uid, 1 if uid == user["id"] else 0, 0, now),
                )
            _log(
                conn,
                level="INFO",
                event="group_created",
                details={"conversation_id": conv_id, "creator": user["id"], "size": len(ids)},
            )
        return json_response(201, {"conversation_id": conv_id})
    finally:
        conn.close()


def _conversation_get(req: Request, state: AppState) -> Response:
    conversation_id = req.path_params["conversation_id"]
    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        convo = _require_member(conn, conversation_id, user["id"])
        members = db_mod.fetch_all(
            conn,
            """
            SELECT u.id, u.username, u.is_bot, u.bot_verified, m.is_admin, m.is_banned, m.joined_at
            FROM conversation_members m
            JOIN users u ON u.id = m.user_id
            WHERE m.conversation_id = ?
            ORDER BY m.joined_at ASC
            """,
            (conversation_id,),
        )
        return json_response(200, {"conversation": convo, "members": [dict(r) for r in members]})
    finally:
        conn.close()


def _conversation_members_add(req: Request, state: AppState) -> Response:
    conversation_id = req.path_params["conversation_id"]
    payload = req.json()
    _require_json_fields(payload, ["user_id"])
    target_user_id = str(payload["user_id"]).strip()
    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        convo = _require_member(conn, conversation_id, user["id"])
        if convo["kind"] != "group":
            raise HttpError(400, "Only group chats support membership changes", code="invalid_input")
        if int(convo["is_admin"]) != 1 and int(user["is_admin"]) != 1:
            raise HttpError(403, "Group admin only", code="forbidden")

        now = now_ts()
        conn.execute(
            """
            INSERT INTO conversation_members (conversation_id, user_id, is_admin, is_banned, joined_at)
            VALUES (?,?,?,?,?)
            ON CONFLICT(conversation_id, user_id) DO UPDATE SET
              is_banned = 0
            """,
            (conversation_id, target_user_id, 0, 0, now),
        )
        _log(conn, level="INFO", event="group_member_added", details={"conversation_id": conversation_id, "by": user["id"], "user_id": target_user_id})
        return json_response(200, {"ok": True})
    finally:
        conn.close()


def _conversation_members_remove(req: Request, state: AppState) -> Response:
    conversation_id = req.path_params["conversation_id"]
    target_user_id = req.path_params["user_id"]
    ban = (req.query.get("ban") or ["0"])[0] in ("1", "true", "yes")
    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        convo = _require_member(conn, conversation_id, user["id"])
        if convo["kind"] != "group":
            raise HttpError(400, "Only group chats support membership changes", code="invalid_input")
        if int(convo["is_admin"]) != 1 and int(user["is_admin"]) != 1:
            raise HttpError(403, "Group admin only", code="forbidden")
        if target_user_id == user["id"]:
            raise HttpError(400, "Cannot remove yourself", code="invalid_input")

        if ban:
            conn.execute(
                "UPDATE conversation_members SET is_banned = 1 WHERE conversation_id = ? AND user_id = ?",
                (conversation_id, target_user_id),
            )
        else:
            conn.execute(
                "DELETE FROM conversation_members WHERE conversation_id = ? AND user_id = ?",
                (conversation_id, target_user_id),
            )
        _log(conn, level="WARN", event="group_member_removed", details={"conversation_id": conversation_id, "by": user["id"], "user_id": target_user_id, "ban": ban})
        return json_response(200, {"ok": True})
    finally:
        conn.close()


def _messages_send(req: Request, state: AppState) -> Response:
    conversation_id = req.path_params["conversation_id"]
    payload = req.json(max_bytes=2_000_000)
    _require_json_fields(payload, ["kind"])
    kind = str(payload["kind"]).strip().lower()
    text = payload.get("text")
    attachment_id = payload.get("attachment_id")
    is_disappearing = bool(payload.get("is_disappearing", False))

    if kind not in ("text", "image", "file", "voice"):
        raise HttpError(400, "Invalid kind", code="invalid_input")
    if kind == "text":
        if text is None:
            raise HttpError(400, "Missing field: text", code="missing_field")
        text = str(text)
    else:
        if attachment_id is None:
            raise HttpError(400, "Missing field: attachment_id", code="missing_field")
        attachment_id = str(attachment_id)

    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        convo = _require_member(conn, conversation_id, user["id"])
        now = now_ts()

        contains_everyone_mention = 0
        if convo["kind"] == "group" and kind == "text" and "@everyone" in (text or ""):
            contains_everyone_mention = 1

        msg_id = _new_id()
        with db_mod.transaction(conn):
            conn.execute(
                """
                INSERT INTO messages (
                  id, conversation_id, sender_id, kind, text, attachment_id,
                  is_disappearing, created_at, contains_everyone_mention
                ) VALUES (?,?,?,?,?,?,?,?,?)
                """,
                (
                    msg_id,
                    conversation_id,
                    user["id"],
                    kind,
                    text if kind == "text" else None,
                    attachment_id if kind != "text" else None,
                    1 if is_disappearing else 0,
                    now,
                    contains_everyone_mention,
                ),
            )
            _inc_metric(conn, "messages_total", 1)
            _log(
                conn,
                level="INFO",
                event="message_sent",
                details={
                    "message_id": msg_id,
                    "conversation_id": conversation_id,
                    "sender_id": user["id"],
                    "kind": kind,
                },
            )
            _queue_push_for_message(conn, conversation_id, msg_id)
        return json_response(201, {"message_id": msg_id})
    finally:
        conn.close()


def _messages_list(req: Request, state: AppState) -> Response:
    conversation_id = req.path_params["conversation_id"]
    since = int((req.query.get("since") or ["0"])[0] or "0")
    limit = int((req.query.get("limit") or ["50"])[0] or "50")
    limit = max(1, min(200, limit))
    include_deleted = (req.query.get("include_deleted") or ["0"])[0] in ("1", "true", "yes")

    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        convo = _require_member(conn, conversation_id, user["id"])
        if convo["kind"] == "dm":
            # no extra constraints
            pass

        where_deleted = "" if include_deleted or int(user["is_admin"]) == 1 else "AND m.deleted_at IS NULL"
        rows = db_mod.fetch_all(
            conn,
            f"""
            SELECT
              m.*,
              a.filename AS attachment_filename,
              a.content_type AS attachment_content_type,
              a.size AS attachment_size,
              a.sha256_hex AS attachment_sha256,
              a.storage_path AS attachment_storage_path
            FROM messages m
            LEFT JOIN attachments a ON a.id = m.attachment_id
            WHERE m.conversation_id = ?
              AND m.created_at > ?
              {where_deleted}
            ORDER BY m.created_at ASC
            LIMIT ?
            """,
            (conversation_id, since, limit),
        )

        messages = []
        for r in rows:
            msg = dict(r)
            attachment = None
            if msg.get("attachment_id"):
                attachment = {
                    "id": msg["attachment_id"],
                    "filename": msg.get("attachment_filename"),
                    "content_type": msg.get("attachment_content_type"),
                    "size": msg.get("attachment_size"),
                    "sha256_hex": msg.get("attachment_sha256"),
                    "url": f"/api/v1/attachments/{msg['attachment_id']}",
                }
            reactions = _reactions_for_message(conn, msg["id"], user["id"])
            messages.append(
                {
                    "id": msg["id"],
                    "conversation_id": msg["conversation_id"],
                    "sender_id": msg["sender_id"],
                    "kind": msg["kind"],
                    "text": msg["text"],
                    "attachment": attachment,
                    "is_disappearing": int(msg["is_disappearing"]),
                    "created_at": msg["created_at"],
                    "edited_at": msg["edited_at"],
                    "deleted_at": msg["deleted_at"],
                    "deleted_by": msg["deleted_by"],
                    "contains_everyone_mention": int(msg["contains_everyone_mention"]),
                    "reactions": reactions,
                }
            )
        return json_response(200, {"messages": messages})
    finally:
        conn.close()


def _reactions_for_message(conn, message_id: str, viewer_user_id: str) -> list[dict]:
    rows = db_mod.fetch_all(
        conn,
        """
        SELECT emoji, COUNT(*) AS c,
               SUM(CASE WHEN user_id = ? THEN 1 ELSE 0 END) AS mine
        FROM reactions
        WHERE message_id = ?
        GROUP BY emoji
        ORDER BY c DESC, emoji ASC
        """,
        (viewer_user_id, message_id),
    )
    out = []
    for r in rows:
        out.append({"emoji": r["emoji"], "count": int(r["c"]), "mine": int(r["mine"]) > 0})
    return out


def _message_edit(req: Request, state: AppState) -> Response:
    message_id = req.path_params["message_id"]
    payload = req.json()
    _require_json_fields(payload, ["text"])
    new_text = str(payload["text"])
    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        msg = db_mod.fetch_one(conn, "SELECT * FROM messages WHERE id = ?", (message_id,))
        if msg is None:
            raise HttpError(404, "Message not found", code="not_found")
        if msg["deleted_at"] is not None:
            raise HttpError(409, "Message deleted", code="conflict")
        if msg["sender_id"] != user["id"]:
            raise HttpError(403, "Only sender can edit", code="forbidden")
        if msg["kind"] != "text":
            raise HttpError(400, "Only text messages can be edited", code="invalid_input")

        now = now_ts()
        with db_mod.transaction(conn):
            conn.execute(
                "UPDATE messages SET text = ?, edited_at = ? WHERE id = ?",
                (new_text, now, message_id),
            )
            conn.execute(
                "INSERT INTO message_edits (id, message_id, editor_id, old_text, new_text, edited_at) VALUES (?,?,?,?,?,?)",
                (_new_id(), message_id, user["id"], msg["text"], new_text, now),
            )
            _log(conn, level="INFO", event="message_edited", details={"message_id": message_id, "user_id": user["id"]})
        return json_response(200, {"ok": True})
    finally:
        conn.close()


def _message_delete(req: Request, state: AppState) -> Response:
    message_id = req.path_params["message_id"]
    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        msg = db_mod.fetch_one(
            conn,
            """
            SELECT m.*, c.kind AS conversation_kind
            FROM messages m
            JOIN conversations c ON c.id = m.conversation_id
            WHERE m.id = ?
            """,
            (message_id,),
        )
        if msg is None:
            raise HttpError(404, "Message not found", code="not_found")
        if msg["deleted_at"] is not None:
            return json_response(200, {"ok": True})

        allowed = msg["sender_id"] == user["id"] or int(user["is_admin"]) == 1
        if not allowed and msg["conversation_kind"] == "group":
            member = db_mod.fetch_one(
                conn,
                "SELECT is_admin FROM conversation_members WHERE conversation_id = ? AND user_id = ?",
                (msg["conversation_id"], user["id"]),
            )
            if member is not None and int(member["is_admin"]) == 1:
                allowed = True
        if not allowed:
            raise HttpError(403, "Not allowed", code="forbidden")

        now = now_ts()
        with db_mod.transaction(conn):
            conn.execute(
                "UPDATE messages SET deleted_at = ?, deleted_by = ? WHERE id = ?",
                (now, user["id"], message_id),
            )
            conn.execute(
                "INSERT INTO message_deletion_audit (id, message_id, conversation_id, deleted_by, deleted_at) VALUES (?,?,?,?,?)",
                (_new_id(), message_id, msg["conversation_id"], user["id"], now),
            )
            _log(conn, level="INFO", event="message_deleted", details={"message_id": message_id, "by": user["id"]})
        return json_response(200, {"ok": True})
    finally:
        conn.close()


def _message_react(req: Request, state: AppState) -> Response:
    message_id = req.path_params["message_id"]
    payload = req.json()
    _require_json_fields(payload, ["emoji", "action"])
    emoji = str(payload["emoji"])[:16]
    action = str(payload["action"]).lower()
    if action not in ("add", "remove"):
        raise HttpError(400, "Invalid action", code="invalid_input")

    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        msg = db_mod.fetch_one(conn, "SELECT conversation_id FROM messages WHERE id = ?", (message_id,))
        if msg is None:
            raise HttpError(404, "Message not found", code="not_found")
        _require_member(conn, msg["conversation_id"], user["id"])
        now = now_ts()
        if action == "add":
            conn.execute(
                "INSERT OR IGNORE INTO reactions (message_id, user_id, emoji, created_at) VALUES (?,?,?,?)",
                (message_id, user["id"], emoji, now),
            )
        else:
            conn.execute(
                "DELETE FROM reactions WHERE message_id = ? AND user_id = ? AND emoji = ?",
                (message_id, user["id"], emoji),
            )
        _log(conn, level="INFO", event="reaction", details={"message_id": message_id, "user_id": user["id"], "emoji": emoji, "action": action})
        return json_response(200, {"ok": True})
    finally:
        conn.close()


def _upload_tmp_dir(state: AppState, upload_id: str) -> Path:
    return state.storage_dir / "tmp" / "uploads" / upload_id


def _upload_create(req: Request, state: AppState) -> Response:
    payload = req.json()
    _require_json_fields(payload, ["filename", "total_size", "overall_sha256_hex"])
    filename = str(payload["filename"])[:255]
    content_type = payload.get("content_type")
    if content_type is not None:
        content_type = str(content_type)[:100]
    total_size = int(payload["total_size"])
    overall_sha256_hex = str(payload["overall_sha256_hex"]).lower()

    chunk_size = int(payload.get("chunk_size") or state.config.upload_chunk_bytes)
    if chunk_size <= 0 or chunk_size > int(state.config.upload_chunk_bytes):
        raise HttpError(400, "Invalid chunk_size", code="invalid_input")

    if total_size <= 0 or total_size > int(state.config.max_attachment_bytes):
        raise HttpError(400, "Invalid total_size", code="invalid_input")
    expected_chunks = int(math.ceil(total_size / chunk_size))
    num_chunks = int(payload.get("num_chunks") or expected_chunks)
    if num_chunks != expected_chunks:
        raise HttpError(400, "num_chunks mismatch", code="invalid_input")
    if len(overall_sha256_hex) != 64 or any(c not in "0123456789abcdef" for c in overall_sha256_hex):
        raise HttpError(400, "Invalid overall_sha256_hex", code="invalid_input")

    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        upload_id = _new_id()
        tmp_dir = _upload_tmp_dir(state, upload_id)
        safe_mkdir(tmp_dir)
        now = now_ts()
        conn.execute(
            """
            INSERT INTO uploads (
              id, user_id, filename, content_type, total_size, overall_sha256_hex,
              chunk_size, num_chunks, tmp_dir, state, created_at
            ) VALUES (?,?,?,?,?,?,?,?,?,?,?)
            """,
            (
                upload_id,
                user["id"],
                filename,
                content_type,
                total_size,
                overall_sha256_hex,
                chunk_size,
                num_chunks,
                str(tmp_dir),
                "active",
                now,
            ),
        )
        _log(conn, level="INFO", event="upload_created", details={"upload_id": upload_id, "user_id": user["id"], "total_size": total_size})
        return json_response(201, {"upload_id": upload_id, "chunk_size": chunk_size, "num_chunks": num_chunks})
    finally:
        conn.close()


def _upload_get(req: Request, state: AppState) -> Response:
    upload_id = req.path_params["upload_id"]
    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        upload = db_mod.fetch_one(conn, "SELECT * FROM uploads WHERE id = ? AND user_id = ?", (upload_id, user["id"]))
        if upload is None:
            raise HttpError(404, "Upload not found", code="not_found")
        chunks = db_mod.fetch_all(conn, "SELECT chunk_index FROM upload_chunks WHERE upload_id = ? ORDER BY chunk_index ASC", (upload_id,))
        return json_response(
            200,
            {
                "upload": {"id": upload["id"], "state": upload["state"], "created_at": upload["created_at"], "finalized_at": upload["finalized_at"]},
                "received_chunks": [int(r["chunk_index"]) for r in chunks],
            },
        )
    finally:
        conn.close()


def _upload_chunk_put(req: Request, state: AppState) -> Response:
    upload_id = req.path_params["upload_id"]
    chunk_index = int(req.path_params["chunk_index"])
    claimed_sha = req.header("x-chunk-sha256")
    if not claimed_sha:
        raise HttpError(400, "Missing X-Chunk-Sha256", code="missing_field")
    claimed_sha = claimed_sha.strip().lower()
    if len(claimed_sha) != 64 or any(c not in "0123456789abcdef" for c in claimed_sha):
        raise HttpError(400, "Invalid X-Chunk-Sha256", code="invalid_input")
    if req.content_length < 0:
        raise HttpError(411, "Missing Content-Length", code="length_required")

    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        upload = db_mod.fetch_one(conn, "SELECT * FROM uploads WHERE id = ? AND user_id = ?", (upload_id, user["id"]))
        if upload is None:
            raise HttpError(404, "Upload not found", code="not_found")
        if upload["state"] != "active":
            raise HttpError(409, "Upload not active", code="conflict")
        if chunk_index < 0 or chunk_index >= int(upload["num_chunks"]):
            raise HttpError(400, "Invalid chunk index", code="invalid_input")
        max_chunk = int(upload["chunk_size"])
        if req.content_length <= 0 or req.content_length > max_chunk:
            raise HttpError(413, "Chunk too large", code="payload_too_large")

        tmp_dir = Path(upload["tmp_dir"])
        safe_mkdir(tmp_dir)
        chunk_path = tmp_dir / f"{chunk_index:06d}.part"
        if chunk_path.exists():
            existing_sha = _sha256_file_hex(chunk_path)
            if existing_sha == claimed_sha:
                return json_response(200, {"ok": True, "dedup": True})
            raise HttpError(409, "Chunk already exists with different hash", code="conflict")

        h = hashlib.sha256()
        tmp_write_path = tmp_dir / f"{chunk_index:06d}.part.tmp"
        with open(tmp_write_path, "wb") as f:
            remaining = req.content_length
            while remaining > 0:
                part = req.rfile.read(min(1024 * 256, remaining))
                if not part:
                    break
                f.write(part)
                h.update(part)
                remaining -= len(part)
        if remaining != 0:
            tmp_write_path.unlink(missing_ok=True)
            raise HttpError(400, "Unexpected EOF while reading chunk", code="invalid_input")
        actual_sha = h.hexdigest()
        if actual_sha != claimed_sha:
            tmp_write_path.unlink(missing_ok=True)
            raise HttpError(400, "Chunk hash mismatch", code="hash_mismatch")
        tmp_write_path.replace(chunk_path)

        now = now_ts()
        conn.execute(
            "INSERT OR REPLACE INTO upload_chunks (upload_id, chunk_index, size, sha256_hex, received_at) VALUES (?,?,?,?,?)",
            (upload_id, chunk_index, req.content_length, actual_sha, now),
        )
        _inc_metric(conn, "upload_chunks_total", 1)
        _log(conn, level="INFO", event="upload_chunk_received", details={"upload_id": upload_id, "chunk_index": chunk_index, "size": req.content_length})
        return json_response(200, {"ok": True})
    finally:
        conn.close()


def _upload_finalize(req: Request, state: AppState) -> Response:
    upload_id = req.path_params["upload_id"]
    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        upload = db_mod.fetch_one(conn, "SELECT * FROM uploads WHERE id = ? AND user_id = ?", (upload_id, user["id"]))
        if upload is None:
            raise HttpError(404, "Upload not found", code="not_found")
        if upload["state"] != "active":
            raise HttpError(409, "Upload not active", code="conflict")

        num_chunks = int(upload["num_chunks"])
        received = db_mod.fetch_all(conn, "SELECT chunk_index FROM upload_chunks WHERE upload_id = ?", (upload_id,))
        have = {int(r["chunk_index"]) for r in received}
        missing = [i for i in range(num_chunks) if i not in have]
        if missing:
            raise HttpError(409, "Missing chunks", code="missing_chunks")

        tmp_dir = Path(upload["tmp_dir"])
        final_attachment_id = _new_id()
        final_rel = Path("attachments") / final_attachment_id[:2] / final_attachment_id
        final_dir = state.storage_dir / final_rel.parent
        safe_mkdir(final_dir)
        final_path = state.storage_dir / final_rel
        tmp_final_path = final_path.with_suffix(".tmp")

        h = hashlib.sha256()
        total_written = 0
        with open(tmp_final_path, "wb") as out:
            for i in range(num_chunks):
                chunk_path = tmp_dir / f"{i:06d}.part"
                if not chunk_path.exists():
                    tmp_final_path.unlink(missing_ok=True)
                    raise HttpError(409, "Missing chunk file", code="missing_chunks")
                with open(chunk_path, "rb") as ch:
                    while True:
                        data = ch.read(1024 * 256)
                        if not data:
                            break
                        out.write(data)
                        h.update(data)
                        total_written += len(data)
        if total_written != int(upload["total_size"]):
            tmp_final_path.unlink(missing_ok=True)
            raise HttpError(409, "Final size mismatch", code="conflict")
        if h.hexdigest() != str(upload["overall_sha256_hex"]).lower():
            tmp_final_path.unlink(missing_ok=True)
            raise HttpError(400, "Overall hash mismatch", code="hash_mismatch")

        tmp_final_path.replace(final_path)
        now = now_ts()
        with db_mod.transaction(conn):
            conn.execute(
                """
                INSERT INTO attachments (id, user_id, filename, content_type, size, sha256_hex, storage_path, created_at)
                VALUES (?,?,?,?,?,?,?,?)
                """,
                (
                    final_attachment_id,
                    user["id"],
                    upload["filename"],
                    upload["content_type"],
                    total_written,
                    h.hexdigest(),
                    str(final_rel),
                    now,
                ),
            )
            conn.execute(
                "UPDATE uploads SET state = 'finalized', finalized_at = ? WHERE id = ?",
                (now, upload_id),
            )
            _log(
                conn,
                level="INFO",
                event="upload_finalized",
                details={"upload_id": upload_id, "attachment_id": final_attachment_id},
            )
        shutil.rmtree(tmp_dir, ignore_errors=True)
        return json_response(200, {"attachment_id": final_attachment_id, "url": f"/api/v1/attachments/{final_attachment_id}"})
    finally:
        conn.close()


def _attachment_get(req: Request, state: AppState) -> Response:
    attachment_id = req.path_params["attachment_id"]
    download = (req.query.get("download") or ["0"])[0] in ("1", "true", "yes")
    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        att = db_mod.fetch_one(conn, "SELECT * FROM attachments WHERE id = ?", (attachment_id,))
        if att is None:
            raise HttpError(404, "Attachment not found", code="not_found")
        if int(user["is_admin"]) != 1:
            ok = db_mod.fetch_one(
                conn,
                """
                SELECT 1
                FROM messages m
                JOIN conversation_members cm ON cm.conversation_id = m.conversation_id AND cm.user_id = ?
                WHERE m.attachment_id = ?
                LIMIT 1
                """,
                (user["id"], attachment_id),
            )
            if ok is None:
                raise HttpError(403, "Not allowed", code="forbidden")

        file_path = (state.storage_dir / str(att["storage_path"])).resolve()
        if not file_path.exists():
            raise HttpError(404, "File missing on disk", code="not_found")

        ctype = att["content_type"] or "application/octet-stream"
        disp = "attachment" if download or not str(ctype).startswith("image/") else "inline"
        filename = str(att["filename"]).replace('"', "")
        headers = {
            "content-type": ctype,
            "content-length": str(file_path.stat().st_size),
            "content-disposition": f'{disp}; filename="{filename}"',
            "cache-control": "no-store",
        }
        return Response(status=200, headers=headers, file_path=str(file_path))
    finally:
        conn.close()


def _sha256_file_hex(path: Path) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        while True:
            data = f.read(1024 * 256)
            if not data:
                break
            h.update(data)
    return h.hexdigest()


def _queue_push_for_message(conn, conversation_id: str, message_id: str) -> None:
    msg = db_mod.fetch_one(
        conn,
        """
        SELECT
          m.id, m.kind, m.text, m.sender_id, m.conversation_id,
          m.contains_everyone_mention,
          c.kind AS conversation_kind
        FROM messages m
        JOIN conversations c ON c.id = m.conversation_id
        WHERE m.id = ?
        """,
        (message_id,),
    )
    if msg is None:
        return
    sender_id = msg["sender_id"]

    preview = "New message"
    if msg["kind"] == "text":
        preview = clamp_push_preview(msg["text"] or "", 12) or "New message"
    elif msg["kind"] == "image":
        preview = "Image"
    elif msg["kind"] == "file":
        preview = "File"
    elif msg["kind"] == "voice":
        preview = "Voice note"

    has_everyone_mention = int(msg["contains_everyone_mention"]) == 1 and msg["conversation_kind"] == "group"

    members = db_mod.fetch_all(
        conn,
        "SELECT user_id FROM conversation_members WHERE conversation_id = ? AND is_banned = 0",
        (conversation_id,),
    )
    for m in members:
        uid = m["user_id"]
        if uid == sender_id:
            continue

        mention = None
        if has_everyone_mention:
            setting = db_mod.fetch_one(
                conn,
                "SELECT receive_everyone_mentions FROM user_settings WHERE user_id = ?",
                (uid,),
            )
            if setting is not None and int(setting["receive_everyone_mentions"]) == 1:
                mention = "everyone"

        payload = {"preview": preview, "conversation_id": conversation_id, "message_id": message_id}
        if mention:
            payload["mention"] = mention

        # UnifiedPush devices
        devices = db_mod.fetch_all(conn, "SELECT id FROM unifiedpush_devices WHERE user_id = ?", (uid,))
        for d in devices:
            conn.execute(
                "INSERT INTO push_events (id, user_id, target_kind, target_id, payload_json, created_at) VALUES (?,?,?,?,?,?)",
                (
                    _new_id(),
                    uid,
                    "unifiedpush",
                    d["id"],
                    json.dumps(payload),
                    now_ts(),
                ),
            )
        subs = db_mod.fetch_all(conn, "SELECT id FROM webpush_subscriptions WHERE user_id = ?", (uid,))
        for s in subs:
            conn.execute(
                "INSERT INTO push_events (id, user_id, target_kind, target_id, payload_json, created_at) VALUES (?,?,?,?,?,?)",
                (
                    _new_id(),
                    uid,
                    "webpush",
                    s["id"],
                    json.dumps(payload),
                    now_ts(),
                ),
            )


def _push_unifiedpush_register(req: Request, state: AppState) -> Response:
    payload = req.json()
    _require_json_fields(payload, ["endpoint"])
    endpoint = str(payload["endpoint"]).strip()
    if endpoint == "":
        raise HttpError(400, "Invalid endpoint", code="invalid_input")
    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        now = now_ts()
        device_id = _new_id()
        conn.execute(
            "INSERT INTO unifiedpush_devices (id, user_id, endpoint, created_at, last_seen_at) VALUES (?,?,?,?,?)",
            (device_id, user["id"], endpoint, now, now),
        )
        _log(conn, level="INFO", event="unifiedpush_registered", details={"user_id": user["id"], "device_id": device_id})
        return json_response(201, {"device_id": device_id})
    finally:
        conn.close()


def _push_webpush_subscribe(req: Request, state: AppState) -> Response:
    payload = req.json()
    _require_json_fields(payload, ["endpoint", "p256dh", "auth"])
    endpoint = str(payload["endpoint"]).strip()
    p256dh = str(payload["p256dh"]).strip()
    auth = str(payload["auth"]).strip()
    if endpoint == "" or p256dh == "" or auth == "":
        raise HttpError(400, "Invalid subscription", code="invalid_input")
    conn = db_mod.connect(state.db_path)
    try:
        user = _auth_user(conn, req, state)
        now = now_ts()
        sub_id = _new_id()
        conn.execute(
            "INSERT INTO webpush_subscriptions (id, user_id, endpoint, p256dh, auth, created_at, last_seen_at) VALUES (?,?,?,?,?,?,?)",
            (sub_id, user["id"], endpoint, p256dh, auth, now, now),
        )
        _log(conn, level="INFO", event="webpush_subscribed", details={"user_id": user["id"], "subscription_id": sub_id})
        return json_response(201, {"subscription_id": sub_id})
    finally:
        conn.close()
