# routes.py import json import re import sqlite3 import time from flask import jsonify, request from features import ( DM_LAYOUT_MODES, LAST_ACTIVE_VISIBILITIES, PAGE_TYPES, PAGE_VISIBILITIES, PRESENCE_MODES, PROFILE_BIO_MODES, PROFILE_VISIBILITIES, REACTION_EMOJI_RE, TEAM_MEMBER_ROLES, THEME_MODES, append_call_signal, call_payload, channel_presence_count, dm_payload, dm_presence_count, end_call, feed_message_payload, find_or_create_dm, get_call_with_dm, get_dm_for_user, list_call_signals, list_dms_for_user, normalized_hex_color, page_payload, profile_payload, reaction_summary_for_message, resolve_owned_attachment, resolve_owned_attachment_ids, richer_message_payload, room_presence_count, send_attachment_file, update_presence, validate_dm_access, valid_hex_color, ) ROOM_SLUG_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,46}[a-z0-9])?$") def register_routes(app, ctx): get_db = ctx["get_db"] db_write = ctx["db_write"] safe_json = ctx["safe_json"] csrf_valid = ctx["csrf_valid"] rate_limited = ctx["rate_limited"] current_user = ctx["current_user"] require_user = ctx["require_user"] require_writable_user = ctx["require_writable_user"] require_moderator = ctx["require_moderator"] require_admin = ctx["require_admin"] now_iso = ctx["now_iso"] utc_now = ctx["utc_now"] parse_iso = ctx["parse_iso"] normalize_username = ctx["normalize_username"] audit = ctx["audit"] hash_identifier = ctx["hash_identifier"] sign_call_helper_token = ctx["sign_call_helper_token"] CALLS_BASE = ctx["CALLS_BASE"] CALL_TOKEN_TTL_SECONDS = ctx["CALL_TOKEN_TTL_SECONDS"] CALL_HELPER_SHARED_SECRET = ctx["CALL_HELPER_SHARED_SECRET"] room_by_id = ctx["room_by_id"] get_room_by_slug = ctx["get_room_by_slug"] can_access_room = ctx["can_access_room"] can_write_room = ctx["can_write_room"] can_manage_room = ctx["can_manage_room"] room_payload = ctx["room_payload"] team_slug_from_name = ctx["team_slug_from_name"] page_slug_from_title = ctx["page_slug_from_title"] get_team_by_slug = ctx["get_team_by_slug"] get_channel_by_id = ctx["get_channel_by_id"] get_channel_by_slug = ctx["get_channel_by_slug"] get_channel_by_legacy_room_id = ctx["get_channel_by_legacy_room_id"] team_member_role = ctx["team_member_role"] is_team_member = ctx["is_team_member"] is_team_admin = ctx["is_team_admin"] can_edit_team = ctx["can_edit_team"] can_access_team = ctx["can_access_team"] can_access_channel = ctx["can_access_channel"] can_write_channel = ctx["can_write_channel"] team_payload = ctx["team_payload"] channel_payload = ctx["channel_payload"] user_payload = ctx["user_payload"] get_dm_other_user = ctx["get_dm_other_user"] MESSAGE_SELECT = ctx["MESSAGE_SELECT"] team_page_select = ctx["team_page_select"] constants = ctx["constants"] MAX_MESSAGE_LENGTH = constants["MAX_MESSAGE_LENGTH"] MAX_HISTORY_MESSAGES = constants["MAX_HISTORY_MESSAGES"] MAX_INCREMENTAL_MESSAGES = constants["MAX_INCREMENTAL_MESSAGES"] MAX_FEED_MESSAGES = constants["MAX_FEED_MESSAGES"] MAX_THREAD_REPLIES = constants["MAX_THREAD_REPLIES"] TYPING_TTL_SECONDS = constants["TYPING_TTL_SECONDS"] PRESENCE_TTL_SECONDS = constants["PRESENCE_TTL_SECONDS"] INVITE_TTL_HOURS = constants["INVITE_TTL_HOURS"] TEAM_INVITE_TTL_HOURS = constants["TEAM_INVITE_TTL_HOURS"] MAX_SEARCH_QUERY_LENGTH = constants["MAX_SEARCH_QUERY_LENGTH"] MAX_SEARCH_RESULTS = constants["MAX_SEARCH_RESULTS"] READ_RECEIPT_BATCH_LIMIT = constants["READ_RECEIPT_BATCH_LIMIT"] MAX_MESSAGE_EDIT_WINDOW_SECONDS = constants["MAX_MESSAGE_EDIT_WINDOW_SECONDS"] MAX_DISPLAY_NAME_LENGTH = constants["MAX_DISPLAY_NAME_LENGTH"] MAX_PROFILE_BIO_LENGTH = constants["MAX_PROFILE_BIO_LENGTH"] MAX_WEBSITE_LENGTH = constants["MAX_WEBSITE_LENGTH"] MAX_LOCATION_LENGTH = constants["MAX_LOCATION_LENGTH"] MAX_CUSTOM_CSS_LENGTH = constants["MAX_CUSTOM_CSS_LENGTH"] MAX_PROFILE_BADGE_LENGTH = constants["MAX_PROFILE_BADGE_LENGTH"] MAX_OWNED_TEAMS_PER_USER = constants["MAX_OWNED_TEAMS_PER_USER"] GROUP_CALL_PARTICIPANT_STALE_SECONDS = constants["GROUP_CALL_PARTICIPANT_STALE_SECONDS"] ROOM_VISIBILITIES = constants["ROOM_VISIBILITIES"] ROLES = constants["ROLES"] STATUSES = constants["STATUSES"] PRIVATE_ACCESS_KEY = ctx["PRIVATE_ACCESS_KEY"] html_module = ctx["html"] def private_access_key_valid(candidate: object) -> bool: configured = (PRIVATE_ACCESS_KEY or "").strip() if not configured: return True if not isinstance(candidate, str): return False return candidate.strip() == configured def build_message_payload(row, user): return richer_message_payload( get_db, row, user, user["id"] if user else None, ) def build_local_stub_share_url(kind: str, object_id: int) -> str: return f"/shared/{kind}/{object_id}" def build_room_helper_call_id(room) -> str: return f"room-{room['slug']}" def build_channel_helper_call_id(channel) -> str: return f"channel-{channel['team_slug']}--{channel['slug']}" def get_or_create_group_call_session( scope_kind: str, scope_id: int, scope_slug: str, helper_call_id: str, created_by_user_id: int, ): with get_db() as db: existing = db.execute( """ SELECT id, scope_kind, scope_id, scope_slug, helper_call_id, created_by_user_id, status, created_at, ended_at FROM group_call_sessions WHERE helper_call_id = ? LIMIT 1 """, (helper_call_id,), ).fetchone() if existing and existing["status"] == "active": return existing with get_db() as db: db.execute("BEGIN IMMEDIATE") existing = db.execute( """ SELECT id, scope_kind, scope_id, scope_slug, helper_call_id, created_by_user_id, status, created_at, ended_at FROM group_call_sessions WHERE helper_call_id = ? LIMIT 1 """, (helper_call_id,), ).fetchone() if existing and existing["status"] == "active": db.commit() return existing if existing: db.execute( """ UPDATE group_call_sessions SET status = 'active', ended_at = NULL, created_at = ? WHERE id = ? """, (now_iso(), existing["id"]), ) db.commit() refreshed = db.execute( """ SELECT id, scope_kind, scope_id, scope_slug, helper_call_id, created_by_user_id, status, created_at, ended_at FROM group_call_sessions WHERE id = ? """, (existing["id"],), ).fetchone() return refreshed cursor = db.execute( """ INSERT INTO group_call_sessions ( scope_kind, scope_id, scope_slug, helper_call_id, created_by_user_id, status, created_at, ended_at ) VALUES (?, ?, ?, ?, ?, 'active', ?, NULL) """, ( scope_kind, scope_id, scope_slug, helper_call_id, created_by_user_id, now_iso(), ), ) session_id = cursor.lastrowid db.commit() created = db.execute( """ SELECT id, scope_kind, scope_id, scope_slug, helper_call_id, created_by_user_id, status, created_at, ended_at FROM group_call_sessions WHERE id = ? """, (session_id,), ).fetchone() return created def group_call_session_payload(row): if not row: return None return { "id": row["id"], "scope_kind": row["scope_kind"], "scope_id": row["scope_id"], "scope_slug": row["scope_slug"], "helper_call_id": row["helper_call_id"], "created_by_user_id": row["created_by_user_id"], "status": row["status"], "created_at": row["created_at"], "ended_at": row["ended_at"], } def get_group_call_session_by_id(session_id: int): with get_db() as db: return db.execute( """ SELECT id, scope_kind, scope_id, scope_slug, helper_call_id, created_by_user_id, status, created_at, ended_at FROM group_call_sessions WHERE id = ? LIMIT 1 """, (session_id,), ).fetchone() def can_access_group_call_session(session_row, user: dict) -> bool: if not session_row or not user: return False if session_row["scope_kind"] == "room": room = room_by_id(session_row["scope_id"]) return bool(room and can_access_room(room, user)) if session_row["scope_kind"] == "channel": channel = get_channel_by_id(session_row["scope_id"]) return bool(channel and can_access_channel(channel, user)) return False def can_manage_group_call_session(session_row, user: dict) -> bool: if not session_row or not user: return False if user["role"] in {"moderator", "admin"}: return True if int(session_row["created_by_user_id"]) == int(user["id"]): return True if session_row["scope_kind"] == "room": room = room_by_id(session_row["scope_id"]) return bool(room and can_manage_room(room, user)) if session_row["scope_kind"] == "channel": channel = get_channel_by_id(session_row["scope_id"]) return bool(channel and can_edit_team(channel["team_id"], user)) return False def cleanup_stale_group_call_participants(session_id: int): cutoff = utc_now().timestamp() - GROUP_CALL_PARTICIPANT_STALE_SECONDS db_write( """ UPDATE group_call_participants SET left_at = COALESCE(left_at, ?) WHERE group_call_session_id = ? AND left_at IS NULL AND ( updated_at IS NULL OR strftime('%s', updated_at) < ? ) """, (now_iso(), session_id, int(cutoff)), ) def upsert_group_call_participant(session_id: int, user_id: int): db_write( """ INSERT INTO group_call_participants ( group_call_session_id, user_id, joined_at, updated_at, left_at ) VALUES (?, ?, ?, ?, NULL) ON CONFLICT(group_call_session_id, user_id) DO UPDATE SET updated_at = excluded.updated_at, left_at = NULL """, (session_id, user_id, now_iso(), now_iso()), ) def heartbeat_group_call_participant(session_id: int, user_id: int): db_write( """ UPDATE group_call_participants SET updated_at = ?, left_at = NULL WHERE group_call_session_id = ? AND user_id = ? """, (now_iso(), session_id, user_id), ) def mark_group_call_participant_left(session_id: int, user_id: int): db_write( """ UPDATE group_call_participants SET updated_at = ?, left_at = ? WHERE group_call_session_id = ? AND user_id = ? AND left_at IS NULL """, (now_iso(), now_iso(), session_id, user_id), ) def list_group_call_participants(session_id: int) -> list[dict]: cleanup_stale_group_call_participants(session_id) with get_db() as db: rows = db.execute( """ SELECT gcp.group_call_session_id, gcp.user_id, gcp.joined_at, gcp.updated_at, gcp.left_at, u.username, u.role, up.display_name, us.name_color FROM group_call_participants AS gcp JOIN users AS u ON u.id = gcp.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 gcp.group_call_session_id = ? AND gcp.left_at IS NULL ORDER BY gcp.joined_at ASC, u.username COLLATE NOCASE """, (session_id,), ).fetchall() return [ { "user_id": row["user_id"], "username": row["username"], "display_name": row["display_name"] or row["username"], "role": row["role"], "name_color": row["name_color"], "joined_at": row["joined_at"], "updated_at": row["updated_at"], } for row in rows ] def active_group_call_participant_count(session_id: int) -> int: cleanup_stale_group_call_participants(session_id) with get_db() as db: row = db.execute( """ SELECT COUNT(*) AS count FROM group_call_participants WHERE group_call_session_id = ? AND left_at IS NULL """, (session_id,), ).fetchone() return int(row["count"]) if row else 0 def maybe_end_group_call_if_empty(session_id: int): if active_group_call_participant_count(session_id) > 0: return db_write( """ UPDATE group_call_sessions SET status = 'ended', ended_at = ? WHERE id = ? AND status != 'ended' """, (now_iso(), session_id), ) db_write( """ UPDATE group_call_invites SET status = 'dismissed', responded_at = ? WHERE group_call_session_id = ? AND status = 'pending' """, (now_iso(), session_id), ) def group_call_scope_payload(session_row, user): payload = { "scope_kind": session_row["scope_kind"], "scope_id": session_row["scope_id"], "scope_slug": session_row["scope_slug"], } if session_row["scope_kind"] == "room": room = room_by_id(session_row["scope_id"]) if room: payload["room"] = room_payload(room, user) payload["scope_label"] = room["name"] elif session_row["scope_kind"] == "channel": channel = get_channel_by_id(session_row["scope_id"]) if channel: payload["channel"] = channel_payload(channel, user) payload["team"] = { "slug": channel["team_slug"], "name": channel["team_name"], } payload["scope_label"] = f"# {channel['name']}" return payload def group_call_permissions_payload(session_row, user): can_manage = can_manage_group_call_session(session_row, user) can_access = can_access_group_call_session(session_row, user) return { "can_manage": can_manage, "can_end": can_manage, "can_invite": can_access, } @app.route("/api/attachments", methods=["POST"]) def upload_attachment(): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error limited = rate_limited("attachment") if limited: return limited ctx["maybe_cleanup_orphaned_attachments"]() uploaded = request.files.get("attachment") if not uploaded or not uploaded.filename: return jsonify({"error": "Choose a file to upload."}), 400 original_name = ctx["secure_filename"](uploaded.filename)[: constants["MAX_ATTACHMENT_NAME_LENGTH"]] if not original_name or "." not in original_name: return jsonify({"error": "Attachment must have an allowed file extension."}), 400 extension = original_name.rsplit(".", 1)[1].lower() if extension not in constants["ALLOWED_ATTACHMENT_EXTENSIONS"]: return jsonify({"error": "This file type is not allowed."}), 400 declared_mime = (uploaded.mimetype or "").lower() guessed_mime = ctx["mimetypes"].guess_type(original_name)[0] or "" if declared_mime and declared_mime not in constants["ALLOWED_ATTACHMENT_MIME_TYPES"]: return jsonify({"error": "This file MIME type is not allowed."}), 400 if guessed_mime and guessed_mime not in constants["ALLOWED_ATTACHMENT_MIME_TYPES"]: return jsonify({"error": "This file extension is not allowed."}), 400 content = uploaded.stream.read(constants["MAX_ATTACHMENT_BYTES"] + 1) if not content: return jsonify({"error": "Attachment cannot be empty."}), 400 if len(content) > constants["MAX_ATTACHMENT_BYTES"]: return jsonify({"error": "Attachments must be 100 MB or smaller."}), 400 stored_name = f"{ctx['secrets'].token_urlsafe(24)}.{extension}" destination = ctx["UPLOAD_DIR"] / stored_name try: destination.write_bytes(content) ctx["os"].chmod(destination, 0o600) except OSError: app.logger.exception("Could not save attachment.") return jsonify({"error": "Could not save attachment."}), 500 mime_type = declared_mime or guessed_mime or "application/octet-stream" try: cursor = db_write( """ INSERT INTO attachments ( uploader_user_id, stored_name, original_name, mime_type, size_bytes, created_at ) VALUES (?, ?, ?, ?, ?, ?) """, (user["id"], stored_name, original_name, mime_type, len(content), now_iso()), ) except Exception: destination.unlink(missing_ok=True) raise audit("attachment_uploaded", user["id"], "attachment", str(cursor.lastrowid), original_name) return jsonify( { "attachment": { "id": cursor.lastrowid, "original_name": original_name, "mime_type": mime_type, "size_bytes": len(content), "url": f"/api/attachments/{cursor.lastrowid}/download", } } ), 201 @app.route("/api/attachments//download") def download_attachment(attachment_id: int): user, error = require_user() if error: return error return send_attachment_file( get_db, ctx["UPLOAD_DIR"], attachment_id, user, can_access_room, can_access_channel, get_channel_by_id, ) @app.route("/api/rooms//messages") def get_room_messages(slug: str): room = get_room_by_slug(slug) if not room: return jsonify({"error": "Room not found."}), 404 user = current_user() if not can_access_room(room, user): if not user: return jsonify({"error": "Login required for this private room.", "code": "login_required"}), 401 return jsonify({"error": "You do not have access to this room.", "code": "forbidden"}), 403 after_raw = request.args.get("after_id", "").strip() after_id = 0 if after_raw: try: after_id = max(0, int(after_raw)) except ValueError: return jsonify({"error": "after_id must be a non-negative integer."}), 400 with get_db() as db: if after_raw: rows = db.execute( f""" {MESSAGE_SELECT} WHERE m.room_id = ? AND m.parent_message_id IS NULL AND m.direct_conversation_id IS NULL AND m.id > ? ORDER BY m.id ASC LIMIT ? """, (room["id"], after_id, MAX_INCREMENTAL_MESSAGES), ).fetchall() else: rows = db.execute( f""" SELECT * FROM ( {MESSAGE_SELECT} WHERE m.room_id = ? AND m.parent_message_id IS NULL AND m.direct_conversation_id IS NULL ORDER BY m.id DESC LIMIT ? ) ORDER BY id ASC """, (room["id"], MAX_HISTORY_MESSAGES), ).fetchall() messages = [build_message_payload(row, user) for row in rows] latest_id = messages[-1]["id"] if messages else after_id return jsonify( { "messages": messages, "latest_id": latest_id, "has_more": len(messages) == MAX_INCREMENTAL_MESSAGES, } ) @app.route("/api/rooms//messages", methods=["POST"]) def send_room_message(slug: str): limited = rate_limited("send") if limited: return limited if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error room = get_room_by_slug(slug) if not room: return jsonify({"error": "Room not found."}), 404 if not can_write_room(room, user): return jsonify({"error": "You cannot post in this room.", "code": "forbidden"}), 403 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 text = data.get("text", "") if not isinstance(text, str): return jsonify({"error": "Message must be text."}), 400 text = text.strip() if len(text) > MAX_MESSAGE_LENGTH: return jsonify({"error": "Message is too long."}), 400 attachment_ids, attachment_error = resolve_owned_attachment_ids( get_db, data.get("attachment_ids", []), user["id"], ) if attachment_error: return attachment_error if not attachment_ids and data.get("attachment_id") is not None: single_attachment_id, single_attachment_error = resolve_owned_attachment( get_db, data.get("attachment_id"), user["id"], ) if single_attachment_error: return single_attachment_error attachment_ids = [single_attachment_id] if single_attachment_id else [] attachment_id = attachment_ids[0] if attachment_ids else None if not text and not attachment_id: return jsonify({"error": "Message cannot be empty."}), 400 channel = get_channel_by_legacy_room_id(room["id"]) channel_id = channel["id"] if channel else None with get_db() as db: db.execute("BEGIN IMMEDIATE") cursor = db.execute( """ INSERT INTO messages ( room_id, channel_id, direct_conversation_id, user_id, username, text, attachment_id, parent_message_id, created_at ) VALUES (?, ?, NULL, ?, ?, ?, ?, NULL, ?) """, ( room["id"], channel_id, user["id"], user["username"], text, attachment_id, now_iso(), ), ) message_id = cursor.lastrowid for position, extra_attachment_id in enumerate(attachment_ids): db.execute( """ INSERT OR IGNORE INTO message_attachments ( message_id, attachment_id, position ) VALUES (?, ?, ?) """, (message_id, extra_attachment_id, position), ) db.commit() with get_db() as db: row = db.execute(f"{MESSAGE_SELECT} WHERE m.id = ?", (message_id,)).fetchone() audit("room_message_sent", user["id"], "room", str(room["id"]), str(message_id)) return jsonify({"message": build_message_payload(row, user)}), 201 @app.route("/api/rooms//messages", methods=["DELETE"]) def clear_room_messages(slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 moderator, error = require_moderator() if error: return error limited = rate_limited("admin") if limited: return limited room = get_room_by_slug(slug) if not room: return jsonify({"error": "Room not found."}), 404 cursor = db_write( """ DELETE FROM messages WHERE room_id = ? AND direct_conversation_id IS NULL """, (room["id"],), ) audit("room_messages_cleared", moderator["id"], "room", str(room["id"]), str(cursor.rowcount)) return jsonify({"status": "cleared", "deleted_messages": cursor.rowcount}) @app.route("/api/teams//channels//messages") def api_get_channel_messages(team_slug: str, channel_slug: str): channel = get_channel_by_slug(team_slug, channel_slug) if not channel: return jsonify({"error": "Channel not found."}), 404 user = current_user() if not can_access_channel(channel, user): if not user: return jsonify({"error": "Login required.", "code": "login_required"}), 401 return jsonify({"error": "You do not have access to this channel.", "code": "forbidden"}), 403 after_raw = request.args.get("after_id", "").strip() after_id = 0 if after_raw: try: after_id = max(0, int(after_raw)) except ValueError: return jsonify({"error": "after_id must be a non-negative integer."}), 400 with get_db() as db: if after_raw: rows = db.execute( f""" {MESSAGE_SELECT} WHERE m.channel_id = ? AND m.parent_message_id IS NULL AND m.direct_conversation_id IS NULL AND m.id > ? ORDER BY m.id ASC LIMIT ? """, (channel["id"], after_id, MAX_INCREMENTAL_MESSAGES), ).fetchall() else: rows = db.execute( f""" SELECT * FROM ( {MESSAGE_SELECT} WHERE m.channel_id = ? AND m.parent_message_id IS NULL AND m.direct_conversation_id IS NULL ORDER BY m.id DESC LIMIT ? ) ORDER BY id ASC """, (channel["id"], MAX_HISTORY_MESSAGES), ).fetchall() messages = [build_message_payload(row, user) for row in rows] latest_id = messages[-1]["id"] if messages else after_id return jsonify( { "team": {"slug": channel["team_slug"], "name": channel["team_name"]}, "channel": channel_payload(channel, user), "messages": messages, "latest_id": latest_id, "has_more": len(messages) == MAX_INCREMENTAL_MESSAGES, } ) @app.route("/api/teams//channels//messages", methods=["POST"]) def api_send_channel_message(team_slug: str, channel_slug: str): limited = rate_limited("send") if limited: return limited if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error channel = get_channel_by_slug(team_slug, channel_slug) if not channel: return jsonify({"error": "Channel not found."}), 404 if not can_write_channel(channel, user): return jsonify({"error": "You cannot post in this channel.", "code": "forbidden"}), 403 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 text = data.get("text", "") if not isinstance(text, str): return jsonify({"error": "Message must be text."}), 400 text = text.strip() if len(text) > MAX_MESSAGE_LENGTH: return jsonify({"error": "Message is too long."}), 400 attachment_ids, attachment_error = resolve_owned_attachment_ids( get_db, data.get("attachment_ids", []), user["id"], ) if attachment_error: return attachment_error if not attachment_ids and data.get("attachment_id") is not None: single_attachment_id, single_attachment_error = resolve_owned_attachment( get_db, data.get("attachment_id"), user["id"], ) if single_attachment_error: return single_attachment_error attachment_ids = [single_attachment_id] if single_attachment_id else [] attachment_id = attachment_ids[0] if attachment_ids else None if not text and not attachment_id: return jsonify({"error": "Message cannot be empty."}), 400 with get_db() as db: db.execute("BEGIN IMMEDIATE") cursor = db.execute( """ INSERT INTO messages ( room_id, channel_id, direct_conversation_id, user_id, username, text, attachment_id, parent_message_id, created_at ) VALUES (?, ?, NULL, ?, ?, ?, ?, NULL, ?) """, ( channel["legacy_room_id"], channel["id"], user["id"], user["username"], text, attachment_id, now_iso(), ), ) message_id = cursor.lastrowid for position, extra_attachment_id in enumerate(attachment_ids): db.execute( """ INSERT OR IGNORE INTO message_attachments ( message_id, attachment_id, position ) VALUES (?, ?, ?) """, (message_id, extra_attachment_id, position), ) db.commit() with get_db() as db: row = db.execute(f"{MESSAGE_SELECT} WHERE m.id = ?", (message_id,)).fetchone() audit("channel_message_sent", user["id"], "channel", str(channel["id"]), str(message_id)) return jsonify({"message": build_message_payload(row, user)}), 201 @app.route("/api/messages/", methods=["PATCH"]) def edit_message(message_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 limited = rate_limited("send") if limited: return limited user, error = require_writable_user() if error: return error try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 new_text = data.get("text", "") if not isinstance(new_text, str): return jsonify({"error": "Message must be text."}), 400 new_text = new_text.strip() attachment_ids, attachment_error = resolve_owned_attachment_ids( get_db, data.get("attachment_ids", []), user["id"], ) if attachment_error: return attachment_error if len(new_text) > MAX_MESSAGE_LENGTH: return jsonify({"error": "Message is too long."}), 400 if not new_text and not attachment_ids: return jsonify({"error": "Message cannot be empty."}), 400 with get_db() as db: message = db.execute(f"{MESSAGE_SELECT} WHERE m.id = ?", (message_id,)).fetchone() if not message: return jsonify({"error": "Message not found."}), 404 is_owner = message["user_id"] == user["id"] is_moderator = user["role"] in {"moderator", "admin"} if not is_owner and not is_moderator: return jsonify({"error": "You cannot edit this message.", "code": "forbidden"}), 403 created_at = parse_iso(message["created_at"]) if is_owner and created_at: age = (utc_now() - created_at).total_seconds() if age > MAX_MESSAGE_EDIT_WINDOW_SECONDS: return jsonify({"error": "Edit window has expired."}), 403 old_attachments = ctx["attachment_list_for_message"](message_id) with get_db() as db: db.execute("BEGIN IMMEDIATE") db.execute( "UPDATE messages SET text = ?, attachment_id = ? WHERE id = ?", (new_text, attachment_ids[0] if attachment_ids else None, message_id), ) db.execute("DELETE FROM message_attachments WHERE message_id = ?", (message_id,)) for position, attachment_id in enumerate(attachment_ids): db.execute( """ INSERT INTO message_attachments ( message_id, attachment_id, position ) VALUES (?, ?, ?) """, (message_id, attachment_id, position), ) db.execute( """ INSERT INTO message_edits ( message_id, edited_by_user_id, previous_text, new_text, previous_attachments_json, new_attachments_json, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( message_id, user["id"], message["text"], new_text, json.dumps(old_attachments), json.dumps(attachment_ids), now_iso(), ), ) db.commit() with get_db() as db: updated = db.execute(f"{MESSAGE_SELECT} WHERE m.id = ?", (message_id,)).fetchone() audit("message_edited", user["id"], "message", str(message_id)) return jsonify({"message": build_message_payload(updated, user)}) @app.route("/api/messages/", methods=["DELETE"]) def delete_message(message_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 moderator, error = require_moderator() if error: return error limited = rate_limited("admin") if limited: return limited cursor = db_write("DELETE FROM messages WHERE id = ?", (message_id,)) if cursor.rowcount == 0: return jsonify({"error": "Message not found."}), 404 audit("message_deleted", moderator["id"], "message", str(message_id)) return jsonify({"status": "deleted"}) @app.route("/api/messages/history") def api_message_history(): user = current_user() kind = str(request.args.get("kind", "")).strip() slug = str(request.args.get("slug", "")).strip() team_slug = str(request.args.get("team_slug", "")).strip() before_id_raw = str(request.args.get("before_id", "")).strip() try: before_id = int(before_id_raw) except ValueError: return jsonify({"error": "before_id is required."}), 400 if kind == "room": room = get_room_by_slug(slug) if not room or not can_access_room(room, user): return jsonify({"error": "Room not found."}), 404 with get_db() as db: rows = db.execute( f""" SELECT * FROM ( {MESSAGE_SELECT} WHERE m.room_id = ? AND m.parent_message_id IS NULL AND m.direct_conversation_id IS NULL AND m.id < ? ORDER BY m.id DESC LIMIT ? ) ORDER BY id ASC """, (room["id"], before_id, MAX_HISTORY_MESSAGES), ).fetchall() elif kind == "channel": channel = get_channel_by_slug(team_slug, slug) if not channel or not can_access_channel(channel, user): return jsonify({"error": "Channel not found."}), 404 with get_db() as db: rows = db.execute( f""" SELECT * FROM ( {MESSAGE_SELECT} WHERE m.channel_id = ? AND m.parent_message_id IS NULL AND m.direct_conversation_id IS NULL AND m.id < ? ORDER BY m.id DESC LIMIT ? ) ORDER BY id ASC """, (channel["id"], before_id, MAX_HISTORY_MESSAGES), ).fetchall() elif kind == "dm": dm_id_raw = str(request.args.get("dm_id", "")).strip() try: dm_id = int(dm_id_raw) except ValueError: return jsonify({"error": "dm_id is required for DM history."}), 400 if not validate_dm_access(get_db, dm_id, user): return jsonify({"error": "DM not found."}), 404 with get_db() as db: rows = db.execute( f""" SELECT * FROM ( {MESSAGE_SELECT} WHERE m.direct_conversation_id = ? AND m.parent_message_id IS NULL AND m.id < ? ORDER BY m.id DESC LIMIT ? ) ORDER BY id ASC """, (dm_id, before_id, MAX_HISTORY_MESSAGES), ).fetchall() else: return jsonify({"error": "Invalid history kind."}), 400 messages = [build_message_payload(row, user) for row in rows] return jsonify({"messages": messages, "has_more": len(messages) == MAX_HISTORY_MESSAGES}) @app.route("/api/messages//thread") def api_get_message_thread(message_id: int): user = current_user() with get_db() as db: root = db.execute( f""" {MESSAGE_SELECT} WHERE m.id = ? AND m.parent_message_id IS NULL """, (message_id,), ).fetchone() if not root: return jsonify({"error": "Thread root message not found."}), 404 if root["direct_conversation_id"]: if not validate_dm_access(get_db, root["direct_conversation_id"], user): if not user: return jsonify({"error": "Login required.", "code": "login_required"}), 401 return jsonify({"error": "You do not have access to this thread.", "code": "forbidden"}), 403 else: channel = None if root["channel_id"]: channel = get_channel_by_id(root["channel_id"]) if not channel and root["room_id"]: channel = get_channel_by_legacy_room_id(root["room_id"]) if not channel or not can_access_channel(channel, user): if not user: return jsonify({"error": "Login required.", "code": "login_required"}), 401 return jsonify({"error": "You do not have access to this thread.", "code": "forbidden"}), 403 with get_db() as db: replies = db.execute( f""" {MESSAGE_SELECT} WHERE m.parent_message_id = ? ORDER BY m.id ASC LIMIT ? """, (message_id, MAX_THREAD_REPLIES), ).fetchall() payload = { "root": build_message_payload(root, user), "replies": [build_message_payload(reply, user) for reply in replies], } if root["direct_conversation_id"]: payload["dm"] = dm_payload(get_db, root["direct_conversation_id"], user["id"]) else: channel = ( get_channel_by_id(root["channel_id"]) if root["channel_id"] else get_channel_by_legacy_room_id(root["room_id"]) ) payload["team"] = {"slug": channel["team_slug"], "name": channel["team_name"]} payload["channel"] = channel_payload(channel, user) return jsonify(payload) @app.route("/api/messages//thread", methods=["POST"]) def api_reply_to_thread(message_id: int): limited = rate_limited("send") if limited: return limited if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error with get_db() as db: root = db.execute( """ SELECT id, room_id, channel_id, direct_conversation_id, parent_message_id FROM messages WHERE id = ? """, (message_id,), ).fetchone() if not root or root["parent_message_id"] is not None: return jsonify({"error": "Thread root message not found."}), 404 if root["direct_conversation_id"]: if not validate_dm_access(get_db, root["direct_conversation_id"], user): return jsonify({"error": "You cannot reply in this thread.", "code": "forbidden"}), 403 else: channel = None if root["channel_id"]: channel = get_channel_by_id(root["channel_id"]) if not channel and root["room_id"]: channel = get_channel_by_legacy_room_id(root["room_id"]) if not channel or not can_write_channel(channel, user): return jsonify({"error": "You cannot reply in this thread.", "code": "forbidden"}), 403 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 text = data.get("text", "") if not isinstance(text, str): return jsonify({"error": "Reply must be text."}), 400 text = text.strip() if len(text) > MAX_MESSAGE_LENGTH: return jsonify({"error": "Reply is too long."}), 400 attachment_ids, attachment_error = resolve_owned_attachment_ids( get_db, data.get("attachment_ids", []), user["id"], ) if attachment_error: return attachment_error if not attachment_ids and data.get("attachment_id") is not None: single_attachment_id, single_attachment_error = resolve_owned_attachment( get_db, data.get("attachment_id"), user["id"], ) if single_attachment_error: return single_attachment_error attachment_ids = [single_attachment_id] if single_attachment_id else [] attachment_id = attachment_ids[0] if attachment_ids else None if not text and not attachment_id: return jsonify({"error": "Reply cannot be empty."}), 400 with get_db() as db: db.execute("BEGIN IMMEDIATE") cursor = db.execute( """ INSERT INTO messages ( room_id, channel_id, direct_conversation_id, user_id, username, text, attachment_id, parent_message_id, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( root["room_id"], root["channel_id"], root["direct_conversation_id"], user["id"], user["username"], text, attachment_id, root["id"], now_iso(), ), ) reply_id = cursor.lastrowid for position, extra_attachment_id in enumerate(attachment_ids): db.execute( """ INSERT OR IGNORE INTO message_attachments ( message_id, attachment_id, position ) VALUES (?, ?, ?) """, (reply_id, extra_attachment_id, position), ) db.commit() with get_db() as db: reply = db.execute(f"{MESSAGE_SELECT} WHERE m.id = ?", (reply_id,)).fetchone() audit("thread_reply_sent", user["id"], "message", str(message_id), str(reply_id)) return jsonify({"reply": build_message_payload(reply, user)}), 201 @app.route("/api/feed") def get_feed(): user = current_user() after_raw = request.args.get("after_id", "").strip() after_id = 0 if after_raw: try: after_id = max(0, int(after_raw)) except ValueError: return jsonify({"error": "after_id must be a non-negative integer."}), 400 base_select = f""" SELECT m.id, m.room_id, m.channel_id, m.direct_conversation_id, m.user_id, m.username, m.text, m.created_at, m.attachment_id, m.parent_message_id, ( SELECT COUNT(*) FROM messages AS thread_reply WHERE thread_reply.parent_message_id = m.id ) AS thread_reply_count, r.slug AS room_slug, r.name AS room_name, r.visibility AS room_visibility, c.slug AS channel_slug, c.name AS channel_name, t.slug AS team_slug, t.name AS team_name, a.original_name AS attachment_original_name, a.mime_type AS attachment_mime_type, a.size_bytes AS attachment_size_bytes, u.role AS user_role, us.name_color AS user_name_color FROM messages AS m JOIN users AS u ON u.id = m.user_id LEFT JOIN user_settings AS us ON us.user_id = u.id LEFT JOIN rooms AS r ON r.id = m.room_id LEFT JOIN channels AS c ON c.id = m.channel_id AND c.is_active = 1 LEFT JOIN teams AS t ON t.id = c.team_id AND t.is_active = 1 LEFT JOIN attachments AS a ON a.id = m.attachment_id """ params = [] if user and user["role"] in {"moderator", "admin"}: access_clause = """ ( (m.direct_conversation_id IS NULL AND r.id IS NOT NULL AND r.is_active = 1) OR (m.direct_conversation_id IS NULL AND c.id IS NOT NULL AND c.is_active = 1 AND t.is_active = 1) OR ( m.direct_conversation_id IS NOT NULL AND EXISTS ( SELECT 1 FROM direct_conversation_members AS dcm WHERE dcm.direct_conversation_id = m.direct_conversation_id AND dcm.user_id = ? ) ) ) """ params.append(user["id"]) elif user: access_clause = """ ( ( m.direct_conversation_id IS NULL AND r.id IS NOT NULL AND r.is_active = 1 AND ( r.visibility = 'public' OR EXISTS ( SELECT 1 FROM room_members AS rm WHERE rm.room_id = r.id AND rm.user_id = ? ) ) ) OR ( m.direct_conversation_id IS NULL AND c.id IS NOT NULL AND c.is_active = 1 AND t.is_active = 1 AND c.legacy_room_id IS NULL AND EXISTS ( SELECT 1 FROM team_members AS tm WHERE tm.team_id = c.team_id AND tm.user_id = ? ) ) OR ( m.direct_conversation_id IS NOT NULL AND EXISTS ( SELECT 1 FROM direct_conversation_members AS dcm WHERE dcm.direct_conversation_id = m.direct_conversation_id AND dcm.user_id = ? ) ) ) """ params.extend([user["id"], user["id"], user["id"]]) else: access_clause = """ m.direct_conversation_id IS NULL AND r.id IS NOT NULL AND r.is_active = 1 AND r.visibility = 'public' """ if after_raw: query = f""" {base_select} WHERE m.parent_message_id IS NULL AND {access_clause} AND m.id > ? ORDER BY m.id ASC LIMIT ? """ query_params = [*params, after_id, MAX_FEED_MESSAGES] with get_db() as db: rows = db.execute(query, tuple(query_params)).fetchall() entries = [feed_message_payload(get_db, row, user) for row in rows] latest_id = entries[-1]["id"] if entries else after_id return jsonify( { "entries": entries, "latest_id": latest_id, "has_more": len(entries) == MAX_FEED_MESSAGES, } ) query = f""" SELECT * FROM ( {base_select} WHERE m.parent_message_id IS NULL AND {access_clause} ORDER BY m.id DESC LIMIT ? ) ORDER BY id ASC """ query_params = [*params, MAX_FEED_MESSAGES] with get_db() as db: rows = db.execute(query, tuple(query_params)).fetchall() entries = [feed_message_payload(get_db, row, user) for row in rows] latest_id = entries[-1]["id"] if entries else 0 return jsonify({"entries": entries, "latest_id": latest_id, "has_more": False}) @app.route("/api/search") def api_search(): user = current_user() query = str(request.args.get("q", "")).strip() if not query: return jsonify({"results": [], "query": ""}) if len(query) > MAX_SEARCH_QUERY_LENGTH: return jsonify({"error": "Search query is too long."}), 400 pattern = f"%{query.lower()}%" with get_db() as db: rows = db.execute( """ SELECT m.id, m.room_id, m.channel_id, m.direct_conversation_id, m.user_id, m.username, m.text, m.created_at, m.attachment_id, m.parent_message_id, ( SELECT COUNT(*) FROM messages AS thread_reply WHERE thread_reply.parent_message_id = m.id ) AS thread_reply_count, r.slug AS room_slug, r.name AS room_name, r.visibility AS room_visibility, c.slug AS channel_slug, c.name AS channel_name, t.slug AS team_slug, t.name AS team_name, a.original_name AS attachment_original_name, a.mime_type AS attachment_mime_type, a.size_bytes AS attachment_size_bytes, u.role AS user_role, us.name_color AS user_name_color FROM messages AS m JOIN users AS u ON u.id = m.user_id LEFT JOIN user_settings AS us ON us.user_id = u.id LEFT JOIN rooms AS r ON r.id = m.room_id LEFT JOIN channels AS c ON c.id = m.channel_id LEFT JOIN teams AS t ON t.id = c.team_id LEFT JOIN attachments AS a ON a.id = m.attachment_id WHERE LOWER(m.text) LIKE ? ORDER BY m.id DESC LIMIT ? """, (pattern, MAX_SEARCH_RESULTS), ).fetchall() visible = [] for row in rows: allowed = False if row["direct_conversation_id"]: allowed = bool(user and validate_dm_access(get_db, row["direct_conversation_id"], user)) elif row["channel_id"]: channel = get_channel_by_id(row["channel_id"]) allowed = bool(channel and can_access_channel(channel, user)) elif row["room_id"]: room = room_by_id(row["room_id"]) allowed = bool(room and can_access_room(room, user)) if allowed: visible.append(feed_message_payload(get_db, row, user)) return jsonify({"query": query, "results": visible}) @app.route("/api/read-receipts", methods=["POST"]) def api_mark_read_receipts(): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_user() if error: return error try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 ids = data.get("message_ids", []) if not isinstance(ids, list): return jsonify({"error": "message_ids must be an array."}), 400 parsed_ids = [] for item in ids[:READ_RECEIPT_BATCH_LIMIT]: try: parsed_ids.append(int(item)) except (TypeError, ValueError): continue if not parsed_ids: return jsonify({"status": "ok", "updated": 0}) updated = 0 with get_db() as db: db.execute("BEGIN IMMEDIATE") for message_id in parsed_ids: row = db.execute( """ SELECT id, room_id, channel_id, direct_conversation_id FROM messages WHERE id = ? """, (message_id,), ).fetchone() if not row: continue allowed = False if row["direct_conversation_id"]: allowed = validate_dm_access(get_db, row["direct_conversation_id"], user) elif row["channel_id"]: channel = get_channel_by_id(row["channel_id"]) allowed = bool(channel and can_access_channel(channel, user)) elif row["room_id"]: room = room_by_id(row["room_id"]) allowed = bool(room and can_access_room(room, user)) if not allowed: continue db.execute( """ INSERT INTO message_receipts ( message_id, user_id, seen_at ) VALUES (?, ?, ?) ON CONFLICT(message_id, user_id) DO UPDATE SET seen_at = excluded.seen_at """, (message_id, user["id"], now_iso()), ) updated += 1 db.commit() return jsonify({"status": "ok", "updated": updated}) @app.route("/api/messages//receipts") def api_get_message_receipts(message_id: int): user, error = require_user() if error: return error with get_db() as db: message = db.execute( """ SELECT id, room_id, channel_id, direct_conversation_id, user_id FROM messages WHERE id = ? """, (message_id,), ).fetchone() if not message: return jsonify({"error": "Message not found."}), 404 allowed = False if message["direct_conversation_id"]: allowed = validate_dm_access(get_db, message["direct_conversation_id"], user) elif message["channel_id"]: channel = get_channel_by_id(message["channel_id"]) allowed = bool(channel and can_access_channel(channel, user)) elif message["room_id"]: room = room_by_id(message["room_id"]) allowed = bool(room and can_access_room(room, user)) if not allowed: return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 with get_db() as db: rows = db.execute( """ SELECT u.id, u.username, mr.seen_at FROM message_receipts AS mr JOIN users AS u ON u.id = mr.user_id WHERE mr.message_id = ? ORDER BY mr.seen_at DESC, u.username COLLATE NOCASE """, (message_id,), ).fetchall() return jsonify( { "message_id": message_id, "receipts": [ { "user_id": row["id"], "username": row["username"], "seen_at": row["seen_at"], } for row in rows ], } ) @app.route("/api/messages//reactions") def api_get_message_reactions(message_id: int): user = current_user() with get_db() as db: row = db.execute( """ SELECT id, room_id, channel_id, direct_conversation_id FROM messages WHERE id = ? """, (message_id,), ).fetchone() if not row: return jsonify({"error": "Message not found."}), 404 allowed = False if row["direct_conversation_id"]: allowed = bool(user and validate_dm_access(get_db, row["direct_conversation_id"], user)) elif row["channel_id"]: channel = get_channel_by_id(row["channel_id"]) allowed = bool(channel and can_access_channel(channel, user)) elif row["room_id"]: room = room_by_id(row["room_id"]) allowed = bool(room and can_access_room(room, user)) if not allowed: if not user: return jsonify({"error": "Login required.", "code": "login_required"}), 401 return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 with get_db() as db: detail_rows = db.execute( """ SELECT mr.emoji, u.id AS user_id, u.username, mr.created_at FROM message_reactions AS mr JOIN users AS u ON u.id = mr.user_id WHERE mr.message_id = ? ORDER BY mr.emoji ASC, u.username COLLATE NOCASE """, (message_id,), ).fetchall() grouped = {} for item in detail_rows: grouped.setdefault(item["emoji"], []).append( { "user_id": item["user_id"], "username": item["username"], "created_at": item["created_at"], } ) summary = [] for emoji, users in grouped.items(): summary.append( { "emoji": emoji, "count": len(users), "reacted": bool(user and any(x["user_id"] == user["id"] for x in users)), "users": users, } ) summary.sort(key=lambda x: (-x["count"], x["emoji"])) return jsonify({"message_id": message_id, "reactions": summary}) @app.route("/api/messages//reactions", methods=["POST"]) def api_add_reaction(message_id: int): limited = rate_limited("reaction") if limited: return limited if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 emoji = str(data.get("emoji", "")).strip() if not emoji or len(emoji) > 16 or not REACTION_EMOJI_RE.fullmatch(emoji): return jsonify({"error": "Invalid reaction."}), 400 with get_db() as db: row = db.execute( """ SELECT id, room_id, channel_id, direct_conversation_id FROM messages WHERE id = ? """, (message_id,), ).fetchone() if not row: return jsonify({"error": "Message not found."}), 404 allowed = False if row["direct_conversation_id"]: allowed = validate_dm_access(get_db, row["direct_conversation_id"], user) elif row["channel_id"]: channel = get_channel_by_id(row["channel_id"]) allowed = bool(channel and can_access_channel(channel, user)) elif row["room_id"]: room = room_by_id(row["room_id"]) allowed = bool(room and can_access_room(room, user)) if not allowed: return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 db_write( """ INSERT OR IGNORE INTO message_reactions ( message_id, user_id, emoji, created_at ) VALUES (?, ?, ?, ?) """, (message_id, user["id"], emoji, now_iso()), ) audit("message_reaction_added", user["id"], "message", str(message_id), emoji) return jsonify( { "message_id": message_id, "reactions": reaction_summary_for_message(get_db, message_id, user), } ), 201 @app.route("/api/messages//reactions/", methods=["DELETE"]) def api_remove_reaction(message_id: int, emoji: str): limited = rate_limited("reaction") if limited: return limited if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error db_write( """ DELETE FROM message_reactions WHERE message_id = ? AND user_id = ? AND emoji = ? """, (message_id, user["id"], emoji), ) audit("message_reaction_removed", user["id"], "message", str(message_id), emoji) return jsonify( { "message_id": message_id, "reactions": reaction_summary_for_message(get_db, message_id, user), } ) @app.route("/api/messages//bookmark", methods=["POST"]) def api_bookmark_message(message_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_user() if error: return error return jsonify({"status": "stubbed", "message_id": message_id}) @app.route("/api/messages//bookmark", methods=["DELETE"]) def api_unbookmark_message(message_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_user() if error: return error return jsonify({"status": "stubbed", "message_id": message_id}) @app.route("/api/bookmarks") def api_list_bookmarks(): user, error = require_user() if error: return error return jsonify({"messages": []}) @app.route("/api/messages//pin", methods=["POST"]) def api_pin_message(message_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_user() if error: return error return jsonify({"status": "stubbed", "message_id": message_id}) @app.route("/api/messages//pin", methods=["DELETE"]) def api_unpin_message(message_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_user() if error: return error return jsonify({"status": "stubbed", "message_id": message_id}) @app.route("/api/pins") def api_list_pins(): user, error = require_user() if error: return error return jsonify({"messages": []}) @app.route("/api/messages//share", methods=["POST"]) def api_share_message(message_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_user() if error: return error return jsonify( { "share": { "url": build_local_stub_share_url("message", message_id) } } ) @app.route("/api/attachments//share", methods=["POST"]) def api_share_attachment(attachment_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_user() if error: return error return jsonify( { "share": { "url": build_local_stub_share_url("attachment", attachment_id) } } ) @app.route("/api/presence", methods=["POST"]) def api_presence_ping(): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_user() if error: return error try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 view_name = str(data.get("view", "")).strip()[:32] or None room_id = data.get("room_id") channel_id = data.get("channel_id") dm_id = data.get("dm_id") try: room_id = int(room_id) if room_id is not None else None except (TypeError, ValueError): room_id = None try: channel_id = int(channel_id) if channel_id is not None else None except (TypeError, ValueError): channel_id = None try: dm_id = int(dm_id) if dm_id is not None else None except (TypeError, ValueError): dm_id = None update_presence(db_write, user["id"], view_name, room_id, channel_id, dm_id) return jsonify({"status": "ok"}) @app.route("/api/rooms//presence") def api_room_presence(slug: str): room = get_room_by_slug(slug) user = current_user() if not room or not can_access_room(room, user): return jsonify({"error": "Room not found."}), 404 return jsonify( { "room_id": room["id"], "online_count": room_presence_count(get_db, PRESENCE_TTL_SECONDS, room["id"]), } ) @app.route("/api/teams//channels//presence") def api_channel_presence(team_slug: str, channel_slug: str): channel = get_channel_by_slug(team_slug, channel_slug) user = current_user() if not channel or not can_access_channel(channel, user): return jsonify({"error": "Channel not found."}), 404 return jsonify( { "channel_id": channel["id"], "online_count": channel_presence_count(get_db, PRESENCE_TTL_SECONDS, channel["id"]), } ) @app.route("/api/dms//presence") def api_dm_presence(dm_id: int): user = current_user() if not validate_dm_access(get_db, dm_id, user): return jsonify({"error": "DM not found."}), 404 return jsonify( { "dm_id": dm_id, "online_count": dm_presence_count(get_db, PRESENCE_TTL_SECONDS, dm_id), } ) @app.route("/api/rooms//typing", methods=["POST"]) def set_typing(slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error limited = rate_limited("typing") if limited: return limited room = get_room_by_slug(slug) if not room or not can_write_room(room, user): return jsonify({"error": "You cannot use typing indicators here.", "code": "forbidden"}), 403 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 typing = bool(data.get("typing")) if typing: db_write( """ INSERT INTO typing_events ( room_id, user_id, updated_at ) VALUES (?, ?, ?) ON CONFLICT(room_id, user_id) DO UPDATE SET updated_at = excluded.updated_at """, (room["id"], user["id"], time.time()), ) else: db_write( "DELETE FROM typing_events WHERE room_id = ? AND user_id = ?", (room["id"], user["id"]), ) return jsonify({"status": "ok"}) @app.route("/api/rooms//typing") def get_typing(slug: str): room = get_room_by_slug(slug) user = current_user() if not room or not can_access_room(room, user): return jsonify({"error": "Room not found."}), 404 cutoff = time.time() - TYPING_TTL_SECONDS with get_db() as db: db.execute("DELETE FROM typing_events WHERE updated_at < ?", (cutoff,)) rows = db.execute( """ SELECT u.id, u.username FROM typing_events AS t JOIN users AS u ON u.id = t.user_id WHERE t.room_id = ? AND t.user_id != ? AND t.updated_at >= ? ORDER BY u.username COLLATE NOCASE """, (room["id"], user["id"] if user else -1, cutoff), ).fetchall() db.commit() return jsonify({"users": [{"id": row["id"], "username": row["username"]} for row in rows]}) @app.route("/api/teams//channels//typing", methods=["POST"]) def api_set_channel_typing(team_slug: str, channel_slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error limited = rate_limited("typing") if limited: return limited channel = get_channel_by_slug(team_slug, channel_slug) if not channel: return jsonify({"error": "Channel not found."}), 404 if not can_write_channel(channel, user): return jsonify({"error": "You cannot use typing indicators here.", "code": "forbidden"}), 403 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 typing = bool(data.get("typing")) if typing: db_write( """ INSERT INTO channel_typing_events ( channel_id, user_id, updated_at ) VALUES (?, ?, ?) ON CONFLICT(channel_id, user_id) DO UPDATE SET updated_at = excluded.updated_at """, (channel["id"], user["id"], time.time()), ) else: db_write( "DELETE FROM channel_typing_events WHERE channel_id = ? AND user_id = ?", (channel["id"], user["id"]), ) return jsonify({"status": "ok"}) @app.route("/api/teams//channels//typing") def api_get_channel_typing(team_slug: str, channel_slug: str): channel = get_channel_by_slug(team_slug, channel_slug) if not channel: return jsonify({"error": "Channel not found."}), 404 user = current_user() if not can_access_channel(channel, user): if not user: return jsonify({"error": "Login required.", "code": "login_required"}), 401 return jsonify({"error": "You do not have access to this channel.", "code": "forbidden"}), 403 cutoff = time.time() - TYPING_TTL_SECONDS with get_db() as db: db.execute("DELETE FROM channel_typing_events WHERE updated_at < ?", (cutoff,)) rows = db.execute( """ SELECT u.id, u.username FROM channel_typing_events AS t JOIN users AS u ON u.id = t.user_id WHERE t.channel_id = ? AND t.user_id != ? AND t.updated_at >= ? ORDER BY u.username COLLATE NOCASE """, (channel["id"], user["id"] if user else -1, cutoff), ).fetchall() db.commit() return jsonify({"users": [{"id": row["id"], "username": row["username"]} for row in rows]}) @app.route("/api/dms//typing", methods=["POST"]) def api_set_dm_typing(dm_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error limited = rate_limited("typing") if limited: return limited if not validate_dm_access(get_db, dm_id, user): return jsonify({"error": "DM not found."}), 404 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 typing = bool(data.get("typing")) if typing: db_write( """ INSERT INTO dm_typing_events ( direct_conversation_id, user_id, updated_at ) VALUES (?, ?, ?) ON CONFLICT(direct_conversation_id, user_id) DO UPDATE SET updated_at = excluded.updated_at """, (dm_id, user["id"], time.time()), ) else: db_write( """ DELETE FROM dm_typing_events WHERE direct_conversation_id = ? AND user_id = ? """, (dm_id, user["id"]), ) return jsonify({"status": "ok"}) @app.route("/api/dms//typing") def api_get_dm_typing(dm_id: int): user = current_user() if not validate_dm_access(get_db, dm_id, user): return jsonify({"error": "DM not found."}), 404 cutoff = time.time() - TYPING_TTL_SECONDS with get_db() as db: db.execute("DELETE FROM dm_typing_events WHERE updated_at < ?", (cutoff,)) rows = db.execute( """ SELECT u.id, u.username FROM dm_typing_events AS t JOIN users AS u ON u.id = t.user_id WHERE t.direct_conversation_id = ? AND t.user_id != ? AND t.updated_at >= ? ORDER BY u.username COLLATE NOCASE """, (dm_id, user["id"], cutoff), ).fetchall() db.commit() return jsonify({"users": [{"id": row["id"], "username": row["username"]} for row in rows]}) @app.route("/api/rooms") def list_rooms(): user = current_user() with get_db() as db: rows = db.execute( """ SELECT id, slug, name, description, visibility, created_by_user_id, is_active FROM rooms WHERE is_active = 1 ORDER BY CASE WHEN slug = 'general' THEN 0 ELSE 1 END, name COLLATE NOCASE """ ).fetchall() rooms = [room_payload(room, user) for room in rows if can_access_room(room, user)] return jsonify({"rooms": rooms, "authenticated": bool(user)}) @app.route("/api/rooms", methods=["POST"]) def create_room(): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 admin, error = require_admin() if error: return error limited = rate_limited("admin") if limited: return limited try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 name = str(data.get("name", "")).strip() description = str(data.get("description", "")).strip()[:300] visibility = data.get("visibility", "public") if not name or len(name) > 60: return jsonify({"error": "Room name must be 1 to 60 characters."}), 400 if visibility not in ROOM_VISIBILITIES: return jsonify({"error": "Invalid room visibility."}), 400 base_slug = re.sub(r"[^a-z0-9-]+", "-", name.lower()).strip("-") base_slug = base_slug[:48].strip("-") if not base_slug or not ROOM_SLUG_RE.fullmatch(base_slug): return jsonify({"error": "Room name cannot create a valid room slug."}), 400 with get_db() as db: db.execute("BEGIN IMMEDIATE") slug = base_slug suffix = 2 while db.execute("SELECT 1 FROM rooms WHERE slug = ?", (slug,)).fetchone(): slug = f"{base_slug[:44]}-{suffix}" suffix += 1 cursor = db.execute( """ INSERT INTO rooms ( slug, name, description, visibility, created_by_user_id, created_at, is_active ) VALUES (?, ?, ?, ?, ?, ?, 1) """, (slug, name, description, visibility, admin["id"], now_iso()), ) room_id = cursor.lastrowid db.execute( """ INSERT INTO room_members ( room_id, user_id, membership_role, added_at ) VALUES (?, ?, 'owner', ?) """, (room_id, admin["id"], now_iso()), ) general_team = db.execute( "SELECT id FROM teams WHERE slug = 'general' AND is_active = 1" ).fetchone() if general_team: db.execute( """ INSERT INTO channels ( team_id, legacy_room_id, slug, name, description, created_by_user_id, created_at, is_active ) VALUES (?, ?, ?, ?, ?, ?, ?, 1) """, ( general_team["id"], room_id, slug, name, description, admin["id"], now_iso(), ), ) db.commit() audit("room_created", admin["id"], "room", str(room_id), f"{slug}:{visibility}") return jsonify( { "room": { "id": room_id, "slug": slug, "name": name, "description": description, "visibility": visibility, "can_read": True, "can_write": True, "can_manage": True, } } ), 201 @app.route("/api/rooms/", methods=["PATCH"]) def update_room(slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error limited = rate_limited("admin") if limited: return limited room = get_room_by_slug(slug) if not room: return jsonify({"error": "Room not found."}), 404 if room["slug"] == "general" and user["role"] != "admin": return jsonify({"error": "Only an administrator can modify the default room."}), 403 if not can_manage_room(room, user): return jsonify({"error": "Room owner or administrator access required.", "code": "forbidden"}), 403 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 name = str(data.get("name", room["name"])).strip() description = str(data.get("description", room["description"] or "")).strip()[:300] visibility = data.get("visibility", room["visibility"]) if not name or len(name) > 60: return jsonify({"error": "Room name must be 1 to 60 characters."}), 400 if visibility not in ROOM_VISIBILITIES: return jsonify({"error": "Invalid room visibility."}), 400 db_write( "UPDATE rooms SET name = ?, description = ?, visibility = ? WHERE id = ?", (name, description, visibility, room["id"]), ) db_write( "UPDATE channels SET name = ?, description = ? WHERE legacy_room_id = ?", (name, description, room["id"]), ) audit("room_updated", user["id"], "room", str(room["id"]), f"name={name};visibility={visibility}") return jsonify({"status": "updated"}) @app.route("/api/rooms/", methods=["DELETE"]) def delete_room(slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 admin, error = require_admin() if error: return error limited = rate_limited("admin") if limited: return limited room = get_room_by_slug(slug) if not room: return jsonify({"error": "Room not found."}), 404 if room["slug"] == "general": return jsonify({"error": "The default room cannot be deleted."}), 400 db_write("DELETE FROM rooms WHERE id = ?", (room["id"],)) audit("room_deleted", admin["id"], "room", str(room["id"]), room["slug"]) return jsonify({"status": "deleted"}) @app.route("/api/rooms//members") def list_room_members(slug: str): admin, error = require_admin() if error: return error room = get_room_by_slug(slug) if not room: return jsonify({"error": "Room not found."}), 404 if room["visibility"] != "private": return jsonify({"error": "This room is public."}), 400 with get_db() as db: rows = db.execute( """ SELECT u.id, u.username, u.role, rm.membership_role, rm.added_at FROM room_members AS rm JOIN users AS u ON u.id = rm.user_id WHERE rm.room_id = ? ORDER BY CASE rm.membership_role WHEN 'owner' THEN 0 ELSE 1 END, u.username COLLATE NOCASE """, (room["id"],), ).fetchall() return jsonify( { "room": room_payload(room, admin), "members": [ { "id": row["id"], "username": row["username"], "role": row["role"], "membership_role": row["membership_role"], "added_at": row["added_at"], } for row in rows ], } ) @app.route("/api/rooms//members", methods=["POST"]) def add_room_member(slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 admin, error = require_admin() if error: return error room = get_room_by_slug(slug) if not room: return jsonify({"error": "Room not found."}), 404 if room["visibility"] != "private": return jsonify({"error": "Members can only be managed for private rooms."}), 400 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 username = normalize_username(data.get("username")) with get_db() as db: target = db.execute( "SELECT id, username FROM users WHERE username = ?", (username,), ).fetchone() if not target: return jsonify({"error": "User not found."}), 404 db_write( """ INSERT OR IGNORE INTO room_members ( room_id, user_id, membership_role, added_at ) VALUES (?, ?, 'member', ?) """, (room["id"], target["id"], now_iso()), ) audit("room_member_added", admin["id"], "room", str(room["id"]), target["username"]) return jsonify({"status": "added"}), 201 @app.route("/api/rooms//members/", methods=["DELETE"]) def remove_room_member(slug: str, user_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 admin, error = require_admin() if error: return error room = get_room_by_slug(slug) if not room: return jsonify({"error": "Room not found."}), 404 with get_db() as db: member = db.execute( """ SELECT membership_role FROM room_members WHERE room_id = ? AND user_id = ? """, (room["id"], user_id), ).fetchone() if not member: return jsonify({"error": "Room member not found."}), 404 if member["membership_role"] == "owner": return jsonify({"error": "The room owner cannot be removed."}), 400 db_write( "DELETE FROM room_members WHERE room_id = ? AND user_id = ?", (room["id"], user_id), ) audit("room_member_removed", admin["id"], "room", str(room["id"]), str(user_id)) return jsonify({"status": "removed"}) @app.route("/api/rooms//invites", methods=["POST"]) def create_room_invite(slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 admin, error = require_admin() if error: return error room = get_room_by_slug(slug) if not room: return jsonify({"error": "Room not found."}), 404 if room["visibility"] != "private": return jsonify({"error": "Invite links are only available for private rooms."}), 400 token = ctx["secrets"].token_urlsafe(32) expires_at = (utc_now() + ctx["timedelta"](hours=INVITE_TTL_HOURS)).isoformat() db_write( """ INSERT INTO room_invites ( room_id, token_hash, created_by_user_id, created_at, expires_at ) VALUES (?, ?, ?, ?, ?) """, (room["id"], hash_identifier(token), admin["id"], now_iso(), expires_at), ) audit("room_invite_created", admin["id"], "room", str(room["id"])) return jsonify({"invite": {"url": f"/?invite={token}", "expires_at": expires_at}}), 201 @app.route("/api/invites//accept", methods=["POST"]) def accept_room_invite(token: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error token_hash = hash_identifier(token) with get_db() as db: invite = db.execute( """ SELECT ri.id, ri.room_id, ri.expires_at, ri.used_at, r.slug, r.name, r.description, r.visibility, r.created_by_user_id, r.is_active FROM room_invites AS ri JOIN rooms AS r ON r.id = ri.room_id WHERE ri.token_hash = ? """, (token_hash,), ).fetchone() if not invite: return jsonify({"error": "Invite link is invalid."}), 404 if invite["used_at"] or parse_iso(invite["expires_at"]) <= utc_now(): return jsonify({"error": "Invite link has expired or was already used."}), 400 if invite["visibility"] != "private" or not invite["is_active"]: return jsonify({"error": "This room is no longer available."}), 404 with get_db() as db: db.execute("BEGIN IMMEDIATE") fresh = db.execute( "SELECT id, used_at, expires_at FROM room_invites WHERE id = ?", (invite["id"],), ).fetchone() if not fresh or fresh["used_at"] or parse_iso(fresh["expires_at"]) <= utc_now(): db.rollback() return jsonify({"error": "Invite link has expired or was already used."}), 400 db.execute( """ INSERT OR IGNORE INTO room_members ( room_id, user_id, membership_role, added_at ) VALUES (?, ?, 'member', ?) """, (invite["room_id"], user["id"], now_iso()), ) db.execute( """ UPDATE room_invites SET used_at = ?, used_by_user_id = ? WHERE id = ? """, (now_iso(), user["id"], invite["id"]), ) db.commit() audit("room_invite_accepted", user["id"], "room", str(invite["room_id"])) room = get_room_by_slug(invite["slug"]) return jsonify({"status": "joined", "room": room_payload(room, user)}) @app.route("/api/admin/users") def admin_list_users(): moderator, error = require_moderator() if error: return error with get_db() as db: rows = db.execute( """ SELECT u.id, u.username, u.role, u.status, u.restricted_until, u.created_at, u.last_login_at, COUNT(m.id) AS message_count FROM users AS u LEFT JOIN messages AS m ON m.user_id = u.id GROUP BY u.id ORDER BY CASE u.role WHEN 'admin' THEN 0 WHEN 'moderator' THEN 1 ELSE 2 END, u.username COLLATE NOCASE """ ).fetchall() return jsonify( { "users": [ { "id": row["id"], "username": row["username"], "role": row["role"], "status": row["status"], "restricted_until": row["restricted_until"], "created_at": row["created_at"], "last_login_at": row["last_login_at"], "message_count": row["message_count"], } for row in rows ], "requested_by": moderator["username"], } ) @app.route("/api/admin/users/", methods=["PATCH"]) def admin_update_user(user_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 admin, error = require_admin() if error: return error limited = rate_limited("admin") if limited: return limited try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 with get_db() as db: target = db.execute( """ SELECT id, username, role, status, restricted_until FROM users WHERE id = ? """, (user_id,), ).fetchone() if not target: return jsonify({"error": "User not found."}), 404 if target["id"] == admin["id"]: return jsonify({"error": "You cannot modify your own account here."}), 400 role = data.get("role", target["role"]) status = data.get("status", target["status"]) restricted_until = data.get("restricted_until", target["restricted_until"]) if role not in ROLES: return jsonify({"error": "Invalid role."}), 400 if status not in STATUSES: return jsonify({"error": "Invalid account status."}), 400 if status == "restricted" and restricted_until: parsed = parse_iso(restricted_until) if not parsed or parsed <= utc_now(): return jsonify({"error": "restricted_until must be a future ISO timestamp."}), 400 elif status != "restricted": restricted_until = None if target["role"] == "admin" and role != "admin": with get_db() as db: count = db.execute( """ SELECT COUNT(*) AS count FROM users WHERE role = 'admin' AND status = 'active' """ ).fetchone()["count"] if count <= 1: return jsonify({"error": "At least one active administrator is required."}), 400 is_admin_value = 1 if role == "admin" else 0 is_active = 1 if status in {"active", "restricted"} else 0 db_write( """ UPDATE users SET role = ?, status = ?, restricted_until = ?, is_admin = ?, is_active = ? WHERE id = ? """, (role, status, restricted_until, is_admin_value, is_active, user_id), ) if status == "blacklisted": db_write( """ INSERT OR REPLACE INTO blacklist_entries ( kind, value_hash, reason, created_at, created_by_user_id ) VALUES ('user', ?, ?, ?, ?) """, ( hash_identifier(str(user_id)), "Account blacklisted by administrator", now_iso(), admin["id"], ), ) if target["status"] == "blacklisted" and status != "blacklisted": db_write( "DELETE FROM blacklist_entries WHERE kind = 'user' AND value_hash = ?", (hash_identifier(str(user_id)),), ) audit( "user_updated", admin["id"], "user", str(user_id), f"role={role};status={status};restricted_until={restricted_until}", ) return jsonify({"status": "updated"}) @app.route("/api/admin/rooms") def admin_rooms(): moderator, error = require_moderator() if error: return error with get_db() as db: rows = db.execute( """ SELECT id, slug, name, description, visibility, created_by_user_id, is_active FROM rooms WHERE is_active = 1 ORDER BY name COLLATE NOCASE """ ).fetchall() return jsonify( { "rooms": [ { "id": row["id"], "slug": row["slug"], "name": row["name"], "description": row["description"], "visibility": row["visibility"], } for row in rows ] } ) @app.route("/api/admin/blacklist/ip", methods=["POST"]) def blacklist_ip(): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 admin, error = require_admin() if error: return error try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 ip = str(data.get("ip", "")).strip() reason = str(data.get("reason", "")).strip()[:300] if not ip or len(ip) > 128: return jsonify({"error": "A valid IP value is required."}), 400 db_write( """ INSERT OR REPLACE INTO blacklist_entries ( kind, value_hash, reason, created_at, created_by_user_id ) VALUES ('ip', ?, ?, ?, ?) """, ( hash_identifier(ip), reason or "IP blacklisted by administrator", now_iso(), admin["id"], ), ) audit("ip_blacklisted", admin["id"], "ip", hash_identifier(ip), reason) return jsonify({"status": "blacklisted"}) @app.route("/api/teams") def api_list_teams(): user = current_user() with get_db() as db: teams = db.execute( """ SELECT id, slug, name, description, created_by_user_id, created_at, is_active FROM teams WHERE is_active = 1 ORDER BY name COLLATE NOCASE """ ).fetchall() channels = db.execute( """ SELECT c.id, c.team_id, c.legacy_room_id, c.slug, c.name, c.description, c.created_by_user_id, c.created_at, c.is_active, t.slug AS team_slug, t.name AS team_name, t.description AS team_description, t.is_active AS team_is_active FROM channels AS c JOIN teams AS t ON t.id = c.team_id WHERE c.is_active = 1 AND t.is_active = 1 """ ).fetchall() public_page_rows = db.execute( """ SELECT DISTINCT p.team_id FROM pages AS p JOIN teams AS t ON t.id = p.team_id WHERE p.is_active = 1 AND p.visibility = 'public' AND t.is_active = 1 """ ).fetchall() channels_by_team = {} for channel in channels: if can_access_channel(channel, user): channels_by_team[channel["team_id"]] = True public_page_team_ids = {row["team_id"] for row in public_page_rows} visible_teams = [ team_payload(team, user) for team in teams if can_access_team(team, user) or channels_by_team.get(team["id"]) or team["id"] in public_page_team_ids ] return jsonify({"teams": visible_teams, "authenticated": bool(user)}) @app.route("/api/teams", methods=["POST"]) def api_create_team(): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error limited = rate_limited("admin") if limited: return limited with get_db() as db: owned_count = db.execute( """ SELECT COUNT(*) AS count FROM team_members WHERE user_id = ? AND membership_role = 'owner' """, (user["id"],), ).fetchone()["count"] if owned_count >= MAX_OWNED_TEAMS_PER_USER: return jsonify({"error": "You can own at most 3 teams."}), 400 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 name = str(data.get("name", "")).strip() description = str(data.get("description", "")).strip()[:300] requested_slug = str(data.get("slug", "")).strip().lower() if not name or len(name) > 60: return jsonify({"error": "Team name must be 1 to 60 characters."}), 400 base_slug = requested_slug or team_slug_from_name(name) if not base_slug or not ROOM_SLUG_RE.fullmatch(base_slug): return jsonify({"error": "Team name cannot create a valid slug."}), 400 with get_db() as db: db.execute("BEGIN IMMEDIATE") slug = base_slug suffix = 2 while db.execute("SELECT 1 FROM teams WHERE slug = ?", (slug,)).fetchone(): slug = f"{base_slug[:44]}-{suffix}" suffix += 1 cursor = db.execute( """ INSERT INTO teams ( slug, name, description, created_by_user_id, created_at, is_active ) VALUES (?, ?, ?, ?, ?, 1) """, (slug, name, description, user["id"], now_iso()), ) team_id = cursor.lastrowid db.execute( """ INSERT INTO team_members ( team_id, user_id, membership_role, added_at ) VALUES (?, ?, 'owner', ?) """, (team_id, user["id"], now_iso()), ) db.commit() audit("team_created", user["id"], "team", str(team_id), slug) with get_db() as db: team = db.execute( """ SELECT id, slug, name, description, created_by_user_id, created_at, is_active FROM teams WHERE id = ? """, (team_id,), ).fetchone() return jsonify({"team": team_payload(team, user)}), 201 @app.route("/api/teams/", methods=["PATCH"]) def api_update_team(team_slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error limited = rate_limited("admin") if limited: return limited team = get_team_by_slug(team_slug) if not team: return jsonify({"error": "Team not found."}), 404 if not can_edit_team(team["id"], user): return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 name = str(data.get("name", team["name"])).strip() description = str(data.get("description", team["description"] or "")).strip()[:300] if not name or len(name) > 60: return jsonify({"error": "Team name must be 1 to 60 characters."}), 400 db_write( "UPDATE teams SET name = ?, description = ? WHERE id = ?", (name, description, team["id"]), ) updated = get_team_by_slug(team_slug) audit("team_updated", user["id"], "team", str(team["id"]), f"name={name}") return jsonify({"team": team_payload(updated, user)}) @app.route("/api/teams//channels") def api_list_team_channels(team_slug: str): team = get_team_by_slug(team_slug) if not team: return jsonify({"error": "Team not found."}), 404 user = current_user() with get_db() as db: rows = db.execute( """ SELECT c.id, c.team_id, c.legacy_room_id, c.slug, c.name, c.description, c.created_by_user_id, c.created_at, c.is_active, t.slug AS team_slug, t.name AS team_name, t.description AS team_description, t.is_active AS team_is_active FROM channels AS c JOIN teams AS t ON t.id = c.team_id WHERE c.team_id = ? AND c.is_active = 1 AND t.is_active = 1 ORDER BY CASE WHEN c.slug = 'general' THEN 0 ELSE 1 END, c.name COLLATE NOCASE """, (team["id"],), ).fetchall() channels = [channel_payload(channel, user) for channel in rows if can_access_channel(channel, user)] if not channels and not can_access_team(team, user): if not user: return jsonify({"error": "Login required.", "code": "login_required"}), 401 return jsonify({"error": "You do not have access to this team.", "code": "forbidden"}), 403 return jsonify({"team": team_payload(team, user), "channels": channels}) @app.route("/api/teams//channels", methods=["POST"]) def api_create_channel(team_slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error team = get_team_by_slug(team_slug) if not team: return jsonify({"error": "Team not found."}), 404 if not is_team_admin(team["id"], user): return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 name = str(data.get("name", "")).strip() description = str(data.get("description", "")).strip()[:300] requested_slug = str(data.get("slug", "")).strip().lower() if not name or len(name) > 60: return jsonify({"error": "Channel name must be 1 to 60 characters."}), 400 base_slug = requested_slug or team_slug_from_name(name) if not base_slug or not ROOM_SLUG_RE.fullmatch(base_slug): return jsonify({"error": "Channel name cannot create a valid slug."}), 400 with get_db() as db: db.execute("BEGIN IMMEDIATE") slug = base_slug suffix = 2 while db.execute("SELECT 1 FROM channels WHERE team_id = ? AND slug = ?", (team["id"], slug)).fetchone(): slug = f"{base_slug[:44]}-{suffix}" suffix += 1 cursor = db.execute( """ INSERT INTO channels ( team_id, legacy_room_id, slug, name, description, created_by_user_id, created_at, is_active ) VALUES (?, NULL, ?, ?, ?, ?, ?, 1) """, (team["id"], slug, name, description, user["id"], now_iso()), ) channel_id = cursor.lastrowid db.commit() audit("channel_created", user["id"], "channel", str(channel_id), f"{team_slug}/{slug}") channel = get_channel_by_id(channel_id) return jsonify({"channel": channel_payload(channel, user)}), 201 @app.route("/api/teams//channels/", methods=["PATCH"]) def api_update_channel(team_slug: str, channel_slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error limited = rate_limited("admin") if limited: return limited channel = get_channel_by_slug(team_slug, channel_slug) if not channel: return jsonify({"error": "Channel not found."}), 404 if not can_edit_team(channel["team_id"], user): return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 name = str(data.get("name", channel["name"])).strip() description = str(data.get("description", channel["description"] or "")).strip()[:300] if not name or len(name) > 60: return jsonify({"error": "Channel name must be 1 to 60 characters."}), 400 db_write( "UPDATE channels SET name = ?, description = ? WHERE id = ?", (name, description, channel["id"]), ) if channel["legacy_room_id"]: db_write( "UPDATE rooms SET name = ?, description = ? WHERE id = ?", (name, description, channel["legacy_room_id"]), ) updated = get_channel_by_id(channel["id"]) audit("channel_updated", user["id"], "channel", str(channel["id"]), f"{team_slug}/{channel_slug}") return jsonify({"channel": channel_payload(updated, user)}) @app.route("/api/teams//channels/", methods=["DELETE"]) def api_delete_channel(team_slug: str, channel_slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error limited = rate_limited("admin") if limited: return limited channel = get_channel_by_slug(team_slug, channel_slug) if not channel: return jsonify({"error": "Channel not found."}), 404 if not can_edit_team(channel["team_id"], user): return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403 if channel["legacy_room_id"]: return jsonify({"error": "Legacy channels linked to rooms cannot be deleted here."}), 400 db_write("UPDATE channels SET is_active = 0 WHERE id = ?", (channel["id"],)) audit("channel_deleted", user["id"], "channel", str(channel["id"]), f"{team_slug}/{channel_slug}") return jsonify({"status": "deleted"}) @app.route("/api/teams//members") def api_list_team_members(team_slug: str): user, error = require_user() if error: return error team = get_team_by_slug(team_slug) if not team: return jsonify({"error": "Team not found."}), 404 if not is_team_admin(team["id"], user): return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403 with get_db() as db: rows = db.execute( """ SELECT u.id, u.username, u.role, tm.membership_role, tm.added_at FROM team_members AS tm JOIN users AS u ON u.id = tm.user_id WHERE tm.team_id = ? ORDER BY CASE tm.membership_role WHEN 'owner' THEN 0 WHEN 'admin' THEN 1 ELSE 2 END, u.username COLLATE NOCASE """, (team["id"],), ).fetchall() return jsonify( { "team": team_payload(team, user), "members": [ { "id": row["id"], "username": row["username"], "role": row["role"], "membership_role": row["membership_role"], "added_at": row["added_at"], } for row in rows ], } ) @app.route("/api/teams//members", methods=["POST"]) def api_add_team_member(team_slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error team = get_team_by_slug(team_slug) if not team: return jsonify({"error": "Team not found."}), 404 if not is_team_admin(team["id"], user): return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 username = normalize_username(data.get("username")) membership_role = str(data.get("membership_role", "member")).strip().lower() if membership_role not in TEAM_MEMBER_ROLES: return jsonify({"error": "Invalid membership role."}), 400 with get_db() as db: target = db.execute( "SELECT id, username FROM users WHERE username = ?", (username,), ).fetchone() if not target: return jsonify({"error": "User not found."}), 404 if membership_role == "owner" and team_member_role(team["id"], user["id"]) != "owner": return jsonify({"error": "Only an owner can assign owner role."}), 403 db_write( """ INSERT OR IGNORE INTO team_members ( team_id, user_id, membership_role, added_at ) VALUES (?, ?, ?, ?) """, (team["id"], target["id"], membership_role, now_iso()), ) audit("team_member_added", user["id"], "team", str(team["id"]), f"{target['username']}:{membership_role}") return jsonify({"status": "added"}), 201 @app.route("/api/teams//members/", methods=["PATCH"]) def api_update_team_member(team_slug: str, member_user_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error team = get_team_by_slug(team_slug) if not team: return jsonify({"error": "Team not found."}), 404 if not is_team_admin(team["id"], user): return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 membership_role = str(data.get("membership_role", "")).strip().lower() if membership_role not in TEAM_MEMBER_ROLES: return jsonify({"error": "Invalid membership role."}), 400 actor_role = team_member_role(team["id"], user["id"]) target_role = team_member_role(team["id"], member_user_id) if not target_role: return jsonify({"error": "Team member not found."}), 404 if target_role == "owner" and actor_role != "owner": return jsonify({"error": "Only an owner can modify another owner."}), 403 if membership_role == "owner" and actor_role != "owner": return jsonify({"error": "Only an owner can assign owner role."}), 403 db_write( "UPDATE team_members SET membership_role = ? WHERE team_id = ? AND user_id = ?", (membership_role, team["id"], member_user_id), ) audit( "team_member_role_updated", user["id"], "team", str(team["id"]), f"user_id={member_user_id};role={membership_role}", ) return jsonify({"status": "updated"}) @app.route("/api/teams//members/", methods=["DELETE"]) def api_remove_team_member(team_slug: str, member_user_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error team = get_team_by_slug(team_slug) if not team: return jsonify({"error": "Team not found."}), 404 if not is_team_admin(team["id"], user): return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403 actor_role = team_member_role(team["id"], user["id"]) target_role = team_member_role(team["id"], member_user_id) if not target_role: return jsonify({"error": "Team member not found."}), 404 if target_role == "owner": return jsonify({"error": "The team owner cannot be removed."}), 400 if target_role == "admin" and actor_role not in {"owner", "admin"}: return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 db_write( "DELETE FROM team_members WHERE team_id = ? AND user_id = ?", (team["id"], member_user_id), ) audit("team_member_removed", user["id"], "team", str(team["id"]), str(member_user_id)) return jsonify({"status": "removed"}) @app.route("/api/teams//invites", methods=["POST"]) def api_create_team_invite(team_slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error team = get_team_by_slug(team_slug) if not team: return jsonify({"error": "Team not found."}), 404 if not is_team_admin(team["id"], user): return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 membership_role = str(data.get("membership_role", "member")).strip().lower() if membership_role not in {"member", "admin"}: return jsonify({"error": "Invite role must be member or admin."}), 400 token = ctx["secrets"].token_urlsafe(32) expires_at = (utc_now() + ctx["timedelta"](hours=TEAM_INVITE_TTL_HOURS)).isoformat() db_write( """ INSERT INTO team_invites ( team_id, token_hash, created_by_user_id, membership_role, created_at, expires_at ) VALUES (?, ?, ?, ?, ?, ?) """, (team["id"], hash_identifier(token), user["id"], membership_role, now_iso(), expires_at), ) audit("team_invite_created", user["id"], "team", str(team["id"]), membership_role) return jsonify( { "invite": { "url": f"/?team_invite={token}", "expires_at": expires_at, "membership_role": membership_role, } } ), 201 @app.route("/api/team-invites//accept", methods=["POST"]) def api_accept_team_invite(token: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error token_hash = hash_identifier(token) with get_db() as db: invite = db.execute( """ SELECT ti.id, ti.team_id, ti.membership_role, ti.expires_at, ti.used_at, t.slug, t.name, t.description, t.is_active FROM team_invites AS ti JOIN teams AS t ON t.id = ti.team_id WHERE ti.token_hash = ? """, (token_hash,), ).fetchone() if not invite: return jsonify({"error": "Invite link is invalid."}), 404 if invite["used_at"] or parse_iso(invite["expires_at"]) <= utc_now(): return jsonify({"error": "Invite link has expired or was already used."}), 400 if not invite["is_active"]: return jsonify({"error": "This team is no longer available."}), 404 with get_db() as db: db.execute("BEGIN IMMEDIATE") fresh = db.execute( "SELECT id, used_at, expires_at FROM team_invites WHERE id = ?", (invite["id"],), ).fetchone() if not fresh or fresh["used_at"] or parse_iso(fresh["expires_at"]) <= utc_now(): db.rollback() return jsonify({"error": "Invite link has expired or was already used."}), 400 db.execute( """ INSERT INTO team_members ( team_id, user_id, membership_role, added_at ) VALUES (?, ?, ?, ?) ON CONFLICT(team_id, user_id) DO UPDATE SET membership_role = excluded.membership_role """, (invite["team_id"], user["id"], invite["membership_role"], now_iso()), ) db.execute( """ UPDATE team_invites SET used_at = ?, used_by_user_id = ? WHERE id = ? """, (now_iso(), user["id"], invite["id"]), ) db.commit() audit("team_invite_accepted", user["id"], "team", str(invite["team_id"])) team = get_team_by_slug(invite["slug"]) return jsonify({"status": "joined", "team": team_payload(team, user)}) @app.route("/api/teams//pages") def api_list_pages(team_slug: str): team = get_team_by_slug(team_slug) if not team: return jsonify({"error": "Team not found."}), 404 user = current_user() page_type = request.args.get("type", "").strip().lower() visibility = request.args.get("visibility", "").strip().lower() if page_type and page_type not in PAGE_TYPES: return jsonify({"error": "Invalid page type."}), 400 if visibility and visibility not in PAGE_VISIBILITIES: return jsonify({"error": "Invalid page visibility."}), 400 is_member = bool(user and (user["role"] in {"moderator", "admin"} or is_team_member(team["id"], user["id"]))) conditions = ["p.team_id = ?", "p.is_active = 1", "t.is_active = 1"] params = [team["id"]] if page_type: conditions.append("p.page_type = ?") params.append(page_type) if visibility: conditions.append("p.visibility = ?") params.append(visibility) if not is_member: conditions.append("p.visibility = 'public'") query = f""" {team_page_select()} WHERE {" AND ".join(conditions)} ORDER BY p.updated_at DESC, p.id DESC """ with get_db() as db: rows = db.execute(query, tuple(params)).fetchall() return jsonify({"team": team_payload(team, user), "pages": [page_payload(page, user) for page in rows]}) @app.route("/api/teams//pages", methods=["POST"]) def api_create_page(): team_slug = request.view_args["team_slug"] if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error team = get_team_by_slug(team_slug) if not team: return jsonify({"error": "Team not found."}), 404 if not is_team_admin(team["id"], user): return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 title = str(data.get("title", "")).strip() body = str(data.get("body", "")).strip() visibility = str(data.get("visibility", "team")).strip().lower() page_type = str(data.get("page_type", "document")).strip().lower() requested_slug = str(data.get("slug", "")).strip().lower() channel_slug = str(data.get("channel_slug", "")).strip().lower() if not title or len(title) > 160: return jsonify({"error": "Page title must be 1 to 160 characters."}), 400 if len(body) > 100000: return jsonify({"error": "Page content is too long."}), 400 if visibility not in PAGE_VISIBILITIES: return jsonify({"error": "Invalid page visibility."}), 400 if page_type not in PAGE_TYPES: return jsonify({"error": "Invalid page type."}), 400 base_slug = requested_slug or page_slug_from_title(title) if not base_slug or not ROOM_SLUG_RE.fullmatch(base_slug): return jsonify({"error": "Page title cannot create a valid slug."}), 400 channel_id = None if channel_slug: channel = get_channel_by_slug(team_slug, channel_slug) if not channel: return jsonify({"error": "Channel not found."}), 404 channel_id = channel["id"] with get_db() as db: db.execute("BEGIN IMMEDIATE") slug = base_slug suffix = 2 while db.execute("SELECT 1 FROM pages WHERE team_id = ? AND slug = ?", (team["id"], slug)).fetchone(): slug = f"{base_slug[:44]}-{suffix}" suffix += 1 cursor = db.execute( """ INSERT INTO pages ( team_id, channel_id, slug, title, body, visibility, page_type, created_by_user_id, updated_by_user_id, created_at, updated_at, is_active ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1) """, ( team["id"], channel_id, slug, title, body, visibility, page_type, user["id"], user["id"], now_iso(), now_iso(), ), ) page_id = cursor.lastrowid db.commit() audit("page_created", user["id"], "page", str(page_id), f"{team_slug}/{slug}") with get_db() as db: page = db.execute(f"{team_page_select()} WHERE p.id = ?", (page_id,)).fetchone() return jsonify({"page": page_payload(page, user)}), 201 @app.route("/api/teams//pages/") def api_get_page(team_slug: str, page_slug: str): user = current_user() with get_db() as db: page = db.execute( f""" {team_page_select()} WHERE t.slug = ? AND p.slug = ? AND p.is_active = 1 AND t.is_active = 1 """, (team_slug, page_slug), ).fetchone() if not page: return jsonify({"error": "Page not found."}), 404 permitted = bool( page["visibility"] == "public" or (user and (user["role"] in {"moderator", "admin"} or is_team_member(page["team_id"], user["id"]))) ) if not permitted: if not user: return jsonify({"error": "Login required.", "code": "login_required"}), 401 return jsonify({"error": "You do not have access to this page.", "code": "forbidden"}), 403 return jsonify({"page": page_payload(page, user)}) @app.route("/api/teams//pages/", methods=["PATCH"]) def api_update_page(team_slug: str, page_slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error with get_db() as db: page = db.execute( f""" {team_page_select()} WHERE t.slug = ? AND p.slug = ? AND p.is_active = 1 AND t.is_active = 1 """, (team_slug, page_slug), ).fetchone() if not page: return jsonify({"error": "Page not found."}), 404 if not is_team_admin(page["team_id"], user): return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 title = str(data.get("title", page["title"])).strip() body = str(data.get("body", page["body"])).strip() visibility = str(data.get("visibility", page["visibility"])).strip().lower() page_type = str(data.get("page_type", page["page_type"])).strip().lower() channel_slug = str(data.get("channel_slug", "")).strip().lower() if not title or len(title) > 160: return jsonify({"error": "Page title must be 1 to 160 characters."}), 400 if len(body) > 100000: return jsonify({"error": "Page content is too long."}), 400 if visibility not in PAGE_VISIBILITIES: return jsonify({"error": "Invalid page visibility."}), 400 if page_type not in PAGE_TYPES: return jsonify({"error": "Invalid page type."}), 400 channel_id = page["channel_id"] if "channel_slug" in data: if channel_slug: channel = get_channel_by_slug(team_slug, channel_slug) if not channel: return jsonify({"error": "Channel not found."}), 404 channel_id = channel["id"] else: channel_id = None db_write( """ UPDATE pages SET title = ?, body = ?, visibility = ?, page_type = ?, channel_id = ?, updated_by_user_id = ?, updated_at = ? WHERE id = ? """, (title, body, visibility, page_type, channel_id, user["id"], now_iso(), page["id"]), ) audit("page_updated", user["id"], "page", str(page["id"]), f"{team_slug}/{page_slug}") with get_db() as db: updated = db.execute(f"{team_page_select()} WHERE p.id = ?", (page["id"],)).fetchone() return jsonify({"page": page_payload(updated, user)}) @app.route("/api/teams//pages/", methods=["DELETE"]) def api_delete_page(team_slug: str, page_slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error with get_db() as db: page = db.execute( """ SELECT p.id, p.team_id FROM pages AS p JOIN teams AS t ON t.id = p.team_id WHERE t.slug = ? AND p.slug = ? AND p.is_active = 1 AND t.is_active = 1 """, (team_slug, page_slug), ).fetchone() if not page: return jsonify({"error": "Page not found."}), 404 if not is_team_admin(page["team_id"], user): return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403 db_write( """ UPDATE pages SET is_active = 0, updated_by_user_id = ?, updated_at = ? WHERE id = ? """, (user["id"], now_iso(), page["id"]), ) audit("page_deleted", user["id"], "page", str(page["id"]), f"{team_slug}/{page_slug}") return jsonify({"status": "deleted"}) @app.route("/api/teams//pages//versions") def api_list_page_versions(team_slug: str, page_slug: str): user = current_user() with get_db() as db: page = db.execute( f""" {team_page_select()} WHERE t.slug = ? AND p.slug = ? AND p.is_active = 1 AND t.is_active = 1 """, (team_slug, page_slug), ).fetchone() if not page: return jsonify({"error": "Page not found."}), 404 permitted = bool( page["visibility"] == "public" or (user and (user["role"] in {"moderator", "admin"} or is_team_member(page["team_id"], user["id"]))) ) if not permitted: if not user: return jsonify({"error": "Login required.", "code": "login_required"}), 401 return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 return jsonify({"versions": []}) @app.route("/api/page-versions/") def api_get_page_version(version_id: int): return jsonify( { "version": { "id": version_id, "version_number": 1, "title": "version unavailable", "body_html": "

