feat: add pinned inventory location enrichment
This commit is contained in:
+210
-5
@@ -1,7 +1,11 @@
|
||||
import { ConvexStore, type RefreshLease } from '../src/lib/server/store';
|
||||
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 { EsiClient, EsiError, EsiHttpError, EsiRateLimitError } from '../src/lib/server/esi';
|
||||
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 = {
|
||||
@@ -21,11 +25,71 @@ export type WorkerConfig = {
|
||||
leaseMs: number;
|
||||
heartbeatMs: number;
|
||||
sleep: (ms: number) => Promise<void>;
|
||||
enrichmentDeadlineMs?: number;
|
||||
/** Present in the production worker; omitted by legacy/unit configurations. */
|
||||
locationServiceFactory?: (accessToken: string, hooks?: { beforeStructure: () => Promise<void>; onTransient: (error: unknown) => Promise<void> }) => 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<T extends NamesObservation | LocationsObservation>(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<ReturnType<EsiClient['getAssetNames']>>, characterId: number): NamesObservation {
|
||||
return { ...value, names: value.names.map((entry) => ({ itemId: entry.itemId, name: entry.name })), characterId };
|
||||
}
|
||||
function asLocationsObservation(value: Awaited<ReturnType<EsiClient['getAssetLocations']>>, 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; }
|
||||
@@ -33,7 +97,7 @@ export class MutationUncertainError extends Error {
|
||||
|
||||
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|chunks must be staged|chunk already exists|Snapshot is not staging|Snapshot is incomplete|Invalid snapshot|not active|missing|Invalid refreshed scopes/.test(message);
|
||||
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<T>(operation: () => Promise<T>, renew?: () => Promise<boolean>): Promise<T> {
|
||||
@@ -55,6 +119,56 @@ async function retryMutation<T>(operation: () => Promise<T>, renew?: () => Promi
|
||||
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<void>,
|
||||
deadlineAt: number,
|
||||
): Promise<{ names: NamesObservation[]; locations: LocationsObservation[] }> {
|
||||
const context = cooldownContext(claimed.userId, claimed.characterId);
|
||||
const readBatches = async <T extends NamesObservation | LocationsObservation>(reader: (ids: string[]) => Promise<T>): Promise<T[]> => {
|
||||
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<T>(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<T>(ids, 'unavailable', Date.now())); continue; }
|
||||
await config.sleep(delay);
|
||||
if (Date.now() >= deadlineAt) { unavailable = true; batches.push(helperObservationFailure<T>(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<T>(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<T>(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<T>(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<boolean> {
|
||||
const claimed = await config.store.claimSyncJob(config.workerId, config.leaseMs);
|
||||
@@ -164,15 +278,89 @@ export async function processSyncJob(config: WorkerConfig): Promise<boolean> {
|
||||
} 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),
|
||||
() => 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;
|
||||
@@ -211,8 +399,25 @@ export async function runSyncWorker(): Promise<void> {
|
||||
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 });
|
||||
const config: WorkerConfig = { store, cipher, sso, esi, workerId, pollMs, leaseMs, heartbeatMs, sleep: (ms) => new Promise((resolve) => setTimeout(resolve, ms)) };
|
||||
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);
|
||||
|
||||
Reference in New Issue
Block a user