11/**
22 * LEGACY-compat realtime hash commands over WS (RESPONSE / DATA_NOTIFICATION).
3- * Only `user:{userId }` Redis hashes are allowed for parity until group/ page ACL is wired via Postgres.
3+ * `user:{id }` allowed on the isolate; ` page:` / `group:` when `acl` resolves Postgres membership .
44 */
55import {
66 RealtimeCommandType ,
@@ -32,6 +32,75 @@ export function canRealtimeHashAccess(
3232 return false ;
3333}
3434
35+ export type RealtimeHashAclPort = {
36+ resolveBatch (
37+ needs : Map < string , { read : boolean ; write : boolean } > ,
38+ ) : Promise < Map < string , { readOk : boolean ; writeOk : boolean } > > ;
39+ } ;
40+
41+ function accumulateRealtimeHashNeeds (
42+ meta : { cmd : RealtimeClientCommand ; commandId : number } [ ] ,
43+ ) : Map < string , { read : boolean ; write : boolean } > {
44+ const needs = new Map < string , { read : boolean ; write : boolean } > ( ) ;
45+ function add ( key : string , read : boolean , write : boolean ) : void {
46+ const cur = needs . get ( key ) ?? { read : false , write : false } ;
47+ if ( read ) {
48+ cur . read = true ;
49+ }
50+ if ( write ) {
51+ cur . write = true ;
52+ }
53+ needs . set ( key , cur ) ;
54+ }
55+ for ( const { cmd } of meta ) {
56+ switch ( cmd . type ) {
57+ case RealtimeCommandType . HGET : {
58+ const t = parseTripleArgs ( cmd . args ) ;
59+ if ( t != null ) {
60+ add ( redisHashKey ( t [ 0 ] , t [ 1 ] ) , true , false ) ;
61+ }
62+ break ;
63+ }
64+ case RealtimeCommandType . SUBSCRIBE : {
65+ const t = parseTripleArgs ( cmd . args ) ;
66+ if ( t != null ) {
67+ add ( redisHashKey ( t [ 0 ] , t [ 1 ] ) , true , false ) ;
68+ }
69+ break ;
70+ }
71+ case RealtimeCommandType . HSET : {
72+ const t = parseHSetArgs ( cmd . args ) ;
73+ if ( t != null ) {
74+ add ( redisHashKey ( t . prefix , t . suffix ) , false , true ) ;
75+ }
76+ break ;
77+ }
78+ default :
79+ break ;
80+ }
81+ }
82+ return needs ;
83+ }
84+
85+ function syncResolveRealtimeHashAccess (
86+ userId : string ,
87+ needs : Map < string , { read : boolean ; write : boolean } > ,
88+ ) : Map < string , { readOk : boolean ; writeOk : boolean } > {
89+ const out = new Map < string , { readOk : boolean ; writeOk : boolean } > ( ) ;
90+ for ( const [ key , need ] of needs ) {
91+ const i = key . indexOf ( ":" ) ;
92+ const prefix = i > 0 ? key . slice ( 0 , i ) : "" ;
93+ const suffix = i > 0 ? key . slice ( i + 1 ) : "" ;
94+ const allowed =
95+ prefix !== "" && canRealtimeHashAccess ( userId , prefix , suffix ) ;
96+ out . set ( key , {
97+ readOk : ! need . read || allowed ,
98+ writeOk : ! need . write || allowed ,
99+ } ) ;
100+ }
101+ return out ;
102+ }
103+
35104export type RealtimeHashPort = {
36105 hmget ( key : string , fields : readonly string [ ] ) : Promise < ( unknown | null ) [ ] > ;
37106 hset ( key : string , entries : Record < string , unknown > ) : Promise < void > ;
@@ -85,13 +154,22 @@ export async function executeRealtimeWsBatch(input: {
85154 decoded : DecodedRealtimeClientRequest ;
86155 redis : RealtimeHashPort | null ;
87156 hooks : RealtimeBatchHooks ;
157+ acl : RealtimeHashAclPort | null ;
88158} ) : Promise < ExecuteRealtimeWsBatchResult > {
89- const { userId, decoded, redis, hooks } = input ;
159+ const { userId, decoded, redis, hooks, acl } = input ;
90160 const meta = decoded . commands . map ( ( cmd , i ) => ( {
91161 cmd,
92162 commandId : decoded . firstCommandId + i ,
93163 } ) ) ;
94164
165+ const needs = accumulateRealtimeHashNeeds ( meta ) ;
166+ const resolved =
167+ needs . size === 0
168+ ? new Map < string , { readOk : boolean ; writeOk : boolean } > ( )
169+ : acl != null
170+ ? await acl . resolveBatch ( needs )
171+ : syncResolveRealtimeHashAccess ( userId , needs ) ;
172+
95173 const hgetResponses = new Map < number , unknown > ( ) ;
96174
97175 type AllowedHGet = {
@@ -112,7 +190,7 @@ export async function executeRealtimeWsBatch(input: {
112190 continue ;
113191 }
114192 const [ prefix , suffix , field ] = t ;
115- if ( ! canRealtimeHashAccess ( userId , prefix , suffix ) ) {
193+ if ( ! resolved . get ( redisHashKey ( prefix , suffix ) ) ?. readOk ) {
116194 hgetResponses . set ( row . commandId , undefined ) ;
117195 continue ;
118196 }
@@ -168,7 +246,7 @@ export async function executeRealtimeWsBatch(input: {
168246 if ( t == null ) {
169247 return ;
170248 }
171- if ( ! canRealtimeHashAccess ( userId , t . prefix , t . suffix ) ) {
249+ if ( ! resolved . get ( redisHashKey ( t . prefix , t . suffix ) ) ?. writeOk ) {
172250 return ;
173251 }
174252 const fk = realtimeFullKey ( t . prefix , t . suffix , t . field ) ;
@@ -193,7 +271,7 @@ export async function executeRealtimeWsBatch(input: {
193271 }
194272 const [ prefix , suffix , field ] = t ;
195273 const fk = realtimeFullKey ( prefix , suffix , field ) ;
196- if ( ! canRealtimeHashAccess ( userId , prefix , suffix ) ) {
274+ if ( ! resolved . get ( redisHashKey ( prefix , suffix ) ) ?. readOk ) {
197275 subscribeItems . push ( { prefix, suffix, field, value : undefined } ) ;
198276 return ;
199277 }
0 commit comments