row-store

// A row store that React components subscribe to one row at a time.
//
// The consumer owns the feed. It calls `applyDeltas` once per frame with everything that changed
// (a Tauri consumer straight from its event listener; a WebSocket consumer through `createFrameBatcher`).
// A delta to one row wakes that row's subscribers and nobody else, and React batches every
// notification fired inside one `applyDeltas` into one commit, so a batch of two thousand row
// changes is one render pass over the rows that changed.
//
// Views sit on top: a filtered, sorted list of ids with an optional reorder hold so rows do not
// move under a trader's cursor while they work.

// setTimeout's longest delay; a longer one fires at once.
const MAX_TIMER_MS = 2 ** 31 - 1
// How long a batch waits for a frame before a timer flushes it: a hidden tab runs no frames.
const FRAME_FALLBACK_MS = 250

export type RowId = string

/** How the lane loses data. `coalesced` drops stale ticks and counts them; `ordered` never drops. */
export type Lane = "coalesced" | "ordered"

export interface DeltaBatch<T> {
  /** Whole rows. Replaces the row object, so every subscriber of that row wakes. */
  upsert?: readonly T[]
  /** Partial rows merged into a new object. Unknown ids are ignored. */
  patch?: readonly { id: RowId; fields: Partial<T> }[]
  remove?: readonly RowId[]
  /** Authoritative order for an ordered lane. Ids not present keep their previous relative order at the end. */
  order?: readonly RowId[]
  meta?: {
    lane?: Lane
    /** Messages the producer dropped since the last batch (coalesced lane). Accumulated. */
    dropped?: number
    /** Last sequence number applied (ordered lane). */
    seq?: number
    /** A gap was detected and replay is in progress. */
    gap?: boolean
    /** Producer timestamp of the newest message in the batch, ms since epoch. */
    producedAt?: number
  }
}

export interface StoreMeta {
  /** Increments per batch or clear, before any subscriber is notified. */
  version: number
  size: number
  lane: Lane
  dropped: number
  seq: number | null
  gap: boolean
  /** Wall clock of the last applied batch, ms since epoch. */
  lastBatchAt: number | null
  producedAt: number | null
}

export interface ViewOptions<T> {
  comparator?: (a: T, b: T) => number
  filter?: (row: T) => boolean
  /**
   * After `touch()` (the grid calls it on every key and pointer interaction) the order is frozen for this
   * long: new rows append, removed rows vanish, nothing moves. 0 disables the hold.
   */
  reorderHoldMs?: number
  /** Injectable clock, ms. Defaults to Date.now. */
  now?: () => number
}

export interface RowView<T> {
  /** The store this view reads from. */
  readonly store: RowStore<T>
  /** Stable array reference until the order actually changes. */
  getIds(): readonly RowId[]
  subscribe(cb: () => void): () => void
  /** The user interacted; start or extend the reorder hold. */
  touch(): void
  /** True while a hold is in force. */
  isHeld(): boolean
  /** When the current or most recent reorder hold lapses or lapsed, in Date.now() milliseconds; null before any hold. Optional: without it, parked arrival marks age from their arrival, so a hold longer than the highlight window swallows those flashes. */
  holdExpiresAt?(): number | null
  dispose(): void
  /** True after `dispose`: the view no longer follows the store. */
  isDisposed(): boolean
}

/** A render-safe snapshot with an explicitly owned, reversible connection to its store. */
export interface PreparedRowView<T> extends RowView<T> {
  /** Follow batches until the returned release runs. Multiple owners release independently. */
  connect(): () => void
}

/** Publish current row, order and metadata snapshots before notifying any subscriber. */
export interface RowStore<T> {
  /** Stable object reference until the row is replaced or patched. */
  getRow(id: RowId): T | undefined
  /** Stable array reference until the order changes. Insertion order unless a batch carries `order`. */
  getIds(): readonly RowId[]
  /** Stable object reference until the next batch. */
  getMeta(): StoreMeta
  subscribeRow(id: RowId, cb: () => void): () => void
  subscribeOrder(cb: () => void): () => void
  subscribeMeta(cb: () => void): () => void
  /** Apply one batch synchronously. Call it once per frame. */
  applyDeltas(batch: DeltaBatch<T>): void
  /** Prepare a current snapshot without subscriptions, registration or timers. Safe during render. */
  prepareView(opts?: ViewOptions<T>): PreparedRowView<T>
  /** Follow the store immediately, even without subscribers. The caller must dispose the view. */
  createView(opts?: ViewOptions<T>): RowView<T>
  /** Drop every row (a re-snapshot is coming). Wakes every subscriber. */
  clear(): void
  readonly getRowId: (row: T) => RowId
}