page version history is not enabled on this server.

", } } ) @app.route("/api/page-versions//restore", methods=["POST"]) def api_restore_page_version(version_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 return jsonify({"error": "Page version restore is not enabled on this server."}), 501 @app.route("/api/teams//pages//comments") def api_list_page_comments(team_slug: str, page_slug: str): user = current_user() with get_db() as db: page = db.execute( f""" {team_page_select()} WHERE t.slug = ? AND p.slug = ? AND p.is_active = 1 AND t.is_active = 1 """, (team_slug, page_slug), ).fetchone() if not page: return jsonify({"error": "Page not found."}), 404 permitted = bool( page["visibility"] == "public" or (user and (user["role"] in {"moderator", "admin"} or is_team_member(page["team_id"], user["id"]))) ) if not permitted: if not user: return jsonify({"error": "Login required.", "code": "login_required"}), 401 return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 return jsonify({"comments": []}) @app.route("/api/teams//pages//comments", methods=["POST"]) def api_create_page_comment(team_slug: str, page_slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error return jsonify({"error": "Page comments are not enabled on this server."}), 501 @app.route("/api/page-comments/", methods=["DELETE"]) def api_delete_page_comment(comment_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error return jsonify({"error": "Page comments are not enabled on this server."}), 501 @app.route("/api/teams//pages//share", methods=["POST"]) def api_create_page_share(team_slug: str, page_slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_user() if error: return error with get_db() as db: page = db.execute( """ SELECT p.id FROM pages AS p JOIN teams AS t ON t.id = p.team_id WHERE t.slug = ? AND p.slug = ? AND p.is_active = 1 AND t.is_active = 1 """, (team_slug, page_slug), ).fetchone() if not page: return jsonify({"error": "Page not found."}), 404 return jsonify( { "share": { "url": build_local_stub_share_url("page", page["id"]) } } ) @app.route("/api/profile") def api_get_my_profile(): user, error = require_user() if error: return error with get_db() as db: user_row = db.execute( """ SELECT id, username, role, status, restricted_until, created_at, last_login_at FROM users WHERE id = ? """, (user["id"],), ).fetchone() profile_row = db.execute( """ SELECT user_id, display_name, badge_text, bio, bio_mode, website, location, status_message, banner_attachment_id, updated_at FROM user_profiles WHERE user_id = ? """, (user["id"],), ).fetchone() settings_row = db.execute( """ SELECT name_color, presence_mode, profile_background_color, profile_banner_color, profile_visibility, last_active_visibility, dm_layout_mode, show_presence_badge FROM user_settings WHERE user_id = ? """, (user["id"],), ).fetchone() return jsonify({"profile": profile_payload(html_module, user_payload, user_row, profile_row, settings_row, user)}) @app.route("/api/profile", methods=["PATCH"]) def api_update_my_profile(): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 limited = rate_limited("profile") if limited: return limited user, error = require_writable_user() if error: return error try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 display_name = str(data.get("display_name", "")).strip() badge_text = str(data.get("badge_text", "")).strip() bio = str(data.get("bio", "")).strip() bio_mode = str(data.get("bio_mode", "txt")).strip().lower() website = str(data.get("website", "")).strip() location = str(data.get("location", "")).strip() status_message = str(data.get("status_message", "")).strip() banner_attachment_id = data.get("banner_attachment_id") if banner_attachment_id not in (None, ""): try: banner_attachment_id = int(banner_attachment_id) except (TypeError, ValueError): return jsonify({"error": "Invalid banner attachment."}), 400 else: banner_attachment_id = None if len(display_name) > MAX_DISPLAY_NAME_LENGTH: return jsonify({"error": "Display name is too long."}), 400 if len(badge_text) > MAX_PROFILE_BADGE_LENGTH: return jsonify({"error": "Badge text is too long."}), 400 if len(bio) > MAX_PROFILE_BIO_LENGTH: return jsonify({"error": "Bio is too long."}), 400 if bio_mode not in PROFILE_BIO_MODES: return jsonify({"error": "Invalid bio mode."}), 400 if len(website) > MAX_WEBSITE_LENGTH: return jsonify({"error": "Website is too long."}), 400 if len(location) > MAX_LOCATION_LENGTH: return jsonify({"error": "Location is too long."}), 400 if len(status_message) > 160: return jsonify({"error": "Status message is too long."}), 400 if banner_attachment_id is not None: with get_db() as db: attachment = db.execute( """ SELECT id, uploader_user_id FROM attachments WHERE id = ? """, (banner_attachment_id,), ).fetchone() if not attachment: return jsonify({"error": "Banner attachment not found."}), 404 if attachment["uploader_user_id"] != user["id"]: return jsonify({"error": "You can only use your own uploaded image."}), 403 db_write( """ UPDATE user_profiles SET display_name = ?, badge_text = ?, bio = ?, bio_mode = ?, website = ?, location = ?, status_message = ?, banner_attachment_id = ?, updated_at = ? WHERE user_id = ? """, ( display_name or user["username"], badge_text, bio, bio_mode, website, location, status_message, banner_attachment_id, now_iso(), user["id"], ), ) with get_db() as db: user_row = db.execute( """ SELECT id, username, role, status, restricted_until, created_at, last_login_at FROM users WHERE id = ? """, (user["id"],), ).fetchone() profile_row = db.execute( """ SELECT user_id, display_name, badge_text, bio, bio_mode, website, location, status_message, banner_attachment_id, updated_at FROM user_profiles WHERE user_id = ? """, (user["id"],), ).fetchone() settings_row = db.execute( """ SELECT name_color, presence_mode, profile_background_color, profile_banner_color, profile_visibility, last_active_visibility, dm_layout_mode, show_presence_badge FROM user_settings WHERE user_id = ? """, (user["id"],), ).fetchone() audit("profile_updated", user["id"], "user", str(user["id"])) return jsonify({"profile": profile_payload(html_module, user_payload, user_row, profile_row, settings_row, user)}) @app.route("/api/profiles/") def api_get_public_profile(username: str): normalized = normalize_username(username) with get_db() as db: user_row = db.execute( """ SELECT id, username, role, status, restricted_until, created_at, last_login_at FROM users WHERE username = ? """, (normalized,), ).fetchone() if not user_row: return jsonify({"error": "Profile not found."}), 404 profile_row = db.execute( """ SELECT user_id, display_name, badge_text, bio, bio_mode, website, location, status_message, banner_attachment_id, updated_at FROM user_profiles WHERE user_id = ? """, (user_row["id"],), ).fetchone() settings_row = db.execute( """ SELECT name_color, presence_mode, profile_background_color, profile_banner_color, profile_visibility, last_active_visibility, dm_layout_mode, show_presence_badge FROM user_settings WHERE user_id = ? """, (user_row["id"],), ).fetchone() return jsonify({"profile": profile_payload(html_module, user_payload, user_row, profile_row, settings_row, current_user())}) @app.route("/api/settings") def api_get_settings(): user, error = require_user() if error: return error with get_db() as db: row = db.execute( """ SELECT accent_color, name_color, custom_css, theme_mode, profile_background_color, profile_banner_color, profile_visibility, last_active_visibility, presence_mode, dm_layout_mode, show_presence_badge, updated_at FROM user_settings WHERE user_id = ? """, (user["id"],), ).fetchone() return jsonify( { "settings": { "accent_color": row["accent_color"], "name_color": row["name_color"], "custom_css": row["custom_css"], "theme_mode": row["theme_mode"], "profile_background_color": row["profile_background_color"], "profile_banner_color": row["profile_banner_color"], "profile_visibility": row["profile_visibility"], "last_active_visibility": row["last_active_visibility"], "presence_mode": row["presence_mode"], "dm_layout_mode": row["dm_layout_mode"], "show_presence_badge": bool(row["show_presence_badge"]), "updated_at": row["updated_at"], } } ) @app.route("/api/settings", methods=["PATCH"]) def api_update_settings(): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 limited = rate_limited("settings") if limited: return limited user, error = require_writable_user() if error: return error try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 accent_color = data.get("accent_color", "#f7dce8") name_color = data.get("name_color") custom_css = str(data.get("custom_css", "")) theme_mode = str(data.get("theme_mode", "dark")).strip().lower() profile_background_color = data.get("profile_background_color") profile_banner_color = data.get("profile_banner_color") profile_visibility = str(data.get("profile_visibility", "public")).strip().lower() last_active_visibility = str(data.get("last_active_visibility", "everyone")).strip().lower() presence_mode = str(data.get("presence_mode", "online")).strip().lower() dm_layout_mode = str(data.get("dm_layout_mode", "cards")).strip().lower() show_presence_badge_raw = data.get("show_presence_badge", True) if not valid_hex_color(accent_color): return jsonify({"error": "Invalid accent color."}), 400 if name_color not in (None, "") and not valid_hex_color(name_color): return jsonify({"error": "Invalid name color."}), 400 if profile_background_color not in (None, "") and not valid_hex_color(profile_background_color): return jsonify({"error": "Invalid profile background color."}), 400 if profile_banner_color not in (None, "") and not valid_hex_color(profile_banner_color): return jsonify({"error": "Invalid profile banner color."}), 400 if len(custom_css) > MAX_CUSTOM_CSS_LENGTH: return jsonify({"error": "Custom CSS is too long."}), 400 if theme_mode not in THEME_MODES: return jsonify({"error": "Invalid theme mode."}), 400 if profile_visibility not in PROFILE_VISIBILITIES: return jsonify({"error": "Invalid profile visibility."}), 400 if last_active_visibility not in LAST_ACTIVE_VISIBILITIES: return jsonify({"error": "Invalid last active visibility."}), 400 if presence_mode not in PRESENCE_MODES: return jsonify({"error": "Invalid presence mode."}), 400 if dm_layout_mode not in DM_LAYOUT_MODES: return jsonify({"error": "Invalid dm layout mode."}), 400 show_presence_badge = 1 if bool(show_presence_badge_raw) else 0 db_write( """ UPDATE user_settings SET accent_color = ?, name_color = ?, custom_css = ?, theme_mode = ?, profile_background_color = ?, profile_banner_color = ?, profile_visibility = ?, last_active_visibility = ?, presence_mode = ?, dm_layout_mode = ?, show_presence_badge = ?, updated_at = ? WHERE user_id = ? """, ( str(accent_color).lower(), normalized_hex_color(name_color) if name_color else None, custom_css, theme_mode, normalized_hex_color(profile_background_color) if profile_background_color else None, normalized_hex_color(profile_banner_color) if profile_banner_color else None, profile_visibility, last_active_visibility, presence_mode, dm_layout_mode, show_presence_badge, now_iso(), user["id"], ), ) with get_db() as db: row = db.execute( """ SELECT accent_color, name_color, custom_css, theme_mode, profile_background_color, profile_banner_color, profile_visibility, last_active_visibility, presence_mode, dm_layout_mode, show_presence_badge, updated_at FROM user_settings WHERE user_id = ? """, (user["id"],), ).fetchone() audit("settings_updated", user["id"], "user", str(user["id"])) return jsonify( { "settings": { **dict(row), "show_presence_badge": bool(row["show_presence_badge"]), } } ) @app.route("/api/scheduled-messages", methods=["POST"]) def api_create_scheduled_message(): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 text = str(data.get("text", "")).strip() deliver_at = str(data.get("deliver_at", "")).strip() scope_kind = str(data.get("scope_kind", "")).strip() scope_id = data.get("scope_id") if not text: return jsonify({"error": "Scheduled message text is required."}), 400 if len(text) > MAX_MESSAGE_LENGTH: return jsonify({"error": "Scheduled message is too long."}), 400 if not deliver_at: return jsonify({"error": "deliver_at is required."}), 400 return jsonify( { "status": "stubbed", "scheduled_message": { "id": 0, "text": text, "deliver_at": deliver_at, "scope_kind": scope_kind, "scope_id": scope_id, } } ), 201 @app.route("/api/dms") def api_list_dms(): user, error = require_user() if error: return error return jsonify({"dms": list_dms_for_user(get_db, user["id"])}) @app.route("/api/dms", methods=["POST"]) def api_create_or_open_dm(): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 other_user_id = data.get("user_id") username = normalize_username(data.get("username")) with get_db() as db: target = None if other_user_id is not None: try: parsed_id = int(other_user_id) except (TypeError, ValueError): return jsonify({"error": "Invalid user_id."}), 400 target = db.execute( """ SELECT id, username, role, status FROM users WHERE id = ? """, (parsed_id,), ).fetchone() elif username: target = db.execute( """ SELECT id, username, role, status FROM users WHERE username = ? """, (username,), ).fetchone() if not target: return jsonify({"error": "User not found."}), 404 if int(target["id"]) == int(user["id"]): return jsonify({"error": "You cannot DM yourself."}), 400 if target["status"] in {"disabled", "blacklisted"}: return jsonify({"error": "User not available."}), 400 dm_id = find_or_create_dm(get_db, db_write, now_iso, user["id"], target["id"]) audit("dm_opened", user["id"], "dm", str(dm_id), str(target["id"])) return jsonify({"dm": dm_payload(get_db, dm_id, user["id"])}), 201 @app.route("/api/dms//messages") def api_get_dm_messages(dm_id: int): user, error = require_user() if error: return error if not validate_dm_access(get_db, dm_id, user): return jsonify({"error": "DM not found."}), 404 after_raw = request.args.get("after_id", "").strip() after_id = 0 if after_raw: try: after_id = max(0, int(after_raw)) except ValueError: return jsonify({"error": "after_id must be a non-negative integer."}), 400 with get_db() as db: if after_raw: rows = db.execute( f""" {MESSAGE_SELECT} WHERE m.direct_conversation_id = ? AND m.parent_message_id IS NULL AND m.id > ? ORDER BY m.id ASC LIMIT ? """, (dm_id, after_id, MAX_INCREMENTAL_MESSAGES), ).fetchall() else: rows = db.execute( f""" SELECT * FROM ( {MESSAGE_SELECT} WHERE m.direct_conversation_id = ? AND m.parent_message_id IS NULL ORDER BY m.id DESC LIMIT ? ) ORDER BY id ASC """, (dm_id, MAX_HISTORY_MESSAGES), ).fetchall() messages = [build_message_payload(row, user) for row in rows] latest_id = messages[-1]["id"] if messages else after_id return jsonify( { "dm": dm_payload(get_db, dm_id, user["id"]), "messages": messages, "latest_id": latest_id, "has_more": len(messages) == MAX_INCREMENTAL_MESSAGES, } ) @app.route("/api/dms//messages", methods=["POST"]) def api_send_dm_message(dm_id: int): limited = rate_limited("send") if limited: return limited if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error if not validate_dm_access(get_db, dm_id, user): return jsonify({"error": "DM not found."}), 404 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 text = data.get("text", "") if not isinstance(text, str): return jsonify({"error": "Message must be text."}), 400 text = text.strip() if len(text) > MAX_MESSAGE_LENGTH: return jsonify({"error": "Message is too long."}), 400 attachment_ids, attachment_error = resolve_owned_attachment_ids( get_db, data.get("attachment_ids", []), user["id"], ) if attachment_error: return attachment_error if not attachment_ids and data.get("attachment_id") is not None: single_attachment_id, single_attachment_error = resolve_owned_attachment( get_db, data.get("attachment_id"), user["id"], ) if single_attachment_error: return single_attachment_error attachment_ids = [single_attachment_id] if single_attachment_id else [] attachment_id = attachment_ids[0] if attachment_ids else None if not text and not attachment_id: return jsonify({"error": "Message cannot be empty."}), 400 with get_db() as db: db.execute("BEGIN IMMEDIATE") cursor = db.execute( """ INSERT INTO messages ( room_id, channel_id, direct_conversation_id, user_id, username, text, attachment_id, parent_message_id, created_at ) VALUES (NULL, NULL, ?, ?, ?, ?, ?, NULL, ?) """, (dm_id, user["id"], user["username"], text, attachment_id, now_iso()), ) message_id = cursor.lastrowid for position, extra_attachment_id in enumerate(attachment_ids): db.execute( """ INSERT OR IGNORE INTO message_attachments ( message_id, attachment_id, position ) VALUES (?, ?, ?) """, (message_id, extra_attachment_id, position), ) db.commit() with get_db() as db: row = db.execute(f"{MESSAGE_SELECT} WHERE m.id = ?", (message_id,)).fetchone() audit("dm_message_sent", user["id"], "dm", str(dm_id), str(message_id)) return jsonify({"message": build_message_payload(row, user)}), 201 @app.route("/api/group-calls/token", methods=["POST"]) def api_group_call_token(): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error if not CALL_HELPER_SHARED_SECRET: return jsonify({"error": "Call helper is not configured."}), 503 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 scope_kind = str(data.get("scope_kind", "")).strip().lower() helper_call_id = None scope_id = None scope_slug = None room_value = None if scope_kind == "room": room_slug = str(data.get("room_slug", "")).strip() room = get_room_by_slug(room_slug) if not room: return jsonify({"error": "Room not found."}), 404 if not can_access_room(room, user): return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 helper_call_id = build_room_helper_call_id(room) scope_id = room["id"] scope_slug = room["slug"] room_value = room["slug"] elif scope_kind == "channel": team_slug = str(data.get("team_slug", "")).strip() channel_slug = str(data.get("channel_slug", "")).strip() channel = get_channel_by_slug(team_slug, channel_slug) if not channel: return jsonify({"error": "Channel not found."}), 404 if not can_access_channel(channel, user): return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 helper_call_id = build_channel_helper_call_id(channel) scope_id = channel["id"] scope_slug = f"{channel['team_slug']}/{channel['slug']}" room_value = scope_slug else: return jsonify({"error": "scope_kind must be room or channel."}), 400 session_row = get_or_create_group_call_session( scope_kind=scope_kind, scope_id=scope_id, scope_slug=scope_slug, helper_call_id=helper_call_id, created_by_user_id=user["id"], ) upsert_group_call_participant(session_row["id"], user["id"]) payload = { "sub": f"user-{user['id']}", "username": user["username"], "room": room_value, "call_id": helper_call_id, "role": user["role"], "can_call": True, "exp": int(time.time()) + CALL_TOKEN_TTL_SECONDS, } token = sign_call_helper_token(payload) audit( "group_call_token_issued", user["id"], "group_call_session", str(session_row["id"]), helper_call_id, ) return jsonify( { "token": token, "calls_base": CALLS_BASE, "call_id": helper_call_id, "session": group_call_session_payload(session_row), "participants": list_group_call_participants(session_row["id"]), "permissions": group_call_permissions_payload(session_row, user), "scope": group_call_scope_payload(session_row, user), } ) @app.route("/api/group-calls/invite", methods=["POST"]) def api_group_call_invite(): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error if not CALL_HELPER_SHARED_SECRET: return jsonify({"error": "Call helper is not configured."}), 503 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 session_id_raw = data.get("group_call_session_id") invite_username = normalize_username(data.get("username")) try: session_id = int(session_id_raw) except (TypeError, ValueError): return jsonify({"error": "Invalid group_call_session_id."}), 400 if not invite_username: return jsonify({"error": "username is required."}), 400 session_row = get_group_call_session_by_id(session_id) if not session_row or session_row["status"] != "active": return jsonify({"error": "Active group call not found."}), 404 if not can_access_group_call_session(session_row, user): return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 is_active_participant = False with get_db() as db: is_active_participant = bool( db.execute( """ SELECT 1 FROM group_call_participants WHERE group_call_session_id = ? AND user_id = ? AND left_at IS NULL """, (session_row["id"], user["id"]), ).fetchone() ) if not (is_active_participant or can_manage_group_call_session(session_row, user)): return jsonify( { "error": "Only active participants or scope managers can invite users.", "code": "forbidden", } ), 403 with get_db() as db: target = db.execute( """ SELECT id, username, role, status FROM users WHERE username = ? """, (invite_username,), ).fetchone() if not target: return jsonify({"error": "User not found."}), 404 if int(target["id"]) == int(user["id"]): return jsonify({"error": "You are already in the call."}), 400 if target["status"] in {"disabled", "blacklisted"}: return jsonify({"error": "User is not available."}), 400 payload = { "sub": f"user-{target['id']}", "username": target["username"], "room": session_row["scope_slug"], "call_id": session_row["helper_call_id"], "role": target["role"], "can_call": True, "exp": int(time.time()) + CALL_TOKEN_TTL_SECONDS, } helper_token = sign_call_helper_token(payload) expires_at = (utc_now() + ctx["timedelta"](minutes=10)).isoformat() cursor = db_write( """ INSERT INTO group_call_invites ( group_call_session_id, inviter_user_id, invitee_user_id, helper_token, status, created_at, expires_at, responded_at ) VALUES (?, ?, ?, ?, 'pending', ?, ?, NULL) """, ( session_row["id"], user["id"], target["id"], helper_token, now_iso(), expires_at, ), ) audit( "group_call_invite_created", user["id"], "group_call_session", str(session_row["id"]), f"invitee={target['username']}", ) return jsonify( { "invite": { "id": cursor.lastrowid, "invitee_username": target["username"], "group_call_session_id": session_row["id"], "status": "pending", "expires_at": expires_at, } } ), 201 @app.route("/api/group-calls/invites") def api_list_group_call_invites(): user, error = require_user() if error: return error with get_db() as db: rows = db.execute( """ SELECT i.id, i.group_call_session_id, i.status, i.created_at, i.expires_at, i.responded_at, s.scope_kind, s.scope_slug, s.helper_call_id, s.status AS session_status, inviter.username AS inviter_username FROM group_call_invites AS i JOIN group_call_sessions AS s ON s.id = i.group_call_session_id JOIN users AS inviter ON inviter.id = i.inviter_user_id WHERE i.invitee_user_id = ? AND i.status = 'pending' AND i.expires_at > ? AND s.status = 'active' ORDER BY i.id DESC """, (user["id"], now_iso()), ).fetchall() return jsonify( { "invites": [ { "id": row["id"], "group_call_session_id": row["group_call_session_id"], "scope_kind": row["scope_kind"], "scope_slug": row["scope_slug"], "helper_call_id": row["helper_call_id"], "inviter_username": row["inviter_username"], "status": row["status"], "created_at": row["created_at"], "expires_at": row["expires_at"], } for row in rows ] } ) @app.route("/api/group-calls/invites//accept", methods=["POST"]) def api_accept_group_call_invite(invite_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error with get_db() as db: row = db.execute( """ SELECT i.id, i.group_call_session_id, i.invitee_user_id, i.helper_token, i.status, i.expires_at, s.helper_call_id FROM group_call_invites AS i JOIN group_call_sessions AS s ON s.id = i.group_call_session_id WHERE i.id = ? AND i.invitee_user_id = ? LIMIT 1 """, (invite_id, user["id"]), ).fetchone() if not row: return jsonify({"error": "Invite not found."}), 404 if row["status"] != "pending" or parse_iso(row["expires_at"]) <= utc_now(): return jsonify({"error": "Invite expired or unavailable."}), 400 db_write( """ UPDATE group_call_invites SET status = 'accepted', responded_at = ? WHERE id = ? """, (now_iso(), invite_id), ) upsert_group_call_participant(row["group_call_session_id"], user["id"]) audit( "group_call_invite_accepted", user["id"], "group_call_invite", str(invite_id), row["helper_call_id"], ) session_row = get_group_call_session_by_id(row["group_call_session_id"]) return jsonify( { "token": row["helper_token"], "calls_base": CALLS_BASE, "call_id": row["helper_call_id"], "session": group_call_session_payload(session_row), "participants": list_group_call_participants(row["group_call_session_id"]), "permissions": group_call_permissions_payload(session_row, user), "scope": group_call_scope_payload(session_row, user), } ) @app.route("/api/group-calls/invites//dismiss", methods=["POST"]) def api_dismiss_group_call_invite(invite_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_user() if error: return error with get_db() as db: row = db.execute( """ SELECT id, group_call_session_id FROM group_call_invites WHERE id = ? AND invitee_user_id = ? LIMIT 1 """, (invite_id, user["id"]), ).fetchone() if not row: return jsonify({"error": "Invite not found."}), 404 db_write( """ UPDATE group_call_invites SET status = 'dismissed', responded_at = ? WHERE id = ? AND invitee_user_id = ? AND status = 'pending' """, (now_iso(), invite_id, user["id"]), ) audit( "group_call_invite_dismissed", user["id"], "group_call_invite", str(invite_id), ) return jsonify( { "status": "dismissed", "group_call_session_id": row["group_call_session_id"], } ) @app.route("/api/group-calls/") def api_get_group_call_session(session_id: int): user, error = require_user() if error: return error session_row = get_group_call_session_by_id(session_id) if not session_row: return jsonify({"error": "Group call session not found."}), 404 if not can_access_group_call_session(session_row, user): return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 return jsonify( { "session": group_call_session_payload(session_row), "scope": group_call_scope_payload(session_row, user), "participants": list_group_call_participants(session_id), "permissions": group_call_permissions_payload(session_row, user), } ) @app.route("/api/group-calls//participants") def api_get_group_call_participants(session_id: int): user, error = require_user() if error: return error session_row = get_group_call_session_by_id(session_id) if not session_row: return jsonify({"error": "Group call session not found."}), 404 if not can_access_group_call_session(session_row, user): return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 return jsonify( { "session_id": session_id, "participants": list_group_call_participants(session_id), "permissions": group_call_permissions_payload(session_row, user), "scope": group_call_scope_payload(session_row, user), } ) @app.route("/api/group-calls//join", methods=["POST"]) def api_join_group_call_session(session_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error session_row = get_group_call_session_by_id(session_id) if not session_row: return jsonify({"error": "Group call session not found."}), 404 if session_row["status"] != "active": return jsonify({"error": "Group call session is not active."}), 400 if not can_access_group_call_session(session_row, user): return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 upsert_group_call_participant(session_id, user["id"]) audit( "group_call_session_joined", user["id"], "group_call_session", str(session_id), session_row["helper_call_id"], ) return jsonify( { "status": "joined", "session": group_call_session_payload(session_row), "participants": list_group_call_participants(session_id), "permissions": group_call_permissions_payload(session_row, user), "scope": group_call_scope_payload(session_row, user), } ) @app.route("/api/group-calls//heartbeat", methods=["POST"]) def api_heartbeat_group_call_session(session_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_user() if error: return error session_row = get_group_call_session_by_id(session_id) if not session_row: return jsonify({"error": "Group call session not found."}), 404 if session_row["status"] != "active": return jsonify({"error": "Group call session is not active."}), 400 if not can_access_group_call_session(session_row, user): return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 heartbeat_group_call_participant(session_id, user["id"]) return jsonify( { "status": "ok", "session": group_call_session_payload(session_row), "participants": list_group_call_participants(session_id), } ) @app.route("/api/group-calls//leave", methods=["POST"]) def api_leave_group_call_session(session_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_user() if error: return error session_row = get_group_call_session_by_id(session_id) if not session_row: return jsonify({"error": "Group call session not found."}), 404 if not can_access_group_call_session(session_row, user): return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 mark_group_call_participant_left(session_id, user["id"]) maybe_end_group_call_if_empty(session_id) refreshed = get_group_call_session_by_id(session_id) audit( "group_call_session_left", user["id"], "group_call_session", str(session_id), session_row["helper_call_id"], ) return jsonify( { "status": "left", "session": group_call_session_payload(refreshed) if refreshed else None, "participants": list_group_call_participants(session_id), } ) @app.route("/api/group-calls//end", methods=["POST"]) def api_end_group_call_session(session_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error session_row = get_group_call_session_by_id(session_id) if not session_row: return jsonify({"error": "Group call session not found."}), 404 if not can_manage_group_call_session(session_row, user): return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 db_write( """ UPDATE group_call_sessions SET status = 'ended', ended_at = ? WHERE id = ? """, (now_iso(), session_id), ) db_write( """ UPDATE group_call_participants SET updated_at = ?, left_at = COALESCE(left_at, ?) WHERE group_call_session_id = ? AND left_at IS NULL """, (now_iso(), now_iso(), session_id), ) db_write( """ UPDATE group_call_invites SET status = 'dismissed', responded_at = COALESCE(responded_at, ?) WHERE group_call_session_id = ? AND status = 'pending' """, (now_iso(), session_id), ) refreshed = get_group_call_session_by_id(session_id) audit( "group_call_session_ended", user["id"], "group_call_session", str(session_id), session_row["helper_call_id"], ) return jsonify( { "status": "ended", "session": group_call_session_payload(refreshed) if refreshed else None, "participants": [], } ) @app.route("/api/call-token", methods=["POST"]) def api_call_token(): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error if not CALL_HELPER_SHARED_SECRET: return jsonify({"error": "Call helper is not configured."}), 503 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 room_slug = str(data.get("room", "")).strip() dm_id_raw = data.get("dm_id") call_id = None room_value = None if dm_id_raw is not None: try: dm_id = int(dm_id_raw) except (TypeError, ValueError): return jsonify({"error": "Invalid dm_id."}), 400 if not validate_dm_access(get_db, dm_id, user): return jsonify({"error": "DM not found."}), 404 peer = get_dm_other_user(dm_id, user["id"]) if not peer: return jsonify({"error": "Direct-message peer not found."}), 404 call_id = f"dm-{dm_id}" room_value = call_id elif room_slug: room = get_room_by_slug(room_slug) if not room: return jsonify({"error": "Room not found."}), 404 if not can_access_room(room, user): return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 call_id = f"room-{room['slug']}" room_value = room["slug"] else: return jsonify({"error": "room or dm_id is required."}), 400 payload = { "sub": f"user-{user['id']}", "username": user["username"], "room": room_value, "call_id": call_id, "role": user["role"], "can_call": True, "exp": int(time.time()) + CALL_TOKEN_TTL_SECONDS, } token = sign_call_helper_token(payload) audit( "call_helper_token_issued", user["id"], "call", call_id, f"room={room_value}", ) return jsonify( { "token": token, "calls_base": CALLS_BASE, "call_id": call_id, } ) @app.route("/api/dms//calls", methods=["POST"]) def api_start_dm_call(dm_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error if not validate_dm_access(get_db, dm_id, user): return jsonify({"error": "DM not found."}), 404 call_id = create_call_session(db_write, now_iso, dm_id, user["id"]) audit("call_started", user["id"], "call", str(call_id), str(dm_id)) return jsonify({"call": call_payload(get_db, call_id)}), 201 @app.route("/api/calls/") def api_get_call(call_id: int): user, error = require_user() if error: return error call = get_call_with_dm(get_db, call_id) if not call: return jsonify({"error": "Call not found."}), 404 if not validate_dm_access(get_db, call["direct_conversation_id"], user): return jsonify({"error": "Call not found."}), 404 return jsonify({"call": call_payload(get_db, call_id)}) @app.route("/api/calls//signal", methods=["POST"]) def api_post_call_signal(call_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error call = get_call_with_dm(get_db, call_id) if not call: return jsonify({"error": "Call not found."}), 404 if not validate_dm_access(get_db, call["direct_conversation_id"], user): return jsonify({"error": "Call not found."}), 404 if call["status"] != "active": return jsonify({"error": "Call is not active."}), 400 try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 signal_type = str(data.get("signal_type", "")).strip()[:40] payload = data.get("payload", {}) if not signal_type: return jsonify({"error": "signal_type is required."}), 400 if not isinstance(payload, dict): return jsonify({"error": "payload must be an object."}), 400 signal_id = append_call_signal(db_write, now_iso, call_id, user["id"], signal_type, payload) return jsonify({"status": "ok", "signal_id": signal_id}), 201 @app.route("/api/calls//signals") def api_get_call_signals(call_id: int): user, error = require_user() if error: return error call = get_call_with_dm(get_db, call_id) if not call: return jsonify({"error": "Call not found."}), 404 if not validate_dm_access(get_db, call["direct_conversation_id"], user): return jsonify({"error": "Call not found."}), 404 after_raw = str(request.args.get("after_id", "")).strip() after_id = 0 if after_raw: try: after_id = max(0, int(after_raw)) except ValueError: return jsonify({"error": "after_id must be a non-negative integer."}), 400 signals = list_call_signals(get_db, call_id, after_id) latest_id = signals[-1]["id"] if signals else after_id return jsonify({"signals": signals, "latest_id": latest_id}) @app.route("/api/calls//end", methods=["POST"]) def api_end_call(call_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error call = get_call_with_dm(get_db, call_id) if not call: return jsonify({"error": "Call not found."}), 404 if not validate_dm_access(get_db, call["direct_conversation_id"], user): return jsonify({"error": "Call not found."}), 404 end_call(db_write, now_iso, call_id) audit("call_ended", user["id"], "call", str(call_id)) return jsonify({"status": "ended", "call": call_payload(get_db, call_id)})