diff options
Diffstat (limited to 'packages/meshbay-node/src')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/chat/store.py | 43 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/quic_server.py | 5 |
2 files changed, 48 insertions, 0 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/chat/store.py b/packages/meshbay-node/src/meshbay_node/chat/store.py index a5b2d7d..6f23905 100644 --- a/packages/meshbay-node/src/meshbay_node/chat/store.py +++ b/packages/meshbay-node/src/meshbay_node/chat/store.py @@ -45,6 +45,11 @@ class StoredMessage: sender_name: str = "" +def _row(r) -> StoredMessage: + return StoredMessage(id=r[0], sender_id=r[1], iteration=r[2], payload=r[3], + timestamp=r[4], thread_id=r[5], sender_name=r[6] or "") + + class ChatStore: """Async SQLite chat store for one group.""" @@ -111,6 +116,44 @@ class ChatStore: for r in rows ] + async def get_recent(self, limit: int = 100) -> list[StoredMessage]: + """The newest `limit` messages, oldest first so they render in order. + + `get_messages(since=0, limit=n)` cannot answer this: it pages *forward* + from the oldest, so a client opening a group with 300 messages was shown + the first 200 and never the conversation it came for. Reaching the newest + rows means ordering DESC in SQL and reversing here. + """ + cursor = await self._db.execute( + "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name " + "FROM messages ORDER BY id DESC LIMIT ?", + (limit,), + ) + rows = await cursor.fetchall() + return [_row(r) for r in reversed(rows)] + + async def get_before(self, before_id: int, limit: int = 50) -> list[StoredMessage]: + """The `limit` messages immediately older than `before_id`, oldest first. + + The cursor is the row id, not the timestamp. Nothing makes `timestamp` + unique — it is a float from `time.time()` — and a cursor on a value two + rows can share is one that eventually skips a message or repeats it. + `id` is AUTOINCREMENT: unique, and ordered by insertion. + """ + cursor = await self._db.execute( + "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name " + "FROM messages WHERE id < ? ORDER BY id DESC LIMIT ?", + (before_id, limit), + ) + rows = await cursor.fetchall() + return [_row(r) for r in reversed(rows)] + + async def has_before(self, message_id: int) -> bool: + """Whether anything older than `message_id` exists — drives "load older".""" + cursor = await self._db.execute( + "SELECT 1 FROM messages WHERE id < ? LIMIT 1", (message_id,)) + return await cursor.fetchone() is not None + async def get_thread(self, thread_id: str, limit: int = 100) -> list[StoredMessage]: """Get messages in a thread.""" cursor = await self._db.execute( diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py index ed3925d..73e668c 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py @@ -203,6 +203,11 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._do_stream_segment_sync(stream_id, msg) elif mtype == MNP.CHAT_MESSAGE: self._do_chat_message_sync(stream_id, msg) + elif mtype == MNP.PING: + # Liveness is transport-agnostic, and a native client over QUIC + # has the same half-open problem a DataChannel does. + self._send(stream_id, {"type": MNP.PONG, "v": MNP_VERSION, + "token": msg.get("token")}) else: log.warning("Unknown MNP message type: %s", mtype) except Exception as e: |