database.py

"""
database.py — SQLite persistence layer for B24 Messenger.
"""

import sqlite3
import os
import time
import json

DB_PATH = os.environ.get("DB_PATH", "/data/b24.db")

AI_USER_ID = 0
AI_USERNAME = "joy"
AI_HANDLE = "joy@b24.me"


def get_conn():
    os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
    conn = sqlite3.connect(DB_PATH, check_same_thread=False)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    return conn


def init_db():
    conn = get_conn()
    c = conn.cursor()

    c.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT UNIQUE NOT NULL,
            password_hash TEXT NOT NULL,
            handle TEXT UNIQUE NOT NULL,
            avatar_letter TEXT,
            created_at INTEGER NOT NULL
        )
    """)

    c.execute("""
        CREATE TABLE IF NOT EXISTS friendships (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER NOT NULL,
            friend_id INTEGER NOT NULL,
            status TEXT CHECK(status IN ('pending','accepted','blocked')) DEFAULT 'pending',
            requested_by INTEGER NOT NULL,
            created_at INTEGER NOT NULL,
            UNIQUE(user_id, friend_id),
            FOREIGN KEY(user_id) REFERENCES users(id),
            FOREIGN KEY(friend_id) REFERENCES users(id)
        )
    """)

    c.execute("""
        CREATE TABLE IF NOT EXISTS messages (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            sender_id INTEGER NOT NULL,
            recipient_id INTEGER NOT NULL,
            content TEXT NOT NULL,
            read_status INTEGER DEFAULT 0,
            timestamp INTEGER NOT NULL,
            FOREIGN KEY(sender_id) REFERENCES users(id),
            FOREIGN KEY(recipient_id) REFERENCES users(id)
        )
    """)

    c.execute("""
        CREATE TABLE IF NOT EXISTS groups (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            created_by INTEGER NOT NULL,
            created_at INTEGER NOT NULL,
            FOREIGN KEY(created_by) REFERENCES users(id)
        )
    """)

    c.execute("""
        CREATE TABLE IF NOT EXISTS group_members (
            group_id INTEGER NOT NULL,
            user_id INTEGER NOT NULL,
            role TEXT CHECK(role IN ('admin','member')) DEFAULT 'member',
            joined_at INTEGER NOT NULL,
            PRIMARY KEY(group_id, user_id),
            FOREIGN KEY(group_id) REFERENCES groups(id),
            FOREIGN KEY(user_id) REFERENCES users(id)
        )
    """)

    c.execute("""
        CREATE TABLE IF NOT EXISTS group_messages (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            group_id INTEGER NOT NULL,
            sender_id INTEGER NOT NULL,
            content TEXT NOT NULL,
            timestamp INTEGER NOT NULL,
            FOREIGN KEY(group_id) REFERENCES groups(id),
            FOREIGN KEY(sender_id) REFERENCES users(id)
        )
    """)

    c.execute("""
        CREATE TABLE IF NOT EXISTS group_message_reads (
            group_message_id INTEGER NOT NULL,
            user_id INTEGER NOT NULL,
            read_at INTEGER NOT NULL,
            PRIMARY KEY(group_message_id, user_id)
        )
    """)


    # Migrate existing DBs: add media columns if they don't exist yet (safe/idempotent)
    for table in ("messages", "group_messages"):
        existing_cols = [r["name"] for r in c.execute(f"PRAGMA table_info({table})").fetchall()]
        if "media_type" not in existing_cols:
            c.execute(f"ALTER TABLE {table} ADD COLUMN media_type TEXT DEFAULT 'text'")
        if "media_url" not in existing_cols:
            c.execute(f"ALTER TABLE {table} ADD COLUMN media_url TEXT")

    c.execute("""
        CREATE TABLE IF NOT EXISTS favorites (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER NOT NULL,
            url TEXT NOT NULL,
            kind TEXT CHECK(kind IN ('sticker','gif')) DEFAULT 'gif',
            created_at INTEGER NOT NULL,
            UNIQUE(user_id, url),
            FOREIGN KEY(user_id) REFERENCES users(id)
        )
    """)

    # Migrate groups: type distinguishes a normal group (anyone posts) from a
    # channel (only admins post, everyone else is read-only/follower)
    group_cols = [r["name"] for r in c.execute("PRAGMA table_info(groups)").fetchall()]
    if "type" not in group_cols:
        c.execute("ALTER TABLE groups ADD COLUMN type TEXT DEFAULT 'group'")
    if "description" not in group_cols:
        c.execute("ALTER TABLE groups ADD COLUMN description TEXT DEFAULT ''")
    if "invite_code" not in group_cols:
        c.execute("ALTER TABLE groups ADD COLUMN invite_code TEXT")

    # Migrate messages: is_request flags a message as pending approval (non-friend, non-business sender)
    msg_cols = [r["name"] for r in c.execute("PRAGMA table_info(messages)").fetchall()]
    if "is_request" not in msg_cols:
        c.execute("ALTER TABLE messages ADD COLUMN is_request INTEGER DEFAULT 0")
    if "expires_at" not in msg_cols:
        c.execute("ALTER TABLE messages ADD COLUMN expires_at INTEGER")

    # Migrate users: account_type (personal/business/business_noreply) + verified tick
    user_cols = [r["name"] for r in c.execute("PRAGMA table_info(users)").fetchall()]
    if "account_type" not in user_cols:
        c.execute("ALTER TABLE users ADD COLUMN account_type TEXT DEFAULT 'personal'")
    if "verified" not in user_cols:
        c.execute("ALTER TABLE users ADD COLUMN verified TEXT DEFAULT NULL")
    if "points" not in user_cols:
        c.execute("ALTER TABLE users ADD COLUMN points INTEGER DEFAULT 0")
    if "avatar_url" not in user_cols:
        c.execute("ALTER TABLE users ADD COLUMN avatar_url TEXT")
    if "banner_type" not in user_cols:
        c.execute("ALTER TABLE users ADD COLUMN banner_type TEXT DEFAULT 'color'")
    if "banner_value" not in user_cols:
        c.execute("ALTER TABLE users ADD COLUMN banner_value TEXT")
    if "bio" not in user_cols:
        c.execute("ALTER TABLE users ADD COLUMN bio TEXT")
    if "push_token" not in user_cols:
        c.execute("ALTER TABLE users ADD COLUMN push_token TEXT")
    if "privacy_settings" not in user_cols:
        c.execute("ALTER TABLE users ADD COLUMN privacy_settings TEXT")
    if "last_seen_at" not in user_cols:
        c.execute("ALTER TABLE users ADD COLUMN last_seen_at INTEGER")
    if "frozen_last_seen_at" not in user_cols:
        c.execute("ALTER TABLE users ADD COLUMN frozen_last_seen_at INTEGER")

    # Migrate friendships: rebuild with a 5-state status (pending/accepted/declined/blocked/canceled)
    friendship_cols = [r["name"] for r in c.execute("PRAGMA table_info(friendships)").fetchall()]
    needs_rebuild = True
    try:
        table_sql = c.execute(
            "SELECT sql FROM sqlite_master WHERE type='table' AND name='friendships'"
        ).fetchone()
        if table_sql and "'canceled'" in table_sql["sql"]:
            needs_rebuild = False
    except Exception:
        pass

    if needs_rebuild and friendship_cols:
        c.execute("""
            CREATE TABLE IF NOT EXISTS friendships_new (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id INTEGER NOT NULL,
                friend_id INTEGER NOT NULL,
                status TEXT CHECK(status IN ('pending','accepted','declined','blocked','canceled')) DEFAULT 'pending',
                requested_by INTEGER NOT NULL,
                created_at INTEGER NOT NULL,
                UNIQUE(user_id, friend_id),
                FOREIGN KEY(user_id) REFERENCES users(id),
                FOREIGN KEY(friend_id) REFERENCES users(id)
            )
        """)
        c.execute("""
            INSERT INTO friendships_new (id, user_id, friend_id, status, requested_by, created_at)
            SELECT id, user_id, friend_id, status, requested_by, created_at FROM friendships
        """)
        c.execute("DROP TABLE friendships")
        c.execute("ALTER TABLE friendships_new RENAME TO friendships")

    c.execute("SELECT id FROM users WHERE id = ?", (AI_USER_ID,))
    if not c.fetchone():
        c.execute(
            "INSERT INTO users (id, username, password_hash, handle, avatar_letter, created_at) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            (AI_USER_ID, AI_USERNAME, "!", AI_HANDLE, "J", int(time.time()))
        )

    c.execute("""
        CREATE TABLE IF NOT EXISTS calls (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            caller_id INTEGER NOT NULL,
            callee_id INTEGER NOT NULL,
            call_type TEXT NOT NULL CHECK(call_type IN ('voice','video')),
            status TEXT NOT NULL DEFAULT 'ringing',
            started_at INTEGER NOT NULL,
            answered_at INTEGER,
            ended_at INTEGER,
            duration INTEGER,
            FOREIGN KEY(caller_id) REFERENCES users(id),
            FOREIGN KEY(callee_id) REFERENCES users(id)
        )
    """)

    c.execute("""
        CREATE TABLE IF NOT EXISTS reports (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            reporter_id INTEGER NOT NULL,
            reported_user_id INTEGER NOT NULL,
            reason TEXT,
            created_at INTEGER NOT NULL,
            FOREIGN KEY(reporter_id) REFERENCES users(id),
            FOREIGN KEY(reported_user_id) REFERENCES users(id)
        )
    """)

    c.execute("""
        CREATE TABLE IF NOT EXISTS reports (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            reporter_id INTEGER NOT NULL,
            reported_user_id INTEGER NOT NULL,
            reason TEXT,
            created_at INTEGER NOT NULL,
            FOREIGN KEY(reporter_id) REFERENCES users(id),
            FOREIGN KEY(reported_user_id) REFERENCES users(id)
        )
    """)

    c.execute("""
        CREATE TABLE IF NOT EXISTS statuses (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER NOT NULL,
            content_type TEXT NOT NULL CHECK(content_type IN ('photo','video','text','voice','song')),
            media_url TEXT,
            text_content TEXT,
            bg_color TEXT,
            privacy TEXT NOT NULL DEFAULT 'all' CHECK(privacy IN ('all','except','only')),
            created_at INTEGER NOT NULL,
            expires_at INTEGER NOT NULL,
            FOREIGN KEY(user_id) REFERENCES users(id)
        )
    """)

    c.execute("""
        CREATE TABLE IF NOT EXISTS status_privacy_list (
            status_id INTEGER NOT NULL,
            user_id INTEGER NOT NULL,
            PRIMARY KEY(status_id, user_id),
            FOREIGN KEY(status_id) REFERENCES statuses(id),
            FOREIGN KEY(user_id) REFERENCES users(id)
        )
    """)

    c.execute("""
        CREATE TABLE IF NOT EXISTS status_views (
            status_id INTEGER NOT NULL,
            viewer_id INTEGER NOT NULL,
            viewed_at INTEGER NOT NULL,
            PRIMARY KEY(status_id, viewer_id),
            FOREIGN KEY(status_id) REFERENCES statuses(id),
            FOREIGN KEY(viewer_id) REFERENCES users(id)
        )
    """)

    c.execute("""
        CREATE TABLE IF NOT EXISTS completed_tasks (
            user_id INTEGER NOT NULL,
            task_key TEXT NOT NULL,
            points_awarded INTEGER NOT NULL,
            completed_at INTEGER NOT NULL,
            PRIMARY KEY (user_id, task_key),
            FOREIGN KEY(user_id) REFERENCES users(id)
        )
    """)

    c.execute("""
        CREATE TABLE IF NOT EXISTS tasks (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            key TEXT UNIQUE NOT NULL,
            label TEXT NOT NULL,
            points INTEGER NOT NULL,
            category TEXT DEFAULT 'General',
            active INTEGER NOT NULL DEFAULT 1,
            created_at INTEGER NOT NULL
        )
    """)

    _seed_tasks = [
        ("first_message", "Send your first message", 10, "Getting Started"),
        ("add_3_friends", "Add 3 friends", 20, "Getting Started"),
        ("first_group", "Create or join a group", 15, "Social"),
        ("first_status", "Post your first status", 10, "Social"),
        ("first_call", "Make your first call", 15, "Social"),
        ("add_10_friends", "Add 10 friends", 30, "Power User"),
    ]
    for _key, _label, _points, _category in _seed_tasks:
        c.execute(
            "INSERT OR IGNORE INTO tasks (key, label, points, category, active, created_at) VALUES (?, ?, ?, ?, 1, ?)",
            (_key, _label, _points, _category, int(time.time()))
        )

    c.execute("""
        CREATE TABLE IF NOT EXISTS announcements (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            body TEXT,
            created_at INTEGER NOT NULL,
            expires_at INTEGER,
            active INTEGER NOT NULL DEFAULT 1
        )
    """)

    c.execute("""
        CREATE TABLE IF NOT EXISTS app_version (
            id INTEGER PRIMARY KEY,
            version TEXT NOT NULL,
            apk_url TEXT NOT NULL,
            notes TEXT,
            updated_at INTEGER NOT NULL
        )
    """)

    c.execute("""
        CREATE TABLE IF NOT EXISTS ota_update (
            id INTEGER PRIMARY KEY,
            update_id TEXT NOT NULL,
            runtime_version TEXT NOT NULL,
            launch_asset_json TEXT NOT NULL,
            assets_json TEXT NOT NULL,
            notes TEXT,
            created_at INTEGER NOT NULL
        )
    """)

    conn.commit()
    conn.close()


def create_user(username, password_hash, handle, avatar_letter):
    """Returns the new user's id, or None if the username/handle was taken
    by a concurrent request (race-safe — relies on the UNIQUE constraint,
    not just the earlier existence check)."""
    conn = get_conn()
    try:
        cur = conn.execute(
            "INSERT INTO users (username, password_hash, handle, avatar_letter, created_at) "
            "VALUES (?, ?, ?, ?, ?)",
            (username, password_hash, handle, avatar_letter, int(time.time()))
        )
        conn.commit()
        return cur.lastrowid
    except sqlite3.IntegrityError:
        return None
    finally:
        conn.close()


def get_user_by_username(username):
    conn = get_conn()
    try:
        row = conn.execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone()
        return dict(row) if row else None
    finally:
        conn.close()


def set_account_type(user_id, account_type):
    if account_type not in ("personal", "business", "business_noreply"):
        return False
    conn = get_conn()
    try:
        conn.execute("UPDATE users SET account_type = ? WHERE id = ?", (account_type, user_id))
        conn.commit()
        return True
    finally:
        conn.close()


def set_verified(user_id, tick):
    """tick: None to remove, or 'purple'/'cyan'."""
    if tick not in (None, "purple", "cyan"):
        return False
    conn = get_conn()
    try:
        conn.execute("UPDATE users SET verified = ? WHERE id = ?", (tick, user_id))
        conn.commit()
        return True
    finally:
        conn.close()


def set_avatar(user_id, avatar_url):
    conn = get_conn()
    try:
        conn.execute("UPDATE users SET avatar_url = ? WHERE id = ?", (avatar_url, user_id))
        conn.commit()
        return True
    finally:
        conn.close()


def set_username(user_id, new_username):
    new_username = (new_username or "").strip()
    if len(new_username) < 3 or len(new_username) > 24:
        return False, "Username must be 3-24 characters"
    conn = get_conn()
    try:
        taken = conn.execute(
            "SELECT id FROM users WHERE username = ? AND id != ?", (new_username, user_id)
        ).fetchone()
        if taken:
            return False, "That username is already taken"
        new_handle = f"{new_username}@b24.me"
        conn.execute("UPDATE users SET username = ?, handle = ? WHERE id = ?", (new_username, new_handle, user_id))
        conn.commit()
        return True, None
    finally:
        conn.close()


def set_bio(user_id, bio):
    bio = (bio or "")[:150]
    conn = get_conn()
    try:
        conn.execute("UPDATE users SET bio = ? WHERE id = ?", (bio, user_id))
        conn.commit()
        return True
    finally:
        conn.close()


def set_banner(user_id, banner_type, banner_value):
    if banner_type not in ("color", "photo", "video"):
        return False
    conn = get_conn()
    try:
        conn.execute(
            "UPDATE users SET banner_type = ?, banner_value = ? WHERE id = ?",
            (banner_type, banner_value, user_id),
        )
        conn.commit()
        return True
    finally:
        conn.close()


DEFAULT_PRIVACY_SETTINGS = {
    "last_seen": True,
    "freeze_last_seen": False,
    "ghost_mode": False,
    "profile_photo_visible": True,
    "read_receipts": True,
    "anti_delete": False,
}


def get_privacy_settings(user_id):
    conn = get_conn()
    try:
        row = conn.execute("SELECT privacy_settings FROM users WHERE id = ?", (user_id,)).fetchone()
        stored = {}
        if row and row["privacy_settings"]:
            try:
                stored = json.loads(row["privacy_settings"])
            except (TypeError, ValueError):
                stored = {}
        merged = dict(DEFAULT_PRIVACY_SETTINGS)
        merged.update({k: v for k, v in stored.items() if k in DEFAULT_PRIVACY_SETTINGS})
        return merged
    finally:
        conn.close()


def set_privacy_settings(user_id, patch):
    current = get_privacy_settings(user_id)
    current.update({k: v for k, v in (patch or {}).items() if k in DEFAULT_PRIVACY_SETTINGS})
    conn = get_conn()
    try:
        conn.execute(
            "UPDATE users SET privacy_settings = ? WHERE id = ?",
            (json.dumps(current), user_id)
        )
        conn.commit()
        return current
    finally:
        conn.close()


def set_frozen_last_seen(user_id, frozen_at):
    """frozen_at: epoch milliseconds (int), or None to clear and go back to live last-seen."""
    conn = get_conn()
    try:
        conn.execute(
            "UPDATE users SET frozen_last_seen_at = ? WHERE id = ?",
            (frozen_at, user_id)
        )
        conn.commit()
    finally:
        conn.close()


def _avatar_or_hidden(owner_id, avatar_url, viewer_id=None):
    """Mask avatar_url for other users when the owner has profile_photo_visible off."""
    if viewer_id is not None and owner_id == viewer_id:
        return avatar_url
    if not get_privacy_settings(owner_id).get("profile_photo_visible", True):
        return None
    return avatar_url


def touch_last_seen(user_id):
    conn = get_conn()
    try:
        conn.execute("UPDATE users SET last_seen_at = ? WHERE id = ?", (int(time.time()), user_id))
        conn.commit()
    finally:
        conn.close()


def get_user_by_id(user_id):
    conn = get_conn()
    try:
        row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
        return dict(row) if row else None
    finally:
        conn.close()


def send_friend_request(user_id, friend_username):
    conn = get_conn()
    try:
        friend = conn.execute(
            "SELECT id FROM users WHERE username = ?", (friend_username,)
        ).fetchone()
        if not friend:
            return None, "User not found"
        friend_id = friend["id"]
        if friend_id == user_id:
            return None, "Can't add yourself"

        existing = conn.execute(
            "SELECT id FROM friendships WHERE (user_id=? AND friend_id=?) OR (user_id=? AND friend_id=?)",
            (user_id, friend_id, friend_id, user_id)
        ).fetchone()
        if existing:
            return None, "Request already exists"

        conn.execute(
            "INSERT INTO friendships (user_id, friend_id, status, requested_by, created_at) "
            "VALUES (?, ?, 'pending', ?, ?)",
            (user_id, friend_id, user_id, int(time.time()))
        )
        conn.commit()
        return friend_id, None
    finally:
        conn.close()


def respond_friend_request(user_id, requester_id, action):
    """action: 'accept', 'decline', or 'block'.
    Decline lets them retry later; block is silent and permanent."""
    conn = get_conn()
    try:
        status_map = {"accept": "accepted", "decline": "declined", "block": "blocked"}
        status = status_map.get(action)
        if not status:
            return False
        conn.execute(
            "UPDATE friendships SET status = ? WHERE user_id = ? AND friend_id = ? AND status = 'pending'",
            (status, requester_id, user_id)
        )
        conn.commit()
        return True
    finally:
        conn.close()


def block_friend(user_id, target_id):
    """Blocks target_id regardless of prior friendship state (works even
    on an already-accepted friendship, unlike respond_friend_request)."""
    conn = get_conn()
    try:
        existing = conn.execute(
            "SELECT id FROM friendships WHERE (user_id=? AND friend_id=?) OR (user_id=? AND friend_id=?)",
            (user_id, target_id, target_id, user_id)
        ).fetchone()
        if existing:
            conn.execute("UPDATE friendships SET status='blocked' WHERE id=?", (existing["id"],))
        else:
            conn.execute(
                "INSERT INTO friendships (user_id, friend_id, status, requested_by, created_at) VALUES (?, ?, 'blocked', ?, ?)",
                (user_id, target_id, user_id, int(time.time()))
            )
        conn.commit()
        return True
    finally:
        conn.close()


def remove_friend(user_id, target_id):
    """Deletes the friendship row entirely (unfriend)."""
    conn = get_conn()
    try:
        conn.execute(
            "DELETE FROM friendships WHERE (user_id=? AND friend_id=?) OR (user_id=? AND friend_id=?)",
            (user_id, target_id, target_id, user_id)
        )
        conn.commit()
        return True
    finally:
        conn.close()


def list_friends(user_id):
    conn = get_conn()
    try:
        rows = conn.execute("""
            SELECT u.id, u.username, u.handle, u.avatar_letter, u.avatar_url, f.status
            FROM friendships f
            JOIN users u ON u.id = CASE WHEN f.user_id = ? THEN f.friend_id ELSE f.user_id END
            WHERE (f.user_id = ? OR f.friend_id = ?) AND f.status = 'accepted'
        """, (user_id, user_id, user_id)).fetchall()
        result = [dict(r) for r in rows]
        for r in result:
            r["avatar_url"] = _avatar_or_hidden(r["id"], r["avatar_url"], user_id)
        return result
    finally:
        conn.close()


def list_pending_requests(user_id):
    conn = get_conn()
    try:
        rows = conn.execute("""
            SELECT f.id as request_id, u.id, u.username, u.handle
            FROM friendships f
            JOIN users u ON u.id = f.requested_by
            WHERE f.friend_id = ? AND f.status = 'pending'
        """, (user_id,)).fetchall()
        return [dict(r) for r in rows]
    finally:
        conn.close()


def list_sent_requests(user_id):
    """Requests *I* sent, grouped by outcome. 'Failed' surfaces requests the
    other person declined -- declined-from-their-side is a failed request
    from mine. 'Canceled' is a request I pulled back before they answered."""
    conn = get_conn()
    try:
        rows = conn.execute("""
            SELECT f.id as request_id, f.status, f.created_at,
                   u.id, u.username, u.handle
            FROM friendships f
            JOIN users u ON u.id = f.friend_id
            WHERE f.requested_by = ? AND f.status IN ('pending', 'canceled', 'declined')
            ORDER BY f.created_at DESC
        """, (user_id,)).fetchall()
        result = {"pending": [], "canceled": [], "failed": []}
        status_key = {"pending": "pending", "canceled": "canceled", "declined": "failed"}
        for r in rows:
            result[status_key[r["status"]]].append(dict(r))
        return result
    finally:
        conn.close()


def cancel_sent_request(user_id, friend_id):
    """Lets the sender withdraw their own still-pending outgoing request."""
    conn = get_conn()
    try:
        cur = conn.execute(
            "UPDATE friendships SET status = 'canceled' "
            "WHERE user_id = ? AND friend_id = ? AND requested_by = ? AND status = 'pending'",
            (user_id, friend_id, user_id)
        )
        conn.commit()
        return cur.rowcount > 0
    finally:
        conn.close()


def save_message(sender_id, recipient_id, content, media_type='text', media_url=None, is_request=0, expires_at=None):
    conn = get_conn()
    try:
        cur = conn.execute(
            "INSERT INTO messages (sender_id, recipient_id, content, timestamp, media_type, media_url, is_request, expires_at) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            (sender_id, recipient_id, content, int(time.time()), media_type, media_url, is_request, expires_at)
        )
        conn.commit()
        return cur.lastrowid
    finally:
        conn.close()


def set_push_token(user_id, token):
    conn = get_conn()
    try:
        conn.execute("UPDATE users SET push_token = ? WHERE id = ?", (token, user_id))
        conn.commit()
    finally:
        conn.close()


def get_push_token(user_id):
    conn = get_conn()
    try:
        row = conn.execute("SELECT push_token FROM users WHERE id = ?", (user_id,)).fetchone()
        return row["push_token"] if row and row["push_token"] else None
    finally:
        conn.close()


def are_friends(user_a, user_b):
    conn = get_conn()
    try:
        row = conn.execute(
            "SELECT id FROM friendships WHERE "
            "((user_id=? AND friend_id=?) OR (user_id=? AND friend_id=?)) AND status='accepted'",
            (user_a, user_b, user_b, user_a)
        ).fetchone()
        return row is not None
    finally:
        conn.close()


def can_message_directly(sender_id, recipient_id):
    """True if the sender can message without landing in the recipient's
    Message Requests: they're already friends, sender is JOY, or sender is
    a verified business account."""
    if sender_id == AI_USER_ID or recipient_id == AI_USER_ID:
        return True
    if are_friends(sender_id, recipient_id):
        return True

    sender = get_user_by_id(sender_id)
    if sender and sender.get("verified") and sender.get("account_type") in ("business", "business_noreply"):
        return True
    return False


def get_conversation(user_a, user_b, limit=100):
    conn = get_conn()
    try:
        now = int(time.time())
        conn.execute("""
            DELETE FROM messages
            WHERE expires_at IS NOT NULL AND expires_at <= ?
            AND ((sender_id = ? AND recipient_id = ?) OR (sender_id = ? AND recipient_id = ?))
        """, (now, user_a, user_b, user_b, user_a))
        conn.commit()
        rows = conn.execute("""
            SELECT * FROM messages
            WHERE (sender_id = ? AND recipient_id = ?) OR (sender_id = ? AND recipient_id = ?)
            ORDER BY timestamp DESC LIMIT ?
        """, (user_a, user_b, user_b, user_a, limit)).fetchall()
        return [dict(r) for r in reversed(rows)]
    finally:
        conn.close()


def mark_read(user_id, from_id):
    conn = get_conn()
    try:
        conn.execute(
            "UPDATE messages SET read_status = 1 WHERE sender_id = ? AND recipient_id = ?",
            (from_id, user_id)
        )
        conn.commit()
    finally:
        conn.close()


def get_pending_messages(user_id):
    """1:1 messages waiting on the server for this user's phone to pick up."""
    conn = get_conn()
    try:
        rows = conn.execute(
            "SELECT * FROM messages WHERE recipient_id = ? ORDER BY timestamp ASC",
            (user_id,)
        ).fetchall()
        return [dict(r) for r in rows]
    finally:
        conn.close()


