jrshikoku/lib/officialPositionService.ts

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<unknown>;

export type OfficialPositionCacheStore = {
  load: () => Promise<ReadonlyMap<string, OfficialPosition>>;
  save: (entries: ReadonlyMap<string, OfficialPosition>) => Promise<void>;
};

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<string, unknown> =>
  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 = <T extends string>(
  value: unknown,
  allowed: Set<string>,
  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<unknown>(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 <T>(
  promise: Promise<T>,
  signal?: AbortSignal,
): Promise<T | undefined> => {
  if (!signal) return promise;
  if (signal.aborted) return undefined;
  return new Promise<T | undefined>((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<string, OfficialPosition | null>();
  const inFlight = new Map<string, Promise<OfficialPosition | null>>();
  const persistentCache = options?.persistentCache;
  const persistentKeys = new Set<string>();
  const refreshAfterWarmCache = new Set<string>();
  let persistentCacheLoaded = !persistentCache;
  let persistentCacheLoad: Promise<void> | 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<void> => {
    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<string, OfficialPosition>();
      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<OfficialPosition | undefined> => {
    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<OfficialPosition | null> => {
      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<Map<string, OfficialPosition | undefined>> => {
    const unique = new Map<string, OfficialPositionLookupInput>();
    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<string, OfficialPosition | undefined>();
    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,
  );