import { BehaviorSubject, filter, firstValueFrom, map, Subject } from "rxjs";
import type { Observable, Subscription } from "rxjs";

import { logger } from "../core/log";
import { MessageBusError } from "./errors";
import { createMeta } from "./message";
import type { OsMessage } from "./message";
import { topicMigrations, WORKBENCH_TOPIC_MANIFEST } from "./topics";
import type {
  EventTopic,
  PayloadOf,
  ReplyOf,
  StateTopic,
  TopicManifest,
  TopicMigration,
  TopicName,
  ValueOf,
} from "./topics";

/**
 * A state topic's read shape. Piping it drops `getCurrent` and `firstValue` (the
 * result is a plain `Observable`) — read them off the un-piped source.
 * @public
 */
export interface StateSource<T> extends Observable<T> {
  getCurrent(): T | undefined;
  firstValue: Promise<T>;
}

/**
 * The one teardown idiom: abort the `signal` to cancel a `query` or tear down a
 * `subscribe`. One controller scopes any number of them.
 * @public
 */
export interface AbortOptions {
  signal?: AbortSignal;
}

/**
 * `emit` options. `timeout` (default 5s; `null` disables the deadline so the
 * `signal` governs) composes with the `signal` — whichever fires first rejects.
 * @public
 */
export interface EmitOptions extends AbortOptions {
  timeout?: number | null;
}

/**
 * An event `emit`'s result. Awaiting it arms the failure machinery; not awaiting
 * is pure fire-and-forget.
 * @public
 */
export interface EmitResult<R> extends PromiseLike<R> {
  catch<T = never>(
    onRejected?: (reason: unknown) => T | PromiseLike<T>,
  ): Promise<R | T>;
  finally(onFinally?: () => void): Promise<R>;
}

/**
 * The in-process message bus. State topics hold a current value (`query` /
 * `subscribe`); event topics stream occurrences, and a request topic (one that
 * declares a reply) resolves an awaited `emit` with that reply.
 * @public
 */
export interface Bus {
  /** Publish a state topic's current value (fire-and-forget). */
  emit<K extends StateTopic>(type: K, value: ValueOf<K>): void;
  /**
   * Fire an event, or `await` it for a request topic's reply. A topic whose
   * payload is `void` takes no payload argument.
   */
  emit<K extends EventTopic>(
    type: K,
    ...rest: PayloadOf<K> extends void
      ? [payload?: void, options?: EmitOptions]
      : [payload: PayloadOf<K>, options?: EmitOptions]
  ): EmitResult<ReplyOf<K>>;
  /**
   * Read a state topic's value once. Seeded topics resolve at once; a suspending
   * one waits for its first value, bounded by the deadline and `signal` (rejects
   * `TIMEOUT`/`ABORTED`).
   */
  query<K extends StateTopic>(
    type: K,
    options?: AbortOptions,
  ): Promise<ValueOf<K>>;
  /**
   * Run `handler` for each event (it may `reply`). Tear down by aborting
   * `options.signal`; with none, it lives for the app's lifetime.
   */
  subscribe<K extends EventTopic>(
    type: K,
    handler: (message: OsMessage<PayloadOf<K>, ReplyOf<K>>) => void,
    options?: AbortOptions,
  ): void;
  /** React to a state topic's value; tears down when `options.signal` aborts. */
  subscribe<K extends StateTopic>(
    type: K,
    handler: (value: ValueOf<K>) => void,
    options?: AbortOptions,
  ): void;
  /** Compose a state topic as a `StateSource` (Observable + synchronous read). */
  subscribe<K extends StateTopic>(type: K): StateSource<ValueOf<K>>;
  /** Compose an event topic as an `Observable` of its payloads. */
  subscribe<K extends EventTopic>(type: K): Observable<PayloadOf<K>>;
}

// Registered symbols, so every federated copy resolves the same keys. BUS holds
// the shared core (a named global would collide). PROTOCOL is the one eternal
// key: the layout behind REGISTRY may change with the protocol number, but
// PROTOCOL must stay readable by every copy ever shipped.
const BUS = Symbol.for("sanity.os.bus");
const PROTOCOL = Symbol.for("sanity.os.protocol");
const REGISTRY = Symbol.for("sanity.os.registry");
const REQUEST = Symbol.for("sanity.os.request");

