diff --git a/spago.lock b/spago.lock
index 9cffae7..06d32b5 100644
--- a/spago.lock
+++ b/spago.lock
@@ -9,7 +9,7 @@ workspace:
         - datetime: ">=6.1.0 <7.0.0"
         - effect: ">=4.0.0 <5.0.0"
         - exceptions: ">=6.0.0 <7.0.0"
-        - foldable-traversable
+        - foldable-traversable: ">=6.0.0 <7.0.0"
         - foreign: ">=7.0.0 <8.0.0"
         - foreign-object: ">=4.1.0 <5.0.0"
         - integers: ">=6.0.0 <7.0.0"
@@ -18,7 +18,7 @@ workspace:
         - newtype: ">=5.0.0 <6.0.0"
         - node-buffer: ">=9.0.0 <10.0.0"
         - node-event-emitter: ">=3.0.0 <4.0.0"
-        - node-stream-pipes
+        - node-stream-pipes: ">=1.2.3 <2.0.0"
         - node-streams: ">=9.0.0 <10.0.0"
         - nullable: ">=6.0.0 <7.0.0"
         - numbers: ">=9.0.1 <10.0.0"
@@ -28,11 +28,11 @@ workspace:
         - prelude: ">=6.0.1 <7.0.0"
         - record: ">=4.0.0 <5.0.0"
         - record-extra: ">=5.0.1 <6.0.0"
-        - st
+        - st: ">=6.2.0 <7.0.0"
         - strings: ">=6.0.1 <7.0.0"
         - tailrec: ">=6.1.0 <7.0.0"
         - transformers: ">=6.0.0 <7.0.0"
-        - tuples
+        - tuples: ">=7.0.0 <8.0.0"
         - typelevel-prelude: ">=7.0.0 <8.0.0"
         - unsafe-coerce: ">=6.0.0 <7.0.0"
       test_dependencies:
@@ -120,12 +120,10 @@ workspace:
         - typelevel-prelude
         - unfoldable
         - unicode
+        - unordered-collections
         - unsafe-coerce
         - variant
-  extra_packages:
-    node-stream-pipes:
-      git: https://git.orionkindel.com/orion/purescript-node-stream-pipes
-      ref: v1.0.5
+  extra_packages: {}
 packages:
   aff:
     type: registry
@@ -608,9 +606,9 @@ packages:
     dependencies:
       - effect
   node-stream-pipes:
-    type: git
-    url: https://git.orionkindel.com/orion/purescript-node-stream-pipes
-    rev: f2f18c3c13ae2f0f5787ccfb3832fc8c653e83ad
+    type: registry
+    version: 1.2.3
+    integrity: sha256-lXD3x6+p72uBrRHGHrob2jrrBDakhhZE9O9EYE4aFiE=
     dependencies:
       - aff
       - arrays
@@ -618,6 +616,8 @@ packages:
       - either
       - exceptions
       - foldable-traversable
+      - foreign-object
+      - lists
       - maybe
       - mmorph
       - newtype
@@ -627,6 +627,7 @@ packages:
       - node-path
       - node-streams
       - node-zlib
+      - ordered-collections
       - parallel
       - pipes
       - prelude
@@ -634,6 +635,8 @@ packages:
       - strings
       - tailrec
       - transformers
+      - tuples
+      - unordered-collections
       - unsafe-coerce
   node-streams:
     type: registry
@@ -1038,6 +1041,21 @@ packages:
       - foldable-traversable
       - maybe
       - strings
+  unordered-collections:
+    type: registry
+    version: 3.1.0
+    integrity: sha256-H2eQR+ylI+cljz4XzWfEbdF7ee+pnw2IZCeq69AuJ+Q=
+    dependencies:
+      - arrays
+      - enums
+      - functions
+      - integers
+      - lists
+      - prelude
+      - record
+      - tuples
+      - typelevel-prelude
+      - unfoldable
   unsafe-coerce:
     type: registry
     version: 6.0.0
diff --git a/spago.yaml b/spago.yaml
index 1d214b4..8c76703 100644
--- a/spago.yaml
+++ b/spago.yaml
@@ -10,16 +10,13 @@ package:
     strict: true
     pedanticPackages: true
   dependencies:
-    - foldable-traversable
-    - node-stream-pipes
-    - st
-    - tuples
     - aff: ">=7.1.0 <8.0.0"
     - arrays: ">=7.3.0 <8.0.0"
     - bifunctors: ">=6.0.0 <7.0.0"
     - datetime: ">=6.1.0 <7.0.0"
     - effect: ">=4.0.0 <5.0.0"
     - exceptions: ">=6.0.0 <7.0.0"
+    - foldable-traversable: ">=6.0.0 <7.0.0"
     - foreign: ">=7.0.0 <8.0.0"
     - foreign-object: ">=4.1.0 <5.0.0"
     - integers: ">=6.0.0 <7.0.0"
@@ -28,6 +25,7 @@ package:
     - newtype: ">=5.0.0 <6.0.0"
     - node-buffer: ">=9.0.0 <10.0.0"
     - node-event-emitter: ">=3.0.0 <4.0.0"
+    - node-stream-pipes: ">=1.2.3 <2.0.0"
     - node-streams: ">=9.0.0 <10.0.0"
     - nullable: ">=6.0.0 <7.0.0"
     - numbers: ">=9.0.1 <10.0.0"
@@ -37,9 +35,11 @@ package:
     - prelude: ">=6.0.1 <7.0.0"
     - record: ">=4.0.0 <5.0.0"
     - record-extra: ">=5.0.1 <6.0.0"
+    - st: ">=6.2.0 <7.0.0"
     - strings: ">=6.0.1 <7.0.0"
     - tailrec: ">=6.1.0 <7.0.0"
     - transformers: ">=6.0.0 <7.0.0"
+    - tuples: ">=7.0.0 <8.0.0"
     - typelevel-prelude: ">=7.0.0 <8.0.0"
     - unsafe-coerce: ">=6.0.0 <7.0.0"
   test:
@@ -53,7 +53,4 @@ package:
       - simple-json
       - spec
 workspace:
-  extraPackages:
-    node-stream-pipes:
-      git: 'https://git.orionkindel.com/orion/purescript-node-stream-pipes'
-      ref: 'v1.0.5'
+  extraPackages: {}
diff --git a/src/Node.Stream.CSV.Stringify.purs b/src/Node.Stream.CSV.Stringify.purs
index 2a3a605..06a70fc 100644
--- a/src/Node.Stream.CSV.Stringify.purs
+++ b/src/Node.Stream.CSV.Stringify.purs
@@ -54,15 +54,15 @@ recordToForeign = unsafeCoerce
 -- | Requires an ordered array of column names.
 make
   :: forall @config @missing @extra
-  . Union config missing (Config extra)
+   . Union config missing (Config extra)
   => Array String
   -> { | config }
   -> Effect (CSVStringifier ())
 make columns =
   makeImpl
-  <<< unsafeToForeign
-  <<< Object.union (recordToForeign { columns, header: true })
-  <<< recordToForeign
+    <<< unsafeToForeign
+    <<< Object.union (recordToForeign { columns, header: true })
+    <<< recordToForeign
 
 -- | Convert the raw stream to a typed ObjectStream
 toObjectStream :: CSVStringifier () -> Object.Transform (Array String) String
diff --git a/src/Pipes.CSV.purs b/src/Pipes.CSV.purs
index 9ac747c..58555a7 100644
--- a/src/Pipes.CSV.purs
+++ b/src/Pipes.CSV.purs
@@ -2,9 +2,9 @@ module Pipes.CSV where
 
 import Prelude
 
-import Control.Monad.Error.Class (liftEither)
+import Control.Monad.Error.Class (class MonadThrow, liftEither)
 import Control.Monad.Except (runExcept)
-import Control.Monad.Rec.Class (forever)
+import Control.Monad.Rec.Class (class MonadRec, forever)
 import Control.Monad.ST.Global as ST
 import Control.Monad.ST.Ref as STRef
 import Data.Array as Array
