Q142: a conversation is read only up to the message the reader actually saw
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
87c297dd45
commit
b18eb9ad14
11 files changed
+167
-55
No files matched your search
@@ -52,5 +52,5 @@ export const collaborationContracts = {
|
||||
}) satisfies z.ZodType<C.SendMessageInput>,
|
||||
editMessage: z.strictObject({ orgId: id, messageId: id, actorType: sender, actorId: nullableId, bodyMd: z.string() }) satisfies z.ZodType<C.EditMessageInput>,
|
||||
deleteMessage: z.strictObject({ orgId: id, messageId: id, actorType: sender, actorId: nullableId }) satisfies z.ZodType<C.DeleteMessageInput>,
|
||||
markChatSpaceRead: z.strictObject({ orgId: id, employeeId: id, spaceId: id }) satisfies z.ZodType<C.MarkChatSpaceReadInput>,
|
||||
markChatSpaceRead: z.strictObject({ orgId: id, employeeId: id, spaceId: id, upToMessageId: id }) satisfies z.ZodType<C.MarkChatSpaceReadInput>,
|
||||
};
|
||||
@@ -220,22 +220,19 @@ export class ChatService {
|
||||
const employee = this.db.prepare("select id from employees where id = ? and org_id = ?")
|
||||
.get(input.employeeId, input.orgId) as { id: string } | undefined;
|
||||
if (!employee) throw new Error("Employee was not found.");
|
||||
// Read state is per-employee bookkeeping, not access control: the board
|
||||
// observes every conversation in its org (agent↔agent directs included),
|
||||
// so opening one clears its unread count even without membership.
|
||||
const latest = this.db.prepare(`
|
||||
select id from chat_messages
|
||||
where space_id = ? and org_id = ?
|
||||
order by created_at desc
|
||||
limit 1
|
||||
`).get(input.spaceId, input.orgId) as { id: string } | undefined;
|
||||
const now = new Date().toISOString();
|
||||
// Read state is per-employee bookkeeping, not access control: the board reads every conversation in its org.
|
||||
const seen = this.db.prepare("select id, created_at from chat_messages where id = ? and space_id = ? and org_id = ?")
|
||||
.get(input.upToMessageId, input.spaceId, input.orgId) as { id: string; created_at: string } | undefined;
|
||||
if (!seen) throw new Error("The message read is not in this conversation.");
|
||||
// Read up to what was seen, never past it; messages order by time then insertion, and a late mark never moves back.
|
||||
this.db.prepare(`
|
||||
insert into chat_read_state(org_id, employee_id, space_id, last_read_message_id, last_read_at)
|
||||
values (?, ?, ?, ?, ?)
|
||||
on conflict(employee_id, space_id)
|
||||
do update set last_read_message_id = excluded.last_read_message_id, last_read_at = excluded.last_read_at
|
||||
`).run(input.orgId, input.employeeId, input.spaceId, latest?.id ?? null, now);
|
||||
where (excluded.last_read_at, (select rowid from chat_messages where id = excluded.last_read_message_id))
|
||||
> (chat_read_state.last_read_at, coalesce((select rowid from chat_messages where id = chat_read_state.last_read_message_id), -1))
|
||||
`).run(input.orgId, input.employeeId, input.spaceId, seen.id, seen.created_at);
|
||||
}
|
||||
|
||||
editMessage(input: EditMessageInput): void {
|
||||
|
||||
@@ -96,11 +96,11 @@ export class SnapshotService {
|
||||
approvals: list(this.db, "approvals", "created_at desc"),
|
||||
chatSpaces: list(this.db, "chat_spaces", "created_at asc"),
|
||||
chatMemberships: list(this.db, "chat_memberships", "space_id asc, principal_type asc, principal_id asc"),
|
||||
chatMessages: recentList<AppSnapshot["chatMessages"][number]>(this.db, "chat_messages", "created_at desc", 500).reverse(),
|
||||
chatMessages: recentList<AppSnapshot["chatMessages"][number]>(this.db, "chat_messages", "created_at desc, rowid desc", 500).reverse(),
|
||||
// Every thread a loaded message belongs to, plus the most recently active ones, so no shown message loses its thread.
|
||||
chatThreads: all<Row>(this.db.prepare(`
|
||||
select * from chat_threads
|
||||
where id in (select thread_id from (select thread_id from chat_messages order by created_at desc limit 500))
|
||||
where id in (select thread_id from (select thread_id from chat_messages order by created_at desc, rowid desc limit 500))
|
||||
or id in (select id from chat_threads order by updated_at desc limit 200)
|
||||
order by updated_at desc
|
||||
`)).map((row) => camel(row) as unknown as AppSnapshot["chatThreads"][number]),
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { useCallback, useEffect, useMemo, useRef, useState, type FormEvent } from "react";
|
||||
import { useCallback, useMemo, useRef, useState, type FormEvent } from "react";
|
||||
import { Field } from "../shared/ui";
|
||||
import { Dialog } from "../shared/Dialog";
|
||||
import { t } from "../i18n/i18n";
|
||||
@@ -90,21 +90,10 @@ export function ChatPage({ snapshot, actions }: PageProps) {
|
||||
// whichever space just got a message and marking it read unseen.
|
||||
if (!selectedSpaceId && firstAvailableSpaceId) setSelectedSpaceId(firstAvailableSpaceId);
|
||||
const activeSpace = spaces.find((space) => space.id === activeSpaceId) ?? null;
|
||||
const latestActiveMessageId = useMemo(() => {
|
||||
if (!activeSpaceId) return null;
|
||||
const inSpace = messagesAll.filter((message) => message.spaceId === activeSpaceId);
|
||||
if (inSpace.length === 0) return null;
|
||||
return inSpace[inSpace.length - 1].id;
|
||||
}, [messagesAll, activeSpaceId]);
|
||||
|
||||
useEffect(() => {
|
||||
if (!activeSpaceId || !latestActiveMessageId) return;
|
||||
void actions.run(() => window.agentCompany.markChatSpaceRead({
|
||||
orgId,
|
||||
employeeId: employee.id,
|
||||
spaceId: activeSpaceId,
|
||||
}));
|
||||
}, [activeSpaceId, latestActiveMessageId, orgId, employee.id, actions]);
|
||||
// The conversation reports what the reader actually has in view; only that counts as read.
|
||||
const markRead = useCallback((spaceId: string, upToMessageId: string) => {
|
||||
void actions.run(() => window.agentCompany.markChatSpaceRead({ orgId, employeeId: employee.id, spaceId, upToMessageId }));
|
||||
}, [actions, orgId, employee.id]);
|
||||
|
||||
const messagesInSpace = useMemo(
|
||||
() => (activeSpace ? messagesAll.filter((m) => m.spaceId === activeSpace.id) : []),
|
||||
@@ -271,6 +260,7 @@ export function ChatPage({ snapshot, actions }: PageProps) {
|
||||
onDeleteMessage={deleteChatMessage}
|
||||
onOpenFile={openFile}
|
||||
onSubmit={sendInChannel}
|
||||
onRead={markRead}
|
||||
onReplyInThread={replyInThread}
|
||||
send={(activeSpaceId ? sends[activeSpaceId] : undefined) ?? IDLE_SEND}
|
||||
composerRef={composerRef}
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { useLayoutEffect, useMemo, useRef, useState, type FormEvent, type MutableRefObject } from "react";
|
||||
import { useEffect, useLayoutEffect, useMemo, useRef, useState, type FormEvent, type MutableRefObject } from "react";
|
||||
import { AvatarStack, Icon, IconButton } from "../../shared/ui";
|
||||
import type {
|
||||
ApprovalRecord,
|
||||
@@ -40,6 +40,7 @@ export function ChatConversation({
|
||||
onDeleteMessage,
|
||||
onOpenFile,
|
||||
onSubmit,
|
||||
onRead,
|
||||
onReplyInThread,
|
||||
send,
|
||||
composerRef,
|
||||
@@ -66,6 +67,8 @@ export function ChatConversation({
|
||||
onDeleteMessage: (messageId: string) => void;
|
||||
onOpenFile: (fileId: string) => void;
|
||||
onSubmit: (event: FormEvent<HTMLFormElement>) => void;
|
||||
/** Marks the conversation read up to a message the reader has in view; must be stable. */
|
||||
onRead: (spaceId: string, upToMessageId: string) => void;
|
||||
onReplyInThread: (threadId: string, body: string) => Promise<void>;
|
||||
send: SendState;
|
||||
composerRef: MutableRefObject<HTMLTextAreaElement | null>;
|
||||
@@ -112,6 +115,20 @@ export function ChatConversation({
|
||||
if (atBottomRef.current) jumpToLatest("auto");
|
||||
}, [lastMessageId]);
|
||||
|
||||
// Read means seen: the newest message is in view, the window is visible, and no search hides messages.
|
||||
const newestId = messagesInSpace.length > 0 ? messagesInSpace[messagesInSpace.length - 1].id : null;
|
||||
const spaceId = space?.id ?? null;
|
||||
const searching = trimmed.length > 0;
|
||||
useEffect(() => {
|
||||
const markIfSeen = () => {
|
||||
// The DOM is measured, not the last scroll state: clearing a search can grow the list under a reader who was at the bottom.
|
||||
if (spaceId && newestId && !searching && measureAtBottom() && document.visibilityState === "visible") onRead(spaceId, newestId);
|
||||
};
|
||||
markIfSeen();
|
||||
document.addEventListener("visibilitychange", markIfSeen);
|
||||
return () => document.removeEventListener("visibilitychange", markIfSeen);
|
||||
}, [spaceId, newestId, atBottom, searching, onRead]);
|
||||
|
||||
return (
|
||||
<>
|
||||
<section style={{ flex: 1, minWidth: 0, display: "flex", flexDirection: "column", background: "var(--panel)" }}>
|
||||
|
||||
@@ -115,6 +115,34 @@ export interface ConversationItem {
|
||||
unread: number;
|
||||
}
|
||||
|
||||
/** Unread messages per space for one reader, counted by position after the last read message (time then insertion order). */
|
||||
export function unreadBySpace(messages: ChatMessageRecord[], readStates: ChatReadStateRecord[], employeeId: string): Map<string, number> {
|
||||
const countsSince = new Map<string, number>();
|
||||
const readStateBySpace = new Map<string, ChatReadStateRecord>();
|
||||
for (const state of readStates) {
|
||||
if (state.employeeId === employeeId) readStateBySpace.set(state.spaceId, state);
|
||||
}
|
||||
const readPassed = new Set<string>();
|
||||
const loadedIds = new Set(messages.map((message) => message.id));
|
||||
for (const message of messages) {
|
||||
const state = readStateBySpace.get(message.spaceId);
|
||||
const readId = state?.lastReadMessageId ?? null;
|
||||
const loadedRead = readId !== null && loadedIds.has(readId);
|
||||
if (message.id === readId) {
|
||||
readPassed.add(message.spaceId);
|
||||
continue;
|
||||
}
|
||||
if (message.senderType === "employee" && message.senderId === employeeId) continue;
|
||||
// A read message older than the loaded window means every loaded message came after it.
|
||||
const unread = !state ? true
|
||||
: loadedRead ? readPassed.has(message.spaceId)
|
||||
: readId !== null ? true
|
||||
: message.createdAt > state.lastReadAt;
|
||||
if (unread) countsSince.set(message.spaceId, (countsSince.get(message.spaceId) ?? 0) + 1);
|
||||
}
|
||||
return countsSince;
|
||||
}
|
||||
|
||||
/**
|
||||
* Folds spaces, messages, memberships, and per-employee read state into the
|
||||
* conversation list: every channel and direct conversation as its own row,
|
||||
@@ -133,18 +161,8 @@ export function buildConversations(input: {
|
||||
ceoAgentId: string | null;
|
||||
}): ConversationItem[] {
|
||||
const lastBySpace = new Map<string, ChatMessageRecord>();
|
||||
const countsSince = new Map<string, number>();
|
||||
const lastReadBySpace = new Map<string, string>();
|
||||
for (const state of input.readStates) {
|
||||
if (state.employeeId === input.currentEmployeeId) lastReadBySpace.set(state.spaceId, state.lastReadAt);
|
||||
}
|
||||
for (const message of input.messages) {
|
||||
lastBySpace.set(message.spaceId, message);
|
||||
if (message.senderType === "employee" && message.senderId === input.currentEmployeeId) continue;
|
||||
const lastRead = lastReadBySpace.get(message.spaceId);
|
||||
if (lastRead && message.createdAt <= lastRead) continue;
|
||||
countsSince.set(message.spaceId, (countsSince.get(message.spaceId) ?? 0) + 1);
|
||||
}
|
||||
for (const message of input.messages) lastBySpace.set(message.spaceId, message);
|
||||
const countsSince = unreadBySpace(input.messages, input.readStates, input.currentEmployeeId);
|
||||
const membersBySpace = new Map<string, ChatMembershipRecord[]>();
|
||||
for (const membership of input.memberships) {
|
||||
const group = membersBySpace.get(membership.spaceId);
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { unreadBySpace } from "../pages/chat/chat-helpers";
|
||||
import { useEffect, useState, type FormEvent, type ReactNode } from "react";
|
||||
import type { AppSnapshot, EmployeeRecord } from "../../../shared/domain";
|
||||
import { Avatar, Field, Icon, IconButton } from "./ui";
|
||||
@@ -225,19 +226,10 @@ function computeBadges(snapshot: AppSnapshot, orgId: string): Partial<Record<Rou
|
||||
}
|
||||
|
||||
function countUnreadMessages(snapshot: AppSnapshot, orgId: string, employeeId: string): number {
|
||||
const readByspace = new Map<string, string>();
|
||||
for (const state of snapshot.chatReadStates) {
|
||||
if (state.orgId !== orgId || state.employeeId !== employeeId) continue;
|
||||
readByspace.set(state.spaceId, state.lastReadAt);
|
||||
}
|
||||
const messages = snapshot.chatMessages.filter((message) => message.orgId === orgId);
|
||||
const readStates = snapshot.chatReadStates.filter((state) => state.orgId === orgId);
|
||||
let unread = 0;
|
||||
for (const message of snapshot.chatMessages) {
|
||||
if (message.orgId !== orgId) continue;
|
||||
if (message.senderType === "employee" && message.senderId === employeeId) continue;
|
||||
const lastRead = readByspace.get(message.spaceId);
|
||||
if (lastRead && message.createdAt <= lastRead) continue;
|
||||
unread += 1;
|
||||
}
|
||||
for (const count of unreadBySpace(messages, readStates, employeeId).values()) unread += count;
|
||||
return unread;
|
||||
}
|
||||
|
||||
|
||||
@@ -38,4 +38,6 @@ export interface MarkChatSpaceReadInput {
|
||||
orgId: string;
|
||||
employeeId: string;
|
||||
spaceId: string;
|
||||
/** The newest message the reader actually saw; later arrivals stay unread. */
|
||||
upToMessageId: string;
|
||||
}
|
||||
@@ -133,6 +133,39 @@ test("the pinned CEO conversation is the Messages front door", async () => {
|
||||
await expect(root).toContainText("1 reply");
|
||||
await thread.getByRole("button", { name: "Close thread" }).click();
|
||||
await expect(thread).toHaveCount(0);
|
||||
|
||||
// A message that arrives while the reader is up in history stays unread until they scroll down to it.
|
||||
await page.evaluate(async () => {
|
||||
const api = (window as unknown as { agentCompany: {
|
||||
loadSnapshot: () => Promise<{ ok: true; data: { activeOrgId: string; employees: Array<{ id: string; orgId: string }>; chatSpaces: Array<{ id: string; orgId: string; spaceType: string }> } }>;
|
||||
sendMessage: (input: unknown) => Promise<unknown>;
|
||||
} }).agentCompany;
|
||||
const snapshot = (await api.loadSnapshot()).data;
|
||||
const employee = snapshot.employees.find((item) => item.orgId === snapshot.activeOrgId)!;
|
||||
const ceoChat = snapshot.chatSpaces.find((item) => item.orgId === snapshot.activeOrgId && item.spaceType === "direct")!;
|
||||
for (let index = 0; index < 40; index += 1) {
|
||||
await api.sendMessage({ orgId: snapshot.activeOrgId, spaceId: ceoChat.id, threadId: null, senderType: "employee", senderId: employee.id, bodyMd: `History line ${index}`, attachmentFileIds: [] });
|
||||
}
|
||||
});
|
||||
await expect(chatLog.getByText("History line 39")).toBeVisible();
|
||||
await chatLog.evaluate((element) => {
|
||||
element.scrollTop = 0;
|
||||
element.dispatchEvent(new Event("scroll"));
|
||||
});
|
||||
await page.evaluate(async () => {
|
||||
const api = (window as unknown as { agentCompany: {
|
||||
loadSnapshot: () => Promise<{ ok: true; data: { activeOrgId: string; agents: Array<{ id: string; orgId: string; roleLevel: string }>; chatSpaces: Array<{ id: string; orgId: string; spaceType: string }> } }>;
|
||||
sendMessage: (input: unknown) => Promise<unknown>;
|
||||
} }).agentCompany;
|
||||
const snapshot = (await api.loadSnapshot()).data;
|
||||
const ceo = snapshot.agents.find((agent) => agent.orgId === snapshot.activeOrgId && agent.roleLevel === "c_suite")!;
|
||||
const ceoChat = snapshot.chatSpaces.find((item) => item.orgId === snapshot.activeOrgId && item.spaceType === "direct")!;
|
||||
await api.sendMessage({ orgId: snapshot.activeOrgId, spaceId: ceoChat.id, threadId: null, senderType: "agent", senderId: ceo.id, bodyMd: "An update you have not seen yet.", attachmentFileIds: [] });
|
||||
});
|
||||
const ceoRow = page.locator(".convo-item").first();
|
||||
await expect(ceoRow.locator(".nav-badge")).toHaveText("1");
|
||||
await page.getByRole("button", { name: "Jump to latest" }).click();
|
||||
await expect(ceoRow.locator(".nav-badge")).toHaveCount(0);
|
||||
} finally {
|
||||
await app.close();
|
||||
}
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { describe, expect, test } from "vitest";
|
||||
import { DatabaseConnection } from "../../src/main/database/connection";
|
||||
import { OrganizationService } from "../../src/main/services/organization-service";
|
||||
import { ChatService } from "../../src/main/services/chat-service";
|
||||
|
||||
describe("read state", () => {
|
||||
test("marks read up to the message the reader saw, never past it and never backwards", () => {
|
||||
const connection = new DatabaseConnection(join(tmpdir(), `ac-read-state-${randomUUID()}.db`));
|
||||
const chat = new ChatService(connection.db);
|
||||
let orgId = "";
|
||||
connection.transaction(() => {
|
||||
orgId = new OrganizationService(connection.db).createOrganization({ organizationName: "Acme", employeeName: "Yang", mission: "m", goal: "g" });
|
||||
});
|
||||
const employeeId = (connection.db.prepare("select id from employees where org_id = ?").get(orgId) as { id: string }).id;
|
||||
const space = (name: string) => (connection.db.prepare("select id from chat_spaces where org_id = ? and name = ?").get(orgId, name) as { id: string }).id;
|
||||
const general = space("general");
|
||||
const send = (spaceId: string, bodyMd: string) => connection.transaction(() => chat.sendMessage({
|
||||
orgId, spaceId, threadId: null, senderType: "employee", senderId: employeeId, bodyMd, attachmentFileIds: [],
|
||||
}));
|
||||
const createdAt = (id: string) => (connection.db.prepare("select created_at from chat_messages where id = ?").get(id) as { created_at: string }).created_at;
|
||||
const readState = () => connection.db.prepare("select last_read_message_id as id, last_read_at as at from chat_read_state where space_id = ?").get(general);
|
||||
const seen = send(general, "Seen");
|
||||
const later = send(general, "Arrived while scrolled up");
|
||||
connection.db.prepare("update chat_messages set created_at = ? where id = ?").run("2099-01-01T00:00:00.000Z", later);
|
||||
|
||||
chat.markSpaceRead({ orgId, employeeId, spaceId: general, upToMessageId: seen });
|
||||
expect(readState()).toEqual({ id: seen, at: createdAt(seen) });
|
||||
chat.markSpaceRead({ orgId, employeeId, spaceId: general, upToMessageId: later });
|
||||
chat.markSpaceRead({ orgId, employeeId, spaceId: general, upToMessageId: seen });
|
||||
expect(readState()).toEqual({ id: later, at: "2099-01-01T00:00:00.000Z" });
|
||||
// Two messages in the same millisecond still order by insertion, so reading the second advances.
|
||||
const first = send(general, "Same instant one");
|
||||
const second = send(general, "Same instant two");
|
||||
connection.db.prepare("update chat_messages set created_at = '2099-06-01T00:00:00.000Z' where id in (?, ?)").run(first, second);
|
||||
chat.markSpaceRead({ orgId, employeeId, spaceId: general, upToMessageId: first });
|
||||
chat.markSpaceRead({ orgId, employeeId, spaceId: general, upToMessageId: second });
|
||||
expect(readState()).toEqual({ id: second, at: "2099-06-01T00:00:00.000Z" });
|
||||
chat.markSpaceRead({ orgId, employeeId, spaceId: general, upToMessageId: first });
|
||||
expect(readState()).toEqual({ id: second, at: "2099-06-01T00:00:00.000Z" });
|
||||
expect(() => chat.markSpaceRead({ orgId, employeeId, spaceId: general, upToMessageId: send(space("engineering"), "Elsewhere") }))
|
||||
.toThrow("The message read is not in this conversation.");
|
||||
connection.close();
|
||||
});
|
||||
});
|
||||
@@ -4,6 +4,7 @@ import type {
|
||||
AgentRecord,
|
||||
ChatMembershipRecord,
|
||||
ChatMessageRecord,
|
||||
ChatReadStateRecord,
|
||||
ChatSpaceRecord,
|
||||
EmployeeRecord,
|
||||
} from "../../src/shared/domain";
|
||||
@@ -103,3 +104,18 @@ describe("conversation pinning", () => {
|
||||
expect(conversations[0].preview).toContain("CEO:");
|
||||
});
|
||||
});
|
||||
|
||||
describe("unread counts", () => {
|
||||
test("count by position after the last read message, so a same-millisecond arrival is still unread", () => {
|
||||
const { spaces, memberships } = fixtures();
|
||||
const at = "2026-02-01T00:00:00.000Z";
|
||||
const messages = [message("m_read", "spc_general", CEO, at), message("m_same_ms", "spc_general", ENGINEER, at)];
|
||||
const unread = (readStates: ChatReadStateRecord[]) => buildConversations({ spaces, memberships, messages, readStates, employees, agents, currentEmployeeId: FOUNDER, ceoAgentId: CEO })
|
||||
.find((item) => item.space.id === "spc_general")!.unread;
|
||||
const state = (lastReadMessageId: string | null): ChatReadStateRecord => ({ orgId: ORG, employeeId: FOUNDER, spaceId: "spc_general", lastReadMessageId, lastReadAt: at });
|
||||
expect(unread([state("m_read")])).toBe(1);
|
||||
expect(unread([state("m_same_ms")])).toBe(0);
|
||||
// A read message older than the loaded window means every loaded message came after it.
|
||||
expect(unread([state("m_older_than_window")])).toBe(2);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user