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, '!$&');