feat: deliver Haskell dispatch verdicts

bindings/haskell/event-sorcery.cabal

@@ -54,6 +54,7 @@ library
     EventSorcery.Aggregate
     EventSorcery.Dispatch
     EventSorcery.Dispatch.TestSupport
+    EventSorcery.Dispatch.Worker
     EventSorcery.Engine
     EventSorcery.Engine.Internal.FFI
     EventSorcery.Job
@@ -68,6 +69,8 @@ library
     EventSorcery.Engine.Acquisition
     EventSorcery.Engine.Internal
     EventSorcery.Job.Definition
+    EventSorcery.Job.Worker.Definition
+    EventSorcery.Job.Worker.Internal
 
   extra-libraries: event_sorcery_ffi
   build-depends:
@@ -177,3 +180,14 @@ test-suite job-worker-spec
     , bytestring     >=0.11 && <0.13
     , event-sorcery
     , text           >=2.0  && <2.2
+
+test-suite dispatch-worker-spec
+  import:         strict
+  type:           exitcode-stdio-1.0
+  hs-source-dirs: test
+  main-is:        DispatchWorkerSpec.hs
+  build-depends:
+    , base           >=4.18 && <5
+    , bytestring     >=0.11 && <0.13
+    , event-sorcery
+    , text           >=2.0  && <2.2

bindings/haskell/src/EventSorcery/Aggregate.hs

