diff --git a/src/lib/server/wallet/balance-job.ts b/src/lib/server/wallet/balance-job.ts new file mode 100644 index 0000000..d767661 --- /dev/null +++ b/src/lib/server/wallet/balance-job.ts @@ -0,0 +1,526 @@ +import { randomUUID } from 'node:crypto'; +import type { EsiWalletBalanceObservation } from '../esi'; +import { ESI_WALLET_SCOPE, EsiError } from '../esi'; +import { parseIsk } from '../../domain/money'; +import { + sameWalletFence, + validateWalletAmount, + validateWalletAuthorityFence, + validateWalletSnapshot, + type WalletAuthorityFence, + type WalletBalanceRepository, + type WalletBalanceSnapshot, + type WalletCredential, + type WalletFailurePublication, + type WalletPublicationResult, + type WalletSnapshotPublication, +} from './balance-repository'; + +export const WALLET_BALANCE_DATASET = 'wallet_balance' as const; +export const DEFAULT_WALLET_JOB_LEASE_MS = 60_000; +export const DEFAULT_WALLET_QUEUE_LIMIT = 100; +export const DEFAULT_WALLET_COOLDOWN_MS = 0; + +export type WalletJobStatus = 'queued' | 'running' | 'succeeded' | 'failed' | 'cancelled' | 'quarantined'; + +/** A serialized, character-scoped job. The fence is the only job authority. */ +export type WalletBalanceJob = { + id: string; + dataset: typeof WALLET_BALANCE_DATASET; + owner: { kind: 'user'; userId: string }; + authority: 'user'; + characterId: number; + fence: WalletAuthorityFence; + status: WalletJobStatus; + attempts: number; + dueAt: number; + createdAt: number; + updatedAt: number; + leaseExpiresAt: number | null; + workerId: string | null; + leaseToken: string | null; + lastError?: string; +}; + +export type WalletEnqueueRequest = { + userId: string; + characterId: number; + fence: WalletAuthorityFence; + dueAt?: number; + now?: number; +}; + +export type WalletEnqueueResult = + | { status: 'queued'; job: WalletBalanceJob } + | { status: 'duplicate'; job: WalletBalanceJob } + | { status: 'cooldown'; job: null; dueAt: number } + | { status: 'queue_full'; job: null; dueAt: number }; + +export type WalletBalanceClaim = WalletBalanceJob & { + status: 'running'; + workerId: string; + leaseToken: string; + leaseExpiresAt: number; +}; + +export interface WalletBalanceJobQueue { + enqueue(request: WalletEnqueueRequest): Promise; + claim(workerId: string, now?: number): Promise; + renew(claim: Pick, now?: number, leaseMs?: number): Promise; + complete(claim: Pick, status: 'succeeded' | 'failed' | 'cancelled' | 'quarantined', now?: number, error?: string): Promise; + defer(claim: Pick, dueAt: number, now?: number): Promise; + isLive(claim: Pick, now?: number): Promise; + /** Returns the canonical stored claim; caller-supplied owner/fence fields are never trusted. */ + canonicalClaim(claim: Pick, now?: number): Promise; + cooldown(userId: string, characterId: number, now?: number): Promise<{ until: number; group: string } | null>; + setCooldown(userId: string, characterId: number, until: number, group?: string): Promise; + /** Server/adapter clock used by atomic fixture publication; event timestamps are not authority. */ + now(): number; + cooldownAfterSuccess(userId: string, characterId: number, now?: number): Promise; + get(jobId: string): Promise; +} + +/** Typed transport shape matching EsiClient.getWalletBalance; no URL or OAuth logic is added. */ +export type WalletBalanceTransport = { + getWalletBalance(characterId: number, accessToken: string, authorizationKey: import('../esi').AuthorizationKey, signal?: AbortSignal): Promise; +}; + +/** A repository/credential backend was unavailable; this is not an auth decision. */ +export class WalletAuthorityUnavailableError extends Error { + public readonly code = 'authority_unavailable'; + public readonly retryable = true; + + public constructor(message = 'wallet authority backend is unavailable') { + super(message); + this.name = 'WalletAuthorityUnavailableError'; + } +} + +/** A worker lease was lost while remote I/O was in flight; publication is forbidden. */ +export class WalletLeaseLostError extends Error { + public readonly code = 'lease_lost'; + + public constructor() { + super('wallet worker lease was lost before publication'); + this.name = 'WalletLeaseLostError'; + } +} + +/** Reads current authority and an already-issued credential; it cannot grant scopes. */ +export type WalletBalanceAuthorityPort = Pick; +/** Atomic publication seam for a durable queue/store adapter. */ +export type WalletBalancePublicationPort = { + publishSuccess(input: WalletSnapshotPublication): Promise; + publishFailure(input: WalletFailurePublication): Promise; +}; + +export type WalletBalanceJobAdapterOptions = { + jobs: WalletBalanceJobQueue; + authority: WalletBalanceAuthorityPort; + publication: WalletBalancePublicationPort; + transport: WalletBalanceTransport; + now?: () => number; + leaseMs?: number; + /** Renew at most this often while transport/publication I/O is pending. */ + heartbeatMs?: number; +}; + +export type WalletJobRunResult = + | { status: 'idle' } + | { status: 'published'; job: WalletBalanceJob; publication: WalletPublicationResult } + | { status: 'rejected'; job: WalletBalanceJob; reason: string } + | { status: 'unavailable'; job: WalletBalanceJob; error: string } + | { status: 'failed'; job: WalletBalanceJob; error: string }; + +function validNow(value: number): number { + if (!Number.isFinite(value) || value < 0) throw new RangeError('wallet job clock returned an invalid timestamp'); + return value; +} + +function cloneFence(fence: WalletAuthorityFence): WalletAuthorityFence { + return { ...fence, principal: { ...fence.principal }, scopes: [...fence.scopes] }; +} + +function validateFence(fence: WalletAuthorityFence): void { + try { validateWalletAuthorityFence(fence); } catch { throw new TypeError('wallet job authority fence is invalid'); } +} + +function validateJob(job: WalletBalanceJob): void { + if (!job || typeof job.id !== 'string' || !job.id || job.dataset !== WALLET_BALANCE_DATASET || job.authority !== 'user' || job.owner.kind !== 'user' || job.owner.userId !== job.fence.userId || job.characterId !== job.fence.characterId || !Number.isSafeInteger(job.attempts) || job.attempts < 0 || !Number.isFinite(job.dueAt) || !Number.isFinite(job.createdAt) || !Number.isFinite(job.updatedAt)) throw new TypeError('wallet job is invalid'); + validateFence(job.fence); + if (!['queued', 'running', 'succeeded', 'failed', 'cancelled', 'quarantined'].includes(job.status)) throw new TypeError('wallet job status is invalid'); +} + +function claimMatches(job: WalletBalanceJob | null, claim: Pick, now: number): job is WalletBalanceClaim { + return !!job && job.status === 'running' && job.id === claim.id && job.workerId === claim.workerId && job.leaseToken === claim.leaseToken && job.leaseExpiresAt !== null && job.leaseExpiresAt > now; +} + +/** + * Fixture-only bounded queue. A durable queue must serialize the same dedupe, + * lease, cooldown and publication transitions instead of copying this Map. + */ +export class InMemoryWalletBalanceJobQueue implements WalletBalanceJobQueue { + private readonly jobs = new Map(); + private readonly activeByResource = new Map(); + private readonly cooldowns = new Map(); + private readonly maxPending: number; + private readonly leaseMs: number; + private readonly cooldownMs: number; + private readonly clock: () => number; + + public constructor(options: { maxPending?: number; leaseMs?: number; cooldownMs?: number; now?: () => number } = {}) { + this.maxPending = options.maxPending ?? DEFAULT_WALLET_QUEUE_LIMIT; + this.leaseMs = options.leaseMs ?? DEFAULT_WALLET_JOB_LEASE_MS; + this.cooldownMs = options.cooldownMs ?? DEFAULT_WALLET_COOLDOWN_MS; + this.clock = options.now ?? (() => Date.now()); + if (!Number.isSafeInteger(this.maxPending) || this.maxPending < 1 || !Number.isSafeInteger(this.leaseMs) || this.leaseMs < 1 || !Number.isSafeInteger(this.cooldownMs) || this.cooldownMs < 0) throw new RangeError('wallet queue bounds are invalid'); + } + + public async enqueue(request: WalletEnqueueRequest): Promise { + validateFence(request.fence); + if (typeof request.userId !== 'string' || !request.userId || request.userId !== request.fence.userId || request.characterId !== request.fence.characterId) throw new TypeError('wallet enqueue identity does not match fence'); + const now = validNow(request.now ?? this.now()); + const dueAt = request.dueAt === undefined ? now : validNow(request.dueAt); + const key = `${request.userId}\u0000${request.characterId}`; + const cooldown = this.cooldowns.get(key); + if (cooldown && cooldown.until > now) return { status: 'cooldown', job: null, dueAt: cooldown.until }; + const activeId = this.activeByResource.get(key); + if (activeId) { + const active = this.jobs.get(activeId); + if (active && (active.status === 'queued' || active.status === 'running')) return { status: 'duplicate', job: structuredClone(active) }; + this.activeByResource.delete(key); + } + const pending = [...this.jobs.values()].filter((job) => job.status === 'queued' || job.status === 'running').length; + if (pending >= this.maxPending) return { status: 'queue_full', job: null, dueAt }; + const job: WalletBalanceJob = { + id: randomUUID(), dataset: WALLET_BALANCE_DATASET, owner: { kind: 'user', userId: request.userId }, authority: 'user', characterId: request.characterId, + fence: cloneFence(request.fence), status: 'queued', attempts: 0, dueAt, createdAt: now, updatedAt: now, leaseExpiresAt: null, workerId: null, leaseToken: null, + }; + this.jobs.set(job.id, job); + this.activeByResource.set(key, job.id); + return { status: 'queued', job: structuredClone(job) }; + } + + public now(): number { return validNow(this.clock()); } + + public async claim(workerId: string, now = this.now()): Promise { + if (typeof workerId !== 'string' || !workerId) throw new TypeError('wallet worker ID is invalid'); + const at = validNow(now); + for (const job of this.jobs.values()) { + if (job.status === 'running' && (job.leaseExpiresAt === null || job.leaseExpiresAt <= at)) { + job.status = 'failed'; job.updatedAt = at; job.lastError = 'worker lease expired'; job.leaseExpiresAt = null; job.workerId = null; job.leaseToken = null; this.activeByResource.delete(`${job.owner.userId}\u0000${job.characterId}`); + } + } + const candidate = [...this.jobs.values()].filter((job) => job.status === 'queued' && job.dueAt <= at).sort((a, b) => a.dueAt - b.dueAt || a.createdAt - b.createdAt)[0]; + if (!candidate) return null; + candidate.status = 'running'; candidate.attempts += 1; candidate.updatedAt = at; candidate.workerId = workerId; candidate.leaseToken = randomUUID(); candidate.leaseExpiresAt = at + this.leaseMs; + return structuredClone(candidate) as WalletBalanceClaim; + } + + public async renew(claim: Pick, now = this.now(), leaseMs = this.leaseMs): Promise { + const at = validNow(now); + const job = this.jobs.get(claim.id) ?? null; + if (!claimMatches(job, claim, at) || !Number.isSafeInteger(leaseMs) || leaseMs < 1) return false; + job.leaseExpiresAt = at + leaseMs; job.updatedAt = at; + return true; + } + + public async complete(claim: Pick, status: Exclude, now = this.now(), error?: string): Promise { + const at = validNow(now); + const job = this.jobs.get(claim.id) ?? null; + if (!claimMatches(job, claim, at)) return false; + const mutable: WalletBalanceJob = job; + mutable.status = status; mutable.updatedAt = at; mutable.leaseExpiresAt = null; mutable.workerId = null; mutable.leaseToken = null; if (error) mutable.lastError = error; + this.activeByResource.delete(`${mutable.owner.userId}\u0000${mutable.characterId}`); + return true; + } + + public async defer(claim: Pick, dueAt: number, now = this.now()): Promise { + const at = validNow(now); const target = validNow(dueAt); const job = this.jobs.get(claim.id) ?? null; + if (!claimMatches(job, claim, at)) return false; + const mutable: WalletBalanceJob = job; + mutable.status = 'queued'; mutable.dueAt = target; mutable.updatedAt = at; mutable.leaseExpiresAt = null; mutable.workerId = null; mutable.leaseToken = null; + return true; + } + + public async isLive(claim: Pick, now = this.now()): Promise { + const at = validNow(now); return claimMatches(this.jobs.get(claim.id) ?? null, claim, at); + } + + public async cooldown(userId: string, characterId: number, now = this.now()): Promise<{ until: number; group: string } | null> { + const at = validNow(now); const key = `${userId}\u0000${characterId}`; const value = this.cooldowns.get(key); + if (!value || value.until <= at) return null; + return { ...value }; + } + + /** Fixture/test hook: defer future automatic enqueues without changing auth. */ + public async setCooldown(userId: string, characterId: number, until: number, group = 'wallet'): Promise { + const value = validNow(until); + this.cooldowns.set(`${userId}\u0000${characterId}`, { until: value, group }); + } + + public async canonicalClaim(claim: Pick, now = this.now()): Promise { + const at = validNow(now); + const job = this.jobs.get(claim.id) ?? null; + if (!claimMatches(job, claim, at)) return null; + validateJob(job); + return structuredClone(job) as WalletBalanceClaim; + } + + public async get(jobId: string): Promise { + const job = this.jobs.get(jobId); return job ? structuredClone(job) : null; + } + + /** Called by the publication adapter after a successful remote read. */ + public async cooldownAfterSuccess(userId: string, characterId: number, now = this.now()): Promise { + if (this.cooldownMs > 0) await this.setCooldown(userId, characterId, validNow(now) + this.cooldownMs, 'wallet'); + } +} + +/** + * Fixture publication adapter: it composes the queue lease fence with the + * repository's atomic authority/publication methods. Durable implementations + * should perform this transition in one serialized checkpoint. + */ +export class InMemoryWalletBalancePublicationPort implements WalletBalancePublicationPort { + public constructor(private readonly jobs: WalletBalanceJobQueue, private readonly repository: WalletBalanceRepository) {} + + public async publishSuccess(input: WalletSnapshotPublication): Promise { + try { validateWalletAuthorityFence(input.fence); } catch { return { accepted: false, reason: 'borrowed_resource' }; } + const serverNow = this.jobs.now(); + const claim = await this.jobs.canonicalClaim({ id: input.jobId, workerId: input.workerId, leaseToken: input.leaseToken }, serverNow); + if (!claim) return { accepted: false, reason: 'expired_job' }; + if (claim.characterId !== input.fence.characterId || claim.owner.userId !== input.fence.userId || !sameWalletFence(claim.fence, input.fence)) return { accepted: false, reason: 'borrowed_resource' }; + const publication = { ...input, workerId: claim.workerId, leaseToken: claim.leaseToken, fence: claim.fence, publishedAt: serverNow }; + let result: WalletPublicationResult; + try { result = await this.repository.publishSnapshot(publication); validatePublicationResult(result, 'success', publication); } catch { await this.jobs.complete({ id: claim.id, workerId: claim.workerId, leaseToken: claim.leaseToken }, 'failed', serverNow, 'malformed_publication'); return { accepted: false, reason: 'invalid_snapshot' }; } + if (result.accepted) { + await this.jobs.complete({ id: claim.id, workerId: claim.workerId, leaseToken: claim.leaseToken }, 'succeeded', serverNow); + await this.jobs.cooldownAfterSuccess(claim.owner.userId, claim.characterId, serverNow); + } else await this.jobs.complete({ id: claim.id, workerId: claim.workerId, leaseToken: claim.leaseToken }, 'failed', serverNow, result.reason); + return result; + } + + public async publishFailure(input: WalletFailurePublication): Promise { + try { validateWalletAuthorityFence(input.fence); } catch { return { accepted: false, reason: 'borrowed_resource' }; } + const serverNow = this.jobs.now(); + const claim = await this.jobs.canonicalClaim({ id: input.jobId, workerId: input.workerId, leaseToken: input.leaseToken }, serverNow); + if (!claim) return { accepted: false, reason: 'expired_job' }; + if (claim.characterId !== input.fence.characterId || claim.owner.userId !== input.fence.userId || !sameWalletFence(claim.fence, input.fence)) return { accepted: false, reason: 'borrowed_resource' }; + const publication = { ...input, workerId: claim.workerId, leaseToken: claim.leaseToken, fence: claim.fence, publishedAt: serverNow }; + let result: WalletPublicationResult; + try { result = await this.repository.publishFailure(publication); validatePublicationResult(result, 'failure', publication); } catch { await this.jobs.complete({ id: claim.id, workerId: claim.workerId, leaseToken: claim.leaseToken }, 'failed', serverNow, 'malformed_publication'); return { accepted: false, reason: 'invalid_snapshot' }; } + await this.jobs.complete({ id: claim.id, workerId: claim.workerId, leaseToken: claim.leaseToken }, 'failed', serverNow, input.failure.category); + if (result.accepted && input.failure.retryAt !== undefined) await this.jobs.setCooldown(claim.owner.userId, claim.characterId, input.failure.retryAt, input.failure.cooldownGroup ?? 'wallet'); + return result; + } +} + +function validateObservation(observation: EsiWalletBalanceObservation, now: number): void { + if (!observation || typeof observation !== 'object') throw new TypeError('wallet transport response is invalid'); + validateWalletAmount(observation.amount); + if (!Number.isFinite(observation.retrievedAt) || observation.retrievedAt < 0 || observation.retrievedAt > now || !Number.isFinite(observation.expiresAt) || observation.expiresAt <= observation.retrievedAt) throw new TypeError('wallet transport timestamps are invalid'); + // A transport cannot claim a negative/unknown expiry, but upstream effective + // time remains null because ESI does not provide one for this endpoint. + if (observation.expiresAt <= now) throw new TypeError('wallet transport returned an already expired observation'); +} + +function exactScopes(left: unknown, right: readonly string[]): boolean { + if (!Array.isArray(left) || left.length !== right.length || left.length === 0 || left.length > 64 || left.some((scope) => typeof scope !== 'string' || !scope || scope.length > 256) || new Set(left).size !== left.length) return false; + return [...left].sort().join('\u0000') === [...right].sort().join('\u0000'); +} + +function validCredential(value: WalletCredential | null, fence: WalletAuthorityFence): value is WalletCredential { + if (!value || typeof value !== 'object' || typeof value.accessToken !== 'string' || !value.accessToken || !value.authorizationKey || typeof value.authorizationKey !== 'object') return false; + const key = value.authorizationKey; + return key.userId === fence.userId && key.principal === `character:${fence.principal.id}` && key.characterId === fence.characterId && key.grantRevision === fence.grantRevision && key.authorizationEpoch === fence.authorizationEpoch && key.credentialVersion === fence.credentialVersion && exactScopes(key.scopes, fence.scopes); +} + +const PUBLICATION_REASONS = new Set(['borrowed_resource', 'stale_grant', 'stale_credential', 'stale_authorization_epoch', 'authorization_revoked', 'mismatched_character', 'invalid_snapshot', 'stale_generation', 'expired_job', 'duplicate_publication']); +function samePublishedSnapshot(left: WalletBalanceSnapshot, right: WalletBalanceSnapshot): boolean { + return left.userId === right.userId && left.characterId === right.characterId && left.principal.id === right.principal.id && left.principal.ownerUserId === right.principal.ownerUserId && left.amount === right.amount && left.fetchedAt === right.fetchedAt && left.expiresAt === right.expiresAt && left.upstreamEffectiveAt === right.upstreamEffectiveAt && left.generation === right.generation && left.authorizationRevision === right.authorizationRevision && left.authorizationEpoch === right.authorizationEpoch && left.credentialVersion === right.credentialVersion; +} +function validatePublicationResult(value: WalletPublicationResult, expected: 'success' | 'failure', requested: WalletSnapshotPublication | WalletFailurePublication): void { + if (!value || typeof value !== 'object' || typeof value.accepted !== 'boolean') throw new TypeError('wallet publication returned malformed result'); + if (value.accepted) { + if (value.outcome !== expected) throw new TypeError('wallet publication outcome does not match request'); + if (expected === 'success') { + if (value.snapshot === null) throw new TypeError('successful wallet publication omitted snapshot'); + validateWalletSnapshot(value.snapshot); + const input = requested as WalletSnapshotPublication; + if (value.snapshot.userId !== input.fence.userId || value.snapshot.characterId !== input.fence.characterId || value.snapshot.principal.id !== input.fence.principal.id || value.snapshot.authorizationRevision !== input.fence.grantRevision || value.snapshot.authorizationEpoch !== input.fence.authorizationEpoch || value.snapshot.credentialVersion !== input.fence.credentialVersion || !samePublishedSnapshot(value.snapshot, input.snapshot)) throw new TypeError('successful wallet publication snapshot does not match request'); + } else if (value.snapshot !== null) throw new TypeError('failure wallet publication returned a snapshot'); + } else if (!PUBLICATION_REASONS.has(value.reason)) throw new TypeError('wallet publication rejection is invalid'); +} + +function failureDetails(error: unknown): { category: string; message: string; retryable: boolean; retryAt?: number; cooldownGroup?: string } { + if (error instanceof EsiError) { + const cooldown = 'cooldown' in error && error.cooldown && typeof error.cooldown === 'object' ? error.cooldown as { until?: unknown; group?: unknown; context?: unknown } : undefined; + const cooldownGroup = typeof cooldown?.group === 'string' ? cooldown.group : typeof cooldown?.context === 'string' ? cooldown.context : undefined; + return { category: error.code, message: error.message, retryable: error.code !== 'authorization_context', ...(typeof cooldown?.until === 'number' && Number.isFinite(cooldown.until) ? { retryAt: cooldown.until } : {}), ...(cooldownGroup === undefined ? {} : { cooldownGroup }) }; + } + if (error instanceof Error) return { category: 'refresh_failed', message: 'wallet refresh failed', retryable: true }; + return { category: 'refresh_failed', message: 'wallet refresh failed', retryable: true }; +} + +/** Runs one claimed job and never broadens its OAuth grants. */ +export class WalletBalanceJobAdapter { + private readonly now: () => number; + private readonly leaseMs: number; + private readonly heartbeatMs: number; + + public constructor(private readonly options: WalletBalanceJobAdapterOptions) { + this.now = options.now ?? (() => Date.now()); + this.leaseMs = options.leaseMs ?? DEFAULT_WALLET_JOB_LEASE_MS; + this.heartbeatMs = options.heartbeatMs ?? Math.max(10, Math.floor(this.leaseMs / 3)); + if (!Number.isSafeInteger(this.leaseMs) || this.leaseMs < 1 || !Number.isSafeInteger(this.heartbeatMs) || this.heartbeatMs < 1 || this.heartbeatMs >= this.leaseMs) throw new RangeError('wallet job lease/heartbeat is invalid'); + } + + /** Keep remote I/O inside the current worker lease; a lost lease aborts and forbids publication. */ + private async withLease(claim: WalletBalanceClaim, work: (signal: AbortSignal) => Promise, checkAfter = true): Promise { + const controller = new AbortController(); + let lost = false; + const renew = async (): Promise => { + try { + const renewed = await this.options.jobs.renew(claim, this.options.jobs.now(), this.leaseMs); + if (!renewed) { + lost = true; + controller.abort(); + } + } catch { + lost = true; + controller.abort(); + } + }; + await renew(); + if (lost) throw new WalletLeaseLostError(); + const timer = setInterval(() => { void renew(); }, this.heartbeatMs); + try { + const result = await work(controller.signal); + if (lost || (checkAfter && !(await this.options.jobs.isLive(claim, this.options.jobs.now())))) throw new WalletLeaseLostError(); + return result; + } finally { + clearInterval(timer); + } + } + + private async unavailable(claim: WalletBalanceClaim, error?: unknown): Promise { + const at = this.options.jobs.now(); + const message = error instanceof WalletAuthorityUnavailableError ? error.message : 'wallet authority backend is unavailable'; + try { + await this.options.jobs.defer(claim, at + Math.min(this.leaseMs, 5_000), at); + } catch { + // The original unavailable classification remains safer than quarantining + // a resource when the queue backend itself cannot be reached. + } + return { status: 'unavailable', job: claim, error: message }; + } + + private async publishRefreshFailure(claim: WalletBalanceClaim, error: unknown): Promise { + if (error instanceof WalletLeaseLostError) return { status: 'rejected', job: claim, reason: 'lease_lost' }; + const details = failureDetails(error); + const failedAt = this.options.jobs.now(); + const failure: WalletFailurePublication = { + jobId: claim.id, workerId: claim.workerId, leaseToken: claim.leaseToken, + fence: claim.fence, + publishedAt: failedAt, + failure: { userId: claim.owner.userId, characterId: claim.characterId, failedAt, category: details.category, message: details.message, retryable: details.retryable, authorizationRevision: claim.fence.grantRevision, authorizationEpoch: claim.fence.authorizationEpoch, credentialVersion: claim.fence.credentialVersion, generation: `wallet-${claim.id}-${claim.attempts}`, ...(details.retryAt === undefined ? {} : { retryAt: details.retryAt }), ...(details.cooldownGroup === undefined ? {} : { cooldownGroup: details.cooldownGroup }), jobId: claim.id }, + }; + try { + const publication = await this.withLease(claim, async () => this.options.publication.publishFailure(failure), false); + validatePublicationResult(publication, 'failure', failure); + if (!publication.accepted && publication.reason === 'expired_job') return { status: 'rejected', job: claim, reason: publication.reason }; + return { status: 'failed', job: claim, error: details.message }; + } catch (publicationError) { + if (publicationError instanceof WalletLeaseLostError) return { status: 'rejected', job: claim, reason: 'lease_lost' }; + return this.unavailable(claim, publicationError); + } + } + + public async runNext(workerId: string): Promise { + const at = this.options.jobs.now(); + const claim = await this.options.jobs.claim(workerId, at); + if (!claim) return { status: 'idle' }; + return this.runClaim(claim); + } + + public async runClaim(claim: WalletBalanceClaim): Promise { + const startedAt = this.options.jobs.now(); + let canonical: WalletBalanceClaim | null; + try { + canonical = await this.options.jobs.canonicalClaim(claim, startedAt); + } catch { + return { status: 'unavailable', job: claim, error: 'wallet job authority is unavailable' }; + } + if (!canonical) return { status: 'rejected', job: claim, reason: 'expired_job' }; + try { validateJob(canonical); } catch { return this.unavailable(claim, new WalletAuthorityUnavailableError('wallet job response is malformed')); } + // Never use mutable caller-owned owner/character/fence fields after this point. + claim = canonical; + let current: WalletAuthorityFence | null; + try { + current = await this.options.authority.currentAuthority(claim.owner.userId, claim.characterId); + } catch (error) { + // A failed authority read is not evidence of revocation. Leave the job + // queued for a bounded retry instead of quarantining its resource. + return this.unavailable(claim, error); + } + if (current) { + try { validateWalletAuthorityFence(current); } catch (error) { return this.unavailable(claim, new WalletAuthorityUnavailableError('wallet authority response is malformed')); } + } + if (!current || !sameWalletFence(current, claim.fence)) { + const reason = !current ? 'authorization_revoked' : current.grantRevision !== claim.fence.grantRevision ? 'stale_grant' : current.authorizationEpoch !== claim.fence.authorizationEpoch ? 'stale_authorization_epoch' : current.credentialVersion !== claim.fence.credentialVersion ? 'stale_credential' : 'borrowed_resource'; + await this.options.jobs.complete(claim, 'quarantined', startedAt, reason); + return { status: 'rejected', job: claim, reason }; + } + let credential: WalletCredential | null; + try { + credential = await this.options.authority.credential(claim.fence); + } catch (error) { + return this.unavailable(claim, error); + } + if (!validCredential(credential, claim.fence)) { + await this.options.jobs.complete(claim, 'quarantined', startedAt, 'credential_authority_mismatch'); + return { status: 'rejected', job: claim, reason: 'credential_authority_mismatch' }; + } + + let observation: EsiWalletBalanceObservation; + try { + observation = await this.withLease(claim, (signal) => this.options.transport.getWalletBalance(claim.characterId, credential.accessToken, credential.authorizationKey, signal)); + } catch (error) { + return this.publishRefreshFailure(claim, error); + } + const observedAt = this.options.jobs.now(); + try { + validateObservation(observation, observedAt); + } catch (error) { + return this.publishRefreshFailure(claim, error); + } + const generation = `wallet-${claim.id}-${claim.attempts}`; + const snapshot: WalletBalanceSnapshot = { + userId: claim.owner.userId, + characterId: claim.characterId, + principal: { ...claim.fence.principal }, + amount: observation.amount, + fetchedAt: observation.retrievedAt, + expiresAt: observation.expiresAt, + upstreamEffectiveAt: null, + coverage: { kind: 'snapshot', generation, complete: true }, + generation, + authorizationRevision: claim.fence.grantRevision, + authorizationEpoch: claim.fence.authorizationEpoch, + credentialVersion: claim.fence.credentialVersion, + }; + const publication: WalletSnapshotPublication = { jobId: claim.id, workerId: claim.workerId, leaseToken: claim.leaseToken, fence: claim.fence, snapshot, publishedAt: observedAt }; + try { + const result = await this.withLease(claim, () => this.options.publication.publishSuccess({ ...publication, publishedAt: validNow(this.now()) }), false); + validatePublicationResult(result, 'success', { ...publication, publishedAt: validNow(this.now()) }); + if (!result.accepted) return { status: 'rejected', job: claim, reason: result.reason }; + return { status: 'published', job: claim, publication: result }; + } catch (error) { + if (error instanceof WalletLeaseLostError) return { status: 'rejected', job: claim, reason: 'lease_lost' }; + return this.unavailable(claim, error); + } + } +} + +export type WalletBalanceJobHandler = WalletBalanceJobAdapter; diff --git a/src/lib/server/wallet/balance-repository.ts b/src/lib/server/wallet/balance-repository.ts new file mode 100644 index 0000000..1e9d7e2 --- /dev/null +++ b/src/lib/server/wallet/balance-repository.ts @@ -0,0 +1,434 @@ +import type { AuthorizationKey } from '../esi'; +import { ESI_WALLET_SCOPE } from '../esi'; +import type { + AuthorizationStatus, + CompleteReportProvenance, + DatasetCoverage, + ResourcePrincipal, + SelectedResource, + WalletBalanceComponent, +} from '../../domain/model'; +import { isDecimalId } from '../../domain/inventory-location'; +import { parseIsk } from '../../domain/money'; + +/** A character resource resolved by the server, never by a browser selector. */ +export type WalletAuthorizedResource = { + userId: string; + characterId: number; + characterName?: string; + principal: Extract; + resource: SelectedResource; + authorizationStatus: AuthorizationStatus; + consent: 'granted' | 'missing'; + grantedScopes: readonly string[]; + grantRevision: number; + authorizationEpoch: number; + credentialVersion: number | null; +}; + +/** The immutable wallet snapshot kept independently of authorization metadata. */ +export type WalletBalanceSnapshot = { + userId: string; + characterId: number; + principal: Extract; + amount: string; + fetchedAt: number; + expiresAt: number; + upstreamEffectiveAt: number | null; + coverage: Extract; + authorizationRevision: number; + authorizationEpoch: number; + credentialVersion: number; + generation: string; +}; + +export type WalletRefreshFailure = { + userId: string; + characterId: number; + failedAt: number; + category: string; + retryable: boolean; + authorizationRevision: number; + authorizationEpoch: number; + credentialVersion: number; + /** Monotonic job attempt identity; content cannot mutate under one generation. */ + generation: string; + retryAt?: number; + cooldownGroup?: string; + message?: string; + jobId?: string; +}; + +/** The complete authority fence copied into a job and checked at publication. */ +export type WalletAuthorityFence = { + userId: string; + characterId: number; + principal: Extract; + grantRevision: number; + authorizationEpoch: number; + credentialVersion: number; + scopes: readonly string[]; +}; + +export type WalletCredential = { + accessToken: string; + authorizationKey: AuthorizationKey; +}; + +export type WalletSnapshotPublication = { + jobId: string; + workerId: string; + leaseToken: string; + fence: WalletAuthorityFence; + snapshot: WalletBalanceSnapshot; + publishedAt: number; +}; + +export type WalletFailurePublication = { + jobId: string; + workerId: string; + leaseToken: string; + fence: WalletAuthorityFence; + failure: WalletRefreshFailure; + publishedAt: number; +}; + +export type WalletPublicationRejection = + | 'borrowed_resource' + | 'stale_grant' + | 'stale_credential' + | 'stale_authorization_epoch' + | 'authorization_revoked' + | 'mismatched_character' + | 'invalid_snapshot' + | 'stale_generation' + | 'expired_job' + | 'duplicate_publication'; + +export type WalletPublicationResult = + | { accepted: true; snapshot: WalletBalanceSnapshot | null; outcome: 'success' | 'failure' } + | { accepted: false; reason: WalletPublicationRejection }; + +/** + * Persistence/runtime seam owned by a later serialized checkpoint. Implementations + * must make authority checks and snapshot publication atomic. No OAuth behavior is + * defined here; credential() only reads an already-authorized credential. + */ +export interface WalletBalanceRepository { + authorizedResources(userId: string): Promise; + snapshot(userId: string, characterId: number): Promise; + failure(userId: string, characterId: number): Promise; + currentAuthority(userId: string, characterId: number): Promise; + credential(fence: WalletAuthorityFence): Promise; + publishSnapshot(publication: WalletSnapshotPublication): Promise; + publishFailure(publication: WalletFailurePublication): Promise; +} + +/** Compatibility aliases make the feature-local port easy to adopt without changing shared contracts. */ +export type WalletRepository = WalletBalanceRepository; +export type WalletResource = WalletAuthorizedResource; + +export function walletResourceKey(userId: string, characterId: number): string { + return `${userId}\u0000${characterId}`; +} + +export function walletPrincipalKey(principal: ResourcePrincipal): string { + return `${principal.kind}:${principal.id}`; +} + +export const MAX_WALLET_AMOUNT_LENGTH = 8_192; +export const MAX_WALLET_AUTH_SCOPES = 64; + +function validCounter(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0; +} + +function clonePrincipal(principal: WalletAuthorizedResource['principal']): WalletAuthorizedResource['principal'] { + return { ...principal }; +} + +function cloneResource(resource: WalletAuthorizedResource): WalletAuthorizedResource { + return { + ...resource, + principal: clonePrincipal(resource.principal), + resource: { ...resource.resource, principal: { ...resource.resource.principal }, requiredScopes: [...resource.resource.requiredScopes] }, + grantedScopes: [...resource.grantedScopes], + }; +} + +function cloneSnapshot(snapshot: WalletBalanceSnapshot): WalletBalanceSnapshot { + return { ...snapshot, principal: clonePrincipal(snapshot.principal), coverage: { ...snapshot.coverage } }; +} + +function cloneFence(fence: WalletAuthorityFence): WalletAuthorityFence { + return { ...fence, principal: clonePrincipal(fence.principal), scopes: [...fence.scopes] }; +} + +function cloneFailure(failure: WalletRefreshFailure): WalletRefreshFailure { + return { ...failure }; +} + +function canonicalScopes(value: unknown, allowEmpty = false): string[] | null { + if (!Array.isArray(value) || (!allowEmpty && value.length === 0) || value.length > MAX_WALLET_AUTH_SCOPES) return null; + if (value.some((scope) => typeof scope !== 'string' || scope.length === 0 || scope.length > 256)) return null; + const scopes = [...value] as string[]; + if (new Set(scopes).size !== scopes.length) return null; + return scopes.sort(); +} + +export function validateWalletScopes(value: unknown, allowEmpty = false): readonly string[] { + const scopes = canonicalScopes(value, allowEmpty); + if (!scopes) throw new TypeError('wallet scopes are invalid'); + return scopes; +} + +export function validateWalletAuthorityFence(fence: WalletAuthorityFence): void { + if (!fence || typeof fence.userId !== 'string' || !fence.userId || !Number.isSafeInteger(fence.characterId) || fence.characterId <= 0 || !fence.principal || fence.principal.kind !== 'character' || !isDecimalId(fence.principal.id) || fence.principal.id !== String(fence.characterId) || fence.principal.ownerUserId !== fence.userId || !validCounter(fence.grantRevision) || !validCounter(fence.authorizationEpoch) || !validCounter(fence.credentialVersion) || !canonicalScopes(fence.scopes)?.includes(ESI_WALLET_SCOPE)) throw new TypeError('wallet authority fence is invalid'); +} + +export function validateWalletAmount(value: unknown): string { + if (typeof value !== 'string' || value.length === 0 || value.length > MAX_WALLET_AMOUNT_LENGTH) throw new RangeError('wallet amount exceeds resource bounds'); + const parsed = parseIsk(value); + const canonical = parsed.toString(10); + if (value !== canonical) throw new TypeError('wallet amount is not canonical signed integer cents'); + return canonical; +} + +function assertResource(resource: WalletAuthorizedResource): void { + if (!resource || typeof resource.userId !== 'string' || !resource.userId || !Number.isSafeInteger(resource.characterId) || resource.characterId <= 0) throw new TypeError('wallet resource identity is invalid'); + if (!isDecimalId(resource.principal.id) || resource.principal.ownerUserId !== resource.userId || resource.principal.kind !== 'character' || resource.principal.id !== String(resource.characterId)) throw new TypeError('wallet resource principal is invalid'); + if (!resource.resource || resource.resource.principal.kind !== 'character' || resource.resource.principal.id !== resource.principal.id || resource.resource.principal.ownerUserId !== resource.userId) throw new TypeError('wallet resource selector is invalid'); + if (!Array.isArray(resource.resource.requiredScopes) || !canonicalScopes(resource.resource.requiredScopes)?.includes(ESI_WALLET_SCOPE) || resource.resource.authorizationRevision !== resource.grantRevision || resource.resource.authorizationEpoch !== resource.authorizationEpoch) throw new TypeError('wallet resource scope pin is invalid'); + if (!canonicalScopes(resource.grantedScopes, true)) throw new TypeError('wallet granted scopes are invalid'); + if (!validCounter(resource.grantRevision) || !validCounter(resource.authorizationEpoch) || (resource.credentialVersion !== null && !validCounter(resource.credentialVersion))) throw new TypeError('wallet authority counters are invalid'); + if (!['granted', 'missing'].includes(resource.consent)) throw new TypeError('wallet consent state is invalid'); + if (!['active', 'revocation_pending', 'quarantined', 'revoked', 'transferred'].includes(resource.authorizationStatus)) throw new TypeError('wallet authorization state is invalid'); + if (resource.consent === 'granted' && !resource.grantedScopes.includes(ESI_WALLET_SCOPE)) throw new TypeError('wallet granted consent lacks the required scope'); + if (resource.consent === 'granted' && resource.credentialVersion === null) throw new TypeError('wallet granted consent lacks credential version'); +} + +export function walletFenceForResource(resource: WalletAuthorizedResource): WalletAuthorityFence | null { + assertResource(resource); + if (resource.consent !== 'granted' || resource.authorizationStatus !== 'active' || resource.credentialVersion === null) return null; + return { + userId: resource.userId, + characterId: resource.characterId, + principal: clonePrincipal(resource.principal), + grantRevision: resource.grantRevision, + authorizationEpoch: resource.authorizationEpoch, + credentialVersion: resource.credentialVersion, + scopes: [...resource.grantedScopes].sort(), + }; +} + +export function sameWalletFence(left: WalletAuthorityFence, right: WalletAuthorityFence): boolean { + return left.userId === right.userId + && left.characterId === right.characterId + && left.principal.kind === 'character' + && right.principal.kind === 'character' + && left.principal.id === right.principal.id + && left.principal.ownerUserId === right.principal.ownerUserId + && left.grantRevision === right.grantRevision + && left.authorizationEpoch === right.authorizationEpoch + && left.credentialVersion === right.credentialVersion + && [...left.scopes].sort().join('\u0000') === [...right.scopes].sort().join('\u0000'); +} + +export function validateWalletSnapshot(snapshot: WalletBalanceSnapshot): void { + if (!snapshot || typeof snapshot.userId !== 'string' || !snapshot.userId || !Number.isSafeInteger(snapshot.characterId) || snapshot.characterId <= 0) throw new TypeError('wallet snapshot identity is invalid'); + if (snapshot.principal.kind !== 'character' || snapshot.principal.ownerUserId !== snapshot.userId || snapshot.principal.id !== String(snapshot.characterId)) throw new TypeError('wallet snapshot principal is invalid'); + validateWalletAmount(snapshot.amount); + if (!Number.isFinite(snapshot.fetchedAt) || snapshot.fetchedAt < 0 || !Number.isFinite(snapshot.expiresAt) || snapshot.expiresAt <= snapshot.fetchedAt) throw new TypeError('wallet snapshot timestamps are invalid'); + if (snapshot.upstreamEffectiveAt !== null && (!Number.isFinite(snapshot.upstreamEffectiveAt) || snapshot.upstreamEffectiveAt < 0)) throw new TypeError('wallet upstream timestamp is invalid'); + if (snapshot.coverage.kind !== 'snapshot' || snapshot.coverage.complete !== true || typeof snapshot.coverage.generation !== 'string' || !snapshot.coverage.generation) throw new TypeError('wallet snapshot coverage is invalid'); + if (snapshot.generation !== snapshot.coverage.generation || !snapshot.generation || snapshot.generation.length > MAX_WALLET_AMOUNT_LENGTH) throw new TypeError('wallet snapshot generation is invalid'); + if (!validCounter(snapshot.authorizationRevision) || !validCounter(snapshot.authorizationEpoch) || !validCounter(snapshot.credentialVersion)) throw new TypeError('wallet snapshot authority counters are invalid'); +} + +export function validateWalletRefreshFailure(failure: WalletRefreshFailure): void { + if (!failure || typeof failure.userId !== 'string' || !failure.userId || !Number.isSafeInteger(failure.characterId) || failure.characterId <= 0 || !Number.isFinite(failure.failedAt) || failure.failedAt < 0 || typeof failure.category !== 'string' || !failure.category || typeof failure.retryable !== 'boolean' || !validCounter(failure.authorizationRevision) || !validCounter(failure.authorizationEpoch) || !validCounter(failure.credentialVersion) || typeof failure.generation !== 'string' || !failure.generation || failure.generation.length > MAX_WALLET_AMOUNT_LENGTH || (failure.retryAt !== undefined && (!Number.isFinite(failure.retryAt) || failure.retryAt < failure.failedAt)) || (failure.cooldownGroup !== undefined && (typeof failure.cooldownGroup !== 'string' || !failure.cooldownGroup))) throw new TypeError('wallet refresh failure is invalid'); +} + +const validateFailure = validateWalletRefreshFailure; + +function sameSnapshot(left: WalletBalanceSnapshot, right: WalletBalanceSnapshot): boolean { + return left.userId === right.userId && left.characterId === right.characterId && left.principal.id === right.principal.id && left.principal.ownerUserId === right.principal.ownerUserId && left.amount === right.amount && left.fetchedAt === right.fetchedAt && left.expiresAt === right.expiresAt && left.upstreamEffectiveAt === right.upstreamEffectiveAt && left.generation === right.generation && left.authorizationRevision === right.authorizationRevision && left.authorizationEpoch === right.authorizationEpoch && left.credentialVersion === right.credentialVersion; +} + +function sameFailure(left: WalletRefreshFailure, right: WalletRefreshFailure): boolean { + return left.userId === right.userId && left.characterId === right.characterId && left.failedAt === right.failedAt && left.category === right.category && left.retryable === right.retryable && left.authorizationRevision === right.authorizationRevision && left.authorizationEpoch === right.authorizationEpoch && left.credentialVersion === right.credentialVersion && left.generation === right.generation && left.retryAt === right.retryAt && left.cooldownGroup === right.cooldownGroup && left.message === right.message && left.jobId === right.jobId; +} + +/** + * Fixture-only memory adapter. It intentionally contains no Convex/schema or + * production queue behavior; a durable adapter must preserve these fences and + * checkpoint the stage/publication transition atomically. + */ +export class InMemoryWalletBalanceRepository implements WalletBalanceRepository { + private readonly resources = new Map(); + private readonly snapshots = new Map(); + private readonly failures = new Map(); + private readonly credentials = new Map(); + private readonly publishedJobs = new Map(); + + public constructor(initial: { + resources?: readonly WalletAuthorizedResource[]; + snapshots?: readonly WalletBalanceSnapshot[]; + failures?: readonly WalletRefreshFailure[]; + credentials?: readonly { fence: WalletAuthorityFence; credential: WalletCredential }[]; + } = {}) { + for (const resource of initial.resources ?? []) this.seedResource(resource); + for (const snapshot of initial.snapshots ?? []) this.seedSnapshot(snapshot); + for (const failure of initial.failures ?? []) this.seedFailure(failure); + for (const entry of initial.credentials ?? []) this.seedCredential(entry.fence, entry.credential); + } + + public seedResource(resource: WalletAuthorizedResource): void { + assertResource(resource); + const key = walletResourceKey(resource.userId, resource.characterId); + if (this.resources.has(key)) throw new TypeError('duplicate wallet resource'); + this.resources.set(key, cloneResource(resource)); + } + + /** Fixture-only authority rotation hook for stale-worker tests. */ + public replaceResource(resource: WalletAuthorizedResource): void { + assertResource(resource); + this.resources.set(walletResourceKey(resource.userId, resource.characterId), cloneResource(resource)); + } + + public seedSnapshot(snapshot: WalletBalanceSnapshot): void { + validateWalletSnapshot(snapshot); + this.snapshots.set(walletResourceKey(snapshot.userId, snapshot.characterId), cloneSnapshot(snapshot)); + } + + public seedFailure(failure: WalletRefreshFailure): void { + validateFailure(failure); + this.failures.set(walletResourceKey(failure.userId, failure.characterId), cloneFailure(failure)); + } + + public seedCredential(fence: WalletAuthorityFence, credential: WalletCredential): void { + validateWalletAuthorityFence(fence); + if (typeof credential.accessToken !== 'string' || !credential.accessToken) throw new TypeError('wallet credential is invalid'); + const key = credential.authorizationKey; + if (!key || key.userId !== fence.userId || key.principal !== walletPrincipalKey(fence.principal) || key.characterId !== fence.characterId || key.grantRevision !== fence.grantRevision || key.authorizationEpoch !== fence.authorizationEpoch || key.credentialVersion !== fence.credentialVersion || canonicalScopes(key.scopes)?.join('\u0000') !== canonicalScopes(fence.scopes)?.join('\u0000')) throw new TypeError('wallet credential authority does not match'); + this.credentials.set(walletResourceKey(fence.userId, fence.characterId), { accessToken: credential.accessToken, authorizationKey: { ...key, scopes: [...key.scopes] } }); + } + + public async authorizedResources(userId: string): Promise { + if (typeof userId !== 'string' || !userId) throw new TypeError('wallet user is invalid'); + return [...this.resources.values()].filter((resource) => resource.userId === userId).sort((a, b) => a.characterId - b.characterId).map(cloneResource); + } + + public async snapshot(userId: string, characterId: number): Promise { + if (typeof userId !== 'string' || !userId || !Number.isSafeInteger(characterId) || characterId <= 0) throw new TypeError('wallet snapshot lookup is invalid'); + const snapshot = this.snapshots.get(walletResourceKey(userId, characterId)); + return snapshot ? cloneSnapshot(snapshot) : null; + } + + public async failure(userId: string, characterId: number): Promise { + const failure = this.failures.get(walletResourceKey(userId, characterId)); + if (!failure || failure.userId !== userId || failure.characterId !== characterId) return null; + const current = await this.currentAuthority(userId, characterId); + if (!current || failure.authorizationRevision !== current.grantRevision || failure.authorizationEpoch !== current.authorizationEpoch || failure.credentialVersion !== current.credentialVersion) return null; + return cloneFailure(failure); + } + + public async currentAuthority(userId: string, characterId: number): Promise { + const resource = this.resources.get(walletResourceKey(userId, characterId)); + return resource ? walletFenceForResource(resource) : null; + } + + public async credential(fence: WalletAuthorityFence): Promise { + validateWalletAuthorityFence(fence); + const current = await this.currentAuthority(fence.userId, fence.characterId); + if (!current || !sameWalletFence(current, fence)) return null; + const credential = this.credentials.get(walletResourceKey(fence.userId, fence.characterId)); + if (!credential) return null; + const key = credential.authorizationKey; + if (typeof credential.accessToken !== 'string' || !credential.accessToken || !key || key.userId !== fence.userId || key.principal !== walletPrincipalKey(fence.principal) || key.characterId !== fence.characterId || key.grantRevision !== fence.grantRevision || key.authorizationEpoch !== fence.authorizationEpoch || key.credentialVersion !== fence.credentialVersion || canonicalScopes(key.scopes)?.join('\u0000') !== canonicalScopes(fence.scopes)?.join('\u0000')) return null; + return { accessToken: credential.accessToken, authorizationKey: { ...key, scopes: [...key.scopes] } }; + } + + public async publishSnapshot(publication: WalletSnapshotPublication): Promise { + validateWalletAuthorityFence(publication.fence); + validateWalletSnapshot(publication.snapshot); + if (!Number.isFinite(publication.publishedAt) || publication.publishedAt < 0 || publication.snapshot.fetchedAt > publication.publishedAt) return { accepted: false, reason: 'invalid_snapshot' }; + if (publication.snapshot.userId !== publication.fence.userId || publication.snapshot.characterId !== publication.fence.characterId || publication.snapshot.principal.id !== publication.fence.principal.id || publication.snapshot.principal.ownerUserId !== publication.fence.principal.ownerUserId || publication.snapshot.authorizationRevision !== publication.fence.grantRevision || publication.snapshot.authorizationEpoch !== publication.fence.authorizationEpoch || publication.snapshot.credentialVersion !== publication.fence.credentialVersion) return { accepted: false, reason: 'mismatched_character' }; + const current = await this.currentAuthority(publication.fence.userId, publication.fence.characterId); + if (!current) return { accepted: false, reason: 'authorization_revoked' }; + if (!sameWalletFence(current, publication.fence)) { + if (current.grantRevision !== publication.fence.grantRevision) return { accepted: false, reason: 'stale_grant' }; + if (current.authorizationEpoch !== publication.fence.authorizationEpoch) return { accepted: false, reason: 'stale_authorization_epoch' }; + if (current.credentialVersion !== publication.fence.credentialVersion) return { accepted: false, reason: 'stale_credential' }; + return { accepted: false, reason: 'borrowed_resource' }; + } + const previousJob = this.publishedJobs.get(publication.jobId); + if (previousJob) { + if (previousJob.outcome === 'success' && sameWalletFence(previousJob.fence, publication.fence) && sameSnapshot(previousJob.snapshot, publication.snapshot)) return { accepted: true, snapshot: cloneSnapshot(previousJob.snapshot), outcome: 'success' }; + return { accepted: false, reason: 'duplicate_publication' }; + } + const key = walletResourceKey(publication.fence.userId, publication.fence.characterId); + const prior = this.snapshots.get(key); + const priorFailure = this.failures.get(key); + if (priorFailure && priorFailure.authorizationRevision === publication.fence.grantRevision && priorFailure.authorizationEpoch === publication.fence.authorizationEpoch && priorFailure.credentialVersion === publication.fence.credentialVersion && priorFailure.failedAt >= publication.snapshot.fetchedAt) return { accepted: false, reason: 'stale_generation' }; + if (prior) { + if (prior.generation === publication.snapshot.generation) { + if (!sameSnapshot(prior, publication.snapshot)) return { accepted: false, reason: 'stale_generation' }; + this.publishedJobs.set(publication.jobId, { fence: cloneFence(publication.fence), outcome: 'success', snapshot: cloneSnapshot(prior) }); + return { accepted: true, snapshot: cloneSnapshot(prior), outcome: 'success' }; + } + if (prior.fetchedAt >= publication.snapshot.fetchedAt) return { accepted: false, reason: 'stale_generation' }; + } + this.snapshots.set(key, cloneSnapshot(publication.snapshot)); + if (!priorFailure || priorFailure.failedAt < publication.snapshot.fetchedAt) this.failures.delete(key); + this.publishedJobs.set(publication.jobId, { fence: cloneFence(publication.fence), outcome: 'success', snapshot: cloneSnapshot(publication.snapshot) }); + return { accepted: true, snapshot: cloneSnapshot(publication.snapshot), outcome: 'success' }; + } + + public async publishFailure(publication: WalletFailurePublication): Promise { + validateWalletAuthorityFence(publication.fence); + validateFailure(publication.failure); + if (publication.failure.userId !== publication.fence.userId || publication.failure.characterId !== publication.fence.characterId || publication.failure.authorizationRevision !== publication.fence.grantRevision || publication.failure.authorizationEpoch !== publication.fence.authorizationEpoch || publication.failure.credentialVersion !== publication.fence.credentialVersion || !Number.isFinite(publication.publishedAt) || publication.failure.failedAt > publication.publishedAt) return { accepted: false, reason: 'mismatched_character' }; + const current = await this.currentAuthority(publication.fence.userId, publication.fence.characterId); + if (!current) return { accepted: false, reason: 'authorization_revoked' }; + if (!sameWalletFence(current, publication.fence)) return { accepted: false, reason: current.grantRevision !== publication.fence.grantRevision ? 'stale_grant' : current.authorizationEpoch !== publication.fence.authorizationEpoch ? 'stale_authorization_epoch' : current.credentialVersion !== publication.fence.credentialVersion ? 'stale_credential' : 'borrowed_resource' }; + const previousJob = this.publishedJobs.get(publication.jobId); + if (previousJob) { + if (previousJob.outcome === 'failure' && sameWalletFence(previousJob.fence, publication.fence) && sameFailure(previousJob.failure, publication.failure)) return { accepted: true, snapshot: null, outcome: 'failure' }; + return { accepted: false, reason: 'duplicate_publication' }; + } + const key = walletResourceKey(publication.failure.userId, publication.failure.characterId); + const prior = this.failures.get(key); + const priorSnapshot = this.snapshots.get(key); + if (priorSnapshot && priorSnapshot.authorizationRevision === publication.fence.grantRevision && priorSnapshot.authorizationEpoch === publication.fence.authorizationEpoch && priorSnapshot.credentialVersion === publication.fence.credentialVersion && priorSnapshot.fetchedAt >= publication.failure.failedAt) return { accepted: false, reason: 'stale_generation' }; + if (prior) { + if (prior.generation === publication.failure.generation) { + if (!sameFailure(prior, publication.failure)) return { accepted: false, reason: 'stale_generation' }; + this.publishedJobs.set(publication.jobId, { fence: cloneFence(publication.fence), outcome: 'failure', failure: cloneFailure(prior) }); + return { accepted: true, snapshot: null, outcome: 'failure' }; + } + if (prior.failedAt >= publication.failure.failedAt) return { accepted: false, reason: 'stale_generation' }; + } + this.failures.set(key, cloneFailure(publication.failure)); + this.publishedJobs.set(publication.jobId, { fence: cloneFence(publication.fence), outcome: 'failure', failure: cloneFailure(publication.failure) }); + return { accepted: true, snapshot: null, outcome: 'failure' }; + } +} + +/** Build the shared component shape without inventing an upstream effective time. */ +export function walletComponent(snapshot: WalletBalanceSnapshot): WalletBalanceComponent { + validateWalletSnapshot(snapshot); + return { + component: 'wallet_balance', + principal: clonePrincipal(snapshot.principal), + amount: snapshot.amount, + fetchedAt: snapshot.fetchedAt, + upstreamEffectiveAt: snapshot.upstreamEffectiveAt, + coverage: { ...snapshot.coverage }, + authorizationRevision: snapshot.authorizationRevision, + credentialVersion: snapshot.credentialVersion, + }; +} + +export type WalletBalanceSharedComponent = WalletBalanceComponent; +export type WalletBalanceSharedProvenance = CompleteReportProvenance; diff --git a/src/lib/server/wallet/balance-service.ts b/src/lib/server/wallet/balance-service.ts new file mode 100644 index 0000000..fdcf241 --- /dev/null +++ b/src/lib/server/wallet/balance-service.ts @@ -0,0 +1,339 @@ +import type { + CompleteReportProvenance, + DatasetKind, + ReportDatasetPin, + ReportProvenance, + ResourcePrincipal, + SelectedResource, + WalletBalanceComponent, +} from '../../domain/model'; +import { compareDecimalIds, isDecimalId } from '../../domain/inventory-location'; +import { sumIsk } from '../../domain/money'; +import { ESI_WALLET_SCOPE } from '../esi'; +import { + validateWalletRefreshFailure, + validateWalletScopes, + validateWalletSnapshot, + walletComponent, + walletFenceForResource, + type WalletAuthorizedResource, + type WalletBalanceRepository, + type WalletBalanceSnapshot, + type WalletRefreshFailure, +} from './balance-repository'; +import type { WalletBalanceJobQueue } from './balance-job'; + +export const DEFAULT_WALLET_MAX_CHARACTERS = 100; +export const DEFAULT_WALLET_ENQUEUE_LIMIT = 100; +export const WALLET_SELECTOR_MAX_IDS = 100; + +export type WalletBalanceRequest = { + userId: string; + /** Character IDs are selectors only; authorization is resolved by the repository. */ + characterIds?: readonly (string | number)[]; +}; + +export type WalletFreshness = 'fresh' | 'stale' | 'missing' | 'failed' | 'consent_required' | 'unavailable'; + +export type WalletBalanceCharacterView = { + principal: Extract; + characterId: number; + characterName: string | null; + consent: 'granted' | 'missing'; + freshness: WalletFreshness; + amount: string | null; + fetchedAt: number | null; + expiresAt: number | null; + upstreamEffectiveAt: number | null; + authorizationRevision: number; + authorizationEpoch: number; + credentialVersion: number | null; + failure: WalletRefreshFailure | null; +}; + +type WalletBalanceCommonView = { + characters: readonly WalletBalanceCharacterView[]; + components: readonly WalletBalanceComponent[]; + /** Explicitly wallet-only; this is never an asset/wealth total. */ + walletOnlyTotal: string | null; + missing: readonly Extract[]; + /** Queue outcomes are diagnostic only and never turn stale data into ready data. */ + refresh: { attempted: number; queued: number; duplicate: number; cooldown: number; skipped: number; failed: number }; +}; + +export type WalletBalanceReadyView = WalletBalanceCommonView & { + status: 'ready'; + walletTotal: string; + walletOnlyTotal: string; + provenance: CompleteReportProvenance; +}; +export type WalletBalancePartialView = WalletBalanceCommonView & { + status: 'partial'; + walletTotal: null; + walletOnlyTotal: null; + provenance: ReportProvenance; +}; +export type WalletBalanceBlockedView = WalletBalanceCommonView & { + status: 'blocked' | 'unavailable'; + walletTotal: null; + walletOnlyTotal: null; + reason: string; + provenance?: never; +}; +/** Shared result consumed by both a future UI route and assistant adapter. */ +export type WalletBalanceView = WalletBalanceReadyView | WalletBalancePartialView | WalletBalanceBlockedView; + +export type WalletBalanceServiceOptions = { + repository: WalletBalanceRepository; + queue?: WalletBalanceJobQueue; + now?: () => number; + maxCharacters?: number; + enqueueLimit?: number; +}; + +function finiteNow(now: () => number): number { + const value = now(); + if (!Number.isFinite(value) || value < 0) throw new RangeError('wallet service clock returned an invalid timestamp'); + return value; +} + +function clonePrincipal(principal: Extract): Extract { + return { ...principal }; +} + +function validateResource(resource: WalletAuthorizedResource, userId: string): void { + if (!resource || resource.userId !== userId || resource.principal.kind !== 'character' || resource.principal.ownerUserId !== userId || resource.principal.id !== String(resource.characterId) || !Number.isSafeInteger(resource.characterId) || resource.characterId <= 0 || !resource.resource || resource.resource.principal.kind !== 'character' || resource.resource.principal.id !== resource.principal.id || resource.resource.principal.ownerUserId !== userId || !validateWalletScopes(resource.resource.requiredScopes).includes(ESI_WALLET_SCOPE)) throw new TypeError('wallet repository returned malformed or borrowed resource'); + validateWalletScopes(resource.grantedScopes, true); + if (!Number.isSafeInteger(resource.grantRevision) || resource.grantRevision < 0 || !Number.isSafeInteger(resource.authorizationEpoch) || resource.authorizationEpoch < 0 || (resource.credentialVersion !== null && (!Number.isSafeInteger(resource.credentialVersion) || resource.credentialVersion < 0))) throw new TypeError('wallet repository returned malformed authority counters'); + if (!['granted', 'missing'].includes(resource.consent) || !['active', 'revocation_pending', 'quarantined', 'revoked', 'transferred'].includes(resource.authorizationStatus)) throw new TypeError('wallet repository returned malformed consent state'); + if (resource.resource.authorizationRevision !== resource.grantRevision || resource.resource.authorizationEpoch !== resource.authorizationEpoch) throw new TypeError('wallet resource pin does not match authority counters'); + if (resource.consent === 'granted' && (!resource.grantedScopes.includes(ESI_WALLET_SCOPE) || resource.credentialVersion === null)) throw new TypeError('wallet granted resource lacks exact authority'); +} + +function normalizeSelectedIds(value: readonly (string | number)[] | undefined): string[] | null { + if (value === undefined) return null; + if (!Array.isArray(value) || value.length === 0 || value.length > WALLET_SELECTOR_MAX_IDS) throw new RangeError('wallet character selection cap exceeded'); + const ids = value.map((entry, index) => { + if (typeof entry === 'number') { + if (!Number.isSafeInteger(entry) || entry <= 0) throw new TypeError(`wallet characterIds[${index}] is invalid`); + return String(entry); + } + if (!isDecimalId(entry)) throw new TypeError(`wallet characterIds[${index}] is invalid`); + return entry; + }); + if (new Set(ids).size !== ids.length) throw new TypeError('wallet character selection contains duplicates'); + return ids.sort(compareDecimalIds); +} + +function sameResource(left: SelectedResource, right: SelectedResource): boolean { + return left.principal.kind === 'character' && right.principal.kind === 'character' && left.principal.id === right.principal.id && left.principal.ownerUserId === right.principal.ownerUserId && left.authorizationRevision === right.authorizationRevision && left.authorizationEpoch === right.authorizationEpoch && [...left.requiredScopes].sort().join('\u0000') === [...right.requiredScopes].sort().join('\u0000'); +} + +function incompleteCoverage(): Extract { + return { kind: 'snapshot', generation: null, complete: false }; +} + +function resourceMaps(resources: readonly WalletAuthorizedResource[]) { + const authorizationRevisions: Record = {}; + const authorizationEpochs: Record = {}; + const snapshotGenerations: Record = {}; + for (const resource of resources) { + authorizationRevisions[`character:${resource.characterId}`] = resource.grantRevision; + authorizationEpochs[`character:${resource.characterId}`] = resource.authorizationEpoch; + } + return { authorizationRevisions, authorizationEpochs, snapshotGenerations }; +} + +function provenance(resources: readonly WalletAuthorizedResource[], snapshots: ReadonlyMap, now: number, complete: boolean): ReportProvenance | CompleteReportProvenance { + const selectedResources = resources.map((resource) => ({ ...resource.resource, principal: clonePrincipal(resource.principal), requiredScopes: [...resource.resource.requiredScopes] })); + const maps = resourceMaps(resources); + const datasetPins: ReportDatasetPin[] = resources.map((resource) => { + const snapshot = snapshots.get(resource.characterId); + if (snapshot) maps.snapshotGenerations[`character:${resource.characterId}`] = snapshot.generation; + return { + selectedResource: resource.resource, + principal: resource.principal, + dataset: 'wallet_balance', + coverage: snapshot?.coverage ?? incompleteCoverage(), + authorizationRevision: resource.grantRevision, + authorizationEpoch: resource.authorizationEpoch, + fetchedAt: snapshot?.fetchedAt ?? null, + upstreamEffectiveAt: snapshot?.upstreamEffectiveAt ?? null, + }; + }); + const base: ReportProvenance = { + reportId: `wallet-${now}-${selectedResources.map((resource) => resource.principal.id).join(',')}`, + generatedAt: now, + fetchedAt: Object.fromEntries(resources.map((resource) => [`character:${resource.characterId}`, snapshots.get(resource.characterId)?.fetchedAt ?? null])), + upstreamEffectiveAt: Object.fromEntries(resources.map((resource) => [`character:${resource.characterId}`, snapshots.get(resource.characterId)?.upstreamEffectiveAt ?? null])), + input: { + selectedResources, + requiredDatasetKinds: ['wallet_balance'] as readonly [DatasetKind], + datasetPins, + authorizationRevisions: maps.authorizationRevisions, + authorizationEpochs: maps.authorizationEpochs, + snapshotGenerations: maps.snapshotGenerations, + appendCheckpoints: {}, + referenceVersion: null, + priceVersion: null, + ...(complete ? { pinSet: 'exact_selected_resource_dataset_set' as const } : {}), + } as ReportProvenance['input'], + sourceSkew: false, + assumptions: ['Wallet balance is a separately labelled cash component; assets, orders, journals and wealth are not included.', 'ESI supplied no upstream effective timestamp; upstreamEffectiveAt is null.'], + }; + return complete ? base as CompleteReportProvenance : base; +} + +function characterView(resource: WalletAuthorizedResource, snapshot: WalletBalanceSnapshot | null, failure: WalletRefreshFailure | null, now: number): WalletBalanceCharacterView { + const allowed = resource.consent === 'granted' && resource.authorizationStatus === 'active' && resource.credentialVersion !== null; + if (!allowed) return { + principal: clonePrincipal(resource.principal), characterId: resource.characterId, characterName: resource.characterName ?? null, consent: 'missing', freshness: 'consent_required', amount: null, fetchedAt: null, expiresAt: null, upstreamEffectiveAt: null, + authorizationRevision: resource.grantRevision, authorizationEpoch: resource.authorizationEpoch, credentialVersion: resource.credentialVersion, failure: null, + }; + if (!snapshot) return { + principal: clonePrincipal(resource.principal), characterId: resource.characterId, characterName: resource.characterName ?? null, consent: 'granted', freshness: failure ? 'failed' : 'missing', amount: null, fetchedAt: null, expiresAt: null, upstreamEffectiveAt: null, + authorizationRevision: resource.grantRevision, authorizationEpoch: resource.authorizationEpoch, credentialVersion: resource.credentialVersion, failure, + }; + const failed = failure !== null && failure.failedAt >= snapshot.fetchedAt; + return { + principal: clonePrincipal(resource.principal), characterId: resource.characterId, characterName: resource.characterName ?? null, consent: 'granted', freshness: failed ? 'failed' : now < snapshot.expiresAt ? 'fresh' : 'stale', amount: snapshot.amount, fetchedAt: snapshot.fetchedAt, expiresAt: snapshot.expiresAt, upstreamEffectiveAt: snapshot.upstreamEffectiveAt, + authorizationRevision: resource.grantRevision, authorizationEpoch: resource.authorizationEpoch, credentialVersion: resource.credentialVersion, failure: failed ? failure : null, + }; +} + +/** + * Deterministic wallet read core shared by an eventual UI route and assistant + * adapter. Opening the view may enqueue stale observations, but never waits for + * a refresh or replaces a valid cache with an unnecessary job. + */ +export class WalletBalanceService { + private readonly now: () => number; + private readonly maxCharacters: number; + private readonly enqueueLimit: number; + + public constructor(private readonly options: WalletBalanceServiceOptions) { + this.now = options.now ?? (() => Date.now()); + this.maxCharacters = options.maxCharacters ?? DEFAULT_WALLET_MAX_CHARACTERS; + this.enqueueLimit = options.enqueueLimit ?? DEFAULT_WALLET_ENQUEUE_LIMIT; + if (!Number.isSafeInteger(this.maxCharacters) || this.maxCharacters < 1 || !Number.isSafeInteger(this.enqueueLimit) || this.enqueueLimit < 1) throw new RangeError('wallet service bounds are invalid'); + } + + public async read(request: WalletBalanceRequest): Promise { + if (!request || typeof request.userId !== 'string' || !request.userId) throw new TypeError('wallet user is required'); + const now = finiteNow(this.now); + const selectedIds = normalizeSelectedIds(request.characterIds); + const all = await this.options.repository.authorizedResources(request.userId); + if (!Array.isArray(all) || all.length > this.maxCharacters) throw new RangeError('wallet authorized resource cap exceeded'); + const seen = new Set(); + for (const resource of all) { + validateResource(resource, request.userId); + if (seen.has(resource.principal.id)) throw new TypeError('wallet repository returned duplicate characters'); + seen.add(resource.principal.id); + } + let resources: WalletAuthorizedResource[]; + if (selectedIds === null) resources = [...all]; + else { + resources = []; + for (const id of selectedIds) { + const resource = all.find((candidate) => candidate.principal.id === id); + if (!resource) return { status: 'blocked', characters: [], components: [], walletTotal: null, walletOnlyTotal: null, missing: [], reason: 'Selected character is not an authorized linked resource', refresh: { attempted: 0, queued: 0, duplicate: 0, cooldown: 0, skipped: 0, failed: 0 } }; + resources.push(resource); + } + } + resources.sort((a, b) => compareDecimalIds(a.principal.id, b.principal.id)); + if (resources.length === 0) return { status: 'blocked', characters: [], components: [], walletTotal: null, walletOnlyTotal: null, missing: [], reason: 'No authorized linked wallet resources are available', refresh: { attempted: 0, queued: 0, duplicate: 0, cooldown: 0, skipped: 0, failed: 0 } }; + + const snapshots = new Map(); + const failures = new Map(); + const views: WalletBalanceCharacterView[] = []; + const refresh = { attempted: 0, queued: 0, duplicate: 0, cooldown: 0, skipped: 0, failed: 0 }; + for (const resource of resources) { + let snapshot: WalletBalanceSnapshot | null = null; + let failure: WalletRefreshFailure | null = null; + try { + snapshot = await this.options.repository.snapshot(request.userId, resource.characterId); + failure = await this.options.repository.failure(request.userId, resource.characterId); + } catch { + views.push({ principal: clonePrincipal(resource.principal), characterId: resource.characterId, characterName: resource.characterName ?? null, consent: resource.consent, freshness: 'unavailable', amount: null, fetchedAt: null, expiresAt: null, upstreamEffectiveAt: null, authorizationRevision: resource.grantRevision, authorizationEpoch: resource.authorizationEpoch, credentialVersion: resource.credentialVersion, failure: null }); + refresh.skipped += 1; + continue; + } + if (snapshot) { + validateWalletSnapshot(snapshot); + if (snapshot.fetchedAt > now || (snapshot.upstreamEffectiveAt !== null && snapshot.upstreamEffectiveAt > snapshot.fetchedAt)) throw new TypeError('wallet repository returned future or inconsistent snapshot provenance'); + if (snapshot.userId !== request.userId || snapshot.characterId !== resource.characterId || !sameResource({ ...resource.resource, principal: resource.principal }, snapshotToResource(snapshot, resource))) throw new TypeError('wallet repository snapshot authority does not match selected resource'); + if (snapshot.authorizationRevision !== resource.grantRevision || snapshot.authorizationEpoch !== resource.authorizationEpoch || snapshot.credentialVersion !== resource.credentialVersion) { + // A snapshot from an earlier authority epoch remains stored for audit, + // but it is not readable data for this request. + snapshot = null; + } else snapshots.set(resource.characterId, snapshot); + } + if (failure) { + validateWalletRefreshFailure(failure); + if (failure.failedAt > now) throw new TypeError('wallet repository returned future failure provenance'); + if (failure.userId !== request.userId || failure.characterId !== resource.characterId || failure.authorizationRevision !== resource.grantRevision || failure.authorizationEpoch !== resource.authorizationEpoch || failure.credentialVersion !== resource.credentialVersion) failure = null; + } + if (failure) failures.set(resource.characterId, failure); + const view = characterView(resource, snapshot, failure, now); + views.push(view); + const stale = view.consent === 'granted' && (view.freshness === 'stale' || view.freshness === 'missing' || (view.freshness === 'failed' && failure?.retryable !== false)); + if (stale && this.options.queue && refresh.attempted < this.enqueueLimit) { + refresh.attempted += 1; + const fence = walletFenceForResource(resource); + if (!fence) { refresh.skipped += 1; continue; } + try { + const result = await this.options.queue.enqueue({ userId: request.userId, characterId: resource.characterId, fence, now }); + if (result.status === 'queued') refresh.queued += 1; + else if (result.status === 'duplicate') refresh.duplicate += 1; + else if (result.status === 'cooldown') refresh.cooldown += 1; + else refresh.failed += 1; + } catch { refresh.failed += 1; } + } else if (stale) refresh.skipped += 1; + } + + const components = resources.map((resource) => snapshots.get(resource.characterId)).filter((snapshot): snapshot is WalletBalanceSnapshot => snapshot !== undefined && resourceIsGranted(resources, snapshot.characterId)).map(walletComponent); + const missing = views.filter((view) => view.freshness !== 'fresh').map((view) => clonePrincipal(view.principal)); + const complete = views.length === resources.length && views.every((view) => view.freshness === 'fresh' && view.consent === 'granted') && components.length === resources.length; + const walletTotal = complete ? sumIsk(components.map((component) => component.amount)) : null; + const report = provenance(resources, snapshots, now, complete); + if (complete) return { + status: 'ready', + characters: views, + components, + walletTotal: walletTotal!, + walletOnlyTotal: walletTotal!, + missing, + provenance: report as CompleteReportProvenance, + refresh, + }; + return { + status: 'partial', + characters: views, + components, + walletTotal: null, + walletOnlyTotal: null, + missing, + provenance: report as ReportProvenance, + refresh, + }; + } + + public execute(request: WalletBalanceRequest): Promise { return this.read(request); } +} + +function resourceIsGranted(resources: readonly WalletAuthorizedResource[], characterId: number): boolean { + const resource = resources.find((candidate) => candidate.characterId === characterId); + return !!resource && resource.consent === 'granted' && resource.authorizationStatus === 'active'; +} + +function snapshotToResource(snapshot: WalletBalanceSnapshot, resource: WalletAuthorizedResource): SelectedResource { + return { ...resource.resource, principal: { ...snapshot.principal }, requiredScopes: [...resource.resource.requiredScopes] }; +} + +export async function readWalletBalance(options: WalletBalanceServiceOptions, request: WalletBalanceRequest): Promise { + return new WalletBalanceService(options).read(request); +} + +export const queryWalletBalance = readWalletBalance; +export type WalletBalanceResult = WalletBalanceView; +export type WalletViewResult = WalletBalanceView; diff --git a/tests/fixtures/wallet/index.ts b/tests/fixtures/wallet/index.ts new file mode 100644 index 0000000..f29e440 --- /dev/null +++ b/tests/fixtures/wallet/index.ts @@ -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 { + const principal: Extract = { 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 } }; +} diff --git a/tests/wallet-balance.test.ts b/tests/wallet-balance.test.ts new file mode 100644 index 0000000..836c69f --- /dev/null +++ b/tests/wallet-balance.test.ts @@ -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((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' }); + }); +});