Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 11 additions & 5 deletions apps/realtime/src/access-revalidation.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import {
ACCESS_REVALIDATION_SWEEP_INTERVAL_MS,
startAccessRevalidationSweep,
} from '@/access-revalidation'
import { registerRoomPermissionPublisher } from '@/handlers/room-permission'
import type { IRoomManager, UserPresence } from '@/rooms'

interface FakeSocket {
Expand Down Expand Up @@ -131,6 +132,8 @@ describe('access-revalidation sweep', () => {
})

it('keeps a downgraded reader subscribed and reports the loss of write access', async () => {
const publish = vi.fn()
registerRoomPermissionPublisher('workspace-file-doc', publish)
const socket = makeSocket('sock-1', 'user-1', 'workspace-file-doc:file-1')
socket.rooms.add('table:t-1')
const manager = makeManager([socket])
Expand All @@ -139,20 +142,23 @@ describe('access-revalidation sweep', () => {
await sweep.runOnce()
sweep.stop()
expect(socket.leave).not.toHaveBeenCalled()
expect(socket.emit).toHaveBeenCalledWith('file-doc-permission', {
fileId: 'file-1',
canWrite: false,
})
expect(publish).toHaveBeenCalledWith(
socket,
{ type: 'workspace-file-doc', id: 'file-1' },
'read'
)
})

it('retains a document subscription for retry when authorization is unavailable', async () => {
const publish = vi.fn()
registerRoomPermissionPublisher('workspace-file-doc', publish)
const socket = makeSocket('sock-1', 'user-1', 'workspace-file-doc:file-1')
mockResolveRole.mockRejectedValue(new Error('database unavailable'))
const sweep = startAccessRevalidationSweep(makeManager([socket]))
await sweep.runOnce()
sweep.stop()
expect(socket.leave).not.toHaveBeenCalled()
expect(socket.emit).not.toHaveBeenCalledWith('file-doc-permission', expect.anything())
expect(publish).not.toHaveBeenCalled()
})

it('evicts only the revoked socket, not co-members of the room', async () => {
Expand Down
18 changes: 2 additions & 16 deletions apps/realtime/src/access-revalidation.ts
Original file line number Diff line number Diff line change
@@ -1,11 +1,9 @@
import { createLogger } from '@sim/logger'
import { ROOM_MEMBERSHIP_ACTIONS, satisfiesRoomMembership } from '@sim/platform-authz/room-policy'
import type { AccessRevokedBroadcast } from '@sim/realtime-protocol/events'
import { FILE_DOC_EVENTS, type FileDocPermission } from '@sim/realtime-protocol/file-doc'
import {
isProjectRoom,
parseRoomName,
projectFileDocTarget,
ROOM_TYPES,
type RoomRef,
type RoomType,
Expand All @@ -17,6 +15,7 @@ import {
runRoomEvictionHandler,
setEvictionCleanupSink,
} from '@/handlers/room-eviction'
import { publishRoomPermission } from '@/handlers/room-permission'
import type { AuthenticatedSocket } from '@/middleware/auth'
import { ROLE_REVALIDATION_TTL_MS, resolveCurrentRoomPermission } from '@/middleware/permissions'
import type { IRoomManager } from '@/rooms'
Expand Down Expand Up @@ -331,20 +330,7 @@ export function startAccessRevalidationSweep(roomManager: IRoomManager): AccessR
])
// {@link SCAN_TIMED_OUT} is the only symbol this race can yield; matching on
// the type narrows it out of the permission comparison below.
if (
typeof role !== 'symbol' &&
(room.type === ROOM_TYPES.PROJECT_FILE_DOC || room.type === ROOM_TYPES.WORKSPACE_FILE_DOC)
) {
const target =
room.type === ROOM_TYPES.PROJECT_FILE_DOC
? projectFileDocTarget(room)
: { fileId: room.id }
if (target)
socket.emit(FILE_DOC_EVENTS.PERMISSION, {
...target,
canWrite: role === 'write' || role === 'admin',
} satisfies FileDocPermission)
}
if (typeof role !== 'symbol') publishRoomPermission(socket, room, role)
if (typeof role === 'symbol') {
logger.warn(
`Authorization check timed out for user ${userId} on ${name}; skipping this pass`
Expand Down
51 changes: 49 additions & 2 deletions apps/realtime/src/handlers/file-doc.redis-store.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -20,17 +20,18 @@ import * as syncProtocol from 'y-protocols/sync'
import * as Y from 'yjs'
import type { IRoomManager } from '@/rooms'

const { mockAuthorizeRoom, mockFetchFileDocSeed } = vi.hoisted(() => ({
const { mockAuthorizeRoom, mockFetchFileDocSeed, mockFetchFileDocPersist } = vi.hoisted(() => ({
mockAuthorizeRoom: vi.fn(),
mockFetchFileDocSeed: vi.fn(),
mockFetchFileDocPersist: vi.fn(),
}))

vi.mock('@sim/platform-authz/rooms', () => ({ authorizeRoom: mockAuthorizeRoom }))

vi.mock('@/handlers/file-doc-app', () => ({
fetchFileDocSeed: mockFetchFileDocSeed,
fetchFileDocMerge: vi.fn(),
fetchFileDocPersist: vi.fn().mockResolvedValue({ status: 'persisted', version: 1 }),
fetchFileDocPersist: mockFetchFileDocPersist,
}))

/** One in-memory Redis backing per test — only the stream/lock ops the store actually uses. */
Expand Down Expand Up @@ -269,6 +270,7 @@ describe('file-doc join readiness (shared store enabled)', () => {
workspacePermission: 'write',
})
mockFetchFileDocSeed.mockResolvedValue(null)
mockFetchFileDocPersist.mockResolvedValue({ status: 'persisted', version: 2 })
})

afterEach(() => {
Expand Down Expand Up @@ -326,4 +328,49 @@ describe('file-doc join readiness (shared store enabled)', () => {
expect(mockFetchFileDocSeed).not.toHaveBeenCalled()
expect(statesDeliveredTo(socket)).toEqual(['seeded by the writer that held the lock'])
})

it('persists unattributed stream state under the local editor when the last collaborator leaves', async () => {
backing.kv.set(`filedoc:generation:${ROOM_NAME}`, 'doc-1')
backing.kv.set(`filedoc:syncver:${ROOM_NAME}`, '1')
const doc = new Y.Doc()
doc.getText(FIELD).insert(0, 'AAA')
doc.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.flag, true)
doc.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.docIdKey, 'doc-1')
backing.streams.set(STREAM_KEY, [
{
id: `${++backing.seq}-0`,
message: { u: Buffer.from(Y.encodeStateAsUpdate(doc)).toString('base64'), g: 'doc-1' },
},
])
const afterSeed = Y.encodeStateVector(doc)
doc.getText(FIELD).insert(0, 'BBB')
const edit = Y.encodeStateAsUpdate(doc, afterSeed)
doc.destroy()

const { handlers } = setup('socket-1', sockets)
await handlers[FILE_DOC_EVENTS.JOIN]({
fileId: FILE_ID,
clientId: 1,
schemaVersion: FILE_DOC_SCHEMA_VERSION,
})
const acknowledge = vi.fn()
await (handlers[FILE_DOC_EVENTS.UPDATE] as (...args: unknown[]) => Promise<void>)(
{ fileId: FILE_ID, docId: 'doc-1', updateId: 'update-1', update: edit },
acknowledge
)
await vi.waitFor(() =>
expect(acknowledge).toHaveBeenCalledWith({ status: 'accepted', updateId: 'update-1' })
)

cleanupFileDocForSocket('socket-1', createIo(sockets), true)
await vi.waitFor(() =>
expect(mockFetchFileDocPersist).toHaveBeenCalledWith(
'ws-1',
FILE_ID,
'user-1',
expect.any(Uint8Array),
1
)
)
})
})
88 changes: 88 additions & 0 deletions apps/realtime/src/handlers/file-doc.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ import {
setupWorkspaceFileDocHandlers,
} from '@/handlers/file-doc'
import { FileDocInvalidatedError, getFileDocStore } from '@/handlers/file-doc-store'
import { publishRoomPermission } from '@/handlers/room-permission'
import * as permissions from '@/middleware/permissions'
import {
beginRoomPermissionRead,
Expand Down Expand Up @@ -642,6 +643,93 @@ describe('setupWorkspaceFileDocHandlers', () => {
}
})

it('publishes write access to a socket only when it changes, not on every frame', async () => {
// A distinct user/file: role decisions recorded under fake timers outlive this test.
vi.useFakeTimers()
try {
const { io } = createIo()
const { socket, handlers } = setup('socket-capability', io, { userId: 'user-capability' })
await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-capability', clientId: 7 })
await vi.advanceTimersByTimeAsync(0)
const presence = awarenessFrame(7, 'Capability').frame
const sendFrames = async () => {
for (let sent = 0; sent < 3; sent++) {
void handlers[FILE_DOC_EVENTS.MESSAGE](presence)
await vi.advanceTimersByTimeAsync(0)
}
}
const published = () =>
socket.emit.mock.calls
.filter(([event]) => event === FILE_DOC_EVENTS.PERMISSION)
.map(([, payload]) => (payload as { canWrite: boolean }).canWrite)
const changeRole = async (workspacePermission: 'read' | 'write') => {
mockAuthorizeRoom.mockResolvedValue({
allowed: true,
status: 200,
workspaceId: 'ws-1',
workspacePermission,
})
await vi.advanceTimersByTimeAsync(ROLE_REVALIDATION_TTL_MS + 1_000)
}

await sendFrames()
expect(published()).toEqual([true])

await changeRole('read')
await sendFrames()
expect(published()).toEqual([true, false])

await changeRole('write')
await sendFrames()
expect(published()).toEqual([true, false, true])
} finally {
vi.useRealTimers()
}
})

it('tells an admitted writer it can write even if its cached role lapses before the reply', async () => {
// A distinct user/file: role decisions recorded under fake timers outlive this test.
vi.useFakeTimers({ toFake: ['Date'] })
try {
const { io } = createIo()
const data: Record<string, unknown> = {}
// Registration records presence just before the reply; expire the role cache right there.
Object.defineProperty(data, 'fileDocPresence', {
set: () => vi.setSystemTime(Date.now() + ROLE_REVALIDATION_TTL_MS + 1),
})
const { socket, handlers } = setup('socket-lapse', io, { userId: 'user-lapse', data })
await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-lapse', clientId: 9 })

expect(socket.emit).toHaveBeenCalledWith(
FILE_DOC_EVENTS.JOIN_SUCCESS,
expect.objectContaining({ fileId: 'file-lapse', canWrite: true })
)
} finally {
vi.useRealTimers()
}
})

it('publishes a re-validated permission only when it changes what the socket last heard', async () => {
const { io } = createIo()
const { socket, handlers } = setup('socket-sweep', io)
await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 8 })
await handlers[FILE_DOC_EVENTS.MESSAGE](awarenessFrame(8, 'Sweep').frame)
const room = { type: ROOM_TYPES.WORKSPACE_FILE_DOC, id: 'file-1' } as const
const subscriber = socket as unknown as Parameters<typeof publishRoomPermission>[0]
const published = () =>
socket.emit.mock.calls
.filter(([event]) => event === FILE_DOC_EVENTS.PERMISSION)
.map(([, payload]) => (payload as { canWrite: boolean }).canWrite)

publishRoomPermission(subscriber, room, 'write')
publishRoomPermission(subscriber, room, 'write')
expect(published()).toEqual([true])

publishRoomPermission(subscriber, room, 'read')
publishRoomPermission(subscriber, room, 'read')
expect(published()).toEqual([true, false])
})

it('applies + fans out an agent-streamed frame (SYNC_NO_PERSIST) but never persists it', async () => {
mockFetchFileDocSeed.mockResolvedValue(seedResult('# From server'))
const { io, sent } = createIo()
Expand Down
Loading
Loading