Function createEffect

packages/db/src/query/effect.ts:184–294

(config: EffectConfig<TRow, TKey>)

 * ```
 */
export function createEffect<
  TRow extends object = Record<string, unknown>,
  TKey extends string | number = string | number,
>(config: EffectConfig<TRow, TKey>): Effect {
  const id = config.id ?? `live-query-effect-${++effectCounter}`

  // AbortController for signalling disposal to handlers
  const abortController = new AbortController()

  const ctx: EffectContext = {
    effectId: id,
    signal: abortController.signal,
  }

  // Track in-flight async handler promises so dispose() can await them
  const inFlightHandlers = new Set<Promise<void>>()
  let disposed = false

  // Callback invoked by the pipeline runner with each batch of delta events
  const onBatchProcessed = (events: Array<DeltaEvent<TRow, TKey>>) => {
    if (disposed) return
    if (events.length === 0) return

    // Batch handler
    if (config.onBatch) {
      try {
        const result = config.onBatch(events, ctx)
        if (result instanceof Promise) {
          const tracked = result.catch((error) => {
            reportError(error, events[0]!, config.onError)
          })
          trackPromise(tracked, inFlightHandlers)
        }
      } catch (error) {
        // For batch handler errors, report with first event as context
        reportError(error, events[0]!, config.onError)
      }
    }

    for (const event of events) {
      if (abortController.signal.aborted) break

      const handler = getHandlerForEvent(event, config)
      if (!handler) continue

      try {
        const result = handler(event, ctx)
        if (result instanceof Promise) {
          const tracked = result.catch((error) => {
            reportError(error, event, config.onError)
          })
          trackPromise(tracked, inFlightHandlers)
        }
      } catch (error) {
        reportError(error, event, config.onError)
      }
    }
  }