# retoor <retoor@molodetz.nl>
import asyncio
import json
import logging
import random
import traceback
import time
from aiohttp import web
from snek.system.model import now
from snek.system.view import BaseView
logger = logging.getLogger(__name__)
def safe_get(obj, key, default=None):
if obj is None:
return default
try:
if isinstance(obj, dict):
return obj.get(key, default)
if hasattr(obj, "fields") and hasattr(obj, "__getitem__"):
val = obj[key]
return val if val is not None else default
return getattr(obj, key, default)
except (KeyError, TypeError, AttributeError):
return default
def safe_str(obj):
if obj is None:
return ""
try:
return str(obj)
except Exception:
return ""
class RPCView(BaseView):
class RPCApi:
def __init__(self, view, ws):
self.view = view
self.app = safe_get(view, "app")
self.services = safe_get(self.app, "services") if self.app else None
self.ws = ws
self.user_session = {}
self._scheduled_tasks = []
self._finalize_task = None
self._is_closed = False
async def _session_ensure(self):
try:
uid = await self.view.session_get("uid") if self.view else None
if uid and uid not in self.user_session:
self.user_session[uid] = {"said_hello": False}
except Exception as ex:
logger.warning(f"Failed to ensure session: {safe_str(ex)}")
async def session_get(self, key, default=None):
try:
await self._session_ensure()
uid = self.user_uid
if not uid or uid not in self.user_session:
return default
return self.user_session[uid].get(key, default)
except Exception as ex:
logger.warning(f"Failed to get session key {key}: {safe_str(ex)}")
return default
async def session_set(self, key, value):
try:
await self._session_ensure()
uid = self.user_uid
if not uid:
return False
if uid not in self.user_session:
self.user_session[uid] = {}
self.user_session[uid][key] = value
return True
except Exception as ex:
logger.warning(f"Failed to set session key {key}: {safe_str(ex)}")
return False
async def db_insert(self, table_name, record):
self._require_login()
self._require_services()
if not table_name or not isinstance(table_name, str):
raise ValueError("Invalid table name")
if not record or not isinstance(record, dict):
raise ValueError("Invalid record")
return await self.services.db.insert(self.user_uid, table_name, record)
async def db_update(self, table_name, record):
self._require_login()
self._require_services()
if not table_name or not isinstance(table_name, str):
raise ValueError("Invalid table name")
if not record or not isinstance(record, dict):
raise ValueError("Invalid record")
return await self.services.db.update(self.user_uid, table_name, record)
async def set_typing(self, channel_uid, color=None):
logger.debug(f"set_typing: called channel_uid={channel_uid}, user_uid={self.user_uid}, color={color}")
self._require_login()
self._require_services()
if not channel_uid or not isinstance(channel_uid, str):
logger.warning(f"set_typing: invalid channel_uid: {channel_uid}")
return False
try:
user = await self.services.user.get(uid=self.user_uid)
if not user:
logger.warning(f"set_typing: user not found: {self.user_uid}")
return False
if not color:
color = safe_get(user, "color", "#000000")
logger.debug(f"set_typing: broadcasting typing indicator to channel={channel_uid}, user={safe_get(user, 'username')}")
result = await self.services.socket.broadcast(
channel_uid,
{
"channel_uid": channel_uid,
"event": "set_typing",
"data": {
"event": "set_typing",
"user_uid": safe_get(user, "uid", ""),
"username": safe_get(user, "username", ""),
"nick": safe_get(user, "nick", ""),
"channel_uid": channel_uid,
"color": color,
},
},
)
logger.debug(f"set_typing: broadcast result={result}")
return result
except Exception as ex:
logger.warning(f"Failed to set typing: {safe_str(ex)}")
return False
async def db_delete(self, table_name, record):
self._require_login()
self._require_services()
if not table_name or not isinstance(table_name, str):
raise ValueError("Invalid table name")
if not record or not isinstance(record, dict):
raise ValueError("Invalid record")
return await self.services.db.delete(self.user_uid, table_name, record)
async def db_get(self, table_name, record):
self._require_login()
self._require_services()
if not table_name or not isinstance(table_name, str):
raise ValueError("Invalid table name")
return await self.services.db.get(self.user_uid, table_name, record)
async def db_find(self, table_name, record):
self._require_login()
self._require_services()
if not table_name or not isinstance(table_name, str):
raise ValueError("Invalid table name")
return await self.services.db.find(self.user_uid, table_name, record)
async def db_upsert(self, table_name, record, keys):
self._require_login()
self._require_services()
if not table_name or not isinstance(table_name, str):
raise ValueError("Invalid table name")
if not record or not isinstance(record, dict):
raise ValueError("Invalid record")
return await self.services.db.upsert(
self.user_uid, table_name, record, keys
)
async def db_query(self, sql, args):
self._require_login()
self._require_services()
if not sql or not isinstance(sql, str):
raise ValueError("Invalid SQL query")
return await self.services.db.query(self.user_uid, sql, args)
@property
def user_uid(self):
try:
if not self.view or not hasattr(self.view, "session"):
return None
session = self.view.session
if session is None:
return None
if hasattr(session, "get"):
return session.get("uid")
return None
except Exception:
return None
@property
def request(self):
return safe_get(self.view, "request")
def _require_login(self):
if not self.is_logged_in:
raise PermissionError("Not logged in")
def _require_services(self):
if not self.services:
raise RuntimeError("Services not available")
@property
def is_logged_in(self):
try:
if not self.view or not hasattr(self.view, "session"):
return False
session = self.view.session
if session is None:
return False
return bool(session.get("logged_in", False))
except Exception:
return False
async def mark_as_read(self, channel_uid):
self._require_login()
self._require_services()
if not channel_uid or not isinstance(channel_uid, str):
return False
try:
await self.services.channel_member.mark_as_read(channel_uid, self.user_uid)
return True
except Exception as ex:
logger.warning(f"Failed to mark as read: {safe_str(ex)}")
return False
async def login(self, username, password):
self._require_services()
if not username or not isinstance(username, str):
raise ValueError("Invalid username")
if not password or not isinstance(password, str):
raise ValueError("Invalid password")
try:
success = await self.services.user.validate_login(username, password)
if not success:
raise PermissionError("Invalid username or password")
user = await self.services.user.get(username=username)
if not user:
raise PermissionError("User not found")
if not self.view or not hasattr(self.view, "session") or self.view.session is None:
raise RuntimeError("Session not available")
self.view.session["uid"] = safe_get(user, "uid")
self.view.session["logged_in"] = True
self.view.session["username"] = safe_get(user, "username", "")
self.view.session["user_nick"] = safe_get(user, "nick", "")
record = dict(user.record) if hasattr(user, "record") else {}
record.pop("password", None)
record.pop("deleted_at", None)
user_uid = None
if self.view and hasattr(self.view, "session") and self.view.session:
user_uid = self.view.session.get("uid") if hasattr(self.view.session, "get") else None
if user_uid:
await self.services.socket.add(self.ws, user_uid)
try:
async for subscription in self.services.channel_member.find(
user_uid=user_uid,
deleted_at=None,
is_banned=False,
):
if subscription and subscription["channel_uid"]:
await self.services.socket.subscribe(
self.ws,
subscription["channel_uid"],
user_uid,
)
except Exception as ex:
logger.warning(f"Failed to subscribe to channels: {safe_str(ex)}")
return record
except (PermissionError, ValueError):
raise
except Exception as ex:
logger.error(f"Login error: {safe_str(ex)}")
raise PermissionError("Login failed")
async def search_user(self, query):
self._require_login()
self._require_services()
if not query or not isinstance(query, str):
return []
try:
results = await self.services.user.search(query)
return [safe_get(user, "username", "") for user in (results or []) if user]
except Exception as ex:
logger.warning(f"Failed to search user: {safe_str(ex)}")
return []
async def get_user(self, user_uid=None):
self._require_login()
self._require_services()
try:
if not user_uid:
user_uid = self.user_uid
if not user_uid:
return None
user = await self.services.user.get(uid=user_uid)
if not user:
return None
record = dict(user.record) if hasattr(user, "record") else {}
record.pop("password", None)
record.pop("deleted_at", None)
if user_uid != self.user_uid:
record.pop("email", None)
return record
except Exception as ex:
logger.warning(f"Failed to get user: {safe_str(ex)}")
return None
async def get_messages(self, channel_uid, offset=0, timestamp=None):
self._require_login()
self._require_services()
if not channel_uid or not isinstance(channel_uid, str):
return []
try:
offset = int(offset) if offset else 0
if offset < 0:
offset = 0
messages = []
result = await self.services.channel_message.offset(
channel_uid, offset, timestamp or None
)
for message in (result or []):
if message:
try:
extended_dict = await self.services.channel_message.to_extended_dict(message)
if extended_dict:
messages.append(extended_dict)
except Exception as ex:
logger.warning(f"Failed to extend message: {safe_str(ex)}")
return messages
except Exception as ex:
logger.warning(f"Failed to get messages: {safe_str(ex)}")
return []
async def get_first_unread_message_uid(self, channel_uid):
self._require_login()
self._require_services()
if not channel_uid or not isinstance(channel_uid, str):
return None
try:
channel_member = await self.services.channel_member.get(
channel_uid=channel_uid, user_uid=self.user_uid
)
if not channel_member:
return None
last_read_at = safe_get(channel_member, "last_read_at")
if not last_read_at:
return None
async for message in self.services.channel_message.query(
"SELECT uid FROM channel_message WHERE channel_uid=:channel_uid AND created_at > :last_read_at AND deleted_at IS NULL ORDER BY created_at ASC LIMIT 1",
{"channel_uid": channel_uid, "last_read_at": last_read_at}
):
return safe_get(message, "uid")
return None
except Exception as ex:
logger.warning(f"Failed to get first unread message: {safe_str(ex)}")
return None
async def get_channels(self):
self._require_login()
self._require_services()
channels = []
try:
async for subscription in self.services.channel_member.find(
user_uid=self.user_uid, is_banned=False
):
if not subscription:
continue
try:
channel_uid = subscription["channel_uid"]
if not channel_uid:
continue
channel = await self.services.channel.get(uid=channel_uid)
if not channel:
continue
color = None
try:
last_message = await channel.get_last_message()
if last_message:
last_message_user = await last_message.get_user()
color = safe_get(last_message_user, "color") if last_message_user else None
except Exception:
pass
channels.append({
"name": subscription["label"] or "",
"uid": channel_uid,
"tag": channel["tag"] or "",
"new_count": subscription["new_count"] or 0,
"is_moderator": subscription["is_moderator"] or False,
"is_read_only": subscription["is_read_only"] or False,
"color": color,
})
except Exception as ex:
logger.warning(f"Failed to process channel subscription: {safe_str(ex)}")
return channels
except Exception as ex:
logger.warning(f"Failed to get channels: {safe_str(ex)}")
return []
async def write_container(self, channel_uid, content, timeout=3):
self._require_login()
self._require_services()
if not channel_uid or not isinstance(channel_uid, str):
raise ValueError("Invalid channel_uid")
try:
channel_member = await self.services.channel_member.get(
channel_uid=channel_uid, user_uid=self.user_uid
)
if not channel_member:
raise PermissionError("Not allowed")
container_name = await self.services.container.get_container_name(channel_uid)
if not container_name:
raise RuntimeError("Container not found")
timeout = max(1, min(int(timeout) if timeout else 3, 30))
class SessionCall:
def __init__(self, app, channel_uid, container_name):
self.app = app
self.channel_uid = channel_uid
self.container_name = container_name
self.time_last_output = time.time()
self.output = b""
async def stdout_event_handler(self, data):
try:
self.time_last_output = time.time()
if data:
self.output += data if isinstance(data, bytes) else data.encode("utf-8", "ignore")
return True
except Exception:
return False
async def communicate(self, content, timeout=3):
try:
if self.app and hasattr(self.app, "services") and self.app.services:
await self.app.services.container.add_event_listener(
self.container_name, "stdout", self.stdout_event_handler
)
await self.app.services.container.write_stdin(self.channel_uid, content)
start_time = time.time()
while time.time() - self.time_last_output < timeout:
if time.time() - start_time > timeout * 2:
break
await asyncio.sleep(0.1)
await self.app.services.container.remove_event_listener(
self.container_name, "stdout", self.stdout_event_handler
)
return self.output
except Exception as ex:
logger.warning(f"Container communication error: {safe_str(ex)}")
return self.output
sc = SessionCall(self.app, channel_uid, container_name)
result = await sc.communicate(content, timeout)
return result.decode("utf-8", "ignore") if result else ""
except (PermissionError, ValueError):
raise
except Exception as ex:
logger.warning(f"Failed to write container: {safe_str(ex)}")
return ""
async def get_container(self, channel_uid):
self._require_login()
self._require_services()
if not channel_uid or not isinstance(channel_uid, str):
return None
try:
channel_member = await self.services.channel_member.get(
channel_uid=channel_uid, user_uid=self.user_uid
)
if not channel_member:
raise PermissionError("Not allowed")
container = await self.services.container.get(channel_uid)
if not container:
return None
deploy = safe_get(container, "deploy", {})
resources = safe_get(deploy, "resources", {})
limits = safe_get(resources, "limits", {})
return {
"name": await self.services.container.get_container_name(channel_uid),
"cpus": safe_get(limits, "cpus", "0.5"),
"memory": safe_get(limits, "memory", "256M"),
"image": "ubuntu:latest",
"volumes": [],
"status": safe_get(container, "status", "unknown"),
}
except PermissionError:
raise
except Exception as ex:
logger.warning(f"Failed to get container: {safe_str(ex)}")
return None
async def _finalize_message_task(self, message_uid):
if not message_uid:
return False
try:
for _ in range(7):
await asyncio.sleep(1)
if not self._finalize_task:
return False
except asyncio.CancelledError:
logger.debug("Finalization cancelled")
return False
except Exception as ex:
logger.warning(f"Finalization wait error: {safe_str(ex)}")
return False
self._finalize_task = None
try:
if not self.services:
return False
message = await self.services.channel_message.get(message_uid, is_final=False)
if not message:
return False
if safe_get(message, "user_uid") != self.user_uid:
return False
if safe_get(message, "is_final", False):
return True
await self.services.chat.finalize(safe_get(message, "uid"))
return True
except Exception as ex:
logger.warning(f"Failed to finalize message: {safe_str(ex)}")
return False
async def _queue_finalize_message(self, message_uid):
if not message_uid:
return
try:
if self._finalize_task:
self._finalize_task.cancel()
try:
await self._finalize_task
except asyncio.CancelledError:
pass
except Exception:
pass
self._finalize_task = asyncio.create_task(
self._finalize_message_task(message_uid)
)
except Exception as ex:
logger.warning(f"Failed to queue finalize: {safe_str(ex)}")
async def send_message(self, channel_uid, message, is_final=True):
logger.debug(f"send_message called: channel_uid={channel_uid}, message_len={len(str(message)) if message else 0}, is_final={is_final}, user_uid={self.user_uid}")
self._require_login()
self._require_services()
if not channel_uid or not isinstance(channel_uid, str):
logger.warning(f"send_message: invalid channel_uid: {channel_uid}")
return None
if message is None:
logger.debug("send_message: message is None")
return None
try:
message_text = safe_str(message).strip()
logger.debug(f"send_message: processing message text, length={len(message_text)}, is_final={is_final}")
if not is_final:
logger.debug(f"send_message: checking for existing non-final message in channel={channel_uid}")
check_message = await self.services.channel_message.get(
channel_uid=channel_uid, user_uid=self.user_uid, is_final=False, deleted_at=None
)
if check_message:
msg_uid = safe_get(check_message, "uid")
logger.debug(f"send_message: found existing non-final message uid={msg_uid}, updating text")
await self._queue_finalize_message(msg_uid)
return await self.update_message_text(msg_uid, message_text)
if not message_text:
logger.debug("send_message: empty message text after strip")
return None
logger.info(f"send_message: sending new message to channel={channel_uid}, is_final={is_final}")
result = await self.services.chat.send(
self.user_uid, channel_uid, message_text, is_final
)
if not result:
logger.warning(f"send_message: chat.send returned None for channel={channel_uid}")
return None
result_uid = safe_get(result, "uid")
result_is_final = safe_get(result, "is_final", True)
logger.info(f"send_message: message sent successfully, uid={result_uid}, is_final={result_is_final}")
if not result_is_final:
await self._queue_finalize_message(result_uid)
elif self._finalize_task:
self._finalize_task.cancel()
return result_uid
except Exception as ex:
logger.warning(f"Failed to send message: {safe_str(ex)}", exc_info=True)
return None
async def start_container(self, channel_uid):
self._require_login()
self._require_services()
if not channel_uid or not isinstance(channel_uid, str):
raise ValueError("Invalid channel_uid")
channel_member = await self.services.channel_member.get(
channel_uid=channel_uid, user_uid=self.user_uid
)
if not channel_member:
raise PermissionError("Not allowed")
return await self.services.container.start(channel_uid)
async def stop_container(self, channel_uid):
self._require_login()
self._require_services()
if not channel_uid or not isinstance(channel_uid, str):
raise ValueError("Invalid channel_uid")
channel_member = await self.services.channel_member.get(
channel_uid=channel_uid, user_uid=self.user_uid
)
if not channel_member:
raise PermissionError("Not allowed")
return await self.services.container.stop(channel_uid)
async def get_container_status(self, channel_uid):
self._require_login()
self._require_services()
if not channel_uid or not isinstance(channel_uid, str):
return None
try:
channel_member = await self.services.channel_member.get(
channel_uid=channel_uid, user_uid=self.user_uid
)
if not channel_member:
raise PermissionError("Not allowed")
return await self.services.container.get_status(channel_uid)
except PermissionError:
raise
except Exception as ex:
logger.warning(f"Failed to get container status: {safe_str(ex)}")
return None
async def finalize_message(self, message_uid):
self._require_login()
self._require_services()
if not message_uid or not isinstance(message_uid, str):
return False
try:
message = await self.services.channel_message.get(message_uid)
if not message:
return False
if safe_get(message, "user_uid") != self.user_uid:
raise PermissionError("Not allowed")
if not safe_get(message, "is_final", False):
await self.services.chat.finalize(safe_get(message, "uid"))
return True
except PermissionError:
raise
except Exception as ex:
logger.warning(f"Failed to finalize message: {safe_str(ex)}")
return False
async def update_message_text(self, message_uid, text):
self._require_login()
self._require_services()
if not message_uid or not isinstance(message_uid, str):
return {"error": "Invalid message_uid", "success": False}
try:
message = await self.services.channel_message.get(message_uid)
if not message:
return {"error": "Message not found", "success": False}
if safe_get(message, "is_final", False):
return {"error": "Message is final", "success": False}
if safe_get(message, "deleted_at"):
return {"error": "Message is deleted", "success": False}
if safe_get(message, "user_uid") != self.user_uid:
raise PermissionError("Not allowed")
seconds_since_update = 0
try:
if hasattr(message, "get_seconds_since_last_update"):
seconds_since_update = message.get_seconds_since_last_update()
except Exception:
pass
if seconds_since_update > 8:
return {
"error": "Message too old",
"seconds_since_last_update": seconds_since_update,
"success": False,
}
text_str = safe_str(text) if text else ""
if not text_str:
message["deleted_at"] = now()
else:
message["deleted_at"] = None
message["message"] = text_str
await self.services.channel_message.save(message)
record = dict(message.record) if hasattr(message, "record") else {}
record["text"] = safe_get(message, "message", "")
record["message_uid"] = message_uid
channel_uid = safe_get(message, "channel_uid")
if channel_uid:
try:
await self.services.socket.broadcast(
channel_uid,
{
"channel_uid": channel_uid,
"event": "update_message_text",
"data": record,
},
)
except Exception as ex:
logger.warning(f"Failed to broadcast message update: {safe_str(ex)}")
return {"success": True}
except PermissionError:
raise
except Exception as ex:
logger.warning(f"Failed to update message text: {safe_str(ex)}")
return {"error": "Update failed", "success": False}
async def clear_channel(self, channel_uid):
self._require_login()
self._require_services()
if not channel_uid or not isinstance(channel_uid, str):
raise ValueError("Invalid channel_uid")
user = await self.services.user.get(uid=self.user_uid)
if not user or not safe_get(user, "is_admin", False):
raise PermissionError("Not allowed")
channel = await self.services.channel.get(uid=channel_uid)
if not channel:
raise ValueError("Channel not found")
channel["history_start"] = now()
await self.services.channel.save(channel)
return await self.services.channel_message.clear(channel_uid)
async def echo(self, *args):
return args
async def query(self, *args):
self._require_login()
self._require_services()
if not args or len(args) == 0:
return []
query_str = args[0]
if not query_str or not isinstance(query_str, str):
return []
lowercase = query_str.lower()
forbidden_keywords = ["drop", "alter", "update", "delete", "replace", "insert", "truncate"]
if any(keyword in lowercase for keyword in forbidden_keywords) and "select" not in lowercase:
raise PermissionError("Not allowed")
try:
results = []
async for record in self.services.channel.query(query_str):
if record:
record_dict = dict(record)
record_dict.pop("email", None)
record_dict.pop("password", None)
record_dict.pop("message", None)
record_dict.pop("html", None)
results.append(record_dict)
return results
except Exception as ex:
logger.warning(f"Query failed: {safe_str(ex)}")
return []
async def __call__(self, data):
call_id = None
start_time = time.time()
try:
if self._is_closed:
logger.debug("RPC call on closed connection")
return
if not data or not isinstance(data, dict):
logger.warning("Invalid RPC data received")
return
call_id = data.get("callId")
method_name = data.get("method")
args = data.get("args", [])
logger.debug(f"RPC call received: method={method_name}, callId={call_id}, args_count={len(args) if args else 0}, user_uid={self.user_uid}")
if not method_name or not isinstance(method_name, str):
logger.warning(f"Invalid method name: {method_name}")
await self._send_json({"callId": call_id, "success": False, "data": "Invalid method"})
return
if method_name.startswith("_"):
logger.warning(f"Attempted to call private method: {method_name}")
await self._send_json({"callId": call_id, "success": False, "data": "Not allowed"})
return
if args is None:
args = []
elif not isinstance(args, (list, tuple)):
args = [args]
safe_method_name = method_name.replace(".", "_")
if not hasattr(self, safe_method_name):
logger.warning(f"Method not found: {method_name}")
await self._send_json({"callId": call_id, "success": False, "data": "Method not found"})
return
method = getattr(self, safe_method_name, None)
if not method or not callable(method):
logger.warning(f"Method not callable: {method_name}")
await self._send_json({"callId": call_id, "success": False, "data": "Method not callable"})
return
success = True
result = None
try:
logger.debug(f"Executing RPC method: {method_name}")
result = await method(*args)
elapsed = (time.time() - start_time) * 1000
logger.info(f"RPC method completed: method={method_name}, callId={call_id}, elapsed={elapsed:.2f}ms, success=True")
except (PermissionError, ValueError) as ex:
result = {"error": safe_str(ex)}
success = False
elapsed = (time.time() - start_time) * 1000
logger.warning(f"RPC method permission/value error: method={method_name}, callId={call_id}, elapsed={elapsed:.2f}ms, error={safe_str(ex)}")
except Exception as ex:
result = {"error": "Internal error"}
success = False
elapsed = (time.time() - start_time) * 1000
logger.exception(f"RPC method failed: method={method_name}, callId={call_id}, elapsed={elapsed:.2f}ms, error={safe_str(ex)}")
if result != "noresponse":
await self._send_json({"callId": call_id, "success": success, "data": result})
except Exception as ex:
logger.exception(f"RPC call failed: {safe_str(ex)}")
try:
await self._send_json({"callId": call_id, "success": False, "data": "Internal error"})
except Exception:
pass
async def _send_json(self, obj):
if self._is_closed:
return False
if not self.ws:
return False
try:
json_str = json.dumps(obj, default=str)
await self.ws.send_str(json_str)
return True
except ConnectionResetError:
self._is_closed = True
logger.debug("Connection reset during send")
except Exception as ex:
self._is_closed = True
logger.warning(f"Failed to send JSON: {safe_str(ex)}")
try:
if self.services and hasattr(self.services, "socket"):
await self.services.socket.delete(self.ws)
except Exception:
pass
return False
async def get_online_users(self, channel_uid):
self._require_login()
self._require_services()
if not channel_uid or not isinstance(channel_uid, str):
return []
try:
results = []
async for record in self.services.channel.get_recent_users(channel_uid):
if record:
results.append(record)
return sorted(results, key=lambda x: safe_get(x, "nick", ""))
except Exception as ex:
logger.warning(f"Failed to get online users: {safe_str(ex)}")
return []
async def echo_raw(self, obj):
if not self.ws:
return "noresponse"
try:
await self.ws.send_json(obj if obj else {})
except Exception as ex:
logger.warning(f"Failed to echo: {safe_str(ex)}")
return "noresponse"
async def get_recent_users(self, channel_uid):
self._require_login()
self._require_services()
if not channel_uid or not isinstance(channel_uid, str):
return []
try:
results = []
async for record in self.services.channel.get_recent_users(channel_uid):
if record:
results.append({
"uid": safe_get(record, "uid", ""),
"username": safe_get(record, "username", ""),
"nick": safe_get(record, "nick", ""),
"last_ping": safe_get(record, "last_ping"),
})
return results
except Exception as ex:
logger.warning(f"Failed to get recent users: {safe_str(ex)}")
return []
async def get_users(self, channel_uid):
self._require_login()
self._require_services()
if not channel_uid or not isinstance(channel_uid, str):
return []
try:
results = []
async for record in self.services.channel.get_users(channel_uid):
if record:
results.append({
"uid": safe_get(record, "uid", ""),
"username": safe_get(record, "username", ""),
"nick": safe_get(record, "nick", ""),
"last_ping": safe_get(record, "last_ping"),
})
return results
except Exception as ex:
logger.warning(f"Failed to get users: {safe_str(ex)}")
return []
async def _schedule(self, seconds, call):
if not call:
return
try:
seconds = max(0, min(float(seconds) if seconds else 0, 300))
await asyncio.sleep(seconds)
if self.services and self.user_uid and not self._is_closed:
await self.services.socket.send_to_user(self.user_uid, call)
except asyncio.CancelledError:
pass
except Exception as ex:
logger.warning(f"Schedule failed: {safe_str(ex)}")
async def ping(self, call_id=None, *args):
try:
if self.user_uid and self.services:
user = await self.services.user.get(uid=self.user_uid)
if user:
user["last_ping"] = now()
await self.services.user.save(user)
return {"pong": args if args else []}
except Exception as ex:
logger.warning(f"Ping failed: {safe_str(ex)}")
return {"pong": []}
async def stars_render(self, channel_uid, message):
self._require_services()
if not channel_uid or not isinstance(channel_uid, str):
return False
try:
users = await self.get_online_users(channel_uid)
for user in (users or []):
if not user:
continue
user_uid = safe_get(user, "uid")
if not user_uid:
continue
try:
await self.services.socket.send_to_user(
user_uid,
{
"event": "stars_render",
"data": {
"channel_uid": channel_uid,
"message": safe_str(message) if message else "",
},
},
)
except Exception as ex:
logger.debug(f"Failed to send stars_render to user: {safe_str(ex)}")
return True
except Exception as ex:
logger.warning(f"stars_render failed: {safe_str(ex)}")
return False
async def get(self):
ws = None
rpc = None
scheduled_tasks = []
async def schedule(uid, seconds, call):
if not uid or not call:
return
try:
seconds = max(0, min(float(seconds) if seconds else 0, 300))
await asyncio.sleep(seconds)
if self.services:
await self.services.socket.send_to_user(uid, call)
except asyncio.CancelledError:
pass
except Exception as ex:
logger.debug(f"Schedule failed: {safe_str(ex)}")
try:
ws = web.WebSocketResponse(heartbeat=30.0, autoping=True)
await ws.prepare(self.request)
except Exception as ex:
logger.warning(f"Failed to prepare WebSocket: {safe_str(ex)}")
return web.Response(status=500, text="WebSocket initialization failed")
user_uid = None
try:
session = getattr(self.request, "session", None)
if session and session.get("logged_in"):
user_uid = session.get("uid")
if user_uid and self.services:
await self.services.socket.add(ws, user_uid)
try:
async for subscription in self.services.channel_member.find(
user_uid=user_uid,
deleted_at=None,
is_banned=False,
):
if subscription:
channel_uid = safe_get(subscription, "channel_uid")
if channel_uid:
await self.services.socket.subscribe(ws, channel_uid, user_uid)
except Exception as ex:
logger.warning(f"Failed to subscribe to channels: {safe_str(ex)}")
except Exception as ex:
logger.warning(f"Session initialization failed: {safe_str(ex)}")
try:
app = getattr(self.request, "app", None)
uptime_seconds = getattr(app, "uptime_seconds", 999) if app else 999
if not scheduled_tasks and uptime_seconds < 15 and user_uid:
jitter = random.uniform(0, 3)
task1 = asyncio.create_task(schedule(
user_uid, jitter,
{"event": "refresh", "data": {"message": "Finishing deployment"}},
))
scheduled_tasks.append(task1)
uptime = getattr(app, "uptime", "") if app else ""
task2 = asyncio.create_task(schedule(
user_uid, 15 + jitter,
{"event": "deployed", "data": {"uptime": uptime}},
))
scheduled_tasks.append(task2)
except Exception as ex:
logger.debug(f"Deployment schedule failed: {safe_str(ex)}")
rpc = RPCView.RPCApi(self, ws)
try:
async for msg in ws:
if msg is None:
continue
try:
if msg.type == web.WSMsgType.TEXT:
try:
data = msg.json()
if data:
await rpc(data)
except json.JSONDecodeError as ex:
logger.warning(f"Invalid JSON received: {safe_str(ex)}")
except Exception as ex:
logger.warning(f"RPC call failed: {safe_str(ex)}")
elif msg.type == web.WSMsgType.BINARY:
logger.debug("Binary message received, ignoring")
elif msg.type == web.WSMsgType.ERROR:
logger.debug(f"WebSocket error: {safe_str(ws.exception())}")
break
elif msg.type == web.WSMsgType.CLOSE:
logger.debug("WebSocket close message received")
break
elif msg.type == web.WSMsgType.CLOSING:
logger.debug("WebSocket closing")
break
elif msg.type == web.WSMsgType.CLOSED:
logger.debug("WebSocket closed")
break
except Exception as ex:
logger.warning(f"Message processing error: {safe_str(ex)}")
except asyncio.CancelledError:
logger.debug("WebSocket handler cancelled")
except ConnectionResetError:
logger.debug("Connection reset by peer")
except Exception as ex:
logger.warning(f"WebSocket loop error: {safe_str(ex)}")
finally:
if rpc:
rpc._is_closed = True
if rpc._finalize_task:
try:
rpc._finalize_task.cancel()
except Exception:
pass
for task in rpc._scheduled_tasks:
try:
if task and not task.done():
task.cancel()
except Exception:
pass
for task in scheduled_tasks:
try:
if task and not task.done():
task.cancel()
except Exception:
pass
try:
if self.services and ws:
await self.services.socket.delete(ws)
except Exception as ex:
logger.debug(f"Socket cleanup failed: {safe_str(ex)}")
try:
if ws and not ws.closed:
await ws.close()
except Exception:
pass
return ws
View Git Blame Copy Permalink