/**
 * A Jetstream consumer, narrowed to a set of collections.
 *
 * Jetstream v2 serves the network as JSON envelopes on one websocket,
 * filtered by collection server-side. Each frame is `{$type: 'message',
 * payload: {...}}`; the payload's `$type` names the event kind and its
 * fields ride flat, not under a `commit` object. The v2 cursor is the
 * event's sequence number, the server refuses one it can no longer serve
 * with a pre-upgrade 400 named CursorTooOld, and deletes arrive as marker
 * events the consumer must fold. This consumer maps every frame onto one
 * of four shapes and hands it to the handler: a record write, a record
 * delete, an identity event, or an account event.
 *
 * Workers-clean: no node imports. Without an injected `decompress` the
 * subscription runs uncompressed; with one, it asks for zstd.
 */

/** Backoff bounds for reconnecting after the socket closes. */
const RECONNECT_MIN_MS = 1_000;
const RECONNECT_MAX_MS = 30_000;

/** Dead handshakes in a row before the stored cursor is given up on. */
const CURSOR_FAILURES = 3;

/** @typedef {import('./store.js').RecordWrite} RecordWrite */
/** @typedef {import('./store.js').RecordDelete} RecordDelete */

/**
 * @typedef {Object} SubscribeOptions
 * @property {string} service - Jetstream base URL, e.g. wss://jetstream.example
 * @property {string[]} collections
 * @property {AbortSignal} [signal]
 * @property {(event: RecordWrite|RecordDelete|IdentityEvent|AccountEvent) => Promise<void>|void} onEvent
 * @property {(seq: number) => Promise<void>|void} [onSeq] - every frame, so a
 *   restart resumes past the ones that indexed nothing
 * @property {(message: string) => void} [onNotice]
 * @property {string[]} [dids] - only these accounts
 * @property {number} [cursor] - Jetstream sequence to resume after
 * @property {(bytes: Uint8Array) => Promise<Uint8Array>} [decompress] - what
 *   unpacks a zstd frame; without it the subscription runs plain JSON
 * @property {(refused: number) => Promise<number|undefined>|number|undefined} [onCursorTooOld]
 *   what resumes after the server refuses the stored cursor; its return
 *   value becomes the new cursor, and the tail continues from there
 */

/**
 * The handle an account now answers to.
 * @typedef {Object} IdentityEvent
 * @property {'identity'} kind
 * @property {string} did
 * @property {string} handle
 * @property {number} seq
 */

/**
 * An account state change. `status: 'deleted'` is the one the fold needs:
 * every record the account held is gone from the network.
 * @typedef {Object} AccountEvent
 * @property {'account'} kind
 * @property {string} did
 * @property {string|undefined} status
 * @property {number} seq
 */

/**
 * The event kind a frame payload names, by its `$type` suffix. A frame
 * without one reads as kind 'commit' only when the v1 `kind` field says so.
 * @param {Record<string, *>} event
 * @returns {string}
 */
function kindOf(event) {
  const type = typeof event.$type === 'string' ? event.$type : '';
  const kind = type.slice(type.lastIndexOf('#') + 1);
  if (kind) return kind;
  return typeof event.kind === 'string' ? event.kind : '';
}

/**
 * The record write one commit event carries, or null for anything else:
 * other kinds and malformed frames. A delete becomes a RecordDelete so the
 * fold can remove what its author withdrew.
 * @param {unknown} body - one parsed Jetstream frame
 * @param {string[]} collections
 * @returns {RecordWrite|RecordDelete|null}
 */
export function jetstreamWrite(body, collections) {
  const event = /** @type {Record<string, *>} */ (body ?? {});
  const payload = event.payload ?? event;
  if (kindOf(payload) !== 'commit' || typeof payload.did !== 'string') {
    return null;
  }
  const { collection, rkey, operation } = payload;
  if (typeof collection !== 'string' || !collections.includes(collection)) {
    return null;
  }
  if (typeof rkey !== 'string') return null;
  const uri = `at://${payload.did}/${collection}/${rkey}`;
  const seq =
    typeof payload.seq === 'number'
      ? payload.seq
      : typeof payload.time_us === 'number'
        ? payload.time_us
        : 0;
  if (operation === 'delete') {
    return { did: payload.did, collection, rkey, uri, seq, deleted: true };
  }
  if (operation !== 'create' && operation !== 'update') return null;
  const { cid, record } = payload;
  if (typeof cid !== 'string') return null;
  if (!record || typeof record !== 'object') return null;
  return { did: payload.did, collection, rkey, cid, uri, seq, value: record };
}

/**
 * @param {string} service - Jetstream base URL, e.g. wss://host
 * @param {string[]} collections
 * @param {number} [cursor]
 * @param {string[]} [dids]
 * @returns {URL}
 */
export function jetstreamUrl(service, collections, cursor, dids) {
  const url = new URL(
    '/xrpc/network.bsky.jetstream.subscribeEvents',
    service.replace(/^http/, 'ws'),
  );
  for (const collection of collections) {
    url.searchParams.append('collections', collection);
  }
  for (const did of dids ?? []) {
    url.searchParams.append('dids', did);
  }
  if (cursor !== undefined && cursor > 0) {
    url.searchParams.set('cursor', String(cursor));
  }
  return url;
}

/**
 * Consume a Jetstream endpoint until the signal aborts, reconnecting from
 * the last sequence seen. Resolves only when aborted.
 *
 * A stored cursor the server refuses as too old (a v2 CursorTooOld arrives
 * as a failed handshake, not a readable message) reads here as several
 * connects in a row that carry no events. After that the consumer gives up
 * the cursor and tails live; older records are the backfill's business.
 * @param {SubscribeOptions} options
 * @returns {Promise<void>}
 */
