Skip to content

Commit ec936c8

Browse files
mzxchandramzxchandra
committed
refactor: carry file owners through collaborative documents
1 parent 3db7e57 commit ec936c8

32 files changed

Lines changed: 786 additions & 401 deletions

‎apps/realtime/src/handlers/file-doc-editor.integration.ts‎

Lines changed: 11 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,11 @@ beforeAll(async () => {
2222
await store.init()
2323
})
2424

25-
async function fixture() {
26-
const name = `project-file-doc:${generateId()}/${generateId()}`
25+
async function fixture(scope: 'project' | 'workspace' = 'project') {
26+
const name =
27+
scope === 'project'
28+
? `project-file-doc:${generateId()}/${generateId()}`
29+
: `workspace-file-doc:${generateId()}`
2730
const doc = new Y.Doc()
2831
const generation = generateId()
2932
doc.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.flag, true)
@@ -75,16 +78,15 @@ describe('Project document editor attribution over real Redis', () => {
7578
}
7679
)
7780

78-
check(
79-
'Project updates without an authenticated editor are rejected before acceptance',
80-
async () => {
81-
const f = await fixture()
81+
for (const scope of ['workspace', 'project'] as const) {
82+
check(`${scope} acknowledged updates require an authenticated editor`, async () => {
83+
const f = await fixture(scope)
8284
await expect(
8385
store.publishClientUpdateAndWait(f.name, 'unknown', update(f.doc, 'X'), f.generation)
84-
).rejects.toThrow('Project document updates require an authenticated editor')
86+
).rejects.toThrow('Document updates require an authenticated editor')
8587
expect(await redis.xLen(`filedoc:stream:${f.name}`)).toBe(1)
86-
}
87-
)
88+
})
89+
}
8890

