glo/features.py

# features.py
import json
import re
import time
from datetime import timedelta
from pathlib import Path
from typing import Optional

from flask import abort, jsonify, send_from_directory
import markdown

from sanitizer import sanitize_html_fragment

IMAGE_MIME_PREFIX = "image/"
AUDIO_MIME_PREFIX = "audio/"
VIDEO_MIME_PREFIX = "video/"

IMAGE_EXTENSIONS = {"png", "jpg", "jpeg", "gif", "webp"}
AUDIO_EXTENSIONS = {"mp3", "wav", "ogg", "m4a", "aac", "flac"}
VIDEO_EXTENSIONS = {"mp4", "webm", "mov", "m4v", "ogv"}

DEFAULT_ROLE_NAME_COLORS = {
    "admin": "#ff7b7b",
    "moderator": "#8ab4ff",
    "user": "#f5f5f7",
}

PROFILE_BIO_MODES = {"txt", "md", "html"}
THEME_MODES = {"dark", "light"}
PAGE_VISIBILITIES = {"team", "public"}
PAGE_TYPES = {"project", "document", "update", "resource"}
TEAM_MEMBER_ROLES = {"owner", "admin", "member"}
PROFILE_VISIBILITIES = {"public", "members", "private"}
LAST_ACTIVE_VISIBILITIES = {"everyone", "members", "nobody"}
PRESENCE_MODES = {"online", "silent", "focus", "dnd"}
REACTION_EMOJI_RE = re.compile(r"^.{1,16}$")
HEX_COLOR_RE = re.compile(r"^#[0-9a-fA-F]{6}$")
PAGE_LINK_RE = re.compile(r"\]\((?:/teams/[^/]+/pages/)?([a-z0-9-]{1,48})\)", re.IGNORECASE)


def valid_hex_color(value: object) -> bool:
    return bool(isinstance(value, str) and HEX_COLOR_RE.fullmatch(value.strip()))


def normalized_hex_color(value: object) -> Optional[str]:
    if not valid_hex_color(value):
        return None
    return str(value).strip().lower()


def render_markdown_to_safe_html(text: str) -> str:
    rendered = markdown.markdown(
        text or "",
        extensions=[
            "extra",
            "sane_lists",
            "nl2br",
            "tables",
            "fenced_code",
        ],
        output_format="html5",
    )
    return sanitize_html_fragment(rendered)


def safe_profile_bio_html(html_module, bio: str, bio_mode: str) -> str:
    mode = bio_mode if bio_mode in PROFILE_BIO_MODES else "txt"
    if mode == "html":
        return sanitize_html_fragment(bio or "")
    if mode == "md":
        return render_markdown_to_safe_html(bio or "")
    return "<p>" + html_module.escape(bio or "").replace("\n", "<br>") + "</p>"


def extract_page_reference_slugs(markdown_text: str) -> list[str]:
    text = str(markdown_text or "")
    matches = PAGE_LINK_RE.findall(text)
    seen = set()
    slugs = []
    for slug in matches:
        normalized = slug.lower().strip()
        if not normalized or normalized in seen:
            continue
        seen.add(normalized)
        slugs.append(normalized)
    return slugs


def attachment_kind_from_name_and_mime(
    original_name: Optional[str],
    mime_type: Optional[str],
) -> str:
    mime_type = str(mime_type or "").lower()
    name = str(original_name or "")
    ext = name.rsplit(".", 1)[-1].lower() if "." in name else ""
    if mime_type.startswith(IMAGE_MIME_PREFIX) or ext in IMAGE_EXTENSIONS:
        return "image"
    if mime_type.startswith(AUDIO_MIME_PREFIX) or ext in AUDIO_EXTENSIONS:
        return "audio"
    if mime_type.startswith(VIDEO_MIME_PREFIX) or ext in VIDEO_EXTENSIONS:
        return "video"
    return "file"


def load_page_children(get_db, page_id: int) -> list[dict]:
    with get_db() as db:
        rows = db.execute(
            """
            SELECT
                id,
                slug,
                title,
                updated_at
            FROM pages
            WHERE parent_page_id = ?
              AND is_active = 1
            ORDER BY title COLLATE NOCASE, id ASC
            """,
            (page_id,),
        ).fetchall()
    return [dict(row) for row in rows]


def load_page_backlinks(get_db, page_id: int, page_slug: str, team_id: int) -> list[dict]:
    pattern_1 = f"%]({page_slug})%"
    pattern_2 = f"%/pages/{page_slug})%"
    with get_db() as db:
        rows = db.execute(
            """
            SELECT
                id,
                slug,
                title,
                updated_at
            FROM pages
            WHERE team_id = ?
              AND is_active = 1
              AND id != ?
              AND (body LIKE ? OR body LIKE ?)
            ORDER BY updated_at DESC, id DESC
            """,
            (team_id, page_id, pattern_1, pattern_2),
        ).fetchall()
    return [dict(row) for row in rows]


