Function createLoadSubsetDedupe

packages/electric-db-collection/src/electric.ts:390–540

* Creates a deduplicated loadSubset handler for progressive/on-demand modes * Returns null for eager mode, or a DeduplicatedLoadSubset instance for other modes. * Handles fetching snapshots in progressive mode during buffering phase, * and requesting snapshots in on-demand mode. * * When cursor

({
  stream,
  syncMode,
  isBufferingInitialSync,
  begin,
  write,
  commit,
  collectionId,
  encodeColumnName,
  signal,
}: {
  stream: ShapeStream<T>
  syncMode: ElectricSyncMode
  isBufferingInitialSync: () => boolean
  begin: () => void
  write: (mutation: {
    type: `insert` | `update` | `delete`
    value: T
    metadata: Record<string, unknown>
  }) => void
  commit: () => void
  collectionId?: string
  /**
   * Optional function to encode column names (e.g., camelCase to snake_case).
   * This is typically the `encode` function from shapeOptions.columnMapper.
   */
  encodeColumnName?: ColumnEncoder
  /**
   * Abort signal to check if the stream has been aborted during cleanup.
   * When aborted, errors from requestSnapshot are silently ignored.
   */
  signal: AbortSignal
})

 * - One for whereCurrent (rows = cursor, for tie-breaking) without limit
 */
function createLoadSubsetDedupe<T extends Row<unknown>>({
  stream,
  syncMode,
  isBufferingInitialSync,
  begin,
  write,
  commit,
  collectionId,
  encodeColumnName,
  signal,
}: {
  stream: ShapeStream<T>
  syncMode: ElectricSyncMode
  isBufferingInitialSync: () => boolean
  begin: () => void
  write: (mutation: {
    type: `insert` | `update` | `delete`
    value: T
    metadata: Record<string, unknown>
  }) => void
  commit: () => void
  collectionId?: string
  /**
   * Optional function to encode column names (e.g., camelCase to snake_case).
   * This is typically the `encode` function from shapeOptions.columnMapper.
   */
  encodeColumnName?: ColumnEncoder
  /**
   * Abort signal to check if the stream has been aborted during cleanup.
   * When aborted, errors from requestSnapshot are silently ignored.
   */
  signal: AbortSignal
}): DeduplicatedLoadSubset | null {
  if (syncMode === `eager`) {
    return null
  }

  const compileOptions = encodeColumnName ? { encodeColumnName } : undefined
  const logPrefix = collectionId ? `[${collectionId}] ` : ``

  /**
   * Handles errors from snapshot operations. Returns true if the error was
   * handled (signal aborted during cleanup), false if it should be re-thrown.
   */
  function handleSnapshotError(error: unknown, operation: string): boolean {
    if (signal.aborted) {
      debug(`${logPrefix}Ignoring ${operation} error during cleanup: %o`, error)
      return true
    }
    debug(`${logPrefix}Error in ${operation}: %o`, error)
    return false
  }

  const loadSubset = async (opts: LoadSubsetOptions) => {
    if (isBufferingInitialSync()) {
      const snapshotParams = compileSQL<T>(opts, compileOptions)
      try {
        const { data: rows } = await stream.fetchSnapshot(snapshotParams)