export interface RowStoreOptions<T> {
  getRowId: (row: T) => RowId
  lane?: Lane
  now?: () => number
}

type Listener = () => void

export function createRowStore<T>(options: RowStoreOptions<T>): RowStore<T> {
  const { getRowId } = options
  const now = options.now ?? Date.now
  const rows = new Map<RowId, T>()
  let ids: RowId[] = []
  // Snapshots never alias the live array: the first batch appends to ids in place.
  let idsSnapshot: readonly RowId[] = ids.slice()
  let meta: StoreMeta = { version: 0, size: 0, lane: options.lane ?? "coalesced", dropped: 0, seq: null, gap: false, lastBatchAt: null, producedAt: null }
  const rowListeners = new Map<RowId, Set<Listener>>()
  const orderListeners = new Set<Listener>()
  const metaListeners = new Set<Listener>()
  const views = new Set<ViewImpl<T>>()

  function notifyRow(id: RowId) {
    const set = rowListeners.get(id)
    if (set) for (const cb of set) cb()
  }

  function applyDeltas(batch: DeltaBatch<T>) {
    const touched = new Set<RowId>()
    const removed = new Set<RowId>()
    let orderChanged = false

    if (batch.upsert) {
      for (const row of batch.upsert) {
        const id = getRowId(row)
        if (!rows.has(id)) {
          ids.push(id)
          orderChanged = true
        }
        rows.set(id, row)
        touched.add(id)
      }
    }
    if (batch.patch) {
      for (const { id, fields } of batch.patch) {
        const existing = rows.get(id)
        if (existing === undefined) continue
        rows.set(id, { ...existing, ...fields })
        touched.add(id)
      }
    }
    if (batch.remove?.length) {
      const drop = new Set<RowId>()
      for (const id of batch.remove) {
        if (rows.delete(id)) {
          drop.add(id)
          removed.add(id)
          touched.add(id)
        }
      }
      if (drop.size) {
        ids = ids.filter((id) => !drop.has(id))
        orderChanged = true
      }
    }
    if (batch.order) {
      const next: RowId[] = []
      const seen = new Set<RowId>()
      for (const id of batch.order) {
        if (rows.has(id) && !seen.has(id)) {
          next.push(id)
          seen.add(id)
        }
      }
      for (const id of ids) if (!seen.has(id)) next.push(id)
      if (!sameOrder(ids, next)) {
        ids = next
        orderChanged = true
      }
    }

    const m = batch.meta
    meta = {
      version: meta.version + 1,
      size: rows.size,
      lane: m?.lane ?? meta.lane,
      dropped: meta.dropped + (m?.dropped ?? 0),
      seq: m?.seq ?? meta.seq,
      gap: m?.gap ?? meta.gap,
      lastBatchAt: now(),
      producedAt: m?.producedAt ?? meta.producedAt,
    }

    if (orderChanged) idsSnapshot = ids.slice()
    for (const view of views) view.onBatch(touched, removed, orderChanged, meta.version)
    for (const id of touched) notifyRow(id)
    if (orderChanged) for (const cb of orderListeners) cb()
    for (const cb of metaListeners) cb()
  }

  const store: RowStore<T> = {
    getRowId,
    getRow: (id) => rows.get(id),
    getIds: () => idsSnapshot,
    getMeta: () => meta,
    subscribeRow(id, cb) {
      let set = rowListeners.get(id)
      if (!set) rowListeners.set(id, (set = new Set()))
      set.add(cb)
      return () => {
        set!.delete(cb)
        if (!set!.size) rowListeners.delete(id)
      }
    },
    subscribeOrder(cb) {
      orderListeners.add(cb)
      return () => orderListeners.delete(cb)
    },
    subscribeMeta(cb) {
      metaListeners.add(cb)
      return () => metaListeners.delete(cb)
    },
    applyDeltas,
    prepareView(opts = {}) {
      const view = new ViewImpl<T>(store, opts, () => views.add(view), () => views.delete(view))
      return view
    },
    createView(opts) {
      const view = store.prepareView(opts)
      view.connect()
      return view
    },
    clear() {
      const all = ids
      rows.clear()
      ids = []
      idsSnapshot = []
      meta = { ...meta, version: meta.version + 1, size: 0, lastBatchAt: now() }
      for (const view of views) view.onBatch(new Set(all), new Set(all), true, meta.version)
      for (const id of all) notifyRow(id)
      for (const cb of orderListeners) cb()
      for (const cb of metaListeners) cb()
    },
  }
  return store
}