// The structural contract between copies. Bumping it is a breaking event,
// expected never — the check exists to fail loudly instead of corrupting.
const OS_PROTOCOL = 1;

const log = logger.child("os");

const DEFAULT_APP_ID = "workbench";
const DEFAULT_TIMEOUT_MS = 5000;

// What a suspending topic holds until its first value; never exposed to readers.
const NO_VALUE = Symbol.for("sanity.os.no-value");

type Outcome =
  | { readonly ok: true; readonly value: unknown }
  | { readonly ok: false; readonly error: unknown };

// One in-flight request, carried in the emit closure and on the message — no
// registry map to leak. No Promise exists until the first await, so a
// fire-and-forget emit can't raise an unhandled rejection.
interface PendingRequest {
  readonly responderAbort: AbortController;
  settled: boolean;
  capturedOutcome?: Outcome;
  deliver?: (outcome: Outcome) => void;
  replyPromise?: Promise<unknown>;
}

interface Registry {
  readonly appId: string;
  readonly topics: Map<string, "state" | "event">;
  readonly stateSubjects: Map<string, BehaviorSubject<unknown>>;
  readonly stateSources: Map<string, StateSource<unknown>>;
  readonly eventSubjects: Map<string, Subject<OsMessage<unknown, unknown>>>;
  readonly responderCounts: Map<string, number>;
  // One controller per app id — disconnectApp tears an app's footprint down
  // in one abort.
  readonly appAborts: Map<string, AbortController>;
  readonly migrations: ReadonlyMap<string, readonly TopicMigration[]>;
}

type InternalBus = Bus & { [REGISTRY]: Registry; [PROTOCOL]: number };

function resolveStateSubject(
  registry: Registry,
  type: string,
): BehaviorSubject<unknown> {
  let subject = registry.stateSubjects.get(type);
  if (!subject) {
    // Read before declared: hold the sentinel so the read waits instead of
    // resolving a misleading `undefined`.
    log.warn(`state topic "${type}" read before any value was published`);
    subject = new BehaviorSubject<unknown>(NO_VALUE);
    registry.stateSubjects.set(type, subject);
  }
  return subject;
}

// The one constructor for a state topic's read shape. Its contract carries two
// load-bearing invariants: `getCurrent` reports `undefined` until the first
// value (a Suspense binding waits on it), and stays reference-stable per
// underlying value (`useSyncExternalStore` spins otherwise).
function toStateSource(
  values: Observable<unknown>,
  getCurrent: () => unknown,
): StateSource<unknown> {
  const source = values as StateSource<unknown>;
  source.getCurrent = getCurrent;
  source.firstValue = firstValueFrom(values);
  return source;
}

function resolveStateSource(
  registry: Registry,
  type: string,
): StateSource<unknown> {
  let source = registry.stateSources.get(type);
  if (!source) {
    const subject = resolveStateSubject(registry, type);
    source = toStateSource(
      subject.pipe(filter((value) => value !== NO_VALUE)),
      () => {
        const current = subject.getValue();
        return current === NO_VALUE ? undefined : current;
      },
    );
    registry.stateSources.set(type, source);
  }
  return source;
}

// A connecting copy may know topics the installed bus doesn't, so add them; if
// a topic name is state in one copy and event in another, reject the whole
// manifest before changing anything.
function registerTopics(registry: Registry, manifest: TopicManifest): void {
  for (const [type, entry] of Object.entries(manifest)) {
    const existing = registry.topics.get(type);
    if (existing && existing !== entry.kind) {
      throw new MessageBusError(
        "PROTOCOL_MISMATCH",
        `topic "${type}" is declared "${entry.kind}" but the installed bus knows it as "${existing}"`,
      );
    }
  }
  for (const [type, entry] of Object.entries(manifest)) {
    registry.topics.set(type, entry.kind);
    if (entry.kind === "state" && !registry.stateSubjects.has(type)) {
      registry.stateSubjects.set(
        type,
        new BehaviorSubject<unknown>(
          entry.seed === undefined ? NO_VALUE : entry.seed,
        ),
      );
    }
  }
}