def ack_messages(user_id, message_ids):
    """Recipient confirms these are saved locally -> safe to delete server-side.
    JOY threads are never deleted; JOY needs that history to answer well."""
    if not message_ids:
        return 0
    conn = get_conn()
    try:
        placeholders = ",".join("?" for _ in message_ids)
        cur = conn.execute(
            f"DELETE FROM messages WHERE id IN ({placeholders}) "
            f"AND recipient_id = ? AND sender_id != ? AND recipient_id != ?",
            (*message_ids, user_id, AI_USER_ID, AI_USER_ID)
        )
        conn.commit()
        return cur.rowcount
    finally:
        conn.close()


def purge_stale_messages(days=30):
    """Safety net for messages nobody ever acked (uninstalled app, dead account).
    Never touches JOY conversations."""
    cutoff = int(time.time()) - days * 86400
    conn = get_conn()
    try:
        cur = conn.execute(
            "DELETE FROM messages WHERE timestamp < ? AND sender_id != ? AND recipient_id != ?",
            (cutoff, AI_USER_ID, AI_USER_ID)
        )
        conn.commit()
        return cur.rowcount
    finally:
        conn.close()


def create_group(name, created_by, member_ids, group_type='group', description=''):
    import secrets
    if group_type not in ('group', 'channel', 'broadcast'):
        group_type = 'group'
    invite_code = secrets.token_urlsafe(6)
    conn = get_conn()
    try:
        cur = conn.execute(
            "INSERT INTO groups (name, created_by, created_at, type, description, invite_code) VALUES (?, ?, ?, ?, ?, ?)",
            (name, created_by, int(time.time()), group_type, description, invite_code)
        )
        group_id = cur.lastrowid
        conn.execute(
            "INSERT INTO group_members (group_id, user_id, role, joined_at) VALUES (?, ?, 'admin', ?)",
            (group_id, created_by, int(time.time()))
        )
        for uid in member_ids:
            if uid != created_by:
                conn.execute(
                    "INSERT OR IGNORE INTO group_members (group_id, user_id, role, joined_at) "
                    "VALUES (?, ?, 'member', ?)",
                    (group_id, uid, int(time.time()))
                )
        conn.commit()
        return group_id, invite_code
    finally:
        conn.close()