function sameOrder(a: readonly RowId[], b: readonly RowId[]): boolean {
  if (a === b) return true
  if (a.length !== b.length) return false
  for (let i = 0; i < a.length; i++) if (a[i] !== b[i]) return false
  return true
}

class ViewImpl<T> implements PreparedRowView<T> {
  private ids: readonly RowId[] = []
  private published: readonly RowId[] = []
  private version: number
  private readonly listeners = new Set<Listener>()
  private holdUntil = 0
  private settlePending = false
  private timer: ReturnType<typeof setTimeout> | null = null
  private disposed = false
  private connections = 0
  private readonly now: () => number
  readonly store: RowStore<T>
  private readonly opts: ViewOptions<T>
  private readonly attach: () => void
  private readonly detach: () => void

  constructor(store: RowStore<T>, opts: ViewOptions<T>, attach: () => void, detach: () => void) {
    this.store = store
    this.opts = opts
    this.attach = attach
    this.detach = detach
    this.now = opts.now ?? Date.now
    this.version = store.getMeta().version
    this.ids = this.compute()
    this.published = this.ids
  }

  getIds() {
    if (this.disposed) return this.ids
    const version = this.store.getMeta().version
    const expired = !this.connections && this.settlePending && !this.isHeld()
    if (version !== this.version || expired) {
      this.cache(this.isHeld() ? this.computeHeld() : this.compute())
      this.version = version
      if (expired) this.settlePending = false
    }
    return this.ids
  }

  subscribe(cb: Listener) {
    if (this.disposed) return () => {}
    this.listeners.add(cb)
    return () => {
      this.listeners.delete(cb)
    }
  }

  connect() {
    if (this.disposed) return () => {}
    let released = false
    const release = () => {
      if (released || this.disposed) return
      released = true
      if (--this.connections === 0) {
        this.detach()
        this.clearTimer()
      }
    }
    if (this.connections === 0) {
      // Refresh while disconnected, including a hold that expired while effects were hidden.
      this.getIds()
      this.attach()
    }
    this.connections++
    try {
      this.publish()
      this.armTimer()
    } catch (error) {
      release()
      throw error
    }
    return release
  }

  isHeld() {
    return this.now() < this.holdUntil
  }

  holdExpiresAt() {
    // The hold is kept in the view's own clock; the reported deadline is wall
    // time, so a grid can compare it with Date.now whatever clock was injected.
    return this.holdUntil > 0 ? Date.now() + (this.holdUntil - this.now()) : null
  }

  touch() {
    if (this.disposed) return
    // A hold that isn't a positive number holds nothing, and one longer than a timer can wait, Infinity included, holds
    // as long as one can, about 24.8 days: past that a timer fires at once.
    const asked = this.opts.reorderHoldMs ?? 0
    const hold = asked > 0 ? Math.min(asked, MAX_TIMER_MS) : 0
    if (hold <= 0) return
    if (!this.connections) this.getIds()
    this.holdUntil = this.now() + hold
    this.settlePending = true
    this.clearTimer()
    this.armTimer()
  }

  private armTimer() {
    if (this.disposed || !this.connections || !this.settlePending || this.timer !== null) return
    // A quiet feed still settles: recompute when the hold lapses.
    this.timer = setTimeout(() => {
      this.timer = null
      this.settlePending = false
      this.cache(this.compute())
      this.publish()
    }, Math.max(0, this.holdUntil - this.now()))
  }

  private clearTimer() {
    if (this.timer !== null) clearTimeout(this.timer)
    this.timer = null
  }

  dispose() {
    if (this.disposed) return
    this.disposed = true
    this.clearTimer()
    this.connections = 0
    this.listeners.clear()
    this.detach()
  }

  isDisposed() {
    return this.disposed
  }

