feat: init

spago.yaml

@@ -3,20 +3,26 @@ package:
     strict: true
     pedantic_packages: true
   dependencies:
-    - prelude
     - aff
+    - aff-promise
+    - bifunctors
     - effect
     - either
-    - maybe
+    - exceptions
     - foldable-traversable
-    - console
+    - foreign
-    - newtype
+    - foreign-object
-    - strings
+    - js-bigints
-    - stringutils
+    - maybe
+    - node-event-emitter
+    - node-streams
+    - nullable
+    - ordered-collections
+    - prelude
+    - record
     - transformers
-    - tuples
     - typelevel-prelude
-  name: project
+  name: worker
 workspace:
   extra_packages: {}
   package_set:

src/Main.purs

@@ -1,7 +0,0 @@
-module Main where
-
-import Prelude
-import Effect (Effect)
-
-main :: Effect Unit
-main = pure unit

src/Node.Worker.Environment.js

@@ -0,0 +1,8 @@
+import * as worker from 'node:worker_threads'
+
+/** @type {(k: worker.Serializable) => () => unknown} */
+export const getEnvironmentDataImpl = k => () => worker.getEnvironmentData(k)
+
+/** @type {(k: worker.Serializable) => (v: worker.Serializable) => () => void} */
+export const setEnvironmentDataImpl = k => v => () =>
+  worker.setEnvironmentData(k, v)

src/Node.Worker.Environment.purs

@@ -0,0 +1,26 @@
+module Node.Worker.Environment where
+
+import Prelude
+
+import Control.Monad.Error.Class (liftEither)
+import Data.Bifunctor (lmap)
+import Data.Maybe (Maybe)
+import Effect (Effect)
+import Effect.Exception (error)
+import Foreign (Foreign)
+import Node.Worker.Serializable (class Serializable, deserialize, serialize, undef)
+
+foreign import getEnvironmentDataImpl :: Foreign -> Effect Foreign
+foreign import setEnvironmentDataImpl :: Foreign -> Foreign -> Effect Unit
+
+lookup :: forall k @v. Serializable k => Serializable v => k -> Effect (Maybe v)
+lookup =
+  flip bind (liftEither <<< lmap error <<< deserialize)
+    <<< getEnvironmentDataImpl
+    <<< serialize
+
+unset :: forall k. Serializable k => k -> Effect Unit
+unset k = setEnvironmentDataImpl (serialize k) undef
+
+set :: forall k @v. Serializable k => Serializable v => k -> v -> Effect Unit
+set k v = setEnvironmentDataImpl (serialize k) (serialize v)

src/Node.Worker.Parent.js

@@ -0,0 +1,60 @@
+import * as stream from 'node:stream'
+import * as worker from 'node:worker_threads'
+
+/**
+ * @template O
+ * @typedef {(
+ *   O extends Record<string, unknown>
+ *     ? {[K in keyof O]: O[K] | null}
+ *     : O
+ * )} PartialNull
+ */
+
+/**
+ * @template O
+ * @typedef {(
+ *   O extends Record<string, unknown>
+ *     ? {[K in keyof O]: O[K] extends infer T | null ? T : O[K]}
+ *     : O
+ * )} OmitNulls
+ */
+
+/** @type {<O>(o: O) => OmitNulls<O>} */
+const omitNulls = o =>
+  // @ts-ignore
+  Array.from(Object.entries(o)).reduce((o, [k, v]) => {
+    // @ts-ignore
+    if (k !== null) o[k] = v
+    return o
+  }, {})
+
+/** @type {(script: string) => (opts: PartialNull<worker.WorkerOptions>) => () => worker.Worker} */
+export const spawnImpl = script => opts => () =>
+  new worker.Worker(script, omitNulls(opts))
+
+/** @type {(w: worker.Worker) => (msg: worker.Serializable) => () => void} */
+export const workerPostMessageImpl = w => a => () => w.postMessage(a)
+
+/** @type {(w: worker.Worker) => stream.Readable} */
+export const stdout = w => w.stdout
+
+/** @type {(w: worker.Worker) => stream.Readable} */
+export const stderr = w => w.stderr
+
+/** @type {(w: worker.Worker) => stream.Writable | null} */
+export const stdinImpl = w => w.stdin
+
+/** @type {(w: worker.Worker) => () => Promise<number>} */
+export const terminateImpl = w => () => w.terminate()
+
+/** @type {(w: worker.Worker) => worker.ResourceLimits | undefined} */
+export const resourceLimitsImpl = w => w.resourceLimits
+
+/** @type {(w: worker.Worker) => number} */
+export const threadId = w => w.threadId
+
+/** @type {(w: worker.Worker) => () => void } */
+export const ref = w => () => w.ref()
+
+/** @type {(w: worker.Worker) => () => void } */
+export const unref = w => () => w.unref()

