diff --git a/modules/database/README.mdx b/modules/database/README.mdx index b84cded13..10703937b 100644 --- a/modules/database/README.mdx +++ b/modules/database/README.mdx @@ -64,7 +64,7 @@ Events arrive as `change` with `{ version, operation, schema, documentId, occurr The leader also publishes `database:change:${schema}` on the Redis bus. **Do not also relay `database:change:*` on `/events/` if the same client is on `/database/`** — that duplicates notifications. Keep the bus for other modules; just do not dual-subscribe. -Client subscribers must authenticate. Schemas with document-level authorization reject schema-wide subscriptions and require a document ID plus a `read` check. Client sockets also require CMS `crudOperations.read.enabled`. Admin consumers use `POST /realtime/ticket` for a 30-second handshake token; that token cannot mint another ticket or call REST/GraphQL. Session JWTs and masterkeys must not be sent from browser code. +Client subscribers must authenticate. Schemas with document-level authorization reject schema-wide subscriptions and require a document ID plus a `read` check. Client sockets also require CMS `crudOperations.read.enabled` (checked again at emit: deny skips client delivery and keeps membership; authorization UNAVAILABLE keeps membership without emitting). Admin consumers use `POST /realtime/ticket` for a 30-second handshake token; that token cannot mint another ticket or call REST/GraphQL. Session JWTs and masterkeys must not be sent from browser code. Admin sockets must be enabled (`admin.transports.sockets`) and the Admin socket port (`ADMIN_SOCKET_PORT`, default 3031) reachable from the UI. diff --git a/modules/database/src/realtime/MongoChangeStreamCoordinator.ts b/modules/database/src/realtime/MongoChangeStreamCoordinator.ts index d29938d15..45644fa78 100644 --- a/modules/database/src/realtime/MongoChangeStreamCoordinator.ts +++ b/modules/database/src/realtime/MongoChangeStreamCoordinator.ts @@ -66,7 +66,9 @@ export class MongoChangeStreamCoordinator { private retryAttempt = 0; private watching = false; private opening = false; + private acquiring = false; private ignoreClose = false; + private lockGeneration = 0; private changeQueue: Promise = Promise.resolve(); private watchedCollectionsKey = ''; private readonly rebacCache = new RealtimeRebacCache(); @@ -150,7 +152,15 @@ export class MongoChangeStreamCoordinator { } return; } + if (this.acquiring) return; + this.acquiring = true; try { + if (this.lock) { + if (!this.watching) { + await this.openStream(); + } + return; + } const acquired = await this.options.grpcSdk.state!.tryAcquireLock( LEADER_LOCK, LOCK_TTL_MS, @@ -160,13 +170,39 @@ export class MongoChangeStreamCoordinator { this.scheduleRetry(); return; } - this.lock = acquired; + if (this.lock) { + try { + await this.options.grpcSdk.state!.releaseLock(acquired); + } catch { + // lock may already have expired + } + if (!this.watching) { + await this.openStream(); + } + return; + } + try { + this.lock = await acquired.extend(LOCK_TTL_MS); + } catch { + try { + await this.options.grpcSdk.state!.releaseLock(acquired); + } catch { + // lock may already have expired + } + this.lock = null; + this.streamState = 'idle'; + this.scheduleRetry(); + return; + } + this.bumpLockGeneration(); this.startRenewal(); await this.openStream(); } catch (err) { this.lastError = err instanceof Error ? err.message : String(err); this.streamState = 'degraded'; this.scheduleRetry(); + } finally { + this.acquiring = false; } } @@ -179,40 +215,59 @@ export class MongoChangeStreamCoordinator { private async renewLock() { if (!this.lock) return; + const generation = this.lockGeneration; try { this.lock = await this.lock.extend(LOCK_TTL_MS); } catch { - this.lock = null; - await this.stopStream('idle'); + if (this.lockGeneration !== generation) return; + await this.fenceLock('idle'); this.scheduleRetry(); } } private async openStream() { - if (this.watching || this.closed || this.opening) return; + if (this.watching || this.closed || this.opening || !this.lock) return; this.opening = true; this.streamState = 'starting'; this.ignoreClose = false; + const generation = this.lockGeneration; try { const resumeAfter = parseResumeToken( await this.options.grpcSdk.state!.getKey(RESUME_TOKEN_KEY), ); - if (this.watching || this.closed) return; + if ( + this.watching || + this.closed || + !this.lock || + generation !== this.lockGeneration + ) { + return; + } const collections = this.collectionNames(); const pipeline = buildWatchPipeline(collections); this.watchedCollectionsKey = optedInCollectionsKey(collections); const stream = this.options.watch({ resumeAfter, pipeline }); + if (generation !== this.lockGeneration || this.closed) { + try { + await stream.close(); + } catch { + // already closed + } + return; + } this.stream = stream; this.watching = true; this.streamState = 'live'; this.retryAttempt = 0; stream.on('change', (change: unknown) => { - this.enqueueChange(change as RawChangeEvent); + this.enqueueChange(change as RawChangeEvent, generation); }); stream.on('error', (err: unknown) => { + if (generation !== this.lockGeneration) return; void this.handleStreamError(err); }); stream.on('close', () => { + if (generation !== this.lockGeneration) return; this.watching = false; if (!this.closed && !this.ignoreClose && this.lock) { this.scheduleRetry(); @@ -226,12 +281,13 @@ export class MongoChangeStreamCoordinator { } } - private enqueueChange(change: RawChangeEvent) { + private enqueueChange(change: RawChangeEvent, generation: number) { this.changeQueue = this.changeQueue.then(async () => { - if (this.closed || !this.watching) return; + if (this.closed || !this.watching || generation !== this.lockGeneration) return; try { - await this.handleChange(change); + await this.handleChange(change, generation); } catch (err) { + if (generation !== this.lockGeneration) return; this.lastError = err instanceof Error ? err.message : String(err); ConduitGrpcSdk.Logger.error(err as Error); this.watching = false; @@ -241,12 +297,13 @@ export class MongoChangeStreamCoordinator { }); } - private async handleChange(change: RawChangeEvent) { + private async handleChange(change: RawChangeEvent, generation: number) { + if (generation !== this.lockGeneration) return; const token = serializeResumeToken(change._id); const schema = this.resolveSchema(change.ns?.coll); const event = schema ? normalizeChangeEvent(change, schema.name) : null; if (!event || !schema) { - if (token) { + if (token && generation === this.lockGeneration) { await this.persistResumeToken(token); } if (change.operationType && WATCH_RESTART_OPERATIONS.has(change.operationType)) { @@ -283,6 +340,9 @@ export class MongoChangeStreamCoordinator { ) { const adminRooms = roomsForPublicChange(schema.name, event.documentId); await this.safePush('admin', adminRooms, payload); + if (!schema.cmsReadEnabled) { + return; + } if (!schema.authorizationEnabled) { await this.safePush('router', adminRooms, payload); return; @@ -381,14 +441,25 @@ export class MongoChangeStreamCoordinator { } private async releaseLeader() { + await this.fenceLock(this.streamState); + } + + private async fenceLock(nextState: RealtimeStatusCode) { + this.bumpLockGeneration(); this.clearRenewTimer(); - if (!this.lock) return; + const lock = this.lock; + this.lock = null; + await this.stopStream(nextState); + if (!lock) return; try { - await this.options.grpcSdk.state!.releaseLock(this.lock); + await this.options.grpcSdk.state!.releaseLock(lock); } catch { // lock may already have expired } - this.lock = null; + } + + private bumpLockGeneration() { + this.lockGeneration += 1; } private clearTimers() { diff --git a/modules/database/src/realtime/__tests__/coordinator.test.ts b/modules/database/src/realtime/__tests__/coordinator.test.ts index cda5dc0e1..3720d5a41 100644 --- a/modules/database/src/realtime/__tests__/coordinator.test.ts +++ b/modules/database/src/realtime/__tests__/coordinator.test.ts @@ -7,6 +7,7 @@ import { roomsForPublicChange } from '../rooms.js'; class MemoryStore { private sets = new Map>(); + readonly ttls = new Map(); async sadd(key: string, ...members: string[]) { const set = this.sets.get(key) ?? new Set(); members.forEach(member => set.add(member)); @@ -26,15 +27,29 @@ class MemoryStore { return this.sets.get(key)?.size ?? 0; } async del(...keys: string[]) { - keys.forEach(key => this.sets.delete(key)); + keys.forEach(key => { + this.sets.delete(key); + this.ttls.delete(key); + }); return keys.length; } + async expire(key: string, seconds: number) { + this.ttls.set(key, seconds); + } + async persist(key: string) { + this.ttls.delete(key); + } } function createCoordinator(overrides?: { allow?: boolean; authorizationAvailable?: boolean; - schemas?: { name: string; collectionName: string; authorizationEnabled: boolean }[]; + schemas?: { + name: string; + collectionName: string; + authorizationEnabled: boolean; + cmsReadEnabled?: boolean; + }[]; getKeyDelayMs?: number; }) { const stream = new EventEmitter() as EventEmitter & { close: () => Promise }; @@ -84,9 +99,14 @@ function createCoordinator(overrides?: { watch, hello: async () => ({ setName: 'rs0' }), getOptedInSchemas: () => - overrides?.schemas ?? [ - { name: 'Order', collectionName: 'orders', authorizationEnabled: false }, - ], + ( + overrides?.schemas ?? [ + { name: 'Order', collectionName: 'orders', authorizationEnabled: false }, + ] + ).map(schema => ({ + cmsReadEnabled: true, + ...schema, + })), subscriptions, enabled: () => true, engine: () => 'MongoDB', @@ -358,7 +378,7 @@ describe('MongoChangeStreamCoordinator', () => { jest.useRealTimers(); }); - it.each(['drop', 'rename', 'invalidate'] as const)( + it.each(['drop', 'rename', 'invalidate', 'dropDatabase'] as const)( 'reopens the watch on %s', async operationType => { jest.useFakeTimers(); @@ -419,4 +439,67 @@ describe('MongoChangeStreamCoordinator', () => { ]); await coordinator.shutdown(); }); + + it('does not open a watch when the lock cannot be extended after acquire', async () => { + const { coordinator, watch, lock } = createCoordinator(); + lock.extend.mockRejectedValueOnce(new Error('extend failed')); + await coordinator.reconcile(); + expect(watch).not.toHaveBeenCalled(); + expect(coordinator.getState()).toBe('idle'); + await coordinator.shutdown(); + }); + + it('ignores draining watch events after lock renew failure', async () => { + jest.useFakeTimers(); + const { coordinator, stream, lock, adminPush, state } = createCoordinator(); + lock.extend.mockResolvedValueOnce(lock).mockRejectedValueOnce(new Error('lost lock')); + await coordinator.reconcile(); + expect(coordinator.getState()).toBe('live'); + await jest.advanceTimersByTimeAsync(5_000); + expect(coordinator.getState()).toBe('idle'); + stream.emit( + 'change', + insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c', { _data: 'stale' }), + ); + await coordinator.waitForIdle(); + expect(adminPush).not.toHaveBeenCalled(); + expect(state.get('realtime:resumeToken')).toBeUndefined(); + await coordinator.shutdown(); + jest.useRealTimers(); + }); + + it('does not emit to clients when CMS read is denied and keeps membership', async () => { + const can = jest.fn(async () => ({ allow: true })); + const { coordinator, stream, routerPush, adminPush, subscriptions, grpcSdk } = + createCoordinator({ + schemas: [ + { + name: 'Order', + collectionName: 'orders', + authorizationEnabled: true, + cmsReadEnabled: false, + }, + ], + }); + grpcSdk.authorization = { can }; + await subscriptions.addAuthorizedDocument( + 'sock-1', + 'Order', + '64b64c4c4c4c4c4c4c4c4c4c', + 'user-1', + ); + await coordinator.reconcile(); + stream.emit( + 'change', + insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c', { _data: 'token' }), + ); + await coordinator.waitForIdle(); + expect(adminPush).toHaveBeenCalledTimes(1); + expect(routerPush).not.toHaveBeenCalled(); + expect(can).not.toHaveBeenCalled(); + expect(await subscriptions.listUsers('Order', '64b64c4c4c4c4c4c4c4c4c4c')).toEqual([ + 'user-1', + ]); + await coordinator.shutdown(); + }); }); diff --git a/modules/database/src/realtime/__tests__/recovery.test.ts b/modules/database/src/realtime/__tests__/recovery.test.ts index eee45b918..4e5aaee37 100644 --- a/modules/database/src/realtime/__tests__/recovery.test.ts +++ b/modules/database/src/realtime/__tests__/recovery.test.ts @@ -1,12 +1,15 @@ import { describe, expect, it } from '@jest/globals'; -import { canReadDocument } from '../authorize.js'; import { authorizedDocumentRoom } from '../rooms.js'; import { isRecoverableDisconnect, restoreAuthorizedSubscriptions } from '../recovery.js'; import { createSocketHandlers } from '../sockets.js'; -import { RealtimeSubscriptionTracker } from '../subscriptions.js'; +import { + RealtimeSubscriptionTracker, + RECOVERY_REDIS_TTL_SECONDS, +} from '../subscriptions.js'; class MemoryStore { private sets = new Map>(); + readonly ttls = new Map(); async sadd(key: string, ...members: string[]) { const set = this.sets.get(key) ?? new Set(); members.forEach(member => set.add(member)); @@ -22,7 +25,16 @@ class MemoryStore { return this.sets.get(key)?.size ?? 0; } async del(...keys: string[]) { - keys.forEach(key => this.sets.delete(key)); + keys.forEach(key => { + this.sets.delete(key); + this.ttls.delete(key); + }); + } + async expire(key: string, seconds: number) { + this.ttls.set(key, seconds); + } + async persist(key: string) { + this.ttls.delete(key); } } @@ -48,7 +60,6 @@ describe('database socket recovery', () => { isAvailable: () => true, authorization: { can: async () => ({ allow: true }) }, }, - canRead: canReadDocument, }); expect(await tracker.listUsers('Order', 'doc-1')).toEqual(['user-1']); }); @@ -79,4 +90,47 @@ describe('database socket recovery', () => { } as never); expect(await tracker.listUsers('Order', 'doc-1')).toEqual(['user-1']); }); + + it('keeps membership when recovered authorization is unavailable', async () => { + const tracker = new RealtimeSubscriptionTracker(new MemoryStore()); + const room = authorizedDocumentRoom('Order', 'doc-1', 'user-1'); + await tracker.addAuthorizedDocument('sock-1', 'Order', 'doc-1', 'user-1'); + const { leaveRooms } = await restoreAuthorizedSubscriptions({ + socketId: 'sock-1', + rooms: [room], + contextSubs: [], + subscriptions: tracker, + grpcSdk: { + isAvailable: () => false, + }, + }); + expect(leaveRooms).toEqual([]); + expect(await tracker.listUsers('Order', 'doc-1')).toEqual(['user-1']); + }); + + it('expires Redis membership when a recoverable disconnect never recovers', async () => { + const store = new MemoryStore(); + const tracker = new RealtimeSubscriptionTracker(store); + const handlers = createSocketHandlers({ + mode: 'client', + grpcSdk: { + isAvailable: () => true, + authorization: { can: async () => ({ allow: true }) }, + } as never, + schemaLookup: { getSchema: () => undefined }, + subscriptions: tracker, + isGloballyEnabled: () => true, + }); + await tracker.addAuthorizedDocument('sock-1', 'Order', 'doc-1', 'user-1'); + expect(store.ttls.size).toBe(0); + await handlers.disconnect({ + request: { socketId: 'sock-1', params: ['transport close'] }, + } as never); + expect(await tracker.listUsers('Order', 'doc-1')).toEqual(['user-1']); + expect(store.ttls.get('realtime:socket:sock-1')).toBe(RECOVERY_REDIS_TTL_SECONDS); + expect(store.ttls.get('realtime:doc:Order:doc-1')).toBe(RECOVERY_REDIS_TTL_SECONDS); + expect(store.ttls.get('realtime:userdoc:Order:doc-1:user-1')).toBe( + RECOVERY_REDIS_TTL_SECONDS, + ); + }); }); diff --git a/modules/database/src/realtime/__tests__/subscriptions.test.ts b/modules/database/src/realtime/__tests__/subscriptions.test.ts index 45ee7d1c9..13bcc7298 100644 --- a/modules/database/src/realtime/__tests__/subscriptions.test.ts +++ b/modules/database/src/realtime/__tests__/subscriptions.test.ts @@ -1,8 +1,12 @@ import { describe, expect, it } from '@jest/globals'; -import { RealtimeSubscriptionTracker } from '../subscriptions.js'; +import { + RealtimeSubscriptionTracker, + RECOVERY_REDIS_TTL_SECONDS, +} from '../subscriptions.js'; class MemoryStore { private sets = new Map>(); + readonly ttls = new Map(); async sadd(key: string, ...members: string[]) { const set = this.sets.get(key) ?? new Set(); @@ -27,9 +31,20 @@ class MemoryStore { } async del(...keys: string[]) { - keys.forEach(key => this.sets.delete(key)); + keys.forEach(key => { + this.sets.delete(key); + this.ttls.delete(key); + }); return keys.length; } + + async expire(key: string, seconds: number) { + this.ttls.set(key, seconds); + } + + async persist(key: string) { + this.ttls.delete(key); + } } describe('RealtimeSubscriptionTracker', () => { @@ -46,10 +61,36 @@ describe('RealtimeSubscriptionTracker', () => { expect(await tracker.listUsers('Order', 'doc-1')).toEqual([]); }); + it('does not expire shared document keys while another socket is still live', async () => { + const store = new MemoryStore(); + const tracker = new RealtimeSubscriptionTracker(store); + await tracker.addAuthorizedDocument('s1', 'Order', 'doc-1', 'user-1'); + await tracker.addAuthorizedDocument('s2', 'Order', 'doc-1', 'user-1'); + await tracker.armRecoverableTtl('s1'); + expect(store.ttls.get('realtime:socket:s1')).toBe(RECOVERY_REDIS_TTL_SECONDS); + expect(store.ttls.has('realtime:doc:Order:doc-1')).toBe(false); + expect(store.ttls.has('realtime:userdoc:Order:doc-1:user-1')).toBe(false); + }); + it('removes revoked users from the document set', async () => { const tracker = new RealtimeSubscriptionTracker(new MemoryStore()); await tracker.addAuthorizedDocument('s1', 'Order', 'doc-1', 'user-1'); await tracker.removeUser('Order', 'doc-1', 'user-1'); expect(await tracker.listUsers('Order', 'doc-1')).toEqual([]); }); + + it('arms Redis TTL on recoverable disconnect keys and persists on restore', async () => { + const store = new MemoryStore(); + const tracker = new RealtimeSubscriptionTracker(store); + await tracker.addAuthorizedDocument('s1', 'Order', 'doc-1', 'user-1'); + expect(store.ttls.size).toBe(0); + await tracker.armRecoverableTtl('s1'); + expect(store.ttls.get('realtime:socket:s1')).toBe(RECOVERY_REDIS_TTL_SECONDS); + expect(store.ttls.get('realtime:doc:Order:doc-1')).toBe(RECOVERY_REDIS_TTL_SECONDS); + expect(store.ttls.get('realtime:userdoc:Order:doc-1:user-1')).toBe( + RECOVERY_REDIS_TTL_SECONDS, + ); + await tracker.addAuthorizedDocument('s1', 'Order', 'doc-1', 'user-1'); + expect(store.ttls.size).toBe(0); + }); }); diff --git a/modules/database/src/realtime/authorize.ts b/modules/database/src/realtime/authorize.ts index cf7ed899c..d2a16e1cf 100644 --- a/modules/database/src/realtime/authorize.ts +++ b/modules/database/src/realtime/authorize.ts @@ -1,6 +1,6 @@ import { status } from '@grpc/grpc-js'; import { GrpcError } from '@conduitplatform/grpc-sdk'; -import type { OptedInSchema, SubscribeRequest } from './types.js'; +import type { OptedInSchema, RebacDecision, SubscribeRequest } from './types.js'; export class RealtimeSubscriptionError extends GrpcError { constructor(code: number, message: string) { @@ -61,14 +61,14 @@ export function optionalDocumentId(value: unknown): string | undefined { return value; } -export async function canReadDocument( +export async function readDocumentDecision( grpcSdk: AuthorizationSdk, schema: string, documentId: string, userId: string, -): Promise { +): Promise { if (!grpcSdk.authorization || !grpcSdk.isAvailable('authorization')) { - return false; + return 'unavailable'; } try { const decision = await grpcSdk.authorization.can({ @@ -76,12 +76,21 @@ export async function canReadDocument( actions: ['read'], resource: `${schema}:${documentId}`, }); - return decision.allow === true; + return decision.allow === true ? 'allow' : 'deny'; } catch { - return false; + return 'unavailable'; } } +export async function canReadDocument( + grpcSdk: AuthorizationSdk, + schema: string, + documentId: string, + userId: string, +): Promise { + return (await readDocumentDecision(grpcSdk, schema, documentId, userId)) === 'allow'; +} + export function assertSchemaAvailable( lookup: SchemaLookup, schemaName: string, @@ -123,6 +132,7 @@ export function toOptedInSchema(schema: { modelOptions?: { conduit?: { realtime?: { enabled?: boolean }; + cms?: { crudOperations?: { read?: { enabled?: boolean } } }; authorization?: { enabled?: boolean }; }; }; @@ -132,5 +142,7 @@ export function toOptedInSchema(schema: { name: schema.name, collectionName: schema.collectionName, authorizationEnabled: schema.modelOptions.conduit.authorization?.enabled === true, + cmsReadEnabled: + schema.modelOptions.conduit.cms?.crudOperations?.read?.enabled === true, }; } diff --git a/modules/database/src/realtime/rebacCache.ts b/modules/database/src/realtime/rebacCache.ts index b33a9438c..c57595227 100644 --- a/modules/database/src/realtime/rebacCache.ts +++ b/modules/database/src/realtime/rebacCache.ts @@ -1,7 +1,6 @@ +import type { RebacDecision } from './types.js'; import type { AuthorizationSdk } from './authorize.js'; -export type RebacDecision = 'allow' | 'deny' | 'unavailable'; - const DEFAULT_TTL_MS = 12_000; const DEFAULT_MAX_ENTRIES = 10_000; diff --git a/modules/database/src/realtime/recovery.ts b/modules/database/src/realtime/recovery.ts index 3bc9c3868..7ff79735b 100644 --- a/modules/database/src/realtime/recovery.ts +++ b/modules/database/src/realtime/recovery.ts @@ -1,5 +1,5 @@ import type { Indexable, ParsedSocketRequest } from '@conduitplatform/grpc-sdk'; -import type { AuthorizationSdk } from './authorize.js'; +import { readDocumentDecision, type AuthorizationSdk } from './authorize.js'; import { authorizedDocumentRoom, parseAuthorizedDocumentRoom } from './rooms.js'; import type { RealtimeSubscriptionTracker } from './subscriptions.js'; @@ -93,12 +93,6 @@ export async function restoreAuthorizedSubscriptions(options: { contextSubs: AuthorizedSub[]; subscriptions: RealtimeSubscriptionTracker; grpcSdk: AuthorizationSdk; - canRead: ( - grpcSdk: AuthorizationSdk, - schema: string, - documentId: string, - userId: string, - ) => Promise; }): Promise<{ leaveRooms: string[] }> { const seen = new Set(); const subs: AuthorizedSub[] = []; @@ -119,14 +113,14 @@ export async function restoreAuthorizedSubscriptions(options: { const leaveRooms: string[] = []; for (const sub of subs) { - const allowed = await options.canRead( + const decision = await readDocumentDecision( options.grpcSdk, sub.schema, sub.documentId, sub.userId, ); const room = authorizedDocumentRoom(sub.schema, sub.documentId, sub.userId); - if (!allowed) { + if (decision === 'deny') { await options.subscriptions.removeAuthorizedDocument( options.socketId, sub.schema, diff --git a/modules/database/src/realtime/sockets.ts b/modules/database/src/realtime/sockets.ts index e2a80931c..3a06ab121 100644 --- a/modules/database/src/realtime/sockets.ts +++ b/modules/database/src/realtime/sockets.ts @@ -78,7 +78,9 @@ export function createSocketHandlers(options: RealtimeSocketOptions) { }, disconnect: async (call: ParsedSocketRequest): Promise => { const reason = call.request.params?.[0]; - if (!isRecoverableDisconnect(reason)) { + if (isRecoverableDisconnect(reason)) { + await options.subscriptions.armRecoverableTtl(call.request.socketId); + } else { await options.subscriptions.disconnect(call.request.socketId); } return { event: 'disconnected', data: { ok: true } }; @@ -90,7 +92,6 @@ export function createSocketHandlers(options: RealtimeSocketOptions) { contextSubs: authorizedSubsFromContext(call.request.context as Indexable), subscriptions: options.subscriptions, grpcSdk: options.grpcSdk as unknown as AuthorizationSdk, - canRead: canReadDocument, }); if (leaveRooms.length > 0) { return { event: 'leave-room', rooms: leaveRooms }; diff --git a/modules/database/src/realtime/subscriptions.ts b/modules/database/src/realtime/subscriptions.ts index da7bfb867..bc54e8efd 100644 --- a/modules/database/src/realtime/subscriptions.ts +++ b/modules/database/src/realtime/subscriptions.ts @@ -1,9 +1,13 @@ +export const RECOVERY_REDIS_TTL_SECONDS = 120; + export type SubscriptionStore = { sadd(key: string, ...members: string[]): Promise; srem(key: string, ...members: string[]): Promise; smembers(key: string): Promise; scard(key: string): Promise; del(...keys: string[]): Promise; + expire(key: string, seconds: number): Promise; + persist(key: string): Promise; }; function socketKey(socketId: string): string { @@ -31,12 +35,15 @@ export class RealtimeSubscriptionTracker { documentId: string, userId: string, ): Promise { - await this.store.sadd( - socketKey(socketId), - subscriptionRecord(schema, documentId, userId), - ); - await this.store.sadd(userDocSocketsKey(schema, documentId, userId), socketId); - await this.store.sadd(docUsersKey(schema, documentId), userId); + const socket = socketKey(socketId); + const userDoc = userDocSocketsKey(schema, documentId, userId); + const docUsers = docUsersKey(schema, documentId); + await this.store.sadd(socket, subscriptionRecord(schema, documentId, userId)); + await this.store.sadd(userDoc, socketId); + await this.store.sadd(docUsers, userId); + await this.store.persist(socket); + await this.store.persist(userDoc); + await this.store.persist(docUsers); } async removeAuthorizedDocument( @@ -89,4 +96,41 @@ export class RealtimeSubscriptionTracker { } await this.store.del(socketKey(socketId)); } + + async armRecoverableTtl( + socketId: string, + ttlSeconds: number = RECOVERY_REDIS_TTL_SECONDS, + ): Promise { + const socket = socketKey(socketId); + const records = await this.store.smembers(socket); + await this.store.expire(socket, ttlSeconds); + for (const record of records) { + try { + const parsed = JSON.parse(record) as { + schema: string; + documentId: string; + userId: string; + }; + const userDoc = userDocSocketsKey( + parsed.schema, + parsed.documentId, + parsed.userId, + ); + const others = (await this.store.smembers(userDoc)).filter(id => id !== socketId); + if (others.length > 0) continue; + await this.store.expire(userDoc, ttlSeconds); + const users = await this.store.smembers( + docUsersKey(parsed.schema, parsed.documentId), + ); + if (users.length <= 1) { + await this.store.expire( + docUsersKey(parsed.schema, parsed.documentId), + ttlSeconds, + ); + } + } catch { + // ignore malformed records + } + } + } } diff --git a/modules/database/src/realtime/types.ts b/modules/database/src/realtime/types.ts index 2b6e3e8c7..004523198 100644 --- a/modules/database/src/realtime/types.ts +++ b/modules/database/src/realtime/types.ts @@ -28,10 +28,13 @@ export type RealtimeStatus = { message?: string; }; +export type RebacDecision = 'allow' | 'deny' | 'unavailable'; + export type OptedInSchema = { name: string; collectionName: string; authorizationEnabled: boolean; + cmsReadEnabled: boolean; }; export type SubscribeRequest = {