diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/chat/store.py | 39 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 34 |
2 files changed, 59 insertions, 14 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/chat/store.py b/packages/meshbay-node/src/meshbay_node/chat/store.py index 1dbcc2b..a5b2d7d 100644 --- a/packages/meshbay-node/src/meshbay_node/chat/store.py +++ b/packages/meshbay-node/src/meshbay_node/chat/store.py @@ -17,17 +17,22 @@ log = logging.getLogger(__name__) _SCHEMA = """ CREATE TABLE IF NOT EXISTS messages ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - sender_id TEXT NOT NULL, - iteration INTEGER NOT NULL, - payload BLOB NOT NULL, - timestamp REAL NOT NULL, - thread_id TEXT DEFAULT NULL + id INTEGER PRIMARY KEY AUTOINCREMENT, + sender_id TEXT NOT NULL, + iteration INTEGER NOT NULL, + payload BLOB NOT NULL, + timestamp REAL NOT NULL, + thread_id TEXT DEFAULT NULL, + sender_name TEXT DEFAULT '' ); CREATE INDEX IF NOT EXISTS idx_messages_ts ON messages(timestamp); CREATE INDEX IF NOT EXISTS idx_messages_thread ON messages(thread_id); """ +_MIGRATE_SENDER_NAME = ( + "ALTER TABLE messages ADD COLUMN sender_name TEXT DEFAULT ''" +) + @dataclass class StoredMessage: @@ -37,6 +42,7 @@ class StoredMessage: payload: bytes timestamp: float thread_id: str | None + sender_name: str = "" class ChatStore: @@ -50,6 +56,10 @@ class ChatStore: self._db_path.parent.mkdir(parents=True, exist_ok=True) self._db = await aiosqlite.connect(str(self._db_path)) await self._db.executescript(_SCHEMA) + try: + await self._db.execute(_MIGRATE_SENDER_NAME) + except Exception: + pass await self._db.commit() async def close(self) -> None: @@ -70,13 +80,14 @@ class ChatStore: iteration: int, payload: bytes, thread_id: str | None = None, + sender_name: str = "", ) -> int: """Store a message. Returns the row id.""" ts = time.time() cursor = await self._db.execute( - "INSERT INTO messages (sender_id, iteration, payload, timestamp, thread_id) " - "VALUES (?, ?, ?, ?, ?)", - (sender_id, iteration, payload, ts, thread_id), + "INSERT INTO messages (sender_id, iteration, payload, timestamp, thread_id, sender_name) " + "VALUES (?, ?, ?, ?, ?, ?)", + (sender_id, iteration, payload, ts, thread_id, sender_name), ) await self._db.commit() return cursor.lastrowid @@ -88,28 +99,30 @@ class ChatStore: ) -> list[StoredMessage]: """Get messages after a timestamp, most recent last.""" cursor = await self._db.execute( - "SELECT id, sender_id, iteration, payload, timestamp, thread_id " + "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name " "FROM messages WHERE timestamp > ? ORDER BY timestamp ASC LIMIT ?", (since, limit), ) rows = await cursor.fetchall() return [ StoredMessage(id=r[0], sender_id=r[1], iteration=r[2], - payload=r[3], timestamp=r[4], thread_id=r[5]) + payload=r[3], timestamp=r[4], thread_id=r[5], + sender_name=r[6] or "") for r in rows ] async def get_thread(self, thread_id: str, limit: int = 100) -> list[StoredMessage]: """Get messages in a thread.""" cursor = await self._db.execute( - "SELECT id, sender_id, iteration, payload, timestamp, thread_id " + "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name " "FROM messages WHERE thread_id = ? ORDER BY timestamp ASC LIMIT ?", (thread_id, limit), ) rows = await cursor.fetchall() return [ StoredMessage(id=r[0], sender_id=r[1], iteration=r[2], - payload=r[3], timestamp=r[4], thread_id=r[5]) + payload=r[3], timestamp=r[4], thread_id=r[5], + sender_name=r[6] or "") for r in rows ] diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index 6231310..15c8f66 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -116,6 +116,8 @@ class WebRTCPeerSession: self._do_chat_history(msg) elif mtype == MNP.FILE_UPLOAD: self._do_file_upload(msg) + elif mtype == MNP.FILE_DELETE: + self._do_file_delete(msg) else: log.warning("Unknown MNP message type on DataChannel: %s", mtype) except Exception as e: @@ -276,6 +278,7 @@ class WebRTCPeerSession: iteration=msg.get("iteration", 0), payload=raw, thread_id=msg.get("thread_id"), + sender_name=sender_name, )) peers = self._ctx.get("_peers", {}) @@ -321,7 +324,7 @@ class WebRTCPeerSession: { "id": m.id, "sender_id": m.sender_id, - "sender_name": names.get(m.sender_id, ""), + "sender_name": m.sender_name or names.get(m.sender_id, ""), "payload": m.payload.decode("utf-8", errors="replace") if isinstance(m.payload, bytes) else m.payload, "timestamp": m.timestamp, @@ -373,6 +376,35 @@ class WebRTCPeerSession: tmp_path.rename(final_path) log.info("Upload complete: %s (%d chunks)", safe_name, total_chunks) + def _do_file_delete(self, msg: dict) -> None: + ctx = self._group_ctx() + file_id = msg.get("file_id", "") + if not file_id: + self._send({"type": "error", "detail": "Missing file_id"}) + return + + node_user_id = self._ctx.get("node_user_id") + if node_user_id and self._user_id != node_user_id: + self._send({"type": "error", "detail": "Only node admin can delete files"}) + return + + entry = ctx["index"].get_entry(file_id) + if not entry: + self._send({"type": "error", "detail": "File not found"}) + return + + file_path = ctx["shared_root"] / entry.path / entry.name + if file_path.exists(): + file_path.unlink() + log.info("File deleted: %s", entry.name) + + ctx["index"].remove_entry(file_id) + self._send({ + "type": MNP.FILE_DELETE_ACK, + "v": MNP_VERSION, + "file_id": file_id, + }) + def _send(self, obj: dict) -> None: if self._channel and self._channel.readyState == "open": self._channel.send(_pack(obj)) |