src/Node.Worker.Parent.purs

@@ -0,0 +1,179 @@
+module Node.Worker.Parent
+  ( Worker
+  , stdout
+  , stderr
+  , stdin
+  , exitH
+  , errorH
+  , terminate
+  , threadId
+  , ref
+  , unref
+  , resourceLimits
+  , messageerrorH
+  , onlineH
+  , SpawnOptions
+  , spawnFile
+  , spawnFile'
+  , spawnScript
+  , spawnScript'
+  ) where
+
+import Prelude
+
+import Control.Promise (Promise)
+import Control.Promise as Promise
+import Data.Either (hush)
+import Data.Map (Map)
+import Data.Maybe (Maybe(..))
+import Data.Nullable (Nullable)
+import Data.Nullable as Nullable
+import Effect (Effect)
+import Effect.Aff (Aff)
+import Effect.Exception (Error)
+import Effect.Uncurried (mkEffectFn1)
+import Foreign (Foreign, unsafeToForeign)
+import Foreign.Object (Object)
+import Foreign.Object as Object
+import Node.EventEmitter (EventHandle(..))
+import Node.EventEmitter.UtilTypes (EventHandle1, EventHandle0)
+import Node.Stream (Readable, Writable)
+import Node.Worker (ResourceLimits)
+import Node.Worker.Port (class PortLike)
+import Node.Worker.Serializable (class Serializable, deserialize, serialize)
+import Prim.Row (class Nub, class Union)
+import Record as Record
+import Record.Apply (applyRecord)
+import Type.Prelude (Proxy(..))
+
+foreign import data Worker :: Type
+
+foreign import spawnImpl :: String -> Record SpawnImplOptions -> Effect Worker
+foreign import workerPostMessageImpl :: Worker -> Foreign -> Effect Unit
+foreign import stdinImpl :: Worker -> Nullable (Writable ())
+foreign import terminateImpl :: Worker -> Effect (Promise Int)
+foreign import resourceLimitsImpl :: Worker -> Foreign
+
+foreign import stderr :: Worker -> Readable ()
+foreign import stdout :: Worker -> Readable ()
+foreign import threadId :: Worker -> Int
+foreign import ref :: Worker -> Effect Unit
+foreign import unref :: Worker -> Effect Unit
+
+instance PortLike Worker where
+  postMessage w a = workerPostMessageImpl w (serialize a)
+  messageH = EventHandle "message" $ \psCb -> mkEffectFn1 (psCb <<< deserialize)
+
+type SpawnImplOptions =
+  ( argv :: Nullable (Array Foreign)
+  , env :: Nullable (Object String)
+  , eval :: Nullable (Boolean)
+  , execArgv :: Nullable (Array String)
+  , stdin :: Nullable (Boolean)
+  , stdout :: Nullable (Boolean)
+  , stderr :: Nullable (Boolean)
+  , workerData :: Nullable (Foreign)
+  , trackUnmanagedFds :: Nullable (Boolean)
+  , transferList :: Nullable (Array Foreign)
+  , resourceLimits :: Nullable (ResourceLimits)
+  , name :: Nullable (String)
+  )
+
+type SpawnOptions r =
+  ( argv :: Maybe (Array String)
+  , env :: Maybe (Map String String)
+  , execArgv :: Maybe (Array String)
+  , stdin :: Maybe (Boolean)
+  , stdout :: Maybe (Boolean)
+  , stderr :: Maybe (Boolean)
+  , trackUnmanagedFds :: Maybe (Boolean)
+  , transferList :: Maybe (Array Foreign)
+  , resourceLimits :: Maybe (ResourceLimits)
+  , name :: Maybe (String)
+  | r
+  )
+
+emptySpawnOptions :: Record (SpawnOptions ())
+emptySpawnOptions =
+  { argv: Nothing
+  , env: Nothing
+  , execArgv: Nothing
+  , stdin: Nothing
+  , stdout: Nothing
+  , stderr: Nothing
+  , trackUnmanagedFds: Nothing
+  , transferList: Nothing
+  , resourceLimits: Nothing
+  , name: Nothing
+  }
+
+spawnOptionsToImpl :: forall a. Serializable a => Record (SpawnOptions (workerData :: Maybe a, eval :: Boolean)) -> Record SpawnImplOptions
+spawnOptionsToImpl = applyRecord
+  { argv: Nullable.toNullable <<< map (map unsafeToForeign)
+  , env: Nullable.toNullable <<< map Object.fromFoldableWithIndex
+  , eval: Nullable.notNull
+  , execArgv: Nullable.toNullable
+  , stdin: Nullable.toNullable
+  , stdout: Nullable.toNullable
+  , stderr: Nullable.toNullable
+  , trackUnmanagedFds: Nullable.toNullable
+  , transferList: Nullable.toNullable
+  , resourceLimits: Nullable.toNullable
+  , name: Nullable.toNullable
+  , workerData: Nullable.toNullable <<< map serialize
+  }
+
+spawnFile :: forall a r optsdup. Serializable a => Union r (SpawnOptions ()) optsdup => Nub optsdup (SpawnOptions ()) => Record r -> String -> a -> Effect Worker
+spawnFile opts' file d =
+  let
+    opts :: Record (SpawnOptions ())
+    opts = Record.merge opts' emptySpawnOptions
+
+    implOpts :: Record SpawnImplOptions
+    implOpts =
+      spawnOptionsToImpl
+        $ Record.insert (Proxy @"workerData") (Just d)
+        $ Record.insert (Proxy @"eval") false opts
+  in
+    spawnImpl file implOpts
+
+spawnFile' :: forall r optsdup. Union r (SpawnOptions ()) optsdup => Nub optsdup (SpawnOptions ()) => Record r -> String -> Effect Worker
+spawnFile' opts' file = spawnFile opts' file (Nothing :: Maybe String)
+
+spawnScript :: forall a r optsdup. Serializable a => Union r (SpawnOptions ()) optsdup => Nub optsdup (SpawnOptions ()) => Record r -> String -> a -> Effect Worker
+spawnScript opts' script d =
+  let
+    opts :: Record (SpawnOptions ())
+    opts = Record.merge opts' emptySpawnOptions
+
+    implOpts :: Record SpawnImplOptions
+    implOpts =
+      spawnOptionsToImpl
+        $ Record.insert (Proxy @"workerData") (Just d)
+        $ Record.insert (Proxy @"eval") true opts
+  in
+    spawnImpl script implOpts
+
+spawnScript' :: forall r optsdup. Union r (SpawnOptions ()) optsdup => Nub optsdup (SpawnOptions ()) => Record r -> String -> Effect Worker
+spawnScript' opts' script = spawnScript opts' script (Nothing :: Maybe String)
+
+exitH :: EventHandle1 Worker Int
+exitH = EventHandle "exit" mkEffectFn1
+
+errorH :: EventHandle1 Worker Error
+errorH = EventHandle "error" mkEffectFn1
+
+messageerrorH :: EventHandle1 Worker Error
+messageerrorH = EventHandle "messageerror" mkEffectFn1
+
+onlineH :: EventHandle0 Worker
+onlineH = EventHandle "online" (const $ pure unit)
+
+stdin :: Worker -> Maybe (Writable ())
+stdin = Nullable.toMaybe <<< stdinImpl
+
+terminate :: Worker -> Aff Int
+terminate = Promise.toAffE <<< terminateImpl
+
+resourceLimits :: Worker -> Maybe ResourceLimits
+resourceLimits = join <<< hush <<< deserialize <<< resourceLimitsImpl