def add_group_members(group_id, user_ids, added_by):
    """Adds users to an existing group. Returns the list of ids actually added."""
    conn = get_conn()
    try:
        member = conn.execute(
            "SELECT role FROM group_members WHERE group_id = ? AND user_id = ?",
            (group_id, added_by)
        ).fetchone()
        if not member:
            return None  # not a member — caller should treat as forbidden
        added = []
        for uid in user_ids:
            existing = conn.execute(
                "SELECT 1 FROM group_members WHERE group_id = ? AND user_id = ?", (group_id, uid)
            ).fetchone()
            if existing:
                continue
            conn.execute(
                "INSERT INTO group_members (group_id, user_id, role, joined_at) VALUES (?, ?, 'member', ?)",
                (group_id, uid, int(time.time()))
            )
            added.append(uid)
        conn.commit()
        return added
    finally:
        conn.close()


def get_group_by_invite_code(code):
    conn = get_conn()
    try:
        row = conn.execute("SELECT * FROM groups WHERE invite_code = ?", (code,)).fetchone()
        return dict(row) if row else None
    finally:
        conn.close()


def get_group_members(group_id):
    conn = get_conn()
    try:
        rows = conn.execute(
            "SELECT user_id, role FROM group_members WHERE group_id = ?", (group_id,)
        ).fetchall()
        return [dict(r) for r in rows]
    finally:
        conn.close()


