diff --git a/server/adapters/cache/runtime-cache.test.ts b/server/adapters/cache/runtime-cache.test.ts index 0405854e..40ace62b 100644 --- a/server/adapters/cache/runtime-cache.test.ts +++ b/server/adapters/cache/runtime-cache.test.ts @@ -48,6 +48,20 @@ describe('runtime cache', () => { expect(await cache.getOrLoad(stringPolicy, 'a', loader)).toMatchObject({ value: 'value-2', tier: 'source' }) }) + it('coalesces concurrent loads for the same key', async () => { + let release: (value: string) => void = () => undefined + const loader = vi.fn(() => new Promise((resolve) => (release = resolve))) + const cache = createRuntimeCache({ mode: 'memory' }) + + const first = cache.getOrLoad(stringPolicy, 'a', loader) + const second = cache.getOrLoad(stringPolicy, 'a', loader) + expect(loader).toHaveBeenCalledTimes(1) + + release('shared') + expect(await first).toMatchObject({ value: 'shared', tier: 'source' }) + expect(await second).toMatchObject({ value: 'shared', tier: 'coalesced' }) + }) + it('uses the shorter negative-cache TTL', async () => { let now = 1_000 const loader = vi.fn(async () => null) diff --git a/server/adapters/cache/runtime-cache.ts b/server/adapters/cache/runtime-cache.ts index dac0e1a6..108a2050 100644 --- a/server/adapters/cache/runtime-cache.ts +++ b/server/adapters/cache/runtime-cache.ts @@ -44,6 +44,7 @@ export function createRuntimeCache(options: RuntimeCacheOptions): CacheService { } const stores = new Map>() + const loads = new Map>>() const now = options.now ?? Date.now function memoryStore(namespace: string): Map { @@ -151,16 +152,32 @@ export function createRuntimeCache(options: RuntimeCacheOptions): CacheService { const inMemory = memoryGet(policy, key) if (inMemory !== undefined) return observed(policy.namespace, 'memory', startedAt, inMemory) - if (usesDistributed(policy)) { - const distributed = await distributedGet(policy, key) - if (distributed !== undefined) return observed(policy.namespace, 'distributed', startedAt, distributed) + const loadKey = cacheKey(policy, key) + const existing = loads.get(loadKey) as Promise> | undefined + if (existing) { + const result = await existing + return observed(policy.namespace, 'coalesced', startedAt, result.value) } - const value = await loader() - const freshUntil = now() + ttlFor(policy, value) - memoryPut(policy, key, value, freshUntil) - if (usesDistributed(policy)) await distributedPut(policy, key, value, freshUntil) - return observed(policy.namespace, 'source', startedAt, value) + const load = (async (): Promise> => { + if (usesDistributed(policy)) { + const distributed = await distributedGet(policy, key) + if (distributed !== undefined) return { value: distributed, tier: 'distributed' } + } + + const value = await loader() + const freshUntil = now() + ttlFor(policy, value) + memoryPut(policy, key, value, freshUntil) + if (usesDistributed(policy)) await distributedPut(policy, key, value, freshUntil) + return { value, tier: 'source' } + })() + loads.set(loadKey, load as Promise>) + try { + const result = await load + return observed(policy.namespace, result.tier, startedAt, result.value) + } finally { + if (loads.get(loadKey) === load) loads.delete(loadKey) + } }, async replace(policy: CachePolicy, key: string, value: T): Promise { diff --git a/server/adapters/repos/webdav-path.ts b/server/adapters/repos/webdav-path.ts index bc01cbaa..d85b18b1 100644 --- a/server/adapters/repos/webdav-path.ts +++ b/server/adapters/repos/webdav-path.ts @@ -4,6 +4,8 @@ import { member, organization } from '../../db/auth-schema' import { matters } from '../../db/schema' import type { Database } from '../../platform/interface' import { + type CachePolicy, + type CacheService, type Matter, WebDavPathError, type WebDavPathRepo, @@ -17,7 +19,38 @@ type WorkspaceMatterRow = { matter: Matter | null } -export function createWebDavPathRepo(db: Database): WebDavPathRepo { +const WEB_DAV_WORKSPACES_CACHE_POLICY: CachePolicy = { + namespace: 'webdav-workspaces', + version: 1, + ttlMs: 1_000, + maxEntries: 256, + distributed: false, + validate(value): value is WebDavWorkspace[] { + return ( + Array.isArray(value) && + value.every( + (workspace) => + typeof workspace === 'object' && + workspace !== null && + typeof workspace.id === 'string' && + typeof workspace.name === 'string' && + typeof workspace.slug === 'string' && + typeof workspace.pathSegment === 'string', + ) + ) + }, +} + +export function createWebDavPathRepo(db: Database, cache?: CacheService): WebDavPathRepo { + async function listUserWorkspaces(userId: string): Promise { + if (!cache) return toWebDavWorkspaces(await userWorkspaceRows(db, userId)) + return ( + await cache.getOrLoad(WEB_DAV_WORKSPACES_CACHE_POLICY, userId, async () => + toWebDavWorkspaces(await userWorkspaceRows(db, userId)), + ) + ).value + } + async function resolveWebDavPath(userId: string, rawPath: string): Promise { const parts = decodeDavPath(rawPath) if (parts.length === 0) return { workspace: null, mountRoot: true, parent: '', name: '', matter: null } @@ -25,15 +58,21 @@ export function createWebDavPathRepo(db: Database): WebDavPathRepo { const matterParts = parts.slice(1) const name = matterParts.at(-1) ?? '' const parent = matterParts.slice(0, -1).join('/') + if (cache) { + const workspaces = await listUserWorkspaces(userId) + const workspace = findWorkspace(workspaces, parts[0]) + if (!workspace) throw new WebDavPathError('Workspace not found', 404) + if (parts.length === 1) return { workspace, mountRoot: false, parent: '', name: '', matter: null } + const matter = await workspaceMatterRow(db, workspace.id, parent, name) + return { workspace, mountRoot: false, parent, name, matter } + } + const rows = parts.length === 1 ? (await userWorkspaceRows(db, userId)).map((workspace) => ({ workspace, matter: null })) : await userWorkspaceMatterRows(db, userId, parent, name) const workspaces = toWebDavWorkspaces(rows.map((row) => row.workspace)) - const workspace = - workspaces.find( - (candidate) => candidate.slug === parts[0] || candidate.id === parts[0] || candidate.pathSegment === parts[0], - ) ?? null + const workspace = findWorkspace(workspaces, parts[0]) if (!workspace) throw new WebDavPathError('Workspace not found', 404) if (parts.length === 1) return { workspace, mountRoot: false, parent: '', name: '', matter: null } @@ -43,7 +82,7 @@ export function createWebDavPathRepo(db: Database): WebDavPathRepo { return { async listUserWorkspaces(userId) { - return toWebDavWorkspaces(await userWorkspaceRows(db, userId)) + return listUserWorkspaces(userId) }, async listChildren(orgId, parent) { @@ -72,6 +111,32 @@ export function createWebDavPathRepo(db: Database): WebDavPathRepo { } } +function findWorkspace(workspaces: WebDavWorkspace[], segment: string): WebDavWorkspace | null { + return ( + workspaces.find( + (candidate) => candidate.slug === segment || candidate.id === segment || candidate.pathSegment === segment, + ) ?? null + ) +} + +async function workspaceMatterRow(db: Database, orgId: string, parent: string, name: string): Promise { + const rows = await db + .select() + .from(matters) + .where( + and( + eq(matters.orgId, orgId), + eq(matters.parent, parent), + eq(matters.name, name), + eq(matters.status, ObjectStatus.ACTIVE), + isNull(matters.trashedAt), + isNull(matters.purgedAt), + ), + ) + .limit(1) + return rows[0] ?? null +} + async function userWorkspaceRows(db: Database, userId: string): Promise { return db .select({ id: organization.id, name: organization.name, slug: organization.slug }) diff --git a/server/composition.ts b/server/composition.ts index 9afdd2b6..95a604e6 100644 --- a/server/composition.ts +++ b/server/composition.ts @@ -117,7 +117,7 @@ export function createDeps(platform: Platform, options: CreateDepsOptions = {}): teams: createTeamRepo(db), teamInvites: createTeamInviteRepo(db), userAdmin: createUserAdminRepo(db), - webdavPath: createWebDavPathRepo(db), + webdavPath: createWebDavPathRepo(db, cache), webdavState: createWebDavStateRepo(db), zip: createZipGateway(), zipPlan: createZipPlanRepo(db), diff --git a/server/usecases/ports/cache.ts b/server/usecases/ports/cache.ts index ba1262ce..5eb1524a 100644 --- a/server/usecases/ports/cache.ts +++ b/server/usecases/ports/cache.ts @@ -1,6 +1,6 @@ export type CacheMode = 'off' | 'memory' | 'distributed' -export type CacheTier = 'bypass' | 'memory' | 'distributed' | 'source' +export type CacheTier = 'bypass' | 'memory' | 'distributed' | 'source' | 'coalesced' export interface CachePolicy { namespace: string