src/Node.Worker.Port.js

@@ -0,0 +1,16 @@
+import * as worker from 'node:worker_threads'
+
+/** @type {(p: worker.MessagePort) => (a: worker.Serializable) => () => void} */
+export const postMessageImpl = p => a => () => p.postMessage(a)
+
+/** @type {(p: worker.MessagePort) => () => void} */
+export const ref = p => () => p.ref()
+
+/** @type {(p: worker.MessagePort) => () => void} */
+export const unref = p => () => p.unref()
+
+/** @type {(p: worker.MessagePort) => () => void} */
+export const close = p => () => p.close()
+
+/** @type {() => worker.MessagePort} */
+export const newImpl = () => new worker.MessagePort()

src/Node.Worker.Port.purs

@@ -0,0 +1,47 @@
+module Node.Worker.Port
+  ( MessagePort
+  , ref
+  , unref
+  , new
+  , close
+  , closeH
+  , messageerrorH
+  , class PortLike
+  , postMessage
+  , messageH
+  ) where
+
+import Prelude
+
+import Data.Either (Either)
+import Effect (Effect)
+import Effect.Exception (Error)
+import Effect.Uncurried (EffectFn1, mkEffectFn1)
+import Foreign (Foreign)
+import Node.EventEmitter (EventHandle(..))
+import Node.EventEmitter.UtilTypes (EventHandle1, EventHandle0)
+import Node.Worker.Serializable (class Serializable, deserialize, serialize)
+
+class PortLike p where
+  postMessage :: forall a. Serializable a => p -> a -> Effect Unit
+  messageH :: forall a. Serializable a => EventHandle p (Either String a -> Effect Unit) (EffectFn1 Foreign Unit)
+
+instance PortLike MessagePort where
+  postMessage p a = postMessageImpl p (serialize a)
+  messageH = EventHandle "message" $ \psCb -> mkEffectFn1 (psCb <<< deserialize)
+
+foreign import data MessagePort :: Type
+foreign import newImpl :: Effect MessagePort
+foreign import postMessageImpl :: MessagePort -> Foreign -> Effect Unit
+foreign import close :: MessagePort -> Effect Unit
+foreign import ref :: MessagePort -> Effect Unit
+foreign import unref :: MessagePort -> Effect Unit
+
+new :: Effect MessagePort
+new = newImpl
+
+closeH :: EventHandle0 MessagePort
+closeH = EventHandle "close" (const $ pure unit)
+
+messageerrorH :: EventHandle1 MessagePort Error
+messageerrorH = EventHandle "messageerror" mkEffectFn1

