import type { IndexCollection, IndexEvent } from './model.ts';
import { parseJetstreamMessage, subscribeUrl } from './jetstream.ts';
import { backfillAll, type BackfillDeps } from './backfill.ts';
import type { Upstream } from '../upstream.ts';

/** The slice of the WebSocket API this module needs, so tests can inject a fake without a
 * real socket. workerd's accepted client socket satisfies this. */
export interface WebSocketLike {
  addEventListener(type: 'message', listener: (event: { data: unknown }) => void): void;
  addEventListener(type: 'close' | 'error', listener: () => void): void;
  close(): void;
}

/** Opens Jetstream with a fetch upgrade rather than `new WebSocket`, so the handshake goes through
 * `Upstream` and carries our User-Agent. The socket is open once this resolves. */
export async function connectJetstream(fetchFn: Upstream, url: string): Promise<WebSocketLike> {
  const res = await fetchFn(url.replace(/^ws/, 'http'), { headers: { upgrade: 'websocket' } });
  const ws = (res as Response & { webSocket?: (WebSocketLike & { accept(): void }) | null }).webSocket;
  if (!ws) throw new Error(`Jetstream refused the WebSocket upgrade: HTTP ${res.status}`);
  ws.accept();
  return ws;
}

/** What ingestion writes to. `IndexModel` implements it. */
export interface IngestIndex {
  apply(event: IndexEvent): void;
  pruneRepo(collection: IndexCollection, did: string, keep: ReadonlySet<string>): void;
  /** Jetstream `time_us`: every event before it has been applied. */
  cursor(): number | undefined;
  setCursor(timeUs: number): void;
}

export interface IngestDeps {
  jetstreamUrl: string;
  /** Resolves once the socket is open. */
  connect(url: string): Promise<WebSocketLike>;
  index: IngestIndex;
  backfill: BackfillDeps;
  now(): number;
  /** How old a cursor can get before Jetstream can no longer replay from it. Default 24h. */
  replayWindowMs?: number;
  /** How long to drain the live tail before returning. Default 20s. */
  drainBudgetMs?: number;
  /** How close to "now" an event's `time` must be to call the drain caught up, and how long the
   * socket must stay silent to call it quiet. Default 5s. */
  caughtUpSkewMs?: number;
  /** How far one quiet drain may move the cursor past where it started. Jetstream scans replay
   * at roughly 12 minutes of network time per second (measured 2026-11-06), so a drain that
   * heard nothing has certainly scanned this much. Default 10 minutes. */
  maxQuietAdvanceMs?: number;
}

/**
 * One ingestion pass. If the cursor is missing or outside Jetstream's replay window, backfill
 * from the relay and each PDS first, then replay Jetstream from the moment the backfill started,
 * so events that land during the backfill are not lost. Then drain the live tail until caught
 * up or out of budget, saving the cursor as it goes. The socket is not held open past this call;
 * the DO's alarm (every 60s) is the only thing that keeps calling it.
 */
export async function ingest(deps: IngestDeps): Promise<void> {
  const replayWindowMs = deps.replayWindowMs ?? 24 * 60 * 60 * 1000;

  let from = deps.index.cursor();
  if (from === undefined || deps.now() - from / 1000 > replayWindowMs) {
    const mark = deps.now() * 1000;
    for await (const item of backfillAll(deps.backfill)) {
      if (item.kind === 'event') deps.index.apply(item.event);
      else deps.index.pruneRepo(item.collection, item.did, item.uris);
    }
    deps.index.setCursor(mark);
    from = mark;
  }

  await drainLiveTail(deps, from, {
    drainBudgetMs: deps.drainBudgetMs ?? 20_000,
    caughtUpSkewMs: deps.caughtUpSkewMs ?? 5_000,
    maxQuietAdvanceMs: deps.maxQuietAdvanceMs ?? 10 * 60 * 1000,
  });
}

function eventTimeUs(raw: unknown): number | undefined {
  const time = (raw as { payload?: { time?: unknown } } | null)?.payload?.time;
  if (typeof time !== 'string') return undefined;
  const ms = Date.parse(time);
  return Number.isFinite(ms) ? ms * 1000 : undefined;
}

async function drainLiveTail(
  deps: IngestDeps,
  fromUs: number,
  options: { drainBudgetMs: number; caughtUpSkewMs: number; maxQuietAdvanceMs: number },
): Promise<void> {
  const ws = await deps.connect(subscribeUrl(deps.jetstreamUrl, String(fromUs)));
  const openedAt = deps.now();
  return new Promise((resolve) => {
    let latestUs = fromUs;
    let lastMessageAt: number | undefined;
    let done = false;

    const advance = (timeUs: number) => {
      if (timeUs <= latestUs) return;
      latestUs = timeUs;
      deps.index.setCursor(timeUs);
    };

    // A dev.nibling.* stream is often silent for minutes, and Jetstream sends nothing on a
    // filtered stream with no matches. Without this, a quiet network never moves the cursor and
    // ages into a full backfill. Only a drain that stayed open and silent counts.
    const finish = (budgetSpent: boolean) => {
      if (done) return;
      done = true;
      clearTimeout(timer);
      if (budgetSpent && deps.now() - (lastMessageAt ?? openedAt) >= options.caughtUpSkewMs) {
        const quietMs = Math.min(openedAt - options.caughtUpSkewMs, fromUs / 1000 + options.maxQuietAdvanceMs);
        advance(quietMs * 1000);
      }
      ws.close();
      resolve();
    };
    const timer = setTimeout(() => finish(true), options.drainBudgetMs);

    ws.addEventListener('message', (event) => {
      if (done) return;
      lastMessageAt = deps.now();
      let raw: unknown;
      try {
        raw = typeof event.data === 'string' ? JSON.parse(event.data) : event.data;
      } catch {
        return;
      }
      const parsed = parseJetstreamMessage(raw);
      if (parsed) deps.index.apply(parsed);

      const timeUs = eventTimeUs(raw);
      if (timeUs === undefined) return;
      advance(timeUs);
      if (deps.now() - timeUs / 1000 <= options.caughtUpSkewMs) finish(false);
    });
    ws.addEventListener('close', () => finish(false));
    ws.addEventListener('error', () => finish(false));
  });
}