export async function subscribeJetstream(options) {
  const {
    service,
    collections,
    signal,
    onEvent,
    onSeq = () => {},
    onNotice = () => {},
    decompress,
  } = options;
  let cursor = options.cursor;
  let backoff = RECONNECT_MIN_MS;
  let deadHandshakes = 0;

  while (!signal?.aborted) {
    const closed = await new Promise((resolve) => {
      const url = jetstreamUrl(service, collections, cursor, options.dids);
      if (decompress) {
        url.searchParams.set('requireCompression', 'true');
      }
      const socket = new WebSocket(url, decompress ? ['zstd'] : []);
      /** @type {Promise<void>} */
      let queue = Promise.resolve();
      let sawEvent = false;

      const abort = () => socket.close();
      signal?.addEventListener('abort', abort, { once: true });
      const finish = (/** @type {string|null} */ reason) => {
        signal?.removeEventListener('abort', abort);
        resolve(reason);
      };

      socket.addEventListener('open', () => {
        backoff = RECONNECT_MIN_MS;
        onNotice(`watching ${url.host} from ${cursor ?? 'now'}`);
      });
      // Each message is handled to completion before the next one starts,
      // the same ordering promise the native consumer makes.
      socket.addEventListener('message', (event) => {
        queue = queue.then(async () => {
          const body = JSON.parse(
            String(await frameBody(event.data, decompress)),
          );
          const write = jetstreamWrite(body, collections);
          if (write) {
            sawEvent = true;
            await onEvent(write);
          }
          const identity = identityOf(body);
          if (identity) {
            sawEvent = true;
            await onEvent(identity);
          }
          const account = accountOf(body);
          if (account) {
            sawEvent = true;
            await onEvent(account);
          }
          const at = seqOf(body);
          if (typeof at === 'number') {
            cursor = at;
            await onSeq(at);
          }
        });
        queue = queue.catch((err) => {
          onNotice(
            `event dropped: ${err instanceof Error ? err.message : err}`,
          );
        });
      });
      socket.addEventListener('error', () => finish('connection failed'));
      socket.addEventListener('close', () => {
        queue.then(() => finish(sawEvent ? null : 'no events'));
      });
    });

    if (signal?.aborted) return;
    if (closed) onNotice(`${service}: ${closed}`);
    if (closed === 'no events' && cursor !== undefined) {
      deadHandshakes += 1;
      if (deadHandshakes >= CURSOR_FAILURES) {
        if (options.onCursorTooOld) {
          onNotice(`cursor ${cursor} refused; asking the caller to resume`);
          const fresh = await options.onCursorTooOld(cursor);
          cursor = fresh ?? undefined;
        } else {
          onNotice(`cursor ${cursor} refused; tailing live`);
          cursor = undefined;
        }
        deadHandshakes = 0;
      }
    } else {
      deadHandshakes = 0;
    }
    await new Promise((resolve) => setTimeout(resolve, backoff));
    backoff = Math.min(backoff * 2, RECONNECT_MAX_MS);
  }
}

/**
 * The sequence one frame names: the payload's, the envelope's, then the
 * v1 time fallback. An envelope carries its own cursor, which is the same
 * number the payload names.
 * @param {Record<string, *>} body
 * @returns {number|undefined}
 */
function seqOf(body) {
  const payload = body.payload ?? body;
  if (typeof payload.seq === 'number') return payload.seq;
  if (typeof body.cursor === 'number') return body.cursor;
  if (typeof payload.time_us === 'number') return payload.time_us;
  return undefined;
}

/**
 * A body that arrives compressed is binary, handed to the injected
 * decompressor. Plain JSON arrives as text.
 * @param {unknown} data
 * @param {((bytes: Uint8Array) => Promise<Uint8Array>)|undefined} decompress
 * @returns {Promise<string>}
 */
async function frameBody(data, decompress) {
  if (typeof data === 'string') return data;
  const bytes =
    data instanceof ArrayBuffer
      ? new Uint8Array(data)
      : data instanceof Uint8Array
        ? data
        : typeof Blob !== 'undefined' && data instanceof Blob
          ? new Uint8Array(await data.arrayBuffer())
          : null;
  if (!bytes) throw new Error('jetstream sent an unreadable frame');
  if (!decompress) return new TextDecoder().decode(bytes);
  return new TextDecoder().decode(await decompress(bytes));
}

/**
 * The identity event behind a frame body, or null. The handle is what the
 * index keeps; the rest of the identity body is not ours to interpret.
 * @param {Record<string, *>} body
 * @returns {IdentityEvent|null}
 */
function identityOf(body) {
  const event = body.payload ?? body;
  if (kindOf(event) !== 'identity' || typeof event.did !== 'string') {
    return null;
  }
  const handle = event.identity?.handle;
  if (typeof handle !== 'string' || !handle) return null;
  return {
    kind: 'identity',
    did: event.did,
    handle,
    seq: typeof event.seq === 'number' ? event.seq : 0,
  };
}

/**
 * The account event behind a frame body, or null. Only the status rides
 * along, and only the deleted one means anything to the fold.
 * @param {Record<string, *>} body
 * @returns {AccountEvent|null}
 */
function accountOf(body) {
  const event = body.payload ?? body;
  if (kindOf(event) !== 'account' || typeof event.did !== 'string') {
    return null;
  }
  const status = event.status ?? event.account?.status;
  return {
    kind: 'account',
    did: event.did,
    status: typeof status === 'string' ? status : undefined,
    seq: typeof event.seq === 'number' ? event.seq : 0,
  };
}