  /** Called by the store inside applyDeltas, before row listeners fire. */
  onBatch(touched: Set<RowId>, removed: Set<RowId>, orderChanged: boolean, version: number) {
    // Another view's listener may already have read our current snapshot. It cannot publish it.
    if (this.version === version) {
      this.publish()
      return
    }
    this.version = version
    if (!touched.size && !orderChanged) return
    if (this.isHeld()) {
      this.cache(this.computeHeld(removed))
      this.publish()
      return
    }
    // Only recompute when the batch could have changed this view.
    let relevant = orderChanged
    if (!relevant) for (const id of touched) if (this.passes(id) || this.ids.includes(id)) { relevant = true; break }
    if (relevant) {
      this.cache(this.compute())
      this.publish()
    }
  }

  private computeHeld(removed?: Set<RowId>): readonly RowId[] {
    // Frozen order: drop removals and rows outside the filter, then append sorted newcomers.
    const present = new Set(this.ids)
    const kept = this.ids.filter((id) => !removed?.has(id) && this.passes(id))
    const newcomers = this.store.getIds().filter((id) => !present.has(id) && this.passes(id))
    if (this.opts.comparator) newcomers.sort(this.compare)
    return kept.length === this.ids.length && !newcomers.length ? this.ids : [...kept, ...newcomers]
  }

  private passes(id: RowId): boolean {
    const row = this.store.getRow(id)
    if (row === undefined) return false
    return this.opts.filter ? this.opts.filter(row) : true
  }

  private readonly compare = (a: RowId, b: RowId) => this.opts.comparator!(this.store.getRow(a)!, this.store.getRow(b)!)

  private compute(): readonly RowId[] {
    const all = this.store.getIds()
    const list = this.opts.filter ? all.filter((id) => this.passes(id)) : all.slice()
    if (this.opts.comparator) list.sort(this.compare)
    return list
  }

  private cache(next: readonly RowId[]) {
    if (!sameOrder(this.ids, next)) this.ids = next
  }

  private publish() {
    if (sameOrder(this.published, this.ids)) return
    this.published = this.ids
    for (const cb of this.listeners) cb()
  }
}

export interface FrameBatcher<T> {
  /** Queue a delta; the merged batch applies on the next animation frame, or after 250 ms when no frame comes. */
  push(delta: DeltaBatch<T>): void
  /** Apply now, without waiting for the frame. */
  flush(): void
  /** Drop everything queued. */
  cancel(): void
  pending(): boolean
}

/**
 * Coalesce many small deltas into one batch per frame, last write wins per row. For consumers whose
 * feed delivers messages one at a time. A consumer whose core already batches per frame calls
 * `applyDeltas` directly and skips this: another frame of latency buys nothing.
 */
