312 lines
25 KiB
TypeScript
312 lines
25 KiB
TypeScript
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' });
|
|
});
|
|
});
|