diff --git a/eslint.config.mts b/eslint.config.mts index 8b165ce..d087793 100644 --- a/eslint.config.mts +++ b/eslint.config.mts @@ -158,6 +158,7 @@ export default tseslint.config( 'src/socket/handlers/presenceRateLimit.test.ts', 'src/socket/handlers/presenceRichActivity.test.ts', 'src/socket/handlers/pushAudience.test.ts', + 'src/socket/handlers/readMarkers.test.ts', 'src/socket/handlers/scopeTemplates.test.ts', 'src/socket/handlers/sendAcrossRestart.test.ts', 'src/socket/handlers/sessions.test.ts', diff --git a/src/db/sqlite/connection.ts b/src/db/sqlite/connection.ts index 79e63b5..efeebcb 100644 --- a/src/db/sqlite/connection.ts +++ b/src/db/sqlite/connection.ts @@ -566,6 +566,16 @@ function createSchema(d: DatabaseSync): void { CREATE INDEX IF NOT EXISTS idx_mentions_unseen ON mentions(server_user_id, created_at) WHERE seen_at IS NULL; + -- Position is the message's own (created_at, id) pair, never when anybody looked. No foreign + -- key: deleting the message somebody stopped at must not forget where they were. + CREATE TABLE IF NOT EXISTS read_markers ( + server_user_id TEXT NOT NULL, + conversation_id TEXT NOT NULL, + read_through_at TEXT NOT NULL, + read_through_message_id TEXT NOT NULL, + PRIMARY KEY (server_user_id, conversation_id) + ); + -- A report about a person rather than about one message. -- -- Its own table rather than a nullable message_id on reports, because the diff --git a/src/db/sqlite/index.ts b/src/db/sqlite/index.ts index a171d34..8323708 100644 --- a/src/db/sqlite/index.ts +++ b/src/db/sqlite/index.ts @@ -22,6 +22,7 @@ export * from "./contactPrefs"; export * from "./pushDevices"; export * from "./friends"; export * from "./mentions"; +export * from "./readMarkers"; export * from "./webhooks"; export * from "./mls"; export * from "./discordImports"; diff --git a/src/db/sqlite/mergeGuest.test.ts b/src/db/sqlite/mergeGuest.test.ts index ebe853a..c77fc68 100644 --- a/src/db/sqlite/mergeGuest.test.ts +++ b/src/db/sqlite/mergeGuest.test.ts @@ -6,6 +6,7 @@ import { after, before, describe, it } from "node:test"; import { eitherHasBlocked, blockUser, listBlocks } from "./blocks"; import { getSqliteDb, initSqlite } from "./connection"; +import { listReadMarkers, setReadMarker } from "./readMarkers"; import { createGroupConversation, listConversationMemberIds, openDirectConversation } from "./conversations"; import { recordMentions } from "./mentions"; import { AUTHOR_COLUMNS, mergeGuestIntoAccount } from "./mergeGuest"; @@ -174,6 +175,39 @@ describe("merging a guest into an account that is already a member", () => { assert.deepEqual(mentions.map((m) => m.server_user_id), [account.server_user_id]); }); + it("keeps the further read marker where both had one", async () => { + const { guest, account, guestId, accountId } = await pair("marker"); + /* created_at given rather than taken from the clock: which marker is further is the + whole question, and two sends in one millisecond would decide it by id instead. */ + const at = (conversation: string, iso: string) => + insertMessage({ + conversation_id: conversation, sender_server_id: "someone-else", + text: iso, attachments: null, reactions: null, created_at: new Date(iso), + }); + + const early = await at("general", "2026-03-01T00:00:00.000Z"); + const late = await at("general", "2026-03-01T00:00:10.000Z"); + const guestOnly = await at("mk-guest-only", "2026-03-01T00:00:00.000Z"); + const behind = await at("forum", "2026-03-01T00:00:00.000Z"); + const ahead = await at("forum", "2026-03-01T00:00:10.000Z"); + + await setReadMarker({ serverUserId: account.server_user_id, conversationId: "general", messageId: early.message_id }); + await setReadMarker({ serverUserId: guest.server_user_id, conversationId: "general", messageId: late.message_id }); + await setReadMarker({ serverUserId: guest.server_user_id, conversationId: "mk-guest-only", messageId: guestOnly.message_id }); + await setReadMarker({ serverUserId: guest.server_user_id, conversationId: "forum", messageId: behind.message_id }); + await setReadMarker({ serverUserId: account.server_user_id, conversationId: "forum", messageId: ahead.message_id }); + + mergeGuestIntoAccount(guestId, accountId); + + const kept = new Map( + (await listReadMarkers(account.server_user_id)).map((m) => [m.conversation_id, m.read_through_message_id]), + ); + assert.equal(kept.get("general"), late.message_id, "the guest had read further and the merge lost it"); + assert.equal(kept.get("forum"), ahead.message_id, "the account had read further and the guest overwrote it"); + assert.equal(kept.get("mk-guest-only"), guestOnly.message_id, "a marker only the guest held did not move"); + assert.deepEqual(await listReadMarkers(guest.server_user_id), [], "the guest kept markers after the merge"); + }); + it("carries blocks both ways, revokes the guest's sessions and keeps the stricter mute", async () => { const { guest, guestId, accountId } = await pair("safety"); await upsertUser("account-safety-pest", "Pest"); @@ -215,7 +249,7 @@ describe("merging a guest into an account that is already a member", () => { // Keyed on the pair or on the row itself, and handled one by one in the merge. const handledApart = new Set([ "users.server_user_id", "roles.server_user_id", "conversation_members.server_user_id", - "mentions.server_user_id", "refresh_tokens.server_user_id", + "mentions.server_user_id", "read_markers.server_user_id", "refresh_tokens.server_user_id", "mls_devices.server_user_id", "mls_key_packages.server_user_id", "mls_welcomes.server_user_id", "mls_removed_devices.server_user_id", "push_devices.server_user_id", ]); diff --git a/src/db/sqlite/mergeGuest.ts b/src/db/sqlite/mergeGuest.ts index cf0acb7..d2981ee 100644 --- a/src/db/sqlite/mergeGuest.ts +++ b/src/db/sqlite/mergeGuest.ts @@ -108,6 +108,26 @@ function mergeInTransaction( ).run(from, to); db.prepare(`UPDATE mentions SET server_user_id = ? WHERE server_user_id = ?`).run(to, from); + /* Read markers keep the further of the two rather than the account's, which the two + above do: merging two of your own identities should not lose read progress. */ + db.prepare( + `DELETE FROM read_markers WHERE server_user_id = ? + AND conversation_id IN ( + SELECT g.conversation_id FROM read_markers g + JOIN read_markers a + ON a.server_user_id = ? AND a.conversation_id = g.conversation_id + WHERE g.server_user_id = ? + AND (a.read_through_at > g.read_through_at + OR (a.read_through_at = g.read_through_at + AND a.read_through_message_id >= g.read_through_message_id)))`, + ).run(from, to, from); + // Whatever the guest still holds is further along, so the account's row gives way. + db.prepare( + `DELETE FROM read_markers WHERE server_user_id = ? + AND conversation_id IN (SELECT conversation_id FROM read_markers WHERE server_user_id = ?)`, + ).run(to, from); + db.prepare(`UPDATE read_markers SET server_user_id = ? WHERE server_user_id = ?`).run(to, from); + for (const [table, column] of AUTHOR_COLUMNS) { db.prepare(`UPDATE ${table} SET ${column} = ? WHERE ${column} = ?`).run(to, from); } diff --git a/src/db/sqlite/readMarkers.test.ts b/src/db/sqlite/readMarkers.test.ts new file mode 100644 index 0000000..0315412 --- /dev/null +++ b/src/db/sqlite/readMarkers.test.ts @@ -0,0 +1,171 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { after, before, describe, it } from "node:test"; + +import { initSqlite, getSqliteDb } from "./connection"; +import { insertMessage } from "./messages"; +import { + countUnreadByConversation, + deleteReadMarkers, + listReadMarkers, + setReadMarker, +} from "./readMarkers"; + +/** A real database, like card.test.ts: the counting is a join and a string compare, + and neither is worth faking. */ +let dir: string; + +const ALICE = "alice"; +const BOB = "bob"; + +before(async () => { + dir = mkdtempSync(join(tmpdir(), "gryt-read-")); + process.env.DATA_DIR = dir; + await initSqlite(); +}); + +after(() => { + delete process.env.DATA_DIR; + /* The connection stays open, there being no close for it, and Windows will not unlink + an open file. Every db test here fails its cleanup that way; the OS sweeps temp. */ + try { + rmSync(dir, { recursive: true, force: true }); + } catch { + // Failing a suite on its own cleanup hides the result it just produced. + } +}); + +/** created_at is given, so the order is the test's rather than the clock's. */ +async function say(conversation: string, sender: string, at: string, extra: { thread_id?: string } = {}) { + const stored = await insertMessage({ + conversation_id: conversation, + sender_server_id: sender, + text: `from ${sender} at ${at}`, + attachments: null, + reactions: null, + created_at: new Date(at), + ...extra, + }); + return stored.message_id; +} + +describe("read markers", () => { + it("counts nothing for a conversation nobody has opened", async () => { + await say("c-untouched", BOB, "2026-01-01T00:00:00.000Z"); + const counts = await countUnreadByConversation(ALICE); + assert.equal(counts["c-untouched"], undefined, "a channel never opened is lighting up"); + assert.deepEqual(await listReadMarkers(ALICE), []); + }); + + it("reads everything when no message is named, as Mark as read does", async () => { + await say("c-all", BOB, "2026-01-03T00:00:00.000Z"); + await say("c-all", BOB, "2026-01-03T00:00:01.000Z"); + assert.equal(await setReadMarker({ serverUserId: ALICE, conversationId: "c-all" }), true); + assert.equal((await countUnreadByConversation(ALICE))["c-all"], undefined); + assert.equal(await setReadMarker({ serverUserId: ALICE, conversationId: "c-empty" }), false); + }); + + it("counts what arrived after the marker, and not the marker itself", async () => { + const first = await say("c-basic", BOB, "2026-01-02T00:00:00.000Z"); + await say("c-basic", BOB, "2026-01-02T00:00:01.000Z"); + await say("c-basic", BOB, "2026-01-02T00:00:02.000Z"); + + assert.equal(await setReadMarker({ serverUserId: ALICE, conversationId: "c-basic", messageId: first }), true); + const counts = await countUnreadByConversation(ALICE); + assert.equal(counts["c-basic"], 2); + }); + + it("does not count your own messages", async () => { + const first = await say("c-mine", BOB, "2026-01-03T00:00:00.000Z"); + await setReadMarker({ serverUserId: ALICE, conversationId: "c-mine", messageId: first }); + await say("c-mine", ALICE, "2026-01-03T00:00:01.000Z"); + assert.equal((await countUnreadByConversation(ALICE))["c-mine"], undefined); + + await say("c-mine", BOB, "2026-01-03T00:00:02.000Z"); + assert.equal((await countUnreadByConversation(ALICE))["c-mine"], 1, "somebody else's message is not counted"); + }); + + it("leaves thread replies to the thread's own count", async () => { + const root = await say("c-thread", BOB, "2026-01-04T00:00:00.000Z"); + await setReadMarker({ serverUserId: ALICE, conversationId: "c-thread", messageId: root }); + await say("c-thread", BOB, "2026-01-04T00:00:01.000Z", { thread_id: "t-1" }); + assert.equal((await countUnreadByConversation(ALICE))["c-thread"], undefined); + }); + + it("only ever moves forwards", async () => { + const a = await say("c-back", BOB, "2026-01-05T00:00:00.000Z"); + const b = await say("c-back", BOB, "2026-01-05T00:00:01.000Z"); + + await setReadMarker({ serverUserId: ALICE, conversationId: "c-back", messageId: b }); + assert.equal((await countUnreadByConversation(ALICE))["c-back"], undefined); + + /* The second device finished last but had read less. Without the guard this would + reopen a conversation somebody had already cleared. */ + const moved = await setReadMarker({ serverUserId: ALICE, conversationId: "c-back", messageId: a }); + assert.equal(moved, false, "the marker went backwards"); + assert.equal((await countUnreadByConversation(ALICE))["c-back"], undefined); + }); + + it("breaks a same-millisecond tie on the message id", async () => { + const same = "2026-01-06T00:00:00.000Z"; + const ids = [await say("c-tie", BOB, same), await say("c-tie", BOB, same), await say("c-tie", BOB, same)]; + const sorted = [...ids].sort(); + + await setReadMarker({ serverUserId: ALICE, conversationId: "c-tie", messageId: sorted[0] }); + /* Ids are random UUIDs, so this is the only ordering the pair has. Comparing on + created_at alone would count all three or none. */ + assert.equal((await countUnreadByConversation(ALICE))["c-tie"], 2); + + await setReadMarker({ serverUserId: ALICE, conversationId: "c-tie", messageId: sorted[2] }); + assert.equal((await countUnreadByConversation(ALICE))["c-tie"], undefined); + }); + + it("moves nothing for a message this server does not have", async () => { + const real = await say("c-ghost", BOB, "2026-01-07T00:00:00.000Z"); + await setReadMarker({ serverUserId: ALICE, conversationId: "c-ghost", messageId: real }); + await say("c-ghost", BOB, "2026-01-07T00:00:01.000Z"); + + assert.equal( + await setReadMarker({ serverUserId: ALICE, conversationId: "c-ghost", messageId: "not-a-message" }), + false, + ); + assert.equal((await countUnreadByConversation(ALICE))["c-ghost"], 1, "an unknown id disturbed the marker"); + }); + + it("keeps the marker when the message it points at is deleted", async () => { + const first = await say("c-gone", BOB, "2026-01-08T00:00:00.000Z"); + await say("c-gone", BOB, "2026-01-08T00:00:01.000Z"); + await setReadMarker({ serverUserId: ALICE, conversationId: "c-gone", messageId: first }); + + getSqliteDb().prepare(`DELETE FROM messages WHERE conversation_id = ? AND message_id = ?`).run("c-gone", first); + + /* No foreign key here, deliberately: deleting the message somebody stopped at must + not forget where they were. With a cascade this would read as caught up. */ + assert.equal((await listReadMarkers(ALICE)).some((m) => m.conversation_id === "c-gone"), true); + assert.equal((await countUnreadByConversation(ALICE))["c-gone"], 1); + }); + + it("stores the message's own time, not the time of the read", async () => { + const at = "2026-01-09T12:34:56.000Z"; + const id = await say("c-when", BOB, at); + await setReadMarker({ serverUserId: ALICE, conversationId: "c-when", messageId: id }); + + const marker = (await listReadMarkers(ALICE)).find((m) => m.conversation_id === "c-when"); + assert.equal(marker?.read_through_at, at, "the marker is not the message's timestamp"); + assert.equal(marker?.read_through_message_id, id); + }); + + it("keeps people apart, and forgets one on request", async () => { + const id = await say("c-two", BOB, "2026-01-10T00:00:00.000Z"); + await say("c-two", BOB, "2026-01-10T00:00:01.000Z"); + await setReadMarker({ serverUserId: ALICE, conversationId: "c-two", messageId: id }); + + assert.equal((await listReadMarkers(BOB)).some((m) => m.conversation_id === "c-two"), false); + + assert.equal(await deleteReadMarkers(ALICE, "c-two"), 1); + assert.equal((await countUnreadByConversation(ALICE))["c-two"], undefined); + assert.ok((await listReadMarkers(ALICE)).length > 0, "every marker went, not just the one asked for"); + }); +}); diff --git a/src/db/sqlite/readMarkers.ts b/src/db/sqlite/readMarkers.ts new file mode 100644 index 0000000..7bca518 --- /dev/null +++ b/src/db/sqlite/readMarkers.ts @@ -0,0 +1,114 @@ +import { getSqliteDb } from "./connection"; + +/** + * Where somebody has read up to in a conversation, so unread survives a restart and + * follows them to another device. Channels only: a DM's would be a read receipt. + */ +export interface ReadMarker { + conversation_id: string; + /** The message's own created_at, not when anybody looked. */ + read_through_at: string; + read_through_message_id: string; +} + +/** + * Moves the marker to a message, and only forwards. Two devices reading the same + * conversation race, and the one that finishes last is not the one that read furthest. + */ +export async function setReadMarker(args: { + serverUserId: string; + conversationId: string; + /** Omitted for "mark as read": everything in the conversation. */ + messageId?: string; +}): Promise { + const db = getSqliteDb(); + + /* The message supplies the position. An id this server does not hold -- deleted, or + never here -- moves nothing, rather than parking the marker somewhere unorderable. */ + const at = ( + args.messageId + ? db + .prepare(`SELECT created_at, message_id FROM messages WHERE conversation_id = ? AND message_id = ?`) + .get(args.conversationId, args.messageId) + : db + .prepare( + `SELECT created_at, message_id FROM messages WHERE conversation_id = ? + ORDER BY created_at DESC, message_id DESC LIMIT 1`, + ) + .get(args.conversationId) + ) as { created_at: string; message_id: string } | undefined; + if (!at) return false; + + /* ON CONFLICT with a WHERE, so a marker never goes backwards. created_at is an ISO + string to the millisecond, so message_id breaks the ties it cannot. */ + const result = db + .prepare( + `INSERT INTO read_markers (server_user_id, conversation_id, read_through_at, read_through_message_id) + VALUES (?, ?, ?, ?) + ON CONFLICT(server_user_id, conversation_id) DO UPDATE SET + read_through_at = excluded.read_through_at, + read_through_message_id = excluded.read_through_message_id + WHERE excluded.read_through_at > read_markers.read_through_at + OR (excluded.read_through_at = read_markers.read_through_at + AND excluded.read_through_message_id > read_markers.read_through_message_id)`, + ) + .run(args.serverUserId, args.conversationId, at.created_at, at.message_id); + + return Number(result.changes ?? 0) > 0; +} + +/** Every marker this person has, for a client that wants to count for itself. */ +export async function listReadMarkers(serverUserId: string): Promise { + const db = getSqliteDb(); + return db + .prepare( + `SELECT conversation_id, read_through_at, read_through_message_id + FROM read_markers WHERE server_user_id = ?`, + ) + .all(serverUserId) as unknown as ReadMarker[]; +} + +/** + * How many messages have arrived since each marker. One query rather than one per + * conversation, joined on the index messages already has for its timeline. + */ +export async function countUnreadByConversation( + serverUserId: string, +): Promise> { + const db = getSqliteDb(); + + /* A conversation with no marker is left out, and the client reads absent as caught + up: joining a server should not light up every channel in it. */ + const rows = db + .prepare( + `SELECT m.conversation_id AS conversation_id, COUNT(*) AS n + FROM read_markers r + JOIN messages m + ON m.conversation_id = r.conversation_id + AND (m.created_at > r.read_through_at + OR (m.created_at = r.read_through_at + AND m.message_id > r.read_through_message_id)) + WHERE r.server_user_id = ? + -- Replies belong to their thread's count, which the client keeps apart. + AND m.thread_id IS NULL + -- Your own messages are not news to you. + AND m.sender_server_id != ? + GROUP BY m.conversation_id`, + ) + .all(serverUserId, serverUserId) as Array<{ conversation_id: string; n: number }>; + + const counts: Record = {}; + for (const row of rows) counts[row.conversation_id] = row.n; + return counts; +} + +/** Leaving a server should not leave its markers behind. */ +export async function deleteReadMarkers(serverUserId: string, conversationId?: string): Promise { + const db = getSqliteDb(); + const result = conversationId + ? db + .prepare(`DELETE FROM read_markers WHERE server_user_id = ? AND conversation_id = ?`) + .run(serverUserId, conversationId) + : db.prepare(`DELETE FROM read_markers WHERE server_user_id = ?`).run(serverUserId); + return Number(result.changes ?? 0); +} diff --git a/src/socket/handlers/readMarkers.test.ts b/src/socket/handlers/readMarkers.test.ts new file mode 100644 index 0000000..fed4c0e --- /dev/null +++ b/src/socket/handlers/readMarkers.test.ts @@ -0,0 +1,137 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { after, before, describe, it } from "node:test"; + +import { upsertServerChannel } from "../../db/sqlite/channels"; +import { initSqlite } from "../../db/sqlite/connection"; +import { insertMessage } from "../../db/sqlite/messages"; +import { setReadMarker } from "../../db/sqlite/readMarkers"; +import { createServerConfigIfNotExists, setServerRole } from "../../db/sqlite/servers"; +import { upsertUser } from "../../db/sqlite/users"; +import type { Clients } from "../../types"; +import { refreshClientPermissions } from "../utils/standing"; +import { registerReadMarkerHandlers } from "./readMarkers"; +import type { EventHandlerMap, HandlerContext } from "./types"; + +/* The access check is the part worth a handler test: a marker says where somebody got + to, not that the conversation is still theirs to see. */ + +const HOST = "read.test:5001"; +const CHANNEL = "chan-general"; +const GONE = "chan-no-longer-visible"; + +let dir: string; +const sockets = new Map boolean }>(); +const clientsInfo: Clients = {}; +let handlers: EventHandlerMap; +let userId: string; +let received: (event: string) => unknown[]; +let clientId: string; +let lastMessageId: string; + +before(async () => { + dir = mkdtempSync(join(tmpdir(), "gryt-readh-")); + process.env.DATA_DIR = dir; + await initSqlite(); + await createServerConfigIfNotExists(); + await upsertServerChannel({ channelId: CHANNEL, name: "general", type: "text" }); + const alice = await upsertUser("account-alice", "Alice"); + userId = alice.server_user_id; + await setServerRole(userId, "owner"); + + const first = await insertMessage({ + conversation_id: CHANNEL, + sender_server_id: "someone-else", + text: "hello", + attachments: null, + reactions: null, + created_at: new Date("2026-02-01T00:00:00.000Z"), + }); + lastMessageId = first.message_id; + await insertMessage({ + conversation_id: CHANNEL, + sender_server_id: "someone-else", + text: "and another", + attachments: null, + reactions: null, + created_at: new Date("2026-02-01T00:00:01.000Z"), + }); + + clientId = "socket-1"; + const emitted: { event: string; payload?: unknown }[] = []; + const record = { emit(event: string, payload?: unknown) { emitted.push({ event, payload }); return true; } }; + sockets.set(clientId, record); + const socket = { id: clientId, handshake: { headers: { host: HOST }, address: "127.0.0.1" }, emit: record.emit, join() {}, leave() {}, to: () => ({ emit() {} }) }; + clientsInfo[clientId] = { serverUserId: userId, grytUserId: "account-alice", nickname: "Alice" } as Clients[string]; + await refreshClientPermissions(clientsInfo, clientId); + const ctx = { io: { sockets: { sockets } }, socket, clientId, serverId: "read-test", clientsInfo, sfuClient: null, getClientIp: () => "10.0.0.1", clientAddressIsOwn: () => true } as unknown as HandlerContext; + handlers = registerReadMarkerHandlers(ctx); + received = (event: string) => emitted.filter((e) => e.event === event).map((e) => e.payload); +}); + +after(() => { + delete process.env.DATA_DIR; + try { rmSync(dir, { recursive: true, force: true }); } catch { /* Windows holds the file */ } +}); + +type Reply = { markers: { conversation_id: string }[]; counts: Record }; + +describe("read markers over the socket", () => { + it("starts with nothing, and says so", async () => { + await handlers["read:list"](); + const reply = received("read:list").at(-1) as Reply; + assert.deepEqual(reply.markers, []); + assert.deepEqual(reply.counts, {}); + }); + + it("reads up to a message and counts what is past it", async () => { + await handlers["read:set"]({ conversationId: CHANNEL, messageId: lastMessageId }); + const reply = received("read:list").at(-1) as Reply; + assert.deepEqual(reply.markers.map((m) => m.conversation_id), [CHANNEL]); + assert.equal(reply.counts[CHANNEL], 1); + }); + + it("writes nothing for a conversation the caller cannot reach", async () => { + await handlers["read:set"]({ conversationId: GONE, messageId: lastMessageId }); + const reply = received("read:list").at(-1) as Reply; + assert.equal(reply.markers.some((m) => m.conversation_id === GONE), false, "a guessed id got a marker"); + }); + + it("leaves out a marker for a conversation the caller can no longer reach", async () => { + /* Straight to the db, which is the state somebody left behind when a channel was + deleted or a role taken away. The handler has to drop it on the way out. */ + const stored = await insertMessage({ + conversation_id: GONE, + sender_server_id: "someone-else", + text: "from before", + attachments: null, + reactions: null, + created_at: new Date("2026-02-01T00:00:00.000Z"), + }); + assert.equal(await setReadMarker({ serverUserId: userId, conversationId: GONE, messageId: stored.message_id }), true); + + await handlers["read:list"](); + const reply = received("read:list").at(-1) as Reply; + assert.equal(reply.markers.some((m) => m.conversation_id === GONE), false, "the marker leaked a channel"); + assert.equal(reply.counts[GONE], undefined, "the count leaked a channel"); + }); + + it("ignores a client that has not finished signing in", async () => { + const real = clientsInfo[clientId].serverUserId; + clientsInfo[clientId].serverUserId = "temp_not-yet"; + const before = received("read:list").length; + await handlers["read:list"](); + await handlers["read:set"]({ conversationId: CHANNEL, messageId: lastMessageId }); + assert.equal(received("read:list").length, before, "a temp client got an answer"); + clientsInfo[clientId].serverUserId = real; + }); + + it("wants a conversation", async () => { + const before = received("read:list").length; + await handlers["read:set"]({ messageId: lastMessageId }); + await handlers["read:set"](undefined); + assert.equal(received("read:list").length, before, "a half-filled payload was answered"); + }); +}); diff --git a/src/socket/handlers/readMarkers.ts b/src/socket/handlers/readMarkers.ts new file mode 100644 index 0000000..357b73e --- /dev/null +++ b/src/socket/handlers/readMarkers.ts @@ -0,0 +1,77 @@ +import type { HandlerContext, EventHandlerMap } from "./types"; +import { countUnreadByConversation, listReadMarkers, setReadMarker } from "../../db"; +import { checkRateLimit, type RateLimitRule } from "../../utils/rateLimiter"; +import { resolveConversationAccess } from "../utils/conversationAccess"; + +/* `read:set` is per channel opened, so it is allowed more than a list. Both follow + mentions, which is the same shape of traffic from the same clients. */ +const RL_LIST: RateLimitRule = { limit: 20, windowMs: 30_000, scorePerAction: 0.5, maxScore: 8, scoreDecayMs: 2000 }; +const RL_SET: RateLimitRule = { limit: 60, windowMs: 60_000, scorePerAction: 0.3, maxScore: 10, scoreDecayMs: 1500 }; + +export function registerReadMarkerHandlers(ctx: HandlerContext): EventHandlerMap { + const { socket, clientId, clientsInfo, getClientIp } = ctx; + + /** + * Markers and counts in one reply: a client that has history counts for itself, and + * one that has just installed has none to count. + */ + async function sendState(userId: string) { + const [markers, counts] = await Promise.all([ + listReadMarkers(userId), + countUnreadByConversation(userId), + ]); + + /* Filtered on read, as mentions are: a marker says where somebody got to, not that + the conversation is still theirs to see. A gate can be taken away and given back. */ + const allowed = await Promise.all( + markers.map((m) => resolveConversationAccess(m.conversation_id, userId)), + ); + const visible = markers.filter((_, i) => allowed[i].allowed); + + const open = new Set(visible.map((m) => m.conversation_id)); + const open_counts: Record = {}; + for (const [id, n] of Object.entries(counts)) { + if (open.has(id)) open_counts[id] = n; + } + + socket.emit("read:list", { markers: visible, counts: open_counts }); + } + + return { + /** Everything this person has read up to, and how much has arrived since. */ + "read:list": async () => { + const userId = clientsInfo[clientId]?.serverUserId; + if (!userId || userId.startsWith("temp_")) return; + + const rl = checkRateLimit("read:list", userId, getClientIp(), RL_LIST); + if (!rl.allowed) return; + + await sendState(userId); + }, + + /** + * Read up to a message, or with none, everything there. The reply carries the whole state back, so two devices in + * the same account converge without either guessing what the other did. + */ + "read:set": async (payload: { conversationId?: string; messageId?: string } | undefined) => { + const userId = clientsInfo[clientId]?.serverUserId; + if (!userId || userId.startsWith("temp_")) return; + + const rl = checkRateLimit("read:set", userId, getClientIp(), RL_SET); + if (!rl.allowed) return; + + const { conversationId, messageId } = payload ?? {}; + if (!conversationId) return; + + /* Before the write, or a guessed conversation id tells you whether a message is + in it by whether the marker moved. */ + const access = await resolveConversationAccess(conversationId, userId); + // A DM's marker would be a read receipt on an encrypted message. + if (!access.allowed || access.kind !== "channel") return; + + await setReadMarker({ serverUserId: userId, conversationId, messageId }); + + await sendState(userId); + }, + }; +} diff --git a/src/socket/index.ts b/src/socket/index.ts index 813e262..70570db 100644 --- a/src/socket/index.ts +++ b/src/socket/index.ts @@ -55,6 +55,7 @@ import { registerDmKeyHandlers } from "./handlers/dmKeys"; import { registerMlsHandlers } from "./handlers/mls"; import { registerMlsPersonKeyHandlers } from "./handlers/mlsPersonKey"; import { registerMentionHandlers } from "./handlers/mentions"; +import { registerReadMarkerHandlers } from "./handlers/readMarkers"; import { addressIsOwn, resolveClientIp, trustedProxyHops } from "../config/clientAddress"; export { broadcastChatNew, broadcastCustomEmojisUpdate, broadcastEmojiQueueUpdate, broadcastServerUiUpdate }; @@ -354,6 +355,7 @@ export function socketHandler(io: Server, socket: Socket, sfuClient: SFUClient | ...registerMlsHandlers(ctx), ...registerMlsPersonKeyHandlers(ctx), ...registerMentionHandlers(ctx), + ...registerReadMarkerHandlers(ctx), ...registerPushHandlers(ctx), };