8991
check(
9092
'real compaction preserves the editor for a headless replica reconstructing accepted bytes',

‎apps/realtime/src/handlers/file-doc-owner.ts‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
import type { FileDocOwner, FileDocTarget } from '@sim/realtime-protocol/file-doc-target'
2+
import {
3+
fetchFileDocPersist,
4+
fetchFileDocSeed,
5+
fetchProjectFileDocPersist,
6+
fetchProjectFileDocSeed,
7+
type PersistResult,
8+
} from '@/handlers/file-doc-app'
9+
import type { FileDocEditor } from '@/handlers/file-doc-store'
10+
11+
interface PersistenceActor {
12+
userId: string
13+
connectionId: string | null
14+
}
15+
16+
interface FileDocOwnerAdapter {
17+
requiresCurrentActor: boolean
18+
tracksLifecycle: boolean
19+
seed(
20+
target: FileDocTarget,
21+
actor?: FileDocEditor
22+
): Promise<{ update: Uint8Array; version: number } | null>
23+
persist(
24+
target: FileDocTarget,
25+
actor: PersistenceActor,
26+
state: Uint8Array,
27+
version?: number
28+
): Promise<PersistResult>
29+
}
30+
31+
function requireOwner(target: FileDocTarget, entityType: FileDocOwner['entityType']): string {
32+
if (!target.owner || target.owner.entityType !== entityType || !target.owner.entityId)
33+
throw new Error('Document has no canonical owner context')
34+
return target.owner.entityId
35+
}
36+
37+
function projectRequest(target: FileDocTarget, actor?: PersistenceActor) {
38+
if (!actor?.userId || !actor.connectionId)
39+
throw new Error('Document callback requires an authenticated editor')
40+
return {
41+
projectId: requireOwner(target, 'project'),
42+
fileId: target.fileId,
43+
userId: actor.userId,
44+
connectionId: actor.connectionId,
45+
}
46+
}
47+
48+
const OWNER_ADAPTERS: Record<FileDocOwner['entityType'], FileDocOwnerAdapter> = {
49+
workspace: {
50+
requiresCurrentActor: false,
51+
tracksLifecycle: false,
52+
seed: (target) => fetchFileDocSeed(requireOwner(target, 'workspace'), target.fileId),
53+
persist: (target, actor, state, version) =>
54+
fetchFileDocPersist(
55+
requireOwner(target, 'workspace'),
56+
target.fileId,
57+
actor.userId,
58+
state,
59+
version
60+
),
61+
},
62+
project: {
63+
requiresCurrentActor: true,
64+
tracksLifecycle: true,
65+
seed: (target, actor) => fetchProjectFileDocSeed(projectRequest(target, actor)),
66+
persist: (target, actor, state, version) =>
67+
fetchProjectFileDocPersist(projectRequest(target, actor), state, version),
68+
},
69+
}
70+
71+
/** Only registered scopes can use callbacks; legacy workspace addresses resolve their owner at join. */
72+
export function fileDocOwnerAdapter(owner?: FileDocOwner): FileDocOwnerAdapter {
73+
const type = owner?.entityType ?? 'workspace'
74+
if (!Object.hasOwn(OWNER_ADAPTERS, type)) throw new Error('Unsupported document owner')
75+
return OWNER_ADAPTERS[type]
76+
}

‎apps/realtime/src/handlers/file-doc-project.integration.ts‎

Lines changed: 89 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
import { mkdir, writeFile } from 'node:fs/promises'
22
import { createServer, type Server as HttpServer } from 'node:http'
33
import { dirname, resolve } from 'node:path'
4+
import { createLogger } from '@sim/logger'
45
import {
56
FILE_DOC_EVENTS,
67
FILE_DOC_SCHEMA_VERSION,
@@ -22,6 +23,7 @@ import {
2223
import type { AuthenticatedSocket } from '@/middleware/auth'
2324
import { beginRoomPermissionRead, commitRoomPermission } from '@/middleware/permissions'
2425
import { MemoryRoomManager } from '@/rooms'
26+
import { createHttpHandler } from '@/routes/http'
2527

2628
const projectId = generateId()
2729
const fileId = generateId()
@@ -98,6 +100,7 @@ beforeAll(async () => {
98100
http = createServer()
99101
io = new Server(http)
100102
const manager = new MemoryRoomManager(io)
103+
http.on('request', createHttpHandler(manager, createLogger('OwnerProtocolFixture')))
101104
io.on('connection', (socket) => {
102105
const authed = socket as AuthenticatedSocket
103106
authed.userId = socket.handshake.auth.actor
@@ -109,7 +112,7 @@ beforeAll(async () => {
109112
socketUrl = await listen(http)
110113
})
111114

112-
async function join(actor: string, assertedProject = projectId) {
115+
async function join(actor: string, assertedProject = projectId, owner?: unknown) {
113116
const socket = connect(socketUrl, { transports: ['websocket'], auth: { actor } })
114117
clients.push(socket)
115118
await new Promise<void>((resolve) => socket.once('connect', resolve))
@@ -122,7 +125,7 @@ async function join(actor: string, assertedProject = projectId) {
122125
client.destroy()
123126
socket.emit(FILE_DOC_EVENTS.JOIN, {
124127
fileId,
125-
projectId: assertedProject,
128+
...(owner === undefined ? { projectId: assertedProject } : { owner }),
126129
clientId,
127130
schemaVersion: FILE_DOC_SCHEMA_VERSION,
128131
})
@@ -193,6 +196,44 @@ describe('Project documents across the Socket.IO and internal HTTP boundary', ()
193196
}
194197
)
195198

199+
check(
200+
'owner-shaped clients share the existing room and preserve editor attribution',
201+
async () => {
202+
const owner = { entityType: 'project', entityId: projectId }
203+
const writer = await join('writer', projectId, owner)
204+
expect(writer.joined.docId).toBe(docId)
205+
const edit = new Y.Doc()
206+
Y.applyUpdate(edit, Buffer.from(seed, 'base64'))
207+
edit.getText('body').insert(0, 'owner-shaped edit')
208+
const reply = await writer.socket.timeout(1500).emitWithAck(FILE_DOC_EVENTS.UPDATE, {
209+
fileId,
210+
owner,
211+
docId,
212+
updateId: generateId(),
213+
update: Y.encodeStateAsUpdate(edit),
214+
})
215+
expect(reply).toMatchObject({ status: 'accepted' })
216+
await flushAllFileDocRooms()
217+
expect(capturedActors.at(-1)).toBe('writer')
218+
edit.destroy()
219+
await expect(join('reader', projectId, { ...owner, entityId: generateId() })).rejects.toThrow(
220+
'NOT_FOUND'
221+
)
222+
await expect(
223+
join('reader', projectId, { entityType: 'organization', entityId: projectId })
224+
).rejects.toThrow('INVALID_PAYLOAD')
225+
const mismatched = await writer.socket.timeout(1500).emitWithAck(FILE_DOC_EVENTS.UPDATE, {
226+
fileId,
227+
owner,
228+
projectId: generateId(),
229+
docId,
230+
updateId: generateId(),
231+
update: Y.encodeStateAsUpdate(document),
232+
})
233+
expect(mismatched).toMatchObject({ status: 'rejected', code: 'INVALID_UPDATE' })
234+
}
235+
)
236+
196237
check('a downgraded writer remains a reader but cannot submit another update', async () => {
197238
const writer = await join('writer')
198239
actors.set('writer', 'read')
@@ -236,6 +277,52 @@ describe('Project documents across the Socket.IO and internal HTTP boundary', ()
236277
expect(restored.joined.docId).toBe(docId)
237278
}
238279
)
280+
281+
check(
282+
'HTTP owner targeting fences the same generation and refuses unsupported or conflicting owners',
283+
async () => {
284+
const owner = { entityType: 'project', entityId: projectId }
285+
async function post(action: string, body: object) {
286+
return fetch(`${socketUrl}/api/file-doc/${action}`, {
287+
method: 'POST',
288+
headers: { 'content-type': 'application/json', 'x-api-key': env.INTERNAL_API_SECRET },
289+
body: JSON.stringify(body),
290+
})
291+
}
292+
for (const action of ['apply-edit', 'invalidate', 'retire']) {
293+
const body = {
294+
fileId,
295+
owner,
296+
markdown: '',
297+
version: 100,
298+
retiredDocId: docId,
299+
replacementDocId: generateId(),
300+
}
301+
expect((await post(action, { ...body, projectId: generateId() })).status).toBe(400)
302+
expect(
303+
(
304+
await post(action, {
305+
...body,
306+
owner: { entityType: 'organization', entityId: projectId },
307+
})
308+
).status
309+
).toBe(400)
310+
}
311+
const joined = await join('reader', projectId, owner)
312+
const invalidation = new Promise<{ docId: string }>((resolve) =>
313+
joined.socket.once(FILE_DOC_EVENTS.INVALIDATED, resolve)
314+
)
315+
const response = await post('retire', {
316+
fileId,
317+
owner,
318+
retiredDocId: docId,
319+
replacementDocId: generateId(),
320+
})
321+
expect(response.status).toBe(200)
322+
expect(await response.json()).toEqual({ status: 'applied' })
323+
expect((await invalidation).docId).toBe(docId)
324+
}
325+
)
239326
})
240327

241328
afterAll(async () => {

‎apps/realtime/src/handlers/file-doc-store.test.ts‎

Lines changed: 47 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -461,7 +461,13 @@ describe('FileDocStore', () => {
461461
'replaced by a newer durable version'
462462
)
463463
await expect(
464-
store.publishClientUpdateAndWait(NAME, 'stale-update', updateFor('stale acknowledged write'))
464+
store.publishClientUpdateAndWait(
465+
NAME,
466+
'stale-update',
467+
updateFor('stale acknowledged write'),
468+
undefined,
469+
{ userId: 'editor', connectionId: 'socket-editor' }
470+
)
465471
).rejects.toThrow('replaced by a newer durable version')
466472

467473
const fresh = seedFor('fresh generation')
@@ -488,15 +494,24 @@ describe('FileDocStore', () => {
488494
await store.seedIfEmpty(NAME, seedFor('base'), 10)
489495
const generation = await store.getDocumentGeneration(NAME)
490496
const delta = updateFor('edit')
491-
await store.publishClientUpdateAndWait(NAME, 'accepted-update', delta, generation)
497+
await store.publishClientUpdateAndWait(NAME, 'accepted-update', delta, generation, {
498+
userId: 'editor',
499+
connectionId: 'socket-editor',
500+
})
492501
state.backing!.streams.delete(`filedoc:stream:${NAME}`)
493502

494503
await expect(store.publishAndWait(NAME, delta, generation)).rejects.toThrow('replaced')
495504
await expect(
496-
store.publishClientUpdateAndWait(NAME, 'new-update', delta, generation)
505+
store.publishClientUpdateAndWait(NAME, 'new-update', delta, generation, {
506+
userId: 'editor',
507+
connectionId: 'socket-editor',
508+
})
497509
).rejects.toThrow('replaced')
498510
await expect(
499-
store.publishClientUpdateAndWait(NAME, 'accepted-update', delta, generation)
511+
store.publishClientUpdateAndWait(NAME, 'accepted-update', delta, generation, {
512+
userId: 'editor',
513+
connectionId: 'socket-editor',
514+
})
500515
).rejects.toThrow('replaced')
501516
expect(state.backing!.streams.has(`filedoc:stream:${NAME}`)).toBe(false)
502517
})
@@ -753,8 +768,14 @@ describe('FileDocStore', () => {
753768
const store = await newStore()
754769
const update = updateFor('retry-safe')
755770

756-
await store.publishClientUpdateAndWait(NAME, 'update-1', update)
757-
await store.publishClientUpdateAndWait(NAME, 'update-1', update)
771+
await store.publishClientUpdateAndWait(NAME, 'update-1', update, undefined, {
772+
userId: 'editor',
773+
connectionId: 'socket-editor',
774+
})
775+
await store.publishClientUpdateAndWait(NAME, 'update-1', update, undefined, {
776+
userId: 'editor',
777+
connectionId: 'socket-editor',
778+
})
758779

759780
expect(state.backing!.streams.get(`filedoc:stream:${NAME}`)).toHaveLength(2)
760781
})
@@ -763,8 +784,14 @@ describe('FileDocStore', () => {
763784
seedLegacyStream()
764785
const store = await newStore()
765786

766-
await store.publishClientUpdateAndWait(NAME, 'update-1', updateFor('first'))
767-
await store.publishClientUpdateAndWait(NAME, 'update-1', updateFor('second'))
787+
await store.publishClientUpdateAndWait(NAME, 'update-1', updateFor('first'), undefined, {
788+
userId: 'editor',
789+
connectionId: 'socket-editor',
790+
})
791+
await store.publishClientUpdateAndWait(NAME, 'update-1', updateFor('second'), undefined, {
792+
userId: 'editor',
793+
connectionId: 'socket-editor',
794+
})
768795

769796
expect(state.backing!.streams.get(`filedoc:stream:${NAME}`)).toHaveLength(3)
770797
})
@@ -814,7 +841,10 @@ describe('FileDocStore', () => {
814841
state.backing!.onSnapshot = async () => {
815842
await replacer.invalidateDocument(NAME, 20)
816843
await replacer.seedIfEmpty(NAME, freshSeed, 20)
817-
await replacer.publishClientUpdateAndWait(NAME, 'fresh-edit', freshEdit, 'new-generation')
844+
await replacer.publishClientUpdateAndWait(NAME, 'fresh-edit', freshEdit, 'new-generation', {
845+
userId: 'editor',
846+
connectionId: 'socket-editor',
847+
})
818848
expect(state.backing!.streams.get(streamKey)?.map((entry) => entry.id)).toEqual([
819849
'1000-0',
820850
'1000-1',
@@ -827,7 +857,10 @@ describe('FileDocStore', () => {
827857
'1000-0',
828858
'1000-1',
829859
])
830-
await replacer.publishClientUpdateAndWait(NAME, 'fresh-edit', freshEdit, 'new-generation')
860+
await replacer.publishClientUpdateAndWait(NAME, 'fresh-edit', freshEdit, 'new-generation', {
861+
userId: 'editor',
862+
connectionId: 'socket-editor',
863+
})
831864
expect(state.backing!.streams.get(streamKey)).toHaveLength(2)
832865
const persisted = await replacer.getStreamState(NAME)
833866
expect(persisted).not.toBeNull()
@@ -846,7 +879,10 @@ describe('FileDocStore', () => {
846879

847880
await expect(store.attachRoom(NAME, doc)).rejects.toThrow('not initialized')
848881
await expect(
849-
store.publishClientUpdateAndWait(NAME, 'update-1', updateFor('x'))
882+
store.publishClientUpdateAndWait(NAME, 'update-1', updateFor('x'), undefined, {
883+
userId: 'editor',
884+
connectionId: 'socket-editor',
885+
})
850886
).rejects.toThrow('not initialized')
851887
await expect(store.seedIfEmpty(NAME, seedFor('seed'))).rejects.toThrow('not initialized')
852888
await expect(store.getStreamState(NAME)).rejects.toThrow('not initialized')

‎apps/realtime/src/handlers/file-doc-store.ts‎

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -608,11 +608,8 @@ export class FileDocStore {
608608
expectedGeneration = this.rooms.get(name)?.generation ?? '',
609609
editor?: FileDocEditor
610610
): Promise<void> {
611-
if (
612-
name.startsWith('project-file-doc:') &&
613-
(!editor?.userId.trim() || !editor.connectionId.trim())
614-
) {
615-
throw new Error('Project document updates require an authenticated editor')
611+
if (!editor?.userId.trim() || !editor.connectionId.trim()) {
612+
throw new Error('Document updates require an authenticated editor')
616613
}
617614
if (!this.enabled) return
618615
if (!this.write) throw new Error('FileDocStore is not initialized')

0 commit comments

Comments
 (0)