From 590c87c00a5ffe4dbc916f4e87fe9cf71d10877b Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Sun, 11 Oct 2026 22:07:53 -0700 Subject: [PATCH] refactor(stores): one row cache and keyset page helper, batched cache round trips, dead store methods removed (#4191) * refactor(stores): one read-through RowCache for UserStore and AppStore UserStore and AppStore each carried the same multi-key row cache: key builder, cache read, tombstone-guarded backfill, broadcast refresh, invalidate and tombstone-on-delete. They now share RowCache on PuterStore, which also single-flights concurrent misses for one key and batches getByIds/getByUids: one pipelined tombstone check for the loaded rows and one pipelined backfill instead of per-row awaits. Cache keys and TTLs are unchanged. * perf(apps): invalidate cached app lists by generation, not by tracker scan Every app write SMEMBERS'd a global set of every cached list key, parsed each key back into its filters and deleted the matches. List pages are now keyed by their owner's generation token (or a shared one for unscoped lists), and a write replaces the tokens it affects: O(1) per write, and a page loaded before a write can no longer be cached after it. * refactor(stores): one keyset page helper for store listings Thirteen store listings each cut a LIMIT n+1 read to a page and sealed the last row's position, and each opened an id cursor its own way. They now use keysetPage() and openIdCursor() from util/pagination. Cursor payloads and sealing are unchanged; a cursor without a usable id still fails with 400 in ShareStore and still restarts the listing elsewhere (strict option). * refactor(stores): delete store methods nothing calls AppStore.existsByIndexUrl, OIDCStore.unlinkByUserId, TeamStore.listByOwner/listActiveSeatUuids, SubdomainStore.getByDomain/ listByDomain, FSEntryStore.ensureDirectoriesForUser, ShareStore.listByIssuer/listByRecipientEmail/deleteByRecipientEmail/ getDailyShareCount/listByHolderAndFsentries and PermissionStore.invalidateScanCache/deleteUserUserPermsByPermissionPrefix had no production callers. ShareStore.listByFsentries was a copy of listReaching; its callers use the latter. SubdomainStore SMEMBERS'd a per-user prefix-list tracker on every write and broadcast its deletion, but nothing ever wrote that set or cached a prefix listing. The reads are gone. * refactor(stores): shared escapeLike and bumpGeneration helpers escapeLike replaces the LIKE escaping copied into PermissionStore (twice), ShareStore and FSEntryStore, and now also guards SubdomainStore's prefix filters, which passed caller prefixes into LIKE unescaped. bumpGeneration sends INCR and EXPIRE in one pipeline instead of two awaited commands in PermissionStore, EventSubscriptionStore and PresenceStore. * refactor(stores): read affected rows off WriteResult Every database client returns affectedRows and anyRowsAffected, so the `affectedRows ?? changes ?? 0` fallbacks in ShareStore, NotificationStore, UserStore and UserBlockStore read those directly. * perf(stores): batch flat-permission KV deletes and durable cache drops PermissionStore deleted each flat KV entry with its own kv.del; one kv.batchDel now covers the set. DurableSubscriptionStore's suspend and reap dropped each row from the cache separately; EventSubscriptionStore gains dropDurables, which drops them per owner in one pass, and dropDurable calls it. suspend keeps one conditional UPDATE per row, which is how concurrent settles learn which rows each of them suspended. * refactor(fs): fold FSEntryStore's duplicate getters, sort maps and cache keys - getEntryByUuid and getEntryById share one by-column getter. - listChildren and listChildrenPage read one sort table; the offset listing now coalesces NULL modified/size like the cursor listing, so both order NULLs the same way on every engine. - Entry cache keys come from one builder instead of hand-built strings. - getEntriesByIds, getEntriesByPaths and the per-user path reader fetch their cache hits in one pipelined round trip instead of a GET per key, through readCacheValues, which RowCache.readMany now uses too. * fix(apps): cache the row a lost createFromOrigin race reads from the primary Keeps AppStore's private #writeCache as a RowCache write, so code that still calls it (#4166's read-back) loads in either merge order. --- src/backend/services/share/ShareService.ts | 2 +- src/backend/stores/app/AppStore.js | 301 ++++++---------- src/backend/stores/app/AppStore.test.js | 70 +++- .../stores/events/DurableSubscriptionStore.ts | 30 +- .../stores/events/EventHandlerStore.ts | 14 +- .../events/EventSubscriptionStore.test.ts | 19 + .../stores/events/EventSubscriptionStore.ts | 51 ++- .../stores/events/KvShareHandleStore.ts | 25 +- src/backend/stores/events/PresenceStore.ts | 12 +- src/backend/stores/fs/FSEntryStore.test.ts | 56 ++- src/backend/stores/fs/FSEntryStore.ts | 303 +++++++--------- .../stores/notification/NotificationStore.js | 12 +- src/backend/stores/oidc/OIDCStore.js | 7 - .../stores/permission/PermissionStore.test.ts | 49 ++- .../stores/permission/PermissionStore.ts | 106 ++---- src/backend/stores/share/ShareStore.js | 173 +++------ src/backend/stores/share/ShareStore.test.js | 95 ----- .../stores/subdomain/SubdomainStore.test.ts | 55 ++- .../stores/subdomain/SubdomainStore.ts | 78 +---- src/backend/stores/team/TeamStore.test.ts | 40 --- src/backend/stores/team/TeamStore.ts | 113 ++---- src/backend/stores/types.test.ts | 304 ++++++++++++++++ src/backend/stores/types.ts | 331 +++++++++++++++++- src/backend/stores/user/UserStore.ts | 271 ++++---------- .../stores/userBlock/UserBlockStore.ts | 7 +- src/backend/util/pagination.test.ts | 63 ++++ src/backend/util/pagination.ts | 49 +++ src/backend/util/redisGeneration.test.ts | 68 ++++ src/backend/util/redisGeneration.ts | 41 +++ src/backend/util/sqlLike.test.ts | 32 ++ src/backend/util/sqlLike.ts | 26 ++ 31 files changed, 1586 insertions(+), 1217 deletions(-) create mode 100644 src/backend/stores/types.test.ts create mode 100644 src/backend/util/redisGeneration.test.ts create mode 100644 src/backend/util/redisGeneration.ts create mode 100644 src/backend/util/sqlLike.test.ts create mode 100644 src/backend/util/sqlLike.ts diff --git a/src/backend/services/share/ShareService.ts b/src/backend/services/share/ShareService.ts index 72506a11f..128234518 100644 --- a/src/backend/services/share/ShareService.ts +++ b/src/backend/services/share/ShareService.ts @@ -2409,7 +2409,7 @@ export class ShareService extends PuterService { const inherited: Array<{ row: ShareIndexRow; via: string }> = issuedHere( - await this.stores.share.listByFsentries([...viaById.keys()]), + await this.stores.share.listReaching([...viaById.keys()]), ).map((row: ShareIndexRow) => ({ row, via: viaById.get(Number(row.fsentry_id)) as string, diff --git a/src/backend/stores/app/AppStore.js b/src/backend/stores/app/AppStore.js index 6741d835d..8c2fca6a4 100644 --- a/src/backend/stores/app/AppStore.js +++ b/src/backend/stores/app/AppStore.js @@ -38,8 +38,10 @@ import { const CACHE_KEY_PREFIX = 'apps'; const CACHE_TTL_SECONDS = 24 * 60 * 60; const LIST_CACHE_KEY_PREFIX = `${CACHE_KEY_PREFIX}:list`; -const LIST_CACHE_TRACKER_KEY = `${LIST_CACHE_KEY_PREFIX}:keys`; const LIST_CACHE_TTL_SECONDS = 15 * 60; +// Outlives every page cached under it, so a lapsed generation can't bring +// one back. +const LIST_GENERATION_TTL_SECONDS = 2 * LIST_CACHE_TTL_SECONDS; const FILETYPE_CACHE_KEY_PREFIX = 'apps:by-filetype'; const FILETYPE_CACHE_TTL_SECONDS = 60; // Filetype associations are matched against the bare lowercase extension @@ -158,6 +160,14 @@ const APP_COLUMN_DEFAULTS = { }; export class AppStore extends PuterStore { + /** @type {import('../types').RowCache} */ + #cache = this.rowCache({ + prefix: CACHE_KEY_PREFIX, + props: APP_ID_PROPERTIES, + ttlSeconds: CACHE_TTL_SECONDS, + revive: (app) => this.#normalizeRow(app), + }); + // -- Reads -------------------------------------------------------- async getByUid(uid) { @@ -262,10 +272,9 @@ export class AppStore extends PuterStore { } /** - * Shared engine behind `getByIds` / `getByUids`. Dedupes input values, - * reads cache via a pipelined MGET, and resolves remaining misses with a - * single `SELECT … WHERE IN (…)` per chunk. The returned map is - * keyed by `prop`'s value. + * Shared engine behind `getByIds` / `getByUids`: a pipelined cache read, + * then one `SELECT … WHERE IN (…)` per chunk of misses. The returned + * map is keyed by `prop`'s value. */ async #getManyByProperty(prop, values) { // `prop` is interpolated into SQL, so it may only ever be one of the @@ -273,76 +282,32 @@ export class AppStore extends PuterStore { if (!BATCH_LOOKUP_COLUMNS.has(prop)) { throw new Error(`AppStore: unsupported batch lookup key ${prop}`); } - - const result = new Map(); - const uniqueValues = [ - ...new Set( - (Array.isArray(values) ? values : []).filter( - (value) => value !== null && value !== undefined, - ), - ), - ]; - if (uniqueValues.length === 0) return result; - - const missingValues = []; - try { - const pipeline = this.clients.redis.pipeline(); - for (const value of uniqueValues) { - pipeline.get(this.#cacheKey(prop, value)); - } - const cacheResults = (await pipeline.exec()) ?? []; - for (let i = 0; i < uniqueValues.length; i++) { - const value = uniqueValues[i]; - const raw = cacheResults[i]?.[1]; - if (typeof raw === 'string') { - try { - result.set(value, JSON.parse(raw)); - continue; - } catch { - // Fall through to DB on any parse failure. - } + const present = (Array.isArray(values) ? values : []).filter( + (value) => value !== null && value !== undefined, + ); + return this.#cache.getMany(prop, present, async (missing, primary) => { + const apps = []; + for ( + let offset = 0; + offset < missing.length; + offset += BULK_QUERY_CHUNK_SIZE + ) { + const chunk = missing.slice( + offset, + offset + BULK_QUERY_CHUNK_SIZE, + ); + const placeholders = chunk.map(() => '?').join(', '); + const sql = `SELECT *, ${this.#createdEpochColumn()} FROM \`apps\` WHERE \`${prop}\` IN (${placeholders})`; + const rows = primary + ? await this.clients.db.pread(sql, chunk) + : await this.clients.db.read(sql, chunk); + for (const row of rows) { + const app = this.#normalizeRow(row); + if (app) apps.push(app); } - missingValues.push(value); } - } catch { - missingValues.push(...uniqueValues); - } - - for ( - let offset = 0; - offset < missingValues.length; - offset += BULK_QUERY_CHUNK_SIZE - ) { - const chunk = missingValues.slice( - offset, - offset + BULK_QUERY_CHUNK_SIZE, - ); - const placeholders = chunk.map(() => '?').join(', '); - const rows = await this.clients.db.read( - `SELECT *, ${this.#createdEpochColumn()} FROM \`apps\` WHERE \`${prop}\` IN (${placeholders})`, - chunk, - ); - for (const row of rows) { - const app = this.#normalizeRow(row); - if (!app) continue; - // Tombstoned: let the primary say if it's really gone. - if ( - await this.isCacheKeyTombstoned([ - this.#cacheKey(prop, app[prop]), - ]) - ) { - const fresh = await this.#readFromDb(prop, app[prop], { - primary: true, - }); - if (fresh) result.set(fresh[prop], fresh); - continue; - } - result.set(app[prop], app); - this.#writeCache(app).catch(() => {}); - } - } - - return result; + return apps; + }); } async existsByName(name) { @@ -353,14 +318,6 @@ export class AppStore extends PuterStore { return rows.length > 0; } - async existsByIndexUrl(indexUrl) { - const rows = await this.clients.db.read( - 'SELECT `id` FROM `apps` WHERE `index_url` = ? LIMIT 1', - [indexUrl], - ); - return rows.length > 0; - } - /** * Find the oldest app whose `index_url` matches one of `candidates`. Used * by the driver to detect duplicate puter-hosted index_url rows @@ -549,10 +506,18 @@ export class AppStore extends PuterStore { const whereClause = where.length ? `WHERE ${where.join(' AND ')}` : ''; const limit = filters.limit ?? 500; const offset = filters.offset ?? 0; - const cacheKey = this.#listCacheKey(whereClause, params, limit, offset); + const generation = await this.#listGeneration(filters.ownerUserId); + const cacheKey = + generation && + `${LIST_CACHE_KEY_PREFIX}:${generation}:${JSON.stringify([ + whereClause, + params, + limit, + offset, + ])}`; try { - const cached = await this.clients.redis.get(cacheKey); + const cached = cacheKey && (await this.clients.redis.get(cacheKey)); if (cached) { const parsed = JSON.parse(cached); if (Array.isArray(parsed)) { @@ -572,7 +537,7 @@ export class AppStore extends PuterStore { const rows = await this.clients.db.read(sql, sqlParams); const apps = rows.map((r) => this.#normalizeRow(r)); - this.#writeListCache(cacheKey, apps).catch(() => {}); + if (cacheKey) void this.#writeListCache(cacheKey, apps); return apps; } @@ -718,7 +683,9 @@ export class AppStore extends PuterStore { [uid], ); if (rows.length === 0) throw error; - return this.#normalizeRow(rows[0]); + const app = this.#normalizeRow(rows[0]); + await this.#writeCache(app); + return app; } const insertId = result?.insertId; if (!insertId) @@ -781,7 +748,7 @@ export class AppStore extends PuterStore { // data instead of returning a copy of the app under its old // name. await this.publishCacheKeys({ - keys: [this.#cacheKey('name', before.name)], + keys: [this.#cache.key('name', before.name)], broadcast: true, }); } @@ -1013,27 +980,26 @@ export class AppStore extends PuterStore { // -- Cache invalidation ------------------------------------------- async invalidate(app) { - const keys = this.#cacheKeysForApp(app); - await this.publishCacheKeys({ keys, broadcast: true }); + await this.#cache.invalidate(app); await this.#invalidateListCachesForApps([app]); } /** Invalidate a deleted row; pass it as read _before_ the delete. */ async markDeleted(app) { - await this.tombstoneCacheKeys(this.#cacheKeysForApp(app)); + await this.#cache.markDeleted(app); await this.#invalidateListCachesForApps([app]); } async invalidateById(id) { const app = - (await this.#readCache('id', id)) ?? + (await this.#cache.read(this.#cache.key('id', id))) ?? (await this.#readFromDb('id', id)); if (app) await this.invalidate(app); } async invalidateByUid(uid) { const app = - (await this.#readCache('uid', uid)) ?? + (await this.#cache.read(this.#cache.key('uid', uid))) ?? (await this.#readFromDb('uid', uid)); if (app) await this.invalidate(app); } @@ -1050,21 +1016,9 @@ export class AppStore extends PuterStore { async #getByProperty(prop, value) { if (value === undefined || value === null) return null; - - const cached = await this.#readCache(prop, value); - if (cached) return cached; - - // Only the primary reliably knows a tombstoned row is gone. - const tombstoned = await this.isCacheKeyTombstoned([ - this.#cacheKey(prop, value), - ]); - const normalized = await this.#readFromDb(prop, value, { - primary: tombstoned, - }); - if (!normalized) return null; - - this.#writeCache(normalized).catch(() => {}); - return normalized; + return this.#cache.get(this.#cache.key(prop, value), (primary) => + this.#readFromDb(prop, value, { primary }), + ); } /** @@ -1174,119 +1128,66 @@ export class AppStore extends PuterStore { return this.#normalizeRow(rows[0]); } - #cacheKey(prop, value) { - return `${CACHE_KEY_PREFIX}:${prop}:${value}`; - } - - #listCacheKey(whereClause, params, limit, offset = 0) { - return `${LIST_CACHE_KEY_PREFIX}:${JSON.stringify([ - whereClause, - params, - limit, - offset, - ])}`; - } - - #parseListCacheKey(cacheKey) { - if (!cacheKey.startsWith(`${LIST_CACHE_KEY_PREFIX}:`)) return null; + /** + * List pages are cached under their owner's generation, or the `all` + * generation when unscoped; a write replaces the generation instead of + * finding the pages. Null when redis is unavailable. + */ + async #listGeneration(ownerUserId) { + const key = this.#listGenerationKey(ownerUserId ?? 'all'); try { - const raw = cacheKey.slice(LIST_CACHE_KEY_PREFIX.length + 1); - const parsed = JSON.parse(raw); - if (!Array.isArray(parsed) || parsed.length < 3) return null; - return { whereClause: parsed[0], params: parsed[1] }; + const current = await this.clients.redis.get(key); + if (current) return current; + const fresh = uuidv4(); + const created = await this.clients.redis.set( + key, + fresh, + 'EX', + LIST_GENERATION_TTL_SECONDS, + 'NX', + ); + return created ? fresh : await this.clients.redis.get(key); } catch { return null; } } - #listCacheMatchesApp(cacheKey, app) { - if (!app) return false; - const parsed = this.#parseListCacheKey(cacheKey); - if (!parsed) return true; - - const { whereClause, params } = parsed; - if (!whereClause) return true; - if (!Array.isArray(params)) return true; - - const columns = whereClause - .replace(/^WHERE\s+/u, '') - .split(' AND ') - .map((part) => part.match(/^`([^`]+)` = \?$/u)?.[1]); - - if (columns.some((column) => !column)) return true; - - for (let i = 0; i < columns.length; i++) { - if (app[columns[i]] !== params[i]) return false; - } - return true; + #listGenerationKey(scope) { + return `${LIST_CACHE_KEY_PREFIX}:gen:${scope}`; } #cacheKeysForApp(app) { - const keys = []; - for (const prop of APP_ID_PROPERTIES) { - if (app[prop] !== undefined && app[prop] !== null) { - keys.push(this.#cacheKey(prop, app[prop])); - } - } - return keys; + return this.#cache.keysFor(app); } - async #readCache(prop, value) { - try { - const raw = await this.clients.redis.get( - this.#cacheKey(prop, value), - ); - return raw ? this.#normalizeRow(JSON.parse(raw)) : null; - } catch { - return null; - } - } - - async #writeCache(app) { - const keys = this.#cacheKeysForApp(app); - const serialized = JSON.stringify(app); - // A replica behind the delete still returns the row. - await this.writeCacheUnlessDeleted(keys, async () => { - await Promise.all( - keys.map((k) => - this.clients.redis.set( - k, - serialized, - 'EX', - CACHE_TTL_SECONDS, - ), - ), - ); - }); + /** Caches `app` on this node unless it was deleted. Never throws. */ + #writeCache(app) { + return this.#cache.write([app]); } async #writeListCache(cacheKey, apps) { - const pipeline = this.clients.redis.pipeline(); - pipeline.set( - cacheKey, - JSON.stringify(apps), - 'EX', - LIST_CACHE_TTL_SECONDS, - ); - pipeline.sadd(LIST_CACHE_TRACKER_KEY, cacheKey); - pipeline.expire(LIST_CACHE_TRACKER_KEY, LIST_CACHE_TTL_SECONDS); - await pipeline.exec(); + try { + await this.clients.redis.set( + cacheKey, + JSON.stringify(apps), + 'EX', + LIST_CACHE_TTL_SECONDS, + ); + } catch { + // The next read fills it. + } } + /** Orphans every cached page these apps' owners, or anyone, could see. */ async #invalidateListCachesForApps(apps) { - let keys = []; - try { - keys = await this.clients.redis.smembers(LIST_CACHE_TRACKER_KEY); - } catch { - return; + const scopes = new Set(['all']); + for (const app of apps) { + if (app?.owner_user_id != null) scopes.add(app.owner_user_id); } - if (!Array.isArray(keys)) keys = []; - keys = keys.filter((key) => - apps.some((app) => this.#listCacheMatchesApp(key, app)), - ); - if (keys.length === 0) return; await this.publishCacheKeys({ - keys, + keys: [...scopes].map((scope) => this.#listGenerationKey(scope)), + serializedData: uuidv4(), + ttlSeconds: LIST_GENERATION_TTL_SECONDS, broadcast: true, }); } diff --git a/src/backend/stores/app/AppStore.test.js b/src/backend/stores/app/AppStore.test.js index 75816deca..0ab3029bf 100644 --- a/src/backend/stores/app/AppStore.test.js +++ b/src/backend/stores/app/AppStore.test.js @@ -596,15 +596,11 @@ describe('AppStore CRUD and cache invalidation', () => { expect((await appStore.getById(bad.id)).metadata).toBeNull(); }); - it('existsByName / existsByIndexUrl answer the driver uniqueness checks', async () => { + it('existsByName answers the driver uniqueness check', async () => { const app = await createApp(); expect(await appStore.existsByName(app.name)).toBe(true); expect(await appStore.existsByName('definitely-not-taken')).toBe(false); - expect(await appStore.existsByIndexUrl(app.index_url)).toBe(true); - expect( - await appStore.existsByIndexUrl('https://unused.example.com/'), - ).toBe(false); }); it('rejects a duplicate app name at the database level', async () => { @@ -928,6 +924,26 @@ describe('AppStore CRUD and cache invalidation', () => { expect(second.uid).toBe(uid); }); + it('caches the row a lost bootstrap race reads from the primary', async () => { + const uid = `app-origin-${Math.random().toString(36).slice(2, 10)}`; + const origin = `https://${uid}.example.com`; + const first = await appStore.createFromOrigin(uid, origin); + // Let the read-through backfill land before clearing it. + await new Promise((r) => setTimeout(r, 20)); + await redis.del( + `apps:id:${first.id}`, + `apps:uid:${uid}`, + `apps:name:${uid}`, + ); + + await appStore.createFromOrigin(uid, origin); + + const cached = JSON.parse( + (await redis.get(`apps:uid:${uid}`)) ?? 'null', + ); + expect(cached?.id).toBe(first.id); + }); + // -- list / count --------------------------------------------------- it('filters list() by owner, creating app and name', async () => { @@ -1041,15 +1057,51 @@ describe('AppStore CRUD and cache invalidation', () => { await createApp({ title: 'Recovered' }, { ownerUserId: badOwner }); await appStore.list({ ownerUserId: badOwner }); - const keys = await redis.smembers('apps:list:keys'); - const key = keys.find((k) => k.includes(String(badOwner))); - expect(key).toBeDefined(); - await redis.set(key, 'not-json'); + const generation = await redis.get(`apps:list:gen:${badOwner}`); + const keys = await redis.keys(`apps:list:${generation}:*`); + expect(keys).toHaveLength(1); + await redis.set(keys[0], 'not-json'); const rows = await appStore.list({ ownerUserId: badOwner }); expect(rows[0].title).toBe('Recovered'); }); + it('drops an unscoped cached list when any app changes', async () => { + const app = await createApp({ title: 'Unscoped' }); + const find = async () => + (await appStore.list({ afterId: app.id - 1, limit: 1 }))[0]; + + expect((await find()).title).toBe('Unscoped'); + await db.write('UPDATE `apps` SET `title` = ? WHERE `id` = ?', [ + 'BehindTheCache', + app.id, + ]); + expect((await find()).title).toBe('Unscoped'); + + await createApp({}, { ownerUserId: 4701 }); + expect((await find()).title).toBe('BehindTheCache'); + }); + + it('invalidates cached lists without scanning them', async () => { + const owner = 4800; + const app = await createApp({}, { ownerUserId: owner }); + for (let i = 0; i < 5; i++) { + await appStore.list({ ownerUserId: owner, limit: i + 1 }); + } + + const smembers = vi.spyOn(redis, 'smembers'); + const keysScan = vi.spyOn(redis, 'keys'); + await appStore.update(app.id, { title: 'Bumped' }); + expect(smembers).not.toHaveBeenCalled(); + expect(keysScan).not.toHaveBeenCalled(); + smembers.mockRestore(); + keysScan.mockRestore(); + + expect( + (await appStore.list({ ownerUserId: owner, limit: 3 }))[0].title, + ).toBe('Bumped'); + }); + it('counts apps by owner, creator and per-caller visibility', async () => { const countOwner = 4600; const viewer = 4601; diff --git a/src/backend/stores/events/DurableSubscriptionStore.ts b/src/backend/stores/events/DurableSubscriptionStore.ts index 40f96caf9..36f1f9adb 100644 --- a/src/backend/stores/events/DurableSubscriptionStore.ts +++ b/src/backend/stores/events/DurableSubscriptionStore.ts @@ -26,8 +26,8 @@ import { HttpError } from '../../core/http/HttpError.js'; import type { DeliveryClass } from '../../services/events/registry.js'; import type { FsOp } from '../../services/events/subjects.js'; import { - openCursor, - sealCursor, + keysetPage, + openIdCursor, type PageResult, } from '../../util/pagination.js'; import { PuterStore } from '../types.js'; @@ -429,11 +429,8 @@ export class DurableSubscriptionStore extends PuterStore { }); } - const owners = new Set(); - for (const row of suspended) { - await this.stores.eventSubscription.dropDurable(row); - owners.add(row.ownerUserId); - } + await this.stores.eventSubscription.dropDurables(suspended); + const owners = new Set(suspended.map((row) => row.ownerUserId)); const bumps = await Promise.all( [...owners].map((owner) => this.#bump(owner)), ); @@ -587,9 +584,7 @@ export class DurableSubscriptionStore extends PuterStore { ), DURABLE_LIST_LIMIT_CAP, ); - const after = asNumber( - openCursor(options.cursor, this.config.jwt_secret_v2)?.id, - ); + const after = openIdCursor(options.cursor, this.config.jwt_secret_v2); // The scope half of the predicate is what the total counts over; the // cursor half only positions one page inside it. @@ -613,15 +608,11 @@ export class DurableSubscriptionStore extends PuterStore { [...params, limit + 1], ); - const page = rows.slice(0, limit); + const page = keysetPage(rows, limit, this.config.jwt_secret_v2); const result: PageResult = { - items: page.map(toRow), + items: page.rows.map(toRow), }; - if (rows.length > limit) - result.cursor = sealCursor( - { id: Number(page[page.length - 1].id) }, - this.config.jwt_secret_v2, - ); + if (page.cursor) result.cursor = page.cursor; if (options.includeTotal) { const [count] = await this.clients.db.read( @@ -923,13 +914,12 @@ export class DurableSubscriptionStore extends PuterStore { rows.map((row) => row.subId), ); - const owners = new Set(); + await this.stores.eventSubscription.dropDurables(rows); for (const row of rows) { - await this.stores.eventSubscription.dropDurable(row); // Whatever was still owed to a row that no longer exists. await this.stores.pendingDelivery.purge(row.subId).catch(() => {}); - owners.add(row.ownerUserId); } + const owners = new Set(rows.map((row) => row.ownerUserId)); for (const ownerUserId of owners) await this.#bump(ownerUserId); return rows.length; } diff --git a/src/backend/stores/events/EventHandlerStore.ts b/src/backend/stores/events/EventHandlerStore.ts index 4ec68bbd1..b0cdb497a 100644 --- a/src/backend/stores/events/EventHandlerStore.ts +++ b/src/backend/stores/events/EventHandlerStore.ts @@ -25,7 +25,7 @@ import { import { HttpError } from '../../core/http/HttpError.js'; import { decodeCursor, - encodeCursor, + keysetPage, type PageResult, } from '../../util/pagination.js'; import { isUniqueViolation } from '../../util/dbError.js'; @@ -477,9 +477,12 @@ export class EventHandlerStore extends PuterStore { [...params, limit + 1], ); - const page = rows.slice(0, limit); + // An app uid rather than a sequence position, so it needn't be sealed. + const page = keysetPage(rows, limit, undefined, (last) => ({ + appUid: last.app_uid, + })); const result: PageResult = { - items: page.map((row) => ({ + items: page.rows.map((row) => ({ appUid: String(row.app_uid), appName: String(row.app_name), appTitle: String(row.app_title), @@ -488,10 +491,7 @@ export class EventHandlerStore extends PuterStore { updatedAt: Number(row.updated_at) || 0, })), }; - if (rows.length > limit) - result.cursor = encodeCursor({ - appUid: page[page.length - 1]!.app_uid, - }); + if (page.cursor) result.cursor = page.cursor; return result; } diff --git a/src/backend/stores/events/EventSubscriptionStore.test.ts b/src/backend/stores/events/EventSubscriptionStore.test.ts index 3cf9daafb..6952925c5 100644 --- a/src/backend/stores/events/EventSubscriptionStore.test.ts +++ b/src/backend/stores/events/EventSubscriptionStore.test.ts @@ -913,6 +913,25 @@ describe('reading rows for dispatch', () => { expect(kept).toEqual([session]); }); + it('drops many cached rows in one pass per owner', async () => { + const rows = durableRows(3); + await store.cacheDurable(rows); + + const hdel = vi.spyOn(redis, 'hdel'); + try { + await store.dropDurables(rows); + expect(hdel).toHaveBeenCalledTimes(1); + } finally { + hdel.mockRestore(); + } + await expect(store.getForTokens(USER, ['f#anchor'])).resolves.toEqual( + [], + ); + await expect(store.watchedTokens(USER, ['f#anchor'])).resolves.toEqual( + [], + ); + }); + it('bounds each token on its own', async () => { await store.cacheDurable(durableRows(100)); await store.add(makeSub({ token: 'f#other', anchorUid: 'other' })); diff --git a/src/backend/stores/events/EventSubscriptionStore.ts b/src/backend/stores/events/EventSubscriptionStore.ts index 50c8dd8bf..e57e1733d 100644 --- a/src/backend/stores/events/EventSubscriptionStore.ts +++ b/src/backend/stores/events/EventSubscriptionStore.ts @@ -19,6 +19,7 @@ import { EVENTS_SESSION_SUBSCRIPTIONS_PER_SOCKET } from '../../controllers/events/limits.js'; import { HttpError } from '../../core/http/HttpError.js'; +import { bumpGeneration } from '../../util/redisGeneration.js'; import { PuterStore } from '../types.js'; import type { DispatchSubscription, @@ -818,13 +819,33 @@ export class EventSubscriptionStore extends PuterStore { token: string; subId: string; }): Promise { - await this.clients.redis.hdel( - durableMapKey(row.ownerUserId), - row.subId, - ); - await this.#dropRows(row.ownerUserId, [ - { token: row.token, subId: row.subId }, - ]); + await this.dropDurables([row]); + } + + /** `dropDurable` for many rows: one pass per owner rather than per row. */ + async dropDurables( + rows: ReadonlyArray<{ + ownerUserId: number; + token: string; + subId: string; + }>, + ): Promise { + const byOwner = new Map< + number, + Array<{ token: string; subId: string }> + >(); + for (const { ownerUserId, token, subId } of rows) { + const owned = byOwner.get(ownerUserId); + if (owned) owned.push({ token, subId }); + else byOwner.set(ownerUserId, [{ token, subId }]); + } + for (const [ownerUserId, owned] of byOwner) { + await this.clients.redis.hdel( + durableMapKey(ownerUserId), + ...owned.map((row) => row.subId), + ); + await this.#dropRows(ownerUserId, owned); + } } /** @@ -1135,15 +1156,17 @@ export class EventSubscriptionStore extends PuterStore { // -- Generation -------------------------------------------------- /** - * Advance the user's subscription-set generation. A single-key `INCR`, so - * it is cluster-safe and costs one command; the broadcast that carries it - * is what actually invalidates other processes. + * Advance the user's subscription-set generation. Single-key, so it is + * cluster-safe; the broadcast that carries it is what actually invalidates + * other processes. */ async bumpGeneration(userId: number | string): Promise { - const key = generationKey(userId); - const next = await this.clients.redis.incr(key); - await this.clients.redis.expire(key, GENERATION_TTL_SECONDS); - return typeof next === 'number' ? next : 0; + const next = await bumpGeneration( + this.clients.redis, + generationKey(userId), + GENERATION_TTL_SECONDS, + ); + return Number.isFinite(next) ? next : 0; } async getGeneration(userId: number | string): Promise { diff --git a/src/backend/stores/events/KvShareHandleStore.ts b/src/backend/stores/events/KvShareHandleStore.ts index e17da439e..9db58d99e 100644 --- a/src/backend/stores/events/KvShareHandleStore.ts +++ b/src/backend/stores/events/KvShareHandleStore.ts @@ -19,8 +19,8 @@ import { mintKvHandleId } from '../../services/events/kvShares.js'; import { - openCursor, - sealCursor, + keysetPage, + openIdCursor, type PageResult, } from '../../util/pagination.js'; import { PuterStore } from '../types.js'; @@ -78,11 +78,6 @@ export interface FindLiveKvShareHandleInput { const nowSeconds = (): number => Math.floor(Date.now() / 1000); -const asNumber = (value: unknown): number | null => { - const parsed = Number(value); - return Number.isFinite(parsed) ? parsed : null; -}; - const toRow = (row: Record): KvShareHandle => ({ handle: String(row.handle), ownerUserId: Number(row.owner_user_id), @@ -254,9 +249,7 @@ export class KvShareHandleStore extends PuterStore { ), KV_HANDLE_LIST_LIMIT_CAP, ); - const after = asNumber( - openCursor(options.cursor, this.config.jwt_secret_v2)?.id, - ); + const after = openIdCursor(options.cursor, this.config.jwt_secret_v2); const filter = ['`owner_user_id` = ?']; const filterParams: unknown[] = [ownerUserId]; @@ -278,13 +271,11 @@ export class KvShareHandleStore extends PuterStore { [...params, limit + 1], ); - const page = rows.slice(0, limit); - const result: PageResult = { items: page.map(toRow) }; - if (rows.length > limit) - result.cursor = sealCursor( - { id: Number(page[page.length - 1].id) }, - this.config.jwt_secret_v2, - ); + const page = keysetPage(rows, limit, this.config.jwt_secret_v2); + const result: PageResult = { + items: page.rows.map(toRow), + }; + if (page.cursor) result.cursor = page.cursor; if (options.includeTotal) { const [count] = await this.clients.db.read( diff --git a/src/backend/stores/events/PresenceStore.ts b/src/backend/stores/events/PresenceStore.ts index 1dec4f92e..b212ae7e6 100644 --- a/src/backend/stores/events/PresenceStore.ts +++ b/src/backend/stores/events/PresenceStore.ts @@ -18,6 +18,7 @@ */ import { CONCURRENT_SLOT_TTL_MS } from '../../core/http/middleware/rateLimit.js'; +import { bumpGeneration } from '../../util/redisGeneration.js'; import { PuterStore } from '../types.js'; import { KV_GLOBAL_APP_KEY } from '../systemKv/SystemKVStore.js'; @@ -458,11 +459,12 @@ export class PresenceStore extends PuterStore { // -- Generation -------------------------------------------------- - /** Advance the user's presence generation. One key, so one command. */ + /** Advance the user's presence generation. */ async bumpGeneration(userId: number): Promise { - const key = generationKey(userId); - const next = await this.clients.redis.incr(key); - await this.clients.redis.expire(key, GENERATION_TTL_SECONDS); - return typeof next === 'number' ? next : Number(next); + return bumpGeneration( + this.clients.redis, + generationKey(userId), + GENERATION_TTL_SECONDS, + ); } } diff --git a/src/backend/stores/fs/FSEntryStore.test.ts b/src/backend/stores/fs/FSEntryStore.test.ts index 588345a11..c1f76d4ff 100644 --- a/src/backend/stores/fs/FSEntryStore.test.ts +++ b/src/backend/stores/fs/FSEntryStore.test.ts @@ -697,6 +697,37 @@ describe('FSEntryStore batched lookups', () => { expect(again.size).toBe(1); }); + it('reads a warm batch from the cache in one round trip', async () => { + const user = await makeUser(); + const files = await Promise.all( + [1, 2, 3, 4].map((n) => + createFile(user, `${user.home}/Documents/warm-${n}.txt`), + ), + ); + await store.getEntriesByIds(files.map((f) => f.id)); + await store.getEntriesByPaths(files.map((f) => f.path)); + + const redis = server.clients.redis; + const get = vi.spyOn(redis, 'get'); + const pipeline = vi.spyOn(redis, 'pipeline'); + const read = vi.spyOn(server.clients.db, 'read'); + try { + const byId = await store.getEntriesByIds(files.map((f) => f.id)); + const byPath = await store.getEntriesByPaths( + files.map((f) => f.path), + ); + expect(byId.size).toBe(files.length); + expect(byPath.size).toBe(files.length); + expect(get).not.toHaveBeenCalled(); + expect(pipeline).toHaveBeenCalledTimes(2); + expect(read).not.toHaveBeenCalled(); + } finally { + get.mockRestore(); + pipeline.mockRestore(); + read.mockRestore(); + } + }); + it('hides entries outside the caller namespace unless crossNamespace is set', async () => { const owner = await makeUser(); const other = await makeUser(); @@ -818,9 +849,9 @@ describe('FSEntryStore directory resolution', () => { await expect( store.resolveParentDirectoriesBatch(1, []), ).resolves.toEqual([]); - await expect(store.ensureDirectoriesForUser(1, [])).resolves.toEqual( - [], - ); + await expect( + store.ensureDirectoriesForUserWithCreated(1, []), + ).resolves.toEqual({ entries: [], createdDirectoryEntries: [] }); await expect( store.resolveParentDirectoriesBatchWithCreated(1, []), ).resolves.toEqual({ parentEntries: [], createdDirectoryEntries: [] }); @@ -854,13 +885,14 @@ describe('FSEntryStore directory resolution', () => { it('ensures directories and refuses the root path', async () => { const user = await makeUser(); - const entries = await store.ensureDirectoriesForUser(user.userId, [ - { path: `${user.home}/Documents/ens/deep`, createPaths: true }, - ]); + const { entries } = await store.ensureDirectoriesForUserWithCreated( + user.userId, + [{ path: `${user.home}/Documents/ens/deep`, createPaths: true }], + ); expect(entries[0]?.path).toBe(`${user.home}/Documents/ens/deep`); const root = await caught(() => - store.ensureDirectoriesForUser(user.userId, [ + store.ensureDirectoriesForUserWithCreated(user.userId, [ { path: '/', createPaths: true }, ]), ); @@ -873,7 +905,7 @@ describe('FSEntryStore directory resolution', () => { await createFile(user, `${user.home}/Documents/occupier`); const missing = await caught(() => - store.ensureDirectoriesForUser(user.userId, [ + store.ensureDirectoriesForUserWithCreated(user.userId, [ { path: `${user.home}/Documents/never`, createPaths: false, @@ -884,7 +916,7 @@ describe('FSEntryStore directory resolution', () => { expect(missing.message).toContain('Directory path does not exist'); const occupied = await caught(() => - store.ensureDirectoriesForUser(user.userId, [ + store.ensureDirectoriesForUserWithCreated(user.userId, [ { path: `${user.home}/Documents/occupier`, createPaths: false, @@ -1001,7 +1033,7 @@ describe('FSEntryStore entry creation', () => { it('refuses to clobber an existing entry without overwrite, or a directory with it', async () => { const user = await makeUser(); await createFile(user, `${user.home}/Documents/keep.txt`); - await store.ensureDirectoriesForUser(user.userId, [ + await store.ensureDirectoriesForUserWithCreated(user.userId, [ { path: `${user.home}/Documents/keepdir`, createPaths: true }, ]); @@ -1750,7 +1782,7 @@ describe('FSEntryStore search', () => { it('restricts results to a path scope and caps the limit', async () => { const user = await makeUser(); - await store.ensureDirectoriesForUser(user.userId, [ + await store.ensureDirectoriesForUserWithCreated(user.userId, [ { path: `${user.home}/Documents/scope`, createPaths: true }, ]); await createFile(user, `${user.home}/Documents/scope/inside.txt`); @@ -1949,7 +1981,7 @@ describe('FSEntryStore home and prefix rewrites', () => { it('rewrites a path prefix and reports how many rows moved', async () => { const user = await makeUser(); - await store.ensureDirectoriesForUser(user.userId, [ + await store.ensureDirectoriesForUserWithCreated(user.userId, [ { path: `${user.home}/Documents/old/inner`, createPaths: true }, ]); await createFile(user, `${user.home}/Documents/old/inner/f.txt`); diff --git a/src/backend/stores/fs/FSEntryStore.ts b/src/backend/stores/fs/FSEntryStore.ts index 6a360e8a9..63aae9176 100644 --- a/src/backend/stores/fs/FSEntryStore.ts +++ b/src/backend/stores/fs/FSEntryStore.ts @@ -24,13 +24,14 @@ import { v4 as uuidv4 } from 'uuid'; import { HttpError } from '../../core/http/HttpError.js'; import type { LayerInstances } from '../../types.js'; import { runWithConcurrencyLimit } from '../../util/concurrency.js'; +import { escapeLike } from '../../util/sqlLike.js'; import { + keysetPage, normalizeLimit, openCursor, - sealCursor, } from '../../util/pagination.js'; import type { puterStores } from '../index.js'; -import { PuterStore } from '../types.js'; +import { PuterStore, readCacheValues } from '../types.js'; import { AncestorChain, FSEntry, @@ -66,6 +67,18 @@ const DESCENDANT_SORT_EXPRESSIONS = { type DescendantSortField = keyof typeof DESCENDANT_SORT_EXPRESSIONS; +// One directory's children sort by bare name; otherwise as above. Anything +// unrecognised sorts by name. +const CHILD_SORT_EXPRESSIONS: Record = { + ...DESCENDANT_SORT_EXPRESSIONS, + name: 'name', +}; +const childSortExpression = (sortBy: unknown): string => + typeof sortBy === 'string' && + Object.prototype.hasOwnProperty.call(CHILD_SORT_EXPRESSIONS, sortBy) + ? CHILD_SORT_EXPRESSIONS[sortBy as DescendantSortField] + : CHILD_SORT_EXPRESSIONS.name; + // Cursors are opaque to callers but arrive from the wire, so the sort they pin // is validated rather than trusted. const toDescendantSortField = (value: unknown): DescendantSortField => { @@ -80,8 +93,12 @@ const toDescendantSortField = (value: unknown): DescendantSortField => { }); }; -// The value the next page seeks from — must match the sort expression above. -const descendantSortValue = (row: FSEntryRow, sortBy: DescendantSortField) => { +// The value the next page seeks from — must match the sort expressions above. +const sortValue = ( + row: FSEntryRow, + sortBy: DescendantSortField, + names: 'name' | 'path', +) => { switch (sortBy) { case 'modified': return row.modified ?? 0; @@ -91,11 +108,15 @@ const descendantSortValue = (row: FSEntryRow, sortBy: DescendantSortField) => { return row.is_dir; case 'name': default: - return row.path; + return row[names]; } }; const ENTRY_CACHE_TTL_SECONDS = 60; +const entryCacheKey = ( + by: 'id' | 'uuid' | 'path', + value: string | number, +): string => `prodfsv2:fsentry:${by === 'path' ? 'path:any' : by}:${value}`; // A renamed home's descendants stay cached under the old path until their TTL // (longer behind a lagging replica), and ACL grants a home by path prefix, so // no other account may take the vacated name until well after that. @@ -329,19 +350,34 @@ export class FSEntryStore extends PuterStore { #entryCacheKeys(entry: FSEntry): string[] { return [ - `prodfsv2:fsentry:id:${entry.id}`, - `prodfsv2:fsentry:uuid:${entry.uuid}`, - `prodfsv2:fsentry:path:any:${entry.path}`, + entryCacheKey('id', entry.id), + entryCacheKey('uuid', entry.uuid), + entryCacheKey('path', entry.path), ]; } async #readEntryFromCache(cacheKey: string): Promise { try { - const cached = await this.clients.redis.get(cacheKey); - if (!cached) { - return null; - } - return JSON.parse(cached) as FSEntry; + return this.#parseCachedEntry( + await this.clients.redis.get(cacheKey), + ); + } catch { + return null; + } + } + + /** `#readEntryFromCache` for many keys, in one pipelined round trip. */ + async #readEntriesFromCache( + keys: string[], + ): Promise> { + const values = await readCacheValues(this.clients.redis, keys); + return values.map((raw) => this.#parseCachedEntry(raw)); + } + + #parseCachedEntry(raw: string | null): FSEntry | null { + if (!raw) return null; + try { + return JSON.parse(raw) as FSEntry; } catch { return null; } @@ -418,9 +454,7 @@ export class FSEntryStore extends PuterStore { path: string, ): Promise { const normalizedPath = this.#normalizePath(path); - const cacheKeys: string[] = [ - `prodfsv2:fsentry:path:any:${normalizedPath}`, - ]; + const cacheKeys: string[] = [entryCacheKey('path', normalizedPath)]; const rows = (await this.clients.db.read( `SELECT ${this.#selectFsentriesColumns()} FROM fsentries WHERE path = ? LIMIT 1`, @@ -458,7 +492,7 @@ export class FSEntryStore extends PuterStore { } const cached = await this.#readEntryFromCache( - `prodfsv2:fsentry:uuid:${uuid}`, + entryCacheKey('uuid', uuid), ); if (cached) { await this.#invalidateEntryCache(cached); @@ -488,9 +522,7 @@ export class FSEntryStore extends PuterStore { return; } - const cached = await this.#readEntryFromCache( - `prodfsv2:fsentry:id:${id}`, - ); + const cached = await this.#readEntryFromCache(entryCacheKey('id', id)); if (cached) { await this.#invalidateEntryCache(cached); return; @@ -641,22 +673,14 @@ export class FSEntryStore extends PuterStore { if (skipCache) { missingPaths.push(...normalizedPaths); } else { - const cacheReads = await Promise.all( - normalizedPaths.map(async (path) => { - const cacheKey = `prodfsv2:fsentry:path:any:${path}`; - const cachedEntry = - await this.#readEntryFromCache(cacheKey); - return { path, cachedEntry }; - }), + const cached = await this.#readEntriesFromCache( + normalizedPaths.map((path) => entryCacheKey('path', path)), ); - - for (const cacheRead of cacheReads) { - if (cacheRead.cachedEntry) { - entriesByPath.set(cacheRead.path, cacheRead.cachedEntry); - } else { - missingPaths.push(cacheRead.path); - } - } + normalizedPaths.forEach((path, i) => { + const entry = cached[i]; + if (entry) entriesByPath.set(path, entry); + else missingPaths.push(path); + }); } const chunks = this.#chunk(missingPaths, BULK_QUERY_CHUNK_SIZE); @@ -1054,7 +1078,7 @@ export class FSEntryStore extends PuterStore { } const normalizedPath = this.#normalizePath(path); - const cacheKey = `prodfsv2:fsentry:path:any:${normalizedPath}`; + const cacheKey = entryCacheKey('path', normalizedPath); const cached = await this.#readEntryFromCache(cacheKey); if (cached) { return cached; @@ -1291,20 +1315,14 @@ export class FSEntryStore extends PuterStore { } const missingPaths: string[] = []; - const cacheReads = await Promise.all( - normalizedPaths.map(async (path) => { - const cacheKey = `prodfsv2:fsentry:path:any:${path}`; - const cachedEntry = await this.#readEntryFromCache(cacheKey); - return { path, cachedEntry }; - }), + const cached = await this.#readEntriesFromCache( + normalizedPaths.map((path) => entryCacheKey('path', path)), ); - for (const { path, cachedEntry } of cacheReads) { - if (cachedEntry) { - entriesByPath.set(path, cachedEntry); - } else { - missingPaths.push(path); - } - } + normalizedPaths.forEach((path, i) => { + const entry = cached[i]; + if (entry) entriesByPath.set(path, entry); + else missingPaths.push(path); + }); if (missingPaths.length > 0) { const chunks = this.#chunk(missingPaths, BULK_QUERY_CHUNK_SIZE); @@ -1394,29 +1412,8 @@ export class FSEntryStore extends PuterStore { return entry; } - async getEntryByUuid(id: string): Promise { - const cacheKey = `prodfsv2:fsentry:uuid:${id}`; - const cached = await this.#readEntryFromCache(cacheKey); - // Treat a cached row with no `path` as a miss — the cache may - // have captured a legacy NULL-path row pre-heal. Falling through - // to the DB read + heal lets the next caller hit the warm cache. - if (cached?.path) return cached; - - const rows = (await this.clients.db.read( - `SELECT ${this.#selectFsentriesColumns()} FROM fsentries WHERE uuid = ? LIMIT 1`, - [id], - )) as unknown as FSEntryRow[]; - const row = rows[0]; - if (!row) { - return null; - } - let entry = this.#mapFSEntryRow(row); - if (!entry.path) { - const healed = await this.#healEntryPathByLineageUp(entry.uuid); - if (healed) entry = healed; - } - await this.#writeEntryToCache(entry); - return entry; + async getEntryByUuid(uuid: string): Promise { + return this.#getEntryBy('uuid', uuid); } /** Uncached primary read, for decisions a stale row must not make. */ @@ -1429,13 +1426,24 @@ export class FSEntryStore extends PuterStore { } async getEntryById(id: number): Promise { - const cacheKey = `prodfsv2:fsentry:id:${id}`; - const cached = await this.#readEntryFromCache(cacheKey); + return this.#getEntryBy('id', id); + } + + async #getEntryBy( + column: 'id' | 'uuid', + value: number | string, + ): Promise { + const cached = await this.#readEntryFromCache( + entryCacheKey(column, value), + ); + // Treat a cached row with no `path` as a miss — the cache may + // have captured a legacy NULL-path row pre-heal. Falling through + // to the DB read + heal lets the next caller hit the warm cache. if (cached?.path) return cached; const rows = (await this.clients.db.read( - `SELECT ${this.#selectFsentriesColumns()} FROM fsentries WHERE id = ? LIMIT 1`, - [id], + `SELECT ${this.#selectFsentriesColumns()} FROM fsentries WHERE ${column} = ? LIMIT 1`, + [value], )) as unknown as FSEntryRow[]; const row = rows[0]; if (!row) { @@ -1451,7 +1459,7 @@ export class FSEntryStore extends PuterStore { } /** - * Batched lookup by id. Dedupes input ids, reads cache via per-id GETs, and + * Batched lookup by id. Dedupes input ids, reads cache in one pipeline, and * resolves remaining misses with a single `SELECT … WHERE id IN (…)` per * chunk. Use this in place of `Promise.all(ids.map(getEntryById))` to avoid * one connection per row on large id sets. @@ -1470,22 +1478,15 @@ export class FSEntryStore extends PuterStore { if (uniqueIds.length === 0) return result; const missingIds: number[] = []; - const cacheReads = await Promise.all( - uniqueIds.map(async (id) => { - const entry = await this.#readEntryFromCache( - `prodfsv2:fsentry:id:${id}`, - ); - return { id, entry }; - }), + const cached = await this.#readEntriesFromCache( + uniqueIds.map((id) => entryCacheKey('id', id)), ); - for (const { id, entry } of cacheReads) { + uniqueIds.forEach((id, i) => { + const entry = cached[i]; // Same NULL-path-cache fallthrough as getEntryById/Uuid. - if (entry?.path) { - result.set(id, entry); - } else { - missingIds.push(id); - } - } + if (entry?.path) result.set(id, entry); + else missingIds.push(id); + }); if (missingIds.length === 0) return result; @@ -1708,17 +1709,6 @@ export class FSEntryStore extends PuterStore { }; } - async ensureDirectoriesForUser( - userId: number, - requests: { path: string; createPaths: boolean }[], - ): Promise { - const { entries } = await this.ensureDirectoriesForUserWithCreated( - userId, - requests, - ); - return entries; - } - async ensureDirectoriesForUserWithCreated( userId: number, requests: { path: string; createPaths: boolean }[], @@ -2692,20 +2682,7 @@ export class FSEntryStore extends PuterStore { ? Math.max(0, Number(options.offset)) : 0; - // Map sort field to a safe column name; reject anything else. - const sortColumn = (() => { - switch (options.sortBy) { - case 'modified': - return 'modified'; - case 'size': - return 'size'; - case 'type': - return 'is_dir'; // directories first when DESC - case 'name': - default: - return 'name'; - } - })(); + const sortColumn = childSortExpression(options.sortBy); const sortDirection = options.sortOrder === 'desc' ? 'DESC' : 'ASC'; const rows = (await this.clients.db.read( @@ -2756,21 +2733,7 @@ export class FSEntryStore extends PuterStore { requestedOrder ?? (payload?.o as typeof requestedOrder) ?? 'asc'; const limit = normalizeLimit(options.limit, { cap: 10_000 }) ?? 1000; - // NULLs would silently drop out of keyset comparisons, so nullable - // sort columns are coalesced in both ORDER BY and the seek condition. - const sortExpr = (() => { - switch (sortBy) { - case 'modified': - return 'COALESCE(modified, 0)'; - case 'size': - return 'COALESCE(size, -1)'; - case 'type': - return 'is_dir'; - case 'name': - default: - return 'name'; - } - })(); + const sortExpr = childSortExpression(sortBy); const dir = sortOrder === 'desc' ? 'DESC' : 'ASC'; const cmp = sortOrder === 'desc' ? '<' : '>'; @@ -2790,33 +2753,19 @@ export class FSEntryStore extends PuterStore { params, )) as unknown as FSEntryRow[]; - const hasMore = rows.length > limit; - const pageRows = hasMore ? rows.slice(0, limit) : rows; - const entries = await this.#finalizeChildEntries(pageRows); - - let cursor: string | undefined; - if (hasMore) { - const last = pageRows[pageRows.length - 1]!; - const v = (() => { - switch (sortBy) { - case 'modified': - return last.modified ?? 0; - case 'size': - return last.size ?? -1; - case 'type': - return last.is_dir; - case 'name': - default: - return last.name; - } - })(); - cursor = sealCursor( - { v, id: Number(last.id), s: sortBy, o: sortOrder }, - this.config.jwt_secret_v2, - ); - } - - return { entries, ...(cursor ? { cursor } : {}) }; + const page = keysetPage( + rows, + limit, + this.config.jwt_secret_v2, + (last) => ({ + v: sortValue(last, sortBy, 'name'), + id: Number(last.id), + s: sortBy, + o: sortOrder, + }), + ); + const entries = await this.#finalizeChildEntries(page.rows); + return { entries, ...(page.cursor ? { cursor: page.cursor } : {}) }; } async countChildren(parentUid: string): Promise { @@ -2839,12 +2788,8 @@ export class FSEntryStore extends PuterStore { return entries; } - // Escape a string for safe use inside a LIKE pattern. We use backslash as - // the LIKE escape char so `%` and `_` in user paths aren't treated as wildcards. - // Uses `!` as the LIKE escape character — both MySQL and SQLite treat `!` as - // a plain character inside string literals, so no dialect-specific quoting. #escapeLikePattern(value: string): string { - return value.replace(/([!%_])/g, '!$1'); + return escapeLike(value); } // All descendants of a directory path (recursive). Paths in fsentries are @@ -3134,25 +3079,19 @@ export class FSEntryStore extends PuterStore { params, )) as unknown as FSEntryRow[]; - const hasMore = rows.length > limit; - const pageRows = hasMore ? rows.slice(0, limit) : rows; - const entries = await this.#finalizeChildEntries(pageRows); - - let cursor: string | undefined; - if (hasMore) { - const last = pageRows[pageRows.length - 1]!; - cursor = sealCursor( - { - v: descendantSortValue(last, sortBy), - ...(tiebreak ? { id: Number(last.id) } : {}), - s: sortBy, - o: sortOrder, - }, - this.config.jwt_secret_v2, - ); - } - - return { entries, ...(cursor ? { cursor } : {}) }; + const page = keysetPage( + rows, + limit, + this.config.jwt_secret_v2, + (last) => ({ + v: sortValue(last, sortBy, 'path'), + ...(tiebreak ? { id: Number(last.id) } : {}), + s: sortBy, + o: sortOrder, + }), + ); + const entries = await this.#finalizeChildEntries(page.rows); + return { entries, ...(page.cursor ? { cursor: page.cursor } : {}) }; } async countDescendantsToDepth( diff --git a/src/backend/stores/notification/NotificationStore.js b/src/backend/stores/notification/NotificationStore.js index f8b98fae4..d74781699 100644 --- a/src/backend/stores/notification/NotificationStore.js +++ b/src/backend/stores/notification/NotificationStore.js @@ -241,7 +241,7 @@ export class NotificationStore extends PuterStore { 'UPDATE `notification` SET `value` = ?, `shown` = NULL WHERE `uid` = ? AND `user_id` = ? AND `acknowledged` IS NULL', [serialized, uid, userId], ); - return (result?.affectedRows ?? result?.changes ?? 0) > 0; + return result.anyRowsAffected; } async markAcknowledged(uid, userId) { @@ -250,7 +250,7 @@ export class NotificationStore extends PuterStore { 'UPDATE `notification` SET `acknowledged` = ? WHERE `uid` = ? AND `user_id` = ? AND `acknowledged` IS NULL', [now, uid, userId], ); - return (result?.affectedRows ?? result?.changes ?? 0) > 0; + return result.anyRowsAffected; } async markShown(uid, userId) { @@ -259,7 +259,7 @@ export class NotificationStore extends PuterStore { 'UPDATE `notification` SET `shown` = ? WHERE `uid` = ? AND `user_id` = ? AND `shown` IS NULL', [now, uid, userId], ); - return (result?.affectedRows ?? result?.changes ?? 0) > 0; + return result.anyRowsAffected; } /** @@ -279,7 +279,7 @@ export class NotificationStore extends PuterStore { `WHERE \`user_id\` = ? AND \`shown\` IS NULL AND \`uid\` IN (${placeholders})`, [now, userId, ...unique], ); - return result?.affectedRows ?? result?.changes ?? 0; + return result.affectedRows; } async deleteByUid(uid, userId) { @@ -287,7 +287,7 @@ export class NotificationStore extends PuterStore { 'DELETE FROM `notification` WHERE `uid` = ? AND `user_id` = ?', [uid, userId], ); - return (result?.affectedRows ?? result?.changes ?? 0) > 0; + return result.anyRowsAffected; } /** @@ -322,7 +322,7 @@ export class NotificationStore extends PuterStore { }); const result = await this.clients.db.write(statement, [batch]); - return result?.affectedRows ?? result?.changes ?? 0; + return result.affectedRows; } // -- Internals ---------------------------------------------------- diff --git a/src/backend/stores/oidc/OIDCStore.js b/src/backend/stores/oidc/OIDCStore.js index 2c7f6015d..5ab6bcd93 100644 --- a/src/backend/stores/oidc/OIDCStore.js +++ b/src/backend/stores/oidc/OIDCStore.js @@ -89,13 +89,6 @@ export class OIDCStore extends PuterStore { } } - async unlinkByUserId(userId, provider) { - await this.clients.db.write( - 'DELETE FROM `user_oidc_providers` WHERE `user_id` = ? AND `provider` = ?', - [userId, provider], - ); - } - async unlinkAllByUserId(userId) { await this.clients.db.write( 'DELETE FROM `user_oidc_providers` WHERE `user_id` = ?', diff --git a/src/backend/stores/permission/PermissionStore.test.ts b/src/backend/stores/permission/PermissionStore.test.ts index ef1e313fe..ac91139d5 100644 --- a/src/backend/stores/permission/PermissionStore.test.ts +++ b/src/backend/stores/permission/PermissionStore.test.ts @@ -20,7 +20,7 @@ import { readFileSync } from 'fs'; import { v4 as uuidv4 } from 'uuid'; -import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; import { PuterServer } from '../../server.ts'; import { setupTestServer } from '../../testUtil.ts'; import { decodeCursor } from '../../util/pagination.ts'; @@ -83,6 +83,34 @@ describe('PermissionStore', () => { expect(await store.getFlatUserPerms(holder.id, [])).toEqual([]); }); + it('drops many flat entries in one batched KV delete', async () => { + const holder = await makeUser(); + const perms = [1, 2, 3].map(() => `fs:${uuidv4()}:read`); + for (const perm of perms) { + await store.setFlatUserPerm(holder.id, perm, { + permission: perm, + }); + } + + const kv = server.stores.kv; + const del = vi.spyOn(kv, 'del'); + const batchDel = vi.spyOn(kv, 'batchDel'); + try { + await store.delFlatUserPerms( + perms.map((permission) => ({ + holderUserId: holder.id, + permission, + })), + ); + expect(batchDel).toHaveBeenCalledTimes(1); + expect(del).not.toHaveBeenCalled(); + } finally { + del.mockRestore(); + batchDel.mockRestore(); + } + expect(await store.getFlatUserPerms(holder.id, perms)).toEqual([]); + }); + it('dedupes repeated permission strings and skips missing keys', async () => { const holder = await makeUser(); const present = `fs:${uuidv4()}:read`; @@ -949,13 +977,6 @@ describe('PermissionStore', () => { ).toHaveLength(1); }); - it('drops a scan cache entry on explicit invalidation', async () => { - const key = store.buildScanCacheKey(`actor-${uuidv4()}`, ['p'], 0); - await store.setScanCache(key, { allowed: false }); - await store.invalidateScanCache(key); - expect(await store.getScanCache(key)).toBeNull(); - }); - it('round-trips a multi-permission check cache, omitting unset entries', async () => { const actorUid = `actor-${uuidv4()}`; await store.setMultiCheckCache( @@ -1086,8 +1107,8 @@ describe('PermissionStore', () => { const keeper = `fs:${uuidv4()}:read`; await store.upsertUserUserPerm(holder.id, issuer.id, keeper, {}); - const removed = await store.deleteUserUserPermsByPermissionPrefix( - `fs:${uid}`, + const removed = await store.deleteUserUserPermsByPermissionPrefixes( + [`fs:${uid}`], ); expect(removed).toHaveLength(3); @@ -1113,10 +1134,10 @@ describe('PermissionStore', () => { // `_` and `%` are LIKE wildcards; unescaped they would match this. expect( - await store.deleteUserUserPermsByPermissionPrefix('fs:%'), + await store.deleteUserUserPermsByPermissionPrefixes(['fs:%']), ).toEqual([]); expect( - await store.deleteUserUserPermsByPermissionPrefix('fs:_'), + await store.deleteUserUserPermsByPermissionPrefixes(['fs:_']), ).toEqual([]); expect( await store.readLinkedUserUserPerms(holder.id, [victim]), @@ -1125,9 +1146,9 @@ describe('PermissionStore', () => { it('returns an empty list when the prefix matches nothing', async () => { expect( - await store.deleteUserUserPermsByPermissionPrefix( + await store.deleteUserUserPermsByPermissionPrefixes([ `fs:${uuidv4()}`, - ), + ]), ).toEqual([]); }); diff --git a/src/backend/stores/permission/PermissionStore.ts b/src/backend/stores/permission/PermissionStore.ts index 1c67dc8e3..a2effda16 100644 --- a/src/backend/stores/permission/PermissionStore.ts +++ b/src/backend/stores/permission/PermissionStore.ts @@ -28,7 +28,9 @@ import { PERMISSION_SCAN_CACHE_TTL_SECONDS, } from '../../services/permission/consts'; import { kv } from '../../util/kvSingleton'; -import { openCursor, sealCursor } from '../../util/pagination'; +import { keysetPage, openIdCursor } from '../../util/pagination'; +import { bumpGeneration } from '../../util/redisGeneration'; +import { escapeLike } from '../../util/sqlLike'; import type { UserRow } from '../user/UserStore'; // Short TTLs: FK CASCADE on user/app delete + PermissionService rewriters @@ -141,17 +143,7 @@ export interface FlatPermRef { const subtreeRoot = (prefix: string): string => prefix.endsWith(':') ? prefix.slice(0, -1) : prefix; -/** - * Match a permission and everything beneath it. - * - * `_` and `%` are LIKE wildcards, so an unescaped one would widen the match - * beyond the intended subtree. `!` as the escape character, matching - * FSEntryStore: a backslash one would have to be written `ESCAPE '\\'` in the - * SQL text, and MySQL processes backslash escapes inside string literals, so - * the `'\'` a JS `'\\'` produces reads as an escaped quote and leaves the - * literal unterminated. SQLite and Postgres accept it, which is why only MySQL - * would have seen the parse error. - */ +/** Match a permission and everything beneath it. */ const subtreeClause = ( permissions: string[], ): { where: string; params: string[] } => ({ @@ -160,7 +152,7 @@ const subtreeClause = ( .join(' OR '), params: permissions.flatMap((permission) => [ permission, - `${permission.replace(/([!%_])/g, '!$1')}:%`, + `${escapeLike(permission)}:%`, ]), }); @@ -320,17 +312,15 @@ export class PermissionStore extends PuterStore { /** Local half of a flat delete. Never emits, so a remote one can't loop. */ async #applyFlatUserPermDeletes(entries: FlatPermRef[]): Promise { - await Promise.all( - entries.map(({ holderUserId, permission }) => - this.stores.kv.del({ - key: PermissionUtil.join( - PERM_KEY_PREFIX, - String(holderUserId), - permission, - ), - }), + await this.stores.kv.batchDel({ + keys: entries.map(({ holderUserId, permission }) => + PermissionUtil.join( + PERM_KEY_PREFIX, + String(holderUserId), + permission, + ), ), - ); + }); } // -- SQL: user-to-user permissions ------------------------------- @@ -510,39 +500,12 @@ export class PermissionStore extends PuterStore { })); } - /** - * Delete every user-to-user grant at or beneath `permission`, clearing the - * flat KV view too, and return the rows removed so the caller can audit - * them and bust caches. - * - * The subject lives in the permission text rather than a column, so no - * foreign key can cascade it — this is how a deleted fsentry's grants get - * withdrawn. `permission` has no index, so this is a table scan: fine on - * deletion, never on a hot path. - */ - async deleteUserUserPermsByPermissionPrefix(permission: string): Promise< - Array<{ - holder_user_id: number; - issuer_user_id: number; - permission: string; - }> - > { - return this.deleteUserUserPermsByPermissionPrefixes([permission]); - } - - /** As above, for several prefixes in one scan. */ /** The group analogue: a deleted node's group grants must go with it. */ async deleteUserGroupPermsByPermissionPrefixes( permissions: string[], ): Promise> { if (permissions.length === 0) return []; - const where = permissions - .map(() => "(`permission` = ? OR `permission` LIKE ? ESCAPE '!')") - .join(' OR '); - const params = permissions.flatMap((permission) => [ - permission, - `${permission.replace(/([!%_])/g, '!$1')}:%`, - ]); + const { where, params } = subtreeClause(permissions); const rows = (await this.clients.db.read( 'SELECT `group_id`, `permission` FROM `user_to_group_permissions` ' + @@ -558,6 +521,16 @@ export class PermissionStore extends PuterStore { return rows; } + /** + * Delete every user-to-user grant at or beneath any of `permissions`, in + * one scan, clearing the flat KV view too, and return the rows removed so + * the caller can audit them and bust caches. + * + * The subject lives in the permission text rather than a column, so no + * foreign key can cascade it — this is how a deleted fsentry's grants get + * withdrawn. `permission` has no index, so this is a table scan: fine on + * deletion, never on a hot path. + */ async deleteUserUserPermsByPermissionPrefixes( permissions: string[], ): Promise< @@ -684,12 +657,10 @@ export class PermissionStore extends PuterStore { Math.max(1, Math.floor(Number(opts.limit) || AUDIT_PAGE_SIZE)), MAX_AUDIT_PAGE_SIZE, ); - const decoded = openCursor( - opts.cursor, - this.config.jwt_secret_v2, - 'audit cursor', - ); - const beforeId = Number(decoded?.id ?? 0) || null; + const beforeId = + openIdCursor(opts.cursor, this.config.jwt_secret_v2, { + label: 'audit cursor', + }) || null; const { sql, params } = this.#auditWhere(filter); const rows = await this.clients.db.read( @@ -700,15 +671,10 @@ export class PermissionStore extends PuterStore { [...params, ...(beforeId === null ? [] : [beforeId]), size + 1], ); - const hasMore = rows.length > size; - const items = rows.slice(0, size).map((row) => this.#auditRow(row)); - const last = items[items.length - 1]; + const page = keysetPage(rows, size, this.config.jwt_secret_v2); return { - items, - cursor: - hasMore && last - ? sealCursor({ id: last.id }, this.config.jwt_secret_v2) - : undefined, + items: page.rows.map((row) => this.#auditRow(row)), + cursor: page.cursor, }; } @@ -1565,16 +1531,16 @@ export class PermissionStore extends PuterStore { async #applyCacheGenerationBump(actorUid: string): Promise { const key = this.#cacheGenerationKey(actorUid); try { - const next = await this.clients.redis.incr(key); // Keep the counter alive well past the cache TTL so it can't // reset to 0 and revive same-generation stale entries. - await this.clients.redis.expire( + const next = await bumpGeneration( + this.clients.redis, key, PERMISSION_CACHE_GENERATION_TTL_SECONDS, ); // Make this node consistent immediately; other nodes pick up the // new value when their local copy expires (≤ local TTL). - if (typeof next === 'number') { + if (Number.isFinite(next)) { kv.set(this.#localCacheGenerationKey(actorUid), next, { EX: PERMISSION_CACHE_GENERATION_LOCAL_TTL_SECONDS, }); @@ -1628,10 +1594,6 @@ export class PermissionStore extends PuterStore { ); } - async invalidateScanCache(cacheKey: string): Promise { - await this.publishCacheKeys({ keys: [cacheKey] }); - } - // -- Per-permission check cache (for `checkMany`) ----------------- // // Cached as `1`/`0` per (actor, permission) pair so a batch lookup diff --git a/src/backend/stores/share/ShareStore.js b/src/backend/stores/share/ShareStore.js index 214df0c66..d030d14c9 100644 --- a/src/backend/stores/share/ShareStore.js +++ b/src/backend/stores/share/ShareStore.js @@ -19,12 +19,8 @@ import { v4 as uuidv4 } from 'uuid'; import { HttpError } from '../../core/http/HttpError.js'; -import { - decodeCursor, - encodeCursor, - openCursor, - sealCursor, -} from '../../util/pagination'; +import { decodeCursor, keysetPage, openIdCursor } from '../../util/pagination'; +import { escapeLike } from '../../util/sqlLike'; import { PuterStore } from '../types'; /** Default page size for the keyset listings. */ @@ -63,22 +59,6 @@ export class ShareStore extends PuterStore { return this.#normalizeRow(rows[0]) ?? null; } - async listByRecipientEmail(email) { - const rows = await this.clients.db.read( - 'SELECT * FROM `share` WHERE `recipient_email` = ? ORDER BY `created_at` DESC', - [email], - ); - return rows.map((r) => this.#normalizeRow(r)); - } - - async listByIssuer(issuerUserId) { - const rows = await this.clients.db.read( - 'SELECT * FROM `share` WHERE `issuer_user_id` = ? ORDER BY `created_at` DESC', - [issuerUserId], - ); - return rows.map((r) => this.#normalizeRow(r)); - } - /** * Active shares held by a user, keyset-paginated. `id` is the tiebreaker, * so a row added mid-iteration can't shift earlier pages. @@ -112,17 +92,10 @@ export class ShareStore extends PuterStore { [...holderParams, afterId, size + 1], ); - const hasMore = rows.length > size; - const items = (hasMore ? rows.slice(0, size) : rows).map((r) => - this.#normalizeRow(r), - ); - const last = items[items.length - 1]; + const page = keysetPage(rows, size, this.config.jwt_secret_v2); return { - items, - cursor: - hasMore && last - ? sealCursor({ id: last.id }, this.config.jwt_secret_v2) - : undefined, + items: page.rows.map((r) => this.#normalizeRow(r)), + cursor: page.cursor, }; } @@ -186,15 +159,10 @@ export class ShareStore extends PuterStore { const merged = [...issued, ...delegated, ...unrecorded].sort( (a, b) => Number(a.id) - Number(b.id), ); - const hasMore = merged.length > size; - const items = merged.slice(0, size).map((r) => this.#normalizeRow(r)); - const last = items[items.length - 1]; + const page = keysetPage(merged, size, this.config.jwt_secret_v2); return { - items, - cursor: - hasMore && last - ? sealCursor({ id: last.id }, this.config.jwt_secret_v2) - : undefined, + items: page.rows.map((r) => this.#normalizeRow(r)), + cursor: page.cursor, }; } @@ -279,18 +247,16 @@ export class ShareStore extends PuterStore { ], ); - const hasMore = rows.length > size; - const items = rows.slice(0, size).map((row) => ({ - appUid: row.app_uid === '' ? null : String(row.app_uid), - count: Number(row.count), + // An app uid rather than a sequence position, so it needn't be sealed. + const page = keysetPage(rows, size, undefined, (last) => ({ + appUid: String(last.app_uid), })); - const last = items[items.length - 1]; return { - items, - cursor: - hasMore && last - ? encodeCursor({ appUid: last.appUid ?? '' }) - : undefined, + items: page.rows.map((row) => ({ + appUid: row.app_uid === '' ? null : String(row.app_uid), + count: Number(row.count), + })), + cursor: page.cursor, }; } @@ -307,11 +273,21 @@ export class ShareStore extends PuterStore { /** Everyone with an active share on one node, whoever issued it. */ async listByFsentry(fsentryId) { - return this.listByFsentries([fsentryId]); + return this.listReaching([fsentryId]); } - /** As above, across several nodes in one query. */ - async listByFsentries(fsentryIds) { + /** + * Active shares on any of `fsentryIds` — a node plus its ancestors, which + * the caller has already resolved to row ids. + * + * This sits behind every file-write event, so it must stay on + * `idx_share_fsentry`: a plain `IN` does, whereas joining `fsentries` and + * OR-ing a path match does not, and the optimizer falls back to a scan of + * `share`. + * + * @param {number[]} fsentryIds + */ + async listReaching(fsentryIds) { if (fsentryIds.length === 0) return []; const placeholders = fsentryIds.map(() => '?').join(', '); const rows = await this.clients.db.read( @@ -322,21 +298,6 @@ export class ShareStore extends PuterStore { return rows.map((r) => this.#normalizeRow(r)); } - /** - * Every active share the holder has on any of `fsentryIds`. Used to find - * which shared root an entry was reached through, in one round trip. - */ - async listByHolderAndFsentries(holderUserId, fsentryIds) { - if (fsentryIds.length === 0) return []; - const placeholders = fsentryIds.map(() => '?').join(', '); - const rows = await this.clients.db.read( - `SELECT * FROM \`share\` WHERE \`holder_user_id\` = ? AND ` + - `\`fsentry_id\` IN (${placeholders}) ORDER BY \`id\``, - [holderUserId, ...fsentryIds], - ); - return rows.map((r) => this.#normalizeRow(r)); - } - /** * Active shares on a directory and everything beneath it. Walks by parent * linkage, not path prefix — `fsentries.path` is lazily backfilled and NULL @@ -393,28 +354,6 @@ export class ShareStore extends PuterStore { return rows.map((r) => this.#normalizeRow(r)); } - /** - * Active shares on any of `fsentryIds` — a node plus its ancestors, which - * the caller has already resolved to row ids. - * - * This sits behind every file-write event, so it must stay on - * `idx_share_fsentry`: a plain `IN` does, whereas joining `fsentries` and - * OR-ing a path match does not, and the optimizer falls back to a scan of - * `share`. - * - * @param {number[]} fsentryIds - */ - async listReaching(fsentryIds) { - if (fsentryIds.length === 0) return []; - const placeholders = fsentryIds.map(() => '?').join(', '); - const rows = await this.clients.db.read( - `SELECT * FROM \`share\` WHERE \`fsentry_id\` IN (${placeholders}) ` + - 'AND `holder_user_id` IS NOT NULL ORDER BY `id`', - fsentryIds, - ); - return rows.map((r) => this.#normalizeRow(r)); - } - /** * Team shares reaching these nodes, one row per member. Members hold * through the group, so without this a shared folder never pushes them a @@ -950,7 +889,7 @@ export class ShareStore extends PuterStore { ? [holderUserId, fsentryId, issuerUserId] : [holderUserId, fsentryId], ); - return (result?.affectedRows ?? result?.changes ?? 0) > 0; + return result.anyRowsAffected; } /** @@ -998,7 +937,7 @@ export class ShareStore extends PuterStore { `DELETE FROM \`share\` WHERE \`fsentry_id\` IN (${placeholders})`, fsentryIds, ); - return result?.affectedRows ?? result?.changes ?? 0; + return result.affectedRows; } /** @@ -1019,12 +958,11 @@ export class ShareStore extends PuterStore { * @param {number} newOwnerId @param {string} path */ async reassignEntryOwnerUnder(newOwnerId, path) { - const escaped = path.replace(/([!%_])/g, '!$1'); await this.clients.db.write( 'UPDATE `share` SET `entry_owner_user_id` = ? WHERE `fsentry_id` ' + 'IN (SELECT `id` FROM `fsentries` WHERE `path` = ? OR ' + "`path` LIKE ? ESCAPE '!')", - [newOwnerId, path, `${escaped}/%`], + [newOwnerId, path, `${escapeLike(path)}/%`], ); } @@ -1063,7 +1001,7 @@ export class ShareStore extends PuterStore { `DELETE FROM \`share\` WHERE \`uid\` IN (${placeholders})`, retired.map((row) => row.uid), ); - return result?.affectedRows ?? result?.changes ?? 0; + return result.affectedRows; } async deleteByUid(uid) { @@ -1071,15 +1009,7 @@ export class ShareStore extends PuterStore { 'DELETE FROM `share` WHERE `uid` = ?', [uid], ); - return (result?.affectedRows ?? result?.changes ?? 0) > 0; - } - - async deleteByRecipientEmail(email) { - const result = await this.clients.db.write( - 'DELETE FROM `share` WHERE `recipient_email` = ?', - [email], - ); - return (result?.affectedRows ?? result?.changes ?? 0) > 0; + return result.anyRowsAffected; } // -- Anyone with the link ----------------------------------------- @@ -1231,7 +1161,7 @@ export class ShareStore extends PuterStore { 'DELETE FROM `share` WHERE `fsentry_id` = ? AND `anyone` = 1', [fsentryId], ); - const removed = (result?.affectedRows ?? result?.changes ?? 0) > 0; + const removed = result.anyRowsAffected; if (removed) await this.#markLinkRevoked(); return removed; } @@ -1254,15 +1184,6 @@ export class ShareStore extends PuterStore { // Counted in KV, not by querying `share`: the ceiling is on shares // *created*, so a COUNT of live rows would let a revoke recycle the slot. - /** @param {number} userId */ - async getDailyShareCount(userId) { - const { res } = await this.stores.kv.get({ - key: this.#dailyQuotaKey(userId), - }); - const count = /** @type {{ count?: unknown } | null} */ (res)?.count; - return typeof count === 'number' ? count : 0; - } - /** * @param {number} userId * @param {number} [amount] @@ -1336,25 +1257,17 @@ export class ShareStore extends PuterStore { } /** - * The id a keyset page resumes after; 0 for the first page. A cursor that - * decodes but names no usable id — another endpoint's cursor, say — is - * refused rather than read as page one, which would silently restart a - * client's iteration from the top. + * The id a keyset page resumes after; 0 for the first page. A cursor + * without a usable id is refused rather than read as page one, which would + * silently restart a client's iteration from the top. */ #afterId(cursor) { - const decoded = openCursor( - cursor, - this.config.jwt_secret_v2, - 'share cursor', + return ( + openIdCursor(cursor, this.config.jwt_secret_v2, { + label: 'share cursor', + strict: true, + }) ?? 0 ); - if (decoded === undefined) return 0; - const id = Number(decoded.id); - if (!Number.isInteger(id) || id < 0) { - throw new HttpError(400, 'invalid share cursor', { - legacyCode: 'bad_request', - }); - } - return id; } /** diff --git a/src/backend/stores/share/ShareStore.test.js b/src/backend/stores/share/ShareStore.test.js index 004d4e264..23754d91c 100644 --- a/src/backend/stores/share/ShareStore.test.js +++ b/src/backend/stores/share/ShareStore.test.js @@ -126,72 +126,6 @@ describe('ShareStore', () => { expect(await store.getByUid('no-such-share')).toBeNull(); }); - it('lists only the shares addressed to the requested recipient', async () => { - const email = `recipient-${Date.now()}@test.local`; - const mine = await store.create({ - issuerUserId: issuer.id, - recipientEmail: email, - data: { n: 1 }, - }); - await store.create({ - issuerUserId: issuer.id, - recipientEmail: `other-${Date.now()}@test.local`, - data: { n: 2 }, - }); - - const rows = await store.listByRecipientEmail(email); - expect(rows.map((r) => r.uid)).toEqual([mine.uid]); - expect(rows[0].data).toEqual({ n: 1 }); - }); - - it('returns an empty list for a recipient with no shares', async () => { - expect(await store.listByRecipientEmail('nobody@test.local')).toEqual( - [], - ); - }); - - it('never leaks another issuer rows through listByIssuer', async () => { - const mine = await store.create({ - issuerUserId: issuer.id, - recipientEmail: 'iso-a@test.local', - data: {}, - }); - const theirs = await store.create({ - issuerUserId: otherIssuer.id, - recipientEmail: 'iso-b@test.local', - data: {}, - }); - - const uids = (await store.listByIssuer(otherIssuer.id)).map( - (r) => r.uid, - ); - expect(uids).toContain(theirs.uid); - expect(uids).not.toContain(mine.uid); - }); - - it('orders a recipient inbox newest-first', async () => { - const email = `ordered-${Date.now()}@test.local`; - const older = await store.create({ - issuerUserId: issuer.id, - recipientEmail: email, - data: { which: 'older' }, - }); - const newer = await store.create({ - issuerUserId: issuer.id, - recipientEmail: email, - data: { which: 'newer' }, - }); - // `created_at` defaults to a second-granularity timestamp, so both - // rows can land in the same second. Age one explicitly. - await server.clients.db.write( - "UPDATE `share` SET `created_at` = '2020-01-01 00:00:00' WHERE `uid` = ?", - [older.uid], - ); - - const rows = await store.listByRecipientEmail(email); - expect(rows.map((r) => r.uid)).toEqual([newer.uid, older.uid]); - }); - // -- deletes ------------------------------------------------------ it('deletes a single share by uid and reports whether a row was removed', async () => { @@ -207,35 +141,6 @@ describe('ShareStore', () => { expect(await store.deleteByUid(created.uid)).toBe(false); }); - it('deletes every share for a recipient in one call', async () => { - const email = `bulk-${Date.now()}@test.local`; - await store.create({ - issuerUserId: issuer.id, - recipientEmail: email, - data: {}, - }); - await store.create({ - issuerUserId: otherIssuer.id, - recipientEmail: email, - data: {}, - }); - const survivor = await store.create({ - issuerUserId: issuer.id, - recipientEmail: `keep-${Date.now()}@test.local`, - data: {}, - }); - - expect(await store.deleteByRecipientEmail(email)).toBe(true); - expect(await store.listByRecipientEmail(email)).toEqual([]); - expect(await store.getByUid(survivor.uid)).not.toBeNull(); - }); - - it('reports false when a recipient bulk delete matches nothing', async () => { - expect(await store.deleteByRecipientEmail('ghost@test.local')).toBe( - false, - ); - }); - // -- active shares (the index) ------------------------------------- describe('active shares', () => { diff --git a/src/backend/stores/subdomain/SubdomainStore.test.ts b/src/backend/stores/subdomain/SubdomainStore.test.ts index 5e89857fe..99a7b425a 100644 --- a/src/backend/stores/subdomain/SubdomainStore.test.ts +++ b/src/backend/stores/subdomain/SubdomainStore.test.ts @@ -405,23 +405,6 @@ describe('SubdomainStore listing and counting', () => { expect(await store.existsBySubdomain(`${p}missing`)).toBe(false); }); - it('resolves custom domains by exact match and by listing', async () => { - const userId = await makeUser(); - const p = prefix(); - const row = await createFor(userId, `${p}dom`); - await store.update(row.uuid, { domain: `${p}example.com` }); - - expect((await store.getByDomain(`${p}example.com`))?.uuid).toBe( - row.uuid, - ); - expect(await store.getByDomain('nobody.example')).toBeNull(); - expect( - (await store.listByDomain(`${p}example.com`)).map( - (r: { uuid: string }) => r.uuid, - ), - ).toEqual([row.uuid]); - }); - it('lists and counts by prefix, scoped to the owner', async () => { const userId = await makeUser(); const other = await makeUser(); @@ -454,6 +437,28 @@ describe('SubdomainStore listing and counting', () => { ).toBe(0); }); + it('matches a prefix literally, not as a LIKE pattern', async () => { + const userId = await makeUser(); + const p = prefix(); + await createFor(userId, `${p}a_b`); + await createFor(userId, `${p}axb`); + + expect( + (await store.listByUserIdAndPrefix(userId, `${p}a_`)).map( + (r) => r.subdomain, + ), + ).toEqual([`${p}a_b`]); + expect(await store.countByUserIdAndPrefix(userId, `${p}a_`)).toBe(1); + expect( + ( + await store.listByUserId(userId, { excludePrefix: `${p}a_` }) + ).map((r) => r.subdomain), + ).toEqual([`${p}axb`]); + expect( + await store.count({ userId, excludePrefix: `${p}a_` }), + ).toBe(1); + }); + it('narrows a prefix listing by a single app id and paginates it', async () => { const userId = await makeUser(); const p = prefix(); @@ -593,6 +598,22 @@ describe('SubdomainStore reads and writes', () => { store.deleteByUuid('no-such-uuid'), ).resolves.toBeUndefined(); }); + + it('writes without scanning any set of cached keys', async () => { + const userId = await makeUser(); + const smembers = vi.spyOn(server.clients.redis, 'smembers'); + try { + const row = await store.create({ + userId, + subdomain: `sds-scan-${Math.random().toString(36).slice(2, 8)}`, + }); + await store.update(row.uuid, { domain: 'scan.example' }); + await store.deleteByUuid(row.uuid); + expect(smembers).not.toHaveBeenCalled(); + } finally { + smembers.mockRestore(); + } + }); }); // -- getBySubdomains (batched) -- diff --git a/src/backend/stores/subdomain/SubdomainStore.ts b/src/backend/stores/subdomain/SubdomainStore.ts index 4133df418..9e989d128 100644 --- a/src/backend/stores/subdomain/SubdomainStore.ts +++ b/src/backend/stores/subdomain/SubdomainStore.ts @@ -18,6 +18,7 @@ */ import { v4 as uuidv4 } from 'uuid'; +import { escapeLike } from '../../util/sqlLike'; import { PuterStore } from '../types'; /** @@ -358,8 +359,8 @@ export class SubdomainStore extends PuterStore { values.push(afterId); } if (excludePrefix) { - conditions.push('`subdomain` NOT LIKE ?'); - values.push(`${excludePrefix}%`); + conditions.push("`subdomain` NOT LIKE ? ESCAPE '!'"); + values.push(`${escapeLike(excludePrefix)}%`); } const whereClause = conditions.length ? `WHERE ${conditions.join(' AND ')}` @@ -393,8 +394,8 @@ export class SubdomainStore extends PuterStore { values.push(appOwner); } if (excludePrefix) { - conditions.push('`subdomain` NOT LIKE ?'); - values.push(`${excludePrefix}%`); + conditions.push("`subdomain` NOT LIKE ? ESCAPE '!'"); + values.push(`${escapeLike(excludePrefix)}%`); } const whereClause = conditions.length ? `WHERE ${conditions.join(' AND ')}` @@ -422,21 +423,6 @@ export class SubdomainStore extends PuterStore { return rows[0]?.n ?? 0; } - async getByDomain(domain: string) { - const rows = await this.clients.db.read( - 'SELECT * FROM `subdomains` WHERE `domain` = ? LIMIT 1', - [domain], - ); - return rows[0] ?? null; - } - - async listByDomain(domain: string) { - return this.clients.db.read( - 'SELECT * FROM `subdomains` WHERE `domain` = ?', - [domain], - ); - } - async listByUserIdAndPrefix( userId: number, prefix: string, @@ -450,8 +436,8 @@ export class SubdomainStore extends PuterStore { ): Promise { if (!userId || prefix == null) return []; - const conditions = ['`user_id` = ?', '`subdomain` LIKE ?']; - const values: unknown[] = [userId, `${prefix}%`]; + const conditions = ['`user_id` = ?', "`subdomain` LIKE ? ESCAPE '!'"]; + const values: unknown[] = [userId, `${escapeLike(prefix)}%`]; const appOwnerFilter = appOwnerCondition(extra); if (appOwnerFilter) { conditions.push(appOwnerFilter.condition); @@ -493,8 +479,8 @@ export class SubdomainStore extends PuterStore { ): Promise { if (!userId || prefix == null) return 0; - const conditions = ['`user_id` = ?', '`subdomain` LIKE ?']; - const values: unknown[] = [userId, `${prefix}%`]; + const conditions = ['`user_id` = ?', "`subdomain` LIKE ? ESCAPE '!'"]; + const values: unknown[] = [userId, `${escapeLike(prefix)}%`]; const appOwnerFilter = appOwnerCondition(extra); if (appOwnerFilter) { conditions.push(appOwnerFilter.condition); @@ -572,7 +558,6 @@ export class SubdomainStore extends PuterStore { protected: isProtected ? 1 : 0, }; await this.#refreshCache(row); - await this.#invalidatePrefixListsForUser(userId); await this.#invalidateRootDirEntry(row.root_dir_id); return row; @@ -624,16 +609,6 @@ export class SubdomainStore extends PuterStore { if (after) { await this.#refreshCache({ ...after, ...allowed }); } - // A patched root_dir_id / associated_app_id / domain changes the rows - // the prefix-list cache would return, so drop those caches for the - // owning user(s). Covers pre- and post-rename owners in case the - // caller ever allowed re-assignment (currently we don't, but cheap). - const affectedUsers = new Set( - [before?.user_id, after?.user_id].filter((v) => v != null), - ); - for (const uid of affectedUsers) { - await this.#invalidatePrefixListsForUser(uid); - } // FSEntry rows embed a `subdomains_agg` JSON of associated subdomains, // so any rename / root_dir reassignment must drop the stale entry // caches on both the old and new root_dir_id. @@ -673,9 +648,6 @@ export class SubdomainStore extends PuterStore { ttlSeconds: NEGATIVE_CACHE_TTL_SECONDS, broadcast: true, }); - if (row.user_id != null) { - await this.#invalidatePrefixListsForUser(row.user_id); - } await this.#invalidateRootDirEntry(row.root_dir_id); } } @@ -713,12 +685,6 @@ export class SubdomainStore extends PuterStore { const after = await this.getByUuid(uuid, { primary: true }); if (after) await this.#refreshCache(after); - const affectedUsers = new Set( - [before.user_id, after?.user_id].filter((v) => v != null), - ); - for (const uid of affectedUsers) { - await this.#invalidatePrefixListsForUser(Number(uid)); - } const affectedRootDirIds = new Set( [before.root_dir_id, after?.root_dir_id].filter((v) => v != null), ); @@ -747,14 +713,9 @@ export class SubdomainStore extends PuterStore { [toAppId, fromAppId], ); - const userIds = new Set(); for (const row of rows) { row.app_owner = toAppId; await this.#refreshCache(row); - if (row.user_id != null) userIds.add(Number(row.user_id)); - } - for (const userId of userIds) { - await this.#invalidatePrefixListsForUser(userId); } return rows; } @@ -765,10 +726,6 @@ export class SubdomainStore extends PuterStore { return `${CACHE_KEY_PREFIX}:name:${subdomain}`; } - #prefixListTrackerKey(userId: number) { - return `${CACHE_KEY_PREFIX}:listByUserPrefixKeys:${userId}`; - } - /** * Write-through on a primary read (heals a stale negative marker); `NX` on * a replica read so an in-flight fresher write can't be clobbered by a @@ -852,21 +809,4 @@ export class SubdomainStore extends PuterStore { /* best-effort */ } } - - async #invalidatePrefixListsForUser(userId: number) { - if (userId == null) return; - const trackerKey = this.#prefixListTrackerKey(userId); - let cacheKeys = []; - try { - cacheKeys = await this.clients.redis.smembers(trackerKey); - } catch { - return; - } - const keysToInvalidate = [...cacheKeys, trackerKey]; - if (keysToInvalidate.length === 0) return; - await this.publishCacheKeys({ - keys: keysToInvalidate, - broadcast: true, - }); - } } diff --git a/src/backend/stores/team/TeamStore.test.ts b/src/backend/stores/team/TeamStore.test.ts index 163cf5db7..dd5592742 100644 --- a/src/backend/stores/team/TeamStore.test.ts +++ b/src/backend/stores/team/TeamStore.test.ts @@ -222,11 +222,6 @@ describe('TeamStore', () => { for (const group of seeded) { await expect(store.getByUid(group.uid)).resolves.toBeNull(); } - const owners = new Set(seeded.map((g) => g.owner_user_id)); - for (const ownerId of owners) { - const listed = await store.listByOwner(ownerId); - expect(listed.map((t) => t.uid)).not.toContain(seeded[0].uid); - } }); // -- update ------------------------------------------------------- @@ -289,8 +284,6 @@ describe('TeamStore', () => { await expect(store.getByUid(created.uid)).resolves.toBeNull(); await expect(store.getByHandle(handle)).resolves.toBeNull(); - const listed = await store.listByOwner(owner.id); - expect(listed.map((t) => t.uid)).not.toContain(created.uid); }); it('releases the handle on soft delete so it can be claimed again', async () => { @@ -719,39 +712,6 @@ describe('TeamStore', () => { expect(await store.countActiveSeats(team.id)).toBe(2); }); - it('lists the uuids of the seats it bills for', async () => { - const { team, members } = await seatedTeam(2); - const uuids = await store.listActiveSeatUuids(team.uid); - const expected = await Promise.all( - members.map(async (m) => - (await server.stores.user.getById(m.id))!.uuid, - ), - ); - expect(uuids.sort()).toEqual(expected.sort()); - }); - - it('leaves out a suspended seat, matching countActiveSeats', async () => { - const { team, members } = await seatedTeam(2); - await suspend(members[0].id); - const uuids = await store.listActiveSeatUuids(team.uid); - expect(uuids).toHaveLength(1); - expect(uuids).toHaveLength( - await store.countActiveSeats(team.id), - ); - }); - - it('leaves out the payer, who is not a seat', async () => { - const { team } = await seatedTeam(1); - const payer = await makeUser(); - await server.stores.user.update(payer.id, { password: 'hashed' }); - await store.addMember(team.uid, payer.id, { orgOwned: false }); - expect(await store.listActiveSeatUuids(team.uid)).toHaveLength(1); - }); - - it('is empty for an unknown team, not an error', async () => { - expect(await store.listActiveSeatUuids('no-such-team')).toEqual([]); - }); - it('is zero for a team with no seats, not an error', async () => { const { team } = await seatedTeam(0); expect(await store.countActiveSeats(team.id)).toBe(0); diff --git a/src/backend/stores/team/TeamStore.ts b/src/backend/stores/team/TeamStore.ts index 4e81d2cdc..0643975a3 100644 --- a/src/backend/stores/team/TeamStore.ts +++ b/src/backend/stores/team/TeamStore.ts @@ -19,9 +19,9 @@ import { v4 as uuidv4 } from 'uuid'; import { + keysetPage, normalizeLimit, - openCursor, - sealCursor, + openIdCursor, type PageResult, } from '../../util/pagination.js'; import { PuterStore } from '../types'; @@ -248,15 +248,6 @@ export class TeamStore extends PuterStore { return (await this.getByHandle(handle)) === null; } - /** Teams this user owns, oldest first. */ - async listByOwner(ownerUserId: number): Promise { - const rows = await this.clients.db.read( - `SELECT * FROM \`group\` WHERE \`owner_user_id\` = ? AND ${this.#live()} ORDER BY \`id\``, - [ownerUserId, TEAM_KIND], - ); - return rows as unknown as TeamRow[]; - } - // -- Writes ------------------------------------------------------- /** Throws on an unusable or taken handle; the unique index is the arbiter. */ @@ -442,12 +433,9 @@ export class TeamStore extends PuterStore { const limit = normalizeLimit(opts.limit, { cap: MEMBER_PAGE_CAP }) ?? MEMBER_PAGE_SIZE; - const page = openCursor( - opts.cursor, - this.config.jwt_secret_v2, - 'team member cursor', - ); - const after = typeof page?.id === 'number' ? page.id : null; + const after = openIdCursor(opts.cursor, this.config.jwt_secret_v2, { + label: 'team member cursor', + }); // One row past the limit is how we know a further page exists. const rows = (await this.clients.db.read( @@ -464,15 +452,8 @@ export class TeamStore extends PuterStore { : [teamUid, TEAM_KIND, after, limit + 1], )) as unknown as TeamMemberRow[]; - const items = rows.slice(0, limit); - const cursor = - rows.length > limit - ? sealCursor( - { id: items[items.length - 1].id }, - this.config.jwt_secret_v2, - ) - : undefined; - return { items, cursor }; + const page = keysetPage(rows, limit, this.config.jwt_secret_v2); + return { items: page.rows, cursor: page.cursor }; } /** @@ -487,12 +468,9 @@ export class TeamStore extends PuterStore { const limit = normalizeLimit(opts.limit, { cap: MEMBER_PAGE_CAP }) ?? MEMBER_PAGE_SIZE; - const page = openCursor( - opts.cursor, - this.config.jwt_secret_v2, - 'team directory cursor', - ); - const after = typeof page?.id === 'number' ? page.id : null; + const after = openIdCursor(opts.cursor, this.config.jwt_secret_v2, { + label: 'team directory cursor', + }); const rows = (await this.clients.db.read( 'SELECT ug.`id`, u.`username`, u.`uuid` FROM `jct_user_group` ug ' + @@ -507,15 +485,8 @@ export class TeamStore extends PuterStore { : [teamUid, TEAM_KIND, after, limit + 1], )) as unknown as { id: number; username: string; uuid: string }[]; - const items = rows.slice(0, limit); - const cursor = - rows.length > limit - ? sealCursor( - { id: items[items.length - 1].id }, - this.config.jwt_secret_v2, - ) - : undefined; - return { items, cursor }; + const page = keysetPage(rows, limit, this.config.jwt_secret_v2); + return { items: page.rows, cursor: page.cursor }; } /** Internal ids of this user's live teams, for the share listing. */ @@ -641,12 +612,9 @@ export class TeamStore extends PuterStore { const limit = normalizeLimit(opts.limit, { cap: MEMBER_PAGE_CAP }) ?? MEMBER_PAGE_SIZE; - const page = openCursor( - opts.cursor, - this.config.jwt_secret_v2, - 'team cursor', - ); - const after = typeof page?.id === 'number' ? page.id : null; + const after = openIdCursor(opts.cursor, this.config.jwt_secret_v2, { + label: 'team cursor', + }); const open = opts.openOnly ? ' AND g.`directory_enabled` = 1' : ''; const rows = (await this.clients.db.read( @@ -661,14 +629,11 @@ export class TeamStore extends PuterStore { : [userId, TEAM_KIND, after, limit + 1], )) as unknown as TeamRow[]; - const items = rows.slice(0, limit); - const cursor = - rows.length > limit - ? sealCursor( - { id: items[items.length - 1].id }, - this.config.jwt_secret_v2, - ) - : undefined; + const { rows: items, cursor } = keysetPage( + rows, + limit, + this.config.jwt_secret_v2, + ); let total: number | undefined; if (opts.includeTotal) { @@ -749,24 +714,6 @@ export class TeamStore extends PuterStore { return Number(rows[0]?.n ?? 0); } - /** - * Uuids of the seats a team is actually billed for, keyed by `uid` because - * that is what the billing side holds. Same filter as `countActiveSeats`; - * the quantities are per tier now, so a count is no longer enough. - */ - async listActiveSeatUuids(teamUid: string): Promise { - const rows = (await this.clients.db.read( - 'SELECT u.`uuid` AS `uuid` FROM `jct_user_group` ug ' + - 'JOIN `user` u ON u.`id` = ug.`user_id` ' + - 'JOIN `group` g ON g.`id` = ug.`group_id` ' + - `WHERE g.\`uid\` = ? AND g.${this.#live()} ` + - 'AND ug.`org_owned` = 1 ' + - 'AND (u.`suspended` IS NULL OR u.`suspended` = 0)', - [teamUid, TEAM_KIND], - )) as unknown as { uuid: string }[]; - return rows.map((r) => r.uuid).filter(Boolean); - } - /** Live teams this user owns. Soft-deleted ones do not count. */ async countOwned(ownerUserId: number): Promise { const rows = (await this.clients.db.read( @@ -911,12 +858,9 @@ export class TeamStore extends PuterStore { const limit = normalizeLimit(opts.limit, { cap: AUDIT_PAGE_CAP }) ?? AUDIT_PAGE_SIZE; - const page = openCursor( - opts.cursor, - this.config.jwt_secret_v2, - 'team audit cursor', - ); - const before = typeof page?.id === 'number' ? page.id : null; + const before = openIdCursor(opts.cursor, this.config.jwt_secret_v2, { + label: 'team audit cursor', + }); // Unix seconds in SQL: the mysql driver reads a stored UTC datetime // as local, which shifts every row away from the sign-ins. @@ -936,15 +880,8 @@ export class TeamStore extends PuterStore { : [...params, before, limit + 1], )) as unknown as TeamAuditRow[]; - const items = rows.slice(0, limit); - const cursor = - rows.length > limit - ? sealCursor( - { id: items[items.length - 1].id }, - this.config.jwt_secret_v2, - ) - : undefined; - return { items, cursor }; + const page = keysetPage(rows, limit, this.config.jwt_secret_v2); + return { items: page.rows, cursor: page.cursor }; } /** Removes a member, returning whether a row was there to remove. */ diff --git a/src/backend/stores/types.test.ts b/src/backend/stores/types.test.ts new file mode 100644 index 000000000..fc738a61e --- /dev/null +++ b/src/backend/stores/types.test.ts @@ -0,0 +1,304 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import { + afterAll, + beforeAll, + beforeEach, + describe, + expect, + it, + vi, +} from 'vitest'; +import { setupTestServer } from '../testUtil.ts'; +import type { PuterServer } from '../server.ts'; +import type { IConfig } from '../types'; +import { PuterStore } from './types.ts'; + +interface Widget { + id: number; + name: string | null; + owned?: boolean; +} + +/** A store over an in-memory "table" that records every load. */ +class WidgetStore extends PuterStore { + replica = new Map(); + primary = new Map(); + loads: Array<{ ids: number[]; primary: boolean }> = []; + + cache = this.rowCache({ + prefix: `widgets-${Math.random().toString(36).slice(2, 8)}`, + props: ['id', 'name'], + ttlSeconds: 60, + // An unowned widget is never cached under its name. + writeKeysFor: (row) => + row.owned === false + ? [this.cache.key('id', row.id)] + : this.cache.keysFor(row), + }); + + put(row: Widget) { + this.replica.set(row.id, { ...row }); + this.primary.set(row.id, { ...row }); + } + + async load(ids: number[], primary: boolean): Promise { + this.loads.push({ ids, primary }); + const table = primary ? this.primary : this.replica; + return ids.flatMap((id) => { + const row = table.get(id); + return row ? [{ ...row }] : []; + }); + } + + get(id: number, opts: { skipRead?: boolean; primary?: boolean } = {}) { + return this.cache.get( + this.cache.key('id', id), + async (primary) => (await this.load([id], primary))[0] ?? null, + opts, + ); + } + + tombstone(keys: string[]) { + return this.tombstoneCacheKeys(keys); + } +} + +describe('RowCache', () => { + let server: PuterServer; + let store: WidgetStore; + let nextId = 1; + + beforeAll(async () => { + server = await setupTestServer(); + }); + + afterAll(async () => { + await server?.shutdown(); + }); + + beforeEach(() => { + store = new WidgetStore({} as IConfig, server.clients); + }); + + const widget = (patch: Partial = {}): Widget => { + const id = nextId++; + const row = { id, name: `w${id}`, ...patch }; + store.put(row); + return row; + }; + + // Lets fire-and-forget backfills land. + const settle = () => new Promise((resolve) => setTimeout(resolve, 20)); + + it('builds one key per non-empty identifying property', () => { + expect(store.cache.keysFor({ id: 7, name: 'seven' })).toEqual([ + store.cache.key('id', 7), + store.cache.key('name', 'seven'), + ]); + expect(store.cache.keysFor({ id: 7, name: '' })).toEqual([ + store.cache.key('id', 7), + ]); + expect(store.cache.keysFor({ id: 7, name: null })).toEqual([ + store.cache.key('id', 7), + ]); + }); + + it('loads a miss from the replica, backfills it, then serves it from cache', async () => { + const row = widget(); + + expect(await store.get(row.id)).toEqual(row); + await settle(); + expect(store.loads).toEqual([{ ids: [row.id], primary: false }]); + expect( + await store.cache.read(store.cache.key('name', row.name)), + ).toEqual(row); + + expect(await store.get(row.id)).toEqual(row); + expect(store.loads).toHaveLength(1); + }); + + it('caches a row only under the keys `writeKeysFor` allows', async () => { + const row = widget({ owned: false }); + await store.get(row.id); + await settle(); + expect(await store.cache.read(store.cache.key('id', row.id))).toEqual( + row, + ); + expect( + await store.cache.read(store.cache.key('name', row.name)), + ).toBeNull(); + }); + + it('reads a tombstoned key from the primary and does not cache it back', async () => { + const row = widget(); + await store.tombstone([store.cache.key('id', row.id)]); + store.primary.delete(row.id); + + expect(await store.get(row.id)).toBeNull(); + expect(store.loads).toEqual([{ ids: [row.id], primary: true }]); + + // The replica still has it; the tombstone keeps it out of the cache. + await store.cache.write([row]); + expect( + await store.cache.read(store.cache.key('id', row.id)), + ).toBeNull(); + }); + + it('skips the cache read and reads the primary when asked', async () => { + const row = widget(); + await store.get(row.id); + store.primary.set(row.id, { ...row, name: 'renamed' }); + + expect((await store.get(row.id, { primary: true }))?.name).toBe( + 'renamed', + ); + expect(store.loads.at(-1)).toEqual({ ids: [row.id], primary: true }); + }); + + it('shares one load between concurrent misses and hands each caller its own copy', async () => { + const row = widget(); + const [a, b, c] = await Promise.all([ + store.get(row.id), + store.get(row.id), + store.get(row.id), + ]); + + expect(store.loads).toHaveLength(1); + expect(a).toEqual(row); + expect(b).toEqual(row); + expect(c).toEqual(row); + a!.name = 'mutated'; + expect(b!.name).toBe(row.name); + expect(b).not.toBe(c); + }); + + it('rejects every waiting caller when the shared load fails', async () => { + const row = widget(); + const failing = vi + .spyOn(store, 'load') + .mockRejectedValue(new Error('db down')); + const results = await Promise.allSettled([ + store.get(row.id), + store.get(row.id), + ]); + failing.mockRestore(); + + expect(results.map((r) => r.status)).toEqual(['rejected', 'rejected']); + // The failed load isn't remembered. + expect(await store.get(row.id)).toEqual(row); + }); + + it('getMany pipelines its reads, tombstone checks and backfills', async () => { + const rows = Array.from({ length: 25 }, () => widget()); + const warm = rows[0]!; + await store.get(warm.id); + await settle(); + store.loads = []; + + const pipelineSpy = vi.spyOn(server.clients.redis, 'pipeline'); + const found = await store.cache.getMany( + 'id', + [...rows.map((r) => r.id), warm.id, 999_999], + (ids, primary) => store.load(ids, primary), + ); + await settle(); + const pipelines = pipelineSpy.mock.calls.length; + pipelineSpy.mockRestore(); + + expect(found.size).toBe(rows.length); + expect(found.get(warm.id)).toEqual(warm); + // One load for every miss, none for the warm row. + expect(store.loads).toHaveLength(1); + expect(store.loads[0]!.ids).toHaveLength(rows.length); + // Cache read, tombstone check, and a three-step backfill: the count + // does not grow with the number of rows. + expect(pipelines).toBeLessThanOrEqual(5); + for (const row of rows) { + expect( + await store.cache.read(store.cache.key('name', row.name)), + ).toEqual(row); + } + }); + + it('getMany re-reads tombstoned misses from the primary', async () => { + const live = widget(); + const gone = widget(); + await store.tombstone([store.cache.key('id', gone.id)]); + store.primary.delete(gone.id); + + const found = await store.cache.getMany( + 'id', + [live.id, gone.id], + (ids, primary) => store.load(ids, primary), + ); + + expect([...found.keys()]).toEqual([live.id]); + expect(store.loads).toEqual([ + { ids: [live.id, gone.id], primary: false }, + { ids: [gone.id], primary: true }, + ]); + }); + + it('refreshes peers and skips a deleted row', async () => { + const row = widget(); + const emit = vi.spyOn(server.clients.event, 'emit'); + + await store.cache.refresh({ ...row, name: 'fresh' }); + expect(emit).toHaveBeenCalledWith( + 'outer.cacheUpdate', + expect.objectContaining({ + cacheKey: [ + store.cache.key('id', row.id), + store.cache.key('name', 'fresh'), + ], + }), + {}, + ); + + emit.mockClear(); + await store.tombstone([store.cache.key('id', row.id)]); + emit.mockClear(); + await store.cache.refresh(row); + expect(emit).not.toHaveBeenCalled(); + emit.mockRestore(); + }); + + it('tombstones only the keys a deleted row owned', async () => { + const row = widget({ owned: false }); + await store.cache.markDeleted(row); + + // The name key may still find another row, so it stays cacheable. + await store.cache.write([{ id: 424242, name: row.name }]); + expect( + await store.cache.read(store.cache.key('name', row.name)), + ).not.toBeNull(); + await store.cache.write([row]); + expect( + await store.cache.read(store.cache.key('id', row.id)), + ).toBeNull(); + + await store.cache.clearTombstones(row); + await store.cache.write([row]); + expect(await store.cache.read(store.cache.key('id', row.id))).toEqual( + row, + ); + }); +}); diff --git a/src/backend/stores/types.ts b/src/backend/stores/types.ts index cb0ec3afc..31fd92d45 100644 --- a/src/backend/stores/types.ts +++ b/src/backend/stores/types.ts @@ -54,6 +54,293 @@ export const CACHE_TOMBSTONE_SUFFIX = ':deleted'; export const isCacheTombstoneKey = (key: string): boolean => key.endsWith(CACHE_TOMBSTONE_SUFFIX); +type PublishCacheKeysParams = { + keys: string[]; + serializedData?: string; + ttlSeconds?: number; + broadcast?: boolean; +}; + +/** The store primitives a {@link RowCache} writes through. */ +interface RowCacheHost { + redis: () => LayerInstances['redis']; + publish: (params: PublishCacheKeysParams) => Promise; + tombstone: (keys: string[]) => Promise; + clearTombstones: (keys: string[]) => Promise; + tombstoned: (keys: string[]) => Promise>; + writeUnlessDeleted: ( + keys: string[], + write: () => Promise, + ) => Promise; +} + +export interface RowCacheOptions { + /** Keys are `::`. */ + prefix: string; + /** The identifying properties a row is cached under. */ + props: readonly string[]; + ttlSeconds: number; + /** The subset of `keysFor(row)` a read may cache `row` under. */ + writeKeysFor?: (row: T) => string[]; + /** Rehydrates a parsed hit. */ + revive?: (row: T) => T | null; +} + +type InFlightLoad = { json: string | null } | { error: unknown }; + +/** One pipelined round of GETs; a miss, or any failure, reads as null. */ +export const readCacheValues = async ( + redis: Pick['redis'], 'pipeline'>, + keys: string[], +): Promise> => { + if (keys.length === 0) return []; + try { + const pipeline = redis.pipeline(); + for (const key of keys) pipeline.get(key); + const results = (await pipeline.exec()) ?? []; + return keys.map((_, i) => { + const raw = results[i]?.[1]; + return typeof raw === 'string' ? raw : null; + }); + } catch { + return keys.map((): string | null => null); + } +}; + +/** + * Read-through cache for rows found by several identifying properties, each key + * holding the whole row. Writes skip tombstoned keys, so a lagging replica + * can't cache a deleted row back; refresh and invalidate reach peers. + */ +export class RowCache { + readonly #inFlight = new Map>(); + + constructor( + private readonly host: RowCacheHost, + private readonly options: RowCacheOptions, + ) {} + + key(prop: string, value: unknown): string { + return `${this.options.prefix}:${prop}:${String(value)}`; + } + + /** Every key `row` can be found under; invalidation drops all of them. */ + keysFor(row: T): string[] { + const keys: string[] = []; + for (const prop of this.options.props) { + const value = (row as Record)[prop]; + if (value === undefined || value === null || value === '') continue; + keys.push(this.key(prop, value)); + } + return keys; + } + + async read(key: string): Promise { + try { + return this.#parse(await this.host.redis().get(key)); + } catch { + return null; + } + } + + /** One pipelined read; a miss is `null`. */ + async readMany(keys: string[]): Promise> { + const values = await readCacheValues(this.host.redis(), keys); + return values.map((raw) => this.#parse(raw)); + } + + /** + * The row under `key`, else `load`ed and cached. `load` reads the primary + * when asked to or when `key` is tombstoned. Concurrent replica loads of + * one key share a query; each caller still gets its own copy. + */ + async get( + key: string, + load: (primary: boolean) => Promise, + { + skipRead = false, + primary = false, + }: { skipRead?: boolean; primary?: boolean } = {}, + ): Promise { + if (!skipRead && !primary) { + const hit = await this.read(key); + if (hit) return hit; + } + if (primary) { + const row = await load(true); + if (row) void this.write([row]); + return row; + } + + const pending = this.#inFlight.get(key); + if (pending) { + const shared = await pending; + if ('error' in shared) throw shared.error; + return shared.json === null ? null : this.#parse(shared.json); + } + + let settle!: (result: InFlightLoad) => void; + this.#inFlight.set( + key, + new Promise((resolve) => { + settle = resolve; + }), + ); + try { + const tombstoned = (await this.host.tombstoned([key])).size > 0; + const row = await load(tombstoned); + // Snapshot before the caller can mutate the row. + const json = row ? JSON.stringify(row) : null; + settle({ json }); + if (row && json) void this.#write([{ row, json }]); + return row; + } catch (error) { + settle({ error }); + throw error; + } finally { + this.#inFlight.delete(key); + } + } + + /** + * Rows by `prop`, keyed by requested value for hits and by the row's own + * value for loads: one pipelined read, one `load` for the misses, a primary + * `load` for misses whose key is tombstoned, one pipelined backfill. + */ + async getMany( + prop: string, + values: V[], + load: (values: V[], primary: boolean) => Promise, + ): Promise> { + const result = new Map(); + const unique = [...new Set(values)]; + if (unique.length === 0) return result; + + const hits = await this.readMany(unique.map((v) => this.key(prop, v))); + const missing = unique.filter((value, i) => { + const hit = hits[i]; + if (hit) result.set(value, hit); + return !hit; + }); + if (missing.length === 0) return result; + + const valueOf = (row: T) => (row as Record)[prop] as V; + const rows = await load(missing, false); + const deleted = await this.host.tombstoned( + rows.map((row) => this.key(prop, valueOf(row))), + ); + const live = rows.filter( + (row) => !deleted.has(this.key(prop, valueOf(row))), + ); + for (const row of live) result.set(valueOf(row), row); + if (live.length < rows.length) { + // Only the primary reliably knows a tombstoned row is gone. + const stale = rows + .filter((row) => deleted.has(this.key(prop, valueOf(row)))) + .map(valueOf); + for (const row of await load(stale, true)) { + result.set(valueOf(row), row); + } + } + void this.write(live); + return result; + } + + /** + * Cache `rows` on this node, skipping tombstoned ones and undoing any a + * delete lands on mid-write. Best effort; never throws. + */ + async write(rows: T[]): Promise { + await this.#write( + rows.map((row) => ({ row, json: JSON.stringify(row) })), + ); + } + + async #write(serialized: Array<{ row: T; json: string }>): Promise { + const entries = serialized + .map(({ row, json }) => ({ keys: this.#writeKeysFor(row), json })) + .filter((entry) => entry.keys.length > 0); + if (entries.length === 0) return; + try { + const before = await this.host.tombstoned( + entries.flatMap((e) => e.keys), + ); + const live = entries.filter( + (e) => !e.keys.some((key) => before.has(key)), + ); + if (live.length === 0) return; + const pipeline = this.host.redis().pipeline(); + for (const { keys, json } of live) { + for (const key of keys) { + pipeline.set(key, json, 'EX', this.options.ttlSeconds); + } + } + await pipeline.exec(); + // A row cached under a live tombstone is never re-checked. + const after = await this.host.tombstoned( + live.flatMap((e) => e.keys), + ); + const undo = live + .filter((e) => e.keys.some((key) => after.has(key))) + .flatMap((e) => e.keys); + if (undo.length > 0) { + await this.host.publish({ keys: undo, broadcast: true }); + } + } catch { + // The next read fills it. + } + } + + /** Cache `row` here and on peers, unless it was deleted. */ + async refresh(row: T): Promise { + const keys = this.#writeKeysFor(row); + const json = JSON.stringify(row); + await this.host.writeUnlessDeleted(keys, () => + this.host.publish({ + keys, + serializedData: json, + ttlSeconds: this.options.ttlSeconds, + broadcast: true, + }), + ); + } + + async invalidate(row: T): Promise { + await this.host.publish({ keys: this.keysFor(row), broadcast: true }); + } + + /** Invalidate a deleted row; pass it as read _before_ the delete. */ + async markDeleted(row: T): Promise { + // Only keys the row owned: another row may still be found by the rest. + const owned = this.#writeKeysFor(row); + await this.host.tombstone(owned); + await this.host.publish({ + keys: this.keysFor(row).filter((key) => !owned.includes(key)), + broadcast: true, + }); + } + + /** A recreated row may be cached again under its keys. */ + async clearTombstones(row: T): Promise { + await this.host.clearTombstones(this.keysFor(row)); + } + + #writeKeysFor(row: T): string[] { + return this.options.writeKeysFor?.(row) ?? this.keysFor(row); + } + + #parse(raw: unknown): T | null { + if (typeof raw !== 'string' || raw === '') return null; + try { + const row = JSON.parse(raw) as T | null; + if (!row || typeof row !== 'object') return null; + return this.options.revive ? this.options.revive(row) : row; + } catch { + return null; + } + } +} + export const PuterStore = class PuterStore implements WithLifecycle { constructor( protected config: IConfig, @@ -79,12 +366,9 @@ export const PuterStore = class PuterStore implements WithLifecycle { * nodes via `outer.cacheUpdate`. Pipelined for cluster-mode safety (no * multi-key DEL/MSET that would CROSSSLOT on Valkey). */ - protected async publishCacheKeys(params: { - keys: string[]; - serializedData?: string; - ttlSeconds?: number; - broadcast?: boolean; - }): Promise { + protected async publishCacheKeys( + params: PublishCacheKeysParams, + ): Promise { const { keys, serializedData } = params; if (keys.length === 0) return; @@ -173,17 +457,44 @@ export const PuterStore = class PuterStore implements WithLifecycle { * open. */ protected async isCacheKeyTombstoned(keys: string[]): Promise { - if (keys.length === 0) return false; + return (await this.tombstonedCacheKeys(keys)).size > 0; + } + + /** Which of `keys` were deleted within the tombstone window. Fails open. */ + protected async tombstonedCacheKeys(keys: string[]): Promise> { + const deleted = new Set(); + if (keys.length === 0) return deleted; try { const pipeline = this.clients.redis.pipeline(); for (const key of keys) { pipeline.exists(key + CACHE_TOMBSTONE_SUFFIX); } - const results = await pipeline.exec(); - return (results ?? []).some(([, exists]) => Number(exists) > 0); + const results = (await pipeline.exec()) ?? []; + keys.forEach((key, i) => { + if (Number(results[i]?.[1]) > 0) deleted.add(key); + }); } catch { - return false; + // Fail open. } + return deleted; + } + + /** A {@link RowCache} over this store's redis and peer broadcast. */ + protected rowCache( + options: RowCacheOptions, + ): RowCache { + return new RowCache( + { + redis: () => this.clients.redis, + publish: (params) => this.publishCacheKeys(params), + tombstone: (keys) => this.tombstoneCacheKeys(keys), + clearTombstones: (keys) => this.clearCacheTombstones(keys), + tombstoned: (keys) => this.tombstonedCacheKeys(keys), + writeUnlessDeleted: (keys, write) => + this.writeCacheUnlessDeleted(keys, write), + }, + options, + ); } } satisfies IPuterStore; diff --git a/src/backend/stores/user/UserStore.ts b/src/backend/stores/user/UserStore.ts index 5d0e26901..16d5227d4 100644 --- a/src/backend/stores/user/UserStore.ts +++ b/src/backend/stores/user/UserStore.ts @@ -196,6 +196,13 @@ const assertLatin1Writable = (fields: Record): void => { * property to that tuple. */ export class UserStore extends PuterStore { + readonly #cache = this.rowCache({ + prefix: CACHE_KEY_PREFIX, + props: USER_ID_PROPERTIES, + ttlSeconds: CACHE_TTL_SECONDS, + writeKeysFor: (user) => this.#cacheKeysToWrite(user), + }); + // -- Reads -------------------------------------------------------- async getById( @@ -287,76 +294,35 @@ export class UserStore extends PuterStore { * Missing ids (no DB row) are simply absent from the returned map. */ async getByIds(ids: number[]): Promise> { - const result = new Map(); - const uniqueIds = [ - ...new Set( - (Array.isArray(ids) ? ids : []).filter( - (id): id is number => typeof id === 'number', - ), - ), - ]; - if (uniqueIds.length === 0) return result; - - const missingIds: number[] = []; - try { - const pipeline = this.clients.redis.pipeline(); - for (const id of uniqueIds) { - pipeline.get(this.#cacheKey('id', id)); - } - const cacheResults = (await pipeline.exec()) ?? []; - for (let i = 0; i < uniqueIds.length; i++) { - const id = uniqueIds[i]; - const raw = cacheResults[i]?.[1]; - if (typeof raw === 'string') { - try { - result.set(id, JSON.parse(raw) as UserRow); - continue; - } catch { - // Fall through to DB on any parse failure. - } - } - missingIds.push(id); - } - } catch { - missingIds.push(...uniqueIds); - } - - for ( - let offset = 0; - offset < missingIds.length; - offset += BULK_QUERY_CHUNK_SIZE - ) { - const chunk = missingIds.slice( - offset, - offset + BULK_QUERY_CHUNK_SIZE, - ); - const placeholders = chunk.map(() => '?').join(', '); - const rows = (await this.clients.db.tryHardRead( - `SELECT * FROM \`user\` WHERE \`id\` IN (${placeholders})`, - chunk, - )) as Array>; - for (const row of rows) { - const user = this.#normalizeRow(row); - // Tombstoned: let the primary say if it's really gone. - if ( - await this.isCacheKeyTombstoned([ - this.#cacheKey('id', user.id), - ]) + const numericIds = (Array.isArray(ids) ? ids : []).filter( + (id): id is number => typeof id === 'number', + ); + return this.#cache.getMany( + 'id', + numericIds, + async (missing, primary) => { + const users: UserRow[] = []; + for ( + let offset = 0; + offset < missing.length; + offset += BULK_QUERY_CHUNK_SIZE ) { - const fresh = await this.getByProperty('id', user.id, { - force: true, - }); - if (fresh) result.set(fresh.id, fresh); - continue; + const chunk = missing.slice( + offset, + offset + BULK_QUERY_CHUNK_SIZE, + ); + const placeholders = chunk.map(() => '?').join(', '); + const sql = `SELECT * FROM \`user\` WHERE \`id\` IN (${placeholders})`; + const rows = ( + primary + ? await this.clients.db.pread(sql, chunk) + : await this.clients.db.tryHardRead(sql, chunk) + ) as Array>; + for (const row of rows) users.push(this.#normalizeRow(row)); } - result.set(user.id, user); - this.#writeCache(user).catch(() => { - // Best-effort cache backfill. - }); - } - } - - return result; + return users; + }, + ); } /** @@ -450,11 +416,6 @@ export class UserStore extends PuterStore { const cached = options.cached ?? true; const force = options.force ?? false; - if (cached && !force) { - const hit = await this.#readCache(prop, value); - if (hit) return hit; - } - // Reject lookup values that can't exist in a latin1 column before // the driver turns them into a collation-mix error at MySQL. if ( @@ -465,10 +426,10 @@ export class UserStore extends PuterStore { return null; } - // Replication-aware read: on `force`, go straight to the primary - // (`pread`) to bypass replica lag for hot reads (e.g., immediately - // after a signup). Otherwise `tryHardRead` parallels primary + - // replica and prefers whichever returns rows. + // Replication-aware read: on `force` (or a tombstoned key), go + // straight to the primary (`pread`) to bypass replica lag for hot + // reads (e.g., immediately after a signup). Otherwise `tryHardRead` + // parallels primary + replica and prefers whichever returns rows. // `id`, `uuid` and `username` are UNIQUE, so at most one row matches and // the optimizer drops the ordering. `email` is not — multiple rows may // legitimately hold the same address while unconfirmed, so without an @@ -479,23 +440,16 @@ export class UserStore extends PuterStore { `SELECT * FROM \`user\` WHERE \`${prop}\` = ?` + (prop === 'email' ? ` ${EMAIL_OWNER_ORDER}` : '') + ' LIMIT 1'; - // Only the primary reliably knows a tombstoned row is gone. - const tombstoned = - !force && - (await this.isCacheKeyTombstoned([this.#cacheKey(prop, value)])); - const rows = - force || tombstoned - ? await this.clients.db.pread(sql, [value]) - : await this.clients.db.tryHardRead(sql, [value]); - const row = rows[0]; - if (!row) return null; - - const user = this.#normalizeRow(row); - // Fire-and-forget cache write — don't block the caller on redis. - this.#writeCache(user).catch(() => { - // Best-effort cache; swallow errors. - }); - return user; + return this.#cache.get( + this.#cache.key(prop, value), + async (primary) => { + const rows = primary + ? await this.clients.db.pread(sql, [value]) + : await this.clients.db.tryHardRead(sql, [value]); + return rows[0] ? this.#normalizeRow(rows[0]) : null; + }, + { skipRead: !cached, primary: force }, + ); } // -- Writes ------------------------------------------------------- @@ -609,14 +563,12 @@ export class UserStore extends PuterStore { throw new Error('Failed to create user — no insertId returned'); // A reused username/address retires its predecessor's tombstone. - await this.clearCacheTombstones( - this.#cacheKeysForUser({ - id: insertId, - uuid: fields.uuid, - username: fields.username, - email: fields.email, - } as UserRow), - ); + await this.#cache.clearTombstones({ + id: insertId, + uuid: fields.uuid, + username: fields.username, + email: fields.email, + } as UserRow); const user = await this.getById(insertId, { force: true }); if (!user) throw new Error('Failed to fetch created user'); return user; @@ -755,16 +707,8 @@ export class UserStore extends PuterStore { [...values, userId], ); - if (guard) { - const affected = - (result as { affectedRows?: number; changes?: number }) - ?.affectedRows ?? - (result as { affectedRows?: number; changes?: number }) - ?.changes ?? - 0; - // Nothing was written, so there are no cache keys to retire. - if (affected === 0) return false; - } + // Nothing was written, so there are no cache keys to retire. + if (guard && !result.anyRowsAffected) return false; const fresh = await this.getByProperty('id', userId, { force: true }); @@ -774,16 +718,16 @@ export class UserStore extends PuterStore { // stopped owning its address keeps the address in its key list but // no longer gets cached under it, and the old value would survive. const live = new Set(fresh ? this.#cacheKeysToWrite(fresh) : []); - const retired = this.#cacheKeysForUser(before).filter( - (key) => !live.has(key), - ); + const retired = this.#cache + .keysFor(before) + .filter((key) => !live.has(key)); if (retired.length > 0) { await this.publishCacheKeys({ keys: retired, broadcast: true }); } } if (fresh) { - await this.#refreshCache(fresh); + await this.#cache.refresh(fresh); } else { await this.invalidateById(userId); } @@ -804,7 +748,7 @@ export class UserStore extends PuterStore { ); if (user) { const refreshed: UserRow = { ...user, metadata: merged }; - await this.#refreshCache(refreshed); + await this.#cache.refresh(refreshed); } } @@ -873,26 +817,22 @@ export class UserStore extends PuterStore { } async invalidate(user: UserRow): Promise { - const keys = this.#cacheKeysForUser(user); - await this.publishCacheKeys({ keys, broadcast: true }); + await this.#cache.invalidate(user); } /** Invalidate by id — fetches the cached row first so we know all its keys. */ async invalidateById(id: number): Promise { - const cached = await this.#readCache('id', id); + const cached = await this.#cache.read(this.#cache.key('id', id)); if (cached) await this.invalidate(cached); } - /** Invalidate a deleted row; pass it as read _before_ the delete. */ + /** + * Invalidate a deleted row; pass it as read _before_ the delete. Only the + * keys it owned are tombstoned: an unconfirmed address may be held by + * several rows, and the owner still needs to be cacheable. + */ async markDeleted(user: UserRow): Promise { - // Only keys this row owned: an unconfirmed address may be held by - // several rows, and the owner still needs to be cacheable. - const owned = this.#cacheKeysToWrite(user); - const rest = this.#cacheKeysForUser(user).filter( - (key) => !owned.includes(key), - ); - await this.tombstoneCacheKeys(owned); - await this.publishCacheKeys({ keys: rest, broadcast: true }); + await this.#cache.markDeleted(user); } /** @@ -900,30 +840,17 @@ export class UserStore extends PuterStore { * keys. */ async markDeletedById(id: number): Promise { - const cached = await this.#readCache('id', id); + const idKey = this.#cache.key('id', id); + const cached = await this.#cache.read(idKey); if (cached) return this.markDeleted(cached); // No cached copy to name the rest, but the id key is always this row's. - await this.tombstoneCacheKeys([this.#cacheKey('id', id)]); + await this.tombstoneCacheKeys([idKey]); } // -- Internals ---------------------------------------------------- - #cacheKey(prop: UserIdProperty, value: unknown): string { - return `${CACHE_KEY_PREFIX}:${prop}:${String(value)}`; - } - - #cacheKeysForUser(user: UserRow): string[] { - const keys: string[] = []; - for (const prop of USER_ID_PROPERTIES) { - const value = user[prop]; - if (value === undefined || value === null || value === '') continue; - keys.push(this.#cacheKey(prop, value)); - } - return keys; - } - /** - * The subset of `#cacheKeysForUser` that may point _at_ this row. Same + * The subset of the row's cache keys that may point _at_ this row. Same * keys, minus the address when the row doesn't own it: several rows may * hold one address, and `EMAIL_OWNER_ORDER` makes SQL resolve it to the * owner. Caching a placeholder under the address would shadow that for the @@ -934,60 +861,12 @@ export class UserStore extends PuterStore { * while the row still owned the address is never orphaned. */ #cacheKeysToWrite(user: UserRow): string[] { - const keys = this.#cacheKeysForUser(user); + const keys = this.#cache.keysFor(user); if (user.email_confirmed || user.password != null) return keys; - const addressKey = this.#cacheKey('email', user.email); + const addressKey = this.#cache.key('email', user.email); return keys.filter((key) => key !== addressKey); } - async #readCache( - prop: UserIdProperty, - value: unknown, - ): Promise { - try { - const raw = await this.clients.redis.get( - this.#cacheKey(prop, value), - ); - if (!raw) return null; - const parsed = JSON.parse(raw) as UserRow; - // Cached rows were normalized on the write path, so booleans are booleans. - return parsed; - } catch { - return null; - } - } - - async #writeCache(user: UserRow): Promise { - const keys = this.#cacheKeysToWrite(user); - const serialized = JSON.stringify(user); - // A replica behind the delete still returns the row. - await this.writeCacheUnlessDeleted(keys, async () => { - await Promise.all( - keys.map((key) => - this.clients.redis.set( - key, - serialized, - 'EX', - CACHE_TTL_SECONDS, - ), - ), - ); - }); - } - - async #refreshCache(user: UserRow): Promise { - const keys = this.#cacheKeysToWrite(user); - // `#write` reads the replica, so it can refresh a row already deleted. - await this.writeCacheUnlessDeleted(keys, () => - this.publishCacheKeys({ - keys, - serializedData: JSON.stringify(user), - ttlSeconds: CACHE_TTL_SECONDS, - broadcast: true, - }), - ); - } - /** * Coerce raw DB row values into consistent JS types. MySQL returns * BOOLEAN/TINYINT as 0|1; SQLite returns INTEGER. JSON columns come as diff --git a/src/backend/stores/userBlock/UserBlockStore.ts b/src/backend/stores/userBlock/UserBlockStore.ts index 8f6255ac9..f0f6a2b24 100644 --- a/src/backend/stores/userBlock/UserBlockStore.ts +++ b/src/backend/stores/userBlock/UserBlockStore.ts @@ -105,11 +105,6 @@ export class UserBlockStore extends PuterStore { 'DELETE FROM `user_block` WHERE `blocker_user_id` = ? AND `blocked_user_id` = ?', [blockerUserId, blockedUserId], ); - return ( - ((result as { affectedRows?: number; changes?: number }) - ?.affectedRows ?? - (result as { changes?: number })?.changes ?? - 0) > 0 - ); + return result.anyRowsAffected; } } diff --git a/src/backend/util/pagination.test.ts b/src/backend/util/pagination.test.ts index 0801c971d..181f72668 100644 --- a/src/backend/util/pagination.test.ts +++ b/src/backend/util/pagination.test.ts @@ -2,9 +2,11 @@ import { describe, expect, it } from 'vitest'; import { decodeCursor, encodeCursor, + keysetPage, normalizeLimit, normalizeOffset, openCursor, + openIdCursor, sealCursor, } from './pagination'; import { HttpError } from '../core/http'; @@ -139,4 +141,65 @@ describe('pagination util', () => { expect(normalizeOffset(5000, { cap: 5000 })).toBe(5000); }); }); + + describe('openIdCursor', () => { + const secret = 'keyset-secret'; + + it('returns null without a cursor', () => { + expect(openIdCursor(undefined, secret)).toBeNull(); + expect(openIdCursor('', secret)).toBeNull(); + }); + + it('reads the id from a sealed or plain cursor', () => { + expect(openIdCursor(sealCursor({ id: 41 }, secret), secret)).toBe( + 41, + ); + expect(openIdCursor(encodeCursor({ id: 7 }), secret)).toBe(7); + }); + + it('reads a cursor without a usable id as the first page', () => { + expect( + openIdCursor(encodeCursor({ appUid: 'app-1' }), secret), + ).toBeNull(); + expect(openIdCursor(encodeCursor({ id: -3 }), secret)).toBeNull(); + }); + + it('refuses a cursor without a usable id when strict', () => { + expect(() => + openIdCursor(encodeCursor({ id: 'x' }), secret, { + label: 'share cursor', + strict: true, + }), + ).toThrowError('invalid share cursor'); + }); + + it('refuses an unreadable cursor either way', () => { + expect(() => openIdCursor('!!!', secret)).toThrowError(HttpError); + }); + }); + + describe('keysetPage', () => { + const secret = 'keyset-secret'; + const rows = [{ id: 3 }, { id: 5 }, { id: 9 }]; + + it('returns every row and no cursor when nothing follows', () => { + expect(keysetPage(rows, 3, secret)).toEqual({ rows }); + }); + + it('cuts to the limit and points the cursor at the last row kept', () => { + const page = keysetPage(rows, 2, secret); + expect(page.rows).toEqual([{ id: 3 }, { id: 5 }]); + expect(openIdCursor(page.cursor, secret)).toBe(5); + }); + + it('takes a custom position, plain without a secret', () => { + const page = keysetPage( + [{ app: 'a' }, { app: 'b' }], + 1, + undefined, + (last) => ({ appUid: last.app }), + ); + expect(decodeCursor(page.cursor)).toEqual({ appUid: 'a' }); + }); + }); }); diff --git a/src/backend/util/pagination.ts b/src/backend/util/pagination.ts index ccf9675c8..69c4a1757 100644 --- a/src/backend/util/pagination.ts +++ b/src/backend/util/pagination.ts @@ -190,3 +190,52 @@ export const normalizeOffset = ( } return floored; }; + +// -- Keyset pages ----------------------------------------------------- + +/** + * The id a keyset page resumes from, or null on the first page. A cursor that + * opens without a usable id (another listing's, say) throws when `strict`, and + * otherwise reads as the first page. + */ +export const openIdCursor = ( + cursor: string | Record | null | undefined, + secret: string | undefined, + { + label = 'cursor', + strict = false, + }: { label?: string; strict?: boolean } = {}, +): number | null => { + const decoded = openCursor(cursor, secret, label); + if (decoded === undefined) return null; + const id = Number(decoded.id); + if (Number.isSafeInteger(id) && id >= 0) return id; + if (strict) { + throw new HttpError(400, `invalid ${label}`, { + legacyCode: 'bad_request', + }); + } + return null; +}; + +/** + * One page of `rows`, fetched with `LIMIT limit + 1`: the first `limit` rows, + * and a cursor at the last of them when the extra row says another page + * follows. `position` is the cursor payload; it defaults to the row's `id`. + * Without `secret` the cursor is a plain `encodeCursor` one. + */ +export const keysetPage = ( + rows: R[], + limit: number, + secret: string | undefined, + position: (last: R) => Record = (last) => ({ + id: Number((last as { id?: unknown }).id), + }), +): { rows: R[]; cursor?: string } => { + if (rows.length <= limit) return { rows }; + const page = rows.slice(0, limit); + return { + rows: page, + cursor: sealCursor(position(page[page.length - 1]!), secret), + }; +}; diff --git a/src/backend/util/redisGeneration.test.ts b/src/backend/util/redisGeneration.test.ts new file mode 100644 index 000000000..c5bcce0a8 --- /dev/null +++ b/src/backend/util/redisGeneration.test.ts @@ -0,0 +1,68 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import MockRedis from 'ioredis-mock'; +import type { Cluster } from 'ioredis'; +import { describe, expect, it, vi } from 'vitest'; +import { bumpGeneration } from './redisGeneration'; + +const redis = new MockRedis() as unknown as Cluster; +const freshKey = () => `gen-test:${Math.random().toString(36).slice(2)}`; + +describe('bumpGeneration', () => { + it('increments from zero and renews the TTL each time', async () => { + const key = freshKey(); + expect(await bumpGeneration(redis, key, 60)).toBe(1); + await redis.expire(key, 5); + expect(await bumpGeneration(redis, key, 60)).toBe(2); + expect(await redis.ttl(key)).toBeGreaterThan(5); + }); + + it('sends the increment and the expiry in one round trip', async () => { + const pipeline = vi.spyOn(redis, 'pipeline'); + const incr = vi.spyOn(redis, 'incr'); + try { + await bumpGeneration(redis, freshKey(), 60); + expect(pipeline).toHaveBeenCalledTimes(1); + expect(incr).not.toHaveBeenCalled(); + } finally { + pipeline.mockRestore(); + incr.mockRestore(); + } + }); + + it('throws the error redis reports for either command', async () => { + const failing = { + pipeline: () => { + const chain = { + incr: () => chain, + expire: () => chain, + exec: async () => [ + [new Error('WRONGTYPE not an integer'), null], + [null, 1], + ], + }; + return chain; + }, + } as unknown as Cluster; + await expect(bumpGeneration(failing, freshKey(), 60)).rejects.toThrow( + 'WRONGTYPE', + ); + }); +}); diff --git a/src/backend/util/redisGeneration.ts b/src/backend/util/redisGeneration.ts new file mode 100644 index 000000000..295f5ee64 --- /dev/null +++ b/src/backend/util/redisGeneration.ts @@ -0,0 +1,41 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import type { Cluster } from 'ioredis'; + +/** + * Increment a generation counter and renew its TTL in one round trip; the two + * commands travel together, so the counter can't be left without an expiry. + * Returns the new value. + */ +export const bumpGeneration = async ( + redis: Pick, + key: string, + ttlSeconds: number, +): Promise => { + const results = await redis + .pipeline() + .incr(key) + .expire(key, ttlSeconds) + .exec(); + for (const [error] of results ?? []) { + if (error) throw error; + } + return Number(results?.[0]?.[1]); +}; diff --git a/src/backend/util/sqlLike.test.ts b/src/backend/util/sqlLike.test.ts new file mode 100644 index 000000000..2b4a0637e --- /dev/null +++ b/src/backend/util/sqlLike.test.ts @@ -0,0 +1,32 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import { describe, expect, it } from 'vitest'; +import { escapeLike } from './sqlLike'; + +describe('escapeLike', () => { + it('escapes LIKE wildcards and the escape character itself', () => { + expect(escapeLike('50%_off!')).toBe('50!%!_off!!'); + }); + + it('leaves everything else alone', () => { + expect(escapeLike('workers.puter.')).toBe('workers.puter.'); + expect(escapeLike('/user/a b\\c')).toBe('/user/a b\\c'); + }); +}); diff --git a/src/backend/util/sqlLike.ts b/src/backend/util/sqlLike.ts new file mode 100644 index 000000000..6029299ca --- /dev/null +++ b/src/backend/util/sqlLike.ts @@ -0,0 +1,26 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +/** + * `value` as literal text inside a LIKE pattern, for use with `ESCAPE '!'`: `%` + * and `_` would otherwise widen the match. `!` rather than a backslash, which + * MySQL also reads as an escape inside the string literal itself. + */ +export const escapeLike = (value: string): string => + value.replace(/[!%_]/g, '!$&');