jrshikoku/lib/requestCoordinator.ts

/**
 * Coordinates one owner for a request source.
 *
 * A second request for the same source is skipped while the first one is
 * in-flight. Switching sources aborts the old request and advances the
 * generation so a late response cannot be committed.
 */
export type RequestToken = Readonly<{
  generation: number;
  sequence: number;
  source: string;
}>;

export type RequestHandle<T> = {
  token: RequestToken;
  signal: AbortSignal;
  promise: Promise<T>;
};

type ActiveRequest = {
  token: RequestToken;
  controller: AbortController;
};

export class RequestCoordinator {
  private active: ActiveRequest | null = null;
  private latestToken: RequestToken | null = null;
  private disposed = false;
  private generation = 0;
  private sequence = 0;

  start<T>(
    source: string,
    operation: (signal: AbortSignal) => Promise<T> | T,
  ): RequestHandle<T> | null {
    if (this.disposed) return null;

    if (this.active?.token.source === source) {
      return null;
    }

    this.active?.controller.abort();
    this.generation += 1;

    const token: RequestToken = {
      generation: this.generation,
      sequence: ++this.sequence,
      source,
    };
    const controller = new AbortController();
    this.latestToken = token;
    const active: ActiveRequest = { token, controller };
    this.active = active;

    const promise = Promise.resolve()
      .then(() => operation(controller.signal))
      .finally(() => {
        if (this.active?.token.sequence === token.sequence) {
          this.active = null;
        }
      });

    return {
      token,
      signal: controller.signal,
      promise,
    };
  }

  isCurrent(token: RequestToken): boolean {
    const latest = this.latestToken;
    return (
      !this.disposed &&
      latest !== null &&
      latest.generation === token.generation &&
      latest.sequence === token.sequence
    );
  }

  /**
   * Stop the current polling request without permanently disposing the
   * coordinator. A later focused screen can start a fresh request.
   */
  cancel(): void {
    this.active?.controller.abort();
    this.active = null;
    this.latestToken = null;
    this.generation += 1;
  }

  dispose(): void {
    if (this.disposed) return;
    this.cancel();
    this.disposed = true;
  }
}

export type CurrentTrainSnapshotSource =
  | "primary"
  | "gas_fallback"
  | "mock"
  | "recording";

export type CurrentTrainSnapshot<T> = {
  data: T;
  source: CurrentTrainSnapshotSource;
};

/**
 * Keeps the existing primary -> GAS fallback order while making cancellation
 * explicit. An aborted primary request must not start a fallback request.
 */
export async function loadWithCurrentTrainFallback<T>(args: {
  signal: AbortSignal;
  primary: (signal: AbortSignal) => Promise<T>;
  fallback: (signal: AbortSignal) => Promise<T>;
}): Promise<CurrentTrainSnapshot<T>> {
  try {
    return {
      data: await args.primary(args.signal),
      source: "primary",
    };
  } catch (error) {
    if (args.signal.aborted) throw error;
    return {
      data: await args.fallback(args.signal),
      source: "gas_fallback",
    };
  }
}

/**
 * Merge the partial GAS fallback without discarding fields from the last
 * complete response. A missing train number is never used as a map key.
 */
export const mergeCurrentTrainFallback = <T extends { num?: string }>(
  fallback: T[],
  previous: T[],
): T[] => {
  const previousByTrainNumber = new Map<string, T[]>();
  for (const train of previous) {
    if (!train.num) continue;
    const rows = previousByTrainNumber.get(train.num) ?? [];
    rows.push(train);
    previousByTrainNumber.set(train.num, rows);
  }

  const expandedAmbiguousTrainNumbers = new Set<string>();
  return fallback.flatMap((train) => {
    const candidates = train.num
      ? previousByTrainNumber.get(train.num) ?? []
      : [];
    const fallbackLine =
      typeof (train as { Line?: unknown }).Line === "string"
        ? (train as unknown as { Line: string }).Line
        : undefined;
    const lineCandidates = fallbackLine
      ? candidates.filter(
          (candidate) =>
            (candidate as { Line?: unknown }).Line === fallbackLine,
        )
      : candidates;
    // A fallback row without line context must not inherit one arbitrary row
    // when the complete snapshot contains the same train on multiple lines.
    // Keep the last line-specific snapshot so the UI can still expose every
    // candidate instead of silently replacing it with an ambiguous row.
    if (lineCandidates.length > 1) {
      if (train.num && expandedAmbiguousTrainNumbers.has(train.num)) return [];
      if (train.num) expandedAmbiguousTrainNumbers.add(train.num);
      return lineCandidates;
    }
    const existing = lineCandidates[0];
    return existing ? [{ ...existing, ...train }] : [train];
  });
};