def save_group_message(group_id, sender_id, content, media_type='text', media_url=None):
    conn = get_conn()
    try:
        cur = conn.execute(
            "INSERT INTO group_messages (group_id, sender_id, content, timestamp, media_type, media_url) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            (group_id, sender_id, content, int(time.time()), media_type, media_url)
        )
        conn.commit()
        return cur.lastrowid
    finally:
        conn.close()


def get_group_conversation(group_id, limit=100):
    conn = get_conn()
    try:
        rows = conn.execute("""
            SELECT gm.*, u.username as sender_username, u.verified as sender_verified
            FROM group_messages gm
            JOIN users u ON u.id = gm.sender_id
            WHERE gm.group_id = ? ORDER BY gm.timestamp DESC LIMIT ?
        """, (group_id, limit)).fetchall()
        result = []
        for r in reversed(rows):
            d = dict(r)
            d["sender_name"] = d["sender_username"]
            d["sender_color"] = _color_for_id(d["sender_id"])
            result.append(d)
        return result
    finally:
        conn.close()


def list_user_groups(user_id):
    conn = get_conn()
    try:
        rows = conn.execute("""
            SELECT g.id, g.name, g.type FROM groups g
            JOIN group_members gm ON gm.group_id = g.id
            WHERE gm.user_id = ?
        """, (user_id,)).fetchall()
        return [dict(r) for r in rows]
    finally:
        conn.close()


def get_pending_group_messages(user_id):
    """Group messages this user hasn't acked yet, across every group they're in."""
    conn = get_conn()
    try:
        group_ids = [g["id"] for g in conn.execute(
            "SELECT group_id as id FROM group_members WHERE user_id = ?", (user_id,)
        ).fetchall()]
        if not group_ids:
            return []
        placeholders = ",".join("?" for _ in group_ids)
        rows = conn.execute(f"""
            SELECT gm.* FROM group_messages gm
            WHERE gm.group_id IN ({placeholders})
            AND gm.id NOT IN (
                SELECT group_message_id FROM group_message_reads WHERE user_id = ?
            )
            ORDER BY gm.timestamp ASC
        """, (*group_ids, user_id)).fetchall()
        return [dict(r) for r in rows]
    finally:
        conn.close()


def ack_group_messages(user_id, message_ids):
    """Member confirms these are saved locally. Once every current member has
    acked a message, it's deleted from the server."""
    if not message_ids:
        return 0
    conn = get_conn()
    try:
        for mid in message_ids:
            conn.execute(
                "INSERT OR IGNORE INTO group_message_reads (group_message_id, user_id, read_at) VALUES (?, ?, ?)",
                (mid, user_id, int(time.time()))
            )
        conn.commit()

        placeholders = ",".join("?" for _ in message_ids)
        rows = conn.execute(
            f"SELECT DISTINCT group_message_id FROM group_message_reads WHERE group_message_id IN ({placeholders})",
            message_ids
        ).fetchall()
        for row in rows:
            mid = row["group_message_id"]
            gm = conn.execute("SELECT group_id FROM group_messages WHERE id = ?", (mid,)).fetchone()
            if not gm:
                continue
            member_ids = {m["user_id"] for m in get_group_members(gm["group_id"])}
            acked_ids = {r["user_id"] for r in conn.execute(
                "SELECT user_id FROM group_message_reads WHERE group_message_id = ?", (mid,)
            ).fetchall()}
            if member_ids and member_ids.issubset(acked_ids):
                conn.execute("DELETE FROM group_messages WHERE id = ?", (mid,))
                conn.execute("DELETE FROM group_message_reads WHERE group_message_id = ?", (mid,))
        conn.commit()
        return len(rows)
    finally:
        conn.close()


def purge_stale_group_messages(days=30):
    cutoff = int(time.time()) - days * 86400
    conn = get_conn()
    try:
        cur = conn.execute("DELETE FROM group_messages WHERE timestamp < ?", (cutoff,))
        conn.execute("""
            DELETE FROM group_message_reads WHERE group_message_id NOT IN (
                SELECT id FROM group_messages
            )
        """)
        conn.commit()
        return cur.rowcount
    finally:
        conn.close()


# ---------- admin ----------

def count_users():
    conn = get_conn()
    try:
        return conn.execute("SELECT COUNT(*) as c FROM users").fetchone()["c"]
    finally:
        conn.close()


def count_messages():
    conn = get_conn()
    try:
        direct = conn.execute("SELECT COUNT(*) as c FROM messages").fetchone()["c"]
        group = conn.execute("SELECT COUNT(*) as c FROM group_messages").fetchone()["c"]
        return direct + group
    finally:
        conn.close()


def count_groups():
    conn = get_conn()
    try:
        return conn.execute("SELECT COUNT(*) as c FROM groups").fetchone()["c"]
    finally:
        conn.close()


def list_all_users(limit=100, offset=0):
    conn = get_conn()
    try:
        rows = conn.execute(
            "SELECT id, username, handle, avatar_letter, avatar_url, created_at, verified, account_type FROM users "
            "ORDER BY created_at DESC LIMIT ? OFFSET ?",
            (limit, offset)
        ).fetchall()
        return [dict(r) for r in rows]
    finally:
        conn.close()


