import type { ObservedFetchOptions } from "@/lib/observability/network/types"; import { officialPositionLookupKey, type OfficialPosition, type OfficialPositionLookupInput, } from "@/lib/domain/railway"; import type { OfficialPositionResolverConfig } from "@/lib/jrDataSystemEnvironment"; import { getOfficialPositionResolverConfig } from "@/lib/jrDataSystemEnvironment"; import { STORAGE_KEYS } from "@/constants/storage"; import { parseOfficialPositionCache, serializeOfficialPositionCache, } from "@/lib/officialPositionCache"; export type { OfficialPosition, OfficialPositionLookupInput, } from "@/lib/domain/railway"; export type OfficialPositionFetcher = ( url: string, options: ObservedFetchOptions, ) => Promise; export type OfficialPositionCacheStore = { load: () => Promise>; save: (entries: ReadonlyMap) => Promise; }; export type OfficialPositionDiagnostics = { requestCount: number; cacheHitCount: number; inFlightDedupeCount: number; unresolvedCount: number; networkFailureCount: number; }; export type ResolveOfficialPositionOptions = { forceRefresh?: boolean; signal?: AbortSignal; }; const POSITION_TYPES = new Set([ "unclassified", "station", "between", "prediction", "other", ]); const ENRICHMENT_STATUSES = new Set(["unreviewed", "partial", "reviewed"]); const isRecord = (value: unknown): value is Record => typeof value === "object" && value !== null; const requiredString = (value: unknown, label: string): string => { if (typeof value !== "string" || !value.trim()) { throw new Error(label + " must be a non-empty string"); } return value; }; const stringOrNull = (value: unknown, label: string): string | null => { if (value === null || value === undefined || value === "") return null; if (typeof value === "string") return value; if (typeof value === "number" && Number.isFinite(value)) return String(value); throw new Error(label + " must be a string or null"); }; const positionNumberString = (value: unknown): string => { if (typeof value === "string") { if (!value.trim()) throw new Error("pos_num must not be empty"); return value; } if (typeof value === "number" && Number.isFinite(value)) return String(value); throw new Error("pos_num must be a string or finite number"); }; const enumValue = ( value: unknown, allowed: Set, label: string, ): T => { if (typeof value !== "string" || !allowed.has(value)) { throw new Error(label + " is invalid"); } return value as T; }; export const parseOfficialPositionResponse = ( value: unknown, ): OfficialPosition => { const root = isRecord(value) ? value.data : undefined; const position = isRecord(root) ? root.position : undefined; if (!isRecord(position)) { throw new Error("OfficialPosition response is missing data.position"); } const revision = position.revision; if ( typeof revision !== "number" || !Number.isInteger(revision) || revision < 0 ) { throw new Error("revision must be a non-negative integer"); } // The SQL-backed resolver calls this field `note`; the retired position // endpoint called the same user-facing value `description`. Accept both so // a backend rollout cannot make the description disappear from the card. const note = position.note ?? position.description; return { positionId: requiredString(position.position_id, "position_id"), pos: requiredString(position.pos, "pos"), line: requiredString(position.line, "line"), posNum: positionNumberString(position.pos_num), positionType: enumValue(position.position_type, POSITION_TYPES, "position_type"), platform: stringOrNull(position.platform, "platform"), track: stringOrNull(position.track, "track"), note: stringOrNull(note, "note/description"), enrichmentStatus: enumValue( position.enrichment_status, ENRICHMENT_STATUSES, "enrichment_status", ), firstSeenAt: stringOrNull(position.first_seen_at, "first_seen_at"), lastSeenAt: stringOrNull(position.last_seen_at, "last_seen_at"), revision, }; }; const getStatus = (error: unknown): number | undefined => { if (typeof error === "object" && error !== null) { const status = (error as { status?: unknown }).status; return typeof status === "number" ? status : undefined; } return undefined; }; const isDevelopment = () => Boolean((globalThis as { __DEV__?: boolean }).__DEV__); const positionLabel = (input: OfficialPositionLookupInput) => "line=" + input.line + " pos=" + input.pos + " pos_num=" + String(input.posNum); const makeBackendUrl = ( config: OfficialPositionResolverConfig, input: OfficialPositionLookupInput, ): string => { const params = new URLSearchParams(); params.set("line", input.line); params.set("pos", input.pos); params.set("pos_num", String(input.posNum)); return ( config.backendApiBaseUrl.replace(/\/+$/, "") + "/api/v1/official-positions/resolve?" + params.toString() ); }; const defaultFetchJson: OfficialPositionFetcher = async ( url, requestOptions, ) => { const { observedFetchJson } = await import("@/lib/observability/network/observedFetch"); return observedFetchJson(url, requestOptions); }; /** * The first position snapshot can be rendered from this local warm cache. * The storage module is loaded lazily so the domain resolver remains usable in * Node tests and other non-native contexts. */ const defaultPersistentCacheStore: OfficialPositionCacheStore = { load: async () => { const { AS } = await import("@/storageControl"); return parseOfficialPositionCache( await AS.getItem(STORAGE_KEYS.OFFICIAL_POSITION_CACHE), ); }, save: async (entries) => { const { AS } = await import("@/storageControl"); await AS.setItem( STORAGE_KEYS.OFFICIAL_POSITION_CACHE, serializeOfficialPositionCache(entries), ); }, }; const abortAware = async ( promise: Promise, signal?: AbortSignal, ): Promise => { if (!signal) return promise; if (signal.aborted) return undefined; return new Promise((resolve, reject) => { const onAbort = () => { signal.removeEventListener("abort", onAbort); resolve(undefined); }; signal.addEventListener("abort", onAbort, { once: true }); promise.then( (value) => { signal.removeEventListener("abort", onAbort); resolve(value); }, (error) => { signal.removeEventListener("abort", onAbort); reject(error); }, ); }); }; export const createOfficialPositionResolver = (options?: { fetchJson?: OfficialPositionFetcher; maxConcurrency?: number; persistentCache?: OfficialPositionCacheStore; }) => { const fetchJson = options?.fetchJson ?? defaultFetchJson; const defaultConcurrency = Math.max( 1, Math.min(8, Math.floor(options?.maxConcurrency ?? 6)), ); const cache = new Map(); const inFlight = new Map>(); const persistentCache = options?.persistentCache; const persistentKeys = new Set(); const refreshAfterWarmCache = new Set(); let persistentCacheLoaded = !persistentCache; let persistentCacheLoad: Promise | undefined; let persistentCacheSaveScheduled = false; const diagnostics: OfficialPositionDiagnostics = { requestCount: 0, cacheHitCount: 0, inFlightDedupeCount: 0, unresolvedCount: 0, networkFailureCount: 0, }; const configKey = (config: OfficialPositionResolverConfig) => config.backendApiBaseUrl; const logDiagnostics = () => { if (isDevelopment()) { console.debug("OfficialPosition diagnostics", diagnostics); } }; const hydratePersistentCache = async (): Promise => { if (!persistentCache || persistentCacheLoaded) return; if (!persistentCacheLoad) { persistentCacheLoad = (async () => { try { const entries = await persistentCache.load(); for (const [key, value] of entries) { cache.set(key, value); persistentKeys.add(key); } } catch { // A malformed or unavailable local cache must never block the live API. } finally { persistentCacheLoaded = true; } })(); } await persistentCacheLoad; }; const schedulePersistentCacheSave = () => { if (!persistentCache || persistentCacheSaveScheduled) return; persistentCacheSaveScheduled = true; void Promise.resolve().then(async () => { persistentCacheSaveScheduled = false; const entries = new Map(); for (const [key, value] of cache) { if (value) entries.set(key, value); } try { await persistentCache.save(entries); } catch { // Persistence is an optimization; the in-memory resolver remains valid. } }); }; const resolve = async ( input: OfficialPositionLookupInput, config: OfficialPositionResolverConfig, resolveOptions?: ResolveOfficialPositionOptions, ): Promise => { let tupleKey: string; try { tupleKey = officialPositionLookupKey(input); } catch { diagnostics.unresolvedCount += 1; logDiagnostics(); return undefined; } const key = configKey(config) + ":" + tupleKey; if (persistentCache && !persistentCacheLoaded) { await hydratePersistentCache(); } const previous = cache.get(key); if (!resolveOptions?.forceRefresh && cache.has(key)) { const shouldRefresh = refreshAfterWarmCache.has(key); if (persistentKeys.delete(key)) { // Show the persisted value immediately. The next polling cycle is // allowed to refresh it from the SQL-backed API. refreshAfterWarmCache.add(key); } if (!shouldRefresh) { diagnostics.cacheHitCount += 1; logDiagnostics(); return abortAware( Promise.resolve(previous ?? undefined), resolveOptions?.signal, ); } refreshAfterWarmCache.delete(key); } if (resolveOptions?.forceRefresh) { persistentKeys.delete(key); refreshAfterWarmCache.delete(key); } const existing = inFlight.get(key); if (existing) { diagnostics.inFlightDedupeCount += 1; logDiagnostics(); return abortAware( existing.then((value) => value ?? undefined), resolveOptions?.signal, ); } const request = (async (): Promise => { diagnostics.requestCount += 1; logDiagnostics(); const url = makeBackendUrl(config, input); try { const raw = await fetchJson(url, { endpoint: "official_position_resolve", source: "backend_api", userVisible: false, preload: false, fetchPriority: "low", timeoutMs: 8000, retry: true, urlPathTemplate: "/api/v1/official-positions/resolve", }); const parsed = parseOfficialPositionResponse(raw); cache.set(key, parsed); persistentKeys.delete(key); refreshAfterWarmCache.delete(key); schedulePersistentCacheSave(); return parsed; } catch (error) { if (getStatus(error) === 404) { diagnostics.unresolvedCount += 1; logDiagnostics(); cache.set(key, null); persistentKeys.delete(key); refreshAfterWarmCache.delete(key); schedulePersistentCacheSave(); if (isDevelopment()) { console.debug("OfficialPosition unresolved: " + positionLabel(input)); } return null; } diagnostics.networkFailureCount += 1; if (persistentCache && previous) { // Keep the warm value visible but retry it on the next poll. refreshAfterWarmCache.add(key); } logDiagnostics(); if (isDevelopment()) { console.debug( "OfficialPosition resolver failed: " + positionLabel(input), ); } return previous ?? null; } })(); inFlight.set(key, request); try { return await abortAware( request.then((value) => value ?? undefined), resolveOptions?.signal, ); } finally { if (inFlight.get(key) === request) inFlight.delete(key); } }; const resolveMany = async ( inputs: readonly OfficialPositionLookupInput[], config: OfficialPositionResolverConfig, resolveOptions?: ResolveOfficialPositionOptions & { concurrency?: number }, ): Promise> => { const unique = new Map(); for (const input of inputs) { try { unique.set(officialPositionLookupKey(input), input); } catch { diagnostics.unresolvedCount += 1; } } const entries = Array.from(unique.entries()); const result = new Map(); let cursor = 0; const workerCount = Math.min( entries.length, Math.max( 1, Math.min(8, Math.floor(resolveOptions?.concurrency ?? defaultConcurrency)), ), ); const worker = async () => { while (cursor < entries.length) { const index = cursor++; const entry = entries[index]; if (!entry) continue; const [key, input] = entry; result.set(key, await resolve(input, config, resolveOptions)); } }; await Promise.all(Array.from({ length: workerCount }, () => worker())); return result; }; const getDiagnostics = (): OfficialPositionDiagnostics => ({ ...diagnostics }); return { resolve, resolveMany, getDiagnostics, }; }; const defaultResolver = createOfficialPositionResolver({ persistentCache: defaultPersistentCacheStore, }); export const resolveOfficialPositions = ( inputs: readonly OfficialPositionLookupInput[], config?: OfficialPositionResolverConfig, options?: ResolveOfficialPositionOptions & { concurrency?: number }, ) => defaultResolver.resolveMany( inputs, config ?? getOfficialPositionResolverConfig("production_release"), options, );