import json import time from db import db_write, get_db from config import utc_now, now_iso, GROUP_CALL_PARTICIPANT_STALE_SECONDS def create_call_session(dm_id: int, started_by_user_id: int) -> int: cursor = db_write( """ INSERT INTO call_sessions ( direct_conversation_id, started_by_user_id, status, created_at, ended_at ) VALUES (?, ?, 'active', ?, NULL) """, (dm_id, started_by_user_id, now_iso()), ) return cursor.lastrowid def append_call_signal(call_id: int, from_user_id: int, signal_type: str, payload: dict) -> int: cursor = db_write( """ INSERT INTO call_signals ( call_session_id, from_user_id, signal_type, payload_json, created_at ) VALUES (?, ?, ?, ?, ?) """, ( call_id, from_user_id, signal_type, json.dumps(payload or {}), now_iso(), ), ) return cursor.lastrowid def list_call_signals(call_id: int, after_id: int = 0) -> list[dict]: with get_db() as db: rows = db.execute( """ SELECT cs.id, cs.call_session_id, cs.from_user_id, u.username, cs.signal_type, cs.payload_json, cs.created_at FROM call_signals AS cs JOIN users AS u ON u.id = cs.from_user_id WHERE cs.call_session_id = ? AND cs.id > ? ORDER BY cs.id ASC LIMIT 200 """, (call_id, after_id), ).fetchall() items = [] for row in rows: try: payload = json.loads(row["payload_json"] or "{}") except json.JSONDecodeError: payload = {} items.append( { "id": row["id"], "call_session_id": row["call_session_id"], "from_user_id": row["from_user_id"], "username": row["username"], "signal_type": row["signal_type"], "payload": payload, "created_at": row["created_at"], } ) return items def end_call(call_id: int) -> None: db_write( """ UPDATE call_sessions SET status = 'ended', ended_at = ? WHERE id = ? """, (now_iso(), call_id), ) def get_call_with_dm(call_id: int): with get_db() as db: row = db.execute( """ SELECT cs.id, cs.direct_conversation_id, cs.started_by_user_id, cs.status, cs.created_at, cs.ended_at FROM call_sessions AS cs WHERE cs.id = ? """, (call_id,), ).fetchone() return row def call_payload(call_id: int): row = get_call_with_dm(call_id) if not row: return None return { "id": row["id"], "direct_conversation_id": row["direct_conversation_id"], "started_by_user_id": row["started_by_user_id"], "status": row["status"], "created_at": row["created_at"], "ended_at": row["ended_at"], } def build_room_helper_call_id(room) -> str: return f"room-{room['slug']}" def build_channel_helper_call_id(channel) -> str: return f"channel-{channel['team_slug']}--{channel['slug']}" def get_or_create_group_call_session( scope_kind: str, scope_id: int, scope_slug: str, helper_call_id: str, created_by_user_id: int, ): with get_db() as db: existing = db.execute( """ SELECT id, scope_kind, scope_id, scope_slug, helper_call_id, created_by_user_id, status, created_at, ended_at FROM group_call_sessions WHERE helper_call_id = ? LIMIT 1 """, (helper_call_id,), ).fetchone() if existing and existing["status"] == "active": return existing with get_db() as db: db.execute("BEGIN IMMEDIATE") existing = db.execute( """ SELECT id, scope_kind, scope_id, scope_slug, helper_call_id, created_by_user_id, status, created_at, ended_at FROM group_call_sessions WHERE helper_call_id = ? LIMIT 1 """, (helper_call_id,), ).fetchone() if existing and existing["status"] == "active": db.commit() return existing if existing: db.execute( """ UPDATE group_call_sessions SET status = 'active', ended_at = NULL, created_at = ? WHERE id = ? """, (now_iso(), existing["id"]), ) db.commit() refreshed = db.execute( """ SELECT id, scope_kind, scope_id, scope_slug, helper_call_id, created_by_user_id, status, created_at, ended_at FROM group_call_sessions WHERE id = ? """, (existing["id"],), ).fetchone() return refreshed cursor = db.execute( """ INSERT INTO group_call_sessions ( scope_kind, scope_id, scope_slug, helper_call_id, created_by_user_id, status, created_at, ended_at ) VALUES (?, ?, ?, ?, ?, 'active', ?, NULL) """, ( scope_kind, scope_id, scope_slug, helper_call_id, created_by_user_id, now_iso(), ), ) session_id = cursor.lastrowid db.commit() created = db.execute( """ SELECT id, scope_kind, scope_id, scope_slug, helper_call_id, created_by_user_id, status, created_at, ended_at FROM group_call_sessions WHERE id = ? """, (session_id,), ).fetchone() return created def group_call_session_payload(row): if not row: return None return { "id": row["id"], "scope_kind": row["scope_kind"], "scope_id": row["scope_id"], "scope_slug": row["scope_slug"], "helper_call_id": row["helper_call_id"], "created_by_user_id": row["created_by_user_id"], "status": row["status"], "created_at": row["created_at"], "ended_at": row["ended_at"], } def get_group_call_session_by_id(session_id: int): with get_db() as db: return db.execute( """ SELECT id, scope_kind, scope_id, scope_slug, helper_call_id, created_by_user_id, status, created_at, ended_at FROM group_call_sessions WHERE id = ? LIMIT 1 """, (session_id,), ).fetchone() def cleanup_stale_group_call_participants(session_id: int): cutoff = utc_now().timestamp() - GROUP_CALL_PARTICIPANT_STALE_SECONDS db_write( """ UPDATE group_call_participants SET left_at = COALESCE(left_at, ?) WHERE group_call_session_id = ? AND left_at IS NULL AND ( updated_at IS NULL OR strftime('%s', updated_at) < ? ) """, (now_iso(), session_id, int(cutoff)), ) def upsert_group_call_participant(session_id: int, user_id: int): db_write( """ INSERT INTO group_call_participants ( group_call_session_id, user_id, joined_at, updated_at, left_at ) VALUES (?, ?, ?, ?, NULL) ON CONFLICT(group_call_session_id, user_id) DO UPDATE SET updated_at = excluded.updated_at, left_at = NULL """, (session_id, user_id, now_iso(), now_iso()), ) def heartbeat_group_call_participant(session_id: int, user_id: int): db_write( """ UPDATE group_call_participants SET updated_at = ?, left_at = NULL WHERE group_call_session_id = ? AND user_id = ? """, (now_iso(), session_id, user_id), ) def mark_group_call_participant_left(session_id: int, user_id: int): db_write( """ UPDATE group_call_participants SET updated_at = ?, left_at = ? WHERE group_call_session_id = ? AND user_id = ? AND left_at IS NULL """, (now_iso(), now_iso(), session_id, user_id), ) def list_group_call_participants(session_id: int) -> list[dict]: cleanup_stale_group_call_participants(session_id) with get_db() as db: rows = db.execute( """ SELECT gcp.group_call_session_id, gcp.user_id, gcp.joined_at, gcp.updated_at, gcp.left_at, u.username, u.role, up.display_name, us.name_color FROM group_call_participants AS gcp JOIN users AS u ON u.id = gcp.user_id LEFT JOIN user_profiles AS up ON up.user_id = u.id LEFT JOIN user_settings AS us ON us.user_id = u.id WHERE gcp.group_call_session_id = ? AND gcp.left_at IS NULL ORDER BY gcp.joined_at ASC, u.username COLLATE NOCASE """, (session_id,), ).fetchall() return [ { "user_id": row["user_id"], "username": row["username"], "display_name": row["display_name"] or row["username"], "role": row["role"], "name_color": row["name_color"], "joined_at": row["joined_at"], "updated_at": row["updated_at"], } for row in rows ] def active_group_call_participant_count(session_id: int) -> int: cleanup_stale_group_call_participants(session_id) with get_db() as db: row = db.execute( """ SELECT COUNT(*) AS count FROM group_call_participants WHERE group_call_session_id = ? AND left_at IS NULL """, (session_id,), ).fetchone() return int(row["count"]) if row else 0 def maybe_end_group_call_if_empty(session_id: int): if active_group_call_participant_count(session_id) > 0: return db_write( """ UPDATE group_call_sessions SET status = 'ended', ended_at = ? WHERE id = ? AND status != 'ended' """, (now_iso(), session_id), ) db_write( """ UPDATE group_call_invites SET status = 'dismissed', responded_at = ? WHERE group_call_session_id = ? AND status = 'pending' """, (now_iso(), session_id), )