* Accumulate D2 output multiplicities into per-key effect changes. * Tracks both insert values (new) and delete values (old) separately * so that update and exit events can include previousValue.
(
acc: Map<unknown, EffectChanges<T>>,
[[key, tupleData], multiplicity]: [
[unknown, [any, string | undefined]],
number,
],
)
| 1025 | * so that update and exit events can include previousValue. |
| 1026 | */ |
| 1027 | function accumulateEffectChanges<T>( |
| 1028 | acc: Map<unknown, EffectChanges<T>>, |
| 1029 | [[key, tupleData], multiplicity]: [ |
| 1030 | [unknown, [any, string | undefined]], |
| 1031 | number, |
| 1032 | ], |
| 1033 | ): Map<unknown, EffectChanges<T>> { |
| 1034 | const [value] = tupleData as [T, string | undefined] |
| 1035 | |
| 1036 | const changes: EffectChanges<T> = acc.get(key) || { |
| 1037 | deletes: 0, |
| 1038 | inserts: 0, |
| 1039 | } |
| 1040 | |
| 1041 | if (multiplicity < 0) { |
| 1042 | changes.deletes += Math.abs(multiplicity) |
| 1043 | // Keep only the first delete value — this is the pre-batch state |
| 1044 | changes.deleteValue ??= value |
| 1045 | } else if (multiplicity > 0) { |
| 1046 | changes.inserts += multiplicity |
| 1047 | // Always overwrite with the latest insert — this is the post-batch state |
| 1048 | changes.insertValue = value |
| 1049 | } |
| 1050 | |
| 1051 | acc.set(key, changes) |
| 1052 | return acc |
| 1053 | } |
| 1054 | |
| 1055 | /** Classify accumulated per-key changes into a DeltaEvent */ |
| 1056 | function classifyDelta<TRow extends object, TKey extends string | number>( |