def db_file_size_bytes():
    try:
        return os.path.getsize(DB_PATH)
    except OSError:
        return 0


def get_group_type(group_id):
    conn = get_conn()
    try:
        row = conn.execute("SELECT type FROM groups WHERE id = ?", (group_id,)).fetchone()
        return row["type"] if row else None
    finally:
        conn.close()


def get_member_role(group_id, user_id):
    conn = get_conn()
    try:
        row = conn.execute(
            "SELECT role FROM group_members WHERE group_id = ? AND user_id = ?",
            (group_id, user_id)
        ).fetchone()
        return row["role"] if row else None
    finally:
        conn.close()

def add_favorite(user_id, url, kind='gif'):
    conn = get_conn()
    try:
        conn.execute(
            "INSERT OR IGNORE INTO favorites (user_id, url, kind, created_at) VALUES (?, ?, ?, ?)",
            (user_id, url, kind, int(time.time()))
        )
        conn.commit()
        return True
    finally:
        conn.close()


def remove_favorite(user_id, url):
    conn = get_conn()
    try:
        conn.execute("DELETE FROM favorites WHERE user_id = ? AND url = ?", (user_id, url))
        conn.commit()
        return True
    finally:
        conn.close()


def list_favorites(user_id):
    conn = get_conn()
    try:
        rows = conn.execute(
            "SELECT url, kind, created_at FROM favorites WHERE user_id = ? ORDER BY created_at DESC",
            (user_id,)
        ).fetchall()
        return [dict(r) for r in rows]
    finally:
        conn.close()

def create_call(caller_id, callee_id, call_type):
    conn = get_conn()
    try:
        cur = conn.execute(
            "INSERT INTO calls (caller_id, callee_id, call_type, started_at) VALUES (?, ?, ?, ?)",
            (caller_id, callee_id, call_type, int(time.time()))
        )
        conn.commit()
        return cur.lastrowid
    finally:
        conn.close()


def update_call_status(call_id, status):
    conn = get_conn()
    try:
        if status == 'answered':
            conn.execute("UPDATE calls SET status=?, answered_at=? WHERE id=?", (status, int(time.time()), call_id))
        else:
            conn.execute("UPDATE calls SET status=? WHERE id=?", (status, call_id))
        conn.commit()
    finally:
        conn.close()


def end_call(call_id):
    conn = get_conn()
    try:
        row = conn.execute("SELECT answered_at FROM calls WHERE id=?", (call_id,)).fetchone()
        now = int(time.time())
        duration = (now - row["answered_at"]) if row and row["answered_at"] else None
        conn.execute(
            "UPDATE calls SET status='ended', ended_at=?, duration=? WHERE id=?",
            (now, duration, call_id)
        )
        conn.commit()
    finally:
        conn.close()


def get_call_history(user_id, limit=50):
    conn = get_conn()
    try:
        rows = conn.execute(
            "SELECT * FROM calls WHERE caller_id=? OR callee_id=? ORDER BY started_at DESC LIMIT ?",
            (user_id, user_id, limit)
        ).fetchall()
        return [dict(r) for r in rows]
    finally:
        conn.close()


def _color_for_id(seed):
    """Deterministic pastel-ish color per user/group id, so the same
    contact always shows the same avatar color across devices."""
    palette = ["#9333ea", "#f97316", "#0ea5a4", "#4f46e5", "#db2777",
               "#16a34a", "#ea580c", "#0891b2", "#7c3aed", "#c026d3"]
    return palette[seed % len(palette)]


def get_chat_list(user_id):
    conn = get_conn()
    try:
        chats = []

        # --- JOY, always present ---
        joy_last = conn.execute("""
            SELECT content, media_type, timestamp FROM messages
            WHERE (sender_id = ? AND recipient_id = ?) OR (sender_id = ? AND recipient_id = ?)
            ORDER BY timestamp DESC LIMIT 1
        """, (user_id, AI_USER_ID, AI_USER_ID, user_id)).fetchone()

        chats.append({
            "id": str(AI_USER_ID),
            "name": "Joy",
            "preview": _preview_text(joy_last),
            "time": joy_last["timestamp"] if joy_last else 0,
            "unread": 0,
            "color": "#9333ea",
            "category": "all",
            "isGroup": False,
        })

        # --- 1:1 friends ---
        friends = conn.execute("""
            SELECT u.id, u.username, u.account_type, u.verified, u.avatar_url, u.bio
            FROM friendships f
            JOIN users u ON u.id = CASE WHEN f.user_id = ? THEN f.friend_id ELSE f.user_id END
            WHERE (f.user_id = ? OR f.friend_id = ?) AND f.status = 'accepted'
        """, (user_id, user_id, user_id)).fetchall()

        for friend in friends:
            fid = friend["id"]
            last_msg = conn.execute("""
                SELECT content, media_type, timestamp FROM messages
                WHERE (sender_id = ? AND recipient_id = ?) OR (sender_id = ? AND recipient_id = ?)
                ORDER BY timestamp DESC LIMIT 1
            """, (user_id, fid, fid, user_id)).fetchone()

            unread = conn.execute("""
                SELECT COUNT(*) as c FROM messages
                WHERE sender_id = ? AND recipient_id = ? AND read_status = 0
            """, (fid, user_id)).fetchone()["c"]

            category = "business" if friend["account_type"] in ("business", "business_noreply") else "all"

            chats.append({
                "id": str(fid),
                "name": friend["username"],
                "preview": _preview_text(last_msg),
                "time": last_msg["timestamp"] if last_msg else 0,
                "unread": unread,
                "color": _color_for_id(fid),
                "category": category,
                "isGroup": False,
                "verified": friend["verified"],
                "avatarUrl": _avatar_or_hidden(fid, friend["avatar_url"], user_id),
                "bio": friend["bio"],
            })

        # --- Groups ---
        groups = conn.execute("""
            SELECT g.id, g.name, g.type FROM groups g
            JOIN group_members gm ON gm.group_id = g.id
            WHERE gm.user_id = ?
        """, (user_id,)).fetchall()

        for group in groups:
            gid = group["id"]
            last_msg = conn.execute("""
                SELECT content, media_type, timestamp FROM group_messages
                WHERE group_id = ? ORDER BY timestamp DESC LIMIT 1
            """, (gid,)).fetchone()

            chats.append({
                "id": str(gid),
                "name": group["name"],
                "preview": _preview_text(last_msg),
                "time": last_msg["timestamp"] if last_msg else 0,
                "unread": 0,  # group read-tracking not implemented yet
                "color": _color_for_id(gid + 1000),
                "category": "groups",
                "isGroup": True,
                "groupType": group["type"],
            })

        chats.sort(key=lambda c: c["time"], reverse=True)
        return chats
    finally:
        conn.close()


def _preview_text(row):
    if not row:
        return ""
    media_labels = {"image": "Photo", "voice": "Voice note", "file": "File"}
    if row["media_type"] in media_labels:
        return media_labels[row["media_type"]]
    return row["content"]


def search_content(user_id, query):
    """Searches the user's friends by username, plus their direct and
    group message content. Returns (friends, messages)."""
    conn = get_conn()
    try:
        like = f"%{query}%"

        friend_rows = conn.execute("""
            SELECT u.id, u.username, u.verified
            FROM friendships f
            JOIN users u ON u.id = CASE WHEN f.user_id = ? THEN f.friend_id ELSE f.user_id END
            WHERE (f.user_id = ? OR f.friend_id = ?) AND f.status = 'accepted'
              AND u.username LIKE ?
            LIMIT 20
        """, (user_id, user_id, user_id, like)).fetchall()
        friends = [{
            "id": r["id"], "name": r["username"], "username": r["username"],
            "color": _color_for_id(r["id"]), "verified": r["verified"],
        } for r in friend_rows]

        messages = []

        dm_rows = conn.execute("""
            SELECT m.id, m.content, m.timestamp,
                   CASE WHEN m.sender_id = ? THEN m.recipient_id ELSE m.sender_id END as other_id
            FROM messages m
            WHERE (m.sender_id = ? OR m.recipient_id = ?) AND m.content LIKE ?
            ORDER BY m.timestamp DESC LIMIT 20
        """, (user_id, user_id, user_id, like)).fetchall()
        for r in dm_rows:
            other = conn.execute("SELECT username FROM users WHERE id = ?", (r["other_id"],)).fetchone()
            messages.append({
                "id": f"dm{r['id']}",
                "chatId": str(r["other_id"]),
                "chatName": other["username"] if other else "Unknown",
                "snippet": r["content"][:80],
            })

        gm_rows = conn.execute("""
            SELECT gmsg.id, gmsg.content, gmsg.timestamp, gmsg.group_id
            FROM group_messages gmsg
            JOIN group_members mem ON mem.group_id = gmsg.group_id AND mem.user_id = ?
            WHERE gmsg.content LIKE ?
            ORDER BY gmsg.timestamp DESC LIMIT 20
        """, (user_id, like)).fetchall()
        for r in gm_rows:
            grp = conn.execute("SELECT name FROM groups WHERE id = ?", (r["group_id"],)).fetchone()
            messages.append({
                "id": f"gm{r['id']}",
                "chatId": str(r["group_id"]),
                "chatName": grp["name"] if grp else "Group",
                "snippet": r["content"][:80],
            })

        return friends, messages
    finally:
        conn.close()


