from .common import ( MAX_OWNED_TEAMS_PER_USER, PAGE_TYPES, PAGE_VISIBILITIES, ROOM_SLUG_RE, TEAM_INVITE_TTL_HOURS, TEAM_MEMBER_ROLES, now_iso, parse_iso, utc_now, csrf_valid, rate_limited, current_user, require_user, require_writable_user, audit, hash_identifier, normalize_username, safe_json, jsonify, request, get_db, db_write, team_page_select, team_slug_from_name, page_slug_from_title, get_team_by_slug, get_channel_by_id, get_channel_by_slug, team_member_role, is_team_member, is_team_admin, can_edit_team, can_access_team, can_access_channel, team_payload, channel_payload, page_payload, build_local_stub_share_url, ) def register_team_routes(app): @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 = app.secrets.token_urlsafe(32) expires_at = (utc_now() + app.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, is_team_member, is_team_admin) 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, is_team_member, is_team_admin)}), 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, is_team_member, is_team_admin)}) @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, is_team_member, is_team_admin)}) @app.route("/api/teams//pages/", methods=["DELETE"]) def api_delete_page(team_slug: str, page_slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error with get_db() as db: page = db.execute( """ SELECT p.id, p.team_id FROM pages AS p JOIN teams AS t ON t.id = p.team_id WHERE t.slug = ? AND p.slug = ? AND p.is_active = 1 AND t.is_active = 1 """, (team_slug, page_slug), ).fetchone() if not page: return jsonify({"error": "Page not found."}), 404 if not is_team_admin(page["team_id"], user): return jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403 db_write( """ UPDATE pages SET is_active = 0, updated_by_user_id = ?, updated_at = ? WHERE id = ? """, (user["id"], now_iso(), page["id"]), ) audit("page_deleted", user["id"], "page", str(page["id"]), f"{team_slug}/{page_slug}") return jsonify({"status": "deleted"}) @app.route("/api/teams//pages//versions") def api_list_page_versions(team_slug: str, page_slug: str): user = current_user() with get_db() as db: page = db.execute( f""" {team_page_select()} WHERE t.slug = ? AND p.slug = ? AND p.is_active = 1 AND t.is_active = 1 """, (team_slug, page_slug), ).fetchone() if not page: return jsonify({"error": "Page not found."}), 404 permitted = bool( page["visibility"] == "public" or (user and (user["role"] in {"moderator", "admin"} or is_team_member(page["team_id"], user["id"]))) ) if not permitted: if not user: return jsonify({"error": "Login required.", "code": "login_required"}), 401 return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 return jsonify({"versions": []}) @app.route("/api/page-versions/") def api_get_page_version(version_id: int): return jsonify( { "version": { "id": version_id, "version_number": 1, "title": "version unavailable", "body_html": "

page version history is not enabled on this server.

", } } ) @app.route("/api/page-versions//restore", methods=["POST"]) def api_restore_page_version(version_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 return jsonify({"error": "Page version restore is not enabled on this server."}), 501 @app.route("/api/teams//pages//comments") def api_list_page_comments(team_slug: str, page_slug: str): user = current_user() with get_db() as db: page = db.execute( f""" {team_page_select()} WHERE t.slug = ? AND p.slug = ? AND p.is_active = 1 AND t.is_active = 1 """, (team_slug, page_slug), ).fetchone() if not page: return jsonify({"error": "Page not found."}), 404 permitted = bool( page["visibility"] == "public" or (user and (user["role"] in {"moderator", "admin"} or is_team_member(page["team_id"], user["id"]))) ) if not permitted: if not user: return jsonify({"error": "Login required.", "code": "login_required"}), 401 return jsonify({"error": "Forbidden.", "code": "forbidden"}), 403 return jsonify({"comments": []}) @app.route("/api/teams//pages//comments", methods=["POST"]) def api_create_page_comment(team_slug: str, page_slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error return jsonify({"error": "Page comments are not enabled on this server."}), 501 @app.route("/api/page-comments/", methods=["DELETE"]) def api_delete_page_comment(comment_id: int): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_writable_user() if error: return error return jsonify({"error": "Page comments are not enabled on this server."}), 501 @app.route("/api/teams//pages//share", methods=["POST"]) def api_create_page_share(team_slug: str, page_slug: str): if not csrf_valid(): return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 user, error = require_user() if error: return error with get_db() as db: page = db.execute( """ SELECT p.id FROM pages AS p JOIN teams AS t ON t.id = p.team_id WHERE t.slug = ? AND p.slug = ? AND p.is_active = 1 AND t.is_active = 1 """, (team_slug, page_slug), ).fetchone() if not page: return jsonify({"error": "Page not found."}), 404 return jsonify( { "share": { "url": build_local_stub_share_url("page", page["id"]) } } )