function resolveEventSubject(
  registry: Registry,
  type: string,
): Subject<OsMessage<unknown, unknown>> {
  let subject = registry.eventSubjects.get(type);
  if (!subject) {
    subject = new Subject<OsMessage<unknown, unknown>>();
    registry.eventSubjects.set(type, subject);
  }
  return subject;
}

/** Settle a request once, from whichever side fires first (reply/throw/timeout/abort). */
function settle(request: PendingRequest, outcome: Outcome): void {
  if (request.settled) return;
  request.settled = true;
  if (request.deliver) request.deliver(outcome);
  else request.capturedOutcome = outcome; // caller hasn't awaited yet — stash it
}

function createMessage(
  appId: string,
  type: string,
  payload: unknown,
  request: PendingRequest,
): OsMessage<unknown, unknown> {
  const message: OsMessage<unknown, unknown> & { [REQUEST]?: PendingRequest } =
    {
      type: type as TopicName,
      payload,
      meta: createMeta(appId),
      reply: (value) => {
        if (request.settled) {
          log.warn(
            `reply ignored for "${type}": no waiting caller or already replied`,
          );
          return;
        }
        settle(request, { ok: true, value });
      },
      get signal() {
        return request.responderAbort.signal;
      },
    };

  // The reply routes through the request riding the message — no registry lookup.
  message[REQUEST] = request;
  return message;
}

function createReplyPromise(
  request: PendingRequest,
  options: EmitOptions & { hadResponder: boolean },
): Promise<unknown> {
  return new Promise<unknown>((resolve, reject) => {
    const deliver = (outcome: Outcome) =>
      outcome.ok ? resolve(outcome.value) : reject(outcome.error);

    // A same-tick reply already settled this.
    if (request.capturedOutcome) {
      request.responderAbort.abort();
      deliver(request.capturedOutcome);
      return;
    }
    // NO_RESPONDER is judged at send time: nothing was listening then.
    if (!options.hadResponder) {
      reject(new MessageBusError("NO_RESPONDER"));
      return;
    }

    const timeoutMs =
      options.timeout === undefined ? DEFAULT_TIMEOUT_MS : options.timeout;
    let timer: ReturnType<typeof setTimeout> | undefined;
    const onAbort = () =>
      settle(request, { ok: false, error: new MessageBusError("ABORTED") });

    request.deliver = (outcome) => {
      if (timer !== undefined) clearTimeout(timer);
      options.signal?.removeEventListener("abort", onAbort);
      request.responderAbort.abort(); // caller is done waiting — let the responder bail
      deliver(outcome);
    };

    if (timeoutMs !== null) {
      timer = setTimeout(
        () =>
          settle(request, { ok: false, error: new MessageBusError("TIMEOUT") }),
        timeoutMs,
      );
    }
    if (options.signal) {
      if (options.signal.aborted) onAbort();
      else options.signal.addEventListener("abort", onAbort, { once: true });
    }
  });
}

function emitEvent(
  registry: Registry,
  type: string,
  payload: unknown,
  options: EmitOptions | undefined,
  appId: string,
): EmitResult<unknown> {
  const hadResponder = (registry.responderCounts.get(type) ?? 0) > 0;
  const request: PendingRequest = {
    responderAbort: new AbortController(),
    settled: false,
  };

  resolveEventSubject(registry, type).next(
    createMessage(appId, type, payload, request),
  );

  // Failure machinery arms on the first await (see PendingRequest).
  const awaitReply = () =>
    (request.replyPromise ??= createReplyPromise(request, {
      ...options,
      hadResponder,
    }));
  return {
    // oxlint-disable-next-line unicorn/no-thenable -- intentional PromiseLike: awaiting it arms the failure machinery; not awaiting is fire-and-forget.
    then: (onFulfilled, onRejected) =>
      awaitReply().then(onFulfilled, onRejected),
    catch: (onRejected) => awaitReply().then(undefined, onRejected),
    finally: (onFinally) => awaitReply().finally(onFinally),
  };
}

