A Workflow scheduler + runner

Workflow.purs

module Workflow where

import Prelude

import Control.Alt ((<|>))
import Control.Monad.Aff.Bus (BusR, BusW, make, read, split, write) as Bus
import Control.Monad.Rec.Class (forever)
import Data.Argonaut.Decode (class DecodeJson)
import Data.Argonaut.Encode.Class (class EncodeJson)
import Data.Newtype (wrap)
import Data.Traversable (traverse_)
import Data.Tuple (curry, uncurry)
import Data.Tuple.Nested ((/\), type (/\))
import Effect.Aff (Aff, Milliseconds, delay, finally, forkAff, invincible, killFiber)
import Effect.Aff.Class (class MonadAff)
import Effect.Exception (error)
import Unsafe.Coerce (unsafeCoerce)

--Scheduler is able to listen to the redis queue and run jobs as well as write jobs to the redis queue

-- TODO : Retries

data PollOptions

type PollInterval = Milliseconds

class Job j <= JobStorageAdapter j a | a -> j, j -> a where
  push :: ∀ m. MonadAff m => a -> j -> Milliseconds -> m Unit
  pull :: a -> PollOptions -> Aff (Array j)
  remove :: a -> j -> Aff Unit

class (DecodeJson a, EncodeJson a) <= Job a where
  run :: ∀ m. MonadAff m => a -> m Unit

scheduler :: ∀ j a . JobStorageAdapter j a => Bus.BusR (j /\ Milliseconds) -> PollInterval -> PollOptions -> a -> _
scheduler bus ms opts jsa = do
  f1 <- forkAff $ Bus.read bus >>= (uncurry $ push jsa)
  map (f1 /\ _) $
    forkAff $
      invincible $
        forever $
          delay ms
            *> pull jsa opts
            >>= traverse_ (\j -> forkAff $ run j `finally` remove jsa j)

-- Can partially apply the `bus` argument to obtain a schedule function which takes the job and time to schedule
schedule_ :: ∀ j a. JobStorageAdapter j a => Bus.BusW (j /\ Milliseconds) -> j -> Milliseconds -> _
schedule_ = curry <<< flip Bus.write

main jsa = do
  (readBus /\ writeBus) <- Bus.split <$> Bus.make
  (f1 /\ f2) <- scheduler readBus (wrap 5000.0) (unsafeCoerce unit) jsa
  blockTillShouldBeKilled <|>
    ( killFiber (error "Die now") f1
    *> killFiber (error "Die now") f2 )
  where
  blockTillShouldBeKilled = forever $ delay (wrap 5000.0) -- *> check some avar and throw if set