export function createFrameBatcher<T>(
  apply: (batch: DeltaBatch<T>) => void,
  options: { getRowId: (row: T) => RowId; raf?: (cb: () => void) => number; caf?: (handle: number) => void },
): FrameBatcher<T> {
  const raf = options.raf ?? ((cb) => requestAnimationFrame(cb))
  const caf = options.caf ?? ((h) => (typeof cancelAnimationFrame === "function" ? cancelAnimationFrame(h) : undefined))
  const upserts = new Map<RowId, T>()
  const patches = new Map<RowId, Partial<T>>()
  const removes = new Set<RowId>()
  let order: readonly RowId[] | undefined
  let meta: NonNullable<DeltaBatch<T>["meta"]> | undefined
  let handle: number | null = null
  // A hidden document runs no frames, so a timer stands in: whichever comes first flushes, and the other is cancelled.
  let timer: ReturnType<typeof setTimeout> | null = null
  // The frame scheduled now. A frame the flush cancelled can still run where it can't be cancelled, as with a `raf`
  // passed without its `caf`, and it finds another frame scheduled, or none, and flushes nothing.
  let scheduled: object | null = null

  function reset() {
    upserts.clear()
    patches.clear()
    removes.clear()
    order = undefined
    meta = undefined
  }

  function flush() {
    scheduled = null
    if (handle !== null) {
      caf(handle)
      handle = null
    }
    if (timer !== null) {
      clearTimeout(timer)
      timer = null
    }
    if (!upserts.size && !patches.size && !removes.size && !order && !meta) return
    const batch: DeltaBatch<T> = {}
    if (upserts.size) batch.upsert = [...upserts.values()]
    if (patches.size) batch.patch = [...patches].map(([id, fields]) => ({ id, fields }))
    if (removes.size) batch.remove = [...removes]
    if (order) batch.order = order
    if (meta) batch.meta = meta
    reset()
    apply(batch)
  }

  return {
    push(delta) {
      for (const row of delta.upsert ?? []) {
        const id = options.getRowId(row)
        removes.delete(id)
        patches.delete(id)
        upserts.set(id, row)
      }
      for (const { id, fields } of delta.patch ?? []) {
        if (removes.has(id)) continue
        const pendingUpsert = upserts.get(id)
        if (pendingUpsert !== undefined) upserts.set(id, { ...pendingUpsert, ...fields })
        else patches.set(id, { ...(patches.get(id) ?? {}), ...fields })
      }
      for (const id of delta.remove ?? []) {
        upserts.delete(id)
        patches.delete(id)
        removes.add(id)
      }
      if (delta.order) order = delta.order
      if (delta.meta) {
        // Omitted fields keep their previous values all the way through: a frame that
        // never mentions gap must not write one, or a replay's recorded gap is
        // silenced mid-flight. Mentions apply in order, so a gap that opens and
        // closes inside one frame ends closed, exactly as two frames would.
        const gapMentioned = delta.meta.gap !== undefined || meta?.gap !== undefined
        meta = {
          ...meta,
          ...delta.meta,
          dropped: (meta?.dropped ?? 0) + (delta.meta.dropped ?? 0),
          ...(gapMentioned ? { gap: delta.meta.gap ?? meta?.gap } : {}),
        }
        if (!gapMentioned) delete (meta as { gap?: boolean }).gap
      }
      if (handle === null && timer === null) {
        const frame = {}
        scheduled = frame
        const requested = raf(() => {
          if (scheduled !== frame) return
          handle = null
          flush()
        })
        // A frame that ran as it was requested has flushed already, and leaves nothing to hold.
        if (scheduled === frame) {
          handle = requested
          timer = setTimeout(() => {
            timer = null
            flush()
          }, FRAME_FALLBACK_MS)
        }
      }
    },
    flush,
    cancel() {
      scheduled = null
      if (handle !== null) caf(handle)
      if (timer !== null) clearTimeout(timer)
      handle = null
      timer = null
      reset()
    },
    pending: () => handle !== null || timer !== null,
  }
}

import { useCallback, useLayoutEffect, useState, useSyncExternalStore } from "react"
import type { RowId, RowStore, RowView, StoreMeta, ViewOptions } from "@/lib/row-store"

/** One row. The component re-renders when this row is replaced or patched, and for nothing else. */
export function useRow<T>(store: RowStore<T>, id: RowId): T | undefined {
  const subscribe = useCallback((cb: () => void) => store.subscribeRow(id, cb), [store, id])
  const get = useCallback(() => store.getRow(id), [store, id])
  return useSyncExternalStore(subscribe, get, get)
}

/** The ids of a store (insertion or authoritative order) or of a view (filtered, sorted, held). */
export function useRowIds<T>(source: RowStore<T> | RowView<T>): readonly RowId[] {
  const subscribe = useCallback((cb: () => void) => ("subscribeOrder" in source ? source.subscribeOrder(cb) : source.subscribe(cb)), [source])
  const get = useCallback(() => source.getIds(), [source])
  return useSyncExternalStore(subscribe, get, get)
}

/** Version, size, lane, drop count, sequence, gap, and timestamps. Changes once per batch. */
export function useStoreMeta<T>(store: RowStore<T>): StoreMeta {
  const subscribe = useCallback((cb: () => void) => store.subscribeMeta(cb), [store])
  const get = useCallback(() => store.getMeta(), [store])
  return useSyncExternalStore(subscribe, get, get)
}

/**
 * A view whose connection belongs to this component. Preparation is pure, so abandoned renders and
 * server rendering leave no live resources. Cleanup disconnects; effect replay reconnects the same view.
 * Keep options stable (a `useMemo`): new store/options identities select a new view synchronously.
 * Null options give null, for a component that borrows a caller-owned view instead.
 */
export function useView<T>(store: RowStore<T>, options: ViewOptions<T> | null): RowView<T> | null {
  const [prepared, setPrepared] = useState(() => ({ store, options, view: options ? store.prepareView(options) : null }))
  let current = prepared
  if (prepared.store !== store || prepared.options !== options) {
    current = { store, options, view: options ? store.prepareView(options) : null }
    setPrepared(current)
  }
  const view = current.view
  useLayoutEffect(() => view?.connect(), [view])
  return view
}