src/Node.Worker.Serializable.js

@@ -0,0 +1,2 @@
+/** @type {undefined} */
+export const undef = undefined

src/Node.Worker.Serializable.purs

@@ -0,0 +1,123 @@
+module Node.Worker.Serializable where
+
+import Prelude
+
+import Control.Monad.Error.Class (throwError)
+import Control.Monad.Except (runExcept)
+import Data.Bifunctor (lmap)
+import Data.Either (Either(..))
+import Data.Maybe (Maybe(..))
+import Data.Symbol (class IsSymbol, reflectSymbol)
+import Data.Traversable (traverse)
+import Foreign (Foreign, readArray, readBoolean, readInt, readNullOrUndefined, readNumber, readString, typeOf, unsafeFromForeign, unsafeReadTagged, unsafeToForeign)
+import Foreign.Index (readProp)
+import Foreign.Object (Object)
+import JS.BigInt (BigInt)
+import Prim.Row as Row
+import Prim.RowList (class RowToList, Cons, Nil, RowList)
+import Record (get)
+import Record.Builder (Builder) as Record
+import Record.Builder as Record.Builder
+import Type.Prelude (Proxy(..))
+
+foreign import undef :: Foreign
+
+class Serializable a where
+  serialize :: a -> Foreign
+  deserialize :: Foreign -> Either String a
+
+instance Serializable Foreign where
+  serialize = identity
+  deserialize = pure
+
+instance Serializable String where
+  serialize = unsafeToForeign
+  deserialize = lmap show <<< runExcept <<< readString
+
+instance Serializable Int where
+  serialize = unsafeToForeign
+  deserialize = lmap show <<< runExcept <<< readInt
+
+instance Serializable Number where
+  serialize = unsafeToForeign
+  deserialize = lmap show <<< runExcept <<< readNumber
+
+instance Serializable BigInt where
+  serialize = unsafeToForeign
+  deserialize = lmap show <<< runExcept <<< unsafeReadTagged "BigInt"
+
+instance Serializable Boolean where
+  serialize = unsafeToForeign
+  deserialize = lmap show <<< runExcept <<< readBoolean
+
+instance Serializable a => Serializable (Maybe a) where
+  serialize (Just a) = unsafeToForeign a
+  serialize Nothing = undef
+  deserialize = flip bind (traverse deserialize) <<< lmap show <<< runExcept <<< readNullOrUndefined
+
+instance Serializable a => Serializable (Array a) where
+  serialize a = unsafeToForeign a
+  deserialize = flip bind (traverse deserialize) <<< lmap show <<< runExcept <<< readArray
+
+instance Serializable a => Serializable (Object a) where
+  serialize a = unsafeToForeign a
+  deserialize a =
+    if typeOf a == "object" then
+      pure $ unsafeFromForeign a
+    else
+      throwError $ "expected object, found " <> typeOf a
+
+instance (RowToList r rl, SerializeRecord rl r () to, DeserializeRecord rl () r) => Serializable (Record r) where
+  serialize a = unsafeToForeign $ serializeFields (Proxy @rl) a
+  deserialize a = map (flip Record.Builder.build {}) (deserializeFields (Proxy @rl) a)
+
+class DeserializeRecord (rl :: RowList Type) (from :: Row Type) (to :: Row Type) | rl -> from to where
+  deserializeFields :: Proxy rl -> Foreign -> Either String (Record.Builder (Record from) (Record to))
+
+instance
+  ( IsSymbol name
+  , Serializable ty
+  , DeserializeRecord tail from from'
+  , Row.Lacks name from'
+  , Row.Cons name ty from' to
+  ) =>
+  DeserializeRecord (Cons name ty tail) from to where
+  deserializeFields _ obj =
+    let
+      first = do
+        value <- deserialize =<< (lmap show $ runExcept $ readProp name obj)
+        pure $ Record.Builder.insert (Proxy @name) value
+      rest = deserializeFields (Proxy @tail) obj
+      name = reflectSymbol (Proxy @name)
+      combine (Left e) (Right _) = Left e
+      combine (Left e1) (Left e2) = Left (e1 <> e2)
+      combine (Right _) (Left e) = Left e
+      combine (Right fun) (Right a) = Right (fun <<< a)
+    in
+      combine first rest
+
+instance DeserializeRecord Nil () () where
+  deserializeFields _ _ = pure identity
+
+class SerializeRecord (rl :: RowList Type) row (from :: Row Type) (to :: Row Type) | rl -> row from to where
+  serializeFields :: Proxy rl -> Record row -> Record.Builder (Record from) (Record to)
+
+instance
+  ( IsSymbol name
+  , Serializable ty
+  , SerializeRecord tail row from from'
+  , Row.Cons name ty whatever row
+  , Row.Lacks name from'
+  , Row.Cons name Foreign from' to
+  ) =>
+  SerializeRecord (Cons name ty tail) row from to where
+  serializeFields _ rec = result
+    where
+    namep = Proxy :: Proxy name
+    value = serialize $ get namep rec
+    tailp = Proxy :: Proxy tail
+    rest = serializeFields tailp rec
+    result = Record.Builder.insert namep value <<< rest
+
+instance SerializeRecord Nil row () () where
+  serializeFields _ _ = identity

