glo/routes.py

# routes.py
import json
import re
import sqlite3
import time

from flask import jsonify, request

from features import (
    PAGE_TYPES,
    PAGE_VISIBILITIES,
    PROFILE_BIO_MODES,
    REACTION_EMOJI_RE,
    TEAM_MEMBER_ROLES,
    THEME_MODES,
    append_call_signal,
    call_payload,
    channel_presence_count,
    dm_payload,
    dm_presence_count,
    end_call,
    feed_message_payload,
    find_or_create_dm,
    get_call_with_dm,
    get_dm_for_user,
    list_call_signals,
    list_dms_for_user,
    normalized_hex_color,
    page_payload,
    profile_payload,
    reaction_summary_for_message,
    resolve_owned_attachment,
    resolve_owned_attachment_ids,
    richer_message_payload,
    room_presence_count,
    send_attachment_file,
    update_presence,
    validate_dm_access,
    valid_hex_color,
)

ROOM_SLUG_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,46}[a-z0-9])?$")


def register_routes(app, ctx):
    get_db = ctx["get_db"]
    db_write = ctx["db_write"]
    safe_json = ctx["safe_json"]
    csrf_valid = ctx["csrf_valid"]
    rate_limited = ctx["rate_limited"]
    current_user = ctx["current_user"]
    require_user = ctx["require_user"]
    require_writable_user = ctx["require_writable_user"]
    require_moderator = ctx["require_moderator"]
    require_admin = ctx["require_admin"]
    now_iso = ctx["now_iso"]
    utc_now = ctx["utc_now"]
    parse_iso = ctx["parse_iso"]
    normalize_username = ctx["normalize_username"]
    audit = ctx["audit"]
    hash_identifier = ctx["hash_identifier"]
    sign_call_helper_token = ctx["sign_call_helper_token"]
    CALLS_BASE = ctx["CALLS_BASE"]
    CALL_TOKEN_TTL_SECONDS = ctx["CALL_TOKEN_TTL_SECONDS"]
    CALL_HELPER_SHARED_SECRET = ctx["CALL_HELPER_SHARED_SECRET"]
    room_by_id = ctx["room_by_id"]
    get_room_by_slug = ctx["get_room_by_slug"]
    can_access_room = ctx["can_access_room"]
    can_write_room = ctx["can_write_room"]
    can_manage_room = ctx["can_manage_room"]
    room_payload = ctx["room_payload"]
    team_slug_from_name = ctx["team_slug_from_name"]
    page_slug_from_title = ctx["page_slug_from_title"]
    get_team_by_slug = ctx["get_team_by_slug"]
    get_channel_by_id = ctx["get_channel_by_id"]
    get_channel_by_slug = ctx["get_channel_by_slug"]
    get_channel_by_legacy_room_id = ctx["get_channel_by_legacy_room_id"]
    team_member_role = ctx["team_member_role"]
    is_team_member = ctx["is_team_member"]
    is_team_admin = ctx["is_team_admin"]
    can_edit_team = ctx["can_edit_team"]
    can_access_team = ctx["can_access_team"]
    can_access_channel = ctx["can_access_channel"]
    can_write_channel = ctx["can_write_channel"]
    team_payload = ctx["team_payload"]
    channel_payload = ctx["channel_payload"]
    user_payload = ctx["user_payload"]
    get_dm_other_user = ctx["get_dm_other_user"]
    MESSAGE_SELECT = ctx["MESSAGE_SELECT"]
    team_page_select = ctx["team_page_select"]
    constants = ctx["constants"]

    MAX_MESSAGE_LENGTH = constants["MAX_MESSAGE_LENGTH"]
    MAX_HISTORY_MESSAGES = constants["MAX_HISTORY_MESSAGES"]
    MAX_INCREMENTAL_MESSAGES = constants["MAX_INCREMENTAL_MESSAGES"]
    MAX_FEED_MESSAGES = constants["MAX_FEED_MESSAGES"]
    MAX_THREAD_REPLIES = constants["MAX_THREAD_REPLIES"]
    TYPING_TTL_SECONDS = constants["TYPING_TTL_SECONDS"]
    PRESENCE_TTL_SECONDS = constants["PRESENCE_TTL_SECONDS"]
    INVITE_TTL_HOURS = constants["INVITE_TTL_HOURS"]
    TEAM_INVITE_TTL_HOURS = constants["TEAM_INVITE_TTL_HOURS"]
    MAX_SEARCH_QUERY_LENGTH = constants["MAX_SEARCH_QUERY_LENGTH"]
    MAX_SEARCH_RESULTS = constants["MAX_SEARCH_RESULTS"]
    READ_RECEIPT_BATCH_LIMIT = constants["READ_RECEIPT_BATCH_LIMIT"]
    MAX_MESSAGE_EDIT_WINDOW_SECONDS = constants["MAX_MESSAGE_EDIT_WINDOW_SECONDS"]
    MAX_DISPLAY_NAME_LENGTH = constants["MAX_DISPLAY_NAME_LENGTH"]
    MAX_PROFILE_BIO_LENGTH = constants["MAX_PROFILE_BIO_LENGTH"]
    MAX_WEBSITE_LENGTH = constants["MAX_WEBSITE_LENGTH"]
    MAX_LOCATION_LENGTH = constants["MAX_LOCATION_LENGTH"]
    MAX_CUSTOM_CSS_LENGTH = constants["MAX_CUSTOM_CSS_LENGTH"]
    MAX_PROFILE_BADGE_LENGTH = constants["MAX_PROFILE_BADGE_LENGTH"]
    MAX_OWNED_TEAMS_PER_USER = constants["MAX_OWNED_TEAMS_PER_USER"]
    ROOM_VISIBILITIES = constants["ROOM_VISIBILITIES"]
    ROLES = constants["ROLES"]
    STATUSES = constants["STATUSES"]

    PRIVATE_ACCESS_KEY = ctx["PRIVATE_ACCESS_KEY"]

    ROLES = constants["ROLES"]
    STATUSES = constants["STATUSES"]

    html_module = ctx["html"]
    
    def private_access_key_valid(candidate: object) -> bool:
        configured = (PRIVATE_ACCESS_KEY or "").strip()
        if not configured:
            return True
        if not isinstance(candidate, str):
            return False
        return candidate.strip() == configured

    def build_message_payload(row, user):
        return richer_message_payload(
            get_db,
            row,
            user,
            user["id"] if user else None,
        )

    def build_room_helper_call_id(room) -> str:
        return f"room-{room['slug']}"

    def build_channel_helper_call_id(channel) -> str:
        return f"channel-{channel['team_slug']}--{channel['slug']}"

    def get_or_create_group_call_session(
        scope_kind: str,
        scope_id: int,
        scope_slug: str,
        helper_call_id: str,
        created_by_user_id: int,
    ):
        with get_db() as db:
            existing = db.execute(
                """
                SELECT
                    id,
                    scope_kind,
                    scope_id,
                    scope_slug,
                    helper_call_id,
                    created_by_user_id,
                    status,
                    created_at,
                    ended_at
                FROM group_call_sessions
                WHERE helper_call_id = ?
                LIMIT 1
                """,
                (helper_call_id,),
            ).fetchone()

        if existing and existing["status"] == "active":
            return existing

        with get_db() as db:
            db.execute("BEGIN IMMEDIATE")
            existing = db.execute(
                """
                SELECT
                    id,
                    scope_kind,
                    scope_id,
                    scope_slug,
                    helper_call_id,
                    created_by_user_id,
                    status,
                    created_at,
                    ended_at
                FROM group_call_sessions
                WHERE helper_call_id = ?
                LIMIT 1
                """,
                (helper_call_id,),
            ).fetchone()

            if existing and existing["status"] == "active":
                db.commit()
                return existing

            if existing:
                db.execute(
                    """
                    UPDATE group_call_sessions
                    SET
                        status = 'active',
                        ended_at = NULL,
                        created_at = ?
                    WHERE id = ?
                    """,
                    (now_iso(), existing["id"]),
                )
                db.commit()
                refreshed = db.execute(
                    """
                    SELECT
                        id,
                        scope_kind,
                        scope_id,
                        scope_slug,
                        helper_call_id,
                        created_by_user_id,
                        status,
                        created_at,
                        ended_at
                    FROM group_call_sessions
                    WHERE id = ?
                    """,
                    (existing["id"],),
                ).fetchone()
                return refreshed

            cursor = db.execute(
                """
                INSERT INTO group_call_sessions (
                    scope_kind,
                    scope_id,
                    scope_slug,
                    helper_call_id,
                    created_by_user_id,
                    status,
                    created_at,
                    ended_at
                ) VALUES (?, ?, ?, ?, ?, 'active', ?, NULL)
                """,
                (
                    scope_kind,
                    scope_id,
                    scope_slug,
                    helper_call_id,
                    created_by_user_id,
                    now_iso(),
                ),
            )
            session_id = cursor.lastrowid
            db.commit()
            created = db.execute(
                """
                SELECT
                    id,
                    scope_kind,
                    scope_id,
                    scope_slug,
                    helper_call_id,
                    created_by_user_id,
                    status,
                    created_at,
                    ended_at
                FROM group_call_sessions
                WHERE id = ?
                """,
                (session_id,),
            ).fetchone()
            return created

    def group_call_session_payload(row):
        return {
            "id": row["id"],
            "scope_kind": row["scope_kind"],
            "scope_id": row["scope_id"],
            "scope_slug": row["scope_slug"],
            "helper_call_id": row["helper_call_id"],
            "created_by_user_id": row["created_by_user_id"],
            "status": row["status"],
            "created_at": row["created_at"],
            "ended_at": row["ended_at"],
        }

    @app.route("/api/attachments", methods=["POST"])
    def upload_attachment():
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        limited = rate_limited("attachment")
        if limited:
            return limited

        ctx["maybe_cleanup_orphaned_attachments"]()

        uploaded = request.files.get("attachment")
        if not uploaded or not uploaded.filename:
            return jsonify({"error": "Choose a file to upload."}), 400

        original_name = ctx["secure_filename"](uploaded.filename)[: constants["MAX_ATTACHMENT_NAME_LENGTH"]]
        if not original_name or "." not in original_name:
            return jsonify({"error": "Attachment must have an allowed file extension."}), 400

        extension = original_name.rsplit(".", 1)[1].lower()
        if extension not in constants["ALLOWED_ATTACHMENT_EXTENSIONS"]:
            return jsonify({"error": "This file type is not allowed."}), 400

        declared_mime = (uploaded.mimetype or "").lower()
        guessed_mime = ctx["mimetypes"].guess_type(original_name)[0] or ""

        if declared_mime and declared_mime not in constants["ALLOWED_ATTACHMENT_MIME_TYPES"]:
            return jsonify({"error": "This file MIME type is not allowed."}), 400

        if guessed_mime and guessed_mime not in constants["ALLOWED_ATTACHMENT_MIME_TYPES"]:
            return jsonify({"error": "This file extension is not allowed."}), 400

        content = uploaded.stream.read(constants["MAX_ATTACHMENT_BYTES"] + 1)
        if not content:
            return jsonify({"error": "Attachment cannot be empty."}), 400

        if len(content) > constants["MAX_ATTACHMENT_BYTES"]:
            return jsonify({"error": "Attachments must be 100 MB or smaller."}), 400

        stored_name = f"{ctx['secrets'].token_urlsafe(24)}.{extension}"
        destination = ctx["UPLOAD_DIR"] / stored_name

        try:
            destination.write_bytes(content)
            ctx["os"].chmod(destination, 0o600)
        except OSError:
            app.logger.exception("Could not save attachment.")
            return jsonify({"error": "Could not save attachment."}), 500

        mime_type = declared_mime or guessed_mime or "application/octet-stream"

        try:
            cursor = db_write(
                """
                INSERT INTO attachments (
                    uploader_user_id,
                    stored_name,
                    original_name,
                    mime_type,
                    size_bytes,
                    created_at
                ) VALUES (?, ?, ?, ?, ?, ?)
                """,
                (user["id"], stored_name, original_name, mime_type, len(content), now_iso()),
            )
        except Exception:
            destination.unlink(missing_ok=True)
            raise

        audit("attachment_uploaded", user["id"], "attachment", str(cursor.lastrowid), original_name)

        return jsonify(
            {
                "attachment": {
                    "id": cursor.lastrowid,
                    "original_name": original_name,
                    "mime_type": mime_type,
                    "size_bytes": len(content),
                    "url": f"/api/attachments/{cursor.lastrowid}/download",
                }
            }
        ), 201

    @app.route("/api/attachments/<int:attachment_id>/download")
    def download_attachment(attachment_id: int):
        user, error = require_user()
        if error:
            return error

        return send_attachment_file(
            get_db,
            ctx["UPLOAD_DIR"],
            attachment_id,
            user,
            can_access_room,
            can_access_channel,
            get_channel_by_id,
        )

    @app.route("/api/rooms/<string:slug>/messages")
    def get_room_messages(slug: str):
        room = get_room_by_slug(slug)
        if not room:
            return jsonify({"error": "Room not found."}), 404

        user = current_user()
        if not can_access_room(room, user):
            if not user:
                return jsonify({"error": "Login required for this private room.", "code": "login_required"}), 401
            return jsonify({"error": "You do not have access to this room.", "code": "forbidden"}), 403

        after_raw = request.args.get("after_id", "").strip()
        after_id = 0
        if after_raw:
            try:
                after_id = max(0, int(after_raw))
            except ValueError:
                return jsonify({"error": "after_id must be a non-negative integer."}), 400

        with get_db() as db:
            if after_raw:
                rows = db.execute(
                    f"""
                    {MESSAGE_SELECT}
                    WHERE m.room_id = ?
                      AND m.parent_message_id IS NULL
                      AND m.direct_conversation_id IS NULL
                      AND m.id > ?
                    ORDER BY m.id ASC
                    LIMIT ?
                    """,
                    (room["id"], after_id, MAX_INCREMENTAL_MESSAGES),
                ).fetchall()
            else:
                rows = db.execute(
                    f"""
                    SELECT *
                    FROM (
                        {MESSAGE_SELECT}
                        WHERE m.room_id = ?
                          AND m.parent_message_id IS NULL
                          AND m.direct_conversation_id IS NULL
                        ORDER BY m.id DESC
                        LIMIT ?
                    )
                    ORDER BY id ASC
                    """,
                    (room["id"], MAX_HISTORY_MESSAGES),
                ).fetchall()

        messages = [build_message_payload(row, user) for row in rows]
        latest_id = messages[-1]["id"] if messages else after_id
        return jsonify(
            {
                "messages": messages,
                "latest_id": latest_id,
                "has_more": len(messages) == MAX_INCREMENTAL_MESSAGES,
            }
        )

    @app.route("/api/rooms/<string:slug>/messages", methods=["POST"])
    def send_room_message(slug: str):
        limited = rate_limited("send")
        if limited:
            return limited

        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        room = get_room_by_slug(slug)
        if not room:
            return jsonify({"error": "Room not found."}), 404

        if not can_write_room(room, user):
            return jsonify({"error": "You cannot post in this room.", "code": "forbidden"}), 403

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        text = data.get("text", "")
        if not isinstance(text, str):
            return jsonify({"error": "Message must be text."}), 400

        text = text.strip()
        if len(text) > MAX_MESSAGE_LENGTH:
            return jsonify({"error": "Message is too long."}), 400

        attachment_ids, attachment_error = resolve_owned_attachment_ids(
            get_db,
            data.get("attachment_ids", []),
            user["id"],
        )
        if attachment_error:
            return attachment_error

        if not attachment_ids and data.get("attachment_id") is not None:
            single_attachment_id, single_attachment_error = resolve_owned_attachment(
                get_db,
                data.get("attachment_id"),
                user["id"],
            )
            if single_attachment_error:
                return single_attachment_error
            attachment_ids = [single_attachment_id] if single_attachment_id else []

        attachment_id = attachment_ids[0] if attachment_ids else None

        if not text and not attachment_id:
            return jsonify({"error": "Message cannot be empty."}), 400

        channel = get_channel_by_legacy_room_id(room["id"])
        channel_id = channel["id"] if channel else None

        with get_db() as db:
            db.execute("BEGIN IMMEDIATE")
            cursor = db.execute(
                """
                INSERT INTO messages (
                    room_id,
                    channel_id,
                    direct_conversation_id,
                    user_id,
                    username,
                    text,
                    attachment_id,
                    parent_message_id,
                    created_at
                ) VALUES (?, ?, NULL, ?, ?, ?, ?, NULL, ?)
                """,
                (
                    room["id"],
                    channel_id,
                    user["id"],
                    user["username"],
                    text,
                    attachment_id,
                    now_iso(),
                ),
            )
            message_id = cursor.lastrowid

            for position, extra_attachment_id in enumerate(attachment_ids):
                db.execute(
                    """
                    INSERT OR IGNORE INTO message_attachments (
                        message_id,
                        attachment_id,
                        position
                    ) VALUES (?, ?, ?)
                    """,
                    (message_id, extra_attachment_id, position),
                )

            db.commit()

        with get_db() as db:
            row = db.execute(f"{MESSAGE_SELECT} WHERE m.id = ?", (message_id,)).fetchone()

        audit("room_message_sent", user["id"], "room", str(room["id"]), str(message_id))
        return jsonify({"message": build_message_payload(row, user)}), 201

    @app.route("/api/rooms/<string:slug>/messages", methods=["DELETE"])
    def clear_room_messages(slug: str):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        moderator, error = require_moderator()
        if error:
            return error

        limited = rate_limited("admin")
        if limited:
            return limited

        room = get_room_by_slug(slug)
        if not room:
            return jsonify({"error": "Room not found."}), 404

        cursor = db_write(
            """
            DELETE FROM messages
            WHERE room_id = ?
              AND direct_conversation_id IS NULL
            """,
            (room["id"],),
        )

        audit("room_messages_cleared", moderator["id"], "room", str(room["id"]), str(cursor.rowcount))
        return jsonify({"status": "cleared", "deleted_messages": cursor.rowcount})

    @app.route("/api/teams/<string:team_slug>/channels/<string:channel_slug>/messages")
    def api_get_channel_messages(team_slug: str, channel_slug: str):
        channel = get_channel_by_slug(team_slug, channel_slug)
        if not channel:
            return jsonify({"error": "Channel not found."}), 404

        user = current_user()
        if not can_access_channel(channel, user):
            if not user:
                return jsonify({"error": "Login required.", "code": "login_required"}), 401
            return jsonify({"error": "You do not have access to this channel.", "code": "forbidden"}), 403

        after_raw = request.args.get("after_id", "").strip()
        after_id = 0
        if after_raw:
            try:
                after_id = max(0, int(after_raw))
            except ValueError:
                return jsonify({"error": "after_id must be a non-negative integer."}), 400

        with get_db() as db:
            if after_raw:
                rows = db.execute(
                    f"""
                    {MESSAGE_SELECT}
                    WHERE m.channel_id = ?
                      AND m.parent_message_id IS NULL
                      AND m.direct_conversation_id IS NULL
                      AND m.id > ?
                    ORDER BY m.id ASC
                    LIMIT ?
                    """,
                    (channel["id"], after_id, MAX_INCREMENTAL_MESSAGES),
                ).fetchall()
            else:
                rows = db.execute(
                    f"""
                    SELECT *
                    FROM (
                        {MESSAGE_SELECT}
                        WHERE m.channel_id = ?
                          AND m.parent_message_id IS NULL
                          AND m.direct_conversation_id IS NULL
                        ORDER BY m.id DESC
                        LIMIT ?
                    )
                    ORDER BY id ASC
                    """,
                    (channel["id"], MAX_HISTORY_MESSAGES),
                ).fetchall()

        messages = [build_message_payload(row, user) for row in rows]
        latest_id = messages[-1]["id"] if messages else after_id

        return jsonify(
            {
                "team": {"slug": channel["team_slug"], "name": channel["team_name"]},
                "channel": channel_payload(channel, user),
                "messages": messages,
                "latest_id": latest_id,
                "has_more": len(messages) == MAX_INCREMENTAL_MESSAGES,
            }
        )

    @app.route("/api/teams/<string:team_slug>/channels/<string:channel_slug>/messages", methods=["POST"])
    def api_send_channel_message(team_slug: str, channel_slug: str):
        limited = rate_limited("send")
        if limited:
            return limited

        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        channel = get_channel_by_slug(team_slug, channel_slug)
        if not channel:
            return jsonify({"error": "Channel not found."}), 404

        if not can_write_channel(channel, user):
            return jsonify({"error": "You cannot post in this channel.", "code": "forbidden"}), 403

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        text = data.get("text", "")
        if not isinstance(text, str):
            return jsonify({"error": "Message must be text."}), 400

        text = text.strip()
        if len(text) > MAX_MESSAGE_LENGTH:
            return jsonify({"error": "Message is too long."}), 400

        attachment_ids, attachment_error = resolve_owned_attachment_ids(
            get_db,
            data.get("attachment_ids", []),
            user["id"],
        )
        if attachment_error:
            return attachment_error

        if not attachment_ids and data.get("attachment_id") is not None:
            single_attachment_id, single_attachment_error = resolve_owned_attachment(
                get_db,
                data.get("attachment_id"),
                user["id"],
            )
            if single_attachment_error:
                return single_attachment_error
            attachment_ids = [single_attachment_id] if single_attachment_id else []

        attachment_id = attachment_ids[0] if attachment_ids else None

        if not text and not attachment_id:
            return jsonify({"error": "Message cannot be empty."}), 400

        with get_db() as db:
            db.execute("BEGIN IMMEDIATE")
            cursor = db.execute(
                """
                INSERT INTO messages (
                    room_id,
                    channel_id,
                    direct_conversation_id,
                    user_id,
                    username,
                    text,
                    attachment_id,
                    parent_message_id,
                    created_at
                ) VALUES (?, ?, NULL, ?, ?, ?, ?, NULL, ?)
                """,
                (
                    channel["legacy_room_id"],
                    channel["id"],
                    user["id"],
                    user["username"],
                    text,
                    attachment_id,
                    now_iso(),
                ),
            )
            message_id = cursor.lastrowid

            for position, extra_attachment_id in enumerate(attachment_ids):
                db.execute(
                    """
                    INSERT OR IGNORE INTO message_attachments (
                        message_id,
                        attachment_id,
                        position
                    ) VALUES (?, ?, ?)
                    """,
                    (message_id, extra_attachment_id, position),
                )

            db.commit()

        with get_db() as db:
            row = db.execute(f"{MESSAGE_SELECT} WHERE m.id = ?", (message_id,)).fetchone()

        audit("channel_message_sent", user["id"], "channel", str(channel["id"]), str(message_id))
        return jsonify({"message": build_message_payload(row, user)}), 201

    @app.route("/api/messages/<int:message_id>", methods=["PATCH"])
    def edit_message(message_id: int):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        limited = rate_limited("send")
        if limited:
            return limited

        user, error = require_writable_user()
        if error:
            return error

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        new_text = data.get("text", "")
        if not isinstance(new_text, str):
            return jsonify({"error": "Message must be text."}), 400

        new_text = new_text.strip()

        attachment_ids, attachment_error = resolve_owned_attachment_ids(
            get_db,
            data.get("attachment_ids", []),
            user["id"],
        )
        if attachment_error:
            return attachment_error

        if len(new_text) > MAX_MESSAGE_LENGTH:
            return jsonify({"error": "Message is too long."}), 400

        if not new_text and not attachment_ids:
            return jsonify({"error": "Message cannot be empty."}), 400

        with get_db() as db:
            message = db.execute(f"{MESSAGE_SELECT} WHERE m.id = ?", (message_id,)).fetchone()

        if not message:
            return jsonify({"error": "Message not found."}), 404

        is_owner = message["user_id"] == user["id"]
        is_moderator = user["role"] in {"moderator", "admin"}
        if not is_owner and not is_moderator:
            return jsonify({"error": "You cannot edit this message.", "code": "forbidden"}), 403

        created_at = parse_iso(message["created_at"])
        if is_owner and created_at:
            age = (utc_now() - created_at).total_seconds()
            if age > MAX_MESSAGE_EDIT_WINDOW_SECONDS:
                return jsonify({"error": "Edit window has expired."}), 403

        old_attachments = ctx["attachment_list_for_message"](message_id)

        with get_db() as db:
            db.execute("BEGIN IMMEDIATE")
            db.execute(
                "UPDATE messages SET text = ?, attachment_id = ? WHERE id = ?",
                (new_text, attachment_ids[0] if attachment_ids else None, message_id),
            )
            db.execute("DELETE FROM message_attachments WHERE message_id = ?", (message_id,))

            for position, attachment_id in enumerate(attachment_ids):
                db.execute(
                    """
                    INSERT INTO message_attachments (
                        message_id,
                        attachment_id,
                        position
                    ) VALUES (?, ?, ?)
                    """,
                    (message_id, attachment_id, position),
                )

            db.execute(
                """
                INSERT INTO message_edits (
                    message_id,
                    edited_by_user_id,
                    previous_text,
                    new_text,
                    previous_attachments_json,
                    new_attachments_json,
                    created_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    message_id,
                    user["id"],
                    message["text"],
                    new_text,
                    json.dumps(old_attachments),
                    json.dumps(attachment_ids),
                    now_iso(),
                ),
            )
            db.commit()

        with get_db() as db:
            updated = db.execute(f"{MESSAGE_SELECT} WHERE m.id = ?", (message_id,)).fetchone()

        audit("message_edited", user["id"], "message", str(message_id))
        return jsonify({"message": build_message_payload(updated, user)})

    @app.route("/api/messages/<int:message_id>", methods=["DELETE"])
    def delete_message(message_id: int):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        moderator, error = require_moderator()
        if error:
            return error

        limited = rate_limited("admin")
        if limited:
            return limited

        cursor = db_write("DELETE FROM messages WHERE id = ?", (message_id,))
        if cursor.rowcount == 0:
            return jsonify({"error": "Message not found."}), 404

        audit("message_deleted", moderator["id"], "message", str(message_id))
        return jsonify({"status": "deleted"})

    @app.route("/api/messages/history")
    def api_message_history():
        user = current_user()
        kind = str(request.args.get("kind", "")).strip()
        slug = str(request.args.get("slug", "")).strip()
        team_slug = str(request.args.get("team_slug", "")).strip()
        before_id_raw = str(request.args.get("before_id", "")).strip()

        try:
            before_id = int(before_id_raw)
        except ValueError:
            return jsonify({"error": "before_id is required."}), 400

        if kind == "room":
            room = get_room_by_slug(slug)
            if not room or not can_access_room(room, user):
                return jsonify({"error": "Room not found."}), 404

            with get_db() as db:
                rows = db.execute(
                    f"""
                    SELECT *
                    FROM (
                        {MESSAGE_SELECT}
                        WHERE m.room_id = ?
                          AND m.parent_message_id IS NULL
                          AND m.direct_conversation_id IS NULL
                          AND m.id < ?
                        ORDER BY m.id DESC
                        LIMIT ?
                    )
                    ORDER BY id ASC
                    """,
                    (room["id"], before_id, MAX_HISTORY_MESSAGES),
                ).fetchall()

        elif kind == "channel":
            channel = get_channel_by_slug(team_slug, slug)
            if not channel or not can_access_channel(channel, user):
                return jsonify({"error": "Channel not found."}), 404

            with get_db() as db:
                rows = db.execute(
                    f"""
                    SELECT *
                    FROM (
                        {MESSAGE_SELECT}
                        WHERE m.channel_id = ?
                          AND m.parent_message_id IS NULL
                          AND m.direct_conversation_id IS NULL
                          AND m.id < ?
                        ORDER BY m.id DESC
                        LIMIT ?
                    )
                    ORDER BY id ASC
                    """,
                    (channel["id"], before_id, MAX_HISTORY_MESSAGES),
                ).fetchall()

        elif kind == "dm":
            dm_id_raw = str(request.args.get("dm_id", "")).strip()
            try:
                dm_id = int(dm_id_raw)
            except ValueError:
                return jsonify({"error": "dm_id is required for DM history."}), 400

            if not validate_dm_access(get_db, dm_id, user):
                return jsonify({"error": "DM not found."}), 404

            with get_db() as db:
                rows = db.execute(
                    f"""
                    SELECT *
                    FROM (
                        {MESSAGE_SELECT}
                        WHERE m.direct_conversation_id = ?
                          AND m.parent_message_id IS NULL
                          AND m.id < ?
                        ORDER BY m.id DESC
                        LIMIT ?
                    )
                    ORDER BY id ASC
                    """,
                    (dm_id, before_id, MAX_HISTORY_MESSAGES),
                ).fetchall()
        else:
            return jsonify({"error": "Invalid history kind."}), 400

        messages = [build_message_payload(row, user) for row in rows]
        return jsonify({"messages": messages, "has_more": len(messages) == MAX_HISTORY_MESSAGES})

    @app.route("/api/messages/<int:message_id>/thread")
    def api_get_message_thread(message_id: int):
        user = current_user()

        with get_db() as db:
            root = db.execute(
                f"""
                {MESSAGE_SELECT}
                WHERE m.id = ?
                  AND m.parent_message_id IS NULL
                """,
                (message_id,),
            ).fetchone()

        if not root:
            return jsonify({"error": "Thread root message not found."}), 404

        if root["direct_conversation_id"]:
            if not validate_dm_access(get_db, root["direct_conversation_id"], user):
                if not user:
                    return jsonify({"error": "Login required.", "code": "login_required"}), 401
                return jsonify({"error": "You do not have access to this thread.", "code": "forbidden"}), 403
        else:
            channel = None
            if root["channel_id"]:
                channel = get_channel_by_id(root["channel_id"])
            if not channel and root["room_id"]:
                channel = get_channel_by_legacy_room_id(root["room_id"])

            if not channel or not can_access_channel(channel, user):
                if not user:
                    return jsonify({"error": "Login required.", "code": "login_required"}), 401
                return jsonify({"error": "You do not have access to this thread.", "code": "forbidden"}), 403

        with get_db() as db:
            replies = db.execute(
                f"""
                {MESSAGE_SELECT}
                WHERE m.parent_message_id = ?
                ORDER BY m.id ASC
                LIMIT ?
                """,
                (message_id, MAX_THREAD_REPLIES),
            ).fetchall()

        payload = {
            "root": build_message_payload(root, user),
            "replies": [build_message_payload(reply, user) for reply in replies],
        }

        if root["direct_conversation_id"]:
            payload["dm"] = dm_payload(get_db, root["direct_conversation_id"], user["id"])
        else:
            channel = (
                get_channel_by_id(root["channel_id"])
                if root["channel_id"]
                else get_channel_by_legacy_room_id(root["room_id"])
            )
            payload["team"] = {"slug": channel["team_slug"], "name": channel["team_name"]}
            payload["channel"] = channel_payload(channel, user)

        return jsonify(payload)

    @app.route("/api/messages/<int:message_id>/thread", methods=["POST"])
    def api_reply_to_thread(message_id: int):
        limited = rate_limited("send")
        if limited:
            return limited

        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        with get_db() as db:
            root = db.execute(
                """
                SELECT
                    id,
                    room_id,
                    channel_id,
                    direct_conversation_id,
                    parent_message_id
                FROM messages
                WHERE id = ?
                """,
                (message_id,),
            ).fetchone()

        if not root or root["parent_message_id"] is not None:
            return jsonify({"error": "Thread root message not found."}), 404

        if root["direct_conversation_id"]:
            if not validate_dm_access(get_db, root["direct_conversation_id"], user):
                return jsonify({"error": "You cannot reply in this thread.", "code": "forbidden"}), 403
        else:
            channel = None
            if root["channel_id"]:
                channel = get_channel_by_id(root["channel_id"])
            if not channel and root["room_id"]:
                channel = get_channel_by_legacy_room_id(root["room_id"])
            if not channel or not can_write_channel(channel, user):
                return jsonify({"error": "You cannot reply in this thread.", "code": "forbidden"}), 403

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        text = data.get("text", "")
        if not isinstance(text, str):
            return jsonify({"error": "Reply must be text."}), 400

        text = text.strip()
        if len(text) > MAX_MESSAGE_LENGTH:
            return jsonify({"error": "Reply is too long."}), 400

        attachment_ids, attachment_error = resolve_owned_attachment_ids(
            get_db,
            data.get("attachment_ids", []),
            user["id"],
        )
        if attachment_error:
            return attachment_error

        if not attachment_ids and data.get("attachment_id") is not None:
            single_attachment_id, single_attachment_error = resolve_owned_attachment(
                get_db,
                data.get("attachment_id"),
                user["id"],
            )
            if single_attachment_error:
                return single_attachment_error
            attachment_ids = [single_attachment_id] if single_attachment_id else []

        attachment_id = attachment_ids[0] if attachment_ids else None

        if not text and not attachment_id:
            return jsonify({"error": "Reply cannot be empty."}), 400

        with get_db() as db:
            db.execute("BEGIN IMMEDIATE")
            cursor = db.execute(
                """
                INSERT INTO messages (
                    room_id,
                    channel_id,
                    direct_conversation_id,
                    user_id,
                    username,
                    text,
                    attachment_id,
                    parent_message_id,
                    created_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    root["room_id"],
                    root["channel_id"],
                    root["direct_conversation_id"],
                    user["id"],
                    user["username"],
                    text,
                    attachment_id,
                    root["id"],
                    now_iso(),
                ),
            )
            reply_id = cursor.lastrowid

            for position, extra_attachment_id in enumerate(attachment_ids):
                db.execute(
                    """
                    INSERT OR IGNORE INTO message_attachments (
                        message_id,
                        attachment_id,
                        position
                    ) VALUES (?, ?, ?)
                    """,
                    (reply_id, extra_attachment_id, position),
                )

            db.commit()

        with get_db() as db:
            reply = db.execute(f"{MESSAGE_SELECT} WHERE m.id = ?", (reply_id,)).fetchone()

        audit("thread_reply_sent", user["id"], "message", str(message_id), str(reply_id))
        return jsonify({"reply": build_message_payload(reply, user)}), 201

    @app.route("/api/feed")
    def get_feed():
        user = current_user()
        after_raw = request.args.get("after_id", "").strip()
        after_id = 0

        if after_raw:
            try:
                after_id = max(0, int(after_raw))
            except ValueError:
                return jsonify({"error": "after_id must be a non-negative integer."}), 400

        base_select = f"""
            SELECT
                m.id,
                m.room_id,
                m.channel_id,
                m.direct_conversation_id,
                m.user_id,
                m.username,
                m.text,
                m.created_at,
                m.attachment_id,
                m.parent_message_id,
                (
                    SELECT COUNT(*)
                    FROM messages AS thread_reply
                    WHERE thread_reply.parent_message_id = m.id
                ) AS thread_reply_count,
                r.slug AS room_slug,
                r.name AS room_name,
                r.visibility AS room_visibility,
                c.slug AS channel_slug,
                c.name AS channel_name,
                t.slug AS team_slug,
                t.name AS team_name,
                a.original_name AS attachment_original_name,
                a.mime_type AS attachment_mime_type,
                a.size_bytes AS attachment_size_bytes,
                u.role AS user_role,
                us.name_color AS user_name_color
            FROM messages AS m
            JOIN users AS u
                ON u.id = m.user_id
            LEFT JOIN user_settings AS us
                ON us.user_id = u.id
            LEFT JOIN rooms AS r
                ON r.id = m.room_id
            LEFT JOIN channels AS c
                ON c.id = m.channel_id
               AND c.is_active = 1
            LEFT JOIN teams AS t
                ON t.id = c.team_id
               AND t.is_active = 1
            LEFT JOIN attachments AS a
                ON a.id = m.attachment_id
        """

        params = []
        if user and user["role"] in {"moderator", "admin"}:
            access_clause = """
                (
                    (m.direct_conversation_id IS NULL AND r.id IS NOT NULL AND r.is_active = 1)
                    OR
                    (m.direct_conversation_id IS NULL AND c.id IS NOT NULL AND c.is_active = 1 AND t.is_active = 1)
                    OR
                    (
                        m.direct_conversation_id IS NOT NULL
                        AND EXISTS (
                            SELECT 1
                            FROM direct_conversation_members AS dcm
                            WHERE dcm.direct_conversation_id = m.direct_conversation_id
                              AND dcm.user_id = ?
                        )
                    )
                )
            """
            params.append(user["id"])
        elif user:
            access_clause = """
                (
                    (
                        m.direct_conversation_id IS NULL
                        AND r.id IS NOT NULL
                        AND r.is_active = 1
                        AND (
                            r.visibility = 'public'
                            OR EXISTS (
                                SELECT 1
                                FROM room_members AS rm
                                WHERE rm.room_id = r.id
                                  AND rm.user_id = ?
                            )
                        )
                    )
                    OR
                    (
                        m.direct_conversation_id IS NULL
                        AND c.id IS NOT NULL
                        AND c.is_active = 1
                        AND t.is_active = 1
                        AND c.legacy_room_id IS NULL
                        AND EXISTS (
                            SELECT 1
                            FROM team_members AS tm
                            WHERE tm.team_id = c.team_id
                              AND tm.user_id = ?
                        )
                    )
                    OR
                    (
                        m.direct_conversation_id IS NOT NULL
                        AND EXISTS (
                            SELECT 1
                            FROM direct_conversation_members AS dcm
                            WHERE dcm.direct_conversation_id = m.direct_conversation_id
                              AND dcm.user_id = ?
                        )
                    )
                )
            """
            params.extend([user["id"], user["id"], user["id"]])
        else:
            access_clause = """
                m.direct_conversation_id IS NULL
                AND r.id IS NOT NULL
                AND r.is_active = 1
                AND r.visibility = 'public'
            """

        if after_raw:
            query = f"""
                {base_select}
                WHERE m.parent_message_id IS NULL
                  AND {access_clause}
                  AND m.id > ?
                ORDER BY m.id ASC
                LIMIT ?
            """
            query_params = [*params, after_id, MAX_FEED_MESSAGES]

            with get_db() as db:
                rows = db.execute(query, tuple(query_params)).fetchall()

            entries = [feed_message_payload(get_db, row, user) for row in rows]
            latest_id = entries[-1]["id"] if entries else after_id
            return jsonify(
                {
                    "entries": entries,
                    "latest_id": latest_id,
                    "has_more": len(entries) == MAX_FEED_MESSAGES,
                }
            )

        query = f"""
            SELECT *
            FROM (
                {base_select}
                WHERE m.parent_message_id IS NULL
                  AND {access_clause}
                ORDER BY m.id DESC
                LIMIT ?
            )
            ORDER BY id ASC
        """
        query_params = [*params, MAX_FEED_MESSAGES]

        with get_db() as db:
            rows = db.execute(query, tuple(query_params)).fetchall()

        entries = [feed_message_payload(get_db, row, user) for row in rows]
        latest_id = entries[-1]["id"] if entries else 0
        return jsonify({"entries": entries, "latest_id": latest_id, "has_more": False})

    @app.route("/api/search")
    def api_search():
        user = current_user()
        query = str(request.args.get("q", "")).strip()

        if not query:
            return jsonify({"results": [], "query": ""})

        if len(query) > MAX_SEARCH_QUERY_LENGTH:
            return jsonify({"error": "Search query is too long."}), 400

        pattern = f"%{query.lower()}%"

        with get_db() as db:
            rows = db.execute(
                """
                SELECT
                    m.id,
                    m.room_id,
                    m.channel_id,
                    m.direct_conversation_id,
                    m.user_id,
                    m.username,
                    m.text,
                    m.created_at,
                    m.attachment_id,
                    m.parent_message_id,
                    (
                        SELECT COUNT(*)
                        FROM messages AS thread_reply
                        WHERE thread_reply.parent_message_id = m.id
                    ) AS thread_reply_count,
                    r.slug AS room_slug,
                    r.name AS room_name,
                    r.visibility AS room_visibility,
                    c.slug AS channel_slug,
                    c.name AS channel_name,
                    t.slug AS team_slug,
                    t.name AS team_name,
                    a.original_name AS attachment_original_name,
                    a.mime_type AS attachment_mime_type,
                    a.size_bytes AS attachment_size_bytes,
                    u.role AS user_role,
                    us.name_color AS user_name_color
                FROM messages AS m
                JOIN users AS u
                    ON u.id = m.user_id
                LEFT JOIN user_settings AS us
                    ON us.user_id = u.id
                LEFT JOIN rooms AS r
                    ON r.id = m.room_id
                LEFT JOIN channels AS c
                    ON c.id = m.channel_id
                LEFT JOIN teams AS t
                    ON t.id = c.team_id
                LEFT JOIN attachments AS a
                    ON a.id = m.attachment_id
                WHERE LOWER(m.text) LIKE ?
                ORDER BY m.id DESC
                LIMIT ?
                """,
                (pattern, MAX_SEARCH_RESULTS),
            ).fetchall()

        visible = []
        for row in rows:
            allowed = False
            if row["direct_conversation_id"]:
                allowed = bool(user and validate_dm_access(get_db, row["direct_conversation_id"], user))
            elif row["channel_id"]:
                channel = get_channel_by_id(row["channel_id"])
                allowed = bool(channel and can_access_channel(channel, user))
            elif row["room_id"]:
                room = room_by_id(row["room_id"])
                allowed = bool(room and can_access_room(room, user))

            if allowed:
                visible.append(feed_message_payload(get_db, row, user))

        return jsonify({"query": query, "results": visible})

    @app.route("/api/read-receipts", methods=["POST"])
    def api_mark_read_receipts():
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_user()
        if error:
            return error

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        ids = data.get("message_ids", [])
        if not isinstance(ids, list):
            return jsonify({"error": "message_ids must be an array."}), 400

        parsed_ids = []
        for item in ids[:READ_RECEIPT_BATCH_LIMIT]:
            try:
                parsed_ids.append(int(item))
            except (TypeError, ValueError):
                continue

        if not parsed_ids:
            return jsonify({"status": "ok", "updated": 0})

        updated = 0
        with get_db() as db:
            db.execute("BEGIN IMMEDIATE")

            for message_id in parsed_ids:
                row = db.execute(
                    """
                    SELECT id, room_id, channel_id, direct_conversation_id
                    FROM messages
                    WHERE id = ?
                    """,
                    (message_id,),
                ).fetchone()

                if not row:
                    continue

                allowed = False
                if row["direct_conversation_id"]:
                    allowed = validate_dm_access(get_db, row["direct_conversation_id"], user)
                elif row["channel_id"]:
                    channel = get_channel_by_id(row["channel_id"])
                    allowed = bool(channel and can_access_channel(channel, user))
                elif row["room_id"]:
                    room = room_by_id(row["room_id"])
                    allowed = bool(room and can_access_room(room, user))

                if not allowed:
                    continue

                db.execute(
                    """
                    INSERT INTO message_receipts (
                        message_id,
                        user_id,
                        seen_at
                    ) VALUES (?, ?, ?)
                    ON CONFLICT(message_id, user_id)
                    DO UPDATE SET seen_at = excluded.seen_at
                    """,
                    (message_id, user["id"], now_iso()),
                )
                updated += 1

            db.commit()

        return jsonify({"status": "ok", "updated": updated})

    @app.route("/api/messages/<int:message_id>/receipts")
    def api_get_message_receipts(message_id: int):
        user, error = require_user()
        if error:
            return error

        with get_db() as db:
            message = db.execute(
                """
                SELECT id, room_id, channel_id, direct_conversation_id, user_id
                FROM messages
                WHERE id = ?
                """,
                (message_id,),
            ).fetchone()

        if not message:
            return jsonify({"error": "Message not found."}), 404

        allowed = False
        if message["direct_conversation_id"]:
            allowed = validate_dm_access(get_db, message["direct_conversation_id"], user)
        elif message["channel_id"]:
            channel = get_channel_by_id(message["channel_id"])
            allowed = bool(channel and can_access_channel(channel, user))
        elif message["room_id"]:
            room = room_by_id(message["room_id"])
            allowed = bool(room and can_access_room(room, user))

        if not allowed:
            return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403

        with get_db() as db:
            rows = db.execute(
                """
                SELECT
                    u.id,
                    u.username,
                    mr.seen_at
                FROM message_receipts AS mr
                JOIN users AS u
                    ON u.id = mr.user_id
                WHERE mr.message_id = ?
                ORDER BY mr.seen_at DESC, u.username COLLATE NOCASE
                """,
                (message_id,),
            ).fetchall()

        return jsonify(
            {
                "message_id": message_id,
                "receipts": [
                    {
                        "user_id": row["id"],
                        "username": row["username"],
                        "seen_at": row["seen_at"],
                    }
                    for row in rows
                ],
            }
        )

    @app.route("/api/messages/<int:message_id>/reactions")
    def api_get_message_reactions(message_id: int):
        user = current_user()

        with get_db() as db:
            row = db.execute(
                """
                SELECT
                    id,
                    room_id,
                    channel_id,
                    direct_conversation_id
                FROM messages
                WHERE id = ?
                """,
                (message_id,),
            ).fetchone()

        if not row:
            return jsonify({"error": "Message not found."}), 404

        allowed = False
        if row["direct_conversation_id"]:
            allowed = bool(user and validate_dm_access(get_db, row["direct_conversation_id"], user))
        elif row["channel_id"]:
            channel = get_channel_by_id(row["channel_id"])
            allowed = bool(channel and can_access_channel(channel, user))
        elif row["room_id"]:
            room = room_by_id(row["room_id"])
            allowed = bool(room and can_access_room(room, user))

        if not allowed:
            if not user:
                return jsonify({"error": "Login required.", "code": "login_required"}), 401
            return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403

        with get_db() as db:
            detail_rows = db.execute(
                """
                SELECT
                    mr.emoji,
                    u.id AS user_id,
                    u.username,
                    mr.created_at
                FROM message_reactions AS mr
                JOIN users AS u
                    ON u.id = mr.user_id
                WHERE mr.message_id = ?
                ORDER BY mr.emoji ASC, u.username COLLATE NOCASE
                """,
                (message_id,),
            ).fetchall()

        grouped = {}
        for item in detail_rows:
            grouped.setdefault(item["emoji"], []).append(
                {
                    "user_id": item["user_id"],
                    "username": item["username"],
                    "created_at": item["created_at"],
                }
            )

        summary = []
        for emoji, users in grouped.items():
            summary.append(
                {
                    "emoji": emoji,
                    "count": len(users),
                    "reacted": bool(user and any(x["user_id"] == user["id"] for x in users)),
                    "users": users,
                }
            )

        summary.sort(key=lambda x: (-x["count"], x["emoji"]))
        return jsonify({"message_id": message_id, "reactions": summary})

    @app.route("/api/messages/<int:message_id>/reactions", methods=["POST"])
    def api_add_reaction(message_id: int):
        limited = rate_limited("reaction")
        if limited:
            return limited

        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        emoji = str(data.get("emoji", "")).strip()
        if not emoji or len(emoji) > 16 or not REACTION_EMOJI_RE.fullmatch(emoji):
            return jsonify({"error": "Invalid reaction."}), 400

        with get_db() as db:
            row = db.execute(
                """
                SELECT
                    id,
                    room_id,
                    channel_id,
                    direct_conversation_id
                FROM messages
                WHERE id = ?
                """,
                (message_id,),
            ).fetchone()

        if not row:
            return jsonify({"error": "Message not found."}), 404

        allowed = False
        if row["direct_conversation_id"]:
            allowed = validate_dm_access(get_db, row["direct_conversation_id"], user)
        elif row["channel_id"]:
            channel = get_channel_by_id(row["channel_id"])
            allowed = bool(channel and can_access_channel(channel, user))
        elif row["room_id"]:
            room = room_by_id(row["room_id"])
            allowed = bool(room and can_access_room(room, user))

        if not allowed:
            return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403

        db_write(
            """
            INSERT OR IGNORE INTO message_reactions (
                message_id,
                user_id,
                emoji,
                created_at
            ) VALUES (?, ?, ?, ?)
            """,
            (message_id, user["id"], emoji, now_iso()),
        )

        audit("message_reaction_added", user["id"], "message", str(message_id), emoji)
        return jsonify(
            {
                "message_id": message_id,
                "reactions": reaction_summary_for_message(get_db, message_id, user),
            }
        ), 201

    @app.route("/api/messages/<int:message_id>/reactions/<path:emoji>", methods=["DELETE"])
    def api_remove_reaction(message_id: int, emoji: str):
        limited = rate_limited("reaction")
        if limited:
            return limited

        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        db_write(
            """
            DELETE FROM message_reactions
            WHERE message_id = ?
              AND user_id = ?
              AND emoji = ?
            """,
            (message_id, user["id"], emoji),
        )

        audit("message_reaction_removed", user["id"], "message", str(message_id), emoji)
        return jsonify(
            {
                "message_id": message_id,
                "reactions": reaction_summary_for_message(get_db, message_id, user),
            }
        )

    @app.route("/api/presence", methods=["POST"])
    def api_presence_ping():
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_user()
        if error:
            return error

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        view_name = str(data.get("view", "")).strip()[:32] or None
        room_id = data.get("room_id")
        channel_id = data.get("channel_id")
        dm_id = data.get("dm_id")

        try:
            room_id = int(room_id) if room_id is not None else None
        except (TypeError, ValueError):
            room_id = None

        try:
            channel_id = int(channel_id) if channel_id is not None else None
        except (TypeError, ValueError):
            channel_id = None

        try:
            dm_id = int(dm_id) if dm_id is not None else None
        except (TypeError, ValueError):
            dm_id = None

        update_presence(db_write, user["id"], view_name, room_id, channel_id, dm_id)
        return jsonify({"status": "ok"})

    @app.route("/api/rooms/<string:slug>/presence")
    def api_room_presence(slug: str):
        room = get_room_by_slug(slug)
        user = current_user()
        if not room or not can_access_room(room, user):
            return jsonify({"error": "Room not found."}), 404
        return jsonify(
            {
                "room_id": room["id"],
                "online_count": room_presence_count(get_db, PRESENCE_TTL_SECONDS, room["id"]),
            }
        )

    @app.route("/api/teams/<string:team_slug>/channels/<string:channel_slug>/presence")
    def api_channel_presence(team_slug: str, channel_slug: str):
        channel = get_channel_by_slug(team_slug, channel_slug)
        user = current_user()
        if not channel or not can_access_channel(channel, user):
            return jsonify({"error": "Channel not found."}), 404
        return jsonify(
            {
                "channel_id": channel["id"],
                "online_count": channel_presence_count(get_db, PRESENCE_TTL_SECONDS, channel["id"]),
            }
        )

    @app.route("/api/dms/<int:dm_id>/presence")
    def api_dm_presence(dm_id: int):
        user = current_user()
        if not validate_dm_access(get_db, dm_id, user):
            return jsonify({"error": "DM not found."}), 404
        return jsonify(
            {
                "dm_id": dm_id,
                "online_count": dm_presence_count(get_db, PRESENCE_TTL_SECONDS, dm_id),
            }
        )

    @app.route("/api/rooms/<string:slug>/typing", methods=["POST"])
    def set_typing(slug: str):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        limited = rate_limited("typing")
        if limited:
            return limited

        room = get_room_by_slug(slug)
        if not room or not can_write_room(room, user):
            return jsonify({"error": "You cannot use typing indicators here.", "code": "forbidden"}), 403

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        typing = bool(data.get("typing"))

        if typing:
            db_write(
                """
                INSERT INTO typing_events (
                    room_id,
                    user_id,
                    updated_at
                ) VALUES (?, ?, ?)
                ON CONFLICT(room_id, user_id)
                DO UPDATE SET updated_at = excluded.updated_at
                """,
                (room["id"], user["id"], time.time()),
            )
        else:
            db_write(
                "DELETE FROM typing_events WHERE room_id = ? AND user_id = ?",
                (room["id"], user["id"]),
            )

        return jsonify({"status": "ok"})

    @app.route("/api/rooms/<string:slug>/typing")
    def get_typing(slug: str):
        room = get_room_by_slug(slug)
        user = current_user()

        if not room or not can_access_room(room, user):
            return jsonify({"error": "Room not found."}), 404

        cutoff = time.time() - TYPING_TTL_SECONDS
        with get_db() as db:
            db.execute("DELETE FROM typing_events WHERE updated_at < ?", (cutoff,))
            rows = db.execute(
                """
                SELECT
                    u.id,
                    u.username
                FROM typing_events AS t
                JOIN users AS u
                    ON u.id = t.user_id
                WHERE t.room_id = ?
                  AND t.user_id != ?
                  AND t.updated_at >= ?
                ORDER BY u.username COLLATE NOCASE
                """,
                (room["id"], user["id"] if user else -1, cutoff),
            ).fetchall()
            db.commit()

        return jsonify({"users": [{"id": row["id"], "username": row["username"]} for row in rows]})

    @app.route("/api/teams/<string:team_slug>/channels/<string:channel_slug>/typing", methods=["POST"])
    def api_set_channel_typing(team_slug: str, channel_slug: str):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        limited = rate_limited("typing")
        if limited:
            return limited

        channel = get_channel_by_slug(team_slug, channel_slug)
        if not channel:
            return jsonify({"error": "Channel not found."}), 404

        if not can_write_channel(channel, user):
            return jsonify({"error": "You cannot use typing indicators here.", "code": "forbidden"}), 403

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        typing = bool(data.get("typing"))

        if typing:
            db_write(
                """
                INSERT INTO channel_typing_events (
                    channel_id,
                    user_id,
                    updated_at
                ) VALUES (?, ?, ?)
                ON CONFLICT(channel_id, user_id)
                DO UPDATE SET updated_at = excluded.updated_at
                """,
                (channel["id"], user["id"], time.time()),
            )
        else:
            db_write(
                "DELETE FROM channel_typing_events WHERE channel_id = ? AND user_id = ?",
                (channel["id"], user["id"]),
            )

        return jsonify({"status": "ok"})

    @app.route("/api/teams/<string:team_slug>/channels/<string:channel_slug>/typing")
    def api_get_channel_typing(team_slug: str, channel_slug: str):
        channel = get_channel_by_slug(team_slug, channel_slug)
        if not channel:
            return jsonify({"error": "Channel not found."}), 404

        user = current_user()
        if not can_access_channel(channel, user):
            if not user:
                return jsonify({"error": "Login required.", "code": "login_required"}), 401
            return jsonify({"error": "You do not have access to this channel.", "code": "forbidden"}), 403

        cutoff = time.time() - TYPING_TTL_SECONDS
        with get_db() as db:
            db.execute("DELETE FROM channel_typing_events WHERE updated_at < ?", (cutoff,))
            rows = db.execute(
                """
                SELECT
                    u.id,
                    u.username
                FROM channel_typing_events AS t
                JOIN users AS u
                    ON u.id = t.user_id
                WHERE t.channel_id = ?
                  AND t.user_id != ?
                  AND t.updated_at >= ?
                ORDER BY u.username COLLATE NOCASE
                """,
                (channel["id"], user["id"] if user else -1, cutoff),
            ).fetchall()
            db.commit()

        return jsonify({"users": [{"id": row["id"], "username": row["username"]} for row in rows]})

    @app.route("/api/dms/<int:dm_id>/typing", methods=["POST"])
    def api_set_dm_typing(dm_id: int):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        limited = rate_limited("typing")
        if limited:
            return limited

        if not validate_dm_access(get_db, dm_id, user):
            return jsonify({"error": "DM not found."}), 404

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        typing = bool(data.get("typing"))

        if typing:
            db_write(
                """
                INSERT INTO dm_typing_events (
                    direct_conversation_id,
                    user_id,
                    updated_at
                ) VALUES (?, ?, ?)
                ON CONFLICT(direct_conversation_id, user_id)
                DO UPDATE SET updated_at = excluded.updated_at
                """,
                (dm_id, user["id"], time.time()),
            )
        else:
            db_write(
                """
                DELETE FROM dm_typing_events
                WHERE direct_conversation_id = ?
                  AND user_id = ?
                """,
                (dm_id, user["id"]),
            )

        return jsonify({"status": "ok"})

    @app.route("/api/dms/<int:dm_id>/typing")
    def api_get_dm_typing(dm_id: int):
        user = current_user()
        if not validate_dm_access(get_db, dm_id, user):
            return jsonify({"error": "DM not found."}), 404

        cutoff = time.time() - TYPING_TTL_SECONDS
        with get_db() as db:
            db.execute("DELETE FROM dm_typing_events WHERE updated_at < ?", (cutoff,))
            rows = db.execute(
                """
                SELECT
                    u.id,
                    u.username
                FROM dm_typing_events AS t
                JOIN users AS u
                    ON u.id = t.user_id
                WHERE t.direct_conversation_id = ?
                  AND t.user_id != ?
                  AND t.updated_at >= ?
                ORDER BY u.username COLLATE NOCASE
                """,
                (dm_id, user["id"], cutoff),
            ).fetchall()
            db.commit()

        return jsonify({"users": [{"id": row["id"], "username": row["username"]} for row in rows]})

    @app.route("/api/rooms")
    def list_rooms():
        user = current_user()

        with get_db() as db:
            rows = db.execute(
                """
                SELECT
                    id,
                    slug,
                    name,
                    description,
                    visibility,
                    created_by_user_id,
                    is_active
                FROM rooms
                WHERE is_active = 1
                ORDER BY
                    CASE WHEN slug = 'general' THEN 0 ELSE 1 END,
                    name COLLATE NOCASE
                """
            ).fetchall()

        rooms = [room_payload(room, user) for room in rows if can_access_room(room, user)]
        return jsonify({"rooms": rooms, "authenticated": bool(user)})

    @app.route("/api/rooms", methods=["POST"])
    def create_room():
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        admin, error = require_admin()
        if error:
            return error

        limited = rate_limited("admin")
        if limited:
            return limited

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        name = str(data.get("name", "")).strip()
        description = str(data.get("description", "")).strip()[:300]
        visibility = data.get("visibility", "public")

        if not name or len(name) > 60:
            return jsonify({"error": "Room name must be 1 to 60 characters."}), 400
        if visibility not in ROOM_VISIBILITIES:
            return jsonify({"error": "Invalid room visibility."}), 400

        base_slug = re.sub(r"[^a-z0-9-]+", "-", name.lower()).strip("-")
        base_slug = base_slug[:48].strip("-")

        if not base_slug or not ROOM_SLUG_RE.fullmatch(base_slug):
            return jsonify({"error": "Room name cannot create a valid room slug."}), 400

        with get_db() as db:
            db.execute("BEGIN IMMEDIATE")
            slug = base_slug
            suffix = 2
            while db.execute("SELECT 1 FROM rooms WHERE slug = ?", (slug,)).fetchone():
                slug = f"{base_slug[:44]}-{suffix}"
                suffix += 1

            cursor = db.execute(
                """
                INSERT INTO rooms (
                    slug,
                    name,
                    description,
                    visibility,
                    created_by_user_id,
                    created_at,
                    is_active
                ) VALUES (?, ?, ?, ?, ?, ?, 1)
                """,
                (slug, name, description, visibility, admin["id"], now_iso()),
            )
            room_id = cursor.lastrowid

            db.execute(
                """
                INSERT INTO room_members (
                    room_id,
                    user_id,
                    membership_role,
                    added_at
                ) VALUES (?, ?, 'owner', ?)
                """,
                (room_id, admin["id"], now_iso()),
            )

            general_team = db.execute(
                "SELECT id FROM teams WHERE slug = 'general' AND is_active = 1"
            ).fetchone()

            if general_team:
                db.execute(
                    """
                    INSERT INTO channels (
                        team_id,
                        legacy_room_id,
                        slug,
                        name,
                        description,
                        created_by_user_id,
                        created_at,
                        is_active
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, 1)
                    """,
                    (
                        general_team["id"],
                        room_id,
                        slug,
                        name,
                        description,
                        admin["id"],
                        now_iso(),
                    ),
                )

            db.commit()

        audit("room_created", admin["id"], "room", str(room_id), f"{slug}:{visibility}")

        return jsonify(
            {
                "room": {
                    "id": room_id,
                    "slug": slug,
                    "name": name,
                    "description": description,
                    "visibility": visibility,
                    "can_read": True,
                    "can_write": True,
                    "can_manage": True,
                }
            }
        ), 201

    @app.route("/api/rooms/<string:slug>", methods=["PATCH"])
    def update_room(slug: str):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        limited = rate_limited("admin")
        if limited:
            return limited

        room = get_room_by_slug(slug)
        if not room:
            return jsonify({"error": "Room not found."}), 404

        if room["slug"] == "general" and user["role"] != "admin":
            return jsonify({"error": "Only an administrator can modify the default room."}), 403

        if not can_manage_room(room, user):
            return jsonify({"error": "Room owner or administrator access required.", "code": "forbidden"}), 403

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        name = str(data.get("name", room["name"])).strip()
        description = str(data.get("description", room["description"] or "")).strip()[:300]
        visibility = data.get("visibility", room["visibility"])

        if not name or len(name) > 60:
            return jsonify({"error": "Room name must be 1 to 60 characters."}), 400
        if visibility not in ROOM_VISIBILITIES:
            return jsonify({"error": "Invalid room visibility."}), 400

        db_write(
            "UPDATE rooms SET name = ?, description = ?, visibility = ? WHERE id = ?",
            (name, description, visibility, room["id"]),
        )
        db_write(
            "UPDATE channels SET name = ?, description = ? WHERE legacy_room_id = ?",
            (name, description, room["id"]),
        )

        audit("room_updated", user["id"], "room", str(room["id"]), f"name={name};visibility={visibility}")
        return jsonify({"status": "updated"})

    @app.route("/api/rooms/<string:slug>", methods=["DELETE"])
    def delete_room(slug: str):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        admin, error = require_admin()
        if error:
            return error

        limited = rate_limited("admin")
        if limited:
            return limited

        room = get_room_by_slug(slug)
        if not room:
            return jsonify({"error": "Room not found."}), 404

        if room["slug"] == "general":
            return jsonify({"error": "The default room cannot be deleted."}), 400

        db_write("DELETE FROM rooms WHERE id = ?", (room["id"],))
        audit("room_deleted", admin["id"], "room", str(room["id"]), room["slug"])
        return jsonify({"status": "deleted"})

    @app.route("/api/rooms/<string:slug>/members")
    def list_room_members(slug: str):
        admin, error = require_admin()
        if error:
            return error

        room = get_room_by_slug(slug)
        if not room:
            return jsonify({"error": "Room not found."}), 404

        if room["visibility"] != "private":
            return jsonify({"error": "This room is public."}), 400

        with get_db() as db:
            rows = db.execute(
                """
                SELECT
                    u.id,
                    u.username,
                    u.role,
                    rm.membership_role,
                    rm.added_at
                FROM room_members AS rm
                JOIN users AS u
                    ON u.id = rm.user_id
                WHERE rm.room_id = ?
                ORDER BY
                    CASE rm.membership_role
                        WHEN 'owner' THEN 0
                        ELSE 1
                    END,
                    u.username COLLATE NOCASE
                """,
                (room["id"],),
            ).fetchall()

        return jsonify(
            {
                "room": room_payload(room, admin),
                "members": [
                    {
                        "id": row["id"],
                        "username": row["username"],
                        "role": row["role"],
                        "membership_role": row["membership_role"],
                        "added_at": row["added_at"],
                    }
                    for row in rows
                ],
            }
        )

    @app.route("/api/rooms/<string:slug>/members", methods=["POST"])
    def add_room_member(slug: str):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        admin, error = require_admin()
        if error:
            return error

        room = get_room_by_slug(slug)
        if not room:
            return jsonify({"error": "Room not found."}), 404

        if room["visibility"] != "private":
            return jsonify({"error": "Members can only be managed for private rooms."}), 400

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        username = normalize_username(data.get("username"))

        with get_db() as db:
            target = db.execute(
                "SELECT id, username FROM users WHERE username = ?",
                (username,),
            ).fetchone()

        if not target:
            return jsonify({"error": "User not found."}), 404

        db_write(
            """
            INSERT OR IGNORE INTO room_members (
                room_id,
                user_id,
                membership_role,
                added_at
            ) VALUES (?, ?, 'member', ?)
            """,
            (room["id"], target["id"], now_iso()),
        )

        audit("room_member_added", admin["id"], "room", str(room["id"]), target["username"])
        return jsonify({"status": "added"}), 201

    @app.route("/api/rooms/<string:slug>/members/<int:user_id>", methods=["DELETE"])
    def remove_room_member(slug: str, user_id: int):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        admin, error = require_admin()
        if error:
            return error

        room = get_room_by_slug(slug)
        if not room:
            return jsonify({"error": "Room not found."}), 404

        with get_db() as db:
            member = db.execute(
                """
                SELECT membership_role
                FROM room_members
                WHERE room_id = ?
                  AND user_id = ?
                """,
                (room["id"], user_id),
            ).fetchone()

        if not member:
            return jsonify({"error": "Room member not found."}), 404

        if member["membership_role"] == "owner":
            return jsonify({"error": "The room owner cannot be removed."}), 400

        db_write(
            "DELETE FROM room_members WHERE room_id = ? AND user_id = ?",
            (room["id"], user_id),
        )

        audit("room_member_removed", admin["id"], "room", str(room["id"]), str(user_id))
        return jsonify({"status": "removed"})

    @app.route("/api/rooms/<string:slug>/invites", methods=["POST"])
    def create_room_invite(slug: str):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        admin, error = require_admin()
        if error:
            return error

        room = get_room_by_slug(slug)
        if not room:
            return jsonify({"error": "Room not found."}), 404

        if room["visibility"] != "private":
            return jsonify({"error": "Invite links are only available for private rooms."}), 400

        token = ctx["secrets"].token_urlsafe(32)
        expires_at = (utc_now() + ctx["timedelta"](hours=INVITE_TTL_HOURS)).isoformat()

        db_write(
            """
            INSERT INTO room_invites (
                room_id,
                token_hash,
                created_by_user_id,
                created_at,
                expires_at
            ) VALUES (?, ?, ?, ?, ?)
            """,
            (room["id"], hash_identifier(token), admin["id"], now_iso(), expires_at),
        )

        audit("room_invite_created", admin["id"], "room", str(room["id"]))
        return jsonify({"invite": {"url": f"/?invite={token}", "expires_at": expires_at}}), 201

    @app.route("/api/invites/<string:token>/accept", methods=["POST"])
    def accept_room_invite(token: str):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        token_hash = hash_identifier(token)

        with get_db() as db:
            invite = db.execute(
                """
                SELECT
                    ri.id,
                    ri.room_id,
                    ri.expires_at,
                    ri.used_at,
                    r.slug,
                    r.name,
                    r.description,
                    r.visibility,
                    r.created_by_user_id,
                    r.is_active
                FROM room_invites AS ri
                JOIN rooms AS r
                    ON r.id = ri.room_id
                WHERE ri.token_hash = ?
                """,
                (token_hash,),
            ).fetchone()

        if not invite:
            return jsonify({"error": "Invite link is invalid."}), 404

        if invite["used_at"] or parse_iso(invite["expires_at"]) <= utc_now():
            return jsonify({"error": "Invite link has expired or was already used."}), 400

        if invite["visibility"] != "private" or not invite["is_active"]:
            return jsonify({"error": "This room is no longer available."}), 404

        with get_db() as db:
            db.execute("BEGIN IMMEDIATE")
            fresh = db.execute(
                "SELECT id, used_at, expires_at FROM room_invites WHERE id = ?",
                (invite["id"],),
            ).fetchone()

            if not fresh or fresh["used_at"] or parse_iso(fresh["expires_at"]) <= utc_now():
                db.rollback()
                return jsonify({"error": "Invite link has expired or was already used."}), 400

            db.execute(
                """
                INSERT OR IGNORE INTO room_members (
                    room_id,
                    user_id,
                    membership_role,
                    added_at
                ) VALUES (?, ?, 'member', ?)
                """,
                (invite["room_id"], user["id"], now_iso()),
            )
            db.execute(
                """
                UPDATE room_invites
                SET
                    used_at = ?,
                    used_by_user_id = ?
                WHERE id = ?
                """,
                (now_iso(), user["id"], invite["id"]),
            )
            db.commit()

        audit("room_invite_accepted", user["id"], "room", str(invite["room_id"]))
        room = get_room_by_slug(invite["slug"])
        return jsonify({"status": "joined", "room": room_payload(room, user)})

    @app.route("/api/admin/users")
    def admin_list_users():
        moderator, error = require_moderator()
        if error:
            return error

        with get_db() as db:
            rows = db.execute(
                """
                SELECT
                    u.id,
                    u.username,
                    u.role,
                    u.status,
                    u.restricted_until,
                    u.created_at,
                    u.last_login_at,
                    COUNT(m.id) AS message_count
                FROM users AS u
                LEFT JOIN messages AS m
                    ON m.user_id = u.id
                GROUP BY u.id
                ORDER BY
                    CASE u.role
                        WHEN 'admin' THEN 0
                        WHEN 'moderator' THEN 1
                        ELSE 2
                    END,
                    u.username COLLATE NOCASE
                """
            ).fetchall()

        return jsonify(
            {
                "users": [
                    {
                        "id": row["id"],
                        "username": row["username"],
                        "role": row["role"],
                        "status": row["status"],
                        "restricted_until": row["restricted_until"],
                        "created_at": row["created_at"],
                        "last_login_at": row["last_login_at"],
                        "message_count": row["message_count"],
                    }
                    for row in rows
                ],
                "requested_by": moderator["username"],
            }
        )

    @app.route("/api/admin/users/<int:user_id>", methods=["PATCH"])
    def admin_update_user(user_id: int):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        admin, error = require_admin()
        if error:
            return error

        limited = rate_limited("admin")
        if limited:
            return limited

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        with get_db() as db:
            target = db.execute(
                """
                SELECT
                    id,
                    username,
                    role,
                    status,
                    restricted_until
                FROM users
                WHERE id = ?
                """,
                (user_id,),
            ).fetchone()

        if not target:
            return jsonify({"error": "User not found."}), 404

        if target["id"] == admin["id"]:
            return jsonify({"error": "You cannot modify your own account here."}), 400

        role = data.get("role", target["role"])
        status = data.get("status", target["status"])
        restricted_until = data.get("restricted_until", target["restricted_until"])

        if role not in ROLES:
            return jsonify({"error": "Invalid role."}), 400
        if status not in STATUSES:
            return jsonify({"error": "Invalid account status."}), 400

        if status == "restricted" and restricted_until:
            parsed = parse_iso(restricted_until)
            if not parsed or parsed <= utc_now():
                return jsonify({"error": "restricted_until must be a future ISO timestamp."}), 400
        elif status != "restricted":
            restricted_until = None

        if target["role"] == "admin" and role != "admin":
            with get_db() as db:
                count = db.execute(
                    """
                    SELECT COUNT(*) AS count
                    FROM users
                    WHERE role = 'admin'
                      AND status = 'active'
                    """
                ).fetchone()["count"]
            if count <= 1:
                return jsonify({"error": "At least one active administrator is required."}), 400

        is_admin_value = 1 if role == "admin" else 0
        is_active = 1 if status in {"active", "restricted"} else 0

        db_write(
            """
            UPDATE users
            SET
                role = ?,
                status = ?,
                restricted_until = ?,
                is_admin = ?,
                is_active = ?
            WHERE id = ?
            """,
            (role, status, restricted_until, is_admin_value, is_active, user_id),
        )

        if status == "blacklisted":
            db_write(
                """
                INSERT OR REPLACE INTO blacklist_entries (
                    kind,
                    value_hash,
                    reason,
                    created_at,
                    created_by_user_id
                ) VALUES ('user', ?, ?, ?, ?)
                """,
                (
                    hash_identifier(str(user_id)),
                    "Account blacklisted by administrator",
                    now_iso(),
                    admin["id"],
                ),
            )

        if target["status"] == "blacklisted" and status != "blacklisted":
            db_write(
                "DELETE FROM blacklist_entries WHERE kind = 'user' AND value_hash = ?",
                (hash_identifier(str(user_id)),),
            )

        audit(
            "user_updated",
            admin["id"],
            "user",
            str(user_id),
            f"role={role};status={status};restricted_until={restricted_until}",
        )
        return jsonify({"status": "updated"})

    @app.route("/api/admin/rooms")
    def admin_rooms():
        moderator, error = require_moderator()
        if error:
            return error

        with get_db() as db:
            rows = db.execute(
                """
                SELECT
                    id,
                    slug,
                    name,
                    description,
                    visibility,
                    created_by_user_id,
                    is_active
                FROM rooms
                WHERE is_active = 1
                ORDER BY name COLLATE NOCASE
                """
            ).fetchall()

        return jsonify(
            {
                "rooms": [
                    {
                        "id": row["id"],
                        "slug": row["slug"],
                        "name": row["name"],
                        "description": row["description"],
                        "visibility": row["visibility"],
                    }
                    for row in rows
                ]
            }
        )

    @app.route("/api/admin/blacklist/ip", methods=["POST"])
    def blacklist_ip():
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        admin, error = require_admin()
        if error:
            return error

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        ip = str(data.get("ip", "")).strip()
        reason = str(data.get("reason", "")).strip()[:300]

        if not ip or len(ip) > 128:
            return jsonify({"error": "A valid IP value is required."}), 400

        db_write(
            """
            INSERT OR REPLACE INTO blacklist_entries (
                kind,
                value_hash,
                reason,
                created_at,
                created_by_user_id
            ) VALUES ('ip', ?, ?, ?, ?)
            """,
            (
                hash_identifier(ip),
                reason or "IP blacklisted by administrator",
                now_iso(),
                admin["id"],
            ),
        )

        audit("ip_blacklisted", admin["id"], "ip", hash_identifier(ip), reason)
        return jsonify({"status": "blacklisted"})

    @app.route("/api/teams")
    def api_list_teams():
        user = current_user()

        with get_db() as db:
            teams = db.execute(
                """
                SELECT
                    id,
                    slug,
                    name,
                    description,
                    created_by_user_id,
                    created_at,
                    is_active
                FROM teams
                WHERE is_active = 1
                ORDER BY name COLLATE NOCASE
                """
            ).fetchall()

            channels = db.execute(
                """
                SELECT
                    c.id,
                    c.team_id,
                    c.legacy_room_id,
                    c.slug,
                    c.name,
                    c.description,
                    c.created_by_user_id,
                    c.created_at,
                    c.is_active,
                    t.slug AS team_slug,
                    t.name AS team_name,
                    t.description AS team_description,
                    t.is_active AS team_is_active
                FROM channels AS c
                JOIN teams AS t
                    ON t.id = c.team_id
                WHERE c.is_active = 1
                  AND t.is_active = 1
                """
            ).fetchall()

            public_page_rows = db.execute(
                """
                SELECT DISTINCT p.team_id
                FROM pages AS p
                JOIN teams AS t
                    ON t.id = p.team_id
                WHERE p.is_active = 1
                  AND p.visibility = 'public'
                  AND t.is_active = 1
                """
            ).fetchall()

        channels_by_team = {}
        for channel in channels:
            if can_access_channel(channel, user):
                channels_by_team[channel["team_id"]] = True

        public_page_team_ids = {row["team_id"] for row in public_page_rows}

        visible_teams = [
            team_payload(team, user)
            for team in teams
            if can_access_team(team, user) or channels_by_team.get(team["id"]) or team["id"] in public_page_team_ids
        ]

        return jsonify({"teams": visible_teams, "authenticated": bool(user)})

    @app.route("/api/teams", methods=["POST"])
    def api_create_team():
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        limited = rate_limited("admin")
        if limited:
            return limited

        with get_db() as db:
            owned_count = db.execute(
                """
                SELECT COUNT(*) AS count
                FROM team_members
                WHERE user_id = ?
                  AND membership_role = 'owner'
                """,
                (user["id"],),
            ).fetchone()["count"]

        if owned_count >= MAX_OWNED_TEAMS_PER_USER:
            return jsonify({"error": "You can own at most 3 teams."}), 400

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        name = str(data.get("name", "")).strip()
        description = str(data.get("description", "")).strip()[:300]
        requested_slug = str(data.get("slug", "")).strip().lower()

        if not name or len(name) > 60:
            return jsonify({"error": "Team name must be 1 to 60 characters."}), 400

        base_slug = requested_slug or team_slug_from_name(name)
        if not base_slug or not ROOM_SLUG_RE.fullmatch(base_slug):
            return jsonify({"error": "Team name cannot create a valid slug."}), 400

        with get_db() as db:
            db.execute("BEGIN IMMEDIATE")
            slug = base_slug
            suffix = 2
            while db.execute("SELECT 1 FROM teams WHERE slug = ?", (slug,)).fetchone():
                slug = f"{base_slug[:44]}-{suffix}"
                suffix += 1

            cursor = db.execute(
                """
                INSERT INTO teams (
                    slug,
                    name,
                    description,
                    created_by_user_id,
                    created_at,
                    is_active
                ) VALUES (?, ?, ?, ?, ?, 1)
                """,
                (slug, name, description, user["id"], now_iso()),
            )
            team_id = cursor.lastrowid

            db.execute(
                """
                INSERT INTO team_members (
                    team_id,
                    user_id,
                    membership_role,
                    added_at
                ) VALUES (?, ?, 'owner', ?)
                """,
                (team_id, user["id"], now_iso()),
            )

            db.commit()

        audit("team_created", user["id"], "team", str(team_id), slug)

        with get_db() as db:
            team = db.execute(
                """
                SELECT
                    id,
                    slug,
                    name,
                    description,
                    created_by_user_id,
                    created_at,
                    is_active
                FROM teams
                WHERE id = ?
                """,
                (team_id,),
            ).fetchone()

        return jsonify({"team": team_payload(team, user)}), 201

    @app.route("/api/teams/<string:team_slug>", methods=["PATCH"])
    def api_update_team(team_slug: str):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        limited = rate_limited("admin")
        if limited:
            return limited

        team = get_team_by_slug(team_slug)
        if not team:
            return jsonify({"error": "Team not found."}), 404

        if not can_edit_team(team["id"], user):
            return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        name = str(data.get("name", team["name"])).strip()
        description = str(data.get("description", team["description"] or "")).strip()[:300]

        if not name or len(name) > 60:
            return jsonify({"error": "Team name must be 1 to 60 characters."}), 400

        db_write(
            "UPDATE teams SET name = ?, description = ? WHERE id = ?",
            (name, description, team["id"]),
        )

        updated = get_team_by_slug(team_slug)
        audit("team_updated", user["id"], "team", str(team["id"]), f"name={name}")
        return jsonify({"team": team_payload(updated, user)})

    @app.route("/api/teams/<string:team_slug>/channels")
    def api_list_team_channels(team_slug: str):
        team = get_team_by_slug(team_slug)
        if not team:
            return jsonify({"error": "Team not found."}), 404

        user = current_user()

        with get_db() as db:
            rows = db.execute(
                """
                SELECT
                    c.id,
                    c.team_id,
                    c.legacy_room_id,
                    c.slug,
                    c.name,
                    c.description,
                    c.created_by_user_id,
                    c.created_at,
                    c.is_active,
                    t.slug AS team_slug,
                    t.name AS team_name,
                    t.description AS team_description,
                    t.is_active AS team_is_active
                FROM channels AS c
                JOIN teams AS t
                    ON t.id = c.team_id
                WHERE c.team_id = ?
                  AND c.is_active = 1
                  AND t.is_active = 1
                ORDER BY
                    CASE
                        WHEN c.slug = 'general' THEN 0
                        ELSE 1
                    END,
                    c.name COLLATE NOCASE
                """,
                (team["id"],),
            ).fetchall()

        channels = [channel_payload(channel, user) for channel in rows if can_access_channel(channel, user)]

        if not channels and not can_access_team(team, user):
            if not user:
                return jsonify({"error": "Login required.", "code": "login_required"}), 401
            return jsonify({"error": "You do not have access to this team.", "code": "forbidden"}), 403

        return jsonify({"team": team_payload(team, user), "channels": channels})

    @app.route("/api/teams/<string:team_slug>/channels", methods=["POST"])
    def api_create_channel(team_slug: str):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        team = get_team_by_slug(team_slug)
        if not team:
            return jsonify({"error": "Team not found."}), 404

        if not is_team_admin(team["id"], user):
            return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        name = str(data.get("name", "")).strip()
        description = str(data.get("description", "")).strip()[:300]
        requested_slug = str(data.get("slug", "")).strip().lower()

        if not name or len(name) > 60:
            return jsonify({"error": "Channel name must be 1 to 60 characters."}), 400

        base_slug = requested_slug or team_slug_from_name(name)
        if not base_slug or not ROOM_SLUG_RE.fullmatch(base_slug):
            return jsonify({"error": "Channel name cannot create a valid slug."}), 400

        with get_db() as db:
            db.execute("BEGIN IMMEDIATE")
            slug = base_slug
            suffix = 2
            while db.execute("SELECT 1 FROM channels WHERE team_id = ? AND slug = ?", (team["id"], slug)).fetchone():
                slug = f"{base_slug[:44]}-{suffix}"
                suffix += 1

            cursor = db.execute(
                """
                INSERT INTO channels (
                    team_id,
                    legacy_room_id,
                    slug,
                    name,
                    description,
                    created_by_user_id,
                    created_at,
                    is_active
                ) VALUES (?, NULL, ?, ?, ?, ?, ?, 1)
                """,
                (team["id"], slug, name, description, user["id"], now_iso()),
            )
            channel_id = cursor.lastrowid
            db.commit()

        audit("channel_created", user["id"], "channel", str(channel_id), f"{team_slug}/{slug}")
        channel = get_channel_by_id(channel_id)
        return jsonify({"channel": channel_payload(channel, user)}), 201

    @app.route("/api/teams/<string:team_slug>/channels/<string:channel_slug>", methods=["PATCH"])
    def api_update_channel(team_slug: str, channel_slug: str):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        limited = rate_limited("admin")
        if limited:
            return limited

        channel = get_channel_by_slug(team_slug, channel_slug)
        if not channel:
            return jsonify({"error": "Channel not found."}), 404

        if not can_edit_team(channel["team_id"], user):
            return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        name = str(data.get("name", channel["name"])).strip()
        description = str(data.get("description", channel["description"] or "")).strip()[:300]

        if not name or len(name) > 60:
            return jsonify({"error": "Channel name must be 1 to 60 characters."}), 400

        db_write(
            "UPDATE channels SET name = ?, description = ? WHERE id = ?",
            (name, description, channel["id"]),
        )

        if channel["legacy_room_id"]:
            db_write(
                "UPDATE rooms SET name = ?, description = ? WHERE id = ?",
                (name, description, channel["legacy_room_id"]),
            )

        updated = get_channel_by_id(channel["id"])
        audit("channel_updated", user["id"], "channel", str(channel["id"]), f"{team_slug}/{channel_slug}")
        return jsonify({"channel": channel_payload(updated, user)})

    @app.route("/api/teams/<string:team_slug>/channels/<string:channel_slug>", methods=["DELETE"])
    def api_delete_channel(team_slug: str, channel_slug: str):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        limited = rate_limited("admin")
        if limited:
            return limited

        channel = get_channel_by_slug(team_slug, channel_slug)
        if not channel:
            return jsonify({"error": "Channel not found."}), 404

        if not can_edit_team(channel["team_id"], user):
            return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403

        if channel["legacy_room_id"]:
            return jsonify({"error": "Legacy channels linked to rooms cannot be deleted here."}), 400

        db_write("UPDATE channels SET is_active = 0 WHERE id = ?", (channel["id"],))
        audit("channel_deleted", user["id"], "channel", str(channel["id"]), f"{team_slug}/{channel_slug}")
        return jsonify({"status": "deleted"})

    @app.route("/api/teams/<string:team_slug>/members")
    def api_list_team_members(team_slug: str):
        user, error = require_user()
        if error:
            return error

        team = get_team_by_slug(team_slug)
        if not team:
            return jsonify({"error": "Team not found."}), 404

        if not is_team_admin(team["id"], user):
            return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403

        with get_db() as db:
            rows = db.execute(
                """
                SELECT
                    u.id,
                    u.username,
                    u.role,
                    tm.membership_role,
                    tm.added_at
                FROM team_members AS tm
                JOIN users AS u
                    ON u.id = tm.user_id
                WHERE tm.team_id = ?
                ORDER BY
                    CASE tm.membership_role
                        WHEN 'owner' THEN 0
                        WHEN 'admin' THEN 1
                        ELSE 2
                    END,
                    u.username COLLATE NOCASE
                """,
                (team["id"],),
            ).fetchall()

        return jsonify(
            {
                "team": team_payload(team, user),
                "members": [
                    {
                        "id": row["id"],
                        "username": row["username"],
                        "role": row["role"],
                        "membership_role": row["membership_role"],
                        "added_at": row["added_at"],
                    }
                    for row in rows
                ],
            }
        )

    @app.route("/api/teams/<string:team_slug>/members", methods=["POST"])
    def api_add_team_member(team_slug: str):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        team = get_team_by_slug(team_slug)
        if not team:
            return jsonify({"error": "Team not found."}), 404

        if not is_team_admin(team["id"], user):
            return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        username = normalize_username(data.get("username"))
        membership_role = str(data.get("membership_role", "member")).strip().lower()

        if membership_role not in TEAM_MEMBER_ROLES:
            return jsonify({"error": "Invalid membership role."}), 400

        with get_db() as db:
            target = db.execute(
                "SELECT id, username FROM users WHERE username = ?",
                (username,),
            ).fetchone()

        if not target:
            return jsonify({"error": "User not found."}), 404

        if membership_role == "owner" and team_member_role(team["id"], user["id"]) != "owner":
            return jsonify({"error": "Only an owner can assign owner role."}), 403

        db_write(
            """
            INSERT OR IGNORE INTO team_members (
                team_id,
                user_id,
                membership_role,
                added_at
            ) VALUES (?, ?, ?, ?)
            """,
            (team["id"], target["id"], membership_role, now_iso()),
        )

        audit("team_member_added", user["id"], "team", str(team["id"]), f"{target['username']}:{membership_role}")
        return jsonify({"status": "added"}), 201

    @app.route("/api/teams/<string:team_slug>/members/<int:member_user_id>", methods=["PATCH"])
    def api_update_team_member(team_slug: str, member_user_id: int):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        team = get_team_by_slug(team_slug)
        if not team:
            return jsonify({"error": "Team not found."}), 404

        if not is_team_admin(team["id"], user):
            return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        membership_role = str(data.get("membership_role", "")).strip().lower()
        if membership_role not in TEAM_MEMBER_ROLES:
            return jsonify({"error": "Invalid membership role."}), 400

        actor_role = team_member_role(team["id"], user["id"])
        target_role = team_member_role(team["id"], member_user_id)

        if not target_role:
            return jsonify({"error": "Team member not found."}), 404

        if target_role == "owner" and actor_role != "owner":
            return jsonify({"error": "Only an owner can modify another owner."}), 403

        if membership_role == "owner" and actor_role != "owner":
            return jsonify({"error": "Only an owner can assign owner role."}), 403

        db_write(
            "UPDATE team_members SET membership_role = ? WHERE team_id = ? AND user_id = ?",
            (membership_role, team["id"], member_user_id),
        )

        audit(
            "team_member_role_updated",
            user["id"],
            "team",
            str(team["id"]),
            f"user_id={member_user_id};role={membership_role}",
        )
        return jsonify({"status": "updated"})

    @app.route("/api/teams/<string:team_slug>/members/<int:member_user_id>", methods=["DELETE"])
    def api_remove_team_member(team_slug: str, member_user_id: int):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        team = get_team_by_slug(team_slug)
        if not team:
            return jsonify({"error": "Team not found."}), 404

        if not is_team_admin(team["id"], user):
            return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403

        actor_role = team_member_role(team["id"], user["id"])
        target_role = team_member_role(team["id"], member_user_id)

        if not target_role:
            return jsonify({"error": "Team member not found."}), 404

        if target_role == "owner":
            return jsonify({"error": "The team owner cannot be removed."}), 400

        if target_role == "admin" and actor_role not in {"owner", "admin"}:
            return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403

        db_write(
            "DELETE FROM team_members WHERE team_id = ? AND user_id = ?",
            (team["id"], member_user_id),
        )

        audit("team_member_removed", user["id"], "team", str(team["id"]), str(member_user_id))
        return jsonify({"status": "removed"})

    @app.route("/api/teams/<string:team_slug>/invites", methods=["POST"])
    def api_create_team_invite(team_slug: str):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        team = get_team_by_slug(team_slug)
        if not team:
            return jsonify({"error": "Team not found."}), 404

        if not is_team_admin(team["id"], user):
            return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        membership_role = str(data.get("membership_role", "member")).strip().lower()
        if membership_role not in {"member", "admin"}:
            return jsonify({"error": "Invite role must be member or admin."}), 400

        token = ctx["secrets"].token_urlsafe(32)
        expires_at = (utc_now() + ctx["timedelta"](hours=TEAM_INVITE_TTL_HOURS)).isoformat()

        db_write(
            """
            INSERT INTO team_invites (
                team_id,
                token_hash,
                created_by_user_id,
                membership_role,
                created_at,
                expires_at
            ) VALUES (?, ?, ?, ?, ?, ?)
            """,
            (team["id"], hash_identifier(token), user["id"], membership_role, now_iso(), expires_at),
        )

        audit("team_invite_created", user["id"], "team", str(team["id"]), membership_role)
        return jsonify(
            {
                "invite": {
                    "url": f"/?team_invite={token}",
                    "expires_at": expires_at,
                    "membership_role": membership_role,
                }
            }
        ), 201

    @app.route("/api/team-invites/<string:token>/accept", methods=["POST"])
    def api_accept_team_invite(token: str):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        token_hash = hash_identifier(token)

        with get_db() as db:
            invite = db.execute(
                """
                SELECT
                    ti.id,
                    ti.team_id,
                    ti.membership_role,
                    ti.expires_at,
                    ti.used_at,
                    t.slug,
                    t.name,
                    t.description,
                    t.is_active
                FROM team_invites AS ti
                JOIN teams AS t
                    ON t.id = ti.team_id
                WHERE ti.token_hash = ?
                """,
                (token_hash,),
            ).fetchone()

        if not invite:
            return jsonify({"error": "Invite link is invalid."}), 404

        if invite["used_at"] or parse_iso(invite["expires_at"]) <= utc_now():
            return jsonify({"error": "Invite link has expired or was already used."}), 400

        if not invite["is_active"]:
            return jsonify({"error": "This team is no longer available."}), 404

        with get_db() as db:
            db.execute("BEGIN IMMEDIATE")
            fresh = db.execute(
                "SELECT id, used_at, expires_at FROM team_invites WHERE id = ?",
                (invite["id"],),
            ).fetchone()

            if not fresh or fresh["used_at"] or parse_iso(fresh["expires_at"]) <= utc_now():
                db.rollback()
                return jsonify({"error": "Invite link has expired or was already used."}), 400

            db.execute(
                """
                INSERT INTO team_members (
                    team_id,
                    user_id,
                    membership_role,
                    added_at
                ) VALUES (?, ?, ?, ?)
                ON CONFLICT(team_id, user_id)
                DO UPDATE SET membership_role = excluded.membership_role
                """,
                (invite["team_id"], user["id"], invite["membership_role"], now_iso()),
            )
            db.execute(
                """
                UPDATE team_invites
                SET used_at = ?, used_by_user_id = ?
                WHERE id = ?
                """,
                (now_iso(), user["id"], invite["id"]),
            )
            db.commit()

        audit("team_invite_accepted", user["id"], "team", str(invite["team_id"]))
        team = get_team_by_slug(invite["slug"])
        return jsonify({"status": "joined", "team": team_payload(team, user)})

    @app.route("/api/teams/<string:team_slug>/pages")
    def api_list_pages(team_slug: str):
        team = get_team_by_slug(team_slug)
        if not team:
            return jsonify({"error": "Team not found."}), 404

        user = current_user()
        page_type = request.args.get("type", "").strip().lower()
        visibility = request.args.get("visibility", "").strip().lower()

        if page_type and page_type not in PAGE_TYPES:
            return jsonify({"error": "Invalid page type."}), 400
        if visibility and visibility not in PAGE_VISIBILITIES:
            return jsonify({"error": "Invalid page visibility."}), 400

        is_member = bool(user and (user["role"] in {"moderator", "admin"} or is_team_member(team["id"], user["id"])))

        conditions = ["p.team_id = ?", "p.is_active = 1", "t.is_active = 1"]
        params = [team["id"]]

        if page_type:
            conditions.append("p.page_type = ?")
            params.append(page_type)

        if visibility:
            conditions.append("p.visibility = ?")
            params.append(visibility)

        if not is_member:
            conditions.append("p.visibility = 'public'")

        query = f"""
            {team_page_select()}
            WHERE {" AND ".join(conditions)}
            ORDER BY p.updated_at DESC, p.id DESC
        """

        with get_db() as db:
            rows = db.execute(query, tuple(params)).fetchall()

        return jsonify({"team": team_payload(team, user), "pages": [page_payload(page, user) for page in rows]})

    @app.route("/api/teams/<string:team_slug>/pages", methods=["POST"])
    def api_create_page():
        team_slug = request.view_args["team_slug"]

        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        team = get_team_by_slug(team_slug)
        if not team:
            return jsonify({"error": "Team not found."}), 404

        if not is_team_admin(team["id"], user):
            return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        title = str(data.get("title", "")).strip()
        body = str(data.get("body", "")).strip()
        visibility = str(data.get("visibility", "team")).strip().lower()
        page_type = str(data.get("page_type", "document")).strip().lower()
        requested_slug = str(data.get("slug", "")).strip().lower()
        channel_slug = str(data.get("channel_slug", "")).strip().lower()

        if not title or len(title) > 160:
            return jsonify({"error": "Page title must be 1 to 160 characters."}), 400
        if len(body) > 100000:
            return jsonify({"error": "Page content is too long."}), 400
        if visibility not in PAGE_VISIBILITIES:
            return jsonify({"error": "Invalid page visibility."}), 400
        if page_type not in PAGE_TYPES:
            return jsonify({"error": "Invalid page type."}), 400

        base_slug = requested_slug or page_slug_from_title(title)
        if not base_slug or not ROOM_SLUG_RE.fullmatch(base_slug):
            return jsonify({"error": "Page title cannot create a valid slug."}), 400

        channel_id = None
        if channel_slug:
            channel = get_channel_by_slug(team_slug, channel_slug)
            if not channel:
                return jsonify({"error": "Channel not found."}), 404
            channel_id = channel["id"]

        with get_db() as db:
            db.execute("BEGIN IMMEDIATE")
            slug = base_slug
            suffix = 2
            while db.execute("SELECT 1 FROM pages WHERE team_id = ? AND slug = ?", (team["id"], slug)).fetchone():
                slug = f"{base_slug[:44]}-{suffix}"
                suffix += 1

            cursor = db.execute(
                """
                INSERT INTO pages (
                    team_id,
                    channel_id,
                    slug,
                    title,
                    body,
                    visibility,
                    page_type,
                    created_by_user_id,
                    updated_by_user_id,
                    created_at,
                    updated_at,
                    is_active
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1)
                """,
                (
                    team["id"],
                    channel_id,
                    slug,
                    title,
                    body,
                    visibility,
                    page_type,
                    user["id"],
                    user["id"],
                    now_iso(),
                    now_iso(),
                ),
            )
            page_id = cursor.lastrowid
            db.commit()

        audit("page_created", user["id"], "page", str(page_id), f"{team_slug}/{slug}")

        with get_db() as db:
            page = db.execute(f"{team_page_select()} WHERE p.id = ?", (page_id,)).fetchone()

        return jsonify({"page": page_payload(page, user)}), 201

    @app.route("/api/teams/<string:team_slug>/pages/<string:page_slug>")
    def api_get_page(team_slug: str, page_slug: str):
        user = current_user()

        with get_db() as db:
            page = db.execute(
                f"""
                {team_page_select()}
                WHERE t.slug = ?
                  AND p.slug = ?
                  AND p.is_active = 1
                  AND t.is_active = 1
                """,
                (team_slug, page_slug),
            ).fetchone()

        if not page:
            return jsonify({"error": "Page not found."}), 404

        permitted = bool(
            page["visibility"] == "public"
            or (user and (user["role"] in {"moderator", "admin"} or is_team_member(page["team_id"], user["id"])))
        )

        if not permitted:
            if not user:
                return jsonify({"error": "Login required.", "code": "login_required"}), 401
            return jsonify({"error": "You do not have access to this page.", "code": "forbidden"}), 403

        return jsonify({"page": page_payload(page, user)})

    @app.route("/api/teams/<string:team_slug>/pages/<string:page_slug>", methods=["PATCH"])
    def api_update_page(team_slug: str, page_slug: str):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        with get_db() as db:
            page = db.execute(
                f"""
                {team_page_select()}
                WHERE t.slug = ?
                  AND p.slug = ?
                  AND p.is_active = 1
                  AND t.is_active = 1
                """,
                (team_slug, page_slug),
            ).fetchone()

        if not page:
            return jsonify({"error": "Page not found."}), 404

        if not is_team_admin(page["team_id"], user):
            return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        title = str(data.get("title", page["title"])).strip()
        body = str(data.get("body", page["body"])).strip()
        visibility = str(data.get("visibility", page["visibility"])).strip().lower()
        page_type = str(data.get("page_type", page["page_type"])).strip().lower()
        channel_slug = str(data.get("channel_slug", "")).strip().lower()

        if not title or len(title) > 160:
            return jsonify({"error": "Page title must be 1 to 160 characters."}), 400
        if len(body) > 100000:
            return jsonify({"error": "Page content is too long."}), 400
        if visibility not in PAGE_VISIBILITIES:
            return jsonify({"error": "Invalid page visibility."}), 400
        if page_type not in PAGE_TYPES:
            return jsonify({"error": "Invalid page type."}), 400

        channel_id = page["channel_id"]
        if "channel_slug" in data:
            if channel_slug:
                channel = get_channel_by_slug(team_slug, channel_slug)
                if not channel:
                    return jsonify({"error": "Channel not found."}), 404
                channel_id = channel["id"]
            else:
                channel_id = None

        db_write(
            """
            UPDATE pages
            SET
                title = ?,
                body = ?,
                visibility = ?,
                page_type = ?,
                channel_id = ?,
                updated_by_user_id = ?,
                updated_at = ?
            WHERE id = ?
            """,
            (title, body, visibility, page_type, channel_id, user["id"], now_iso(), page["id"]),
        )

        audit("page_updated", user["id"], "page", str(page["id"]), f"{team_slug}/{page_slug}")

        with get_db() as db:
            updated = db.execute(f"{team_page_select()} WHERE p.id = ?", (page["id"],)).fetchone()

        return jsonify({"page": page_payload(updated, user)})

    @app.route("/api/teams/<string:team_slug>/pages/<string:page_slug>", methods=["DELETE"])
    def api_delete_page(team_slug: str, page_slug: str):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        with get_db() as db:
            page = db.execute(
                """
                SELECT
                    p.id,
                    p.team_id
                FROM pages AS p
                JOIN teams AS t
                    ON t.id = p.team_id
                WHERE t.slug = ?
                  AND p.slug = ?
                  AND p.is_active = 1
                  AND t.is_active = 1
                """,
                (team_slug, page_slug),
            ).fetchone()

        if not page:
            return jsonify({"error": "Page not found."}), 404

        if not is_team_admin(page["team_id"], user):
            return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403

        db_write(
            """
            UPDATE pages
            SET
                is_active = 0,
                updated_by_user_id = ?,
                updated_at = ?
            WHERE id = ?
            """,
            (user["id"], now_iso(), page["id"]),
        )

        audit("page_deleted", user["id"], "page", str(page["id"]), f"{team_slug}/{page_slug}")
        return jsonify({"status": "deleted"})

    @app.route("/api/profile")
    def api_get_my_profile():
        user, error = require_user()
        if error:
            return error

        with get_db() as db:
            user_row = db.execute(
                """
                SELECT
                    id,
                    username,
                    role,
                    status,
                    restricted_until,
                    created_at
                FROM users
                WHERE id = ?
                """,
                (user["id"],),
            ).fetchone()

            profile_row = db.execute(
                """
                SELECT
                    user_id,
                    display_name,
                    badge_text,
                    bio,
                    bio_mode,
                    website,
                    location,
                    updated_at
                FROM user_profiles
                WHERE user_id = ?
                """,
                (user["id"],),
            ).fetchone()

            settings_row = db.execute(
                """
                SELECT
                    name_color
                FROM user_settings
                WHERE user_id = ?
                """,
                (user["id"],),
            ).fetchone()

        return jsonify({"profile": profile_payload(html_module, user_payload, user_row, profile_row, settings_row, user)})

    @app.route("/api/profile", methods=["PATCH"])
    def api_update_my_profile():

        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        limited = rate_limited("profile")
        if limited:
            return limited

        user, error = require_writable_user()
        if error:
            return error

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        display_name = str(data.get("display_name", "")).strip()
        badge_text = str(data.get("badge_text", "")).strip()
        bio = str(data.get("bio", "")).strip()
        bio_mode = str(data.get("bio_mode", "txt")).strip().lower()
        website = str(data.get("website", "")).strip()
        location = str(data.get("location", "")).strip()

        if len(display_name) > MAX_DISPLAY_NAME_LENGTH:
            return jsonify({"error": "Display name is too long."}), 400
        if len(badge_text) > MAX_PROFILE_BADGE_LENGTH:
            return jsonify({"error": "Badge text is too long."}), 400
        if len(bio) > MAX_PROFILE_BIO_LENGTH:
            return jsonify({"error": "Bio is too long."}), 400
        if bio_mode not in PROFILE_BIO_MODES:
            return jsonify({"error": "Invalid bio mode."}), 400
        if len(website) > MAX_WEBSITE_LENGTH:
            return jsonify({"error": "Website is too long."}), 400
        if len(location) > MAX_LOCATION_LENGTH:
            return jsonify({"error": "Location is too long."}), 400

        db_write(
            """
            UPDATE user_profiles
            SET
                display_name = ?,
                badge_text = ?,
                bio = ?,
                bio_mode = ?,
                website = ?,
                location = ?,
                updated_at = ?
            WHERE user_id = ?
            """,
            (
                display_name or user["username"],
                badge_text,
                bio,
                bio_mode,
                website,
                location,
                now_iso(),
                user["id"],
            ),
        )

        with get_db() as db:
            user_row = db.execute(
                """
                SELECT
                    id,
                    username,
                    role,
                    status,
                    restricted_until,
                    created_at
                FROM users
                WHERE id = ?
                """,
                (user["id"],),
            ).fetchone()

            profile_row = db.execute(
                """
                SELECT
                    user_id,
                    display_name,
                    badge_text,
                    bio,
                    bio_mode,
                    website,
                    location,
                    updated_at
                FROM user_profiles
                WHERE user_id = ?
                """,
                (user["id"],),
            ).fetchone()

            settings_row = db.execute(
                "SELECT name_color FROM user_settings WHERE user_id = ?",
                (user["id"],),
            ).fetchone()

        audit("profile_updated", user["id"], "user", str(user["id"]))
        return jsonify({"profile": profile_payload(html_module, user_payload, user_row, profile_row, settings_row, user)})
    
    @app.route("/api/profiles/<string:username>")
    def api_get_public_profile(username: str):
        normalized = normalize_username(username)

        with get_db() as db:
            user_row = db.execute(
                """
                SELECT
                    id,
                    username,
                    role,
                    status,
                    restricted_until,
                    created_at
                FROM users
                WHERE username = ?
                """,
                (normalized,),
            ).fetchone()

            if not user_row:
                return jsonify({"error": "Profile not found."}), 404

            profile_row = db.execute(
                """
                SELECT
                    user_id,
                    display_name,
                    badge_text,
                    bio,
                    bio_mode,
                    website,
                    location,
                    updated_at
                FROM user_profiles
                WHERE user_id = ?
                """,
                (user_row["id"],),
            ).fetchone()

            settings_row = db.execute(
                "SELECT name_color FROM user_settings WHERE user_id = ?",
                (user_row["id"],),
            ).fetchone()

        return jsonify({"profile": profile_payload(html_module, user_payload, user_row, profile_row, settings_row, current_user())})

    @app.route("/api/settings")
    def api_get_settings():
        user, error = require_user()
        if error:
            return error

        with get_db() as db:
            row = db.execute(
                """
                SELECT
                    accent_color,
                    name_color,
                    custom_css,
                    theme_mode,
                    updated_at
                FROM user_settings
                WHERE user_id = ?
                """,
                (user["id"],),
            ).fetchone()

        return jsonify(
            {
                "settings": {
                    "accent_color": row["accent_color"],
                    "name_color": row["name_color"],
                    "custom_css": row["custom_css"],
                    "theme_mode": row["theme_mode"],
                    "updated_at": row["updated_at"],
                }
            }
        )

    @app.route("/api/settings", methods=["PATCH"])
    def api_update_settings():
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        limited = rate_limited("settings")
        if limited:
            return limited

        user, error = require_writable_user()
        if error:
            return error

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        accent_color = data.get("accent_color", "#f7dce8")
        name_color = data.get("name_color")
        custom_css = str(data.get("custom_css", ""))
        theme_mode = str(data.get("theme_mode", "dark")).strip().lower()

        if not valid_hex_color(accent_color):
            return jsonify({"error": "Invalid accent color."}), 400
        if name_color not in (None, "") and not valid_hex_color(name_color):
            return jsonify({"error": "Invalid name color."}), 400
        if len(custom_css) > MAX_CUSTOM_CSS_LENGTH:
            return jsonify({"error": "Custom CSS is too long."}), 400
        if theme_mode not in THEME_MODES:
            return jsonify({"error": "Invalid theme mode."}), 400

        db_write(
            """
            UPDATE user_settings
            SET
                accent_color = ?,
                name_color = ?,
                custom_css = ?,
                theme_mode = ?,
                updated_at = ?
            WHERE user_id = ?
            """,
            (
                str(accent_color).lower(),
                normalized_hex_color(name_color) if name_color else None,
                custom_css,
                theme_mode,
                now_iso(),
                user["id"],
            ),
        )

        with get_db() as db:
            row = db.execute(
                """
                SELECT
                    accent_color,
                    name_color,
                    custom_css,
                    theme_mode,
                    updated_at
                FROM user_settings
                WHERE user_id = ?
                """,
                (user["id"],),
            ).fetchone()

        audit("settings_updated", user["id"], "user", str(user["id"]))
        return jsonify({"settings": dict(row)})

    @app.route("/api/dms")
    def api_list_dms():
        user, error = require_user()
        if error:
            return error
        return jsonify({"dms": list_dms_for_user(get_db, user["id"])})

    @app.route("/api/dms", methods=["POST"])
    def api_create_or_open_dm():
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        other_user_id = data.get("user_id")
        username = normalize_username(data.get("username"))

        with get_db() as db:
            target = None
            if other_user_id is not None:
                try:
                    parsed_id = int(other_user_id)
                except (TypeError, ValueError):
                    return jsonify({"error": "Invalid user_id."}), 400
                target = db.execute(
                    """
                    SELECT id, username, role, status
                    FROM users
                    WHERE id = ?
                    """,
                    (parsed_id,),
                ).fetchone()
            elif username:
                target = db.execute(
                    """
                    SELECT id, username, role, status
                    FROM users
                    WHERE username = ?
                    """,
                    (username,),
                ).fetchone()

        if not target:
            return jsonify({"error": "User not found."}), 404
        if int(target["id"]) == int(user["id"]):
            return jsonify({"error": "You cannot DM yourself."}), 400
        if target["status"] in {"disabled", "blacklisted"}:
            return jsonify({"error": "User not available."}), 400

        dm_id = find_or_create_dm(get_db, db_write, now_iso, user["id"], target["id"])
        audit("dm_opened", user["id"], "dm", str(dm_id), str(target["id"]))
        return jsonify({"dm": dm_payload(get_db, dm_id, user["id"])}), 201

    @app.route("/api/dms/<int:dm_id>/messages")
    def api_get_dm_messages(dm_id: int):
        user, error = require_user()
        if error:
            return error

        if not validate_dm_access(get_db, dm_id, user):
            return jsonify({"error": "DM not found."}), 404

        after_raw = request.args.get("after_id", "").strip()
        after_id = 0
        if after_raw:
            try:
                after_id = max(0, int(after_raw))
            except ValueError:
                return jsonify({"error": "after_id must be a non-negative integer."}), 400

        with get_db() as db:
            if after_raw:
                rows = db.execute(
                    f"""
                    {MESSAGE_SELECT}
                    WHERE m.direct_conversation_id = ?
                      AND m.parent_message_id IS NULL
                      AND m.id > ?
                    ORDER BY m.id ASC
                    LIMIT ?
                    """,
                    (dm_id, after_id, MAX_INCREMENTAL_MESSAGES),
                ).fetchall()
            else:
                rows = db.execute(
                    f"""
                    SELECT *
                    FROM (
                        {MESSAGE_SELECT}
                        WHERE m.direct_conversation_id = ?
                          AND m.parent_message_id IS NULL
                        ORDER BY m.id DESC
                        LIMIT ?
                    )
                    ORDER BY id ASC
                    """,
                    (dm_id, MAX_HISTORY_MESSAGES),
                ).fetchall()

        messages = [build_message_payload(row, user) for row in rows]
        latest_id = messages[-1]["id"] if messages else after_id

        return jsonify(
            {
                "dm": dm_payload(get_db, dm_id, user["id"]),
                "messages": messages,
                "latest_id": latest_id,
                "has_more": len(messages) == MAX_INCREMENTAL_MESSAGES,
            }
        )

    @app.route("/api/dms/<int:dm_id>/messages", methods=["POST"])
    def api_send_dm_message(dm_id: int):
        limited = rate_limited("send")
        if limited:
            return limited

        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        if not validate_dm_access(get_db, dm_id, user):
            return jsonify({"error": "DM not found."}), 404

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        text = data.get("text", "")
        if not isinstance(text, str):
            return jsonify({"error": "Message must be text."}), 400

        text = text.strip()
        if len(text) > MAX_MESSAGE_LENGTH:
            return jsonify({"error": "Message is too long."}), 400

        attachment_ids, attachment_error = resolve_owned_attachment_ids(
            get_db,
            data.get("attachment_ids", []),
            user["id"],
        )
        if attachment_error:
            return attachment_error

        if not attachment_ids and data.get("attachment_id") is not None:
            single_attachment_id, single_attachment_error = resolve_owned_attachment(
                get_db,
                data.get("attachment_id"),
                user["id"],
            )
            if single_attachment_error:
                return single_attachment_error
            attachment_ids = [single_attachment_id] if single_attachment_id else []

        attachment_id = attachment_ids[0] if attachment_ids else None

        if not text and not attachment_id:
            return jsonify({"error": "Message cannot be empty."}), 400

        with get_db() as db:
            db.execute("BEGIN IMMEDIATE")
            cursor = db.execute(
                """
                INSERT INTO messages (
                    room_id,
                    channel_id,
                    direct_conversation_id,
                    user_id,
                    username,
                    text,
                    attachment_id,
                    parent_message_id,
                    created_at
                ) VALUES (NULL, NULL, ?, ?, ?, ?, ?, NULL, ?)
                """,
                (dm_id, user["id"], user["username"], text, attachment_id, now_iso()),
            )
            message_id = cursor.lastrowid

            for position, extra_attachment_id in enumerate(attachment_ids):
                db.execute(
                    """
                    INSERT OR IGNORE INTO message_attachments (
                        message_id,
                        attachment_id,
                        position
                    ) VALUES (?, ?, ?)
                    """,
                    (message_id, extra_attachment_id, position),
                )

            db.commit()

        with get_db() as db:
            row = db.execute(f"{MESSAGE_SELECT} WHERE m.id = ?", (message_id,)).fetchone()

        audit("dm_message_sent", user["id"], "dm", str(dm_id), str(message_id))
        return jsonify({"message": build_message_payload(row, user)}), 201

    @app.route("/api/call-token", methods=["POST"])
    def api_call_token():
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        if not CALL_HELPER_SHARED_SECRET:
            return jsonify({"error": "Call helper is not configured."}), 503

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        room_slug = str(data.get("room", "")).strip()
        dm_id_raw = data.get("dm_id")

        call_id = None
        room_value = None

        if dm_id_raw is not None:
            try:
                dm_id = int(dm_id_raw)
            except (TypeError, ValueError):
                return jsonify({"error": "Invalid dm_id."}), 400

            if not validate_dm_access(get_db, dm_id, user):
                return jsonify({"error": "DM not found."}), 404

            peer = get_dm_other_user(dm_id, user["id"])
            if not peer:
                return jsonify({"error": "Direct-message peer not found."}), 404

            call_id = f"dm-{dm_id}"
            room_value = call_id

        elif room_slug:
            room = get_room_by_slug(room_slug)
            if not room:
                return jsonify({"error": "Room not found."}), 404
            if not can_access_room(room, user):
                return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403

            call_id = f"room-{room['slug']}"
            room_value = room["slug"]

        else:
            return jsonify({"error": "room or dm_id is required."}), 400

        payload = {
            "sub": f"user-{user['id']}",
            "username": user["username"],
            "room": room_value,
            "call_id": call_id,
            "role": user["role"],
            "can_call": True,
            "exp": int(time.time()) + CALL_TOKEN_TTL_SECONDS,
        }

        token = sign_call_helper_token(payload)

        audit(
            "call_helper_token_issued",
            user["id"],
            "call",
            call_id,
            f"room={room_value}",
        )

        return jsonify(
            {
                "token": token,
                "calls_base": CALLS_BASE,
                "call_id": call_id,
            }
        )

    @app.route("/api/group-calls/token", methods=["POST"])
    def api_group_call_token():
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        if not CALL_HELPER_SHARED_SECRET:
            return jsonify({"error": "Call helper is not configured."}), 503

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        scope_kind = str(data.get("scope_kind", "")).strip().lower()
        helper_call_id = None
        scope_id = None
        scope_slug = None
        room_value = None

        if scope_kind == "room":
            room_slug = str(data.get("room_slug", "")).strip()
            room = get_room_by_slug(room_slug)
            if not room:
                return jsonify({"error": "Room not found."}), 404
            if not can_access_room(room, user):
                return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403

            helper_call_id = build_room_helper_call_id(room)
            scope_id = room["id"]
            scope_slug = room["slug"]
            room_value = room["slug"]

        elif scope_kind == "channel":
            team_slug = str(data.get("team_slug", "")).strip()
            channel_slug = str(data.get("channel_slug", "")).strip()
            channel = get_channel_by_slug(team_slug, channel_slug)
            if not channel:
                return jsonify({"error": "Channel not found."}), 404
            if not can_access_channel(channel, user):
                return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403

            helper_call_id = build_channel_helper_call_id(channel)
            scope_id = channel["id"]
            scope_slug = f"{channel['team_slug']}/{channel['slug']}"
            room_value = scope_slug

        else:
            return jsonify({"error": "scope_kind must be room or channel."}), 400

        session_row = get_or_create_group_call_session(
            scope_kind=scope_kind,
            scope_id=scope_id,
            scope_slug=scope_slug,
            helper_call_id=helper_call_id,
            created_by_user_id=user["id"],
        )

        payload = {
            "sub": f"user-{user['id']}",
            "username": user["username"],
            "room": room_value,
            "call_id": helper_call_id,
            "role": user["role"],
            "can_call": True,
            "exp": int(time.time()) + CALL_TOKEN_TTL_SECONDS,
        }

        token = sign_call_helper_token(payload)

        audit(
            "group_call_token_issued",
            user["id"],
            "group_call_session",
            str(session_row["id"]),
            helper_call_id,
        )

        return jsonify(
            {
                "token": token,
                "calls_base": CALLS_BASE,
                "call_id": helper_call_id,
                "session": group_call_session_payload(session_row),
            }
        )

    @app.route("/api/group-calls/invite", methods=["POST"])
    def api_group_call_invite():
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        if not CALL_HELPER_SHARED_SECRET:
            return jsonify({"error": "Call helper is not configured."}), 503

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        session_id_raw = data.get("group_call_session_id")
        invite_username = normalize_username(data.get("username"))

        try:
            session_id = int(session_id_raw)
        except (TypeError, ValueError):
            return jsonify({"error": "Invalid group_call_session_id."}), 400

        if not invite_username:
            return jsonify({"error": "username is required."}), 400

        with get_db() as db:
            session_row = db.execute(
                """
                SELECT
                    id,
                    scope_kind,
                    scope_id,
                    scope_slug,
                    helper_call_id,
                    created_by_user_id,
                    status,
                    created_at,
                    ended_at
                FROM group_call_sessions
                WHERE id = ?
                LIMIT 1
                """,
                (session_id,),
            ).fetchone()

        if not session_row or session_row["status"] != "active":
            return jsonify({"error": "Active group call not found."}), 404

        allowed = False
        if session_row["scope_kind"] == "room":
            room = room_by_id(session_row["scope_id"])
            allowed = bool(room and can_access_room(room, user))
        elif session_row["scope_kind"] == "channel":
            channel = get_channel_by_id(session_row["scope_id"])
            allowed = bool(channel and can_access_channel(channel, user))

        if not allowed:
            return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403

        with get_db() as db:
            target = db.execute(
                """
                SELECT
                    id,
                    username,
                    role,
                    status
                FROM users
                WHERE username = ?
                """,
                (invite_username,),
            ).fetchone()

        if not target:
            return jsonify({"error": "User not found."}), 404

        if int(target["id"]) == int(user["id"]):
            return jsonify({"error": "You are already in the call."}), 400

        if target["status"] in {"disabled", "blacklisted"}:
            return jsonify({"error": "User is not available."}), 400

        payload = {
            "sub": f"user-{target['id']}",
            "username": target["username"],
            "room": session_row["scope_slug"],
            "call_id": session_row["helper_call_id"],
            "role": target["role"],
            "can_call": True,
            "exp": int(time.time()) + CALL_TOKEN_TTL_SECONDS,
        }

        helper_token = sign_call_helper_token(payload)
        expires_at = (utc_now() + ctx["timedelta"](minutes=10)).isoformat()

        cursor = db_write(
            """
            INSERT INTO group_call_invites (
                group_call_session_id,
                inviter_user_id,
                invitee_user_id,
                helper_token,
                status,
                created_at,
                expires_at,
                responded_at
            ) VALUES (?, ?, ?, ?, 'pending', ?, ?, NULL)
            """,
            (
                session_row["id"],
                user["id"],
                target["id"],
                helper_token,
                now_iso(),
                expires_at,
            ),
        )

        audit(
            "group_call_invite_created",
            user["id"],
            "group_call_session",
            str(session_row["id"]),
            f"invitee={target['username']}",
        )

        return jsonify(
            {
                "invite": {
                    "id": cursor.lastrowid,
                    "invitee_username": target["username"],
                    "group_call_session_id": session_row["id"],
                    "status": "pending",
                    "expires_at": expires_at,
                }
            }
        ), 201

    @app.route("/api/group-calls/invites")
    def api_list_group_call_invites():
        user, error = require_user()
        if error:
            return error

        with get_db() as db:
            rows = db.execute(
                """
                SELECT
                    i.id,
                    i.group_call_session_id,
                    i.status,
                    i.created_at,
                    i.expires_at,
                    i.responded_at,
                    s.scope_kind,
                    s.scope_slug,
                    s.helper_call_id,
                    inviter.username AS inviter_username
                FROM group_call_invites AS i
                JOIN group_call_sessions AS s
                    ON s.id = i.group_call_session_id
                JOIN users AS inviter
                    ON inviter.id = i.inviter_user_id
                WHERE i.invitee_user_id = ?
                  AND i.status = 'pending'
                  AND i.expires_at > ?
                ORDER BY i.id DESC
                """,
                (user["id"], now_iso()),
            ).fetchall()

        return jsonify(
            {
                "invites": [
                    {
                        "id": row["id"],
                        "group_call_session_id": row["group_call_session_id"],
                        "scope_kind": row["scope_kind"],
                        "scope_slug": row["scope_slug"],
                        "helper_call_id": row["helper_call_id"],
                        "inviter_username": row["inviter_username"],
                        "status": row["status"],
                        "created_at": row["created_at"],
                        "expires_at": row["expires_at"],
                    }
                    for row in rows
                ]
            }
        )

    @app.route("/api/group-calls/invites/<int:invite_id>/accept", methods=["POST"])
    def api_accept_group_call_invite(invite_id: int):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        with get_db() as db:
            row = db.execute(
                """
                SELECT
                    i.id,
                    i.group_call_session_id,
                    i.invitee_user_id,
                    i.helper_token,
                    i.status,
                    i.expires_at,
                    s.helper_call_id
                FROM group_call_invites AS i
                JOIN group_call_sessions AS s
                    ON s.id = i.group_call_session_id
                WHERE i.id = ?
                  AND i.invitee_user_id = ?
                LIMIT 1
                """,
                (invite_id, user["id"]),
            ).fetchone()

        if not row:
            return jsonify({"error": "Invite not found."}), 404

        if row["status"] != "pending" or parse_iso(row["expires_at"]) <= utc_now():
            return jsonify({"error": "Invite expired or unavailable."}), 400

        db_write(
            """
            UPDATE group_call_invites
            SET
                status = 'accepted',
                responded_at = ?
            WHERE id = ?
            """,
            (now_iso(), invite_id),
        )

        audit(
            "group_call_invite_accepted",
            user["id"],
            "group_call_invite",
            str(invite_id),
            row["helper_call_id"],
        )

        return jsonify(
            {
                "token": row["helper_token"],
                "calls_base": CALLS_BASE,
                "call_id": row["helper_call_id"],
            }
        )

    @app.route("/api/group-calls/invites/<int:invite_id>/dismiss", methods=["POST"])
    def api_dismiss_group_call_invite(invite_id: int):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_user()
        if error:
            return error

        db_write(
            """
            UPDATE group_call_invites
            SET
                status = 'dismissed',
                responded_at = ?
            WHERE id = ?
              AND invitee_user_id = ?
              AND status = 'pending'
            """,
            (now_iso(), invite_id, user["id"]),
        )

        audit(
            "group_call_invite_dismissed",
            user["id"],
            "group_call_invite",
            str(invite_id),
        )

        return jsonify({"status": "dismissed"})

    @app.route("/api/group-calls/<int:session_id>/end", methods=["POST"])
    def api_end_group_call_session(session_id: int):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        with get_db() as db:
            session_row = db.execute(
                """
                SELECT
                    id,
                    scope_kind,
                    scope_id,
                    helper_call_id,
                    status
                FROM group_call_sessions
                WHERE id = ?
                LIMIT 1
                """,
                (session_id,),
            ).fetchone()

        if not session_row:
            return jsonify({"error": "Group call session not found."}), 404

        allowed = False
        if session_row["scope_kind"] == "room":
            room = room_by_id(session_row["scope_id"])
            allowed = bool(room and can_manage_room(room, user))
        elif session_row["scope_kind"] == "channel":
            channel = get_channel_by_id(session_row["scope_id"])
            allowed = bool(channel and can_edit_team(channel["team_id"], user))

        if not allowed and user["role"] not in {"moderator", "admin"}:
            return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403

        db_write(
            """
            UPDATE group_call_sessions
            SET
                status = 'ended',
                ended_at = ?
            WHERE id = ?
            """,
            (now_iso(), session_id),
        )

        audit(
            "group_call_session_ended",
            user["id"],
            "group_call_session",
            str(session_id),
            session_row["helper_call_id"],
        )

        return jsonify({"status": "ended"})

    @app.route("/api/dms/<int:dm_id>/calls", methods=["POST"])
    def api_start_dm_call(dm_id: int):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        if not validate_dm_access(get_db, dm_id, user):
            return jsonify({"error": "DM not found."}), 404

        call_id = create_call_session(db_write, now_iso, dm_id, user["id"])
        audit("call_started", user["id"], "call", str(call_id), str(dm_id))
        return jsonify({"call": call_payload(get_db, call_id)}), 201

    @app.route("/api/calls/<int:call_id>")
    def api_get_call(call_id: int):
        user, error = require_user()
        if error:
            return error

        call = get_call_with_dm(get_db, call_id)
        if not call:
            return jsonify({"error": "Call not found."}), 404

        if not validate_dm_access(get_db, call["direct_conversation_id"], user):
            return jsonify({"error": "Call not found."}), 404

        return jsonify({"call": call_payload(get_db, call_id)})

    @app.route("/api/calls/<int:call_id>/signal", methods=["POST"])
    def api_post_call_signal(call_id: int):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        call = get_call_with_dm(get_db, call_id)
        if not call:
            return jsonify({"error": "Call not found."}), 404

        if not validate_dm_access(get_db, call["direct_conversation_id"], user):
            return jsonify({"error": "Call not found."}), 404

        if call["status"] != "active":
            return jsonify({"error": "Call is not active."}), 400

        try:
            data = safe_json()
        except ValueError as error:
            return jsonify({"error": str(error)}), 400

        signal_type = str(data.get("signal_type", "")).strip()[:40]
        payload = data.get("payload", {})

        if not signal_type:
            return jsonify({"error": "signal_type is required."}), 400
        if not isinstance(payload, dict):
            return jsonify({"error": "payload must be an object."}), 400

        signal_id = append_call_signal(db_write, now_iso, call_id, user["id"], signal_type, payload)
        return jsonify({"status": "ok", "signal_id": signal_id}), 201

    @app.route("/api/calls/<int:call_id>/signals")
    def api_get_call_signals(call_id: int):
        user, error = require_user()
        if error:
            return error

        call = get_call_with_dm(get_db, call_id)
        if not call:
            return jsonify({"error": "Call not found."}), 404

        if not validate_dm_access(get_db, call["direct_conversation_id"], user):
            return jsonify({"error": "Call not found."}), 404

        after_raw = str(request.args.get("after_id", "")).strip()
        after_id = 0
        if after_raw:
            try:
                after_id = max(0, int(after_raw))
            except ValueError:
                return jsonify({"error": "after_id must be a non-negative integer."}), 400

        signals = list_call_signals(get_db, call_id, after_id)
        latest_id = signals[-1]["id"] if signals else after_id
        return jsonify({"signals": signals, "latest_id": latest_id})

    @app.route("/api/calls/<int:call_id>/end", methods=["POST"])
    def api_end_call(call_id: int):
        if not csrf_valid():
            return jsonify({"error": "Security token expired.", "code": "csrf"}), 403

        user, error = require_writable_user()
        if error:
            return error

        call = get_call_with_dm(get_db, call_id)
        if not call:
            return jsonify({"error": "Call not found."}), 404

        if not validate_dm_access(get_db, call["direct_conversation_id"], user):
            return jsonify({"error": "Call not found."}), 404

        end_call(db_write, now_iso, call_id)
        audit("call_ended", user["id"], "call", str(call_id))
        return jsonify({"status": "ended", "call": call_payload(get_db, call_id)})