const isStateTopic = (registry: Registry, type: string): boolean =>
  registry.topics.get(type) === "state";

function emit(
  registry: Registry,
  type: string,
  payload: unknown,
  options: EmitOptions | undefined,
  appId: string,
): EmitResult<unknown> | undefined {
  if (isStateTopic(registry, type)) {
    const subject = resolveStateSubject(registry, type);
    // Producers may republish freely (e.g. on every machine state entry);
    // subscribers only ever see actual changes.
    if (!Object.is(subject.getValue(), payload)) subject.next(payload);
    return undefined;
  }
  return emitEvent(registry, type, payload, options, appId);
}

function query(
  registry: Registry,
  type: string,
  options: AbortOptions | undefined,
): Promise<unknown> {
  const source = resolveStateSource(registry, type);
  const current = source.getCurrent();
  if (current !== undefined) return Promise.resolve(current);

  const signal = options?.signal;
  if (signal?.aborted) return Promise.reject(new MessageBusError("ABORTED"));
  // The deadline keeps a never-producing topic from hanging the caller.
  return new Promise((resolve, reject) => {
    const timer = setTimeout(
      () =>
        reject(new MessageBusError("TIMEOUT", `query("${type}") timed out`)),
      DEFAULT_TIMEOUT_MS,
    );
    const onAbort = () => {
      clearTimeout(timer);
      reject(new MessageBusError("ABORTED"));
    };
    signal?.addEventListener("abort", onAbort, { once: true });
    void source.firstValue.then((value) => {
      clearTimeout(timer);
      signal?.removeEventListener("abort", onAbort);
      resolve(value);
    });
  });
}

function unsubscribeOnAbort(
  subscription: Subscription,
  signal: AbortSignal | undefined,
): void {
  if (!signal) return;
  if (signal.aborted) {
    subscription.unsubscribe();
    return;
  }
  signal.addEventListener("abort", () => subscription.unsubscribe(), {
    once: true,
  });
}

function invokeResponder(
  handler: (message: OsMessage<unknown, unknown>) => unknown,
  message: OsMessage<unknown, unknown> & { [REQUEST]?: PendingRequest },
): void {
  const request = message[REQUEST];
  const fail = (error: unknown) => {
    if (request) {
      settle(request, {
        ok: false,
        error: new MessageBusError("HANDLER_THREW", undefined, {
          cause: error,
        }),
      });
    }
  };
  try {
    const result = handler(message);
    // An async responder that rejects surfaces as HANDLER_THREW to the caller.
    if (result instanceof Promise) result.catch(fail);
  } catch (error) {
    fail(error);
  }
}

function subscribe(
  registry: Registry,
  type: string,
  handler: ((arg: never) => void) | undefined,
  options: AbortOptions | undefined,
): StateSource<unknown> | Observable<unknown> | void {
  const isState = isStateTopic(registry, type);

  if (!handler) {
    return isState
      ? resolveStateSource(registry, type)
      : resolveEventSubject(registry, type).pipe(
          map((message) => message.payload),
        );
  }

  // Handler form returns nothing — teardown is via `options.signal`, the one
  // idiom shared with `query`.
  if (isState) {
    const subscription = resolveStateSource(registry, type).subscribe(
      handler as (value: unknown) => void,
    );
    unsubscribeOnAbort(subscription, options?.signal);
    return;
  }

  // Event handler = responder: counts toward NO_RESPONDER and may `reply`.
  registry.responderCounts.set(
    type,
    (registry.responderCounts.get(type) ?? 0) + 1,
  );
  const subscription = resolveEventSubject(registry, type).subscribe(
    (message) =>
      invokeResponder(
        handler as (message: OsMessage<unknown, unknown>) => unknown,
        message,
      ),
  );
  subscription.add(() => {
    const count = registry.responderCounts.get(type) ?? 0;
    if (count > 0) registry.responderCounts.set(type, count - 1);
  });
  unsubscribeOnAbort(subscription, options?.signal);
}