def create_report(reporter_id, reported_user_id, reason):
    conn = get_conn()
    try:
        conn.execute(
            "INSERT INTO reports (reporter_id, reported_user_id, reason, created_at) VALUES (?, ?, ?, ?)",
            (reporter_id, reported_user_id, reason, int(time.time()))
        )
        conn.commit()
    finally:
        conn.close()

def get_reports_for_user(user_id):
    conn = get_conn()
    try:
        rows = conn.execute('''
            SELECT r.id, r.reporter_id, u.username as reporter_username, r.reason, r.created_at
            FROM reports r
            LEFT JOIN users u ON u.id = r.reporter_id
            WHERE r.reported_user_id = ?
            ORDER BY r.created_at DESC
        ''', (user_id,)).fetchall()
        return [dict(row) for row in rows]
    finally:
        conn.close()

def get_report_counts():
    conn = get_conn()
    try:
        rows = conn.execute('''
            SELECT reported_user_id, COUNT(*) as count
            FROM reports
            GROUP BY reported_user_id
        ''').fetchall()
        return {row["reported_user_id"]: row["count"] for row in rows}
    finally:
        conn.close()


# ---------- statuses ----------

def create_status(user_id, content_type, media_url=None, text_content=None,
                   bg_color=None, privacy='all', privacy_user_ids=None, duration_hours=24):
    conn = get_conn()
    try:
        now = int(time.time())
        expires_at = now + duration_hours * 3600
        cur = conn.execute(
            "INSERT INTO statuses (user_id, content_type, media_url, text_content, bg_color, privacy, created_at, expires_at) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            (user_id, content_type, media_url, text_content, bg_color, privacy, now, expires_at)
        )
        status_id = cur.lastrowid
        if privacy in ('except', 'only') and privacy_user_ids:
            for uid in privacy_user_ids:
                conn.execute(
                    "INSERT OR IGNORE INTO status_privacy_list (status_id, user_id) VALUES (?, ?)",
                    (status_id, uid)
                )
        conn.commit()
        return status_id
    finally:
        conn.close()


def _can_view_status(conn, status_row, viewer_id):
    if status_row["user_id"] == viewer_id:
        return True
    privacy = status_row["privacy"]
    if privacy == 'all':
        return True
    listed = conn.execute(
        "SELECT 1 FROM status_privacy_list WHERE status_id = ? AND user_id = ?",
        (status_row["id"], viewer_id)
    ).fetchone() is not None
    if privacy == 'except':
        return not listed
    if privacy == 'only':
        return listed
    return False


def get_status_feed(user_id):
    """Active (non-expired) statuses from the user's friends, grouped by
    author, filtered per-status by privacy (all / except / only)."""
    conn = get_conn()
    try:
        now = int(time.time())
        friend_rows = conn.execute("""
            SELECT u.id, u.username, u.avatar_letter, u.avatar_url
            FROM friendships f
            JOIN users u ON u.id = CASE WHEN f.user_id = ? THEN f.friend_id ELSE f.user_id END
            WHERE (f.user_id = ? OR f.friend_id = ?) AND f.status = 'accepted'
        """, (user_id, user_id, user_id)).fetchall()

        result = []
        for friend in friend_rows:
            fid = friend["id"]
            rows = conn.execute(
                "SELECT * FROM statuses WHERE user_id = ? AND expires_at > ? ORDER BY created_at ASC",
                (fid, now)
            ).fetchall()
            visible = [dict(r) for r in rows if _can_view_status(conn, r, user_id)]
            if not visible:
                continue

            placeholders = ",".join("?" * len(visible))
            viewed_ids = {
                r["status_id"] for r in conn.execute(
                    f"SELECT status_id FROM status_views WHERE viewer_id = ? AND status_id IN ({placeholders})",
                    [user_id] + [s["id"] for s in visible]
                ).fetchall()
            }
            for s in visible:
                s["viewed"] = s["id"] in viewed_ids

            result.append({
                "user_id": fid,
                "username": friend["username"],
                "avatar_letter": friend["avatar_letter"],
                "avatar_url": _avatar_or_hidden(fid, friend["avatar_url"], user_id),
                "statuses": visible,
                "all_viewed": all(s["viewed"] for s in visible),
            })
        return result
    finally:
        conn.close()


def get_my_statuses(user_id):
    conn = get_conn()
    try:
        now = int(time.time())
        rows = conn.execute(
            "SELECT * FROM statuses WHERE user_id = ? AND expires_at > ? ORDER BY created_at ASC",
            (user_id, now)
        ).fetchall()
        return [dict(r) for r in rows]
    finally:
        conn.close()


def record_status_view(status_id, viewer_id):
    conn = get_conn()
    try:
        conn.execute(
            "INSERT OR IGNORE INTO status_views (status_id, viewer_id, viewed_at) VALUES (?, ?, ?)",
            (status_id, viewer_id, int(time.time()))
        )
        conn.commit()
    finally:
        conn.close()


def get_status_viewers(status_id, owner_id):
    """Only returns data if owner_id actually owns the status."""
    conn = get_conn()
    try:
        status = conn.execute("SELECT user_id FROM statuses WHERE id = ?", (status_id,)).fetchone()
        if not status or status["user_id"] != owner_id:
            return None
        rows = conn.execute("""
            SELECT u.id, u.username, u.avatar_letter, u.avatar_url, sv.viewed_at
            FROM status_views sv
            JOIN users u ON u.id = sv.viewer_id
            WHERE sv.status_id = ?
            ORDER BY sv.viewed_at DESC
        """, (status_id,)).fetchall()
        return [dict(r) for r in rows]
    finally:
        conn.close()


def delete_status(status_id, user_id):
    """Returns {"deleted": True, "media_url": ...} on success (media_url may be
    None for text statuses), or None if not found / not owned by user_id."""
    conn = get_conn()
    try:
        row = conn.execute("SELECT user_id, media_url FROM statuses WHERE id = ?", (status_id,)).fetchone()
        if not row or row["user_id"] != user_id:
            return None
        media_url = row["media_url"]
        conn.execute("DELETE FROM status_views WHERE status_id = ?", (status_id,))
        conn.execute("DELETE FROM status_privacy_list WHERE status_id = ?", (status_id,))
        conn.execute("DELETE FROM statuses WHERE id = ?", (status_id,))
        conn.commit()
        return {"deleted": True, "media_url": media_url}
    finally:
        conn.close()


def delete_expired_statuses():
    """Deletes all status rows past their expires_at. Returns the list of
    media_url values (None entries included for text statuses) so the caller
    can remove any local files on disk."""
    conn = get_conn()
    try:
        now = int(time.time())
        rows = conn.execute("SELECT id, media_url FROM statuses WHERE expires_at <= ?", (now,)).fetchall()
        if not rows:
            return []
        ids = [r["id"] for r in rows]
        placeholders = ",".join("?" * len(ids))
        conn.execute(f"DELETE FROM status_views WHERE status_id IN ({placeholders})", ids)
        conn.execute(f"DELETE FROM status_privacy_list WHERE status_id IN ({placeholders})", ids)
        conn.execute(f"DELETE FROM statuses WHERE id IN ({placeholders})", ids)
        conn.commit()
        return [r["media_url"] for r in rows]
    finally:
        conn.close()


# ---------- unfriend / block ----------

def remove_friendship(user_id, friend_id):
    conn = get_conn()
    try:
        conn.execute(
            "DELETE FROM friendships WHERE (user_id=? AND friend_id=?) OR (user_id=? AND friend_id=?)",
            (user_id, friend_id, friend_id, user_id)
        )
        conn.commit()
        return True
    finally:
        conn.close()


def block_user(user_id, target_id):
    """Blocks target_id directly, regardless of prior friendship state."""
    conn = get_conn()
    try:
        existing = conn.execute(
            "SELECT id FROM friendships WHERE (user_id=? AND friend_id=?) OR (user_id=? AND friend_id=?)",
            (user_id, target_id, target_id, user_id)
        ).fetchone()
        if existing:
            conn.execute(
                "UPDATE friendships SET status='blocked', user_id=?, friend_id=? WHERE id=?",
                (user_id, target_id, existing["id"])
            )
        else:
            conn.execute(
                "INSERT INTO friendships (user_id, friend_id, status, requested_by, created_at) "
                "VALUES (?, ?, 'blocked', ?, ?)",
                (user_id, target_id, user_id, int(time.time()))
            )
        conn.commit()
        return True
    finally:
        conn.close()


# ---------- chat mute ----------

