import { ConvexStore, type ClaimedSyncJob, type RefreshLease } from '../src/lib/server/store'; import { TokenCipher } from '../src/lib/server/security'; import { EveSso, type VerifiedGrant } from '../src/lib/server/sso'; import { ESI_ASSETS_SCOPE, EsiClient, EsiError, EsiHttpError, EsiRateLimitError } from '../src/lib/server/esi'; import { InventoryLocationService } from '../src/lib/server/inventory/location-service'; import { CachedPrivateStructurePort, EsiPrivateStructurePort, PrivateStructureCache } from '../src/lib/server/inventory/private-structure-cache'; import { SqliteReferenceRepository } from '../src/lib/server/reference/repository'; import type { EnrichedLocationNode, LocationsObservation, NamesObservation } from '../src/lib/domain/inventory-location'; /** Conservative prototype budgets; these are intentionally configurable and bounded. */ export const SYNC_WORKER_DEFAULTS = { pollMs: 1_000, leaseMs: 10 * 60_000, heartbeatMs: 60_000, maxIdlePolls: 0, } as const; export type WorkerConfig = { store: ConvexStore; cipher: TokenCipher; sso: EveSso; esi: EsiClient; workerId: string; pollMs: number; leaseMs: number; heartbeatMs: number; sleep: (ms: number) => Promise; enrichmentDeadlineMs?: number; /** Present in the production worker; omitted by legacy/unit configurations. */ locationServiceFactory?: (accessToken: string, hooks?: { beforeStructure: () => Promise; onTransient: (error: unknown) => Promise }) => InventoryLocationService | null; }; function fixedFailure(): string { return 'Asset synchronization failed'; } const DEFAULT_SYNC_COOLDOWN_GROUP = 'esi:assets'; function cooldownContext(userId: string, characterId: number): string { return `character:${userId}:${characterId}`; } const HELPER_BATCH_SIZE = 1_000; const LOCATION_CHUNK_SIZE = 100; const MAX_ENRICHMENT_PERSIST_BYTES = 2 * 1024 * 1024; // Keep serialized rows below Convex's 512 KiB validator limit with argument // envelope headroom; the measured worker ceiling is 192 KiB per chunk. const MAX_ENRICHMENT_CHUNK_BYTES = 192 * 1024; const MAX_PERSISTED_ENRICHMENT_ROWS = 2_500; function helperObservationFailure(requestedIds: readonly string[], state: 'forbidden' | 'unavailable', now: number): T { return { requestedIds: [...requestedIds], names: [], locations: [], missing: [], retrievedAt: now, expiresAt: now + 60 * 60_000, complete: false, state, } as unknown as T; } function asNamesObservation(value: Awaited>, characterId: number): NamesObservation { return { ...value, names: value.names.map((entry) => ({ itemId: entry.itemId, name: entry.name })), characterId }; } function asLocationsObservation(value: Awaited>, characterId: number): LocationsObservation { return { ...value, locations: value.locations.map((entry) => ({ itemId: entry.itemId, position: { ...entry.position } })), characterId }; } function isOptionalEnrichmentEnvelopeError(error: unknown): boolean { const message = error instanceof Error ? error.message : String(error); return /Invalid location enrichment (?:chunk|rows)|Location enrichment exceeds persistence (?:byte|row) bound|Location enrichment chunk exceeds byte bound/.test(message); } export function prepareEnrichmentChunks(rows: readonly EnrichedLocationNode[]): EnrichedLocationNode[][] | null { if (rows.length > MAX_PERSISTED_ENRICHMENT_ROWS) return null; const chunks: EnrichedLocationNode[][] = []; let current: EnrichedLocationNode[] = []; let totalBytes = 2; for (const row of rows) { const singleBytes = new TextEncoder().encode(JSON.stringify([row])).byteLength; if (singleBytes > MAX_ENRICHMENT_CHUNK_BYTES) return null; const candidate = [...current, row]; const candidateBytes = new TextEncoder().encode(JSON.stringify(candidate)).byteLength; if (current.length > 0 && (candidate.length > LOCATION_CHUNK_SIZE || candidateBytes > MAX_ENRICHMENT_CHUNK_BYTES)) { const serialized = JSON.stringify(current); totalBytes += new TextEncoder().encode(serialized).byteLength; if (totalBytes > MAX_ENRICHMENT_PERSIST_BYTES) return null; chunks.push(current); current = [row]; } else { current = candidate; } } if (current.length > 0) { totalBytes += new TextEncoder().encode(JSON.stringify(current)).byteLength; if (totalBytes > MAX_ENRICHMENT_PERSIST_BYTES) return null; chunks.push(current); } return chunks; } export class EnrichmentBudgetExceeded extends Error { constructor() { super('Optional location enrichment budget exhausted'); this.name = 'EnrichmentBudgetExceeded'; } } export class MutationUncertainError extends Error { constructor(cause: unknown) { super('Synchronization mutation acknowledgement was uncertain'); this.name = 'MutationUncertainError'; this.cause = cause; } } function permanentMutationFailure(error: unknown): boolean { const message = error instanceof Error ? error.message : String(error); return /lease is not valid|authorization (?:is )?stale|authorization changed|generation is not current|already leased|Invalid asset chunk|Asset character mismatch|Invalid location enrichment chunk|Invalid location enrichment rows|Location enrichment chunk already exists|Location enrichment chunks must be staged|Location enrichment is incomplete|chunks must be staged|chunk already exists|Snapshot is not staging|Snapshot is incomplete|Invalid snapshot|not active|missing|Invalid refreshed scopes|exceeds byte bound|does not match staged asset facts|unrequested item ID|duplicate item IDs|does not cover the staged asset generation|deterministic|truthful|invalid pin/.test(message); } async function retryMutation(operation: () => Promise, renew?: () => Promise): Promise { let lastError: unknown; for (let attempt = 0; attempt < 2; attempt += 1) { try { return await operation(); } catch (error) { lastError = error; if (permanentMutationFailure(error)) throw error; if (attempt === 0 && renew) { try { if (!(await renew())) throw new Error('Synchronization lease is not valid'); } catch (renewError) { if (permanentMutationFailure(renewError)) throw renewError; lastError = renewError; } } } } throw new MutationUncertainError(lastError); } async function observeAssetHelpers( config: WorkerConfig, claimed: ClaimedSyncJob, authority: { userId: string; principal: string; characterId: number; scopes: readonly string[]; grantRevision: number; authorizationEpoch: number; credentialVersion: number }, accessToken: string, assetIds: readonly string[], revalidate: () => Promise, deadlineAt: number, ): Promise<{ names: NamesObservation[]; locations: LocationsObservation[] }> { const context = cooldownContext(claimed.userId, claimed.characterId); const readBatches = async (reader: (ids: string[]) => Promise): Promise => { const batches: T[] = []; let unavailable = false; for (let offset = 0; offset < assetIds.length; offset += HELPER_BATCH_SIZE) { const ids = assetIds.slice(offset, offset + HELPER_BATCH_SIZE); if (Date.now() >= deadlineAt) unavailable = true; if (!unavailable) await revalidate(); if (unavailable) { batches.push(helperObservationFailure(ids, 'unavailable', Date.now())); continue; } const cooldown = await config.store.cooldown(context, claimed.cooldownGroup); if (cooldown && cooldown.until > Date.now()) { const delay = cooldown.until - Date.now(); if (Date.now() + delay >= deadlineAt) { unavailable = true; batches.push(helperObservationFailure(ids, 'unavailable', Date.now())); continue; } await config.sleep(delay); if (Date.now() >= deadlineAt) { unavailable = true; batches.push(helperObservationFailure(ids, 'unavailable', Date.now())); continue; } } try { batches.push(await reader(ids)); } catch (error) { if (error instanceof EsiHttpError && (error.status === 401)) throw error; if (error instanceof EsiHttpError && (error.status === 403 || error.status === 404)) { batches.push(helperObservationFailure(ids, error.status === 403 ? 'forbidden' : 'unavailable', Date.now())); } else if (error instanceof EsiRateLimitError || (error instanceof EsiError && (error.code === 'network' || error.code === 'deadline' || (error instanceof EsiHttpError && (error.status === 408 || error.status === 420 || error.status === 425 || error.status === 429 || error.status >= 500))))) { if (error instanceof EsiRateLimitError && config.store.recordSyncCooldown) await config.store.recordSyncCooldown(context, error.retryAt, error.cooldown.group ?? claimed.cooldownGroup); batches.push(helperObservationFailure(ids, 'unavailable', Date.now())); unavailable = true; } else if (error instanceof EsiError || error instanceof TypeError) { // Schema/validation failures are optional-helper failures, not reasons // to discard the raw asset snapshot or replay a malformed response. batches.push(helperObservationFailure(ids, 'unavailable', Date.now())); unavailable = true; } else throw error; } } return batches; }; const names = await readBatches(async (ids) => asNamesObservation(await config.esi.getAssetNames(authority.characterId, ids, accessToken, authority), authority.characterId)); const locations = await readBatches(async (ids) => asLocationsObservation(await config.esi.getAssetLocations(authority.characterId, ids, accessToken, authority), authority.characterId)); return { names, locations }; } /** Process one at-least-once delivery. Every write is fenced by the durable lease. */ export async function processSyncJob(config: WorkerConfig): Promise { const claimed = await config.store.claimSyncJob(config.workerId, config.leaseMs); if (!claimed) return false; let lostLease = false; let refreshLease: RefreshLease | null = null; let refreshAttempted = false; const heartbeat = setInterval(() => { void config.store.renewSyncJob(claimed.jobId, claimed.leaseToken, config.workerId, config.leaseMs).then((ok) => { if (!ok) lostLease = true; }).catch(() => { lostLease = true; }); }, config.heartbeatMs); const stopHeartbeat = () => clearInterval(heartbeat); try { const refresh = await config.store.acquireRefreshLease(claimed.userId, claimed.characterId, config.workerId); refreshLease = refresh; if (!refresh) { await config.store.deferSyncJob(claimed.jobId, claimed.leaseToken, Date.now() + 1_000); return true; } let previous: VerifiedGrant; try { previous = await config.cipher.open(refresh.encryptedTokens, `character:${claimed.characterId}`); if (previous.characterId !== claimed.characterId) throw new Error('authorization mismatch'); } catch { await config.store.failSyncJob(claimed.jobId, claimed.leaseToken).catch(() => {}); return true; } const needsRefresh = refresh.tokenExpiresAt <= Date.now() + 60_000; let grant: VerifiedGrant = previous; let authorityVersion = { grantRevision: refresh.grantRevision, authorizationEpoch: refresh.authorizationEpoch, credentialVersion: refresh.credentialVersion, scopes: refresh.scopes }; if (!needsRefresh) { // A lease acquired for a still-valid token is only a serialization guard; // release it before paging so another job is not blocked for two minutes. await retryMutation(() => config.store.releaseRefreshLease ? config.store.releaseRefreshLease(claimed.userId, claimed.characterId, refresh.leaseToken, config.workerId) : Promise.resolve(true)); refreshLease = null; } if (needsRefresh) { try { const dispatched = await config.store.markRefreshDispatched(claimed.userId, claimed.characterId, refresh.leaseToken, config.workerId); if (!dispatched) { await config.store.failSyncJob(claimed.jobId, claimed.leaseToken).catch(() => {}); return true; } refreshAttempted = true; // Exactly one remote refresh attempt. An error is treated as uncertain: // rotating refresh tokens cannot safely be replayed after a lost result. grant = await config.sso.refresh(previous.refreshToken); } catch { await config.store.markRefreshUncertain(claimed.userId, claimed.characterId, refresh.leaseToken, config.workerId, refresh.expectedVersion).catch(() => {}); return true; } if (grant.characterId !== claimed.characterId || grant.ownerHash !== previous.ownerHash) { await config.store.markRefreshUncertain(claimed.userId, claimed.characterId, refresh.leaseToken, config.workerId, refresh.expectedVersion).catch(() => {}); return true; } const committed = await config.store.commitRefresh( claimed.userId, claimed.characterId, refresh.leaseToken, config.workerId, refresh.expectedVersion, { encryptedTokens: await config.cipher.seal(grant, `character:${claimed.characterId}`), tokenExpiresAt: grant.expiresAt, scopes: grant.scopes, jobId: claimed.jobId, jobLeaseToken: claimed.leaseToken }, ); if (!committed) { await config.store.markRefreshUncertain(claimed.userId, claimed.characterId, refresh.leaseToken, config.workerId, refresh.expectedVersion).catch(() => {}); await config.store.failSyncJob(claimed.jobId, claimed.leaseToken).catch(() => {}); return true; } authorityVersion = committed; // A refreshed credential cannot publish into a generation fenced to the // old credential version. Discard only that staging generation. if (claimed.generation !== undefined) await retryMutation( () => config.store.restartSyncJobGeneration(claimed.jobId, claimed.leaseToken), () => config.store.renewSyncJob(claimed.jobId, claimed.leaseToken, config.workerId, config.leaseMs), ); } const currentAuthority = await retryMutation( () => config.store.refreshSyncJobAuthority(claimed.jobId, claimed.leaseToken, authorityVersion), () => config.store.renewSyncJob(claimed.jobId, claimed.leaseToken, config.workerId, config.leaseMs), ); if (!grant.scopes.includes('esi-assets.read_assets.v1') || !currentAuthority.scopes.includes('esi-assets.read_assets.v1') || lostLease) { await config.store.failSyncJob(claimed.jobId, claimed.leaseToken).catch(() => {}); return true; } authorityVersion = currentAuthority; const generation = await retryMutation( () => config.store.beginSyncJob(claimed.jobId, claimed.leaseToken), () => config.store.renewSyncJob(claimed.jobId, claimed.leaseToken, config.workerId, config.leaseMs), ); const context = cooldownContext(claimed.userId, claimed.characterId); const cooldown = await config.store.cooldown(context, claimed.cooldownGroup); if (cooldown && cooldown.until > Date.now()) await config.sleep(cooldown.until - Date.now()); const authority = { userId: claimed.userId, principal: grant.ownerHash, characterId: claimed.characterId, scopes: authorityVersion.scopes, grantRevision: authorityVersion.grantRevision, authorizationEpoch: authorityVersion.authorizationEpoch, credentialVersion: authorityVersion.credentialVersion, } as const; const snapshot = await config.esi.getAssets(claimed.characterId, grant.accessToken, authority); const chunks = Math.ceil(snapshot.assets.length / 250); const requestedEnrichmentMs = config.enrichmentDeadlineMs; const configuredEnrichmentMs = Math.max(10_000, Math.min(10 * 60_000, Number.isFinite(requestedEnrichmentMs) ? requestedEnrichmentMs! : 120_000)); const leaseDeadline = typeof claimed.leaseExpiresAt === 'number' && claimed.leaseExpiresAt > Date.now() ? claimed.leaseExpiresAt - 5_000 : Number.POSITIVE_INFINITY; const enrichmentDeadlineAt = Math.min(Date.now() + configuredEnrichmentMs, leaseDeadline); const revalidate = async () => { if (lostLease) throw new Error('lease lost'); if (!(await config.store.renewSyncJob(claimed.jobId, claimed.leaseToken, config.workerId, config.leaseMs))) throw new Error('lease lost'); await config.store.refreshSyncJobAuthority(claimed.jobId, claimed.leaseToken, authorityVersion); }; for (let index = 0; index < chunks; index += 1) { if (lostLease) throw new Error('lease lost'); if (index > 0) await revalidate(); await retryMutation( () => config.store.stageSyncAssets(claimed.jobId, claimed.leaseToken, index, snapshot.assets.slice(index * 250, (index + 1) * 250)), () => config.store.renewSyncJob(claimed.jobId, claimed.leaseToken, config.workerId, config.leaseMs), ); } let enrichment: { status: 'complete' | 'partial' | 'failed' | 'unavailable'; version?: string; inputPin?: any; chunkCount: number; unresolvedCount: number; error?: string } | undefined; if (config.locationServiceFactory) { const assetIds = [...new Set(snapshot.assets.map((asset) => asset.itemId))].sort((left, right) => { const a = BigInt(left); const b = BigInt(right); return a < b ? -1 : a > b ? 1 : 0; }); const locationService = config.locationServiceFactory(grant.accessToken, { beforeStructure: async () => { if (Date.now() >= enrichmentDeadlineAt) throw new EnrichmentBudgetExceeded(); await revalidate(); }, onTransient: async (error) => { if (error instanceof EsiRateLimitError && config.store.recordSyncCooldown) await config.store.recordSyncCooldown(context, error.retryAt, error.cooldown.group ?? claimed.cooldownGroup); }, }); if (!locationService) { enrichment = { status: 'failed', chunkCount: 0, unresolvedCount: snapshot.assets.length, error: 'Reference enrichment unavailable' }; } else { const observations = await observeAssetHelpers(config, claimed, authority, grant.accessToken, assetIds, revalidate, enrichmentDeadlineAt); if (lostLease) throw new Error('lease lost'); let locationResult; try { if (Date.now() >= enrichmentDeadlineAt) throw new EnrichmentBudgetExceeded(); await revalidate(); locationResult = await locationService.enrich({ assets: snapshot.assets, authority, assetGeneration: generation, namesBatches: observations.names, locationsBatches: observations.locations, structureRevision: `${authority.grantRevision}:${authority.authorizationEpoch}:${authority.credentialVersion}`, }); } catch (error) { if (error instanceof EnrichmentBudgetExceeded) { enrichment = { status: 'failed', chunkCount: 0, unresolvedCount: snapshot.assets.length, error: 'Location enrichment deadline exceeded' }; } else if (error instanceof EsiRateLimitError || (error instanceof EsiError && (error.code === 'network' || error.code === 'deadline' || (error instanceof EsiHttpError && (error.status === 408 || error.status === 420 || error.status === 425 || error.status === 429 || error.status >= 500))))) { enrichment = { status: 'failed', chunkCount: 0, unresolvedCount: snapshot.assets.length, error: 'Optional location enrichment unavailable' }; } else if ((error instanceof EsiError && !(error instanceof EsiHttpError && error.status === 401)) || error instanceof TypeError) { // Optional endpoint/schema validation failures are represented as a // failed enrichment; authority, cancellation, and lease fences are // ordinary Errors and remain fatal below. enrichment = { status: 'failed', chunkCount: 0, unresolvedCount: snapshot.assets.length, error: 'Optional location enrichment validation failed' }; } else throw error; } if (locationResult) { const preparedChunks = prepareEnrichmentChunks(locationResult.rows); if (!preparedChunks) { enrichment = { status: 'failed', chunkCount: 0, unresolvedCount: snapshot.assets.length, error: 'Location enrichment persistence bound exceeded' }; } else { const locationChunks = preparedChunks.length; try { for (let index = 0; index < locationChunks; index += 1) { if (lostLease) throw new Error('lease lost'); await retryMutation( () => config.store.stageSyncEnrichment(claimed.jobId, claimed.leaseToken, index, preparedChunks[index]!), () => config.store.renewSyncJob(claimed.jobId, claimed.leaseToken, config.workerId, config.leaseMs), ); } enrichment = { status: locationResult.status, version: locationResult.enrichmentVersion, inputPin: locationResult.inputPin, chunkCount: locationChunks, unresolvedCount: locationResult.unresolved.length }; } catch (error) { if (!isOptionalEnrichmentEnvelopeError(error)) throw error; await config.store.resetSyncEnrichment(claimed.jobId, claimed.leaseToken); enrichment = { status: 'failed', chunkCount: 0, unresolvedCount: snapshot.assets.length, error: 'Location enrichment persistence envelope exceeded' }; } } } } } if (config.locationServiceFactory) await revalidate(); await retryMutation( () => config.store.commitSyncJob(claimed.jobId, claimed.leaseToken, chunks, snapshot.syncedAt, snapshot.expiresAt, enrichment), () => config.store.renewSyncJob(claimed.jobId, claimed.leaseToken, config.workerId, config.leaseMs), ); return true; } catch (error) { if (refreshAttempted && refreshLease) await config.store.markRefreshUncertain(claimed.userId, claimed.characterId, refreshLease.leaseToken, config.workerId, refreshLease.expectedVersion).catch(() => {}); if (error instanceof MutationUncertainError) { await config.store.deferSyncJob(claimed.jobId, claimed.leaseToken, Date.now() + 1_000).catch(() => {}); } else if (error instanceof EsiRateLimitError) { const retryAt = Math.min(Date.now() + 24 * 60 * 60_000, Math.max(Date.now() + 1_000, error.retryAt)); await config.store.deferSyncJob(claimed.jobId, claimed.leaseToken, retryAt, { context: cooldownContext(claimed.userId, claimed.characterId), until: retryAt, group: error.cooldown.group ?? DEFAULT_SYNC_COOLDOWN_GROUP }).catch(() => {}); } else if (error instanceof EsiError && (error.code === 'network' || error.code === 'deadline' || (error instanceof EsiHttpError && (error.status === 408 || error.status === 420 || error.status === 425 || error.status === 429 || error.status >= 500)))) { const retryAt = Math.min(Date.now() + 24 * 60 * 60_000, Date.now() + 1_000 * 2 ** Math.max(0, claimed.attempts - 1)); await config.store.deferSyncJob(claimed.jobId, claimed.leaseToken, retryAt).catch(() => {}); } else { await config.store.failSyncJob(claimed.jobId, claimed.leaseToken).catch(() => {}); } return true; } finally { stopHeartbeat(); } } function env(name: string): string { const value = process.env[name]; if (!value) throw new Error(`Sync worker configuration missing: ${name}`); return value; } export async function runSyncWorker(): Promise { const workerId = process.env.SYNC_WORKER_ID ?? `sync-worker-${crypto.randomUUID()}`; const pollMs = Math.max(250, Math.min(30_000, Number(process.env.SYNC_WORKER_POLL_MS ?? SYNC_WORKER_DEFAULTS.pollMs))); const leaseMs = Math.max(60_000, Math.min(60 * 60_000, Number(process.env.SYNC_WORKER_LEASE_MS ?? SYNC_WORKER_DEFAULTS.leaseMs))); const heartbeatMs = Math.max(10_000, Math.min(10 * 60_000, Number(process.env.SYNC_WORKER_HEARTBEAT_MS ?? SYNC_WORKER_DEFAULTS.heartbeatMs))); const store = new ConvexStore(env('CONVEX_URL'), undefined, env('CONVEX_SYNC_WORKER_SECRET')); const cipher = new TokenCipher(env('TOKEN_ENCRYPTION_KEY')); const userAgent = env('ESI_USER_AGENT'); const sso = new EveSso({ clientId: env('EVE_CLIENT_ID'), clientSecret: process.env.EVE_CLIENT_SECRET || undefined, redirectUri: process.env.EVE_CALLBACK_URL ?? 'http://localhost/unused-worker-callback', userAgent }); const esi = new EsiClient({ userAgent }); if (process.env.SYNC_WORKER_STARTUP_CHECK === '1') return; // The sync worker receives only a read-only reference catalog mount. It has // no reference compiler/import credentials and never writes to this path. let reference: SqliteReferenceRepository | null = null; let referenceUnavailable = false; const privateStructureCache = new PrivateStructureCache(); const configuredEnrichmentDeadline = Number(process.env.SYNC_WORKER_ENRICHMENT_DEADLINE_MS ?? 120_000); const config: WorkerConfig = { store, cipher, sso, esi, workerId, pollMs, leaseMs, heartbeatMs, sleep: (ms) => new Promise((resolve) => setTimeout(resolve, ms)), enrichmentDeadlineMs: Number.isFinite(configuredEnrichmentDeadline) ? configuredEnrichmentDeadline : 120_000, locationServiceFactory: (accessToken, hooks) => { if (referenceUnavailable) return null; try { reference ??= new SqliteReferenceRepository(env('REFERENCE_DIR'), 'production'); } catch { referenceUnavailable = true; return null; } const structures = new CachedPrivateStructurePort((context, structureId) => new EsiPrivateStructurePort(esi, accessToken, undefined, undefined, hooks ? { onTransient: hooks.onTransient } : undefined).access(context, structureId), privateStructureCache, hooks?.beforeStructure); return new InventoryLocationService({ reference, structures }); }, }; for (;;) { const handled = await processSyncJob(config); if (!handled) await config.sleep(pollMs); } } if (import.meta.main) await runSyncWorker();