mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-10-20 17:13:44 +00:00
🗃️ perf: Cache Group Memberships for ACL Principal Resolution (#14075)
Some checks are pending
Docker Dev Branch Images Build / build (Dockerfile, lc-dev, node) (push) Waiting to run
Docker Dev Branch Images Build / build (Dockerfile.multi, lc-dev-api, api-build) (push) Waiting to run
GitNexus Index / index (push) Waiting to run
GitNexus Index / post-index (push) Blocked by required conditions
Some checks are pending
Docker Dev Branch Images Build / build (Dockerfile, lc-dev, node) (push) Waiting to run
Docker Dev Branch Images Build / build (Dockerfile.multi, lc-dev-api, api-build) (push) Waiting to run
GitNexus Index / index (push) Waiting to run
GitNexus Index / post-index (push) Blocked by required conditions
* 🗃️ 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 <peter.rothlaender@ginkgo.com> Co-authored-by: Joachim Keltsch <joachim.keltsch@daimlertruck.com> * 🧵 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 <peter.rothlaender@ginkgo.com> Co-authored-by: Joachim Keltsch <joachim.keltsch@daimlertruck.com>
This commit is contained in:
parent
ed8547018c
commit
7b7fa496aa
14 changed files with 1173 additions and 29 deletions
10
.env.example
10
.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
|
||||
|
|
|
|||
10
api/cache/getLogStores.js
vendored
10
api/cache/getLogStores.js
vendored
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -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<void>}
|
||||
*/
|
||||
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;
|
||||
|
|
|
|||
|
|
@ -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([
|
||||
{
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
1
packages/api/src/cache/index.ts
vendored
1
packages/api/src/cache/index.ts
vendored
|
|
@ -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';
|
||||
|
|
|
|||
61
packages/api/src/cache/principals.ts
vendored
Normal file
61
packages/api/src/cache/principals.ts
vendored
Normal file
|
|
@ -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<string | null>;
|
||||
releaseLock?: (key: string, token: string) => Promise<void>;
|
||||
};
|
||||
|
||||
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;
|
||||
}
|
||||
|
|
@ -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.
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -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<unknown>;
|
||||
set: (k: string, v: unknown) => Promise<void>;
|
||||
};
|
||||
getCache?: (key: string) => CacheStore | undefined;
|
||||
}
|
||||
|
||||
export function createRoleMethods(
|
||||
|
|
|
|||
|
|
@ -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<t.IGroup>;
|
||||
let User: mongoose.Model<t.IUser>;
|
||||
let Role: mongoose.Model<t.IRole>;
|
||||
let methods: ReturnType<typeof createUserGroupMethods>;
|
||||
|
||||
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<t.IGroup>('Group', groupSchema);
|
||||
User = mongoose.models.User || mongoose.model<t.IUser>('User', userSchema);
|
||||
Role = mongoose.models.Role || mongoose.model<t.IRole>('Role', roleSchema);
|
||||
|
|
@ -481,6 +483,631 @@ describe('userGroup methods', () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe('getUserPrincipals caching', () => {
|
||||
function createFakeCache() {
|
||||
const store = new Map<string, unknown>();
|
||||
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<typeof createFakeCache>) {
|
||||
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<void>((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<void>((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' });
|
||||
|
|
|
|||
|
|
@ -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<Types.ObjectId[]>; markStale: () => void };
|
||||
|
||||
/** Same-process dedup of concurrent cache builds, keyed by scoped member cache key. */
|
||||
const pendingGroupLookups = new Map<string, PendingGroupLookup>();
|
||||
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<void>,
|
||||
): Promise<void> {
|
||||
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<string, unknown>): {
|
||||
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<string, unknown>));
|
||||
if (!touchesMembers) {
|
||||
continue;
|
||||
}
|
||||
if (operator === '$addToSet' || operator === '$push' || operator === '$pull') {
|
||||
indeterminate = !collect((operand as Record<string, unknown>).memberIds) || indeterminate;
|
||||
} else if (operator === '$pullAll') {
|
||||
const values = (operand as Record<string, unknown>).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<string, 0 | 1>,
|
||||
|
|
@ -117,6 +218,227 @@ export function createUserGroupMethods(mongoose: typeof import('mongoose')): {
|
|||
session?: ClientSession,
|
||||
) => Promise<IGroup | null>;
|
||||
} {
|
||||
const getPrincipalsCache = (): CacheStore | undefined =>
|
||||
deps.getCache?.(CacheKeys.USER_PRINCIPALS);
|
||||
|
||||
async function queryGroupIds(
|
||||
memberId: string,
|
||||
session?: ClientSession,
|
||||
): Promise<Types.ObjectId[]> {
|
||||
const Group = mongoose.models.Group as Model<IGroup>;
|
||||
const groupsQuery = Group.find({ memberIds: memberId }, { _id: 1 });
|
||||
if (session) {
|
||||
groupsQuery.session(session);
|
||||
}
|
||||
const groups = await groupsQuery.lean<Array<Pick<IGroup, '_id'>>>();
|
||||
return groups.map((group) => group._id);
|
||||
}
|
||||
|
||||
async function readCachedGroupIds(
|
||||
cache: CacheStore,
|
||||
cacheKey: string,
|
||||
): Promise<Types.ObjectId[] | undefined> {
|
||||
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<Types.ObjectId[]> {
|
||||
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:<key>`), even for member ids ending in `:lock`.
|
||||
*/
|
||||
const lockKey = `${CacheKeys.USER_PRINCIPALS}_LOCK:${cacheKey}`;
|
||||
const lookup = (async (): Promise<Types.ObjectId[]> => {
|
||||
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<string>): Promise<void> {
|
||||
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>): 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<string | Types.ObjectId | undefined | null>,
|
||||
): Promise<void> {
|
||||
const cache = getPrincipalsCache();
|
||||
if (!cache?.delete) {
|
||||
return;
|
||||
}
|
||||
const cacheKeys = new Set<string>();
|
||||
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<void> {
|
||||
const cache = getPrincipalsCache();
|
||||
if (!cache?.clear) {
|
||||
return;
|
||||
}
|
||||
const clearAll = async (): Promise<void> => {
|
||||
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<IGroup>, session?: ClientSession): Promise<IGroup> {
|
||||
const Group = mongoose.models.Group as Model<IGroup>;
|
||||
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<IGroup>();
|
||||
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<IGroup>();
|
||||
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<IGroup>();
|
||||
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<IGroup>;
|
||||
const groupsQuery = Group.find({ memberIds: memberId }, { _id: 1 });
|
||||
if (session) {
|
||||
groupsQuery.session(session);
|
||||
}
|
||||
const userGroups = await groupsQuery.lean<Array<Pick<IGroup, '_id'>>>();
|
||||
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<void> {
|
||||
const Group = mongoose.models.Group as Model<IGroup>;
|
||||
await Group.updateMany({ memberIds: userId }, { $pullAll: { memberIds: [userId] } });
|
||||
await invalidateMemberGroupsCache([userId]);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -796,7 +1139,28 @@ export function createUserGroupMethods(mongoose: typeof import('mongoose')): {
|
|||
): Promise<IGroup | null> {
|
||||
const Group = mongoose.models.Group as Model<IGroup>;
|
||||
const options = { new: true, ...(session ? { session } : {}) };
|
||||
return Group.findByIdAndUpdate(groupId, { $set: data }, options).lean<IGroup>();
|
||||
if (data.memberIds === undefined) {
|
||||
return Group.findByIdAndUpdate(groupId, { $set: data }, options).lean<IGroup>();
|
||||
}
|
||||
/** 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<IGroup>();
|
||||
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<import('mongoose').UpdateWriteOpResult> {
|
||||
const Group = mongoose.models.Group as Model<IGroup>;
|
||||
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<IGroup | null> {
|
||||
const Group = mongoose.models.Group as Model<IGroup>;
|
||||
const options = session ? { session } : {};
|
||||
return await Group.findByIdAndDelete(groupId, options).lean<IGroup>();
|
||||
const group = await Group.findByIdAndDelete(groupId, options).lean<IGroup>();
|
||||
await runAfterTransaction(session, () => invalidateMemberGroupsCache(group?.memberIds ?? []));
|
||||
return group;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -898,11 +1274,13 @@ export function createUserGroupMethods(mongoose: typeof import('mongoose')): {
|
|||
): Promise<IGroup | null> {
|
||||
const Group = mongoose.models.Group as Model<IGroup>;
|
||||
const options = { new: true, ...(session ? { session } : {}) };
|
||||
return await Group.findByIdAndUpdate(
|
||||
const group = await Group.findByIdAndUpdate(
|
||||
groupId,
|
||||
{ $pull: { memberIds: memberId } },
|
||||
options,
|
||||
).lean<IGroup>();
|
||||
await runAfterTransaction(session, () => invalidateMemberGroupsCache([memberId]));
|
||||
return group;
|
||||
}
|
||||
|
||||
return {
|
||||
|
|
|
|||
16
packages/data-schemas/src/types/cache.ts
Normal file
16
packages/data-schemas/src/types/cache.ts
Normal file
|
|
@ -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<unknown>;
|
||||
set: (key: string, value: unknown) => Promise<unknown>;
|
||||
delete?: (key: string) => Promise<unknown>;
|
||||
clear?: () => Promise<unknown>;
|
||||
/** Acquires a cross-process build lock; resolves a release token, or null when already held. */
|
||||
acquireLock?: (key: string) => Promise<string | null>;
|
||||
releaseLock?: (key: string, token: string) => Promise<unknown>;
|
||||
/** Max time to wait for another process holding the build lock to fill the cache. */
|
||||
lockWaitMs?: number;
|
||||
}
|
||||
|
|
@ -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';
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue