import type { JsonValue } from "@earendil-works/chord";
import { diffRevisions, type Op } from "@earendil-works/chord/delta";

/** One json value: its current revision, then the ops of each next one. Durable watches are feeds. */
export type Feed = {
  readonly value: unknown;
  /** Never invoked inline; ops apply to `value` as read just before `start`. */
  start(listener: (value: unknown, ops: readonly Op[]) => Promise<void>): void;
  stop(): Promise<unknown>;
};

export type ClientMessage =
  | { t: "auth"; token: string }
  | { t: "sub"; topic: string }
  | { t: "unsub"; topic: string };

export type ServerMessage =
  | { t: "ready" }
  /** `older` counts the entries before `value.entries`, which `older` messages bring, newest first, before any ops. */
  | { t: "snapshot"; topic: string; value: unknown; older?: number }
  | { t: "older"; topic: string; entries: unknown[] }
  | { t: "ops"; topic: string; ops: readonly Op[] }
  | { t: "error"; topic?: string; message: string };

export type PubsubOptions = {
  authorize(token: string): boolean;
  /** `undefined` for unknown topics. */
  resolve(topic: string): Promise<Feed | undefined>;
};

const AUTH_TIMEOUT_MS = 10_000;

/** How much of a history its snapshot carries, of the newest; the rest follows in chunks of about `CHUNK_BYTES`. */
const TAIL_BYTES = 32_768;
const CHUNK_BYTES = 262_144;

/**
 * A snapshot's frames. A value with `entries`, as a conversation is, sends its newest entries first and then the rest,
 * so that a reader far away sees the latest soonest; each frame is encoded only as it is sent.
 */
function* snapshot(topic: string, value: unknown): Generator<string> {
  const entries = (value as { entries?: unknown } | null)?.entries;
  if (!Array.isArray(entries)) return yield JSON.stringify({ t: "snapshot", topic, value });
  // the encoded entries from `end` back that fit in `budget`; at least one fits
  const take = (end: number, budget: number) => {
    const taken: string[] = [];
    for (let at = end, bytes = 0; at > 0;) {
      const encoded = JSON.stringify(entries[at - 1]);
      if (taken.length > 0 && bytes + encoded.length > budget) break;
      taken.push(encoded);
      bytes += encoded.length;
      at--;
    }
    return taken.reverse();
  };
  const tail = take(entries.length, TAIL_BYTES);
  const from = entries.length - tail.length;
  const { entries: _, ...others } = value as { entries: unknown[] };
  const rest = JSON.stringify(others).slice(1, -1);
  yield `{"t":"snapshot","topic":${JSON.stringify(topic)},${
    from ? `"older":${from},` : ""
  }"value":{${rest}${rest ? "," : ""}"entries":[${tail.join(",")}]}}`;
  for (let end = from; end > 0;) {
    const chunk = take(end, CHUNK_BYTES);
    yield `{"t":"older","topic":${JSON.stringify(topic)},"entries":[${chunk.join(",")}]}`;
    end -= chunk.length;
  }
}

export function servePubsub(socket: WebSocket, { authorize, resolve }: PubsubOptions): void {
  let authorized = false;
  const subscriptions = new Map<string, Promise<Feed | undefined>>();
  const sendFrame = (frame: string) => {
    if (socket.readyState === WebSocket.OPEN) socket.send(frame);
  };
  const send = (message: ServerMessage) => sendFrame(JSON.stringify(message));
  const authTimeout = setTimeout(() => socket.close(4001, "unauthorized"), AUTH_TIMEOUT_MS);

  async function subscribe(topic: string) {
    if (subscriptions.has(topic)) return;
    const pending = resolve(topic);
    const entry = pending.catch(() => undefined);
    subscriptions.set(topic, entry);
    let feed: Feed | undefined;
    try {
      feed = await pending;
    } catch (error) {
      feed = undefined;
      console.error(`[pubsub] ${topic}:`, error);
    }
    // unsubscribed meanwhile: unsubscribe() stops the feed
    if (subscriptions.get(topic) !== entry) return;
    if (feed === undefined) {
      subscriptions.delete(topic);
      send({ t: "error", topic, message: `no topic ${topic}` });
      return;
    }
    for (const frame of snapshot(topic, feed.value)) sendFrame(frame);
    feed.start((_, ops) => Promise.resolve(send({ t: "ops", topic, ops })));
  }

  async function unsubscribe(topic: string) {
    const pending = subscriptions.get(topic);
    subscriptions.delete(topic);
    await (await pending)?.stop();
  }

  socket.addEventListener("message", (event) => {
    let message: ClientMessage;
    try {
      message = JSON.parse(String(event.data));
    } catch {
      socket.close(4000, "malformed message");
      return;
    }
    if (!authorized) {
      clearTimeout(authTimeout);
      if (message.t !== "auth" || !authorize(message.token)) {
        socket.close(4001, "unauthorized");
        return;
      }
      authorized = true;
      send({ t: "ready" });
      return;
    }
    if (message.t === "sub") void subscribe(message.topic);
    else if (message.t === "unsub") void unsubscribe(message.topic);
  });

  socket.addEventListener("close", () => {
    clearTimeout(authTimeout);
    for (const topic of [...subscriptions.keys()]) void unsubscribe(topic);
  });
}

/** A json value the server changes, which any number of feeds follow as a snapshot and then diffs. */
export function published<T>(initial: T) {
  let value = initial;
  const listeners = new Set<(value: unknown, ops: readonly Op[]) => Promise<void>>();
  return {
    get value() {
      return value;
    },
    set(next: T) {
      if (next === value) return;
      const ops = diffRevisions(value as JsonValue, next as JsonValue);
      value = next;
      if (ops.length === 0) return;
      for (const listener of listeners) void listener(value, ops);
    },
    feed(): Feed {
      let listener: ((value: unknown, ops: readonly Op[]) => Promise<void>) | undefined;
      return {
        get value() {
          return value;
        },
        start(next) {
          listener = next;
          listeners.add(next);
        },
        stop() {
          if (listener) listeners.delete(listener);
          return Promise.resolve();
        },
      };
    },
  };
}