import { LiveQueryObserverDisposedError } from './errors.js'
import { codedMessage, devBuild } from './error-message.js'
import { runAllCallbacks } from './utils/callbacks.js'
import {
  getLiveQueryStatusFlags,
  getPublicCollection,
  isFailedStatus,
  isSingleResultCollection,
} from './live-query-primitives.js'
import { describeProblem } from './collection/lifecycle.js'
import type { Collection } from './collection/index.js'
import type { LiveQueryObserverClient } from './live-query-observer-client.js'
import type { DbClient, DehydratedLiveQueryResult } from './client.js'
import type { ChangeMessage, CollectionStatus, HealthNode } from './types.js'

/**
 * The canonical, adapter-agnostic view of a live query at a point in time.
 *
 * `getSnapshot()` returns a stable object identity that only changes when the
 * query changes, so `useSyncExternalStore`-style consumers can compare by
 * reference. Each snapshot retains a captured view of `state`/`data`, so
 * reading an older snapshot cannot expose rows from a later revision.
 * Within one row revision, metadata-only changes retain the row containers.
 * Cleanup/reset may discard that revision even when its rows were empty.
 */
export interface LiveQuerySnapshot<
  T extends object,
  TKey extends string | number,
> {
  /** Keyed results, or `undefined` for a disabled query. */
  state: ReadonlyMap<TKey, T> | undefined
  /** Ordered results (single row for `findOne`), or `undefined` when disabled. */
  data: T | ReadonlyArray<T> | undefined
  /** The underlying collection, or `undefined` when disabled. */
  collection: Collection<T, TKey, any> | undefined
  /**
   * Monotonic counter bumped whenever the visible layout (the ordered key
   * sequence) changes — membership, ordering, or an order-only move. Lets
   * consumers detect a reorder that changed no row value (which `data`/`state`
   * identity alone can't express once row values are structurally shared).
   *
   * It is NOT in lockstep with snapshot identity: a value-only update produces a
   * new snapshot while `layoutRevision` stays put. A `layoutRevision` change
   * always accompanies a new snapshot, but not vice versa.
   */
  layoutRevision: number
  status: CollectionStatus | `disabled`
  isLoading: boolean
  isReady: boolean
  isIdle: boolean
  /** The first load failed: there is nothing to show. */
  isLoadingError: boolean
  /** Local storage or an invariant broke. Only a cascade cleanup and a fresh start recover. */
  isCrashed: boolean
  isCleanedUp: boolean
  isEnabled: boolean
  /**
   * The data is shown, and something keeps it from staying current: a sync
   * problem or a failed request somewhere in `health`.
   */
  isSyncError: boolean
  /**
   * The health tree for the query and every Collection it depends on, or
   * `undefined` for a disabled query. After the first ready, a source
   * problem shows here and the data stays.
   */
  health: HealthNode | undefined
}

/**
 * Listener payload: changes, `[]` for an internal layout-only publication, or
 * `undefined` for a synthetic status/ready notification.
 */
export type LiveQueryObserverListener<
  T extends object,
  TKey extends string | number,
> = (changes: Array<ChangeMessage<T, TKey>> | undefined) => void

/**
 * Wraps a resolved live-query `Collection` (or `null` for a disabled query) with
 * the shared lifecycle every framework adapter needs: start sync on first
 * subscribe, subscribe to changes and status transitions, expose a stable
 * snapshot for wholesale consumers, and deliver the raw change set for
 * granular consumers.
 *
 * Input resolution (query fn / config / collection / disabled) stays in the
 * adapter — it is framework-reactive. The observer owns everything after the
 * input is resolved to a concrete collection.
 *
 * @internal Unstable contract for TanStack DB's official framework adapters —
 * not a public extension point yet; may change in any release.
 */
export interface LiveQueryObserver<
  T extends object,
  TKey extends string | number,
