Catalog sync

/** Every page, or an exception: never a partial catalog. */
export async function fetchWholeCatalog(o: CallOptions = {}, pageSize = 200): Promise<{ films: CatalogFilm[]; episodes: CatalogEpisode[] }> {
  const films = new Map<string, CatalogFilm>();
  const episodes = new Map<string, CatalogEpisode>();
  let cursor: string | null = null;
  for (let page = 0; page < 10_000; page++) {
    const qs: string = `limit=${pageSize}${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ''}`;
    const { data }: { data: CatalogPage } = await call<CatalogPage>('GET', `/api/delivery/v1/catalog?${qs}`, { policy: BACKGROUND, ...o });
    for (const f of data.films) films.set(f.id, f); // seen twice (catalog changed mid-walk): later copy wins
    for (const e of data.episodes) episodes.set(episodeKey(e), e);
    cursor = data.next_cursor ?? null;
    if (!cursor) return { films: [...films.values()], episodes: [...episodes.values()] };
  }
  throw new Error('catalog walk did not end');
}

/** One run. Returns false — and changes nothing — when the catalog could not be read completely. */
export async function syncCatalog(store: CatalogStore, o: CallOptions = {}): Promise<boolean> {
  let catalog: { films: CatalogFilm[]; episodes: CatalogEpisode[] };
  try {
    catalog = await fetchWholeCatalog(o);
  } catch (e) {
    // Keep the previous copy; alert if it keeps failing. A failure is never an empty catalog.
    log('catalog_sync_failed', { status: (e as { status?: number }).status, request_id: (e as { requestId?: string }).requestId });
    return false;
  }
  const seen = new Set<string>();
  const films = [...catalog.films].sort((a, b) => a.display_order - b.display_order); // keep Wandura's order
  for (const [i, f] of films.entries()) { await store.upsertFilm(f, i); seen.add(filmKey(f)); }
  for (const e of catalog.episodes) { await store.upsertEpisode(e); seen.add(episodeKey(e)); }
  for (const k of seen) await store.setAvailable(k, true);

  // Withdrawals: unavailable only after two complete walks in a row without the row. Never delete it.
  const before = await store.previousMisses();
  const missing = new Set<string>();
  for (const k of await store.availableKeys()) {
    if (seen.has(k)) continue;
    if (before.has(k)) await store.setAvailable(k, false);
    else missing.add(k);
  }
  await store.savePreviousMisses(missing);
  log('catalog_synced', { films: films.length, episodes: catalog.episodes.length, pending_withdrawals: missing.size });
  return true;
}

/** One API call under the retry policy. Resolves with the parsed body; throws WanduraApiError. */
export async function call<T>(method: 'GET' | 'POST', path: string, o: CallOptions = {}): Promise<{ data: T; headers: Headers }> {
  const apiKey = o.apiKey ?? process.env.WANDURA_API_KEY;
  if (!apiKey) throw new Error('WANDURA_API_KEY is not set');
  const p = o.policy ?? BACKGROUND;
  const budget = o.budget ?? sharedBudget;
  const started = Date.now();

  for (let attempt = 0; ; attempt++) {
    budget.noteRequest();
    let res: Response | null = null;
    let body: unknown = null;
    try {
      res = await (o.fetchImpl ?? fetch)(`${BASE_URL}${path}`, {
        method,
        headers: { Authorization: `Bearer ${apiKey}`, Accept: 'application/json', ...(o.body !== undefined ? { 'Content-Type': 'application/json' } : {}) },
        body: o.body !== undefined ? JSON.stringify(o.body) : undefined,
        signal: AbortSignal.timeout(p.timeoutMs),
      });
      body = await res.json().catch(() => null);
    } catch {
      res = null; // network error or timeout
    }
    if (res?.ok) return { data: body as T, headers: res.headers };

    const status = res?.status ?? null;
    const parsed = status === null ? { message: 'network error or timeout' } : parseError(status, body);
    const ra = Number(res?.headers.get('retry-after'));
    const err = new WanduraApiError({ ...parsed, status, retryAfterS: ra > 0 ? ra : null, requestId: parsed.requestId ?? (res ? requestIdOf(res.headers) : undefined) });
    log('wandura_call_failed', { method, path, status, error_code: err.errorCode, reason: err.reason, request_id: err.requestId, attempt, key: keyPrefix(apiKey) });

    const wait = backoffMs(attempt, err.retryAfterS, p);
    const late = Date.now() - started + wait > p.deadlineMs;
    if (!isRetryable(status, err.errorCode) || attempt >= p.maxRetries || late || !budget.tryRetry()) throw err;
    await (o.sleep ?? realSleep)(wait);
  }
}

def fetch_whole_catalog(page_size: int = 200, **call_opts: Any) -> Tuple[List[dict], List[dict]]:
    """Every page, or an exception: never a partial catalog."""
    films: Dict[str, dict] = {}
    episodes: Dict[str, dict] = {}
    cursor = None
    for _ in range(10_000):
        qs = f"limit={page_size}" + (f"&cursor={quote(cursor, safe='')}" if cursor else "")
        data, _headers = call("GET", f"/api/delivery/v1/catalog?{qs}", policy=BACKGROUND, **call_opts)
        for f in data["films"]:
            films[f["id"]] = f  # seen twice (catalog changed mid-walk): the later copy wins
        for e in data["episodes"]:
            episodes[episode_key(e)] = e
        cursor = data.get("next_cursor")
        if not cursor:
            return list(films.values()), list(episodes.values())
    raise RuntimeError("catalog walk did not end")

def sync_catalog(store: MemoryStore, **call_opts: Any) -> bool:
    """One run. Returns False - and changes nothing - when the catalog could not be read completely."""
    try:
        films, episodes = fetch_whole_catalog(**call_opts)
    except WanduraApiError as e:
        # Keep the previous copy; alert if it keeps failing. A failure is never an empty catalog.
        log("catalog_sync_failed", status=e.status, error_code=e.error_code, request_id=e.request_id)
        return False
    seen: Set[str] = set()
    for order, f in enumerate(sorted(films, key=lambda f: f["display_order"])):  # keep Wandura's order
        store.films[film_key(f)] = {**f, "_order": order}
        seen.add(film_key(f))
    for e in episodes:
        store.episodes[episode_key(e)] = e
        seen.add(episode_key(e))
    for k in seen:
        store.available[k] = True
    # Withdrawals: unavailable only after two complete walks in a row without the row. Never delete it.
    missing = set()
    for k, avail in list(store.available.items()):
        if avail and k not in seen:
            if k in store.misses:
                store.available[k] = False
            else:
                missing.add(k)
    store.misses = missing
    log("catalog_synced", films=len(films), episodes=len(episodes), pending_withdrawals=len(missing))
    return True