import { describe, expect, test } from 'bun:test'; import { makeFunctionReference } from 'convex/server'; import { convexTest } from 'convex-test'; import schema from '../src/convex/schema'; import { EsiRateLimitError } from '../src/lib/server/esi'; import { createReferenceEnrichmentPin, deterministicEnrichmentVersion } from '../src/lib/domain/inventory-location'; import { InventoryLocationService } from '../src/lib/server/inventory/location-service'; import { processSyncJob } from '../workers/sync'; process.env.CONVEX_BRIDGE_SECRET = 'test-bridge-secret'; process.env.CONVEX_SYNC_WORKER_SECRET = 'test-worker-secret'; process.env.APP_CAPABILITY_ALLOWLIST = 'asset_search,asset_valuation,asset_sync,assistant_dispatch,session_token,identity,subscriptions'; const modules = { 'src/convex/functions.ts': () => import('../src/convex/functions'), 'src/convex/status.ts': () => import('../src/convex/status'), 'src/convex/_generated/server.ts': async () => ({}), }; const mutation = (name: string) => makeFunctionReference<'mutation', any, any>(`functions:${name}`); const query = (name: string) => makeFunctionReference<'query', any, any>(`functions:${name}`); const bridge = { bridgeSecret: 'test-bridge-secret' }; const worker = { workerSecret: 'test-worker-secret' }; async function login(t: ReturnType, input: Record = {}) { return await t.mutation(mutation('bridgeCompleteLogin'), { ...bridge, characterId: 700, name: 'Queue Pilot', ownerHash: 'queue-owner', scopes: ['esi-assets.read_assets.v1'], intent: 'initial_login', capability: 'asset_search', requestedScopes: ['esi-assets.read_assets.v1'], approvedExistingScopes: [], encryptedTokens: 'encrypted-token', tokenExpiresAt: Date.now() + 60_000, sessionHash: 'queue-session', sessionExpiresAt: Date.now() + 60_000, ...input, }); } async function enqueue(t: ReturnType, account: any, characterId = 700, sessionHash = 'queue-session') { return await t.mutation(mutation('bridgeEnqueueAssetSync'), { ...bridge, userId: account.userId, characterId, initiatingSessionHash: sessionHash }); } async function cloneCharacters(t: ReturnType, userId: string, sourceCharacterId: number, characterIds: number[]) { await t.run(async (ctx: any) => { const source = await ctx.db.query('characters').withIndex('by_user_character', (q: any) => q.eq('userId', userId).eq('characterId', sourceCharacterId)).first(); const token = await ctx.db.query('tokenRecords').withIndex('by_user_character', (q: any) => q.eq('userId', userId).eq('characterId', sourceCharacterId)).first(); const { _id: _characterId, _creationTime: _characterCreated, ...characterFields } = source; const { _id: _tokenId, _creationTime: _tokenCreated, ...tokenFields } = token; for (const characterId of characterIds) { await ctx.db.insert('characters', { ...characterFields, characterId, name: `Clone ${characterId}` }); await ctx.db.insert('tokenRecords', { ...tokenFields, characterId }); } }); } describe('T07 durable asset scheduling', () => { test('derives authority, isolates tenants, and deduplicates one active job', async () => { const t = convexTest({ schema, modules }); const first = await login(t); const duplicate = await enqueue(t, first); const same = await enqueue(t, first); expect(same.id).toBe(duplicate.id); const other = await login(t, { characterId: 701, sessionHash: 'other-session', ownerHash: 'other-owner' }); const otherJob = await enqueue(t, other, 701, 'other-session'); expect(otherJob.id).not.toBe(duplicate.id); await expect(t.query(query('bridgeSyncJob'), { ...bridge, userId: other.userId, jobId: duplicate.id })).resolves.toBeNull(); await expect(t.mutation(mutation('bridgeEnqueueAssetSync'), { ...bridge, userId: other.userId, characterId: 700, initiatingSessionHash: 'other-session' })).rejects.toThrow('active'); }); test('claims only user asset jobs, reclaims an expired lease, and keeps unsupported jobs untouched', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 708, sessionHash: 'claim-filter-session' }); const job = await enqueue(t, account, 708, 'claim-filter-session'); await t.run(async (ctx: any) => { await ctx.db.insert('jobs', { userId: account.userId, kind: 'asset_sync', dataset: 'assets', authority: 'system', ownerKind: 'system', status: 'queued', createdAt: Date.now(), updatedAt: Date.now(), dueAt: Date.now(), attempts: 0, maxAttempts: 5, priority: 0 }); await ctx.db.insert('jobs', { userId: account.userId, kind: 'reference_sync', dataset: 'reference', authority: 'system', ownerKind: 'system', status: 'queued', createdAt: Date.now(), updatedAt: Date.now(), dueAt: Date.now(), attempts: 0, maxAttempts: 5, priority: 0 }); }); const claimed = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-filter', leaseMs: 6 * 60_000 }); expect(claimed.jobId).toBe(job.id); await t.run(async (ctx: any) => { const row = await ctx.db.get(claimed.jobId); await ctx.db.patch(row._id, { leaseExpiresAt: 0 }); }); const reclaimed = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-reclaim', leaseMs: 6 * 60_000 }); expect(reclaimed.jobId).toBe(job.id); expect(reclaimed.attempts).toBe(2); }); test('claims, renews beyond the former five-minute lease, and cancellation fences delivery', async () => { const t = convexTest({ schema, modules }); const account = await login(t); const job = await enqueue(t, account); const claimed = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-a', leaseMs: 6 * 60_000 }); expect(claimed).toMatchObject({ jobId: job.id, status: 'running', attempts: 1 }); await t.run(async (ctx: any) => await ctx.db.patch(claimed.jobId, { leaseExpiresAt: Date.now() + 301_000 })); await expect(t.mutation(mutation('bridgeRenewSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, workerId: 'worker-a', leaseMs: 6 * 60_000 })).resolves.toBe(true); await expect(t.mutation(mutation('bridgeCancelSyncJob'), { ...bridge, userId: account.userId, jobId: claimed.jobId, initiatingSessionHash: 'queue-session' })).resolves.toMatchObject({ status: 'cancelled' }); await expect(t.mutation(mutation('bridgeRenewSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, workerId: 'worker-a', leaseMs: 6 * 60_000 })).resolves.toBe(false); }); test('enforces the per-user queue cap and fair claims across users', async () => { const t = convexTest({ schema, modules }); const first = await login(t, { characterId: 709, sessionHash: 'cap-session' }); await cloneCharacters(t, first.userId, 709, [710, 711, 712, 713, 714, 715, 716, 717, 718]); for (const characterId of [709, 710, 711, 712, 713, 714, 715, 716]) await enqueue(t, first, characterId, 'cap-session'); await expect(enqueue(t, first, 717, 'cap-session')).rejects.toThrow('queue is full'); const second = await login(t, { characterId: 719, sessionHash: 'fair-second-session', ownerHash: 'fair-second-owner' }); await enqueue(t, second, 719, 'fair-second-session'); const claimedFirst = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-fair-a' }); const claimedSecond = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-fair-b' }); expect(claimedFirst.userId).toBe(first.userId); expect(claimedSecond.userId).toBe(second.userId); }); test('shared provider cooldown blocks a different character until its eligible time', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 720, sessionHash: 'cooldown-session' }); await cloneCharacters(t, account.userId, 720, [721]); const first = await enqueue(t, account, 720, 'cooldown-session'); const claimed = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-cooldown-a' }); const until = Date.now() + 30_000; await t.mutation(mutation('bridgeDeferSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, dueAt: until, cooldownContext: `character:${account.userId}:720`, cooldownUntil: until, cooldownGroup: 'esi:assets' }); await enqueue(t, account, 721, 'cooldown-session'); await expect(t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-cooldown-b' })).resolves.toBeNull(); await t.run(async (ctx: any) => { const cooldown = await ctx.db.query('syncCooldowns').withIndex('by_context', (q: any) => q.eq('context', 'group:esi:assets')).first(); if (cooldown) await ctx.db.patch(cooldown._id, { until: 0 }); const jobs = await ctx.db.query('jobs').withIndex('by_user_status', (q: any) => q.eq('userId', account.userId).eq('status', 'queued')).collect(); for (const job of jobs) await ctx.db.patch(job._id, { dueAt: 0 }); }); await expect(t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-cooldown-c' })).resolves.toMatchObject({ userId: account.userId }); expect(first.status).toBe('queued'); }); test('stages idempotently and fences stale success and failure after credential rotation', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 703, sessionHash: 'publish-session' }); const queued = await enqueue(t, account, 703, 'publish-session'); const claimed = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-publish' }); const generation = await t.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken }); const row = { itemId: '9007199254740993', characterId: 703, typeId: 34, quantity: 1, locationId: '60003760', locationType: 'station', locationFlag: 'Hangar', isSingleton: true }; await t.mutation(mutation('bridgeStageSyncAssets'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, index: 0, assets: [row] }); await t.mutation(mutation('bridgeStageSyncAssets'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, index: 0, assets: [row] }); await t.mutation(mutation('bridgeSaveTokens'), { ...bridge, userId: account.userId, characterId: 703, encryptedTokens: 'rotated', tokenExpiresAt: Date.now() + 60_000, scopes: ['esi-assets.read_assets.v1'], expectedVersion: 1 }); const rotatedCharacters = await t.query(query('bridgeCharacters'), { ...bridge, userId: account.userId }); const rotated = rotatedCharacters.find((entry: any) => entry.characterId === 703); expect(rotated.syncGeneration).toBeUndefined(); expect(rotated.datasetStatuses.find((entry: any) => entry.dataset === 'assets')).toMatchObject({ status: 'blocked' }); await expect(t.mutation(mutation('bridgeCommitSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, chunkCount: 1, syncedAt: Date.now(), expiresAt: Date.now() + 60_000 })).rejects.toThrow(); await expect(t.mutation(mutation('bridgeFailSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken })).rejects.toThrow(); const page = await t.query(query('bridgeAssetsPage'), { ...bridge, userId: account.userId, characterId: 703, paginationOpts: { numItems: 10, cursor: null } }); expect(page.generation).toBeNull(); expect(generation).toEqual(expect.any(String)); expect(queued.status).toBe('queued'); }); test('duplicate commit delivery is idempotent and cancellation fences stale success and failure', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 722, sessionHash: 'duplicate-session' }); await enqueue(t, account, 722, 'duplicate-session'); const claimed = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-duplicate' }); await t.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken }); await t.mutation(mutation('bridgeCommitSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, chunkCount: 0, syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000 }); await expect(t.mutation(mutation('bridgeCommitSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, chunkCount: 0, syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000 })).resolves.toBe(true); const second = await login(t, { characterId: 723, sessionHash: 'cancel-fence-session' }); await enqueue(t, second, 723, 'cancel-fence-session'); const cancelled = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-cancel-fence' }); await t.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: cancelled.jobId, leaseToken: cancelled.leaseToken }); await t.mutation(mutation('bridgeCancelSyncJob'), { ...bridge, userId: second.userId, jobId: cancelled.jobId, initiatingSessionHash: 'cancel-fence-session' }); await expect(t.mutation(mutation('bridgeCommitSyncJob'), { ...worker, jobId: cancelled.jobId, leaseToken: cancelled.leaseToken, chunkCount: 0, syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000 })).rejects.toThrow(); await expect(t.mutation(mutation('bridgeFailSyncJob'), { ...worker, jobId: cancelled.jobId, leaseToken: cancelled.leaseToken })).rejects.toThrow(); }); test('stages private enrichment with the exact generation and exposes an atomic read pin', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 726, sessionHash: 'enrichment-session' }); await enqueue(t, account, 726, 'enrichment-session'); const claimed = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-enrichment' }); const generation = await t.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken }); const asset = { itemId: '1', characterId: 726, typeId: 34, quantity: 1, locationId: '60003760', locationType: 'station', locationFlag: 'Hangar' }; await t.mutation(mutation('bridgeStageSyncAssets'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, index: 0, assets: [asset] }); const row = { ...asset, parentItemId: null, root: { kind: 'external', id: '60003760', locationType: 'station', locationFlag: 'Hangar' }, depth: 0, breadcrumbs: [], fitted: { classification: 'unknown', rawFlag: 'Hangar', reason: 'ambiguous_special_bay' }, fittedClassification: 'unknown', isFitted: null, unresolved: [], customName: null, nameStatus: 'not_requested', position: null, locationStatus: 'not_requested', location: { kind: 'station', id: '60003760', name: 'Jita', parentId: null, evidence: 'system_listed', sourceVerified: false, referenceVersion: 'ref-v1', dockingPermission: 'unknown', system: null, region: null }, }; await t.mutation(mutation('bridgeStageSyncEnrichment'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, index: 0, rows: [row] }); const inputPin = createReferenceEnrichmentPin({ assetGeneration: generation, referenceVersion: 'ref-v1', authority: { userId: account.userId, principal: 'queue-owner', characterId: 726, scopes: ['esi-assets.read_assets.v1'], grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1 }, referenceFacts: [{ kind: 'station', id: '60003760', name: 'Jita', parentId: null, evidence: 'system_listed', sourceVerified: false, systemId: null, regionId: null, referenceVersion: 'ref-v1' }] }); await t.mutation(mutation('bridgeCommitSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, chunkCount: 1, syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000, enrichmentStatus: 'partial', enrichmentVersion: deterministicEnrichmentVersion(inputPin), enrichmentInputPin: inputPin, enrichmentChunkCount: 1, enrichmentUnresolvedCount: 0, }); const page = await t.query(query('bridgeAssetsPage'), { ...bridge, userId: account.userId, characterId: 726, paginationOpts: { numItems: 10, cursor: null } }); expect(page).toMatchObject({ generation, enrichmentStatus: 'partial', enrichmentVersion: deterministicEnrichmentVersion(inputPin) }); const manifest = await t.query(query('bridgeEnrichmentManifest'), { ...bridge, userId: account.userId, characterId: 726, expectedGeneration: generation }); expect(manifest).toMatchObject({ generation, status: 'partial', version: deterministicEnrichmentVersion(inputPin), unresolvedCount: 0 }); const chunks = await t.query(query('bridgeEnrichmentChunksPage'), { ...bridge, userId: account.userId, characterId: 726, expectedGeneration: generation, paginationOpts: { numItems: 10, cursor: null } }); expect(chunks.chunks[0]?.rows[0]).toMatchObject({ itemId: '1', location: { dockingPermission: 'unknown' } }); await expect(t.query(query('bridgeEnrichmentManifest'), { ...bridge, userId: 'other-user', characterId: 726, expectedGeneration: generation })).resolves.toBeNull(); }); test('enrichment first page reports the active generation after a rollover', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 728, sessionHash: 'enrichment-rollover-session' }); await enqueue(t, account, 728, 'enrichment-rollover-session'); const first = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'enrichment-rollover-worker-1' }); const generationOne = await t.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: first.jobId, leaseToken: first.leaseToken }); await t.mutation(mutation('bridgeCommitSyncJob'), { ...worker, jobId: first.jobId, leaseToken: first.leaseToken, chunkCount: 0, syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000 }); await enqueue(t, account, 728, 'enrichment-rollover-session'); const second = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'enrichment-rollover-worker-2' }); const generationTwo = await t.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: second.jobId, leaseToken: second.leaseToken }); await t.mutation(mutation('bridgeCommitSyncJob'), { ...worker, jobId: second.jobId, leaseToken: second.leaseToken, chunkCount: 0, syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000 }); expect(generationTwo).not.toBe(generationOne); const page = await t.query(query('bridgeEnrichmentChunksPage'), { ...bridge, userId: account.userId, characterId: 728, expectedGeneration: generationOne, paginationOpts: { numItems: 10, cursor: null } }); expect(page).toMatchObject({ chunks: [], generation: generationTwo, isDone: true }); }); test('commit rejects forged public metadata even when the root ID is correct', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 727, sessionHash: 'forged-reference-facts-session' }); await enqueue(t, account, 727, 'forged-reference-facts-session'); const claimed = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'forged-reference-facts-worker' }); const generation = await t.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken }); const asset = { itemId: '2', characterId: 727, typeId: 34, quantity: 1, locationId: '60003760', locationType: 'station', locationFlag: 'Hangar' }; await t.mutation(mutation('bridgeStageSyncAssets'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, index: 0, assets: [asset] }); const row = { ...asset, parentItemId: null, root: { kind: 'external', id: '60003760', locationType: 'station', locationFlag: 'Hangar' }, depth: 0, breadcrumbs: [], fitted: { classification: 'unknown', rawFlag: 'Hangar', reason: 'ambiguous_special_bay' }, fittedClassification: 'unknown', isFitted: null, unresolved: [], customName: null, nameStatus: 'not_requested', position: null, locationStatus: 'not_requested', location: { kind: 'station', id: '60003760', name: 'Forged', parentId: null, evidence: 'system_listed', sourceVerified: false, referenceVersion: 'ref-v1', dockingPermission: 'unknown', system: null, region: null } }; await t.mutation(mutation('bridgeStageSyncEnrichment'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, index: 0, rows: [row] }); const inputPin = createReferenceEnrichmentPin({ assetGeneration: generation, referenceVersion: 'ref-v1', authority: { userId: account.userId, principal: 'queue-owner', characterId: 727, scopes: ['esi-assets.read_assets.v1'], grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1 }, referenceFacts: [{ kind: 'station', id: '60003760', name: 'Jita', parentId: null, evidence: 'system_listed', sourceVerified: false, systemId: null, regionId: null, referenceVersion: 'ref-v1' }] }); await expect(t.mutation(mutation('bridgeCommitSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, chunkCount: 1, syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000, enrichmentStatus: 'partial', enrichmentVersion: deterministicEnrichmentVersion(inputPin), enrichmentInputPin: inputPin, enrichmentChunkCount: 1, enrichmentUnresolvedCount: 0 })).rejects.toThrow('Public location facts'); }); test('private structure rows must match pinned source facts at commit', async () => { const t = convexTest({ schema, modules }); const priorAllowlist = process.env.APP_CAPABILITY_ALLOWLIST; process.env.APP_CAPABILITY_ALLOWLIST = `${priorAllowlist ?? ''},inventory_locations`; const account = await login(t, { characterId: 731, sessionHash: 'private-facts-session', capability: 'inventory_locations', scopes: ['esi-assets.read_assets.v1', 'esi-universe.read_structures.v1'], requestedScopes: ['esi-assets.read_assets.v1', 'esi-universe.read_structures.v1'] }); process.env.APP_CAPABILITY_ALLOWLIST = priorAllowlist; await enqueue(t, account, 731, 'private-facts-session'); const claimed = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'private-facts-worker' }); const generation = await t.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken }); const asset = { itemId: '8000000000000001', characterId: 731, typeId: 34, quantity: 1, locationId: '8000000000000001', locationType: 'other', locationFlag: 'Hangar' }; await t.mutation(mutation('bridgeStageSyncAssets'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, index: 0, assets: [asset] }); const service = new InventoryLocationService({ reference: { pinnedVersion: 'ref-v1', pin() { return this; }, close() {}, async types() { return []; }, async systems() { return []; }, async regions() { return []; }, async stations() { return []; }, async constellations() { return []; }, async searchTypes() { return []; }, async searchTypesByName() { return []; } } as any, structures: { async access() { return { structureId: asset.locationId, characterId: 731, status: 'accessible', evidence: 'esi_structure_access', name: 'TRUSTED PRIVATE', observedAt: 100, expiresAt: 1_000, dockingPermission: 'unknown' }; } }, now: () => 200 }); const authority = { userId: account.userId, principal: 'queue-owner', characterId: 731, scopes: ['esi-assets.read_assets.v1', 'esi-universe.read_structures.v1'], grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1 }; const enrichment = await service.enrich({ assets: [asset], authority, assetGeneration: generation }); const forged = structuredClone(enrichment.rows) as any[]; forged[0] = { ...forged[0], location: { ...forged[0].location, name: 'FORGED PRIVATE' } }; await t.mutation(mutation('bridgeStageSyncEnrichment'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, index: 0, rows: forged }); await expect(t.mutation(mutation('bridgeCommitSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, chunkCount: 1, syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000, enrichmentStatus: 'partial', enrichmentVersion: enrichment.enrichmentVersion, enrichmentInputPin: enrichment.inputPin, enrichmentChunkCount: 1, enrichmentUnresolvedCount: enrichment.unresolved.length })).rejects.toThrow('Structure row facts'); }); test('service output with forbidden structure evidence stages and commits as explicit partial enrichment', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 729, sessionHash: 'forbidden-persistence-session' }); await enqueue(t, account, 729, 'forbidden-persistence-session'); const claimed = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'forbidden-persistence-worker' }); const generation = await t.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken }); const asset = { itemId: '8000000000000001', characterId: 729, typeId: 34, quantity: 1, locationId: '8000000000000001', locationType: 'other', locationFlag: 'Hangar' }; await t.mutation(mutation('bridgeStageSyncAssets'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, index: 0, assets: [asset] }); const service = new InventoryLocationService({ reference: { pinnedVersion: 'ref-v1', pin() { return this; }, close() {}, async types() { return []; }, async systems() { return []; }, async regions() { return []; }, async stations() { return []; }, async constellations() { return []; }, async searchTypes() { return []; }, async searchTypesByName() { return []; } } as any, structures: { async access() { return { structureId: asset.locationId, characterId: 729, status: 'forbidden', evidence: 'esi_forbidden', name: null, observedAt: 100, expiresAt: 1_000, dockingPermission: 'unknown' }; } }, now: () => 200, }); const enrichment = await service.enrich({ assets: [asset], authority: { userId: account.userId, principal: 'queue-owner', characterId: 729, scopes: ['esi-assets.read_assets.v1'], grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1 }, assetGeneration: generation }); expect(enrichment.rows[0]?.location).toMatchObject({ kind: 'structure', evidence: 'unknown', name: null }); expect(enrichment.rows[0]?.unresolved).toContain('forbidden_structure'); await t.mutation(mutation('bridgeStageSyncEnrichment'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, index: 0, rows: enrichment.rows }); await t.mutation(mutation('bridgeCommitSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, chunkCount: 1, syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000, enrichmentStatus: enrichment.status, enrichmentVersion: enrichment.enrichmentVersion, enrichmentInputPin: enrichment.inputPin, enrichmentChunkCount: 1, enrichmentUnresolvedCount: enrichment.unresolved.length }); const chunks = await t.query(query('bridgeEnrichmentChunksPage'), { ...bridge, userId: account.userId, characterId: 729, expectedGeneration: generation, paginationOpts: { numItems: 10, cursor: null } }); expect(chunks.chunks[0]?.rows[0]).toMatchObject({ location: { evidence: 'unknown', name: null }, unresolved: ['forbidden_structure'] }); }); test('indexed persistence bounds a 2.5k raw generation without repeated full scans', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 750, sessionHash: 'large-persistence-session' }); await enqueue(t, account, 750, 'large-persistence-session'); const claimed = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'large-persistence-worker' }); await t.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken }); for (let index = 0; index < 10; index += 1) { const assets = Array.from({ length: 250 }, (_, offset) => ({ itemId: String(index * 250 + offset + 1), characterId: 750, typeId: 34, quantity: 1, locationId: '60003760', locationType: 'station', locationFlag: 'Hangar' })); await t.mutation(mutation('bridgeStageSyncAssets'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, index, assets }); } await t.mutation(mutation('bridgeCommitSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, chunkCount: 10, syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000, enrichmentStatus: 'failed', enrichmentChunkCount: 0, enrichmentUnresolvedCount: 2_500, enrichmentError: 'bounded test fallback' }); const page = await t.query(query('bridgeAssetsPage'), { ...bridge, userId: account.userId, characterId: 750, paginationOpts: { numItems: 1, cursor: null } }); expect(page.page).toHaveLength(250); }); test('missing parents, unknown references, and helper misses persist as explicit partial rows', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 730, sessionHash: 'partial-row-session' }); await enqueue(t, account, 730, 'partial-row-session'); const claimed = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'partial-row-worker' }); const generation = await t.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken }); const rows = [ { itemId: '100', characterId: 730, typeId: 34, quantity: 1, locationId: '999', locationType: 'item', locationFlag: 'Cargo' }, { itemId: '101', characterId: 730, typeId: 35, quantity: 1, locationId: '60009999', locationType: 'station', locationFlag: 'Hangar' }, ]; await t.mutation(mutation('bridgeStageSyncAssets'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, index: 0, assets: rows }); const observation = { requestedIds: ['100'], names: [], missing: ['100'], retrievedAt: 100, expiresAt: 1_000, complete: false }; const forbiddenObservation = { state: 'forbidden' as const, requestedIds: ['101'], names: [], missing: [], retrievedAt: 100, expiresAt: 1_000, complete: false }; const locationObservation = { requestedIds: ['100'], locations: [], missing: ['100'], retrievedAt: 100, expiresAt: 1_000, complete: false }; const unavailableLocationObservation = { state: 'unavailable' as const, requestedIds: ['101'], locations: [], missing: [], retrievedAt: 100, expiresAt: 1_000, complete: false }; const service = new InventoryLocationService({ reference: { pinnedVersion: 'ref-v1', pin() { return this; }, close() {}, async types() { return []; }, async systems() { return []; }, async regions() { return []; }, async stations() { return []; }, async constellations() { return []; }, async searchTypes() { return []; }, async searchTypesByName() { return []; } } as any, now: () => 200 }); const enrichment = await service.enrich({ assets: rows, authority: { userId: account.userId, principal: 'queue-owner', characterId: 730, scopes: ['esi-assets.read_assets.v1'], grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1 }, assetGeneration: generation, namesBatches: [observation, forbiddenObservation], locationsBatches: [locationObservation, unavailableLocationObservation] }); expect(enrichment.rows.map((row) => row.unresolved).flat()).toEqual(expect.arrayContaining(['missing_parent', 'reference_unknown'])); await t.mutation(mutation('bridgeStageSyncEnrichment'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, index: 0, rows: enrichment.rows }); await t.mutation(mutation('bridgeCommitSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, chunkCount: 1, syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000, enrichmentStatus: 'partial', enrichmentVersion: enrichment.enrichmentVersion, enrichmentInputPin: enrichment.inputPin, enrichmentChunkCount: 1, enrichmentUnresolvedCount: enrichment.unresolved.length }); const chunks = await t.query(query('bridgeEnrichmentChunksPage'), { ...bridge, userId: account.userId, characterId: 730, expectedGeneration: generation, paginationOpts: { numItems: 10, cursor: null } }); expect(chunks.chunks[0]?.rows.flatMap((row: any) => row.unresolved)).toEqual(expect.arrayContaining(['missing_parent', 'reference_unknown'])); }); test('authorization quarantine fences both stale success and stale failure', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 727, sessionHash: 'quarantine-session' }); await enqueue(t, account, 727, 'quarantine-session'); const claimed = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-quarantine' }); await t.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken }); await t.mutation(mutation('bridgeQuarantineAuthorization'), { ...bridge, userId: account.userId, characterId: 727 }); await expect(t.mutation(mutation('bridgeCommitSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, chunkCount: 0, syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000 })).rejects.toThrow(); await expect(t.mutation(mutation('bridgeFailSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken })).rejects.toThrow(); }); test('worker batches helper calls and publishes enrichment with the staged authority pin', async () => { const calls: string[] = []; const claimed = { id: 'job-location', jobId: 'job-location', userId: 'user-location', characterId: 730, leaseToken: 'lease-location', status: 'running', attempts: 1, dueAt: Date.now(), createdAt: Date.now(), updatedAt: Date.now(), grantRevision: 2, authorizationEpoch: 3, credentialVersion: 4, scopes: ['esi-assets.read_assets.v1'] } as any; const assets = Array.from({ length: 1_001 }, (_, index) => ({ itemId: String(index + 1), characterId: 730, typeId: 34, quantity: 1, locationId: '60003760', locationType: 'station', locationFlag: 'Hangar' })); const store = { claimSyncJob: async () => claimed, renewSyncJob: async () => true, acquireRefreshLease: async () => ({ leaseToken: 'refresh-location', expiresAt: Date.now() + 120_000, expectedVersion: 1, encryptedTokens: 'sealed', tokenExpiresAt: Date.now() + 120_000, scopes: claimed.scopes, grantRevision: 2, authorizationEpoch: 3, credentialVersion: 4 }), refreshSyncJobAuthority: async () => ({ grantRevision: 2, authorizationEpoch: 3, credentialVersion: 4, scopes: claimed.scopes }), beginSyncJob: async () => 'generation-location', cooldown: async () => null, stageSyncAssets: async () => {}, stageSyncEnrichment: async (_job: string, _lease: string, index: number) => { calls.push(`enrichment:${index}`); }, commitSyncJob: async (_job: string, _lease: string, _chunks: number, _synced: number, _expires: number, enrichment: any) => { calls.push(`commit:${enrichment.status}:${enrichment.chunkCount}`); return true; }, failSyncJob: async () => true, deferSyncJob: async () => true, }; const esi = { getAssets: async () => ({ assets, syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000 }), getAssetNames: async (_character: number, ids: string[]) => ({ requestedIds: ids, names: [], missing: ids, retrievedAt: 1, expiresAt: 60_000, complete: false }), getAssetLocations: async (_character: number, ids: string[]) => ({ requestedIds: ids, locations: [], missing: ids, retrievedAt: 1, expiresAt: 60_000, complete: false }), }; let capturedNames = 0; let capturedLocations = 0; await processSyncJob({ store: store as any, cipher: { open: async () => ({ characterId: 730, ownerHash: 'owner', accessToken: 'access', refreshToken: 'refresh', expiresAt: Date.now() + 120_000, scopes: claimed.scopes }) } as any, sso: {} as any, esi: esi as any, workerId: 'worker-location', pollMs: 1, leaseMs: 600_000, heartbeatMs: 60_000, sleep: async () => {}, locationServiceFactory: () => ({ enrich: async (request: any) => { capturedNames = request.namesBatches.length; capturedLocations = request.locationsBatches.length; return { status: 'partial', rows: [], unresolved: [{ itemId: null, reasons: ['unavailable_structure'] }], enrichmentVersion: 'containment-v1-worker', inputPin: { assetGeneration: 'generation-location' } }; } } as any), }); expect(capturedNames).toBe(2); expect(capturedLocations).toBe(2); expect(calls).toEqual(['commit:partial:0']); }); test('worker retries applied-then-lost staging and commit acknowledgements idempotently', async () => { const events: string[] = []; let stageCalls = 0; let commitCalls = 0; const claimed = { id: 'job-lost', jobId: 'job-lost', userId: 'user-lost', characterId: 728, leaseToken: 'lease-lost', status: 'running', attempts: 1, dueAt: Date.now(), createdAt: Date.now(), updatedAt: Date.now(), grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1, scopes: ['esi-assets.read_assets.v1'] } as any; const store = { claimSyncJob: async () => claimed, renewSyncJob: async () => { events.push('renew'); return true; }, acquireRefreshLease: async () => ({ leaseToken: 'refresh-lost', expiresAt: Date.now() + 120_000, expectedVersion: 1, encryptedTokens: 'sealed', tokenExpiresAt: Date.now() + 120_000, scopes: ['esi-assets.read_assets.v1'], grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1 }), refreshSyncJobAuthority: async () => ({ grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1, scopes: ['esi-assets.read_assets.v1'] }), beginSyncJob: async () => 'generation-lost', cooldown: async () => null, stageSyncAssets: async () => { stageCalls += 1; events.push(`stage${stageCalls}`); if (stageCalls === 1) throw new Error('response lost after stage'); }, commitSyncJob: async () => { commitCalls += 1; events.push(`commit${commitCalls}`); if (commitCalls === 1) throw new Error('response lost after commit'); return true; }, failSyncJob: async () => { events.push('fail'); return true; }, deferSyncJob: async () => { events.push('defer'); return true; }, }; await processSyncJob({ store: store as any, cipher: { open: async () => ({ characterId: 728, ownerHash: 'owner', accessToken: 'access', refreshToken: 'refresh', expiresAt: Date.now() + 120_000, scopes: ['esi-assets.read_assets.v1'] }) } as any, sso: {} as any, esi: { getAssets: async () => ({ assets: [{ itemId: '1', characterId: 728, typeId: 34, quantity: 1, locationId: '2', locationType: 'station', locationFlag: 'Hangar' }], syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000 }) } as any, workerId: 'worker-lost', pollMs: 1, leaseMs: 600_000, heartbeatMs: 60_000, sleep: async () => {}, }); expect(stageCalls).toBe(2); expect(commitCalls).toBe(2); expect(events).toEqual(['stage1', 'renew', 'stage2', 'commit1', 'renew', 'commit2']); }); test('ambiguous begin and commit mutations defer for lease recovery instead of failing terminally', async () => { const beginEvents: string[] = []; const beginStore = { claimSyncJob: async () => ({ id: 'job-begin-uncertain', jobId: 'job-begin-uncertain', userId: 'user-begin', characterId: 740, leaseToken: 'lease-begin', status: 'running', attempts: 1, dueAt: Date.now(), createdAt: Date.now(), updatedAt: Date.now(), grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1, scopes: ['esi-assets.read_assets.v1'] } as any), renewSyncJob: async () => { beginEvents.push('renew'); return true; }, acquireRefreshLease: async () => ({ leaseToken: 'refresh-begin', expiresAt: Date.now() + 120_000, expectedVersion: 1, encryptedTokens: 'sealed', tokenExpiresAt: Date.now() + 120_000, scopes: ['esi-assets.read_assets.v1'], grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1 }), refreshSyncJobAuthority: async () => ({ grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1, scopes: ['esi-assets.read_assets.v1'] }), beginSyncJob: async () => { beginEvents.push('begin'); throw new Error('response lost after begin'); }, deferSyncJob: async () => { beginEvents.push('defer'); return true; }, failSyncJob: async () => { beginEvents.push('fail'); return true; }, }; await processSyncJob({ store: beginStore as any, cipher: { open: async () => ({ characterId: 740, ownerHash: 'owner', accessToken: 'access', refreshToken: 'refresh', expiresAt: Date.now() + 120_000, scopes: ['esi-assets.read_assets.v1'] }) } as any, sso: {} as any, esi: {} as any, workerId: 'worker-begin', pollMs: 1, leaseMs: 600_000, heartbeatMs: 60_000, sleep: async () => {} }); expect(beginEvents).toEqual(['begin', 'renew', 'begin', 'defer']); const commitEvents: string[] = []; const commitStore = { claimSyncJob: async () => ({ id: 'job-commit-uncertain', jobId: 'job-commit-uncertain', userId: 'user-commit', characterId: 741, leaseToken: 'lease-commit', status: 'running', attempts: 1, dueAt: Date.now(), createdAt: Date.now(), updatedAt: Date.now(), grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1, scopes: ['esi-assets.read_assets.v1'] } as any), renewSyncJob: async () => { commitEvents.push('renew'); return true; }, acquireRefreshLease: async () => ({ leaseToken: 'refresh-commit', expiresAt: Date.now() + 120_000, expectedVersion: 1, encryptedTokens: 'sealed', tokenExpiresAt: Date.now() + 120_000, scopes: ['esi-assets.read_assets.v1'], grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1 }), refreshSyncJobAuthority: async () => ({ grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1, scopes: ['esi-assets.read_assets.v1'] }), beginSyncJob: async () => 'generation-commit-uncertain', cooldown: async () => null, stageSyncAssets: async () => { commitEvents.push('stage'); }, commitSyncJob: async () => { commitEvents.push('commit'); throw new Error('response lost after commit'); }, deferSyncJob: async () => { commitEvents.push('defer'); return true; }, failSyncJob: async () => { commitEvents.push('fail'); return true; }, }; await processSyncJob({ store: commitStore as any, cipher: { open: async () => ({ characterId: 741, ownerHash: 'owner', accessToken: 'access', refreshToken: 'refresh', expiresAt: Date.now() + 120_000, scopes: ['esi-assets.read_assets.v1'] }) } as any, sso: {} as any, esi: { getAssets: async () => ({ assets: [], syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000 }) } as any, workerId: 'worker-commit', pollMs: 1, leaseMs: 600_000, heartbeatMs: 60_000, sleep: async () => {} }); expect(commitEvents).toEqual(['commit', 'renew', 'commit', 'defer']); }); test('ambiguous restart after refresh commit defers without replaying the remote refresh', async () => { const events: string[] = []; const store = { claimSyncJob: async () => ({ id: 'job-restart-uncertain', jobId: 'job-restart-uncertain', userId: 'user-restart', characterId: 742, leaseToken: 'lease-restart', status: 'running', attempts: 1, generation: 'old-generation', dueAt: Date.now(), createdAt: Date.now(), updatedAt: Date.now(), grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1, scopes: ['esi-assets.read_assets.v1'] } as any), renewSyncJob: async () => { events.push('renew'); return true; }, acquireRefreshLease: async () => ({ leaseToken: 'refresh-restart', expiresAt: Date.now() + 120_000, expectedVersion: 1, encryptedTokens: 'sealed', tokenExpiresAt: Date.now() + 1_000, scopes: ['esi-assets.read_assets.v1'], grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1 }), markRefreshDispatched: async () => { events.push('dispatch'); return true; }, commitRefresh: async () => { events.push('refresh-commit'); return { credentialVersion: 2, grantRevision: 1, authorizationEpoch: 1, scopes: ['esi-assets.read_assets.v1'] }; }, restartSyncJobGeneration: async () => { events.push('restart'); throw new Error('response lost after restart'); }, markRefreshUncertain: async () => { events.push('uncertain'); return true; }, deferSyncJob: async () => { events.push('defer'); return true; }, failSyncJob: async () => { events.push('fail'); return true; }, }; await processSyncJob({ store: store as any, cipher: { open: async () => ({ characterId: 742, ownerHash: 'owner', accessToken: 'access', refreshToken: 'refresh', expiresAt: Date.now() + 1_000, scopes: ['esi-assets.read_assets.v1'] }), seal: async () => 'rotated-sealed' } as any, sso: { refresh: async () => { events.push('remote'); return { characterId: 742, ownerHash: 'owner', accessToken: 'new-access', refreshToken: 'new-refresh', expiresAt: Date.now() + 120_000, scopes: ['esi-assets.read_assets.v1'] }; } } as any, esi: {} as any, workerId: 'worker-restart', pollMs: 1, leaseMs: 600_000, heartbeatMs: 60_000, sleep: async () => {} }); expect(events).toEqual(['dispatch', 'remote', 'refresh-commit', 'restart', 'renew', 'restart', 'uncertain', 'defer']); }); test('worker publishes a complete empty snapshot through the leased lifecycle', async () => { const events: string[] = []; const claimed = { id: 'job-1', jobId: 'job-1', userId: 'user-1', characterId: 700, leaseToken: 'lease-1', status: 'running', attempts: 1, dueAt: Date.now(), createdAt: Date.now(), updatedAt: Date.now(), grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1, scopes: ['esi-assets.read_assets.v1'] } as any; const store = { claimSyncJob: async () => claimed, renewSyncJob: async () => true, acquireRefreshLease: async () => ({ leaseToken: 'refresh-1', expiresAt: Date.now() + 120_000, expectedVersion: 1, encryptedTokens: 'sealed', tokenExpiresAt: Date.now() + 120_000, scopes: ['esi-assets.read_assets.v1'], grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1 }), releaseRefreshLease: async () => { events.push('release'); return true; }, refreshSyncJobAuthority: async () => ({ grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1, scopes: ['esi-assets.read_assets.v1'] }), beginSyncJob: async () => { events.push('begin'); return 'generation-1'; }, cooldown: async () => null, stageSyncAssets: async () => { events.push('stage'); }, commitSyncJob: async () => { events.push('commit'); return true; }, failSyncJob: async () => { events.push('fail'); return true; }, deferSyncJob: async () => { events.push('defer'); return true; }, }; const handled = await processSyncJob({ store: store as any, cipher: { open: async () => ({ characterId: 700, ownerHash: 'owner', accessToken: 'access', refreshToken: 'refresh', expiresAt: Date.now() + 120_000, scopes: ['esi-assets.read_assets.v1'] }) } as any, sso: { refresh: async () => { throw new Error('refresh should not run'); } } as any, esi: { getAssets: async () => ({ assets: [], syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000 }) } as any, workerId: 'worker-1', pollMs: 1, leaseMs: 600_000, heartbeatMs: 60_000, sleep: async () => {}, }); expect(handled).toBe(true); expect(events).toEqual(['release', 'begin', 'commit']); }); test('worker durably marks refresh dispatch before a failed remote attempt', async () => { const events: string[] = []; const claimed = { id: 'job-refresh', jobId: 'job-refresh', userId: 'user-refresh', characterId: 725, leaseToken: 'lease-refresh', status: 'running', attempts: 1, dueAt: Date.now(), createdAt: Date.now(), updatedAt: Date.now(), grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1, scopes: ['esi-assets.read_assets.v1'] } as any; const store = { claimSyncJob: async () => claimed, renewSyncJob: async () => true, acquireRefreshLease: async () => ({ leaseToken: 'refresh-dispatch', expiresAt: Date.now() + 120_000, expectedVersion: 1, encryptedTokens: 'sealed', tokenExpiresAt: Date.now() + 1_000, scopes: ['esi-assets.read_assets.v1'], grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1 }), markRefreshDispatched: async () => { events.push('dispatch'); return true; }, markRefreshUncertain: async () => { events.push('uncertain'); return true; }, failSyncJob: async () => { events.push('fail'); return true; }, deferSyncJob: async () => { events.push('defer'); return true; }, }; await processSyncJob({ store: store as any, cipher: { open: async () => ({ characterId: 725, ownerHash: 'owner', accessToken: 'access', refreshToken: 'refresh', expiresAt: Date.now() + 1_000, scopes: ['esi-assets.read_assets.v1'] }) } as any, sso: { refresh: async () => { events.push('remote'); throw new Error('unknown remote outcome'); } } as any, esi: {} as any, workerId: 'worker-refresh', pollMs: 1, leaseMs: 600_000, heartbeatMs: 60_000, sleep: async () => {}, }); expect(events).toEqual(['dispatch', 'remote', 'uncertain']); }); test('worker defers rate limits into the durable cooldown instead of failing the job', async () => { const events: Array<{ name: string; dueAt?: number; cooldown?: unknown }> = []; const claimed = { id: 'job-2', jobId: 'job-2', userId: 'user-2', characterId: 701, leaseToken: 'lease-2', status: 'running', attempts: 2, dueAt: Date.now(), createdAt: Date.now(), updatedAt: Date.now(), grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1, scopes: ['esi-assets.read_assets.v1'] } as any; const retryAt = Date.now() + 15_000; const store = { claimSyncJob: async () => claimed, renewSyncJob: async () => true, acquireRefreshLease: async () => ({ leaseToken: 'refresh-2', expiresAt: Date.now() + 120_000, expectedVersion: 1, encryptedTokens: 'sealed', tokenExpiresAt: Date.now() + 120_000, scopes: ['esi-assets.read_assets.v1'], grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1 }), refreshSyncJobAuthority: async () => ({ grantRevision: 1, authorizationEpoch: 1, credentialVersion: 1, scopes: ['esi-assets.read_assets.v1'] }), beginSyncJob: async () => 'generation-2', cooldown: async () => null, stageSyncAssets: async () => {}, commitSyncJob: async () => true, failSyncJob: async () => { events.push({ name: 'fail' }); return true; }, deferSyncJob: async (_jobId: string, _lease: string, dueAt: number, cooldown?: unknown) => { events.push({ name: 'defer', dueAt, cooldown }); return true; }, }; await processSyncJob({ store: store as any, cipher: { open: async () => ({ characterId: 701, ownerHash: 'owner', accessToken: 'access', refreshToken: 'refresh', expiresAt: Date.now() + 120_000, scopes: ['esi-assets.read_assets.v1'] }) } as any, sso: { refresh: async () => { throw new Error('refresh should not run'); } } as any, esi: { getAssets: async () => { throw new EsiRateLimitError('/latest/characters/701/assets', retryAt, 15_000, 'esi:public'); } } as any, workerId: 'worker-2', pollMs: 1, leaseMs: 600_000, heartbeatMs: 60_000, sleep: async () => {}, }); expect(events).toHaveLength(1); expect(events[0]?.name).toBe('defer'); expect(events[0]?.dueAt).toBe(retryAt); expect(events[0]?.cooldown).toMatchObject({ context: 'character:user-2:701', until: retryAt }); }); test('terminal retry exhaustion clears syncing state while preserving the prior generation', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 724, sessionHash: 'terminal-session' }); await enqueue(t, account, 724, 'terminal-session'); const first = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-terminal-first' }); await t.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: first.jobId, leaseToken: first.leaseToken }); await t.mutation(mutation('bridgeCommitSyncJob'), { ...worker, jobId: first.jobId, leaseToken: first.leaseToken, chunkCount: 0, syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000 }); await enqueue(t, account, 724, 'terminal-session'); const second = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-terminal-second' }); await t.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: second.jobId, leaseToken: second.leaseToken }); await t.run(async (ctx: any) => await ctx.db.patch(second.jobId, { attempts: 5 })); await expect(t.mutation(mutation('bridgeDeferSyncJob'), { ...worker, jobId: second.jobId, leaseToken: second.leaseToken, dueAt: Date.now() + 1_000 })).resolves.toBe(false); const characters = await t.query(query('bridgeCharacters'), { ...bridge, userId: account.userId }); const character = characters.find((entry: any) => entry.characterId === 724); expect(character).toMatchObject({ state: 'error', activeGeneration: expect.any(String) }); expect(character.syncGeneration).toBeUndefined(); }); test('reclaims a lease after refresh commit before generation restart and publishes a fresh generation', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 703, sessionHash: 'refresh-crash-session' }); const queued = await enqueue(t, account, 703, 'refresh-crash-session'); const claimed = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-refresh-crash' }); const oldGeneration = await t.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken }); const refresh = await t.mutation(mutation('bridgeAcquireRefreshLease'), { ...worker, userId: account.userId, characterId: 703, ownerId: 'worker-refresh-crash' }); await t.mutation(mutation('bridgeMarkRefreshDispatched'), { ...worker, userId: account.userId, characterId: 703, ownerId: 'worker-refresh-crash', leaseToken: refresh.leaseToken }); await t.mutation(mutation('bridgeCommitRefresh'), { ...worker, userId: account.userId, characterId: 703, ownerId: 'worker-refresh-crash', leaseToken: refresh.leaseToken, expectedVersion: refresh.expectedVersion, encryptedTokens: 'rotated-token', tokenExpiresAt: Date.now() + 120_000, scopes: ['esi-assets.read_assets.v1'], jobId: claimed.jobId, jobLeaseToken: claimed.leaseToken }); await t.run(async (ctx: any) => await ctx.db.patch(queued.id, { leaseExpiresAt: 0 })); const reclaimed = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-refresh-reclaimed' }); const freshGeneration = await t.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: reclaimed.jobId, leaseToken: reclaimed.leaseToken }); expect(freshGeneration).not.toBe(oldGeneration); await t.mutation(mutation('bridgeCommitSyncJob'), { ...worker, jobId: reclaimed.jobId, leaseToken: reclaimed.leaseToken, chunkCount: 0, syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000 }); const characters = await t.query(query('bridgeCharacters'), { ...bridge, userId: account.userId }); expect(characters.find((entry: any) => entry.characterId === 703)).toMatchObject({ state: 'ready', activeGeneration: freshGeneration }); }); test('refresh commit validates allowlisted subset scopes and fences reduced grants', async () => { const makeLease = async (t: ReturnType, account: any, ownerId: string) => { const lease = await t.mutation(mutation('bridgeAcquireRefreshLease'), { ...worker, userId: account.userId, characterId: account.characterId, ownerId }); await t.mutation(mutation('bridgeMarkRefreshDispatched'), { ...worker, userId: account.userId, characterId: account.characterId, ownerId, leaseToken: lease.leaseToken }); return lease; }; const invalid = convexTest({ schema, modules }); const invalidAccount = await login(invalid, { characterId: 730, sessionHash: 'scope-invalid-session' }); const invalidLease = await makeLease(invalid, { ...invalidAccount, characterId: 730 }, 'scope-invalid'); await expect(invalid.mutation(mutation('bridgeCommitRefresh'), { ...worker, userId: invalidAccount.userId, characterId: 730, ownerId: 'scope-invalid', leaseToken: invalidLease.leaseToken, expectedVersion: invalidLease.expectedVersion, encryptedTokens: 'bad', tokenExpiresAt: Date.now() + 60_000, scopes: ['esi-unknown.scope'] })).rejects.toThrow('Invalid refreshed scopes'); await expect(invalid.mutation(mutation('bridgeSaveTokens'), { ...bridge, userId: invalidAccount.userId, characterId: 730, encryptedTokens: 'expanded', tokenExpiresAt: Date.now() + 60_000, scopes: ['esi-assets.read_assets.v1', 'esi-wallet.read_character_wallet.v1'], expectedVersion: 1 })).resolves.toBe(false); const expanded = convexTest({ schema, modules }); const expandedAccount = await login(expanded, { characterId: 731, sessionHash: 'scope-expanded-session' }); const expandedLease = await makeLease(expanded, { ...expandedAccount, characterId: 731 }, 'scope-expanded'); await expect(expanded.mutation(mutation('bridgeCommitRefresh'), { ...worker, userId: expandedAccount.userId, characterId: 731, ownerId: 'scope-expanded', leaseToken: expandedLease.leaseToken, expectedVersion: expandedLease.expectedVersion, encryptedTokens: 'bad', tokenExpiresAt: Date.now() + 60_000, scopes: ['esi-assets.read_assets.v1', 'esi-wallet.read_character_wallet.v1'] })).rejects.toThrow('Invalid refreshed scopes'); const reduced = convexTest({ schema, modules }); const reducedAccount = await login(reduced, { characterId: 732, sessionHash: 'scope-reduced-session' }); const reducedJob = await enqueue(reduced, reducedAccount, 732, 'scope-reduced-session'); const reducedClaim = await reduced.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'scope-reduced-worker' }); await reduced.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: reducedClaim.jobId, leaseToken: reducedClaim.leaseToken }); await reduced.mutation(mutation('bridgeCommitSyncJob'), { ...worker, jobId: reducedClaim.jobId, leaseToken: reducedClaim.leaseToken, chunkCount: 0, syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000 }); const reducedLease = await makeLease(reduced, { ...reducedAccount, characterId: 732 }, 'scope-reduced'); await expect(reduced.mutation(mutation('bridgeCommitRefresh'), { ...worker, userId: reducedAccount.userId, characterId: 732, ownerId: 'scope-reduced', leaseToken: reducedLease.leaseToken, expectedVersion: reducedLease.expectedVersion, encryptedTokens: 'reduced', tokenExpiresAt: Date.now() + 60_000, scopes: [] })).resolves.toMatchObject({ saved: true }); const reducedCharacter = (await reduced.query(query('bridgeCharacters'), { ...bridge, userId: reducedAccount.userId })).find((entry: any) => entry.characterId === 732); expect(reducedCharacter).toMatchObject({ state: 'error' }); expect(reducedCharacter.activeGeneration).toBeUndefined(); const unchanged = convexTest({ schema, modules }); const unchangedAccount = await login(unchanged, { characterId: 733, sessionHash: 'scope-unchanged-session' }); const unchangedLease = await makeLease(unchanged, { ...unchangedAccount, characterId: 733 }, 'scope-unchanged'); await expect(unchanged.mutation(mutation('bridgeCommitRefresh'), { ...worker, userId: unchangedAccount.userId, characterId: 733, ownerId: 'scope-unchanged', leaseToken: unchangedLease.leaseToken, expectedVersion: unchangedLease.expectedVersion, encryptedTokens: 'unchanged', tokenExpiresAt: Date.now() + 60_000, scopes: ['esi-assets.read_assets.v1'] })).resolves.toMatchObject({ saved: true, grantRevision: 1, credentialVersion: 2 }); }); test('first-generation asset scope loss quarantines staging and clears the matching pointer', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 736, sessionHash: 'first-scope-loss-session' }); await enqueue(t, account, 736, 'first-scope-loss-session'); const claimed = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'first-scope-loss-worker' }); const generation = await t.mutation(mutation('bridgeBeginSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken }); const refresh = await t.mutation(mutation('bridgeAcquireRefreshLease'), { ...worker, userId: account.userId, characterId: 736, ownerId: 'first-scope-loss-worker' }); await t.mutation(mutation('bridgeMarkRefreshDispatched'), { ...worker, userId: account.userId, characterId: 736, ownerId: 'first-scope-loss-worker', leaseToken: refresh.leaseToken }); await t.mutation(mutation('bridgeCommitRefresh'), { ...worker, userId: account.userId, characterId: 736, ownerId: 'first-scope-loss-worker', leaseToken: refresh.leaseToken, expectedVersion: refresh.expectedVersion, encryptedTokens: 'reduced-first', tokenExpiresAt: Date.now() + 60_000, scopes: [], jobId: claimed.jobId, jobLeaseToken: claimed.leaseToken }); const characters = await t.query(query('bridgeCharacters'), { ...bridge, userId: account.userId }); const character = characters.find((entry: any) => entry.characterId === 736); expect(character.state).toBe('error'); expect(character.syncGeneration).toBeUndefined(); expect(character.syncLeaseExpiresAt).toBeUndefined(); expect(character.datasetStatuses.find((entry: any) => entry.dataset === 'assets')).toMatchObject({ status: 'blocked' }); const job = await t.query(query('bridgeSyncJob'), { ...bridge, userId: account.userId, jobId: claimed.jobId }); expect(job).toMatchObject({ status: 'quarantined' }); await expect(t.mutation(mutation('bridgeCommitSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, chunkCount: 0, syncedAt: Date.now() - 1_000, expiresAt: Date.now() + 60_000 })).rejects.toThrow(); await expect(t.mutation(mutation('bridgeFailSyncJob'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken })).rejects.toThrow(); await t.run(async (ctx: any) => { const row = await ctx.db.query('assetGenerations').withIndex('by_user_character_generation', (q: any) => q.eq('userId', account.userId).eq('characterId', 736).eq('generation', generation)).first(); expect(row.status).toBe('failed'); }); }); test('refresh release is fenced and lets a subsequent worker acquire immediately', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 734, sessionHash: 'release-session' }); const lease = await t.mutation(mutation('bridgeAcquireRefreshLease'), { ...worker, userId: account.userId, characterId: 734, ownerId: 'release-owner' }); await expect(t.mutation(mutation('bridgeReleaseRefreshLease'), { ...worker, userId: account.userId, characterId: 734, ownerId: 'foreign-owner', leaseToken: lease.leaseToken })).resolves.toBe(false); await expect(t.mutation(mutation('bridgeReleaseRefreshLease'), { ...worker, userId: account.userId, characterId: 734, ownerId: 'release-owner', leaseToken: lease.leaseToken })).resolves.toBe(true); await expect(t.mutation(mutation('bridgeReleaseRefreshLease'), { ...worker, userId: account.userId, characterId: 734, ownerId: 'release-owner', leaseToken: lease.leaseToken })).resolves.toBe(false); const next = await t.mutation(mutation('bridgeAcquireRefreshLease'), { ...worker, userId: account.userId, characterId: 734, ownerId: 'next-owner' }); expect(typeof next.leaseToken).toBe('string'); expect(next.expectedVersion).toBe(1); await t.mutation(mutation('bridgeMarkRefreshDispatched'), { ...worker, userId: account.userId, characterId: 734, ownerId: 'next-owner', leaseToken: next.leaseToken }); await expect(t.mutation(mutation('bridgeReleaseRefreshLease'), { ...worker, userId: account.userId, characterId: 734, ownerId: 'next-owner', leaseToken: next.leaseToken })).resolves.toBe(false); }); test('authority refresh rejects a same-scope credential rotation instead of adopting newer DB counters', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 704, sessionHash: 'stale-authority-session' }); const job = await enqueue(t, account, 704, 'stale-authority-session'); const claimed = await t.mutation(mutation('bridgeClaimSyncJob'), { ...worker, workerId: 'worker-stale' }); await t.mutation(mutation('bridgeSaveTokens'), { ...bridge, userId: account.userId, characterId: 704, encryptedTokens: 'new-token', tokenExpiresAt: Date.now() + 60_000, scopes: ['esi-assets.read_assets.v1'], expectedVersion: 1 }); await expect(t.mutation(mutation('bridgeRefreshSyncJobAuthority'), { ...worker, jobId: claimed.jobId, leaseToken: claimed.leaseToken, expectedGrantRevision: 1, expectedAuthorizationEpoch: 1, expectedCredentialVersion: 1, expectedScopes: ['esi-assets.read_assets.v1'] })).rejects.toThrow(); expect(job.status).toBe('queued'); }); test('expired dispatched refresh leases fail closed without replaying a rotated token', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 705, sessionHash: 'dispatch-session' }); const lease = await t.mutation(mutation('bridgeAcquireRefreshLease'), { ...worker, userId: account.userId, characterId: 705, ownerId: 'worker-dispatch' }); await t.mutation(mutation('bridgeMarkRefreshDispatched'), { ...worker, userId: account.userId, characterId: 705, ownerId: 'worker-dispatch', leaseToken: lease.leaseToken }); await t.run(async (ctx: any) => { const row = await ctx.db.query('refreshLeases').withIndex('by_user_character', (q: any) => q.eq('userId', account.userId).eq('characterId', 705)).first(); await ctx.db.patch(row._id, { expiresAt: 0 }); }); await expect(t.mutation(mutation('bridgeAcquireRefreshLease'), { ...worker, userId: account.userId, characterId: 705, ownerId: 'worker-replay' })).resolves.toBeNull(); const characters = await t.query(query('bridgeCharacters'), { ...bridge, userId: account.userId }); expect(characters.find((character: any) => character.characterId === 705)).toMatchObject({ authorizationStatus: 'reconnect_required' }); await expect(t.query(query('bridgeTokenRecord'), { ...bridge, userId: account.userId, characterId: 705 })).resolves.toBeNull(); }); test('worker credentials cannot invoke general login, enqueue, or cancellation bridges', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 706, sessionHash: 'least-privilege-session' }); await expect(t.mutation(mutation('bridgeCompleteLogin'), { ...worker, characterId: 707, name: 'Nope', ownerHash: 'nope', scopes: ['esi-assets.read_assets.v1'], intent: 'initial_login', capability: 'asset_search', requestedScopes: ['esi-assets.read_assets.v1'], approvedExistingScopes: [], encryptedTokens: 'x', tokenExpiresAt: Date.now() + 60_000, sessionHash: 'bad-worker-session', sessionExpiresAt: Date.now() + 60_000 })).rejects.toThrow(); await expect(t.mutation(mutation('bridgeEnqueueAssetSync'), { ...worker, userId: account.userId, characterId: 706, initiatingSessionHash: 'least-privilege-session' })).rejects.toThrow(); await expect(t.mutation(mutation('bridgeCancelSyncJob'), { ...worker, userId: account.userId, jobId: 'missing', initiatingSessionHash: 'least-privilege-session' })).rejects.toThrow(); }); test('refresh lease is durable, CAS committed, and uncertain rotation fails closed', async () => { const t = convexTest({ schema, modules }); const account = await login(t, { characterId: 702, sessionHash: 'refresh-session' }); const lease = await t.mutation(mutation('bridgeAcquireRefreshLease'), { ...worker, userId: account.userId, characterId: 702, ownerId: 'worker-a' }); expect(lease).toMatchObject({ expectedVersion: 1 }); await expect(t.mutation(mutation('bridgeAcquireRefreshLease'), { ...worker, userId: account.userId, characterId: 702, ownerId: 'worker-b' })).resolves.toBeNull(); await expect(t.mutation(mutation('bridgeMarkRefreshUncertain'), { ...worker, userId: account.userId, characterId: 702, ownerId: 'worker-a', leaseToken: lease.leaseToken, expectedVersion: 1 })).resolves.toBe(true); const characters = await t.query(query('bridgeCharacters'), { ...bridge, userId: account.userId }); expect(characters[0]).toMatchObject({ authorizationStatus: 'reconnect_required' }); await expect(t.query(query('bridgeTokenRecord'), { ...bridge, userId: account.userId, characterId: 702 })).resolves.toBeNull(); }); });