> {
  /** Stable per-revision snapshot for wholesale materialization. */
  getSnapshot: () => LiveQuerySnapshot<T, TKey>
  /** Server snapshot. preloadRequested marks an explicit streaming Suspense load. */
  getServerSnapshot: (preloadRequested?: boolean) => LiveQuerySnapshot<T, TKey>
  /**
   * Subscribe to changes. The listener receives the change set (or `undefined`
   * for the synthetic notify a ready collection emits on attach). Granular
   * adapters apply the changes; wholesale adapters can ignore them and re-read
   * `getSnapshot()`. Raw changes can carry fresh equal row objects while a
   * snapshot retains its capture for the unchanged revision. Returns an
   * unsubscribe function.
   */
  subscribe: (listener: LiveQueryObserverListener<T, TKey>) => () => void
  /** Resolve once the collection has loaded its first data. */
  preload: () => Promise<void>
  /** The transport or preload error for this query, if it has not produced data. */
  getError: () => unknown
  /** Capture the ordered query result without serializing its source collections. */
  dehydrate: () => DehydratedLiveQueryResult<T, TKey>
  /** Idempotent teardown. */
  dispose: () => void
}

/**
 * One logical subscription. Records — not raw callbacks — identify
 * subscriptions, so the same listener function can be subscribed twice and
 * each subscription tears down independently.
 */
interface SubscriptionRecord<T extends object, TKey extends string | number> {
  listener: LiveQueryObserverListener<T, TKey>
  active: boolean
}

interface Publication<T extends object, TKey extends string | number> {
  changes: Array<ChangeMessage<T, TKey>> | undefined
  targets: Array<SubscriptionRecord<T, TKey>>
  entries?: Array<[TKey, T]>
  status: CollectionStatus
  collectionRevision?: number
  collectionLayoutRevision?: number
  layoutChanged: boolean
}

const DISABLED_SNAPSHOT: LiveQuerySnapshot<any, any> = {
  state: undefined,
  data: undefined,
  collection: undefined,
  layoutRevision: 0,
  status: `disabled`,
  isLoading: false,
  isReady: true,
  isIdle: false,
  isLoadingError: false,
  isCrashed: false,
  isCleanedUp: false,
  isEnabled: false,
  isSyncError: false,
  health: undefined,
}

class LiveQueryObserverImpl<
  T extends object,
  TKey extends string | number,