// This copy's per-topic chains, pinning its versions when a caller supplies none.
const bundledMigrations = () => new Map(Object.entries(topicMigrations));

/**
 * Build an isolated bus with no `globalThis` install — the test seam.
 * `migrations` lets a test stand up synthetic topic chains; production passes
 * nothing.
 */
export function createBus(
  appId: string = DEFAULT_APP_ID,
  config: {
    migrations?: ReadonlyMap<string, readonly TopicMigration[]>;
  } = {},
): Bus {
  const registry: Registry = {
    appId,
    topics: new Map(),
    stateSubjects: new Map(),
    stateSources: new Map(),
    eventSubjects: new Map(),
    responderCounts: new Map(),
    appAborts: new Map(),
    migrations: config.migrations ?? bundledMigrations(),
  };

  registerTopics(registry, WORKBENCH_TOPIC_MANIFEST);

  const api = {
    emit: (type: string, payload: unknown, options?: EmitOptions) =>
      emit(registry, type, payload, options, registry.appId),
    query: (type: string, options?: AbortOptions) =>
      query(registry, type, options),
    subscribe: (
      type: string,
      handler?: (arg: never) => void,
      options?: AbortOptions,
    ) => subscribe(registry, type, handler, options),
  };

  const instance = api as unknown as InternalBus;
  instance[REGISTRY] = registry;
  instance[PROTOCOL] = OS_PROTOCOL;
  return instance;
}

// A topic's version is the highest `to` its chain reaches — derived, never
// hand-declared.
export function topicVersions(
  migrations: ReadonlyMap<string, readonly TopicMigration[]>,
): Map<string, number> {
  const versions = new Map<string, number>();
  for (const [topic, steps] of migrations) {
    let version = 1;
    for (const step of steps) if (step.to > version) version = step.to;
    versions.set(topic, version);
  }
  return versions;
}

// Recast values between a client's topic versions and the installed core's.
// Chains are append-only shared history, so the deeper side carries every step
// between the two versions: a client behind the install walks the core's chain,
// one ahead walks its own.
export function topicAdapters(
  coreMigrations: ReadonlyMap<string, readonly TopicMigration[]>,
  clientMigrations: ReadonlyMap<string, readonly TopicMigration[]>,
): {
  toCore: (type: string, value: unknown) => unknown;
  toClient: (type: string, value: unknown) => unknown;
} {
  const coreVersions = topicVersions(coreMigrations);
  const clientVersions = topicVersions(clientMigrations);
  const clientVersion = (type: string) => clientVersions.get(type) ?? 1;
  const coreVersion = (type: string) => coreVersions.get(type) ?? 1;
  const chainFor = (type: string) =>
    (clientVersion(type) > coreVersion(type)
      ? clientMigrations
      : coreMigrations
    ).get(type);
  return {
    toCore: (type, value) =>
      applyChain(chainFor(type), value, clientVersion(type), coreVersion(type)),
    toClient: (type, value) =>
      applyChain(chainFor(type), value, coreVersion(type), clientVersion(type)),
  };
}

// A version the topic didn't change at has no step and is skipped, so a chain
// only declares the versions that actually moved.
export function applyChain(
  steps: readonly TopicMigration[] | undefined,
  value: unknown,
  from: number,
  to: number,
): unknown {
  if (!steps || steps.length === 0 || from === to) return value;

  let current = value;
  if (from < to) {
    for (let version = from; version < to; version++) {
      const step = steps.find((candidate) => candidate.from === version);
      if (step) current = step.up(current);
    }
  } else {
    for (let version = from; version > to; version--) {
      const step = steps.find((candidate) => candidate.from === version - 1);
      if (step) current = step.down(current);
    }
  }
  return current;
}

