diff --git a/apps/mobile/.maestro/README.md b/apps/mobile/.maestro/README.md new file mode 100644 index 0000000..84cde78 --- /dev/null +++ b/apps/mobile/.maestro/README.md @@ -0,0 +1,54 @@ +# Maestro E2E — chat "alive" features (M8) + +Flows for the typing-indicator / read-receipt / reaction work. + +## What runs today + +`chat-smoke.yaml` — a **single-client** smoke. One device can only observe the +**optimistic / local** surface of these features: + +| Asserted (single client) | Not assertable (needs a 2nd actor) | +| ---------------------------------------------------- | ------------------------------------------------ | +| Sending a message renders my bubble | A peer's **typing** indicator (self is excluded) | +| Long-press → quick-react tray → my **reaction pill** | **Read** double-tick (needs a peer to read) | +| Composer input drives the typing emitter | Reaction **fan-in** from another user | + +The cross-user half is deliberately out of scope here — it needs the 2-client +rig below. The deterministic core of all three features is already covered by +the vitest units in `packages/chat/src/*.test.ts`. + +## Running + +```sh +# install: https://maestro.mobile.dev +export MAESTRO_APP_ID=c # match app.json → ios/android identifier +maestro test apps/mobile/.maestro/chat-smoke.yaml +``` + +Preconditions: a **built app** on a booted simulator/emulator, a **signed-in** +session, and **≥1 chat** on the conversations list (seed via +`packages/database/prisma/seed.ts`). The flow does not sign in — add sign-in +steps (or seed an authenticated session) for a cold-start CI run. + +## E2E anchors added for these flows + +- `chat-row` — Glacier `ListRow` (conversations list) +- `composer-input`, `composer-send` — Glacier `Composer` + +Message text and reaction emoji are asserted by their visible text. + +## Follow-ups (tracked, not built) + +1. **2-client journey (rig).** Drive user A in Maestro while a second actor + (user B) produces real traffic, then assert A's UI reflects it: + B sends → A shows the message + typing pulse; A reacts → pill; B reads → + A's tick flips to the primary double-check. Requires B to be a real MLS + group member (so its frames decrypt on A). See the cost note in the PR/plan. + +2. **MLS out-of-order regression** (binary-gated). Lives in `chat-mls-core` + (needs `node/index.node` via `pnpm run build:node`; the suite auto-skips + when the binary is absent). Fire an interleaved same-sender + message+reaction and assert both decrypt — guards the OpenMLS default + `SenderRatchetConfiguration` out-of-order tolerance (=5) that + `reactions-ride-send` relies on (`chat-mls-core/src/engine.rs:165` & `:267` + leave it unset). A future config change lowering it must fail this test. diff --git a/apps/mobile/.maestro/chat-smoke.yaml b/apps/mobile/.maestro/chat-smoke.yaml new file mode 100644 index 0000000..7fd220e --- /dev/null +++ b/apps/mobile/.maestro/chat-smoke.yaml @@ -0,0 +1,84 @@ +# Single-client smoke — M8 "alive chat" (typing / read receipts / reactions). +# +# A single device can only assert the OPTIMISTIC / local surface: +# • sending a message renders my bubble +# • reacting via the long-press tray renders my reaction pill +# Typing indicators and read-receipt double-ticks are CROSS-USER and cannot be +# observed with one client — see ./README.md (2-client rig is a follow-up). +# +# Preconditions: app installed on a booted sim, DB seeded (alice + password123), +# and at least one chat on alice's conversations list. The flow signs in itself +# (see the runFlow block below). Set MAESTRO_APP_ID (e.g. io.sessions.app). Run: +# maestro --device test apps/mobile/.maestro/chat-smoke.yaml +appId: ${MAESTRO_APP_ID} +--- +- launchApp + +# Cold-start sign-in. launchApp relaunches the process and the Better Auth +# session does not rehydrate, so we land on the auth gate. Gated on the login +# screen being visible — a no-op if a session ever does persist. +# Creds come from the DB seed (packages/database/prisma/seed.ts): SEED_PASSWORD. +- runFlow: + when: + visible: "Login" + commands: + - tapOn: "Email Address" + - inputText: "alice@example.com" + - tapOn: "Password" + - inputText: "password123" + # Password field has enterKeyHint="go" + onSubmitEditing → submit via the + # keyboard return key. Avoids hideKeyboard (unsupported on the RN input) + # and a Login-button tap that the keyboard can overlap. + - pressKey: Enter + +- extendedWaitUntil: + visible: + id: "chat-row" + timeout: 20000 +# Open the first conversation (testID on the Glacier ListRow). +- tapOn: + id: "chat-row" + index: 0 + +# --- Send a message --- +- tapOn: + id: "composer-input" +- inputText: "Maestro smoke" +- tapOn: + id: "composer-send" +# Optimistic (status "sending") bubble renders NO footer (chat-bubble.tsx), so +# its accessibility label is exactly "Maestro smoke" and this text match works. +# This asserts the text we typed actually rendered. +- assertVisible: "Maestro smoke" + +# On ACK the store swaps the message .id (clientMsgId → serverSerial, +# store.ts confirmOptimisticMessage), which flips the FlashList row key and +# REMOUNTS the bubble. The confirmed bubble now shows a footer (timestamp + +# receipt), and because the row Pressable is one iOS a11y container its label +# folds into "Maestro smoke, 2:11 PM, " — so a bare text match no longer hits. +# Anchor on the stable testID instead. The app tags the newest message I sent +# ONLY once it's CONFIRMED (serverSerial present), so this wait blocks through +# the whole ack window — not just until the optimistic bubble paints. That +# matters because the quick-react tray is gated on serverSerial and the actions +# sheet snapshots the message at long-press time: long-pressing a still-optimistic +# bubble opens a tray-less sheet that never recovers. Waiting for the confirmed +# anchor guarantees the long-press lands on a reactable, settled row. +- waitForAnimationToEnd: + timeout: 4000 +- extendedWaitUntil: + visible: + id: "latest-sent-message" + timeout: 10000 + +# --- React to the message (long-press → quick-react tray → pill) --- +- longPressOn: + id: "latest-sent-message" +# Quick-react tray sits in the actions sheet; each emoji is its own a11y node, +# so an exact emoji match works. NOTE: "❤️" here must be U+2764 U+FE0F (heart + +# variation selector) to match the rendered node — a bare U+2764 won't. +- assertVisible: "❤️" +- tapOn: "❤️" +# The reaction pill folds UNDER the bubble, inside the same row Pressable, so it +# merges into the row's a11y label ("Maestro smoke … ❤️ …"). Match a substring +# regex rather than the bare emoji, which would anchor-fail against the label. +- assertVisible: ".*❤️.*" diff --git a/apps/mobile/app/(tabs)/index.tsx b/apps/mobile/app/(tabs)/index.tsx index d7ff4e7..612fae7 100644 --- a/apps/mobile/app/(tabs)/index.tsx +++ b/apps/mobile/app/(tabs)/index.tsx @@ -81,6 +81,7 @@ function ChatRow({ chat, myId, onPress }: ChatRowProps) { return ( Mortstack - ); } + +function SettingsToggleRow({ + label, + value, + onValueChange, +}: { + label: string; + value: boolean; + onValueChange: (next: boolean) => void; +}) { + return ( + + + {label} + + + + + + ); +} diff --git a/apps/mobile/lib/chat/components/MessageActionsSheet.tsx b/apps/mobile/lib/chat/components/MessageActionsSheet.tsx index 2bae9ee..5e346c2 100644 --- a/apps/mobile/lib/chat/components/MessageActionsSheet.tsx +++ b/apps/mobile/lib/chat/components/MessageActionsSheet.tsx @@ -6,7 +6,7 @@ import { Alert } from "react-native"; import * as Clipboard from "expo-clipboard"; import * as Haptics from "expo-haptics"; import { Feather, MaterialCommunityIcons } from "@expo/vector-icons"; -import { Separator, Sheet, XStack, useTheme } from "tamagui"; +import { Separator, Sheet, Text, XStack, useTheme } from "tamagui"; import type { Message } from "@repo/chat"; import { Label } from "@repo/ui/glacier/typography"; @@ -14,10 +14,15 @@ import { Label } from "@repo/ui/glacier/typography"; const INSPECTOR_ENABLED = __DEV__ || process.env.EXPO_PUBLIC_DEMO_CRYPTO_INSPECTOR === "1"; +// Quick-react row. Reactions target a persisted message (serverSerial), so the +// row is hidden for a still-sending bubble. +const QUICK_EMOJIS = ["👍", "❤️", "😂", "😮", "😢", "🙏"]; + export interface MessageActionsSheetProps { message: Message | null; isMine: boolean; onClose: () => void; + onReact: (message: Message, emoji: string) => void; onDelete: (message: Message) => void; onRetry: (message: Message) => void; onReport: (message: Message, reason: "SPAM" | "HARASSMENT" | "OTHER") => void; @@ -57,6 +62,7 @@ export function MessageActionsSheet({ message, isMine, onClose, + onReact, onDelete, onRetry, onReport, @@ -113,6 +119,34 @@ export function MessageActionsSheet({ > + {/* Quick-react row — reactions ride the E2EE send path (encrypted like + a message). Hidden until the target has a serverSerial. */} + {message?.serverSerial ? ( + <> + + {QUICK_EMOJIS.map((emoji) => ( + { + onReact(message, emoji); + void Haptics.impactAsync(Haptics.ImpactFeedbackStyle.Light); + onClose(); + }} + > + {emoji} + + ))} + + + + ) : null} + } label="Copy" diff --git a/apps/mobile/lib/chat/components/ReactionPills.tsx b/apps/mobile/lib/chat/components/ReactionPills.tsx new file mode 100644 index 0000000..80c51f7 --- /dev/null +++ b/apps/mobile/lib/chat/components/ReactionPills.tsx @@ -0,0 +1,121 @@ +// Reaction pills folded under a message bubble. Groups a target's reactions by +// emoji with a count, tints the pill when it includes mine, and pops in with a +// spring on first mount (reduced-motion → no pop). Tapping a pill toggles my +// reaction with that emoji (add ⇆ del) via the screen's onToggle. +import { useEffect } from "react"; +import { Pressable } from "react-native"; +import Animated, { + useAnimatedStyle, + useReducedMotion, + useSharedValue, + withSpring, +} from "react-native-reanimated"; +import { Text, XStack } from "tamagui"; + +import type { Reaction } from "@repo/chat"; + +interface Grouped { + emoji: string; + count: number; + mine: boolean; + sending: boolean; +} + +function groupByEmoji( + reactions: Reaction[], + myAuthUserId: string | null, +): Grouped[] { + const map = new Map(); + for (const r of reactions) { + const g = map.get(r.emoji) ?? { + emoji: r.emoji, + count: 0, + mine: false, + sending: false, + }; + g.count += 1; + if (r.senderAuthUserId === myAuthUserId) g.mine = true; + if (r.status === "sending") g.sending = true; + map.set(r.emoji, g); + } + return [...map.values()]; +} + +function Pill({ + group, + reduced, + onPress, +}: { + group: Grouped; + reduced: boolean; + onPress: () => void; +}) { + const scale = useSharedValue(reduced ? 1 : 0); + + useEffect(() => { + if (reduced) { + scale.value = 1; + return; + } + scale.value = withSpring(1, { damping: 12, stiffness: 220 }); + }, [reduced, scale]); + + const style = useAnimatedStyle(() => ({ + transform: [{ scale: scale.value }], + })); + + return ( + + + + {group.emoji} + {group.count > 1 ? ( + + {group.count} + + ) : null} + + + + ); +} + +export function ReactionPills({ + reactions, + myAuthUserId, + onToggle, +}: { + reactions: Reaction[]; + myAuthUserId: string | null; + onToggle: (emoji: string) => void; +}) { + const reduced = useReducedMotion(); + const groups = groupByEmoji(reactions, myAuthUserId); + if (groups.length === 0) return null; + return ( + <> + {groups.map((g) => ( + onToggle(g.emoji)} + /> + ))} + + ); +} diff --git a/apps/mobile/lib/chat/components/TypingIndicator.tsx b/apps/mobile/lib/chat/components/TypingIndicator.tsx new file mode 100644 index 0000000..9cc91d1 --- /dev/null +++ b/apps/mobile/lib/chat/components/TypingIndicator.tsx @@ -0,0 +1,88 @@ +// Three-dot typing pulse in an incoming-style bubble frame (chat/DESIGN.md +// §Typing): 1.2s staggered loop, dots in on-surface-variant. Respects +// prefers-reduced-motion — dots fade in place instead of bouncing. +import { useEffect } from "react"; +import { StyleSheet, View } from "react-native"; +import Animated, { + cancelAnimation, + useAnimatedStyle, + useReducedMotion, + useSharedValue, + withDelay, + withRepeat, + withSequence, + withTiming, +} from "react-native-reanimated"; +import { YStack, useTheme } from "tamagui"; + +const DOT_COUNT = 3; +const PHASE_MS = 400; +const STAGGER_MS = 150; + +function Dot({ + index, + color, + reduced, +}: { + index: number; + color: string; + reduced: boolean; +}) { + const progress = useSharedValue(0); + + useEffect(() => { + progress.value = withDelay( + index * STAGGER_MS, + withRepeat( + withSequence( + withTiming(1, { duration: PHASE_MS }), + withTiming(0, { duration: PHASE_MS }), + ), + -1, + ), + ); + return () => cancelAnimation(progress); + }, [index, progress]); + + const style = useAnimatedStyle(() => { + const opacity = 0.35 + progress.value * 0.65; + if (reduced) return { opacity }; + return { opacity, transform: [{ translateY: -3 * progress.value }] }; + }); + + return ( + + ); +} + +export function TypingIndicator() { + const theme = useTheme(); + const reduced = useReducedMotion(); + const dotColor = theme.onSurfaceVariant?.val ?? "#888"; + + return ( + + + {Array.from({ length: DOT_COUNT }, (_, i) => ( + + ))} + + + ); +} + +const styles = StyleSheet.create({ + row: { flexDirection: "row", gap: 4, alignItems: "center" }, + dot: { width: 6, height: 6, borderRadius: 3 }, +}); diff --git a/apps/mobile/store/settings.ts b/apps/mobile/store/settings.ts new file mode 100644 index 0000000..1644918 --- /dev/null +++ b/apps/mobile/store/settings.ts @@ -0,0 +1,37 @@ +import { create } from "zustand"; +import { + createJSONStorage, + persist, + type StateStorage, +} from "zustand/middleware"; +import * as SecureStore from "expo-secure-store"; + +// SecureStore-backed adapter for zustand persist. SecureStore is async (persist +// rehydrates asynchronously — the default below applies until it resolves) and +// keys must be [A-Za-z0-9._-], which the `name` satisfies. +const secureStorage: StateStorage = { + getItem: (name) => SecureStore.getItemAsync(name), + setItem: (name, value) => SecureStore.setItemAsync(name, value), + removeItem: (name) => SecureStore.deleteItemAsync(name), +}; + +interface SettingsState { + /** Symmetric read-receipts toggle (M8). When OFF: the client emits no read + * receipts AND suppresses rendering peers' receipts (Signal/WhatsApp model). + * Defaults ON. */ + readReceipts: boolean; + setReadReceipts: (on: boolean) => void; +} + +export const useSettingsStore = create()( + persist( + (set) => ({ + readReceipts: true, + setReadReceipts: (on) => set({ readReceipts: on }), + }), + { + name: "app_settings_v1", + storage: createJSONStorage(() => secureStorage), + }, + ), +); diff --git a/packages/chat-transport/src/client.ts b/packages/chat-transport/src/client.ts index be68408..ce1c9c5 100644 --- a/packages/chat-transport/src/client.ts +++ b/packages/chat-transport/src/client.ts @@ -14,6 +14,8 @@ export type ConnectionState = export type IncomingMessage = Extract; export type IncomingError = Extract; export type IncomingMlsWelcome = Extract; +export type IncomingTyping = Extract; +export type IncomingRead = Extract; export interface ChatTransportOptions { // Cloudflare Worker URL. Use ws:// for local wrangler, wss:// in prod. @@ -61,10 +63,19 @@ export interface ChatTransport { close(): void; subscribe(chatIds: string[]): void; send(input: SendInput): Promise; + /** Fire-and-forget typing signal — ephemeral, dropped when the socket isn't + * open (no queue). */ + sendTyping(input: { chatId: string; on: boolean }): void; + /** Fire-and-forget read watermark — dropped when offline; the server + * reconciles ChatMember.lastReadSerial on next load. `upto` is a + * serverMsgId string. */ + sendRead(input: { chatId: string; upto: string }): void; onMessage(handler: (msg: IncomingMessage) => void): () => void; onState(handler: (state: ConnectionState) => void): () => void; onError(handler: (err: IncomingError) => void): () => void; onMlsWelcome(handler: (m: IncomingMlsWelcome) => void): () => void; + onTyping(handler: (m: IncomingTyping) => void): () => void; + onRead(handler: (m: IncomingRead) => void): () => void; } interface PendingSend { @@ -100,6 +111,8 @@ export function createChatTransport(opts: ChatTransportOptions): ChatTransport { const stateHandlers = new Set<(s: ConnectionState) => void>(); const errorHandlers = new Set<(e: IncomingError) => void>(); const mlsWelcomeHandlers = new Set<(m: IncomingMlsWelcome) => void>(); + const typingHandlers = new Set<(m: IncomingTyping) => void>(); + const readHandlers = new Set<(m: IncomingRead) => void>(); function setState(next: ConnectionState) { if (state === next) return; @@ -172,6 +185,12 @@ export function createChatTransport(opts: ChatTransportOptions): ChatTransport { case "msg": for (const h of messageHandlers) h(env); return; + case "typ": + for (const h of typingHandlers) h(env); + return; + case "read": + for (const h of readHandlers) h(env); + return; case "mls-welcome": for (const h of mlsWelcomeHandlers) h(env); return; @@ -333,6 +352,27 @@ export function createChatTransport(opts: ChatTransportOptions): ChatTransport { return () => mlsWelcomeHandlers.delete(handler); } + // Ephemeral signals — best-effort, no pending queue. sendFrame() no-ops when + // the socket isn't open, which is the desired behaviour: a typing signal or + // read receipt lost to a disconnect is not worth replaying. + function sendTyping(input: { chatId: string; on: boolean }) { + sendFrame({ t: "typ", chatId: input.chatId, on: input.on }); + } + + function sendRead(input: { chatId: string; upto: string }) { + sendFrame({ t: "read", chatId: input.chatId, upto: input.upto }); + } + + function onTyping(handler: (m: IncomingTyping) => void) { + typingHandlers.add(handler); + return () => typingHandlers.delete(handler); + } + + function onRead(handler: (m: IncomingRead) => void) { + readHandlers.add(handler); + return () => readHandlers.delete(handler); + } + return { get state() { return state; @@ -341,10 +381,14 @@ export function createChatTransport(opts: ChatTransportOptions): ChatTransport { close, subscribe, send, + sendTyping, + sendRead, onMessage, onState, onError, onMlsWelcome, + onTyping, + onRead, }; } diff --git a/packages/chat-transport/src/envelope.ts b/packages/chat-transport/src/envelope.ts index dae842b..5713e7d 100644 --- a/packages/chat-transport/src/envelope.ts +++ b/packages/chat-transport/src/envelope.ts @@ -26,6 +26,17 @@ export type ClientToServer = nonce: Uint8Array; unencrypted?: boolean; } + // Typing signal — ephemeral, never persisted. `on:true` on first keystroke + // and re-sent as a heartbeat (~3s) while composing; `on:false` on send / blur + // / idle. The Chat DO is a stateless relay — the receiver holds a short expiry + // timer, so a dropped `on:false` (sender crash) self-clears with no server + // TTL. Content-blind: whether someone is typing is metadata, no plaintext. + | { t: "typ"; chatId: string; on: boolean } + // Read receipt — high-water-mark, not per-message. `upto` is the greatest + // serverMsgId (= serverSerial as string) the user has read. Server persists it + // to ChatMember.lastReadSerial and fans out. Gated client-side by the + // symmetric read-receipts privacy toggle (never emitted when off). + | { t: "read"; chatId: string; upto: string } // Heartbeat — answered with `pong`. Auto-handled when supported by the // platform; manual fallback for clients without auto-ping. | { t: "ping" }; @@ -57,6 +68,14 @@ export type ServerToClient = // background poll. Best-effort: the 30s poll remains the correctness // guarantee in case the push fails or the client is offline at send time. | { t: "mls-welcome"; ts: number } + // Typing fanout — a peer in `chatId` started/stopped composing. Server stamps + // `userId`; receiver renders the three-dot pulse (chat/DESIGN.md §Typing). + // Never sent for the connection's own userId. + | { t: "typ"; chatId: string; userId: string; on: boolean } + // Read fanout — `userId` has read up to `upto` (serverMsgId string) in + // `chatId`. Receiver advances that member's watermark; the sender's own + // outgoing bubbles flip sent → read once a peer's `upto` covers their serial. + | { t: "read"; chatId: string; userId: string; upto: string } // Soft error — connection stays open. Use error.code for routing. | { t: "err"; code: ChatErrorCode; msg?: string } | { t: "pong" }; diff --git a/packages/chat/package.json b/packages/chat/package.json index 1d7d7c8..e568cdc 100644 --- a/packages/chat/package.json +++ b/packages/chat/package.json @@ -8,7 +8,8 @@ "scripts": { "lint": "eslint .", "check-types": "tsc --noEmit", - "test": "echo \"no tests yet\" && exit 0" + "test": "vitest run", + "test:watch": "vitest" }, "dependencies": { "@msgpack/msgpack": "^3.1.3", @@ -22,7 +23,8 @@ "@repo/typescript-config": "workspace:*", "@types/react": "~19.1.10", "eslint": "^9.39.1", - "typescript": "5.9.2" + "typescript": "5.9.2", + "vitest": "^2.1.9" }, "peerDependencies": { "react": "*", diff --git a/packages/chat/src/crypto-pipe.test.ts b/packages/chat/src/crypto-pipe.test.ts new file mode 100644 index 0000000..6a4ea33 --- /dev/null +++ b/packages/chat/src/crypto-pipe.test.ts @@ -0,0 +1,112 @@ +import { describe, expect, it, vi } from "vitest"; +import { encode } from "@msgpack/msgpack"; + +// chat-crypto calls requireNativeModule() at import, which throws in Node. Stub +// it so importing crypto-pipe works — these tests use the v=2 (MLS) path with a +// passthrough engine and never touch ChatCrypto.box. vi.mock is hoisted above +// the import below by vitest, so the stub is in place before crypto-pipe loads. +vi.mock("@repo/chat-crypto", () => ({ + ChatCrypto: { box: vi.fn(), boxOpen: vi.fn() }, +})); + +import { + DecryptError, + FRAME_VERSION_V2, + decryptInbound, + encryptOutboundMls, + isReactionFrame, + type MlsApi, +} from "./crypto-pipe"; + +// Passthrough MLS engine: encrypt returns plaintext unchanged, processMessage +// returns it as an application message. Round-trips buildFrame→parseFrame +// through the real v=2 code path with zero native crypto. +const fakeMls: MlsApi = { + encryptApp: (_g, plaintext) => plaintext, + processMessage: (_g, bytes) => ({ kind: "application", plaintext: bytes }), +}; +const GID = new Uint8Array([1, 2, 3, 4]); + +function decrypt(ciphertext: Uint8Array) { + return decryptInbound({ + ciphertext, + nonce: new Uint8Array(0), + senderAccountId: "acct", + seed: new Uint8Array(0), + candidateSenderX25519Pubs: [], + mls: fakeMls, + mlsGroupId: GID, + }); +} + +// A raw v=2 wire frame from an arbitrary plaintext object (bypasses the encoder +// so we can craft legacy / malformed shapes the encoder wouldn't produce). +function wire(obj: unknown): Uint8Array { + const body = encode(obj); + const out = new Uint8Array(1 + body.length); + out[0] = FRAME_VERSION_V2; + out.set(body, 1); + return out; +} + +describe("ContentFrame codec (v=2 passthrough)", () => { + it("round-trips a message frame (kind omitted on the wire)", () => { + const env = encryptOutboundMls({ + body: { text: "hello" }, + groupId: GID, + mls: fakeMls, + }); + const { frame } = decrypt(env.ciphertext); + expect(isReactionFrame(frame)).toBe(false); + if (isReactionFrame(frame)) throw new Error("unreachable"); + expect(frame.text).toBe("hello"); + expect(frame.kind).toBeUndefined(); + }); + + it("round-trips a reaction frame", () => { + const env = encryptOutboundMls({ + body: { kind: "rx", target: "42", emoji: "👍", op: "add" }, + groupId: GID, + mls: fakeMls, + }); + const { frame } = decrypt(env.ciphertext); + expect(isReactionFrame(frame)).toBe(true); + if (!isReactionFrame(frame)) throw new Error("unreachable"); + expect(frame.target).toBe("42"); + expect(frame.emoji).toBe("👍"); + expect(frame.op).toBe("add"); + }); + + it("treats a legacy frame with no `kind` as a message", () => { + const { frame } = decrypt( + wire({ v: FRAME_VERSION_V2, text: "legacy", ts: 123 }), + ); + expect(isReactionFrame(frame)).toBe(false); + if (isReactionFrame(frame)) throw new Error("unreachable"); + expect(frame.text).toBe("legacy"); + }); + + it("rejects a reaction frame with a malformed op", () => { + const bad = wire({ + v: FRAME_VERSION_V2, + ts: 1, + kind: "rx", + target: "1", + emoji: "x", + op: "nope", + }); + expect(() => decrypt(bad)).toThrow(DecryptError); + }); + + it("rejects a reaction frame with a non-string target", () => { + const bad = wire({ + v: FRAME_VERSION_V2, + ts: 1, + kind: "rx", + target: 5, + emoji: "x", + op: "add", + }); + expect(() => decrypt(bad)).toThrow(DecryptError); + }); +}); diff --git a/packages/chat/src/crypto-pipe.ts b/packages/chat/src/crypto-pipe.ts index 9d70112..394b4c3 100644 --- a/packages/chat/src/crypto-pipe.ts +++ b/packages/chat/src/crypto-pipe.ts @@ -36,18 +36,74 @@ export interface MlsApi { // informational, never trusted for ordering (server assigns serverMsgId // which carries authoritative ts). // -// `sender` is the sender's display-name snapshot at send time. Optional -// because (a) v=1 doesn't populate it, (b) the user may not have a Profile -// yet, and (c) the receiver-side NSE/FCM treats absence as "show generic -// title". Carrying it inside the ciphertext is necessary because the NSE -// has no network access at notification-decrypt time and can't look it up. -export interface ChatFrame { +// M8: the frame is now a discriminated union on `kind` so one encrypted +// channel carries both messages and reactions (reactions are E2EE for the +// same reason messages are — a plaintext emoji on the wire would contradict +// "server never sees plaintext", ADR/CONTEXT hero). The server + transport +// stay content-blind; only the receiving client decodes `kind`. +// +// EXPAND/CONTRACT: legacy (pre-M8) frames are `{ v, text, ts, sender? }` with +// NO `kind`. parseFrame treats a missing/absent `kind` as "msg", and the +// message encoder OMITS `kind` on the wire so message frames stay byte-for-byte +// identical to legacy — the native NSE msgpack readers (which skip unknown +// keys but require `text`) keep working unchanged for messages. +export interface ChatMsgFrame { v: number; - text: string; ts: number; + /** Absent on the wire for messages (legacy-identical); narrows the union. */ + kind?: "msg"; + text: string; + // `sender` is the sender's display-name snapshot at send time. Optional + // because (a) v=1 doesn't populate it, (b) the user may not have a Profile + // yet, and (c) the receiver-side NSE/FCM treats absence as "show generic + // title". Carried inside the ciphertext because the NSE has no network at + // notification-decrypt time and can't look it up. sender?: string; } +// Reaction frame (M8). Rides the same ciphertext as a message (Option A — no +// new transport frame), so the server can't distinguish it and stays blind. +// No `text`: the receiver folds it onto the target bubble instead of rendering +// a row. `target` is the reacted-to message's serverSerial (string). +export interface ChatReactionFrame { + v: number; + ts: number; + kind: "rx"; + target: string; + emoji: string; + op: "add" | "del"; +} + +export type ChatFrame = ChatMsgFrame | ChatReactionFrame; + +// Caller-supplied frame content (the part that isn't the version/timestamp, +// which the encoders stamp). Discriminated the same way as ChatFrame. +export type ChatFrameBody = + | { kind?: "msg"; text: string; sender?: string } + | { kind: "rx"; target: string; emoji: string; op: "add" | "del" }; + +export function isReactionFrame(f: ChatFrame): f is ChatReactionFrame { + return f.kind === "rx"; +} + +// Stamp version + timestamp onto a caller body → a full ChatFrame. Message +// frames deliberately omit `kind` to stay legacy-identical on the wire. +function buildFrame(v: number, ts: number, body: ChatFrameBody): ChatFrame { + if (body.kind === "rx") { + return { + v, + ts, + kind: "rx", + target: body.target, + emoji: body.emoji, + op: body.op, + }; + } + const f: ChatMsgFrame = { v, ts, text: body.text }; + if (body.sender !== undefined) f.sender = body.sender; + return f; +} + export interface RecipientDevice { deviceId: string; // Always present (every M3+ device publishes one). Used by the v=1 path. @@ -97,7 +153,7 @@ export class DecryptError extends Error { // ── encrypt ──────────────────────────────────────────────────────────────── export interface EncryptOutboundOpts { - text: string; + body: ChatFrameBody; targets: FanoutTarget[]; // Required for v=1 (libsodium box uses our X25519 secret). seed: Uint8Array; @@ -109,7 +165,7 @@ export interface EncryptOutboundOpts { // caller routes through encryptOutboundMls instead. export function encryptOutbound(opts: EncryptOutboundOpts): OutboundEnvelope[] { const ts = opts.now ?? Date.now(); - const frame: ChatFrame = { v: FRAME_VERSION_V1, text: opts.text, ts }; + const frame = buildFrame(FRAME_VERSION_V1, ts, opts.body); const plaintext = encode(frame); const out: OutboundEnvelope[] = []; @@ -134,13 +190,10 @@ export function encryptOutbound(opts: EncryptOutboundOpts): OutboundEnvelope[] { } export interface EncryptOutboundMlsOpts { - text: string; + body: ChatFrameBody; groupId: Uint8Array; mls: MlsApi; now?: number; - // Sender display name attached to the plaintext frame. Read by the - // recipient's NSE/FCM decryptor for the push-notification title. - sender?: string; } // v=2 MLS group-native send. One plaintext → one application message → @@ -156,11 +209,10 @@ export function encryptOutboundMls( opts: EncryptOutboundMlsOpts, ): OutboundMlsEnvelope { const ts = opts.now ?? Date.now(); - const frame: ChatFrame = { v: FRAME_VERSION_V2, text: opts.text, ts }; - // Only attach `sender` when provided — keeps the key out of the msgpack map - // entirely (vs encoding `null`), so receivers can rely on simple "key - // present" semantics and the wire stays minimal. - if (opts.sender !== undefined) frame.sender = opts.sender; + // buildFrame omits `kind` for messages (legacy-identical) and only attaches + // `sender` when the body carries it — keeping the key out of the msgpack map + // entirely (vs encoding `null`) so the wire stays minimal. + const frame = buildFrame(FRAME_VERSION_V2, ts, opts.body); const plaintext = encode(frame); const mlsBytes = opts.mls.encryptApp(opts.groupId, plaintext); // Prepend version byte so decryptInbound dispatches without needing the @@ -281,7 +333,6 @@ function parseFrame(plaintext: Uint8Array, expectedV: number): ChatFrame { typeof decoded !== "object" || decoded === null || !("v" in decoded) || - !("text" in decoded) || !("ts" in decoded) ) { throw new DecryptError("decrypted payload missing required frame fields"); @@ -292,14 +343,38 @@ function parseFrame(plaintext: Uint8Array, expectedV: number): ChatFrame { `plaintext frame.v=${String(f.v)} disagrees with wire version=${expectedV}`, ); } - if (typeof f.text !== "string" || typeof f.ts !== "number") { + if (typeof f.ts !== "number") { + throw new DecryptError("frame field types mismatch"); + } + + // Reaction frame (M8). A missing/absent `kind` is a legacy/message frame — + // fall through to the text branch, preserving pre-M8 compatibility. + if (f.kind === "rx") { + if ( + typeof f.target !== "string" || + typeof f.emoji !== "string" || + (f.op !== "add" && f.op !== "del") + ) { + throw new DecryptError("reaction frame missing/invalid target|emoji|op"); + } + return { + v: expectedV, + ts: f.ts, + kind: "rx", + target: f.target, + emoji: f.emoji, + op: f.op, + }; + } + + if (typeof f.text !== "string") { throw new DecryptError("frame field types mismatch"); } const senderRaw = "sender" in f ? f.sender : undefined; if (senderRaw !== undefined && typeof senderRaw !== "string") { throw new DecryptError("frame.sender, when present, must be a string"); } - const out: ChatFrame = { v: expectedV, text: f.text, ts: f.ts }; + const out: ChatMsgFrame = { v: expectedV, ts: f.ts, text: f.text }; if (senderRaw !== undefined) out.sender = senderRaw; return out; } diff --git a/packages/chat/src/encrypted-transport.ts b/packages/chat/src/encrypted-transport.ts index de59bfa..ae1b78b 100644 --- a/packages/chat/src/encrypted-transport.ts +++ b/packages/chat/src/encrypted-transport.ts @@ -3,6 +3,8 @@ import type { ConnectionState, IncomingError, IncomingMessage, + IncomingRead, + IncomingTyping, SendResult, } from "@repo/chat-transport/client"; @@ -12,13 +14,23 @@ import { encryptOutboundMls, FRAME_VERSION_V2, type ChatFrame, + type ChatFrameBody, type FanoutTarget, type MlsApi, } from "./crypto-pipe"; -export interface EncryptedSendInput { +export interface ReactionInput { + /** serverSerial (string) of the message being reacted to. */ + target: string; + emoji: string; + op: "add" | "del"; +} + +// Discriminated on payload: exactly one of `text` (a message) or `reaction` +// (a reaction that rides the same encrypted frame — Option A). Both share the +// routing fields below. +export type EncryptedSendInput = { chatId: string; - text: string; /** Required for v=1 chats; ignored for v=2 (MLS routes by groupId). */ targets: FanoutTarget[]; /** Idempotency key forwarded to the underlying transport. Outbox worker @@ -26,7 +38,10 @@ export interface EncryptedSendInput { * server's (chatId, clientMsgId) unique constraint dedupes. Optional — * when absent the transport generates a nanoid internally. */ clientMsgId?: string; -} +} & ( + | { text: string; reaction?: undefined } + | { reaction: ReactionInput; text?: undefined } +); export interface EncryptedSendResultV1 extends SendResult { recipientAccountId: string; @@ -55,9 +70,18 @@ export interface EncryptedChatTransport { close(): void; subscribe(chatIds: string[]): void; send(input: EncryptedSendInput): Promise; + /** Ephemeral typing signal — plaintext metadata, bypasses the crypto pipe + * (whether someone is typing isn't content). */ + sendTyping(input: { chatId: string; on: boolean }): void; + /** Read watermark — plaintext metadata (a serverMsgId), bypasses crypto. */ + sendRead(input: { chatId: string; upto: string }): void; onMessage(handler: (msg: EncryptedIncomingMessage) => void): () => void; onState(handler: (state: ConnectionState) => void): () => void; onError(handler: (err: IncomingError) => void): () => void; + /** Inbound typing fanout (peer started/stopped composing). */ + onTyping(handler: (m: IncomingTyping) => void): () => void; + /** Inbound read fanout (peer advanced their read watermark). */ + onRead(handler: (m: IncomingRead) => void): () => void; /** Server-pushed MLS Welcome wake-up. Caller should call * MlsClient.pollPendingWelcomes() immediately. Best-effort: the 30s * background poll remains the correctness fallback. */ @@ -211,6 +235,28 @@ export function createEncryptedTransport( for (const h of decryptedHandlers) h(decoded); } + // Build the plaintext frame body for a send. Reactions carry no text/sender; + // messages resolve the display-name snapshot only on the v=2 path (`withSender`) + // — the v=1 legacy path never populated it. + async function frameBody( + input: EncryptedSendInput, + withSender: boolean, + ): Promise { + if (input.reaction) { + return { + kind: "rx", + target: input.reaction.target, + emoji: input.reaction.emoji, + op: input.reaction.op, + }; + } + if (withSender && opts.getMyDisplayName) { + const sender = await opts.getMyDisplayName().catch(() => null); + return sender ? { text: input.text, sender } : { text: input.text }; + } + return { text: input.text }; + } + async function send( input: EncryptedSendInput, ): Promise { @@ -223,14 +269,11 @@ export function createEncryptedTransport( // Resolve display name in parallel with anything else the caller // wants to await — currently nothing, but if a snapshot/ratchet // persist call lands here later it should not block on this read. - const sender = opts.getMyDisplayName - ? await opts.getMyDisplayName().catch(() => null) - : null; + const body = await frameBody(input, true); const env = encryptOutboundMls({ - text: input.text, + body, groupId, mls: opts.mls, - ...(sender ? { sender } : {}), }); const r = await underlying.send({ chatId: input.chatId, @@ -243,8 +286,9 @@ export function createEncryptedTransport( } const seed = await opts.getMySeed(); + const body = await frameBody(input, false); const envelopes = encryptOutbound({ - text: input.text, + body, targets: input.targets, seed, }); @@ -298,9 +342,14 @@ export function createEncryptedTransport( close: () => underlying.close(), subscribe: (chatIds) => underlying.subscribe(chatIds), send, + // Ephemeral signals delegate straight through — no encryption. + sendTyping: (input) => underlying.sendTyping(input), + sendRead: (input) => underlying.sendRead(input), onMessage, onState: (handler) => underlying.onState(handler), onError: (handler) => underlying.onError(handler), + onTyping: (handler) => underlying.onTyping(handler), + onRead: (handler) => underlying.onRead(handler), onMlsWelcome: (handler) => underlying.onMlsWelcome((env) => handler(env.ts)), }; diff --git a/packages/chat/src/hooks.ts b/packages/chat/src/hooks.ts index 5fe213d..484c819 100644 --- a/packages/chat/src/hooks.ts +++ b/packages/chat/src/hooks.ts @@ -1,13 +1,13 @@ // Public React hooks over the chat store. Selectors use useShallow so // callers don't re-render on unrelated slice changes. -import { useCallback } from "react"; +import { useCallback, useEffect, useRef } from "react"; import { encode } from "@msgpack/msgpack"; import { useShallow } from "zustand/react/shallow"; import { useChatTransport, useOutbox, useOutboxWorker } from "./provider"; import { useChatStore } from "./store"; -import type { ChatRecord, Message } from "./types"; +import type { ChatRecord, Message, Reaction } from "./types"; export interface UseChatsResult { chats: ChatRecord[]; @@ -223,3 +223,294 @@ export function useDeleteMessage(): UseDeleteMessageResult { ); return { delete: del }; } + +// React-to-message hook (M8). Toggles the given emoji on a message: if the +// user already has it on that target → op "del", else → op "add". The reaction +// rides the same encrypted outbox path as a message (Option A) — it's an MLS +// application message carrying a "rx" frame, so per-chat FIFO/generation order +// is preserved across retries. Optimistic: the pill appears (add) or disappears +// (del) immediately; the worker reconciles on ack. +export interface UseReactToMessageResult { + react(input: { + chatId: string; + /** serverSerial (string) of the message being reacted to. */ + target: string; + emoji: string; + senderAuthUserId: string; + }): void; +} + +export function useReactToMessage(): UseReactToMessageResult { + const addOptimistic = useChatStore((s) => s.addOptimisticReaction); + const removeOwn = useChatStore((s) => s.removeOwnReaction); + const confirm = useChatStore((s) => s.confirmReaction); + const fail = useChatStore((s) => s.failReaction); + const transport = useChatTransport(); + const outbox = useOutbox(); + const worker = useOutboxWorker(); + + const react = useCallback( + (input: { + chatId: string; + target: string; + emoji: string; + senderAuthUserId: string; + }) => { + // Toggle: derive op from current state. getState() reads synchronously — + // this is an event handler, not render, so it's safe + avoids a subscription. + const perChat = useChatStore.getState().reactions.get(input.chatId); + const mine = (perChat?.get(input.target) ?? []).some( + (r) => + r.senderAuthUserId === input.senderAuthUserId && + r.emoji === input.emoji, + ); + const op: "add" | "del" = mine ? "del" : "add"; + const clientMsgId = randomClientMsgId(); + + if (op === "add") { + addOptimistic({ + chatId: input.chatId, + clientMsgId, + target: input.target, + emoji: input.emoji, + senderAuthUserId: input.senderAuthUserId, + }); + } else { + removeOwn({ + chatId: input.chatId, + target: input.target, + emoji: input.emoji, + senderAuthUserId: input.senderAuthUserId, + }); + } + + if (outbox && worker) { + const payload = encode({ + kind: "rx", + target: input.target, + emoji: input.emoji, + op, + }); + void (async () => { + try { + await outbox.enqueue({ + id: clientMsgId, + chatId: input.chatId, + payload, + idempotencyKey: clientMsgId, + }); + worker.kick(); + } catch (err) { + console.warn("[chat] reaction enqueue failed", err); + // Only "add" has a pill to roll back; a failed "del" reconciles on + // reload (in-memory, demo-scoped). + if (op === "add") fail({ chatId: input.chatId, clientMsgId }); + } + })(); + return; + } + + // Fallback: direct send without retry (tests, pre-outbox screens). + void (async () => { + try { + const results = await transport.send({ + chatId: input.chatId, + reaction: { target: input.target, emoji: input.emoji, op }, + targets: [], + clientMsgId, + }); + if (!results[0]) throw new Error("transport returned no results"); + if (op === "add") confirm({ chatId: input.chatId, clientMsgId }); + } catch (err) { + console.warn("[chat] reaction send failed", err); + if (op === "add") fail({ chatId: input.chatId, clientMsgId }); + } + })(); + }, + [addOptimistic, removeOwn, confirm, fail, outbox, transport, worker], + ); + return { react }; +} + +// Selector hook: reactions folded onto a chat's messages, keyed by target +// serverSerial. Stable empty-map fallback to avoid re-render loops. +const EMPTY_REACTIONS: ReadonlyMap = new Map(); + +export function useReactions(chatId: string): ReadonlyMap { + return useChatStore((s) => s.reactions.get(chatId) ?? EMPTY_REACTIONS); +} + +// Selector hook: userIds currently typing in a chat (excluding an optional +// self id). Filters by expiry at read time as a belt-and-suspenders alongside +// the provider's sweep timer. useShallow keeps the array reference stable while +// the typer set is unchanged, avoiding re-render loops. +const EMPTY_TYPERS: string[] = []; + +export function useTypers(chatId: string, excludeUserId?: string): string[] { + return useChatStore( + useShallow((s) => { + const perChat = s.typing.get(chatId); + if (!perChat || perChat.size === 0) return EMPTY_TYPERS; + const now = Date.now(); + const out: string[] = []; + for (const [userId, expiresAt] of perChat) { + if (expiresAt > now && userId !== excludeUserId) out.push(userId); + } + return out.length === 0 ? EMPTY_TYPERS : out; + }), + ); +} + +// Selector hook: whether my outgoing message (by serverSerial) has been read by +// any peer — i.e. some member other than me has a read watermark ≥ its serial. +// Returns a boolean (Object.is-stable, no useShallow needed). +export function useIsReadByPeer( + chatId: string, + serverSerial: string | undefined, + myAuthUserId: string | null, +): boolean { + return useChatStore((s) => { + if (!serverSerial) return false; + const perChat = s.readReceipts.get(chatId); + if (!perChat) return false; + let target: bigint; + try { + target = BigInt(serverSerial); + } catch { + return false; + } + for (const [userId, upto] of perChat) { + if (userId === myAuthUserId) continue; + try { + if (BigInt(upto) >= target) return true; + } catch { + // skip a malformed watermark + } + } + return false; + }); +} + +// Selector: the greatest read watermark among a chat's members excluding self, +// as a serverSerial string (or null). Lets a thread derive per-message read +// state without a hook-per-row — compare each message's serial to this once. +export function usePeerReadWatermark( + chatId: string, + excludeUserId: string | null, +): string | null { + return useChatStore((s) => { + const perChat = s.readReceipts.get(chatId); + if (!perChat) return null; + let max: bigint | null = null; + let maxStr: string | null = null; + for (const [userId, upto] of perChat) { + if (userId === excludeUserId) continue; + try { + const v = BigInt(upto); + if (max === null || v > max) { + max = v; + maxStr = upto; + } + } catch { + // skip a malformed watermark + } + } + return maxStr; + }); +} + +// Typing-emitter hook. Wire onActivity() to the composer's onChangeText and +// stop() to send/blur. Emits `on:true` on first activity, re-emits as a ~3s +// heartbeat while composing (keeps the receiver's 6s TTL alive), and `on:false` +// after a 4s idle gap or an explicit stop(). +const TYPING_HEARTBEAT_MS = 3_000; +const TYPING_IDLE_MS = 4_000; + +export interface UseTypingEmitterResult { + onActivity(): void; + stop(): void; +} + +export function useTypingEmitter(chatId: string): UseTypingEmitterResult { + const transport = useChatTransport(); + const activeRef = useRef(false); + const heartbeatRef = useRef | null>(null); + const idleRef = useRef | null>(null); + + const clearTimers = useCallback(() => { + if (heartbeatRef.current) { + clearInterval(heartbeatRef.current); + heartbeatRef.current = null; + } + if (idleRef.current) { + clearTimeout(idleRef.current); + idleRef.current = null; + } + }, []); + + const stop = useCallback(() => { + clearTimers(); + if (!activeRef.current) return; + activeRef.current = false; + transport.sendTyping({ chatId, on: false }); + }, [chatId, clearTimers, transport]); + + const onActivity = useCallback(() => { + if (!activeRef.current) { + activeRef.current = true; + transport.sendTyping({ chatId, on: true }); + heartbeatRef.current = setInterval(() => { + transport.sendTyping({ chatId, on: true }); + }, TYPING_HEARTBEAT_MS); + } + if (idleRef.current) clearTimeout(idleRef.current); + idleRef.current = setTimeout(stop, TYPING_IDLE_MS); + }, [chatId, stop, transport]); + + // Stop + clear on unmount or chat change so a "typing…" doesn't leak into a + // thread the user left. Cleanup runs the previous chatId's stop(). + useEffect(() => stop, [stop]); + + return { onActivity, stop }; +} + +// Read-emitter hook. Call markRead(serverSerial) when the newest visible +// message changes (thread focus / list viewability). Monotonic + deduped so we +// only emit when the watermark advances. Gated by `enabled` — the symmetric +// read-receipts privacy toggle (#8); when off we emit nothing (and, per the +// symmetric model, the UI also suppresses peers' receipts). +export interface UseReadEmitterResult { + markRead(serverSerial: string): void; +} + +export function useReadEmitter( + chatId: string, + enabled: boolean, +): UseReadEmitterResult { + const transport = useChatTransport(); + const lastSentRef = useRef(null); + + const markRead = useCallback( + (serverSerial: string) => { + if (!enabled) return; + let serial: bigint; + try { + serial = BigInt(serverSerial); + } catch { + return; + } + if (lastSentRef.current !== null && serial <= lastSentRef.current) return; + lastSentRef.current = serial; + transport.sendRead({ chatId, upto: serverSerial }); + }, + [chatId, enabled, transport], + ); + + // Reset the watermark when switching chats so the first read in a new thread + // always emits. + useEffect(() => { + lastSentRef.current = null; + }, [chatId]); + + return { markRead }; +} diff --git a/packages/chat/src/index.ts b/packages/chat/src/index.ts index 990ab5d..9a00c73 100644 --- a/packages/chat/src/index.ts +++ b/packages/chat/src/index.ts @@ -7,9 +7,13 @@ export { encryptOutbound, encryptOutboundMls, decryptInbound, + isReactionFrame, DecryptError, FrameVersionError, type ChatFrame, + type ChatMsgFrame, + type ChatReactionFrame, + type ChatFrameBody, type FanoutTarget, type MlsApi, type OutboundEnvelope, @@ -46,12 +50,22 @@ export { useSendMessage, useRetryMessage, useDeleteMessage, + useReactToMessage, + useReactions, + useTypers, + useIsReadByPeer, + usePeerReadWatermark, + useTypingEmitter, + useReadEmitter, type UseChatsResult, type UseChatResult, type UseMessagesResult, type UseSendMessageResult, type UseRetryMessageResult, type UseDeleteMessageResult, + type UseReactToMessageResult, + type UseTypingEmitterResult, + type UseReadEmitterResult, } from "./hooks"; export { ChatStoreProvider, @@ -66,6 +80,7 @@ export type { Member, Message, MessageStatus, + Reaction, ChatListInput, ChatListOutput, ChatCreateInput, diff --git a/packages/chat/src/outbox-worker.ts b/packages/chat/src/outbox-worker.ts index e5381bd..7242a50 100644 --- a/packages/chat/src/outbox-worker.ts +++ b/packages/chat/src/outbox-worker.ts @@ -44,6 +44,11 @@ export interface OutboxWorkerStoreApi { chatId: string; clientMsgId: string; }): void; + // Reaction reconciliation (M8). Optional so environments without reactions + // (tests, legacy debug) still satisfy the interface. A "del" row has no + // optimistic pill keyed by clientMsgId, so these no-op for it. + confirmReaction?(input: { chatId: string; clientMsgId: string }): void; + failReaction?(input: { chatId: string; clientMsgId: string }): void; } export interface OutboxWorkerDeps { @@ -114,6 +119,63 @@ export function createOutboxWorker(deps: OutboxWorkerDeps): OutboxWorker { deps.onTerminalFailure?.(row.id, reason); return; } + + // Reaction row (M8) — rides the same encrypted send path as a message but + // carries a "rx" payload instead of `text`. Reconciles the optimistic pill + // rather than a message bubble. + const asRx = decoded as { + kind?: unknown; + target?: unknown; + emoji?: unknown; + op?: unknown; + }; + if (asRx.kind === "rx") { + if ( + typeof asRx.target !== "string" || + typeof asRx.emoji !== "string" || + (asRx.op !== "add" && asRx.op !== "del") + ) { + const reason = "reaction payload missing/invalid target|emoji|op"; + await deps.outbox.markPermanentlyFailed(row.id, reason); + deps.store.failReaction?.({ chatId: row.chat_id, clientMsgId: row.id }); + deps.onTerminalFailure?.(row.id, reason); + return; + } + try { + const results = await deps.transport.send({ + chatId: row.chat_id, + reaction: { target: asRx.target, emoji: asRx.emoji, op: asRx.op }, + targets: [], + clientMsgId: row.id, + }); + if (!results[0]) throw new Error("transport returned no results"); + await deps.outbox.markSent(row.id); + deps.store.confirmReaction?.({ + chatId: row.chat_id, + clientMsgId: row.id, + }); + } catch (err) { + const reason = err instanceof Error ? err.message : String(err); + const attemptsAfter = row.attempts + 1; + if (attemptsAfter >= MAX_ATTEMPTS) { + await deps.outbox.markPermanentlyFailed(row.id, reason); + deps.store.failReaction?.({ + chatId: row.chat_id, + clientMsgId: row.id, + }); + deps.onTerminalFailure?.(row.id, reason); + } else { + await deps.outbox.markFailed( + row.id, + reason, + backoffFor(attemptsAfter), + ); + deps.onTransientFailure?.(row.id, reason); + } + } + return; + } + if ( typeof decoded !== "object" || decoded === null || diff --git a/packages/chat/src/provider.tsx b/packages/chat/src/provider.tsx index 094855d..35f8427 100644 --- a/packages/chat/src/provider.tsx +++ b/packages/chat/src/provider.tsx @@ -8,6 +8,7 @@ import { createContext, useContext, useEffect, type ReactNode } from "react"; +import { isReactionFrame } from "./crypto-pipe"; import type { EncryptedChatTransport } from "./encrypted-transport"; import type { BoundOutboxApi, OutboxWorker } from "./outbox-worker"; import { useChatStore } from "./store"; @@ -71,6 +72,10 @@ export function ChatStoreProvider({ const refresh = useChatStore((s) => s.refresh); const reset = useChatStore((s) => s.reset); const addIncomingMessage = useChatStore((s) => s.addIncomingMessage); + const applyIncomingReaction = useChatStore((s) => s.applyIncomingReaction); + const setTyping = useChatStore((s) => s.setTyping); + const setReadReceipt = useChatStore((s) => s.setReadReceipt); + const sweepExpiredTyping = useChatStore((s) => s.sweepExpiredTyping); const setPersistApi = useChatStore((s) => s.setPersistApi); const hydrateMessages = useChatStore((s) => s.hydrateMessages); @@ -141,10 +146,21 @@ export function ChatStoreProvider({ }, [authenticated, api, refresh, transport]); // Live message ingestion. The transport already decrypts (v=1 libsodium - // or v=2 MLS) and surfaces a ChatFrame with plaintext text + ts. + // or v=2 MLS) and surfaces a ChatFrame. A reaction frame (kind "rx") folds + // onto its target bubble instead of rendering as a message row. useEffect(() => { if (!authenticated) return; return transport.onMessage((msg) => { + if (isReactionFrame(msg.frame)) { + applyIncomingReaction({ + chatId: msg.chatId, + target: msg.frame.target, + emoji: msg.frame.emoji, + op: msg.frame.op, + senderAuthUserId: msg.senderId, + }); + return; + } addIncomingMessage({ chatId: msg.chatId, serverMsgId: msg.serverMsgId, @@ -153,7 +169,32 @@ export function ChatStoreProvider({ ts: msg.ts, }); }); - }, [authenticated, addIncomingMessage, transport]); + }, [authenticated, addIncomingMessage, applyIncomingReaction, transport]); + + // Live typing + read-receipt ingestion. These are ephemeral/metadata frames + // (plaintext, not encrypted) fanned out by the Chat DO. + useEffect(() => { + if (!authenticated) return; + const offTyping = transport.onTyping((m) => { + setTyping({ chatId: m.chatId, userId: m.userId, on: m.on }); + }); + const offRead = transport.onRead((m) => { + setReadReceipt({ chatId: m.chatId, userId: m.userId, upto: m.upto }); + }); + return () => { + offTyping(); + offRead(); + }; + }, [authenticated, setTyping, setReadReceipt, transport]); + + // Typing expiry sweep — clears a stuck indicator when a peer's `on:false` + // never arrives (sender crash / dropped frame). 2s cadence sits well under + // the store's 6s TTL. + useEffect(() => { + if (!authenticated) return; + const id = setInterval(() => sweepExpiredTyping(), 2_000); + return () => clearInterval(id); + }, [authenticated, sweepExpiredTyping]); return ( diff --git a/packages/chat/src/store.test.ts b/packages/chat/src/store.test.ts new file mode 100644 index 0000000..a94d2e9 --- /dev/null +++ b/packages/chat/src/store.test.ts @@ -0,0 +1,125 @@ +import { beforeEach, describe, expect, it } from "vitest"; + +import { useChatStore } from "./store"; + +const CHAT = "chat-1"; +const s = () => useChatStore.getState(); + +beforeEach(() => { + s().reset(); +}); + +describe("reaction reducers", () => { + it("adds an optimistic reaction (sending) and confirms it (sent)", () => { + s().addOptimisticReaction({ + chatId: CHAT, + clientMsgId: "c1", + target: "5", + emoji: "👍", + senderAuthUserId: "u1", + }); + expect(s().reactions.get(CHAT)?.get("5")?.[0]?.status).toBe("sending"); + s().confirmReaction({ chatId: CHAT, clientMsgId: "c1" }); + expect(s().reactions.get(CHAT)?.get("5")?.[0]?.status).toBe("sent"); + }); + + it("dedups a repeated (sender,target,emoji) add — LWW is a no-op", () => { + s().applyIncomingReaction({ + chatId: CHAT, + target: "5", + emoji: "👍", + op: "add", + senderAuthUserId: "u2", + }); + s().applyIncomingReaction({ + chatId: CHAT, + target: "5", + emoji: "👍", + op: "add", + senderAuthUserId: "u2", + }); + expect(s().reactions.get(CHAT)?.get("5")).toHaveLength(1); + }); + + it("removes on op del", () => { + s().applyIncomingReaction({ + chatId: CHAT, + target: "5", + emoji: "👍", + op: "add", + senderAuthUserId: "u2", + }); + s().applyIncomingReaction({ + chatId: CHAT, + target: "5", + emoji: "👍", + op: "del", + senderAuthUserId: "u2", + }); + expect(s().reactions.get(CHAT)).toBeUndefined(); + }); + + it("distinct senders on the same emoji both count", () => { + for (const u of ["u2", "u3"]) { + s().applyIncomingReaction({ + chatId: CHAT, + target: "5", + emoji: "👍", + op: "add", + senderAuthUserId: u, + }); + } + expect(s().reactions.get(CHAT)?.get("5")).toHaveLength(2); + }); + + it("rolls back a terminally-failed optimistic reaction", () => { + s().addOptimisticReaction({ + chatId: CHAT, + clientMsgId: "c1", + target: "5", + emoji: "👍", + senderAuthUserId: "u1", + }); + s().failReaction({ chatId: CHAT, clientMsgId: "c1" }); + expect(s().reactions.get(CHAT)).toBeUndefined(); + }); +}); + +describe("typing reducers", () => { + it("sets then clears a typer", () => { + s().setTyping({ chatId: CHAT, userId: "u2", on: true }); + expect(s().typing.get(CHAT)?.has("u2")).toBe(true); + s().setTyping({ chatId: CHAT, userId: "u2", on: false }); + expect(s().typing.get(CHAT)).toBeUndefined(); + }); + + it("sweep prunes an expired typer", () => { + s().setTyping({ chatId: CHAT, userId: "u2", on: true }); + // TTL is 6s; sweeping 10s in the future prunes it. + s().sweepExpiredTyping(Date.now() + 10_000); + expect(s().typing.get(CHAT)).toBeUndefined(); + }); + + it("sweep keeps a fresh typer", () => { + s().setTyping({ chatId: CHAT, userId: "u2", on: true }); + s().sweepExpiredTyping(Date.now()); + expect(s().typing.get(CHAT)?.has("u2")).toBe(true); + }); +}); + +describe("read-receipt reducer (monotonic)", () => { + it("advances forward and ignores regressions", () => { + s().setReadReceipt({ chatId: CHAT, userId: "u2", upto: "10" }); + expect(s().readReceipts.get(CHAT)?.get("u2")).toBe("10"); + s().setReadReceipt({ chatId: CHAT, userId: "u2", upto: "5" }); + expect(s().readReceipts.get(CHAT)?.get("u2")).toBe("10"); + s().setReadReceipt({ chatId: CHAT, userId: "u2", upto: "20" }); + expect(s().readReceipts.get(CHAT)?.get("u2")).toBe("20"); + }); + + it("compares as BigInt, not lexicographically (9 < 100)", () => { + s().setReadReceipt({ chatId: CHAT, userId: "u2", upto: "9" }); + s().setReadReceipt({ chatId: CHAT, userId: "u2", upto: "100" }); + expect(s().readReceipts.get(CHAT)?.get("u2")).toBe("100"); + }); +}); diff --git a/packages/chat/src/store.ts b/packages/chat/src/store.ts index 7f288f0..2879389 100644 --- a/packages/chat/src/store.ts +++ b/packages/chat/src/store.ts @@ -12,7 +12,13 @@ import { create } from "zustand"; -import type { ChatApi, ChatRecord, Message, MessagePersistApi } from "./types"; +import type { + ChatApi, + ChatRecord, + Message, + MessagePersistApi, + Reaction, +} from "./types"; export interface ChatStoreState { chats: Map; @@ -23,6 +29,16 @@ export interface ChatStoreState { /** Messages per chatId, ordered oldest → newest. The thread screen * renders inverted via FlashList. */ messages: Map; + /** Reactions per chatId → target serverSerial → Reaction[]. In-memory only + * (same as plaintext messages, M4-3). Folded onto bubbles by the UI. */ + reactions: Map>; + /** Typing indicators per chatId → userId → expiry epoch-ms (receiver-side + * TTL). A typer past expiry is treated as stopped; sweepExpiredTyping prunes + * so a dropped `on:false` (sender crash) self-clears with no server TTL. */ + typing: Map>; + /** Read watermarks per chatId → userId → their lastReadSerial (string). + * Drives whether my outgoing messages show as read by a peer. */ + readReceipts: Map>; bootstrapStatus: "idle" | "loading" | "ready" | "error"; bootstrapError: string | null; /** Optional local persistence (M4-followup #25). Set by the provider on @@ -79,6 +95,51 @@ export interface ChatStoreActions { chatId: string; clientMsgId: string; }): void; + // ── Reactions (M8) ────────────────────────────────────────────────────── + /** Optimistic add: insert a "sending" reaction pill immediately. Ignores a + * duplicate clientMsgId or an existing (sender, target, emoji). */ + addOptimisticReaction(input: { + chatId: string; + clientMsgId: string; + target: string; + emoji: string; + senderAuthUserId: string; + }): void; + /** Flip an optimistic reaction to "sent" on outbox ack. Keyed by + * clientMsgId; no-op if not found (e.g. an op:"del" row has no pill). */ + confirmReaction(input: { chatId: string; clientMsgId: string }): void; + /** Roll back an optimistic reaction that terminally failed to send. */ + failReaction(input: { chatId: string; clientMsgId: string }): void; + /** Optimistic un-react: remove my own pill immediately (paired with an + * op:"del" send). No rollback on send failure — reconciles on reload + * (in-memory, demo-scoped). */ + removeOwnReaction(input: { + chatId: string; + target: string; + emoji: string; + senderAuthUserId: string; + }): void; + /** Apply an inbound reaction frame. op "add" inserts (dedup per + * sender+target+emoji); op "del" removes the matching pill. */ + applyIncomingReaction(input: { + chatId: string; + target: string; + emoji: string; + op: "add" | "del"; + senderAuthUserId: string; + }): void; + // ── Typing (M8) ─────────────────────────────────────────────────────────── + /** Apply an inbound typing signal. `on` sets/refreshes the userId's expiry + * (now + TTL); `!on` clears it immediately. */ + setTyping(input: { chatId: string; userId: string; on: boolean }): void; + /** Prune expired typers across all chats. Called on a timer by the provider + * so a stuck indicator (dropped `on:false`) self-clears. No-op — skips the + * set() — when nothing expired, to avoid needless re-renders. */ + sweepExpiredTyping(now?: number): void; + // ── Read receipts (M8) ──────────────────────────────────────────────────── + /** Advance a peer's read watermark. Monotonic — only moves forward (BigInt + * compare), so out-of-order `read` fanout can't regress it. */ + setReadReceipt(input: { chatId: string; userId: string; upto: string }): void; /** Inject (or clear) the persistence API. Idempotent. */ setPersistApi(api: MessagePersistApi | null): void; /** Merge a batch of persisted messages into the per-chat slice. Used by @@ -92,10 +153,17 @@ export interface ChatStoreActions { export type ChatStore = ChatStoreState & ChatStoreActions; +// Receiver-side typing expiry. The sender re-emits `on:true` on a ~3s heartbeat +// (see the emit hook, #6), so this must exceed that interval with margin. +const TYPING_TTL_MS = 6_000; + const emptyState: ChatStoreState = { chats: new Map(), chatOrder: [], messages: new Map(), + reactions: new Map(), + typing: new Map(), + readReceipts: new Map(), bootstrapStatus: "idle", bootstrapError: null, persistApi: null, @@ -175,10 +243,19 @@ export const useChatStore = create((set, get) => ({ chats.delete(chatId); const messages = new Map(get().messages); messages.delete(chatId); + const reactions = new Map(get().reactions); + reactions.delete(chatId); + const typing = new Map(get().typing); + typing.delete(chatId); + const readReceipts = new Map(get().readReceipts); + readReceipts.delete(chatId); set({ chats, chatOrder: get().chatOrder.filter((id) => id !== chatId), messages, + reactions, + typing, + readReceipts, }); }, @@ -295,6 +372,197 @@ export const useChatStore = create((set, get) => ({ set({ messages }); }, + // ── Reactions (M8) ──────────────────────────────────────────────────────── + addOptimisticReaction(input) { + const reactions = new Map(get().reactions); + const perChat = new Map(reactions.get(input.chatId)); + const list = perChat.get(input.target) ?? []; + if (list.some((r) => r.clientMsgId === input.clientMsgId)) return; + // One reaction per (sender, target, emoji) — don't stack a duplicate. + if ( + list.some( + (r) => + r.senderAuthUserId === input.senderAuthUserId && + r.emoji === input.emoji, + ) + ) { + return; + } + const reaction: Reaction = { + clientMsgId: input.clientMsgId, + target: input.target, + emoji: input.emoji, + senderAuthUserId: input.senderAuthUserId, + status: "sending", + }; + perChat.set(input.target, [...list, reaction]); + reactions.set(input.chatId, perChat); + set({ reactions }); + }, + + confirmReaction(input) { + const reactions = new Map(get().reactions); + const perChat = reactions.get(input.chatId); + if (!perChat) return; + const nextPerChat = new Map(perChat); + let changed = false; + for (const [target, list] of perChat) { + const idx = list.findIndex((r) => r.clientMsgId === input.clientMsgId); + if (idx < 0) continue; + const next = [...list]; + next[idx] = { ...next[idx]!, status: "sent" }; + nextPerChat.set(target, next); + changed = true; + break; + } + if (!changed) return; + reactions.set(input.chatId, nextPerChat); + set({ reactions }); + }, + + failReaction(input) { + // Terminal failure → drop the optimistic pill entirely. + const reactions = new Map(get().reactions); + const perChat = reactions.get(input.chatId); + if (!perChat) return; + const nextPerChat = new Map(perChat); + let changed = false; + for (const [target, list] of perChat) { + const filtered = list.filter((r) => r.clientMsgId !== input.clientMsgId); + if (filtered.length === list.length) continue; + if (filtered.length === 0) nextPerChat.delete(target); + else nextPerChat.set(target, filtered); + changed = true; + break; + } + if (!changed) return; + if (nextPerChat.size === 0) reactions.delete(input.chatId); + else reactions.set(input.chatId, nextPerChat); + set({ reactions }); + }, + + removeOwnReaction(input) { + const reactions = new Map(get().reactions); + const perChat = reactions.get(input.chatId); + if (!perChat) return; + const list = perChat.get(input.target); + if (!list) return; + const filtered = list.filter( + (r) => + !( + r.senderAuthUserId === input.senderAuthUserId && + r.emoji === input.emoji + ), + ); + if (filtered.length === list.length) return; + const nextPerChat = new Map(perChat); + if (filtered.length === 0) nextPerChat.delete(input.target); + else nextPerChat.set(input.target, filtered); + if (nextPerChat.size === 0) reactions.delete(input.chatId); + else reactions.set(input.chatId, nextPerChat); + set({ reactions }); + }, + + applyIncomingReaction(input) { + const reactions = new Map(get().reactions); + const perChat = new Map(reactions.get(input.chatId)); + const list = perChat.get(input.target) ?? []; + + if (input.op === "del") { + const filtered = list.filter( + (r) => + !( + r.senderAuthUserId === input.senderAuthUserId && + r.emoji === input.emoji + ), + ); + if (filtered.length === list.length) return; + if (filtered.length === 0) perChat.delete(input.target); + else perChat.set(input.target, filtered); + if (perChat.size === 0) reactions.delete(input.chatId); + else reactions.set(input.chatId, perChat); + set({ reactions }); + return; + } + + // op "add" — dedup per (sender, target, emoji). Last-write-wins is trivial + // here: a repeat add is a no-op. + if ( + list.some( + (r) => + r.senderAuthUserId === input.senderAuthUserId && + r.emoji === input.emoji, + ) + ) { + return; + } + const reaction: Reaction = { + clientMsgId: `in-${input.senderAuthUserId}-${input.emoji}-${input.target}`, + target: input.target, + emoji: input.emoji, + senderAuthUserId: input.senderAuthUserId, + status: "sent", + }; + perChat.set(input.target, [...list, reaction]); + reactions.set(input.chatId, perChat); + set({ reactions }); + }, + + // ── Typing (M8) ──────────────────────────────────────────────────────────── + setTyping(input) { + const typing = new Map(get().typing); + const perChat = new Map(typing.get(input.chatId)); + if (input.on) { + perChat.set(input.userId, Date.now() + TYPING_TTL_MS); + } else { + if (!perChat.has(input.userId)) return; + perChat.delete(input.userId); + } + if (perChat.size === 0) typing.delete(input.chatId); + else typing.set(input.chatId, perChat); + set({ typing }); + }, + + sweepExpiredTyping(now = Date.now()) { + const typing = new Map(get().typing); + let changed = false; + for (const [chatId, perChat] of typing) { + const next = new Map(perChat); + let localChanged = false; + for (const [userId, expiresAt] of perChat) { + if (expiresAt <= now) { + next.delete(userId); + localChanged = true; + } + } + if (!localChanged) continue; + changed = true; + if (next.size === 0) typing.delete(chatId); + else typing.set(chatId, next); + } + if (!changed) return; + set({ typing }); + }, + + // ── Read receipts (M8) ────────────────────────────────────────────────────── + setReadReceipt(input) { + const readReceipts = new Map(get().readReceipts); + const perChat = new Map(readReceipts.get(input.chatId)); + const current = perChat.get(input.userId); + // Monotonic: ignore a watermark that doesn't advance. Guard BigInt against + // a malformed serial (defensive — the server sends a numeric serverMsgId). + let advances: boolean; + try { + advances = current === undefined || BigInt(input.upto) > BigInt(current); + } catch { + return; + } + if (!advances) return; + perChat.set(input.userId, input.upto); + readReceipts.set(input.chatId, perChat); + set({ readReceipts }); + }, + setPersistApi(api) { set({ persistApi: api }); }, @@ -321,6 +589,9 @@ export const useChatStore = create((set, get) => ({ chats: new Map(), chatOrder: [], messages: new Map(), + reactions: new Map(), + typing: new Map(), + readReceipts: new Map(), // Preserve persistApi across resets — caller (provider) controls its // lifetime independently from auth state. persistApi: get().persistApi, diff --git a/packages/chat/src/types.ts b/packages/chat/src/types.ts index 626e0c9..06cd9c2 100644 --- a/packages/chat/src/types.ts +++ b/packages/chat/src/types.ts @@ -24,6 +24,23 @@ export interface ChatRecord { export type MessageStatus = "sending" | "sent" | "failed"; +// A reaction folded onto a message bubble. The wire frame is `ChatReactionFrame` +// (crypto-pipe.ts) carried inside the same E2EE ciphertext as a message; this is +// the resolved view-model the store keeps and the UI renders as a pill. +// +// In-memory only for now (like plaintext messages — see the store's M4-3 note); +// a reactions-persistence column is a follow-up. `status` drives the optimistic +// pill: "sending" until the outbox worker acks, "sent" after. +export interface Reaction { + /** clientMsgId of the reaction send — optimistic reconciliation key. */ + clientMsgId: string; + /** serverSerial (string) of the reacted-to message. */ + target: string; + emoji: string; + senderAuthUserId: string; + status: MessageStatus; +} + export interface Message { /** serverMsgId for confirmed messages, clientMsgId for in-flight. */ id: string; diff --git a/packages/chat/vitest.config.ts b/packages/chat/vitest.config.ts new file mode 100644 index 0000000..709f0fa --- /dev/null +++ b/packages/chat/vitest.config.ts @@ -0,0 +1,13 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + // Pure-logic units only: the msgpack frame codec (crypto-pipe, exercised via + // a passthrough MlsApi mock so no native crypto is needed) and the Zustand + // store reducers. React hooks + the native chat-crypto module are out of + // scope here — the frame tests mock @repo/chat-crypto since it calls + // requireNativeModule() at import. + include: ["src/**/*.test.ts"], + environment: "node", + }, +}); diff --git a/packages/database/prisma/migrations/20260704120000_chatmember_last_read_serial/migration.sql b/packages/database/prisma/migrations/20260704120000_chatmember_last_read_serial/migration.sql new file mode 100644 index 0000000..23636c6 --- /dev/null +++ b/packages/database/prisma/migrations/20260704120000_chatmember_last_read_serial/migration.sql @@ -0,0 +1,11 @@ +-- Read-receipt high-water-mark on ChatMember (M8 typing/receipts/reactions). +-- +-- Replaces `lastReadAt TIMESTAMP(3)` (dormant, never written) with a +-- serial-keyed watermark `lastReadSerial BIGINT`. Receipts key off the DO's +-- monotonic serverSerial (ADR-012), not wall-clock: skew-free and directly +-- comparable to a message's serial for "is my message read?" + unread counts. +-- +-- Pre-launch — no read-state data to preserve, so a drop/add is safe. + +ALTER TABLE "ChatMember" DROP COLUMN "lastReadAt"; +ALTER TABLE "ChatMember" ADD COLUMN "lastReadSerial" BIGINT; diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma index d774cb3..2797e71 100644 --- a/packages/database/prisma/schema.prisma +++ b/packages/database/prisma/schema.prisma @@ -503,9 +503,12 @@ model ChatMember { // Member identified by Better Auth user id. Matches the value the chat-ws // Worker derives from the bearer token at WS upgrade time. userId String - // For unread counts + read-receipt UX. Updated by the API on read events. - lastReadAt DateTime? - joinedAt DateTime @default(now()) + // Read-receipt + unread-count high-water-mark: the greatest ChatMessage + // serverSerial this member has read. Keyed on serial (the DO's monotonic + // authority, ADR-012), not wall-clock — skew-free and directly comparable to + // a message's serial. Updated by the API on `read` frames (chat-ws). + lastReadSerial BigInt? + joinedAt DateTime @default(now()) @@unique([chatId, userId]) @@index([userId]) diff --git a/packages/db-edge/src/chat-persist.ts b/packages/db-edge/src/chat-persist.ts index c0587a4..3ec3069 100644 --- a/packages/db-edge/src/chat-persist.ts +++ b/packages/db-edge/src/chat-persist.ts @@ -77,4 +77,22 @@ export class ChatPersistClient { `) as Array<{ userId: string }>; return rows.map((r) => r.userId); } + + // Advance a member's read high-water-mark. Monotonic by construction: the + // WHERE guard only moves it forward, so out-of-order `read` frames (reconnect + // replay, multi-device races) can't regress it. No-op when the member row is + // absent or already at/ahead of `upto`. + async updateLastReadSerial( + chatId: string, + userId: string, + upto: bigint, + ): Promise { + await this.sql` + UPDATE "ChatMember" + SET "lastReadSerial" = ${upto} + WHERE "chatId" = ${chatId} + AND "userId" = ${userId} + AND ("lastReadSerial" IS NULL OR "lastReadSerial" < ${upto}) + `; + } } diff --git a/packages/ui/src/glacier/chat-bubble.tsx b/packages/ui/src/glacier/chat-bubble.tsx index 59ee08e..8643ab4 100644 --- a/packages/ui/src/glacier/chat-bubble.tsx +++ b/packages/ui/src/glacier/chat-bubble.tsx @@ -42,6 +42,9 @@ export interface ChatBubbleProps { status?: BubbleStatus; /** Read-receipt tick node (outgoing only; screen owns the icon). */ receipt?: ReactNode; + /** Reaction pills, rendered attached below the bubble body (screen owns the + * node — it needs reanimated, which this package doesn't depend on). */ + reactions?: ReactNode; onRetryPress?: () => void; } @@ -53,6 +56,7 @@ export function ChatBubble({ sender, status = "sent", receipt, + reactions, onRetryPress, }: ChatBubbleProps) { const sending = status === "sending"; @@ -110,6 +114,17 @@ export function ChatBubble({ )} + {/* Reaction pills — attached under the bubble, aligned to its side. */} + {reactions ? ( + + {reactions} + + ) : null} + {/* Footer: timestamp + receipt, or the failed affordance */} {failed ? ( ; onPress?: () => void; onLongPress?: () => void; + /** E2E anchor forwarded to the row frame. */ + testID?: string; } export function ListRow({ @@ -88,9 +90,10 @@ export function ListRow({ avatar, onPress, onLongPress, + testID, }: ListRowProps) { return ( - + diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 3593513..037e0e6 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -189,7 +189,7 @@ importers: version: 0.32.17(expo@54.0.34)(react-native@0.81.5(@babel/core@7.29.0)(@react-native/metro-config@0.85.3(@babel/core@7.29.0))(@types/react@19.1.17)(react@19.1.0))(react@19.1.0)(typescript@5.9.2) expo-router: specifier: ~6.0.0 - version: 6.0.23(cc35041b6ef808321d4e4724234aa178) + version: 6.0.23(d270f903a381d04171c7e3105bcc3cd4) expo-secure-store: specifier: ~15.0.0 version: 15.0.8(expo@54.0.34) @@ -297,6 +297,9 @@ importers: typescript: specifier: 5.9.2 version: 5.9.2 + vitest: + specifier: ^2.1.9 + version: 2.1.9(@types/node@22.19.18)(lightningcss@1.32.0)(terser@5.47.1) packages/chat-calls: dependencies: @@ -10212,7 +10215,7 @@ snapshots: '@bcoe/v8-coverage@0.2.3': optional: true - '@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0)': + '@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0)': dependencies: '@better-auth/utils': 0.4.0 '@better-fetch/fetch': 1.1.21 @@ -10226,39 +10229,39 @@ snapshots: optionalDependencies: '@cloudflare/workers-types': 4.20260509.1 - '@better-auth/drizzle-adapter@1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)': + '@better-auth/drizzle-adapter@1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)': dependencies: - '@better-auth/core': 1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0) + '@better-auth/core': 1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0) '@better-auth/utils': 0.4.0 - '@better-auth/kysely-adapter@1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)(kysely@0.28.17)': + '@better-auth/kysely-adapter@1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)(kysely@0.28.17)': dependencies: - '@better-auth/core': 1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0) + '@better-auth/core': 1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0) '@better-auth/utils': 0.4.0 optionalDependencies: kysely: 0.28.17 - '@better-auth/memory-adapter@1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)': + '@better-auth/memory-adapter@1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)': dependencies: - '@better-auth/core': 1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0) + '@better-auth/core': 1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0) '@better-auth/utils': 0.4.0 - '@better-auth/mongo-adapter@1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)': + '@better-auth/mongo-adapter@1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)': dependencies: - '@better-auth/core': 1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0) + '@better-auth/core': 1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0) '@better-auth/utils': 0.4.0 - '@better-auth/prisma-adapter@1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)(@prisma/client@6.19.3(prisma@6.19.3(typescript@5.9.2))(typescript@5.9.2))(prisma@6.19.3(typescript@5.9.2))': + '@better-auth/prisma-adapter@1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)(@prisma/client@6.19.3(prisma@6.19.3(typescript@5.9.2))(typescript@5.9.2))(prisma@6.19.3(typescript@5.9.2))': dependencies: - '@better-auth/core': 1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0) + '@better-auth/core': 1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0) '@better-auth/utils': 0.4.0 optionalDependencies: '@prisma/client': 6.19.3(prisma@6.19.3(typescript@5.9.2))(typescript@5.9.2) prisma: 6.19.3(typescript@5.9.2) - '@better-auth/telemetry@1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)': + '@better-auth/telemetry@1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)': dependencies: - '@better-auth/core': 1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0) + '@better-auth/core': 1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0) '@better-auth/utils': 0.4.0 '@better-fetch/fetch': 1.1.21 @@ -10738,7 +10741,7 @@ snapshots: wrap-ansi: 7.0.0 ws: 8.20.0 optionalDependencies: - expo-router: 6.0.23(cc35041b6ef808321d4e4724234aa178) + expo-router: 6.0.23(d270f903a381d04171c7e3105bcc3cd4) react-native: 0.81.5(@babel/core@7.29.0)(@react-native/metro-config@0.85.3(@babel/core@7.29.0))(@types/react@19.1.17)(react@19.1.0) transitivePeerDependencies: - bufferutil @@ -14073,7 +14076,7 @@ snapshots: '@tanstack/query-core': 5.100.9 react: 19.1.0 - '@testing-library/react-native@13.3.3(jest@29.7.0(@types/node@22.19.18))(react-native@0.81.5(@babel/core@7.29.0)(@react-native/metro-config@0.85.3(@babel/core@7.29.0))(@types/react@19.1.17)(react@19.1.0))(react-test-renderer@19.2.0(react@19.1.0))(react@19.1.0)': + '@testing-library/react-native@13.3.3(jest@29.7.0)(react-native@0.81.5(@babel/core@7.29.0)(@react-native/metro-config@0.85.3(@babel/core@7.29.0))(@types/react@19.1.17)(react@19.1.0))(react-test-renderer@19.2.0(react@19.1.0))(react@19.1.0)': dependencies: jest-matcher-utils: 30.4.1 picocolors: 1.1.1 @@ -14744,13 +14747,13 @@ snapshots: better-auth@1.6.9(@cloudflare/workers-types@4.20260509.1)(@prisma/client@6.19.3(prisma@6.19.3(typescript@5.9.2))(typescript@5.9.2))(next@16.1.0(@babel/core@7.29.0)(babel-plugin-react-compiler@1.0.0)(react-dom@19.1.0(react@19.1.0))(react@19.1.0))(prisma@6.19.3(typescript@5.9.2))(react-dom@19.1.0(react@19.1.0))(react@19.1.0)(vitest@2.1.9(@types/node@22.19.18)(lightningcss@1.32.0)(terser@5.47.1)): dependencies: - '@better-auth/core': 1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0) - '@better-auth/drizzle-adapter': 1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0) - '@better-auth/kysely-adapter': 1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)(kysely@0.28.17) - '@better-auth/memory-adapter': 1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0) - '@better-auth/mongo-adapter': 1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0) - '@better-auth/prisma-adapter': 1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)(@prisma/client@6.19.3(prisma@6.19.3(typescript@5.9.2))(typescript@5.9.2))(prisma@6.19.3(typescript@5.9.2)) - '@better-auth/telemetry': 1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@3.25.76))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21) + '@better-auth/core': 1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0) + '@better-auth/drizzle-adapter': 1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0) + '@better-auth/kysely-adapter': 1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)(kysely@0.28.17) + '@better-auth/memory-adapter': 1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0) + '@better-auth/mongo-adapter': 1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0) + '@better-auth/prisma-adapter': 1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)(@prisma/client@6.19.3(prisma@6.19.3(typescript@5.9.2))(typescript@5.9.2))(prisma@6.19.3(typescript@5.9.2)) + '@better-auth/telemetry': 1.6.9(@better-auth/core@1.6.9(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@cloudflare/workers-types@4.20260509.1)(better-call@1.3.5(zod@4.4.3))(jose@6.2.3)(kysely@0.28.17)(nanostores@1.3.0))(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21) '@better-auth/utils': 0.4.0 '@better-fetch/fetch': 1.1.21 '@noble/ciphers': 2.2.0 @@ -15532,9 +15535,9 @@ snapshots: '@typescript-eslint/eslint-plugin': 8.59.2(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2) '@typescript-eslint/parser': 8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2) eslint: 9.39.4(jiti@2.7.0) - eslint-import-resolver-typescript: 3.10.1(eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint@9.39.4(jiti@2.7.0)))(eslint@9.39.4(jiti@2.7.0)) + eslint-import-resolver-typescript: 3.10.1(eslint-plugin-import@2.32.0)(eslint@9.39.4(jiti@2.7.0)) eslint-plugin-expo: 1.0.0(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2) - eslint-plugin-import: 2.32.0(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint-import-resolver-typescript@3.10.1(eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint@9.39.4(jiti@2.7.0)))(eslint@9.39.4(jiti@2.7.0)))(eslint@9.39.4(jiti@2.7.0)) + eslint-plugin-import: 2.32.0(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint-import-resolver-typescript@3.10.1)(eslint@9.39.4(jiti@2.7.0)) eslint-plugin-react: 7.37.5(eslint@9.39.4(jiti@2.7.0)) eslint-plugin-react-hooks: 5.2.0(eslint@9.39.4(jiti@2.7.0)) globals: 16.5.0 @@ -15556,7 +15559,7 @@ snapshots: transitivePeerDependencies: - supports-color - eslint-import-resolver-typescript@3.10.1(eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint@9.39.4(jiti@2.7.0)))(eslint@9.39.4(jiti@2.7.0)): + eslint-import-resolver-typescript@3.10.1(eslint-plugin-import@2.32.0)(eslint@9.39.4(jiti@2.7.0)): dependencies: '@nolyfill/is-core-module': 1.0.39 debug: 4.4.3 @@ -15567,18 +15570,18 @@ snapshots: tinyglobby: 0.2.16 unrs-resolver: 1.11.1 optionalDependencies: - eslint-plugin-import: 2.32.0(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint-import-resolver-typescript@3.10.1(eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint@9.39.4(jiti@2.7.0)))(eslint@9.39.4(jiti@2.7.0)))(eslint@9.39.4(jiti@2.7.0)) + eslint-plugin-import: 2.32.0(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint-import-resolver-typescript@3.10.1)(eslint@9.39.4(jiti@2.7.0)) transitivePeerDependencies: - supports-color - eslint-module-utils@2.12.1(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint-import-resolver-node@0.3.10)(eslint-import-resolver-typescript@3.10.1(eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint@9.39.4(jiti@2.7.0)))(eslint@9.39.4(jiti@2.7.0)))(eslint@9.39.4(jiti@2.7.0)): + eslint-module-utils@2.12.1(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint-import-resolver-node@0.3.10)(eslint-import-resolver-typescript@3.10.1)(eslint@9.39.4(jiti@2.7.0)): dependencies: debug: 3.2.7 optionalDependencies: '@typescript-eslint/parser': 8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2) eslint: 9.39.4(jiti@2.7.0) eslint-import-resolver-node: 0.3.10 - eslint-import-resolver-typescript: 3.10.1(eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint@9.39.4(jiti@2.7.0)))(eslint@9.39.4(jiti@2.7.0)) + eslint-import-resolver-typescript: 3.10.1(eslint-plugin-import@2.32.0)(eslint@9.39.4(jiti@2.7.0)) transitivePeerDependencies: - supports-color @@ -15591,7 +15594,7 @@ snapshots: - supports-color - typescript - eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint-import-resolver-typescript@3.10.1(eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint@9.39.4(jiti@2.7.0)))(eslint@9.39.4(jiti@2.7.0)))(eslint@9.39.4(jiti@2.7.0)): + eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint-import-resolver-typescript@3.10.1)(eslint@9.39.4(jiti@2.7.0)): dependencies: '@rtsao/scc': 1.1.0 array-includes: 3.1.9 @@ -15602,7 +15605,7 @@ snapshots: doctrine: 2.1.0 eslint: 9.39.4(jiti@2.7.0) eslint-import-resolver-node: 0.3.10 - eslint-module-utils: 2.12.1(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint-import-resolver-node@0.3.10)(eslint-import-resolver-typescript@3.10.1(eslint-plugin-import@2.32.0(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint@9.39.4(jiti@2.7.0)))(eslint@9.39.4(jiti@2.7.0)))(eslint@9.39.4(jiti@2.7.0)) + eslint-module-utils: 2.12.1(@typescript-eslint/parser@8.59.2(eslint@9.39.4(jiti@2.7.0))(typescript@5.9.2))(eslint-import-resolver-node@0.3.10)(eslint-import-resolver-typescript@3.10.1)(eslint@9.39.4(jiti@2.7.0)) hasown: 2.0.3 is-core-module: 2.16.2 is-glob: 4.0.3 @@ -15921,7 +15924,7 @@ snapshots: - supports-color - typescript - expo-router@6.0.23(cc35041b6ef808321d4e4724234aa178): + expo-router@6.0.23(d270f903a381d04171c7e3105bcc3cd4): dependencies: '@expo/metro-runtime': 6.1.2(expo@54.0.34)(react-dom@19.1.0(react@19.1.0))(react-native@0.81.5(@babel/core@7.29.0)(@react-native/metro-config@0.85.3(@babel/core@7.29.0))(@types/react@19.1.17)(react@19.1.0))(react@19.1.0) '@expo/schema-utils': 0.1.8 @@ -15954,7 +15957,7 @@ snapshots: use-latest-callback: 0.2.6(react@19.1.0) vaul: 1.1.2(@types/react-dom@19.1.11(@types/react@19.1.17))(@types/react@19.1.17)(react-dom@19.1.0(react@19.1.0))(react@19.1.0) optionalDependencies: - '@testing-library/react-native': 13.3.3(jest@29.7.0(@types/node@22.19.18))(react-native@0.81.5(@babel/core@7.29.0)(@react-native/metro-config@0.85.3(@babel/core@7.29.0))(@types/react@19.1.17)(react@19.1.0))(react-test-renderer@19.2.0(react@19.1.0))(react@19.1.0) + '@testing-library/react-native': 13.3.3(jest@29.7.0)(react-native@0.81.5(@babel/core@7.29.0)(@react-native/metro-config@0.85.3(@babel/core@7.29.0))(@types/react@19.1.17)(react@19.1.0))(react-test-renderer@19.2.0(react@19.1.0))(react@19.1.0) react-dom: 19.1.0(react@19.1.0) react-native-gesture-handler: 2.28.0(react-native@0.81.5(@babel/core@7.29.0)(@react-native/metro-config@0.85.3(@babel/core@7.29.0))(@types/react@19.1.17)(react@19.1.0))(react@19.1.0) react-native-reanimated: 4.1.7(react-native-worklets@0.8.3(@babel/core@7.29.0)(@react-native/metro-config@0.85.3(@babel/core@7.29.0))(react-native@0.81.5(@babel/core@7.29.0)(@react-native/metro-config@0.85.3(@babel/core@7.29.0))(@types/react@19.1.17)(react@19.1.0))(react@19.1.0))(react-native@0.81.5(@babel/core@7.29.0)(@react-native/metro-config@0.85.3(@babel/core@7.29.0))(@types/react@19.1.17)(react@19.1.0))(react@19.1.0) diff --git a/services/chat-ws/src/durable/chat.ts b/services/chat-ws/src/durable/chat.ts index 71df5a9..3c61bc5 100644 --- a/services/chat-ws/src/durable/chat.ts +++ b/services/chat-ws/src/durable/chat.ts @@ -44,6 +44,13 @@ export class Chat extends DurableObject { private bufferLoaded = false; private nextSerial: bigint | null = null; private flushTimer: number | null = null; + // Authoritative ChatMember set, cached to gate ephemeral typ/read signals + // without a Postgres round-trip per frame (typing heartbeats fire ~every 3s). + // Short TTL bounds staleness for add/remove-member (a removed member can spoof + // for at most MEMBER_CACHE_TTL_MS; a new member is briefly denied) — an + // acceptable window for plaintext metadata signals. + private memberCache: { ids: Set; at: number } | null = null; + private static readonly MEMBER_CACHE_TTL_MS = 30_000; // ── RPC: attach / detach UserInbox membership ───────────────────────────── async attachInbox(userId: string): Promise { @@ -60,6 +67,73 @@ export class Chat extends DurableObject { await this.ctx.storage.put(ATTACHED_KEY, [...set]); } + // ── RPC: typing relay (ephemeral — no persistence, no storage, no alarm) ── + // A UserInbox forwards a typing signal. Pure fanout to every *other* attached + // member. Expiry lives on the receiver client (short timer refreshed by the + // sender's ~3s `on:true` heartbeat), so a dropped `on:false` self-clears + // without any server-side TTL — keeps this hibernation-proof and stateless. + async acceptTyping(input: { userId: string; on: boolean }): Promise { + const chatId = this.ctx.id.name ?? this.ctx.id.toString(); + // Authorization: typing/read frames carry a client-supplied chatId with no + // crypto barrier (unlike `send`, whose secrecy rests on E2EE). Gate on the + // authoritative member set so a client can't spoof presence into a chat it + // isn't in. + if (!(await this.isMember(chatId, input.userId))) return; + const attached = await this.loadAttached(); + await Promise.all( + [...attached] + .filter((userId) => userId !== input.userId) + .map((userId) => { + const stub = this.env.USER_INBOX.get( + this.env.USER_INBOX.idFromName(userId), + ); + return stub.deliverTyping({ + chatId, + userId: input.userId, + on: input.on, + }); + }), + ); + } + + // ── RPC: read receipt (persist watermark + fanout) ──────────────────────── + // A UserInbox forwards a read high-water-mark. Persist it (monotonic) to + // ChatMember.lastReadSerial, then fan out to every *other* attached member so + // their outgoing bubbles flip sent → read live. Offline members reconcile the + // watermark from ChatMember on their next chat load. + async acceptRead(input: { userId: string; upto: string }): Promise { + const chatId = this.ctx.id.name ?? this.ctx.id.toString(); + // Same authorization gate as acceptTyping — a non-member must not be able to + // spoof a read receipt (fanned out to real members) into this chat. + if (!(await this.isMember(chatId, input.userId))) return; + let serial: bigint; + try { + serial = BigInt(input.upto); + } catch { + // Malformed watermark — drop silently (content-blind soft handling). + return; + } + + const db = getPersistClient(this.env); + await db.updateLastReadSerial(chatId, input.userId, serial); + + const attached = await this.loadAttached(); + await Promise.all( + [...attached] + .filter((userId) => userId !== input.userId) + .map((userId) => { + const stub = this.env.USER_INBOX.get( + this.env.USER_INBOX.idFromName(userId), + ); + return stub.deliverRead({ + chatId, + userId: input.userId, + upto: input.upto, + }); + }), + ); + } + // ── RPC: a UserInbox forwards an outbound send ──────────────────────────── async acceptSend(input: { senderId: string; @@ -310,6 +384,21 @@ export class Chat extends DurableObject { return new Set(arr); } + // Membership check against the authoritative ChatMember set, memoised with a + // short TTL so hot paths (typing heartbeats) don't hit Postgres per frame. + private async isMember(chatId: string, userId: string): Promise { + const now = Date.now(); + if ( + !this.memberCache || + now - this.memberCache.at > Chat.MEMBER_CACHE_TTL_MS + ) { + const db = getPersistClient(this.env); + const ids = await db.memberIds(chatId); + this.memberCache = { ids: new Set(ids), at: now }; + } + return this.memberCache.ids.has(userId); + } + private async dispatchErr( senderIds: string[], code: "PERSIST_FAILED", diff --git a/services/chat-ws/src/durable/user-inbox.ts b/services/chat-ws/src/durable/user-inbox.ts index ed2a609..70aecd7 100644 --- a/services/chat-ws/src/durable/user-inbox.ts +++ b/services/chat-ws/src/durable/user-inbox.ts @@ -20,6 +20,8 @@ import { validateSendFrame } from "../validators"; // RPC surface (called by Chat DO): // - deliverBatch(frames[]) — fan a list of `msg` frames to all sockets // - ackBatch(frames[]) — fan a list of `ack` frames to all sockets +// - deliverTyping(payload) — single `typ` frame (peer typing, ephemeral) +// - deliverRead(payload) — single `read` frame (peer advanced read watermark) // - error({ code, msg }) — single `err` frame, soft error (connection stays) interface SocketAttachment { @@ -38,6 +40,8 @@ type MsgFrame = Extract; type AckFrame = Extract; type MsgPayload = Omit; type AckPayload = Omit; +type TypPayload = Omit, "t">; +type ReadPayload = Omit, "t">; export class UserInbox extends DurableObject { constructor(ctx: DurableObjectState, env: Env) { @@ -114,6 +118,12 @@ export class UserInbox extends DurableObject { case "send": await this.handleSend(ws, att.userId, env); return; + case "typ": + await this.handleTyping(att.userId, env); + return; + case "read": + await this.handleRead(att.userId, env); + return; case "ping": ws.send(encodeFrame({ t: "pong" } satisfies ServerToClient)); return; @@ -189,6 +199,26 @@ export class UserInbox extends DurableObject { } } + // ── Typing / read → Chat DO ─────────────────────────────────────────────── + private async handleTyping( + userId: string, + env: Extract, + ): Promise { + if (!env.chatId) return; + const stub = this.env.CHAT.get(this.env.CHAT.idFromName(env.chatId)); + // Fire-and-forget: typing is best-effort, a failed relay just drops. + await stub.acceptTyping({ userId, on: env.on }); + } + + private async handleRead( + userId: string, + env: Extract, + ): Promise { + if (!env.chatId || !env.upto) return; + const stub = this.env.CHAT.get(this.env.CHAT.idFromName(env.chatId)); + await stub.acceptRead({ userId, upto: env.upto }); + } + // ── Presence RPC (called by Chat DO before push fanout) ────────────────── // Returns the set of deviceIds with an open WS attached to this user. The // empty string is filtered out — pre-M6 clients connect without a `did` @@ -251,6 +281,22 @@ export class UserInbox extends DurableObject { } } + // Fan a single `typ` frame to every socket of this user (a peer started or + // stopped composing in a shared chat). Ephemeral, best-effort — no retry. + async deliverTyping(payload: TypPayload): Promise { + this.broadcast( + encodeFrame({ t: "typ", ...payload } satisfies ServerToClient), + ); + } + + // Fan a single `read` frame to every socket of this user (a peer advanced + // their read watermark). Flips this user's outgoing bubbles sent → read. + async deliverRead(payload: ReadPayload): Promise { + this.broadcast( + encodeFrame({ t: "read", ...payload } satisfies ServerToClient), + ); + } + // Fan a single `mls-welcome` wake-up to every socket of this user. Called // by the Worker's /internal/notify endpoint after the API publishes one or // more Welcomes addressed to this user. Lets the client skip the 30s @@ -293,6 +339,20 @@ export class UserInbox extends DurableObject { } // ── Internal helpers ────────────────────────────────────────────────────── + // Send one pre-encoded frame to every open socket of this user. Best-effort: + // a failed send is dropped (the close handler cleans up dead sockets). + private broadcast(frame: ReturnType): void { + const sockets = this.ctx.getWebSockets(); + for (const ws of sockets) { + if (ws.readyState !== WebSocket.READY_STATE_OPEN) continue; + try { + ws.send(frame); + } catch { + // ignore + } + } + } + private sendErr(ws: WebSocket, code: ChatErrorCode, msg?: string): void { try { ws.send(encodeFrame({ t: "err", code, msg } satisfies ServerToClient));