diff --git a/CHANGELOG.md b/CHANGELOG.md index 55bf0ac..ab72c6f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,12 @@ - The Files tab has **My files**, **Shared** and, for admins, **All** tabs instead of a Home screen; the same tabs appear in the Open, Save and Move dialogs - Who can open a workflow now follows its folder only. For a live workflow in a folder you own, *This tab* shows **Folder access** +- Chat keeps its whole history (`chat.jsonl`, `chat_keep` to trim) instead of the last 200 messages. The panel shows the latest 50 and loads older ones as you scroll up, with date dividers. The old `chat.json` is imported once + +### Fixed + +- Switching straight between the RigShare and Files sidebar tabs could leave the previous tab's content on screen + ### Removed - Per-workflow access lists (restrict the folder instead). Lists set on individual workflows before are dropped diff --git a/docs/api.md b/docs/api.md index b2a657c..5f6b9d9 100644 --- a/docs/api.md +++ b/docs/api.md @@ -46,7 +46,8 @@ The panel uses these; they take the same session cookie or API key. | `GET /rigshare/api/status` | Server name and whether setup is needed (public) | | `GET /rigshare/api/me` | The current account | | `GET/POST /rigshare/api/me/keys`, `DELETE /rigshare/api/me/keys/{id}` | Your API keys | -| `GET /rigshare/api/rooms` | Shared workflows | +| `GET /rigshare/api/chat?before=&limit=50` | Older chat messages, oldest first, and whether more exist | +| `GET /rigshare/api/rooms` | Live workflows | | `GET /rigshare/api/room?key=…` | One room, with its workflow JSON | | `GET /rigshare/api/snapshots?room=…` | History of a room | | `GET /rigshare/api/people` | Accounts that can be added to an access list | diff --git a/docs/install.md b/docs/install.md index 43b2f67..2c8a6a1 100644 --- a/docs/install.md +++ b/docs/install.md @@ -51,7 +51,7 @@ The data folder is `/rigshare`; point it elsewhere with `RIGS | `config.json` | Server settings, see [Settings](settings.md) | | `rooms/` | Current state of every shared workflow | | `snapshots/` | Workflow history | -| `chat.json` | Recent chat | +| `chat.jsonl` | The whole chat history, one message per line | Back up the folder to keep accounts and history. Deleting `secret.key` signs everyone out. diff --git a/docs/settings.md b/docs/settings.md index ca58bfc..cd75680 100644 --- a/docs/settings.md +++ b/docs/settings.md @@ -33,7 +33,7 @@ Admins change these under **Admin → Server settings**. They are stored in `rig | `room_expiry_days` | `14` | Forget the live state of workflows nobody opened for this long | | `snapshot_interval_sec` | `300` | Automatic snapshot interval while editing; `0` turns it off | | `snapshot_keep` | `30` | Automatic snapshots kept per workflow; named ones are always kept | -| `chat_history` | `200` | Chat messages kept | +| `chat_keep` | `0` | Chat messages kept: `0` keeps the whole history, a number trims it to the latest ones at startup | ## Environment variables diff --git a/rigshare_core/chatlog.py b/rigshare_core/chatlog.py new file mode 100644 index 0000000..2bd111f --- /dev/null +++ b/rigshare_core/chatlog.py @@ -0,0 +1,100 @@ +"""Chat history: an append-only log, read back one page at a time. + +Messages are stored one JSON object per line in ``chat.jsonl``. Each gets a +sequence number (``seq``, its position in the log), so clients can ask for +"the 50 messages before #1234" while scrolling back. Only the byte offset of +each line is kept in memory, so the history can grow without limit. +""" + +import json +import logging +import os +import threading + +log = logging.getLogger("ComfyUI-RigShare") + + +class ChatLog: + def __init__(self, folder, keep=0, legacy=None): + """``keep`` > 0 trims the log to that many messages at startup (0 keeps everything). + ``legacy``: messages from the old ``chat.json``, imported once.""" + self.path = os.path.join(folder, "chat.jsonl") + self._lock = threading.Lock() + if legacy and not os.path.exists(self.path): + with open(self.path, "w", encoding="utf-8") as f: + for msg in legacy: + if isinstance(msg, dict): + f.write(json.dumps({k: v for k, v in msg.items() if k != "seq"}, ensure_ascii=False) + "\n") + self.offsets = [] + self._index() + if keep and len(self.offsets) > keep: + self._trim(keep) + + def _index(self): + self.offsets = [] + if not os.path.exists(self.path): + return + torn = None + with open(self.path, "rb") as f: + pos = 0 + for line in f: + if not line.endswith(b"\n"): + torn = pos # cut short by a crash mid-write + elif line.strip(): + self.offsets.append(pos) + pos += len(line) + if torn is not None: + with open(self.path, "r+b") as f: + f.truncate(torn) + + def _trim(self, keep): + messages = self.read(len(self.offsets) - keep, len(self.offsets)) + tmp = self.path + ".tmp" + with open(tmp, "w", encoding="utf-8") as f: + for msg in messages: + msg.pop("seq", None) + f.write(json.dumps(msg, ensure_ascii=False) + "\n") + os.replace(tmp, self.path) + self._index() + + def __len__(self): + return len(self.offsets) + + def append(self, msg): + """Store a message; returns it with its ``seq``.""" + with self._lock: + msg = {**msg, "seq": len(self.offsets)} + data = (json.dumps({k: v for k, v in msg.items() if k != "seq"}, ensure_ascii=False) + "\n").encode("utf-8") + try: + with open(self.path, "ab") as f: + pos = f.tell() + f.write(data) + self.offsets.append(pos) + except OSError as e: + log.error(f"[RigShare] could not save chat message: {e}") + return msg + + def read(self, start, end): + """Messages ``start`` <= seq < ``end``.""" + start, end = max(0, start), min(end, len(self.offsets)) + if start >= end: + return [] + out = [] + with open(self.path, "rb") as f: + f.seek(self.offsets[start]) + for seq in range(start, end): + line = f.readline() + try: + msg = json.loads(line) + except ValueError: + continue + msg["seq"] = seq + out.append(msg) + return out + + def page(self, before=None, limit=50): + """The ``limit`` messages before ``before`` (newest when None), oldest first, + and whether older ones exist.""" + end = len(self.offsets) if before is None else max(0, min(int(before), len(self.offsets))) + start = max(0, end - max(1, min(int(limit), 200))) + return self.read(start, end), start > 0 diff --git a/rigshare_core/hub.py b/rigshare_core/hub.py index 76ef5b1..ffdf1e1 100644 --- a/rigshare_core/hub.py +++ b/rigshare_core/hub.py @@ -15,10 +15,10 @@ import json import logging import secrets import time -from collections import deque from aiohttp import web, WSMsgType +from .chatlog import ChatLog from .graphdoc import apply_patch, remap_conflicts from .stats import collect_stats from .workspace import Workspace @@ -28,6 +28,7 @@ log = logging.getLogger("ComfyUI-RigShare") COLORS = ["#f87171", "#fb923c", "#facc15", "#4ade80", "#2dd4bf", "#38bdf8", "#818cf8", "#c084fc", "#f472b6", "#a3e635", "#fbbf24", "#22d3ee"] MAX_MESSAGE = 16 * 1024 * 1024 +CHAT_PAGE = 50 def clean_text(text, limit): @@ -99,10 +100,8 @@ class Hub: data.get("doc"), data.get("version", 0), data.get("updated")) except (KeyError, TypeError): continue - self.chat = deque(maxlen=int(store.config.get("chat_history", 200))) - for msg in store._read("chat.json", []): - self.chat.append(msg) - self.chat_dirty = False + self.chat = ChatLog(store.folder, keep=int(store.config.get("chat_keep") or 0), + legacy=store._read("chat.json", None)) self.stats = None self._task = None self.workspace = None # set by __init__ (folders, private spaces) @@ -131,9 +130,6 @@ class Hub: if expiry > 0 and now - room.updated > expiry and not self.members(room.key): self.rooms.pop(room.key, None) self.store.delete_room(room.key) - if self.chat_dirty: - self.store._write("chat.json", list(self.chat)) - self.chat_dirty = False except Exception as e: log.error(f"[RigShare] housekeeping error: {e}") @@ -384,8 +380,7 @@ class Hub: async def system_chat(self, text): msg = {"id": secrets.token_hex(6), "system": True, "text": text, "ts": int(time.time() * 1000)} - self.chat.append(msg) - self.chat_dirty = True + msg = self.chat.append(msg) await self.broadcast({"type": "chat", "message": msg}) # ----- websocket ---------------------------------------------------- @@ -440,7 +435,7 @@ class Hub: "self": client.public(), "users": self.users_for(client), "rooms": self.room_list(client), - "chat": list(self.chat), + **dict(zip(("chat", "chat_more"), self.chat.page(limit=CHAT_PAGE))), "stats": self.stats, "server": self.server_info(), }) @@ -496,8 +491,7 @@ class Hub: return msg = {"id": secrets.token_hex(6), "from": client.id, "name": client.name, "color": client.color, "text": text, "ts": int(time.time() * 1000)} - self.chat.append(msg) - self.chat_dirty = True + msg = self.chat.append(msg) await self.broadcast({"type": "chat", "message": msg}) elif kind == "rename": diff --git a/rigshare_core/routes.py b/rigshare_core/routes.py index d8830ac..4ee29ce 100644 --- a/rigshare_core/routes.py +++ b/rigshare_core/routes.py @@ -344,6 +344,18 @@ def setup(server, store, hub): return user.get("display_name") or user["username"] return "guest" + @routes.get("/rigshare/api/chat") + async def chat_history(request): + """Older chat messages: ``before`` is the seq of the oldest one already shown.""" + try: + before = request.query.get("before") + before = int(before) if before not in (None, "") else None + limit = int(request.query.get("limit", 50)) + except ValueError: + return error("before and limit must be numbers") + messages, more = hub.chat.page(before, limit) + return web.json_response({"messages": messages, "more": more}) + @routes.get("/rigshare/api/rooms") async def list_rooms(request): return web.json_response([r for r in hub.room_list() if room_access(request, hub.rooms[r["key"]])]) diff --git a/rigshare_core/store.py b/rigshare_core/store.py index a703376..7fb145a 100644 --- a/rigshare_core/store.py +++ b/rigshare_core/store.py @@ -53,7 +53,7 @@ DEFAULT_CONFIG = { "room_expiry_days": 14, "snapshot_interval_sec": 300, "snapshot_keep": 30, - "chat_history": 200, + "chat_keep": 0, # 0 keeps the whole chat; N trims it to the last N messages at startup } diff --git a/tests/python/test_chatlog.py b/tests/python/test_chatlog.py new file mode 100644 index 0000000..e874a83 --- /dev/null +++ b/tests/python/test_chatlog.py @@ -0,0 +1,41 @@ +"""Chat history: append-only log paged by sequence number.""" + +import json +import os + +from rigshare_core.chatlog import ChatLog + + +def test_pages_back_through_the_whole_history(tmp_path): + chat = ChatLog(str(tmp_path)) + for i in range(130): + assert chat.append({"text": f"m{i}", "ts": i})["seq"] == i + page, more = chat.page(limit=50) + assert [m["text"] for m in page] == [f"m{i}" for i in range(80, 130)] and more + seen = page + while more: + page, more = chat.page(before=seen[0]["seq"], limit=50) + seen = page + seen + assert [m["seq"] for m in seen] == list(range(130)), "every message exactly once, in order" + assert chat.page(before=0) == ([], False) + + +def test_survives_restart_a_torn_line_and_imports_old_chat(tmp_path): + folder = str(tmp_path) + chat = ChatLog(folder, legacy=[{"text": "old 1"}, {"text": "old 2"}]) + chat.append({"text": "new"}) + with open(os.path.join(folder, "chat.jsonl"), "ab") as f: + f.write(b'{"text": "torn') # the server died mid-write + chat = ChatLog(folder, legacy=[{"text": "never imported twice"}]) + assert [m["text"] for m in chat.page()[0]] == ["old 1", "old 2", "new"] + chat.append({"text": "after"}) + assert chat.page()[0][-1] == {"text": "after", "seq": 3} + + +def test_optional_trim_and_unicode(tmp_path): + chat = ChatLog(str(tmp_path)) + for i in range(10): + chat.append({"text": f"héllo {i} 🎉"}) + chat = ChatLog(str(tmp_path), keep=4) + assert [m["text"] for m in chat.page()[0]] == [f"héllo {i} 🎉" for i in range(6, 10)] + assert len(chat) == 4 diff --git a/tests/python/test_server.py b/tests/python/test_server.py index bb94ad8..2197b6c 100644 --- a/tests/python/test_server.py +++ b/tests/python/test_server.py @@ -292,3 +292,35 @@ def test_moves_follow_open_tabs(run): for sock in socks.values(): await sock.close() run(scenario()) + + +def test_chat_history_pages(run): + async def scenario(): + async with rig_server() as rig: + await rig.setup_admin() + sock = await rig.ws("boss") + for i in range(70): + await sock.send({"type": "chat", "text": f"hello {i}"}) + await asyncio.sleep(0.3) + await sock.close() + sock = await rig.ws("boss") + await sock.close() + # The welcome carries only the newest page; the rest comes over HTTP. + h = rig.headers["boss"] + session = rig.http + ws = await session.ws_connect(rig.url("/rigshare/ws"), headers=h) + await ws.send_json({"type": "hello"}) + while True: + welcome = json.loads((await ws.receive()).data) + if welcome["type"] == "welcome": + break + await ws.close() + assert len(welcome["chat"]) == 50 and welcome["chat_more"] + oldest = welcome["chat"][0]["seq"] + older = await (await session.get(rig.url(f"/rigshare/api/chat?before={oldest}&limit=50"), headers=h)).json() + texts = [m["text"] for m in older["messages"]] + assert not older["more"] and older["messages"][0]["seq"] == 0 + seqs = [m["seq"] for m in older["messages"] + welcome["chat"]] + assert "hello 0" in texts and seqs == list(range(len(seqs))), "no gaps, no repeats" + assert (await session.get(rig.url("/rigshare/api/chat?before=x"))).status in (400, 401) + run(scenario()) diff --git a/web/rigshare/client.js b/web/rigshare/client.js index 77d1a99..f5ddb8a 100644 --- a/web/rigshare/client.js +++ b/web/rigshare/client.js @@ -28,7 +28,8 @@ export class RigClient extends EventTarget { this.status = "offline"; // offline | connecting | online | denied this.self = null; this.users = []; - this.chat = []; + this.chat = []; // loaded chat messages, oldest first + this.chatMore = false; // older messages exist on the server this.rooms = []; this.stats = null; this.server = {}; @@ -105,6 +106,7 @@ export class RigClient extends EventTarget { this.self = msg.self; this.users = msg.users; this.chat = msg.chat || []; + this.chatMore = !!msg.chat_more; this.rooms = msg.rooms || []; this.stats = msg.stats; this.server = msg.server || {}; @@ -146,7 +148,6 @@ export class RigClient extends EventTarget { break; case "chat": this.chat.push(msg.message); - if (this.chat.length > 500) this.chat.shift(); this.emit("chat", msg.message); break; default: @@ -200,6 +201,18 @@ export class RigClient extends EventTarget { return { "X-RigShare-Guest": this.guestId }; } + /** Fetch the page of chat before the oldest loaded message; returns them (oldest first). */ + async loadOlderChat(limit = 50) { + const oldest = this.chat[0]?.seq; + if (!this.chatMore || oldest === undefined) return []; + const data = await this.request("GET", `/rigshare/api/chat?before=${oldest}&limit=${limit}`); + // A reconnect may have replaced the history meanwhile. + if (this.chat[0]?.seq !== oldest) return []; + this.chat.unshift(...data.messages); + this.chatMore = data.more; + return data.messages; + } + async request(method, path, body) { const base = location.pathname.replace(/[^/]*$/, ""); const res = await fetch(base + path.replace(/^\//, ""), { diff --git a/web/rigshare/main.js b/web/rigshare/main.js index a9bfef1..502f50d 100644 --- a/web/rigshare/main.js +++ b/web/rigshare/main.js @@ -281,7 +281,10 @@ app.registerExtension({ type: "custom", render: (el) => { el.classList.add("rs-host"); - el.append(filesPanel); + // Same slot element as the RigShare tab when switching between them + // (ComfyUI only calls destroy on unmount): take it over cleanly. + if (el.contains(panel.root)) panel.unmount(); + el.replaceChildren(filesPanel); if (!filesLoaded || client.online) { filesLoaded = true; filesBrowser.reload(); } }, }); diff --git a/web/rigshare/panel.js b/web/rigshare/panel.js index 319ca6a..e20833f 100644 --- a/web/rigshare/panel.js +++ b/web/rigshare/panel.js @@ -36,7 +36,8 @@ export class RigSharePanel { this.body = h("div", { class: "rs-body" }); this.root = h("div", { class: "rs-panel" }, this.header, this.nav, this.body); - this.chatLog = h("div", { class: "rs-chat-log" }); + this.chatTop = h("div", { class: "rs-chat-top" }); + this.chatLog = h("div", { class: "rs-chat-log", onscroll: () => this.maybeLoadOlder() }); this.chatInput = h("textarea", { class: "rs-input rs-chat-input", placeholder: "Message… (Enter to send)", maxLength: 1000, rows: 1, onkeydown: (e) => { @@ -63,8 +64,11 @@ export class RigSharePanel { mount(container) { container.classList.add("rs-host"); - container.append(this.root); + // ComfyUI reuses the same slot element when switching straight between + // two custom sidebar tabs: replace whatever the other tab left there. + container.replaceChildren(this.root); this.visible = true; + this.renderedView = null; // reopened: the chat starts at the newest message if (this.view === "chat") this.clearUnread(); this.render(); } @@ -192,9 +196,14 @@ export class RigSharePanel { }; if ((this.view === "admin" && !c.perms.admin) || !VIEWS.some((v) => v.id === this.view) && this.view !== "account") this.view = "workflows"; const scroll = this.body.scrollTop; + // Re-attaching the chat resets its scroll: keep the reader's place in the history. + const stay = this.view === "chat" && this.renderedView === "chat"; + const chatScroll = this.chatLog.scrollTop; this.body.replaceChildren(...(views[this.view] ?? views.workflows)().filter(Boolean)); this.body.scrollTop = scroll; - if (this.view === "chat") this.scrollChat(); + if (stay) this.chatLog.scrollTop = chatScroll; + else if (this.view === "chat") this.scrollChat(); + this.renderedView = this.view; } section(icon, title, ...children) { @@ -447,6 +456,71 @@ export class RigSharePanel { if (this.client.send({ type: "chat", text })) this.chatInput.value = ""; } + dayLabel(ts) { + const d = new Date(ts); + const today = new Date(); + const yesterday = new Date(today.getTime() - 86400000); + if (d.toDateString() === today.toDateString()) return "Today"; + if (d.toDateString() === yesterday.toDateString()) return "Yesterday"; + return d.toLocaleDateString([], { weekday: "short", year: "numeric", month: "short", day: "numeric" }); + } + + /** Chat lines for ``messages``, with a date divider wherever the day changes after ``prev``. */ + chatLines(messages, prev) { + const out = []; + for (const m of messages) { + if (!prev || new Date(prev.ts).toDateString() !== new Date(m.ts).toDateString()) { + out.push(h("div", { class: "rs-chat-day" }, h("span", {}, this.dayLabel(m.ts)))); + } + out.push(this.chatLine(m)); + prev = m; + } + return out; + } + + renderChatTop() { + const c = this.client; + this.chatTop.replaceChildren(this.loadingOlder + ? h("span", { class: "rs-muted rs-small" }, h("i", { class: "pi pi-spin pi-spinner" }), " Loading older messages…") + : c.chatMore + ? h("button", { class: "rs-btn rs-btn-sm rs-btn-ghost", onclick: () => this.loadOlder() }, "Load older messages") + : c.chat.length ? h("span", { class: "rs-muted rs-small" }, "Beginning of the chat") : null); + } + + maybeLoadOlder() { + if (this.chatLog.scrollTop < 120 && this.client.chatMore && !this.loadingOlder) this.loadOlder(); + } + + /** Infinite scroll back: prepend the previous page and keep the view where it was. */ + async loadOlder() { + if (this.loadingOlder || !this.client.chatMore) return; + this.loadingOlder = true; + this.renderChatTop(); + try { + const next = this.client.chat.find((m) => m.seq !== undefined); + const older = await this.client.loadOlderChat(); + if (older.length) { + const log = this.chatLog; + const before = log.scrollHeight - log.scrollTop; + // The first message's date divider is redrawn with the older page. + const firstDay = this.chatTop.nextSibling; + if (firstDay?.classList?.contains("rs-chat-day") && next + && new Date(older.at(-1).ts).toDateString() === new Date(next.ts).toDateString()) firstDay.remove(); + this.chatTop.after(...this.chatLines(older)); + log.scrollTop = log.scrollHeight - before; + } + } catch (e) { + this.toast("error", e.message || String(e)); + } finally { + this.loadingOlder = false; + this.renderChatTop(); + } + // Still not enough to scroll: keep going. + requestAnimationFrame(() => { + if (this.chatLog.scrollHeight <= this.chatLog.clientHeight + 120) this.maybeLoadOlder(); + }); + } + chatLine(m) { if (m.system) return h("div", { class: "rs-msg rs-msg-system" }, h("span", { class: "rs-time" }, clock(m.ts)), m.text); const mine = m.from === this.client.self?.id; @@ -458,8 +532,12 @@ export class RigSharePanel { } renderChatLog() { - this.chatLog.replaceChildren(...this.client.chat.slice(-200).map((m) => this.chatLine(m))); + this.chatLog.replaceChildren(this.chatTop, ...this.chatLines(this.client.chat)); + this.renderChatTop(); this.scrollChat(); + requestAnimationFrame(() => { + if (this.chatLog.isConnected && this.chatLog.scrollHeight <= this.chatLog.clientHeight + 120) this.maybeLoadOlder(); + }); } scrollChat() { @@ -468,8 +546,9 @@ export class RigSharePanel { onChat(m) { const atBottom = this.chatLog.scrollHeight - this.chatLog.scrollTop - this.chatLog.clientHeight < 40; - this.chatLog.append(this.chatLine(m)); - while (this.chatLog.childElementCount > 300) this.chatLog.firstChild.remove(); + const chat = this.client.chat; + this.chatLog.append(...this.chatLines([m], chat[chat.length - 2])); + if (chat.length === 1) this.renderChatTop(); if (atBottom || m.from === this.client.self?.id) this.scrollChat(); const seen = this.visible && this.view === "chat" && !document.hidden; if (!seen && !m.system && m.from !== this.client.self?.id) { diff --git a/web/rigshare/styles.js b/web/rigshare/styles.js index be36490..677e471 100644 --- a/web/rigshare/styles.js +++ b/web/rigshare/styles.js @@ -155,6 +155,9 @@ body.rs-no-manager [aria-label="ComfyUI Manager"] { display: none !important; } .rs-msg.mine .rs-msg-text { border-radius: 10px 4px 10px 10px; background: color-mix(in srgb, var(--rs-accent) 22%, var(--rs-bg2)); margin-left: auto; } .rs-msg-system { align-self: center; color: var(--rs-muted); font-size: 11px; font-style: italic; } .rs-time { color: var(--rs-muted); font-size: 10px; font-variant-numeric: tabular-nums; margin-right: 4px; } +.rs-chat-top { display: flex; justify-content: center; align-items: center; min-height: 28px; flex: none; } +.rs-chat-day { display: flex; align-items: center; gap: 8px; color: var(--rs-muted); font-size: 11px; flex: none; } +.rs-chat-day::before, .rs-chat-day::after { content: ""; flex: 1; border-top: 1px solid var(--rs-border); } .rs-chat-compose { display: flex; gap: 6px; align-items: flex-end; border-top: 1px solid var(--rs-border); padding-top: 8px; } .rs-chat-input { resize: none; max-height: 120px; min-height: 32px; }