@@ -1,3 +1,5 @@
+{-# LANGUAGE FunctionalDependencies #-}
+
 module EventSorcery.Aggregate (
   DecodeCause (..),
   DispatchIntent,
@@ -8,6 +10,7 @@ module EventSorcery.Aggregate (
   Member,
   SchemaVersion (..),
   dispatchIntent,
+  dispatchIntentJob,
   dispatchJobId,
 ) where
 
@@ -18,11 +21,10 @@ import Data.Proxy (Proxy)
 import Data.Text (Text)
 import Data.Type.Equality (type (~))
 import Data.Word (Word16)
-import EventSorcery.Dispatch (JobDispatch)
+import EventSorcery.Dispatch (DispatchOutcome, JobDispatch)
 import EventSorcery.Job.Definition (
   Job (..),
   JobId,
-  jobType,
  )
 import Prelude (Bool (..), Either, Eq, Ord, Show)
 
@@ -39,11 +41,12 @@ newtype DecodeCause = DecodeCause Text
   deriving stock (Eq, Show)
 
 
-data DispatchIntent job = DispatchIntent JobId Text ByteString
+data DispatchIntent job = DispatchIntent JobId job
 
 
-class Dispatches entity job where
+class Dispatches entity job | job -> entity where
   injectDispatchIntent :: DispatchIntent job -> Event entity
+  injectDispatchOutcome :: DispatchOutcome job -> Command entity
 
 
 type Member item items = Elem item items ~ 'True
@@ -57,6 +60,7 @@ type family Elem (item :: Type) (items :: [Type]) :: Bool where
 
 data Effect entity where
   Events :: NonEmpty (Event entity) -> Effect entity
+  Unchanged :: Effect entity
   Dispatch
     :: (Job job, Member job (Jobs entity), Dispatches entity job)
     => JobDispatch job
@@ -90,10 +94,13 @@ class EventSourced entity where
     -> Either (CommandError entity) (Effect entity)
 
 
-dispatchIntent :: forall job. Job job => JobId -> job -> DispatchIntent job
-dispatchIntent identifier job =
-  DispatchIntent identifier (jobType @job) (encodeJob job)
+dispatchIntent :: JobId -> job -> DispatchIntent job
+dispatchIntent = DispatchIntent
 
 
 dispatchJobId :: DispatchIntent job -> JobId
-dispatchJobId (DispatchIntent identifier _ _) = identifier
+dispatchJobId (DispatchIntent identifier _) = identifier
+
+
+dispatchIntentJob :: DispatchIntent job -> job
+dispatchIntentJob (DispatchIntent _ job) = job

bindings/haskell/src/EventSorcery/Dispatch/Worker.hs

@@ -0,0 +1,232 @@
+module EventSorcery.Dispatch.Worker (
+  DeliveryPolicy,
+  DispatchWorker,
+  OriginDeliveryError (..),
+  OriginJob (..),
+  OriginPort,
+  deliveryPolicy,
+  dispatchWorker,
+  originPort,
+  runDispatchJobOnce,
+  storeOriginPort,
+) where
+
+import Data.Word (Word32)
+import EventSorcery.Aggregate (
+  Dispatches (injectDispatchOutcome),
+  EventSourced,
+ )
+import EventSorcery.Dispatch (DispatchFailure (..), DispatchOutcome)
+import EventSorcery.Dispatch.Internal qualified as Dispatch
+import EventSorcery.Job (
+  DeadReason (RetriesExhausted),
+  Job (JobError, JobOutput),
+  JobId,
+  JobInstant,
+ )
+import EventSorcery.Job.Execution (
+  DurableJob (renderJobError),
+  JobAttempt (JobAttempt),
+ )
+import EventSorcery.Job.Worker (
+  JobRunError,
+  JobRunResult,
+  JobWorker,
+ )
+import EventSorcery.Job.Worker.Internal (
+  SettlementStrategy (..),
+  VerdictDelivery (..),
+  runJobOnceWith,
+ )
+import EventSorcery.Store (
+  Store,
+  StoreError (..),
+  executeCommand,
+ )
+import EventSorcery.Stream (StreamKey)
+import Prelude (Either (..), IO, pure)
+
+
+class
+  ( DurableJob job
+  , EventSourced (Origin job)
+  , Dispatches (Origin job) job
+  ) =>
+  OriginJob job
+  where
+  type Origin job
+  originKey :: job -> StreamKey (Origin job)
+
+
+data OriginDeliveryError failure
+  = OriginUnavailable failure
+  | OriginRefused failure
+
+
+newtype OriginPort job failure
+  = OriginPort
+      (job -> DispatchOutcome job -> IO (Either (OriginDeliveryError failure) ()))
+
+
+data DeliveryPolicy failure
+  = DeliveryPolicy
+      (JobInstant -> failure -> JobInstant)
+      (JobInstant -> failure -> JobInstant)
+      (JobId -> failure -> IO ())
+
+
+data DispatchWorker job failure
+  = DispatchWorker
+      (JobWorker job)
+      (OriginPort job failure)
+      (DeliveryPolicy failure)
+
+
+originPort
+  :: (job -> DispatchOutcome job -> IO (Either (OriginDeliveryError failure) ()))
+  -> OriginPort job failure
+originPort = OriginPort
+
+
+deliveryPolicy
+  :: (JobInstant -> failure -> JobInstant)
+  -> (JobInstant -> failure -> JobInstant)
+  -> (JobId -> failure -> IO ())
+  -> DeliveryPolicy failure
+deliveryPolicy = DeliveryPolicy
+
+
+dispatchWorker
+  :: JobWorker job
+  -> OriginPort job failure
+  -> DeliveryPolicy failure
+  -> DispatchWorker job failure
+dispatchWorker = DispatchWorker
+
+
+storeOriginPort
+  :: OriginJob job
+  => Store (Origin job)
+  -> OriginPort job (StoreError (Origin job))
+storeOriginPort store = OriginPort \job outcome -> do
+  delivered <-
+    executeCommand
+      store
+      (originKey job)
+      (injectDispatchOutcome outcome)
+
+  pure case delivered of
+    Left failure@(StoreCommandRejected _) -> Left (OriginRefused failure)
+    Left failure@(StoreDecisionRejected _) -> Left (OriginRefused failure)
+    Left failure -> Left (OriginUnavailable failure)
+    Right _ -> Right ()
+
+
+runDispatchJobOnce
+  :: forall job failure
+   . OriginJob job
+  => DispatchWorker job failure
+  -> JobId
+  -> JobInstant
+  -> IO (Either JobRunError (JobRunResult (JobOutput job) (JobError job)))
+runDispatchJobOnce (DispatchWorker worker port policy) identifier now =
+  runJobOnceWith (dispatchSettlement port policy now) worker identifier now
+
+
+dispatchSettlement
+  :: OriginJob job
+  => OriginPort job failure
+  -> DeliveryPolicy failure
+  -> JobInstant
+  -> SettlementStrategy job
+dispatchSettlement port policy now =
+  SettlementStrategy
+    { beforeSuccess = \identifier job attempt output ->
+        deliverVerdict
+          policy
+          now
+          port
+          job
+          (Dispatch.confirmedOutcome identifier output (attemptCount attempt))
+    , beforeRejection = \identifier job attempt failure ->
+        deliverVerdict
+          policy
+          now
+          port
+          job
+          ( Dispatch.failedOutcome
+              identifier
+              (Rejected failure)
+              (attemptCount attempt)
+          )
+    , beforeExhaustion = \identifier job attempt failure -> do
+        delivered <-
+          deliver
+            port
+            job
+            ( Dispatch.failedOutcome
+                identifier
+                ( DeadLettered
+                    RetriesExhausted
+                    (renderJobError job failure)
+                )
+                (attemptCount attempt)
+            )
+
+        case delivered of
+          Left (OriginUnavailable deliveryFailure) ->
+            report policy identifier deliveryFailure
+          Left (OriginRefused deliveryFailure) ->
+            report policy identifier deliveryFailure
+          Right () -> pure ()
+    }
+
+
+deliverVerdict
+  :: DeliveryPolicy failure
+  -> JobInstant
+  -> OriginPort job failure
+  -> job
+  -> DispatchOutcome job
+  -> IO VerdictDelivery
+deliverVerdict policy now port job outcome = do
+  delivered <- deliver port job outcome
+
+  pure case delivered of
+    Left (OriginUnavailable failure) ->
+      VerdictDeferred (retryUnavailable policy now failure)
+    Left (OriginRefused failure) ->
+      VerdictDeferred (retryRefused policy now failure)
+    Right () -> VerdictAccepted
+
+
+deliver
+  :: OriginPort job failure
+  -> job
+  -> DispatchOutcome job
+  -> IO (Either (OriginDeliveryError failure) ())
+deliver (OriginPort send) = send
+
+
+retryUnavailable
+  :: DeliveryPolicy failure
+  -> JobInstant
+  -> failure
+  -> JobInstant
+retryUnavailable (DeliveryPolicy schedule _ _) = schedule
+
+
+retryRefused
+  :: DeliveryPolicy failure
+  -> JobInstant
+  -> failure
+  -> JobInstant
+retryRefused (DeliveryPolicy _ schedule _) = schedule
+
+
+report :: DeliveryPolicy failure -> JobId -> failure -> IO ()
+report (DeliveryPolicy _ _ send) = send
+
+
+attemptCount :: JobAttempt -> Word32
+attemptCount (JobAttempt attempt) = attempt