diff --git a/apps/web/content/docs/content-engine/admincp.mdx b/apps/web/content/docs/content-engine/admincp.mdx
index 93169a884..0678f2c82 100644
--- a/apps/web/content/docs/content-engine/admincp.mdx
+++ b/apps/web/content/docs/content-engine/admincp.mdx
@@ -103,6 +103,12 @@ The **More actions** menu of each row collects the less frequent actions. Two of
Each action asks for confirmation, refreshes the list and shows a toast. On the edit page, `ContentFormActions` adds a **Hide** or **Unhide** button next to **Save changes**, and `ContentFormStatus` shows the same **Hidden** badge with a short explanation.
+## Live updates
+
+Every generated list stays current while it's open, with no configuration. When a record of the same type changes in another tab or session, such as another admin's edit, a member's post or a scheduled publish, the page on screen reads again with its filters, search, sort and page. Selected rows that are still on the page stay ticked, and a burst of changes costs one read. If that read fails, the list keeps its rows and shows **Could not refresh this list** with **Try again**.
+
+Live updates need the app to serve the [WebSocket](/docs/dev/websocket/setup). [Live refresh](/docs/content-engine/live-refresh) explains who receives what.
+
## Group form fields into sections
By default the form shows every field except `field.blocks()`, one after another, in a dialog:
diff --git a/apps/web/content/docs/content-engine/comments-api.mdx b/apps/web/content/docs/content-engine/comments-api.mdx
index 822dc5241..77137ca5e 100644
--- a/apps/web/content/docs/content-engine/comments-api.mdx
+++ b/apps/web/content/docs/content-engine/comments-api.mdx
@@ -33,6 +33,8 @@ const comments = useContentComments({ queryKey, thread, ...actions })
| `record` | `ZodType` | No | Validates every loaded comment's public record and gives it a type. The thread route types records loosely, so pass it when your own rendering reads their fields |
| `field` | `string` | No | The rich-text field with the comment's text. Defaults to `"body"` |
| `onChanged` | `(queryClient) => void \| Promise` | No | Refreshes your other queries after a successful write, such as a comment count elsewhere on the page |
+| `branches` | `(ids) => Promise` | No | [`GET /thread/branches`](#get-threadbranches) with `root`, plus `ids` joined by commas only when there are any. Writes then re-read only the branch they touched instead of every loaded page. `live` needs it |
+| `live` | `{ contentTypeId, rootId }` | No | The comment content type's `id` and the root record's id. Keeps the thread current over the WebSocket while it is mounted; ignored without `branches`. See [Live comments](/docs/content-engine/threads-live) |
### Callback contract
@@ -48,16 +50,20 @@ const comments = useContentComments({ queryKey, thread, ...actions })
| ---------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `tree` | The loaded comments as a tree: `ContentCommentTreeNode[]`, replies in `children` |
| `items` | The same entries, flat and depth first |
-| `status` | `"pending"` before the first page, `"error"` when it failed, then `"success"` |
+| `status` | `"pending"` before the first page, `"error"` when it failed, then `"success"` - also after a failed background refresh, which `refreshError` reports |
| `hasMore`, `loadMore()`, `isLoadingMore` | The next page of top-level comments. `loadMore` resolves to `false` when it failed |
-| `isRefreshing`, `retry()` | A refetch over data on screen, and retrying a failed first load |
+| `isRefreshing`, `retry()` | A refetch over data on screen (not **Load more**), and retrying a failed first load |
+| `refreshError`, `retryRefresh()` | A background read of the comments on screen failed, and they stayed as they were. `retryRefresh()` asks again for what it missed and resolves when done |
+| `hasNewComments`, `showNewComments()` | With `live`: comments appeared after the last loaded one, at the end of the thread. `showNewComments()` reads only those and adds them below; it resolves to `false` when that failed |
+| `liveStatus` | `"off"` (no `live`, or no open WebSocket), `"joining"`, `"live"`, or `"denied"` when the server refused or revoked the subscription |
+| `trackDraft(id, open)` | Call with `true` while an edit or reply form is open on `id` and `false` when it closes. A live update that removes that comment keeps it on screen with `removed: true` and its record until then. `useContentCommentItem` calls it for you |
| `replyCount`, `truncated` | Every public comment at any depth, and whether a page held more replies than one answer carries |
| `permissionsOf(id)` | `{ canDelete, canEdit, canHide, canReport }`, all `false` until known or after a failed batch |
| `canComment`, `canReply(node)` | Whether the composer and **Reply** show: signed in, `create` given, under `maxDepth`, not hidden |
| `features` | Which actions you wired up |
| `busy` | The action running right now, `{ action, id }` or `null`. A second action waits for it |
-| `remove(id)`, `hide(id, reason)`, `unhide(id)` | Run an action and refresh after success. Resolve to `{ ok: true }` or `{ ok: false, status }`; `status` is `409` when another action was still running and `0` for a network failure |
-| `refresh()` | Refreshes the thread, its permissions and `onChanged`'s queries |
+| `remove(id)`, `hide(id, reason)`, `unhide(id)` | Run an action and refresh after success - only that comment's branch with `branches`. Resolve to `{ ok: true }` or `{ ok: false, status }`; `status` is `409` when another action was still running and `0` for a network failure |
+| `refresh()` | Re-reads every loaded page, its permissions and `onChanged`'s queries |
| `userId` | The signed-in visitor's id, or `null` |
### Thread entries
@@ -158,9 +164,10 @@ const MyComment = ({ node }) => {
| Export | What it is |
| ------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `ContentCommentsRoot` | Provides `comments` to the components below; takes a `fallback` shown while the messages load |
-| `useContentCommentItem(node)` | One comment's state: `permissions` (the server's answers limited to wired actions), `canReply`, `editing`, `startEditing()`, `stopEditing()`, `isReplying`, `setIsReplying`, `isBusy` |
+| `useContentCommentItem(node)` | One comment's state: `permissions` (the server's answers limited to wired actions), `canReply`, `editing`, `startEditing()`, `stopEditing()`, `isReplying`, `setIsReplying`, `isBusy`, and `isRemoved` when a live update removed the comment under an open form |
| `ContentCommentActions` | The **More actions** menu with the hide-reason prompt. `children` adds your own menu items |
-| `ContentCommentComposer` | The comment form, lazy-loaded: `mode="create"` with `parentId`, or `mode="edit"` with `id` and `editable` |
+| `ContentCommentComposer` | The comment form, lazy-loaded: `mode="create"` with `parentId`, or `mode="edit"` with `id` and `editable`. `unavailable` (a translated reason) keeps the text but hides the submit button |
+| `ContentCommentsLiveStatus` | **Show new comments** and the refresh-failed notice, with a polite screen reader announcement. Renders nothing while there is nothing to say |
| `ContentCommentItem` | One comment as the default section draws it |
| `ContentCommentPlaceholder` | The anonymous stand-in for a removed comment |
| `ContentCommentsComposer`, `ContentCommentsLoadMore` | The new-comment card and the load-more button |
@@ -173,7 +180,30 @@ The Blog plugin's `plugins/blog/src/pages/post-comments.tsx` is a complete custo
- **Writes.** The composer posts through [`useContentPublicWrite`](/docs/content-engine/public-write-form-api): a refused write keeps the text and shows the reason, and with `editorial` a stale edit shows the version-conflict alert. A success reads the answer's `status` and `approval`: **Comment posted** only when it is public, **saved as a draft** or **waiting for a moderator** otherwise.
- **Actions.** Delete, hide and unhide show **Comment deleted**, **Comment hidden** or **Comment visible again** after the server accepts them. When it refuses, the section shows one message for the cause: signed out, not allowed, gone, busy or unknown.
-- **Loading.** A failed first load shows **Try again**; a failed **Load more** keeps the loaded comments and says it failed.
+- **Loading.** A failed first load shows **Try again**; a failed **Load more** keeps the loaded comments and says it failed. A failed background refresh keeps them too; `ContentCommentsLiveStatus` offers **Try again**.
+
+## `GET /thread/branches`
+
+Every content type with `thread` also gets `GET /{publicApi.path}/thread/branches`. It re-reads a few top-level comments already on screen, each with every reply below it, using the same root check, visibility, placeholders, depth and truncation as `/thread`. `useContentComments` calls it through `branches`.
+
+```txt
+GET /api/newsroom/content/article-comments/thread/branches?root=1&ids=4,9
+```
+
+| Query | Required | Description |
+| ------ | -------- | -------------------------------------------------------------------------------------------- |
+| `root` | Yes | The root record's id |
+| `ids` | No | Up to 10 top-level comment ids, joined by commas. Duplicates are ignored. Without it, only the counts are read |
+
+| Response field | Description |
+| ---------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------- |
+| `branches` | One entry per requested id, in the order asked: `{ id, status: "ok", items }` or `{ id, status: "gone" }` |
+| `items` | The top-level comment first, then every reply below it, as [thread entries](#thread-entries) |
+| `pageInfo.totalCount`, `replyCount`, `truncated` | The thread's current counts, as on `/thread` |
+
+`gone` means this reader no longer sees the branch: deleted, hidden or unpublished with no visible reply left, not a top-level comment, or under another root. An unreadable root answers `404`, as on `/thread`; more than 10 or malformed ids answer `400`.
+
+Guests and readers who can't see hidden comments get the same answer, served from the [public read cache](/docs/content-engine/caching) with `Cache-Control: no-cache`. Readers who may see hidden comments, and readers of a hidden root, get a `private, no-store` answer.
## Related
diff --git a/apps/web/content/docs/content-engine/live-refresh.mdx b/apps/web/content/docs/content-engine/live-refresh.mdx
new file mode 100644
index 000000000..ebc915edf
--- /dev/null
+++ b/apps/web/content/docs/content-engine/live-refresh.mdx
@@ -0,0 +1,146 @@
+---
+title: Live refresh
+description: How VitNode Content Engine keeps AdminCP lists and comment threads current over the shared WebSocket. Covers which writes send a signal, who may subscribe, reconnects, several server instances and custom write routes.
+icon: RefreshCw
+---
+
+Live refresh keeps screens that show Content Engine records current without a reload. Every generated AdminCP list takes part on its own, and a comment section joins in with [Live comments](/docs/content-engine/threads-live). Both use the app's one [WebSocket](/docs/dev/websocket) connection; nothing opens a second one.
+
+The WebSocket only carries small hints, such as "this list changed" or "this branch of this thread changed". Each open screen then reads the data again over HTTP with the viewer's own session. That costs one extra request per change, and buys a simple rule: what a reader sees always comes from the same routes and checks as a first load.
+
+## AdminCP lists
+
+Every generated list joins its content type's room while it's on screen, with no configuration. When a record of that type changes in another tab or session (created, edited, published, hidden, deleted, duplicated, scheduled, or one of its translations changed), the list:
+
+- reads the page on screen again, with its filters, search, sorting and page number, and keeps the old rows on screen meanwhile;
+- forgets the cached pages nobody is looking at, so opening one later reads it fresh;
+- keeps selected rows that are still on the page and drops the ones that left it;
+- collects a burst of changes into one read, 0.4 seconds after the last change and never more than 2 seconds after the first;
+- shows **Could not refresh this list** with **Try again** when that read fails, and keeps the rows.
+
+A delete can push the current page past the end. The list route answers with the last page instead, so the list doesn't end up empty.
+
+## What sends a signal
+
+Signals come from `emitContentEvent`, which every generated route and service effect calls once a write has committed. It works in this order:
+
+1. The transaction commits. A write that rolls back never gets further.
+2. The content event is emitted, and the content type's cached pages and public reads are expired.
+3. The signals are published.
+
+So whoever reads again because of a signal gets the new state, never a cached copy.
+
+| Room | Who may join | Sent for |
+| --- | --- | --- |
+| AdminCP list of one content type | An AdminCP session with the content type's `can_view`, the permission of the list route | Every content event of that type |
+| Thread of one comment type under one root record | Anyone who may read that thread: guests for a public article; authors, editors and moderators for a hidden one | Changes to comments that were or are public, naming the top-level branch they sit in |
+| The same thread room | The same | The root record being hidden, unhidden, published, unpublished, restored or deleted |
+
+A thread signal never carries text, authors or reasons. It is sent only for a comment that was or is public, so a draft, a comment waiting for a moderator, or an edit of a hidden comment sends nothing to the thread. The AdminCP list still hears about it.
+
+### Custom write routes
+
+A custom route that writes through `model.service(c)` or `model.editorialService(c)` emits nothing by itself, so open screens don't hear about it. After the write commits, call `contentEditorialEffects` for an editorial write, or `emitContentEvent` for a plain one, as the generated routes do. Pass the row whenever you have it. For a delete of a threaded record it's required, because the deleted record can't be read back to find its thread:
+
+```ts title="plugins/newsroom/src/api/modules/comments/routes/remove.route.ts (excerpt)"
+import {
+ contentPublicStateOf,
+ emitContentEvent,
+} from "@vitnode/core/content/server";
+
+import { commentContentType } from "@/content/comment";
+import { commentContent } from "@/database/comments";
+import { manifest } from "@/manifest.gen";
+
+// Inside your route handler, after the write:
+const row = await commentContent.service(c).delete(id);
+if (row) {
+ await emitContentEvent(
+ c,
+ commentContentType,
+ "deleted",
+ { contentId: row.id },
+ {
+ pluginId: manifest.id,
+ publicState: contentPublicStateOf(commentContentType, row, null),
+ row,
+ },
+ );
+}
+```
+
+For a change that has no content event, `publishContentRefresh(c, { action, contentId, definition, publicState, row })` from the same import sends only the signals. Call it after the caches are expired.
+
+## Who stays subscribed
+
+A room name is only an address. The server checks every join with the rules above and answers `FORBIDDEN` or `NOT_FOUND` without adding the connection. After that:
+
+- **A revoked session** (signing out a device, a password change or reset, an admin revoke) closes the user's WebSocket connections on every instance. **Signing out or in** within a tab reconnects the shared connection under the new identity. Either way, the tabs join their rooms again and the server checks them again.
+- **A lost permission or an ended session**, including an AdminCP session that expired or signed out, is caught within about 80 seconds: the server checks every subscription again once it's a minute old. A thread subscription opened while signed in fails that check once its session is gone, even on a public article.
+- **A thread root that becomes hidden, unpublished or deleted** is checked at once. Readers who lost access are removed from the room, and their open thread reads again, gets `404` and clears.
+
+Tabs of one browser share the connection, but each tab joins for itself and acts only on its own confirmed joins. A tab with the comments closed doesn't react to a signal another tab asked for.
+
+## Reconnects and missed changes
+
+Delivery is best effort and nothing is replayed. A signal sent while a tab was disconnected is lost, so every join is followed by one read that covers the gap:
+
+- an AdminCP list reads its page once;
+- a comment section reads its loaded pages once.
+
+After a reconnect, that read waits a random moment of up to 1.5 seconds, so a server restart doesn't make every open tab read at the same time.
+
+## Several server instances
+
+With `REDIS_URL` set, a signal published on one instance reaches subscribers on every instance, and a root change re-checks their rooms there too. Without Redis, only tabs connected to the instance that handled the write hear about it, which is fine for a single server. See [Rooms on several instances](/docs/dev/websocket/rooms#rooms-on-several-instances).
+
+## Cost of a busy thread
+
+A popular article can have hundreds of readers with its thread open, and every signal reaches all of them, so each one has to stay cheap. Per open section:
+
+| Event | Requests |
+| --- | --- |
+| A change in a loaded branch | One `GET /thread/branches` for that branch |
+| Five quick changes to the same branch | One, within 2 seconds |
+| Changes to several loaded branches at once | One per 10 branches, at most 2 running at once |
+| A new top-level comment | None until **Show new comments**, then one `GET /thread` for what follows the last loaded comment |
+| A change in a branch that isn't loaded | None |
+| A join or reconnect | One read of every loaded page |
+
+Guests and readers without hidden access all get the same branch answer, so the API serves it from its [public read cache](/docs/content-engine/caching), which the write expired before the signal went out. Readers who may see hidden comments get a private answer that is never shared.
+
+Limits: one connection holds at most 64 subscriptions across all of its tabs and may send 120 joins a minute, and `/thread/branches` reads at most 10 branches per request. These are limits in the code, not benchmark results.
+
+## Follow a scope from your own screen
+
+`useContentLiveRefresh` from `@vitnode/core/tanstack/content` is the hook behind both features. Use it when your own screen shows Content Engine data, such as a custom admin page. Pass a scope and re-read on every signal and every join:
+
+```tsx title="plugins/newsroom/src/views/story-board.tsx (excerpt)"
+import { useContentLiveRefresh } from "@vitnode/core/tanstack/content";
+
+import { storyContentType } from "@/content/story";
+
+// Inside the component, next to the query that loads the board:
+const status = useContentLiveRefresh(
+ { contentTypeId: storyContentType.id, kind: "admin" },
+ {
+ onSignal: () => void refetch(),
+ onSubscribed: () => void refetch(),
+ },
+);
+```
+
+`refetch` is the refetch of the screen's own TanStack Query. A scope is `{ contentTypeId, kind: "admin" }` or `{ contentTypeId, kind: "thread", rootId }`, with the same join rules as the table above. Pass `null` to stay off. The hook returns `"off"` (no scope or no open connection), `"joining"`, `"live"` or `"denied"` (refused or revoked), and calls `onDenied` when the server refuses or revokes the subscription.
+
+## Limitations
+
+- Signals are hints, not a durable log. A reader who was offline catches up on the next join, not change by change.
+- Moderators don't get live updates for changes the public can't see, such as a new comment waiting for approval. They see them on the next read.
+- A subscription the server refused or revoked stays off until the connection reconnects or the screen is opened again.
+- The Forum plugin draws its own topic thread and doesn't use live comments yet.
+
+## Related
+
+- [Live comments](/docs/content-engine/threads-live): turn it on for a comment section.
+- [Events](/docs/content-engine/events): the content events behind the signals.
+- [Reconnects](/docs/dev/websocket/reconnects): what a WebSocket reconnect keeps and loses.
diff --git a/apps/web/content/docs/content-engine/meta.json b/apps/web/content/docs/content-engine/meta.json
index 13ec5f095..ab1294791 100644
--- a/apps/web/content/docs/content-engine/meta.json
+++ b/apps/web/content/docs/content-engine/meta.json
@@ -23,6 +23,7 @@
"threads",
"threads-editing",
"threads-frontend",
+ "threads-live",
"notifications",
"---Organization---",
"relations",
@@ -52,6 +53,7 @@
"admincp",
"services",
"events",
+ "live-refresh",
"---Deployment---",
"database",
"caching",
diff --git a/apps/web/content/docs/content-engine/threads-live-new-comments.webp b/apps/web/content/docs/content-engine/threads-live-new-comments.webp
new file mode 100644
index 000000000..616af7e7d
Binary files /dev/null and b/apps/web/content/docs/content-engine/threads-live-new-comments.webp differ
diff --git a/apps/web/content/docs/content-engine/threads-live.mdx b/apps/web/content/docs/content-engine/threads-live.mdx
new file mode 100644
index 000000000..ca2b08b98
--- /dev/null
+++ b/apps/web/content/docs/content-engine/threads-live.mdx
@@ -0,0 +1,122 @@
+---
+title: Live comments
+description: Keep a VitNode Content Engine comment section current while people read it. Add branches and live to useContentComments, and replies, edits and removals from other readers arrive over the shared WebSocket without a reload and without losing a half-written reply.
+icon: Radio
+implementation:
+ objective: Make an existing shared comment section update live over the app's WebSocket.
+ decisions:
+ - Which comment section should update live? Offer the components in the project that call `useContentComments`.
+ constraints:
+ - Add `branches` (`GET /thread/branches` with `root`, plus `ids` joined by commas only when there are any) and `live` (the comment content type's `id` and the root record's id) to the existing `useContentComments` call; `live` does nothing without `branches`.
+ - "The app must serve the WebSocket (`vitnode({ webSocket: true })` or a standalone API); without it the section keeps working, just without live updates."
+ - Set `REDIS_URL` when the API runs on more than one instance.
+ related:
+ - title: Live refresh
+ href: /docs/content-engine/live-refresh
+ - title: Comments API
+ href: /docs/content-engine/comments-api
+ - title: WebSocket setup
+ href: /docs/dev/websocket/setup
+---
+
+import { ImgDocs } from "@/components/fumadocs/img";
+import newComments from "./threads-live-new-comments.webp";
+
+Live comments keep a comment section current while people read it. When someone replies, edits, hides or deletes a comment, every open copy of the article re-reads only the branch that changed. New top-level comments wait behind a **Show new comments** button, because nobody enjoys a page that rearranges itself mid-sentence. Loaded pages, open reply forms and the reading position stay where they were.
+
+## Before you begin
+
+- The working comment section from [Comments](/docs/content-engine/threads). [Replies](/docs/content-engine/threads-frontend) and [Comment actions](/docs/content-engine/threads-editing) are optional; live updates cover whatever the section already does.
+- The app's [WebSocket endpoint](/docs/dev/websocket/setup). Without it, the section works as before, just without live updates.
+
+
+
+
+
+### Read single branches
+
+Add `branches` to the `useContentComments` call in `src/pages/article-comments.tsx`. It calls `/thread/branches`, which the Content Engine adds to every content type with `thread`, so there is nothing to change on the comment type. This is an excerpt:
+
+```tsx title="plugins/newsroom/src/pages/article-comments.tsx"
+ const comments = useContentComments({
+ ...articleCommentsThread(articleId),
+ branches: async ids => // [!code ++:13]
+ await fetcher({
+ plugin: manifest.id,
+ method: "get",
+ module: "content/article-comments",
+ path: "/thread/branches",
+ args: {
+ query: {
+ root: String(articleId),
+ ...(ids.length > 0 ? { ids: ids.join(",") } : {}),
+ },
+ },
+ }),
+```
+
+A branch is one top-level comment with every reply below it. `branches` already makes the reader's own writes cheaper: a reply, edit, hide or delete re-reads only its branch instead of every loaded page.
+
+
+
+
+
+### Subscribe to the thread
+
+Add `live` to the same call, with the comment type's id and the article's id. Import `commentContentType` unless [Comment actions](/docs/content-engine/threads-editing) already did for reports:
+
+```tsx title="plugins/newsroom/src/pages/article-comments.tsx"
+import { commentContentType } from "@/content/comment"; // [!code ++]
+
+ live: { contentTypeId: commentContentType.id, rootId: articleId }, // [!code ++]
+```
+
+That's the whole change. `live` needs `branches` and is ignored without it. `ContentComments` shows **Show new comments** and refresh errors on its own. A custom layout renders `ContentCommentsLiveStatus` inside its `ContentCommentsRoot`, as the Blog plugin does; see [Comments API](/docs/content-engine/comments-api#your-own-layout).
+
+
+
+
+
+## Check the result
+
+Use two browser windows, one of them private, so they hold different sessions. Sign in as a different member in each, and open the same article with at most 20 top-level comments, so its only page is loaded:
+
+1. In window A, reply to a comment. Window B shows the reply within a second or two, without a reload.
+2. In window B, click **Reply** under a comment and type a few words without sending. In window A, reply to the same comment. Window B shows the new reply and keeps your unsent text.
+3. In window A, post a new top-level comment. Window B shows **New comments were posted.** at the end of the thread. Click **Show new comments** to add it below the others.
+
+A signed-out window gets the same updates; it just can't reply.
+
+
+
+## What arrives live
+
+| Change in another session | What an open section does |
+| --- | --- |
+| A reply, edit, hide, unhide or delete in a loaded branch | Re-reads that branch and puts it back in place |
+| A burst of changes to one branch | Re-reads it once, 0.4 seconds after the last change and never more than 2 seconds after the first |
+| A new top-level comment, with every page loaded | Shows **Show new comments**; clicking it reads only what follows the last loaded comment |
+| A new top-level comment, with more pages on the server | Nothing yet. Comments are oldest first, so **Load more comments** reaches it |
+| A change in a branch that isn't loaded | Nothing; it's read fresh when its page loads |
+| A comment that is a draft or waits for a moderator | Nothing. It isn't public, so other readers aren't told |
+| The article is hidden, unpublished or deleted | Readers who lost access see the thread disappear |
+
+Each reader's own session decides what they see, exactly as on the first load. A hidden comment still shows to its author and moderators, and becomes a placeholder for everybody else when it has replies.
+
+### Open forms keep their text
+
+If a comment is removed while someone edits it or replies to it, the form stays open with the text. It says "This comment was removed while you were editing it" or "The comment you are replying to was removed", and hides the submit button so the text can be copied elsewhere. Closing the form re-reads the branch, and the comment leaves the screen or turns into a placeholder.
+
+### Failed refreshes
+
+When a live refresh fails, the comments on screen stay and the section says "Could not refresh the comments, so some may be out of date." with **Try again**. A comment that was hidden, unpublished or deleted still leaves the screen, or turns into a placeholder when it has replies. When the article itself becomes unreadable, the thread is cleared rather than left on screen.
+
+## Related
+
+- [Live refresh](/docs/content-engine/live-refresh): which writes send a signal, who may subscribe, reconnects, several instances and custom write routes.
+- [Comments API](/docs/content-engine/comments-api): every `useContentComments` option and controller field.
diff --git a/apps/web/content/docs/dev/websocket/rooms.mdx b/apps/web/content/docs/dev/websocket/rooms.mdx
index e562b18db..c3b264bbf 100644
--- a/apps/web/content/docs/dev/websocket/rooms.mdx
+++ b/apps/web/content/docs/dev/websocket/rooms.mdx
@@ -207,4 +207,4 @@ Both listeners return a function that removes them. Without Redis, all of this w
## Related
- [Reconnects and missed messages](/docs/dev/websocket/reconnects)
-- [WebSocket](/docs/dev/websocket)
+- [Live refresh](/docs/content-engine/live-refresh): Content Engine's authorized rooms for AdminCP lists and comment threads, built on these primitives.
diff --git a/apps/web/src/locales/@vitnode/core/pl.json b/apps/web/src/locales/@vitnode/core/pl.json
index f9eceaeb6..9337fb3f9 100644
--- a/apps/web/src/locales/@vitnode/core/pl.json
+++ b/apps/web/src/locales/@vitnode/core/pl.json
@@ -3377,6 +3377,14 @@
"name": "[usunięty]",
"by_author": "Komentarz usunięty przez użytkownika",
"by_moderator": "Komentarz usunięty przez moderatora"
+ },
+ "live": {
+ "new_comments": "Pojawiły się nowe komentarze.",
+ "show_new": "Pokaż nowe komentarze",
+ "refresh_failed": "Nie udało się odświeżyć komentarzy, więc część z nich może być nieaktualna.",
+ "retry": "Spróbuj ponownie",
+ "edit_removed": "Ten komentarz został usunięty w trakcie edycji. Twój tekst nadal tu jest, ale nie da się go już zapisać.",
+ "reply_removed": "Komentarz, na który odpowiadasz, został usunięty. Twój tekst nadal tu jest, ale nie da się go już opublikować."
}
}
},
diff --git a/packages/vitnode/src/api/modules/content/content.module.ts b/packages/vitnode/src/api/modules/content/content.module.ts
index f141e7c15..710695906 100644
--- a/packages/vitnode/src/api/modules/content/content.module.ts
+++ b/packages/vitnode/src/api/modules/content/content.module.ts
@@ -1,5 +1,6 @@
import { buildModule } from "@/api/builders";
import { contentLiveWebSocket } from "@/content/server/live/socket";
+import { contentRefreshWebSocket } from "@/content/server/refresh/socket";
import { contentEditorialCleanupCron } from "./cron/content-editorial-cleanup.cron";
import { contentNotificationsQueueTask } from "./tasks/content-notifications.task";
@@ -17,5 +18,5 @@ export const contentModule = buildModule({
contentSearchDependentsQueueTask,
contentNotificationsQueueTask,
],
- webSockets: [contentLiveWebSocket],
+ webSockets: [contentLiveWebSocket, contentRefreshWebSocket],
});
diff --git a/packages/vitnode/src/content/index.ts b/packages/vitnode/src/content/index.ts
index e8f62ade8..ce6c22ef2 100644
--- a/packages/vitnode/src/content/index.ts
+++ b/packages/vitnode/src/content/index.ts
@@ -419,6 +419,14 @@ export {
resolveContentTranslationIndexes,
toSnakeCase,
} from "./indexes";
+export type {
+ ContentRefreshClientMessage,
+ ContentRefreshErrorCode,
+ ContentRefreshScope,
+ ContentRefreshServerMessage,
+ ContentRefreshSignal,
+} from "./live/refresh";
+export { contentRefreshChannel, contentRefreshRoom } from "./live/refresh";
export {
contentLocalesMatch,
isContentLocaleShaped,
diff --git a/packages/vitnode/src/content/live/refresh.ts b/packages/vitnode/src/content/live/refresh.ts
new file mode 100644
index 000000000..063d5856d
--- /dev/null
+++ b/packages/vitnode/src/content/live/refresh.ts
@@ -0,0 +1,104 @@
+import type { ContentId } from "@/content/ids";
+
+import { contentIdKey } from "@/content/ids";
+import { createWebSocketChannel } from "@/ws/types";
+
+/**
+ * What one live refresh subscription watches:
+ *
+ * - `admin`: the generated AdminCP list of one content type. Joining needs an
+ * AdminCP session with the content type's `can_view`.
+ * - `thread`: the public thread of one threaded content type under one root
+ * record. Joining needs the same access the `GET /thread` route asks for.
+ *
+ * Browser-safe.
+ */
+export type ContentRefreshScope =
+ | { contentTypeId: string; kind: "admin" }
+ | { contentTypeId: string; kind: "thread"; rootId: ContentId };
+
+/**
+ * A hint that something a subscriber shows changed. It never carries values:
+ * HTTP stays the source of truth, and the receiver reads it again.
+ *
+ * - `list`: an admin list of this content type changed.
+ * - `branch`: a public change inside one top-level branch of a thread.
+ * `branchId` is the top-level comment's id - public, or shown as a
+ * placeholder, before or after the change. `removedId` is set when a
+ * comment that was public stopped being so, so a receiver whose refetch
+ * fails can still take it off screen.
+ * - `thread`: the thread's root record changed in a way that can move who may
+ * read it. Read the thread again, and drop it when that is refused.
+ */
+export type ContentRefreshSignal =
+ | { branchId: ContentId; kind: "branch"; removedId?: ContentId }
+ | { kind: "list" }
+ | { kind: "thread" };
+
+export interface ContentRefreshClientMessage {
+ clientId: string;
+ scope: ContentRefreshScope;
+ type: "join" | "leave";
+}
+
+export type ContentRefreshErrorCode =
+ | "FORBIDDEN"
+ | "INVALID_MESSAGE"
+ | "NOT_FOUND"
+ | "RATE_LIMITED"
+ | "TOO_MANY_SUBSCRIPTIONS";
+
+export type ContentRefreshServerMessage =
+ | {
+ clientId: string;
+ code: "FORBIDDEN" | "NOT_FOUND";
+ room: string;
+ type: "revoked";
+ }
+ | {
+ clientId: string;
+ code: ContentRefreshErrorCode;
+ room?: string;
+ type: "error";
+ }
+ | { clientId: string; room: string; type: "joined" }
+ | { room: string; signal: ContentRefreshSignal; type: "changed" };
+
+/** Core's channel for Content Engine live refresh, on the shared WebSocket. */
+export const contentRefreshChannel = createWebSocketChannel<
+ ContentRefreshClientMessage,
+ ContentRefreshServerMessage
+>({ pluginId: "@vitnode/core", module: "content", id: "refresh" });
+
+export const CONTENT_REFRESH_ROOM_PREFIX = "content-refresh:";
+
+/**
+ * The room one scope maps to - a routing key, never a permission. Content type
+ * ids hold no `:`, and ids are compared as decimal or UUID strings, so a
+ * bigint id is never rounded through a `number`.
+ */
+export const contentRefreshRoom = (scope: ContentRefreshScope): string =>
+ scope.kind === "admin"
+ ? `${CONTENT_REFRESH_ROOM_PREFIX}admin:${scope.contentTypeId}`
+ : `${CONTENT_REFRESH_ROOM_PREFIX}thread:${scope.contentTypeId}:${contentIdKey(scope.rootId)}`;
+
+/** Subscriptions one connection may hold at once, over every tab sharing it. */
+export const CONTENT_REFRESH_MAX_SUBSCRIPTIONS = 64;
+
+/** Join requests one connection may send per minute. */
+export const CONTENT_REFRESH_MAX_JOINS_PER_MINUTE = 120;
+
+/** How long a successful join is trusted before the server checks it again. */
+export const CONTENT_REFRESH_REAUTH_MS = 60_000;
+
+/** How often the server looks for subscriptions due for that check. */
+export const CONTENT_REFRESH_SWEEP_MS = 20_000;
+
+/** A burst of signals is applied this long after the last one... */
+export const CONTENT_REFRESH_DEBOUNCE_MS = 400;
+
+/** ...but never later than this after the first, so traffic cannot starve it. */
+export const CONTENT_REFRESH_MAX_WAIT_MS = 2_000;
+
+/** Top-level branches one `GET /thread/branches` request reads. */
+export const CONTENT_THREAD_BRANCHES_MAX = 10;
diff --git a/packages/vitnode/src/content/server/editorial-effects.ts b/packages/vitnode/src/content/server/editorial-effects.ts
index 007d97996..a5b615383 100644
--- a/packages/vitnode/src/content/server/editorial-effects.ts
+++ b/packages/vitnode/src/content/server/editorial-effects.ts
@@ -167,7 +167,11 @@ export const contentEditorialEffects = async (
// be handling the request. Passed on every path, interactive and
// scheduled, so the envelope's owner is a property of the event rather
// than of how it was triggered.
- { pluginId, publicState: outcome.publicState },
+ {
+ pluginId,
+ publicState: outcome.publicState,
+ row: outcome.row,
+ },
);
// Logged here rather than left to each caller. The scheduled path
diff --git a/packages/vitnode/src/content/server/emit.ts b/packages/vitnode/src/content/server/emit.ts
index dfa4ee7c5..8c23ec9d5 100644
--- a/packages/vitnode/src/content/server/emit.ts
+++ b/packages/vitnode/src/content/server/emit.ts
@@ -29,6 +29,7 @@ import type { ContentPublicState } from "./page-cache";
import { contentEventName } from "../events";
import { invalidateContentPages } from "./page-cache";
+import { publishContentRefresh } from "./refresh/publish";
type ContentPayload =
| ContentCreatedPayload
@@ -56,10 +57,15 @@ interface ContentEventEmitter {
}
/**
- * Emits `content.{definition.id}.{action}` on the request's event bus and then
- * invalidates the cached pages that show the content type. Call it after the
- * write's transaction has committed. The generated routes already emit every
- * built-in event; use this from custom write paths. Server-only.
+ * Emits `content.{definition.id}.{action}` on the request's event bus,
+ * invalidates the cached pages and public reads that show the content type,
+ * and only then sends the live refresh signals (`publishContentRefresh`) to
+ * open AdminCP lists and threads - so a client that reacts reads the new
+ * state. Call it after the write's transaction has committed, never inside
+ * it: a rolled-back write must not be announced. The generated routes already
+ * emit every built-in event; use this from custom write paths. A plain
+ * `model.service(c)` or `model.editorialService(c)` write emits nothing on
+ * its own. Server-only.
*
* @returns The event bus result, with a page-cache invalidation failure, if
* any, appended to `failures` rather than thrown.
@@ -77,6 +83,12 @@ export const emitContentEvent = async (
* the content type's cached lists.
*/
publicState?: ContentPublicState;
+ /**
+ * The record after the write, or as it was just before a delete. Live
+ * refresh routes a threaded record to its thread with it; a deleted one
+ * cannot be read back, so pass it on every delete of a threaded type.
+ */
+ row?: null | Record;
},
): Promise => {
const name = contentEventName(definition.id, action) as VitNodeEventName;
@@ -99,6 +111,15 @@ export const emitContentEvent = async (
),
]);
+ // After the caches above, so whoever reads again on this signal misses them.
+ await publishContentRefresh(c, {
+ action,
+ contentId: payload.contentId,
+ definition,
+ publicState: options?.publicState,
+ row: options?.row,
+ });
+
return pageCacheFailure === null
? result
: { ...result, failures: [...result.failures, pageCacheFailure] };
diff --git a/packages/vitnode/src/content/server/index.ts b/packages/vitnode/src/content/server/index.ts
index 9279aa799..eab8f2574 100644
--- a/packages/vitnode/src/content/server/index.ts
+++ b/packages/vitnode/src/content/server/index.ts
@@ -214,8 +214,10 @@ export type {
} from "./notifications/registry";
export {
contentListPageCacheTags,
+ contentPublicStateOf,
contentRecordPageCacheTags,
} from "./page-cache";
+export type { ContentPublicState } from "./page-cache";
export {
assertContentPreviewIsServable,
contentPreviewSecret,
@@ -303,6 +305,11 @@ export {
} from "./query";
export { LABEL_PREFIX, resolveReferenceTargets, toLabel } from "./references";
export type { ReferenceTarget } from "./references";
+export {
+ onContentRefreshRootChanged,
+ publishContentRefresh,
+} from "./refresh/publish";
+export type { ContentRefreshChange } from "./refresh/publish";
export {
contentLocaleHeaders,
resolveContentRequestLocale,
@@ -396,8 +403,16 @@ export {
assertContentThreadPlacement,
ContentThreadPlacementError,
readContentThread,
+ readContentThreadBranches,
+} from "./thread";
+export type {
+ ContentThreadBranch,
+ ContentThreadBranches,
+ ContentThreadNode,
+ ContentThreadPage,
} from "./thread";
-export type { ContentThreadNode, ContentThreadPage } from "./thread";
+export { resolveContentThreadRoot } from "./thread-access";
+export type { ContentThreadRootAccess } from "./thread-access";
export { createContentTranslationEditorialService } from "./translation-editorial-service";
export type {
ContentRevisionDetailForLocale,
diff --git a/packages/vitnode/src/content/server/public-routes.ts b/packages/vitnode/src/content/server/public-routes.ts
index b39639c76..127c11ec2 100644
--- a/packages/vitnode/src/content/server/public-routes.ts
+++ b/packages/vitnode/src/content/server/public-routes.ts
@@ -10,7 +10,7 @@ import { z } from "@hono/zod-openapi";
import { and, eq } from "drizzle-orm";
import { HTTPException } from "hono/http-exception";
-import type { ContentIdOf } from "../ids";
+import type { ContentId, ContentIdOf } from "../ids";
import type {
AnyContentTypeDefinition,
ContentPublicFilterInput,
@@ -19,6 +19,7 @@ import type {
import type { ContentModel } from "./model";
import type { ContentPreviewTokenPayload } from "./preview-token";
import type { ContentPublicService } from "./public-service";
+import type { ContentThreadNode } from "./thread";
import { setPageCacheTags } from "../../api/lib/page-cache";
import { buildRoute } from "../../api/lib/route";
@@ -37,6 +38,7 @@ import { contentContainerFieldOf } from "../containers/association";
import { createContentContainerService } from "../containers/server/service";
import { ContentEngineError } from "../errors";
import { parseContentId } from "../ids";
+import { CONTENT_THREAD_BRANCHES_MAX } from "../live/refresh";
import { partitionContentFields } from "../localization";
import {
contentColumnsToValues,
@@ -47,7 +49,6 @@ import { publicOrderableColumns } from "../registry";
import { buildContentDeliveryRoutes } from "./delivery-routes";
import { resolveContentPublicRowFiles } from "./files";
import {
- canViewHiddenContentRecord,
contentListHiddenAccess,
contentThreadHiddenAccess,
findViewableHiddenBySlug,
@@ -79,7 +80,8 @@ import {
contentSnapshotRow,
projectTranslationRevisionSnapshot,
} from "./revision-snapshot";
-import { readContentThread } from "./thread";
+import { readContentThread, readContentThreadBranches } from "./thread";
+import { resolveContentThreadRoot } from "./thread-access";
/**
* The public routes of one content type: list and detail, plus preview,
@@ -524,6 +526,38 @@ export const buildContentPublicRoutes = <
},
});
+ /** Rows for a thread read: public ones, plus hidden ones this reader may see. */
+ const threadProjector =
+ (c: Context, hiddenParents: boolean) =>
+ async (ids: ContentId[], { hiddenIds }: { hiddenIds: ContentId[] }) => {
+ const read = service(c).findManyByIds;
+ const rows = read
+ ? await read(ids as never, {
+ hidden: hiddenIds.length > 0,
+ hiddenParents,
+ })
+ : [];
+
+ return new Map(
+ rows.map(row => [String((row as unknown as { id: unknown }).id), row]),
+ );
+ };
+
+ const serializeThreadNode = <
+ T extends { deleted: ContentThreadNode["deleted"] },
+ >({
+ deleted,
+ ...item
+ }: T) => ({
+ ...item,
+ deleted: deleted
+ ? {
+ byAuthor: deleted.byAuthor,
+ createdAt: deleted.createdAt.toISOString(),
+ }
+ : null,
+ });
+
const threadNode = z.object({
deleted: z
.object({ byAuthor: z.boolean(), createdAt: z.string() })
@@ -578,42 +612,11 @@ export const buildContentPublicRoutes = <
},
handler: async c => {
const { after, first, root } = c.req.valid("query");
- const rootField = definition.fields[definition.thread.rootField];
- const target =
- rootField?.kind === "relation" ? rootField.target() : undefined;
- const rootModel = (c.get("core")?.contentModels ?? []).find(
- entry => entry.model.definition.id === target?.id,
- )?.model;
- const rootId = target ? parseContentId(target.idStrategy, root) : null;
- // An existence check: `select: []` reads the same published, visible row
- // without hydrating a single field of it.
- const findReadableRoot = async () => {
- const rootService = rootModel?.publicService?.(c);
- if (rootId === null || !target || !rootService) return null;
-
- if (await rootService.findById(rootId, { select: [] })) {
- return { hidden: false };
- }
-
- const hidden = await rootService.findById(rootId, {
- hidden: true,
- select: [],
- });
-
- return hidden &&
- (await canViewHiddenContentRecord(c, {
- contentTypeId: target.id,
- itemId: rootId,
- }))
- ? { hidden: true }
- : null;
- };
- const readable = await findReadableRoot();
- if (!readable || rootId === null) {
+ const readable = await resolveContentThreadRoot(c, definition, root);
+ if (!readable) {
throw new HTTPException(404, { message: "Thread not found." });
}
- const read = service(c).findManyByIds;
const hiddenAccess = await contentThreadHiddenAccess(c, {
model,
pluginId,
@@ -622,35 +625,13 @@ export const buildContentPublicRoutes = <
after,
first,
hiddenAccess,
- project: async (ids, { hiddenIds }) => {
- const rows = read
- ? await read(ids as never, {
- hidden: hiddenIds.length > 0,
- hiddenParents: readable.hidden,
- })
- : [];
-
- return new Map(
- rows.map(row => [
- String((row as unknown as { id: unknown }).id),
- row,
- ]),
- );
- },
- root: rootId,
+ project: threadProjector(c, readable.hidden),
+ root: readable.rootId,
});
const body = {
...page,
- items: page.items.map(({ deleted, ...item }) => ({
- ...item,
- deleted: deleted
- ? {
- byAuthor: deleted.byAuthor,
- createdAt: deleted.createdAt.toISOString(),
- }
- : null,
- })),
+ items: page.items.map(serializeThreadNode),
};
if (hiddenAccess || readable.hidden) {
@@ -659,13 +640,137 @@ export const buildContentPublicRoutes = <
setPageCacheTags(c, [
...contentListPageCacheTags(definition),
- ...(target ? contentListPageCacheTags(target) : []),
+ ...contentListPageCacheTags(readable.target),
]);
return c.json(body, 200);
},
});
+ const threadBranch = z.discriminatedUnion("status", [
+ z.object({
+ id: schemas.params.shape.id,
+ items: z.array(threadNode),
+ status: z.literal("ok"),
+ }),
+ z.object({ id: schemas.params.shape.id, status: z.literal("gone") }),
+ ]);
+
+ /**
+ * The current state of a few top-level branches already on a reader's
+ * screen, so a live update re-reads one branch instead of every page. The
+ * same root check, visibility, placeholders, depth and truncation as
+ * `/thread`. Guests and readers with no hidden access get the same answer,
+ * so it is read through the public read cache that every comment write
+ * expires before it is announced; anyone else's answer is private.
+ */
+ const threadBranches = buildRoute({
+ pluginId,
+ route: {
+ method: "get",
+ path: "/thread/branches",
+ description: `Re-read some top-level ${name} branches of one thread, each with every answer below it`,
+ request: {
+ query: z.object({
+ root: z.string().min(1).max(64),
+ ids: z
+ .string()
+ .max(CONTENT_THREAD_BRANCHES_MAX * 65)
+ .optional(),
+ }),
+ },
+ responses: {
+ 200: {
+ content: {
+ "application/json": {
+ schema: z.object({
+ branches: z.array(threadBranch),
+ pageInfo: z.object({ totalCount: z.number() }),
+ replyCount: z.number(),
+ truncated: z.boolean(),
+ }),
+ },
+ },
+ description:
+ "Each requested branch, in the order asked, or `gone` when the reader no longer sees it. Without `ids`, only the counts",
+ },
+ 400: { description: "Malformed or too many ids" },
+ 404: { description: "The root record does not exist or is not public" },
+ },
+ },
+ handler: async c => {
+ const query = c.req.valid("query");
+ const unique = new Map();
+ for (const raw of (query.ids ?? "").split(",")) {
+ const trimmed = raw.trim();
+ if (trimmed === "") continue;
+ const id = parseContentId(definition.idStrategy, trimmed);
+ if (id === null) {
+ throw new HTTPException(400, { message: "Invalid branch id." });
+ }
+ if (!unique.has(String(id))) unique.set(String(id), id);
+ }
+ const ids = [...unique.values()];
+ if (ids.length > CONTENT_THREAD_BRANCHES_MAX) {
+ throw new HTTPException(400, {
+ message: `Ask for at most ${CONTENT_THREAD_BRANCHES_MAX} branches at once.`,
+ });
+ }
+
+ const readable = await resolveContentThreadRoot(
+ c,
+ definition,
+ query.root,
+ );
+ if (!readable) {
+ throw new HTTPException(404, { message: "Thread not found." });
+ }
+
+ const hiddenAccess = await contentThreadHiddenAccess(c, {
+ model,
+ pluginId,
+ });
+ const load = async () => {
+ const result = await readContentThreadBranches(c, model, {
+ hiddenAccess,
+ ids,
+ project: threadProjector(c, readable.hidden),
+ root: readable.rootId,
+ });
+
+ return {
+ ...result,
+ branches: result.branches.map(branch =>
+ branch.status === "ok"
+ ? { ...branch, items: branch.items.map(serializeThreadNode) }
+ : branch,
+ ),
+ };
+ };
+
+ if (hiddenAccess || readable.hidden) {
+ return c.json(await load(), 200, {
+ "Cache-Control": "private, no-store",
+ });
+ }
+
+ const body = await rememberContentPublicRead(c, {
+ contentTypeId: definition.id,
+ key: {
+ ids: ids.map(String),
+ root: String(readable.rootId),
+ route: "thread-branches",
+ },
+ load,
+ });
+
+ // Every viewer of a busy thread asks for the same branch after one
+ // signal: the read cache above answers them, and no shared HTTP cache
+ // may hold a copy that outlives the next write.
+ return c.json(body, 200, { "Cache-Control": "no-cache" });
+ },
+ });
+
const detail = buildRoute({
pluginId,
route: {
@@ -755,7 +860,7 @@ export const buildContentPublicRoutes = <
...(definition.delivery.enabled
? buildContentDeliveryRoutes(model, { pluginId })
: []),
- ...(definition.thread.enabled ? [thread] : []),
+ ...(definition.thread.enabled ? [thread, threadBranches] : []),
...(definition.publicWrite.enabled ||
definition.reports.enabled ||
definition.visibility.enabled
diff --git a/packages/vitnode/src/content/server/public-visibility-routes.ts b/packages/vitnode/src/content/server/public-visibility-routes.ts
index b9ed4c1d6..56355e6f2 100644
--- a/packages/vitnode/src/content/server/public-visibility-routes.ts
+++ b/packages/vitnode/src/content/server/public-visibility-routes.ts
@@ -286,6 +286,7 @@ export const contentPublicDeleteRoute = <
{
pluginId,
publicState: contentPublicStateOf(definition, deleted, null),
+ row: deleted,
},
);
await syncContentSearch(c, definition, {
@@ -309,6 +310,7 @@ export const contentPublicDeleteRoute = <
{
pluginId,
publicState: contentPublicStateOf(definition, row, null),
+ row,
},
);
await syncContentSearch(c, definition, {
diff --git a/packages/vitnode/src/content/server/refresh/auth.test.ts b/packages/vitnode/src/content/server/refresh/auth.test.ts
new file mode 100644
index 000000000..a0ee87476
--- /dev/null
+++ b/packages/vitnode/src/content/server/refresh/auth.test.ts
@@ -0,0 +1,216 @@
+import type { Context, Next } from "hono";
+
+import { describe, expect, it } from "vitest";
+
+import type { EnvVariablesVitNode } from "@/api/middlewares/global.middleware";
+import type { ContentRefreshScope } from "@/content/live/refresh";
+
+import { SessionModel } from "@/api/models/session";
+import { CONTENT_PERMISSIONS } from "@/content/const";
+import { core_content_revisions } from "@/database/content";
+import { core_content_containers } from "@/database/content-containers";
+import { core_content_submissions } from "@/database/content-submissions";
+import {
+ BOARD,
+ POLICY_LIVE,
+ POLICY_PLUGIN_ID,
+ policyBoardRows,
+ policyReplies,
+ policyTopicRow,
+ policyTopics,
+} from "@/tests/content-policies";
+import {
+ createSessionWorld,
+ fakeSession,
+ sessionCookies,
+} from "@/tests/sessions";
+import { grantStaffPermissions } from "@/tests/staff-permissions";
+
+import { authorizeContentRefresh } from "./auth";
+
+const TOPIC = { DRAFT: 2, HIDDEN: 3, LIVE: 1 } as const;
+const REPLY_TYPE = policyReplies.definition.id;
+const TOPIC_TYPE = policyTopics.definition.id;
+
+const DEVICE = { id: 3, publicId: "laptop" };
+const admin = fakeSession(7, DEVICE);
+const member = fakeSession(8, DEVICE, ["user"]);
+const author = fakeSession(9, DEVICE, ["user"]);
+
+const CAN_VIEW_TOPICS = {
+ module: policyTopics.definition.permissionModule,
+ permission: CONTENT_PERMISSIONS.view,
+ plugin: POLICY_PLUGIN_ID,
+};
+
+const setup = async () => {
+ const world = await createSessionWorld({
+ extraTables: [
+ [core_content_containers, policyBoardRows()],
+ [
+ policyTopics.table,
+ [
+ policyTopicRow(TOPIC.LIVE, BOARD.OPEN),
+ policyTopicRow(TOPIC.DRAFT, BOARD.OPEN, {
+ publishedAt: null,
+ status: "draft",
+ }),
+ policyTopicRow(TOPIC.HIDDEN, BOARD.OPEN, {
+ author: author.userId,
+ hiddenAt: POLICY_LIVE,
+ hiddenBy: 99,
+ }),
+ ],
+ ],
+ [policyReplies.table, []],
+ [core_content_revisions, []],
+ [core_content_submissions, []],
+ ],
+ sessions: [admin, member, author],
+ });
+
+ world.app.use("*", async (c: Context, next: Next) => {
+ c.set("core", {
+ ...c.get("core"),
+ contentModels: [
+ { model: policyTopics, pluginId: POLICY_PLUGIN_ID },
+ { model: policyReplies, pluginId: POLICY_PLUGIN_ID },
+ ],
+ contentTypes: [
+ { definition: policyTopics.definition, pluginId: POLICY_PLUGIN_ID },
+ { definition: policyReplies.definition, pluginId: POLICY_PLUGIN_ID },
+ ],
+ moderation: { scopes: [], targets: [] },
+ permissionStaff: [],
+ } as unknown as EnvVariablesVitNode["core"]);
+ await next();
+ });
+ world.app.post("/authorize", async c =>
+ c.json(
+ await authorizeContentRefresh(c, await c.req.json()),
+ ),
+ );
+ // An open WebSocket keeps the user it was opened with, even after sign-out.
+ world.app.post("/authorize-after-sign-out", async c => {
+ await new SessionModel(c).deleteSession();
+
+ return c.json(
+ await authorizeContentRefresh(c, await c.req.json()),
+ );
+ });
+
+ const authorize = async (
+ scope: ContentRefreshScope,
+ {
+ as,
+ kind = "admin",
+ signedOut = false,
+ }: {
+ as?: typeof admin;
+ kind?: "admin" | "user";
+ signedOut?: boolean;
+ } = {},
+ ) => {
+ const response = await world.app.request(
+ signedOut ? "/authorize-after-sign-out" : "/authorize",
+ {
+ body: JSON.stringify(scope),
+ headers: as ? { cookie: sessionCookies(as, kind) } : {},
+ method: "POST",
+ },
+ );
+
+ return (await response.json()) as Awaited<
+ ReturnType
+ >;
+ };
+
+ return { authorize, world };
+};
+
+const thread = (rootId: number | string): ContentRefreshScope => ({
+ contentTypeId: REPLY_TYPE,
+ kind: "thread",
+ rootId,
+});
+
+const NOT_FOUND = { code: "NOT_FOUND", ok: false };
+const FORBIDDEN = { code: "FORBIDDEN", ok: false };
+
+describe("authorizeContentRefresh for a thread", () => {
+ it("lets a guest follow the thread of a public root", async () => {
+ const { authorize } = await setup();
+
+ expect(await authorize(thread(TOPIC.LIVE))).toStrictEqual({ ok: true });
+ });
+
+ it.each([
+ ["a draft", thread(TOPIC.DRAFT)],
+ ["a hidden", thread(TOPIC.HIDDEN)],
+ ["a missing", thread(999)],
+ ["a malformed", thread("abc")],
+ [
+ "a content type without threads, for",
+ { contentTypeId: TOPIC_TYPE, kind: "thread", rootId: TOPIC.LIVE },
+ ],
+ [
+ "an unknown content type, for",
+ { contentTypeId: "test.nope", kind: "thread", rootId: TOPIC.LIVE },
+ ],
+ ] satisfies [string, ContentRefreshScope][])(
+ "answers NOT_FOUND for %s root",
+ async (_label, scope) => {
+ const { authorize } = await setup();
+
+ expect(await authorize(scope)).toStrictEqual(NOT_FOUND);
+ },
+ );
+
+ it("lets a signed-in reader follow a public thread only while their session exists", async () => {
+ const { authorize, world } = await setup();
+ const reader = { as: member, kind: "user" } as const;
+
+ expect(await authorize(thread(TOPIC.LIVE), reader)).toStrictEqual({
+ ok: true,
+ });
+ expect(
+ await authorize(thread(TOPIC.LIVE), { ...reader, signedOut: true }),
+ ).toStrictEqual(FORBIDDEN);
+ expect(world.rowsFor(member).user).toBe(0);
+ });
+
+ it("lets the author of a hidden root follow its thread", async () => {
+ const { authorize } = await setup();
+
+ expect(
+ await authorize(thread(TOPIC.HIDDEN), { as: author, kind: "user" }),
+ ).toStrictEqual({ ok: true });
+ });
+});
+
+describe("authorizeContentRefresh for an AdminCP list", () => {
+ const list: ContentRefreshScope = {
+ contentTypeId: TOPIC_TYPE,
+ kind: "admin",
+ };
+
+ it.each([
+ ["without a session", undefined, "admin"],
+ ["with only a member session", member, "user"],
+ ["with an admin session lacking can_view", admin, "admin"],
+ ] as const)("refuses %s", async (_label, as, kind) => {
+ const { authorize } = await setup();
+
+ expect(await authorize(list, { as, kind })).toStrictEqual(FORBIDDEN);
+ });
+
+ it("accepts an AdminCP session with the content type's can_view", async () => {
+ const { authorize, world } = await setup();
+ await grantStaffPermissions(world.cache, {
+ permissions: [CAN_VIEW_TOPICS],
+ userId: admin.userId,
+ });
+
+ expect(await authorize(list, { as: admin })).toStrictEqual({ ok: true });
+ });
+});
diff --git a/packages/vitnode/src/content/server/refresh/auth.ts b/packages/vitnode/src/content/server/refresh/auth.ts
new file mode 100644
index 000000000..d8670ed76
--- /dev/null
+++ b/packages/vitnode/src/content/server/refresh/auth.ts
@@ -0,0 +1,81 @@
+import type { Context } from "hono";
+
+import type { ContentRefreshScope } from "@/content/live/refresh";
+
+import { checkStaffPermissionOfUser } from "@/api/lib/check-staff-permission";
+import { SessionModel } from "@/api/models/session";
+import { SessionAdminModel } from "@/api/models/session-admin";
+import { CONTENT_PERMISSIONS } from "@/content/const";
+import { contentRefreshRoom } from "@/content/live/refresh";
+
+import { resolveContentThreadRoot } from "../thread-access";
+
+export type ContentRefreshAuthorization =
+ | { code: "FORBIDDEN" | "NOT_FOUND"; ok: false }
+ | { ok: true };
+
+/**
+ * Whether the connection behind `c` may follow `scope`, by the same rules as
+ * the HTTP route it mirrors:
+ *
+ * - `admin`: an AdminCP session (read fresh, so a signed-out or expired one
+ * fails) whose user holds the content type's `can_view` - the permission
+ * the generated `GET /` list route requires.
+ * - `thread`: `resolveContentThreadRoot`, the rule behind `GET /thread`. A
+ * guest passes for a public root; a connection opened signed in fails once
+ * that session is gone.
+ */
+export const authorizeContentRefresh = async (
+ c: Context,
+ scope: ContentRefreshScope,
+): Promise => {
+ const entry = (c.get("core")?.contentTypes ?? []).find(
+ candidate => candidate.definition.id === scope.contentTypeId,
+ );
+ if (!entry) return { code: "NOT_FOUND", ok: false };
+
+ if (scope.kind === "thread") {
+ const { definition } = entry;
+ if (!definition.thread.enabled || !definition.publicApi.enabled) {
+ return { code: "NOT_FOUND", ok: false };
+ }
+
+ // The connection keeps the user it was opened with. Signed out since, it
+ // follows nothing until the client reconnects as whoever it is now.
+ const user = c.get("user");
+ if (user && (await new SessionModel(c).getUser())?.id !== user.id) {
+ return { code: "FORBIDDEN", ok: false };
+ }
+
+ return (await resolveContentThreadRoot(c, definition, scope.rootId))
+ ? { ok: true }
+ : { code: "NOT_FOUND", ok: false };
+ }
+
+ const session = await new SessionAdminModel(c).getSession({ extend: false });
+ if (!session) return { code: "FORBIDDEN", ok: false };
+
+ const allowed = await checkStaffPermissionOfUser(c, session.user, {
+ module: entry.definition.permissionModule,
+ permission: CONTENT_PERMISSIONS.view,
+ plugin: entry.pluginId,
+ type: "admin",
+ });
+
+ return allowed ? { ok: true } : { code: "FORBIDDEN", ok: false };
+};
+
+/**
+ * Connections that are sure to get the same answer for `scope`, so a re-check
+ * of a busy room asks once per group: a thread's answer depends only on who is
+ * signed in. `null` for an admin scope, which reads each connection's own
+ * AdminCP session.
+ */
+export const contentRefreshShareKey = (
+ c: Context,
+ scope: ContentRefreshScope,
+): null | string => {
+ if (scope.kind === "admin") return null;
+
+ return `${contentRefreshRoom(scope)}|${c.get("user")?.id ?? "guest"}`;
+};
diff --git a/packages/vitnode/src/content/server/refresh/instances.test.ts b/packages/vitnode/src/content/server/refresh/instances.test.ts
new file mode 100644
index 000000000..e13eea202
--- /dev/null
+++ b/packages/vitnode/src/content/server/refresh/instances.test.ts
@@ -0,0 +1,98 @@
+import type { WSContext } from "hono/ws";
+
+import { describe, expect, it } from "vitest";
+
+import type { CacheClient } from "@/api/lib/cache";
+
+import { contentRefreshRoom } from "@/content/live/refresh";
+import {
+ BOARD,
+ createPolicyWorld,
+ policyReplies,
+ policyTopicRow,
+ USER,
+} from "@/tests/content-policies";
+import {
+ initRealtimePubSub,
+ markWebSocketEnabled,
+ wsRegistry,
+} from "@/ws/registry";
+
+/** One Redis channel between this instance and another one. */
+const bus = () => {
+ const published: string[] = [];
+ let deliver: ((raw: string) => void) | undefined;
+ const client = {
+ duplicate: () => {
+ const subscriber = {
+ connect: async () => await Promise.resolve(),
+ on: () => subscriber,
+ subscribe: async (
+ _channel: string,
+ listener: (raw: string) => void,
+ ) => {
+ deliver = listener;
+
+ return await Promise.resolve();
+ },
+ };
+
+ return subscriber;
+ },
+ publish: async (_channel: string, raw: string) => {
+ published.push(raw);
+
+ return await Promise.resolve(1);
+ },
+ };
+
+ return {
+ client: client as unknown as CacheClient,
+ /** Hands a message to this process as if another instance sent it. */
+ fromOtherInstance: (message: Record) => {
+ deliver?.(JSON.stringify({ ...message, origin: "other-instance" }));
+ },
+ published: () =>
+ published.map(raw => JSON.parse(raw) as Record),
+ };
+};
+
+describe("content refresh across instances", () => {
+ it("relays a root change to a thread followed only on another instance", async () => {
+ const pubsub = bus();
+ initRealtimePubSub(pubsub.client);
+ markWebSocketEnabled();
+ await Promise.resolve();
+ const room = contentRefreshRoom({
+ contentTypeId: policyReplies.definition.id,
+ kind: "thread",
+ rootId: 10,
+ });
+ const { send } = await createPolicyWorld({
+ topics: [policyTopicRow(10, BOARD.OPEN)],
+ });
+
+ await send("POST", "/topics/10/hide", { body: {}, user: USER.ROOT });
+
+ const message = pubsub.published().find(entry => entry.room === room);
+ expect(message).toMatchObject({
+ data: { room, signal: { kind: "thread" }, type: "changed" },
+ type: "room",
+ });
+
+ const received: unknown[] = [];
+ const ws = {
+ send: (raw: string) => {
+ received.push((JSON.parse(raw) as { data: unknown }).data);
+ },
+ } as unknown as WSContext;
+ wsRegistry.add(ws, null);
+ wsRegistry.join(ws, room, "tab");
+ pubsub.fromOtherInstance(message ?? {});
+ wsRegistry.remove(ws);
+
+ expect(received).toStrictEqual([
+ { room, signal: { kind: "thread" }, type: "changed" },
+ ]);
+ });
+});
diff --git a/packages/vitnode/src/content/server/refresh/publish.test.ts b/packages/vitnode/src/content/server/refresh/publish.test.ts
new file mode 100644
index 000000000..0e81c811f
--- /dev/null
+++ b/packages/vitnode/src/content/server/refresh/publish.test.ts
@@ -0,0 +1,302 @@
+import type { WSContext } from "hono/ws";
+
+import { afterEach, beforeAll, describe, expect, it, vi } from "vitest";
+
+import type {
+ ContentRefreshScope,
+ ContentRefreshServerMessage,
+} from "@/content/live/refresh";
+
+import { CacheModel } from "@/api/lib/cache";
+import { contentRefreshRoom } from "@/content/live/refresh";
+import {
+ BOARD,
+ createPolicyWorld,
+ POLICY_LIVE,
+ policyReplies,
+ policyReplyRow,
+ policyTopicRow,
+ USER,
+} from "@/tests/content-policies";
+import { markWebSocketEnabled, realtime, wsRegistry } from "@/ws/registry";
+
+import { onContentRefreshRootChanged } from "./publish";
+
+const REPLY_TYPE = policyReplies.definition.id;
+const TOPIC = { APPROVE: 30, LIVE: 10 } as const;
+
+const threadOf = (rootId: number): ContentRefreshScope => ({
+ contentTypeId: REPLY_TYPE,
+ kind: "thread",
+ rootId,
+});
+const REPLIES_LIST: ContentRefreshScope = {
+ contentTypeId: REPLY_TYPE,
+ kind: "admin",
+};
+
+/**
+ * Topic 10:
+ *
+ * 100 ─ 101 ─ 102
+ * └ 103 (hidden)
+ * 110
+ */
+const world = async () =>
+ await createPolicyWorld({
+ replies: [
+ policyReplyRow(100, TOPIC.LIVE),
+ policyReplyRow(101, TOPIC.LIVE, { parent: 100 }),
+ policyReplyRow(102, TOPIC.LIVE, { parent: 101 }),
+ policyReplyRow(103, TOPIC.LIVE, {
+ hiddenAt: POLICY_LIVE,
+ hiddenBy: USER.ROOT,
+ parent: 101,
+ }),
+ policyReplyRow(110, TOPIC.LIVE),
+ ],
+ topics: [
+ policyTopicRow(TOPIC.LIVE, BOARD.OPEN),
+ policyTopicRow(TOPIC.APPROVE, BOARD.APPROVE),
+ ],
+ });
+
+const sockets: WSContext[] = [];
+
+/** A browser tab following `scope`, as a successful join leaves it. */
+const listen = (scope: ContentRefreshScope) => {
+ const received: ContentRefreshServerMessage[] = [];
+ const ws = {
+ send: (raw: string) => {
+ received.push(
+ (JSON.parse(raw) as { data: ContentRefreshServerMessage }).data,
+ );
+ },
+ } as unknown as WSContext;
+ wsRegistry.add(ws, null);
+ wsRegistry.join(ws, contentRefreshRoom(scope), "tab");
+ sockets.push(ws);
+
+ return { received, ws };
+};
+
+const unique = (messages: T[]): T[] => [
+ ...new Map(
+ messages.map(message => [JSON.stringify(message), message]),
+ ).values(),
+];
+
+const signalsOf = (messages: ContentRefreshServerMessage[]) =>
+ messages.flatMap(message =>
+ message.type === "changed" ? [message.signal] : [],
+ );
+
+beforeAll(() => {
+ markWebSocketEnabled();
+});
+
+afterEach(() => {
+ sockets.splice(0).forEach(ws => {
+ wsRegistry.remove(ws);
+ });
+ vi.restoreAllMocks();
+});
+
+describe("publishing content refresh signals", () => {
+ it.each([
+ {
+ expected: { branchId: 100, kind: "branch" },
+ name: "a new reply deep in a branch",
+ request: [
+ "POST",
+ "/replies",
+ { values: { body: "New", parent: 102, topic: TOPIC.LIVE } },
+ ],
+ user: USER.MEMBER,
+ },
+ {
+ expected: { branchId: 100, kind: "branch" },
+ name: "an edit",
+ request: ["PATCH", "/replies/102", { values: { body: "Edited" } }],
+ user: USER.MOD_OPEN,
+ },
+ {
+ expected: { branchId: 100, kind: "branch", removedId: 102 },
+ name: "a delete",
+ request: ["POST", "/replies/102/delete"],
+ user: USER.ROOT,
+ },
+ {
+ expected: { branchId: 100, kind: "branch", removedId: 102 },
+ name: "a hide",
+ request: ["POST", "/replies/102/hide", {}],
+ user: USER.ROOT,
+ },
+ {
+ expected: { branchId: 100, kind: "branch" },
+ name: "an unhide",
+ request: ["POST", "/replies/103/unhide"],
+ user: USER.ROOT,
+ },
+ {
+ expected: { branchId: 110, kind: "branch", removedId: 110 },
+ name: "a top-level delete",
+ request: ["POST", "/replies/110/delete"],
+ user: USER.ROOT,
+ },
+ ] as const)(
+ "sends $name to the thread with its top-level branch, and the list to the AdminCP",
+ async ({ expected, request: [method, path, body], user }) => {
+ const { send } = await world();
+ const thread = listen(threadOf(TOPIC.LIVE));
+ const list = listen(REPLIES_LIST);
+
+ const response = await send(method, path, { body, user });
+
+ expect(response.status).toBeLessThan(300);
+ // A create announces `created` and `published`: one hint per event.
+ expect(unique(thread.received)).toStrictEqual([
+ {
+ room: contentRefreshRoom(threadOf(TOPIC.LIVE)),
+ signal: expected,
+ type: "changed",
+ },
+ ]);
+ expect(unique(signalsOf(list.received))).toStrictEqual([
+ { kind: "list" },
+ ]);
+ },
+ );
+
+ it("keeps a reply that waits for a moderator off the thread, but not off the AdminCP list", async () => {
+ const { send } = await world();
+ const thread = listen(threadOf(TOPIC.APPROVE));
+ const list = listen(REPLIES_LIST);
+
+ const response = await send("POST", "/replies", {
+ body: { values: { body: "Pending", topic: TOPIC.APPROVE } },
+ user: USER.MEMBER,
+ });
+
+ expect(response.status).toBe(201);
+ await expect(response.json()).resolves.toMatchObject({ status: "draft" });
+ expect(thread.received).toStrictEqual([]);
+ expect(unique(signalsOf(list.received))).toStrictEqual([{ kind: "list" }]);
+ });
+
+ it("sends nothing to a thread room nobody on this instance follows", async () => {
+ const { send } = await world();
+ const toRoom = vi.spyOn(realtime, "toRoom");
+
+ await send("POST", "/replies", {
+ body: { values: { body: "Unheard", topic: TOPIC.LIVE } },
+ user: USER.MEMBER,
+ });
+
+ expect(unique(toRoom.mock.calls.map(([room]) => room))).toStrictEqual([
+ contentRefreshRoom(REPLIES_LIST),
+ ]);
+ });
+
+ it("tells a thread its root was hidden, and re-checks the room's members", async () => {
+ const { send } = await world();
+ const thread = listen(threadOf(TOPIC.LIVE));
+ const rootChanged = vi.fn();
+ const stop = onContentRefreshRootChanged(rootChanged);
+
+ const response = await send("POST", `/topics/${TOPIC.LIVE}/hide`, {
+ body: {},
+ user: USER.ROOT,
+ });
+ stop();
+
+ expect(response.status).toBe(200);
+ expect(signalsOf(thread.received)).toStrictEqual([{ kind: "thread" }]);
+ expect(rootChanged).toHaveBeenCalledExactlyOnceWith(
+ contentRefreshRoom(threadOf(TOPIC.LIVE)),
+ );
+ });
+
+ it("signals only after the event went out and the cached public reads expired", async () => {
+ const { emitted, send } = await world();
+ const setSystem = vi.spyOn(CacheModel.prototype, "setSystem");
+ const seen: { bumped: boolean; events: string[] }[] = [];
+ const thread = listen(threadOf(TOPIC.LIVE));
+ const { send: deliver } = thread.ws as unknown as {
+ send: (raw: string) => void;
+ };
+ Object.assign(thread.ws, {
+ send: (raw: string) => {
+ seen.push({
+ bumped: setSystem.mock.calls.some(
+ ([key]) => key === `content:gen:${REPLY_TYPE}`,
+ ),
+ events: emitted.map(event => event.name),
+ });
+ deliver(raw);
+ },
+ });
+
+ await send("POST", "/replies", {
+ body: { values: { body: "Ordered", parent: 100, topic: TOPIC.LIVE } },
+ user: USER.MEMBER,
+ });
+
+ expect(seen).toStrictEqual([
+ { bumped: true, events: [`content.${REPLY_TYPE}.created`] },
+ {
+ bumped: true,
+ events: [
+ `content.${REPLY_TYPE}.created`,
+ `content.${REPLY_TYPE}.published`,
+ ],
+ },
+ ]);
+ });
+
+ it.each([
+ ["a missing reply", "PATCH", "/replies/999", { body: "Nope" }, 404],
+ [
+ "a parent from another topic",
+ "POST",
+ "/replies",
+ { body: "Nope", parent: 100, topic: TOPIC.APPROVE },
+ 400,
+ ],
+ ] as const)(
+ "publishes nothing for a refused write to %s",
+ async (_label, method, path, values, status) => {
+ const { send } = await world();
+ const thread = listen(threadOf(TOPIC.LIVE));
+ const list = listen(REPLIES_LIST);
+
+ const response = await send(method, path, {
+ body: { values },
+ user: USER.MEMBER,
+ });
+
+ expect(response.status).toBe(status);
+ expect([...thread.received, ...list.received]).toStrictEqual([]);
+ },
+ );
+
+ it("never fails a committed write when sending a signal throws", async () => {
+ const { events, send } = await world();
+ listen(threadOf(TOPIC.LIVE));
+ vi.spyOn(realtime, "toRoom").mockImplementation(() => {
+ throw new Error("Socket layer down.");
+ });
+ vi.spyOn(console, "error").mockImplementation(() => undefined);
+
+ const response = await send("POST", "/replies", {
+ body: { values: { body: "Still saved", topic: TOPIC.LIVE } },
+ user: USER.MEMBER,
+ });
+
+ expect(response.status).toBe(201);
+ expect(events()).toStrictEqual([
+ `content.${REPLY_TYPE}.created`,
+ `content.${REPLY_TYPE}.published`,
+ ]);
+ });
+});
diff --git a/packages/vitnode/src/content/server/refresh/publish.ts b/packages/vitnode/src/content/server/refresh/publish.ts
new file mode 100644
index 000000000..069b5262a
--- /dev/null
+++ b/packages/vitnode/src/content/server/refresh/publish.ts
@@ -0,0 +1,258 @@
+import type { PgColumn, PgTable } from "drizzle-orm/pg-core";
+import type { Context } from "hono";
+
+import { eq } from "drizzle-orm";
+
+import type { ContentEventAction } from "@/content/events";
+import type { ContentId } from "@/content/ids";
+import type {
+ ContentRefreshScope,
+ ContentRefreshServerMessage,
+ ContentRefreshSignal,
+} from "@/content/live/refresh";
+import type { AnyContentTypeDefinition } from "@/content/types";
+
+import { parseContentId } from "@/content/ids";
+import {
+ contentRefreshChannel,
+ contentRefreshRoom,
+} from "@/content/live/refresh";
+import {
+ isRealtimePubSubEnabled,
+ isWebSocketEnabled,
+ realtime,
+ wsRegistry,
+} from "@/ws/registry";
+
+import type { ContentPublicState } from "../page-cache";
+
+import { findContentModel } from "../model";
+import { contentDependentDefinitions } from "../parent-visibility";
+import { contentThreadTarget } from "../thread-access";
+import { isContentBaseRowPublic } from "../visibility";
+
+/** One committed content change, as `publishContentRefresh` reads it. */
+export interface ContentRefreshChange {
+ action: ContentEventAction;
+ contentId: ContentId;
+ definition: AnyContentTypeDefinition;
+ /** Whether the record was public before and after; derived from `row` when absent. */
+ publicState?: ContentPublicState;
+ /**
+ * The record after the write - or, for a delete, as it was just before it.
+ * A threaded record needs its root and parent to be routed, and a deleted
+ * one can no longer be read back, so pass it on every delete.
+ */
+ row?: null | Record;
+}
+
+/** Actions after which a thread's root may be readable by different people. */
+const ROOT_ACCESS_ACTIONS: ReadonlySet = new Set([
+ "deleted",
+ "hidden",
+ "published",
+ "restored",
+ "unhidden",
+ "unpublished",
+]);
+
+/** Actions that can only take a record off the public site. */
+const REMOVING_ACTIONS: ReadonlySet = new Set([
+ "deleted",
+ "hidden",
+ "unpublished",
+]);
+
+type RootChangedListener = (room: string) => Promise | void;
+
+const rootChangedListeners = new Set();
+
+/**
+ * Runs `listener` with the room of every thread whose root changed in a way
+ * that can move access, on this instance. The refresh socket re-checks that
+ * room's members with it. Returns an unsubscribe function.
+ */
+export const onContentRefreshRootChanged = (
+ listener: RootChangedListener,
+): (() => void) => {
+ rootChangedListeners.add(listener);
+
+ return () => {
+ rootChangedListeners.delete(listener);
+ };
+};
+
+const send = (scope: ContentRefreshScope, signal: ContentRefreshSignal) => {
+ const room = contentRefreshRoom(scope);
+ realtime.toRoom(room, contentRefreshChannel, {
+ room,
+ signal,
+ type: "changed",
+ } satisfies ContentRefreshServerMessage);
+
+ return room;
+};
+
+/** Without Redis, nobody outside this instance can be listening. */
+const hasListeners = (scope: ContentRefreshScope): boolean =>
+ isRealtimePubSubEnabled() ||
+ wsRegistry.membersOf(contentRefreshRoom(scope)).length > 0;
+
+const publicStateOf = ({
+ action,
+ publicState,
+ row,
+}: ContentRefreshChange): ContentPublicState => {
+ if (publicState) return publicState;
+ // Unknown: assume a record that can only have left the site was on it, and
+ // judge everything else by the row. A draft's edit stays unannounced.
+ const rowIsPublic = row ? isContentBaseRowPublic(row) : false;
+
+ return {
+ isPublic: !REMOVING_ACTIONS.has(action) && rowIsPublic,
+ wasPublic: REMOVING_ACTIONS.has(action) || rowIsPublic,
+ };
+};
+
+const MAX_PARENT_HOPS = 64;
+
+/**
+ * Where one threaded record sits: its root and the top-level reply its branch
+ * hangs from. Reads the record when `row` lacks the thread columns, then walks
+ * up its parents - each one a primary-key read, at most `MAX_PARENT_HOPS`.
+ */
+const threadLocation = async (
+ c: Context,
+ definition: AnyContentTypeDefinition,
+ contentId: ContentId,
+ row: null | Record | undefined,
+): Promise => {
+ const target = contentThreadTarget(definition);
+ const model = findContentModel(
+ c.get("core")?.contentModels ?? [],
+ definition.id,
+ )?.model;
+ if (!target || !model) return null;
+
+ const { parentField, rootField } = definition.thread;
+ const table: PgTable = model.table;
+ const columns = model.columns as Record;
+ const db = c.get("db");
+ const read = async (id: unknown) => {
+ const [found] = await db
+ .select({ parent: columns[parentField], root: columns[rootField] })
+ .from(table)
+ .where(eq(columns.id, id as never))
+ .limit(1);
+
+ return found as undefined | { parent: unknown; root: unknown };
+ };
+
+ const own =
+ row && Object.hasOwn(row, rootField) && Object.hasOwn(row, parentField)
+ ? { parent: row[parentField], root: row[rootField] }
+ : await read(contentId);
+ if (!own) return null;
+
+ const rootId = parseContentId(target.idStrategy, own.root);
+ if (
+ rootId === null ||
+ !hasListeners({ contentTypeId: definition.id, kind: "thread", rootId })
+ ) {
+ return null;
+ }
+
+ let branchId: ContentId = contentId;
+ let parent = own.parent;
+ for (let hop = 0; parent !== null && parent !== undefined; hop++) {
+ const parentId = parseContentId(definition.idStrategy, parent);
+ if (parentId === null || hop >= MAX_PARENT_HOPS) return null;
+ branchId = parentId;
+ parent = (await read(parentId))?.parent;
+ }
+
+ return { branchId, rootId };
+};
+
+/**
+ * Announces one committed content change to the live refresh rooms that show
+ * it, over the shared WebSocket (and Redis, across instances):
+ *
+ * - every change to the content type's AdminCP list room, as `list`;
+ * - a change that was or is public, for a threaded type, to its thread's
+ * room, as `branch` with the top-level branch it sits in;
+ * - a change that can move who reads a record, to the thread rooms hanging
+ * off it, as `thread` - and this instance re-checks those rooms' members.
+ *
+ * A signal names no values, no author and no private record: a branch id is
+ * only sent when that branch was or is on the public thread. Call it after the
+ * transaction committed and the caches were expired, so a receiver that reads
+ * again gets the new state; `emitContentEvent` does, for every write that goes
+ * through it. Best effort - never throws, and nothing is replayed to a client
+ * that was not connected. Server-only.
+ */
+export const publishContentRefresh = async (
+ c: Context,
+ change: ContentRefreshChange,
+): Promise => {
+ if (!isWebSocketEnabled()) return;
+
+ try {
+ const { action, contentId, definition, row } = change;
+ send({ contentTypeId: definition.id, kind: "admin" }, { kind: "list" });
+
+ const state = publicStateOf(change);
+ const touchesPublic = state.wasPublic || state.isPublic;
+
+ if (definition.thread.enabled && touchesPublic) {
+ const location = await threadLocation(c, definition, contentId, row);
+ if (location) {
+ send(
+ {
+ contentTypeId: definition.id,
+ kind: "thread",
+ rootId: location.rootId,
+ },
+ {
+ branchId: location.branchId,
+ kind: "branch",
+ ...(state.wasPublic && !state.isPublic
+ ? { removedId: contentId }
+ : {}),
+ },
+ );
+ }
+ }
+
+ if (
+ touchesPublic &&
+ (ROOT_ACCESS_ACTIONS.has(action) || state.wasPublic !== state.isPublic)
+ ) {
+ for (const dependent of contentDependentDefinitions(c, definition)) {
+ if (contentThreadTarget(dependent)?.id !== definition.id) continue;
+ const scope: ContentRefreshScope = {
+ contentTypeId: dependent.id,
+ kind: "thread",
+ rootId: contentId,
+ };
+ if (!hasListeners(scope)) continue;
+ const room = send(scope, { kind: "thread" });
+ await Promise.all(
+ [...rootChangedListeners].map(async listener => listener(room)),
+ );
+ }
+ }
+ } catch (error) {
+ // The write has committed; a missed hint costs freshness, never data.
+ try {
+ await c
+ .get("log")
+ .error(
+ `[content-refresh] ${change.definition.id}: live signal failed: ${error instanceof Error ? error.message : String(error)}`,
+ );
+ } catch {
+ // oxlint-disable-next-line no-console
+ console.error("[VitNode] Content live refresh signal failed:", error);
+ }
+ }
+};
diff --git a/packages/vitnode/src/content/server/refresh/server.test.ts b/packages/vitnode/src/content/server/refresh/server.test.ts
new file mode 100644
index 000000000..1e77760ca
--- /dev/null
+++ b/packages/vitnode/src/content/server/refresh/server.test.ts
@@ -0,0 +1,344 @@
+import type { Context } from "hono";
+import type { WSContext } from "hono/ws";
+
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+import type {
+ ContentRefreshScope,
+ ContentRefreshServerMessage,
+} from "@/content/live/refresh";
+
+import {
+ CONTENT_REFRESH_MAX_JOINS_PER_MINUTE,
+ CONTENT_REFRESH_MAX_SUBSCRIPTIONS,
+ contentRefreshRoom,
+} from "@/content/live/refresh";
+import { wsRegistry } from "@/ws/registry";
+
+import type { ContentRefreshAuthorization } from "./auth";
+
+import { createContentRefreshServer } from "./server";
+
+const REAUTH_MS = 60_000;
+
+const ADMIN: ContentRefreshScope = {
+ contentTypeId: "blog.post",
+ kind: "admin",
+};
+const THREAD: ContentRefreshScope = {
+ contentTypeId: "blog.comment",
+ kind: "thread",
+ rootId: 7,
+};
+
+const cleanups: (() => void)[] = [];
+
+afterEach(() => {
+ cleanups.splice(0).forEach(cleanup => cleanup());
+});
+
+const setup = ({
+ shareKey = () => null,
+}: {
+ shareKey?: (c: Context, scope: ContentRefreshScope) => null | string;
+} = {}) => {
+ let at = 1_000_000;
+ const denied = new Map();
+ const authorize = vi.fn(
+ async (c: Context): Promise =>
+ await Promise.resolve(denied.get(c) ?? { ok: true }),
+ );
+ const server = createContentRefreshServer({
+ authorize,
+ now: () => at,
+ reauthMs: REAUTH_MS,
+ shareKey,
+ });
+ const stopClose = wsRegistry.onConnectionClose(server.onConnectionClose);
+ const sockets: WSContext[] = [];
+ cleanups.push(() => {
+ server.stop();
+ stopClose();
+ sockets.forEach(ws => wsRegistry.remove(ws));
+ });
+
+ const socket = () => {
+ const received: ContentRefreshServerMessage[] = [];
+ const ws = {
+ readyState: 1,
+ send: (raw: string) => {
+ received.push(
+ (JSON.parse(raw) as { data: ContentRefreshServerMessage }).data,
+ );
+ },
+ } as unknown as WSContext;
+ const c = {} as Context;
+ wsRegistry.add(ws, null);
+ sockets.push(ws);
+
+ const send = async (data: unknown) => {
+ await server.handle({ c, data, ws });
+ };
+
+ return {
+ c,
+ join: async (scope: ContentRefreshScope, clientId = "tab-a") => {
+ await send({ clientId, scope, type: "join" });
+ },
+ leave: async (scope: ContentRefreshScope, clientId = "tab-a") => {
+ await send({ clientId, scope, type: "leave" });
+ },
+ received,
+ send,
+ ws,
+ };
+ };
+
+ const members = (scope: ContentRefreshScope) =>
+ wsRegistry
+ .membersOf(contentRefreshRoom(scope))
+ .map(({ memberId }) => memberId);
+
+ return {
+ advance: (ms: number) => {
+ at += ms;
+ },
+ authorize,
+ deny: (
+ target: { c: Context },
+ code: "FORBIDDEN" | "NOT_FOUND" = "FORBIDDEN",
+ ) => {
+ denied.set(target.c, { code, ok: false });
+ },
+ members,
+ server,
+ socket,
+ };
+};
+
+describe("content refresh joins", () => {
+ it.each([
+ ["an admin list", ADMIN, "content-refresh:admin:blog.post"],
+ [
+ "a thread, with a bigint root kept as its exact string",
+ {
+ contentTypeId: "blog.comment",
+ kind: "thread",
+ rootId: "9223372036854775807",
+ } satisfies ContentRefreshScope,
+ "content-refresh:thread:blog.comment:9223372036854775807",
+ ],
+ ])("acks an authorized join to %s", async (_label, scope, room) => {
+ const { members, server, socket } = setup();
+ const tab = socket();
+
+ await tab.join(scope);
+
+ expect(tab.received).toStrictEqual([
+ { clientId: "tab-a", room, type: "joined" },
+ ]);
+ expect(members(scope)).toStrictEqual(["tab-a"]);
+ expect(server.memberCount()).toBe(1);
+ });
+
+ it.each(["FORBIDDEN", "NOT_FOUND"] as const)(
+ "answers a refused join with %s and no membership",
+ async code => {
+ const { deny, members, server, socket } = setup();
+ const tab = socket();
+ deny(tab, code);
+
+ await tab.join(THREAD);
+
+ expect(tab.received).toStrictEqual([
+ {
+ clientId: "tab-a",
+ code,
+ room: contentRefreshRoom(THREAD),
+ type: "error",
+ },
+ ]);
+ expect(members(THREAD)).toStrictEqual([]);
+ expect(server.memberCount()).toBe(0);
+ },
+ );
+
+ it.each([
+ [
+ "a scope without a content type",
+ { clientId: "tab-a", scope: { kind: "admin" }, type: "join" },
+ "tab-a",
+ ],
+ [
+ "an unknown field",
+ { clientId: "tab-a", extra: 1, scope: ADMIN, type: "join" },
+ "tab-a",
+ ],
+ [
+ "an unknown type",
+ { clientId: "tab-a", scope: ADMIN, type: "subscribe" },
+ "tab-a",
+ ],
+ ["no object at all", "join", ""],
+ ])("answers %s with INVALID_MESSAGE", async (_label, data, clientId) => {
+ const { server, socket } = setup();
+ const tab = socket();
+
+ await tab.send(data);
+
+ expect(tab.received).toStrictEqual([
+ { clientId, code: "INVALID_MESSAGE", type: "error" },
+ ]);
+ expect(server.memberCount()).toBe(0);
+ });
+
+ it("keeps tabs sharing a socket apart: one leaving keeps the other", async () => {
+ const { members, server, socket } = setup();
+ const tab = socket();
+
+ await tab.join(THREAD, "tab-a");
+ await tab.join(THREAD, "tab-b");
+ await tab.leave(THREAD, "tab-a");
+
+ expect(members(THREAD)).toStrictEqual(["tab-b"]);
+ expect(server.memberCount()).toBe(1);
+
+ await tab.leave(THREAD, "tab-b");
+
+ expect(members(THREAD)).toStrictEqual([]);
+ expect(server.memberCount()).toBe(0);
+ });
+
+ it("refuses a subscription over the limit with TOO_MANY_SUBSCRIPTIONS", async () => {
+ const { authorize, server, socket } = setup();
+ const tab = socket();
+ for (let index = 0; index < CONTENT_REFRESH_MAX_SUBSCRIPTIONS; index++) {
+ await tab.join({ contentTypeId: `type-${index}`, kind: "admin" });
+ }
+ authorize.mockClear();
+
+ await tab.join({ contentTypeId: "one-more", kind: "admin" });
+
+ expect(tab.received.at(-1)).toMatchObject({
+ code: "TOO_MANY_SUBSCRIPTIONS",
+ type: "error",
+ });
+ expect(authorize).not.toHaveBeenCalled();
+ expect(server.memberCount()).toBe(CONTENT_REFRESH_MAX_SUBSCRIPTIONS);
+ });
+
+ it("refuses joins over the per-minute limit with RATE_LIMITED, until the minute passes", async () => {
+ const { advance, socket } = setup();
+ const tab = socket();
+ for (let index = 0; index < CONTENT_REFRESH_MAX_JOINS_PER_MINUTE; index++) {
+ await tab.join(ADMIN);
+ }
+
+ await tab.join(ADMIN);
+
+ expect(tab.received.at(-1)).toMatchObject({
+ code: "RATE_LIMITED",
+ type: "error",
+ });
+
+ advance(60_000);
+ await tab.join(ADMIN);
+
+ expect(tab.received.at(-1)).toMatchObject({ type: "joined" });
+ });
+});
+
+describe("content refresh re-checks", () => {
+ it("revokes a membership that lost access once it is due, and keeps the others", async () => {
+ const { advance, authorize, deny, members, socket, server } = setup();
+ const kept = socket();
+ const lost = socket();
+ await kept.join(THREAD);
+ await lost.join(THREAD, "tab-b");
+ authorize.mockClear();
+ deny(lost, "NOT_FOUND");
+
+ advance(REAUTH_MS - 1);
+ await server.sweep();
+
+ expect(authorize).not.toHaveBeenCalled();
+
+ advance(1);
+ await server.sweep();
+
+ expect(lost.received.at(-1)).toStrictEqual({
+ clientId: "tab-b",
+ code: "NOT_FOUND",
+ room: contentRefreshRoom(THREAD),
+ type: "revoked",
+ });
+ expect(members(THREAD)).toStrictEqual(["tab-a"]);
+ expect(kept.received.at(-1)).toMatchObject({ type: "joined" });
+ });
+
+ it("re-checks one room at once, asking once per share key", async () => {
+ const shareKey = vi.fn(
+ (c: Context) => (c as unknown as { key?: string }).key ?? null,
+ );
+ const { authorize, deny, members, server, socket } = setup({ shareKey });
+ const guests = [socket(), socket()];
+ const member = socket();
+ const elsewhere = socket();
+ for (const guest of guests) {
+ Object.assign(guest.c, { key: "guest" });
+ await guest.join(THREAD);
+ }
+ Object.assign(member.c, { key: "user-1" });
+ await member.join(THREAD);
+ await elsewhere.join(ADMIN);
+ authorize.mockClear();
+ guests.forEach(guest => {
+ deny(guest);
+ });
+
+ await server.revalidateRoom(contentRefreshRoom(THREAD));
+
+ expect(authorize).toHaveBeenCalledTimes(2);
+ expect(authorize.mock.calls.map(([c]) => c)).not.toContain(elsewhere.c);
+ expect(members(THREAD)).toHaveLength(1);
+ expect(members(ADMIN)).toStrictEqual(["tab-a"]);
+ });
+
+ it("forgets every membership of a closed socket", async () => {
+ const { authorize, server, socket } = setup();
+ const tab = socket();
+ await tab.join(THREAD);
+ authorize.mockClear();
+
+ wsRegistry.remove(tab.ws);
+ await server.revalidateRoom(contentRefreshRoom(THREAD));
+
+ expect(server.memberCount()).toBe(0);
+ expect(authorize).not.toHaveBeenCalled();
+ });
+
+ it("does not register a socket that closed while its join was checked", async () => {
+ const { authorize, members, server, socket } = setup();
+ const tab = socket();
+ let answer: (value: ContentRefreshAuthorization) => void = () => undefined;
+ authorize.mockImplementationOnce(
+ async () =>
+ await new Promise(resolve => {
+ answer = resolve;
+ }),
+ );
+
+ const joining = tab.join(THREAD);
+ await vi.waitFor(() => {
+ expect(authorize).toHaveBeenCalledOnce();
+ });
+ Object.assign(tab.ws, { readyState: 3 });
+ wsRegistry.remove(tab.ws);
+ answer({ ok: true });
+ await joining;
+
+ expect(tab.received).toStrictEqual([]);
+ expect(members(THREAD)).toStrictEqual([]);
+ expect(server.memberCount()).toBe(0);
+ });
+});
diff --git a/packages/vitnode/src/content/server/refresh/server.ts b/packages/vitnode/src/content/server/refresh/server.ts
new file mode 100644
index 000000000..30c89ac0d
--- /dev/null
+++ b/packages/vitnode/src/content/server/refresh/server.ts
@@ -0,0 +1,330 @@
+import type { Context } from "hono";
+import type { WSContext } from "hono/ws";
+
+import { z } from "zod";
+
+import type {
+ ContentRefreshClientMessage,
+ ContentRefreshErrorCode,
+ ContentRefreshScope,
+ ContentRefreshServerMessage,
+} from "@/content/live/refresh";
+
+import { contentAnyIdSchema } from "@/content/ids";
+import {
+ CONTENT_REFRESH_MAX_JOINS_PER_MINUTE,
+ CONTENT_REFRESH_MAX_SUBSCRIPTIONS,
+ CONTENT_REFRESH_REAUTH_MS,
+ CONTENT_REFRESH_SWEEP_MS,
+ contentRefreshChannel,
+ contentRefreshRoom,
+} from "@/content/live/refresh";
+import { wsRegistry } from "@/ws/registry";
+
+import type { ContentRefreshAuthorization } from "./auth";
+
+import { authorizeContentRefresh, contentRefreshShareKey } from "./auth";
+
+const contentTypeId = z.string().min(1).max(100);
+
+const scopeSchema = z.discriminatedUnion("kind", [
+ z.strictObject({ contentTypeId, kind: z.literal("admin") }),
+ z.strictObject({
+ contentTypeId,
+ kind: z.literal("thread"),
+ rootId: contentAnyIdSchema,
+ }),
+]) satisfies z.ZodType;
+
+const clientId = z.string().min(1).max(64);
+
+const clientMessageSchema = z.strictObject({
+ clientId,
+ scope: scopeSchema,
+ type: z.enum(["join", "leave"]),
+}) satisfies z.ZodType;
+
+const JOIN_WINDOW_MS = 60_000;
+
+interface Membership {
+ c: Context;
+ checkedAt: number;
+ clientIds: Set;
+ room: string;
+ scope: ContentRefreshScope;
+ ws: WSContext;
+}
+
+const logError = (message: string, error: unknown): void => {
+ // oxlint-disable-next-line no-console
+ console.error(message, error);
+};
+
+const sendTo = (ws: WSContext, data: ContentRefreshServerMessage): void => {
+ try {
+ ws.send(JSON.stringify({ data, id: contentRefreshChannel.id }));
+ } catch {
+ // The socket closed meanwhile; its close handling cleans up.
+ }
+};
+
+const clientIdOf = (data: unknown): string => {
+ if (typeof data !== "object" || data === null || !("clientId" in data)) {
+ return "";
+ }
+ const parsed = clientId.safeParse(data.clientId);
+
+ return parsed.success ? parsed.data : "";
+};
+
+/**
+ * The server half of Content Engine live refresh: who follows which scope on
+ * this instance. A room name is only a routing key - every join is
+ * authorized, each membership is checked again every
+ * `CONTENT_REFRESH_REAUTH_MS` (so a signed-out, expired or demoted viewer stops
+ * receiving signals) and at once when its thread's root changes. A failed
+ * check answers `revoked` and leaves the room. Memberships are per tab
+ * (`clientId`), so tabs sharing one connection never ride on each other's
+ * joins.
+ */
+export const createContentRefreshServer = ({
+ authorize = authorizeContentRefresh,
+ now = Date.now,
+ reauthMs = CONTENT_REFRESH_REAUTH_MS,
+ shareKey = contentRefreshShareKey,
+}: {
+ authorize?: (
+ c: Context,
+ scope: ContentRefreshScope,
+ ) => Promise;
+ now?: () => number;
+ reauthMs?: number;
+ shareKey?: (c: Context, scope: ContentRefreshScope) => null | string;
+} = {}) => {
+ const bySocket = new Map>();
+ const byRoom = new Map>();
+ const joins = new WeakMap();
+ const queues = new WeakMap>();
+ let sweeper: ReturnType | undefined;
+
+ const countOf = (ws: WSContext): number =>
+ [...(bySocket.get(ws)?.values() ?? [])].reduce(
+ (total, membership) => total + membership.clientIds.size,
+ 0,
+ );
+
+ const removeFromRoom = (membership: Membership): void => {
+ const members = byRoom.get(membership.room);
+ members?.delete(membership);
+ if (members?.size === 0) byRoom.delete(membership.room);
+ };
+
+ const drop = (membership: Membership, id: string): void => {
+ wsRegistry.leave(membership.ws, membership.room, id);
+ membership.clientIds.delete(id);
+ if (membership.clientIds.size > 0) return;
+
+ const rooms = bySocket.get(membership.ws);
+ rooms?.delete(membership.room);
+ if (rooms?.size === 0) bySocket.delete(membership.ws);
+ removeFromRoom(membership);
+ };
+
+ const revoke = (
+ membership: Membership,
+ code: "FORBIDDEN" | "NOT_FOUND",
+ ): void => {
+ for (const id of [...membership.clientIds]) {
+ drop(membership, id);
+ sendTo(membership.ws, {
+ clientId: id,
+ code,
+ room: membership.room,
+ type: "revoked",
+ });
+ }
+ };
+
+ /** Re-checks memberships, asking once per group of identical answers. */
+ const recheck = async (memberships: Membership[]): Promise => {
+ const shared = new Map>();
+
+ await Promise.all(
+ memberships.map(async membership => {
+ const key = shareKey(membership.c, membership.scope);
+ let pending = key === null ? undefined : shared.get(key);
+ if (!pending) {
+ pending = authorize(membership.c, membership.scope).catch(
+ (error: unknown): ContentRefreshAuthorization => {
+ logError("Content refresh authorization error:", error);
+
+ return { code: "FORBIDDEN", ok: false };
+ },
+ );
+ if (key !== null) shared.set(key, pending);
+ }
+ const result = await pending;
+ if (result.ok) membership.checkedAt = now();
+ else revoke(membership, result.code);
+ }),
+ );
+ };
+
+ const sweep = async (): Promise => {
+ const due = [...bySocket.values()]
+ .flatMap(rooms => [...rooms.values()])
+ .filter(membership => membership.checkedAt + reauthMs <= now());
+ if (due.length > 0) await recheck(due);
+ };
+
+ const startSweeper = (): void => {
+ if (sweeper) return;
+ sweeper = setInterval(() => {
+ sweep().catch((error: unknown) => {
+ logError("Content refresh sweep error:", error);
+ });
+ }, CONTENT_REFRESH_SWEEP_MS);
+ sweeper.unref?.();
+ };
+
+ const allowJoin = (ws: WSContext): boolean => {
+ const recent = (joins.get(ws) ?? []).filter(
+ at => at > now() - JOIN_WINDOW_MS,
+ );
+ const allowed = recent.length < CONTENT_REFRESH_MAX_JOINS_PER_MINUTE;
+ if (allowed) recent.push(now());
+ joins.set(ws, recent);
+
+ return allowed;
+ };
+
+ const handleMessage = async (
+ c: Context,
+ ws: WSContext,
+ message: ContentRefreshClientMessage,
+ ): Promise => {
+ const room = contentRefreshRoom(message.scope);
+ const existing = bySocket.get(ws)?.get(room);
+
+ if (message.type === "leave") {
+ if (existing?.clientIds.has(message.clientId)) {
+ drop(existing, message.clientId);
+ }
+
+ return;
+ }
+
+ const fail = (code: ContentRefreshErrorCode): void => {
+ sendTo(ws, { clientId: message.clientId, code, room, type: "error" });
+ };
+
+ if (!allowJoin(ws)) {
+ fail("RATE_LIMITED");
+
+ return;
+ }
+ if (
+ !existing?.clientIds.has(message.clientId) &&
+ countOf(ws) >= CONTENT_REFRESH_MAX_SUBSCRIPTIONS
+ ) {
+ fail("TOO_MANY_SUBSCRIPTIONS");
+
+ return;
+ }
+
+ const result = await authorize(c, message.scope);
+ // Closing or closed while authorizing: its close cleanup already ran.
+ if (ws.readyState >= 2) return;
+ if (!result.ok) {
+ const current = bySocket.get(ws)?.get(room);
+ if (current?.clientIds.has(message.clientId)) {
+ drop(current, message.clientId);
+ }
+ fail(result.code);
+
+ return;
+ }
+
+ const rooms = bySocket.get(ws) ?? new Map();
+ bySocket.set(ws, rooms);
+ let membership = rooms.get(room);
+ if (!membership) {
+ membership = {
+ c,
+ checkedAt: now(),
+ clientIds: new Set(),
+ room,
+ scope: message.scope,
+ ws,
+ };
+ rooms.set(room, membership);
+ byRoom.set(
+ room,
+ (byRoom.get(room) ?? new Set()).add(membership),
+ );
+ }
+ membership.checkedAt = now();
+ membership.clientIds.add(message.clientId);
+ wsRegistry.join(ws, room, message.clientId);
+
+ sendTo(ws, { clientId: message.clientId, room, type: "joined" });
+ startSweeper();
+ };
+
+ return {
+ handle: async ({
+ c,
+ data,
+ ws,
+ }: {
+ c: Context;
+ data: unknown;
+ ws: WSContext;
+ }): Promise => {
+ const parsed = clientMessageSchema.safeParse(data);
+ if (!parsed.success) {
+ sendTo(ws, {
+ clientId: clientIdOf(data),
+ code: "INVALID_MESSAGE",
+ type: "error",
+ });
+
+ return;
+ }
+
+ // One message at a time per connection, so a leave never overtakes
+ // the join it undoes.
+ const previous = queues.get(ws) ?? Promise.resolve();
+ const settled = previous
+ .then(async () => {
+ await handleMessage(c, ws, parsed.data);
+ })
+ .catch((error: unknown) => {
+ logError("Content refresh message error:", error);
+ });
+ queues.set(ws, settled);
+ await settled;
+ },
+ memberCount: (): number =>
+ [...bySocket.keys()].reduce((total, ws) => total + countOf(ws), 0),
+ onConnectionClose: (ws: WSContext): void => {
+ const rooms = bySocket.get(ws);
+ if (!rooms) return;
+ bySocket.delete(ws);
+ for (const membership of rooms.values()) removeFromRoom(membership);
+ },
+ /**
+ * Checks every local member of `room` again now - after its thread's
+ * root changed - and revokes the ones that lost access.
+ */
+ revalidateRoom: async (room: string): Promise => {
+ const members = [...(byRoom.get(room) ?? [])];
+ if (members.length > 0) await recheck(members);
+ },
+ stop: (): void => {
+ clearInterval(sweeper);
+ sweeper = undefined;
+ },
+ sweep,
+ };
+};
diff --git a/packages/vitnode/src/content/server/refresh/socket.ts b/packages/vitnode/src/content/server/refresh/socket.ts
new file mode 100644
index 000000000..4ddeb61de
--- /dev/null
+++ b/packages/vitnode/src/content/server/refresh/socket.ts
@@ -0,0 +1,41 @@
+import { z } from "zod";
+
+import { buildWebSocket } from "@/api/lib/websocket";
+import {
+ CONTENT_REFRESH_ROOM_PREFIX,
+ contentRefreshChannel,
+} from "@/content/live/refresh";
+import { onRemoteRoomMessage, wsRegistry } from "@/ws/registry";
+
+import { onContentRefreshRootChanged } from "./publish";
+import { createContentRefreshServer } from "./server";
+
+const contentRefreshServer = createContentRefreshServer();
+
+const threadChangedMessage = z.object({
+ signal: z.object({ kind: z.literal("thread") }),
+});
+
+wsRegistry.onConnectionClose(contentRefreshServer.onConnectionClose);
+onContentRefreshRootChanged(contentRefreshServer.revalidateRoom);
+
+// A root changed on another instance: its members here are checked again too.
+onRemoteRoomMessage((room, id, data) => {
+ if (
+ id === contentRefreshChannel.id &&
+ room.startsWith(CONTENT_REFRESH_ROOM_PREFIX) &&
+ threadChangedMessage.safeParse(data).success
+ ) {
+ contentRefreshServer.revalidateRoom(room).catch((error: unknown) => {
+ // oxlint-disable-next-line no-console
+ console.error("Content refresh revalidation error:", error);
+ });
+ }
+});
+
+export const contentRefreshWebSocket = buildWebSocket({
+ description:
+ "Content Engine live refresh: authorized subscriptions to AdminCP lists and public threads, answered with small change signals.",
+ id: "refresh",
+ onMessage: contentRefreshServer.handle,
+});
diff --git a/packages/vitnode/src/content/server/routes.ts b/packages/vitnode/src/content/server/routes.ts
index 0e45eb2cf..39ef89736 100644
--- a/packages/vitnode/src/content/server/routes.ts
+++ b/packages/vitnode/src/content/server/routes.ts
@@ -914,7 +914,11 @@ export const buildContentRoutes = <
definition,
"created",
{ contentId: row.id },
- { pluginId, publicState: contentPublicStateOf(definition, null, row) },
+ {
+ pluginId,
+ publicState: contentPublicStateOf(definition, null, row),
+ row,
+ },
);
// A new record is a draft, so this normally indexes a private document -
@@ -2046,7 +2050,11 @@ export const buildContentRoutes = <
definition,
"deleted",
{ contentId: row.id },
- { pluginId, publicState: contentPublicStateOf(definition, row, null) },
+ {
+ pluginId,
+ publicState: contentPublicStateOf(definition, row, null),
+ row,
+ },
);
// Drafts are indexed as private documents, so every delete removes one - a
diff --git a/packages/vitnode/src/content/server/thread-access.ts b/packages/vitnode/src/content/server/thread-access.ts
new file mode 100644
index 000000000..f23a91717
--- /dev/null
+++ b/packages/vitnode/src/content/server/thread-access.ts
@@ -0,0 +1,68 @@
+import type { Context } from "hono";
+
+import type { ContentId } from "../ids";
+import type { AnyContentTypeDefinition } from "../types";
+
+import { parseContentId } from "../ids";
+import { canViewHiddenContentRecord } from "./hidden-access";
+
+/** A thread root the current reader may open, from {@link resolveContentThreadRoot}. */
+export interface ContentThreadRootAccess {
+ /** `true` when the root is hidden and readable only to this reader. */
+ hidden: boolean;
+ rootId: ContentId;
+ /** The content type the thread hangs off. */
+ target: AnyContentTypeDefinition;
+}
+
+/** The content type a thread's `root` relation points at, if it has one. */
+export const contentThreadTarget = (
+ definition: AnyContentTypeDefinition,
+): AnyContentTypeDefinition | undefined => {
+ if (!definition.thread.enabled) return undefined;
+ const rootField = definition.fields[definition.thread.rootField];
+
+ return rootField?.kind === "relation" ? rootField.target() : undefined;
+};
+
+/**
+ * Whether the current reader may open the thread of `definition` under
+ * `root`: the root is published and visible, or hidden and readable to them
+ * (an author, an editor or a moderator). The single rule behind `GET /thread`,
+ * `GET /thread/branches` and live thread subscriptions. `null` for a missing,
+ * malformed, unpublished or unreadable root - callers answer all of them the
+ * same way, so none confirms that a draft exists. Server-only.
+ */
+export const resolveContentThreadRoot = async (
+ c: Context,
+ definition: AnyContentTypeDefinition,
+ root: unknown,
+): Promise => {
+ const target = contentThreadTarget(definition);
+ if (!target) return null;
+
+ const rootId = parseContentId(target.idStrategy, root);
+ const rootService = (c.get("core")?.contentModels ?? [])
+ .find(entry => entry.model.definition.id === target.id)
+ ?.model.publicService?.(c);
+ if (rootId === null || !rootService) return null;
+
+ // An existence check: `select: []` reads the same published, visible row
+ // without hydrating a single field of it.
+ if (await rootService.findById(rootId, { select: [] })) {
+ return { hidden: false, rootId, target };
+ }
+
+ const hidden = await rootService.findById(rootId, {
+ hidden: true,
+ select: [],
+ });
+
+ return hidden &&
+ (await canViewHiddenContentRecord(c, {
+ contentTypeId: target.id,
+ itemId: rootId,
+ }))
+ ? { hidden: true, rootId, target }
+ : null;
+};
diff --git a/packages/vitnode/src/content/server/thread-routes.test.ts b/packages/vitnode/src/content/server/thread-routes.test.ts
index f888deb18..73f291409 100644
--- a/packages/vitnode/src/content/server/thread-routes.test.ts
+++ b/packages/vitnode/src/content/server/thread-routes.test.ts
@@ -1,16 +1,18 @@
import { describe, expect, it } from "vitest";
+import { CONTENT_THREAD_BRANCHES_MAX } from "@/content/live/refresh";
import {
BOARD,
createPolicyWorld,
POLICY_LIVE,
policyReplyRow,
policyTopicRow,
+ USER,
} from "@/tests/content-policies";
-import type { ContentThreadPage } from "./thread";
+import type { ContentThreadBranches, ContentThreadPage } from "./thread";
-const TOPIC = { DRAFT: 12, HIDDEN: 13, LIVE: 10 } as const;
+const TOPIC = { DRAFT: 12, HIDDEN: 13, LIVE: 10, OTHER: 11 } as const;
const draft = { publishedAt: null, status: "draft" };
@@ -21,6 +23,8 @@ const draft = { publishedAt: null, status: "draft" };
* 103 (draft) ─ 104
* 105 (hidden, no answers)
* 106 ─ 107 (draft)
+ *
+ * Topic 11: 200
*/
const { send } = await createPolicyWorld({
replies: [
@@ -32,9 +36,11 @@ const { send } = await createPolicyWorld({
policyReplyRow(105, TOPIC.LIVE, { hiddenAt: POLICY_LIVE }),
policyReplyRow(106, TOPIC.LIVE),
policyReplyRow(107, TOPIC.LIVE, { parent: 106, ...draft }),
+ policyReplyRow(200, TOPIC.OTHER),
],
topics: [
policyTopicRow(TOPIC.LIVE, BOARD.OPEN),
+ policyTopicRow(TOPIC.OTHER, BOARD.OPEN),
policyTopicRow(TOPIC.DRAFT, BOARD.OPEN, draft),
policyTopicRow(TOPIC.HIDDEN, BOARD.OPEN, { hiddenAt: POLICY_LIVE }),
],
@@ -110,3 +116,99 @@ describe("GET /replies/thread", () => {
]);
});
});
+
+type Branches = ContentThreadBranches<{ body: string; id: number }>;
+
+const readBranches = async (ids: string, user?: number) => {
+ const response = await send(
+ "GET",
+ `/replies/thread/branches?root=${TOPIC.LIVE}${ids === "" ? "" : `&ids=${ids}`}`,
+ { user },
+ );
+ expect(response.status).toBe(200);
+
+ return {
+ body: (await response.json()) as Branches,
+ cacheControl: response.headers.get("cache-control"),
+ };
+};
+
+describe("GET /replies/thread/branches", () => {
+ it("answers each branch with the same entries /thread places on a page", async () => {
+ const page = await read(`root=${TOPIC.LIVE}`);
+ const { body } = await readBranches("106,100,103");
+
+ const slice = (from: number, to: number) =>
+ page.items.slice(
+ page.items.findIndex(item => item.id === from),
+ to === -1 ? undefined : page.items.findIndex(item => item.id === to),
+ );
+ expect(body.branches).toStrictEqual([
+ { id: 106, items: slice(106, -1), status: "ok" },
+ { id: 100, items: slice(100, 103), status: "ok" },
+ { id: 103, items: slice(103, 106), status: "ok" },
+ ]);
+ expect(body.replyCount).toBe(page.replyCount);
+ expect(body.pageInfo.totalCount).toBe(page.pageInfo.totalCount);
+ });
+
+ it("answers gone for a reply, another root's reply, a hidden reply without answers and a missing id", async () => {
+ const { body } = await readBranches("101,200,105,999");
+
+ expect(body.branches).toStrictEqual([
+ { id: 101, status: "gone" },
+ { id: 200, status: "gone" },
+ { id: 105, status: "gone" },
+ { id: 999, status: "gone" },
+ ]);
+ });
+
+ it("answers only the counts without ids", async () => {
+ const { body } = await readBranches("");
+
+ expect(body).toStrictEqual({
+ branches: [],
+ pageInfo: { totalCount: 2 },
+ replyCount: 5,
+ truncated: false,
+ });
+ });
+
+ it.each([
+ ["a draft", String(TOPIC.DRAFT)],
+ ["a hidden", String(TOPIC.HIDDEN)],
+ ["a missing", "999"],
+ ])("answers 404 for %s root", async (_label, root) => {
+ expect(
+ (await send("GET", `/replies/thread/branches?root=${root}&ids=100`))
+ .status,
+ ).toBe(404);
+ });
+
+ it.each([
+ [
+ "too many ids",
+ Array.from(
+ { length: CONTENT_THREAD_BRANCHES_MAX + 1 },
+ (_, index) => 100 + index,
+ ).join(","),
+ ],
+ ["a malformed id", "100,abc"],
+ ])("answers 400 for %s", async (_label, ids) => {
+ expect(
+ (
+ await send(
+ "GET",
+ `/replies/thread/branches?root=${TOPIC.LIVE}&ids=${ids}`,
+ )
+ ).status,
+ ).toBe(400);
+ });
+
+ it.each([
+ ["a guest", undefined, "no-cache"],
+ ["a reader who sees hidden replies", USER.ROOT, "private, no-store"],
+ ])("lets %s cache it as %s", async (_label, user, cacheControl) => {
+ expect((await readBranches("100", user)).cacheControl).toBe(cacheControl);
+ });
+});
diff --git a/packages/vitnode/src/content/server/thread.ts b/packages/vitnode/src/content/server/thread.ts
index 60a9a72f4..7c8b8c571 100644
--- a/packages/vitnode/src/content/server/thread.ts
+++ b/packages/vitnode/src/content/server/thread.ts
@@ -233,43 +233,37 @@ interface NodeRow extends Record {
status: string;
}
+interface ThreadReadOptions {
+ /**
+ * Hidden replies this reader may see, from `contentThreadHiddenAccess`.
+ * They are kept with `hidden: true`; omit it for the public thread.
+ */
+ hiddenAccess?: ContentPublicHiddenAccess;
+ /**
+ * Rows for these ids, in any order: public ones, plus the hidden ones in
+ * `hiddenIds` the reader may see.
+ */
+ project: (
+ ids: ContentId[],
+ options: { hiddenIds: ContentId[] },
+ ) => Promise