@@ -0,0 +1,1473 @@
1+-- @inline export mkStrom arity=1
2+-- @inline export runStrom arity=1
3+-- @inline export mapStrom arity=2
4+-- @inline export filterStrom arity=2
5+-- @inline export takeStrom arity=2
6+-- @inline export runFold arity=3
7+-- @inline export runDrain arity=1
8+-- @inline export rangeStrom arity=2
9+-- @inline export scanStrom arity=3
10+-- @inline export zipWithStrom arity=3
11+-- @inline export mapPar arity=3
12+-- @inline export bindStrom arity=2
13+module Yoga.Om.Strom
14+ ( Strom
15+ -- Construction
16+ , empty
17+ , succeed
18+ , fail
19+ -- , die -- TODO: Fix type constraints
20+ , fromAff
21+ , fromOm
22+ , fromArray
23+ , fromFoldable
24+ , rangeStrom
25+ , iterateStrom
26+ , iterateStromInfinite
27+ , repeatStrom
28+ , repeatStromInfinite
29+ , repeatOmStrom
30+ , repeatOmStromInfinite
31+ , unfoldStrom
32+ , unfoldOmStrom
33+ -- Running
34+ , runCollect
35+ , runDrain
36+ , runFold
37+ , traverseStrom_
38+ , forStrom_
39+ , forMStrom_
40+ , traverseMStrom_
41+ -- Transformations
42+ , mapStrom
43+ , mapMStrom
44+ , bindStrom
45+ , scanStrom
46+ , mapAccumStrom
47+ , tapStrom
48+ , tapMStrom
49+ , filterStrom
50+ -- Selection
51+ , takeStrom
52+ , takeWhileStrom
53+ , takeUntilStrom
54+ , dropStrom
55+ , dropWhileStrom
56+ , filterMStrom
57+ , collectStrom
58+ , collectMStrom
59+ , changesStrom
60+ -- Grouping
61+ , groupedStrom
62+ , chunkedStrom
63+ , groupByStrom
64+ , partition
65+ , partitionMap
66+ -- Timing
67+ , debounce
68+ , throttle
69+ , delayStrom
70+ -- Combining
71+ , appendStrom
72+ , concatStrom
73+ , zipStrom
74+ , zipWithStrom
75+ , interleave
76+ , intersperse
77+ , merge
78+ , mergeAll
79+ , mergeND
80+ , mergeAllND
81+ , race
82+ , raceAll
83+ -- Parallel Processing
84+ , mapPar
85+ , mapMPar
86+ , foreachPar
87+ -- Error Handling
88+ , catchAll
89+ -- , catchSome -- TODO: Fix type constraints
90+ , retry
91+ , retryN
92+ , timeout
93+ , ensuring
94+ -- Resource Safety
95+ , bracket
96+ , bracketExit
97+ , acquireRelease
98+ ) where
99+
100+import Prelude
101+
102+import Control.Alt (class Alt, (<|>))
103+import Control.Alternative (class Alternative)
104+import Control.Monad.Error.Class (throwError)
105+import Control.Monad.Rec.Class (Step(..), tailRecM)
106+import Control.Monad.ST as ST
107+import Control.Monad.ST.Ref as STRef
108+import Control.Plus (class Plus)
109+import Data.Array as Array
110+import Data.Array.ST as STArray
111+import Data.Array.ST.Iterator as Iterator
112+import Data.Either (Either(..), either)
113+import Data.Foldable (class Foldable, foldl)
114+import Data.Functor (map) as Functor
115+import Data.List as List
116+import Data.Maybe (Maybe(..))
117+import Data.Time.Duration (Milliseconds(..))
118+import Data.Traversable (traverse)
119+import Data.Tuple (Tuple(..))
120+import Data.Variant (Variant)
121+import Effect.Aff (Aff, Error, delay, forkAff, joinFiber, killFiber)
122+import Effect.Aff.AVar as AVar
123+import Effect.Aff.Class (liftAff)
124+import Effect.Exception as Exception
125+import Effect.Now (now)
126+import Effect.Class (liftEffect)
127+import Data.DateTime.Instant (unInstant)
128+import Yoga.Om (Om)
129+import Yoga.Om as Om
130+
131+--------------------------------------------------------------------------------
132+-- Core Types
133+--------------------------------------------------------------------------------
134+
135+-- | A Strom is a stream of values that can:
136+-- | - Emit zero or more values of type `a`
137+-- | - Fail with errors tracked in the `err` row
138+-- | - Require a context `ctx`
139+-- | - Perform effects in Om
140+-- |
141+-- | Think of it as: `Om ctx err (Array a)` but lazy and chunked
142+-- |
143+-- | The Step type is interpreted as:
144+-- | - Loop (chunk, next) => emit chunk (if present) and continue with next
145+-- | - Done finalChunk => emit final chunk (if present) and stop
146+newtype Strom ctx err a = Strom
147+ { pull :: Om ctx err (Step (Tuple (Maybe (Chunk a)) (Strom ctx err a)) (Maybe (Chunk a)))
148+ }
149+
150+-- | Chunks for efficient batch processing
151+type Chunk a = Array a
152+
153+-- | Helper to create a Strom from a pull function
154+mkStrom :: forall ctx err a. Om ctx err (Step (Tuple (Maybe (Chunk a)) (Strom ctx err a)) (Maybe (Chunk a))) -> Strom ctx err a
155+mkStrom pull = Strom { pull }
156+
157+-- | Helper to unwrap a Strom
158+runStrom :: forall ctx err a. Strom ctx err a -> Om ctx err (Step (Tuple (Maybe (Chunk a)) (Strom ctx err a)) (Maybe (Chunk a)))
159+runStrom (Strom s) = s.pull
160+
161+--------------------------------------------------------------------------------
162+-- Non-Deterministic Merge Types
163+--------------------------------------------------------------------------------
164+
165+-- | Stream identifier for tracking which stream produced data
166+data StreamId = StreamId1 | StreamId2
167+
168+derive instance eqStreamId :: Eq StreamId
169+derive instance ordStreamId :: Ord StreamId
170+
171+-- | Messages passed through the coordination queue
172+data QueueItem a
173+ = DataChunk (Array a) StreamId
174+ | StreamDone StreamId
175+
176+derive instance functorQueueItem :: Functor QueueItem
177+
178+--------------------------------------------------------------------------------
179+-- Construction
180+--------------------------------------------------------------------------------
181+
182+-- | An empty stream
183+empty :: forall ctx err a. Strom ctx err a
184+empty = mkStrom $ pure $ Done Nothing
185+
186+-- | A stream with a single element
187+succeed :: forall ctx err a. a -> Strom ctx err a
188+succeed a = mkStrom $ pure $ Done $ Just [ a ]
189+
190+-- | A stream that fails immediately with an empty result (no error thrown)
191+fail :: forall ctx err a. Strom ctx err a
192+fail = empty
193+
194+-- TODO: Add die function with proper type constraints
195+-- die :: forall ctx err errors a. SingletonVariantRecord err (exception :: Error | errors) => Record err -> Strom ctx (exception :: Error | errors) a
196+-- die error = mkStrom $ Om.throw error
197+
198+-- | Create a stream from an Aff computation
199+fromAff :: forall ctx err a. Aff a -> Strom ctx err a
200+fromAff aff = mkStrom do
201+ value <- liftAff aff
202+ pure $ Done $ Just [ value ]
203+
204+-- | Create a stream from an Om computation
205+fromOm :: forall ctx err a. Om ctx err a -> Strom ctx err a
206+fromOm om = mkStrom do
207+ value <- om
208+ pure $ Done $ Just [ value ]
209+
210+-- | Create a stream from an array
211+fromArray :: forall ctx err a. Array a -> Strom ctx err a
212+fromArray arr =
213+ if Array.null arr then mkStrom $ pure $ Done Nothing
214+ else mkStrom $ pure $ Done $ Just arr
215+
216+-- | Create a stream from any foldable
217+fromFoldable :: forall ctx err f a. Foldable f => f a -> Strom ctx err a
218+fromFoldable = fromArray <<< Array.fromFoldable
219+
220+-- | Create a stream of integers in a range [start, end)
221+rangeStrom :: forall ctx err. Int -> Int -> Strom ctx err Int
222+rangeStrom start end = rangeHelper start end
223+ where
224+ rangeHelper current limit
225+ | current >= limit = empty
226+ | otherwise = mkStrom do
227+ let
228+ chunkSize = 1000
229+ chunkEnd = min (current + chunkSize) limit
230+ chunk = Array.range current (chunkEnd - 1)
231+ if chunkEnd >= limit then pure $ Done $ Just chunk -- Final chunk
232+ else pure $ Loop $ Tuple (Just chunk) (rangeHelper chunkEnd limit) -- Emit chunk and continue
233+
234+-- | Iterate producing values - limited to ~10,000 elements for stack safety
235+-- | For truly infinite streams, use `iterateStromInfinite` (with Aff overhead)
236+iterateStrom :: forall ctx err a. (a -> a) -> a -> Strom ctx err a
237+iterateStrom f initial = fromArray $ ST.run do
238+ arr <- STArray.unsafeThaw []
239+ valRef <- STRef.new initial
240+ iter <- Iterator.iterator (\i -> if i < 10000 then Just i else Nothing)
241+ Iterator.iterate iter \_ -> do
242+ v <- STRef.read valRef
243+ void $ STArray.push v arr
244+ void $ STRef.modify f valRef
245+ STArray.unsafeFreeze arr
246+
247+-- | Truly infinite iteration - stack-safe via Aff async boundaries
248+-- | Adds tiny delay (0ms) at each chunk to reset the stack
249+-- | Use this when you need more than 10k elements
250+iterateStromInfinite :: forall ctx err a. (a -> a) -> a -> Strom ctx err a
251+iterateStromInfinite f initial = iterateHelper initial
252+ where
253+ iterateHelper current = mkStrom do
254+ -- Async boundary resets the stack!
255+ liftAff $ delay (Milliseconds 0.0)
256+ let
257+ chunkSize = 10000
258+ -- Build chunk using STArray with Iterator (stack-safe!)
259+ result = ST.run do
260+ arr <- STArray.unsafeThaw []
261+ valRef <- STRef.new current
262+ iter <- Iterator.iterator (\i -> if i < chunkSize then Just i else Nothing)
263+ Iterator.iterate iter \_ -> do
264+ val <- STRef.read valRef
265+ void $ STArray.push val arr
266+ void $ STRef.modify f valRef
267+ finalVal <- STRef.read valRef
268+ frozen <- STArray.unsafeFreeze arr
269+ pure { arr: frozen, nextVal: finalVal }
270+ pure $ Loop $ Tuple (Just result.arr) (iterateHelper result.nextVal)
271+
272+-- | Repeat a value - limited to ~10,000 elements for stack safety
273+repeatStrom :: forall ctx err a. a -> Strom ctx err a
274+repeatStrom a = fromArray (Array.replicate 10000 a)
275+
276+-- | Truly infinite repeat - stack-safe via Aff async boundaries
277+-- | Adds tiny delay (0ms) at each chunk to reset the stack
278+repeatStromInfinite :: forall ctx err a. a -> Strom ctx err a
279+repeatStromInfinite a = repeatHelper unit
280+ where
281+ repeatHelper _ = mkStrom do
282+ -- Async boundary resets the stack!
283+ liftAff $ delay (Milliseconds 0.0)
284+ let chunk = Array.replicate 10000 a
285+ pure $ Loop $ Tuple (Just chunk) (repeatHelper unit)
286+
287+-- | Repeat an Om computation - limited to 100 iterations for stack safety
288+-- | For truly infinite repetition, use `repeatOmStromInfinite` (with Aff overhead)
289+repeatOmStrom :: forall ctx err a. Om ctx err a -> Strom ctx err a
290+repeatOmStrom om = repeatOmHelper 0
291+ where
292+ maxCount = 100
293+ repeatOmHelper count
294+ | count >= maxCount = empty
295+ | otherwise = mkStrom do
296+ value <- om
297+ pure $ Loop $ Tuple (Just [ value ]) (repeatOmHelper (count + 1))
298+
299+-- | Truly infinite Om repetition - stack-safe via Aff async boundaries
300+-- | Adds tiny delay (0ms) at each pull to reset the stack
301+repeatOmStromInfinite :: forall ctx err a. Om ctx err a -> Strom ctx err a
302+repeatOmStromInfinite om = repeatOmHelper unit
303+ where
304+ repeatOmHelper _ = mkStrom do
305+ -- Async boundary resets the stack!
306+ liftAff $ delay (Milliseconds 0.0)
307+ value <- om
308+ pure $ Loop $ Tuple (Just [ value ]) (repeatOmHelper unit)
309+
310+-- | Unfold a stream from a seed value - can be infinite if f never returns Nothing
311+-- | Emits chunks lazily, terminates when f returns Nothing
312+unfoldStrom :: forall ctx err a b. (b -> Maybe (Tuple a b)) -> b -> Strom ctx err a
313+unfoldStrom f seed = unfoldHelper seed
314+ where
315+ unfoldHelper currentSeed = mkStrom do
316+ let chunk = buildChunk currentSeed 10000
317+ case chunk.finalSeed of
318+ Nothing -> pure $ Done $ if Array.null chunk.values then Nothing else Just chunk.values
319+ Just nextSeed ->
320+ if Array.null chunk.values then runStrom (unfoldHelper nextSeed)
321+ else pure $ Loop $ Tuple (Just chunk.values) (unfoldHelper nextSeed)
322+
323+ buildChunk seed' size = ST.run do
324+ arr <- STArray.unsafeThaw []
325+ seedRef <- STRef.new seed'
326+ stoppedRef <- STRef.new false
327+ iter <- Iterator.iterator (\i -> if i < size then Just i else Nothing)
328+ Iterator.iterate iter \_ -> do
329+ stopped <- STRef.read stoppedRef
330+ when (not stopped) do
331+ s <- STRef.read seedRef
332+ case f s of
333+ Nothing -> void $ STRef.write true stoppedRef
334+ Just (Tuple value nextS) -> do
335+ void $ STArray.push value arr
336+ void $ STRef.write nextS seedRef
337+ frozen <- STArray.unsafeFreeze arr
338+ stopped <- STRef.read stoppedRef
339+ finalSeed <- STRef.read seedRef
340+ pure { values: frozen, finalSeed: if stopped then Nothing else Just finalSeed }
341+
342+-- | Unfold a stream with an effectful function
343+unfoldOmStrom :: forall ctx err a b. (b -> Om ctx err (Maybe (Tuple a b))) -> b -> Strom ctx err a
344+unfoldOmStrom f seed = mkStrom go
345+ where
346+ go = do
347+ result <- f seed
348+ case result of
349+ Nothing -> pure $ Done Nothing
350+ Just (Tuple value nextSeed) -> pure $ Loop $ Tuple (Just [ value ]) (unfoldOmStrom f nextSeed)
351+
352+--------------------------------------------------------------------------------
353+-- Transformations
354+--------------------------------------------------------------------------------
355+
356+-- | Map over elements
357+mapStrom :: forall ctx err a b. (a -> b) -> Strom ctx err a -> Strom ctx err b
358+mapStrom f stream = mkStrom do
359+ step <- runStrom stream
360+ case step of
361+ Done Nothing -> pure $ Done Nothing
362+ Done (Just chunk) -> pure $ Done $ Just $ Functor.map f chunk
363+ Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple (Functor.map (Functor.map f) maybeChunk) (mapStrom f next)
364+
365+-- | Map with a monadic effect
366+mapMStrom :: forall ctx err a b. (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
367+mapMStrom f stream = mkStrom do
368+ step <- runStrom stream
369+ case step of
370+ Done Nothing -> pure $ Done Nothing
371+ Done (Just chunk) -> do
372+ mapped <- traverse f chunk
373+ pure $ Done $ Just mapped
374+ Loop (Tuple maybeChunk next) -> do
375+ mappedChunk <- case maybeChunk of
376+ Nothing -> pure Nothing
377+ Just chunk -> Just <$> traverse f chunk
378+ pure $ Loop $ Tuple mappedChunk (mapMStrom f next)
379+
380+-- | Tap (observe without modifying)
381+tapStrom :: forall ctx err a. (a -> Unit) -> Strom ctx err a -> Strom ctx err a
382+tapStrom f stream = mkStrom do
383+ step <- runStrom stream
384+ case step of
385+ Done Nothing -> pure $ Done Nothing
386+ Done (Just chunk) -> do
387+ let _ = Functor.map f chunk
388+ pure $ Done $ Just chunk
389+ Loop (Tuple maybeChunk next) -> do
390+ let
391+ tappedChunk = case maybeChunk of
392+ Nothing -> Nothing
393+ Just chunk ->
394+ let
395+ _ = Functor.map f chunk
396+ in
397+ Just chunk
398+ pure $ Loop $ Tuple tappedChunk (tapStrom f next)
399+
400+-- | Tap with monadic effect (observe with effects)
401+tapMStrom :: forall ctx err a. (a -> Om ctx err Unit) -> Strom ctx err a -> Strom ctx err a
402+tapMStrom f stream = mkStrom do
403+ step <- runStrom stream
404+ case step of
405+ Done Nothing -> pure $ Done Nothing
406+ Done (Just chunk) -> do
407+ _ <- traverse f chunk
408+ pure $ Done $ Just chunk
409+ Loop (Tuple maybeChunk next) -> do
410+ tappedChunk <- case maybeChunk of
411+ Nothing -> pure Nothing
412+ Just chunk -> do
413+ _ <- traverse f chunk
414+ pure $ Just chunk
415+ pure $ Loop $ Tuple tappedChunk (tapMStrom f next)
416+
417+-- | Bind for streams (monadic flatMap/chain)
418+-- | Use >>= operator for cleaner syntax: `stream >>= f`
419+bindStrom :: forall ctx err a b. (a -> Strom ctx err b) -> Strom ctx err a -> Strom ctx err b
420+bindStrom f stream = stream >>= f
421+
422+-- | Scan with accumulator (like foldl but emits intermediate results)
423+scanStrom :: forall ctx err a b. (b -> a -> b) -> b -> Strom ctx err a -> Strom ctx err b
424+scanStrom f initial stream = scanHelper initial stream
425+ where
426+ scanHelper acc s = mkStrom do
427+ step <- runStrom s
428+ case step of
429+ Done Nothing -> pure $ Done Nothing
430+ Done (Just chunk) -> do
431+ -- Build results by folding and collecting intermediate values
432+ let
433+ Tuple _ results = Array.foldl
434+ ( \(Tuple prevAcc accList) item ->
435+ let
436+ newAcc = f prevAcc item
437+ in
438+ Tuple newAcc (Array.snoc accList newAcc)
439+ )
440+ (Tuple acc [])
441+ chunk
442+ pure $ Done $ if Array.null results then Nothing else Just results
443+ Loop (Tuple maybeChunk next) ->
444+ case maybeChunk of
445+ Nothing -> pure $ Loop $ Tuple Nothing (scanHelper acc next)
446+ Just chunk -> do
447+ let
448+ Tuple newAcc results = Array.foldl
449+ ( \(Tuple prevAcc accList) item ->
450+ let
451+ currAcc = f prevAcc item
452+ in
453+ Tuple currAcc (Array.snoc accList currAcc)
454+ )
455+ (Tuple acc [])
456+ chunk
457+ if Array.null results then pure $ Loop $ Tuple Nothing (scanHelper newAcc next)
458+ else pure $ Loop $ Tuple (Just results) (scanHelper newAcc next)
459+
460+-- | Map with accumulator (stateful map)
461+mapAccumStrom :: forall ctx err a b s. (s -> a -> Tuple s b) -> s -> Strom ctx err a -> Strom ctx err b
462+mapAccumStrom f initial stream = mapAccumHelper initial stream
463+ where
464+ mapAccumHelper state s = mkStrom do
465+ step <- runStrom s
466+ case step of
467+ Done Nothing -> pure $ Done Nothing
468+ Done (Just chunk) -> do
469+ let
470+ Tuple finalState results = Array.foldl
471+ ( \(Tuple st acc) a ->
472+ let
473+ Tuple newSt b = f st a
474+ in
475+ Tuple newSt (Array.snoc acc b)
476+ )
477+ (Tuple state [])
478+ chunk
479+ pure $ Done $ if Array.null results then Nothing else Just results
480+ Loop (Tuple maybeChunk next) ->
481+ case maybeChunk of
482+ Nothing -> pure $ Loop $ Tuple Nothing (mapAccumHelper state next)
483+ Just chunk -> do
484+ let
485+ Tuple newState results = Array.foldl
486+ ( \(Tuple st acc) a ->
487+ let
488+ Tuple newSt b = f st a
489+ in
490+ Tuple newSt (Array.snoc acc b)
491+ )
492+ (Tuple state [])
493+ chunk
494+ if Array.null results then pure $ Loop $ Tuple Nothing (mapAccumHelper newState next)
495+ else pure $ Loop $ Tuple (Just results) (mapAccumHelper newState next)
496+
497+-- | Filter elements
498+filterStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
499+filterStrom predicate stream = mkStrom do
500+ step <- runStrom stream
501+ case step of
502+ Done Nothing -> pure $ Done Nothing
503+ Done (Just chunk) -> do
504+ let filtered = Array.filter predicate chunk
505+ pure $ Done $ if Array.null filtered then Nothing else Just filtered
506+ Loop (Tuple maybeChunk next) -> do
507+ let
508+ filteredChunk = case maybeChunk of
509+ Nothing -> Nothing
510+ Just chunk ->
511+ let
512+ filtered = Array.filter predicate chunk
513+ in
514+ if Array.null filtered then Nothing else Just filtered
515+ pure $ Loop $ Tuple filteredChunk (filterStrom predicate next)
516+
517+--------------------------------------------------------------------------------
518+-- Selection
519+--------------------------------------------------------------------------------
520+
521+-- | Take n elements (with proper state tracking across chunks)
522+takeStrom :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a
523+takeStrom n stream = takeHelper n stream
524+ where
525+ takeHelper remaining s
526+ | remaining <= 0 = empty
527+ | otherwise = mkStrom do
528+ step <- runStrom s
529+ case step of
530+ Done Nothing -> pure $ Done Nothing
531+ Done (Just chunk) -> do
532+ let
533+ taken = Array.take remaining chunk
534+ numTaken = Array.length taken
535+ if numTaken == 0 then pure $ Done Nothing
536+ else pure $ Done $ Just taken
537+ Loop (Tuple maybeChunk next) ->
538+ case maybeChunk of
539+ Nothing -> pure $ Loop $ Tuple Nothing (takeHelper remaining next)
540+ Just chunk -> do
541+ let
542+ chunkLen = Array.length chunk
543+ taken = Array.take remaining chunk
544+ numTaken = Array.length taken
545+ newRemaining = remaining - numTaken
546+ if numTaken == 0 then pure $ Done Nothing
547+ else if newRemaining <= 0 then pure $ Done $ Just taken
548+ else pure $ Loop $ Tuple (Just taken) (takeHelper newRemaining next)
549+
550+-- | Take while predicate is true
551+takeWhileStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
552+takeWhileStrom predicate stream = mkStrom do
553+ step <- runStrom stream
554+ case step of
555+ Done Nothing -> pure $ Done Nothing
556+ Done (Just chunk) -> do
557+ let taken = Array.takeWhile predicate chunk
558+ if Array.length taken < Array.length chunk then pure $ Done $ Just taken
559+ else pure $ Done $ Just taken
560+ Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple maybeChunk (takeWhileStrom predicate next)
561+
562+-- | Drop n elements (with proper state tracking across chunks)
563+dropStrom :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a
564+dropStrom n stream
565+ | n <= 0 = stream
566+ | otherwise = dropHelper n stream
567+ where
568+ dropHelper remaining s
569+ | remaining <= 0 = s
570+ | otherwise = mkStrom do
571+ step <- runStrom s
572+ case step of
573+ Done Nothing -> pure $ Done Nothing
574+ Done (Just chunk) -> do
575+ let dropped = Array.drop remaining chunk
576+ pure $ Done $ if Array.null dropped then Nothing else Just dropped
577+ Loop (Tuple maybeChunk next) ->
578+ case maybeChunk of
579+ Nothing -> pure $ Loop $ Tuple Nothing (dropHelper remaining next)
580+ Just chunk ->
581+ let
582+ chunkLen = Array.length chunk
583+ in
584+ if remaining >= chunkLen then pure $ Loop $ Tuple Nothing (dropHelper (remaining - chunkLen) next)
585+ else
586+ let
587+ dropped = Array.drop remaining chunk
588+ in
589+ pure $ Loop $ Tuple (Just dropped) next
590+
591+-- | Take until predicate is true (includes the matching element)
592+takeUntilStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
593+takeUntilStrom predicate stream = mkStrom do
594+ step <- runStrom stream
595+ case step of
596+ Done Nothing -> pure $ Done Nothing
597+ Done (Just chunk) -> do
598+ case Array.findIndex predicate chunk of
599+ Just idx -> pure $ Done $ Just $ Array.take (idx + 1) chunk -- Include the matching element
600+ Nothing -> pure $ Done $ Just chunk
601+ Loop (Tuple maybeChunk next) ->
602+ case maybeChunk of
603+ Nothing -> pure $ Loop $ Tuple Nothing (takeUntilStrom predicate next)
604+ Just chunk -> case Array.findIndex predicate chunk of
605+ Just idx -> pure $ Done $ Just $ Array.take (idx + 1) chunk -- Found match, stop here
606+ Nothing -> pure $ Loop $ Tuple (Just chunk) (takeUntilStrom predicate next) -- Keep going
607+
608+-- | Drop while predicate is true
609+dropWhileStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
610+dropWhileStrom predicate stream = dropWhileHelper true stream
611+ where
612+ dropWhileHelper stillDropping s
613+ | not stillDropping = s
614+ | otherwise = mkStrom do
615+ step <- runStrom s
616+ case step of
617+ Done Nothing -> pure $ Done Nothing
618+ Done (Just chunk) -> do
619+ let dropped = Array.dropWhile predicate chunk
620+ pure $ Done $ if Array.null dropped then Nothing else Just dropped
621+ Loop (Tuple maybeChunk next) ->
622+ case maybeChunk of
623+ Nothing -> pure $ Loop $ Tuple Nothing (dropWhileHelper stillDropping next)
624+ Just chunk -> do
625+ let dropped = Array.dropWhile predicate chunk
626+ if Array.null dropped then pure $ Loop $ Tuple Nothing (dropWhileHelper true next)
627+ else pure $ Loop $ Tuple (Just dropped) next -- Found elements to keep, stop dropping
628+
629+-- | Filter with monadic predicate
630+filterMStrom :: forall ctx err a. (a -> Om ctx err Boolean) -> Strom ctx err a -> Strom ctx err a
631+filterMStrom predicate stream = mkStrom do
632+ step <- runStrom stream
633+ case step of
634+ Done Nothing -> pure $ Done Nothing
635+ Done (Just chunk) -> do
636+ filtered <- Array.filterA predicate chunk
637+ pure $ Done $ if Array.null filtered then Nothing else Just filtered
638+ Loop (Tuple maybeChunk next) -> do
639+ filteredChunk <- case maybeChunk of
640+ Nothing -> pure Nothing
641+ Just chunk -> do
642+ filtered <- Array.filterA predicate chunk
643+ pure $ if Array.null filtered then Nothing else Just filtered
644+ pure $ Loop $ Tuple filteredChunk (filterMStrom predicate next)
645+
646+-- | Collect with partial function (mapMaybe)
647+collectStrom :: forall ctx err a b. (a -> Maybe b) -> Strom ctx err a -> Strom ctx err b
648+collectStrom f stream = mkStrom do
649+ step <- runStrom stream
650+ case step of
651+ Done Nothing -> pure $ Done Nothing
652+ Done (Just chunk) -> do
653+ let collected = Array.mapMaybe f chunk
654+ pure $ Done $ if Array.null collected then Nothing else Just collected
655+ Loop (Tuple maybeChunk next) ->
656+ let
657+ collected = case maybeChunk of
658+ Nothing -> Nothing
659+ Just chunk ->
660+ let
661+ c = Array.mapMaybe f chunk
662+ in
663+ if Array.null c then Nothing else Just c
664+ in
665+ pure $ Loop $ Tuple collected (collectStrom f next)
666+
667+-- | Collect with monadic partial function
668+collectMStrom :: forall ctx err a b. (a -> Om ctx err (Maybe b)) -> Strom ctx err a -> Strom ctx err b
669+collectMStrom f stream = mkStrom do
670+ step <- runStrom stream
671+ case step of
672+ Done Nothing -> pure $ Done Nothing
673+ Done (Just chunk) -> do
674+ maybes <- traverse f chunk
675+ let collected = Array.mapMaybe identity maybes
676+ pure $ Done $ if Array.null collected then Nothing else Just collected
677+ Loop (Tuple maybeChunk next) -> do
678+ collected <- case maybeChunk of
679+ Nothing -> pure Nothing
680+ Just chunk -> do
681+ maybes <- traverse f chunk
682+ let c = Array.mapMaybe identity maybes
683+ pure $ if Array.null c then Nothing else Just c
684+ pure $ Loop $ Tuple collected (collectMStrom f next)
685+
686+-- | Remove consecutive duplicates
687+changesStrom :: forall ctx err a. Eq a => Strom ctx err a -> Strom ctx err a
688+changesStrom stream = changesHelper Nothing stream
689+ where
690+ changesHelper lastSeen s = mkStrom do
691+ step <- runStrom s
692+ case step of
693+ Done Nothing -> pure $ Done Nothing
694+ Done (Just chunk) -> do
695+ let
696+ Tuple _ result = Array.foldl
697+ ( \(Tuple prev acc) curr ->
698+ if Just curr == prev then Tuple (Just curr) acc
699+ else Tuple (Just curr) (Array.snoc acc curr)
700+ )
701+ (Tuple lastSeen [])
702+ chunk
703+ pure $ Done $ if Array.null result then Nothing else Just result
704+ Loop (Tuple maybeChunk next) ->
705+ case maybeChunk of
706+ Nothing -> pure $ Loop $ Tuple Nothing (changesHelper lastSeen next)
707+ Just chunk -> do
708+ let
709+ Tuple newLast result = Array.foldl
710+ ( \(Tuple prev acc) curr ->
711+ if Just curr == prev then Tuple (Just curr) acc
712+ else Tuple (Just curr) (Array.snoc acc curr)
713+ )
714+ (Tuple lastSeen [])
715+ chunk
716+ if Array.null result then pure $ Loop $ Tuple Nothing (changesHelper newLast next)
717+ else pure $ Loop $ Tuple (Just result) (changesHelper newLast next)
718+
719+--------------------------------------------------------------------------------
720+-- Grouping
721+--------------------------------------------------------------------------------
722+
723+-- | Group elements into fixed-size arrays
724+groupedStrom :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err (Array a)
725+groupedStrom size stream
726+ | size <= 0 = empty
727+ | otherwise = groupedHelper [] stream
728+ where
729+ groupedHelper buffer s = mkStrom do
730+ step <- runStrom s
731+ case step of
732+ Done Nothing ->
733+ -- End of stream, emit remaining buffer if non-empty
734+ pure $ Done $ if Array.null buffer then Nothing else Just [ buffer ]
735+ Done (Just chunk) -> do
736+ -- Process final chunk with buffer
737+ let
738+ combined = buffer <> chunk
739+ groups = groupsOf size combined
740+ completeGroups = Array.dropEnd 1 groups
741+ lastGroup = case Array.last groups of
742+ Just g -> if Array.length g == size then [ g ] else []
743+ Nothing -> []
744+ remainder = case Array.last groups of
745+ Just g -> if Array.length g < size then g else []
746+ Nothing -> []
747+ if Array.null remainder then pure $ Done $ if Array.null (completeGroups <> lastGroup) then Nothing else Just (completeGroups <> lastGroup)
748+ else pure $ Done $ if Array.null (completeGroups <> [ remainder ]) then Nothing else Just (completeGroups <> [ remainder ])
749+ Loop (Tuple maybeChunk next) ->
750+ case maybeChunk of
751+ Nothing -> pure $ Loop $ Tuple Nothing (groupedHelper buffer next)
752+ Just chunk -> do
753+ let
754+ combined = buffer <> chunk
755+ groups = groupsOf size combined
756+ completeGroups = if Array.null groups then [] else Array.dropEnd 1 groups
757+ newBuffer = case Array.last groups of
758+ Just g -> if Array.length g < size then g else []
759+ Nothing -> []
760+ if Array.null completeGroups then pure $ Loop $ Tuple Nothing (groupedHelper newBuffer next)
761+ else pure $ Loop $ Tuple (Just completeGroups) (groupedHelper newBuffer next)
762+
763+ groupsOf :: forall b. Int -> Array b -> Array (Array b)
764+ groupsOf n arr
765+ | n <= 0 || Array.null arr = []
766+ | otherwise =
767+ let
768+ group = Array.take n arr
769+ rest = Array.drop n arr
770+ in
771+ if Array.null group then [] else Array.cons group (groupsOf n rest)
772+
773+-- | Alias for groupedStrom
774+chunkedStrom :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err (Array a)
775+chunkedStrom = groupedStrom
776+
777+-- | Group consecutive elements by a key function
778+-- | Emits groups when the key changes
779+groupByStrom :: forall ctx err a k. Eq k => (a -> k) -> Strom ctx err a -> Strom ctx err (Array a)
780+groupByStrom keyFn stream = groupByHelper Nothing [] stream
781+ where
782+ groupByHelper lastKey buffer s = mkStrom do
783+ step <- runStrom s
784+ case step of
785+ Done Nothing ->
786+ pure $ Done $ if Array.null buffer then Nothing else Just [ buffer ]
787+ Done (Just chunk) -> do
788+ let
789+ Tuple _ result = Array.foldl
790+ ( \(Tuple currentKey acc) item ->
791+ let
792+ itemKey = keyFn item
793+ in
794+ case currentKey of
795+ Nothing -> Tuple (Just itemKey) { groups: [], buffer: [ item ] }
796+ Just k ->
797+ if k == itemKey then Tuple (Just itemKey) { groups: acc.groups, buffer: Array.snoc acc.buffer item }
798+ else Tuple (Just itemKey) { groups: Array.snoc acc.groups acc.buffer, buffer: [ item ] }
799+ )
800+ (Tuple lastKey { groups: [], buffer })
801+ chunk
802+ allGroups =
803+ if Array.null result.buffer then result.groups
804+ else Array.snoc result.groups result.buffer
805+ pure $ Done $ if Array.null allGroups then Nothing else Just allGroups
806+ Loop (Tuple maybeChunk next) ->
807+ case maybeChunk of
808+ Nothing -> pure $ Loop $ Tuple Nothing (groupByHelper lastKey buffer next)
809+ Just chunk -> do
810+ let
811+ Tuple newKey result = Array.foldl
812+ ( \(Tuple currentKey acc) item ->
813+ let
814+ itemKey = keyFn item
815+ in
816+ case currentKey of
817+ Nothing -> Tuple (Just itemKey) { groups: [], buffer: [ item ] }
818+ Just k ->
819+ if k == itemKey then Tuple (Just itemKey) { groups: acc.groups, buffer: Array.snoc acc.buffer item }
820+ else Tuple (Just itemKey) { groups: Array.snoc acc.groups acc.buffer, buffer: [ item ] }
821+ )
822+ (Tuple lastKey { groups: [], buffer })
823+ chunk
824+ if Array.null result.groups then pure $ Loop $ Tuple Nothing (groupByHelper newKey result.buffer next)
825+ else pure $ Loop $ Tuple (Just result.groups) (groupByHelper newKey result.buffer next)
826+
827+-- | Partition a stream by a predicate into two separate operations
828+-- | Returns a tuple of (trues, falses)
829+-- | Note: This requires running the stream twice, so it's not lazy
830+partition :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Tuple (Strom ctx err a) (Strom ctx err a)
831+partition predicate stream =
832+ Tuple
833+ (filterStrom predicate stream)
834+ (filterStrom (not <<< predicate) stream)
835+
836+-- | Partition and map in one pass
837+partitionMap :: forall ctx err a b c. (a -> Either b c) -> Strom ctx err a -> Tuple (Strom ctx err b) (Strom ctx err c)
838+partitionMap f stream =
839+ let
840+ leftOnly x = case f x of
841+ Left b -> Just b
842+ Right _ -> Nothing
843+ rightOnly x = case f x of
844+ Left _ -> Nothing
845+ Right c -> Just c
846+ in
847+ Tuple
848+ (collectStrom leftOnly stream)
849+ (collectStrom rightOnly stream)
850+
851+--------------------------------------------------------------------------------
852+-- Timing
853+--------------------------------------------------------------------------------
854+
855+-- | Debounce a stream - adds a delay after each element
856+-- | Useful for rate-limiting or giving downstream time to process
857+debounce :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a
858+debounce duration stream = mkStrom do
859+ step <- runStrom stream
860+ case step of
861+ Done Nothing -> pure $ Done Nothing
862+ Done (Just chunk) -> do
863+ Om.delay duration
864+ pure $ Done $ Just chunk
865+ Loop (Tuple maybeChunk next) -> do
866+ case maybeChunk of
867+ Nothing -> pure $ Loop $ Tuple Nothing (debounce duration next)
868+ Just chunk -> do
869+ Om.delay duration
870+ pure $ Loop $ Tuple (Just chunk) (debounce duration next)
871+
872+-- | Throttle a stream - emit at most once per duration period
873+-- | Tracks last emission time and skips elements that arrive too quickly
874+-- | Note: Uses wall-clock time, not processing time
875+throttle :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a
876+throttle (Milliseconds duration) stream = throttleWithState 0.0 stream
877+ where
878+ throttleWithState lastEmit s = mkStrom do
879+ step <- runStrom s
880+ case step of
881+ Done Nothing -> pure $ Done Nothing
882+ Done (Just chunk) -> do
883+ Tuple newLastEmit filtered <- filterByTime lastEmit chunk
884+ pure $ Done $ if Array.null filtered then Nothing else Just filtered
885+ Loop (Tuple maybeChunk next) -> do
886+ case maybeChunk of
887+ Nothing -> pure $ Loop $ Tuple Nothing (throttleWithState lastEmit next)
888+ Just chunk -> do
889+ Tuple newLastEmit filtered <- filterByTime lastEmit chunk
890+ if Array.null filtered then pure $ Loop $ Tuple Nothing (throttleWithState newLastEmit next)
891+ else pure $ Loop $ Tuple (Just filtered) (throttleWithState newLastEmit next)
892+
893+ filterByTime lastEmit chunk = do
894+ Tuple newLastEmit results <- Array.foldl
895+ ( \acc value -> do
896+ Tuple currentLast filtered <- acc
897+ currentTime <- liftEffect now
898+ let Milliseconds currentMillis = unInstant currentTime
899+ let elapsed = currentMillis - currentLast
900+ if elapsed >= duration then
901+ pure $ Tuple currentMillis (Array.snoc filtered value)
902+ else
903+ pure $ Tuple currentLast filtered
904+ )
905+ (pure $ Tuple lastEmit [])
906+ chunk
907+ pure $ Tuple newLastEmit results
908+
909+-- | Delay each element by a duration
910+delayStrom :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a
911+delayStrom duration stream = mkStrom do
912+ Om.delay duration
913+ runStrom stream
914+
915+--------------------------------------------------------------------------------
916+-- Running
917+--------------------------------------------------------------------------------
918+
919+-- | Fold over all elements
920+runFold :: forall ctx err a b. b -> (b -> a -> b) -> Strom ctx err a -> Om ctx err b
921+runFold initial f stream = tailRecM go (Tuple initial stream)
922+ where
923+ go (Tuple acc s) = do
924+ step <- runStrom s
925+ case step of
926+ Done Nothing -> pure $ Done acc
927+ Done (Just chunk) -> pure $ Done $ foldl f acc chunk
928+ Loop (Tuple maybeChunk next) ->
929+ -- Process chunk if present and continue
930+ case maybeChunk of
931+ Nothing -> pure $ Loop $ Tuple acc next
932+ Just chunk -> pure $ Loop $ Tuple (foldl f acc chunk) next
933+
934+-- | Collect all elements into an array (O(n) using chunk accumulation)
935+runCollect :: forall ctx err a. Strom ctx err a -> Om ctx err (Array a)
936+runCollect stream = tailRecM go (Tuple List.Nil stream)
937+ where
938+ go (Tuple chunkList s) = do
939+ step <- runStrom s
940+ case step of
941+ Done Nothing -> pure $ Done $ Array.concat $ Array.fromFoldable $ List.reverse chunkList
942+ Done (Just chunk) -> pure $ Done $ Array.concat $ Array.fromFoldable $ List.reverse $ List.Cons chunk chunkList
943+ Loop (Tuple maybeChunk next) ->
944+ -- Emit the chunk (if present) and continue
945+ case maybeChunk of
946+ Nothing -> pure $ Loop $ Tuple chunkList next
947+ Just chunk -> pure $ Loop $ Tuple (List.Cons chunk chunkList) next
948+
949+-- | Run the stream and discard all elements
950+runDrain :: forall ctx err a. Strom ctx err a -> Om ctx err Unit
951+runDrain = runFold unit (\_ _ -> unit)
952+
953+-- | Traverse the stream with effects, discarding results
954+traverseStrom_ :: forall ctx err a. (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit
955+traverseStrom_ f stream = tailRecM go stream
956+ where
957+ go s = do
958+ step <- runStrom s
959+ case step of
960+ Done Nothing -> pure $ Done unit
961+ Done (Just chunk) -> do
962+ _ <- traverse f chunk
963+ pure $ Done unit
964+ Loop (Tuple maybeChunk next) -> do
965+ -- Process chunk if present, then continue
966+ case maybeChunk of
967+ Nothing -> pure unit
968+ Just chunk -> void $ traverse f chunk
969+ pure $ Loop next
970+
971+-- | Alias for traverseStrom_ with arguments flipped
972+forStrom_ :: forall ctx err a. Strom ctx err a -> (a -> Om ctx err Unit) -> Om ctx err Unit
973+forStrom_ = flip traverseStrom_
974+
975+-- | Alias for traverseStrom_ (FP convention)
976+forMStrom_ :: forall ctx err a. Strom ctx err a -> (a -> Om ctx err Unit) -> Om ctx err Unit
977+forMStrom_ = forStrom_
978+
979+-- | Alias for traverseStrom_ (explicit monadic naming)
980+traverseMStrom_ :: forall ctx err a. (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit
981+traverseMStrom_ = traverseStrom_
982+
983+--------------------------------------------------------------------------------
984+-- Cancellable Execution
985+--------------------------------------------------------------------------------
986+
987+-- | Run a stream in the background, returning a Fiber that can be cancelled
988+-- |
989+-- | To use cancellable streams, convert your stream to Om and use Om.launchOm:
990+-- |
991+-- | Example:
992+-- | ```purescript
993+-- | fiber <- Om.launchOm ctx handlers (Strom.runCollect myStream)
994+-- | -- ... later ...
995+-- | killFiber (error "Cancelled") fiber
996+-- | ```
997+-- |
998+-- | Or for subscription-style:
999+-- | ```purescript
1000+-- | fiber <- Om.launchOm ctx handlers (Strom.traverseStrom_ callback myStream)
1001+-- | -- ... later ...
1002+-- | killFiber (error "Unsubscribed") fiber
1003+-- | ```
1004+-- |
1005+-- | Note: You can use Om.launchOm directly with any Om-returning function like:
1006+-- | - `runCollect` - collect results in background
1007+-- | - `runDrain` - run for side effects
1008+-- | - `traverseStrom_` - process each element with callback
1009+-- | - `runFold` - fold with accumulator
1010+
1011+--------------------------------------------------------------------------------
1012+-- Combining
1013+--------------------------------------------------------------------------------
1014+
1015+-- | Append two streams
1016+-- | Now properly handles multi-chunk streams with new Step protocol
1017+-- | Note: s2 is appended TO s1 (s1 comes first, then s2)
1018+-- | This matches pipeline semantics: stream # appendStrom other means stream followed by other
1019+appendStrom :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1020+appendStrom toAppend baseStream = mkStrom do
1021+ step <- runStrom baseStream
1022+ case step of
1023+ Done Nothing -> runStrom toAppend
1024+ Done (Just chunk) -> pure $ Loop $ Tuple (Just chunk) toAppend -- Emit chunk and continue with toAppend
1025+ Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple maybeChunk (appendStrom toAppend next)
1026+
1027+-- | Concatenate an array of streams
1028+-- | Uses foldr with flip to match appendStrom's pipeline-friendly parameter order
1029+concatStrom :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
1030+concatStrom streams = Array.foldr (flip appendStrom) empty streams
1031+
1032+-- | Zip two streams together
1033+zipStrom :: forall ctx err a b. Strom ctx err a -> Strom ctx err b -> Strom ctx err (Tuple a b)
1034+zipStrom = zipWithStrom Tuple
1035+
1036+-- | Zip two streams with a function
1037+zipWithStrom :: forall ctx err a b c. (a -> b -> c) -> Strom ctx err a -> Strom ctx err b -> Strom ctx err c
1038+zipWithStrom f s1 s2 = mkStrom do
1039+ step1 <- runStrom s1
1040+ step2 <- runStrom s2
1041+ case step1, step2 of
1042+ Done (Just arr1), Done (Just arr2) -> do
1043+ let zipped = Array.zipWith f arr1 arr2
1044+ pure $ Done $ Just zipped
1045+ Done Nothing, _ -> pure $ Done Nothing
1046+ _, Done Nothing -> pure $ Done Nothing
1047+ Loop (Tuple maybeChunk1 next1), Loop (Tuple maybeChunk2 next2) ->
1048+ -- Both looping: zip chunks if both present, continue with both nexts
1049+ case maybeChunk1, maybeChunk2 of
1050+ Just chunk1, Just chunk2 -> pure $ Loop $ Tuple (Just $ Array.zipWith f chunk1 chunk2) (zipWithStrom f next1 next2)
1051+ _, _ -> pure $ Loop $ Tuple Nothing (zipWithStrom f next1 next2)
1052+ Loop (Tuple _ next1), _ -> pure $ Loop $ Tuple Nothing (zipWithStrom f next1 s2)
1053+ _, Loop (Tuple _ next2) -> pure $ Loop $ Tuple Nothing (zipWithStrom f s1 next2)
1054+
1055+-- | Interleave two streams - alternates between elements from each stream
1056+-- | Similar to merge but deterministic - always alternates
1057+interleave :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1058+interleave s1 s2 = interleaveHelper true s1 s2
1059+ where
1060+ interleaveHelper preferFirst first second = mkStrom do
1061+ if preferFirst then do
1062+ step1 <- runStrom first
1063+ case step1 of
1064+ Done Nothing -> runStrom second
1065+ Done (Just chunk1) -> pure $ Loop $ Tuple (Just chunk1) (interleaveHelper false second first)
1066+ Loop (Tuple maybeChunk1 next1) ->
1067+ pure $ Loop $ Tuple maybeChunk1 (interleaveHelper false second next1)
1068+ else do
1069+ step2 <- runStrom second
1070+ case step2 of
1071+ Done Nothing -> runStrom first
1072+ Done (Just chunk2) -> pure $ Loop $ Tuple (Just chunk2) (interleaveHelper true first second)
1073+ Loop (Tuple maybeChunk2 next2) ->
1074+ pure $ Loop $ Tuple maybeChunk2 (interleaveHelper true first next2)
1075+
1076+-- | Intersperse a separator element between each element of the stream
1077+intersperse :: forall ctx err a. a -> Strom ctx err a -> Strom ctx err a
1078+intersperse separator stream = intersperseHelper true stream
1079+ where
1080+ intersperseHelper isFirst s = mkStrom do
1081+ step <- runStrom s
1082+ case step of
1083+ Done Nothing -> pure $ Done Nothing
1084+ Done (Just chunk) -> do
1085+ let
1086+ withSeparators =
1087+ if isFirst then Array.intercalate [ separator ] (Array.singleton <$> chunk)
1088+ else Array.cons separator (Array.intercalate [ separator ] (Array.singleton <$> chunk))
1089+ pure $ Done $ Just withSeparators
1090+ Loop (Tuple maybeChunk next) ->
1091+ case maybeChunk of
1092+ Nothing -> pure $ Loop $ Tuple Nothing (intersperseHelper isFirst next)
1093+ Just chunk -> do
1094+ let
1095+ withSeparators =
1096+ if isFirst then Array.intercalate [ separator ] (Array.singleton <$> chunk)
1097+ else Array.cons separator (Array.intercalate [ separator ] (Array.singleton <$> chunk))
1098+ pure $ Loop $ Tuple (Just withSeparators) (intersperseHelper false next)
1099+
1100+--------------------------------------------------------------------------------
1101+-- Concurrency
1102+--------------------------------------------------------------------------------
1103+
1104+-- | Merge two streams concurrently
1105+-- | Elements are emitted as soon as they're available from either stream
1106+merge :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1107+merge s1 s2 = mkStrom do
1108+ -- Pull from both streams concurrently
1109+ step1 <- runStrom s1
1110+ step2 <- runStrom s2
1111+ case step1, step2 of
1112+ Done Nothing, Done Nothing -> pure $ Done Nothing
1113+ Done Nothing, Done (Just chunk2) -> pure $ Done $ Just chunk2
1114+ Done (Just chunk1), Done Nothing -> pure $ Done $ Just chunk1
1115+ Done (Just chunk1), Done (Just chunk2) -> pure $ Done $ Just (chunk1 <> chunk2)
1116+ Done Nothing, Loop (Tuple maybeChunk2 next2) ->
1117+ pure $ Loop $ Tuple maybeChunk2 next2
1118+ Done (Just chunk1), Loop (Tuple maybeChunk2 next2) ->
1119+ case maybeChunk2 of
1120+ Nothing -> pure $ Loop $ Tuple (Just chunk1) next2
1121+ Just chunk2 -> pure $ Loop $ Tuple (Just $ chunk1 <> chunk2) next2
1122+ Loop (Tuple maybeChunk1 next1), Done Nothing ->
1123+ pure $ Loop $ Tuple maybeChunk1 next1
1124+ Loop (Tuple maybeChunk1 next1), Done (Just chunk2) ->
1125+ case maybeChunk1 of
1126+ Nothing -> pure $ Loop $ Tuple (Just chunk2) next1
1127+ Just chunk1 -> pure $ Loop $ Tuple (Just $ chunk1 <> chunk2) next1
1128+ Loop (Tuple maybeChunk1 next1), Loop (Tuple maybeChunk2 next2) ->
1129+ -- Both streams continue, merge chunks and continue with both
1130+ case maybeChunk1, maybeChunk2 of
1131+ Nothing, Nothing -> pure $ Loop $ Tuple Nothing (merge next1 next2)
1132+ Just c1, Nothing -> pure $ Loop $ Tuple (Just c1) (merge next1 next2)
1133+ Nothing, Just c2 -> pure $ Loop $ Tuple (Just c2) (merge next1 next2)
1134+ Just c1, Just c2 -> pure $ Loop $ Tuple (Just $ c1 <> c2) (merge next1 next2)
1135+
1136+-- | Merge multiple streams concurrently
1137+mergeAll :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
1138+mergeAll streams = Array.foldl merge empty streams
1139+
1140+-- | Race two streams - returns elements from whichever produces first
1141+-- | Uses Om.race for true concurrent racing
1142+race :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1143+race s1 s2 = mkStrom do
1144+ Om.race [ runStrom s1, runStrom s2 ]
1145+
1146+-- | Race multiple streams - returns elements from whichever produces first
1147+raceAll :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
1148+raceAll streams = mkStrom do
1149+ Om.race (Functor.map runStrom streams)
1150+
1151+-- | Non-deterministic merge: Merge two streams concurrently, emitting elements as soon as they're available
1152+-- | Unlike `merge`, this runs both streams in parallel and elements may arrive in any order
1153+-- | The stream that produces results faster will have its elements appear first
1154+-- | Optimized: Spawns producer fibers once and uses a queue for communication
1155+mergeND :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1156+mergeND stream1 stream2 = mkStrom do
1157+ -- Capture context
1158+ ctx <- Om.ask
1159+
1160+ liftAff do
1161+ -- Create communication queue
1162+ queue <- AVar.empty
1163+
1164+ -- Spawn producer fiber for stream1
1165+ fiber1 <- forkAff do
1166+ let
1167+ producer s = do
1168+ stepResult <- Om.runReader ctx (runStrom s)
1169+ step <- either (\e -> throwError (Exception.error "Stream error")) pure stepResult
1170+ case step of
1171+ Done Nothing ->
1172+ AVar.put (Left StreamId1) queue -- Signal stream1 done
1173+ Done (Just chunk) -> do
1174+ AVar.put (Right chunk) queue -- Emit final chunk
1175+ AVar.put (Left StreamId1) queue -- Signal stream1 done
1176+ Loop (Tuple maybeChunk next) -> do
1177+ case maybeChunk of
1178+ Just chunk -> AVar.put (Right chunk) queue
1179+ Nothing -> pure unit
1180+ producer next
1181+ producer stream1
1182+
1183+ -- Spawn producer fiber for stream2
1184+ fiber2 <- forkAff do
1185+ let
1186+ producer s = do
1187+ stepResult <- Om.runReader ctx (runStrom s)
1188+ step <- either (\e -> throwError (Exception.error "Stream error")) pure stepResult
1189+ case step of
1190+ Done Nothing ->
1191+ AVar.put (Left StreamId2) queue -- Signal stream2 done
1192+ Done (Just chunk) -> do
1193+ AVar.put (Right chunk) queue -- Emit final chunk
1194+ AVar.put (Left StreamId2) queue -- Signal stream2 done
1195+ Loop (Tuple maybeChunk next) -> do
1196+ case maybeChunk of
1197+ Just chunk -> AVar.put (Right chunk) queue
1198+ Nothing -> pure unit
1199+ producer next
1200+ producer stream2
1201+
1202+ -- Consumer: pull from queue
1203+ let
1204+ consumer doneCount = mkStrom $ liftAff do
1205+ if doneCount >= 2 then do
1206+ -- Both streams done, cleanup
1207+ killFiber (Exception.error "done") fiber1
1208+ killFiber (Exception.error "done") fiber2
1209+ pure $ Done Nothing
1210+ else do
1211+ msg <- AVar.take queue
1212+ case msg of
1213+ Left _streamId -> do
1214+ -- One stream finished, recurse
1215+ stepResult <- Om.runReader ctx (runStrom $ consumer (doneCount + 1))
1216+ either (\_ -> throwError (Exception.error "Consumer error")) pure stepResult
1217+ Right chunk ->
1218+ -- Got a chunk
1219+ pure $ Loop $ Tuple (Just chunk) (consumer doneCount)
1220+
1221+ stepResult <- Om.runReader ctx (runStrom $ consumer 0)
1222+ either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
1223+
1224+-- | Non-deterministic merge of multiple streams
1225+-- | All streams run concurrently and emit elements as soon as they're available
1226+mergeAllND :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
1227+mergeAllND streams = Array.foldl mergeND empty streams
1228+
1229+--------------------------------------------------------------------------------
1230+-- Parallel Processing
1231+--------------------------------------------------------------------------------
1232+
1233+-- | Map over elements in parallel with bounded concurrency
1234+-- | Processes chunks in parallel groups up to the concurrency limit
1235+mapPar :: forall ctx err a b. Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
1236+mapPar concurrency f stream
1237+ | concurrency <= 1 = mapMStrom f stream -- Sequential fallback
1238+ | otherwise =
1239+ mkStrom do
1240+ step <- runStrom stream
1241+ case step of
1242+ Done Nothing -> pure $ Done Nothing
1243+ Done (Just chunk) -> do
1244+ -- Split chunk into groups of `concurrency` size and process each group in parallel
1245+ let groups = chunksOf concurrency chunk
1246+ results <- traverse (\group -> Om.inParallel (Functor.map f group)) groups
1247+ pure $ Done $ Just $ Array.concat results
1248+ Loop (Tuple maybeChunk next) -> do
1249+ mappedChunk <- case maybeChunk of
1250+ Nothing -> pure Nothing
1251+ Just chunk -> do
1252+ let groups = chunksOf concurrency chunk
1253+ results <- traverse (\group -> Om.inParallel (Functor.map f group)) groups
1254+ pure $ Just $ Array.concat results
1255+ pure $ Loop $ Tuple mappedChunk (mapPar concurrency f next)
1256+ where
1257+ chunksOf :: forall x. Int -> Array x -> Array (Array x)
1258+ chunksOf n arr
1259+ | n <= 0 || Array.null arr = []
1260+ | otherwise =
1261+ let
1262+ group = Array.take n arr
1263+ rest = Array.drop n arr
1264+ in
1265+ if Array.null group then [] else Array.cons group (chunksOf n rest)
1266+
1267+-- | Alias for mapPar with explicit "M" naming
1268+mapMPar :: forall ctx err a b. Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
1269+mapMPar = mapPar
1270+
1271+-- | For each element, execute an effect in parallel with bounded concurrency
1272+foreachPar :: forall ctx err a. Int -> (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit
1273+foreachPar concurrency f stream = runDrain (mapPar concurrency f stream)
1274+
1275+--------------------------------------------------------------------------------
1276+-- Error Handling
1277+--------------------------------------------------------------------------------
1278+
1279+-- | Catch all errors and recover with a handler function
1280+-- | Note: Uses handleErrors' for full error handling
1281+catchAll :: forall ctx err a. (Variant (exception :: Error | err) -> Strom ctx err a) -> Strom ctx err a -> Strom ctx err a
1282+catchAll handler stream = mkStrom do
1283+ Om.handleErrors'
1284+ (\err -> runStrom (handler err))
1285+ (runStrom stream)
1286+
1287+-- TODO: Fix catchSome type constraints
1288+-- catchSome :: forall ctx err handlers a. (Record handlers) -> Strom ctx err a -> Strom ctx err a
1289+-- catchSome handlers stream = mkStrom do
1290+-- Om.handleErrors handlers (runStrom stream)
1291+
1292+-- | Retry a stream on failure up to 3 times
1293+retry :: forall ctx err a. Strom ctx err a -> Strom ctx err a
1294+retry = retryN 3
1295+
1296+-- | Retry a stream N times on failure
1297+retryN :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a
1298+retryN n stream
1299+ | n <= 0 = stream
1300+ | otherwise = catchAll (\_ -> retryN (n - 1) stream) stream
1301+
1302+-- | Run a finaliser after the stream completes (success or failure)
1303+ensuring :: forall ctx err a. Om ctx err Unit -> Strom ctx err a -> Strom ctx err a
1304+ensuring finaliser stream = mkStrom do
1305+ result <- Om.handleErrors'
1306+ ( \err -> do
1307+ finaliser
1308+ throwError err
1309+ )
1310+ ( runStrom stream >>= \step -> do
1311+ finaliser
1312+ pure step
1313+ )
1314+ pure result
1315+
1316+-- | Add a timeout to stream operations
1317+-- | If the stream doesn't complete within the duration, it terminates as empty
1318+timeout :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a
1319+timeout duration stream = mkStrom do
1320+ Om.race [ runStrom stream, Om.delay duration *> runStrom empty ]
1321+
1322+--------------------------------------------------------------------------------
1323+-- Resource Safety
1324+--------------------------------------------------------------------------------
1325+
1326+-- | Bracket pattern: acquire, use, release
1327+-- | Guarantees release runs even if stream fails
1328+bracket
1329+ :: forall ctx err a b
1330+ . Om ctx err a -- acquire
1331+ -> (a -> Om ctx err Unit) -- release
1332+ -> (a -> Strom ctx err b) -- use
1333+ -> Strom ctx err b
1334+bracket acquire release use = mkStrom do
1335+ resource <- acquire
1336+ Om.handleErrors'
1337+ ( \err -> do
1338+ release resource
1339+ throwError err
1340+ )
1341+ (runStrom (use resource))
1342+ >>= \step -> do
1343+ case step of
1344+ Done _ -> release resource
1345+ Loop _ -> pure unit
1346+ pure step
1347+
1348+-- | Bracket with exit information (whether stream succeeded or failed)
1349+bracketExit
1350+ :: forall ctx err a b
1351+ . Om ctx err a -- acquire
1352+ -> (a -> Maybe (Variant (exception :: Error | err)) -> Om ctx err Unit) -- release with exit info
1353+ -> (a -> Strom ctx err b) -- use
1354+ -> Strom ctx err b
1355+bracketExit acquire release use = mkStrom do
1356+ resource <- acquire
1357+ Om.handleErrors'
1358+ ( \err -> do
1359+ release resource (Just err)
1360+ throwError err
1361+ )
1362+ (runStrom (use resource))
1363+ >>= \step -> do
1364+ case step of
1365+ Done _ -> release resource Nothing
1366+ Loop _ -> pure unit
1367+ pure step
1368+
1369+-- | Simplified acquire-release pattern
1370+acquireRelease
1371+ :: forall ctx err a b
1372+ . Om ctx err a -- acquire
1373+ -> (a -> Om ctx err Unit) -- release
1374+ -> (a -> Strom ctx err b) -- use
1375+ -> Strom ctx err b
1376+acquireRelease = bracket
1377+
1378+--------------------------------------------------------------------------------
1379+-- Instances
1380+--------------------------------------------------------------------------------
1381+
1382+instance Functor (Strom ctx err) where
1383+ map f stream = mkStrom
1384+ ( runStrom stream >>= \step ->
1385+ case step of
1386+ Done Nothing -> pure $ Done Nothing
1387+ Done (Just chunk) -> pure $ Done $ Just $ Functor.map f chunk
1388+ Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple (Functor.map (Functor.map f) maybeChunk) (Functor.map f next)
1389+ )
1390+
1391+instance Apply (Strom ctx err) where
1392+ apply fs as = mkStrom
1393+ ( runStrom fs >>= \stepF ->
1394+ runStrom as >>= \stepA ->
1395+ case stepF, stepA of
1396+ Done (Just funcs), Done (Just vals) ->
1397+ let
1398+ results = do
1399+ f <- funcs
1400+ a <- vals
1401+ pure (f a)
1402+ in
1403+ pure $ Done $ Just results
1404+ Done Nothing, _ -> pure $ Done Nothing
1405+ _, Done Nothing -> pure $ Done Nothing
1406+ Loop (Tuple maybeChunkF nextF), Loop (Tuple maybeChunkA nextA) ->
1407+ -- Apply chunks if both present
1408+ case maybeChunkF, maybeChunkA of
1409+ Just chunkF, Just chunkA ->
1410+ let
1411+ applied = do
1412+ f <- chunkF
1413+ a <- chunkA
1414+ pure (f a)
1415+ in
1416+ pure $ Loop $ Tuple (Just applied) (nextF <*> nextA)
1417+ _, _ -> pure $ Loop $ Tuple Nothing (nextF <*> nextA)
1418+ Loop (Tuple _ nextF), _ -> pure $ Loop $ Tuple Nothing (nextF <*> as)
1419+ _, Loop (Tuple _ nextA) -> pure $ Loop $ Tuple Nothing (fs <*> nextA)
1420+ )
1421+
1422+instance Applicative (Strom ctx err) where
1423+ pure = succeed
1424+
1425+instance Bind (Strom ctx err) where
1426+ bind stream f = mkStrom
1427+ ( runStrom stream >>= \step ->
1428+ case step of
1429+ Done Nothing -> pure $ Done Nothing
1430+ Done (Just chunk) ->
1431+ let
1432+ streams = Functor.map f chunk
1433+ in
1434+ runStrom $ concatStrom streams
1435+ Loop (Tuple maybeChunk next) ->
1436+ -- When looping, if we have a chunk, we need to process it through f and concat
1437+ case maybeChunk of
1438+ Nothing -> pure $ Loop $ Tuple Nothing (next >>= f)
1439+ Just chunk ->
1440+ let
1441+ streams = Functor.map f chunk
1442+ prefixStream = concatStrom streams
1443+ in
1444+ runStrom $ appendStrom prefixStream (next >>= f)
1445+ )
1446+
1447+instance Monad (Strom ctx err)
1448+
1449+instance Semigroup (Strom ctx err a) where
1450+ append s1 s2 = mkStrom
1451+ ( runStrom s1 >>= \step ->
1452+ case step of
1453+ Done Nothing -> runStrom s2
1454+ Done (Just chunk) -> pure $ Loop $ Tuple (Just chunk) s2
1455+ Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple maybeChunk (next <> s2)
1456+ )
1457+
1458+instance Monoid (Strom ctx err a) where
1459+ mempty = empty
1460+
1461+instance Alt (Strom ctx err) where
1462+ alt s1 s2 = mkStrom
1463+ ( runStrom s1 >>= \step ->
1464+ case step of
1465+ Done Nothing -> runStrom s2
1466+ Done (Just chunk) -> pure $ Loop $ Tuple (Just chunk) s2
1467+ Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple maybeChunk (next <|> s2)
1468+ )
1469+
1470+instance Plus (Strom ctx err) where
1471+ empty = mkStrom $ pure $ Done Nothing
1472+
1473+instance Alternative (Strom ctx err)