def _ensure_mutes_table(conn):
    conn.execute("""
        CREATE TABLE IF NOT EXISTS chat_mutes (
            user_id INTEGER NOT NULL,
            chat_id TEXT NOT NULL,
            is_group INTEGER NOT NULL DEFAULT 0,
            muted_at INTEGER NOT NULL,
            PRIMARY KEY (user_id, chat_id)
        )
    """)


def set_chat_mute(user_id, chat_id, muted, is_group=False):
    conn = get_conn()
    try:
        _ensure_mutes_table(conn)
        if muted:
            conn.execute(
                "INSERT OR REPLACE INTO chat_mutes (user_id, chat_id, is_group, muted_at) VALUES (?, ?, ?, ?)",
                (user_id, str(chat_id), 1 if is_group else 0, int(time.time()))
            )
        else:
            conn.execute(
                "DELETE FROM chat_mutes WHERE user_id=? AND chat_id=?",
                (user_id, str(chat_id))
            )
        conn.commit()
        return True
    finally:
        conn.close()


# ---------- disappearing messages ----------

def _ensure_disappearing_table(conn):
    conn.execute("""
        CREATE TABLE IF NOT EXISTS disappearing_settings (
            user_id INTEGER NOT NULL,
            chat_id TEXT NOT NULL,
            seconds INTEGER NOT NULL,
            set_at INTEGER NOT NULL,
            PRIMARY KEY (user_id, chat_id)
        )
    """)


def set_disappearing_duration(user_id, chat_id, seconds):
    conn = get_conn()
    try:
        _ensure_disappearing_table(conn)
        if seconds and seconds > 0:
            conn.execute(
                "INSERT OR REPLACE INTO disappearing_settings (user_id, chat_id, seconds, set_at) VALUES (?, ?, ?, ?)",
                (user_id, str(chat_id), int(seconds), int(time.time()))
            )
        else:
            conn.execute(
                "DELETE FROM disappearing_settings WHERE user_id=? AND chat_id=?",
                (user_id, str(chat_id))
            )
        conn.commit()
        return True
    finally:
        conn.close()


def get_disappearing_duration(user_id, chat_id):
    conn = get_conn()
    try:
        _ensure_disappearing_table(conn)
        row = conn.execute(
            "SELECT seconds FROM disappearing_settings WHERE user_id=? AND chat_id=?",
            (user_id, str(chat_id))
        ).fetchone()
        return row["seconds"] if row else None
    finally:
        conn.close()


# ---------- group management ----------

def _ensure_group_perm_columns(conn):
    cols = [r["name"] for r in conn.execute("PRAGMA table_info(groups)").fetchall()]
    if "send_perm" not in cols:
        conn.execute("ALTER TABLE groups ADD COLUMN send_perm TEXT DEFAULT 'all'")
    if "edit_perm" not in cols:
        conn.execute("ALTER TABLE groups ADD COLUMN edit_perm TEXT DEFAULT 'admins'")


def get_group_members_detailed(group_id, viewer_id=None):
    conn = get_conn()
    try:
        rows = conn.execute("""
            SELECT u.id, u.username, u.avatar_letter, u.avatar_url, u.verified, gm.role
            FROM group_members gm
            JOIN users u ON u.id = gm.user_id
            WHERE gm.group_id = ?
            ORDER BY CASE gm.role WHEN 'admin' THEN 0 ELSE 1 END, u.username
        """, (group_id,)).fetchall()
        result = [dict(r) for r in rows]
        for r in result:
            r["avatar_url"] = _avatar_or_hidden(r["id"], r["avatar_url"], viewer_id)
        return result
    finally:
        conn.close()


def _is_group_owner(conn, group_id, user_id):
    row = conn.execute("SELECT created_by FROM groups WHERE id=?", (group_id,)).fetchone()
    return row is not None and row["created_by"] == user_id


def promote_member(group_id, target_id, requester_id):
    conn = get_conn()
    try:
        if not _is_group_owner(conn, group_id, requester_id):
            return False, "Only the owner can promote members"
        conn.execute(
            "UPDATE group_members SET role='admin' WHERE group_id=? AND user_id=?",
            (group_id, target_id)
        )
        conn.commit()
        return True, None
    finally:
        conn.close()


def demote_member(group_id, target_id, requester_id):
    conn = get_conn()
    try:
        if not _is_group_owner(conn, group_id, requester_id):
            return False, "Only the owner can demote members"
        conn.execute(
            "UPDATE group_members SET role='member' WHERE group_id=? AND user_id=?",
            (group_id, target_id)
        )
        conn.commit()
        return True, None
    finally:
        conn.close()


def remove_member(group_id, target_id, requester_id):
    conn = get_conn()
    try:
        requester_role = conn.execute(
            "SELECT role FROM group_members WHERE group_id=? AND user_id=?",
            (group_id, requester_id)
        ).fetchone()
        target_role = conn.execute(
            "SELECT role FROM group_members WHERE group_id=? AND user_id=?",
            (group_id, target_id)
        ).fetchone()
        is_owner = _is_group_owner(conn, group_id, requester_id)
        if not requester_role or (requester_role["role"] != "admin" and not is_owner):
            return False, "Not authorized"
        if target_role and target_role["role"] == "admin" and not is_owner:
            return False, "Only the owner can remove an admin"
        conn.execute(
            "DELETE FROM group_members WHERE group_id=? AND user_id=?",
            (group_id, target_id)
        )
        conn.commit()
        return True, None
    finally:
        conn.close()


def leave_group(group_id, user_id):
    conn = get_conn()
    try:
        conn.execute(
            "DELETE FROM group_members WHERE group_id=? AND user_id=?",
            (group_id, user_id)
        )
        conn.commit()
        return True
    finally:
        conn.close()


def delete_group(group_id, requester_id):
    conn = get_conn()
    try:
        if not _is_group_owner(conn, group_id, requester_id):
            return False
        conn.execute("DELETE FROM group_messages WHERE group_id=?", (group_id,))
        conn.execute("DELETE FROM group_members WHERE group_id=?", (group_id,))
        conn.execute("DELETE FROM groups WHERE id=?", (group_id,))
        conn.commit()
        return True
    finally:
        conn.close()


def update_group_permissions(group_id, requester_id, send_perm=None, edit_perm=None):
    conn = get_conn()
    try:
        _ensure_group_perm_columns(conn)
        if not _is_group_owner(conn, group_id, requester_id):
            role = conn.execute(
                "SELECT role FROM group_members WHERE group_id=? AND user_id=?",
                (group_id, requester_id)
            ).fetchone()
            if not role or role["role"] != "admin":
                return False
        if send_perm is not None:
            conn.execute("UPDATE groups SET send_perm=? WHERE id=?", (send_perm, group_id))
        if edit_perm is not None:
            conn.execute("UPDATE groups SET edit_perm=? WHERE id=?", (edit_perm, group_id))
        conn.commit()
        return True
    finally:
        conn.close()


def get_group_permissions(group_id):
    conn = get_conn()
    try:
        _ensure_group_perm_columns(conn)
        row = conn.execute("SELECT send_perm, edit_perm FROM groups WHERE id=?", (group_id,)).fetchone()
        return dict(row) if row else {"send_perm": "all", "edit_perm": "admins"}
    finally:
        conn.close()


# ---------- group reports ----------

def _ensure_group_report_column(conn):
    cols = [r["name"] for r in conn.execute("PRAGMA table_info(reports)").fetchall()]
    if "reported_group_id" not in cols:
        conn.execute("ALTER TABLE reports ADD COLUMN reported_group_id INTEGER")


def create_group_report(reporter_id, group_id, reason):
    conn = get_conn()
    try:
        _ensure_group_report_column(conn)
        conn.execute(
            "INSERT INTO reports (reporter_id, reported_user_id, reported_group_id, reason, created_at) "
            "VALUES (?, 0, ?, ?, ?)",
            (reporter_id, group_id, reason, int(time.time()))
        )
        conn.commit()
    finally:
        conn.close()


def get_group_invite(group_id, requester_id):
    """Returns {id, name, invite_code} if requester_id is a member, else None."""
    conn = get_conn()
    try:
        member = conn.execute(
            "SELECT 1 FROM group_members WHERE group_id=? AND user_id=?",
            (group_id, requester_id)
        ).fetchone()
        if not member:
            return None
        row = conn.execute(
            "SELECT id, name, invite_code FROM groups WHERE id=?", (group_id,)
        ).fetchone()
        return dict(row) if row else None
    finally:
        conn.close()


def clear_group_messages(group_id, requester_id):
    """Deletes all messages in a group. Admins or the owner only."""
    conn = get_conn()
    try:
        role = conn.execute(
            "SELECT role FROM group_members WHERE group_id=? AND user_id=?",
            (group_id, requester_id)
        ).fetchone()
        is_owner = _is_group_owner(conn, group_id, requester_id)
        if not role or (role["role"] != "admin" and not is_owner):
            return False
        conn.execute("DELETE FROM group_messages WHERE group_id=?", (group_id,))
        conn.commit()
        return True
    finally:
        conn.close()

