Merge remote-tracking branch 'origin/main' into fix/ui-thinking-effort-draft-project-rename
This commit is contained in:
Bohdan Triapitsyn
@@ -64,7 +64,9 @@ These stores coordinate persistent project/session metadata across multiple view
|
||||
|
||||
`useProjectContextStore.ts` caches server-owned project notes, todos, and plan links, keyed by the path-derived project id. It replaced a pair of `window` CustomEvents that made every mounted notes panel re-read the whole project config. Writes are optimistic and roll back on failure; they are serialized per project, because the server's own store does a read-modify-write and two concurrent saves would otherwise race it. A load that resolves while a write is in flight keeps the local value for that field group only, so a slow snapshot cannot undo newer typing while still delivering the plan list it fetched. A failed load sets `error` and preserves the cached snapshot — an unreachable server must never render as "this project has no notes". Note and plan creation are deliberately not optimistic, since ids and timestamps are assigned by the server. Notes, todos, and plans are written through separate routes and tracked by separate in-flight flags, so a todo toggle cannot clobber a note edit in the same window. Pinned notes and plans are assembled into a synthetic context part by `lib/projectContextPinning.ts` at send time; that module tracks per-session what it already sent so an unchanged pinned set is not re-sent every turn.
|
||||
|
||||
`messageQueueStore.ts` keeps a queued message until its own send resolves, so between dispatch and resolution the entry is still visible to every reader. Dispatchers must therefore mark the send (`markSending`/`clearSending`) and read `getSendableQueue()` — or filter `sendingIds` themselves — instead of dispatching straight from `queuedMessages`; otherwise a composer submit merges a message the auto-send hook is already delivering and it is sent twice (the window is seconds over a relay). `clearQueue()` retains in-flight entries for the same reason. `sendingIds` is deliberately not persisted: a restart has no in-flight sends, and a stale flag would strand a queued message. Desktop queues use the configured host id as runtime identity, not the current API URL, because an SSH reconnect allocates a new local forwarding port while the remote host remains the same.
|
||||
`messageQueueStore.ts` has two owners, decided by `isServerOwnedMessageQueue()`. On web, desktop, and mobile the OpenChamber server owns the queue (`packages/web/server/lib/message-queue/`): it delivers queued messages when the session goes idle whether or not any UI is open, and the store is a projection of it — `hydrate()` loads the server snapshot for the active runtime, `openchamber:message-queue.updated` broadcasts keep it current, and every mutation is optimistic locally then settled on the server's copy of that session (a failed round-trip re-reads the server instead of guessing). A per-key server revision rejects stale snapshots. Projection items carry attachment metadata only; `popToInput()`/`takeForSend()` remove the message on the server and get the full payload back, which is why both are async. Messages a previous build left in this browser are uploaded once on the first hydration of a runtime and then dropped from persistence for that runtime (`partialize` skips server-owned runtime keys). VS Code has no server and keeps the local queue with the foreground auto-send hook (`useQueuedMessageAutoSend`, enabled only there); `useMessageQueueHoldSync` tells the server to hold a session's queue while a UI-driven auto-review run is going.
|
||||
|
||||
In the local (VS Code) mode the store keeps a queued message until its own send resolves, so between dispatch and resolution the entry is still visible to every reader. Dispatchers must therefore mark the send (`markSending`/`clearSending`) and read `getSendableQueue()` — or filter `sendingIds` themselves — instead of dispatching straight from `queuedMessages`; otherwise a composer submit merges a message the auto-send hook is already delivering and it is sent twice (the window is seconds over a relay). `clearQueue()` retains in-flight entries for the same reason. `sendingIds` is deliberately not persisted: a restart has no in-flight sends, and a stale flag would strand a queued message; in the server-owned mode it mirrors the server's in-flight item. Desktop queues use the configured host id as runtime identity, not the current API URL, because an SSH reconnect allocates a new local forwarding port while the remote host remains the same.
|
||||
|
||||
`useGlobalSessionsStore.ts` owns cold/global active and archived session coverage. Its entity map and active root, parent/child, and directory indexes are maintained in the same transaction as the compatibility arrays and `sessionsByDirectory`. Full authoritative snapshots may rebuild those indexes once; direct create, update, move, archive, and delete mutations update only affected hierarchy and directory buckets. Metadata-only updates preserve the structure reference. It is complementary to directory child stores: it is not the source of live busy/retry status or session messages.
|
||||
|
||||
|
||||
@@ -0,0 +1,225 @@
|
||||
import { beforeEach, describe, expect, mock, test } from "bun:test"
|
||||
import type { AttachedFile } from "./types/sessionTypes"
|
||||
import type { MessageQueueUpdatedEvent } from "./messageQueueStore"
|
||||
|
||||
type FetchCall = { path: string; method: string; body: ReturnType<typeof JSON.parse> }
|
||||
let calls: FetchCall[] = []
|
||||
let respond: (call: FetchCall) => Response = () => new Response("{}", { status: 200 })
|
||||
|
||||
mock.module("@/lib/runtime-fetch", () => ({
|
||||
runtimeFetch: async (path: string, init?: RequestInit) => {
|
||||
const call = {
|
||||
path,
|
||||
method: init?.method ?? "GET",
|
||||
body: init?.body === undefined ? undefined : JSON.parse(String(init.body)),
|
||||
}
|
||||
calls.push(call)
|
||||
return respond(call)
|
||||
},
|
||||
}))
|
||||
const desktop = await import("@/lib/desktop")
|
||||
mock.module("@/lib/desktop", () => ({ ...desktop, isVSCodeRuntime: () => false }))
|
||||
mock.module("@/lib/runtime-switch", () => ({ getRuntimeKey: () => "runtime-a" }))
|
||||
mock.module("@/lib/persistence", () => ({ updateDesktopSettings: async () => undefined }))
|
||||
|
||||
const {
|
||||
applyMessageQueueUpdatedEvent,
|
||||
createMessageQueueTarget,
|
||||
getMessageQueueKey,
|
||||
useMessageQueueStore,
|
||||
} = await import("./messageQueueStore")
|
||||
|
||||
type ServerItem = MessageQueueUpdatedEvent["properties"]["session"]["items"][number]
|
||||
type ServerSession = MessageQueueUpdatedEvent["properties"]["session"]
|
||||
|
||||
type ServerReply = {
|
||||
revision: number
|
||||
session?: ServerSession
|
||||
sessions?: ServerSession[]
|
||||
item?: ServerItem
|
||||
items?: ServerItem[]
|
||||
}
|
||||
|
||||
const json = (value: ServerReply, status = 200) => new Response(JSON.stringify(value), { status })
|
||||
|
||||
const target = createMessageQueueTarget("session-1", "/repo", "runtime-a")!
|
||||
const key = getMessageQueueKey(target)
|
||||
|
||||
const serverItem = (id: string, content: string, extra: Partial<ServerItem> = {}): ServerItem => ({
|
||||
id,
|
||||
createdAt: 1,
|
||||
content,
|
||||
attachments: [],
|
||||
sendConfig: { providerID: "p", modelID: "m" },
|
||||
...extra,
|
||||
})
|
||||
|
||||
const session = (items: ServerItem[], sendingId: string | null = null): ServerSession => ({
|
||||
sessionId: "session-1",
|
||||
directory: "/repo",
|
||||
items,
|
||||
sendingId,
|
||||
})
|
||||
|
||||
const updated = (revision: number, updatedSession: ServerSession): MessageQueueUpdatedEvent => ({
|
||||
type: "openchamber:message-queue.updated",
|
||||
properties: { revision, session: updatedSession },
|
||||
})
|
||||
|
||||
const attachment: AttachedFile = {
|
||||
id: "att-1",
|
||||
file: new File(["hi"], "note.txt", { type: "text/plain" }),
|
||||
dataUrl: "data:text/plain;base64,aGk=",
|
||||
mimeType: "text/plain",
|
||||
filename: "note.txt",
|
||||
size: 2,
|
||||
source: "local",
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
calls = []
|
||||
respond = () => json({ revision: 1, session: session([]) })
|
||||
// Forgetting also drops the revision guard, so each test starts unordered.
|
||||
useMessageQueueStore.getState().forgetQueue(target)
|
||||
useMessageQueueStore.setState({ queuedMessages: {}, quarantinedLegacyMessages: {}, sendingIds: {} })
|
||||
})
|
||||
|
||||
describe("server-owned message queue", () => {
|
||||
// First: the one-time upload of a legacy local queue happens before this
|
||||
// runtime is known to be server-owned, which the later hydrations establish.
|
||||
test("hydrate uploads messages queued by an older build before reading the server", async () => {
|
||||
useMessageQueueStore.setState({
|
||||
queuedMessages: {
|
||||
[key]: [{ id: "local-1", content: "from before", createdAt: 1, sendConfig: { providerID: "p", modelID: "m" } }],
|
||||
},
|
||||
})
|
||||
respond = (call) => (call.method === "POST"
|
||||
? json({ revision: 2, session: session([serverItem("q1", "from before")]) })
|
||||
: json({ revision: 2, sessions: [session([serverItem("q1", "from before")])] }))
|
||||
await useMessageQueueStore.getState().hydrate()
|
||||
|
||||
expect(calls[0]).toEqual({
|
||||
method: "POST",
|
||||
path: "/api/message-queue/sessions/session-1/items",
|
||||
body: { directory: "/repo", item: { content: "from before", text: "from before", attachments: [], sendConfig: { providerID: "p", modelID: "m" } } },
|
||||
})
|
||||
expect(calls[1]?.path).toBe("/api/message-queue")
|
||||
expect(useMessageQueueStore.getState().queuedMessages[key]?.map((m) => m.id)).toEqual(["q1"])
|
||||
})
|
||||
|
||||
test("hydrate replaces the runtime's projection with the server queue", async () => {
|
||||
respond = () => json({ revision: 3, sessions: [session([serverItem("q1", "hello")], "q1")] })
|
||||
await useMessageQueueStore.getState().hydrate()
|
||||
|
||||
expect(calls.map((call) => `${call.method} ${call.path}`)).toEqual(["GET /api/message-queue"])
|
||||
expect(useMessageQueueStore.getState().queuedMessages[key]?.map((m) => m.content)).toEqual(["hello"])
|
||||
expect(useMessageQueueStore.getState().sendingIds[key]).toEqual(["q1"])
|
||||
})
|
||||
|
||||
test("addToQueue shows the message at once and settles on the server's copy", async () => {
|
||||
respond = () => json({ revision: 5, session: session([serverItem("srv-1", "hi @reviewer", { agentMention: "reviewer" })]) })
|
||||
const pending = useMessageQueueStore.getState().addToQueue(target, {
|
||||
content: "hi @reviewer",
|
||||
text: "hi",
|
||||
agentMention: "reviewer",
|
||||
attachments: [attachment],
|
||||
sendConfig: { providerID: "p", modelID: "m", agent: "build" },
|
||||
})
|
||||
expect(useMessageQueueStore.getState().queuedMessages[key]).toHaveLength(1)
|
||||
await pending
|
||||
|
||||
expect(calls[0]).toEqual({
|
||||
method: "POST",
|
||||
path: "/api/message-queue/sessions/session-1/items",
|
||||
body: {
|
||||
directory: "/repo",
|
||||
item: {
|
||||
content: "hi @reviewer",
|
||||
text: "hi",
|
||||
agentMention: "reviewer",
|
||||
attachments: [{ id: "att-1", filename: "note.txt", mimeType: "text/plain", size: 2, source: "local", dataUrl: attachment.dataUrl }],
|
||||
sendConfig: { providerID: "p", modelID: "m", agent: "build" },
|
||||
},
|
||||
},
|
||||
})
|
||||
expect(useMessageQueueStore.getState().queuedMessages[key]?.map((m) => m.id)).toEqual(["srv-1"])
|
||||
})
|
||||
|
||||
test("addToQueue rolls the optimistic entry back when the server refuses", async () => {
|
||||
respond = () => new Response("nope", { status: 500 })
|
||||
await expect(useMessageQueueStore.getState().addToQueue(target, {
|
||||
content: "x",
|
||||
sendConfig: { providerID: "p", modelID: "m" },
|
||||
})).rejects.toThrow()
|
||||
expect(useMessageQueueStore.getState().queuedMessages[key]).toBe(undefined)
|
||||
})
|
||||
|
||||
test("addToQueue refuses a message with no captured model", async () => {
|
||||
await expect(useMessageQueueStore.getState().addToQueue(target, { content: "x" })).rejects.toThrow()
|
||||
expect(useMessageQueueStore.getState().queuedMessages[key]).toBe(undefined)
|
||||
expect(calls).toHaveLength(0)
|
||||
})
|
||||
|
||||
test("takeForSend brings the full message back, attachments included", async () => {
|
||||
respond = () => json({
|
||||
revision: 7,
|
||||
session: session([]),
|
||||
item: serverItem("q1", "with file", {
|
||||
attachments: [{ id: "att-1", filename: "note.txt", mimeType: "text/plain", size: 2, source: "local", dataUrl: "data:text/plain;base64,aGk=" }],
|
||||
}),
|
||||
})
|
||||
const [taken] = await useMessageQueueStore.getState().takeForSend(target, "q1")
|
||||
|
||||
expect(calls[0]?.path).toBe("/api/message-queue/sessions/session-1/items/q1/take")
|
||||
expect(calls[0]?.method).toBe("POST")
|
||||
expect(taken?.content).toBe("with file")
|
||||
expect(taken?.attachments?.[0]?.dataUrl).toBe("data:text/plain;base64,aGk=")
|
||||
expect(taken?.attachments?.[0]?.file.size).toBe(2)
|
||||
expect(useMessageQueueStore.getState().queuedMessages[key]).toBe(undefined)
|
||||
})
|
||||
|
||||
test("takeForSend without an id takes everything the server is not already sending", async () => {
|
||||
respond = () => json({ revision: 8, session: session([serverItem("q1", "in flight")], "q1"), items: [serverItem("q2", "second")] })
|
||||
const taken = await useMessageQueueStore.getState().takeForSend(target)
|
||||
|
||||
expect(calls[0]?.path).toBe("/api/message-queue/sessions/session-1/take")
|
||||
expect(calls[0]?.method).toBe("POST")
|
||||
expect(taken.map((m) => m.content)).toEqual(["second"])
|
||||
expect(useMessageQueueStore.getState().queuedMessages[key]?.map((m) => m.id)).toEqual(["q1"])
|
||||
})
|
||||
|
||||
test("broadcasts update the projection but never move it backwards", () => {
|
||||
applyMessageQueueUpdatedEvent(updated(4, session([serverItem("q1", "newer")])), "runtime-a")
|
||||
expect(useMessageQueueStore.getState().queuedMessages[key]?.map((m) => m.content)).toEqual(["newer"])
|
||||
|
||||
applyMessageQueueUpdatedEvent(updated(2, session([serverItem("q0", "older")])), "runtime-a")
|
||||
expect(useMessageQueueStore.getState().queuedMessages[key]?.map((m) => m.content)).toEqual(["newer"])
|
||||
|
||||
applyMessageQueueUpdatedEvent(updated(9, session([serverItem("q1", "newer")])), "runtime-b")
|
||||
expect(useMessageQueueStore.getState().queuedMessages[key]?.map((m) => m.content)).toEqual(["newer"])
|
||||
})
|
||||
|
||||
test("removeFromQueue and clearQueue update locally and tell the server", async () => {
|
||||
useMessageQueueStore.setState({ queuedMessages: { [key]: [{ id: "q1", content: "a", createdAt: 1 }, { id: "q2", content: "b", createdAt: 2 }] } })
|
||||
respond = () => json({ revision: 10, session: session([serverItem("q2", "b")]) })
|
||||
useMessageQueueStore.getState().removeFromQueue(target, "q1")
|
||||
expect(useMessageQueueStore.getState().queuedMessages[key]?.map((m) => m.id)).toEqual(["q2"])
|
||||
await Promise.resolve()
|
||||
await Promise.resolve()
|
||||
expect(calls[0]).toEqual({ method: "DELETE", path: "/api/message-queue/sessions/session-1/items/q1", body: undefined })
|
||||
|
||||
respond = () => json({ revision: 11, session: session([]) })
|
||||
useMessageQueueStore.getState().clearQueue(target)
|
||||
expect(useMessageQueueStore.getState().queuedMessages[key]).toBe(undefined)
|
||||
await Promise.resolve()
|
||||
expect(calls[1]).toEqual({ method: "DELETE", path: "/api/message-queue/sessions/session-1", body: undefined })
|
||||
})
|
||||
|
||||
test("reorderQueue sends the complete new order", async () => {
|
||||
useMessageQueueStore.setState({ queuedMessages: { [key]: [{ id: "q1", content: "a", createdAt: 1 }, { id: "q2", content: "b", createdAt: 2 }] } })
|
||||
respond = () => json({ revision: 12, session: session([serverItem("q2", "b"), serverItem("q1", "a")]) })
|
||||
useMessageQueueStore.getState().reorderQueue(target, "q2", "q1")
|
||||
await Promise.resolve()
|
||||
expect(calls[0]).toEqual({ method: "PUT", path: "/api/message-queue/sessions/session-1/order", body: { itemIds: ["q2", "q1"] } })
|
||||
})
|
||||
})
|
||||
@@ -1,11 +1,17 @@
|
||||
import { beforeEach, describe, expect, test } from "bun:test"
|
||||
import {
|
||||
import { beforeEach, describe, expect, mock, test } from "bun:test"
|
||||
|
||||
// The local queue is the VS Code behavior; every other runtime hands the
|
||||
// queue to the server (see messageQueueStore.server.test.ts).
|
||||
const desktop = await import("@/lib/desktop")
|
||||
mock.module("@/lib/desktop", () => ({ ...desktop, isVSCodeRuntime: () => true }))
|
||||
|
||||
const {
|
||||
createMessageQueueTarget,
|
||||
getMessageQueueKey,
|
||||
migrateMessageQueueState,
|
||||
parseMessageQueueKey,
|
||||
useMessageQueueStore,
|
||||
} from "./messageQueueStore"
|
||||
} = await import("./messageQueueStore")
|
||||
|
||||
beforeEach(() => {
|
||||
useMessageQueueStore.setState({ queuedMessages: {}, quarantinedLegacyMessages: {}, sendingIds: {} })
|
||||
|
||||
@@ -1,9 +1,13 @@
|
||||
import { create } from 'zustand';
|
||||
import { devtools, persist } from 'zustand/middleware';
|
||||
import { z } from 'zod';
|
||||
import type { Event } from '@opencode-ai/sdk/v2';
|
||||
import { createDeferredSafeJSONStorage } from './utils/safeStorage';
|
||||
import type { AttachedFile } from './types/sessionTypes';
|
||||
import { updateDesktopSettings } from '@/lib/persistence';
|
||||
import { getRuntimeKey } from '@/lib/runtime-switch';
|
||||
import { isVSCodeRuntime } from '@/lib/desktop';
|
||||
import { runtimeFetch } from '@/lib/runtime-fetch';
|
||||
import { normalizePath } from '@/lib/pathNormalization';
|
||||
|
||||
export type FollowUpBehavior = 'steer' | 'queue';
|
||||
@@ -40,18 +44,38 @@ export const normalizeFollowUpBehavior = (
|
||||
return DEFAULT_FOLLOW_UP_BEHAVIOR;
|
||||
};
|
||||
|
||||
/**
|
||||
* Who delivers the queue. Web, desktop, and mobile talk to an OpenChamber
|
||||
* server that owns the queue and sends it whether or not any UI is open. VS
|
||||
* Code has no server of its own, so the extension UI keeps the local queue
|
||||
* and the foreground auto-send hook.
|
||||
*/
|
||||
export const isServerOwnedMessageQueue = (): boolean => !isVSCodeRuntime();
|
||||
|
||||
export interface QueuedMessageSendConfig {
|
||||
providerID: string;
|
||||
modelID: string;
|
||||
agent?: string;
|
||||
variant?: string;
|
||||
}
|
||||
|
||||
export interface QueuedMessage {
|
||||
id: string;
|
||||
content: string;
|
||||
attachments?: AttachedFile[];
|
||||
createdAt: number;
|
||||
/** Send config captured at queue time — used as-is when auto-sending */
|
||||
sendConfig?: {
|
||||
providerID: string;
|
||||
modelID: string;
|
||||
agent?: string;
|
||||
variant?: string;
|
||||
};
|
||||
sendConfig?: QueuedMessageSendConfig;
|
||||
}
|
||||
|
||||
interface QueuedMessageInput {
|
||||
content: string;
|
||||
attachments?: AttachedFile[];
|
||||
sendConfig?: QueuedMessageSendConfig;
|
||||
/** Text to deliver once the agent mention is stripped; defaults to `content`. */
|
||||
text?: string;
|
||||
/** Agent mentioned at the start of `content`, delivered as an agent part. */
|
||||
agentMention?: string;
|
||||
}
|
||||
|
||||
export type MessageQueueTarget = {
|
||||
@@ -81,6 +105,177 @@ export const parseMessageQueueKey = (key: string): MessageQueueTarget | null =>
|
||||
return createMessageQueueTarget(sessionParts.join('\n'), directory, runtimeKey);
|
||||
};
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Server contract (packages/web/server/lib/message-queue)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const serverSendConfigSchema = z.object({
|
||||
providerID: z.string().min(1),
|
||||
modelID: z.string().min(1),
|
||||
agent: z.string().optional(),
|
||||
variant: z.string().optional(),
|
||||
});
|
||||
|
||||
const serverAttachmentSchema = z.object({
|
||||
id: z.string(),
|
||||
filename: z.string(),
|
||||
mimeType: z.string(),
|
||||
size: z.number(),
|
||||
source: z.enum(['local', 'server', 'vscode']),
|
||||
serverPath: z.string().optional(),
|
||||
/** Present only on a taken item; broadcasts and snapshots omit payloads. */
|
||||
dataUrl: z.string().optional(),
|
||||
});
|
||||
|
||||
const serverItemSchema = z.object({
|
||||
id: z.string().min(1),
|
||||
createdAt: z.number(),
|
||||
content: z.string(),
|
||||
agentMention: z.string().optional(),
|
||||
attachments: z.array(serverAttachmentSchema),
|
||||
sendConfig: serverSendConfigSchema,
|
||||
});
|
||||
|
||||
const serverSessionSchema = z.object({
|
||||
sessionId: z.string().min(1),
|
||||
directory: z.string(),
|
||||
items: z.array(serverItemSchema),
|
||||
sendingId: z.string().nullable(),
|
||||
});
|
||||
|
||||
const serverSnapshotSchema = z.object({
|
||||
revision: z.number(),
|
||||
sessions: z.array(serverSessionSchema),
|
||||
});
|
||||
|
||||
const serverSessionResponseSchema = z.object({
|
||||
revision: z.number(),
|
||||
session: serverSessionSchema,
|
||||
});
|
||||
|
||||
const serverTakeResponseSchema = serverSessionResponseSchema.extend({ item: serverItemSchema });
|
||||
const serverTakeAllResponseSchema = serverSessionResponseSchema.extend({ items: z.array(serverItemSchema) });
|
||||
|
||||
type ServerQueueSession = z.infer<typeof serverSessionSchema>;
|
||||
type ServerQueueItem = z.infer<typeof serverItemSchema>;
|
||||
type ServerQueueAttachment = z.infer<typeof serverAttachmentSchema>;
|
||||
|
||||
const decodeDataUrl = (dataUrl: string): ArrayBuffer | null => {
|
||||
const commaIndex = dataUrl.indexOf(',');
|
||||
if (!dataUrl.startsWith('data:') || commaIndex === -1) return null;
|
||||
const meta = dataUrl.slice(5, commaIndex);
|
||||
const payload = dataUrl.slice(commaIndex + 1);
|
||||
try {
|
||||
if (meta.endsWith(';base64')) {
|
||||
const binary = atob(payload);
|
||||
const buffer = new ArrayBuffer(binary.length);
|
||||
const bytes = new Uint8Array(buffer);
|
||||
for (let index = 0; index < binary.length; index += 1) bytes[index] = binary.charCodeAt(index);
|
||||
return buffer;
|
||||
}
|
||||
const encoded = new TextEncoder().encode(decodeURIComponent(payload));
|
||||
const buffer = new ArrayBuffer(encoded.byteLength);
|
||||
new Uint8Array(buffer).set(encoded);
|
||||
return buffer;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
};
|
||||
|
||||
/** A taken item carries its payload; a projection item has an empty file. */
|
||||
const toAttachedFile = (attachment: ServerQueueAttachment): AttachedFile => {
|
||||
const dataUrl = attachment.dataUrl ?? '';
|
||||
const bytes = dataUrl ? decodeDataUrl(dataUrl) : null;
|
||||
const file: AttachedFile = {
|
||||
id: attachment.id,
|
||||
file: new File(bytes ? [bytes] : [], attachment.filename, { type: attachment.mimeType }),
|
||||
dataUrl,
|
||||
mimeType: attachment.mimeType,
|
||||
filename: attachment.filename,
|
||||
size: attachment.size,
|
||||
source: attachment.source,
|
||||
};
|
||||
if (attachment.serverPath) file.serverPath = attachment.serverPath;
|
||||
return file;
|
||||
};
|
||||
|
||||
const toQueuedMessage = (item: ServerQueueItem): QueuedMessage => ({
|
||||
id: item.id,
|
||||
content: item.content,
|
||||
createdAt: item.createdAt,
|
||||
attachments: item.attachments.length > 0 ? item.attachments.map(toAttachedFile) : undefined,
|
||||
sendConfig: { ...item.sendConfig },
|
||||
});
|
||||
|
||||
type ServerQueueAttachmentInput = Omit<ServerQueueAttachment, 'dataUrl'> & { dataUrl: string };
|
||||
|
||||
type ServerQueueItemInput = {
|
||||
content: string;
|
||||
text: string;
|
||||
agentMention?: string;
|
||||
attachments: ServerQueueAttachmentInput[];
|
||||
sendConfig: QueuedMessageSendConfig;
|
||||
};
|
||||
|
||||
type ServerQueueRequestBody =
|
||||
| { directory: string; item: ServerQueueItemInput }
|
||||
| { itemIds: string[] }
|
||||
| { held: boolean };
|
||||
|
||||
const toServerAttachment = (attachment: AttachedFile): ServerQueueAttachmentInput => {
|
||||
const input: ServerQueueAttachmentInput = {
|
||||
id: attachment.id,
|
||||
filename: attachment.filename,
|
||||
mimeType: attachment.mimeType,
|
||||
size: attachment.size,
|
||||
source: attachment.source,
|
||||
dataUrl: attachment.dataUrl,
|
||||
};
|
||||
if (attachment.serverPath) input.serverPath = attachment.serverPath;
|
||||
return input;
|
||||
};
|
||||
|
||||
const toServerItemInput = (message: QueuedMessageInput, sendConfig: QueuedMessageSendConfig): ServerQueueItemInput => {
|
||||
const item: ServerQueueItemInput = {
|
||||
content: message.content,
|
||||
text: message.text ?? message.content,
|
||||
attachments: (message.attachments ?? []).filter((file) => Boolean(file.dataUrl)).map(toServerAttachment),
|
||||
sendConfig,
|
||||
};
|
||||
if (message.agentMention) item.agentMention = message.agentMention;
|
||||
return item;
|
||||
};
|
||||
|
||||
const requestJson = async <T,>(schema: z.ZodType<T>, path: string, init?: RequestInit): Promise<T> => {
|
||||
const response = await runtimeFetch(path, init);
|
||||
if (!response.ok) {
|
||||
const error: Error & { status?: number } = new Error(`Message queue request failed (${response.status})`);
|
||||
error.status = response.status;
|
||||
throw error;
|
||||
}
|
||||
const parsed = schema.safeParse(await response.json());
|
||||
if (!parsed.success) throw new Error('Invalid message queue response');
|
||||
return parsed.data;
|
||||
};
|
||||
|
||||
const jsonInit = (method: string, body?: ServerQueueRequestBody): RequestInit => {
|
||||
if (body === undefined) return { method };
|
||||
return { method, headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body) };
|
||||
};
|
||||
|
||||
const sessionPath = (sessionId: string) => `/api/message-queue/sessions/${encodeURIComponent(sessionId)}`;
|
||||
|
||||
/**
|
||||
* Runtime keys whose queue the server owns, established by a successful
|
||||
* hydration. Their entries are a projection and must not be persisted: a
|
||||
* stale local copy would resurrect messages the server already delivered.
|
||||
*/
|
||||
const serverOwnedRuntimeKeys = new Set<string>();
|
||||
|
||||
/** Server revision last applied per queue key; older snapshots are ignored. */
|
||||
const appliedRevisions = new Map<string, number>();
|
||||
let hydrationGeneration = 0;
|
||||
|
||||
interface MessageQueueState {
|
||||
queuedMessages: Record<string, QueuedMessage[]>; // runtime + directory + session → queue
|
||||
quarantinedLegacyMessages: Record<string, QueuedMessage[]>;
|
||||
@@ -95,23 +290,39 @@ interface MessageQueueState {
|
||||
* twice. Dispatchers must skip entries listed here.
|
||||
*
|
||||
* Never persisted: a restart has no in-flight sends, and a stale flag would
|
||||
* strand a queued message permanently.
|
||||
* strand a queued message permanently. With a server-owned queue this
|
||||
* mirrors the server's in-flight item.
|
||||
*/
|
||||
sendingIds: Record<string, string[]>;
|
||||
}
|
||||
|
||||
interface MessageQueueActions {
|
||||
addToQueue: (target: MessageQueueTarget, message: Omit<QueuedMessage, 'id' | 'createdAt'>) => void;
|
||||
addToQueue: (target: MessageQueueTarget, message: QueuedMessageInput) => Promise<void>;
|
||||
removeFromQueue: (target: MessageQueueTarget, messageId: string) => void;
|
||||
reorderQueue: (target: MessageQueueTarget, fromId: string, toId: string) => void;
|
||||
popToInput: (target: MessageQueueTarget, messageId: string) => QueuedMessage | null;
|
||||
/** Removes the message and returns it in full, attachments included. */
|
||||
popToInput: (target: MessageQueueTarget, messageId: string) => Promise<QueuedMessage | null>;
|
||||
/**
|
||||
* Removes what the composer is about to send itself — one message or every
|
||||
* message not already being delivered — and returns it in full.
|
||||
*/
|
||||
takeForSend: (target: MessageQueueTarget, messageId?: string) => Promise<QueuedMessage[]>;
|
||||
clearQueue: (target: MessageQueueTarget) => void;
|
||||
/** Drops the local projection only (the session is gone); never a server call. */
|
||||
forgetQueue: (target: MessageQueueTarget) => void;
|
||||
clearAllQueues: () => void;
|
||||
markSending: (target: MessageQueueTarget, messageId: string) => void;
|
||||
clearSending: (target: MessageQueueTarget, messageId: string) => void;
|
||||
getSendableQueue: (target: MessageQueueTarget) => QueuedMessage[];
|
||||
setFollowUpBehavior: (behavior: FollowUpBehavior) => void;
|
||||
getQueueForTarget: (target: MessageQueueTarget) => QueuedMessage[];
|
||||
/** Server-owned queue: load the authoritative queue for the active runtime. */
|
||||
hydrate: () => Promise<void>;
|
||||
/** Server-owned queue: apply one session's authoritative state (broadcast or response). */
|
||||
applyServerSession: (session: ServerQueueSession, revision: number, expectedRuntimeKey: string) => void;
|
||||
/** Server-owned queue: tell the server to hold or release a session's delivery. */
|
||||
setServerHold: (sessionId: string, held: boolean) => Promise<void>;
|
||||
resetForRuntimeSwitch: (previousRuntimeKey: string | null | undefined) => void;
|
||||
}
|
||||
|
||||
type MessageQueueStore = MessageQueueState & MessageQueueActions;
|
||||
@@ -136,190 +347,362 @@ export const migrateMessageQueueState = (persistedState: unknown, version: numbe
|
||||
};
|
||||
};
|
||||
|
||||
const withoutKey = <T,>(record: Record<string, T>, key: string): Record<string, T> => {
|
||||
const { [key]: _removed, ...rest } = record;
|
||||
void _removed;
|
||||
return rest;
|
||||
};
|
||||
|
||||
const removeMessageLocally = (
|
||||
state: Pick<MessageQueueState, 'queuedMessages'>,
|
||||
key: string,
|
||||
messageId: string,
|
||||
): Pick<MessageQueueState, 'queuedMessages'> => {
|
||||
const newQueue = (state.queuedMessages[key] ?? []).filter((m) => m.id !== messageId);
|
||||
if (newQueue.length === 0) return { queuedMessages: withoutKey(state.queuedMessages, key) };
|
||||
return { queuedMessages: { ...state.queuedMessages, [key]: newQueue } };
|
||||
};
|
||||
|
||||
export const useMessageQueueStore = create<MessageQueueStore>()(
|
||||
devtools(
|
||||
persist(
|
||||
(set, get) => ({
|
||||
queuedMessages: {},
|
||||
quarantinedLegacyMessages: {},
|
||||
followUpBehavior: DEFAULT_FOLLOW_UP_BEHAVIOR,
|
||||
sendingIds: {},
|
||||
|
||||
addToQueue: (target, message) => {
|
||||
(set, get) => {
|
||||
const applyServerSession = (session: ServerQueueSession, revision: number, expectedRuntimeKey: string) => {
|
||||
if (expectedRuntimeKey !== getRuntimeKey()) return;
|
||||
const target = createMessageQueueTarget(session.sessionId, session.directory, expectedRuntimeKey);
|
||||
if (!target) return;
|
||||
const key = getMessageQueueKey(target);
|
||||
const id = `queued-${Date.now()}-${Math.random().toString(36).substring(2, 9)}`;
|
||||
const queuedMessage: QueuedMessage = {
|
||||
id,
|
||||
content: message.content,
|
||||
attachments: message.attachments,
|
||||
createdAt: Date.now(),
|
||||
sendConfig: message.sendConfig,
|
||||
};
|
||||
|
||||
if ((appliedRevisions.get(key) ?? -1) > revision) return;
|
||||
appliedRevisions.set(key, revision);
|
||||
set((state) => {
|
||||
const currentQueue = state.queuedMessages[key] ?? [];
|
||||
const queuedMessages = {
|
||||
...state.queuedMessages,
|
||||
[key]: [...currentQueue, queuedMessage].slice(-MAX_MESSAGES_PER_QUEUE),
|
||||
};
|
||||
const keys = Object.keys(queuedMessages);
|
||||
if (keys.length > MAX_QUEUE_TARGETS) {
|
||||
keys.sort((left, right) => (
|
||||
(queuedMessages[left]?.[0]?.createdAt ?? 0) - (queuedMessages[right]?.[0]?.createdAt ?? 0)
|
||||
));
|
||||
for (const staleKey of keys.slice(0, keys.length - MAX_QUEUE_TARGETS)) delete queuedMessages[staleKey];
|
||||
}
|
||||
return {
|
||||
queuedMessages,
|
||||
};
|
||||
const queue = session.items.map(toQueuedMessage);
|
||||
const queuedMessages = queue.length > 0
|
||||
? { ...state.queuedMessages, [key]: queue }
|
||||
: withoutKey(state.queuedMessages, key);
|
||||
const sendingIds = session.sendingId
|
||||
? { ...state.sendingIds, [key]: [session.sendingId] }
|
||||
: withoutKey(state.sendingIds, key);
|
||||
return { queuedMessages, sendingIds };
|
||||
});
|
||||
},
|
||||
};
|
||||
|
||||
removeFromQueue: (target, messageId) => {
|
||||
const key = getMessageQueueKey(target);
|
||||
set((state) => {
|
||||
const currentQueue = state.queuedMessages[key] ?? [];
|
||||
const newQueue = currentQueue.filter((m) => m.id !== messageId);
|
||||
|
||||
if (newQueue.length === 0) {
|
||||
const { [key]: _removed, ...rest } = state.queuedMessages;
|
||||
void _removed;
|
||||
return { queuedMessages: rest };
|
||||
}
|
||||
|
||||
return {
|
||||
queuedMessages: {
|
||||
/** Server state wins; a failed round-trip re-reads it instead of guessing. */
|
||||
const refreshSession = async (target: MessageQueueTarget) => {
|
||||
try {
|
||||
const snapshot = await requestJson(serverSnapshotSchema, '/api/message-queue');
|
||||
const session = snapshot.sessions.find((entry) => entry.sessionId === target.sessionId)
|
||||
?? { sessionId: target.sessionId, directory: target.directory, items: [], sendingId: null };
|
||||
applyServerSession(session, snapshot.revision, target.runtimeKey);
|
||||
} catch {
|
||||
// Offline: keep the optimistic projection; the next broadcast or hydration corrects it.
|
||||
}
|
||||
};
|
||||
|
||||
const serverMutation = async (
|
||||
target: MessageQueueTarget,
|
||||
path: string,
|
||||
init: RequestInit,
|
||||
) => {
|
||||
try {
|
||||
const result = await requestJson(serverSessionResponseSchema, path, init);
|
||||
applyServerSession(result.session, result.revision, target.runtimeKey);
|
||||
} catch (error) {
|
||||
console.warn('[queue] server update failed:', error);
|
||||
await refreshSession(target);
|
||||
}
|
||||
};
|
||||
|
||||
return {
|
||||
queuedMessages: {},
|
||||
quarantinedLegacyMessages: {},
|
||||
followUpBehavior: DEFAULT_FOLLOW_UP_BEHAVIOR,
|
||||
sendingIds: {},
|
||||
|
||||
addToQueue: async (target, message) => {
|
||||
const key = getMessageQueueKey(target);
|
||||
const id = `queued-${Date.now()}-${Math.random().toString(36).substring(2, 9)}`;
|
||||
const queuedMessage: QueuedMessage = {
|
||||
id,
|
||||
content: message.content,
|
||||
attachments: message.attachments,
|
||||
createdAt: Date.now(),
|
||||
sendConfig: message.sendConfig,
|
||||
};
|
||||
|
||||
set((state) => {
|
||||
const currentQueue = state.queuedMessages[key] ?? [];
|
||||
const queuedMessages = {
|
||||
...state.queuedMessages,
|
||||
[key]: newQueue,
|
||||
},
|
||||
};
|
||||
});
|
||||
},
|
||||
[key]: [...currentQueue, queuedMessage].slice(-MAX_MESSAGES_PER_QUEUE),
|
||||
};
|
||||
const keys = Object.keys(queuedMessages);
|
||||
if (keys.length > MAX_QUEUE_TARGETS) {
|
||||
keys.sort((left, right) => (
|
||||
(queuedMessages[left]?.[0]?.createdAt ?? 0) - (queuedMessages[right]?.[0]?.createdAt ?? 0)
|
||||
));
|
||||
for (const staleKey of keys.slice(0, keys.length - MAX_QUEUE_TARGETS)) delete queuedMessages[staleKey];
|
||||
}
|
||||
return {
|
||||
queuedMessages,
|
||||
};
|
||||
});
|
||||
|
||||
reorderQueue: (target, fromId, toId) => {
|
||||
if (fromId === toId) return;
|
||||
const key = getMessageQueueKey(target);
|
||||
set((state) => {
|
||||
const currentQueue = state.queuedMessages[key];
|
||||
if (!currentQueue) return state;
|
||||
if (!isServerOwnedMessageQueue()) return;
|
||||
if (!message.sendConfig) {
|
||||
set((state) => removeMessageLocally(state, key, id));
|
||||
throw new Error('A queued message needs a provider and model to be delivered later.');
|
||||
}
|
||||
try {
|
||||
const result = await requestJson(serverSessionResponseSchema, `${sessionPath(target.sessionId)}/items`, jsonInit('POST', {
|
||||
directory: target.directory,
|
||||
item: toServerItemInput(message, message.sendConfig),
|
||||
}));
|
||||
// The optimistic entry is replaced by the server's copy of the queue.
|
||||
set((state) => removeMessageLocally(state, key, id));
|
||||
applyServerSession(result.session, result.revision, target.runtimeKey);
|
||||
} catch (error) {
|
||||
set((state) => removeMessageLocally(state, key, id));
|
||||
throw error;
|
||||
}
|
||||
},
|
||||
|
||||
removeFromQueue: (target, messageId) => {
|
||||
const key = getMessageQueueKey(target);
|
||||
set((state) => removeMessageLocally(state, key, messageId));
|
||||
if (isServerOwnedMessageQueue()) {
|
||||
void serverMutation(target, `${sessionPath(target.sessionId)}/items/${encodeURIComponent(messageId)}`, jsonInit('DELETE'));
|
||||
}
|
||||
},
|
||||
|
||||
reorderQueue: (target, fromId, toId) => {
|
||||
if (fromId === toId) return;
|
||||
const key = getMessageQueueKey(target);
|
||||
const currentQueue = get().queuedMessages[key];
|
||||
if (!currentQueue) return;
|
||||
const fromIndex = currentQueue.findIndex((m) => m.id === fromId);
|
||||
const toIndex = currentQueue.findIndex((m) => m.id === toId);
|
||||
if (fromIndex === -1 || toIndex === -1) return state;
|
||||
if (fromIndex === -1 || toIndex === -1) return;
|
||||
|
||||
const newQueue = currentQueue.slice();
|
||||
const [moved] = newQueue.splice(fromIndex, 1);
|
||||
newQueue.splice(toIndex, 0, moved);
|
||||
|
||||
return {
|
||||
set((state) => ({
|
||||
queuedMessages: {
|
||||
...state.queuedMessages,
|
||||
[key]: newQueue,
|
||||
},
|
||||
};
|
||||
});
|
||||
},
|
||||
|
||||
popToInput: (target, messageId) => {
|
||||
const key = getMessageQueueKey(target);
|
||||
const state = get();
|
||||
const currentQueue = state.queuedMessages[key] ?? [];
|
||||
const message = currentQueue.find((m) => m.id === messageId);
|
||||
|
||||
if (!message) {
|
||||
return null;
|
||||
}
|
||||
|
||||
// Remove from queue
|
||||
set((prevState) => {
|
||||
const queue = prevState.queuedMessages[key] ?? [];
|
||||
const newQueue = queue.filter((m) => m.id !== messageId);
|
||||
|
||||
if (newQueue.length === 0) {
|
||||
const { [key]: _removed, ...rest } = prevState.queuedMessages;
|
||||
void _removed;
|
||||
return { queuedMessages: rest };
|
||||
}));
|
||||
if (isServerOwnedMessageQueue()) {
|
||||
const itemIds = newQueue.map((message) => message.id);
|
||||
void serverMutation(target, `${sessionPath(target.sessionId)}/order`, jsonInit('PUT', { itemIds }));
|
||||
}
|
||||
|
||||
return {
|
||||
queuedMessages: {
|
||||
...prevState.queuedMessages,
|
||||
[key]: newQueue,
|
||||
},
|
||||
};
|
||||
});
|
||||
},
|
||||
|
||||
return message;
|
||||
},
|
||||
popToInput: async (target, messageId) => {
|
||||
const [message] = await get().takeForSend(target, messageId);
|
||||
return message ?? null;
|
||||
},
|
||||
|
||||
clearQueue: (target) => {
|
||||
const key = getMessageQueueKey(target);
|
||||
set((state) => {
|
||||
// Clearing drops what is still queued, never a message
|
||||
// already handed to the server: that send will resolve
|
||||
// and must find its entry to remove or restore.
|
||||
takeForSend: async (target, messageId) => {
|
||||
const key = getMessageQueueKey(target);
|
||||
if (isServerOwnedMessageQueue()) {
|
||||
if (messageId) {
|
||||
const result = await requestJson(
|
||||
serverTakeResponseSchema,
|
||||
`${sessionPath(target.sessionId)}/items/${encodeURIComponent(messageId)}/take`,
|
||||
jsonInit('POST'),
|
||||
);
|
||||
applyServerSession(result.session, result.revision, target.runtimeKey);
|
||||
return [toQueuedMessage(result.item)];
|
||||
}
|
||||
const result = await requestJson(serverTakeAllResponseSchema, `${sessionPath(target.sessionId)}/take`, jsonInit('POST'));
|
||||
applyServerSession(result.session, result.revision, target.runtimeKey);
|
||||
return result.items.map(toQueuedMessage);
|
||||
}
|
||||
|
||||
const state = get();
|
||||
const sending = state.sendingIds[key] ?? [];
|
||||
const retained = (state.queuedMessages[key] ?? []).filter((m) => sending.includes(m.id));
|
||||
if (retained.length > 0) {
|
||||
return { queuedMessages: { ...state.queuedMessages, [key]: retained } };
|
||||
const taken = (state.queuedMessages[key] ?? []).filter((message) => (
|
||||
(messageId ? message.id === messageId : true) && !sending.includes(message.id)
|
||||
));
|
||||
if (taken.length === 0) return [];
|
||||
const takenIds = new Set(taken.map((message) => message.id));
|
||||
set((prevState) => {
|
||||
const remaining = (prevState.queuedMessages[key] ?? []).filter((message) => !takenIds.has(message.id));
|
||||
if (remaining.length === 0) return { queuedMessages: withoutKey(prevState.queuedMessages, key) };
|
||||
return { queuedMessages: { ...prevState.queuedMessages, [key]: remaining } };
|
||||
});
|
||||
return taken;
|
||||
},
|
||||
|
||||
clearQueue: (target) => {
|
||||
const key = getMessageQueueKey(target);
|
||||
set((state) => {
|
||||
// Clearing drops what is still queued, never a message
|
||||
// already handed to the server: that send will resolve
|
||||
// and must find its entry to remove or restore.
|
||||
const sending = state.sendingIds[key] ?? [];
|
||||
const retained = (state.queuedMessages[key] ?? []).filter((m) => sending.includes(m.id));
|
||||
if (retained.length > 0) {
|
||||
return { queuedMessages: { ...state.queuedMessages, [key]: retained } };
|
||||
}
|
||||
return { queuedMessages: withoutKey(state.queuedMessages, key) };
|
||||
});
|
||||
if (isServerOwnedMessageQueue()) {
|
||||
void serverMutation(target, sessionPath(target.sessionId), jsonInit('DELETE'));
|
||||
}
|
||||
const { [key]: _removed, ...rest } = state.queuedMessages;
|
||||
void _removed;
|
||||
return { queuedMessages: rest };
|
||||
});
|
||||
},
|
||||
},
|
||||
|
||||
clearAllQueues: () => {
|
||||
set({ queuedMessages: {}, sendingIds: {} });
|
||||
},
|
||||
forgetQueue: (target) => {
|
||||
const key = getMessageQueueKey(target);
|
||||
appliedRevisions.delete(key);
|
||||
set((state) => ({
|
||||
queuedMessages: withoutKey(state.queuedMessages, key),
|
||||
sendingIds: withoutKey(state.sendingIds, key),
|
||||
}));
|
||||
},
|
||||
|
||||
markSending: (target, messageId) => {
|
||||
const key = getMessageQueueKey(target);
|
||||
set((state) => {
|
||||
const current = state.sendingIds[key] ?? [];
|
||||
if (current.includes(messageId)) return state;
|
||||
return { sendingIds: { ...state.sendingIds, [key]: [...current, messageId] } };
|
||||
});
|
||||
},
|
||||
clearAllQueues: () => {
|
||||
set({ queuedMessages: {}, sendingIds: {} });
|
||||
},
|
||||
|
||||
clearSending: (target, messageId) => {
|
||||
const key = getMessageQueueKey(target);
|
||||
set((state) => {
|
||||
const current = state.sendingIds[key];
|
||||
if (!current || !current.includes(messageId)) return state;
|
||||
const next = current.filter((id) => id !== messageId);
|
||||
if (next.length === 0) {
|
||||
const { [key]: _removed, ...rest } = state.sendingIds;
|
||||
void _removed;
|
||||
return { sendingIds: rest };
|
||||
markSending: (target, messageId) => {
|
||||
const key = getMessageQueueKey(target);
|
||||
set((state) => {
|
||||
const current = state.sendingIds[key] ?? [];
|
||||
if (current.includes(messageId)) return state;
|
||||
return { sendingIds: { ...state.sendingIds, [key]: [...current, messageId] } };
|
||||
});
|
||||
},
|
||||
|
||||
clearSending: (target, messageId) => {
|
||||
const key = getMessageQueueKey(target);
|
||||
set((state) => {
|
||||
const current = state.sendingIds[key];
|
||||
if (!current || !current.includes(messageId)) return state;
|
||||
const next = current.filter((id) => id !== messageId);
|
||||
if (next.length === 0) return { sendingIds: withoutKey(state.sendingIds, key) };
|
||||
return { sendingIds: { ...state.sendingIds, [key]: next } };
|
||||
});
|
||||
},
|
||||
|
||||
getSendableQueue: (target) => {
|
||||
const key = getMessageQueueKey(target);
|
||||
const state = get();
|
||||
const queue = state.queuedMessages[key] ?? [];
|
||||
const sending = state.sendingIds[key];
|
||||
if (!sending || sending.length === 0) return queue;
|
||||
return queue.filter((message) => !sending.includes(message.id));
|
||||
},
|
||||
|
||||
setFollowUpBehavior: (behavior) => {
|
||||
set({ followUpBehavior: behavior });
|
||||
void updateDesktopSettings({ followUpBehavior: behavior });
|
||||
},
|
||||
|
||||
getQueueForTarget: (target) => {
|
||||
return get().queuedMessages[getMessageQueueKey(target)] ?? [];
|
||||
},
|
||||
|
||||
hydrate: async () => {
|
||||
if (!isServerOwnedMessageQueue()) return;
|
||||
const runtimeKey = getRuntimeKey();
|
||||
const generation = ++hydrationGeneration;
|
||||
const isCurrent = () => generation === hydrationGeneration && runtimeKey === getRuntimeKey();
|
||||
|
||||
// Messages queued by an older build live in this browser only.
|
||||
// Hand them to the server once so they are still delivered;
|
||||
// whatever cannot be uploaded is superseded by the server's queue.
|
||||
const legacyEntries = Object.entries(get().queuedMessages)
|
||||
.map(([key, queue]) => ({ target: parseMessageQueueKey(key), queue }))
|
||||
.filter((entry): entry is { target: MessageQueueTarget; queue: QueuedMessage[] } => (
|
||||
entry.target !== null && entry.target.runtimeKey === runtimeKey && !serverOwnedRuntimeKeys.has(runtimeKey)
|
||||
));
|
||||
for (const { target, queue } of legacyEntries) {
|
||||
for (const message of queue) {
|
||||
if (!message.sendConfig) continue;
|
||||
try {
|
||||
await requestJson(serverSessionResponseSchema, `${sessionPath(target.sessionId)}/items`, jsonInit('POST', {
|
||||
directory: target.directory,
|
||||
item: toServerItemInput(message, message.sendConfig),
|
||||
}));
|
||||
} catch (error) {
|
||||
console.warn('[queue] failed to migrate a locally queued message to the server:', error);
|
||||
}
|
||||
if (!isCurrent()) return;
|
||||
}
|
||||
}
|
||||
return { sendingIds: { ...state.sendingIds, [key]: next } };
|
||||
});
|
||||
},
|
||||
|
||||
getSendableQueue: (target) => {
|
||||
const key = getMessageQueueKey(target);
|
||||
const state = get();
|
||||
const queue = state.queuedMessages[key] ?? [];
|
||||
const sending = state.sendingIds[key];
|
||||
if (!sending || sending.length === 0) return queue;
|
||||
return queue.filter((message) => !sending.includes(message.id));
|
||||
},
|
||||
const snapshot = await requestJson(serverSnapshotSchema, '/api/message-queue');
|
||||
if (!isCurrent()) return;
|
||||
serverOwnedRuntimeKeys.add(runtimeKey);
|
||||
set((state) => {
|
||||
const queuedMessages: Record<string, QueuedMessage[]> = {};
|
||||
const sendingIds: Record<string, string[]> = {};
|
||||
for (const [key, queue] of Object.entries(state.queuedMessages)) {
|
||||
if (parseMessageQueueKey(key)?.runtimeKey !== runtimeKey) queuedMessages[key] = queue;
|
||||
}
|
||||
for (const [key, ids] of Object.entries(state.sendingIds)) {
|
||||
if (parseMessageQueueKey(key)?.runtimeKey !== runtimeKey) sendingIds[key] = ids;
|
||||
}
|
||||
for (const session of snapshot.sessions) {
|
||||
const target = createMessageQueueTarget(session.sessionId, session.directory, runtimeKey);
|
||||
if (!target) continue;
|
||||
const key = getMessageQueueKey(target);
|
||||
if ((appliedRevisions.get(key) ?? -1) > snapshot.revision) {
|
||||
// A broadcast newer than this snapshot already landed; keep it.
|
||||
if (state.queuedMessages[key]) queuedMessages[key] = state.queuedMessages[key];
|
||||
if (state.sendingIds[key]) sendingIds[key] = state.sendingIds[key];
|
||||
continue;
|
||||
}
|
||||
appliedRevisions.set(key, snapshot.revision);
|
||||
if (session.items.length > 0) queuedMessages[key] = session.items.map(toQueuedMessage);
|
||||
if (session.sendingId) sendingIds[key] = [session.sendingId];
|
||||
}
|
||||
return { queuedMessages, sendingIds };
|
||||
});
|
||||
},
|
||||
|
||||
setFollowUpBehavior: (behavior) => {
|
||||
set({ followUpBehavior: behavior });
|
||||
void updateDesktopSettings({ followUpBehavior: behavior });
|
||||
},
|
||||
applyServerSession,
|
||||
|
||||
getQueueForTarget: (target) => {
|
||||
return get().queuedMessages[getMessageQueueKey(target)] ?? [];
|
||||
},
|
||||
}),
|
||||
setServerHold: async (sessionId, held) => {
|
||||
if (!isServerOwnedMessageQueue()) return;
|
||||
const response = await runtimeFetch(`${sessionPath(sessionId)}/hold`, jsonInit('PUT', { held }));
|
||||
if (!response.ok) throw new Error(`Message queue hold request failed (${response.status})`);
|
||||
},
|
||||
|
||||
resetForRuntimeSwitch: (previousRuntimeKey) => {
|
||||
hydrationGeneration += 1;
|
||||
if (!previousRuntimeKey || !serverOwnedRuntimeKeys.has(previousRuntimeKey)) return;
|
||||
// The previous runtime's projection belongs to its server;
|
||||
// switching back re-hydrates it from there.
|
||||
set((state) => {
|
||||
const queuedMessages: Record<string, QueuedMessage[]> = {};
|
||||
const sendingIds: Record<string, string[]> = {};
|
||||
for (const [key, queue] of Object.entries(state.queuedMessages)) {
|
||||
if (parseMessageQueueKey(key)?.runtimeKey === previousRuntimeKey) appliedRevisions.delete(key);
|
||||
else queuedMessages[key] = queue;
|
||||
}
|
||||
for (const [key, ids] of Object.entries(state.sendingIds)) {
|
||||
if (parseMessageQueueKey(key)?.runtimeKey !== previousRuntimeKey) sendingIds[key] = ids;
|
||||
}
|
||||
return { queuedMessages, sendingIds };
|
||||
});
|
||||
},
|
||||
};
|
||||
},
|
||||
{
|
||||
name: 'message-queue-store',
|
||||
version: 2,
|
||||
storage: createDeferredSafeJSONStorage(),
|
||||
partialize: (state) => ({
|
||||
queuedMessages: state.queuedMessages,
|
||||
queuedMessages: Object.fromEntries(
|
||||
Object.entries(state.queuedMessages).filter(([key]) => {
|
||||
const runtimeKey = parseMessageQueueKey(key)?.runtimeKey;
|
||||
return !runtimeKey || !serverOwnedRuntimeKeys.has(runtimeKey);
|
||||
}),
|
||||
),
|
||||
quarantinedLegacyMessages: state.quarantinedLegacyMessages,
|
||||
followUpBehavior: state.followUpBehavior,
|
||||
}),
|
||||
@@ -331,3 +714,21 @@ export const useMessageQueueStore = create<MessageQueueStore>()(
|
||||
}
|
||||
)
|
||||
);
|
||||
|
||||
const serverUpdatedEventSchema = z.object({
|
||||
properties: z.object({ revision: z.number(), session: serverSessionSchema }),
|
||||
});
|
||||
|
||||
export type MessageQueueUpdatedEvent = {
|
||||
type: 'openchamber:message-queue.updated';
|
||||
properties: z.infer<typeof serverUpdatedEventSchema>['properties'];
|
||||
};
|
||||
|
||||
/** `openchamber:message-queue.updated` broadcast → projection. */
|
||||
export const applyMessageQueueUpdatedEvent = (payload: Event | MessageQueueUpdatedEvent, expectedRuntimeKey: string): void => {
|
||||
if (!isServerOwnedMessageQueue()) return;
|
||||
const parsed = serverUpdatedEventSchema.safeParse(payload);
|
||||
if (!parsed.success) return;
|
||||
const { session, revision } = parsed.data.properties;
|
||||
useMessageQueueStore.getState().applyServerSession(session, revision, expectedRuntimeKey);
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user