diff --git a/docs/contributing/disaster-recovery.md b/docs/contributing/disaster-recovery.md index a5c3ddf651..4e8d9c8b97 100644 --- a/docs/contributing/disaster-recovery.md +++ b/docs/contributing/disaster-recovery.md @@ -122,13 +122,17 @@ Restore rebuilds these; do not treat them as recovery media: sealed dump. Storage ids absent from the inventory are not deleted by restore. - R2 restore puts sealed objects back by key; it does not sweep orphans that appeared after the sealed day. -- **StorageRunner inventory still reads `package_runtime_runs` in D1** for - storage ids (plus jobs, archived artifacts, app packages, and service names). - New runs write storage ids only into `RunLog`, so a StorageRunner bucket - referenced solely by a post-migration run record is not yet picked up by the - platform DR inventory. Account deletion/export union `RunLog` storage ids for - purge/portability; extending DR inventory the same way is a separate follow-up - if those orphan-only buckets matter for sealed-day completeness. +- **StorageRunner inventory** unions authoritative D1 sources: `jobs`, + `archived_job_artifacts`, `saved_packages` (app packages), the + `user_storage_buckets` registry (including ad-hoc / execute buckets), and + `package_service_states` (projected service storage ids). Platform DR has only + a `D1Database`, so it does **not** walk package manifests or enumerate + `RunLog` Durable Objects. A service whose Durable Object never projected into + `package_service_states` is therefore absent from sealed-day inventory until + it heartbeats or transitions; account deletion/export cover those via manifest + enumeration. Buckets known only inside a user's `RunLog` (and never registered + in `user_storage_buckets` or an entity table) remain outside DR inventory by + design — RunLog is observability, not a canonical store. ## Credentials and Access diff --git a/packages/worker/migrations/0097-user-storage-buckets.sql b/packages/worker/migrations/0097-user-storage-buckets.sql new file mode 100644 index 0000000000..e9134e90cd --- /dev/null +++ b/packages/worker/migrations/0097-user-storage-buckets.sql @@ -0,0 +1,80 @@ +-- Authoritative per-user durable storage bucket ownership. +-- +-- "Which storage buckets does this user own?" is state, not history. Deriving +-- it from `package_runtime_runs` was wrong: that table stopped being written +-- (RunLog migration) and will drain under the 30-day retention policy. Ad-hoc +-- buckets (caller-supplied `storageId` on execute / storage capabilities) have +-- no other D1 record, so they would become unenumerable for backup, account +-- export, and account deletion. +-- +-- This table is written on mutating StorageRunner access. The +-- `package_runtime_runs` arm below is a one-time rescue of pre-migration +-- buckets and is why this migration could not wait for the legacy drop. + +CREATE TABLE IF NOT EXISTS user_storage_buckets ( + user_id TEXT NOT NULL, + storage_id TEXT NOT NULL, + kind TEXT NOT NULL CHECK (kind IN ('job', 'app', 'service', 'execute', 'unknown')), + created_at TEXT NOT NULL, + last_seen_at TEXT NOT NULL, + PRIMARY KEY (user_id, storage_id) +); + +CREATE INDEX IF NOT EXISTS idx_user_storage_buckets_user +ON user_storage_buckets(user_id); + +INSERT OR IGNORE INTO user_storage_buckets ( + user_id, storage_id, kind, created_at, last_seen_at +) +SELECT + user_id, + storage_id, + 'job', + created_at, + updated_at +FROM jobs +WHERE storage_id IS NOT NULL AND trim(storage_id) != ''; + +INSERT OR IGNORE INTO user_storage_buckets ( + user_id, storage_id, kind, created_at, last_seen_at +) +SELECT + user_id, + storage_id, + 'job', + created_at, + updated_at +FROM archived_job_artifacts +WHERE storage_id IS NOT NULL AND trim(storage_id) != ''; + +INSERT OR IGNORE INTO user_storage_buckets ( + user_id, storage_id, kind, created_at, last_seen_at +) +SELECT + user_id, + id, + 'app', + created_at, + updated_at +FROM saved_packages +WHERE has_app = 1; + +-- One-time rescue of pre-migration buckets (including ad-hoc execute/storage +-- ids) that only survive in run history until ~2026-08-25 retention drain. +INSERT OR IGNORE INTO user_storage_buckets ( + user_id, storage_id, kind, created_at, last_seen_at +) +SELECT + user_id, + storage_id, + CASE surface + WHEN 'job' THEN 'job' + WHEN 'service' THEN 'service' + WHEN 'app_fetch' THEN 'app' + WHEN 'app_realtime' THEN 'app' + ELSE 'unknown' + END, + COALESCE(started_at, created_at, CURRENT_TIMESTAMP), + COALESCE(updated_at, finished_at, started_at, created_at, CURRENT_TIMESTAMP) +FROM package_runtime_runs +WHERE storage_id IS NOT NULL AND trim(storage_id) != ''; diff --git a/packages/worker/src/app/account-data-targets.ts b/packages/worker/src/app/account-data-targets.ts index 0db9d06d30..29c6ae1dcb 100644 --- a/packages/worker/src/app/account-data-targets.ts +++ b/packages/worker/src/app/account-data-targets.ts @@ -116,6 +116,7 @@ export const accountUserDataTargets: ReadonlyArray = [ { kind: 'user_id', table: 'package_runtime_logs' }, { kind: 'user_id', table: 'package_runtime_runs' }, { kind: 'user_id', table: 'package_service_states' }, + { kind: 'user_id', table: 'user_storage_buckets' }, { kind: 'user_id', table: 'usage_rollups' }, { kind: 'user_id', table: 'user_activation_milestones' }, { kind: 'user_id', table: 'user_package_run_successes' }, diff --git a/packages/worker/src/app/account-deletion.node.test.ts b/packages/worker/src/app/account-deletion.node.test.ts index e6fe2100dd..aaecf92312 100644 --- a/packages/worker/src/app/account-deletion.node.test.ts +++ b/packages/worker/src/app/account-deletion.node.test.ts @@ -1,5 +1,6 @@ import { quoteSqlIdentifier } from '@kody-internal/shared/sql-literals.ts' import { readdirSync, readFileSync } from 'node:fs' +import { fileURLToPath } from 'node:url' import { DatabaseSync } from 'node:sqlite' import { expect, test, vi } from 'vitest' import { @@ -15,6 +16,7 @@ import { AccountDeletionWritersActiveError, assertAccountWritable, } from './account-deletion-state.ts' +import { consoleWarn } from '#worker/test-support/console-spies.ts' type RowMap = Record>> @@ -176,44 +178,9 @@ function createTestDb( .map((row) => ({ do_id: row['do_id'] })) return { results: results as Array, meta: { changes: 0 } } } - if ( - lower.includes('select distinct package_id, name from (') && - lower.includes('from package_service_states') && - lower.includes('from package_runtime_runs') - ) { - const seen = new Set() - results = [] - for (const row of rows.package_service_states ?? []) { - if (row['user_id'] !== userId) continue - const key = `${String(row['package_id'])}:${String(row['service_name'])}` - if (seen.has(key)) continue - seen.add(key) - results.push({ - package_id: row['package_id'], - name: row['service_name'], - }) - } - for (const row of rows.package_runtime_runs ?? []) { - if ( - row['user_id'] !== userId || - row['surface'] !== 'service' || - row['name'] == null - ) { - continue - } - const key = `${String(row['package_id'])}:${String(row['name'])}` - if (seen.has(key)) continue - seen.add(key) - results.push({ - package_id: row['package_id'], - name: row['name'], - }) - } - return { results: results as Array, meta: { changes: 0 } } - } if ( lower.includes('from package_service_states as s') && - lower.includes('from package_runtime_runs as r') + lower.includes('left join saved_packages as p') ) { const seen = new Set() results = [] @@ -234,34 +201,54 @@ function createTestDb( name: row['service_name'], }) } + return { results: results as Array, meta: { changes: 0 } } + } + if ( + lower.includes('from package_runtime_runs as r') && + lower.includes("r.surface = 'service'") + ) { + const seen = new Set() + results = [] for (const row of rows.package_runtime_runs ?? []) { - if ( - row['user_id'] !== userId || - row['surface'] !== 'service' || - row['name'] == null - ) { - continue - } + if (row['user_id'] !== userId) continue + if (row['surface'] !== 'service') continue + if (row['name'] == null) continue const savedPackage = (rows.saved_packages ?? []).find( (pkg) => pkg['id'] === row['package_id'] && pkg['user_id'] === row['user_id'], ) - const sourceId = - savedPackage?.['source_id'] ?? row['source_id'] const key = `${String(row['package_id'])}:${String(row['name'])}` if (seen.has(key)) continue seen.add(key) results.push({ package_id: row['package_id'], kody_id: - savedPackage?.['kody_id'] ?? row['package_kody_id'], - source_id: sourceId, + savedPackage?.['kody_id'] ?? + row['package_kody_id'] ?? + null, + source_id: + savedPackage?.['source_id'] ?? row['source_id'] ?? null, name: row['name'], }) } return { results: results as Array, meta: { changes: 0 } } } + if ( + lower.includes( + 'select storage_id as storageid from user_storage_buckets', + ) + ) { + results = (rows.user_storage_buckets ?? []) + .filter((row) => row['user_id'] === userId) + .map((row) => ({ storageId: row['storage_id'] })) + .sort((left, right) => + String(left.storageId).localeCompare( + String(right.storageId), + ), + ) + return { results: results as Array, meta: { changes: 0 } } + } if ( lower.includes('from community_listings') && lower.includes('entity_sources.published_commit') @@ -780,6 +767,14 @@ test('deleteUserAccount cascades user-scoped rows for the requested user', async started_at: null, updated_at: '2026-07-05T00:00:00.000Z', }, + { + user_id: userAaa, + package_id: 'pkg-orphan', + service_name: 'legacy-sync', + status: 'idle', + started_at: null, + updated_at: '2026-07-05T00:00:00.000Z', + }, { user_id: userBbb, package_id: 'pkg-2', @@ -789,6 +784,18 @@ test('deleteUserAccount cascades user-scoped rows for the requested user', async updated_at: '2026-07-05T00:00:00.000Z', }, ], + user_storage_buckets: [ + { + user_id: userAaa, + storage_id: 'exec:run-2', + kind: 'execute', + }, + { + user_id: userBbb, + storage_id: 'service:pkg-2:sync', + kind: 'service', + }, + ], mcp_memories: [ { id: 'mem-1', user_id: userAaa }, { id: 'mem-2', user_id: userBbb }, @@ -1288,6 +1295,13 @@ test('deleteUserAccount cascades user-scoped rows for the requested user', async updated_at: '2026-07-05T00:00:00.000Z', }, ]) + expect(rows.user_storage_buckets).toEqual([ + { + user_id: userBbb, + storage_id: 'service:pkg-2:sync', + kind: 'service', + }, + ]) expect(rows.community_listings).toEqual([ { id: 'listing-2', owner_user_id: userBbb, pinned_commit: 'commit-2' }, ]) @@ -1381,7 +1395,8 @@ test('deleteUserAccount cascades user-scoped rows for the requested user', async expect(result.deletedRowCounts.email_attachments).toBe(1) expect(result.deletedRowCounts.package_runtime_runs).toBe(3) expect(result.deletedRowCounts.package_runtime_logs).toBe(1) - expect(result.deletedRowCounts.package_service_states).toBe(2) + expect(result.deletedRowCounts.package_service_states).toBe(3) + expect(result.deletedRowCounts.user_storage_buckets).toBe(1) expect(result.deletedRowCounts.community_listings).toBe(1) expect(result.deletedRowCounts.community_forks).toBe(2) expect(result.deletedRowCounts.community_ratings).toBe(2) @@ -1905,6 +1920,39 @@ test('account deletion empties the user RunLog DO and leaves other users untouch expect(clearedStorageIds.some((id) => id.includes('bbb-bucket'))).toBe(false) }) +test('account deletion purges a StorageRunner known only via user_storage_buckets', async () => { + const userId = 'user-bucket-only' + const clearStorage = vi.fn(async () => ({ ok: true as const })) + const idFromName = vi.fn((name: string) => name as unknown as DurableObjectId) + const { db } = createTestDb({ + users: [{ id: 1, email: 'bucket@example.com', stable_user_id: userId }], + user_storage_buckets: [ + { + user_id: userId, + storage_id: 'exec:adhoc-only', + kind: 'execute', + }, + ], + }) + + const result = await deleteUserAccount({ + env: createSuccessfulDeletionEnv(db, { + STORAGE_RUNNER: { + idFromName, + get: () => ({ clearStorage }), + }, + }), + dbUserId: 1, + mcpUserId: userId, + }) + + expect(result.clearedDurableObjects.storageRunners).toBe(1) + expect(clearStorage).toHaveBeenCalledTimes(1) + expect(idFromName).toHaveBeenCalledWith( + JSON.stringify([userId, 'exec:adhoc-only']), + ) +}) + test('account deletion purges a PackageServiceInstance known only via package_service_states', async () => { const userId = 'user-states-only' const serviceFetch = vi.fn(async () => Response.json({ ok: true })) @@ -1966,44 +2014,352 @@ test('account deletion purges a PackageServiceInstance known only via package_se ) }) -test('account deletion still purges a service known only via legacy package_runtime_runs', async () => { - const userId = 'user-legacy-only' +test('account deletion purges a PackageServiceInstance known only via legacy package_runtime_runs', async () => { + const userId = 'user-legacy-runs-only' const serviceFetch = vi.fn(async () => Response.json({ ok: true })) + const idFromName = vi.fn((name: string) => name as unknown as DurableObjectId) const { db } = createTestDb({ users: [{ id: 1, email: 'legacy@example.com', stable_user_id: userId }], + saved_packages: [ + { + id: 'pkg-legacy', + user_id: userId, + kody_id: 'legacy-pkg', + source_id: 'src-legacy', + has_app: 0, + }, + ], package_runtime_runs: [ { - id: 'run-legacy', + id: 'run-legacy-only', user_id: userId, package_id: 'pkg-legacy', package_kody_id: 'legacy-pkg', source_id: 'src-legacy', surface: 'service', - name: 'only-in-runs', - storage_id: null, + name: 'only-in-runtime-runs', + status: 'succeeded', + started_at: '2026-07-05T00:00:00.000Z', + finished_at: '2026-07-05T00:00:01.000Z', }, ], }) - const result = await deleteUserAccount({ - env: createSuccessfulDeletionEnv(db, { - PACKAGE_SERVICE_INSTANCE: { - idFromName: (name: string) => name as unknown as DurableObjectId, - get: () => ({ fetch: serviceFetch }), + const listSaved = vi.spyOn( + await import('#worker/package-registry/repo.ts'), + 'listSavedPackagesByUserId', + ) + listSaved.mockResolvedValue([ + { + id: 'pkg-legacy', + userId, + name: 'Legacy Package', + kodyId: 'legacy-pkg', + description: '', + tags: [], + searchText: null, + sourceId: 'src-legacy', + hasApp: false, + hidden: false, + isPrivate: true, + createdAt: '2026-07-05T00:00:00.000Z', + updatedAt: '2026-07-05T00:00:00.000Z', + }, + ]) + const loadManifest = vi.spyOn( + await import('#worker/package-registry/source.ts'), + 'loadPackageManifestBySourceId', + ) + loadManifest.mockResolvedValue({ + source: { + id: 'src-legacy', + userId, + entityKind: 'package', + entityId: 'pkg-legacy', + repoId: 'repo-1', + manifestPath: 'package.json', + sourceRoot: '.', + publishedCommit: null, + }, + manifest: { + name: 'legacy-pkg', + kody: { + services: {}, }, - }), - dbUserId: 1, - mcpUserId: userId, + }, + } as never) + + try { + const result = await deleteUserAccount({ + env: createSuccessfulDeletionEnv(db, { + BUNDLE_ARTIFACTS_KV: { + get: async () => null, + async list() { + return { keys: [], list_complete: true as const } + }, + delete: async () => undefined, + }, + PACKAGE_SERVICE_INSTANCE: { + idFromName, + get: () => ({ fetch: serviceFetch }), + }, + }), + dbUserId: 1, + mcpUserId: userId, + }) + + expect(result.clearedDurableObjects.packageServiceInstances).toBe(1) + expect(serviceFetch).toHaveBeenCalledTimes(1) + const request = serviceFetch.mock.calls[0]?.[0] as Request + expect(new URL(request.url).pathname).toContain('/purge') + const body = (await request.clone().json()) as { + binding: { packageId: string; serviceName: string } + } + expect(body.binding).toMatchObject({ + packageId: 'pkg-legacy', + serviceName: 'only-in-runtime-runs', + }) + expect(idFromName).toHaveBeenCalledWith( + JSON.stringify([userId, 'pkg-legacy', 'only-in-runtime-runs']), + ) + } finally { + loadManifest.mockRestore() + listSaved.mockRestore() + } +}) + +test('account deletion purges a service declared only in the package manifest', async () => { + const userId = 'user-manifest-only' + const serviceFetch = vi.fn(async () => Response.json({ ok: true })) + const { db } = createTestDb({ + users: [{ id: 1, email: 'manifest@example.com', stable_user_id: userId }], + saved_packages: [ + { + id: 'pkg-manifest', + user_id: userId, + kody_id: 'manifest-pkg', + source_id: 'src-manifest', + has_app: 0, + name: 'Manifest Package', + description: '', + tags_json: '[]', + search_text: null, + hidden: 0, + is_private: 1, + created_at: '2026-07-05T00:00:00.000Z', + updated_at: '2026-07-05T00:00:00.000Z', + }, + ], }) - expect(result.clearedDurableObjects.packageServiceInstances).toBe(1) - const request = serviceFetch.mock.calls[0]?.[0] as Request - const body = (await request.clone().json()) as { - binding: { packageId: string; serviceName: string; kodyId: string } + const loadManifest = vi.spyOn( + await import('#worker/package-registry/source.ts'), + 'loadPackageManifestBySourceId', + ) + loadManifest.mockResolvedValue({ + source: { + id: 'src-manifest', + userId, + entityKind: 'package', + entityId: 'pkg-manifest', + repoId: 'repo-1', + manifestPath: 'package.json', + sourceRoot: '.', + publishedCommit: null, + }, + manifest: { + name: 'manifest-pkg', + kody: { + services: { + 'only-in-manifest': { + entry: './services/only-in-manifest.ts', + }, + }, + }, + }, + } as never) + + const listSaved = vi.spyOn( + await import('#worker/package-registry/repo.ts'), + 'listSavedPackagesByUserId', + ) + listSaved.mockResolvedValue([ + { + id: 'pkg-manifest', + userId, + name: 'Manifest Package', + kodyId: 'manifest-pkg', + description: '', + tags: [], + searchText: null, + sourceId: 'src-manifest', + hasApp: false, + hidden: false, + isPrivate: true, + createdAt: '2026-07-05T00:00:00.000Z', + updatedAt: '2026-07-05T00:00:00.000Z', + }, + ]) + + try { + const result = await deleteUserAccount({ + env: createSuccessfulDeletionEnv(db, { + BUNDLE_ARTIFACTS_KV: { + get: async () => null, + async list() { + return { keys: [], list_complete: true as const } + }, + delete: async () => undefined, + }, + PACKAGE_SERVICE_INSTANCE: { + idFromName: (name: string) => name as unknown as DurableObjectId, + get: () => ({ fetch: serviceFetch }), + }, + }), + dbUserId: 1, + mcpUserId: userId, + }) + + expect(result.clearedDurableObjects.packageServiceInstances).toBe(1) + const request = serviceFetch.mock.calls[0]?.[0] as Request + const body = (await request.clone().json()) as { + binding: { packageId: string; serviceName: string; kodyId: string } + } + expect(body.binding).toMatchObject({ + packageId: 'pkg-manifest', + serviceName: 'only-in-manifest', + kodyId: 'manifest-pkg', + }) + } finally { + loadManifest.mockRestore() + listSaved.mockRestore() } - expect(body.binding).toMatchObject({ - packageId: 'pkg-legacy', - serviceName: 'only-in-runs', - kodyId: 'legacy-pkg', +}) + +test('account deletion continues when manifest load fails and still purges state-table services', async () => { + const userId = 'user-manifest-fail' + const serviceFetch = vi.fn(async () => Response.json({ ok: true })) + consoleWarn.mockImplementation(() => {}) + const { db } = createTestDb({ + users: [{ id: 1, email: 'fail@example.com', stable_user_id: userId }], + saved_packages: [ + { + id: 'pkg-fail', + user_id: userId, + kody_id: 'fail-pkg', + source_id: 'src-fail', + has_app: 0, + }, + ], + package_service_states: [ + { + user_id: userId, + package_id: 'pkg-fail', + service_name: 'known-in-states', + status: 'idle', + started_at: null, + updated_at: '2026-07-05T00:00:00.000Z', + }, + ], }) + + const listSaved = vi.spyOn( + await import('#worker/package-registry/repo.ts'), + 'listSavedPackagesByUserId', + ) + listSaved.mockResolvedValue([ + { + id: 'pkg-fail', + userId, + name: 'Fail Package', + kodyId: 'fail-pkg', + description: '', + tags: [], + searchText: null, + sourceId: 'src-fail', + hasApp: false, + hidden: false, + isPrivate: true, + createdAt: '2026-07-05T00:00:00.000Z', + updatedAt: '2026-07-05T00:00:00.000Z', + }, + ]) + const loadManifest = vi.spyOn( + await import('#worker/package-registry/source.ts'), + 'loadPackageManifestBySourceId', + ) + loadManifest.mockRejectedValue(new Error('manifest unavailable')) + + try { + let result + try { + result = await deleteUserAccount({ + env: createSuccessfulDeletionEnv(db, { + BUNDLE_ARTIFACTS_KV: { + get: async () => null, + async list() { + return { keys: [], list_complete: true as const } + }, + delete: async () => undefined, + }, + PACKAGE_SERVICE_INSTANCE: { + idFromName: (name: string) => name as unknown as DurableObjectId, + get: () => ({ fetch: serviceFetch }), + }, + }), + dbUserId: 1, + mcpUserId: userId, + }) + } catch (error) { + expect(error).toBeInstanceOf(AccountDeletionCleanupError) + const cleanupError = error as AccountDeletionCleanupError + expect( + cleanupError.cleanupErrors.some((warning) => + warning.includes('Failed to load package manifest'), + ), + ).toBe(true) + result = cleanupError.partialResult + } + + expect(result.clearedDurableObjects.packageServiceInstances).toBe(1) + const request = serviceFetch.mock.calls[0]?.[0] as Request + const body = (await request.clone().json()) as { + binding: { packageId: string; serviceName: string } + } + expect(body.binding).toMatchObject({ + packageId: 'pkg-fail', + serviceName: 'known-in-states', + }) + expect(consoleWarn).toHaveBeenCalledWith( + expect.stringContaining('Failed to load package manifest'), + ) + } finally { + loadManifest.mockRestore() + listSaved.mockRestore() + consoleWarn.mockReset() + } +}) + +test('export and DR sources no longer read package_runtime_runs; deletion inventory keeps the legacy arm', () => { + const mustNotRead = [ + './account-deletion.ts', + './account-export.ts', + '../dr/exporter.ts', + ] + for (const relative of mustNotRead) { + const source = readFileSync( + fileURLToPath(new URL(relative, import.meta.url)), + 'utf8', + ) + expect( + source.includes('package_runtime_runs'), + `${relative} must not read package_runtime_runs`, + ).toBe(false) + } + const inventory = readFileSync( + fileURLToPath(new URL('./account-user-inventory.ts', import.meta.url)), + 'utf8', + ) + expect(inventory.includes('package_runtime_runs')).toBe(true) + expect(inventory.includes('includeLegacyRuntimeRuns')).toBe(true) }) diff --git a/packages/worker/src/app/account-deletion.ts b/packages/worker/src/app/account-deletion.ts index 145fab6188..fb9e64f062 100644 --- a/packages/worker/src/app/account-deletion.ts +++ b/packages/worker/src/app/account-deletion.ts @@ -9,15 +9,13 @@ import { cleanupAllUserArtifactRepos } from '#worker/repo/artifact-repo-cleanup. import { repoSessionRpc } from '#worker/repo/repo-session-rpc.ts' import { userScopedConnectorSessionKey } from '#worker/remote-connector/connector-session-key.ts' import { mcpClientHubDurableObjectName } from '#worker/user-scoped-durable-object-name.ts' -import { - buildPackageServiceStorageId, - packageServiceRpc, -} from '#worker/package-runtime/package-service.ts' +import { packageServiceRpc } from '#worker/package-runtime/package-service.ts' import { packageRealtimeSessionRpc } from '#worker/package-runtime/realtime-session.ts' +import { clearRunRecords } from '#worker/run-records/service.ts' import { - clearRunRecords, - listRunRecordStorageIds, -} from '#worker/run-records/service.ts' + listAccountUserPackageServices, + listAccountUserStorageIds, +} from '#app/account-user-inventory.ts' import { accountUserDataTargets, buildUserScopedDeleteOrUpdateSql, @@ -195,68 +193,19 @@ async function listUserVectorIds(env: Env, userId: string) { return ids } -async function listUserStorageIds(env: Env, userId: string) { - const [ - jobRows, - archivedRows, - runtimeRows, - packageRows, - serviceRows, - runRecordStorageIds, - ] = await Promise.all([ - env.APP_DB.prepare( - `SELECT storage_id FROM jobs WHERE user_id = ? AND storage_id IS NOT NULL`, - ) - .bind(userId) - .all<{ storage_id: string }>(), - env.APP_DB.prepare( - `SELECT storage_id FROM archived_job_artifacts WHERE user_id = ? AND storage_id IS NOT NULL`, - ) - .bind(userId) - .all<{ storage_id: string }>(), - // Keep reading package_runtime_runs for storage ids: that table still - // holds legacy rows written before RunLog, and is deliberately not - // dropped yet. Remove this D1 read once the follow-up drop migration - // lands. - env.APP_DB.prepare( - `SELECT storage_id FROM package_runtime_runs WHERE user_id = ? AND storage_id IS NOT NULL`, - ) - .bind(userId) - .all<{ storage_id: string }>(), - env.APP_DB.prepare( - `SELECT id FROM saved_packages WHERE user_id = ? AND has_app = 1`, - ) - .bind(userId) - .all<{ id: string }>(), - env.APP_DB.prepare( - `SELECT DISTINCT package_id, name FROM ( - SELECT package_id, service_name AS name - FROM package_service_states - WHERE user_id = ? - UNION - -- Legacy arm: remove once the follow-up drop migration for - -- package_runtime_runs lands. - SELECT package_id, name - FROM package_runtime_runs - WHERE user_id = ? - AND surface = 'service' - AND name IS NOT NULL - )`, - ) - .bind(userId, userId) - .all<{ package_id: string; name: string }>(), - listRunRecordStorageIds({ env, userId }), - ]) - return uniqueStrings([ - ...(jobRows.results ?? []).map((row) => row.storage_id), - ...(archivedRows.results ?? []).map((row) => row.storage_id), - ...(runtimeRows.results ?? []).map((row) => row.storage_id), - ...(packageRows.results ?? []).map((row) => row.id), - ...(serviceRows.results ?? []).map((row) => - buildPackageServiceStorageId(row.package_id, row.name), - ), - ...runRecordStorageIds, - ]) +async function listUserStorageIds( + env: Env, + userId: string, + warnings?: Array, + packageServices?: ReadonlyArray, +) { + return await listAccountUserStorageIds({ + env, + userId, + baseUrl: 'https://account-deletion.invalid', + warnings, + packageServices, + }) } async function listUserSourceSnapshots(env: Env, userId: string) { @@ -327,52 +276,19 @@ async function listUserMcpServers(env: Env, userId: string) { return (rows.results ?? []).map((row) => ({ id: row.id })) } -async function listUserPackageServices(env: Env, userId: string) { - const rows = await env.APP_DB.prepare( - `SELECT DISTINCT - package_id, - kody_id, - source_id, - name - FROM ( - SELECT - s.package_id AS package_id, - p.kody_id AS kody_id, - p.source_id AS source_id, - s.service_name AS name - FROM package_service_states AS s - LEFT JOIN saved_packages AS p - ON p.id = s.package_id AND p.user_id = s.user_id - WHERE s.user_id = ? - UNION - -- Legacy arm: remove once the follow-up drop migration for - -- package_runtime_runs lands. - SELECT - r.package_id AS package_id, - COALESCE(p.kody_id, r.package_kody_id) AS kody_id, - COALESCE(p.source_id, r.source_id) AS source_id, - r.name AS name - FROM package_runtime_runs AS r - LEFT JOIN saved_packages AS p - ON p.id = r.package_id AND p.user_id = r.user_id - WHERE r.user_id = ? - AND r.surface = 'service' - AND r.name IS NOT NULL - )`, - ) - .bind(userId, userId) - .all<{ - package_id: string - kody_id: string | null - source_id: string | null - name: string - }>() - return (rows.results ?? []).map((row) => ({ - packageId: row.package_id, - kodyId: row.kody_id ?? '', - sourceId: row.source_id ?? '', - serviceName: row.name, - })) +async function listUserPackageServices( + env: Env, + userId: string, + warnings?: Array, +) { + return await listAccountUserPackageServices({ + env, + userId, + baseUrl: 'https://account-deletion.invalid', + warnings, + // Legacy pre-#955 service arm via includeLegacyRuntimeRuns (issue #956). + includeLegacyRuntimeRuns: true, + }) } async function listUserBundleKvKeys(input: { @@ -430,6 +346,16 @@ async function collectUserDeletionInventory(input: { input.warnings.push(warning) inventoryErrors.push(warning) } + // Enumerate services first so storage-id listing can reuse the result and + // avoid a second package-manifest pass in the same request. + const packageServices = await listUserPackageServices( + input.env, + input.userId, + input.warnings, + ).catch((error) => { + recordInventoryError('package services', error) + return [] as Array + }) const [ vectorIds, storageIds, @@ -440,13 +366,17 @@ async function collectUserDeletionInventory(input: { remoteConnectors, mcpServers, mcpAgentSessions, - packageServices, ] = await Promise.all([ listUserVectorIds(input.env, input.userId).catch((error) => { recordInventoryError('vector ids', error) return [] as Array }), - listUserStorageIds(input.env, input.userId).catch((error) => { + listUserStorageIds( + input.env, + input.userId, + input.warnings, + packageServices, + ).catch((error) => { recordInventoryError('storage ids', error) return [] as Array }), @@ -487,10 +417,6 @@ async function collectUserDeletionInventory(input: { return [] as Array }, ), - listUserPackageServices(input.env, input.userId).catch((error) => { - recordInventoryError('package services', error) - return [] as Array - }), ]) const bundleKvKeys = await listUserBundleKvKeys({ env: input.env, diff --git a/packages/worker/src/app/account-export.node.test.ts b/packages/worker/src/app/account-export.node.test.ts index c51d93903a..4d48024b6b 100644 --- a/packages/worker/src/app/account-export.node.test.ts +++ b/packages/worker/src/app/account-export.node.test.ts @@ -1,5 +1,6 @@ import { quoteSqlIdentifier } from '@kody-internal/shared/sql-literals.ts' import { readdirSync, readFileSync } from 'node:fs' +import { fileURLToPath } from 'node:url' import { DatabaseSync } from 'node:sqlite' import { expect, test, vi } from 'vitest' import { @@ -8,6 +9,7 @@ import { getAccountExportD1UserColumnCoverage, readAccountExportSection, } from './account-export.ts' +import { consoleWarn } from '#worker/test-support/console-spies.ts' function applyMigrations(db: DatabaseSync) { const migrationsDir = new URL('../../migrations/', import.meta.url) @@ -858,11 +860,14 @@ test('durable object discovery pages high-cardinality storage ids without nested maxRows = Math.max(maxRows, rows.length) return { results: rows as Array } } - if (query.includes('SELECT DISTINCT package_id, name')) { + if (query.includes('FROM package_service_states')) { return { results: [] as Array } } throw new Error(`Unexpected query: ${query}`) }, + async first() { + return null as T | null + }, } }, } @@ -891,6 +896,74 @@ test('durable object discovery pages high-cardinality storage ids without nested expect(maxRows).toBeLessThanOrEqual(101) }) +test('durable object discovery paging does not load package manifests', async () => { + const loadManifest = vi.spyOn( + await import('#worker/package-registry/source.ts'), + 'loadPackageManifestBySourceId', + ) + loadManifest.mockRejectedValue(new Error('manifest should not be loaded')) + try { + const { sqlite, db } = createMigratedDb() + sqlite.exec(` + INSERT INTO users ( + id, username, email, password_hash, created_at, updated_at, + email_verified_at, stable_user_id + ) + VALUES ( + 1, 'user-a', 'a@example.com', 'password-hash-a', '2026-07-05', + '2026-07-05', '2026-07-05', 'user-aaa' + ); + INSERT INTO saved_packages ( + id, user_id, name, kody_id, description, tags_json, source_id, + has_app, hidden, is_private, created_at, updated_at + ) VALUES ( + 'pkg-1', 'user-aaa', 'Pkg', 'pkg', '', '[]', + 'src-1', 0, 0, 1, '2026-07-05', '2026-07-05' + ); + INSERT INTO user_storage_buckets ( + user_id, storage_id, kind, created_at, last_seen_at + ) VALUES ( + 'user-aaa', 'exec:page-1', 'execute', '2026-07-05', '2026-07-05' + ); + INSERT INTO package_service_states ( + user_id, package_id, service_name, status, started_at, updated_at + ) VALUES ( + 'user-aaa', 'pkg-1', 'worker', 'idle', + null, '2026-07-05T00:00:00.000Z' + ); + `) + + const storagePage = await readAccountExportSection({ + env: { APP_DB: db } as Env, + dbUserId: 1, + mcpUserId: 'user-aaa', + section: 'durable_object_summaries', + kind: 'storage_runner', + pageSize: 50, + }) + expect(storagePage.items.length).toBeGreaterThan(0) + + const servicePage = await readAccountExportSection({ + env: { APP_DB: db } as Env, + dbUserId: 1, + mcpUserId: 'user-aaa', + section: 'durable_object_summaries', + kind: 'package_service', + pageSize: 50, + }) + expect(servicePage.items).toEqual([ + { + kind: 'package_service', + packageId: 'pkg-1', + serviceName: 'worker', + }, + ]) + expect(loadManifest).not.toHaveBeenCalled() + } finally { + loadManifest.mockRestore() + } +}) + test('createAccountExport redacts secrets and credential-equivalent hashes', async () => { const { sqlite, db } = createMigratedDb() sqlite.exec(` @@ -1184,12 +1257,10 @@ test('D1 export reads large tables in bounded keyset pages', async () => { insert.run(`message-${String(index).padStart(4, '0')}`, `Mail ${index}`) } sqlite.exec(` - INSERT INTO package_runtime_runs ( - id, user_id, package_id, package_kody_id, surface, name, status, - started_at, storage_id, created_at, updated_at + INSERT INTO user_storage_buckets ( + user_id, storage_id, kind, created_at, last_seen_at ) VALUES ( - 'service-run', 'user-aaa', 'pkg:1', 'pkg', 'service', 'svc x', - 'success', '2026-07-05', 'service:pkg%3A1:svc%20x', + 'user-aaa', 'service:pkg%3A1:svc%20x', 'service', '2026-07-05', '2026-07-05' ); `) @@ -1259,9 +1330,6 @@ test('D1 export reads large tables in bounded keyset pages', async () => { expect( queries.some((query) => query.includes('__account_export_rowid')), ).toBe(false) - expect( - queries.some((query) => query.startsWith('SELECT storage_id FROM jobs')), - ).toBe(false) }) test('account export includes run_records section with runs and log lines', async () => { @@ -1385,8 +1453,10 @@ test('account export includes run_records section with runs and log lines', asyn }) test('account export includes a package service known only via package_service_states', async () => { - const { sqlite, db } = createMigratedDb() - sqlite.exec(` + consoleWarn.mockImplementation(() => {}) + try { + const { sqlite, db } = createMigratedDb() + sqlite.exec(` INSERT INTO users ( id, username, email, password_hash, created_at, updated_at, email_verified_at, stable_user_id @@ -1410,49 +1480,369 @@ test('account export includes a package service known only via package_service_s ); `) - const statusMock = vi.fn(async () => - Response.json({ - package_id: 'pkg-states', - kody_id: 'states-pkg', - service_name: 'only-in-states', - status: 'running', - auto_start: false, - mode: 'bounded', - timeout_ms: 30_000, - stop_requested: false, - active_run_id: null, - next_alarm_at: null, - last_error: null, - last_started_at: '2026-07-05T00:00:00.000Z', - last_stopped_at: null, - last_run_finished_at: null, - last_result: null, - }), + const statusMock = vi.fn(async () => + Response.json({ + package_id: 'pkg-states', + kody_id: 'states-pkg', + service_name: 'only-in-states', + status: 'running', + auto_start: false, + mode: 'bounded', + timeout_ms: 30_000, + stop_requested: false, + active_run_id: null, + next_alarm_at: null, + last_error: null, + last_started_at: '2026-07-05T00:00:00.000Z', + last_stopped_at: null, + last_run_finished_at: null, + last_result: null, + }), + ) + + const accountExport = await createAccountExport({ + env: { + APP_DB: db, + PACKAGE_SERVICE_INSTANCE: { + idFromName: (name: string) => name as unknown as DurableObjectId, + get: () => ({ fetch: statusMock }), + }, + } as unknown as Env, + dbUserId: 1, + mcpUserId: 'user-aaa', + generatedAt: '2026-07-05T00:00:00.000Z', + }) + + expect(accountExport.manifest.sections.package_services?.count).toBe(1) + expect(accountExport.durableObjects.packageServices).toEqual([ + expect.objectContaining({ + packageId: 'pkg-states', + serviceName: 'only-in-states', + status: expect.objectContaining({ + service_name: 'only-in-states', + status: 'running', + }), + }), + ]) + expect(statusMock).toHaveBeenCalledTimes(1) + } finally { + consoleWarn.mockReset() + } +}) + +test('account export includes a storage runner known only via user_storage_buckets', async () => { + const { sqlite, db } = createMigratedDb() + sqlite.exec(` + INSERT INTO users ( + id, username, email, password_hash, created_at, updated_at, + email_verified_at, stable_user_id + ) + VALUES ( + 1, 'user-a', 'a@example.com', 'password-hash-a', '2026-07-05', + '2026-07-05', '2026-07-05', 'user-aaa' + ); + INSERT INTO user_storage_buckets ( + user_id, storage_id, kind, created_at, last_seen_at + ) VALUES ( + 'user-aaa', 'exec:export-only', 'execute', + '2026-07-05', '2026-07-05' + ); + `) + + const exportStorage = vi.fn(async () => ({ + entries: [{ key: 'alpha', value: { n: 1 } }], + truncated: false, + nextStartAfter: null, + pageSize: 100, + estimatedBytes: 10, + })) + + const env = { + APP_DB: db, + STORAGE_RUNNER: { + idFromName: (name: string) => name as unknown as DurableObjectId, + get: () => ({ exportStorage }), + }, + } as unknown as Env + + const manifest = await createAccountExportManifest({ + env, + dbUserId: 1, + mcpUserId: 'user-aaa', + }) + expect(manifest.sections.storage_runners?.count).toBe(1) + + const section = await readAccountExportSection({ + env, + dbUserId: 1, + mcpUserId: 'user-aaa', + section: 'storage_runner', + storageId: 'exec:export-only', + }) + expect(section.items).toEqual([{ key: 'alpha', value: { n: 1 } }]) + expect(exportStorage).toHaveBeenCalledTimes(1) +}) + +test('storage_runners count matches ids enumerable by discovery paging', async () => { + const { sqlite, db } = createMigratedDb() + sqlite.exec(` + INSERT INTO users ( + id, username, email, password_hash, created_at, updated_at, + email_verified_at, stable_user_id + ) + VALUES ( + 1, 'user-a', 'a@example.com', 'password-hash-a', '2026-07-05', + '2026-07-05', '2026-07-05', 'user-aaa' + ); + INSERT INTO jobs ( + id, user_id, name, source_id, storage_id, schedule_json, timezone, + caller_context_json, created_at, updated_at, next_run_at + ) VALUES ( + 'job-1', 'user-aaa', 'Job', 'src-job-1', 'job:job-1', '{}', 'UTC', + '{}', '2026-07-05', '2026-07-05', '2026-07-05' + ); + INSERT INTO user_storage_buckets ( + user_id, storage_id, kind, created_at, last_seen_at + ) VALUES ( + 'user-aaa', 'exec:adhoc', 'execute', '2026-07-05', '2026-07-05' + ); + INSERT INTO saved_packages ( + id, user_id, name, kody_id, description, tags_json, source_id, + has_app, hidden, is_private, created_at, updated_at + ) VALUES ( + 'pkg-1', 'user-aaa', 'Pkg', 'pkg', '', '[]', + 'src-1', 1, 0, 1, '2026-07-05', '2026-07-05' + ); + INSERT INTO package_service_states ( + user_id, package_id, service_name, status, started_at, updated_at + ) VALUES ( + 'user-aaa', 'pkg-1', 'worker', 'idle', + null, '2026-07-05T00:00:00.000Z' + ); + `) + + const env = { + APP_DB: db, + RUN_LOG: { + idFromName: (name: string) => name as unknown as DurableObjectId, + get: () => ({ + listStorageIds: async () => [ + 'exec:adhoc', + 'exec:runlog-only', + 'job:job-1', + ], + }), + }, + } as unknown as Env + const manifest = await createAccountExportManifest({ + env, + dbUserId: 1, + mcpUserId: 'user-aaa', + }) + const expectedCount = manifest.sections.storage_runners?.count + expect(expectedCount).toBeGreaterThan(0) + + const seen = new Set() + let startAfter: string | undefined + for (;;) { + const page = await readAccountExportSection({ + env, + dbUserId: 1, + mcpUserId: 'user-aaa', + section: 'durable_object_summaries', + kind: 'storage_runner', + pageSize: 2, + startAfter, + }) + for (const item of page.items as Array<{ storageId: string }>) { + seen.add(item.storageId) + } + if (!page.truncated) break + startAfter = page.nextStartAfter ?? undefined + } + expect(seen.size).toBe(expectedCount) + expect(seen.has('exec:runlog-only')).toBe(true) +}) + +test('storage_runner section exports a RunLog-only storage id', async () => { + const { sqlite, db } = createMigratedDb() + sqlite.exec(` + INSERT INTO users ( + id, username, email, password_hash, created_at, updated_at, + email_verified_at, stable_user_id + ) + VALUES ( + 1, 'user-a', 'a@example.com', 'password-hash-a', '2026-07-05', + '2026-07-05', '2026-07-05', 'user-aaa' + ); + `) + + const exportStorage = vi.fn(async () => ({ + entries: [{ key: 'runlog', value: { ok: true } }], + truncated: false, + nextStartAfter: null, + pageSize: 100, + estimatedBytes: 8, + })) + const env = { + APP_DB: db, + STORAGE_RUNNER: { + idFromName: (name: string) => name as unknown as DurableObjectId, + get: () => ({ exportStorage }), + }, + RUN_LOG: { + idFromName: (name: string) => name as unknown as DurableObjectId, + get: () => ({ + listStorageIds: async () => ['exec:runlog-export-only'], + }), + }, + } as unknown as Env + + const manifest = await createAccountExportManifest({ + env, + dbUserId: 1, + mcpUserId: 'user-aaa', + }) + expect(manifest.sections.storage_runners?.count).toBe(1) + + const section = await readAccountExportSection({ + env, + dbUserId: 1, + mcpUserId: 'user-aaa', + section: 'storage_runner', + storageId: 'exec:runlog-export-only', + }) + expect(section.items).toEqual([{ key: 'runlog', value: { ok: true } }]) + expect(exportStorage).toHaveBeenCalledTimes(1) +}) + +test('storage_runner and package_service section reads do not load manifests for D1-known rows', async () => { + const loadManifest = vi.spyOn( + await import('#worker/package-registry/source.ts'), + 'loadPackageManifestBySourceId', ) + loadManifest.mockRejectedValue(new Error('manifest should not be loaded')) + try { + const { sqlite, db } = createMigratedDb() + sqlite.exec(` + INSERT INTO users ( + id, username, email, password_hash, created_at, updated_at, + email_verified_at, stable_user_id + ) + VALUES ( + 1, 'user-a', 'a@example.com', 'password-hash-a', '2026-07-05', + '2026-07-05', '2026-07-05', 'user-aaa' + ); + INSERT INTO user_storage_buckets ( + user_id, storage_id, kind, created_at, last_seen_at + ) VALUES ( + 'user-aaa', 'exec:section-only', 'execute', + '2026-07-05', '2026-07-05' + ); + INSERT INTO saved_packages ( + id, user_id, name, kody_id, description, tags_json, source_id, + has_app, hidden, is_private, created_at, updated_at + ) VALUES ( + 'pkg-1', 'user-aaa', 'Pkg', 'pkg', '', '[]', + 'src-1', 0, 0, 1, '2026-07-05', '2026-07-05' + ); + INSERT INTO package_service_states ( + user_id, package_id, service_name, status, started_at, updated_at + ) VALUES ( + 'user-aaa', 'pkg-1', 'worker', 'idle', + null, '2026-07-05T00:00:00.000Z' + ); + `) - const accountExport = await createAccountExport({ - env: { + const exportStorage = vi.fn(async () => ({ + entries: [], + truncated: false, + nextStartAfter: null, + pageSize: 100, + estimatedBytes: 0, + })) + const statusMock = vi.fn(async () => + Response.json({ + package_id: 'pkg-1', + kody_id: 'pkg', + service_name: 'worker', + status: 'idle', + }), + ) + const env = { APP_DB: db, + STORAGE_RUNNER: { + idFromName: (name: string) => name as unknown as DurableObjectId, + get: () => ({ exportStorage }), + }, PACKAGE_SERVICE_INSTANCE: { idFromName: (name: string) => name as unknown as DurableObjectId, get: () => ({ fetch: statusMock }), }, - } as unknown as Env, - dbUserId: 1, - mcpUserId: 'user-aaa', - generatedAt: '2026-07-05T00:00:00.000Z', - }) + } as unknown as Env - expect(accountExport.manifest.sections.package_services?.count).toBe(1) - expect(accountExport.durableObjects.packageServices).toEqual([ - expect.objectContaining({ - packageId: 'pkg-states', - serviceName: 'only-in-states', - status: expect.objectContaining({ - service_name: 'only-in-states', - status: 'running', - }), + await readAccountExportSection({ + env, + dbUserId: 1, + mcpUserId: 'user-aaa', + section: 'storage_runner', + storageId: 'exec:section-only', + }) + await readAccountExportSection({ + env, + dbUserId: 1, + mcpUserId: 'user-aaa', + section: 'package_service', + packageId: 'pkg-1', + serviceName: 'worker', + }) + expect(loadManifest).not.toHaveBeenCalled() + } finally { + loadManifest.mockRestore() + } +}) + +test('storage_runner section treats malformed service storage ids as not found', async () => { + const { sqlite, db } = createMigratedDb() + sqlite.exec(` + INSERT INTO users ( + id, username, email, password_hash, created_at, updated_at, + email_verified_at, stable_user_id + ) + VALUES ( + 1, 'user-a', 'a@example.com', 'password-hash-a', '2026-07-05', + '2026-07-05', '2026-07-05', 'user-aaa' + ); + INSERT INTO saved_packages ( + id, user_id, name, kody_id, description, tags_json, source_id, + has_app, hidden, is_private, created_at, updated_at + ) VALUES ( + 'pkg%', 'user-aaa', 'Malformed', 'malformed', '', '[]', + 'src-malformed', 0, 0, 1, '2026-07-05', '2026-07-05' + ); + INSERT INTO package_service_states ( + user_id, package_id, service_name, status, started_at, updated_at + ) VALUES ( + 'user-aaa', 'pkg%', 'worker%', 'idle', + null, '2026-07-05T00:00:00.000Z' + ); + `) + + await expect( + readAccountExportSection({ + env: { APP_DB: db } as Env, + dbUserId: 1, + mcpUserId: 'user-aaa', + section: 'storage_runner', + storageId: 'service:pkg%:worker%', }), - ]) - expect(statusMock).toHaveBeenCalledTimes(1) + ).rejects.toThrow('Storage runner was not found for account export.') +}) + +test('account export source no longer reads package_runtime_runs', () => { + const source = readFileSync( + fileURLToPath(new URL('./account-export.ts', import.meta.url)), + 'utf8', + ) + expect(source.includes('package_runtime_runs')).toBe(false) }) diff --git a/packages/worker/src/app/account-export.ts b/packages/worker/src/app/account-export.ts index 5629c6736a..f101d41e2b 100644 --- a/packages/worker/src/app/account-export.ts +++ b/packages/worker/src/app/account-export.ts @@ -15,6 +15,8 @@ import { readAccountR2ExportPage, } from '#app/account-r2-export.ts' import { exportJobManagerForUser } from '#worker/jobs/manager-client.ts' +import { listPackageServices } from '#worker/package-registry/manifest.ts' +import { loadPackageManifestBySourceId } from '#worker/package-registry/source.ts' import { buildPackageServiceStorageId, packageServiceRpc, @@ -33,6 +35,10 @@ import { summarizeRunRecords, } from '#worker/run-records/service.ts' import { resolveUserStableId } from '#worker/user-id.ts' +import { + listAccountUserPackageServices, + listAccountUserStorageIds, +} from '#app/account-user-inventory.ts' const accountExportSchemaVersion = 1 const defaultExportPageSize = 100 @@ -444,70 +450,172 @@ async function collectD1TableRows(input: { return section } -async function listUserStorageIds(env: Env, userId: string) { - const [ - jobRows, - archivedRows, - runtimeRows, - packageRows, - serviceRows, - runRecordStorageIds, - ] = await Promise.all([ - selectRows<{ storage_id: string }>( - env, - `SELECT storage_id FROM jobs WHERE user_id = ? AND storage_id IS NOT NULL`, - [userId], - ), - selectRows<{ storage_id: string }>( - env, - `SELECT storage_id FROM archived_job_artifacts WHERE user_id = ? AND storage_id IS NOT NULL`, - [userId], - ), - // Keep reading package_runtime_runs for storage ids: that table still - // holds legacy rows written before RunLog, and is deliberately not - // dropped yet. Remove this D1 read once the follow-up drop migration - // lands. - selectRows<{ storage_id: string }>( - env, - `SELECT storage_id FROM package_runtime_runs WHERE user_id = ? AND storage_id IS NOT NULL`, - [userId], - ), - selectRows<{ id: string }>( - env, - `SELECT id FROM saved_packages WHERE user_id = ? AND has_app = 1`, - [userId], - ), - selectRows<{ package_id: string; name: string }>( - env, - `SELECT DISTINCT package_id, name FROM ( - SELECT package_id, service_name AS name - FROM package_service_states - WHERE user_id = ? - UNION - -- Legacy arm: remove once the follow-up drop migration for - -- package_runtime_runs lands. - SELECT package_id, name - FROM package_runtime_runs - WHERE user_id = ? - AND surface = 'service' - AND name IS NOT NULL - )`, - [userId, userId], - ), - listRunRecordStorageIds({ env, userId }), - ]) - return uniqueStrings([ - ...jobRows.map((row) => row.storage_id), - ...archivedRows.map((row) => row.storage_id), - ...runtimeRows.map((row) => row.storage_id), - ...packageRows.map((row) => row.id), - ...serviceRows.map((row) => - buildPackageServiceStorageId(row.package_id, row.name), - ), - ...runRecordStorageIds, +async function listUserStorageIds( + env: Env, + userId: string, + warnings?: Array, + packageServices?: ReadonlyArray, +) { + return await listAccountUserStorageIds({ + env, + userId, + baseUrl: 'https://account-export.invalid', + warnings, + packageServices, + }) +} + +/** D1 entity/registry storage ids used by discovery paging (no manifests). */ +const exportStorageIdBaseSql = `SELECT id FROM ( + SELECT storage_id AS id FROM jobs + WHERE user_id = ? AND storage_id IS NOT NULL + UNION SELECT storage_id FROM archived_job_artifacts + WHERE user_id = ? AND storage_id IS NOT NULL + UNION SELECT storage_id FROM user_storage_buckets + WHERE user_id = ? + UNION SELECT id FROM saved_packages + WHERE user_id = ? AND has_app = 1 +)` + +function exportStorageIdBaseParams(userId: string) { + return [userId, userId, userId, userId] as const +} + +function tryDecodeURIComponent(value: string) { + try { + return decodeURIComponent(value) + } catch { + return null + } +} + +function parseServiceStorageId(storageId: string) { + if (!storageId.startsWith('service:')) return null + const [packagePart, servicePart] = storageId + .slice('service:'.length) + .split(':') + if (!packagePart || !servicePart) return null + const packageId = tryDecodeURIComponent(packagePart) + const serviceName = tryDecodeURIComponent(servicePart) + if (packageId == null || serviceName == null) return null + if (!packageId || !serviceName) return null + return { packageId, serviceName } +} + +async function listExportD1DiscoverableStorageIds(env: Env, userId: string) { + const [baseRows, serviceRows] = await Promise.all([ + env.APP_DB.prepare(exportStorageIdBaseSql) + .bind(...exportStorageIdBaseParams(userId)) + .all<{ id: string }>(), + env.APP_DB.prepare( + `SELECT package_id, service_name AS name + FROM package_service_states + WHERE user_id = ?`, + ) + .bind(userId) + .all<{ package_id: string; name: string }>(), ]) + const ids = new Set((baseRows.results ?? []).map((row) => row.id)) + for (const row of serviceRows.results ?? []) { + ids.add(buildPackageServiceStorageId(row.package_id, row.name)) + } + return ids } +async function isExportDiscoverableStorageId( + env: Env, + userId: string, + storageId: string, +) { + const inBase = await env.APP_DB.prepare( + `SELECT 1 AS owned FROM (${exportStorageIdBaseSql}) WHERE id = ?`, + ) + .bind(...exportStorageIdBaseParams(userId), storageId) + .first<{ owned: number }>() + if (inBase?.owned === 1) return true + const parsed = parseServiceStorageId(storageId) + if (parsed) { + const row = await env.APP_DB.prepare( + `SELECT 1 AS owned + FROM package_service_states + WHERE user_id = ? AND package_id = ? AND service_name = ?`, + ) + .bind(userId, parsed.packageId, parsed.serviceName) + .first<{ owned: number }>() + if (row?.owned === 1) return true + } + const runRecordStorageIds = await listRunRecordStorageIds({ env, userId }) + return runRecordStorageIds.includes(storageId) +} + +async function resolveExportPackageService(input: { + env: Env + userId: string + packageId: string + serviceName: string + warnings: Array +}) { + const stateRow = await input.env.APP_DB.prepare( + `SELECT + s.package_id AS package_id, + p.kody_id AS kody_id, + p.source_id AS source_id, + s.service_name AS name + FROM package_service_states AS s + LEFT JOIN saved_packages AS p + ON p.id = s.package_id AND p.user_id = s.user_id + WHERE s.user_id = ? + AND s.package_id = ? + AND s.service_name = ?`, + ) + .bind(input.userId, input.packageId, input.serviceName) + .first<{ + package_id: string + kody_id: string | null + source_id: string | null + name: string + }>() + if (stateRow) { + return { + packageId: stateRow.package_id, + kodyId: stateRow.kody_id ?? '', + sourceId: stateRow.source_id ?? '', + serviceName: stateRow.name, + } + } + + const savedPackage = await input.env.APP_DB.prepare( + `SELECT id, kody_id, source_id + FROM saved_packages + WHERE user_id = ? AND id = ?`, + ) + .bind(input.userId, input.packageId) + .first<{ id: string; kody_id: string; source_id: string }>() + if (!savedPackage) return null + try { + const loaded = await loadPackageManifestBySourceId({ + env: input.env, + baseUrl: 'https://account-export.invalid', + userId: input.userId, + sourceId: savedPackage.source_id, + }) + const declared = listPackageServices(loaded.manifest).some( + (service) => service.name === input.serviceName, + ) + if (!declared) return null + return { + packageId: savedPackage.id, + kodyId: savedPackage.kody_id, + sourceId: savedPackage.source_id, + serviceName: input.serviceName, + } + } catch (error) { + input.warnings.push( + `Failed to load package manifest for service export (${input.packageId}/${input.serviceName}): ${getErrorMessage(error)}`, + ) + return null + } +} async function listUserSourceSnapshots(env: Env, userId: string) { const rows = await selectRows<{ id: string @@ -569,52 +677,17 @@ async function listUserRemoteConnectors(env: Env, userId: string) { })) } -async function listUserPackageServices(env: Env, userId: string) { - const rows = await selectRows<{ - package_id: string - kody_id: string | null - source_id: string | null - name: string - }>( +async function listUserPackageServices( + env: Env, + userId: string, + warnings?: Array, +) { + return await listAccountUserPackageServices({ env, - `SELECT DISTINCT - package_id, - kody_id, - source_id, - name - FROM ( - SELECT - s.package_id AS package_id, - p.kody_id AS kody_id, - p.source_id AS source_id, - s.service_name AS name - FROM package_service_states AS s - LEFT JOIN saved_packages AS p - ON p.id = s.package_id AND p.user_id = s.user_id - WHERE s.user_id = ? - UNION - -- Legacy arm: remove once the follow-up drop migration for - -- package_runtime_runs lands. - SELECT - r.package_id AS package_id, - COALESCE(p.kody_id, r.package_kody_id) AS kody_id, - COALESCE(p.source_id, r.source_id) AS source_id, - r.name AS name - FROM package_runtime_runs AS r - LEFT JOIN saved_packages AS p - ON p.id = r.package_id AND p.user_id = r.user_id - WHERE r.user_id = ? - AND r.surface = 'service' - AND r.name IS NOT NULL - )`, - [userId, userId], - ) - return rows.map((row) => ({ - packageId: row.package_id, - kodyId: row.kody_id ?? '', - sourceId: row.source_id ?? '', - serviceName: row.name, - })) + userId, + baseUrl: 'https://account-export.invalid', + warnings, + }) } async function listUserBundleKvKeys(input: { @@ -692,16 +765,32 @@ async function collectInventory(input: { dbUserId: number warnings: Array }): Promise { + // Enumerate services first so storage-id listing can reuse the result and + // avoid a second package-manifest pass in the same request. + const packageServices = await listUserPackageServices( + input.env, + input.userId, + input.warnings, + ).catch((error) => { + input.warnings.push( + `Failed to enumerate package services: ${getErrorMessage(error)}`, + ) + return [] as Array + }) const [ storageIds, sourceSnapshots, savedPackages, remoteConnectors, - packageServices, communityListingIds, r2ObjectCount, ] = await Promise.all([ - listUserStorageIds(input.env, input.userId).catch((error) => { + listUserStorageIds( + input.env, + input.userId, + input.warnings, + packageServices, + ).catch((error) => { input.warnings.push( `Failed to enumerate storage ids: ${getErrorMessage(error)}`, ) @@ -725,12 +814,6 @@ async function collectInventory(input: { ) return [] as Array }), - listUserPackageServices(input.env, input.userId).catch((error) => { - input.warnings.push( - `Failed to enumerate package services: ${getErrorMessage(error)}`, - ) - return [] as Array - }), listUserCommunityListingIds(input.env, input.userId).catch((error) => { input.warnings.push( `Failed to enumerate community listings: ${getErrorMessage(error)}`, @@ -794,80 +877,17 @@ async function countScalar( } async function countUserStorageIds(env: Env, userId: string) { - const baseSql = `SELECT id FROM ( - SELECT storage_id AS id FROM jobs - WHERE user_id = ? AND storage_id IS NOT NULL - UNION SELECT storage_id FROM archived_job_artifacts - WHERE user_id = ? AND storage_id IS NOT NULL - UNION SELECT storage_id FROM package_runtime_runs - WHERE user_id = ? AND storage_id IS NOT NULL - UNION SELECT id FROM saved_packages - WHERE user_id = ? AND has_app = 1 - )` - let count = await countScalar( - env, - `SELECT COUNT(*) AS count FROM (${baseSql})`, - Array(4).fill(userId), - ) - const countedExtraIds = new Set() - let afterPackageId = '' - let afterName = '' - for (;;) { - const page = await env.APP_DB.prepare( - `SELECT DISTINCT package_id, name FROM ( - SELECT package_id, service_name AS name - FROM package_service_states - WHERE user_id = ? - UNION - -- Legacy arm: remove once the follow-up drop migration for - -- package_runtime_runs lands. - SELECT package_id, name - FROM package_runtime_runs - WHERE user_id = ? - AND surface = 'service' - AND name IS NOT NULL - ) - WHERE ( - package_id > ? OR (package_id = ? AND name > ?) - ) - ORDER BY package_id, name - LIMIT 100`, - ) - .bind(userId, userId, afterPackageId, afterPackageId, afterName) - .all<{ package_id: string; name: string }>() - const rows = page.results ?? [] - if (rows.length === 0) break - for (const row of rows) { - const storageId = buildPackageServiceStorageId(row.package_id, row.name) - const existing = await env.APP_DB.prepare( - `SELECT 1 AS owned FROM (${baseSql}) WHERE id = ?`, - ) - .bind(userId, userId, userId, userId, storageId) - .first<{ owned: number }>() - if (existing?.owned !== 1 && !countedExtraIds.has(storageId)) { - countedExtraIds.add(storageId) - count += 1 - } - } - const last = rows.at(-1)! - afterPackageId = last.package_id - afterName = last.name - if (rows.length < 100) break - } - const runRecordStorageIds = await listRunRecordStorageIds({ env, userId }) + // Match durable_object_summaries discovery: D1 base + package_service_states + // + RunLog storage ids (one RunLog RPC, not per-package manifests). + const [d1Ids, runRecordStorageIds] = await Promise.all([ + listExportD1DiscoverableStorageIds(env, userId), + listRunRecordStorageIds({ env, userId }), + ]) + const ids = new Set(d1Ids) for (const storageId of runRecordStorageIds) { - if (countedExtraIds.has(storageId)) continue - const existing = await env.APP_DB.prepare( - `SELECT 1 AS owned FROM (${baseSql}) WHERE id = ?`, - ) - .bind(userId, userId, userId, userId, storageId) - .first<{ owned: number }>() - if (existing?.owned !== 1) { - countedExtraIds.add(storageId) - count += 1 - } + ids.add(storageId) } - return count + return ids.size } async function countUserBundleKvKeys(input: { @@ -986,21 +1006,11 @@ async function collectManifestInventoryCounts(input: { countScalar( input.env, `SELECT COUNT(*) AS count FROM ( - SELECT DISTINCT package_id, name FROM ( - SELECT package_id, service_name AS name - FROM package_service_states - WHERE user_id = ? - UNION - -- Legacy arm: remove once the follow-up drop migration for - -- package_runtime_runs lands. - SELECT package_id, name - FROM package_runtime_runs - WHERE user_id = ? - AND surface = 'service' - AND name IS NOT NULL - ) + SELECT DISTINCT package_id, service_name + FROM package_service_states + WHERE user_id = ? )`, - [input.userId, input.userId], + [input.userId], ), ), safeCount('artifact repos', async () => @@ -1711,65 +1721,11 @@ export async function readAccountExportSection(input: { if (!input.storageId) { throw new Error('storage_id is required when section is storage_runner.') } - let storageOwned = - ( - await input.env.APP_DB.prepare( - `SELECT 1 AS owned FROM ( - SELECT storage_id AS id FROM jobs WHERE user_id = ? - UNION SELECT storage_id FROM archived_job_artifacts WHERE user_id = ? - UNION SELECT storage_id FROM package_runtime_runs WHERE user_id = ? - UNION SELECT id FROM saved_packages WHERE user_id = ? AND has_app = 1 - ) WHERE id = ?`, - ) - .bind( - input.mcpUserId, - input.mcpUserId, - input.mcpUserId, - input.mcpUserId, - input.storageId, - ) - .first<{ owned: number }>() - )?.owned === 1 - if (!storageOwned && input.storageId.startsWith('service:')) { - const [packagePart, servicePart] = input.storageId - .slice('service:'.length) - .split(':') - if (packagePart && servicePart) { - const packageId = decodeURIComponent(packagePart) - const serviceName = decodeURIComponent(servicePart) - const row = await input.env.APP_DB.prepare( - `SELECT 1 AS owned FROM ( - SELECT 1 AS owned - FROM package_service_states - WHERE user_id = ? AND package_id = ? AND service_name = ? - UNION - -- Legacy arm: remove once the follow-up drop migration for - -- package_runtime_runs lands. - SELECT 1 AS owned - FROM package_runtime_runs - WHERE user_id = ? AND surface = 'service' - AND package_id = ? AND name = ? - )`, - ) - .bind( - input.mcpUserId, - packageId, - serviceName, - input.mcpUserId, - packageId, - serviceName, - ) - .first<{ owned: number }>() - storageOwned = row?.owned === 1 - } - } - if (!storageOwned) { - const runRecordStorageIds = await listRunRecordStorageIds({ - env: input.env, - userId: input.mcpUserId, - }) - storageOwned = runRecordStorageIds.includes(input.storageId) - } + const storageOwned = await isExportDiscoverableStorageId( + input.env, + input.mcpUserId, + input.storageId, + ) if (!storageOwned) { throw new Error('Storage runner was not found for account export.') } @@ -1892,67 +1848,18 @@ export async function readAccountExportSection(input: { 'package_id and service_name are required when section is package_service.', ) } - const row = await input.env.APP_DB.prepare( - `SELECT DISTINCT - package_id, - kody_id, - source_id, - name - FROM ( - SELECT - s.package_id AS package_id, - p.kody_id AS kody_id, - p.source_id AS source_id, - s.service_name AS name - FROM package_service_states AS s - LEFT JOIN saved_packages AS p - ON p.id = s.package_id AND p.user_id = s.user_id - WHERE s.user_id = ? - AND s.package_id = ? - AND s.service_name = ? - UNION - -- Legacy arm: remove once the follow-up drop migration for - -- package_runtime_runs lands. - SELECT - r.package_id AS package_id, - COALESCE(p.kody_id, r.package_kody_id) AS kody_id, - COALESCE(p.source_id, r.source_id) AS source_id, - r.name AS name - FROM package_runtime_runs AS r - LEFT JOIN saved_packages AS p - ON p.id = r.package_id AND p.user_id = r.user_id - WHERE r.user_id = ? - AND r.surface = 'service' - AND r.package_id = ? - AND r.name = ? - )`, - ) - .bind( - input.mcpUserId, - input.packageId, - input.serviceName, - input.mcpUserId, - input.packageId, - input.serviceName, - ) - .first<{ - package_id: string - kody_id: string | null - source_id: string | null - name: string - }>() - if (!row) throw new Error('Package service was not found for export.') + const service = await resolveExportPackageService({ + env: input.env, + userId: input.mcpUserId, + packageId: input.packageId, + serviceName: input.serviceName, + warnings, + }) + if (!service) throw new Error('Package service was not found for export.') const [exported] = await exportPackageServices({ env: input.env, userId: input.mcpUserId, - services: [ - { - packageId: row.package_id, - kodyId: row.kody_id ?? '', - sourceId: row.source_id ?? '', - serviceName: row.name, - }, - ], + services: [service], warnings, }) return { @@ -2067,27 +1974,22 @@ export async function readAccountExportSection(input: { } } if (input.kind === 'package_service') { + // Discovery pages are D1-only keyset SQL so each page stays cheap + // and bounded. Manifest-declared services (never projected into + // package_service_states) are included by the one-shot + // listAccountUserPackageServices path used for full export + // inventory and account deletion — not here. const afterPackageId = String(cursor['packageId'] ?? '') const afterName = String(cursor['name'] ?? '') const rows = await input.env.APP_DB.prepare( - `SELECT DISTINCT package_id, name FROM ( - SELECT package_id, service_name AS name - FROM package_service_states - WHERE user_id = ? - UNION - -- Legacy arm: remove once the follow-up drop migration for - -- package_runtime_runs lands. - SELECT package_id, name - FROM package_runtime_runs - WHERE user_id = ? - AND surface = 'service' - AND name IS NOT NULL - ) - WHERE (package_id > ? OR (package_id = ? AND name > ?)) - ORDER BY package_id, name LIMIT ?`, + `SELECT package_id, service_name AS name + FROM package_service_states + WHERE user_id = ? + AND (package_id > ? OR (package_id = ? AND service_name > ?)) + ORDER BY package_id, service_name + LIMIT ?`, ) .bind( - input.mcpUserId, input.mcpUserId, afterPackageId, afterPackageId, @@ -2116,20 +2018,17 @@ export async function readAccountExportSection(input: { warnings, } } + // Discovery pages: D1 keyset SQL (user_storage_buckets + entity + // tables + package_service_states) then RunLog-only ids. Do not call + // listAccountUserStorageIds / listAccountUserPackageServices here — + // those helpers fetch package manifests and are for one-shot full + // export inventory and account deletion completeness. RunLog is one + // Durable Object RPC per request, not per package. const stage = String(cursor['stage'] ?? 'base') if (stage === 'base') { const afterId = String(cursor['afterId'] ?? '') const rows = await input.env.APP_DB.prepare( - `SELECT id FROM ( - SELECT storage_id AS id FROM jobs - WHERE user_id = ? AND storage_id IS NOT NULL - UNION SELECT storage_id FROM archived_job_artifacts - WHERE user_id = ? AND storage_id IS NOT NULL - UNION SELECT storage_id FROM package_runtime_runs - WHERE user_id = ? AND storage_id IS NOT NULL - UNION SELECT id FROM saved_packages - WHERE user_id = ? AND has_app = 1 - ) WHERE id > ? ORDER BY id LIMIT ?`, + `${exportStorageIdBaseSql} WHERE id > ? ORDER BY id LIMIT ?`, ) .bind( input.mcpUserId, @@ -2159,74 +2058,104 @@ export async function readAccountExportSection(input: { warnings, } } - const afterPackageId = String(cursor['packageId'] ?? '') - const afterName = String(cursor['name'] ?? '') - const rows = await input.env.APP_DB.prepare( - `SELECT DISTINCT package_id, name FROM ( - SELECT package_id, service_name AS name + if (stage === 'service') { + const afterPackageId = String(cursor['packageId'] ?? '') + const afterName = String(cursor['name'] ?? '') + const rows = await input.env.APP_DB.prepare( + `SELECT package_id, service_name AS name FROM package_service_states WHERE user_id = ? - UNION - -- Legacy arm: remove once the follow-up drop migration for - -- package_runtime_runs lands. - SELECT package_id, name - FROM package_runtime_runs - WHERE user_id = ? - AND surface = 'service' - AND name IS NOT NULL - ) - WHERE (package_id > ? OR (package_id = ? AND name > ?)) - ORDER BY package_id, name LIMIT ?`, - ) - .bind( - input.mcpUserId, - input.mcpUserId, - afterPackageId, - afterPackageId, - afterName, - pageSize + 1, - ) - .all<{ package_id: string; name: string }>() - const pageRows = rows.results ?? [] - const selected = pageRows.slice(0, pageSize) - const resultItems: Array<{ kind: string; storageId: string }> = [] - for (const row of selected) { - const storageId = buildPackageServiceStorageId(row.package_id, row.name) - const existing = await input.env.APP_DB.prepare( - `SELECT 1 AS owned FROM ( - SELECT storage_id AS id FROM jobs WHERE user_id = ? - UNION SELECT storage_id FROM archived_job_artifacts WHERE user_id = ? - UNION SELECT storage_id FROM package_runtime_runs WHERE user_id = ? - UNION SELECT id FROM saved_packages WHERE user_id = ? AND has_app = 1 - ) WHERE id = ?`, + AND (package_id > ? OR (package_id = ? AND service_name > ?)) + ORDER BY package_id, service_name + LIMIT ?`, ) .bind( input.mcpUserId, - input.mcpUserId, - input.mcpUserId, - input.mcpUserId, - storageId, + afterPackageId, + afterPackageId, + afterName, + pageSize + 1, + ) + .all<{ package_id: string; name: string }>() + const pageRows = rows.results ?? [] + const selected = pageRows.slice(0, pageSize) + const resultItems: Array<{ kind: string; storageId: string }> = [] + const candidateIds = selected.map((row) => + buildPackageServiceStorageId(row.package_id, row.name), + ) + const alreadyInBase = new Set() + if (candidateIds.length > 0) { + const placeholders = candidateIds.map(() => '?').join(', ') + const existing = await input.env.APP_DB.prepare( + `SELECT id FROM (${exportStorageIdBaseSql}) WHERE id IN (${placeholders})`, ) - .first<{ owned: number }>() - if (existing?.owned !== 1) { - resultItems.push({ kind: input.kind, storageId }) + .bind( + ...exportStorageIdBaseParams(input.mcpUserId), + ...candidateIds, + ) + .all<{ id: string }>() + for (const row of existing.results ?? []) { + alreadyInBase.add(row.id) + } + } + for (const storageId of candidateIds) { + if (!alreadyInBase.has(storageId)) { + resultItems.push({ kind: input.kind, storageId }) + } + } + const truncated = pageRows.length > pageSize + return { + section: input.section, + items: resultItems, + truncated: true, + nextStartAfter: JSON.stringify( + truncated + ? { + stage: 'service', + packageId: selected.at(-1)!.package_id, + name: selected.at(-1)!.name, + } + : { stage: 'runlog', afterId: '' }, + ), + pageSize, + warnings, } } - const truncated = pageRows.length > pageSize - return { - section: input.section, - items: resultItems, - truncated, - nextStartAfter: truncated - ? JSON.stringify({ - stage: 'service', - packageId: selected.at(-1)!.package_id, - name: selected.at(-1)!.name, - }) - : null, - pageSize, - warnings, + if (stage === 'runlog') { + const afterId = String(cursor['afterId'] ?? '') + const [d1Ids, runRecordStorageIds] = await Promise.all([ + listExportD1DiscoverableStorageIds(input.env, input.mcpUserId), + listRunRecordStorageIds({ + env: input.env, + userId: input.mcpUserId, + }), + ]) + const exclusive = runRecordStorageIds + .filter((storageId) => !d1Ids.has(storageId)) + .sort((left, right) => left.localeCompare(right)) + const pageRows = exclusive.filter((storageId) => storageId > afterId) + const truncated = pageRows.length > pageSize + const selected = truncated ? pageRows.slice(0, pageSize) : pageRows + return { + section: input.section, + items: selected.map((storageId) => ({ + kind: input.kind, + storageId, + })), + truncated, + nextStartAfter: truncated + ? JSON.stringify({ + stage: 'runlog', + afterId: selected.at(-1)!, + }) + : null, + pageSize, + warnings, + } } + throw new Error( + `Unknown durable_object_summaries storage_runner stage: ${stage}`, + ) } default: { const exhaustive: never = input.section diff --git a/packages/worker/src/app/account-retention-dispositions.node.test.ts b/packages/worker/src/app/account-retention-dispositions.node.test.ts index 5a4cfaa460..b40d53533d 100644 --- a/packages/worker/src/app/account-retention-dispositions.node.test.ts +++ b/packages/worker/src/app/account-retention-dispositions.node.test.ts @@ -50,6 +50,13 @@ test('retention dispositions stay aligned with scheduled policies and documented disposition.kind === 'durable_forever', ), ).toBe(true) + expect( + nonScheduled.some( + (disposition) => + disposition.table === 'user_storage_buckets' && + disposition.kind === 'durable_forever', + ), + ).toBe(true) // The schema-growth heuristic in retention.node.test.ts remains the broader // guardrail for discovering new growth-pattern tables. expect(getRetentionPolicyCoverage().has('mcp_memories')).toBe(true) diff --git a/packages/worker/src/app/account-retention-dispositions.ts b/packages/worker/src/app/account-retention-dispositions.ts index 4fe01bf1dd..300bbb54ee 100644 --- a/packages/worker/src/app/account-retention-dispositions.ts +++ b/packages/worker/src/app/account-retention-dispositions.ts @@ -41,6 +41,12 @@ export const accountRetentionDispositions: ReadonlyArray { diff --git a/packages/worker/src/app/account-user-inventory.ts b/packages/worker/src/app/account-user-inventory.ts new file mode 100644 index 0000000000..0ca273eb37 --- /dev/null +++ b/packages/worker/src/app/account-user-inventory.ts @@ -0,0 +1,249 @@ +import { getErrorMessage } from '@kody-internal/shared/error-message.ts' +import { listPackageServices } from '#worker/package-registry/manifest.ts' +import { listSavedPackagesByUserId } from '#worker/package-registry/repo.ts' +import { loadPackageManifestBySourceId } from '#worker/package-registry/source.ts' +import { buildPackageServiceStorageId } from '#worker/package-runtime/package-service.ts' +import { listRunRecordStorageIds } from '#worker/run-records/service.ts' +import { listUserStorageBucketIds } from '#worker/storage-buckets/service.ts' + +export type AccountUserPackageService = { + packageId: string + kodyId: string + sourceId: string + serviceName: string +} + +function uniqueStrings(values: Iterable) { + return Array.from( + new Set( + Array.from(values) + .map((value) => value?.trim() ?? '') + .filter((value) => value.length > 0), + ), + ) +} + +function packageServiceKey(service: AccountUserPackageService) { + return `${service.packageId}\0${service.serviceName}` +} + +function reportManifestEnumerationWarning( + warnings: Array | undefined, + message: string, +) { + console.warn(message) + if (!warnings) return + if (!warnings.includes(message)) warnings.push(message) +} + +function upsertPackageService( + byKey: Map, + service: AccountUserPackageService, +) { + const key = packageServiceKey(service) + const existing = byKey.get(key) + if (existing) { + byKey.set(key, { + packageId: existing.packageId, + kodyId: existing.kodyId || service.kodyId, + sourceId: existing.sourceId || service.sourceId, + serviceName: existing.serviceName, + }) + return + } + byKey.set(key, service) +} + +/** + * Enumerate package services for account deletion and one-shot full export + * inventory. Not for paginated export discovery — that path uses D1 keyset SQL + * plus RunLog (see account-export durable_object_summaries). + * + * Authoritative sources are the package manifest (`kody.services`) unioned with + * projected `package_service_states` rows. Manifest loads can fail (network / + * source bindings); those failures warn and fall back to state-table rows so + * deletion never aborts for a missing manifest. Pass `warnings` so callers can + * surface that degradation in their result (incomplete deletion must not look + * clean). + * + * Pass `includeLegacyRuntimeRuns: true` on the deletion path only. That arm + * catches pre-#955 services whose DOs still exist but never projected into + * `package_service_states` and are no longer declared in the manifest. + * Migration 0097 backfills storage-bucket ownership, not service identity. + * Remove once `package_runtime_runs` drains (~2026-08-25); tracked by #956. + */ +export async function listAccountUserPackageServices(input: { + env: Env + userId: string + baseUrl: string + warnings?: Array + includeLegacyRuntimeRuns?: boolean +}): Promise> { + const stateRows = await input.env.APP_DB.prepare( + `SELECT + s.package_id AS package_id, + p.kody_id AS kody_id, + p.source_id AS source_id, + s.service_name AS name + FROM package_service_states AS s + LEFT JOIN saved_packages AS p + ON p.id = s.package_id AND p.user_id = s.user_id + WHERE s.user_id = ?`, + ) + .bind(input.userId) + .all<{ + package_id: string + kody_id: string | null + source_id: string | null + name: string + }>() + + const byKey = new Map() + for (const row of stateRows.results ?? []) { + upsertPackageService(byKey, { + packageId: row.package_id, + kodyId: row.kody_id ?? '', + sourceId: row.source_id ?? '', + serviceName: row.name, + }) + } + + try { + const packages = await listSavedPackagesByUserId(input.env.APP_DB, { + userId: input.userId, + }) + for (const savedPackage of packages) { + try { + const loaded = await loadPackageManifestBySourceId({ + env: input.env, + baseUrl: input.baseUrl, + userId: input.userId, + sourceId: savedPackage.sourceId, + }) + for (const service of listPackageServices(loaded.manifest)) { + upsertPackageService(byKey, { + packageId: savedPackage.id, + kodyId: savedPackage.kodyId, + sourceId: savedPackage.sourceId, + serviceName: service.name, + }) + } + } catch (error) { + reportManifestEnumerationWarning( + input.warnings, + `Failed to load package manifest for service enumeration (package ${savedPackage.id}): ${getErrorMessage(error)}`, + ) + } + } + } catch (error) { + reportManifestEnumerationWarning( + input.warnings, + `Failed to enumerate package services from manifests for user ${input.userId}: ${getErrorMessage(error)}`, + ) + } + + if (input.includeLegacyRuntimeRuns) { + // Legacy arm for account deletion only (issue #956): remove once + // package_runtime_runs drains (~2026-08-25). Covers pre-#955 service + // DOs that never projected into package_service_states and are no + // longer declared in the package manifest. 0097 backfills storage + // bucket ids, not (packageId, serviceName) tuples. + const legacyRows = await input.env.APP_DB.prepare( + `SELECT + r.package_id AS package_id, + COALESCE(p.kody_id, r.package_kody_id) AS kody_id, + COALESCE(p.source_id, r.source_id) AS source_id, + r.name AS name + FROM package_runtime_runs AS r + LEFT JOIN saved_packages AS p + ON p.id = r.package_id AND p.user_id = r.user_id + WHERE r.user_id = ? + AND r.surface = 'service' + AND r.name IS NOT NULL`, + ) + .bind(input.userId) + .all<{ + package_id: string + kody_id: string | null + source_id: string | null + name: string + }>() + for (const row of legacyRows.results ?? []) { + upsertPackageService(byKey, { + packageId: row.package_id, + kodyId: row.kody_id ?? '', + sourceId: row.source_id ?? '', + serviceName: row.name, + }) + } + } + + return Array.from(byKey.values()).sort((left, right) => { + const byPackage = left.packageId.localeCompare(right.packageId) + if (byPackage !== 0) return byPackage + return left.serviceName.localeCompare(right.serviceName) + }) +} + +/** + * Enumerate StorageRunner bucket ids for account deletion and one-shot full + * export inventory. Not for paginated export discovery — that path uses D1 + * keyset SQL plus RunLog (see account-export durable_object_summaries). + * + * Unions authoritative entity tables, the user storage-bucket registry, + * declared/projected package services, and current RunLog ids. + * + * Pass `packageServices` when the caller already enumerated services in this + * request to avoid loading package manifests twice. + */ +export async function listAccountUserStorageIds(input: { + env: Env + userId: string + baseUrl: string + warnings?: Array + includeLegacyRuntimeRuns?: boolean + packageServices?: ReadonlyArray +}): Promise> { + const [ + jobRows, + archivedRows, + bucketIds, + packageRows, + packageServices, + runRecordStorageIds, + ] = await Promise.all([ + input.env.APP_DB.prepare( + `SELECT storage_id FROM jobs WHERE user_id = ? AND storage_id IS NOT NULL`, + ) + .bind(input.userId) + .all<{ storage_id: string }>(), + input.env.APP_DB.prepare( + `SELECT storage_id FROM archived_job_artifacts WHERE user_id = ? AND storage_id IS NOT NULL`, + ) + .bind(input.userId) + .all<{ storage_id: string }>(), + listUserStorageBucketIds({ + env: input.env, + userId: input.userId, + }), + input.env.APP_DB.prepare( + `SELECT id FROM saved_packages WHERE user_id = ? AND has_app = 1`, + ) + .bind(input.userId) + .all<{ id: string }>(), + input.packageServices + ? Promise.resolve([...input.packageServices]) + : listAccountUserPackageServices(input), + listRunRecordStorageIds({ env: input.env, userId: input.userId }), + ]) + return uniqueStrings([ + ...(jobRows.results ?? []).map((row) => row.storage_id), + ...(archivedRows.results ?? []).map((row) => row.storage_id), + ...bucketIds, + ...(packageRows.results ?? []).map((row) => row.id), + ...packageServices.map((service) => + buildPackageServiceStorageId(service.packageId, service.serviceName), + ), + ...runRecordStorageIds, + ]) +} diff --git a/packages/worker/src/dr/exporter.node.test.ts b/packages/worker/src/dr/exporter.node.test.ts index 495f6e6cba..a61a1c80be 100644 --- a/packages/worker/src/dr/exporter.node.test.ts +++ b/packages/worker/src/dr/exporter.node.test.ts @@ -1,3 +1,5 @@ +import { readFileSync } from 'node:fs' +import { fileURLToPath } from 'node:url' import { expect, test, vi } from 'vitest' import { backupBlobKey, @@ -9,6 +11,7 @@ import { import { sha256Hex } from '#worker/dr/sha256.ts' import { drExportMaxStorageDumpBufferBytes, + listPlatformStorageInventory, runDrExportTick, shouldRunDrExportCron, } from '#worker/dr/exporter.ts' @@ -23,12 +26,24 @@ const storageMocks = vi.hoisted(() => ({ exportStorage: vi.fn(), })) +const storageBucketMocks = vi.hoisted(() => ({ + listPlatformStorageBuckets: vi.fn( + async () => [] as Array<{ userId: string; storageId: string }>, + ), +})) + vi.mock('#worker/storage-runner.ts', () => ({ storageRunnerRpc: () => ({ exportStorage: storageMocks.exportStorage, }), })) +vi.mock('#worker/storage-buckets/service.ts', () => ({ + listPlatformStorageBuckets: storageBucketMocks.listPlatformStorageBuckets, + listUserStorageBucketIds: vi.fn(async () => []), + registerStorageBucket: vi.fn(), +})) + function etagFor(bytes: Uint8Array) { let hash = 0 for (const value of bytes) hash = (hash * 31 + value) >>> 0 @@ -78,7 +93,6 @@ function createMemoryS3() { function createDb(results: { jobs?: Array<{ userId: string; storageId: string }> archived?: Array<{ userId: string; storageId: string }> - runtime?: Array<{ userId: string; storageId: string }> packages?: Array<{ userId: string; storageId: string }> services?: Array<{ userId: string @@ -103,15 +117,9 @@ function createDb(results: { if (sql.includes('FROM archived_job_artifacts')) { return { results: results.archived ?? [] } } - if ( - sql.includes("surface = 'service'") || - sql.includes('surface = "service"') - ) { + if (sql.includes('FROM package_service_states')) { return { results: results.services ?? [] } } - if (sql.includes('FROM package_runtime_runs')) { - return { results: results.runtime ?? [] } - } if (sql.includes('FROM saved_packages')) { return { results: results.packages ?? [] } } @@ -561,3 +569,33 @@ test('R2 export does not duplicate index lines across budget interruptions', asy dateNow.mockRestore() } }) + +test('DR inventory includes registry buckets and package_service_states services', async () => { + storageBucketMocks.listPlatformStorageBuckets.mockResolvedValueOnce([ + { userId: 'user-a', storageId: 'exec:adhoc-only' }, + ]) + const inventory = await listPlatformStorageInventory( + createDb({ + services: [ + { + userId: 'user-a', + packageId: 'pkg-1', + serviceName: 'worker', + }, + ], + }), + ) + expect(inventory.map((entry) => entry.storageId).sort()).toEqual([ + 'exec:adhoc-only', + 'service:pkg-1:worker', + ]) + expect(inventory.every((entry) => entry.userId === 'user-a')).toBe(true) +}) + +test('DR exporter source no longer reads package_runtime_runs', () => { + const source = readFileSync( + fileURLToPath(new URL('./exporter.ts', import.meta.url)), + 'utf8', + ) + expect(source.includes('package_runtime_runs')).toBe(false) +}) diff --git a/packages/worker/src/dr/exporter.ts b/packages/worker/src/dr/exporter.ts index 523798f635..8c637694b7 100644 --- a/packages/worker/src/dr/exporter.ts +++ b/packages/worker/src/dr/exporter.ts @@ -20,6 +20,7 @@ import { } from '@kody-internal/shared/backup-staging.ts' import { buildPackageServiceStorageId } from '#worker/package-runtime/package-service.ts' import { buildPublishedSourceSnapshotKvKey } from '#worker/package-runtime/published-runtime-artifacts.ts' +import { listPlatformStorageBuckets } from '#worker/storage-buckets/service.ts' import { storageRunnerRpc } from '#worker/storage-runner.ts' import { createDrBackupS3Client, @@ -223,7 +224,11 @@ function ndjsonLine(value: unknown) { export async function listPlatformStorageInventory( db: D1Database, ): Promise> { - const [jobRows, archivedRows, runtimeRows, packageRows, serviceRows] = + // Platform-wide DR has only a D1Database (no per-user Env / network), so + // service buckets come from projected `package_service_states` rather than + // package manifests. Ad-hoc / execute buckets come from the authoritative + // `user_storage_buckets` registry via `listPlatformStorageBuckets`. + const [jobRows, archivedRows, registeredBuckets, packageRows, serviceRows] = await Promise.all([ db .prepare( @@ -237,12 +242,7 @@ export async function listPlatformStorageInventory( FROM archived_job_artifacts WHERE storage_id IS NOT NULL`, ) .all<{ userId: string; storageId: string }>(), - db - .prepare( - `SELECT user_id AS userId, storage_id AS storageId - FROM package_runtime_runs WHERE storage_id IS NOT NULL`, - ) - .all<{ userId: string; storageId: string }>(), + listPlatformStorageBuckets({ db }), db .prepare( `SELECT user_id AS userId, id AS storageId @@ -251,9 +251,9 @@ export async function listPlatformStorageInventory( .all<{ userId: string; storageId: string }>(), db .prepare( - `SELECT DISTINCT user_id AS userId, package_id AS packageId, name AS serviceName - FROM package_runtime_runs - WHERE surface = 'service' AND name IS NOT NULL`, + `SELECT DISTINCT user_id AS userId, package_id AS packageId, + service_name AS serviceName + FROM package_service_states`, ) .all<{ userId: string @@ -272,7 +272,7 @@ export async function listPlatformStorageInventory( } for (const row of jobRows.results ?? []) push(row.userId, row.storageId) for (const row of archivedRows.results ?? []) push(row.userId, row.storageId) - for (const row of runtimeRows.results ?? []) push(row.userId, row.storageId) + for (const row of registeredBuckets) push(row.userId, row.storageId) for (const row of packageRows.results ?? []) push(row.userId, row.storageId) for (const row of serviceRows.results ?? []) { push( diff --git a/packages/worker/src/entitlements/test-schema.ts b/packages/worker/src/entitlements/test-schema.ts index 2b6185f289..ea3cf00a86 100644 --- a/packages/worker/src/entitlements/test-schema.ts +++ b/packages/worker/src/entitlements/test-schema.ts @@ -1,3 +1,5 @@ +import { ensureUserStorageBucketsTestSchema } from '#worker/storage-buckets/test-schema.ts' + /** * Non-destructive schema for entitlement primitives in workers-unit tests, * where the D1 database starts empty and each suite provisions the tables it @@ -13,6 +15,9 @@ * stable_user_id TEXT` (SQLite cannot add NOT NULL without a default); callers * must insert concrete ids before relying on the unique index. Adding `plan` * to preexisting tables uses `NOT NULL DEFAULT 'free'`. + * + * Also provisions `user_storage_buckets` because entitlement suites that touch + * StorageRunner writes register ownership through that table. */ export async function ensureEntitlementTestSchema(db: D1Database) { await db @@ -116,4 +121,5 @@ export async function ensureEntitlementTestSchema(db: D1Database) { )`, ) .run() + await ensureUserStorageBucketsTestSchema(db) } diff --git a/packages/worker/src/storage-buckets/migration.node.test.ts b/packages/worker/src/storage-buckets/migration.node.test.ts new file mode 100644 index 0000000000..40ec429732 --- /dev/null +++ b/packages/worker/src/storage-buckets/migration.node.test.ts @@ -0,0 +1,91 @@ +import { readdirSync, readFileSync } from 'node:fs' +import { DatabaseSync } from 'node:sqlite' +import { expect, test } from 'vitest' + +const migrationsDirectory = new URL('../../migrations/', import.meta.url) +const userStorageBucketsMigration = '0097-user-storage-buckets.sql' + +function applyMigrationsBefore(db: DatabaseSync, exclusiveUpperBound: string) { + for (const fileName of readdirSync(migrationsDirectory) + .filter((file) => file.endsWith('.sql') && file < exclusiveUpperBound) + .sort()) { + db.exec(readFileSync(new URL(fileName, migrationsDirectory), 'utf8')) + } +} + +function applyMigration(db: DatabaseSync, fileName: string) { + db.exec(readFileSync(new URL(fileName, migrationsDirectory), 'utf8')) +} + +test('user_storage_buckets migration backfills jobs, apps, and runtime-only buckets', () => { + const db = new DatabaseSync(':memory:') + applyMigrationsBefore(db, userStorageBucketsMigration) + + db.exec(` + INSERT INTO jobs ( + id, user_id, name, source_id, storage_id, schedule_json, timezone, + caller_context_json, created_at, updated_at, next_run_at + ) VALUES ( + 'job-1', 'user-a', 'nightly', 'source-job-1', 'job:job-1', '{}', 'UTC', + '{}', '2026-07-01T00:00:00.000Z', '2026-07-02T00:00:00.000Z', + '2026-07-03T00:00:00.000Z' + ); + + INSERT INTO saved_packages ( + id, user_id, name, kody_id, description, source_id, has_app, + created_at, updated_at + ) VALUES ( + 'pkg-app-1', 'user-a', '@a/app', 'app', 'App package', 'source-app-1', 1, + '2026-07-04T00:00:00.000Z', '2026-07-05T00:00:00.000Z' + ); + + INSERT INTO package_runtime_runs ( + id, user_id, package_id, package_kody_id, surface, status, + started_at, storage_id, created_at, updated_at + ) VALUES ( + 'run-adhoc-1', 'user-a', 'pkg-none', 'none', 'export', 'success', + '2026-07-06T00:00:00.000Z', 'exec:legacy-only', + '2026-07-06T00:00:00.000Z', '2026-07-06T00:01:00.000Z' + ); + `) + + applyMigration(db, userStorageBucketsMigration) + + const rows = db + .prepare( + `SELECT user_id, storage_id, kind, created_at, last_seen_at + FROM user_storage_buckets + ORDER BY storage_id ASC`, + ) + .all() as Array<{ + user_id: string + storage_id: string + kind: string + created_at: string + last_seen_at: string + }> + + expect(rows).toEqual([ + { + user_id: 'user-a', + storage_id: 'exec:legacy-only', + kind: 'unknown', + created_at: '2026-07-06T00:00:00.000Z', + last_seen_at: '2026-07-06T00:01:00.000Z', + }, + { + user_id: 'user-a', + storage_id: 'job:job-1', + kind: 'job', + created_at: '2026-07-01T00:00:00.000Z', + last_seen_at: '2026-07-02T00:00:00.000Z', + }, + { + user_id: 'user-a', + storage_id: 'pkg-app-1', + kind: 'app', + created_at: '2026-07-04T00:00:00.000Z', + last_seen_at: '2026-07-05T00:00:00.000Z', + }, + ]) +}) diff --git a/packages/worker/src/storage-buckets/service.node.test.ts b/packages/worker/src/storage-buckets/service.node.test.ts new file mode 100644 index 0000000000..f886c6a500 --- /dev/null +++ b/packages/worker/src/storage-buckets/service.node.test.ts @@ -0,0 +1,202 @@ +import { expect, test } from 'vitest' +import { consoleWarn } from '#worker/test-support/console-spies.ts' +import { + clearStorageBucketRegistrationDedupeForTests, + flushStorageBucketRegistrationsForTests, + listPlatformStorageBuckets, + listUserStorageBucketIds, + registerStorageBucket, + storageBucketKindFromStorageId, +} from './service.ts' + +function createCountingDb(input?: { failRun?: boolean }) { + let insertCount = 0 + const rows = new Map() + + function listRows(userFilter: string | null) { + const results = [...rows.values()] + .filter((row) => (userFilter ? row.userId === userFilter : true)) + .sort((left, right) => + `${left.userId}\0${left.storageId}`.localeCompare( + `${right.userId}\0${right.storageId}`, + ), + ) + .map((row) => + userFilter + ? ({ storageId: row.storageId } as T) + : ({ + userId: row.userId, + storageId: row.storageId, + } as T), + ) + return { results, meta: { changes: 0 } } + } + + const db = { + prepare(sql: string) { + return { + bind(...params: Array) { + return { + async run() { + if (sql.includes('INSERT INTO user_storage_buckets')) { + insertCount += 1 + if (input?.failRun) { + throw new Error('simulated upsert failure') + } + const userId = String(params[0]) + const storageId = String(params[1]) + rows.set(`${userId}\u0000${storageId}`, { userId, storageId }) + } + return { meta: { changes: 1 } } + }, + async all() { + if (!sql.includes('FROM user_storage_buckets')) { + return { results: [] as Array, meta: { changes: 0 } } + } + const userFilter = + sql.includes('WHERE user_id = ?') && params[0] != null + ? String(params[0]) + : null + return listRows(userFilter) + }, + } + }, + async all() { + if (!sql.includes('FROM user_storage_buckets')) { + return { results: [] as Array, meta: { changes: 0 } } + } + return listRows(null) + }, + } + }, + } as unknown as D1Database + return { + db, + env: { APP_DB: db } as Env, + get insertCount() { + return insertCount + }, + rows, + } +} + +test('storageBucketKindFromStorageId only trusts unambiguous prefixes', () => { + expect(storageBucketKindFromStorageId('job:abc')).toBe('job') + expect(storageBucketKindFromStorageId('exec:abc')).toBe('execute') + expect(storageBucketKindFromStorageId('package:abc')).toBe('unknown') + expect(storageBucketKindFromStorageId('service:pkg:svc')).toBe('unknown') + expect(storageBucketKindFromStorageId('adhoc-bucket')).toBe('unknown') +}) + +test('registerStorageBucket never throws when binding or table writes are missing', async () => { + consoleWarn.mockImplementation(() => {}) + clearStorageBucketRegistrationDedupeForTests() + + expect(() => + registerStorageBucket({ + env: {} as Env, + userId: 'user-a', + storageId: 'bucket-a', + }), + ).not.toThrow() + + expect(() => + registerStorageBucket({ + env: { APP_DB: undefined } as unknown as Env, + userId: 'user-a', + storageId: 'bucket-a', + }), + ).not.toThrow() + + const failing = createCountingDb({ failRun: true }) + expect(() => + registerStorageBucket({ + env: failing.env, + userId: 'user-a', + storageId: 'bucket-a', + kind: 'execute', + }), + ).not.toThrow() + await flushStorageBucketRegistrationsForTests() + expect(consoleWarn).toHaveBeenCalledWith( + 'storage-bucket-register-failed', + expect.any(Error), + ) +}) + +test('registerStorageBucket dedupes to one D1 write per bucket in an isolate', async () => { + clearStorageBucketRegistrationDedupeForTests() + const counting = createCountingDb() + const pending: Array> = [] + for (let index = 0; index < 5; index += 1) { + registerStorageBucket({ + env: counting.env, + userId: 'user-a', + storageId: 'exec:same', + kind: 'execute', + waitUntil: (promise) => { + pending.push(promise) + }, + }) + } + await Promise.all(pending) + expect(counting.insertCount).toBe(1) +}) + +test('listUserStorageBucketIds returns only the calling user buckets', async () => { + clearStorageBucketRegistrationDedupeForTests() + const counting = createCountingDb() + const pending: Array> = [] + const waitUntil = (promise: Promise) => { + pending.push(promise) + } + registerStorageBucket({ + env: counting.env, + userId: 'user-a', + storageId: 'bucket-a', + waitUntil, + }) + registerStorageBucket({ + env: counting.env, + userId: 'user-b', + storageId: 'bucket-b', + waitUntil, + }) + await Promise.all(pending) + + await expect( + listUserStorageBucketIds({ env: counting.env, userId: 'user-a' }), + ).resolves.toEqual(['bucket-a']) + await expect( + listUserStorageBucketIds({ env: counting.env, userId: 'user-b' }), + ).resolves.toEqual(['bucket-b']) +}) + +test('listPlatformStorageBuckets returns every user bucket', async () => { + clearStorageBucketRegistrationDedupeForTests() + const counting = createCountingDb() + const pending: Array> = [] + const waitUntil = (promise: Promise) => { + pending.push(promise) + } + registerStorageBucket({ + env: counting.env, + userId: 'user-a', + storageId: 'bucket-a', + waitUntil, + }) + registerStorageBucket({ + env: counting.env, + userId: 'user-b', + storageId: 'bucket-b', + waitUntil, + }) + await Promise.all(pending) + + await expect( + listPlatformStorageBuckets({ db: counting.db }), + ).resolves.toEqual([ + { userId: 'user-a', storageId: 'bucket-a' }, + { userId: 'user-b', storageId: 'bucket-b' }, + ]) +}) diff --git a/packages/worker/src/storage-buckets/service.ts b/packages/worker/src/storage-buckets/service.ts new file mode 100644 index 0000000000..502b5fbfc2 --- /dev/null +++ b/packages/worker/src/storage-buckets/service.ts @@ -0,0 +1,146 @@ +/** + * Authoritative per-user durable storage bucket ownership. + * + * Cloudflare cannot enumerate Durable Objects by name, so every consumer that + * needs "which StorageRunner buckets does this user own?" must read D1. That + * inventory is state (`user_storage_buckets`), not run history. + * + * Helper contract (same spirit as `recordUsage`): + * - `registerStorageBucket` is synchronous, never throws, and never rejects + * into the caller. Missing binding / userId / storageId is a silent no-op. + * - Failures during the async upsert are logged with + * `console.warn('storage-bucket-register-failed', error)`. + * - Prefer `waitUntil` when an ExecutionContext is available; otherwise the + * upsert is fire-and-forget (`void`). + * - Register only on write-ish StorageRunner access. In-isolate dedupe keeps + * hot buckets from becoming per-event shared D1 writes. + */ + +export type StorageBucketKind = + | 'job' + | 'app' + | 'service' + | 'execute' + | 'unknown' + +const registerUpsertStatement = ` +INSERT INTO user_storage_buckets ( + user_id, storage_id, kind, created_at, last_seen_at +) VALUES (?1, ?2, ?3, ?4, ?4) +ON CONFLICT (user_id, storage_id) DO UPDATE SET + last_seen_at = excluded.last_seen_at +`.trim() + +/** Bound in-isolate dedupe so a long-lived isolate cannot grow without limit. */ +const maxRegisteredBucketKeys = 4_096 +const registeredBucketKeys = new Set() +const pendingRegistrations = new Set>() + +function registrationDedupeKey(userId: string, storageId: string) { + return `${userId}\u0000${storageId}` +} + +function rememberRegistrationKey(key: string) { + if (registeredBucketKeys.has(key)) return false + if (registeredBucketKeys.size >= maxRegisteredBucketKeys) { + registeredBucketKeys.clear() + } + registeredBucketKeys.add(key) + return true +} + +function scheduleRegistration( + work: Promise, + waitUntil?: (promise: Promise) => void, +) { + const tracked = work.finally(() => { + pendingRegistrations.delete(tracked) + }) + pendingRegistrations.add(tracked) + if (waitUntil) { + waitUntil(tracked) + return + } + void tracked +} + +/** Synchronous, never throws, fire-and-forget. No-op without a binding/userId/storageId. */ +export function registerStorageBucket(input: { + env: Env + userId?: string | null + storageId?: string | null + kind?: StorageBucketKind + waitUntil?: (promise: Promise) => void +}): void { + try { + const userId = input.userId?.trim() + const storageId = input.storageId?.trim() + if (!userId || !storageId) return + const db = input.env.APP_DB + if (!db) return + const key = registrationDedupeKey(userId, storageId) + if (!rememberRegistrationKey(key)) return + const kind = input.kind ?? 'unknown' + const seenAt = new Date().toISOString() + scheduleRegistration( + db + .prepare(registerUpsertStatement) + .bind(userId, storageId, kind, seenAt) + .run() + .then(() => undefined) + .catch((error: unknown) => { + registeredBucketKeys.delete(key) + console.warn('storage-bucket-register-failed', error) + }), + input.waitUntil, + ) + } catch (error) { + console.warn('storage-bucket-register-failed', error) + } +} + +export async function listUserStorageBucketIds(input: { + env: Env + userId: string +}): Promise> { + const result = await input.env.APP_DB.prepare( + `SELECT storage_id AS storageId + FROM user_storage_buckets + WHERE user_id = ? + ORDER BY storage_id ASC`, + ) + .bind(input.userId) + .all<{ storageId: string }>() + return (result.results ?? []).map((row) => row.storageId) +} + +export async function listPlatformStorageBuckets(input: { + db: D1Database +}): Promise> { + const result = await input.db + .prepare( + `SELECT user_id AS userId, storage_id AS storageId + FROM user_storage_buckets + ORDER BY user_id ASC, storage_id ASC`, + ) + .all<{ userId: string; storageId: string }>() + return result.results ?? [] +} + +export function storageBucketKindFromStorageId( + storageId: string, +): StorageBucketKind { + if (storageId.startsWith('job:')) return 'job' + if (storageId.startsWith('exec:')) return 'execute' + return 'unknown' +} + +/** Test helper: await fire-and-forget registration upserts scheduled in this isolate. */ +export async function flushStorageBucketRegistrationsForTests(): Promise { + await Promise.all([...pendingRegistrations]) +} + +/** Test helper: clear the in-isolate registration dedupe set. */ +export function clearStorageBucketRegistrationDedupeForTests(): void { + registeredBucketKeys.clear() +} diff --git a/packages/worker/src/storage-buckets/service.workers.test.ts b/packages/worker/src/storage-buckets/service.workers.test.ts new file mode 100644 index 0000000000..ccb8cb4c38 --- /dev/null +++ b/packages/worker/src/storage-buckets/service.workers.test.ts @@ -0,0 +1,86 @@ +import { env } from 'cloudflare:workers' +import { expect, test } from 'vitest' +import { consoleWarn } from '#worker/test-support/console-spies.ts' +import { + clearStorageBucketRegistrationDedupeForTests, + flushStorageBucketRegistrationsForTests, + listPlatformStorageBuckets, + listUserStorageBucketIds, + registerStorageBucket, +} from './service.ts' +import { ensureUserStorageBucketsTestSchema } from './test-schema.ts' + +test('registerStorageBucket upserts and list helpers scope correctly on real D1', async () => { + await ensureUserStorageBucketsTestSchema(env.APP_DB) + clearStorageBucketRegistrationDedupeForTests() + const userA = `usb-a-${crypto.randomUUID()}` + const userB = `usb-b-${crypto.randomUUID()}` + const bucketA = `exec:${crypto.randomUUID()}` + const bucketB = `job:${crypto.randomUUID()}` + const pending: Array> = [] + const waitUntil = (promise: Promise) => { + pending.push(promise) + } + + registerStorageBucket({ + env, + userId: userA, + storageId: bucketA, + kind: 'execute', + waitUntil, + }) + registerStorageBucket({ + env, + userId: userB, + storageId: bucketB, + kind: 'job', + waitUntil, + }) + await Promise.all(pending) + + await expect( + listUserStorageBucketIds({ env, userId: userA }), + ).resolves.toEqual([bucketA]) + await expect( + listUserStorageBucketIds({ env, userId: userB }), + ).resolves.toEqual([bucketB]) + + const platform = await listPlatformStorageBuckets({ db: env.APP_DB }) + expect(platform).toEqual( + expect.arrayContaining([ + { userId: userA, storageId: bucketA }, + { userId: userB, storageId: bucketB }, + ]), + ) +}) + +test('registerStorageBucket never throws when the table is missing', async () => { + consoleWarn.mockImplementation(() => {}) + clearStorageBucketRegistrationDedupeForTests() + const missingTableDb = { + prepare() { + return { + bind() { + return { + async run() { + throw new Error('no such table: user_storage_buckets') + }, + } + }, + } + }, + } as unknown as D1Database + + expect(() => + registerStorageBucket({ + env: { APP_DB: missingTableDb } as Env, + userId: 'user-missing-table', + storageId: 'bucket-missing-table', + }), + ).not.toThrow() + await flushStorageBucketRegistrationsForTests() + expect(consoleWarn).toHaveBeenCalledWith( + 'storage-bucket-register-failed', + expect.any(Error), + ) +}) diff --git a/packages/worker/src/storage-buckets/test-schema.ts b/packages/worker/src/storage-buckets/test-schema.ts new file mode 100644 index 0000000000..084a34ede2 --- /dev/null +++ b/packages/worker/src/storage-buckets/test-schema.ts @@ -0,0 +1,25 @@ +/** + * Non-destructive schema for `user_storage_buckets` in workers-unit tests, + * where the D1 database starts empty and each suite provisions the tables it + * needs. Mirrors migration 0097. + */ +export async function ensureUserStorageBucketsTestSchema(db: D1Database) { + await db + .prepare( + `CREATE TABLE IF NOT EXISTS user_storage_buckets ( + user_id TEXT NOT NULL, + storage_id TEXT NOT NULL, + kind TEXT NOT NULL CHECK (kind IN ('job', 'app', 'service', 'execute', 'unknown')), + created_at TEXT NOT NULL, + last_seen_at TEXT NOT NULL, + PRIMARY KEY (user_id, storage_id) +)`, + ) + .run() + await db + .prepare( + `CREATE INDEX IF NOT EXISTS idx_user_storage_buckets_user + ON user_storage_buckets(user_id)`, + ) + .run() +} diff --git a/packages/worker/src/storage-runner.ts b/packages/worker/src/storage-runner.ts index a472476b31..6a409d5fbf 100644 --- a/packages/worker/src/storage-runner.ts +++ b/packages/worker/src/storage-runner.ts @@ -7,6 +7,10 @@ import { estimateEntitlementStorageSqlWriteBytes, readUserD1StorageBytes, } from '#worker/entitlements/service.ts' +import { + registerStorageBucket, + storageBucketKindFromStorageId, +} from '#worker/storage-buckets/service.ts' import { storageRunnerDurableObjectName } from '#worker/user-scoped-durable-object-name.ts' const defaultStorageExportPageSize = 250 @@ -466,7 +470,7 @@ export function storageRunnerRpc(input: { userId: string storageId: string }) { - return input.env.STORAGE_RUNNER.get( + const runner = input.env.STORAGE_RUNNER.get( input.env.STORAGE_RUNNER.idFromName( storageRunnerDurableObjectName(input.userId, input.storageId), ), @@ -502,6 +506,62 @@ export function storageRunnerRpc(input: { writable?: boolean }) => Promise } + + // Registration must never run on a path that executes after the owning + // user's D1 rows are removed. Account deletion clears StorageRunner DOs + // via clearStorage, then deletes user_storage_buckets; registering on + // clear would fire-and-forget an upsert that can recreate rows for a + // deleted user. Clearing also is not evidence of use — prior writes + // already registered the bucket. + const registerOwnedBucket = () => { + registerStorageBucket({ + env: input.env, + userId: input.userId, + storageId: input.storageId, + kind: storageBucketKindFromStorageId(input.storageId), + }) + } + + return { + getValue: (payload: { key: string }) => runner.getValue(payload), + setValue: (payload: { key: string; value: unknown }) => { + registerOwnedBucket() + return runner.setValue(payload) + }, + deleteValue: (payload: { key: string }) => { + registerOwnedBucket() + return runner.deleteValue(payload) + }, + clearStorage: () => runner.clearStorage(), + getEstimatedBytes: () => runner.getEstimatedBytes(), + listValues: (payload: { + prefix?: string | null + pageSize?: number + startAfter?: string | null + }) => runner.listValues(payload), + exportStorage: (payload: { + pageSize?: number + startAfter?: string | null + }) => runner.exportStorage(payload), + importStorage: (payload: { + mode: 'replace' + replacePage: 'first' | 'continue' + entries: Array<{ key: string; valueJson: string }> + }) => { + registerOwnedBucket() + return runner.importStorage(payload) + }, + sqlQuery: (payload: { + query: string + params?: Array + writable?: boolean + }) => { + if (payload.writable) { + registerOwnedBucket() + } + return runner.sqlQuery(payload) + }, + } } export async function assertStorageRunnerWriteWithinEntitlement(input: { diff --git a/packages/worker/src/storage-runner.workers.test.ts b/packages/worker/src/storage-runner.workers.test.ts index 716b4f2e93..54129a6eb1 100644 --- a/packages/worker/src/storage-runner.workers.test.ts +++ b/packages/worker/src/storage-runner.workers.test.ts @@ -4,6 +4,12 @@ import { expect, test } from 'vitest' import { EntitlementLimitError } from '#worker/entitlements/errors.ts' import { planLimits } from '#worker/entitlements/plans.ts' import { ensureEntitlementTestSchema } from '#worker/entitlements/test-schema.ts' +import { + clearStorageBucketRegistrationDedupeForTests, + flushStorageBucketRegistrationsForTests, + listUserStorageBucketIds, +} from '#worker/storage-buckets/service.ts' +import { ensureUserStorageBucketsTestSchema } from '#worker/storage-buckets/test-schema.ts' import { createStableUserIdFromEmail } from '#worker/user-id.ts' import { createStorageKodyTools, @@ -12,6 +18,11 @@ import { storageRunnerRpc, } from './storage-runner.ts' +async function ensureStorageRunnerTestSchema() { + await ensureUserStorageBucketsTestSchema(env.APP_DB) + clearStorageBucketRegistrationDedupeForTests() +} + async function ensureStorageBytesEmailTestSchema() { await ensureEntitlementTestSchema(env.APP_DB) await env.APP_DB.prepare( @@ -83,6 +94,7 @@ async function seedPlannedStorageUser(input: { } test('storage runner preserves isolated state per storage id', async () => { + await ensureStorageRunnerTestSchema() const storageIdA = createExecuteStorageId() const storageIdB = createExecuteStorageId() const runnerA = storageRunnerRpc({ @@ -160,6 +172,7 @@ test('storage runner preserves isolated state per storage id', async () => { test('storage runner write tools enforce storage byte entitlements for planned users', async () => { await ensureStorageBytesEmailTestSchema() + clearStorageBucketRegistrationDedupeForTests() const limit = planLimits.pro.maxStorageBytes if (limit === null) throw new Error('Expected a numeric pro storage cap.') const plannedEmail = `storage-planned-${crypto.randomUUID()}@example.com` @@ -226,6 +239,7 @@ test('storage runner write tools enforce storage byte entitlements for planned u }) test('storage runner supports raw SQL with explicit writable access', async () => { + await ensureStorageRunnerTestSchema() const storageId = createExecuteStorageId() const runner = storageRunnerRpc({ env, @@ -273,6 +287,7 @@ test('storage runner supports raw SQL with explicit writable access', async () = }) test('storage runner enforces read-only SQL policy for mutations, multi-statement queries, and literal semicolons', async () => { + await ensureStorageRunnerTestSchema() const storageId = createExecuteStorageId() const runner = storageRunnerRpc({ env, @@ -333,3 +348,133 @@ test('storage runner enforces read-only SQL policy for mutations, multi-statemen rowsWritten: 0, }) }) + +test('storage runner registers buckets on writes but not on reads', async () => { + await ensureStorageRunnerTestSchema() + const userId = `storage-register-${crypto.randomUUID()}` + const writeStorageId = createExecuteStorageId() + const readStorageId = createExecuteStorageId() + const writer = storageRunnerRpc({ + env, + userId, + storageId: writeStorageId, + }) + const reader = storageRunnerRpc({ + env, + userId, + storageId: readStorageId, + }) + + await reader.getValue({ key: 'missing' }) + await reader.listValues({ pageSize: 10 }) + await reader.exportStorage({ pageSize: 10 }) + await reader.sqlQuery({ + query: 'select 1 as ok', + writable: false, + }) + await flushStorageBucketRegistrationsForTests() + await expect(listUserStorageBucketIds({ env, userId })).resolves.toEqual([]) + + await writer.setValue({ key: 'counter', value: 1 }) + await flushStorageBucketRegistrationsForTests() + await expect(listUserStorageBucketIds({ env, userId })).resolves.toEqual([ + writeStorageId, + ]) +}) + +test('storage runner dedupes bucket registration to one D1 write per isolate', async () => { + await ensureStorageRunnerTestSchema() + const userId = `storage-dedupe-${crypto.randomUUID()}` + const storageId = createExecuteStorageId() + let insertCount = 0 + const originalPrepare = env.APP_DB.prepare.bind(env.APP_DB) + env.APP_DB.prepare = ((sql: string) => { + if (sql.includes('INSERT INTO user_storage_buckets')) { + insertCount += 1 + } + return originalPrepare(sql) + }) as typeof env.APP_DB.prepare + + try { + const runner = storageRunnerRpc({ + env, + userId, + storageId, + }) + await runner.setValue({ key: 'a', value: 1 }) + await runner.setValue({ key: 'b', value: 2 }) + await runner.deleteValue({ key: 'a' }) + await runner.sqlQuery({ + query: 'create table if not exists t (id integer primary key)', + writable: true, + }) + await flushStorageBucketRegistrationsForTests() + expect(insertCount).toBe(1) + await expect(listUserStorageBucketIds({ env, userId })).resolves.toEqual([ + storageId, + ]) + } finally { + env.APP_DB.prepare = originalPrepare + } +}) + +test('clearStorage during account-deletion purge must not recreate ownership rows', async () => { + await ensureStorageRunnerTestSchema() + const userId = `storage-delete-race-${crypto.randomUUID()}` + const storageId = createExecuteStorageId() + const runner = storageRunnerRpc({ + env, + userId, + storageId, + }) + + await runner.setValue({ key: 'keep-until-purge', value: 1 }) + await flushStorageBucketRegistrationsForTests() + await expect(listUserStorageBucketIds({ env, userId })).resolves.toEqual([ + storageId, + ]) + + // Account deletion often runs in a fresh isolate, so in-memory dedupe + // cannot paper over a clearStorage registration race. + clearStorageBucketRegistrationDedupeForTests() + + // Hold any ownership upsert until after the D1 delete so a fire-and-forget + // clearStorage registration cannot win the race before the assertion. + let releaseInsert: (() => void) | null = null + const insertGate = new Promise((resolve) => { + releaseInsert = resolve + }) + const originalPrepare = env.APP_DB.prepare.bind(env.APP_DB) + env.APP_DB.prepare = ((sql: string) => { + const statement = originalPrepare(sql) + if (!sql.includes('INSERT INTO user_storage_buckets')) { + return statement + } + return { + bind(...params: Array) { + const bound = statement.bind(...params) + return { + async run() { + await insertGate + return await bound.run() + }, + } + }, + } + }) as typeof env.APP_DB.prepare + + try { + await runner.clearStorage() + await env.APP_DB.prepare( + `DELETE FROM user_storage_buckets WHERE user_id = ?`, + ) + .bind(userId) + .run() + releaseInsert?.() + await flushStorageBucketRegistrationsForTests() + await expect(listUserStorageBucketIds({ env, userId })).resolves.toEqual([]) + } finally { + env.APP_DB.prepare = originalPrepare + releaseInsert?.() + } +}) diff --git a/tools/migration-ledger.json b/tools/migration-ledger.json index 8afa743170..61efb21f87 100644 --- a/tools/migration-ledger.json +++ b/tools/migration-ledger.json @@ -411,6 +411,10 @@ { "filename": "0096-user-package-run-successes.sql", "sha256": "bec0c57fc3254eb17db094499095009985df84e7c309adbebc70704df8928daf" + }, + { + "filename": "0097-user-storage-buckets.sql", + "sha256": "280af0890f1e517830535f9379d1d90b5bc05b9da5b1d8a5c6b9423ca3da842b" } ] }