| 606 | } |
| 607 | |
| 608 | export function electricCollectionOptions<T extends Row<unknown>>( |
| 609 | config: ElectricCollectionConfig<T, any>, |
| 610 | ): Omit< |
| 611 | CollectionConfig<T, string | number, any, ElectricCollectionUtils<T>>, |
| 612 | `utils` |
| 613 | > & { |
| 614 | id?: string |
| 615 | utils: ElectricCollectionUtils<T> |
| 616 | schema?: any |
| 617 | } { |
| 618 | const seenTxids = new Store<Set<Txid>>(new Set([])) |
| 619 | const seenSnapshots = new Store<Array<PostgresSnapshot>>([]) |
| 620 | const internalSyncMode = config.syncMode ?? `eager` |
| 621 | const finalSyncMode = |
| 622 | internalSyncMode === `progressive` ? `on-demand` : internalSyncMode |
| 623 | const pendingMatches = new Store< |
| 624 | Map< |
| 625 | string, |
| 626 | { |
| 627 | matchFn: (message: Message<any>) => boolean |
| 628 | resolve: (value: boolean) => void |
| 629 | reject: (error: Error) => void |
| 630 | timeoutId: ReturnType<typeof setTimeout> |
| 631 | matched: boolean |
| 632 | } |
| 633 | > |
| 634 | >(new Map()) |
| 635 | |
| 636 | // Buffer messages since last up-to-date to handle race conditions |
| 637 | const currentBatchMessages = new Store<Array<Message<any>>>([]) |
| 638 | |
| 639 | // Track whether the current batch has been committed (up-to-date received) |
| 640 | // This allows awaitMatch to resolve immediately for messages from committed batches |
| 641 | const batchCommitted = new Store<boolean>(false) |
| 642 | |
| 643 | /** |
| 644 | * Helper function to remove multiple matches from the pendingMatches store |
| 645 | */ |
| 646 | const removePendingMatches = (matchIds: Array<string>) => { |
| 647 | if (matchIds.length > 0) { |
| 648 | pendingMatches.setState((current) => { |
| 649 | const newMatches = new Map(current) |
| 650 | matchIds.forEach((id) => newMatches.delete(id)) |
| 651 | return newMatches |
| 652 | }) |
| 653 | } |
| 654 | } |
| 655 | |
| 656 | /** |
| 657 | * Helper function to resolve and cleanup matched pending matches |
| 658 | */ |
| 659 | const resolveMatchedPendingMatches = () => { |
| 660 | const matchesToResolve: Array<string> = [] |
| 661 | pendingMatches.state.forEach((match, matchId) => { |
| 662 | if (match.matched) { |
| 663 | clearTimeout(match.timeoutId) |
| 664 | match.resolve(true) |
| 665 | matchesToResolve.push(matchId) |