src/Node.Worker.js

@@ -0,0 +1,16 @@
+import * as worker from 'node:worker_threads'
+
+/** @type {symbol} */
+export const shareEnv = worker.SHARE_ENV
+
+/** @type {() => boolean} */
+export const isMainThread = () => worker.isMainThread
+
+/** @type {() => number} */
+export const threadId = () => worker.threadId
+
+/** @type {() => object} */
+export const resourceLimitsImpl = () => worker.resourceLimits
+
+/** @type {(a: object) => () => void} */
+export const markUntransferableImpl = a => () => worker.markAsUntransferable(a)

src/Node.Worker.purs

@@ -0,0 +1,30 @@
+module Node.Worker (ResourceLimits, resourceLimits, isMainThread, threadId, markUntransferable) where
+
+import Prelude
+
+import Data.Either (hush)
+import Data.Maybe (Maybe)
+import Effect (Effect)
+import Foreign (Foreign)
+import Node.Worker.Serializable (class Serializable, deserialize, serialize)
+
+foreign import shareEnv :: Symbol
+foreign import isMainThread :: Effect Boolean
+foreign import threadId :: Effect Int
+
+foreign import markUntransferableImpl :: Foreign -> Effect Unit
+
+markUntransferable :: forall a. Serializable a => a -> Effect Unit
+markUntransferable = markUntransferableImpl <<< serialize
+
+type ResourceLimits =
+  { maxYoungGenerationSizeMb :: Number
+  , maxOldGenerationSizeMb :: Number
+  , codeRangeSizeMb :: Number
+  , stackSizeMb :: Number
+  }
+
+foreign import resourceLimitsImpl :: Effect Foreign
+
+resourceLimits :: Effect (Maybe ResourceLimits)
+resourceLimits = map (join <<< hush <<< deserialize) $ resourceLimitsImpl