> implements LiveQueryObserver<T, TKey> {
  private readonly collection: Collection<T, TKey, any> | null
  private readonly wholesale: boolean
  private readonly clientBinding: LiveQueryObserverClient<T, TKey> | undefined
  private readonly onPreload: (() => void) | undefined
  private visibleStatus: CollectionStatus | undefined
  private cachedEntries: Array<[TKey, T]> | undefined
  private cachedCollectionRevision: number | undefined
  private cachedCollectionLayoutRevision: number | undefined
  private snapshotDirty = true
  // While attached, the tree from the last status or health change. Every
  // tree change is announced, so a snapshot reuses it instead of rebuilding.
  private publishedHealth: { tree: HealthNode | undefined } | undefined
  // Whether the cached snapshot was built for the hydration read.
  private cachedServerView = false
  private cachedSnapshot: LiveQuerySnapshot<T, TKey> = DISABLED_SNAPSHOT
  // Last materialized visible capture, which can be a hydration seed or lag
  // the live capture until getSnapshot() runs. These are array references.
  private cachedRows:
    | {
        entries: Array<[TKey, T]>
        data: Array<T>
        state?: Map<TKey, T>
      }
    | undefined
  private layoutRevision = 0
  private deliveredLayoutRevision: number | undefined
  private readonly subscriptions = new Set<SubscriptionRecord<T, TKey>>()
  // Publications are dispatched FIFO: an emit that happens while another
  // publication is being delivered (a listener mutating the collection
  // synchronously) is queued, never delivered reentrantly.
  private readonly publicationQueue: Array<Publication<T, TKey>> = []
  private dispatching = false
  private blockDelivery = false
  private attached = false
  private collectionUnsub: (() => void) | null = null
  private preloadPromise: Promise<void> | undefined
  private disposed = false

  // Sync activation belongs to the first subscription (attach), so building
  // an observer cannot activate collection resources on its own. Server
  // request clients still record ownership here because React may render an
  // observer without ever subscribing to it.
  constructor(
    collection: Collection<T, TKey, any> | null,
    wholesale: boolean,
    client: DbClient | undefined,
    queryHash: string | undefined,
    onPreload: (() => void) | undefined,
    suspendUntilReady: boolean,
  ) {
    if (
      collection &&
      (!Number.isFinite(collection._stateRevision) ||
        !Number.isFinite(collection._layoutRevision))
    ) {
      throw new Error(
        devBuild() && process.env.NODE_ENV !== `production`
          ? `LiveQueryObserver requires finite Collection row and layout revision clocks`
          : codedMessage(262),
      )
    }
    this.collection = collection
    this.wholesale = wholesale
    this.onPreload = onPreload
    this.clientBinding = client?._bindLiveQueryObserver(
      collection,
      queryHash,
      suspendUntilReady,
      {
        observer: this,
        read: (server) => this.getSnapshot(server),
        entries: () => (collection ? this.getVisibleEntries(collection) : []),
        isAttached: () => this.attached,
        isDisposed: () => this.disposed,
        dirty: () => {
          this.snapshotDirty = true
        },
        publish: (changes, entries) => {
          const revision =
            entries && collection ? collection._stateRevision : undefined
          const layoutRevision =
            entries && collection ? collection._layoutRevision : undefined
          this.emit(
            this.wholesale ? undefined : changes,
            undefined,
            entries,
            collection?.status,
            revision,
            layoutRevision,
            entries !== undefined,
          )
        },
        preload: () => this.preloadCollection(),
      },
    )
    this.clientBinding?.registerResource()
  }

  getSnapshot(forServerSnapshot = false): LiveQuerySnapshot<T, TKey> {
    const collection = this.collection
    if (!collection) return DISABLED_SNAPSHOT

    this.clientBinding?.sync(forServerSnapshot)
    if (!this.attached) this.refreshDetachedState(collection)

    // The hydration read and an ordinary read differ only while a seed
    // stands in front of a failure.
    if (
      this.cachedServerView !== forServerSnapshot &&
      this.clientBinding?.seed &&
      isFailedStatus(this.visibleStatus ?? collection.status)
    ) {
      this.snapshotDirty = true
    }

    if (this.snapshotDirty) {
      const entries = this.getVisibleEntries(collection)
      // Capture each row revision once. Sharing this value across metadata
      // snapshots keeps rows lazy and avoids retaining a chain of old snapshots.
      let rows = this.cachedRows
      if (!rows || rows.entries !== entries) {
        // Captured entries already contain the previous ordered keys. Compare
        // keys directly: serialized signatures can collide for string keys.
        const previous = rows?.entries
        if (
          previous?.length !== entries.length ||
          entries.some(([key], index) => previous[index]![0] !== key)
        ) {
          this.layoutRevision++
        }
        rows = this.cachedRows = {
          entries,
          data: entries.map(([, value]) => value),
        }
      }
      const singleResult = isSingleResultCollection(collection)
      const liveStatus = this.visibleStatus ?? collection.status
      // The seed stands in only for a result that is still coming, so an
      // ordinary read never hides a failure behind it. The hydration read
      // keeps the server's view: it must reproduce what the server rendered.
      const status = this.clientBinding?.failed
        ? (`error` as const)
        : this.clientBinding?.seed &&
            (forServerSnapshot || !isFailedStatus(liveStatus))
          ? (this.clientBinding.seed.snapshot.status ?? `ready`)
          : liveStatus
      this.cachedServerView = forServerSnapshot

      const health = this.publishedHealth
        ? this.publishedHealth.tree
        : (collection as { health?: HealthNode }).health
      this.cachedSnapshot = {
        get state() {
          return (rows.state ??= new Map(rows.entries))
        },
        data: singleResult ? rows.data[0] : rows.data,
        collection: getPublicCollection(collection),
        layoutRevision: this.layoutRevision,
        status,
        ...getLiveQueryStatusFlags(status, health),
        isEnabled: true,
      }
      this.snapshotDirty = false
    }
    return this.cachedSnapshot
  }

  getServerSnapshot(preloadRequested = false): LiveQuerySnapshot<T, TKey> {
    return (
      this.clientBinding?.getServerSnapshot(preloadRequested) ??
      this.getSnapshot()
    )
  }

  getError(): unknown {
    this.clientBinding?.sync()
    return this.clientBinding?.error
  }

  dehydrate(): DehydratedLiveQueryResult<T, TKey> {
    const collection = this.collection
    if (!collection) return { rows: [] }

    const entries =
      this.clientBinding?.seed &&
      this.clientBinding.seed.snapshot.status === undefined
        ? this.clientBinding.seed.entries
        : this.readEntries(collection).entries

    return {
      rows: entries.map(([key, value]) => ({
        key,
        value,
      })),
    }
  }

  private getVisibleEntries(
    collection: Collection<T, TKey, any>,
  ): Array<[TKey, T]> {
    if (this.clientBinding?.seed) return this.clientBinding.seed.entries
    return this.cachedEntries ?? this.captureEntries(collection)
  }

  private readEntries(collection: Collection<T, TKey, any>): {
    entries: Array<[TKey, T]>
    revision: number
    layoutRevision: number
  } {
    const revision = collection._stateRevision
    const layoutRevision = collection._layoutRevision
    // Reuse the last delivered capture only when both clocks match. A queued
    // newer publication must still capture its own rows before reentrant writes.
    const entries =
      this.cachedEntries &&
      revision === this.cachedCollectionRevision &&
      layoutRevision === this.cachedCollectionLayoutRevision
        ? this.cachedEntries
        : (Array.from(collection.entries()) as Array<[TKey, T]>)
    return { entries, revision, layoutRevision }
  }

  private captureEntries(
    collection: Collection<T, TKey, any>,
  ): Array<[TKey, T]> {
    const { entries, revision, layoutRevision } = this.readEntries(collection)
    this.updateCachedEntries(entries, revision, layoutRevision)
    return this.cachedEntries!
  }

  private updateCachedEntries(
    entries: Array<[TKey, T]>,
    revision: number | undefined,
    layoutRevision: number | undefined,
  ): void {
    // Collection clocks are authoritative; equal writes may re-enrich a row
    // without changing its value. Keep that revision's captured identities.
    if (
      this.cachedEntries === undefined ||
      revision !== this.cachedCollectionRevision ||
      layoutRevision !== this.cachedCollectionLayoutRevision
    ) {
      this.cachedEntries = entries
      this.snapshotDirty = true
    }
    this.cachedCollectionRevision = revision
    this.cachedCollectionLayoutRevision = layoutRevision
  }

  /**
   * A detached read catches up from the Collection's row and layout clocks.
   * Supported inputs are Collections and pooled views from this DB instance.
   */
  private refreshDetachedState(collection: Collection<T, TKey, any>): void {
    const status = collection.status
    this.captureEntries(collection)

    // A health-only change moves no row or status. Trees keep their identity
    // until their content changes, so identity tells whether it went stale.
    if (
      this.visibleStatus !== status ||
      this.cachedSnapshot.health !==
        (collection as { health?: HealthNode }).health
    ) {
      this.visibleStatus = status
      this.snapshotDirty = true
    }
  }

  subscribe(listener: LiveQueryObserverListener<T, TKey>): () => void {
    if (this.disposed) throw new LiveQueryObserverDisposedError()
    if (this.subscriptions.size === 0) this.clientBinding?.assertSources()

    const record: SubscriptionRecord<T, TKey> = { listener, active: true }
    const release = () => {
      if (!record.active) return
      record.active = false
      this.subscriptions.delete(record)
      if (this.subscriptions.size === 0) this.detach()
    }
    this.subscriptions.add(record)
    const attaching = !this.attached
    try {
      if (attaching) {
        this.attach()
      } else {
        // Seed only this late granular listener; a failed seed releases its
        // record without detaching peers that already own subscriptions.
        if (!this.wholesale) this.seed(record)
      }
    } catch (error) {
      // Setup returned no ownership handle. Release every acquired channel
      // and the logical record, so a later explicit subscription can attach.
      if (attaching) this.detach()
      release()
      throw error
    }
    return release
  }

  /** Deliver the collection's current rows to one late subscription as inserts. */
  private seed(record: SubscriptionRecord<T, TKey>): void {
    const collection = this.collection
    if (!collection) return

    const seedChanges: Array<ChangeMessage<T, TKey>> = []
    for (const [key, value] of this.getVisibleEntries(collection)) {
      seedChanges.push({ type: `insert`, key, value })
    }
    if (seedChanges.length === 0) return

    this.emit(seedChanges, [record])
  }

  private attach(): void {
    const collection = this.collection
    if (!collection || this.disposed) return
    // Install ownership before any registration can call back or throw.
    const releases: Array<() => void> = []
    const release = () => runAllCallbacks(releases.splice(0))
    this.collectionUnsub = release
    const ownRelease = (unsubscribe: () => void) => {
      if (this.collectionUnsub === release) releases.push(unsubscribe)
      else unsubscribe()
    }
    this.clientBinding?.registerResource()
    this.clientBinding?.sync()
    this.refreshDetachedState(collection)
    this.attached = true
    this.visibleStatus ??= collection.status
    this.deliveredLayoutRevision = collection._layoutRevision
    const attachedWithHydrationSeed = !!this.clientBinding?.seed
    this.blockDelivery = this.wholesale || attachedWithHydrationSeed

    // Sync activation happens inside subscribeChanges (addSubscriber starts
    // an idle/cleaned-up collection) — the same startSync path the old
    // constructor-time startSyncImmediate() took, but now owned by the first
    // committed subscription and observed by the status listener below.

    // Granular consumers subscribe with initial state so they receive the
    // current rows as inserts followed by deltas through one consistent
    // channel (the collection's per-subscriber change stream requires this to
    // align deltas). Wholesale consumers subscribe WITHOUT initial state —
    // preserving their pre-observer loading policy: no snapshot request means
    // no unfiltered loadSubset({ where: undefined }) against on-demand
    // collections. The explicit `false` marks all state as seen so deletes
    // still flow through as notifies.
    const notify = (
      changes: Array<ChangeMessage<T, TKey>> | undefined,
      status: CollectionStatus = collection.status,
      explicitLayoutChange = false,
    ) => {
      if (this.disposed || this.subscriptions.size === 0) return

      if (this.clientBinding?.seed) {
        // Startup remains behind the hydration cut. A later failure ends the
        // sync run whose result the seed stood in for, and is already an
        // owned publication, so it must be visible synchronously.
        if (isFailedStatus(status) && !this.blockDelivery)
          this.clientBinding.handoff()
        else if (status === `ready` || isFailedStatus(status))
          this.clientBinding.scheduleHandoff()
        return
      }

      if (status === `ready` && !collection.isLoadingSubset) {
        this.clientBinding?.markAuthoritative()
      }

      const layoutRevision = collection._layoutRevision
      let layoutChanged = explicitLayoutChange
      if (
        !explicitLayoutChange &&
        changes !== undefined &&
        changes.length === 0
      ) {
        // Empty ready events predate the explicit layout signal and share its
        // empty-array payload. Only forward an empty batch when the collection
        // confirms that a new layout-only publication occurred.
        if (layoutRevision === this.deliveredLayoutRevision) {
          return
        }
        layoutChanged = true
      }
      if (changes !== undefined) {
        this.deliveredLayoutRevision = layoutRevision
      }
      const captured =
        changes !== undefined || status === `cleaned-up`
          ? this.readEntries(collection)
          : undefined
      this.emit(
        changes,
        undefined,
        captured?.entries,
        status,
        captured?.revision,
        layoutRevision,
        layoutChanged,
      )
    }

    // Status transitions that carry no change events (loading→ready with no
    // rows, error, cleaned-up) are part of the canonical publication path:
    // any status change publishes a synthetic notify so consumers re-read the
    // snapshot. Unlike onFirstReady, `on` returns a real unsubscribe, so a
    // detached attachment leaves nothing behind.
    // A health change below the query changes its health tree, not its rows.
    // Each status change is followed by a health change; tree identity
    // tells whether the health change has anything left to publish.
    const healthOf = () => (collection as { health?: HealthNode }).health
    this.publishedHealth = { tree: healthOf() }
    ownRelease(() => {
      this.publishedHealth = undefined
    })
    ownRelease(
      collection.on(`status:change`, ({ status }) => {
        this.publishedHealth = { tree: healthOf() }
        notify(undefined, status)
      }),
    )
    ownRelease(
      collection.on(`health:change`, () => {
        const health = healthOf()
        if (health === this.publishedHealth?.tree) return
        this.publishedHealth = { tree: health }
        // No rows or status changed, so only this marks the snapshot stale.
        this.snapshotDirty = true
        notify(undefined, collection.status)
      }),
    )
    // Pooled views carry layout changes in their ordinary change stream.
    if (
      (collection as Partial<Collection<T, TKey, any>>)._subscribeLayoutChanges
    ) {
      ownRelease(
        collection._subscribeLayoutChanges(() =>
          notify([], collection.status, true),
        ),
      )
    }
    // `subscribeChanges` delivers the initial state synchronously, so a
    // listener can dispose the observer while the collection subscription is
    // still being created. Register the release hook up front; if detach()
    // ran during that replay (collectionUnsub no longer points at our hook),
    // undo the subscription as soon as the call returns.
    if (this.clientBinding) {
      ownRelease(this.clientBinding.subscribe())
      // A retained ready Collection can settle subset work without another
      // row or status change. That settlement must release its hydration seed.
      ownRelease(
        collection.on(`loadingSubset:change`, () => {
          if (this.clientBinding?.seed) this.clientBinding.scheduleHandoff()
        }),
      )
    }
    const subscription = collection.subscribeChanges(
      (changes) => notify(changes as Array<ChangeMessage<T, TKey>>),
      { includeInitialState: !this.wholesale && !attachedWithHydrationSeed },
    )
    ownRelease(() => subscription.unsubscribe())
    this.blockDelivery = false
    if (this.collectionUnsub !== release) return
    if (this.wholesale || attachedWithHydrationSeed) {
      // Publications raised while subscribeChanges starts sync are part of the
      // subscribe handshake. Apply their final snapshot state now, but suppress
      // listener delivery: useSyncExternalStore performs its consistency read
      // immediately after subscribe returns.
      this.flushPublications(!this.wholesale)
      // The render-time read is still current unless the handshake published.
      this.captureEntries(collection)
    }
    if (this.clientBinding?.seed) {
      if (!this.wholesale) this.seed(Array.from(this.subscriptions)[0]!)
      if (collection.status === `ready` || isFailedStatus(collection.status)) {
        this.clientBinding.scheduleHandoff()
      }
    }
  }

  private detach(): void {
    const release = this.collectionUnsub
    const unregister = this.clientBinding?.takeResourceRelease()
    this.collectionUnsub = null
    this.attached = false
    this.blockDelivery = false
    this.publicationQueue.length = 0
    runAllCallbacks([() => release?.(), () => unregister?.()])
  }

  private emit(
    changes: Array<ChangeMessage<T, TKey>> | undefined,
    targets = Array.from(this.subscriptions),
    entries?: Array<[TKey, T]>,
    status = this.collection?.status ?? `cleaned-up`,
    collectionRevision?: number,
    collectionLayoutRevision?: number,
    layoutChanged = false,
  ): void {
    this.publicationQueue.push({
      changes,
      targets,
      entries,
      status,
      collectionRevision,
      collectionLayoutRevision,
      layoutChanged,
    })
    if (this.dispatching || this.blockDelivery) return

    this.flushPublications()
  }

  private flushPublications(deliver = true): void {
    if (this.dispatching) return

    let failure: { error: unknown } | undefined
    this.dispatching = true
    try {
      // A dispose() during dispatch empties the queue, ending this loop.
      while (this.publicationQueue.length > 0) {
        const publication = this.publicationQueue.shift()!
        if (publication.entries) {
          this.updateCachedEntries(
            publication.entries,
            publication.collectionRevision,
            publication.collectionLayoutRevision,
          )
        }
        if (publication.layoutChanged) {
          this.snapshotDirty = true
        }
        if (this.visibleStatus !== publication.status) {
          this.visibleStatus = publication.status
          this.snapshotDirty = true
        }
        // Targets are captured when the publication is queued: a subscription
        // removed mid-delivery still receives the in-flight publication, and
        // one added later does not. Late-subscriber seeds use the same queue.
        if (deliver) {
          for (const subRecord of publication.targets) {
            if (this.disposed) break
            try {
              subRecord.listener(publication.changes)
            } catch (error) {
              failure ??= { error }
            }
          }
        }
      }
    } finally {
      this.dispatching = false
    }
    if (failure) throw failure.error
  }

  preload(): Promise<void> {
    return this.clientBinding?.preload() ?? this.preloadCollection()
  }

  private preloadCollection(): Promise<void> {
    if (this.preloadPromise) return this.preloadPromise
    this.onPreload?.()
    const promise = this.collection?.preload() ?? Promise.resolve()
    this.preloadPromise = promise
    const clear = () => {
      if (this.preloadPromise === promise) this.preloadPromise = undefined
    }
    void promise.then(clear, clear)
    return promise
  }

  dispose(): void {
    if (this.disposed) return
    this.disposed = true
    this.detach()
    for (const subRecord of this.subscriptions) subRecord.active = false
    this.subscriptions.clear()
    this.publicationQueue.length = 0
  }
}

