wip: checkpoint reviewed T13 wallet balance core
This commit is contained in:
Vendored
+41
@@ -0,0 +1,41 @@
|
||||
import { ESI_WALLET_SCOPE } from '../../../src/lib/server/esi';
|
||||
import type { AuthorizationKey } from '../../../src/lib/server/esi';
|
||||
import type { ResourcePrincipal, SelectedResource } from '../../../src/lib/domain/model';
|
||||
import type {
|
||||
WalletAuthorizedResource,
|
||||
WalletAuthorityFence,
|
||||
WalletBalanceSnapshot,
|
||||
WalletCredential,
|
||||
} from '../../../src/lib/server/wallet/balance-repository';
|
||||
|
||||
export function resource(userId: string, characterId: number, overrides: Partial<WalletAuthorizedResource> = {}): WalletAuthorizedResource {
|
||||
const principal: Extract<ResourcePrincipal, { kind: 'character' }> = { kind: 'character', id: String(characterId), ownerUserId: userId };
|
||||
const selected: SelectedResource = { principal, requiredScopes: [ESI_WALLET_SCOPE], authorizationRevision: overrides.grantRevision ?? 1, authorizationEpoch: overrides.authorizationEpoch ?? 1 };
|
||||
return {
|
||||
userId,
|
||||
characterId,
|
||||
principal: overrides.principal ?? principal,
|
||||
resource: overrides.resource ?? selected,
|
||||
authorizationStatus: 'active',
|
||||
consent: 'granted',
|
||||
grantedScopes: [ESI_WALLET_SCOPE],
|
||||
grantRevision: 1,
|
||||
authorizationEpoch: 1,
|
||||
credentialVersion: 1,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
export function fence(value: WalletAuthorizedResource): WalletAuthorityFence {
|
||||
return { userId: value.userId, characterId: value.characterId, principal: { ...value.principal }, grantRevision: value.grantRevision, authorizationEpoch: value.authorizationEpoch, credentialVersion: value.credentialVersion!, scopes: [...value.grantedScopes] };
|
||||
}
|
||||
|
||||
export function snapshot(value: WalletAuthorizedResource, amount: string, fetchedAt = 100, expiresAt = 1_000, generation = `wallet-${value.characterId}-${fetchedAt}`): WalletBalanceSnapshot {
|
||||
return { userId: value.userId, characterId: value.characterId, principal: { ...value.principal }, amount, fetchedAt, expiresAt, upstreamEffectiveAt: null, coverage: { kind: 'snapshot', generation, complete: true }, generation, authorizationRevision: value.grantRevision, authorizationEpoch: value.authorizationEpoch, credentialVersion: value.credentialVersion!, };
|
||||
}
|
||||
|
||||
export function credential(value: WalletAuthorizedResource, token = `token-${value.userId}-${value.characterId}`): { fence: WalletAuthorityFence; credential: WalletCredential } {
|
||||
const current = fence(value);
|
||||
const authorizationKey: AuthorizationKey = { userId: value.userId, principal: `character:${value.principal.id}`, characterId: value.characterId, scopes: [...value.grantedScopes], grantRevision: value.grantRevision, authorizationEpoch: value.authorizationEpoch, credentialVersion: value.credentialVersion! };
|
||||
return { fence: current, credential: { accessToken: token, authorizationKey } };
|
||||
}
|
||||
@@ -0,0 +1,311 @@
|
||||
import { describe, expect, test } from 'bun:test';
|
||||
import { ESI_WALLET_SCOPE, EsiError, EsiRateLimitError } from '../src/lib/server/esi';
|
||||
import { InMemoryWalletBalanceJobQueue, InMemoryWalletBalancePublicationPort, WalletBalanceJobAdapter } from '../src/lib/server/wallet/balance-job';
|
||||
import { InMemoryWalletBalanceRepository } from '../src/lib/server/wallet/balance-repository';
|
||||
import { WalletBalanceService } from '../src/lib/server/wallet/balance-service';
|
||||
import { credential, fence, resource, snapshot } from './fixtures/wallet';
|
||||
|
||||
describe('wallet balance core', () => {
|
||||
test('sums signed and larger-than-safe-number cents exactly, without wealth/assets', async () => {
|
||||
const one = resource('tenant-a', 1);
|
||||
const two = resource('tenant-a', 2);
|
||||
const repository = new InMemoryWalletBalanceRepository({
|
||||
resources: [one, two],
|
||||
snapshots: [snapshot(one, '-900719925474099301'), snapshot(two, '900719925474099301')],
|
||||
});
|
||||
const result = await new WalletBalanceService({ repository, now: () => 500 }).read({ userId: 'tenant-a' });
|
||||
expect(result.status).toBe('ready');
|
||||
expect(result.walletTotal).toBe('0');
|
||||
expect(result.walletOnlyTotal).toBe('0');
|
||||
expect(result.components.every((component) => component.component === 'wallet_balance')).toBe(true);
|
||||
expect(result.provenance?.input.requiredDatasetKinds).toEqual(['wallet_balance']);
|
||||
expect(result.provenance?.input.referenceVersion).toBeNull();
|
||||
});
|
||||
|
||||
test('shows consent/stale state per character and does not produce a partial total', async () => {
|
||||
const granted = resource('tenant-a', 1);
|
||||
const noConsent = resource('tenant-a', 2, { consent: 'missing', grantedScopes: [], credentialVersion: null });
|
||||
const stale = resource('tenant-a', 3);
|
||||
const queue = new InMemoryWalletBalanceJobQueue();
|
||||
const repository = new InMemoryWalletBalanceRepository({ resources: [granted, noConsent, stale], snapshots: [snapshot(granted, '125'), snapshot(stale, '-25', 100, 200)] });
|
||||
const result = await new WalletBalanceService({ repository, queue, now: () => 500 }).read({ userId: 'tenant-a' });
|
||||
expect(result.status).toBe('partial');
|
||||
expect(result.walletTotal).toBeNull();
|
||||
expect(result.characters.map((character) => [character.characterId, character.consent, character.freshness])).toEqual([[1, 'granted', 'fresh'], [2, 'missing', 'consent_required'], [3, 'granted', 'stale']]);
|
||||
expect(result.missing.map((principal) => principal.id)).toEqual(['2', '3']);
|
||||
expect(result.refresh.queued).toBe(1);
|
||||
expect((await queue.claim('worker', 500))?.characterId).toBe(3);
|
||||
});
|
||||
|
||||
test('opening Wallet does not enqueue a valid cache and queue deduplicates stale refreshes', async () => {
|
||||
const value = resource('tenant-a', 4);
|
||||
const queue = new InMemoryWalletBalanceJobQueue();
|
||||
const repository = new InMemoryWalletBalanceRepository({ resources: [value], snapshots: [snapshot(value, '4', 100, 200)] });
|
||||
const service = new WalletBalanceService({ repository, queue, now: () => 500 });
|
||||
const first = await service.read({ userId: 'tenant-a' });
|
||||
const second = await service.read({ userId: 'tenant-a' });
|
||||
expect(first.refresh.queued).toBe(1);
|
||||
expect(second.refresh.duplicate).toBe(1);
|
||||
const valid = await new WalletBalanceService({ repository, queue: new InMemoryWalletBalanceJobQueue(), now: () => 150 }).read({ userId: 'tenant-a' });
|
||||
expect(valid.status).toBe('ready');
|
||||
expect(valid.refresh.attempted).toBe(0);
|
||||
queue.setCooldown('tenant-a', 4, 10_000);
|
||||
const cooldown = await service.read({ userId: 'tenant-a' });
|
||||
expect(cooldown.refresh.cooldown).toBe(1);
|
||||
});
|
||||
|
||||
test('tenant and selector isolation is server resolved', async () => {
|
||||
const a = resource('tenant-a', 7);
|
||||
const b = resource('tenant-b', 7);
|
||||
const repository = new InMemoryWalletBalanceRepository({ resources: [a, b], snapshots: [snapshot(a, '700'), snapshot(b, '900')] });
|
||||
const result = await new WalletBalanceService({ repository, now: () => 500 }).read({ userId: 'tenant-a', characterIds: ['7'] });
|
||||
expect(result.status).toBe('ready');
|
||||
expect(result.walletTotal).toBe('700');
|
||||
await expect(new WalletBalanceService({ repository, now: () => 500 }).read({ userId: 'tenant-a', characterIds: ['8'] })).resolves.toMatchObject({ status: 'blocked' });
|
||||
expect(await repository.snapshot('tenant-b', 7)).toMatchObject({ amount: '900' });
|
||||
});
|
||||
|
||||
test('job uses exact ESI transport and publication fence, preserving old snapshot on failure', async () => {
|
||||
const value = resource('tenant-a', 8);
|
||||
const repository = new InMemoryWalletBalanceRepository({ resources: [value], snapshots: [snapshot(value, '10', 100, 200)], credentials: [credential(value)] });
|
||||
const jobs = new InMemoryWalletBalanceJobQueue({ leaseMs: 100, now: () => 500 });
|
||||
const publication = new InMemoryWalletBalancePublicationPort(jobs, repository);
|
||||
const observations = [{ amount: '-900719925474099301', retrievedAt: 500, expiresAt: 1_000 }];
|
||||
const adapter = new WalletBalanceJobAdapter({
|
||||
jobs, authority: repository, publication,
|
||||
transport: { getWalletBalance: async (characterId, token, key) => { expect(characterId).toBe(8); expect(token).toBe('token-tenant-a-8'); expect(key.scopes).toContain(ESI_WALLET_SCOPE); return observations.shift()!; } },
|
||||
now: () => 500,
|
||||
});
|
||||
const enqueued = await jobs.enqueue({ userId: 'tenant-a', characterId: 8, fence: fence(value), now: 500 });
|
||||
expect(enqueued.status).toBe('queued');
|
||||
const run = await adapter.runNext('worker-a');
|
||||
expect(run.status).toBe('published');
|
||||
expect((await repository.snapshot('tenant-a', 8))?.amount).toBe('-900719925474099301');
|
||||
|
||||
const failed = resource('tenant-a', 9);
|
||||
const failedRepo = new InMemoryWalletBalanceRepository({ resources: [failed], snapshots: [snapshot(failed, '99', 100, 200)], credentials: [credential(failed)] });
|
||||
const failedJobs = new InMemoryWalletBalanceJobQueue({ now: () => 500 });
|
||||
const failedPublication = new InMemoryWalletBalancePublicationPort(failedJobs, failedRepo);
|
||||
const failing = new WalletBalanceJobAdapter({ jobs: failedJobs, authority: failedRepo, publication: failedPublication, transport: { getWalletBalance: async () => { throw new Error('upstream unavailable'); } }, now: () => 500 });
|
||||
await failedJobs.enqueue({ userId: 'tenant-a', characterId: 9, fence: fence(failed), now: 500 });
|
||||
expect((await failing.runNext('worker-a')).status).toBe('failed');
|
||||
expect((await failedRepo.snapshot('tenant-a', 9))?.amount).toBe('99');
|
||||
expect((await failedRepo.failure('tenant-a', 9))?.category).toBe('refresh_failed');
|
||||
});
|
||||
|
||||
test('borrowed or stale workers are fenced before publication', async () => {
|
||||
const value = resource('tenant-a', 11);
|
||||
const repository = new InMemoryWalletBalanceRepository({ resources: [value], credentials: [credential(value)] });
|
||||
const jobs = new InMemoryWalletBalanceJobQueue({ now: () => 100 });
|
||||
const publication = new InMemoryWalletBalancePublicationPort(jobs, repository);
|
||||
const adapter = new WalletBalanceJobAdapter({ jobs, authority: repository, publication, transport: { getWalletBalance: async () => ({ amount: '1', retrievedAt: 100, expiresAt: 1_000 }) }, now: () => 100 });
|
||||
await jobs.enqueue({ userId: 'tenant-a', characterId: 11, fence: fence(value), now: 100 });
|
||||
const claim = await jobs.claim('worker-a', 100);
|
||||
expect(claim).not.toBeNull();
|
||||
const rotated = resource('tenant-a', 11, { grantRevision: 2 });
|
||||
repository.replaceResource(rotated);
|
||||
const rejected = await adapter.runClaim(claim!);
|
||||
expect(rejected).toMatchObject({ status: 'rejected', reason: 'stale_grant' });
|
||||
expect(await repository.snapshot('tenant-a', 11)).toBeNull();
|
||||
});
|
||||
|
||||
test('transient authority lookup is deferred, not quarantined as revoked', async () => {
|
||||
const value = resource('tenant-a', 12);
|
||||
const repository = new InMemoryWalletBalanceRepository({ resources: [value], credentials: [credential(value)] });
|
||||
const jobs = new InMemoryWalletBalanceJobQueue({ now: () => 500 });
|
||||
const publication = new InMemoryWalletBalancePublicationPort(jobs, repository);
|
||||
let transportCalls = 0;
|
||||
const adapter = new WalletBalanceJobAdapter({ jobs, authority: { currentAuthority: async () => { throw new Error('backend timeout'); }, credential: repository.credential.bind(repository) }, publication, transport: { getWalletBalance: async () => { transportCalls += 1; return { amount: '1', retrievedAt: 500, expiresAt: 1_000 }; } }, now: () => 500 });
|
||||
const enqueued = await jobs.enqueue({ userId: 'tenant-a', characterId: 12, fence: fence(value), now: 500 });
|
||||
const result = await adapter.runNext('worker');
|
||||
expect(result).toMatchObject({ status: 'unavailable' });
|
||||
expect((await jobs.get(enqueued.status === 'queued' ? enqueued.job.id : ''))?.status).toBe('queued');
|
||||
expect(transportCalls).toBe(0);
|
||||
});
|
||||
|
||||
test('mutable tenant/character claim fields cannot substitute the canonical stored claim', async () => {
|
||||
const value = resource('tenant-a', 13);
|
||||
const repository = new InMemoryWalletBalanceRepository({ resources: [value], credentials: [credential(value)] });
|
||||
const jobs = new InMemoryWalletBalanceJobQueue({ now: () => 100 });
|
||||
const publication = new InMemoryWalletBalancePublicationPort(jobs, repository);
|
||||
let authorityUser = '';
|
||||
const authority = { currentAuthority: async (userId: string, characterId: number) => { authorityUser = userId; return repository.currentAuthority(userId, characterId); }, credential: repository.credential.bind(repository) };
|
||||
const adapter = new WalletBalanceJobAdapter({ jobs, authority, publication, transport: { getWalletBalance: async () => ({ amount: '1', retrievedAt: 100, expiresAt: 1_000 }) }, now: () => 100 });
|
||||
await jobs.enqueue({ userId: 'tenant-a', characterId: 13, fence: fence(value), now: 100 });
|
||||
const claim = await jobs.claim('worker', 100);
|
||||
const forged = { ...claim!, owner: { kind: 'user' as const, userId: 'tenant-b' }, fence: { ...claim!.fence, userId: 'tenant-b', principal: { ...claim!.fence.principal, ownerUserId: 'tenant-b' } } };
|
||||
const result = await adapter.runClaim(forged);
|
||||
expect(result.status).toBe('published');
|
||||
expect(authorityUser).toBe('tenant-a');
|
||||
expect((await repository.snapshot('tenant-a', 13))?.amount).toBe('1');
|
||||
expect(await repository.snapshot('tenant-b', 13)).toBeNull();
|
||||
});
|
||||
|
||||
test('publication requires exact snapshot fence pins and monotonic failure ordering', async () => {
|
||||
const value = resource('tenant-a', 14);
|
||||
const repository = new InMemoryWalletBalanceRepository({ resources: [value], snapshots: [snapshot(value, '10', 200, 1_000)] });
|
||||
const current = fence(value);
|
||||
const failure = { userId: 'tenant-a', characterId: 14, failedAt: 300, category: 'upstream', retryable: true, authorizationRevision: 1, authorizationEpoch: 1, credentialVersion: 1, generation: 'wallet-failure-new', jobId: 'failure-new' };
|
||||
expect(await repository.publishFailure({ jobId: 'failure-new', workerId: 'worker', leaseToken: 'lease', fence: current, failure, publishedAt: 300 })).toMatchObject({ accepted: true });
|
||||
const mismatched = snapshot(value, '11', 400, 1_000, 'wallet-success-mismatch');
|
||||
mismatched.authorizationRevision = 2;
|
||||
expect(await repository.publishSnapshot({ jobId: 'success-mismatch', workerId: 'worker', leaseToken: 'lease', fence: current, snapshot: mismatched, publishedAt: 400 })).toMatchObject({ accepted: false, reason: 'mismatched_character' });
|
||||
expect(await repository.publishSnapshot({ jobId: 'success-old', workerId: 'worker', leaseToken: 'lease', fence: current, snapshot: snapshot(value, '11', 250, 1_000, 'wallet-success-old'), publishedAt: 400 })).toMatchObject({ accepted: false, reason: 'stale_generation' });
|
||||
const oldFailure = { ...failure, failedAt: 250, generation: 'wallet-failure-old', jobId: 'failure-old' };
|
||||
expect(await repository.publishFailure({ jobId: 'failure-old', workerId: 'worker', leaseToken: 'lease', fence: current, failure: oldFailure, publishedAt: 400 })).toMatchObject({ accepted: false, reason: 'stale_generation' });
|
||||
expect((await repository.failure('tenant-a', 14))?.generation).toBe('wallet-failure-new');
|
||||
});
|
||||
|
||||
test('publication uses the queue clock, so forged backdated timestamps cannot revive an expired lease', async () => {
|
||||
const value = resource('tenant-a', 15);
|
||||
const repository = new InMemoryWalletBalanceRepository({ resources: [value] });
|
||||
const jobs = new InMemoryWalletBalanceJobQueue({ leaseMs: 10, now: () => 200 });
|
||||
const publication = new InMemoryWalletBalancePublicationPort(jobs, repository);
|
||||
await jobs.enqueue({ userId: 'tenant-a', characterId: 15, fence: fence(value), now: 100 });
|
||||
const claim = await jobs.claim('worker', 100);
|
||||
expect(claim).not.toBeNull();
|
||||
expect(await publication.publishSuccess({ jobId: claim!.id, workerId: claim!.workerId, leaseToken: claim!.leaseToken, fence: claim!.fence, snapshot: snapshot(value, '1', 100, 1_000), publishedAt: 100 })).toMatchObject({ accepted: false, reason: 'expired_job' });
|
||||
});
|
||||
|
||||
test('rate-limit cooldown is stored and non-retryable failures suppress automatic enqueue', async () => {
|
||||
const limited = resource('tenant-a', 16);
|
||||
const limitedRepository = new InMemoryWalletBalanceRepository({ resources: [limited], credentials: [credential(limited)] });
|
||||
const limitedJobs = new InMemoryWalletBalanceJobQueue({ now: () => 500 });
|
||||
const limitedPublication = new InMemoryWalletBalancePublicationPort(limitedJobs, limitedRepository);
|
||||
const limitedAdapter = new WalletBalanceJobAdapter({ jobs: limitedJobs, authority: limitedRepository, publication: limitedPublication, transport: { getWalletBalance: async () => { throw new EsiRateLimitError('/wallet', 1_000, 500, 'wallet-group'); } }, now: () => 500 });
|
||||
await limitedJobs.enqueue({ userId: 'tenant-a', characterId: 16, fence: fence(limited), now: 500 });
|
||||
expect((await limitedAdapter.runNext('worker')).status).toBe('failed');
|
||||
expect(await limitedJobs.cooldown('tenant-a', 16, 500)).toMatchObject({ until: 1_000, group: 'wallet-group' });
|
||||
|
||||
const denied = resource('tenant-a', 17);
|
||||
const deniedRepository = new InMemoryWalletBalanceRepository({ resources: [denied], credentials: [credential(denied)] });
|
||||
const deniedJobs = new InMemoryWalletBalanceJobQueue({ now: () => 500 });
|
||||
const deniedPublication = new InMemoryWalletBalancePublicationPort(deniedJobs, deniedRepository);
|
||||
const deniedAdapter = new WalletBalanceJobAdapter({ jobs: deniedJobs, authority: deniedRepository, publication: deniedPublication, transport: { getWalletBalance: async () => { throw new EsiError('authorization_context', 'denied'); } }, now: () => 500 });
|
||||
await deniedJobs.enqueue({ userId: 'tenant-a', characterId: 17, fence: fence(denied), now: 500 });
|
||||
await deniedAdapter.runNext('worker');
|
||||
const deniedView = await new WalletBalanceService({ repository: deniedRepository, queue: deniedJobs, now: () => 500 }).read({ userId: 'tenant-a' });
|
||||
expect(deniedView.refresh.attempted).toBe(0);
|
||||
});
|
||||
|
||||
test('lease heartbeat renews long transport and lost renewal forbids publication', async () => {
|
||||
let clock = 100;
|
||||
const value = resource('tenant-a', 18);
|
||||
const repository = new InMemoryWalletBalanceRepository({ resources: [value], credentials: [credential(value)] });
|
||||
const jobs = new InMemoryWalletBalanceJobQueue({ leaseMs: 30, now: () => clock });
|
||||
const publication = new InMemoryWalletBalancePublicationPort(jobs, repository);
|
||||
let release!: () => void;
|
||||
const adapter = new WalletBalanceJobAdapter({ jobs, authority: repository, publication, heartbeatMs: 5, now: () => clock, transport: { getWalletBalance: async () => { await new Promise<void>((resolve) => { release = resolve; }); return { amount: '18', retrievedAt: 100, expiresAt: 1_000 }; } } });
|
||||
const enqueued = await jobs.enqueue({ userId: 'tenant-a', characterId: 18, fence: fence(value), now: 100 });
|
||||
const running = adapter.runNext('worker');
|
||||
await Bun.sleep(8);
|
||||
clock = 120;
|
||||
await Bun.sleep(12);
|
||||
expect((await jobs.get(enqueued.status === 'queued' ? enqueued.job.id : ''))?.leaseExpiresAt).toBeGreaterThan(130);
|
||||
release();
|
||||
expect((await running).status).toBe('published');
|
||||
|
||||
const lost = resource('tenant-a', 19);
|
||||
const lostRepository = new InMemoryWalletBalanceRepository({ resources: [lost], credentials: [credential(lost)] });
|
||||
const lostJobs = new InMemoryWalletBalanceJobQueue({ leaseMs: 30, now: () => 100 });
|
||||
const lostPublication = new InMemoryWalletBalancePublicationPort(lostJobs, lostRepository);
|
||||
const originalRenew = lostJobs.renew.bind(lostJobs);
|
||||
let renewals = 0;
|
||||
lostJobs.renew = async (claim, now, leaseMs) => { renewals += 1; if (renewals > 1) return false; return originalRenew(claim, now, leaseMs); };
|
||||
const lostAdapter = new WalletBalanceJobAdapter({ jobs: lostJobs, authority: lostRepository, publication: lostPublication, heartbeatMs: 5, now: () => 100, transport: { getWalletBalance: async () => { await Bun.sleep(20); return { amount: '19', retrievedAt: 100, expiresAt: 1_000 }; } } });
|
||||
await lostJobs.enqueue({ userId: 'tenant-a', characterId: 19, fence: fence(lost), now: 100 });
|
||||
expect((await lostAdapter.runNext('worker')).status).toBe('rejected');
|
||||
expect(await lostRepository.snapshot('tenant-a', 19)).toBeNull();
|
||||
});
|
||||
|
||||
test('raw wallet amounts and selector cardinality are bounded before arithmetic/traversal', async () => {
|
||||
const value = resource('tenant-a', 20);
|
||||
expect(() => new InMemoryWalletBalanceRepository({ resources: [value], snapshots: [snapshot(value, '-0')] })).toThrow();
|
||||
const repository = new InMemoryWalletBalanceRepository({ resources: [value] });
|
||||
await expect(new WalletBalanceService({ repository, now: () => 100 }).read({ userId: 'tenant-a', characterIds: Array.from({ length: 101 }, (_, index) => String(index + 1)) })).rejects.toThrow('cap');
|
||||
});
|
||||
|
||||
test('service rejects malicious persisted future snapshot provenance', async () => {
|
||||
const value = resource('tenant-a', 23);
|
||||
const repository = new InMemoryWalletBalanceRepository({ resources: [value] });
|
||||
repository.snapshot = async () => snapshot(value, '23', 1_000, 2_000, 'future-generation');
|
||||
await expect(new WalletBalanceService({ repository, now: () => 500 }).read({ userId: 'tenant-a' })).rejects.toThrow('future');
|
||||
});
|
||||
|
||||
test('malformed required scopes fail before any refresh enqueue', async () => {
|
||||
const value = resource('tenant-a', 25);
|
||||
const repository = new InMemoryWalletBalanceRepository({ resources: [value] });
|
||||
repository.authorizedResources = async () => [{ ...value, resource: { ...value.resource, requiredScopes: [ESI_WALLET_SCOPE, null as unknown as string] } }];
|
||||
const queue = new InMemoryWalletBalanceJobQueue({ now: () => 100 });
|
||||
let enqueues = 0;
|
||||
queue.enqueue = async () => { enqueues += 1; throw new Error('must not enqueue'); };
|
||||
await expect(new WalletBalanceService({ repository, queue, now: () => 100 }).read({ userId: 'tenant-a' })).rejects.toThrow('scopes');
|
||||
expect(enqueues).toBe(0);
|
||||
});
|
||||
|
||||
test('malformed credential scopes and future transport observations fail closed', async () => {
|
||||
const value = resource('tenant-a', 21);
|
||||
const repository = new InMemoryWalletBalanceRepository({ resources: [value], credentials: [credential(value)] });
|
||||
const jobs = new InMemoryWalletBalanceJobQueue({ now: () => 100 });
|
||||
const publication = new InMemoryWalletBalancePublicationPort(jobs, repository);
|
||||
const bad = credential(value).credential;
|
||||
const authority = { currentAuthority: repository.currentAuthority.bind(repository), credential: async () => ({ ...bad, authorizationKey: { ...bad.authorizationKey, scopes: [ESI_WALLET_SCOPE, ESI_WALLET_SCOPE] } }) };
|
||||
const badAdapter = new WalletBalanceJobAdapter({ jobs, authority, publication, now: () => 100, transport: { getWalletBalance: async () => ({ amount: '1', retrievedAt: 100, expiresAt: 1_000 }) } });
|
||||
const badEnqueued = await jobs.enqueue({ userId: 'tenant-a', characterId: 21, fence: fence(value), now: 100 });
|
||||
expect((await badAdapter.runNext('worker')).status).toBe('rejected');
|
||||
expect((await jobs.get(badEnqueued.status === 'queued' ? badEnqueued.job.id : ''))?.status).toBe('quarantined');
|
||||
|
||||
const future = resource('tenant-a', 22);
|
||||
const futureRepository = new InMemoryWalletBalanceRepository({ resources: [future], credentials: [credential(future)] });
|
||||
const futureJobs = new InMemoryWalletBalanceJobQueue({ now: () => 100 });
|
||||
const futurePublication = new InMemoryWalletBalancePublicationPort(futureJobs, futureRepository);
|
||||
const futureAdapter = new WalletBalanceJobAdapter({ jobs: futureJobs, authority: futureRepository, publication: futurePublication, now: () => 100, transport: { getWalletBalance: async () => ({ amount: '1', retrievedAt: 101, expiresAt: 1_000 }) } });
|
||||
await futureJobs.enqueue({ userId: 'tenant-a', characterId: 22, fence: fence(future), now: 100 });
|
||||
expect((await futureAdapter.runNext('worker')).status).toBe('failed');
|
||||
expect(await futureRepository.snapshot('tenant-a', 22)).toBeNull();
|
||||
});
|
||||
|
||||
test('malformed accepted publication pairing is rejected without storing data', async () => {
|
||||
const value = resource('tenant-a', 24);
|
||||
const repository = new InMemoryWalletBalanceRepository({ resources: [value], credentials: [credential(value)] });
|
||||
const jobs = new InMemoryWalletBalanceJobQueue({ now: () => 100 });
|
||||
const malformedPublication = {
|
||||
publishSuccess: async () => ({ accepted: true, outcome: 'failure', snapshot: snapshot(value, '24', 100, 1_000) } as never),
|
||||
publishFailure: async () => ({ accepted: true, outcome: 'success', snapshot: null } as never),
|
||||
};
|
||||
const adapter = new WalletBalanceJobAdapter({ jobs, authority: repository, publication: malformedPublication, now: () => 100, transport: { getWalletBalance: async () => ({ amount: '24', retrievedAt: 100, expiresAt: 1_000 }) } });
|
||||
const enqueued = await jobs.enqueue({ userId: 'tenant-a', characterId: 24, fence: fence(value), now: 100 });
|
||||
expect((await adapter.runNext('worker')).status).toBe('unavailable');
|
||||
expect(await repository.snapshot('tenant-a', 24)).toBeNull();
|
||||
expect((await jobs.get(enqueued.status === 'queued' ? enqueued.job.id : ''))?.status).toBe('queued');
|
||||
});
|
||||
|
||||
test('duplicate and expired publication cannot overwrite a newer snapshot', async () => {
|
||||
const value = resource('tenant-a', 10);
|
||||
const repository = new InMemoryWalletBalanceRepository({ resources: [value], credentials: [credential(value)] });
|
||||
const jobs = new InMemoryWalletBalanceJobQueue({ leaseMs: 10, now: () => 100 });
|
||||
const publication = new InMemoryWalletBalancePublicationPort(jobs, repository);
|
||||
const adapter = new WalletBalanceJobAdapter({ jobs, authority: repository, publication, transport: { getWalletBalance: async () => ({ amount: '1', retrievedAt: 100, expiresAt: 1_000 }) }, now: () => 100 });
|
||||
const queued = await jobs.enqueue({ userId: 'tenant-a', characterId: 10, fence: fence(value), now: 100 });
|
||||
expect(queued.status).toBe('queued');
|
||||
const claim = await jobs.claim('worker', 100);
|
||||
expect(claim).not.toBeNull();
|
||||
const first = await adapter.runClaim(claim!);
|
||||
expect(first.status).toBe('published');
|
||||
// A second publication for the same job is rejected by the completed lease.
|
||||
if (first.status === 'published') {
|
||||
const duplicate = await publication.publishSuccess({ jobId: first.job.id, workerId: 'worker', leaseToken: claim!.leaseToken, fence: fence(value), snapshot: snapshot(value, '999', 101, 2_000, 'wallet-duplicate'), publishedAt: 101 });
|
||||
expect(duplicate).toMatchObject({ accepted: false, reason: 'expired_job' });
|
||||
const directDuplicate = await repository.publishSnapshot({ jobId: first.job.id, workerId: 'worker', leaseToken: 'bypassed', fence: fence(value), snapshot: snapshot(value, '999', 101, 2_000, 'wallet-duplicate'), publishedAt: 101 });
|
||||
expect(directDuplicate).toMatchObject({ accepted: false, reason: 'duplicate_publication' });
|
||||
const directOutcomeChange = await repository.publishFailure({ jobId: first.job.id, workerId: 'worker', leaseToken: 'bypassed', fence: fence(value), failure: { userId: value.userId, characterId: value.characterId, failedAt: 101, category: 'changed-outcome', retryable: true, authorizationRevision: 1, authorizationEpoch: 1, credentialVersion: 1, generation: 'wallet-outcome-change', jobId: first.job.id }, publishedAt: 101 });
|
||||
expect(directOutcomeChange).toMatchObject({ accepted: false, reason: 'duplicate_publication' });
|
||||
}
|
||||
const old = await repository.publishSnapshot({ jobId: 'old', workerId: 'worker', leaseToken: 'none', fence: fence(value), snapshot: snapshot(value, '0', 99, 1_000, 'wallet-old'), publishedAt: 101 });
|
||||
expect(old).toMatchObject({ accepted: false, reason: 'stale_generation' });
|
||||
const stored = await repository.snapshot('tenant-a', 10);
|
||||
expect(stored).not.toBeNull();
|
||||
const retry = await repository.publishSnapshot({ jobId: 'retry-identical', workerId: 'worker', leaseToken: 'none', fence: fence(value), snapshot: stored!, publishedAt: 101 });
|
||||
expect(retry).toMatchObject({ accepted: true, outcome: 'success' });
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user