def load_page_comments(get_db, page_id: int) -> list[dict]:
    with get_db() as db:
        rows = db.execute(
            """
            SELECT
                pc.id,
                pc.page_id,
                pc.user_id,
                pc.username,
                pc.body,
                pc.created_at,
                pc.updated_at,
                u.role AS user_role,
                us.name_color AS user_name_color
            FROM page_comments AS pc
            JOIN users AS u
                ON u.id = pc.user_id
            LEFT JOIN user_settings AS us
                ON us.user_id = u.id
            WHERE pc.page_id = ?
              AND pc.deleted_at IS NULL
            ORDER BY pc.created_at ASC, pc.id ASC
            """,
            (page_id,),
        ).fetchall()
    return [
        {
            "id": row["id"],
            "page_id": row["page_id"],
            "user_id": row["user_id"],
            "username": row["username"],
            "body": row["body"],
            "created_at": row["created_at"],
            "updated_at": row["updated_at"],
            "user_role": row["user_role"],
            "user_name_color": resolve_name_color(row["user_role"], row["user_name_color"]),
        }
        for row in rows
    ]


def next_page_version_number(get_db, page_id: int) -> int:
    with get_db() as db:
        row = db.execute(
            """
            SELECT COALESCE(MAX(version_number), 0) AS value
            FROM page_versions
            WHERE page_id = ?
            """,
            (page_id,),
        ).fetchone()
    return int(row["value"] or 0) + 1