// `project` builds a fresh object per call, so the result is cached by input
// reference — that's what keeps a projected `getCurrent` stable (see
// `toStateSource`).
function projectCurrent(
  input: () => unknown,
  project: (value: unknown) => unknown,
): () => unknown {
  let lastInput: unknown;
  let lastOutput: unknown;
  let cached = false;
  return () => {
    const current = input();
    if (current === undefined) return undefined;
    if (!cached || current !== lastInput) {
      lastInput = current;
      lastOutput = project(current);
      cached = true;
    }
    return lastOutput;
  };
}

function mapStateSource(
  source: StateSource<unknown>,
  project: (value: unknown) => unknown,
): StateSource<unknown> {
  return toStateSource(
    source.pipe(map(project)),
    projectCurrent(source.getCurrent, project),
  );
}

function adaptMessage(
  message: OsMessage<unknown, unknown>,
  downPayload: (value: unknown) => unknown,
): OsMessage<unknown, unknown> {
  // The reply passes back untouched — replies aren't migrated.
  return {
    type: message.type,
    payload: downPayload(message.payload),
    meta: message.meta,
    reply: message.reply,
    get signal() {
      return message.signal;
    },
  };
}

// The copy's module graph loads fine; the first actual use fails loudly.
function createInertBus(installed: unknown): Bus {
  const fail = (): never => {
    throw new MessageBusError(
      "PROTOCOL_MISMATCH",
      `installed OS bus speaks protocol ${String(installed)}, this copy speaks ${OS_PROTOCOL}`,
    );
  };
  return { emit: fail, query: fail, subscribe: fail } as unknown as Bus;
}

function resolveAppAbort(registry: Registry, appId: string): AbortController {
  let controller = registry.appAborts.get(appId);
  if (!controller) {
    controller = new AbortController();
    registry.appAborts.set(appId, controller);
  }
  return controller;
}

/**
 * Tear down everything an app holds on the bus in one abort. The next `connect`
 * for the same app id starts a fresh lifetime, so a reload replaces a
 * generation of handlers instead of stacking a new one.
 * @public
 */
export function disconnectApp(bus: Bus, appId: string): void {
  const registry = (bus as InternalBus)[REGISTRY];
  const controller = registry.appAborts.get(appId);
  if (!controller) return;
  registry.appAborts.delete(appId);
  controller.abort();
}

/**
 * Options for {@link connect}. `appId` is stamped on every message the client
 * emits and scopes its lifetime (see {@link disconnectApp}); `migrations` pins
 * the client's per-topic versions and defaults to this copy's bundled chains.
 * @public
 */
export interface ConnectOptions {
  appId?: string;
  migrations?: ReadonlyMap<string, readonly TopicMigration[]>;
}

/**
 * Connect a client to the shared core: this copy's identity plus its topic
 * versions, recast per topic against the install in either direction (see
 * {@link topicAdapters}).
 *
 * The one connect-time failure is a core-protocol mismatch, which yields an
 * inert client rather than a throw here — `connect` runs at module evaluation,
 * where a throw would take the whole module graph down with it.
 * @public
 */
