diff --git a/apps/web/package.json b/apps/web/package.json index 1ed057ef18..e4253f3373 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -13,6 +13,8 @@ "test:coverage": "vitest run --coverage" }, "dependencies": { + "@adobe/data": "0.9.83", + "@adobe/data-react": "0.9.83", "@ai-sdk/openai-compatible": "^2.0.52", "@ai-sdk/react": "^3.0.214", "@aws-sdk/client-s3": "^3.1045.0", diff --git a/apps/web/src/app/spike/adobe-data/ChatDatabaseProvider.tsx b/apps/web/src/app/spike/adobe-data/ChatDatabaseProvider.tsx new file mode 100644 index 0000000000..70918d432d --- /dev/null +++ b/apps/web/src/app/spike/adobe-data/ChatDatabaseProvider.tsx @@ -0,0 +1,26 @@ +'use client'; + +import { useState, type ReactNode } from 'react'; +import { DatabaseProvider } from '@adobe/data-react'; +import { createChatDatabase } from '@/state/chat/createChatDatabase'; +import { chatStatePlugin } from '@/state/chat/chat-state-plugin'; + +/** + * SPIKE (@adobe/data adoption evidence) — React binding harness, container half. + * + * `useState(initializer)` (not a module singleton, not `useMemo`) is what makes + * this safe under React 19 StrictMode: the initializer may run twice in + * development, but only the first result is retained, so exactly one Database + * survives per mount and a discarded double-invocation cannot leak observers. + * A module-level singleton would instead be shared across every request on the + * server, which is a cross-tenant state leak — the reason the container is + * created here, inside the client boundary, rather than at import time. + */ +export const ChatDatabaseProvider = ({ children }: { children: ReactNode }) => { + const [handle] = useState(createChatDatabase); + return ( + + {children} + + ); +}; diff --git a/apps/web/src/app/spike/adobe-data/SpikeChatHarness.tsx b/apps/web/src/app/spike/adobe-data/SpikeChatHarness.tsx new file mode 100644 index 0000000000..2836fc6c4e --- /dev/null +++ b/apps/web/src/app/spike/adobe-data/SpikeChatHarness.tsx @@ -0,0 +1,89 @@ +'use client'; + +import { useDatabase, useObservableValues } from '@adobe/data-react'; +import { chatStatePlugin } from '@/state/chat/chat-state-plugin'; + +const CONVERSATION_ID = 'spike-conversation'; +const PAGE_ID = 'spike-page'; + +/** + * SPIKE (@adobe/data adoption evidence) — React binding harness, binding half. + * + * Answers the "Next 15 App Router + React 19" spike question in the shape the + * aidd-react skill prescribes: ONE `useObservableValues` call, no other React + * context, actions passed as callbacks, no business logic in the component. + * + * SSR/hydration: `useObservable` seeds `useState(undefined)` and subscribes in + * an effect, so the server render and the first client render both produce + * `values === undefined` — the skeleton branch below. That makes hydration + * mismatch structurally impossible for observable-derived markup, at the cost + * of the first paint always being the skeleton (no server-rendered content for + * this subtree). Any surface that needs server-rendered chat content must get + * it from a prop/RSC payload, not from the Database. + */ +export const SpikeChatHarness = () => { + const db = useDatabase(chatStatePlugin); + const values = useObservableValues(() => ({ + entry: db.computed.conversationEntry(CONVERSATION_ID), + streams: db.computed.pageStreams(PAGE_ID), + })); + + if (!values) return

Loading chat state…

; + + return ( +
+

{values.entry.loadStatus}

+ + + + + + +
+ ); +}; diff --git a/apps/web/src/app/spike/adobe-data/page.tsx b/apps/web/src/app/spike/adobe-data/page.tsx new file mode 100644 index 0000000000..c2095b0aca --- /dev/null +++ b/apps/web/src/app/spike/adobe-data/page.tsx @@ -0,0 +1,23 @@ +import { ChatDatabaseProvider } from './ChatDatabaseProvider'; +import { SpikeChatHarness } from './SpikeChatHarness'; + +/** + * SPIKE (@adobe/data adoption evidence) — evidence route, NOT a product surface. + * + * A server component (RSC) rendering the client Database boundary, so the spike + * exercises the real Next 15 App Router topology: RSC → 'use client' provider → + * binding component reading `computed` observables. Delete with the spike + * branch; it is deliberately unlinked from any navigation. + */ +export const metadata = { title: '@adobe/data spike harness' }; + +export default function AdobeDataSpikePage() { + return ( +
+

@adobe/data chat-state harness

+ + + +
+ ); +} diff --git a/apps/web/src/state/chat/__tests__/aiActionUndo.test.ts b/apps/web/src/state/chat/__tests__/aiActionUndo.test.ts new file mode 100644 index 0000000000..d6e407d173 --- /dev/null +++ b/apps/web/src/state/chat/__tests__/aiActionUndo.test.ts @@ -0,0 +1,126 @@ +/** + * SPIKE (@adobe/data adoption evidence) — AI actions with atomic undo. + * + * Spike question: "should AI tool-driven edits map to actions→single-transaction + * with the built-in undo/redo stack giving atomic user-facing undo of AI + * changes?" and its companion constraint, "≤1 transaction per action — do + * send/answer/abort corrupt the undo stack?" + */ +import { describe, it, expect } from 'vitest'; +import type { UIMessage } from 'ai'; +import { Observe } from '@adobe/data/observe'; +import { createChatDatabase } from '../createChatDatabase'; + +const msg = (id: string, text: string): UIMessage => ({ + id, + role: 'assistant', + parts: [{ type: 'text', text }], +}); + +const partsOf = (db: ReturnType['db'], id: string) => + db.actions.getEntry('c1').messages.find((m) => m.id === id)?.parts; + +const seededDatabase = () => { + const handle = createChatDatabase(); + handle.db.transactions.applyServerSnapshot({ + conversationId: 'c1', + generationToken: 0, + messages: [msg('m1', 'original')], + }); + return handle; +}; + +const readOnce = (observe: Observe): T => { + let captured: { value: T } | null = null; + observe((value) => { + captured = { value }; + })(); + if (captured === null) throw new Error('observable did not emit synchronously'); + return (captured as { value: T }).value; +}; + +describe('AI action → single undoable transaction', () => { + it('given an AI edit applied through the action, should undo it atomically and leave the original message', () => { + const { db, undoRedo } = seededDatabase(); + + db.actions.aiApplyEdit({ + conversationId: 'c1', + payload: { messageId: 'm1', parts: [{ type: 'text', text: 'ai rewrite' }], editedAt: new Date(0) }, + }); + expect(partsOf(db, 'm1')).toEqual([{ type: 'text', text: 'ai rewrite' }]); + + undoRedo.undo(); + + expect(partsOf(db, 'm1')).toEqual([{ type: 'text', text: 'original' }]); + }); + + it('given an undone AI edit, should redo it', () => { + const { db, undoRedo } = seededDatabase(); + db.actions.aiApplyEdit({ + conversationId: 'c1', + payload: { messageId: 'm1', parts: [{ type: 'text', text: 'ai rewrite' }], editedAt: new Date(0) }, + }); + undoRedo.undo(); + + undoRedo.redo(); + + expect(partsOf(db, 'm1')).toEqual([{ type: 'text', text: 'ai rewrite' }]); + }); + + it('given two AI edits, should undo only the most recent one (no coalescing)', () => { + const { db, undoRedo } = seededDatabase(); + db.actions.aiApplyEdit({ + conversationId: 'c1', + payload: { messageId: 'm1', parts: [{ type: 'text', text: 'first' }], editedAt: new Date(0) }, + }); + db.actions.aiApplyEdit({ + conversationId: 'c1', + payload: { messageId: 'm1', parts: [{ type: 'text', text: 'second' }], editedAt: new Date(1) }, + }); + + undoRedo.undo(); + + expect(partsOf(db, 'm1')).toEqual([{ type: 'text', text: 'first' }]); + }); + + it('given ordinary chat traffic (sends, stream frames, loads), should record nothing on the undo stack', () => { + const { db, undoRedo } = seededDatabase(); + + db.transactions.addOptimisticSend({ conversationId: 'c1', message: msg('m2', 'hello') }); + db.transactions.addStream({ + messageId: 's1', + pageId: 'p1', + conversationId: 'c1', + triggeredBy: { userId: 'u1', displayName: 'Alice' }, + isOwn: true, + }); + db.transactions.appendPart({ messageId: 's1', part: { type: 'text', text: 'tok' } }); + db.transactions.seedConversation('c2'); + + expect(readOnce(undoRedo.undoEnabled)).toBe(false); + }); + + it('given an AI edit surrounded by ordinary chat traffic, should undo the AI edit and nothing else', () => { + const { db, undoRedo } = seededDatabase(); + db.transactions.addOptimisticSend({ conversationId: 'c1', message: msg('m2', 'hello') }); + + db.actions.aiApplyEdit({ + conversationId: 'c1', + payload: { messageId: 'm1', parts: [{ type: 'text', text: 'ai rewrite' }], editedAt: new Date(0) }, + }); + db.transactions.addStream({ + messageId: 's1', + pageId: 'p1', + conversationId: 'c1', + triggeredBy: { userId: 'u1', displayName: 'Alice' }, + isOwn: true, + }); + + undoRedo.undo(); + + expect(partsOf(db, 'm1')).toEqual([{ type: 'text', text: 'original' }]); + expect(db.actions.getEntry('c1').optimisticSends.map((m) => m.id)).toEqual(['m2']); + expect(db.actions.getStream('s1')).not.toBeNull(); + expect(readOnce(undoRedo.undoEnabled)).toBe(false); + }); +}); diff --git a/apps/web/src/state/chat/__tests__/computedObservables.test.ts b/apps/web/src/state/chat/__tests__/computedObservables.test.ts new file mode 100644 index 0000000000..fc7ad9448d --- /dev/null +++ b/apps/web/src/state/chat/__tests__/computedObservables.test.ts @@ -0,0 +1,132 @@ +/** + * SPIKE (@adobe/data adoption evidence) — selectors as `computed` Observe. + * + * The render path under adoption is `useObservableValues(() => ({ entry: + * db.computed.conversationEntry(id) }))`. These tests drive the same + * observables the React harness binds to, without React — so the propagation + * semantics are proven in an environment where render tests are known-broken + * (.pu worktree, dual-React dispatcher). + */ +import { describe, it, expect } from 'vitest'; +import type { UIMessage } from 'ai'; +import { createChatDatabase } from '../createChatDatabase'; + +const msg = (id: string): UIMessage => ({ id, role: 'user', parts: [] }); + +const collect = (observe: (notify: (value: T) => void) => () => void) => { + const values: T[] = []; + const unobserve = observe((value) => values.push(value)); + return { values, unobserve }; +}; + +describe('computed conversationEntry', () => { + it('given a subscription, should emit the seeded empty entry immediately for an unknown conversation', () => { + const { db } = createChatDatabase(); + + const { values, unobserve } = collect(db.computed.conversationEntry('c1')); + + expect(values).toHaveLength(1); + expect(values[0].messages).toEqual([]); + expect(values[0].loadStatus).toBe('idle'); + unobserve(); + }); + + it('given a committed transaction, should emit the new entry', () => { + const { db } = createChatDatabase(); + const { values, unobserve } = collect(db.computed.conversationEntry('c1')); + + db.transactions.applyServerSnapshot({ conversationId: 'c1', generationToken: 0, messages: [msg('m1')] }); + + expect(values[values.length - 1].messages).toEqual([msg('m1')]); + expect(values[values.length - 1].loadStatus).toBe('loaded'); + unobserve(); + }); + + it('given a transaction that changes another conversation, should not re-emit', () => { + const { db } = createChatDatabase(); + db.transactions.seedConversation('c1'); + const { values, unobserve } = collect(db.computed.conversationEntry('c1')); + const before = values.length; + + db.transactions.seedConversation('c2'); + + expect(values).toHaveLength(before); + unobserve(); + }); + + it('given a no-op transaction, should not re-emit', () => { + const { db } = createChatDatabase(); + db.transactions.applyServerSnapshot({ conversationId: 'c1', generationToken: 0, messages: [msg('m1')] }); + const { values, unobserve } = collect(db.computed.conversationEntry('c1')); + const before = values.length; + + // Stale generation → applyLoad returns its input unchanged → zero writes. + db.transactions.applyLoad({ conversationId: 'c1', generation: 999, messages: [msg('ignored')] }); + + expect(values).toHaveLength(before); + unobserve(); + }); + + it('given unobserve, should stop emitting', () => { + const { db } = createChatDatabase(); + const { values, unobserve } = collect(db.computed.conversationEntry('c1')); + unobserve(); + const before = values.length; + + db.transactions.seedConversation('c1'); + + expect(values).toHaveLength(before); + }); +}); + +describe('computed pageStreams', () => { + const stream = (messageId: string, pageId: string, isOwn: boolean) => ({ + messageId, + pageId, + conversationId: 'c1', + triggeredBy: { userId: 'u1', displayName: 'Alice' }, + isOwn, + }); + + it('given streams added on a page, should emit them', () => { + const { db } = createChatDatabase(); + const { values, unobserve } = collect(db.computed.pageStreams('page-a')); + + db.transactions.addStream(stream('s1', 'page-a', false)); + + expect(values[values.length - 1].map((s) => s.messageId)).toEqual(['s1']); + unobserve(); + }); + + it('given a stream added on another page, should not include it', () => { + const { db } = createChatDatabase(); + const { values, unobserve } = collect(db.computed.pageStreams('page-a')); + + db.transactions.addStream(stream('s1', 'page-b', false)); + + expect(values[values.length - 1]).toEqual([]); + unobserve(); + }); + + it('given a streamed token, should emit the appended parts', () => { + const { db } = createChatDatabase(); + db.transactions.addStream(stream('s1', 'page-a', false)); + const { values, unobserve } = collect(db.computed.pageStreams('page-a')); + + db.transactions.appendPart({ messageId: 's1', part: { type: 'text', text: 'tok' } }); + + expect(values[values.length - 1][0].parts).toEqual([{ type: 'text', text: 'tok' }]); + unobserve(); + }); + + it('given mixed own/remote streams, ownPageStreams should emit only the own ones', () => { + const { db } = createChatDatabase(); + db.transactions.addStream(stream('s1', 'page-a', false)); + db.transactions.addStream(stream('s2', 'page-a', true)); + + const { values, unobserve } = collect(db.computed.ownPageStreams('page-a')); + + expect(values[values.length - 1].map((s) => s.messageId)).toEqual(['s2']); + unobserve(); + }); +}); diff --git a/apps/web/src/state/chat/__tests__/conversationMessagesFacade.test.ts b/apps/web/src/state/chat/__tests__/conversationMessagesFacade.test.ts new file mode 100644 index 0000000000..3c37e870e9 --- /dev/null +++ b/apps/web/src/state/chat/__tests__/conversationMessagesFacade.test.ts @@ -0,0 +1,259 @@ +/** + * SPIKE (@adobe/data adoption evidence) — the container-swap proof. + * + * This is `apps/web/src/stores/__tests__/useConversationMessagesStore.test.ts` with EXACTLY ONE + * change: the container it drives. The zustand store import and its + * `setState`-based reset become a freshly-created @adobe/data Database wrapped + * in `createConversationMessagesFacade`. Every `it(...)` title, every arrangement and + * every assertion below is byte-identical to the zustand suite. + * + * DO NOT "fix" a failure here by editing an assertion — a red test is the port + * being wrong, which is the whole point of running this file. + */ +import { describe, it, expect, beforeEach } from 'vitest'; +import type { UIMessage } from 'ai'; +import { createChatDatabase } from '../createChatDatabase'; +import { + createConversationMessagesFacade, + type ConversationMessagesFacade, +} from '../facade/conversationMessagesFacade'; + +let useConversationMessagesStore: ConversationMessagesFacade; + + +const msg = (id: string): UIMessage => ({ id, role: 'user', parts: [] }); + +describe('useConversationMessagesStore', () => { + beforeEach(() => { + useConversationMessagesStore = createConversationMessagesFacade(createChatDatabase().db); + }); + + it('given a conversation never seen before, getEntry should return a seeded empty entry without mutating the store', () => { + const entry = useConversationMessagesStore.getState().getEntry('c1'); + expect(entry).toEqual({ messages: [], optimisticSends: [], loadGeneration: 0, pendingMutationsSinceLoad: [], loadStatus: 'idle', olderCursor: null, hasMoreOlder: false, isLoadingOlder: false }); + expect(useConversationMessagesStore.getState().byConversationId.c1).toBeUndefined(); + }); + + it('given applyServerSnapshot with a current token, should commit the messages as loaded truth', () => { + const { beginServerSnapshot, applyServerSnapshot, getEntry } = useConversationMessagesStore.getState(); + const token = beginServerSnapshot('c1'); + applyServerSnapshot('c1', token, [msg('m1'), msg('m2')]); + const entry = getEntry('c1'); + expect(entry.messages).toEqual([msg('m1'), msg('m2')]); + expect(entry.loadStatus).toBe('loaded'); + }); + + it('given a loud load starting AFTER the snapshot fetch began, the stale snapshot commit must be dropped (the loud load is fresher)', () => { + const { startLoad, beginServerSnapshot, applyServerSnapshot, getEntry } = useConversationMessagesStore.getState(); + const token = beginServerSnapshot('c1'); + startLoad('c1'); + applyServerSnapshot('c1', token, [msg('snapshot')]); + expect(getEntry('c1').messages).toEqual([]); + }); + + it('given a snapshot committing while a loud load is still in flight, the loud load later resolving stale is dropped', () => { + const { startLoad, beginServerSnapshot, applyServerSnapshot, applyLoad, getEntry } = useConversationMessagesStore.getState(); + const inFlight = startLoad('c1'); + // Snapshot fetch begins AFTER the loud load started — it holds the newer view. + const token = beginServerSnapshot('c1'); + applyServerSnapshot('c1', token, [msg('snapshot')]); + applyLoad('c1', inFlight, [msg('stale')]); + expect(getEntry('c1').messages).toEqual([msg('snapshot')]); + }); + + // CR4 (CodeRabbit round 2): two concurrent background heals — the one whose fetch + // started FIRST must not overwrite the one that committed with fresher data, and + // replay cannot recover rows the newer snapshot introduced (its commit clears the + // pending queue). The token is captured BEFORE each fetch; a commit whose token no + // longer matches the entry's generation is dropped. + it('given two racing snapshots, the older-fetched one must not overwrite the newer committed one', () => { + const { beginServerSnapshot, applyServerSnapshot, getEntry } = useConversationMessagesStore.getState(); + const tokenA = beginServerSnapshot('c1'); + const tokenB = beginServerSnapshot('c1'); + // B (fresher fetch) commits first. + applyServerSnapshot('c1', tokenB, [msg('m1'), msg('reply2')]); + // A (older fetch, missing reply2) resolves late — must be dropped. + applyServerSnapshot('c1', tokenA, [msg('m1')]); + expect(getEntry('c1').messages).toEqual([msg('m1'), msg('reply2')]); + }); + + it('given a live delete recorded while the snapshot fetch was in flight, applyServerSnapshot must not resurrect the deleted message', () => { + // The snapshot was FETCHED before this call, so a mutation recorded in the + // window between fetch and commit is newer than the snapshot — it must be + // replayed onto it, not cleared by the generation bump (CodeRabbit P2, PR #2098). + const { startLoad, applyLoad, applyDelete, beginServerSnapshot, applyServerSnapshot, getEntry } = useConversationMessagesStore.getState(); + const gen = startLoad('c1'); + applyLoad('c1', gen, [msg('m1'), msg('m2')]); + // tryRecover's fetch begins; another tab deletes m2 (recorded as a pending mutation). + const token = beginServerSnapshot('c1'); + applyDelete('c1', 'm2'); + // The recovery snapshot resolves — it still contains m2. + applyServerSnapshot('c1', token, [msg('m1'), msg('m2')]); + expect(getEntry('c1').messages).toEqual([msg('m1')]); + }); + + it('given a live remote append recorded while the snapshot fetch was in flight, applyServerSnapshot must keep it', () => { + const { applyRemoteUserMessage, beginServerSnapshot, applyServerSnapshot, getEntry } = useConversationMessagesStore.getState(); + const token = beginServerSnapshot('c1'); + applyRemoteUserMessage('c1', msg('live-append')); + applyServerSnapshot('c1', token, [msg('m1')]); + expect(getEntry('c1').messages).toEqual([msg('m1'), msg('live-append')]); + }); + + it('given applyServerSnapshot containing an optimistic send id, should reconcile it out of optimisticSends', () => { + const { addOptimisticSend, beginServerSnapshot, applyServerSnapshot, getEntry } = useConversationMessagesStore.getState(); + const token = beginServerSnapshot('c1'); + addOptimisticSend('c1', msg('opt1')); + applyServerSnapshot('c1', token, [msg('opt1')]); + const entry = getEntry('c1'); + expect(entry.optimisticSends).toEqual([]); + expect(entry.messages).toEqual([msg('opt1')]); + }); + + it('given seedConversation for a freshly minted id, should mark the entry loaded-empty so no fetch is pending for it', () => { + const { seedConversation, getEntry } = useConversationMessagesStore.getState(); + seedConversation('c-new'); + const entry = getEntry('c-new'); + expect(entry.messages).toEqual([]); + expect(entry.loadStatus).toBe('loaded'); + }); + + it('given startLoad then applyLoad with the returned generation, should commit the loaded messages', () => { + const { startLoad, applyLoad, getEntry } = useConversationMessagesStore.getState(); + const generation = startLoad('c1'); + applyLoad('c1', generation, [msg('m1')]); + expect(getEntry('c1').messages).toEqual([msg('m1')]); + }); + + it('given failLoad after startLoad, should leave the prior entry unchanged', () => { + const { startLoad, applyLoad, failLoad, getEntry } = useConversationMessagesStore.getState(); + const gen1 = startLoad('c1'); + applyLoad('c1', gen1, [msg('m1')]); + const gen2 = startLoad('c1'); + failLoad('c1', gen2); + expect(getEntry('c1').messages).toEqual([msg('m1')]); + }); + + it('given addOptimisticSend, should track the message in optimisticSends', () => { + const { addOptimisticSend, getEntry } = useConversationMessagesStore.getState(); + addOptimisticSend('c1', msg('opt1')); + expect(getEntry('c1').optimisticSends).toEqual([msg('opt1')]); + }); + + it('given applyRemoteUserMessage for an id matching an optimistic send, should reconcile it into messages', () => { + const { addOptimisticSend, applyRemoteUserMessage, getEntry } = useConversationMessagesStore.getState(); + addOptimisticSend('c1', msg('opt1')); + applyRemoteUserMessage('c1', msg('opt1')); + const entry = getEntry('c1'); + expect(entry.messages).toEqual([msg('opt1')]); + expect(entry.optimisticSends).toEqual([]); + }); + + it('given applyEdit for a confirmed message, should update its parts', () => { + const { startLoad, applyLoad, applyEdit, getEntry } = useConversationMessagesStore.getState(); + const gen = startLoad('c1'); + applyLoad('c1', gen, [msg('m1')]); + const editedAt = new Date('2024-01-01T00:00:00.000Z'); + applyEdit('c1', { messageId: 'm1', parts: [{ type: 'text', text: 'edited' }], editedAt }); + expect(getEntry('c1').messages[0]).toMatchObject({ parts: [{ type: 'text', text: 'edited' }], editedAt }); + }); + + it('given applyDelete for a confirmed message, should remove it', () => { + const { startLoad, applyLoad, applyDelete, getEntry } = useConversationMessagesStore.getState(); + const gen = startLoad('c1'); + applyLoad('c1', gen, [msg('m1'), msg('m2')]); + applyDelete('c1', 'm1'); + expect(getEntry('c1').messages).toEqual([msg('m2')]); + }); + + it('given actions against one conversation, should not affect another conversation entry', () => { + const { addOptimisticSend, getEntry } = useConversationMessagesStore.getState(); + addOptimisticSend('c1', msg('opt1')); + expect(getEntry('c2')).toEqual({ messages: [], optimisticSends: [], loadGeneration: 0, pendingMutationsSinceLoad: [], loadStatus: 'idle', olderCursor: null, hasMoreOlder: false, isLoadingOlder: false }); + }); + + it('given isLoadCurrent with the generation returned by startLoad, should return true; with a stale generation, should return false', () => { + const { startLoad, isLoadCurrent } = useConversationMessagesStore.getState(); + const gen1 = startLoad('c1'); + expect(isLoadCurrent('c1', gen1)).toBe(true); + const gen2 = startLoad('c1'); + expect(isLoadCurrent('c1', gen1)).toBe(false); + expect(isLoadCurrent('c1', gen2)).toBe(true); + }); + + it('given startLoadingOlder for a conversation never seen before, should no-op (nothing to mark loading on)', () => { + const { startLoadingOlder, getEntry } = useConversationMessagesStore.getState(); + startLoadingOlder('never-seeded'); + expect(useConversationMessagesStore.getState().byConversationId['never-seeded']).toBeUndefined(); + expect(getEntry('never-seeded').isLoadingOlder).toBe(false); + }); + + it('given startLoadingOlder for a tracked conversation, should set isLoadingOlder true', () => { + const { startLoad, applyLoad, startLoadingOlder, getEntry } = useConversationMessagesStore.getState(); + const gen = startLoad('c1'); + applyLoad('c1', gen, [msg('m1')]); + startLoadingOlder('c1'); + expect(getEntry('c1').isLoadingOlder).toBe(true); + }); + + it('given failLoadingOlder for a conversation never seen before, should no-op', () => { + const { failLoadingOlder } = useConversationMessagesStore.getState(); + failLoadingOlder('never-seeded', 1); + expect(useConversationMessagesStore.getState().byConversationId['never-seeded']).toBeUndefined(); + }); + + it('given failLoadingOlder with a STALE generation (a newer load has since started), should leave isLoadingOlder untouched', () => { + const { startLoad, applyLoad, startLoadingOlder, failLoadingOlder, getEntry } = useConversationMessagesStore.getState(); + const gen1 = startLoad('c1'); + applyLoad('c1', gen1, [msg('m1')]); + startLoad('c1'); // bumps generation — gen1 is now stale + startLoadingOlder('c1'); // a NEW load-older fetch under the current generation + failLoadingOlder('c1', gen1); // the OLD (stale) fetch's failure handler fires late + expect(getEntry('c1').isLoadingOlder).toBe(true); + }); + + it('given failLoadingOlder with the CURRENT generation, should clear isLoadingOlder', () => { + const { startLoad, applyLoad, startLoadingOlder, failLoadingOlder, getEntry } = useConversationMessagesStore.getState(); + const gen = startLoad('c1'); + applyLoad('c1', gen, [msg('m1')]); + startLoadingOlder('c1'); + failLoadingOlder('c1', gen); + expect(getEntry('c1').isLoadingOlder).toBe(false); + }); + + it('given removeOptimisticSendOnFailure for a tracked optimistic send, should remove it from optimisticSends', () => { + const { addOptimisticSend, removeOptimisticSendOnFailure, getEntry } = useConversationMessagesStore.getState(); + addOptimisticSend('c1', msg('opt1')); + removeOptimisticSendOnFailure('c1', 'opt1'); + expect(getEntry('c1').optimisticSends).toEqual([]); + }); + + it('given applyAskUserAnswer then revertAskUserAnswer for the same tool call, should patch to output-available then back to input-available with output dropped', () => { + const { startLoad, applyLoad, applyAskUserAnswer, revertAskUserAnswer, getEntry } = useConversationMessagesStore.getState(); + const gen = startLoad('c1'); + const askUserMessage: UIMessage = { + id: 'm1', + role: 'assistant', + parts: [{ type: 'tool-ask_user', toolCallId: 'tc1', state: 'input-available', input: { questions: [] } } as UIMessage['parts'][number]], + }; + applyLoad('c1', gen, [askUserMessage]); + + applyAskUserAnswer('c1', { messageId: 'm1', toolCallId: 'tc1', output: { answers: [{ header: 'h', question: 'q', otherText: 'hi' }] } }); + expect(getEntry('c1').messages[0].parts[0]).toMatchObject({ state: 'output-available' }); + + revertAskUserAnswer('c1', { messageId: 'm1', toolCallId: 'tc1' }); + const revertedPart = getEntry('c1').messages[0].parts[0] as Record; + expect(revertedPart.state).toBe('input-available'); + expect(revertedPart.output).toBeUndefined(); + }); + + it('given applyConfirmedMessage for a new id, should append it; for an existing id, should replace its content in place', () => { + const { startLoad, applyLoad, applyConfirmedMessage, getEntry } = useConversationMessagesStore.getState(); + const gen = startLoad('c1'); + applyLoad('c1', gen, [msg('m1')]); + applyConfirmedMessage('c1', { id: 'm2', role: 'assistant', parts: [] }); + expect(getEntry('c1').messages.map((m) => m.id)).toEqual(['m1', 'm2']); + applyConfirmedMessage('c1', { id: 'm1', role: 'assistant', parts: [{ type: 'text', text: 'confirmed' }] }); + expect(getEntry('c1').messages[0]).toMatchObject({ id: 'm1', parts: [{ type: 'text', text: 'confirmed' }] }); + }); +}); diff --git a/apps/web/src/state/chat/__tests__/pendingStreamsFacade.test.ts b/apps/web/src/state/chat/__tests__/pendingStreamsFacade.test.ts new file mode 100644 index 0000000000..8534e07db2 --- /dev/null +++ b/apps/web/src/state/chat/__tests__/pendingStreamsFacade.test.ts @@ -0,0 +1,266 @@ +/** + * SPIKE (@adobe/data adoption evidence) — the container-swap proof. + * + * This is `apps/web/src/stores/__tests__/usePendingStreamsStore.test.ts` with EXACTLY ONE + * change: the container it drives. The zustand store import and its + * `setState`-based reset become a freshly-created @adobe/data Database wrapped + * in `createPendingStreamsFacade`. Every `it(...)` title, every arrangement and + * every assertion below is byte-identical to the zustand suite. + * + * DO NOT "fix" a failure here by editing an assertion — a red test is the port + * being wrong, which is the whole point of running this file. + */ +import { describe, it, expect, beforeEach } from 'vitest'; +import { createChatDatabase } from '../createChatDatabase'; +import { createPendingStreamsFacade, type PendingStreamsFacade } from '../facade/pendingStreamsFacade'; + +let usePendingStreamsStore: PendingStreamsFacade; + + +const BASE_STREAM = { + messageId: 'msg-1', + pageId: 'page-a', + conversationId: 'conv-1', + triggeredBy: { userId: 'user-2', displayName: 'Alice' }, + isOwn: false, +}; + +const text = (text: string) => ({ type: 'text' as const, text }); + +describe('usePendingStreamsStore', () => { + beforeEach(() => { + usePendingStreamsStore = createPendingStreamsFacade(createChatDatabase().db); + }); + + describe('initial state', () => { + it('given store is created, should have no streams for any page', () => { + const { getRemotePageStreams } = usePendingStreamsStore.getState(); + expect(getRemotePageStreams('page-a')).toEqual([]); + }); + }); + + describe('addStream', () => { + it('given a new stream, should add it with empty parts', () => { + const { addStream } = usePendingStreamsStore.getState(); + addStream(BASE_STREAM); + + const { getRemotePageStreams } = usePendingStreamsStore.getState(); + const [stream] = getRemotePageStreams('page-a'); + expect(stream).toEqual({ ...BASE_STREAM, parts: [] }); + }); + + it('given two streams for the same page, should store both', () => { + const { addStream } = usePendingStreamsStore.getState(); + addStream(BASE_STREAM); + addStream({ ...BASE_STREAM, messageId: 'msg-2' }); + + const { getRemotePageStreams } = usePendingStreamsStore.getState(); + expect(getRemotePageStreams('page-a')).toHaveLength(2); + }); + + it('given initial parts, should seed the stream with them', () => { + const { addStream } = usePendingStreamsStore.getState(); + addStream({ ...BASE_STREAM, parts: [text('restored')] }); + + const { getRemotePageStreams } = usePendingStreamsStore.getState(); + const [stream] = getRemotePageStreams('page-a'); + expect(stream.parts).toEqual([text('restored')]); + }); + + it('given a duplicate addStream carrying initial parts, should keep the existing entry parts (no double-seed)', () => { + const { addStream } = usePendingStreamsStore.getState(); + addStream({ ...BASE_STREAM, parts: [text('restored')] }); + addStream({ ...BASE_STREAM, parts: [text('restored')] }); + + const { getRemotePageStreams } = usePendingStreamsStore.getState(); + const [stream] = getRemotePageStreams('page-a'); + expect(stream.parts).toEqual([text('restored')]); + }); + + it('given a stream with messageId X already present, should preserve existing parts on duplicate addStream', () => { + const { addStream, appendPart } = usePendingStreamsStore.getState(); + addStream(BASE_STREAM); + appendPart('msg-1', text('partial-text')); + addStream(BASE_STREAM); + + const { getRemotePageStreams } = usePendingStreamsStore.getState(); + const [stream] = getRemotePageStreams('page-a'); + expect(stream.parts).toEqual([text('partial-text')]); + }); + + it('given a duplicate addStream with different metadata, should keep the existing entry unchanged', () => { + const { addStream, appendPart } = usePendingStreamsStore.getState(); + addStream({ ...BASE_STREAM, isOwn: false }); + appendPart('msg-1', text('hello')); + addStream({ ...BASE_STREAM, isOwn: true, conversationId: 'conv-other' }); + + const { getRemotePageStreams } = usePendingStreamsStore.getState(); + const [stream] = getRemotePageStreams('page-a'); + expect(stream.parts).toEqual([text('hello')]); + expect(stream.isOwn).toBe(false); + expect(stream.conversationId).toBe('conv-1'); + }); + }); + + describe('appendPart', () => { + it('given two consecutive text-deltas, should merge them positionally into one text part', () => { + const { addStream, appendPart } = usePendingStreamsStore.getState(); + addStream(BASE_STREAM); + appendPart('msg-1', text('hello')); + appendPart('msg-1', text(' world')); + + const { getRemotePageStreams } = usePendingStreamsStore.getState(); + const [stream] = getRemotePageStreams('page-a'); + expect(stream.parts).toEqual([text('hello world')]); + }); + + it('given a tool part with new toolCallId then output for the same id, should replace the in-place entry rather than duplicate', () => { + const { addStream, appendPart } = usePendingStreamsStore.getState(); + addStream(BASE_STREAM); + const inputPart = { + type: 'tool-list_pages' as const, + toolCallId: 'tc1', + toolName: 'list_pages', + state: 'input-available' as const, + input: { driveId: 'd1' }, + }; + const outputPart = { ...inputPart, state: 'output-available' as const, output: { pages: [] } }; + appendPart('msg-1', inputPart); + appendPart('msg-1', outputPart); + + const { getRemotePageStreams } = usePendingStreamsStore.getState(); + const [stream] = getRemotePageStreams('page-a'); + expect(stream.parts).toEqual([outputPart]); + }); + + it('given unknown messageId, should not throw and should leave state untouched', () => { + const { addStream, appendPart, getRemotePageStreams } = usePendingStreamsStore.getState(); + addStream(BASE_STREAM); + const before = getRemotePageStreams('page-a'); + expect(() => appendPart('unknown', text('lost'))).not.toThrow(); + expect(usePendingStreamsStore.getState().getRemotePageStreams('page-a')).toEqual(before); + }); + }); + + describe('removeStream', () => { + it('given an existing stream, should remove it', () => { + const { addStream, removeStream } = usePendingStreamsStore.getState(); + addStream(BASE_STREAM); + removeStream('msg-1'); + + const { getRemotePageStreams } = usePendingStreamsStore.getState(); + expect(getRemotePageStreams('page-a')).toHaveLength(0); + }); + + it('given unknown messageId, should not throw', () => { + const { removeStream } = usePendingStreamsStore.getState(); + expect(() => removeStream('unknown')).not.toThrow(); + }); + }); + + describe('clearPageStreams', () => { + it("given streams for a page, should remove only that page's streams", () => { + const { addStream, clearPageStreams } = usePendingStreamsStore.getState(); + addStream(BASE_STREAM); + addStream({ ...BASE_STREAM, messageId: 'msg-2' }); + addStream({ ...BASE_STREAM, messageId: 'msg-3', pageId: 'page-b' }); + + clearPageStreams('page-a'); + + const { getRemotePageStreams } = usePendingStreamsStore.getState(); + expect(getRemotePageStreams('page-a')).toHaveLength(0); + expect(getRemotePageStreams('page-b')).toHaveLength(1); + }); + + it('given no streams for the page, should not throw', () => { + const { clearPageStreams } = usePendingStreamsStore.getState(); + expect(() => clearPageStreams('page-missing')).not.toThrow(); + }); + }); + + describe('getRemotePageStreams', () => { + it("given streams for multiple pages, should return only the requested page's streams", () => { + const { addStream, getRemotePageStreams } = usePendingStreamsStore.getState(); + addStream(BASE_STREAM); + addStream({ ...BASE_STREAM, messageId: 'msg-2' }); + addStream({ ...BASE_STREAM, messageId: 'msg-3', pageId: 'page-b' }); + + const streams = getRemotePageStreams('page-a'); + expect(streams).toHaveLength(2); + expect(streams.every((s) => s.pageId === 'page-a')).toBe(true); + }); + + it('given no streams for the page, should return empty array', () => { + const { getRemotePageStreams } = usePendingStreamsStore.getState(); + expect(getRemotePageStreams('page-empty')).toEqual([]); + }); + + it('given appended parts, should reflect accumulated parts in returned stream', () => { + const { addStream, appendPart, getRemotePageStreams } = usePendingStreamsStore.getState(); + addStream(BASE_STREAM); + appendPart('msg-1', text('chunk-a')); + appendPart('msg-1', text('chunk-b')); + + const [stream] = getRemotePageStreams('page-a'); + expect(stream.parts).toEqual([text('chunk-achunk-b')]); + }); + }); + + describe('getOwnStreams', () => { + it('given a stream with isOwn true, should include it', () => { + const { addStream, getOwnStreams } = usePendingStreamsStore.getState(); + addStream({ ...BASE_STREAM, isOwn: true }); + + expect(getOwnStreams('page-a')).toHaveLength(1); + }); + + it('given a stream with isOwn false, should exclude it', () => { + const { addStream, getOwnStreams } = usePendingStreamsStore.getState(); + addStream(BASE_STREAM); + + expect(getOwnStreams('page-a')).toHaveLength(0); + }); + + it('given mixed own and remote streams, should return only own streams for the channel', () => { + const { addStream, getOwnStreams } = usePendingStreamsStore.getState(); + addStream({ ...BASE_STREAM, messageId: 'msg-own', isOwn: true }); + addStream({ ...BASE_STREAM, messageId: 'msg-remote', isOwn: false }); + addStream({ ...BASE_STREAM, messageId: 'msg-other-page', pageId: 'page-b', isOwn: true }); + + const streams = getOwnStreams('page-a'); + expect(streams).toHaveLength(1); + expect(streams[0].messageId).toBe('msg-own'); + }); + + it('given no streams for the channel, should return empty array', () => { + const { getOwnStreams } = usePendingStreamsStore.getState(); + expect(getOwnStreams('page-empty')).toEqual([]); + }); + }); + + describe('setStreamParts', () => { + it('given an existing stream, should replace parts wholesale rather than merge', () => { + const { addStream, setStreamParts, getRemotePageStreams } = usePendingStreamsStore.getState(); + addStream({ ...BASE_STREAM, parts: [text('a')] }); + setStreamParts('msg-1', [text('a'), text('b')], 1); + + const [stream] = getRemotePageStreams('page-a'); + expect(stream.parts).toEqual([text('a'), text('b')]); + }); + + it('given a stale seq (not greater than the last write), should drop the write', () => { + const { addStream, setStreamParts, getRemotePageStreams } = usePendingStreamsStore.getState(); + addStream(BASE_STREAM); + setStreamParts('msg-1', [text('newer')], 5); + setStreamParts('msg-1', [text('stale')], 4); + + const [stream] = getRemotePageStreams('page-a'); + expect(stream.parts).toEqual([text('newer')]); + }); + + it('given an unknown messageId, should not throw', () => { + const { setStreamParts } = usePendingStreamsStore.getState(); + expect(() => setStreamParts('unknown', [text('x')], 1)).not.toThrow(); + }); + }); +}); diff --git a/apps/web/src/state/chat/__tests__/reactBinding.test.tsx b/apps/web/src/state/chat/__tests__/reactBinding.test.tsx new file mode 100644 index 0000000000..70a9ccf45e --- /dev/null +++ b/apps/web/src/state/chat/__tests__/reactBinding.test.tsx @@ -0,0 +1,135 @@ +/** + * SPIKE (@adobe/data adoption evidence) — React 19 binding behaviour. + * + * Spike question: "should useObservableValues-based bindings render and hydrate + * correctly (SSR story, strict mode double-render, RSC boundaries)?" + * + * NOTE for the epic's environment rail: the epic records that React render + * tests fail in .pu worktrees (dual-React dispatcher null). That did NOT + * reproduce for this file — `@adobe/data-react` resolves React from + * `apps/web/node_modules`, the same copy the test runner uses, so client + * render, StrictMode and hydration all execute here. The result is still to be + * re-run on the main checkout before anyone leans on it (recorded on the spike + * page), but the evidence below is real, not a typecheck stand-in. + */ +import { describe, it, expect, afterEach, vi } from 'vitest'; +import { StrictMode, act } from 'react'; +import { renderToStaticMarkup } from 'react-dom/server'; +import { hydrateRoot } from 'react-dom/client'; +import { render, cleanup } from '@testing-library/react'; +import { DatabaseProvider } from '@adobe/data-react'; +import { chatStatePlugin } from '../chat-state-plugin'; +import { createChatDatabase, type ChatDatabaseHandle } from '../createChatDatabase'; +import { SpikeChatHarness } from '@/app/spike/adobe-data/SpikeChatHarness'; + +const CONVERSATION_ID = 'spike-conversation'; + +const tree = (handle: ChatDatabaseHandle) => ( + + + +); + +afterEach(cleanup); + +describe('useObservableValues under server rendering', () => { + it('given a server render, should not throw and should emit the skeleton branch', () => { + const html = renderToStaticMarkup(tree(createChatDatabase())); + + expect(html).toContain('data-testid="spike-skeleton"'); + expect(html).not.toContain('data-testid="spike-messages"'); + }); + + it('given StrictMode, the server render should be byte-identical to the non-strict one', () => { + // Identical markup is what makes hydration mismatch structurally impossible: + // `useObservable` seeds `undefined` and only subscribes in an effect, which + // never runs on the server or in the first client render. + const strict = renderToStaticMarkup({tree(createChatDatabase())}); + const loose = renderToStaticMarkup(tree(createChatDatabase())); + + expect(strict).toBe(loose); + }); +}); + +describe('useObservableValues under client rendering', () => { + it('given a StrictMode mount, should render the observed values after effects run', () => { + const handle = createChatDatabase(); + + const view = render({tree(handle)}); + + expect(view.getByTestId('spike-load-status').textContent).toBe('idle'); + expect(view.queryByTestId('spike-skeleton')).toBeNull(); + }); + + it('given a committed transaction, should propagate to the rendered output', () => { + const handle = createChatDatabase(); + const view = render({tree(handle)}); + + act(() => { + handle.db.transactions.seedConversation(CONVERSATION_ID); + }); + + expect(view.getByTestId('spike-load-status').textContent).toBe('loaded'); + }); + + it('given streamed parts, should propagate each frame to the rendered stream list', () => { + const handle = createChatDatabase(); + const view = render({tree(handle)}); + + act(() => { + handle.db.transactions.addStream({ + messageId: 's1', + pageId: 'spike-page', + conversationId: CONVERSATION_ID, + triggeredBy: { userId: 'u1', displayName: 'Alice' }, + isOwn: true, + }); + }); + act(() => { + handle.db.transactions.appendPart({ messageId: 's1', part: { type: 'text', text: 'tok' } }); + }); + + expect(view.getByTestId('spike-streams').textContent).toContain('s1: 1'); + }); + + it('given an unmount, should stop applying updates', () => { + const handle = createChatDatabase(); + const view = render({tree(handle)}); + view.unmount(); + + // A transaction after unmount must not reach a detached component — React + // would log "update on an unmounted component" via console.error. + const consoleError = vi.spyOn(console, 'error').mockImplementation(() => {}); + act(() => { + handle.db.transactions.seedConversation(CONVERSATION_ID); + }); + + expect(consoleError).not.toHaveBeenCalled(); + consoleError.mockRestore(); + }); +}); + +describe('hydration of a server-rendered harness', () => { + it('given SSR markup hydrated on the client, should not report a hydration mismatch', () => { + const container = document.createElement('div'); + container.innerHTML = renderToStaticMarkup(tree(createChatDatabase())); + document.body.appendChild(container); + const consoleError = vi.spyOn(console, 'error').mockImplementation(() => {}); + + const handle = createChatDatabase(); + let root: ReturnType | null = null; + act(() => { + root = hydrateRoot(container, tree(handle)); + }); + + expect(consoleError).not.toHaveBeenCalled(); + // Post-hydration the effect has run, so the observed values replace the skeleton. + expect(container.querySelector('[data-testid="spike-load-status"]')).not.toBeNull(); + + consoleError.mockRestore(); + act(() => { + root?.unmount(); + }); + container.remove(); + }); +}); diff --git a/apps/web/src/state/chat/__tests__/zustandInterop.test.ts b/apps/web/src/state/chat/__tests__/zustandInterop.test.ts new file mode 100644 index 0000000000..641f046f56 --- /dev/null +++ b/apps/web/src/state/chat/__tests__/zustandInterop.test.ts @@ -0,0 +1,78 @@ +/** + * SPIKE (@adobe/data adoption evidence) — incremental adoption alongside the + * ~two dozen untouched zustand stores. + * + * Spike question: "should @adobe/data adopt incrementally (one Database.Plugin + * hosting chat state while zustand persists elsewhere) without dual-source + * bugs?" + * + * The scenario driven here is a real ask_user answer + failed send, which + * crosses BOTH containers in one flow: the message state lives in the ported + * Database, the answering mutex (`useAskUserAnsweringStore`) and the typed + * error (`useChatErrorStore`) stay in untouched zustand. Neither store is + * modified for this test. + */ +import { describe, it, expect, beforeEach } from 'vitest'; +import type { UIMessage } from 'ai'; +import { useAskUserAnsweringStore } from '@/stores/useAskUserAnsweringStore'; +import { useChatErrorStore } from '@/stores/useChatErrorStore'; +import type { AIErrorCause } from '@/lib/ai/shared/aiErrorCause'; +import { createChatDatabase } from '../createChatDatabase'; +import { createConversationMessagesFacade } from '../facade/conversationMessagesFacade'; + +const msg = (id: string): UIMessage => ({ id, role: 'user', parts: [] }); + +const OUT_OF_CREDITS: AIErrorCause = { + code: 'out_of_credits', + httpStatus: 402, + message: 'Out of credits.', + retryable: false, +}; + +describe('@adobe/data chat container alongside untouched zustand stores', () => { + beforeEach(() => { + useAskUserAnsweringStore.setState({ answeringToolCallIds: new Set() }); + useChatErrorStore.setState({ byConversationId: {} }); + }); + + it('given a send that fails after an ask_user claim, both containers should hold their own half of the state', () => { + const facade = createConversationMessagesFacade(createChatDatabase().db); + + // zustand: claim the answering mutex (unchanged store, unchanged call site). + expect(useAskUserAnsweringStore.getState().claimAnswering('tool-1')).toBe(true); + // @adobe/data: the optimistic send. + facade.getState().addOptimisticSend('c1', msg('m1')); + // zustand: the POST rejects with a typed cause. + useChatErrorStore.getState().setError('c1', OUT_OF_CREDITS); + // @adobe/data: roll the optimistic bubble back. + facade.getState().removeOptimisticSendOnFailure('c1', 'm1'); + + expect(facade.getState().getEntry('c1').optimisticSends).toEqual([]); + expect(useChatErrorStore.getState().getError('c1')).toEqual(OUT_OF_CREDITS); + expect(useAskUserAnsweringStore.getState().answeringToolCallIds.has('tool-1')).toBe(true); + }); + + it('given two chat Databases, the zustand stores should stay global while chat state stays per-container', () => { + const a = createConversationMessagesFacade(createChatDatabase().db); + const b = createConversationMessagesFacade(createChatDatabase().db); + + a.getState().addOptimisticSend('c1', msg('m1')); + useChatErrorStore.getState().setError('c1', OUT_OF_CREDITS); + + expect(a.getState().getEntry('c1').optimisticSends.map((m) => m.id)).toEqual(['m1']); + expect(b.getState().getEntry('c1').optimisticSends).toEqual([]); + // The zustand singleton is shared by construction — that is exactly the + // property that lets un-migrated stores keep working untouched. + expect(useChatErrorStore.getState().getError('c1')).toEqual(OUT_OF_CREDITS); + }); + + it('given the facade, every zustand call site keeps its exact shape (getState().action(...))', () => { + const facade = createConversationMessagesFacade(createChatDatabase().db); + + const generation = facade.getState().startLoad('c1'); + facade.getState().applyLoad('c1', generation, [msg('m1')], { hasMore: false, nextCursor: null }); + + expect(facade.getState().isLoadCurrent('c1', generation)).toBe(true); + expect(facade.getState().byConversationId.c1.messages).toEqual([msg('m1')]); + }); +}); diff --git a/apps/web/src/state/chat/chat-data-plugin.ts b/apps/web/src/state/chat/chat-data-plugin.ts new file mode 100644 index 0000000000..e0f248a967 --- /dev/null +++ b/apps/web/src/state/chat/chat-data-plugin.ts @@ -0,0 +1,95 @@ +import { Database } from '@adobe/data/ecs'; +import type { UIMessage } from 'ai'; +import type { ConversationLoadStatus, PendingMutation } from '@/stores/conversationMessages/seedEmpty'; +import type { PendingStream } from '@/stores/pendingStreams/applyAddStream'; + +type UIMessagePart = UIMessage['parts'][number]; + +/** + * SPIKE (@adobe/data adoption evidence). Data-only half of the chat-state + * plugin: components, archetypes and indexes, with NO transactions. + * + * It is split from `chat-state-plugin.ts` so the column read/write helpers can + * be typed against `Database.Plugin.ToStore` without a + * circular type reference (the behavior plugin's transactions consume those + * helpers, so the helpers cannot depend on the behavior plugin's own type). + * + * Modeling notes (per aidd-ecs data-modeling): + * - one entity per conversation, one entity per live stream — the two keyed + * collections that `useConversationMessagesStore.byConversationId` and + * `usePendingStreamsStore.streams` hold as plain Record/Map today; + * - opaque JS payloads (`UIMessage[]`, part arrays, the recorded pending + * mutation queue) are `mutable: true` components: they are stored by + * reference and replaced wholesale by the pure transition functions, so the + * library must not deep-freeze them into `DeepReadonly`; + * - `streamsByPageId` / `conversationById` are real ECS indexes, which is what + * removes the full-collection scan the epic filed as a `D` finding against + * `usePendingStreamsStore.getRemotePageStreams`. + */ +export const chatDataPlugin = Database.Plugin.create({ + components: { + conversationId: { type: 'string' }, + messages: { default: [] as UIMessage[], mutable: true }, + optimisticSends: { default: [] as UIMessage[], mutable: true }, + loadGeneration: { type: 'integer' }, + pendingMutationsSinceLoad: { default: [] as PendingMutation[], mutable: true }, + loadStatus: { default: 'idle' as ConversationLoadStatus, mutable: true }, + olderCursor: { default: null as string | null, mutable: true }, + hasMoreOlder: { type: 'boolean' }, + isLoadingOlder: { type: 'boolean' }, + + streamMessageId: { type: 'string' }, + streamPageId: { type: 'string' }, + streamConversationId: { type: 'string' }, + streamTriggeredBy: { default: null as unknown as PendingStream['triggeredBy'], mutable: true }, + streamParts: { default: [] as UIMessagePart[], mutable: true }, + streamIsOwn: { type: 'boolean' }, + /** `null` encodes "absent" — the facade projects it back to an omitted key. */ + streamStartedAt: { default: null as string | null, mutable: true }, + /** `null` encodes "no replace-semantics write yet" (`PendingStream.lastSeq` undefined). */ + streamLastSeq: { default: null as number | null, mutable: true }, + }, + archetypes: { + Conversation: [ + 'conversationId', + 'messages', + 'optimisticSends', + 'loadGeneration', + 'pendingMutationsSinceLoad', + 'loadStatus', + 'olderCursor', + 'hasMoreOlder', + 'isLoadingOlder', + ], + PendingStream: [ + 'streamMessageId', + 'streamPageId', + 'streamConversationId', + 'streamTriggeredBy', + 'streamParts', + 'streamIsOwn', + 'streamStartedAt', + 'streamLastSeq', + ], + }, + indexes: { + conversationById: { key: 'conversationId', unique: true, archetype: 'Conversation' }, + streamByMessageId: { key: 'streamMessageId', unique: true, archetype: 'PendingStream' }, + streamsByPageId: { key: 'streamPageId', archetype: 'PendingStream' }, + }, +}); + +export type ChatDataDatabase = Database.Plugin.ToDatabase; +/** + * The read-only projection every projection helper takes. `Database.Read` is + * what a `db.derive` callback receives, and the full `Database` (what `actions` + * receive) is a superset of it — so one helper serves computed, actions and the + * facade without a cast. + */ +export type ChatDataRead = Database.Read; +/** + * The store as seen inside a transaction body. Unlike `Plugin.ToStore`, this + * alias carries the declared `indexes`, which the transitions rely on for + * O(1) entity lookup. + */ +export type ChatDataTransaction = Database.Plugin.ToTransactionContext; diff --git a/apps/web/src/state/chat/chat-state-plugin.ts b/apps/web/src/state/chat/chat-state-plugin.ts new file mode 100644 index 0000000000..a7bbf0f321 --- /dev/null +++ b/apps/web/src/state/chat/chat-state-plugin.ts @@ -0,0 +1,298 @@ +import { Database, applyOperations } from '@adobe/data/ecs'; +import type { UIMessage } from 'ai'; +import { applyStartLoad } from '@/stores/conversationMessages/applyStartLoad'; +import { applyLoad } from '@/stores/conversationMessages/applyLoad'; +import { applyFailLoad } from '@/stores/conversationMessages/applyFailLoad'; +import { applyOptimisticSend } from '@/stores/conversationMessages/applyOptimisticSend'; +import { applyOptimisticSendFailure } from '@/stores/conversationMessages/applyOptimisticSendFailure'; +import { applyOlderPage } from '@/stores/conversationMessages/applyOlderPage'; +import { applyConversationEdit } from '@/stores/conversationMessages/applyConversationEdit'; +import { applyConversationDelete } from '@/stores/conversationMessages/applyConversationDelete'; +import { applyConversationAskUserAnswer } from '@/stores/conversationMessages/applyConversationAskUserAnswer'; +import { applyRemoteUserMessage } from '@/stores/conversationMessages/applyRemoteUserMessage'; +import { applyConfirmedMessage } from '@/stores/conversationMessages/applyConfirmedMessage'; +import { promoteOptimisticSends } from '@/stores/conversationMessages/promoteOptimisticSends'; +import { replayPendingMutations } from '@/stores/conversationMessages/replayPendingMutations'; +import { seedEmpty, type ConversationCacheEntry } from '@/stores/conversationMessages/seedEmpty'; +import { appendPart as appendPartPure } from '@/lib/ai/streams/appendPart'; +import type { MessageEditPayload } from '@/lib/ai/streams/applyMessageEdit'; +import { + revertAskUserAnswer, + type AskUserAnswerPayload, + type AskUserAnswerRevertPayload, +} from '@/lib/ai/streams/applyAskUserAnswer'; +import type { PendingStream } from '@/stores/pendingStreams/applyAddStream'; +import { chatDataPlugin } from './chat-data-plugin'; +import { readConversationEntry, transitionConversation } from './conversationEntry'; +import { readPageStreams, readPendingStream } from './pendingStreamRow'; + +type UIMessagePart = UIMessage['parts'][number]; + +/** + * SPIKE (@adobe/data adoption evidence). Behavior half of the chat-state + * plugin: the two chat stores of E1 PR3 expressed as `transactions` (the + * `applyX` pure transitions, unchanged), `computed` (the render selectors) and + * `actions` (the void helpers + the value-returning glue the effects need). + * + * Two porting patterns show up, and they behave differently: + * + * 1. KEYED transitions (`useConversationMessagesStore`) port 1:1. Each + * `applyX(byConversationId, event)` runs verbatim inside + * `transitionConversation`, against a one-key projection of the conversation + * entity. Zero changes to the transition functions. + * 2. WHOLE-COLLECTION transitions (`usePendingStreamsStore`) do not: their + * input is the entire `Map`, which is exactly the shape ECS replaces. These + * are rewritten against the store — and the rewrite is where the index win + * lands (see `pendingStreamRow.readPageStreams`). Their *sub*-level pure + * helper (`appendPart`) is still reused untouched. + * + * Undo policy: NOTHING here is undoable by default. Only `aiApplyEdit` sets + * `t.undoable`, so a user's Ctrl-Z undoes the AI's edit and never a send, a + * stream frame, or a load — which is the answer to the epic's "≤1 transaction + * per action, don't corrupt the undo stack" constraint. + */ +const chatBehaviorPlugin = Database.Plugin.create({ + extends: chatDataPlugin, + transactions: { + /** Required by `createUndoRedoService` — the replay entry point for undo/redo ops. */ + applyOperations: (t, operations: Parameters[1]) => { + applyOperations(t, operations); + }, + + startLoad: (t, conversationId: string) => { + transitionConversation(t, conversationId, (by) => applyStartLoad(by, conversationId).byConversationId); + }, + applyLoad: ( + t, + event: { + conversationId: string; + generation: number; + messages: UIMessage[]; + pagination?: { hasMore: boolean; nextCursor: string | null }; + }, + ) => { + transitionConversation(t, event.conversationId, (by) => applyLoad(by, event)); + }, + failLoad: (t, event: { conversationId: string; generation: number }) => { + transitionConversation(t, event.conversationId, (by) => applyFailLoad(by, event)); + }, + startLoadingOlder: (t, conversationId: string) => { + transitionConversation(t, conversationId, (by) => { + const existing = by[conversationId]; + if (!existing) return by; + return { ...by, [conversationId]: { ...existing, isLoadingOlder: true } }; + }); + }, + applyOlderPage: ( + t, + event: { + conversationId: string; + generation: number; + messages: UIMessage[]; + hasMoreOlder: boolean; + nextCursor: string | null; + }, + ) => { + transitionConversation(t, event.conversationId, (by) => applyOlderPage(by, event)); + }, + failLoadingOlder: (t, event: { conversationId: string; generation: number }) => { + transitionConversation(t, event.conversationId, (by) => { + const existing = by[event.conversationId]; + if (!existing || existing.loadGeneration !== event.generation) return by; + return { ...by, [event.conversationId]: { ...existing, isLoadingOlder: false } }; + }); + }, + addOptimisticSend: (t, event: { conversationId: string; message: UIMessage }) => { + transitionConversation(t, event.conversationId, (by) => applyOptimisticSend(by, event)); + }, + removeOptimisticSendOnFailure: (t, event: { conversationId: string; messageId: string }) => { + transitionConversation(t, event.conversationId, (by) => applyOptimisticSendFailure(by, event)); + }, + applyEdit: (t, event: { conversationId: string; payload: MessageEditPayload }) => { + transitionConversation(t, event.conversationId, (by) => applyConversationEdit(by, event)); + }, + applyDelete: (t, event: { conversationId: string; messageId: string }) => { + transitionConversation(t, event.conversationId, (by) => applyConversationDelete(by, event)); + }, + applyAskUserAnswer: (t, event: { conversationId: string; payload: AskUserAnswerPayload }) => { + transitionConversation(t, event.conversationId, (by) => applyConversationAskUserAnswer(by, event)); + }, + revertAskUserAnswer: (t, event: { conversationId: string; payload: AskUserAnswerRevertPayload }) => { + transitionConversation(t, event.conversationId, (by) => { + const existing = by[event.conversationId]; + if (!existing) return by; + return { + ...by, + [event.conversationId]: { + ...existing, + messages: revertAskUserAnswer(existing.messages, event.payload), + }, + }; + }); + }, + applyRemoteUserMessage: (t, event: { conversationId: string; message: UIMessage }) => { + transitionConversation(t, event.conversationId, (by) => applyRemoteUserMessage(by, event)); + }, + applyConfirmedMessage: (t, event: { conversationId: string; message: UIMessage }) => { + transitionConversation(t, event.conversationId, (by) => applyConfirmedMessage(by, event)); + }, + promoteOptimisticSends: (t, conversationId: string) => { + transitionConversation(t, conversationId, (by) => promoteOptimisticSends(by, conversationId)); + }, + /** + * The cross-store atomicity the epic hand-managed: `startLoad` + `applyLoad` + * composed inside ONE transaction, so no observer can ever see the + * intermediate 'loading' state this composition passes through. + */ + applyServerSnapshot: ( + t, + event: { conversationId: string; generationToken: number; messages: UIMessage[] }, + ) => { + transitionConversation(t, event.conversationId, (by) => { + const currentGeneration = by[event.conversationId]?.loadGeneration ?? 0; + if (currentGeneration !== event.generationToken) return by; + const pendingSinceFetch = by[event.conversationId]?.pendingMutationsSinceLoad ?? []; + const { byConversationId, generation } = applyStartLoad(by, event.conversationId); + return applyLoad(byConversationId, { + conversationId: event.conversationId, + generation, + messages: replayPendingMutations(event.messages, pendingSinceFetch), + }); + }); + }, + seedConversation: (t, conversationId: string) => { + transitionConversation(t, conversationId, (by) => { + const { byConversationId, generation } = applyStartLoad(by, conversationId); + return applyLoad(byConversationId, { conversationId, generation, messages: [] }); + }); + }, + + /** + * The AI-actions prototype's write half: an edit applied by an AI tool call, + * marked undoable so the built-in undo/redo stack can revert exactly it. + * `coalesce: false` keeps consecutive AI edits as separate undo steps. + */ + aiApplyEdit: (t, event: { conversationId: string; payload: MessageEditPayload }) => { + t.undoable = { coalesce: false }; + transitionConversation(t, event.conversationId, (by) => applyConversationEdit(by, event)); + }, + + addStream: (t, stream: Omit & { parts?: UIMessagePart[] }) => { + if (t.indexes.streamByMessageId.get({ streamMessageId: stream.messageId }) !== null) return; + t.archetypes.PendingStream.insert({ + streamMessageId: stream.messageId, + streamPageId: stream.pageId, + streamConversationId: stream.conversationId, + streamTriggeredBy: stream.triggeredBy, + streamParts: stream.parts ?? [], + streamIsOwn: stream.isOwn, + streamStartedAt: stream.startedAt ?? null, + streamLastSeq: null, + }); + }, + appendPart: (t, event: { messageId: string; part: UIMessagePart }) => { + const entity = t.indexes.streamByMessageId.get({ streamMessageId: event.messageId }); + if (entity === null) return; + const parts = t.get(entity, 'streamParts') ?? []; + const next = appendPartPure(parts, event.part); + if (next === parts) return; + t.update(entity, { streamParts: next }); + }, + setStreamParts: (t, event: { messageId: string; parts: UIMessagePart[]; seq: number }) => { + const entity = t.indexes.streamByMessageId.get({ streamMessageId: event.messageId }); + if (entity === null) return; + if (event.seq <= (t.get(entity, 'streamLastSeq') ?? -1)) return; + t.update(entity, { streamParts: event.parts, streamLastSeq: event.seq }); + }, + removeStream: (t, messageId: string) => { + const entity = t.indexes.streamByMessageId.get({ streamMessageId: messageId }); + if (entity === null) return; + t.delete(entity); + }, + clearPageStreams: (t, pageId: string) => { + for (const entity of [...t.indexes.streamsByPageId.find({ streamPageId: pageId })]) { + t.delete(entity); + } + }, + + /** Test/teardown glue only — the `setState({ byConversationId: {} })` equivalent. */ + resetChatState: (t) => { + t.reset(); + }, + }, + actions: { + /** + * Bumps the load generation and returns it, matching the zustand action's + * contract (callers pass it into the matching `applyLoad`/`failLoad`). + * + * Deliberately NOT a transaction: transactions may only return `void | + * Entity`. The generation is derived from a read taken before the single + * transaction, so it is still deterministic and still one transaction. + */ + startLoad: (db, conversationId: string): number => { + const entity = db.indexes.conversationById.get({ conversationId }); + const next = (entity === null ? 0 : db.get(entity, 'loadGeneration') ?? 0) + 1; + db.transactions.startLoad(conversationId); + return next; + }, + isLoadCurrent: (db, event: { conversationId: string; generation: number }): boolean => { + const entity = db.indexes.conversationById.get({ conversationId: event.conversationId }); + return entity !== null && db.get(entity, 'loadGeneration') === event.generation; + }, + beginServerSnapshot: (db, conversationId: string): number => { + const entity = db.indexes.conversationById.get({ conversationId }); + return entity === null ? 0 : db.get(entity, 'loadGeneration') ?? 0; + }, + getEntry: (db, conversationId: string): ConversationCacheEntry => { + const entity = db.indexes.conversationById.get({ conversationId }); + return entity === null ? seedEmpty() : readConversationEntry(db, entity); + }, + getRemotePageStreams: (db, pageId: string): PendingStream[] => readPageStreams(db, pageId), + getOwnStreams: (db, pageId: string): PendingStream[] => + readPageStreams(db, pageId).filter((stream) => stream.isOwn), + getStream: (db, messageId: string): PendingStream | null => { + const entity = db.indexes.streamByMessageId.get({ streamMessageId: messageId }); + return entity === null ? null : readPendingStream(db, entity); + }, + /** + * AI-actions prototype: an AI tool result becomes ONE action dispatching ONE + * undoable transaction, so the user's undo reverts the whole applied edit + * atomically. Returns nothing — unidirectional flow (aidd-service). + */ + aiApplyEdit: (db, event: { conversationId: string; payload: MessageEditPayload }): void => { + db.transactions.aiApplyEdit(event); + }, + }, +}); + +/** + * Final composition: `computed` MUST live in a plugin separate from `actions`. + * + * SPIKE FINDING (@adobe/data 0.9.83): declaring `computed` and `actions` in the + * SAME `Database.Plugin.create` call silently collapses the inferred action + * declarations to `{}` — `db.actions.X` becomes a compile error — because the + * `computed` factories' constraint type references the action generic and + * poisons its inference. Splitting the two across an `extends` boundary, as + * here, keeps both fully typed with no `any` and no cast. Non-obvious, and it + * dictates plugin layout for every adopter, so it is written up on the spike + * page. + */ +export const chatStatePlugin = Database.Plugin.create({ + extends: chatBehaviorPlugin, + computed: { + /** Render source for a conversation — the `computed` form of `getEntry`. */ + conversationEntry: (db) => (conversationId: string) => + db.derive((read) => { + const entity = read.indexes.conversationById.get({ conversationId }); + return entity === null ? seedEmpty() : readConversationEntry(read, entity); + }), + /** Every live stream on a page — the `computed` form of `getRemotePageStreams`. */ + pageStreams: (db) => (pageId: string) => + db.derive((read) => readPageStreams(read, pageId)), + /** Own-tab live streams on a page — the `computed` form of `getOwnStreams`. */ + ownPageStreams: (db) => (pageId: string) => + db.derive((read) => readPageStreams(read, pageId).filter((stream) => stream.isOwn)), + }, +}); + +export type ChatStateDatabase = Database.Plugin.ToDatabase; diff --git a/apps/web/src/state/chat/conversationEntry.ts b/apps/web/src/state/chat/conversationEntry.ts new file mode 100644 index 0000000000..52ff3c46fc --- /dev/null +++ b/apps/web/src/state/chat/conversationEntry.ts @@ -0,0 +1,70 @@ +import type { Entity } from '@adobe/data/ecs'; +import type { ConversationCacheEntry, ConversationMessagesById } from '@/stores/conversationMessages/seedEmpty'; +import type { ChatDataRead, ChatDataTransaction } from './chat-data-plugin'; + +/** + * SPIKE (@adobe/data adoption evidence). Column ⇄ `ConversationCacheEntry` + * projection for the Conversation archetype. + * + * `?? ` per column is not defensive padding: an ECS entity may + * legitimately lack a component, and the seed value is exactly what + * `seedEmpty()` returns for it, so a partially-populated row reads as the + * same entry the zustand store would have produced. + */ +export const readConversationEntry = (store: ChatDataRead, entity: Entity): ConversationCacheEntry => ({ + messages: store.get(entity, 'messages') ?? [], + optimisticSends: store.get(entity, 'optimisticSends') ?? [], + loadGeneration: store.get(entity, 'loadGeneration') ?? 0, + pendingMutationsSinceLoad: store.get(entity, 'pendingMutationsSinceLoad') ?? [], + loadStatus: store.get(entity, 'loadStatus') ?? 'idle', + olderCursor: store.get(entity, 'olderCursor') ?? null, + hasMoreOlder: store.get(entity, 'hasMoreOlder') ?? false, + isLoadingOlder: store.get(entity, 'isLoadingOlder') ?? false, +}); + +/** All conversation entities, projected back into the store's `byConversationId` shape. */ +export const readAllConversations = (store: ChatDataRead): ConversationMessagesById => { + const byConversationId: ConversationMessagesById = {}; + for (const entity of store.select(['conversationId'])) { + const conversationId = store.get(entity, 'conversationId'); + if (conversationId === undefined) continue; + byConversationId[conversationId] = readConversationEntry(store, entity); + } + return byConversationId; +}; + +/** + * THE container swap, in one function. + * + * Every `useConversationMessagesStore` action was `set(state => ({ byConversationId: + * applyX(state.byConversationId, event) }))`. Here the same pure `applyX` runs + * against a single-key projection of ONE conversation entity and its result is + * written back as columns — the transition function is untouched, only the + * container around it changed. + * + * The pure functions all return their input reference when a transition is a + * no-op, so `after === before` is a precise "nothing changed" test and the + * transaction records zero write operations (which keeps it out of the undo + * stack and out of `db.derive` recomputes). + */ +export const transitionConversation = ( + store: ChatDataTransaction, + conversationId: string, + transition: (byConversationId: ConversationMessagesById) => ConversationMessagesById, +): void => { + const entity = store.indexes.conversationById.get({ conversationId }); + const before: ConversationMessagesById = + entity === null ? {} : { [conversationId]: readConversationEntry(store, entity) }; + + const after = transition(before); + if (after === before) return; + + const next = after[conversationId]; + if (next === undefined) return; + + if (entity === null) { + store.archetypes.Conversation.insert({ conversationId, ...next }); + return; + } + store.update(entity, next); +}; diff --git a/apps/web/src/state/chat/createChatDatabase.ts b/apps/web/src/state/chat/createChatDatabase.ts new file mode 100644 index 0000000000..9a6397f435 --- /dev/null +++ b/apps/web/src/state/chat/createChatDatabase.ts @@ -0,0 +1,21 @@ +import { Database, createUndoRedoService, type UndoRedoService } from '@adobe/data/ecs'; +import { chatStatePlugin, type ChatStateDatabase } from './chat-state-plugin'; + +export interface ChatDatabaseHandle { + readonly db: ChatStateDatabase; + readonly undoRedo: UndoRedoService; +} + +/** + * SPIKE (@adobe/data adoption evidence). Builds one chat Database plus its + * undo/redo service. + * + * A factory (not a module singleton) because tests need an isolated container + * per case and the React harness needs one instance per provider. The + * production facades below hold a lazily-created module singleton, which is the + * shape that lets the existing zustand-consumer components stay unchanged. + */ +export const createChatDatabase = (): ChatDatabaseHandle => { + const db = Database.create(chatStatePlugin); + return { db, undoRedo: createUndoRedoService(db) }; +}; diff --git a/apps/web/src/state/chat/facade/conversationMessagesFacade.ts b/apps/web/src/state/chat/facade/conversationMessagesFacade.ts new file mode 100644 index 0000000000..e39484a63b --- /dev/null +++ b/apps/web/src/state/chat/facade/conversationMessagesFacade.ts @@ -0,0 +1,134 @@ +import type { UIMessage } from 'ai'; +import type { ConversationCacheEntry, ConversationMessagesById } from '@/stores/conversationMessages/seedEmpty'; +import type { MessageEditPayload } from '@/lib/ai/streams/applyMessageEdit'; +import type { AskUserAnswerPayload, AskUserAnswerRevertPayload } from '@/lib/ai/streams/applyAskUserAnswer'; +import { readAllConversations } from '../conversationEntry'; +import type { ChatStateDatabase } from '../chat-state-plugin'; + +/** + * SPIKE (@adobe/data adoption evidence). The zustand-shaped facade over the + * ported Database. + * + * This is the interop proof: `useConversationMessagesStore`'s public surface + * (`getState()` returning `byConversationId` plus the same action names and + * signatures, `setState` for teardown) re-expressed over the ECS container, so + * every existing consumer of the store keeps compiling and behaving while the + * container underneath is @adobe/data instead of zustand — and untouched + * zustand stores elsewhere in the app are unaffected, because nothing about + * this container is global. + * + * It is deliberately NOT a React hook: the rendering path under adoption is + * `useObservableValues` over `db.computed.conversationEntry(id)` (see the + * harness route). This facade exists for the imperative `getState()` call + * sites the epic's effects use. + */ +export interface ConversationMessagesFacadeState { + byConversationId: ConversationMessagesById; + getEntry: (conversationId: string) => ConversationCacheEntry; + startLoad: (conversationId: string) => number; + isLoadCurrent: (conversationId: string, generation: number) => boolean; + applyLoad: ( + conversationId: string, + generation: number, + messages: UIMessage[], + pagination?: { hasMore: boolean; nextCursor: string | null }, + ) => void; + failLoad: (conversationId: string, generation: number) => void; + startLoadingOlder: (conversationId: string) => void; + applyOlderPage: ( + conversationId: string, + generation: number, + messages: UIMessage[], + hasMoreOlder: boolean, + nextCursor: string | null, + ) => void; + failLoadingOlder: (conversationId: string, generation: number) => void; + addOptimisticSend: (conversationId: string, message: UIMessage) => void; + removeOptimisticSendOnFailure: (conversationId: string, messageId: string) => void; + applyEdit: (conversationId: string, payload: MessageEditPayload) => void; + applyDelete: (conversationId: string, messageId: string) => void; + applyAskUserAnswer: (conversationId: string, payload: AskUserAnswerPayload) => void; + revertAskUserAnswer: (conversationId: string, payload: AskUserAnswerRevertPayload) => void; + applyRemoteUserMessage: (conversationId: string, message: UIMessage) => void; + applyConfirmedMessage: (conversationId: string, message: UIMessage) => void; + promoteOptimisticSends: (conversationId: string) => void; + beginServerSnapshot: (conversationId: string) => number; + applyServerSnapshot: (conversationId: string, generationToken: number, messages: UIMessage[]) => void; + seedConversation: (conversationId: string) => void; +} + +export interface ConversationMessagesFacade { + getState: () => ConversationMessagesFacadeState; + /** Only the `{ byConversationId: {} }` teardown form is meaningful on an ECS container. */ + setState: (partial: { byConversationId: ConversationMessagesById }) => void; +} + +export const createConversationMessagesFacade = (db: ChatStateDatabase): ConversationMessagesFacade => { + const getState = (): ConversationMessagesFacadeState => ({ + byConversationId: readAllConversations(db), + getEntry: (conversationId) => db.actions.getEntry(conversationId), + startLoad: (conversationId) => db.actions.startLoad(conversationId), + isLoadCurrent: (conversationId, generation) => db.actions.isLoadCurrent({ conversationId, generation }), + applyLoad: (conversationId, generation, messages, pagination) => { + db.transactions.applyLoad({ conversationId, generation, messages, pagination }); + }, + failLoad: (conversationId, generation) => { + db.transactions.failLoad({ conversationId, generation }); + }, + startLoadingOlder: (conversationId) => { + db.transactions.startLoadingOlder(conversationId); + }, + applyOlderPage: (conversationId, generation, messages, hasMoreOlder, nextCursor) => { + db.transactions.applyOlderPage({ conversationId, generation, messages, hasMoreOlder, nextCursor }); + }, + failLoadingOlder: (conversationId, generation) => { + db.transactions.failLoadingOlder({ conversationId, generation }); + }, + addOptimisticSend: (conversationId, message) => { + db.transactions.addOptimisticSend({ conversationId, message }); + }, + removeOptimisticSendOnFailure: (conversationId, messageId) => { + db.transactions.removeOptimisticSendOnFailure({ conversationId, messageId }); + }, + applyEdit: (conversationId, payload) => { + db.transactions.applyEdit({ conversationId, payload }); + }, + applyDelete: (conversationId, messageId) => { + db.transactions.applyDelete({ conversationId, messageId }); + }, + applyAskUserAnswer: (conversationId, payload) => { + db.transactions.applyAskUserAnswer({ conversationId, payload }); + }, + revertAskUserAnswer: (conversationId, payload) => { + db.transactions.revertAskUserAnswer({ conversationId, payload }); + }, + applyRemoteUserMessage: (conversationId, message) => { + db.transactions.applyRemoteUserMessage({ conversationId, message }); + }, + applyConfirmedMessage: (conversationId, message) => { + db.transactions.applyConfirmedMessage({ conversationId, message }); + }, + promoteOptimisticSends: (conversationId) => { + db.transactions.promoteOptimisticSends(conversationId); + }, + beginServerSnapshot: (conversationId) => db.actions.beginServerSnapshot(conversationId), + applyServerSnapshot: (conversationId, generationToken, messages) => { + db.transactions.applyServerSnapshot({ conversationId, generationToken, messages }); + }, + seedConversation: (conversationId) => { + db.transactions.seedConversation(conversationId); + }, + }); + + return { + getState, + setState: (partial) => { + if (Object.keys(partial.byConversationId).length > 0) { + throw new Error( + 'conversationMessagesFacade.setState only supports the empty-reset form; seed state through transactions.', + ); + } + db.transactions.resetChatState(); + }, + }; +}; diff --git a/apps/web/src/state/chat/facade/pendingStreamsFacade.ts b/apps/web/src/state/chat/facade/pendingStreamsFacade.ts new file mode 100644 index 0000000000..71c7e92c10 --- /dev/null +++ b/apps/web/src/state/chat/facade/pendingStreamsFacade.ts @@ -0,0 +1,69 @@ +import type { UIMessage } from 'ai'; +import type { PendingStream, PendingStreamsMap } from '@/stores/pendingStreams/applyAddStream'; +import type { ChatStateDatabase } from '../chat-state-plugin'; + +type UIMessagePart = UIMessage['parts'][number]; + +/** + * SPIKE (@adobe/data adoption evidence). The zustand-shaped facade over the + * ported pending-streams state — see `conversationMessagesFacade` for why the + * facade exists at all. + */ +export interface PendingStreamsFacadeState { + streams: PendingStreamsMap; + addStream: (stream: Omit & { parts?: UIMessagePart[] }) => void; + appendPart: (messageId: string, part: UIMessagePart) => void; + setStreamParts: (messageId: string, parts: UIMessagePart[], seq: number) => void; + removeStream: (messageId: string) => void; + clearPageStreams: (pageId: string) => void; + getRemotePageStreams: (pageId: string) => PendingStream[]; + getOwnStreams: (pageId: string) => PendingStream[]; +} + +export interface PendingStreamsFacade { + getState: () => PendingStreamsFacadeState; + /** Only the `{ streams: new Map() }` teardown form is meaningful on an ECS container. */ + setState: (partial: { streams: PendingStreamsMap }) => void; +} + +export const createPendingStreamsFacade = (db: ChatStateDatabase): PendingStreamsFacade => { + const readStreams = (): PendingStreamsMap => { + const streams: PendingStreamsMap = new Map(); + for (const entity of db.select(['streamMessageId'])) { + const stream = db.actions.getStream(db.get(entity, 'streamMessageId') ?? ''); + if (stream !== null) streams.set(stream.messageId, stream); + } + return streams; + }; + + return { + getState: () => ({ + streams: readStreams(), + addStream: (stream) => { + db.transactions.addStream(stream); + }, + appendPart: (messageId, part) => { + db.transactions.appendPart({ messageId, part }); + }, + setStreamParts: (messageId, parts, seq) => { + db.transactions.setStreamParts({ messageId, parts, seq }); + }, + removeStream: (messageId) => { + db.transactions.removeStream(messageId); + }, + clearPageStreams: (pageId) => { + db.transactions.clearPageStreams(pageId); + }, + getRemotePageStreams: (pageId) => db.actions.getRemotePageStreams(pageId), + getOwnStreams: (pageId) => db.actions.getOwnStreams(pageId), + }), + setState: (partial) => { + if (partial.streams.size > 0) { + throw new Error( + 'pendingStreamsFacade.setState only supports the empty-reset form; seed state through transactions.', + ); + } + db.transactions.resetChatState(); + }, + }; +}; diff --git a/apps/web/src/state/chat/pendingStreamRow.ts b/apps/web/src/state/chat/pendingStreamRow.ts new file mode 100644 index 0000000000..3137d5e539 --- /dev/null +++ b/apps/web/src/state/chat/pendingStreamRow.ts @@ -0,0 +1,50 @@ +import type { Entity } from '@adobe/data/ecs'; +import type { PendingStream } from '@/stores/pendingStreams/applyAddStream'; +import type { ChatDataRead } from './chat-data-plugin'; + +/** + * SPIKE (@adobe/data adoption evidence). Column ⇄ `PendingStream` projection. + * + * `startedAt`/`lastSeq` are optional on `PendingStream` but a component column + * cannot be "sometimes missing" without leaving the archetype, so absence is + * encoded as `null` and projected back to `undefined` here. Vitest `toEqual` + * ignores explicitly-`undefined` keys, so the existing store assertions + * (`expect(stream).toEqual({ ...BASE_STREAM, parts: [] })`) hold unchanged. + */ +export const readPendingStream = (store: ChatDataRead, entity: Entity): PendingStream | null => { + const messageId = store.get(entity, 'streamMessageId'); + const pageId = store.get(entity, 'streamPageId'); + const conversationId = store.get(entity, 'streamConversationId'); + const triggeredBy = store.get(entity, 'streamTriggeredBy'); + if (messageId === undefined || pageId === undefined || conversationId === undefined || triggeredBy === null || triggeredBy === undefined) { + return null; + } + const startedAt = store.get(entity, 'streamStartedAt') ?? null; + const lastSeq = store.get(entity, 'streamLastSeq') ?? null; + return { + messageId, + pageId, + conversationId, + triggeredBy, + parts: store.get(entity, 'streamParts') ?? [], + isOwn: store.get(entity, 'streamIsOwn') ?? false, + startedAt: startedAt ?? undefined, + lastSeq: lastSeq ?? undefined, + }; +}; + +/** + * Every live stream on a page, in insertion order. + * + * This is the ECS index doing what `getRemotePageStreams` could not: a bucket + * lookup on `streamPageId` instead of iterating the app-wide stream map and + * filtering afterwards (the epic's filed `D —` performance finding). + */ +export const readPageStreams = (store: ChatDataRead, pageId: string): PendingStream[] => { + const streams: PendingStream[] = []; + for (const entity of store.indexes.streamsByPageId.find({ streamPageId: pageId })) { + const stream = readPendingStream(store, entity); + if (stream !== null) streams.push(stream); + } + return streams; +}; diff --git a/bun.lock b/bun.lock index 5e0c93c578..0f6d4f9782 100644 --- a/bun.lock +++ b/bun.lock @@ -307,6 +307,8 @@ "name": "web", "version": "0.1.0", "dependencies": { + "@adobe/data": "0.9.83", + "@adobe/data-react": "0.9.83", "@ai-sdk/openai-compatible": "^2.0.52", "@ai-sdk/react": "^3.0.214", "@aws-sdk/client-s3": "^3.1045.0", @@ -550,6 +552,10 @@ "@adobe/css-tools": ["@adobe/css-tools@4.4.4", "", {}, "sha512-Elp+iwUx5rN5+Y8xLt5/GRoG20WGoDCQ/1Fb+1LiGtvwbDavuSk0jhD/eZdckHAuzcDzccnkv+rEjyWfRx18gg=="], + "@adobe/data": ["@adobe/data@0.9.83", "", { "dependencies": { "@cfworker/json-schema": "^4.1.1", "hash-wasm": "^4.12.0", "jsonpath": "^1.1.1" } }, "sha512-fUzDjURd8N5LXW1hZyvqSWSfGR7/VGyJ6WK0SLOXRw90c5bdrMxtjykG/d8TPtkfjqhHTDJSEf/E2fGKdAGaPA=="], + + "@adobe/data-react": ["@adobe/data-react@0.9.83", "", { "dependencies": { "@adobe/data": "0.9.83" }, "peerDependencies": { "react": ">=17.0.0" } }, "sha512-HnqseguqQZjl4il9YSk37+FsrpafNcwQ78SC/1oje68PseU5NMKRc4DEJHSOWnVcXSa3CPlcaMIG/qZV8Gg/cA=="], + "@ai-sdk/gateway": ["@ai-sdk/gateway@3.0.137", "", { "dependencies": { "@ai-sdk/provider": "3.0.11", "@ai-sdk/provider-utils": "4.0.31", "@vercel/oidc": "3.2.0" }, "peerDependencies": { "zod": "^3.25.76 || ^4.1.8" } }, "sha512-OIOBsRq8hrpML0kdT1QqC2s6nLnmX8G60xvukFJeWUd+jguY2WngzPVQkwlRBJN137LtEa7QyA0NMU3uxxVy3g=="], "@ai-sdk/openai-compatible": ["@ai-sdk/openai-compatible@2.0.52", "", { "dependencies": { "@ai-sdk/provider": "3.0.11", "@ai-sdk/provider-utils": "4.0.31" }, "peerDependencies": { "zod": "^3.25.76 || ^4.1.8" } }, "sha512-WdVesd9VJQv8BLLla+x1AjLa40D3QUfnXH488+ydOcOIX7c4m6RF8Sh3JsAF+5ICa2Vknc9GTBlOvqCKoivE1w=="], @@ -748,6 +754,8 @@ "@capgo/capacitor-social-login": ["@capgo/capacitor-social-login@7.20.0", "", { "peerDependencies": { "@capacitor/core": ">=7.0.0" } }, "sha512-rHlEALFUonLe4sDG+qybDowh2/XDiHJ4DK9pShe1aPTGSrmA5j7kz8VbHYGqgof5xdyeC7pAY3+8PASTW+qGKg=="], + "@cfworker/json-schema": ["@cfworker/json-schema@4.1.1", "", {}, "sha512-gAmrUZSGtKc3AiBL71iNWxDsyUC5uMaKKGdvzYsBoTW/xi42JQHl7eKV2OYzCUqvc+D2RCcf7EXY2iCyFIk6og=="], + "@chevrotain/types": ["@chevrotain/types@11.1.2", "", {}, "sha512-U+HFai5+zmJCkK86QsaJtoITlboZHBqrVketcO2ROv865xfCMSFpELQoz1GkX5GzME8pTa+3kbKrZHQtI0gdbw=="], "@clack/core": ["@clack/core@1.4.3", "", { "dependencies": { "fast-wrap-ansi": "^0.2.0", "sisteransi": "^1.0.5" } }, "sha512-/kr3UWNtdJfxZtPgDqUOmG2pvwlmcLGheex5yiZKdwbzZJxhV+HMNR9QNmyY5cGwTNV6LrR7Jtp+KjhUAP1qBQ=="], @@ -2718,6 +2726,8 @@ "escape-string-regexp": ["escape-string-regexp@4.0.0", "", {}, "sha512-TtpcNJ3XAzx3Gq8sWRzJaVajRs0uVxA2YAkdb1jm2YkPz4G6egUFAyA3n5vtEIZefPk5Wa4UXbKuS5fKkJWdgA=="], + "escodegen": ["escodegen@2.1.0", "", { "dependencies": { "esprima": "^4.0.1", "estraverse": "^5.2.0", "esutils": "^2.0.2" }, "optionalDependencies": { "source-map": "~0.6.1" }, "bin": { "esgenerate": "bin/esgenerate.js", "escodegen": "bin/escodegen.js" } }, "sha512-2NlIDTwUWJN0mRPQOdtQBzbUHvdGY2P1VXSyU83Q3xKxM7WHX2Ql8dKq782Q9TgQUNOLEzEYu9bzLNj1q88I5w=="], + "eslint": ["eslint@9.39.4", "", { "dependencies": { "@eslint-community/eslint-utils": "^4.8.0", "@eslint-community/regexpp": "^4.12.1", "@eslint/config-array": "^0.21.2", "@eslint/config-helpers": "^0.4.2", "@eslint/core": "^0.17.0", "@eslint/eslintrc": "^3.3.5", "@eslint/js": "9.39.4", "@eslint/plugin-kit": "^0.4.1", "@humanfs/node": "^0.16.6", "@humanwhocodes/module-importer": "^1.0.1", "@humanwhocodes/retry": "^0.4.2", "@types/estree": "^1.0.6", "ajv": "^6.14.0", "chalk": "^4.0.0", "cross-spawn": "^7.0.6", "debug": "^4.3.2", "escape-string-regexp": "^4.0.0", "eslint-scope": "^8.4.0", "eslint-visitor-keys": "^4.2.1", "espree": "^10.4.0", "esquery": "^1.5.0", "esutils": "^2.0.2", "fast-deep-equal": "^3.1.3", "file-entry-cache": "^8.0.0", "find-up": "^5.0.0", "glob-parent": "^6.0.2", "ignore": "^5.2.0", "imurmurhash": "^0.1.4", "is-glob": "^4.0.0", "json-stable-stringify-without-jsonify": "^1.0.1", "lodash.merge": "^4.6.2", "minimatch": "^3.1.5", "natural-compare": "^1.4.0", "optionator": "^0.9.3" }, "peerDependencies": { "jiti": "*" }, "optionalPeers": ["jiti"], "bin": { "eslint": "bin/eslint.js" } }, "sha512-XoMjdBOwe/esVgEvLmNsD3IRHkm7fbKIUGvrleloJXUZgDHig2IPWNniv+GwjyJXzuNqVjlr5+4yVUZjycJwfQ=="], "eslint-config-next": ["eslint-config-next@15.3.9", "", { "dependencies": { "@next/eslint-plugin-next": "15.3.9", "@rushstack/eslint-patch": "^1.10.3", "@typescript-eslint/eslint-plugin": "^5.4.2 || ^6.0.0 || ^7.0.0 || ^8.0.0", "@typescript-eslint/parser": "^5.4.2 || ^6.0.0 || ^7.0.0 || ^8.0.0", "eslint-import-resolver-node": "^0.3.6", "eslint-import-resolver-typescript": "^3.5.2", "eslint-plugin-import": "^2.31.0", "eslint-plugin-jsx-a11y": "^6.10.0", "eslint-plugin-react": "^7.37.0", "eslint-plugin-react-hooks": "^5.0.0" }, "peerDependencies": { "eslint": "^7.23.0 || ^8.0.0 || ^9.0.0", "typescript": ">=3.3.1" }, "optionalPeers": ["typescript"] }, "sha512-tY/893UZ6rcfJd+G5c1KdRDceEJu3TerrWp+MCzElJD0KpPfrrHMJ8Tq2dTn0bcdRr/wElp0qrXM2vWVf/9uXw=="], @@ -2974,6 +2984,8 @@ "has-unicode": ["has-unicode@2.0.1", "", {}, "sha512-8Rf9Y83NBReMnx0gFzA8JImQACstCYWUplepDa9xprwwtmgEZUF0h/i5xSA625zB/I37EtrswSST6OXxwaaIJQ=="], + "hash-wasm": ["hash-wasm@4.12.0", "", {}, "sha512-+/2B2rYLb48I/evdOIhP+K/DD2ca2fgBjp6O+GBEnCDk2e4rpeXIK8GvIyRPjTezgmWn9gmKwkQjjx6BtqDHVQ=="], + "hasown": ["hasown@2.0.2", "", { "dependencies": { "function-bind": "^1.1.2" } }, "sha512-0hJU9SCPvmMzIBdZFqNPXWa6dqh7WdH0cII9y+CyS8rG3nL48Bclra9HmKhVVUHyPWNH5Y7xDwAB7bfgSjkUMQ=="], "hast": ["hast@1.0.0", "", {}, "sha512-vFUqlRV5C+xqP76Wwq2SrM0kipnmpxJm7OfvVXpB35Fp+Fn4MV+ozr+JZr5qFvyR1q/U+Foim2x+3P+x9S1PLA=="], @@ -3262,6 +3274,8 @@ "jsonfile": ["jsonfile@6.2.1", "", { "dependencies": { "universalify": "^2.0.0" }, "optionalDependencies": { "graceful-fs": "^4.1.6" } }, "sha512-zwOTdL3rFQ/lRdBnntKVOX6k5cKJwEc1HdilT71BWEu7J41gXIB2MRp+vxduPSwZJPWBxEzv4yH1wYLJGUHX4Q=="], + "jsonpath": ["jsonpath@1.3.0", "", { "dependencies": { "esprima": "1.2.5", "static-eval": "2.1.1", "underscore": "1.13.6" } }, "sha512-0kjkYHJBkAy50Z5QzArZ7udmvxrJzkpKYW27fiF//BrMY7TQibYLl+FYIXN2BiYmwMIVzSfD8aDRj6IzgBX2/w=="], + "jsonwebtoken": ["jsonwebtoken@9.0.3", "", { "dependencies": { "jws": "^4.0.1", "lodash.includes": "^4.3.0", "lodash.isboolean": "^3.0.3", "lodash.isinteger": "^4.0.4", "lodash.isnumber": "^3.0.3", "lodash.isplainobject": "^4.0.6", "lodash.isstring": "^4.0.1", "lodash.once": "^4.0.0", "ms": "^2.1.1", "semver": "^7.5.4" } }, "sha512-MT/xP0CrubFRNLNKvxJ2BYfy53Zkm++5bX9dtuPbqAeQpTVe0MQTFhao8+Cp//EmJp244xt6Drw/GVEGCUj40g=="], "jsx-ast-utils": ["jsx-ast-utils@3.3.5", "", { "dependencies": { "array-includes": "^3.1.6", "array.prototype.flat": "^1.3.1", "object.assign": "^4.1.4", "object.values": "^1.1.6" } }, "sha512-ZZow9HBI5O6EPgSJLUb8n2NKgmVWTwCvHGwFuJlMjvLFqlGG6pjirPhtdsseaLZjSibD8eegzmYpUZwoIlj2cQ=="], @@ -4192,6 +4206,8 @@ "state-local": ["state-local@1.0.7", "", {}, "sha512-HTEHMNieakEnoe33shBYcZ7NX83ACUjCu8c40iOGEZsngj9zRnkqS9j1pqQPXwobB0ZcVTk27REb7COQ0UR59w=="], + "static-eval": ["static-eval@2.1.1", "", { "dependencies": { "escodegen": "^2.1.0" } }, "sha512-MgWpQ/ZjGieSVB3eOJVs4OA2LT/q1vx98KPCTTQPzq/aLr0YUXTsgryTXr4SLfR0ZfUUCiedM9n/ABeDIyy4mA=="], + "statuses": ["statuses@2.0.2", "", {}, "sha512-DvEy55V3DB7uknRo+4iOGT5fP1slR8wQohVdknigZPMpMstaKJQWhwiYBACJE3Ul2pTnATihhBYnRhZQHGBiRw=="], "std-env": ["std-env@3.10.0", "", {}, "sha512-5GS12FdOZNliM5mAOxFRg7Ir0pWz8MdpYm6AY6VPkGpbA7ZzmbzNcBJQ0GPvvyWgcY7QAhCgf9Uy89I03faLkg=="], @@ -5086,6 +5102,10 @@ "jsdom/ws": ["ws@8.19.0", "", { "peerDependencies": { "bufferutil": "^4.0.1", "utf-8-validate": ">=5.0.2" }, "optionalPeers": ["bufferutil", "utf-8-validate"] }, "sha512-blAT2mjOEIi0ZzruJfIhb3nps74PRWTCz1IjglWEEpQl5XS/UNama6u2/rjFkDDouqr4L67ry+1aGIALViWjDg=="], + "jsonpath/esprima": ["esprima@1.2.5", "", { "bin": { "esparse": "./bin/esparse.js", "esvalidate": "./bin/esvalidate.js" } }, "sha512-S9VbPDU0adFErpDai3qDkjq8+G05ONtKzcyNrPKg/ZKa+tf879nX2KexNU95b31UoTJjRLInNBHHHjFPoCd7lQ=="], + + "jsonpath/underscore": ["underscore@1.13.6", "", {}, "sha512-+A5Sja4HP1M08MaXya7p5LvjuM7K6q/2EaC0+iovj/wOcMsTzMvDFbasi/oSapiwOlt252IqsKqPjCl7huKS0A=="], + "jsonwebtoken/semver": ["semver@7.8.0", "", { "bin": { "semver": "bin/semver.js" } }, "sha512-AcM7dV/5ul4EekoQ29Agm5vri8JNqRyj39o0qpX6vDF2GZrtutZl5RwgD1XnZjiTAfncsJhMI48QQH3sN87YNA=="], "jszip/pako": ["pako@1.0.11", "", {}, "sha512-4hLB8Py4zZce5s4yd9XzopqwVv/yGNhV1Bl8NTmCq1763HeK2+EwVTv+leGeL13Dnh2wfbqowVPXCIO0z4taYw=="],