From 7b7fa496aa18c964225deab7ac5c8168655eee3a Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Sun, 5 Jul 2026 08:40:14 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=97=83=EF=B8=8F=20perf:=20Cache=20Group?= =?UTF-8?q?=20Memberships=20for=20ACL=20Principal=20Resolution=20(#14075)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * ๐Ÿ—ƒ๏ธ perf: Cache Group Memberships for ACL Principal Resolution Continues #14069: caches resolved group-membership ids per member key in a new USER_PRINCIPALS namespace so repeated permission checks skip the group query, with exact invalidation on every membership mutation, same-process build dedup, and optional Redis cross-container build locks. Co-authored-by: Peter Rothlaender Co-authored-by: Joachim Keltsch * ๐Ÿงต fix: Defer Transactional Invalidation and Gate No-Op Bulk Clears Codex round-2 findings on the principals cache: membership writes inside a transaction now defer cache invalidation until the session ends so concurrent readers cannot re-cache pre-commit state with no later correction, and bulkUpdateGroups skips invalidation entirely when MongoDB reports no modified or upserted documents. userGroup.spec.ts now runs on a single-node replica set so the deferral is covered by a real transaction. * ๐ŸŒ fix: Evict Cross-Process Stale Rewrites and Scope Entra Sync to Tenant Codex round-3 findings: a delayed second invalidation pass (lock wait plus a short grace) now evicts membership entries that a build in another container re-cached after the first delete, and syncUserEntraGroupMemberships establishes the user's tenant ALS context when the OAuth callback runs it pre-middleware, so tenant-scoped principal keys are invalidated (and sync queries and created groups are scoped) exactly like authenticated reads. Deferred transactional invalidations snapshot the caller's ALS so they keep tenant scoping. --------- Co-authored-by: Peter Rothlaender Co-authored-by: Joachim Keltsch --- .env.example | 10 + api/cache/getLogStores.js | 10 + api/server/services/PermissionService.js | 22 +- api/server/services/PermissionService.spec.js | 14 + packages/api/src/acl/accessControlService.ts | 6 +- packages/api/src/cache/index.ts | 1 + packages/api/src/cache/principals.ts | 61 ++ packages/data-provider/src/config.ts | 4 + packages/data-schemas/src/methods/index.ts | 5 +- packages/data-schemas/src/methods/role.ts | 7 +- .../src/methods/userGroup.spec.ts | 633 +++++++++++++++++- .../data-schemas/src/methods/userGroup.ts | 412 +++++++++++- packages/data-schemas/src/types/cache.ts | 16 + packages/data-schemas/src/types/index.ts | 1 + 14 files changed, 1173 insertions(+), 29 deletions(-) create mode 100644 packages/api/src/cache/principals.ts create mode 100644 packages/data-schemas/src/types/cache.ts diff --git a/.env.example b/.env.example index 3f6798fc6b..d29b6d0c85 100644 --- a/.env.example +++ b/.env.example @@ -994,6 +994,16 @@ HELP_AND_FAQ_URL=https://librechat.ai # Set to empty string to force all namespaces through Redis: FORCED_IN_MEMORY_CACHE_NAMESPACES= # FORCED_IN_MEMORY_CACHE_NAMESPACES=CONFIG_STORE,APP_CONFIG +# TTL in milliseconds for cached group memberships used in ACL permission checks (default: 300000 / 5 minutes; 0 disables) +# Membership changes invalidate affected entries immediately; the TTL bounds staleness from cross-process races. +# USER_PRINCIPALS_CACHE_TTL_MS=300000 +# Redis lock TTL in milliseconds for cross-container cache builds (default: 5000; 0 disables locking) +# Only used when the USER_PRINCIPALS namespace is Redis-backed; non-Redis deployments use in-process deduplication. +# USER_PRINCIPALS_LOCK_TTL_MS=5000 +# Maximum time in milliseconds to wait for another container holding the lock to fill the cache +# before falling back to a direct database read (default: USER_PRINCIPALS_LOCK_TTL_MS) +# USER_PRINCIPALS_LOCK_WAIT_MS=5000 + # Leader Election Configuration (for multi-instance deployments with Redis) # Duration in seconds that the leader lease is valid before it expires (default: 25) # LEADER_LEASE_DURATION=25 diff --git a/api/cache/getLogStores.js b/api/cache/getLogStores.js index 45a6a69994..848d22227c 100644 --- a/api/cache/getLogStores.js +++ b/api/cache/getLogStores.js @@ -7,9 +7,18 @@ const { sessionCache, standardCache, violationCache, + userPrincipalsCache, registerShutdownTask, } = require('@librechat/api'); +/** No-op store registered when the user principals cache is disabled (TTL of 0). */ +const disabledCache = { + get: async () => undefined, + set: async () => undefined, + delete: async () => undefined, + clear: async () => undefined, +}; + const namespaces = { [ViolationTypes.GENERAL]: new Keyv({ store: logFile, namespace: 'violations' }), [ViolationTypes.LOGINS]: violationCache(ViolationTypes.LOGINS), @@ -36,6 +45,7 @@ const namespaces = { [CacheKeys.SAML_SESSION]: sessionCache(CacheKeys.SAML_SESSION), [CacheKeys.ROLES]: standardCache(CacheKeys.ROLES), + [CacheKeys.USER_PRINCIPALS]: userPrincipalsCache() ?? disabledCache, [CacheKeys.APP_CONFIG]: standardCache(CacheKeys.APP_CONFIG), [CacheKeys.CONFIG_STORE]: standardCache(CacheKeys.CONFIG_STORE), [CacheKeys.TOOL_CACHE]: standardCache(CacheKeys.TOOL_CACHE), diff --git a/api/server/services/PermissionService.js b/api/server/services/PermissionService.js index 65336ab002..d798a4bd7b 100644 --- a/api/server/services/PermissionService.js +++ b/api/server/services/PermissionService.js @@ -1,6 +1,11 @@ const mongoose = require('mongoose'); const { isEnabled } = require('@librechat/api'); -const { getTransactionSupport, logger } = require('@librechat/data-schemas'); +const { + getTransactionSupport, + tenantStorage, + getTenantId, + logger, +} = require('@librechat/data-schemas'); const { ResourceType, PrincipalType, PrincipalModel } = require('librechat-data-provider'); const { entraIdPrincipalFeatureEnabled, @@ -501,6 +506,21 @@ const ensureGroupPrincipalExists = async function (principal, authContext = null * @returns {Promise} */ const syncUserEntraGroupMemberships = async (user, accessToken, session = null) => { + const tenantId = user?.tenantId ? String(user.tenantId) : undefined; + if (!tenantId || getTenantId() != null) { + return performEntraGroupMembershipSync(user, accessToken, session); + } + /** + * The OAuth callback runs before `tenantContextMiddleware`, so establish the + * user's tenant context here: group queries, created groups, and principal + * cache invalidation are then scoped exactly like authenticated reads. + */ + return tenantStorage.run({ tenantId, userId: user._id?.toString() }, async () => + performEntraGroupMembershipSync(user, accessToken, session), + ); +}; + +const performEntraGroupMembershipSync = async (user, accessToken, session = null) => { try { if (!entraIdPrincipalFeatureEnabled(user) || !accessToken || !user.idOnTheSource) { return; diff --git a/api/server/services/PermissionService.spec.js b/api/server/services/PermissionService.spec.js index d6f78414a8..4bf213db13 100644 --- a/api/server/services/PermissionService.spec.js +++ b/api/server/services/PermissionService.spec.js @@ -2060,6 +2060,20 @@ describe('syncUserEntraGroupMemberships - $pullAll on Group.memberIds', () => { expect(groups[2].memberIds).toContain(userEntraId); }); + it('establishes the user tenant context for the sync when tenantId is present', async () => { + const { getTenantId } = require('@librechat/data-schemas'); + const observed = []; + getUserEntraGroups.mockImplementation(async () => { + observed.push(getTenantId()); + return []; + }); + + await syncUserEntraGroupMemberships({ ...user, tenantId: 'tenant-42' }, 'fake-token'); + await syncUserEntraGroupMemberships(user, 'fake-token'); + + expect(observed).toEqual(['tenant-42', undefined]); + }); + it('should not modify groups when API returns empty list (early return)', async () => { await Group.create([ { diff --git a/packages/api/src/acl/accessControlService.ts b/packages/api/src/acl/accessControlService.ts index bc5e2a41b7..361bb70098 100644 --- a/packages/api/src/acl/accessControlService.ts +++ b/packages/api/src/acl/accessControlService.ts @@ -1,6 +1,7 @@ import { Types } from 'mongoose'; import { createMethods, logger } from '@librechat/data-schemas'; import { + CacheKeys, AccessRoleIds, PermissionBits, PrincipalType, @@ -10,12 +11,15 @@ import type { AllMethods, IAclEntry } from '@librechat/data-schemas'; import type { ClientSession, DeleteResult } from 'mongoose'; import type { ResolvedPrincipal } from '~/types/principal'; +import { userPrincipalsCache } from '~/cache'; export class AccessControlService { private _dbMethods: AllMethods; constructor(mongoose: typeof import('mongoose')) { - this._dbMethods = createMethods(mongoose); + this._dbMethods = createMethods(mongoose, { + getCache: (key) => (key === CacheKeys.USER_PRINCIPALS ? userPrincipalsCache() : undefined), + }); } /** diff --git a/packages/api/src/cache/index.ts b/packages/api/src/cache/index.ts index f3cfdbf0c5..ff0046b71f 100644 --- a/packages/api/src/cache/index.ts +++ b/packages/api/src/cache/index.ts @@ -3,4 +3,5 @@ export * from './redisClients'; export * from './keyvFiles'; export { default as keyvMongo } from './keyvMongo'; export * from './cacheFactory'; +export * from './principals'; export * from './redisUtils'; diff --git a/packages/api/src/cache/principals.ts b/packages/api/src/cache/principals.ts new file mode 100644 index 0000000000..ad0009d284 --- /dev/null +++ b/packages/api/src/cache/principals.ts @@ -0,0 +1,61 @@ +import { randomUUID } from 'crypto'; +import { Time, CacheKeys } from 'librechat-data-provider'; +import type { Keyv } from 'keyv'; +import { keyvRedisClient, ioredisClient } from './redisClients'; +import { standardCache } from './cacheFactory'; +import { cacheConfig } from './cacheConfig'; +import { math } from '~/utils'; + +const cacheTtl = math(process.env.USER_PRINCIPALS_CACHE_TTL_MS, Time.FIVE_MINUTES); +const lockTtl = math(process.env.USER_PRINCIPALS_LOCK_TTL_MS, 5000); +const lockWait = math(process.env.USER_PRINCIPALS_LOCK_WAIT_MS, lockTtl); + +export type UserPrincipalsCache = Keyv & { + lockWaitMs?: number; + acquireLock?: (key: string) => Promise; + releaseLock?: (key: string, token: string) => Promise; +}; + +const releaseLockScript = ` +if redis.call('GET', KEYS[1]) == ARGV[1] then + return redis.call('DEL', KEYS[1]) +end +return 0 +`; + +let memoizedCache: UserPrincipalsCache | undefined; + +/** + * Cache for resolved group memberships used by ACL principal resolution + * (`getUserPrincipals`). When the namespace is Redis-backed, lock helpers are + * attached so concurrent cold-key builds are deduplicated across containers. + * Returns undefined when disabled via USER_PRINCIPALS_CACHE_TTL_MS=0. + */ +export function userPrincipalsCache(): UserPrincipalsCache | undefined { + if (cacheTtl <= 0) { + return undefined; + } + if (memoizedCache) { + return memoizedCache; + } + + const cache: UserPrincipalsCache = standardCache(CacheKeys.USER_PRINCIPALS, cacheTtl); + const redisClient = ioredisClient; + const isRedisBacked = + keyvRedisClient != null && + !cacheConfig.FORCED_IN_MEMORY_CACHE_NAMESPACES?.includes(CacheKeys.USER_PRINCIPALS); + if (isRedisBacked && redisClient != null && lockTtl > 0) { + cache.lockWaitMs = Math.max(lockWait, 0); + cache.acquireLock = async (key) => { + const token = randomUUID(); + const result = await redisClient.set(key, token, 'PX', lockTtl, 'NX'); + return result === 'OK' ? token : null; + }; + cache.releaseLock = async (key, token) => { + await redisClient.eval(releaseLockScript, 1, key, token); + }; + } + + memoizedCache = cache; + return cache; +} diff --git a/packages/data-provider/src/config.ts b/packages/data-provider/src/config.ts index 543e703401..09ddd28d8f 100644 --- a/packages/data-provider/src/config.ts +++ b/packages/data-provider/src/config.ts @@ -2283,6 +2283,10 @@ export enum CacheKeys { * Key for the roles cache. */ ROLES = 'ROLES', + /** + * Key for cached group memberships used to resolve ACL user principals. + */ + USER_PRINCIPALS = 'USER_PRINCIPALS', /** * Key for the title generation cache. */ diff --git a/packages/data-schemas/src/methods/index.ts b/packages/data-schemas/src/methods/index.ts index 5eee6c4db5..dca2915379 100644 --- a/packages/data-schemas/src/methods/index.ts +++ b/packages/data-schemas/src/methods/index.ts @@ -17,7 +17,7 @@ import { createMCPServerMethods, type MCPServerMethods } from './mcpServer'; import { createPluginAuthMethods, type PluginAuthMethods } from './pluginAuth'; /* Permissions */ import { createAccessRoleMethods, type AccessRoleMethods } from './accessRole'; -import { createUserGroupMethods, type UserGroupMethods } from './userGroup'; +import { createUserGroupMethods, type UserGroupMethods, type UserGroupDeps } from './userGroup'; import { createAclEntryMethods, permissionBitSupersets, type AclEntryMethods } from './aclEntry'; import { createSystemGrantMethods, type SystemGrantMethods } from './systemGrant'; import { @@ -224,6 +224,7 @@ export function createMethods( // Role methods with optional cache injection const roleDeps: RoleDeps = { getCache: deps.getCache }; + const userGroupDeps: UserGroupDeps = { getCache: deps.getCache }; const roleMethods = createRoleMethods(mongoose, roleDeps); // Tier 1: action methods (created as variable for agent dependency) @@ -249,7 +250,7 @@ export function createMethods( ...createAgentApiKeyMethods(mongoose), ...createMCPServerMethods(mongoose), ...createAccessRoleMethods(mongoose), - ...createUserGroupMethods(mongoose), + ...createUserGroupMethods(mongoose, userGroupDeps), ...aclEntryMethods, ...systemGrantMethods, ...createAuditLogMethods(mongoose), diff --git a/packages/data-schemas/src/methods/role.ts b/packages/data-schemas/src/methods/role.ts index 1d7bf9cbbf..214aa185d6 100644 --- a/packages/data-schemas/src/methods/role.ts +++ b/packages/data-schemas/src/methods/role.ts @@ -6,7 +6,7 @@ import { removeNullishValues, } from 'librechat-data-provider'; import type { Model } from 'mongoose'; -import type { IRole, IUser } from '~/types'; +import type { CacheStore, IRole, IUser } from '~/types'; import { scopedCacheKey, getTenantId, runAsSystem, SYSTEM_TENANT_ID } from '~/config/tenantContext'; import { escapeRegExp } from '~/utils/string'; import logger from '~/config/winston'; @@ -27,10 +27,7 @@ export class RoleConflictError extends Error { export interface RoleDeps { /** Returns a cache store for the given key. Injected from getLogStores. */ - getCache?: (key: string) => { - get: (k: string) => Promise; - set: (k: string, v: unknown) => Promise; - }; + getCache?: (key: string) => CacheStore | undefined; } export function createRoleMethods( diff --git a/packages/data-schemas/src/methods/userGroup.spec.ts b/packages/data-schemas/src/methods/userGroup.spec.ts index 4f88042dea..0fcc849a44 100644 --- a/packages/data-schemas/src/methods/userGroup.spec.ts +++ b/packages/data-schemas/src/methods/userGroup.spec.ts @@ -1,7 +1,8 @@ import mongoose, { Types } from 'mongoose'; -import { MongoMemoryServer } from 'mongodb-memory-server'; +import { MongoMemoryReplSet } from 'mongodb-memory-server'; import { PrincipalType, SystemRoles } from 'librechat-data-provider'; import type * as t from '~/types'; +import { tenantStorage } from '~/config/tenantContext'; import { createUserGroupMethods } from './userGroup'; import groupSchema from '~/schema/group'; import userSchema from '~/schema/user'; @@ -13,14 +14,15 @@ jest.mock('~/config/winston', () => ({ debug: jest.fn(), })); -let mongoServer: MongoMemoryServer; +let mongoServer: MongoMemoryReplSet; let Group: mongoose.Model; let User: mongoose.Model; let Role: mongoose.Model; let methods: ReturnType; beforeAll(async () => { - mongoServer = await MongoMemoryServer.create(); + /** Single-node replica set so transactional invalidation deferral is tested for real */ + mongoServer = await MongoMemoryReplSet.create({ replSet: { count: 1 } }); Group = mongoose.models.Group || mongoose.model('Group', groupSchema); User = mongoose.models.User || mongoose.model('User', userSchema); Role = mongoose.models.Role || mongoose.model('Role', roleSchema); @@ -481,6 +483,631 @@ describe('userGroup methods', () => { }); }); + describe('getUserPrincipals caching', () => { + function createFakeCache() { + const store = new Map(); + return { + store, + get: jest.fn(async (key: string) => store.get(key)), + set: jest.fn(async (key: string, value: unknown) => { + store.set(key, value); + }), + delete: jest.fn(async (key: string) => store.delete(key)), + clear: jest.fn(async () => store.clear()), + }; + } + + function createCachedMethods(cache: ReturnType) { + return createUserGroupMethods(mongoose, { getCache: jest.fn(() => cache) }); + } + + const groupPrincipalIds = ( + principals: Array<{ principalType: PrincipalType; principalId?: string | Types.ObjectId }>, + ) => + principals + .filter((p) => p.principalType === PrincipalType.GROUP) + .map((p) => p.principalId!.toString()) + .sort(); + + it('serves repeat lookups from the cache without re-querying groups', async () => { + const user = await createTestUser({ idOnTheSource: 'cache-ext-1' }); + const group = await Group.create({ + name: 'Cached Team', + source: 'entra', + idOnTheSource: 'cache-grp-1', + memberIds: ['cache-ext-1'], + }); + const cache = createFakeCache(); + const cachedMethods = createCachedMethods(cache); + const params = { + userId: user._id.toString(), + role: SystemRoles.USER, + idOnTheSource: 'cache-ext-1', + }; + + const first = await cachedMethods.getUserPrincipals(params); + const findSpy = jest.spyOn(Group, 'find'); + const second = await cachedMethods.getUserPrincipals(params); + expect(findSpy).not.toHaveBeenCalled(); + findSpy.mockRestore(); + + expect(first).toEqual(second); + expect(groupPrincipalIds(second)).toEqual([group._id.toString()]); + expect(cache.set).toHaveBeenCalledTimes(1); + expect(cache.set).toHaveBeenCalledWith('cache-ext-1', [group._id.toString()]); + }); + + it('caches empty memberships and hydrates group ids as ObjectIds', async () => { + const user = await createTestUser({ idOnTheSource: 'cache-ext-empty' }); + const cache = createFakeCache(); + const cachedMethods = createCachedMethods(cache); + const params = { + userId: user._id.toString(), + role: SystemRoles.USER, + idOnTheSource: 'cache-ext-empty', + }; + + await cachedMethods.getUserPrincipals(params); + expect(cache.store.get('cache-ext-empty')).toEqual([]); + + const group = await Group.create({ + name: 'Late Team', + source: 'entra', + idOnTheSource: 'cache-grp-late', + memberIds: ['cache-ext-empty'], + }); + /** Still served from the cached empty entry until invalidation or expiry */ + const cached = await cachedMethods.getUserPrincipals(params); + expect(groupPrincipalIds(cached)).toEqual([]); + + cache.store.set('cache-ext-empty', [group._id.toString()]); + const hydrated = await cachedMethods.getUserPrincipals(params); + const groupPrincipal = hydrated.find((p) => p.principalType === PrincipalType.GROUP); + expect(groupPrincipal!.principalId).toBeInstanceOf(Types.ObjectId); + }); + + it('never caches role resolution', async () => { + const user = await createTestUser({ role: SystemRoles.USER }); + const cache = createFakeCache(); + const cachedMethods = createCachedMethods(cache); + const params = { userId: user._id.toString(), idOnTheSource: null }; + + const before = await cachedMethods.getUserPrincipals(params); + expect(before).toContainEqual({ + principalType: PrincipalType.ROLE, + principalId: SystemRoles.USER, + }); + + await User.updateOne({ _id: user._id }, { $set: { role: SystemRoles.ADMIN } }); + const after = await cachedMethods.getUserPrincipals(params); + expect(after).toContainEqual({ + principalType: PrincipalType.ROLE, + principalId: SystemRoles.ADMIN, + }); + expect(cache.set).toHaveBeenCalledTimes(1); + }); + + it('bypasses the cache for session-scoped reads', async () => { + const user = await createTestUser({ idOnTheSource: 'cache-ext-session' }); + const cache = createFakeCache(); + const cachedMethods = createCachedMethods(cache); + const session = await mongoose.startSession(); + + try { + await cachedMethods.getUserPrincipals( + { userId: user._id.toString(), role: SystemRoles.USER, idOnTheSource: null }, + session, + ); + } finally { + await session.endSession(); + } + + expect(cache.get).not.toHaveBeenCalled(); + expect(cache.set).not.toHaveBeenCalled(); + }); + + it('reflects membership changes immediately via invalidation', async () => { + const user = await createTestUser({ idOnTheSource: 'inv-ext-1' }); + const group = await Group.create({ + name: 'Invalidation Team', + source: 'entra', + idOnTheSource: 'inv-grp-1', + memberIds: [], + }); + const cache = createFakeCache(); + const cachedMethods = createCachedMethods(cache); + const params = { + userId: user._id.toString(), + role: SystemRoles.USER, + idOnTheSource: 'inv-ext-1', + }; + + expect(groupPrincipalIds(await cachedMethods.getUserPrincipals(params))).toEqual([]); + + await cachedMethods.addUserToGroup(user._id, group._id); + expect(groupPrincipalIds(await cachedMethods.getUserPrincipals(params))).toEqual([ + group._id.toString(), + ]); + + await cachedMethods.removeUserFromGroup(user._id, group._id); + expect(groupPrincipalIds(await cachedMethods.getUserPrincipals(params))).toEqual([]); + }); + + it('invalidates members when a group is created with members or deleted', async () => { + const user = await createTestUser({ idOnTheSource: 'inv-ext-2' }); + const cache = createFakeCache(); + const cachedMethods = createCachedMethods(cache); + const params = { + userId: user._id.toString(), + role: SystemRoles.USER, + idOnTheSource: 'inv-ext-2', + }; + + expect(groupPrincipalIds(await cachedMethods.getUserPrincipals(params))).toEqual([]); + + const group = await cachedMethods.createGroup({ + name: 'Created Team', + source: 'entra', + idOnTheSource: 'inv-grp-2', + memberIds: ['inv-ext-2'], + }); + expect(groupPrincipalIds(await cachedMethods.getUserPrincipals(params))).toEqual([ + group._id.toString(), + ]); + + await cachedMethods.deleteGroup(group._id); + expect(groupPrincipalIds(await cachedMethods.getUserPrincipals(params))).toEqual([]); + }); + + it('invalidates member keys extracted from bulk membership updates', async () => { + const user = await createTestUser({ idOnTheSource: 'bulk-ext-1' }); + const group = await Group.create({ + name: 'Bulk Team', + source: 'entra', + idOnTheSource: 'bulk-grp-1', + memberIds: ['bulk-ext-1'], + }); + const cache = createFakeCache(); + const cachedMethods = createCachedMethods(cache); + const params = { + userId: user._id.toString(), + role: SystemRoles.USER, + idOnTheSource: 'bulk-ext-1', + }; + + expect(groupPrincipalIds(await cachedMethods.getUserPrincipals(params))).toEqual([ + group._id.toString(), + ]); + + await cachedMethods.bulkUpdateGroups( + { _id: group._id }, + { $pullAll: { memberIds: ['bulk-ext-1'] } }, + ); + + expect(cache.delete).toHaveBeenCalledWith('bulk-ext-1'); + expect(cache.clear).not.toHaveBeenCalled(); + expect(groupPrincipalIds(await cachedMethods.getUserPrincipals(params))).toEqual([]); + }); + + it('clears the cache when a bulk update replaces memberIds wholesale', async () => { + const group = await Group.create({ + name: 'Wholesale Team', + source: 'entra', + idOnTheSource: 'bulk-grp-2', + memberIds: ['bulk-ext-old'], + }); + const cache = createFakeCache(); + const cachedMethods = createCachedMethods(cache); + + await cachedMethods.bulkUpdateGroups( + { _id: group._id }, + { $set: { memberIds: ['bulk-ext-new'] } }, + ); + + expect(cache.clear).toHaveBeenCalledTimes(1); + }); + + it('invalidates previous and new members when memberIds are replaced by id', async () => { + const user = await createTestUser({ idOnTheSource: 'upd-ext-old' }); + const group = await Group.create({ + name: 'Update Team', + source: 'entra', + idOnTheSource: 'upd-grp-1', + memberIds: ['upd-ext-old'], + }); + const cache = createFakeCache(); + const cachedMethods = createCachedMethods(cache); + const params = { + userId: user._id.toString(), + role: SystemRoles.USER, + idOnTheSource: 'upd-ext-old', + }; + + expect(groupPrincipalIds(await cachedMethods.getUserPrincipals(params))).toEqual([ + group._id.toString(), + ]); + + await cachedMethods.updateGroupById(group._id, { memberIds: ['upd-ext-new'] }); + + expect(cache.delete).toHaveBeenCalledWith('upd-ext-old'); + expect(cache.delete).toHaveBeenCalledWith('upd-ext-new'); + expect(groupPrincipalIds(await cachedMethods.getUserPrincipals(params))).toEqual([]); + }); + + it('ignores cached entries with invalid shapes and rebuilds them', async () => { + const user = await createTestUser({ idOnTheSource: 'shape-ext-1' }); + const group = await Group.create({ + name: 'Shape Team', + source: 'entra', + idOnTheSource: 'shape-grp-1', + memberIds: ['shape-ext-1'], + }); + const cache = createFakeCache(); + cache.store.set('shape-ext-1', ['not-an-object-id']); + const cachedMethods = createCachedMethods(cache); + + const principals = await cachedMethods.getUserPrincipals({ + userId: user._id.toString(), + role: SystemRoles.USER, + idOnTheSource: 'shape-ext-1', + }); + + expect(groupPrincipalIds(principals)).toEqual([group._id.toString()]); + expect(cache.set).toHaveBeenCalledWith('shape-ext-1', [group._id.toString()]); + }); + + it('deduplicates concurrent cache builds for the same member key', async () => { + const user = await createTestUser({ idOnTheSource: 'dedup-ext-1' }); + const cache = { + get: jest.fn(async () => { + await new Promise((resolve) => setTimeout(resolve, 10)); + return undefined; + }), + set: jest.fn(async () => undefined), + }; + const cachedMethods = createUserGroupMethods(mongoose, { getCache: jest.fn(() => cache) }); + const params = { + userId: user._id.toString(), + role: SystemRoles.USER, + idOnTheSource: 'dedup-ext-1', + }; + + const [first, second, third] = await Promise.all([ + cachedMethods.getUserPrincipals(params), + cachedMethods.getUserPrincipals(params), + cachedMethods.getUserPrincipals(params), + ]); + + expect(first).toEqual(second); + expect(second).toEqual(third); + expect(cache.get).toHaveBeenCalledTimes(3); + expect(cache.set).toHaveBeenCalledTimes(1); + }); + + it('shares one lock and DB build across concurrent same-process callers', async () => { + const user = await createTestUser({ idOnTheSource: 'lock-ext-1' }); + const cache = { + get: jest.fn(async () => { + await new Promise((resolve) => setTimeout(resolve, 10)); + return undefined; + }), + set: jest.fn(async () => undefined), + acquireLock: jest.fn(async () => 'lock-token'), + releaseLock: jest.fn(async () => undefined), + lockWaitMs: 5000, + }; + const cachedMethods = createUserGroupMethods(mongoose, { getCache: jest.fn(() => cache) }); + const params = { + userId: user._id.toString(), + role: SystemRoles.USER, + idOnTheSource: 'lock-ext-1', + }; + + const [first, second, third] = await Promise.all([ + cachedMethods.getUserPrincipals(params), + cachedMethods.getUserPrincipals(params), + cachedMethods.getUserPrincipals(params), + ]); + + expect(first).toEqual(second); + expect(second).toEqual(third); + expect(cache.acquireLock).toHaveBeenCalledTimes(1); + expect(cache.acquireLock).toHaveBeenCalledWith('USER_PRINCIPALS_LOCK:lock-ext-1'); + expect(cache.set).toHaveBeenCalledTimes(1); + expect(cache.releaseLock).toHaveBeenCalledTimes(1); + }); + + it('waits for a locked build from another process instead of querying', async () => { + const user = await createTestUser({ idOnTheSource: 'wait-ext-1' }); + const groupId = new Types.ObjectId(); + const cache = { + get: jest.fn().mockResolvedValueOnce(undefined).mockResolvedValue([groupId.toString()]), + set: jest.fn(async () => undefined), + acquireLock: jest.fn(async () => null), + releaseLock: jest.fn(async () => undefined), + lockWaitMs: 5000, + }; + const cachedMethods = createUserGroupMethods(mongoose, { getCache: jest.fn(() => cache) }); + + const principals = await cachedMethods.getUserPrincipals({ + userId: user._id.toString(), + role: SystemRoles.USER, + idOnTheSource: 'wait-ext-1', + }); + + expect(groupPrincipalIds(principals)).toEqual([groupId.toString()]); + expect(cache.acquireLock).toHaveBeenCalledTimes(1); + expect(cache.set).not.toHaveBeenCalled(); + expect(cache.releaseLock).not.toHaveBeenCalled(); + }); + + it('builds immediately when lock acquisition fails', async () => { + const user = await createTestUser({ idOnTheSource: 'fail-ext-1' }); + const cache = { + get: jest.fn(async () => undefined), + set: jest.fn(async () => undefined), + acquireLock: jest.fn(async () => { + throw new Error('redis unavailable'); + }), + releaseLock: jest.fn(async () => undefined), + lockWaitMs: 5000, + }; + const cachedMethods = createUserGroupMethods(mongoose, { getCache: jest.fn(() => cache) }); + + const principals = await cachedMethods.getUserPrincipals({ + userId: user._id.toString(), + idOnTheSource: 'fail-ext-1', + }); + + expect(principals).toContainEqual({ + principalType: PrincipalType.ROLE, + principalId: SystemRoles.USER, + }); + expect(cache.set).toHaveBeenCalledTimes(1); + expect(cache.releaseLock).not.toHaveBeenCalled(); + }); + + it('skips the cache write when the entry is invalidated mid-build', async () => { + const user = await createTestUser({ idOnTheSource: 'stale-ext-1' }); + const group = await Group.create({ + name: 'Stale Team', + source: 'entra', + idOnTheSource: 'stale-grp-1', + memberIds: [], + }); + const cache = createFakeCache(); + let openLockGate: (() => void) | undefined; + const lockGate = new Promise((resolve) => { + openLockGate = resolve; + }); + const lockedCache = Object.assign(cache, { + acquireLock: jest.fn(async () => { + await lockGate; + return 'lock-token'; + }), + releaseLock: jest.fn(async () => undefined), + lockWaitMs: 1000, + }); + const cachedMethods = createUserGroupMethods(mongoose, { + getCache: jest.fn(() => lockedCache), + }); + const params = { + userId: user._id.toString(), + role: SystemRoles.USER, + idOnTheSource: 'stale-ext-1', + }; + + const pendingRead = cachedMethods.getUserPrincipals(params); + await new Promise((resolve) => setTimeout(resolve, 10)); + /** Mutation lands while the pending build is parked on the lock */ + await cachedMethods.addUserToGroup(user._id, group._id); + openLockGate!(); + const principals = await pendingRead; + + expect(groupPrincipalIds(principals)).toEqual([group._id.toString()]); + expect(cache.set).not.toHaveBeenCalled(); + expect(cache.store.has('stale-ext-1')).toBe(false); + + const fresh = await cachedMethods.getUserPrincipals(params); + expect(groupPrincipalIds(fresh)).toEqual([group._id.toString()]); + expect(cache.set).toHaveBeenCalledTimes(1); + }); + + it('starts a fresh build for reads after invalidation instead of joining a stale one', async () => { + const user = await createTestUser({ idOnTheSource: 'rejoin-ext-1' }); + const group = await Group.create({ + name: 'Rejoin Team', + source: 'entra', + idOnTheSource: 'rejoin-grp-1', + memberIds: [], + }); + const cache = createFakeCache(); + let openReleaseGate: (() => void) | undefined; + const releaseGate = new Promise((resolve) => { + openReleaseGate = resolve; + }); + const lockedCache = Object.assign(cache, { + acquireLock: jest.fn(async () => 'lock-token'), + releaseLock: jest + .fn(async () => undefined) + .mockImplementationOnce(async () => { + await releaseGate; + }), + lockWaitMs: 1000, + }); + const cachedMethods = createUserGroupMethods(mongoose, { + getCache: jest.fn(() => lockedCache), + }); + const params = { + userId: user._id.toString(), + role: SystemRoles.USER, + idOnTheSource: 'rejoin-ext-1', + }; + + /** First read finishes its DB work but stays unsettled, parked on releaseLock */ + const parkedRead = cachedMethods.getUserPrincipals(params); + await new Promise((resolve) => setTimeout(resolve, 25)); + await cachedMethods.addUserToGroup(user._id, group._id); + + /** A read issued after the mutation must not join the parked pre-mutation build */ + const freshRead = await cachedMethods.getUserPrincipals(params); + expect(groupPrincipalIds(freshRead)).toEqual([group._id.toString()]); + + openReleaseGate!(); + expect(groupPrincipalIds(await parkedRead)).toEqual([]); + }); + + it('defers invalidation for transactional writes until the session ends', async () => { + const user = await createTestUser({ idOnTheSource: 'txn-ext-1' }); + const group = await Group.create({ + name: 'Transactional Team', + source: 'entra', + idOnTheSource: 'txn-grp-1', + memberIds: [], + }); + const cache = createFakeCache(); + const cachedMethods = createCachedMethods(cache); + const params = { + userId: user._id.toString(), + role: SystemRoles.USER, + idOnTheSource: 'txn-ext-1', + }; + + expect(groupPrincipalIds(await cachedMethods.getUserPrincipals(params))).toEqual([]); + + const session = await mongoose.startSession(); + session.startTransaction(); + await cachedMethods.addUserToGroup(user._id, group._id, session); + + /** Mid-transaction: no invalidation, reads keep the pre-commit membership */ + expect(cache.delete).not.toHaveBeenCalled(); + expect(groupPrincipalIds(await cachedMethods.getUserPrincipals(params))).toEqual([]); + + await session.commitTransaction(); + await session.endSession(); + await new Promise((resolve) => setTimeout(resolve, 20)); + + expect(cache.delete).toHaveBeenCalledWith('txn-ext-1'); + expect(groupPrincipalIds(await cachedMethods.getUserPrincipals(params))).toEqual([ + group._id.toString(), + ]); + }); + + it('caches and invalidates under tenant-scoped keys within a tenant context', async () => { + const user = await createTestUser({ idOnTheSource: 'tenant-ext-1' }); + const group = await Group.create({ + name: 'Tenant Team', + source: 'entra', + idOnTheSource: 'tenant-grp-1', + memberIds: ['tenant-ext-1'], + }); + const cache = createFakeCache(); + const cachedMethods = createCachedMethods(cache); + const params = { + userId: user._id.toString(), + role: SystemRoles.USER, + idOnTheSource: 'tenant-ext-1', + }; + + await tenantStorage.run({ tenantId: 'tenant-a' }, async () => { + expect(groupPrincipalIds(await cachedMethods.getUserPrincipals(params))).toEqual([ + group._id.toString(), + ]); + }); + expect(cache.store.has('tenant-ext-1:tenant-a')).toBe(true); + + await tenantStorage.run({ tenantId: 'tenant-a' }, async () => { + await cachedMethods.removeUserFromGroup(user._id, group._id); + }); + expect(cache.store.has('tenant-ext-1:tenant-a')).toBe(false); + expect(cache.delete).toHaveBeenCalledWith('tenant-ext-1:tenant-a'); + expect(cache.delete).toHaveBeenCalledWith('tenant-ext-1'); + }); + + it('keeps tenant scoping for invalidations deferred past the transaction', async () => { + const user = await createTestUser({ idOnTheSource: 'txn-tenant-ext-1' }); + const group = await Group.create({ + name: 'Tenant Transactional Team', + source: 'entra', + idOnTheSource: 'txn-tenant-grp-1', + memberIds: [], + }); + const cache = createFakeCache(); + const cachedMethods = createCachedMethods(cache); + + let session!: mongoose.ClientSession; + await tenantStorage.run({ tenantId: 'tenant-b' }, async () => { + session = await mongoose.startSession(); + session.startTransaction(); + await cachedMethods.addUserToGroup(user._id, group._id, session); + }); + expect(cache.delete).not.toHaveBeenCalled(); + + /** Session ends outside the tenant context; the snapshot must preserve it */ + await session.commitTransaction(); + await session.endSession(); + await new Promise((resolve) => setTimeout(resolve, 20)); + + expect(cache.delete).toHaveBeenCalledWith('txn-tenant-ext-1:tenant-b'); + expect(cache.delete).toHaveBeenCalledWith('txn-tenant-ext-1'); + }); + + it('runs a delayed second invalidation to evict cross-process stale rewrites', async () => { + const user = await createTestUser({ idOnTheSource: 'second-ext-1' }); + const group = await Group.create({ + name: 'Second Pass Team', + source: 'entra', + idOnTheSource: 'second-grp-1', + memberIds: [], + }); + const cache = createFakeCache(); + const lockedCache = Object.assign(cache, { + acquireLock: jest.fn(async () => 'lock-token'), + releaseLock: jest.fn(async () => undefined), + lockWaitMs: 0, + }); + const cachedMethods = createUserGroupMethods(mongoose, { + getCache: jest.fn(() => lockedCache), + }); + const params = { + userId: user._id.toString(), + role: SystemRoles.USER, + idOnTheSource: 'second-ext-1', + }; + + await cachedMethods.addUserToGroup(user._id, group._id); + expect(cache.delete).toHaveBeenCalledTimes(1); + + /** Another container re-caches pre-mutation memberships after the first delete */ + cache.store.set('second-ext-1', []); + await new Promise((resolve) => setTimeout(resolve, 700)); + + expect(cache.store.has('second-ext-1')).toBe(false); + expect(groupPrincipalIds(await cachedMethods.getUserPrincipals(params))).toEqual([ + group._id.toString(), + ]); + }); + + it('skips the namespace clear for no-op bulk membership replacements', async () => { + await Group.create({ + name: 'Noop Team', + source: 'entra', + idOnTheSource: 'noop-grp-1', + memberIds: ['noop-ext-1'], + }); + const cache = createFakeCache(); + const cachedMethods = createCachedMethods(cache); + + await cachedMethods.bulkUpdateGroups( + { _id: new Types.ObjectId() }, + { $set: { memberIds: ['noop-ext-2'] } }, + ); + + expect(cache.clear).not.toHaveBeenCalled(); + expect(cache.delete).not.toHaveBeenCalled(); + }); + }); + describe('syncUserEntraGroups', () => { it('creates new groups and adds user as member', async () => { const user = await createTestUser({ idOnTheSource: 'user-ext-1' }); diff --git a/packages/data-schemas/src/methods/userGroup.ts b/packages/data-schemas/src/methods/userGroup.ts index 806ed86eda..e78c1985de 100644 --- a/packages/data-schemas/src/methods/userGroup.ts +++ b/packages/data-schemas/src/methods/userGroup.ts @@ -1,11 +1,112 @@ import { Types } from 'mongoose'; -import { PrincipalType } from 'librechat-data-provider'; +import { AsyncLocalStorage } from 'async_hooks'; +import { CacheKeys, PrincipalType } from 'librechat-data-provider'; import type { TUser, TPrincipalSearchResult } from 'librechat-data-provider'; import type { Model, ClientSession, FilterQuery } from 'mongoose'; -import type { IGroup, IRole, IUser } from '~/types'; +import type { CacheStore, IGroup, IRole, IUser } from '~/types'; +import { isValidObjectIdString } from '~/utils/objectId'; +import { scopedCacheKey } from '~/config/tenantContext'; import { escapeRegExp } from '~/utils/string'; -export function createUserGroupMethods(mongoose: typeof import('mongoose')): { +export interface UserGroupDeps { + /** Returns the USER_PRINCIPALS cache store when principal caching is enabled. From getLogStores. */ + getCache?: (key: string) => CacheStore | undefined; +} + +type PendingGroupLookup = { promise: Promise; markStale: () => void }; + +/** Same-process dedup of concurrent cache builds, keyed by scoped member cache key. */ +const pendingGroupLookups = new Map(); +const GROUP_LOCK_POLL_MS = 50; +/** Above this many member keys, invalidation clears the namespace instead of fanning out deletes. */ +const INVALIDATION_CLEAR_THRESHOLD = 1000; +/** Margin added to the lock wait when scheduling the delayed second invalidation pass. */ +const STALE_WRITE_GRACE_MS = 500; + +const isCachedGroupId = (value: unknown): value is string => + typeof value === 'string' && isValidObjectIdString(value); + +/** + * Runs cache invalidation immediately, or defers it to session end for transactional + * writes. Mid-transaction invalidation would let concurrent readers re-cache pre-commit + * state with no later correction; deferring keeps the existing entry serving the + * still-committed old memberships until the transaction commits or aborts. + */ +function runAfterTransaction( + session: ClientSession | undefined, + invalidate: () => Promise, +): Promise { + if (session?.inTransaction()) { + /** The `ended` listener runs in the emitter's context; snapshot the caller's ALS + * (tenant scoping) so the deferred invalidation still derives tenant-scoped keys. */ + const runInCallerContext = AsyncLocalStorage.snapshot(); + session.once('ended', () => { + runInCallerContext(invalidate).catch(() => undefined); + }); + return Promise.resolve(); + } + return invalidate(); +} + +/** Extracts member ids referenced by a group update; indeterminate when they cannot be enumerated. */ +function collectMemberIdsFromGroupUpdate(update: Record): { + memberIds: string[]; + indeterminate: boolean; +} { + const memberIds: string[] = []; + + const collect = (value: unknown): boolean => { + if (typeof value === 'string') { + memberIds.push(value); + return true; + } + if (value instanceof Types.ObjectId) { + memberIds.push(value.toString()); + return true; + } + if (Array.isArray(value)) { + return value.every(collect); + } + if (value !== null && typeof value === 'object') { + const modifiers = value as { $each?: unknown; $in?: unknown }; + if (Array.isArray(modifiers.$each)) { + return modifiers.$each.every(collect); + } + if (Array.isArray(modifiers.$in)) { + return modifiers.$in.every(collect); + } + } + return false; + }; + + let indeterminate = false; + for (const [operator, operand] of Object.entries(update)) { + const touchesMembers = + operator === 'memberIds' || + (operand !== null && + typeof operand === 'object' && + 'memberIds' in (operand as Record)); + if (!touchesMembers) { + continue; + } + if (operator === '$addToSet' || operator === '$push' || operator === '$pull') { + indeterminate = !collect((operand as Record).memberIds) || indeterminate; + } else if (operator === '$pullAll') { + const values = (operand as Record).memberIds; + indeterminate = !Array.isArray(values) || !values.every(collect) || indeterminate; + } else { + /** Wholesale replacement ($set, $unset, direct assignment): affected members are unknown. */ + indeterminate = true; + } + } + + return { memberIds, indeterminate }; +} + +export function createUserGroupMethods( + mongoose: typeof import('mongoose'), + deps: UserGroupDeps = {}, +): { findGroupById: ( groupId: string | Types.ObjectId, projection?: Record, @@ -117,6 +218,227 @@ export function createUserGroupMethods(mongoose: typeof import('mongoose')): { session?: ClientSession, ) => Promise; } { + const getPrincipalsCache = (): CacheStore | undefined => + deps.getCache?.(CacheKeys.USER_PRINCIPALS); + + async function queryGroupIds( + memberId: string, + session?: ClientSession, + ): Promise { + const Group = mongoose.models.Group as Model; + const groupsQuery = Group.find({ memberIds: memberId }, { _id: 1 }); + if (session) { + groupsQuery.session(session); + } + const groups = await groupsQuery.lean>>(); + return groups.map((group) => group._id); + } + + async function readCachedGroupIds( + cache: CacheStore, + cacheKey: string, + ): Promise { + try { + const cached = await cache.get(cacheKey); + if (Array.isArray(cached) && cached.every(isCachedGroupId)) { + return cached.map((groupId) => new Types.ObjectId(groupId)); + } + } catch { + /** Cache failures must not block permission checks. */ + } + return undefined; + } + + /** + * Resolves group ids for a member key (`idOnTheSource` for external users, else the + * raw user id), cached per member. Concurrent same-process misses share one build; + * Redis-backed stores also dedupe cross-process builds via a short-lived lock. + * Session-scoped reads bypass the cache entirely. + */ + async function getMemberGroupIds( + memberId: string, + session?: ClientSession, + ): Promise { + const cache = session ? undefined : getPrincipalsCache(); + if (!cache) { + return queryGroupIds(memberId, session); + } + + const cacheKey = scopedCacheKey(memberId); + const fastPath = await readCachedGroupIds(cache, cacheKey); + if (fastPath) { + return fastPath; + } + + const pending = pendingGroupLookups.get(cacheKey); + if (pending) { + return pending.promise; + } + + /** Invalidation flips this so an in-flight build skips its now-stale cache write. */ + let stale = false; + /** + * The `_LOCK` prefix keeps raw lock keys disjoint from Keyv data keys + * (always `USER_PRINCIPALS:`), even for member ids ending in `:lock`. + */ + const lockKey = `${CacheKeys.USER_PRINCIPALS}_LOCK:${cacheKey}`; + const lookup = (async (): Promise => { + let lockToken: string | null | undefined; + try { + if (cache.acquireLock) { + let lockFailed = false; + try { + lockToken = await cache.acquireLock(lockKey); + } catch { + lockFailed = true; + } + if (lockToken) { + /** Another process may have filled the key between the fast-path miss and the lock. */ + const cached = await readCachedGroupIds(cache, cacheKey); + if (cached) { + return cached; + } + } else if (!lockFailed) { + const waitUntil = Date.now() + Math.max(cache.lockWaitMs ?? 0, 0); + while (Date.now() < waitUntil) { + await new Promise((resolve) => setTimeout(resolve, GROUP_LOCK_POLL_MS)); + const cached = await readCachedGroupIds(cache, cacheKey); + if (cached) { + return cached; + } + } + } + } + + const groupIds = await queryGroupIds(memberId); + if (!stale) { + try { + await cache.set( + cacheKey, + groupIds.map((groupId) => groupId.toString()), + ); + } catch { + /** Cache failures must not block permission checks. */ + } + } + return groupIds; + } finally { + if (lockToken) { + try { + await cache.releaseLock?.(lockKey, lockToken); + } catch { + /** Lock cleanup failures expire via the lock TTL. */ + } + } + } + })(); + + /** + * Registered synchronously (no await between the pending check and this set) so every + * concurrent same-process caller attaches to this single flow instead of starting its own. + */ + pendingGroupLookups.set(cacheKey, { + promise: lookup, + markStale: () => { + stale = true; + }, + }); + try { + return await lookup; + } finally { + if (pendingGroupLookups.get(cacheKey)?.promise === lookup) { + pendingGroupLookups.delete(cacheKey); + } + } + } + + async function dropMemberKeys(cache: CacheStore, cacheKeys: Set): Promise { + for (const cacheKey of cacheKeys) { + const pending = pendingGroupLookups.get(cacheKey); + if (pending) { + pending.markStale(); + pendingGroupLookups.delete(cacheKey); + } + } + try { + await Promise.all([...cacheKeys].map((cacheKey) => cache.delete?.(cacheKey))); + } catch { + /** Cache failures must not block membership updates. */ + } + } + + /** + * Cross-process builds cannot see this process's stale flags, so a build in another + * container that read pre-mutation state can re-cache it after the first delete. + * A delayed second pass (bounded by the lock wait plus one query round-trip) evicts + * such rewrites; only Redis-backed stores (lock-capable) share data across processes. + */ + function scheduleSecondInvalidation(cache: CacheStore, secondPass: () => Promise): void { + if (!cache.acquireLock) { + return; + } + const graceMs = Math.max(cache.lockWaitMs ?? 0, 0) + STALE_WRITE_GRACE_MS; + const timer = setTimeout(() => { + secondPass().catch(() => undefined); + }, graceMs); + timer.unref?.(); + } + + /** + * Drops cached group memberships for the given member keys (both base and + * tenant-scoped variants) after a membership mutation. In-flight builds are + * marked stale (skipping their cache write) and unregistered so later reads + * start a fresh build instead of joining a pre-mutation one. Failures are + * non-fatal; the cache TTL bounds any residual staleness (e.g. mutations + * running under a different tenant context than the member's reads). + */ + async function invalidateMemberGroupsCache( + memberIds: Array, + ): Promise { + const cache = getPrincipalsCache(); + if (!cache?.delete) { + return; + } + const cacheKeys = new Set(); + for (const memberId of memberIds) { + if (!memberId) { + continue; + } + const baseKey = memberId.toString(); + cacheKeys.add(baseKey); + cacheKeys.add(scopedCacheKey(baseKey)); + } + if (cacheKeys.size === 0) { + return; + } + if (cacheKeys.size > INVALIDATION_CLEAR_THRESHOLD && cache.clear) { + return clearMemberGroupsCache(); + } + await dropMemberKeys(cache, cacheKeys); + scheduleSecondInvalidation(cache, () => dropMemberKeys(cache, cacheKeys)); + } + + /** Clears the whole membership cache when affected members cannot be enumerated. */ + async function clearMemberGroupsCache(): Promise { + const cache = getPrincipalsCache(); + if (!cache?.clear) { + return; + } + const clearAll = async (): Promise => { + for (const pending of pendingGroupLookups.values()) { + pending.markStale(); + } + pendingGroupLookups.clear(); + try { + await cache.clear?.(); + } catch { + /** Cache failures must not block membership updates. */ + } + }; + await clearAll(); + scheduleSecondInvalidation(cache, clearAll); + } + /** * Find a group by its ID * @param groupId - The group ID @@ -254,7 +576,11 @@ export function createUserGroupMethods(mongoose: typeof import('mongoose')): { async function createGroup(groupData: Partial, session?: ClientSession): Promise { const Group = mongoose.models.Group as Model; const options = session ? { session } : {}; - return await Group.create([groupData], options).then((groups) => groups[0]); + const group = await Group.create([groupData], options).then((groups) => groups[0]); + await runAfterTransaction(session, () => + invalidateMemberGroupsCache(groupData.memberIds ?? []), + ); + return group; } /** @@ -278,7 +604,22 @@ export function createUserGroupMethods(mongoose: typeof import('mongoose')): { ...(session ? { session } : {}), }; - return await Group.findOneAndUpdate({ idOnTheSource, source }, { $set: updateData }, options); + if (updateData.memberIds === undefined) { + return await Group.findOneAndUpdate({ idOnTheSource, source }, { $set: updateData }, options); + } + /** Atomic pre-image (`new: false`) so members added concurrently before the $set are invalidated too. */ + const previous = await Group.findOneAndUpdate( + { idOnTheSource, source }, + { $set: updateData }, + { ...options, new: false }, + ).lean(); + await runAfterTransaction(session, () => + invalidateMemberGroupsCache([ + ...(previous?.memberIds ?? []), + ...(updateData.memberIds ?? []), + ]), + ); + return await findGroupByExternalId(idOnTheSource, source, {}, session); } /** @@ -315,6 +656,7 @@ export function createUserGroupMethods(mongoose: typeof import('mongoose')): { { $addToSet: { memberIds: userIdOnTheSource } }, options, ).lean(); + await runAfterTransaction(session, () => invalidateMemberGroupsCache([userIdOnTheSource])); return { user: user as IUser, group: updatedGroup }; } @@ -353,6 +695,7 @@ export function createUserGroupMethods(mongoose: typeof import('mongoose')): { { $pullAll: { memberIds: [userIdOnTheSource] } }, options, ).lean(); + await runAfterTransaction(session, () => invalidateMemberGroupsCache([userIdOnTheSource])); return { user: user as IUser, group: updatedGroup }; } @@ -392,6 +735,10 @@ export function createUserGroupMethods(mongoose: typeof import('mongoose')): { * `_id`-projected group query. `idOnTheSource: null` means "known to be absent" * (local user) and also avoids the lookup; only `undefined` triggers it. * + * Group memberships are additionally cached per member key (USER_PRINCIPALS + * namespace) when a cache is injected; membership mutations in this module + * invalidate the affected keys. Role resolution is never cached. + * * @param params - Parameters object * @param params.userId - The user ID * @param params.role - Optional user role (looked up when `undefined`) @@ -440,14 +787,9 @@ export function createUserGroupMethods(mongoose: typeof import('mongoose')): { /** `memberIds` stores `idOnTheSource` for external users, else the raw user id. */ const memberId = memberIdOnTheSource || userId.toString(); - const Group = mongoose.models.Group as Model; - const groupsQuery = Group.find({ memberIds: memberId }, { _id: 1 }); - if (session) { - groupsQuery.session(session); - } - const userGroups = await groupsQuery.lean>>(); - for (const group of userGroups) { - principals.push({ principalType: PrincipalType.GROUP, principalId: group._id }); + const groupIds = await getMemberGroupIds(memberId, session); + for (const groupId of groupIds) { + principals.push({ principalType: PrincipalType.GROUP, principalId: groupId }); } principals.push({ principalType: PrincipalType.PUBLIC }); @@ -766,6 +1108,7 @@ export function createUserGroupMethods(mongoose: typeof import('mongoose')): { async function removeUserFromAllGroups(userId: string | Types.ObjectId): Promise { const Group = mongoose.models.Group as Model; await Group.updateMany({ memberIds: userId }, { $pullAll: { memberIds: [userId] } }); + await invalidateMemberGroupsCache([userId]); } /** @@ -796,7 +1139,28 @@ export function createUserGroupMethods(mongoose: typeof import('mongoose')): { ): Promise { const Group = mongoose.models.Group as Model; const options = { new: true, ...(session ? { session } : {}) }; - return Group.findByIdAndUpdate(groupId, { $set: data }, options).lean(); + if (data.memberIds === undefined) { + return Group.findByIdAndUpdate(groupId, { $set: data }, options).lean(); + } + /** Atomic pre-image (`new: false`) so members added concurrently before the $set are invalidated too. */ + const previous = await Group.findByIdAndUpdate( + groupId, + { $set: data }, + { + ...options, + new: false, + }, + ).lean(); + const nextMemberIds = Array.isArray(data.memberIds) + ? data.memberIds.filter( + (memberId): memberId is string | Types.ObjectId => + typeof memberId === 'string' || memberId instanceof Types.ObjectId, + ) + : []; + await runAfterTransaction(session, () => + invalidateMemberGroupsCache([...(previous?.memberIds ?? []), ...nextMemberIds]), + ); + return previous ? await findGroupById(groupId, {}, session) : null; } /** @@ -811,7 +1175,17 @@ export function createUserGroupMethods(mongoose: typeof import('mongoose')): { options?: { session?: ClientSession }, ): Promise { const Group = mongoose.models.Group as Model; - return Group.updateMany(filter, update, options || {}); + const result = await Group.updateMany(filter, update, options || {}); + if (result.modifiedCount === 0 && result.upsertedCount === 0) { + return result; + } + const { memberIds, indeterminate } = collectMemberIdsFromGroupUpdate(update); + if (indeterminate) { + await runAfterTransaction(options?.session, () => clearMemberGroupsCache()); + } else if (memberIds.length > 0) { + await runAfterTransaction(options?.session, () => invalidateMemberGroupsCache(memberIds)); + } + return result; } function buildGroupQuery(filter: { @@ -881,7 +1255,9 @@ export function createUserGroupMethods(mongoose: typeof import('mongoose')): { ): Promise { const Group = mongoose.models.Group as Model; const options = session ? { session } : {}; - return await Group.findByIdAndDelete(groupId, options).lean(); + const group = await Group.findByIdAndDelete(groupId, options).lean(); + await runAfterTransaction(session, () => invalidateMemberGroupsCache(group?.memberIds ?? [])); + return group; } /** @@ -898,11 +1274,13 @@ export function createUserGroupMethods(mongoose: typeof import('mongoose')): { ): Promise { const Group = mongoose.models.Group as Model; const options = { new: true, ...(session ? { session } : {}) }; - return await Group.findByIdAndUpdate( + const group = await Group.findByIdAndUpdate( groupId, { $pull: { memberIds: memberId } }, options, ).lean(); + await runAfterTransaction(session, () => invalidateMemberGroupsCache([memberId])); + return group; } return { diff --git a/packages/data-schemas/src/types/cache.ts b/packages/data-schemas/src/types/cache.ts new file mode 100644 index 0000000000..5db84dbacf --- /dev/null +++ b/packages/data-schemas/src/types/cache.ts @@ -0,0 +1,16 @@ +/** + * Cache store contract injected into database methods from the api layer + * (e.g. getLogStores). Lock members are only present when the store is + * Redis-backed and cross-process build deduplication is enabled. + */ +export interface CacheStore { + get: (key: string) => Promise; + set: (key: string, value: unknown) => Promise; + delete?: (key: string) => Promise; + clear?: () => Promise; + /** Acquires a cross-process build lock; resolves a release token, or null when already held. */ + acquireLock?: (key: string) => Promise; + releaseLock?: (key: string, token: string) => Promise; + /** Max time to wait for another process holding the build lock to fill the cache. */ + lockWaitMs?: number; +} diff --git a/packages/data-schemas/src/types/index.ts b/packages/data-schemas/src/types/index.ts index 3dc5228532..f185ebf625 100644 --- a/packages/data-schemas/src/types/index.ts +++ b/packages/data-schemas/src/types/index.ts @@ -2,6 +2,7 @@ import type { Types } from 'mongoose'; export type ObjectId = Types.ObjectId; export * from './app'; +export * from './cache'; export * from './user'; export * from './token'; export * from './convo';