export function connect(core: Bus, config: ConnectOptions = {}): Bus {
  const appId = config.appId ?? DEFAULT_APP_ID;

  const installedProtocol = (core as Partial<InternalBus>)[PROTOCOL];
  if (installedProtocol !== OS_PROTOCOL) {
    log.error(
      `OS bus protocol mismatch for "${appId}": installed ${String(installedProtocol)}, this copy speaks ${OS_PROTOCOL}`,
    );
    return createInertBus(installedProtocol);
  }

  const clientMigrations = config.migrations ?? bundledMigrations();
  const registry = (core as InternalBus)[REGISTRY];

  try {
    registerTopics(registry, WORKBENCH_TOPIC_MANIFEST);
  } catch (error) {
    log.error(`OS bus topic manifest conflict for "${appId}"`, { error });
    return createInertBus(installedProtocol);
  }
  const { toCore, toClient } = topicAdapters(
    registry.migrations,
    clientMigrations,
  );
  const isState = (type: string) => isStateTopic(registry, type);

  // Everything the client subscribes to or waits on is scoped to its app's
  // lifetime, composed with any caller signal.
  const appAbort = resolveAppAbort(registry, appId);
  const scoped = (signal: AbortSignal | undefined) =>
    signal ? AbortSignal.any([signal, appAbort.signal]) : appAbort.signal;

  // One stream per topic: a UI binding may call `subscribe` on every render,
  // and a fresh stream each call spins `useSyncExternalStore`.
  const clientStreams = new Map<
    string,
    StateSource<unknown> | Observable<unknown>
  >();

  const api = {
    emit: (type: string, payload: unknown, options?: EmitOptions) =>
      emit(
        registry,
        type,
        toCore(type, payload),
        { ...options, signal: scoped(options?.signal) },
        appId,
      ),
    query: (type: string, options?: AbortOptions) =>
      query(registry, type, { signal: scoped(options?.signal) }).then((value) =>
        toClient(type, value),
      ),
    subscribe: (
      type: string,
      handler?: (arg: never) => void,
      options?: AbortOptions,
    ) => {
      if (!handler) {
        let stream = clientStreams.get(type);
        if (!stream) {
          const raw = subscribe(registry, type, undefined, options);
          stream = isState(type)
            ? mapStateSource(raw as StateSource<unknown>, (value) =>
                toClient(type, value),
              )
            : (raw as Observable<unknown>).pipe(
                map((payload) => toClient(type, payload)),
              );
          clientStreams.set(type, stream);
        }
        return stream;
      }
      const adapted = isState(type)
        ? (value: unknown) =>
            (handler as (value: unknown) => void)(toClient(type, value))
        : (message: OsMessage<unknown, unknown>) =>
            (handler as (message: OsMessage<unknown, unknown>) => void)(
              adaptMessage(message, (value) => toClient(type, value)),
            );
      return subscribe(registry, type, adapted as (arg: never) => void, {
        signal: scoped(options?.signal),
      });
    },
  };

  const instance = api as unknown as InternalBus;
  // Carrying the symbols lets internal seams reach the shared registry and a
  // client itself be `connect`ed.
  instance[REGISTRY] = registry;
  instance[PROTOCOL] = OS_PROTOCOL;
  return instance;
}

/**
 * Resolve the shared core: the first copy to load creates it, every later one
 * reuses it — after a brand check, so a foreign value squatting on the key is
 * replaced rather than trusted.
 */
export function installBus(): Bus {
  const globals = globalThis as { [BUS]?: Bus };
  const existing = globals[BUS];
  if (existing && PROTOCOL in existing) return existing;
  if (existing) log.warn("replaced a foreign value at the OS bus install key");
  return (globals[BUS] = createBus());
}

// Baked in by the app's build; absent in the host's own copy.
declare const __SANITY_APP_ID__: string | undefined;

/**
 * The client every remote and the host import: this copy's identity and topic
 * versions, connected to the shared core.
 * @public
 */
export const os: Bus = connect(installBus(), {
  appId:
    typeof __SANITY_APP_ID__ === "string" ? __SANITY_APP_ID__ : DEFAULT_APP_ID,
});

/**
 * Registers state topics an app adds to `Topics`; `undefined` makes reads wait
 * for the first publish, and a name already used as an event topic throws.
 * @public
 */
export function defineStateTopics(
  target: Bus,
  topics: Partial<{ [K in StateTopic]: ValueOf<K> | undefined }>,
): void {
  const registry = (target as InternalBus)[REGISTRY];
  for (const name of Object.keys(topics)) {
    if (registry.topics.get(name) === "event") {
      throw new MessageBusError(
        "PROTOCOL_MISMATCH",
        `topic "${name}" is declared "state" but the installed bus knows it as "event"`,
      );
    }
  }
  for (const [name, seed] of Object.entries(topics)) {
    registry.topics.set(name, "state");
    const existing = registry.stateSubjects.get(name);
    if (existing) {
      // Never reset a live subject back to "no value".
      if (seed !== undefined && !Object.is(existing.getValue(), seed)) {
        existing.next(seed);
      }
    } else {
      registry.stateSubjects.set(
        name,
        new BehaviorSubject<unknown>(seed === undefined ? NO_VALUE : seed),
      );
    }
  }
}