# ---------- points / gamification (dynamic, DB-backed) ----------

def create_task(key, label, points, category="General"):
    conn = get_conn()
    try:
        cur = conn.execute(
            "INSERT INTO tasks (key, label, points, category, active, created_at) VALUES (?, ?, ?, ?, 1, ?)",
            (key, label, points, category, int(time.time()))
        )
        conn.commit()
        return cur.lastrowid
    except sqlite3.IntegrityError:
        return None
    finally:
        conn.close()


def list_all_tasks():
    conn = get_conn()
    try:
        rows = conn.execute("SELECT * FROM tasks ORDER BY created_at ASC").fetchall()
        return [dict(r) for r in rows]
    finally:
        conn.close()


def toggle_task_active(task_id):
    conn = get_conn()
    try:
        row = conn.execute("SELECT active FROM tasks WHERE id = ?", (task_id,)).fetchone()
        if not row:
            return None
        new_state = 0 if row["active"] else 1
        conn.execute("UPDATE tasks SET active = ? WHERE id = ?", (new_state, task_id))
        conn.commit()
        return bool(new_state)
    finally:
        conn.close()


def get_task_by_id(task_id):
    conn = get_conn()
    try:
        row = conn.execute("SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()
        return dict(row) if row else None
    finally:
        conn.close()


def _get_task_by_key(conn, task_key):
    row = conn.execute("SELECT * FROM tasks WHERE key = ? AND active = 1", (task_key,)).fetchone()
    return dict(row) if row else None


def award_points(user_id, task_key):
    """Idempotent — awards points for task_key only if not already completed
    and the task exists and is active. Returns True if newly awarded."""
    if user_id == AI_USER_ID:
        return False
    conn = get_conn()
    try:
        task = _get_task_by_key(conn, task_key)
        if not task:
            return False
        cur = conn.execute(
            "INSERT OR IGNORE INTO completed_tasks (user_id, task_key, points_awarded, completed_at) VALUES (?, ?, ?, ?)",
            (user_id, task_key, task["points"], int(time.time()))
        )
        if cur.rowcount == 0:
            return False
        conn.execute("UPDATE users SET points = points + ? WHERE id = ?", (task["points"], user_id))
        conn.commit()
        return True
    finally:
        conn.close()


def admin_award_task(task_id, user_id):
    """Admin manually awards a task to a specific user, bypassing any auto-detection.
    Still idempotent — won't double-award the same task to the same user."""
    conn = get_conn()
    try:
        task = conn.execute("SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()
        if not task:
            return None, "Task not found"
        cur = conn.execute(
            "INSERT OR IGNORE INTO completed_tasks (user_id, task_key, points_awarded, completed_at) VALUES (?, ?, ?, ?)",
            (user_id, task["key"], task["points"], int(time.time()))
        )
        if cur.rowcount == 0:
            return False, "Already awarded to this user"
        conn.execute("UPDATE users SET points = points + ? WHERE id = ?", (task["points"], user_id))
        conn.commit()
        return True, None
    finally:
        conn.close()


def maybe_award_friend_count_tasks(user_id):
    """Call after a friendship becomes accepted — checks 3/10 friend thresholds."""
    count = len(list_friends(user_id))
    if count >= 3:
        award_points(user_id, "add_3_friends")
    if count >= 10:
        award_points(user_id, "add_10_friends")


def get_points_summary(user_id):
    conn = get_conn()
    try:
        row = conn.execute("SELECT points FROM users WHERE id = ?", (user_id,)).fetchone()
        points = row["points"] if row and row["points"] is not None else 0

        active_tasks = conn.execute("SELECT * FROM tasks WHERE active = 1").fetchall()
        task_by_key = {t["key"]: dict(t) for t in active_tasks}

        completed_rows = conn.execute(
            "SELECT task_key, points_awarded, completed_at FROM completed_tasks WHERE user_id = ? ORDER BY completed_at ASC",
            (user_id,)
        ).fetchall()
        completed_keys = {r["task_key"] for r in completed_rows}

        completed = []
        for r in completed_rows:
            t = task_by_key.get(r["task_key"])
            completed.append({
                "id": r["task_key"],
                "label": t["label"] if t else r["task_key"],
                "points": r["points_awarded"],
            })

        available = [
            {"id": t["key"], "label": t["label"], "points": t["points"], "category": t["category"]}
            for t in task_by_key.values() if t["key"] not in completed_keys
        ]
        return {"points": points, "completed": completed, "available": available}
    finally:
        conn.close()


# ---------- announcements (single active one at a time) ----------

def create_announcement(title, body=None, expires_in_hours=72):
    """Retires any currently-active announcement, then creates a new active one."""
    conn = get_conn()
    try:
        now = int(time.time())
        expires_at = now + int(expires_in_hours * 3600) if expires_in_hours else None
        conn.execute("UPDATE announcements SET active = 0 WHERE active = 1")
        cur = conn.execute(
            "INSERT INTO announcements (title, body, created_at, expires_at, active) VALUES (?, ?, ?, ?, 1)",
            (title, body, now, expires_at)
        )
        conn.commit()
        return cur.lastrowid
    finally:
        conn.close()


def get_active_announcement():
    """Returns the current active, non-expired announcement dict, or None."""
    conn = get_conn()
    try:
        now = int(time.time())
        row = conn.execute(
            "SELECT * FROM announcements WHERE active = 1 AND (expires_at IS NULL OR expires_at > ?) "
            "ORDER BY created_at DESC LIMIT 1",
            (now,)
        ).fetchone()
        return dict(row) if row else None
    finally:
        conn.close()


def list_announcements(limit=50):
    conn = get_conn()
    try:
        rows = conn.execute(
            "SELECT * FROM announcements ORDER BY created_at DESC LIMIT ?", (limit,)
        ).fetchall()
        return [dict(r) for r in rows]
    finally:
        conn.close()


def deactivate_announcement(announcement_id):
    conn = get_conn()
    try:
        cur = conn.execute("UPDATE announcements SET active = 0 WHERE id = ?", (announcement_id,))
        conn.commit()
        return cur.rowcount > 0
    finally:
        conn.close()


def clear_direct_messages(user_a, user_b):
    """Deletes all direct messages between two users, both directions."""
    conn = get_conn()
    try:
        conn.execute(
            "DELETE FROM messages WHERE (sender_id=? AND recipient_id=?) OR (sender_id=? AND recipient_id=?)",
            (user_a, user_b, user_b, user_a)
        )
        conn.commit()
        return True
    finally:
        conn.close()


def list_blocked_users(user_id):
    """Returns users blocked BY user_id. Relies on block_user() always
    normalizing so the blocker is stored in the user_id column."""
    conn = get_conn()
    try:
        rows = conn.execute("""
            SELECT u.id, u.username, u.handle, u.avatar_letter, u.avatar_url
            FROM friendships f
            JOIN users u ON u.id = f.friend_id
            WHERE f.user_id = ? AND f.status = 'blocked'
        """, (user_id,)).fetchall()
        return [dict(r) for r in rows]
    finally:
        conn.close()


def set_latest_app_version(version, apk_url, notes=None):
    """Always overwrites the single row (id=1) — there's only ever one 'latest' version."""
    conn = get_conn()
    try:
        conn.execute("""
            INSERT INTO app_version (id, version, apk_url, notes, updated_at)
            VALUES (1, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET version=excluded.version, apk_url=excluded.apk_url,
                notes=excluded.notes, updated_at=excluded.updated_at
        """, (version, apk_url, notes, int(time.time())))
        conn.commit()
        return True
    finally:
        conn.close()


def get_latest_app_version():
    conn = get_conn()
    try:
        row = conn.execute("SELECT version, apk_url, notes, updated_at FROM app_version WHERE id = 1").fetchone()
        return dict(row) if row else None
    finally:
        conn.close()


def set_latest_ota_update(update_id, runtime_version, launch_asset_json, assets_json, notes=None):
    """Always overwrites the single row (id=1) -- there's only ever one
    'latest' OTA update, same pattern as set_latest_app_version. Callers
    pass launch_asset_json/assets_json as already-serialized JSON strings
    (the exact objects the Expo Updates manifest needs)."""
    conn = get_conn()
    try:
        conn.execute("""
            INSERT INTO ota_update (id, update_id, runtime_version, launch_asset_json, assets_json, notes, created_at)
            VALUES (1, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET update_id=excluded.update_id, runtime_version=excluded.runtime_version,
                launch_asset_json=excluded.launch_asset_json, assets_json=excluded.assets_json,
                notes=excluded.notes, created_at=excluded.created_at
        """, (update_id, runtime_version, launch_asset_json, assets_json, notes, int(time.time())))
        conn.commit()
        return True
    finally:
        conn.close()


def get_latest_ota_update():
    conn = get_conn()
    try:
        row = conn.execute("""
            SELECT update_id, runtime_version, launch_asset_json, assets_json, notes, created_at
            FROM ota_update WHERE id = 1
        """).fetchone()
        return dict(row) if row else None
    finally:
        conn.close()