def snapshot_page_version(
    get_db,
    page_id: int,
    title: str,
    body: str,
    visibility: str,
    page_type: str,
    channel_id: Optional[int],
    parent_page_id: Optional[int],
    created_by_user_id: int,
    created_at: str,
) -> int:
    version_number = next_page_version_number(get_db, page_id)
    with get_db() as db:
        db.execute("BEGIN IMMEDIATE")
        cursor = db.execute(
            """
            INSERT INTO page_versions (
                page_id,
                version_number,
                title,
                body,
                visibility,
                page_type,
                channel_id,
                parent_page_id,
                created_by_user_id,
                created_at
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                page_id,
                version_number,
                title,
                body,
                visibility,
                page_type,
                channel_id,
                parent_page_id,
                created_by_user_id,
                created_at,
            ),
        )
        db.commit()
    return cursor.lastrowid


def load_page_versions(get_db, page_id: int) -> list[dict]:
    with get_db() as db:
        rows = db.execute(
            """
            SELECT
                pv.id,
                pv.page_id,
                pv.version_number,
                pv.title,
                pv.visibility,
                pv.page_type,
                pv.channel_id,
                pv.parent_page_id,
                pv.created_by_user_id,
                pv.created_at,
                u.username AS created_by_username
            FROM page_versions AS pv
            JOIN users AS u
                ON u.id = pv.created_by_user_id
            WHERE pv.page_id = ?
            ORDER BY pv.version_number DESC
            """,
            (page_id,),
        ).fetchall()
    return [dict(row) for row in rows]


def load_page_version_detail(get_db, version_id: int):
    with get_db() as db:
        row = db.execute(
            """
            SELECT
                pv.id,
                pv.page_id,
                pv.version_number,
                pv.title,
                pv.body,
                pv.visibility,
                pv.page_type,
                pv.channel_id,
                pv.parent_page_id,
                pv.created_by_user_id,
                pv.created_at,
                u.username AS created_by_username
            FROM page_versions AS pv
            JOIN users AS u
                ON u.id = pv.created_by_user_id
            WHERE pv.id = ?
            """,
            (version_id,),
        ).fetchone()
    return row


def user_can_see_last_active(settings_row, viewer: Optional[dict], target_user_id: int, is_team_member_func=None) -> bool:
    visibility = (settings_row["last_active_visibility"] if settings_row and "last_active_visibility" in settings_row.keys() else "everyone") or "everyone"
    if visibility == "nobody":
        return False
    if visibility == "everyone":
        return True
    if visibility == "members":
        if not viewer:
            return False
        if viewer["id"] == target_user_id:
            return True
        if viewer["role"] in {"moderator", "admin"}:
            return True
        if is_team_member_func:
            return True
        return False
    return False


def page_payload(app_module, page, user: Optional[dict]) -> dict:
    can_read = bool(
        page["visibility"] == "public"
        or (
            user
            and (
                user["role"] in {"moderator", "admin"}
                or app_module["is_team_member"](page["team_id"], user["id"])
            )
        )
    )
    rendered_html = render_markdown_to_safe_html(page["body"] or "")
    return {
        "id": page["id"],
        "team_id": page["team_id"],
        "team_slug": page["team_slug"],
        "team_name": page["team_name"],
        "channel_id": page["channel_id"],
        "channel_slug": page["channel_slug"],
        "channel_name": page["channel_name"],
        "parent_page_id": page["parent_page_id"] if "parent_page_id" in page.keys() else None,
        "parent_page_slug": page["parent_page_slug"] if "parent_page_slug" in page.keys() else None,
        "parent_page_title": page["parent_page_title"] if "parent_page_title" in page.keys() else None,
        "slug": page["slug"],
        "title": page["title"],
        "body": page["body"],
        "body_html": rendered_html,
        "visibility": page["visibility"],
        "page_type": page["page_type"],
        "created_by_user_id": page["created_by_user_id"],
        "updated_by_user_id": page["updated_by_user_id"],
        "created_at": page["created_at"],
        "updated_at": page["updated_at"],
        "can_read": can_read,
        "can_edit": bool(
            user
            and (
                user["role"] in {"moderator", "admin"}
                or app_module["is_team_admin"](page["team_id"], user)
            )
        ),
        "children": load_page_children(app_module["get_db"], page["id"]),
        "backlinks": load_page_backlinks(
            app_module["get_db"],
            page["id"],
            page["slug"],
            page["team_id"],
        ),
        "references": extract_page_reference_slugs(page["body"] or ""),
        "comments": load_page_comments(app_module["get_db"], page["id"]),
    }


def attachment_payload_from_row(row) -> Optional[dict]:
    if not row or not row["id"]:
        return None
    kind = attachment_kind_from_name_and_mime(
        row["original_name"],
        row["mime_type"],
    )
    return {
        "id": row["id"],
        "original_name": row["original_name"],
        "mime_type": row["mime_type"],
        "size_bytes": row["size_bytes"],
        "kind": kind,
        "url": f"/api/attachments/{row['id']}/download",
    }


def attachment_payload_legacy_from_message_select_row(row) -> Optional[dict]:
    if not row["attachment_id"]:
        return None
    kind = attachment_kind_from_name_and_mime(
        row["attachment_original_name"],
        row["attachment_mime_type"],
    )
    return {
        "id": row["attachment_id"],
        "original_name": row["attachment_original_name"],
        "mime_type": row["attachment_mime_type"],
        "size_bytes": row["attachment_size_bytes"],
        "kind": kind,
        "url": f"/api/attachments/{row['attachment_id']}/download",
    }


def attachment_list_for_message(get_db, message_id: int) -> list[dict]:
    with get_db() as db:
        rows = db.execute(
            """
            SELECT
                a.id,
                a.original_name,
                a.mime_type,
                a.size_bytes
            FROM message_attachments AS ma
            JOIN attachments AS a
                ON a.id = ma.attachment_id
            WHERE ma.message_id = ?
            ORDER BY ma.position ASC, a.id ASC
            """,
            (message_id,),
        ).fetchall()
    return [attachment_payload_from_row(row) for row in rows]


def cleanup_orphaned_attachments(
    get_db,
    upload_dir: Path,
    utc_now,
    orphan_ttl_hours: int,
    app_logger,
) -> None:
    cutoff = (utc_now() - timedelta(hours=orphan_ttl_hours)).isoformat()
    with get_db() as db:
        orphaned = db.execute(
            """
            SELECT
                a.id,
                a.stored_name
            FROM attachments AS a
            WHERE a.created_at < ?
              AND NOT EXISTS (
                  SELECT 1
                  FROM messages AS m
                  WHERE m.attachment_id = a.id
              )
              AND NOT EXISTS (
                  SELECT 1
                  FROM message_attachments AS ma
                  WHERE ma.attachment_id = a.id
              )
            """,
            (cutoff,),
        ).fetchall()
        for attachment in orphaned:
            db.execute(
                """
                DELETE FROM attachments
                WHERE id = ?
                """,
                (attachment["id"],),
            )
        db.commit()

    for attachment in orphaned:
        try:
            (upload_dir / attachment["stored_name"]).unlink(missing_ok=True)
        except OSError:
            app_logger.warning(
                "Could not remove orphaned attachment file: %s",
                attachment["stored_name"],
            )


def maybe_cleanup_orphaned_attachments(
    state: dict,
    cleanup_interval_seconds: int,
    cleanup_func,
    app_logger,
) -> None:
    now = time.time()
    last = state.get("last_attachment_cleanup", 0.0)
    if now - last < cleanup_interval_seconds:
        return
    state["last_attachment_cleanup"] = now
    try:
        cleanup_func()
    except Exception:
        app_logger.exception("Could not clean orphaned attachments.")


def resolve_owned_attachment_ids(
    get_db,
    attachment_ids: object,
    user_id: int,
) -> tuple[Optional[list[int]], Optional[tuple]]:
    if attachment_ids is None:
        return [], None
    if not isinstance(attachment_ids, list):
        return None, (jsonify({"error": "attachment_ids must be an array."}), 400)

    parsed = []
    seen = set()
    for item in attachment_ids[:10]:
        try:
            attachment_id = int(item)
        except (TypeError, ValueError):
            return None, (jsonify({"error": "Invalid attachment."}), 400)
        if attachment_id in seen:
            continue
        seen.add(attachment_id)
        with get_db() as db:
            row = db.execute(
                """
                SELECT id, uploader_user_id
                FROM attachments
                WHERE id = ?
                """,
                (attachment_id,),
            ).fetchone()
        if not row:
            return None, (jsonify({"error": "Attachment not found."}), 404)
        if row["uploader_user_id"] != user_id:
            return None, (
                jsonify({"error": "You can only attach files you uploaded."}),
                403,
            )
        parsed.append(attachment_id)
    return parsed, None


def resolve_owned_attachment(
    get_db,
    attachment_id: object,
    user_id: int,
) -> tuple[Optional[int], Optional[tuple]]:
    if attachment_id is None:
        return None, None
    try:
        parsed_id = int(attachment_id)
    except (TypeError, ValueError):
        return None, (jsonify({"error": "Invalid attachment."}), 400)

    with get_db() as db:
        attachment = db.execute(
            """
            SELECT
                id,
                uploader_user_id
            FROM attachments
            WHERE id = ?
            """,
            (parsed_id,),
        ).fetchone()
    if not attachment:
        return None, (jsonify({"error": "Attachment not found."}), 404)
    if attachment["uploader_user_id"] != user_id:
        return None, (
            jsonify({"error": "You can only attach files you uploaded."}),
            403,
        )
    return parsed_id, None


def can_download_attachment(
    get_db,
    attachment_id: int,
    user: Optional[dict],
    can_access_room,
    can_access_channel,
    get_channel_by_id,
) -> bool:
    if not user:
        return False
    with get_db() as db:
        room_rows = db.execute(
            """
            SELECT DISTINCT
                r.id,
                r.slug,
                r.name,
                r.description,
                r.visibility,
                r.created_by_user_id,
                r.is_active
            FROM message_attachments AS ma
            JOIN messages AS m
                ON m.id = ma.message_id
            JOIN rooms AS r
                ON r.id = m.room_id
            WHERE ma.attachment_id = ?
              AND r.is_active = 1
            """,
            (attachment_id,),
        ).fetchall()
        channel_rows = db.execute(
            """
            SELECT DISTINCT
                m.channel_id
            FROM message_attachments AS ma
            JOIN messages AS m
                ON m.id = ma.message_id
            WHERE ma.attachment_id = ?
              AND m.channel_id IS NOT NULL
            """,
            (attachment_id,),
        ).fetchall()
        dm_rows = db.execute(
            """
            SELECT DISTINCT dc.id
            FROM message_attachments AS ma
            JOIN messages AS m
                ON m.id = ma.message_id
            JOIN direct_conversations AS dc
                ON dc.id = m.direct_conversation_id
            JOIN direct_conversation_members AS dcm
                ON dcm.direct_conversation_id = dc.id
            WHERE ma.attachment_id = ?
              AND dcm.user_id = ?
            """,
            (attachment_id, user["id"]),
        ).fetchall()

    if dm_rows:
        return True
    if any(can_access_room(room, user) for room in room_rows):
        return True
    for item in channel_rows:
        channel = get_channel_by_id(item["channel_id"])
        if channel and can_access_channel(channel, user):
            return True
    return False


def send_attachment_file(
    get_db,
    upload_dir: Path,
    attachment_id: int,
    user: Optional[dict],
    can_access_room,
    can_access_channel,
    get_channel_by_id,
):
    with get_db() as db:
        attachment = db.execute(
            """
            SELECT
                id,
                stored_name,
                original_name,
                mime_type
            FROM attachments
            WHERE id = ?
            """,
            (attachment_id,),
        ).fetchone()
    if not attachment:
        abort(404)

    if not can_download_attachment(
        get_db,
        attachment_id,
        user,
        can_access_room,
        can_access_channel,
        get_channel_by_id,
    ):
        abort(403)

    return send_from_directory(
        upload_dir,
        attachment["stored_name"],
        as_attachment=True,
        download_name=attachment["original_name"],
        mimetype=attachment["mime_type"],
    )


def role_default_name_color(role: Optional[str]) -> str:
    return DEFAULT_ROLE_NAME_COLORS.get(str(role or "user"), DEFAULT_ROLE_NAME_COLORS["user"])


def resolve_name_color(user_role: Optional[str], explicit_color: Optional[str]) -> str:
    return normalized_hex_color(explicit_color) or role_default_name_color(user_role)


def reaction_summary_for_message(get_db, message_id: int, user: Optional[dict]) -> list[dict]:
    with get_db() as db:
        rows = db.execute(
            """
            SELECT
                emoji,
                COUNT(*) AS count,
                MAX(CASE WHEN user_id = ? THEN 1 ELSE 0 END) AS reacted
            FROM message_reactions
            WHERE message_id = ?
            GROUP BY emoji
            ORDER BY count DESC, emoji ASC
            """,
            (user["id"] if user else -1, message_id),
        ).fetchall()
    return [
        {
            "emoji": row["emoji"],
            "count": row["count"],
            "reacted": bool(row["reacted"]),
        }
        for row in rows
    ]


def richer_message_payload(
    get_db,
    row,
    user: Optional[dict],
    current_user_id: Optional[int],
) -> dict:
    with get_db() as db:
        edited = db.execute(
            """
            SELECT created_at
            FROM message_edits
            WHERE message_id = ?
            ORDER BY id DESC
            LIMIT 1
            """,
            (row["id"],),
        ).fetchone()

        receipt_count = db.execute(
            """
            SELECT COUNT(*) AS count
            FROM message_receipts
            WHERE message_id = ?
            """,
            (row["id"],),
        ).fetchone()["count"]

        seen_by_user = False
        is_bookmarked = False
        if current_user_id:
            seen_by_user = bool(
                db.execute(
                    """
                    SELECT 1
                    FROM message_receipts
                    WHERE message_id = ?
                      AND user_id = ?
                    """,
                    (row["id"], current_user_id),
                ).fetchone()
            )
            is_bookmarked = bool(
                db.execute(
                    """
                    SELECT 1
                    FROM message_bookmarks
                    WHERE message_id = ?
                      AND user_id = ?
                    """,
                    (row["id"], current_user_id),
                ).fetchone()
            )

        pin_row = db.execute(
            """
            SELECT pinned_at, pinned_by_user_id
            FROM message_pins
            WHERE message_id = ?
            """,
            (row["id"],),
        ).fetchone()

    primary_attachment = attachment_payload_legacy_from_message_select_row(row)
    attachments = attachment_list_for_message(get_db, row["id"])
    if primary_attachment and not any(item["id"] == primary_attachment["id"] for item in attachments):
        attachments.insert(0, primary_attachment)

    user_role = row["user_role"] if "user_role" in row.keys() else "user"
    user_name_color = row["user_name_color"] if "user_name_color" in row.keys() else None
    resolved_color = resolve_name_color(user_role, user_name_color)

    return {
        "id": row["id"],
        "room_id": row["room_id"],
        "channel_id": row["channel_id"],
        "direct_conversation_id": row["direct_conversation_id"] if "direct_conversation_id" in row.keys() else None,
        "user_id": row["user_id"],
        "username": row["username"],
        "user_role": user_role,
        "user_name_color": resolved_color,
        "text": row["text"],
        "created_at": row["created_at"],
        "updated_at": edited["created_at"] if edited else None,
        "is_edited": bool(edited),
        "parent_message_id": row["parent_message_id"],
        "thread_reply_count": row["thread_reply_count"],
        "attachment": primary_attachment,
        "attachments": attachments,
        "read_receipt_count": receipt_count,
        "seen_by_current_user": seen_by_user,
        "reactions": reaction_summary_for_message(get_db, row["id"], user),
        "is_dm": bool(("direct_conversation_id" in row.keys()) and row["direct_conversation_id"]),
        "is_pinned": bool(pin_row),
        "pinned_at": pin_row["pinned_at"] if pin_row else None,
        "pinned_by_user_id": pin_row["pinned_by_user_id"] if pin_row else None,
        "is_bookmarked": is_bookmarked,
    }


def feed_message_payload(
    get_db,
    row,
    user: Optional[dict],
) -> dict:
    payload = richer_message_payload(
        get_db,
        row,
        user,
        user["id"] if user else None,
    )
    payload.update(
        {
            "room_slug": row["room_slug"] if "room_slug" in row.keys() else None,
            "room_name": row["room_name"] if "room_name" in row.keys() else None,
            "room_visibility": row["room_visibility"] if "room_visibility" in row.keys() else None,
            "team_slug": row["team_slug"] if "team_slug" in row.keys() else None,
            "team_name": row["team_name"] if "team_name" in row.keys() else None,
            "channel_slug": row["channel_slug"] if "channel_slug" in row.keys() else None,
            "channel_name": row["channel_name"] if "channel_name" in row.keys() else None,
            "dm_id": row["direct_conversation_id"] if "direct_conversation_id" in row.keys() else None,
            "is_mention": bool(
                user and f"@{user['username'].lower()}" in payload["text"].lower()
            ),
        }
    )
    return payload


def profile_payload(
    html_module,
    user_payload_func,
    user_row,
    profile_row,
    settings_row,
    viewer: Optional[dict],
) -> dict:
    is_self = bool(viewer and viewer["id"] == user_row["id"])
    status_payload = user_payload_func(user_row, settings_row)
    bio = profile_row["bio"] or ""
    bio_mode = profile_row["bio_mode"] or "txt"
    explicit_name_color = settings_row["name_color"] if settings_row else None
    resolved = resolve_name_color(user_row["role"], explicit_name_color)

    profile_visibility = settings_row["profile_visibility"] if settings_row else "public"
    if profile_visibility not in PROFILE_VISIBILITIES:
      profile_visibility = "public"

    last_active_visibility = settings_row["last_active_visibility"] if settings_row else "everyone"
    if last_active_visibility not in LAST_ACTIVE_VISIBILITIES:
      last_active_visibility = "everyone"

    presence_mode = settings_row["presence_mode"] if settings_row else "online"
    if presence_mode not in PRESENCE_MODES:
      presence_mode = "online"

    last_active_at = user_row["last_login_at"] if "last_login_at" in user_row.keys() else None

    can_show_last_active = False
    if last_active_visibility == "everyone":
        can_show_last_active = True
    elif last_active_visibility == "members":
        can_show_last_active = bool(viewer and (viewer["id"] == user_row["id"] or viewer["role"] in {"moderator", "admin"}))
    elif last_active_visibility == "nobody":
        can_show_last_active = bool(viewer and viewer["id"] == user_row["id"])

    return {
        "id": user_row["id"],
        "username": user_row["username"],
        "role": user_row["role"],
        "status": status_payload["status"],
        "presence_mode": presence_mode,
        "created_at": user_row["created_at"],
        "last_active_at": last_active_at if can_show_last_active else None,
        "display_name": profile_row["display_name"] if profile_row["display_name"] else user_row["username"],
        "badge_text": (profile_row["badge_text"] or "").strip() if "badge_text" in profile_row.keys() else "",
        "bio": bio,
        "bio_mode": bio_mode,
        "bio_html": safe_profile_bio_html(html_module, bio, bio_mode),
        "website": profile_row["website"] or "",
        "location": profile_row["location"] or "",
        "status_message": profile_row["status_message"] or "",
        "banner_attachment_id": profile_row["banner_attachment_id"],
        "banner_image_url": f"/api/attachments/{profile_row['banner_attachment_id']}/download" if profile_row["banner_attachment_id"] else "",
        "updated_at": profile_row["updated_at"],
        "is_self": is_self,
        "name_color": explicit_name_color,
        "resolved_name_color": resolved,
        "profile_background_color": settings_row["profile_background_color"] if settings_row else None,
        "profile_banner_color": settings_row["profile_banner_color"] if settings_row else None,
        "profile_visibility": profile_visibility,
        "last_active_visibility": last_active_visibility,
    }


def list_dms_for_user(get_db, user_id: int) -> list[dict]:
    with get_db() as db:
        rows = db.execute(
            """
            SELECT
                dc.id,
                dc.created_at,
                other.id AS other_user_id,
                other.username AS other_username,
                other.role AS other_role,
                us.name_color AS other_name_color,
                up.display_name AS other_display_name,
                up.status_message AS other_status_message,
                us.presence_mode AS other_presence_mode,
                (
                    SELECT m.text
                    FROM messages AS m
                    WHERE m.direct_conversation_id = dc.id
                    ORDER BY m.id DESC
                    LIMIT 1
                ) AS last_message_text,
                (
                    SELECT m.created_at
                    FROM messages AS m
                    WHERE m.direct_conversation_id = dc.id
                    ORDER BY m.id DESC
                    LIMIT 1
                ) AS last_message_created_at,
                (
                    SELECT m.id
                    FROM messages AS m
                    WHERE m.direct_conversation_id = dc.id
                    ORDER BY m.id DESC
                    LIMIT 1
                ) AS last_message_id
            FROM direct_conversations AS dc
            JOIN direct_conversation_members AS self_member
                ON self_member.direct_conversation_id = dc.id
            JOIN direct_conversation_members AS other_member
                ON other_member.direct_conversation_id = dc.id
               AND other_member.user_id != self_member.user_id
            JOIN users AS other
                ON other.id = other_member.user_id
            LEFT JOIN user_profiles AS up
                ON up.user_id = other.id
            LEFT JOIN user_settings AS us
                ON us.user_id = other.id
            WHERE self_member.user_id = ?
            ORDER BY COALESCE(last_message_id, 0) DESC, dc.id DESC
            """,
            (user_id,),
        ).fetchall()

    items = []
    for row in rows:
        items.append(
            {
                "id": row["id"],
                "created_at": row["created_at"],
                "other_user": {
                    "id": row["other_user_id"],
                    "username": row["other_username"],
                    "display_name": row["other_display_name"] or row["other_username"],
                    "role": row["other_role"],
                    "name_color": row["other_name_color"],
                    "resolved_name_color": resolve_name_color(row["other_role"], row["other_name_color"]),
                    "status_message": row["other_status_message"] or "",
                    "presence_mode": row["other_presence_mode"] or "online",
                },
                "last_message_text": row["last_message_text"],
                "last_message_created_at": row["last_message_created_at"],
                "last_message_id": row["last_message_id"],
            }
        )
    return items


def find_or_create_dm(get_db, db_write, now_iso, user_a_id: int, user_b_id: int) -> int:
    low_id, high_id = sorted([int(user_a_id), int(user_b_id)])
    with get_db() as db:
        existing = db.execute(
            """
            SELECT dc.id
            FROM direct_conversations AS dc
            JOIN direct_conversation_members AS m1
                ON m1.direct_conversation_id = dc.id AND m1.user_id = ?
            JOIN direct_conversation_members AS m2
                ON m2.direct_conversation_id = dc.id AND m2.user_id = ?
            WHERE dc.member_pair_key = ?
            LIMIT 1
            """,
            (low_id, high_id, f"{low_id}:{high_id}"),
        ).fetchone()
        if existing:
            return existing["id"]

    with get_db() as db:
        db.execute("BEGIN IMMEDIATE")
        existing = db.execute(
            """
            SELECT id
            FROM direct_conversations
            WHERE member_pair_key = ?
            LIMIT 1
            """,
            (f"{low_id}:{high_id}",),
        ).fetchone()
        if existing:
            db.commit()
            return existing["id"]

        cursor = db.execute(
            """
            INSERT INTO direct_conversations (
                member_pair_key,
                created_at
            ) VALUES (?, ?)
            """,
            (f"{low_id}:{high_id}", now_iso()),
        )
        dm_id = cursor.lastrowid
        db.execute(
            """
            INSERT INTO direct_conversation_members (
                direct_conversation_id,
                user_id,
                added_at
            ) VALUES (?, ?, ?)
            """,
            (dm_id, low_id, now_iso()),
        )
        db.execute(
            """
            INSERT INTO direct_conversation_members (
                direct_conversation_id,
                user_id,
                added_at
            ) VALUES (?, ?, ?)
            """,
            (dm_id, high_id, now_iso()),
        )
        db.commit()
        return dm_id


def get_dm_for_user(get_db, dm_id: int, user_id: int):
    with get_db() as db:
        row = db.execute(
            """
            SELECT
                dc.id,
                dc.created_at
            FROM direct_conversations AS dc
            JOIN direct_conversation_members AS dcm
                ON dcm.direct_conversation_id = dc.id
            WHERE dc.id = ?
              AND dcm.user_id = ?
            """,
            (dm_id, user_id),
        ).fetchone()
    return row


def get_dm_other_user(get_db, dm_id: int, user_id: int):
    with get_db() as db:
        row = db.execute(
            """
            SELECT
                u.id,
                u.username,
                u.role,
                up.display_name,
                up.status_message,
                us.name_color,
                us.presence_mode
            FROM direct_conversation_members AS dcm
            JOIN users AS u
                ON u.id = dcm.user_id
            LEFT JOIN user_profiles AS up
                ON up.user_id = u.id
            LEFT JOIN user_settings AS us
                ON us.user_id = u.id
            WHERE dcm.direct_conversation_id = ?
              AND dcm.user_id != ?
            LIMIT 1
            """,
            (dm_id, user_id),
        ).fetchone()
    return row


def validate_dm_access(get_db, dm_id: int, user: Optional[dict]) -> bool:
    if not user:
        return False
    return bool(get_dm_for_user(get_db, dm_id, user["id"]))


def dm_payload(get_db, dm_id: int, user_id: int) -> dict:
    dm = get_dm_for_user(get_db, dm_id, user_id)
    other = get_dm_other_user(get_db, dm_id, user_id)
    return {
        "id": dm["id"],
        "created_at": dm["created_at"],
        "other_user": {
            "id": other["id"],
            "username": other["username"],
            "display_name": other["display_name"] or other["username"],
            "role": other["role"],
            "name_color": other["name_color"],
            "resolved_name_color": resolve_name_color(other["role"], other["name_color"]),
            "status_message": other["status_message"] or "",
            "presence_mode": other["presence_mode"] or "online",
        },
    }


def update_presence(
    db_write,
    user_id: int,
    view_name: Optional[str] = None,
    room_id: Optional[int] = None,
    channel_id: Optional[int] = None,
    direct_conversation_id: Optional[int] = None,
) -> None:
    db_write(
        """
        INSERT INTO presence_events (
            user_id,
            updated_at,
            view_name,
            room_id,
            channel_id,
            direct_conversation_id
        ) VALUES (?, ?, ?, ?, ?, ?)
        ON CONFLICT(user_id)
        DO UPDATE SET
            updated_at = excluded.updated_at,
            view_name = excluded.view_name,
            room_id = excluded.room_id,
            channel_id = excluded.channel_id,
            direct_conversation_id = excluded.direct_conversation_id
        """,
        (
            user_id,
            time.time(),
            view_name,
            room_id,
            channel_id,
            direct_conversation_id,
        ),
    )


def room_presence_count(get_db, ttl_seconds: int, room_id: int) -> int:
    cutoff = time.time() - ttl_seconds
    with get_db() as db:
        row = db.execute(
            """
            SELECT COUNT(*) AS count
            FROM presence_events
            WHERE room_id = ?
              AND updated_at >= ?
            """,
            (room_id, cutoff),
        ).fetchone()
    return row["count"] if row else 0


def channel_presence_count(get_db, ttl_seconds: int, channel_id: int) -> int:
    cutoff = time.time() - ttl_seconds
    with get_db() as db:
        row = db.execute(
            """
            SELECT COUNT(*) AS count
            FROM presence_events
            WHERE channel_id = ?
              AND updated_at >= ?
            """,
            (channel_id, cutoff),
        ).fetchone()
    return row["count"] if row else 0


def dm_presence_count(get_db, ttl_seconds: int, dm_id: int) -> int:
    cutoff = time.time() - ttl_seconds
    with get_db() as db:
        row = db.execute(
            """
            SELECT COUNT(*) AS count
            FROM presence_events
            WHERE direct_conversation_id = ?
              AND updated_at >= ?
            """,
            (dm_id, cutoff),
        ).fetchone()
    return row["count"] if row else 0


def create_call_session(db_write, now_iso, dm_id: int, started_by_user_id: int) -> int:
    cursor = db_write(
        """
        INSERT INTO call_sessions (
            direct_conversation_id,
            started_by_user_id,
            status,
            created_at,
            ended_at
        ) VALUES (?, ?, 'active', ?, NULL)
        """,
        (dm_id, started_by_user_id, now_iso()),
    )
    return cursor.lastrowid


def append_call_signal(
    db_write,
    now_iso,
    call_id: int,
    from_user_id: int,
    signal_type: str,
    payload: dict,
) -> int:
    cursor = db_write(
        """
        INSERT INTO call_signals (
            call_session_id,
            from_user_id,
            signal_type,
            payload_json,
            created_at
        ) VALUES (?, ?, ?, ?, ?)
        """,
        (
            call_id,
            from_user_id,
            signal_type,
            json.dumps(payload or {}),
            now_iso(),
        ),
    )
    return cursor.lastrowid


def list_call_signals(get_db, call_id: int, after_id: int = 0) -> list[dict]:
    with get_db() as db:
        rows = db.execute(
            """
            SELECT
                cs.id,
                cs.call_session_id,
                cs.from_user_id,
                u.username,
                cs.signal_type,
                cs.payload_json,
                cs.created_at
            FROM call_signals AS cs
            JOIN users AS u
                ON u.id = cs.from_user_id
            WHERE cs.call_session_id = ?
              AND cs.id > ?
            ORDER BY cs.id ASC
            LIMIT 200
            """,
            (call_id, after_id),
        ).fetchall()

    items = []
    for row in rows:
        try:
            payload = json.loads(row["payload_json"] or "{}")
        except json.JSONDecodeError:
            payload = {}
        items.append(
            {
                "id": row["id"],
                "call_session_id": row["call_session_id"],
                "from_user_id": row["from_user_id"],
                "username": row["username"],
                "signal_type": row["signal_type"],
                "payload": payload,
                "created_at": row["created_at"],
            }
        )
    return items


def end_call(db_write, now_iso, call_id: int) -> None:
    db_write(
        """
        UPDATE call_sessions
        SET
            status = 'ended',
            ended_at = ?
        WHERE id = ?
        """,
        (now_iso(), call_id),
    )


def get_call_with_dm(get_db, call_id: int):
    with get_db() as db:
        row = db.execute(
            """
            SELECT
                cs.id,
                cs.direct_conversation_id,
                cs.started_by_user_id,
                cs.status,
                cs.created_at,
                cs.ended_at
            FROM call_sessions AS cs
            WHERE cs.id = ?
            """,
            (call_id,),
        ).fetchone()
    return row


def call_payload(get_db, call_id: int) -> Optional[dict]:
    row = get_call_with_dm(get_db, call_id)
    if not row:
        return None
    return {
        "id": row["id"],
        "direct_conversation_id": row["direct_conversation_id"],
        "started_by_user_id": row["started_by_user_id"],
        "status": row["status"],
        "created_at": row["created_at"],
        "ended_at": row["ended_at"],
    }


def dispatch_scheduled_messages(get_db, now_iso_func, app_logger) -> None:
    now_value = now_iso_func()
    with get_db() as db:
        rows = db.execute(
            """
            SELECT
                id,
                room_id,
                channel_id,
                direct_conversation_id,
                user_id,
                username,
                text,
                attachment_ids_json
            FROM scheduled_messages
            WHERE sent_at IS NULL
              AND cancelled_at IS NULL
              AND deliver_at <= ?
            ORDER BY deliver_at ASC, id ASC
            LIMIT 100
            """,
            (now_value,),
        ).fetchall()

        if not rows:
            return

        db.execute("BEGIN IMMEDIATE")
        for row in rows:
            try:
                attachment_ids = json.loads(row["attachment_ids_json"] or "[]")
            except json.JSONDecodeError:
                attachment_ids = []

            attachment_id = attachment_ids[0] if attachment_ids else None

            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, ?)
                """,
                (
                    row["room_id"],
                    row["channel_id"],
                    row["direct_conversation_id"],
                    row["user_id"],
                    row["username"],
                    row["text"],
                    attachment_id,
                    now_value,
                ),
            )
            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.execute(
                """
                UPDATE scheduled_messages
                SET sent_at = ?
                WHERE id = ?
                """,
                (now_value, row["id"]),
            )
        db.commit()