export interface CreateLiveQueryObserverOptions {
  /**
   * How subscribers consume the observer:
   *
   * - `granular` (default): subscribers apply the delivered `ChangeMessage[]`
   *   deltas to their own keyed state (Vue/Svelte/Solid). The observer
   *   subscribes with initial state and seeds late subscribers, so every
   *   subscriber converges from deltas alone.
   * - `wholesale`: subscribers treat notifications as a wake-up and re-read
   *   `getSnapshot()` (React/Angular). The observer subscribes WITHOUT initial
   *   state, preserving those adapters' loading policy — no snapshot request,
   *   so no unfiltered `loadSubset` against on-demand collections. Nothing is
   *   delivered synchronously during `subscribe`, which keeps
   *   `useSyncExternalStore`-style consumers safe by construction.
   */
  mode?: `granular` | `wholesale`
  /** DbClient cache that owns SSR snapshots for this query identity. */
  client?: DbClient
  /** Stable live-query identity used for dehydration and hydration. */
  queryHash?: string
  /** Resume framework-deferred query sources before a server preload. */
  onPreload?: () => void
  /** Read the completed stream rather than an earlier ordinary loading view. */
  suspendUntilReady?: boolean
}

/**
 * The Error an adapter sends to an error boundary for a failed first load or
 * a crash: the cause itself when it is an Error, otherwise an Error with the
 * problem's message (as in `health.error.message`) whose `cause` is the value
 * the sync adapter reported.
 *
 * @internal Shared by TanStack DB's official framework adapters.
 */
export const toBoundaryError = (cause: unknown): Error =>
  cause instanceof Error
    ? cause
    : new Error(describeProblem(cause).message, { cause })

/**
 * Create a {@link LiveQueryObserver} for a resolved live-query collection, or a
 * disabled observer when `collection` is `null`/`undefined`.
 *
 * @internal This is an unstable contract shared by TanStack DB's official
 * framework adapters. It is exported so the adapter packages can use it, but
 * it is not a public extension point yet: its API may change in any release
 * without a semver major.
 */
export function createLiveQueryObserver<
  T extends object,
  TKey extends string | number,
>(
  collection: Collection<T, TKey, any> | null | undefined,
  options: CreateLiveQueryObserverOptions = {},
): LiveQueryObserver<T, TKey> {
  return new LiveQueryObserverImpl<T, TKey>(
    collection ?? null,
    options.mode === `wholesale`,
    options.client,
    options.queryHash,
    options.onPreload,
    options.suspendUntilReady === true,
  )
}