src/Record.Apply.purs

@@ -0,0 +1,35 @@
+module Record.Apply where
+
+import Prelude
+
+import Data.Symbol (class IsSymbol)
+import Prim.Row as Row
+import Prim.RowList (class RowToList, Cons, Nil, RowList)
+import Record as Record
+import Type.Prelude (Proxy(..))
+
+applyRecord :: forall rl src to ap. RowToList src rl => ApplyRecord rl src () to ap => Record ap -> Record src -> Record to
+applyRecord ap src = applyFields (Proxy @rl) ap src {}
+
+class ApplyRecord :: RowList Type -> Row Type -> Row Type -> Row Type -> Row Type -> Constraint
+class ApplyRecord rl src from to ap | rl -> src from to ap where
+  applyFields :: Proxy rl -> Record ap -> Record src -> Record from -> Record to
+
+instance
+  ( IsSymbol k
+  , ApplyRecord tailrl src tail from ap
+  , Row.Cons k (a -> b) aptail ap
+  , Row.Cons k a srctail src
+  , Row.Lacks k from
+  , Row.Cons k b from to
+  ) =>
+  ApplyRecord (Cons k a tailrl) src tail to ap where
+  applyFields _ ap src tail =
+    let
+      f = Record.get (Proxy @k) ap
+      a = Record.get (Proxy @k) src
+    in
+      Record.insert (Proxy @k) (f a) $ applyFields (Proxy @tailrl) ap src tail
+
+instance ApplyRecord Nil src () () ap where
+  applyFields _ _ _ = identity