@@ -14,9 +14,9 @@ import Data.FunctorWithIndex (mapWithIndex)
 import Data.Map as Map
 import Data.Maybe (Maybe(..))
 import Data.Tuple.Nested ((/\))
-import Effect.Aff (Aff)
+import Effect.Aff.Class (class MonadAff)
 import Effect.Class (liftEffect)
-import Effect.Exception (error)
+import Effect.Exception (Error, error)
 import Node.Buffer (Buffer)
 import Node.Stream.CSV.Parse as CSV.Parse
 import Node.Stream.CSV.Stringify as CSV.Stringify
@@ -47,10 +47,13 @@ import Type.Prelude (Proxy(..))
 -- | rows `shouldEqual` [{id: 1, foo: "hi", is_deleted: false}, {id: 2, foo: "bye", is_deleted: true}]
 -- | ```
 parse
-  :: forall @r rl
-  . RowToList r rl
+  :: forall @r rl m
+   . MonadAff m
+  => MonadThrow Error m
+  => MonadRec m
+  => RowToList r rl
   => ReadCSVRecord r rl
-  => Pipe (Maybe Buffer) (Maybe { | r }) Aff Unit
+  => Pipe (Maybe Buffer) (Maybe { | r }) m Unit
 parse = do
   raw <- liftEffect $ CSV.Parse.make {}
   colsST <- liftEffect $ ST.toEffect $ STRef.new Nothing
@@ -74,14 +77,14 @@ parse = do
 
 -- | Transforms buffer chunks of a CSV file to parsed
 -- | arrays of CSV values.
-parseRaw :: Pipe (Maybe Buffer) (Maybe (Array String)) Aff Unit
+parseRaw :: forall m. MonadAff m => MonadThrow Error m => Pipe (Maybe Buffer) (Maybe (Array String)) m Unit
 parseRaw = do
   s <- liftEffect $ CSV.Parse.toObjectStream <$> CSV.Parse.make {}
   Pipes.Stream.fromTransform s
 
 -- | Transforms CSV rows into stringified CSV records
 -- | using the given ordered array of column names.
-stringifyRaw :: Array String -> Pipe (Maybe (Array String)) (Maybe String) Aff Unit
+stringifyRaw :: forall m. MonadAff m => MonadThrow Error m => Array String -> Pipe (Maybe (Array String)) (Maybe String) m Unit
 stringifyRaw columns = do
   s <- liftEffect $ CSV.Stringify.toObjectStream <$> CSV.Stringify.make columns {}
   Pipes.Stream.fromTransform s
@@ -89,7 +92,7 @@ stringifyRaw columns = do
 -- | Transforms purescript records into stringified CSV records.
 -- |
 -- | Columns are inferred from the record's keys, ordered alphabetically.
-stringify :: forall r rl. WriteCSVRecord r rl => RowToList r rl => Keys rl => Pipe (Maybe { | r }) (Maybe String) Aff Unit
+stringify :: forall m r rl. MonadRec m => MonadAff m => MonadThrow Error m => WriteCSVRecord r rl => RowToList r rl => Keys rl => Pipe (Maybe { | r }) (Maybe String) m Unit
 stringify = do
   raw <- liftEffect $ CSV.Stringify.make (Array.fromFoldable $ keys $ Proxy @r) {}
   let
diff --git a/test/Test/Pipes.CSV.purs b/test/Test/Pipes.CSV.purs
index 13a71c1..cfaa152 100644
--- a/test/Test/Pipes.CSV.purs
+++ b/test/Test/Pipes.CSV.purs
@@ -14,7 +14,6 @@ import Effect.Class (liftEffect)
 import Node.Encoding (Encoding(..))
 import Partial.Unsafe (unsafePartial)
 import Pipes (yield, (>->))
-import Pipes (each) as Pipes
 import Pipes.CSV as Pipes.CSV
 import Pipes.Collect as Pipes.Collect
 import Pipes.Node.Buffer as Pipes.Buffer
@@ -26,7 +25,8 @@ import Test.Spec (Spec, describe, it)
 import Test.Spec.Assertions (shouldEqual)
 
 csv :: String
-csv = """created,flag,foo,id
+csv =
+  """created,flag,foo,id
 2020-01-01T00:00:00.0Z,true,a,1
 2024-02-02T08:00:00.0Z,false,apple,2
 1970-01-01T00:00:00.0Z,true,hello,3
@@ -41,31 +41,31 @@ spec =
     it "stringify" do
       let
         objs =
-          [ {id: 1, foo: "a", flag: true, created: dt "2020-01-01T00:00:00Z"}
-          , {id: 2, foo: "apple", flag: false, created: dt "2024-02-02T08:00:00Z"}
-          , {id: 3, foo: "hello", flag: true, created: dt "1970-01-01T00:00:00Z"}
+          [ { id: 1, foo: "a", flag: true, created: dt "2020-01-01T00:00:00Z" }
+          , { id: 2, foo: "apple", flag: false, created: dt "2024-02-02T08:00:00Z" }
+          , { id: 3, foo: "hello", flag: true, created: dt "1970-01-01T00:00:00Z" }
           ]
 
-      csv' <- map fold $ Pipes.Collect.collectArray $ Pipes.Stream.withEOS (Pipes.each objs) >-> Pipes.CSV.stringify >-> Pipes.Stream.unEOS
+      csv' <- map fold $ Pipes.Collect.toArray $ Pipes.Stream.withEOS (Pipes.Construct.eachArray objs) >-> Pipes.CSV.stringify >-> Pipes.Stream.unEOS
       csv' `shouldEqual` csv
     describe "parse" do
       it "parses csv" do
         rows <- map Array.fromFoldable
           $ Pipes.toListM
           $ Pipes.Stream.withEOS (yield csv)
-            >-> Pipes.Stream.inEOS (Pipes.Buffer.fromString UTF8)
-            >-> Pipes.CSV.parse
-            >-> Pipes.Stream.unEOS
+              >-> Pipes.Stream.inEOS (Pipes.Buffer.fromString UTF8)
+              >-> Pipes.CSV.parse
+              >-> Pipes.Stream.unEOS
 
         rows `shouldEqual`
-          [ {id: 1, foo: "a", flag: true, created: dt "2020-01-01T00:00:00Z"}
-          , {id: 2, foo: "apple", flag: false, created: dt "2024-02-02T08:00:00Z"}
-          , {id: 3, foo: "hello", flag: true, created: dt "1970-01-01T00:00:00Z"}
+          [ { id: 1, foo: "a", flag: true, created: dt "2020-01-01T00:00:00Z" }
+          , { id: 2, foo: "apple", flag: false, created: dt "2024-02-02T08:00:00Z" }
+          , { id: 3, foo: "hello", flag: true, created: dt "1970-01-01T00:00:00Z" }
           ]
       it "parses large csv" do
         nums <- liftEffect $ randomSample' 100000 (chooseInt 0 9)
         let
-          csvRows = ["id\n"] <> ((_ <> "\n") <$> show <$> nums)
+          csvRows = [ "id\n" ] <> ((_ <> "\n") <$> show <$> nums)
           csv' =
             let
               go ix
@@ -75,14 +75,14 @@ spec =
               tailRecM go 0
           in16kbChunks =
             Pipes.Util.chunked 16000
-            >-> Pipes.Stream.inEOS (Pipes.map fold)
-            >-> Pipes.Stream.inEOS (Pipes.Buffer.fromString UTF8)
+              >-> Pipes.Stream.inEOS (Pipes.map fold)
+              >-> Pipes.Stream.inEOS (Pipes.Buffer.fromString UTF8)
 
         rows <-
-          Pipes.Collect.collectArray
-          $ Pipes.Stream.withEOS csv'
-            >-> in16kbChunks
-            >-> Pipes.CSV.parse
-            >-> Pipes.Stream.unEOS
+          Pipes.Collect.toArray
+            $ Pipes.Stream.withEOS csv'
+                >-> in16kbChunks
+                >-> Pipes.CSV.parse
+                >-> Pipes.Stream.unEOS
 
-        rows `shouldEqual` ((\id -> {id}) <$> nums)
+        rows `shouldEqual` ((\id -> { id }) <$> nums)