465 lines
12 KiB
Python
465 lines
12 KiB
Python
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),
|
|
)
|