# routes.py import json import re import sqlite3 import time from flask import jsonify, request from features import ( PAGE_TYPES, PAGE_VISIBILITIES, PROFILE_BIO_MODES, REACTION_EMOJI_RE, TEAM_MEMBER_ROLES, THEME_MODES, append_call_signal, attachment_payload_from_row, call_payload, channel_presence_count, cleanup_orphaned_attachments, create_call_session, 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"] 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"] 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_OWNED_TEAMS_PER_USER = constants["MAX_OWNED_TEAMS_PER_USER"] ROOM_VISIBILITIES = constants["ROOM_VISIBILITIES"] ROLES = constants["ROLES"] STATUSES = constants["STATUSES"] html_module = ctx["html"] def build_message_payload(row, user): return richer_message_payload( get_db, row, user, user["id"] if user else None, ) @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/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/profile") def api_get_my_profile(): user, error = require_user() if error: return error with get_db() as db: user_row = db.execute( """ SELECT id, username, role, status, restricted_until, created_at FROM users WHERE id = ? """, (user["id"],), ).fetchone() profile_row = db.execute( """ SELECT user_id, display_name, bio, bio_mode, website, location, updated_at FROM user_profiles WHERE user_id = ? """, (user["id"],), ).fetchone() settings_row = db.execute( """ SELECT name_color FROM user_settings WHERE user_id = ? """, (user["id"],), ).fetchone() return jsonify({"profile": profile_payload(html_module, user_payload, user_row, profile_row, settings_row, user)}) @app.route("/api/profile", methods=["PATCH"]) def api_update_my_profile(): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 limited = rate_limited("profile") if limited: return limited user, error = require_writable_user() if error: return error try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 display_name = str(data.get("display_name", "")).strip() bio = str(data.get("bio", "")).strip() bio_mode = str(data.get("bio_mode", "txt")).strip().lower() website = str(data.get("website", "")).strip() location = str(data.get("location", "")).strip() if len(display_name) > MAX_DISPLAY_NAME_LENGTH: return jsonify({"error": "Display name is too long."}), 400 if len(bio) > MAX_PROFILE_BIO_LENGTH: return jsonify({"error": "Bio is too long."}), 400 if bio_mode not in PROFILE_BIO_MODES: return jsonify({"error": "Invalid bio mode."}), 400 if len(website) > MAX_WEBSITE_LENGTH: return jsonify({"error": "Website is too long."}), 400 if len(location) > MAX_LOCATION_LENGTH: return jsonify({"error": "Location is too long."}), 400 db_write( """ UPDATE user_profiles SET display_name = ?, bio = ?, bio_mode = ?, website = ?, location = ?, updated_at = ? WHERE user_id = ? """, (display_name or user["username"], bio, bio_mode, website, location, now_iso(), user["id"]), ) with get_db() as db: user_row = db.execute( """ SELECT id, username, role, status, restricted_until, created_at FROM users WHERE id = ? """, (user["id"],), ).fetchone() profile_row = db.execute( """ SELECT user_id, display_name, bio, bio_mode, website, location, updated_at FROM user_profiles WHERE user_id = ? """, (user["id"],), ).fetchone() settings_row = db.execute("SELECT name_color FROM user_settings WHERE user_id = ?", (user["id"],)).fetchone() audit("profile_updated", user["id"], "user", str(user["id"])) return jsonify({"profile": profile_payload(html_module, user_payload, user_row, profile_row, settings_row, user)}) @app.route("/api/profiles/") def api_get_public_profile(username: str): normalized = normalize_username(username) with get_db() as db: user_row = db.execute( """ SELECT id, username, role, status, restricted_until, created_at FROM users WHERE username = ? """, (normalized,), ).fetchone() if not user_row: return jsonify({"error": "Profile not found."}), 404 profile_row = db.execute( """ SELECT user_id, display_name, bio, bio_mode, website, location, updated_at FROM user_profiles WHERE user_id = ? """, (user_row["id"],), ).fetchone() settings_row = db.execute("SELECT name_color FROM user_settings WHERE user_id = ?", (user_row["id"],)).fetchone() return jsonify({"profile": profile_payload(html_module, user_payload, user_row, profile_row, settings_row, current_user())}) @app.route("/api/settings") def api_get_settings(): user, error = require_user() if error: return error with get_db() as db: row = db.execute( """ SELECT accent_color, name_color, custom_css, theme_mode, updated_at FROM user_settings WHERE user_id = ? """, (user["id"],), ).fetchone() return jsonify( { "settings": { "accent_color": row["accent_color"], "name_color": row["name_color"], "custom_css": row["custom_css"], "theme_mode": row["theme_mode"], "updated_at": row["updated_at"], } } ) @app.route("/api/settings", methods=["PATCH"]) def api_update_settings(): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 limited = rate_limited("settings") if limited: return limited user, error = require_writable_user() if error: return error try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 accent_color = data.get("accent_color", "#f7dce8") name_color = data.get("name_color") custom_css = str(data.get("custom_css", "")) theme_mode = str(data.get("theme_mode", "dark")).strip().lower() if not valid_hex_color(accent_color): return jsonify({"error": "Invalid accent color."}), 400 if name_color not in (None, "",) and not valid_hex_color(name_color): return jsonify({"error": "Invalid name color."}), 400 if len(custom_css) > MAX_CUSTOM_CSS_LENGTH: return jsonify({"error": "Custom CSS is too long."}), 400 if theme_mode not in THEME_MODES: return jsonify({"error": "Invalid theme mode."}), 400 db_write( """ UPDATE user_settings SET accent_color = ?, name_color = ?, custom_css = ?, theme_mode = ?, updated_at = ? WHERE user_id = ? """, ( str(accent_color).lower(), normalized_hex_color(name_color) if name_color else None, custom_css, theme_mode, now_iso(), user["id"], ), ) with get_db() as db: row = db.execute( """ SELECT accent_color, name_color, custom_css, theme_mode, updated_at FROM user_settings WHERE user_id = ? """, (user["id"],), ).fetchone() audit("settings_updated", user["id"], "user", str(user["id"])) return jsonify({"settings": dict(row)}) @app.route("/api/dms") def api_list_dms(): user, error = require_user() if error: return error return jsonify({"dms": list_dms_for_user(get_db, user["id"])}) @app.route("/api/dms", methods=["POST"]) def api_create_or_open_dm(): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error try: data = safe_json() except ValueError as error: return jsonify({"error": str(error)}), 400 other_user_id = data.get("user_id") username = normalize_username(data.get("username")) with get_db() as db: target = None if other_user_id is not None: try: parsed_id = int(other_user_id) except (TypeError, ValueError): return jsonify({"error": "Invalid user_id."}), 400 target = db.execute( """ SELECT id, username, role, status FROM users WHERE id = ? """, (parsed_id,), ).fetchone() elif username: target = db.execute( """ SELECT id, username, role, status FROM users WHERE username = ? """, (username,), ).fetchone() if not target: return jsonify({"error": "User not found."}), 404 if int(target["id"]) == int(user["id"]): return jsonify({"error": "You cannot DM yourself."}), 400 if target["status"] in {"disabled", "blacklisted"}: return jsonify({"error": "User not available."}), 400 dm_id = find_or_create_dm(get_db, db_write, now_iso, user["id"], target["id"]) audit("dm_opened", user["id"], "dm", str(dm_id), str(target["id"])) return jsonify({"dm": dm_payload(get_db, dm_id, user["id"])}), 201 @app.route("/api/dms//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/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)})