{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE UndecidableInstances #-}

-- | Top-level module to run a single Hydra node.
--
-- Checkout [Hydra
-- Documentation](https://hydra.family/head-protocol/docs/dev/architecture)
-- for some details about the overall architecture of the `Node`.
module Hydra.Node where

import Hydra.Prelude

import Conduit (MonadUnliftIO, ZipSink (..), foldMapC, foldlC, mapC, runConduitRes, sinkList, (.|))
import Control.Concurrent.Class.MonadSTM (
  stateTVar,
  writeTVar,
 )
import Control.Monad.Trans.Writer (execWriter, tell)
import Data.Text (pack)
import Hydra.API.ClientInput (ClientInput)
import Hydra.API.Server (Server, sendMessage)
import Hydra.Cardano.Api (
  getCardanoPaymentVerificationKey,
 )
import Hydra.Chain (Chain (..), ChainEvent (..), ChainStateHistory (lastKnown), PostTxError, initHistory)
import Hydra.Chain.ChainState (IsChainState (..))
import Hydra.Events (EventId, EventSink (..), EventSource (..), getEventId, putEventsToSinks)
import Hydra.Events.Rotation (EventStore (..))
import Hydra.HeadLogic (
  Effect (..),
  HeadState (..),
  Input (..),
  Outcome (..),
  TTL,
  aggregateChainStateHistory,
  aggregateNodeState,
  aggregateState,
 )
import Hydra.HeadLogic qualified as HeadLogic
import Hydra.HeadLogic.Outcome (StateChanged (..), WaitReason (..))
import Hydra.HeadLogic.State (getHeadParameters)
import Hydra.HeadLogic.StateEvent (StateEvent (..))
import Hydra.Ledger (Ledger)
import Hydra.Logging (Tracer, traceWith)
import Hydra.Network (Host (..), Network (..), NetworkCallback (..))
import Hydra.Network.Authenticate (Authenticated (..))
import Hydra.Network.Message (Message (..), NetworkEvent (..))
import Hydra.Node.Environment (Environment (..))
import Hydra.Node.InputQueue (InputQueue (..), Queued (..), createInputQueue)
import Hydra.Node.ParameterMismatch (ParamMismatch (..), ParameterMismatch (..))
import Hydra.Node.State (NodeState (..), initNodeState)
import Hydra.Node.UnsyncedPeriod (UnsyncedPeriod (..))
import Hydra.Node.Util (readFileTextEnvelopeThrow, readSigningKey, readVerificationKey)
import Hydra.Options (CardanoChainConfig (..), ChainConfig (..), RunOptions (..), defaultContestationPeriod, defaultDepositActivation, defaultDepositPeriod)
import Hydra.Tx (HeadParameters (..), Party (..), deriveParty)
import Hydra.Tx.Secret (mkSecret)
import Hydra.Tx.Utils (verificationKeyToOnChainId)

-- * Environment Handling

-- | Initialize the 'Environment' from command line options.
initEnvironment :: RunOptions -> IO Environment
initEnvironment :: RunOptions -> IO Environment
initEnvironment RunOptions
options = do
  -- Wrap the raw key as soon as it leaves disk: every in-process holder
  -- after this point sees only a 'Secret'.
  Secret (SigningKey HydraKey)
sk <- SigningKey HydraKey -> Secret (SigningKey HydraKey)
forall a. a -> Secret a
mkSecret (SigningKey HydraKey -> Secret (SigningKey HydraKey))
-> IO (SigningKey HydraKey) -> IO (Secret (SigningKey HydraKey))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [Char] -> IO (SigningKey HydraKey)
forall a. HasTextEnvelope a => [Char] -> IO a
readFileTextEnvelopeThrow [Char]
hydraSigningKey
  [Party]
otherParties <- ([Char] -> IO Party) -> [[Char]] -> IO [Party]
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
(a -> m b) -> t a -> m (t b)
forall (m :: * -> *) a b. Monad m => (a -> m b) -> [a] -> m [b]
mapM [Char] -> IO Party
loadParty [[Char]]
hydraVerificationKeys
  [OnChainId]
participants <- IO [OnChainId]
getParticipants
  Environment -> IO Environment
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Environment -> IO Environment) -> Environment -> IO Environment
forall a b. (a -> b) -> a -> b
$
    Environment
      { $sel:party:Environment :: Party
party = Secret (SigningKey HydraKey) -> Party
deriveParty Secret (SigningKey HydraKey)
sk
      , $sel:signingKey:Environment :: Secret (SigningKey HydraKey)
signingKey = Secret (SigningKey HydraKey)
sk
      , [Party]
otherParties :: [Party]
$sel:otherParties:Environment :: [Party]
otherParties
      , [OnChainId]
participants :: [OnChainId]
$sel:participants:Environment :: [OnChainId]
participants
      , ContestationPeriod
contestationPeriod :: ContestationPeriod
$sel:contestationPeriod:Environment :: ContestationPeriod
contestationPeriod
      , DepositPeriod
depositPeriod :: DepositPeriod
$sel:depositPeriod:Environment :: DepositPeriod
depositPeriod
      , DepositPeriod
depositActivation :: DepositPeriod
$sel:depositActivation:Environment :: DepositPeriod
depositActivation
      , UnsyncedPeriod
unsyncedPeriod :: UnsyncedPeriod
$sel:unsyncedPeriod:Environment :: UnsyncedPeriod
unsyncedPeriod
      , Text
configuredPeers :: Text
$sel:configuredPeers:Environment :: Text
configuredPeers
      }
 where
  -- XXX: This is mostly a cardano-specific initialization step of loading
  -- --cardano-verification-key options and deriving 'OnChainId's from it. We should be able to call out to the various chain layer
  getParticipants :: IO [OnChainId]
getParticipants =
    case ChainConfig
chainConfig of
      Offline{} -> [OnChainId] -> IO [OnChainId]
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
      Cardano
        CardanoChainConfig
          { [[Char]]
cardanoVerificationKeys :: [[Char]]
$sel:cardanoVerificationKeys:CardanoChainConfig :: CardanoChainConfig -> [[Char]]
cardanoVerificationKeys
          , [Char]
cardanoSigningKey :: [Char]
$sel:cardanoSigningKey:CardanoChainConfig :: CardanoChainConfig -> [Char]
cardanoSigningKey
          } -> do
          CardanoSigningKey
ownSigningKey <- [Char] -> IO CardanoSigningKey
readSigningKey [Char]
cardanoSigningKey
          [VerificationKey PaymentKey]
otherVerificationKeys <- ([Char] -> IO (VerificationKey PaymentKey))
-> [[Char]] -> IO [VerificationKey PaymentKey]
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
(a -> m b) -> t a -> m (t b)
forall (m :: * -> *) a b. Monad m => (a -> m b) -> [a] -> m [b]
mapM [Char] -> IO (VerificationKey PaymentKey)
readVerificationKey [[Char]]
cardanoVerificationKeys
          [OnChainId] -> IO [OnChainId]
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([OnChainId] -> IO [OnChainId]) -> [OnChainId] -> IO [OnChainId]
forall a b. (a -> b) -> a -> b
$ VerificationKey PaymentKey -> OnChainId
verificationKeyToOnChainId (VerificationKey PaymentKey -> OnChainId)
-> [VerificationKey PaymentKey] -> [OnChainId]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (CardanoSigningKey -> VerificationKey PaymentKey
getCardanoPaymentVerificationKey CardanoSigningKey
ownSigningKey VerificationKey PaymentKey
-> [VerificationKey PaymentKey] -> [VerificationKey PaymentKey]
forall a. a -> [a] -> [a]
: [VerificationKey PaymentKey]
otherVerificationKeys)

  contestationPeriod :: ContestationPeriod
contestationPeriod = case ChainConfig
chainConfig of
    Offline{} -> ContestationPeriod
defaultContestationPeriod
    Cardano CardanoChainConfig{$sel:contestationPeriod:CardanoChainConfig :: CardanoChainConfig -> ContestationPeriod
contestationPeriod = ContestationPeriod
cp} -> ContestationPeriod
cp
  depositPeriod :: DepositPeriod
depositPeriod = case ChainConfig
chainConfig of
    Offline{} -> DepositPeriod
defaultDepositPeriod
    Cardano CardanoChainConfig{$sel:depositPeriod:CardanoChainConfig :: CardanoChainConfig -> DepositPeriod
depositPeriod = DepositPeriod
dp} -> DepositPeriod
dp
  depositActivation :: DepositPeriod
depositActivation = case ChainConfig
chainConfig of
    Offline{} -> DepositPeriod
defaultDepositActivation
    Cardano CardanoChainConfig{$sel:depositActivation:CardanoChainConfig :: CardanoChainConfig -> DepositPeriod
depositActivation = DepositPeriod
da} -> DepositPeriod
da
  -- In offline mode, there's no real chain to sync with, so we use a very large
  -- unsynced period to effectively disable the unsynced check.
  unsyncedPeriod :: UnsyncedPeriod
unsyncedPeriod = case ChainConfig
chainConfig of
    Offline{} -> NominalDiffTime -> UnsyncedPeriod
UnsyncedPeriod (Int -> NominalDiffTime
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int
forall a. Bounded a => a
maxBound :: Int))
    Cardano CardanoChainConfig{$sel:unsyncedPeriod:CardanoChainConfig :: CardanoChainConfig -> UnsyncedPeriod
unsyncedPeriod = UnsyncedPeriod
up} -> UnsyncedPeriod
up

  loadParty :: [Char] -> IO Party
loadParty [Char]
p =
    VerificationKey HydraKey -> Party
Party (VerificationKey HydraKey -> Party)
-> IO (VerificationKey HydraKey) -> IO Party
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [Char] -> IO (VerificationKey HydraKey)
forall a. HasTextEnvelope a => [Char] -> IO a
readFileTextEnvelopeThrow [Char]
p

  httpUrl :: Host -> [Char]
httpUrl (Host Text
h PortNumber
p) = [Char]
"http://" [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> Text -> [Char]
forall a. ToString a => a -> [Char]
toString Text
h [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> [Char]
":" [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> PortNumber -> [Char]
forall b a. (Show a, IsString b) => a -> b
show PortNumber
p

  configuredPeers :: Text
configuredPeers =
    [Char] -> Text
pack
      ([Char] -> Text) -> [Char] -> Text
forall a b. (a -> b) -> a -> b
$ [Char] -> [[Char]] -> [Char]
forall a. [a] -> [[a]] -> [a]
intercalate [Char]
","
        ([[Char]] -> [Char]) -> ([Host] -> [[Char]]) -> [Host] -> [Char]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Host -> [Char]) -> [Host] -> [[Char]]
forall a b. (a -> b) -> [a] -> [b]
map (\Host
h -> Host -> [Char]
forall b a. (Show a, IsString b) => a -> b
show Host
h [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> [Char]
"=" [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> Host -> [Char]
httpUrl Host
h)
      ([Host] -> [Char]) -> [Host] -> [Char]
forall a b. (a -> b) -> a -> b
$ (Maybe Host -> [Host]
forall a. Maybe a -> [a]
maybeToList Maybe Host
advertise [Host] -> [Host] -> [Host]
forall a. Semigroup a => a -> a -> a
<> [Host]
peers)

  RunOptions
    { [Char]
hydraSigningKey :: [Char]
$sel:hydraSigningKey:RunOptions :: RunOptions -> [Char]
hydraSigningKey
    , [[Char]]
hydraVerificationKeys :: [[Char]]
$sel:hydraVerificationKeys:RunOptions :: RunOptions -> [[Char]]
hydraVerificationKeys
    , ChainConfig
chainConfig :: ChainConfig
$sel:chainConfig:RunOptions :: RunOptions -> ChainConfig
chainConfig
    , Maybe Host
advertise :: Maybe Host
$sel:advertise:RunOptions :: RunOptions -> Maybe Host
advertise
    , [Host]
peers :: [Host]
$sel:peers:RunOptions :: RunOptions -> [Host]
peers
    } = RunOptions
options

-- | Checks that command line options match a given 'HeadState'. This function
-- takes 'Environment' because it is derived from 'RunOptions' via
-- 'initEnvironment'.
--
-- Throws: 'ParameterMismatch' when state not matching the environment.
checkHeadState ::
  MonadThrow m =>
  Tracer m (HydraNodeLog tx) ->
  Environment ->
  HeadState tx ->
  m ()
checkHeadState :: forall (m :: * -> *) tx.
MonadThrow m =>
Tracer m (HydraNodeLog tx) -> Environment -> HeadState tx -> m ()
checkHeadState Tracer m (HydraNodeLog tx)
tracer Environment
env HeadState tx
headState = do
  Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless ([ParamMismatch] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [ParamMismatch]
paramsMismatch) (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
    Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer ([ParamMismatch] -> HydraNodeLog tx
forall tx. [ParamMismatch] -> HydraNodeLog tx
Misconfiguration [ParamMismatch]
paramsMismatch)
    ParameterMismatch -> m ()
forall e a. Exception e => e -> m a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO (ParameterMismatch -> m ()) -> ParameterMismatch -> m ()
forall a b. (a -> b) -> a -> b
$ [ParamMismatch] -> ParameterMismatch
ParameterMismatch [ParamMismatch]
paramsMismatch
 where
  paramsMismatch :: [ParamMismatch]
paramsMismatch =
    [ParamMismatch]
-> (HeadParameters -> [ParamMismatch])
-> Maybe HeadParameters
-> [ParamMismatch]
forall b a. b -> (a -> b) -> Maybe a -> b
maybe [] HeadParameters -> [ParamMismatch]
validateParameters (Maybe HeadParameters -> [ParamMismatch])
-> Maybe HeadParameters -> [ParamMismatch]
forall a b. (a -> b) -> a -> b
$ HeadState tx -> Maybe HeadParameters
forall tx. HeadState tx -> Maybe HeadParameters
getHeadParameters HeadState tx
headState

  validateParameters :: HeadParameters -> [ParamMismatch]
validateParameters HeadParameters{$sel:contestationPeriod:HeadParameters :: HeadParameters -> ContestationPeriod
contestationPeriod = ContestationPeriod
loadedCp, $sel:depositPeriod:HeadParameters :: HeadParameters -> DepositPeriod
depositPeriod = DepositPeriod
loadedDp, [Party]
parties :: [Party]
$sel:parties:HeadParameters :: HeadParameters -> [Party]
parties} =
    Writer [ParamMismatch] () -> [ParamMismatch]
forall w a. Writer w a -> w
execWriter (Writer [ParamMismatch] () -> [ParamMismatch])
-> Writer [ParamMismatch] () -> [ParamMismatch]
forall a b. (a -> b) -> a -> b
$ do
      Bool -> Writer [ParamMismatch] () -> Writer [ParamMismatch] ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (ContestationPeriod
loadedCp ContestationPeriod -> ContestationPeriod -> Bool
forall a. Eq a => a -> a -> Bool
/= ContestationPeriod
configuredCp) (Writer [ParamMismatch] () -> Writer [ParamMismatch] ())
-> Writer [ParamMismatch] () -> Writer [ParamMismatch] ()
forall a b. (a -> b) -> a -> b
$
        [ParamMismatch] -> Writer [ParamMismatch] ()
forall (m :: * -> *) w. Monad m => w -> WriterT w m ()
tell [ContestationPeriodMismatch{ContestationPeriod
loadedCp :: ContestationPeriod
$sel:loadedCp:ContestationPeriodMismatch :: ContestationPeriod
loadedCp, ContestationPeriod
configuredCp :: ContestationPeriod
$sel:configuredCp:ContestationPeriodMismatch :: ContestationPeriod
configuredCp}]

      Bool -> Writer [ParamMismatch] () -> Writer [ParamMismatch] ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (DepositPeriod
loadedDp DepositPeriod -> DepositPeriod -> Bool
forall a. Eq a => a -> a -> Bool
/= DepositPeriod
configuredDp) (Writer [ParamMismatch] () -> Writer [ParamMismatch] ())
-> Writer [ParamMismatch] () -> Writer [ParamMismatch] ()
forall a b. (a -> b) -> a -> b
$
        [ParamMismatch] -> Writer [ParamMismatch] ()
forall (m :: * -> *) w. Monad m => w -> WriterT w m ()
tell [DepositPeriodMismatch{DepositPeriod
loadedDp :: DepositPeriod
$sel:loadedDp:ContestationPeriodMismatch :: DepositPeriod
loadedDp, DepositPeriod
configuredDp :: DepositPeriod
$sel:configuredDp:ContestationPeriodMismatch :: DepositPeriod
configuredDp}]

      let loadedParties :: [Party]
loadedParties = [Party] -> [Party]
forall a. Ord a => [a] -> [a]
sort [Party]
parties
          configuredParties :: [Party]
configuredParties = [Party] -> [Party]
forall a. Ord a => [a] -> [a]
sort (Party
party Party -> [Party] -> [Party]
forall a. a -> [a] -> [a]
: [Party]
otherParties)
      Bool -> Writer [ParamMismatch] () -> Writer [ParamMismatch] ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when ([Party]
loadedParties [Party] -> [Party] -> Bool
forall a. Eq a => a -> a -> Bool
/= [Party]
configuredParties) (Writer [ParamMismatch] () -> Writer [ParamMismatch] ())
-> Writer [ParamMismatch] () -> Writer [ParamMismatch] ()
forall a b. (a -> b) -> a -> b
$
        [ParamMismatch] -> Writer [ParamMismatch] ()
forall (m :: * -> *) w. Monad m => w -> WriterT w m ()
tell [PartiesMismatch{[Party]
loadedParties :: [Party]
$sel:loadedParties:ContestationPeriodMismatch :: [Party]
loadedParties, [Party]
configuredParties :: [Party]
$sel:configuredParties:ContestationPeriodMismatch :: [Party]
configuredParties}]

  Environment{$sel:contestationPeriod:Environment :: Environment -> ContestationPeriod
contestationPeriod = ContestationPeriod
configuredCp, $sel:depositPeriod:Environment :: Environment -> DepositPeriod
depositPeriod = DepositPeriod
configuredDp, [Party]
$sel:otherParties:Environment :: Environment -> [Party]
otherParties :: [Party]
otherParties, Party
$sel:party:Environment :: Environment -> Party
party :: Party
party} = Environment
env

-- * Create and run a hydra node

-- | A draft version of the 'HydraNode' that holds state, but is not yet
-- connected (see 'connect'). This is commonly created by the 'hydrate' smart
-- constructor.
data DraftHydraNode tx m = DraftHydraNode
  { forall tx (m :: * -> *).
DraftHydraNode tx m -> Tracer m (HydraNodeLog tx)
tracer :: Tracer m (HydraNodeLog tx)
  , forall tx (m :: * -> *). DraftHydraNode tx m -> Environment
env :: Environment
  , forall tx (m :: * -> *). DraftHydraNode tx m -> Ledger tx
ledger :: Ledger tx
  , forall tx (m :: * -> *).
DraftHydraNode tx m -> NodeStateHandler tx m
nodeStateHandler :: NodeStateHandler tx m
  , forall tx (m :: * -> *).
DraftHydraNode tx m -> InputQueue m (Input tx)
inputQueue :: InputQueue m (Input tx)
  , forall tx (m :: * -> *).
DraftHydraNode tx m -> EventSource (StateEvent tx) m
eventSource :: EventSource (StateEvent tx) m
  , forall tx (m :: * -> *).
DraftHydraNode tx m -> [EventSink (StateEvent tx) m]
eventSinks :: [EventSink (StateEvent tx) m]
  , -- XXX: This is an odd field in here, but needed for the chain layer to
    -- bootstrap. Maybe move to NodeStateHandler or make it differently accessible?
    forall tx (m :: * -> *).
DraftHydraNode tx m -> ChainStateHistory tx
chainStateHistory :: ChainStateHistory tx
  }

-- | Hydrate a 'DraftHydraNode' by loading events from source, re-aggregate node
-- state and sending events to sinks while doing so.
hydrate ::
  (IsChainState tx, MonadDelay m, MonadLabelledSTM m, MonadAsync m, MonadThrow m, MonadUnliftIO m) =>
  Tracer m (HydraNodeLog tx) ->
  Environment ->
  Ledger tx ->
  ChainStateType tx ->
  EventStore (StateEvent tx) m ->
  [EventSink (StateEvent tx) m] ->
  m (DraftHydraNode tx m)
hydrate :: forall tx (m :: * -> *).
(IsChainState tx, MonadDelay m, MonadLabelledSTM m, MonadAsync m,
 MonadThrow m, MonadUnliftIO m) =>
Tracer m (HydraNodeLog tx)
-> Environment
-> Ledger tx
-> ChainStateType tx
-> EventStore (StateEvent tx) m
-> [EventSink (StateEvent tx) m]
-> m (DraftHydraNode tx m)
hydrate Tracer m (HydraNodeLog tx)
tracer Environment
env Ledger tx
ledger ChainStateType tx
initialChainState EventStore{EventSource (StateEvent tx) m
eventSource :: EventSource (StateEvent tx) m
$sel:eventSource:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSource e m
eventSource, EventSink (StateEvent tx) m
eventSink :: EventSink (StateEvent tx) m
$sel:eventSink:EventStore :: forall e (m :: * -> *). EventStore e m -> EventSink e m
eventSink} [EventSink (StateEvent tx) m]
eventSinks = do
  Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer HydraNodeLog tx
forall tx. HydraNodeLog tx
LoadingState
  (Last EventId
lastEventId, (NodeState tx
nodeState, ChainStateHistory tx
chainStateHistory)) <-
    ConduitT
  ()
  Void
  (ResourceT m)
  (Last EventId, (NodeState tx, ChainStateHistory tx))
-> m (Last EventId, (NodeState tx, ChainStateHistory tx))
forall (m :: * -> *) r.
MonadUnliftIO m =>
ConduitT () Void (ResourceT m) r -> m r
runConduitRes (ConduitT
   ()
   Void
   (ResourceT m)
   (Last EventId, (NodeState tx, ChainStateHistory tx))
 -> m (Last EventId, (NodeState tx, ChainStateHistory tx)))
-> ConduitT
     ()
     Void
     (ResourceT m)
     (Last EventId, (NodeState tx, ChainStateHistory tx))
-> m (Last EventId, (NodeState tx, ChainStateHistory tx))
forall a b. (a -> b) -> a -> b
$
      EventSource (StateEvent tx) m
-> HasEventId (StateEvent tx) =>
   ConduitT () (StateEvent tx) (ResourceT m) ()
forall e (m :: * -> *).
EventSource e m -> HasEventId e => ConduitT () e (ResourceT m) ()
sourceEvents EventSource (StateEvent tx) m
eventSource
        ConduitT () (StateEvent tx) (ResourceT m) ()
-> ConduitT
     (StateEvent tx)
     Void
     (ResourceT m)
     (Last EventId, (NodeState tx, ChainStateHistory tx))
-> ConduitT
     ()
     Void
     (ResourceT m)
     (Last EventId, (NodeState tx, ChainStateHistory tx))
forall (m :: * -> *) a b c r.
Monad m =>
ConduitT a b m () -> ConduitT b c m r -> ConduitT a c m r
.| ZipSink
  (StateEvent tx)
  (ResourceT m)
  (Last EventId, (NodeState tx, ChainStateHistory tx))
-> ConduitT
     (StateEvent tx)
     Void
     (ResourceT m)
     (Last EventId, (NodeState tx, ChainStateHistory tx))
forall i (m :: * -> *) r. ZipSink i m r -> ConduitT i Void m r
getZipSink
          ( (,)
              (Last EventId
 -> (NodeState tx, ChainStateHistory tx)
 -> (Last EventId, (NodeState tx, ChainStateHistory tx)))
-> ZipSink (StateEvent tx) (ResourceT m) (Last EventId)
-> ZipSink
     (StateEvent tx)
     (ResourceT m)
     ((NodeState tx, ChainStateHistory tx)
      -> (Last EventId, (NodeState tx, ChainStateHistory tx)))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ConduitT (StateEvent tx) Void (ResourceT m) (Last EventId)
-> ZipSink (StateEvent tx) (ResourceT m) (Last EventId)
forall i (m :: * -> *) r. ConduitT i Void m r -> ZipSink i m r
ZipSink ((StateEvent tx -> Last EventId)
-> ConduitT (StateEvent tx) Void (ResourceT m) (Last EventId)
forall (m :: * -> *) b a o.
(Monad m, Monoid b) =>
(a -> b) -> ConduitT a o m b
foldMapC (Maybe EventId -> Last EventId
forall a. Maybe a -> Last a
Last (Maybe EventId -> Last EventId)
-> (StateEvent tx -> Maybe EventId)
-> StateEvent tx
-> Last EventId
forall b c a. (b -> c) -> (a -> b) -> a -> c
. EventId -> Maybe EventId
forall a. a -> Maybe a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (EventId -> Maybe EventId)
-> (StateEvent tx -> EventId) -> StateEvent tx -> Maybe EventId
forall b c a. (b -> c) -> (a -> b) -> a -> c
. StateEvent tx -> EventId
forall a. HasEventId a => a -> EventId
getEventId))
              ZipSink
  (StateEvent tx)
  (ResourceT m)
  ((NodeState tx, ChainStateHistory tx)
   -> (Last EventId, (NodeState tx, ChainStateHistory tx)))
-> ZipSink
     (StateEvent tx) (ResourceT m) (NodeState tx, ChainStateHistory tx)
-> ZipSink
     (StateEvent tx)
     (ResourceT m)
     (Last EventId, (NodeState tx, ChainStateHistory tx))
forall a b.
ZipSink (StateEvent tx) (ResourceT m) (a -> b)
-> ZipSink (StateEvent tx) (ResourceT m) a
-> ZipSink (StateEvent tx) (ResourceT m) b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> ConduitT
  (StateEvent tx)
  Void
  (ResourceT m)
  (NodeState tx, ChainStateHistory tx)
-> ZipSink
     (StateEvent tx) (ResourceT m) (NodeState tx, ChainStateHistory tx)
forall i (m :: * -> *) r. ConduitT i Void m r -> ZipSink i m r
ZipSink ConduitT
  (StateEvent tx)
  Void
  (ResourceT m)
  (NodeState tx, ChainStateHistory tx)
recoverNodeStateC
          )
  Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer (HydraNodeLog tx -> m ()) -> HydraNodeLog tx -> m ()
forall a b. (a -> b) -> a -> b
$ LoadedChainState{$sel:lastKnownChainPoint:BeginInput :: ChainPointType tx
lastKnownChainPoint = ChainStateHistory tx -> ChainPointType tx
forall tx. ChainStateHistory tx -> ChainPointType tx
lastKnown ChainStateHistory tx
chainStateHistory}
  Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer (HydraNodeLog tx -> m ()) -> HydraNodeLog tx -> m ()
forall a b. (a -> b) -> a -> b
$ LoadedState{Last EventId
lastEventId :: Last EventId
$sel:lastEventId:BeginInput :: Last EventId
lastEventId, NodeState tx
nodeState :: NodeState tx
$sel:nodeState:BeginInput :: NodeState tx
nodeState}
  -- Check whether the loaded state matches our configuration (env)
  -- XXX: re-stream events just for this?
  Tracer m (HydraNodeLog tx) -> Environment -> HeadState tx -> m ()
forall (m :: * -> *) tx.
MonadThrow m =>
Tracer m (HydraNodeLog tx) -> Environment -> HeadState tx -> m ()
checkHeadState Tracer m (HydraNodeLog tx)
tracer Environment
env (NodeState tx -> HeadState tx
forall tx. NodeState tx -> HeadState tx
headState NodeState tx
nodeState)
  -- (Re-)submit events to sinks; de-duplication is handled by the sinks
  Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer HydraNodeLog tx
forall tx. HydraNodeLog tx
ReplayingState
  [StateEvent tx]
replayedEvents <- ConduitT () Void (ResourceT m) [StateEvent tx] -> m [StateEvent tx]
forall (m :: * -> *) r.
MonadUnliftIO m =>
ConduitT () Void (ResourceT m) r -> m r
runConduitRes (ConduitT () Void (ResourceT m) [StateEvent tx]
 -> m [StateEvent tx])
-> ConduitT () Void (ResourceT m) [StateEvent tx]
-> m [StateEvent tx]
forall a b. (a -> b) -> a -> b
$ EventSource (StateEvent tx) m
-> HasEventId (StateEvent tx) =>
   ConduitT () (StateEvent tx) (ResourceT m) ()
forall e (m :: * -> *).
EventSource e m -> HasEventId e => ConduitT () e (ResourceT m) ()
sourceEvents EventSource (StateEvent tx) m
eventSource ConduitT () (StateEvent tx) (ResourceT m) ()
-> ConduitT (StateEvent tx) Void (ResourceT m) [StateEvent tx]
-> ConduitT () Void (ResourceT m) [StateEvent tx]
forall (m :: * -> *) a b c r.
Monad m =>
ConduitT a b m () -> ConduitT b c m r -> ConduitT a c m r
.| ConduitT (StateEvent tx) Void (ResourceT m) [StateEvent tx]
forall (m :: * -> *) a o. Monad m => ConduitT a o m [a]
sinkList
  [EventSink (StateEvent tx) m] -> [StateEvent tx] -> m ()
forall (m :: * -> *) e.
(Monad m, HasEventId e) =>
[EventSink e m] -> [e] -> m ()
putEventsToSinks [EventSink (StateEvent tx) m]
eventSinks [StateEvent tx]
replayedEvents

  NodeStateHandler tx m
nodeStateHandler <- Maybe EventId -> NodeState tx -> m (NodeStateHandler tx m)
forall (m :: * -> *) tx.
MonadLabelledSTM m =>
Maybe EventId -> NodeState tx -> m (NodeStateHandler tx m)
createNodeStateHandler (Last EventId -> Maybe EventId
forall a. Last a -> Maybe a
getLast Last EventId
lastEventId) NodeState tx
nodeState
  InputQueue m (Input tx)
inputQueue <- m (InputQueue m (Input tx))
forall (m :: * -> *) e.
(MonadDelay m, MonadAsync m, MonadLabelledSTM m) =>
m (InputQueue m e)
createInputQueue
  DraftHydraNode tx m -> m (DraftHydraNode tx m)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
    DraftHydraNode
      { Tracer m (HydraNodeLog tx)
$sel:tracer:DraftHydraNode :: Tracer m (HydraNodeLog tx)
tracer :: Tracer m (HydraNodeLog tx)
tracer
      , Environment
$sel:env:DraftHydraNode :: Environment
env :: Environment
env
      , Ledger tx
$sel:ledger:DraftHydraNode :: Ledger tx
ledger :: Ledger tx
ledger
      , NodeStateHandler tx m
$sel:nodeStateHandler:DraftHydraNode :: NodeStateHandler tx m
nodeStateHandler :: NodeStateHandler tx m
nodeStateHandler
      , InputQueue m (Input tx)
$sel:inputQueue:DraftHydraNode :: InputQueue m (Input tx)
inputQueue :: InputQueue m (Input tx)
inputQueue
      , EventSource (StateEvent tx) m
$sel:eventSource:DraftHydraNode :: EventSource (StateEvent tx) m
eventSource :: EventSource (StateEvent tx) m
eventSource
      , $sel:eventSinks:DraftHydraNode :: [EventSink (StateEvent tx) m]
eventSinks = EventSink (StateEvent tx) m
eventSink EventSink (StateEvent tx) m
-> [EventSink (StateEvent tx) m] -> [EventSink (StateEvent tx) m]
forall a. a -> [a] -> [a]
: [EventSink (StateEvent tx) m]
eventSinks
      , ChainStateHistory tx
$sel:chainStateHistory:DraftHydraNode :: ChainStateHistory tx
chainStateHistory :: ChainStateHistory tx
chainStateHistory
      }
 where
  initialState :: NodeState tx
initialState = ChainStateType tx -> NodeState tx
forall tx. IsChainState tx => ChainStateType tx -> NodeState tx
initNodeState ChainStateType tx
initialChainState

  recoverNodeStateC :: ConduitT
  (StateEvent tx)
  Void
  (ResourceT m)
  (NodeState tx, ChainStateHistory tx)
recoverNodeStateC =
    (StateEvent tx -> StateChanged tx)
-> ConduitT (StateEvent tx) (StateChanged tx) (ResourceT m) ()
forall (m :: * -> *) a b. Monad m => (a -> b) -> ConduitT a b m ()
mapC StateEvent tx -> StateChanged tx
forall tx. StateEvent tx -> StateChanged tx
stateChanged
      ConduitT (StateEvent tx) (StateChanged tx) (ResourceT m) ()
-> ConduitT
     (StateChanged tx)
     Void
     (ResourceT m)
     (NodeState tx, ChainStateHistory tx)
-> ConduitT
     (StateEvent tx)
     Void
     (ResourceT m)
     (NodeState tx, ChainStateHistory tx)
forall (m :: * -> *) a b c r.
Monad m =>
ConduitT a b m () -> ConduitT b c m r -> ConduitT a c m r
.| ZipSink
  (StateChanged tx)
  (ResourceT m)
  (NodeState tx, ChainStateHistory tx)
-> ConduitT
     (StateChanged tx)
     Void
     (ResourceT m)
     (NodeState tx, ChainStateHistory tx)
forall i (m :: * -> *) r. ZipSink i m r -> ConduitT i Void m r
getZipSink
        ( (,)
            (NodeState tx
 -> ChainStateHistory tx -> (NodeState tx, ChainStateHistory tx))
-> ZipSink (StateChanged tx) (ResourceT m) (NodeState tx)
-> ZipSink
     (StateChanged tx)
     (ResourceT m)
     (ChainStateHistory tx -> (NodeState tx, ChainStateHistory tx))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ConduitT (StateChanged tx) Void (ResourceT m) (NodeState tx)
-> ZipSink (StateChanged tx) (ResourceT m) (NodeState tx)
forall i (m :: * -> *) r. ConduitT i Void m r -> ZipSink i m r
ZipSink ((NodeState tx -> StateChanged tx -> NodeState tx)
-> NodeState tx
-> ConduitT (StateChanged tx) Void (ResourceT m) (NodeState tx)
forall (m :: * -> *) a b o.
Monad m =>
(a -> b -> a) -> a -> ConduitT b o m a
foldlC NodeState tx -> StateChanged tx -> NodeState tx
forall tx.
IsChainState tx =>
NodeState tx -> StateChanged tx -> NodeState tx
aggregateNodeState NodeState tx
initialState)
            ZipSink
  (StateChanged tx)
  (ResourceT m)
  (ChainStateHistory tx -> (NodeState tx, ChainStateHistory tx))
-> ZipSink (StateChanged tx) (ResourceT m) (ChainStateHistory tx)
-> ZipSink
     (StateChanged tx)
     (ResourceT m)
     (NodeState tx, ChainStateHistory tx)
forall a b.
ZipSink (StateChanged tx) (ResourceT m) (a -> b)
-> ZipSink (StateChanged tx) (ResourceT m) a
-> ZipSink (StateChanged tx) (ResourceT m) b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> ConduitT
  (StateChanged tx) Void (ResourceT m) (ChainStateHistory tx)
-> ZipSink (StateChanged tx) (ResourceT m) (ChainStateHistory tx)
forall i (m :: * -> *) r. ConduitT i Void m r -> ZipSink i m r
ZipSink ((ChainStateHistory tx -> StateChanged tx -> ChainStateHistory tx)
-> ChainStateHistory tx
-> ConduitT
     (StateChanged tx) Void (ResourceT m) (ChainStateHistory tx)
forall (m :: * -> *) a b o.
Monad m =>
(a -> b -> a) -> a -> ConduitT b o m a
foldlC ChainStateHistory tx -> StateChanged tx -> ChainStateHistory tx
forall tx.
IsChainState tx =>
ChainStateHistory tx -> StateChanged tx -> ChainStateHistory tx
aggregateChainStateHistory (ChainStateHistory tx
 -> ConduitT
      (StateChanged tx) Void (ResourceT m) (ChainStateHistory tx))
-> ChainStateHistory tx
-> ConduitT
     (StateChanged tx) Void (ResourceT m) (ChainStateHistory tx)
forall a b. (a -> b) -> a -> b
$ ChainStateType tx -> ChainStateHistory tx
forall tx.
IsChainState tx =>
ChainStateType tx -> ChainStateHistory tx
initHistory ChainStateType tx
initialChainState)
        )

wireChainInput :: DraftHydraNode tx m -> (ChainEvent tx -> m ())
wireChainInput :: forall tx (m :: * -> *).
DraftHydraNode tx m -> ChainEvent tx -> m ()
wireChainInput DraftHydraNode tx m
node = Input tx -> m ()
enqueue (Input tx -> m ())
-> (ChainEvent tx -> Input tx) -> ChainEvent tx -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ChainEvent tx -> Input tx
forall tx. ChainEvent tx -> Input tx
ChainInput
 where
  DraftHydraNode{$sel:inputQueue:DraftHydraNode :: forall tx (m :: * -> *).
DraftHydraNode tx m -> InputQueue m (Input tx)
inputQueue = InputQueue{Input tx -> m ()
enqueue :: Input tx -> m ()
$sel:enqueue:InputQueue :: forall (m :: * -> *) e. InputQueue m e -> e -> m ()
enqueue}} = DraftHydraNode tx m
node

wireClientInput :: DraftHydraNode tx m -> (ClientInput tx -> m ())
wireClientInput :: forall tx (m :: * -> *).
DraftHydraNode tx m -> ClientInput tx -> m ()
wireClientInput DraftHydraNode tx m
node = Input tx -> m ()
enqueue (Input tx -> m ())
-> (ClientInput tx -> Input tx) -> ClientInput tx -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ClientInput tx -> Input tx
forall tx. ClientInput tx -> Input tx
ClientInput
 where
  DraftHydraNode{$sel:inputQueue:DraftHydraNode :: forall tx (m :: * -> *).
DraftHydraNode tx m -> InputQueue m (Input tx)
inputQueue = InputQueue{Input tx -> m ()
$sel:enqueue:InputQueue :: forall (m :: * -> *) e. InputQueue m e -> e -> m ()
enqueue :: Input tx -> m ()
enqueue}} = DraftHydraNode tx m
node

wireNetworkInput :: DraftHydraNode tx m -> NetworkCallback (Authenticated (Message tx)) m
wireNetworkInput :: forall tx (m :: * -> *).
DraftHydraNode tx m
-> NetworkCallback (Authenticated (Message tx)) m
wireNetworkInput DraftHydraNode tx m
node =
  NetworkCallback
    { $sel:deliver:NetworkCallback :: Authenticated (Message tx) -> m ()
deliver = \Authenticated{$sel:party:Authenticated :: forall msg. Authenticated msg -> Party
party = Party
sender, $sel:payload:Authenticated :: forall msg. Authenticated msg -> msg
payload = Message tx
msg} ->
        Input tx -> m ()
enqueue (Input tx -> m ()) -> Input tx -> m ()
forall a b. (a -> b) -> a -> b
$ Party -> Message tx -> Input tx
forall tx. Party -> Message tx -> Input tx
mkNetworkInput Party
sender Message tx
msg
    , $sel:onConnectivity:NetworkCallback :: Connectivity -> m ()
onConnectivity =
        Input tx -> m ()
enqueue (Input tx -> m ())
-> (Connectivity -> Input tx) -> Connectivity -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TTL -> NetworkEvent (Message tx) -> Input tx
forall tx. TTL -> NetworkEvent (Message tx) -> Input tx
NetworkInput TTL
1 (NetworkEvent (Message tx) -> Input tx)
-> (Connectivity -> NetworkEvent (Message tx))
-> Connectivity
-> Input tx
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Connectivity -> NetworkEvent (Message tx)
forall msg. Connectivity -> NetworkEvent msg
ConnectivityEvent
    }
 where
  DraftHydraNode{$sel:inputQueue:DraftHydraNode :: forall tx (m :: * -> *).
DraftHydraNode tx m -> InputQueue m (Input tx)
inputQueue = InputQueue{Input tx -> m ()
$sel:enqueue:InputQueue :: forall (m :: * -> *) e. InputQueue m e -> e -> m ()
enqueue :: Input tx -> m ()
enqueue}} = DraftHydraNode tx m
node

-- | Create a network input with corresponding default ttl from given sender.
mkNetworkInput :: Party -> Message tx -> Input tx
mkNetworkInput :: forall tx. Party -> Message tx -> Input tx
mkNetworkInput Party
sender Message tx
msg =
  case Message tx
msg of
    ReqTx{} -> TTL -> NetworkEvent (Message tx) -> Input tx
forall tx. TTL -> NetworkEvent (Message tx) -> Input tx
NetworkInput TTL
defaultTxTTL (NetworkEvent (Message tx) -> Input tx)
-> NetworkEvent (Message tx) -> Input tx
forall a b. (a -> b) -> a -> b
$ ReceivedMessage{Party
sender :: Party
$sel:sender:ConnectivityEvent :: Party
sender, Message tx
msg :: Message tx
$sel:msg:ConnectivityEvent :: Message tx
msg}
    ReqDec{} -> TTL -> NetworkEvent (Message tx) -> Input tx
forall tx. TTL -> NetworkEvent (Message tx) -> Input tx
NetworkInput TTL
defaultTxTTL (NetworkEvent (Message tx) -> Input tx)
-> NetworkEvent (Message tx) -> Input tx
forall a b. (a -> b) -> a -> b
$ ReceivedMessage{Party
sender :: Party
$sel:sender:ConnectivityEvent :: Party
sender, Message tx
msg :: Message tx
$sel:msg:ConnectivityEvent :: Message tx
msg}
    Message tx
_ -> TTL -> NetworkEvent (Message tx) -> Input tx
forall tx. TTL -> NetworkEvent (Message tx) -> Input tx
NetworkInput TTL
defaultTTL (NetworkEvent (Message tx) -> Input tx)
-> NetworkEvent (Message tx) -> Input tx
forall a b. (a -> b) -> a -> b
$ ReceivedMessage{Party
sender :: Party
$sel:sender:ConnectivityEvent :: Party
sender, Message tx
msg :: Message tx
$sel:msg:ConnectivityEvent :: Message tx
msg}

-- | Connect chain, network and API to a hydrated 'DraftHydraNode' to get a fully
-- connected 'HydraNode'.
connect ::
  Monad m =>
  Chain tx m ->
  Network m (Message tx) ->
  Server tx m ->
  DraftHydraNode tx m ->
  m (HydraNode tx m)
connect :: forall (m :: * -> *) tx.
Monad m =>
Chain tx m
-> Network m (Message tx)
-> Server tx m
-> DraftHydraNode tx m
-> m (HydraNode tx m)
connect Chain tx m
chain Network m (Message tx)
network Server tx m
server DraftHydraNode tx m
node =
  HydraNode tx m -> m (HydraNode tx m)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure HydraNode{Tracer m (HydraNodeLog tx)
tracer :: Tracer m (HydraNodeLog tx)
$sel:tracer:HydraNode :: Tracer m (HydraNodeLog tx)
tracer, Environment
env :: Environment
$sel:env:HydraNode :: Environment
env, Ledger tx
ledger :: Ledger tx
$sel:ledger:HydraNode :: Ledger tx
ledger, NodeStateHandler tx m
nodeStateHandler :: NodeStateHandler tx m
$sel:nodeStateHandler:HydraNode :: NodeStateHandler tx m
nodeStateHandler, InputQueue m (Input tx)
inputQueue :: InputQueue m (Input tx)
$sel:inputQueue:HydraNode :: InputQueue m (Input tx)
inputQueue, EventSource (StateEvent tx) m
eventSource :: EventSource (StateEvent tx) m
$sel:eventSource:HydraNode :: EventSource (StateEvent tx) m
eventSource, [EventSink (StateEvent tx) m]
eventSinks :: [EventSink (StateEvent tx) m]
$sel:eventSinks:HydraNode :: [EventSink (StateEvent tx) m]
eventSinks, $sel:oc:HydraNode :: Chain tx m
oc = Chain tx m
chain, $sel:hn:HydraNode :: Network m (Message tx)
hn = Network m (Message tx)
network, Server tx m
server :: Server tx m
$sel:server:HydraNode :: Server tx m
server}
 where
  DraftHydraNode{Tracer m (HydraNodeLog tx)
$sel:tracer:DraftHydraNode :: forall tx (m :: * -> *).
DraftHydraNode tx m -> Tracer m (HydraNodeLog tx)
tracer :: Tracer m (HydraNodeLog tx)
tracer, Environment
$sel:env:DraftHydraNode :: forall tx (m :: * -> *). DraftHydraNode tx m -> Environment
env :: Environment
env, Ledger tx
$sel:ledger:DraftHydraNode :: forall tx (m :: * -> *). DraftHydraNode tx m -> Ledger tx
ledger :: Ledger tx
ledger, NodeStateHandler tx m
$sel:nodeStateHandler:DraftHydraNode :: forall tx (m :: * -> *).
DraftHydraNode tx m -> NodeStateHandler tx m
nodeStateHandler :: NodeStateHandler tx m
nodeStateHandler, InputQueue m (Input tx)
$sel:inputQueue:DraftHydraNode :: forall tx (m :: * -> *).
DraftHydraNode tx m -> InputQueue m (Input tx)
inputQueue :: InputQueue m (Input tx)
inputQueue, EventSource (StateEvent tx) m
$sel:eventSource:DraftHydraNode :: forall tx (m :: * -> *).
DraftHydraNode tx m -> EventSource (StateEvent tx) m
eventSource :: EventSource (StateEvent tx) m
eventSource, [EventSink (StateEvent tx) m]
$sel:eventSinks:DraftHydraNode :: forall tx (m :: * -> *).
DraftHydraNode tx m -> [EventSink (StateEvent tx) m]
eventSinks :: [EventSink (StateEvent tx) m]
eventSinks} = DraftHydraNode tx m
node

-- | Fully connected hydra node with everything wired in.
data HydraNode tx m = HydraNode
  { forall tx (m :: * -> *).
HydraNode tx m -> Tracer m (HydraNodeLog tx)
tracer :: Tracer m (HydraNodeLog tx)
  , forall tx (m :: * -> *). HydraNode tx m -> Environment
env :: Environment
  , forall tx (m :: * -> *). HydraNode tx m -> Ledger tx
ledger :: Ledger tx
  , forall tx (m :: * -> *). HydraNode tx m -> NodeStateHandler tx m
nodeStateHandler :: NodeStateHandler tx m
  , forall tx (m :: * -> *). HydraNode tx m -> InputQueue m (Input tx)
inputQueue :: InputQueue m (Input tx)
  , forall tx (m :: * -> *).
HydraNode tx m -> EventSource (StateEvent tx) m
eventSource :: EventSource (StateEvent tx) m
  , forall tx (m :: * -> *).
HydraNode tx m -> [EventSink (StateEvent tx) m]
eventSinks :: [EventSink (StateEvent tx) m]
  , forall tx (m :: * -> *). HydraNode tx m -> Chain tx m
oc :: Chain tx m
  , forall tx (m :: * -> *). HydraNode tx m -> Network m (Message tx)
hn :: Network m (Message tx)
  , forall tx (m :: * -> *). HydraNode tx m -> Server tx m
server :: Server tx m
  }

runHydraNode ::
  ( MonadCatch m
  , MonadAsync m
  , MonadTime m
  , IsChainState tx
  ) =>
  HydraNode tx m ->
  m ()
runHydraNode :: forall (m :: * -> *) tx.
(MonadCatch m, MonadAsync m, MonadTime m, IsChainState tx) =>
HydraNode tx m -> m ()
runHydraNode node :: HydraNode tx m
node@HydraNode{Tracer m (HydraNodeLog tx)
$sel:tracer:HydraNode :: forall tx (m :: * -> *).
HydraNode tx m -> Tracer m (HydraNodeLog tx)
tracer :: Tracer m (HydraNodeLog tx)
tracer, $sel:nodeStateHandler:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> NodeStateHandler tx m
nodeStateHandler = NodeStateHandler{STM m (NodeState tx)
queryNodeState :: STM m (NodeState tx)
$sel:queryNodeState:NodeStateHandler :: forall tx (m :: * -> *).
NodeStateHandler tx m -> STM m (NodeState tx)
queryNodeState}} = do
  -- On startup, resume an interrupted fanout: if the node is mid-fanout
  -- ('FanoutProgress'), re-emit the next fanout step so an auto-drain whose
  -- driver crashed after its last observed chunk continues instead of stalling
  -- (mirrors the rollback re-post). A step already on chain fails harmlessly
  -- ('StalePartialFanoutTx' is silently ignored) and catch-up observations
  -- re-drive; a passive observer ('AwaitingSelection') posts nothing.
  STM m (NodeState tx) -> m (NodeState tx)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically STM m (NodeState tx)
queryNodeState m (NodeState tx) -> (NodeState tx -> m ()) -> m ()
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \NodeState tx
ns -> case NodeState tx -> HeadState tx
forall tx. NodeState tx -> HeadState tx
headState NodeState tx
ns of
    FanoutProgress PartialFanoutState tx
pfs -> case PartialFanoutState tx -> Outcome tx
forall tx. IsTx tx => PartialFanoutState tx -> Outcome tx
HeadLogic.repostFanoutStep PartialFanoutState tx
pfs of
      Continue{[Effect tx]
effects :: [Effect tx]
$sel:effects:Continue :: forall tx. Outcome tx -> [Effect tx]
effects} -> HydraNode tx m
-> Tracer m (HydraNodeLog tx) -> EventId -> [Effect tx] -> m ()
forall (m :: * -> *) tx.
(MonadAsync m, MonadCatch m, IsChainState tx) =>
HydraNode tx m
-> Tracer m (HydraNodeLog tx) -> EventId -> [Effect tx] -> m ()
processEffects HydraNode tx m
node Tracer m (HydraNodeLog tx)
tracer EventId
0 [Effect tx]
effects
      Outcome tx
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
    HeadState tx
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
  -- NOTE(SN): here we could introduce concurrent head processing, e.g. with
  -- something like 'forM_ [0..1] $ async'
  m () -> m ()
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
    UTCTime
now <- m UTCTime
forall (m :: * -> *). MonadTime m => m UTCTime
getCurrentTime
    UTCTime -> HydraNode tx m -> m ()
forall (m :: * -> *) tx.
(MonadCatch m, MonadAsync m, MonadTime m, IsChainState tx) =>
UTCTime -> HydraNode tx m -> m ()
stepHydraNode UTCTime
now HydraNode tx m
node

stepHydraNode ::
  ( MonadCatch m
  , MonadAsync m
  , MonadTime m
  , IsChainState tx
  ) =>
  UTCTime ->
  HydraNode tx m ->
  m ()
stepHydraNode :: forall (m :: * -> *) tx.
(MonadCatch m, MonadAsync m, MonadTime m, IsChainState tx) =>
UTCTime -> HydraNode tx m -> m ()
stepHydraNode UTCTime
now HydraNode tx m
node = do
  i :: Queued (Input tx)
i@Queued{EventId
queuedId :: EventId
$sel:queuedId:Queued :: forall a. Queued a -> EventId
queuedId, Input tx
queuedItem :: Input tx
$sel:queuedItem:Queued :: forall a. Queued a -> a
queuedItem} <- m (Queued (Input tx))
dequeue
  Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer (HydraNodeLog tx -> m ()) -> HydraNodeLog tx -> m ()
forall a b. (a -> b) -> a -> b
$ BeginInput{$sel:by:BeginInput :: Party
by = Party
party, $sel:inputId:BeginInput :: EventId
inputId = EventId
queuedId, $sel:input:BeginInput :: Input tx
input = Input tx
queuedItem}
  Outcome tx
outcome <- STM m (Outcome tx) -> m (Outcome tx)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (Outcome tx) -> m (Outcome tx))
-> STM m (Outcome tx) -> m (Outcome tx)
forall a b. (a -> b) -> a -> b
$ HydraNode tx m -> Input tx -> UTCTime -> STM m (Outcome tx)
forall tx (m :: * -> *).
IsChainState tx =>
HydraNode tx m -> Input tx -> UTCTime -> STM m (Outcome tx)
processNextInput HydraNode tx m
node Input tx
queuedItem UTCTime
now
  Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer (Party -> Outcome tx -> HydraNodeLog tx
forall tx. Party -> Outcome tx -> HydraNodeLog tx
LogicOutcome Party
party Outcome tx
outcome)
  case Outcome tx
outcome of
    Continue{[StateChanged tx]
stateChanges :: [StateChanged tx]
$sel:stateChanges:Continue :: forall tx. Outcome tx -> [StateChanged tx]
stateChanges, [Effect tx]
$sel:effects:Continue :: forall tx. Outcome tx -> [Effect tx]
effects :: [Effect tx]
effects} -> do
      HydraNode tx m -> [StateChanged tx] -> m ()
forall (m :: * -> *) tx.
(MonadSTM m, MonadTime m) =>
HydraNode tx m -> [StateChanged tx] -> m ()
processStateChanges HydraNode tx m
node [StateChanged tx]
stateChanges
      HydraNode tx m
-> Tracer m (HydraNodeLog tx) -> EventId -> [Effect tx] -> m ()
forall (m :: * -> *) tx.
(MonadAsync m, MonadCatch m, IsChainState tx) =>
HydraNode tx m
-> Tracer m (HydraNodeLog tx) -> EventId -> [Effect tx] -> m ()
processEffects HydraNode tx m
node Tracer m (HydraNodeLog tx)
tracer EventId
queuedId [Effect tx]
effects
      [StateChanged tx] -> m ()
releaseParkedWhenSynced [StateChanged tx]
stateChanges
    Wait{WaitReason tx
reason :: WaitReason tx
$sel:reason:Continue :: forall tx. Outcome tx -> WaitReason tx
reason, [StateChanged tx]
$sel:stateChanges:Continue :: forall tx. Outcome tx -> [StateChanged tx]
stateChanges :: [StateChanged tx]
stateChanges} -> do
      HydraNode tx m -> [StateChanged tx] -> m ()
forall (m :: * -> *) tx.
(MonadSTM m, MonadTime m) =>
HydraNode tx m -> [StateChanged tx] -> m ()
processStateChanges HydraNode tx m
node [StateChanged tx]
stateChanges
      WaitReason tx -> Queued (Input tx) -> m ()
maybeReenqueue WaitReason tx
reason Queued (Input tx)
i
    Error{} -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
  Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer EndInput{$sel:by:BeginInput :: Party
by = Party
party, $sel:inputId:BeginInput :: EventId
inputId = EventId
queuedId}
 where
  maybeReenqueue :: WaitReason tx -> Queued (Input tx) -> m ()
maybeReenqueue WaitReason tx
reason q :: Queued (Input tx)
q@Queued{EventId
$sel:queuedId:Queued :: forall a. Queued a -> EventId
queuedId :: EventId
queuedId, Input tx
$sel:queuedItem:Queued :: forall a. Queued a -> a
queuedItem :: Input tx
queuedItem} =
    case Input tx
queuedItem of
      -- While the node is catching up it can't process any network message
      -- yet, so re-enqueuing here would just spin (and never terminate if the
      -- node stays out of sync). Park the message instead and replay it once
      -- the node is in sync (see 'releaseParkedWhenSynced'). This also avoids
      -- burning the message's retry budget during sync, which would otherwise
      -- drop e.g. ReqTx before catch-up completes.
      NetworkInput TTL
_ NetworkEvent (Message tx)
_
        | WaitOnNodeInSync{} <- WaitReason tx
reason -> Queued (Input tx) -> m ()
park Queued (Input tx)
q
      NetworkInput TTL
ttl NetworkEvent (Message tx)
msg
        | TTL
ttl TTL -> TTL -> Bool
forall a. Ord a => a -> a -> Bool
> TTL
0 -> DiffTime -> Queued (Input tx) -> m ()
reenqueue DiffTime
waitDelay Queued (Input tx)
q{queuedItem = NetworkInput (ttl - 1) msg}
      Input tx
_ -> Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer (HydraNodeLog tx -> m ()) -> HydraNodeLog tx -> m ()
forall a b. (a -> b) -> a -> b
$ DroppedFromQueue{$sel:inputId:BeginInput :: EventId
inputId = EventId
queuedId, $sel:input:BeginInput :: Input tx
input = Input tx
queuedItem}

  -- Replay parked network inputs once the node transitions into sync, so
  -- messages received during catch-up are processed rather than lost.
  releaseParkedWhenSynced :: [StateChanged tx] -> m ()
releaseParkedWhenSynced [StateChanged tx]
stateChanges =
    Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when ((StateChanged tx -> Bool) -> [StateChanged tx] -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any StateChanged tx -> Bool
forall tx. StateChanged tx -> Bool
isNodeSynced [StateChanged tx]
stateChanges) m ()
releaseParked

  isNodeSynced :: StateChanged tx -> Bool
  isNodeSynced :: forall tx. StateChanged tx -> Bool
isNodeSynced = \case
    NodeSynced{} -> Bool
True
    StateChanged tx
_ -> Bool
False

  Environment{Party
$sel:party:Environment :: Environment -> Party
party :: Party
party} = Environment
env

  HydraNode{Tracer m (HydraNodeLog tx)
$sel:tracer:HydraNode :: forall tx (m :: * -> *).
HydraNode tx m -> Tracer m (HydraNodeLog tx)
tracer :: Tracer m (HydraNodeLog tx)
tracer, $sel:inputQueue:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> InputQueue m (Input tx)
inputQueue = InputQueue{m (Queued (Input tx))
dequeue :: m (Queued (Input tx))
$sel:dequeue:InputQueue :: forall (m :: * -> *) e. InputQueue m e -> m (Queued e)
dequeue, DiffTime -> Queued (Input tx) -> m ()
reenqueue :: DiffTime -> Queued (Input tx) -> m ()
$sel:reenqueue:InputQueue :: forall (m :: * -> *) e.
InputQueue m e -> DiffTime -> Queued e -> m ()
reenqueue, Queued (Input tx) -> m ()
park :: Queued (Input tx) -> m ()
$sel:park:InputQueue :: forall (m :: * -> *) e. InputQueue m e -> Queued e -> m ()
park, m ()
releaseParked :: m ()
$sel:releaseParked:InputQueue :: forall (m :: * -> *) e. InputQueue m e -> m ()
releaseParked}, Environment
$sel:env:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Environment
env :: Environment
env} = HydraNode tx m
node

-- | The maximum number of times to re-enqueue a network messages upon 'Wait'.
-- outcome.
defaultTTL :: TTL
defaultTTL :: TTL
defaultTTL = TTL
6000

-- | The maximum number of times to re-enqueue 'ReqTx' and 'ReqDec' network
-- messages upon 'Wait'.
defaultTxTTL :: TTL
defaultTxTTL :: TTL
defaultTxTTL = TTL
5

-- | The time to wait between re-enqueuing a 'Wait' outcome.
waitDelay :: DiffTime
waitDelay :: DiffTime
waitDelay = DiffTime
0.1

-- | Monadic interface around 'Hydra.Logic.update'.
processNextInput ::
  IsChainState tx =>
  HydraNode tx m ->
  Input tx ->
  UTCTime ->
  STM m (Outcome tx)
processNextInput :: forall tx (m :: * -> *).
IsChainState tx =>
HydraNode tx m -> Input tx -> UTCTime -> STM m (Outcome tx)
processNextInput HydraNode{NodeStateHandler tx m
$sel:nodeStateHandler:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> NodeStateHandler tx m
nodeStateHandler :: NodeStateHandler tx m
nodeStateHandler, Ledger tx
$sel:ledger:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Ledger tx
ledger :: Ledger tx
ledger, Environment
$sel:env:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Environment
env :: Environment
env} Input tx
e UTCTime
now =
  (NodeState tx -> (Outcome tx, NodeState tx)) -> STM m (Outcome tx)
forall a. (NodeState tx -> (a, NodeState tx)) -> STM m a
modifyNodeState ((NodeState tx -> (Outcome tx, NodeState tx))
 -> STM m (Outcome tx))
-> (NodeState tx -> (Outcome tx, NodeState tx))
-> STM m (Outcome tx)
forall a b. (a -> b) -> a -> b
$ \NodeState tx
s ->
    let outcome :: Outcome tx
outcome = Environment
-> Ledger tx -> UTCTime -> NodeState tx -> Input tx -> Outcome tx
forall tx.
IsChainState tx =>
Environment
-> Ledger tx -> UTCTime -> NodeState tx -> Input tx -> Outcome tx
HeadLogic.update Environment
env Ledger tx
ledger UTCTime
now NodeState tx
s Input tx
e
     in (Outcome tx
outcome, NodeState tx -> Outcome tx -> NodeState tx
forall tx.
IsChainState tx =>
NodeState tx -> Outcome tx -> NodeState tx
aggregateState NodeState tx
s Outcome tx
outcome)
 where
  NodeStateHandler{forall a. (NodeState tx -> (a, NodeState tx)) -> STM m a
modifyNodeState :: forall a. (NodeState tx -> (a, NodeState tx)) -> STM m a
$sel:modifyNodeState:NodeStateHandler :: forall tx (m :: * -> *).
NodeStateHandler tx m
-> forall a. (NodeState tx -> (a, NodeState tx)) -> STM m a
modifyNodeState} = NodeStateHandler tx m
nodeStateHandler

processStateChanges :: (MonadSTM m, MonadTime m) => HydraNode tx m -> [StateChanged tx] -> m ()
processStateChanges :: forall (m :: * -> *) tx.
(MonadSTM m, MonadTime m) =>
HydraNode tx m -> [StateChanged tx] -> m ()
processStateChanges HydraNode tx m
node [StateChanged tx]
stateChanges = do
  [StateEvent tx]
events <- [StateChanged tx]
-> (StateChanged tx -> m (StateEvent tx)) -> m [StateEvent tx]
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
t a -> (a -> m b) -> m (t b)
forM [StateChanged tx]
stateChanges ((StateChanged tx -> m (StateEvent tx)) -> m [StateEvent tx])
-> (StateChanged tx -> m (StateEvent tx)) -> m [StateEvent tx]
forall a b. (a -> b) -> a -> b
$ \StateChanged tx
stateChanged -> do
    UTCTime
time <- m UTCTime
forall (m :: * -> *). MonadTime m => m UTCTime
getCurrentTime
    EventId
eventId <- STM m EventId -> m EventId
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically STM m EventId
getNextEventId
    StateEvent tx -> m (StateEvent tx)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure StateEvent{EventId
eventId :: EventId
$sel:eventId:StateEvent :: EventId
eventId, StateChanged tx
$sel:stateChanged:StateEvent :: StateChanged tx
stateChanged :: StateChanged tx
stateChanged, UTCTime
time :: UTCTime
$sel:time:StateEvent :: UTCTime
time}
  [EventSink (StateEvent tx) m] -> [StateEvent tx] -> m ()
forall (m :: * -> *) e.
(Monad m, HasEventId e) =>
[EventSink e m] -> [e] -> m ()
putEventsToSinks [EventSink (StateEvent tx) m]
eventSinks [StateEvent tx]
events
 where
  HydraNode
    { [EventSink (StateEvent tx) m]
$sel:eventSinks:HydraNode :: forall tx (m :: * -> *).
HydraNode tx m -> [EventSink (StateEvent tx) m]
eventSinks :: [EventSink (StateEvent tx) m]
eventSinks
    , $sel:nodeStateHandler:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> NodeStateHandler tx m
nodeStateHandler = NodeStateHandler{STM m EventId
getNextEventId :: STM m EventId
$sel:getNextEventId:NodeStateHandler :: forall tx (m :: * -> *). NodeStateHandler tx m -> STM m EventId
getNextEventId}
    } = HydraNode tx m
node

processEffects ::
  ( MonadAsync m
  , MonadCatch m
  , IsChainState tx
  ) =>
  HydraNode tx m ->
  Tracer m (HydraNodeLog tx) ->
  Word64 ->
  [Effect tx] ->
  m ()
processEffects :: forall (m :: * -> *) tx.
(MonadAsync m, MonadCatch m, IsChainState tx) =>
HydraNode tx m
-> Tracer m (HydraNodeLog tx) -> EventId -> [Effect tx] -> m ()
processEffects HydraNode tx m
node Tracer m (HydraNodeLog tx)
tracer EventId
inputId [Effect tx]
effects = do
  ((Effect tx, Word32) -> m ()) -> [(Effect tx, Word32)] -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (Effect tx, Word32) -> m ()
processEffect ([(Effect tx, Word32)] -> m ()) -> [(Effect tx, Word32)] -> m ()
forall a b. (a -> b) -> a -> b
$ [Effect tx] -> [Word32] -> [(Effect tx, Word32)]
forall a b. [a] -> [b] -> [(a, b)]
zip [Effect tx]
effects [Word32
0 ..]
 where
  processEffect :: (Effect tx, Word32) -> m ()
processEffect (Effect tx
effect, Word32
effectId) = do
    Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer (HydraNodeLog tx -> m ()) -> HydraNodeLog tx -> m ()
forall a b. (a -> b) -> a -> b
$ Party -> EventId -> Word32 -> Effect tx -> HydraNodeLog tx
forall tx.
Party -> EventId -> Word32 -> Effect tx -> HydraNodeLog tx
BeginEffect Party
party EventId
inputId Word32
effectId Effect tx
effect
    case Effect tx
effect of
      ClientEffect ClientMessage tx
i -> Server tx m -> ClientMessage tx -> m ()
forall tx (m :: * -> *). Server tx m -> ClientMessage tx -> m ()
sendMessage Server tx m
server ClientMessage tx
i
      NetworkEffect Message tx
msg -> Network m (Message tx) -> Message tx -> m ()
forall (m :: * -> *) msg. Network m msg -> msg -> m ()
broadcast Network m (Message tx)
hn Message tx
msg
      OnChainEffect{PostChainTx tx
postChainTx :: PostChainTx tx
$sel:postChainTx:ClientEffect :: forall tx. Effect tx -> PostChainTx tx
postChainTx} ->
        MonadThrow m => PostChainTx tx -> m ()
PostChainTx tx -> m ()
postTx PostChainTx tx
postChainTx
          m () -> (PostTxError tx -> m ()) -> m ()
forall e a. Exception e => m a -> (e -> m a) -> m a
forall (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
m a -> (e -> m a) -> m a
`catch` \(PostTxError tx
postTxError :: PostTxError tx) ->
            Input tx -> m ()
enqueue (Input tx -> m ())
-> (ChainEvent tx -> Input tx) -> ChainEvent tx -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ChainEvent tx -> Input tx
forall tx. ChainEvent tx -> Input tx
ChainInput (ChainEvent tx -> m ()) -> ChainEvent tx -> m ()
forall a b. (a -> b) -> a -> b
$ PostTxError{PostChainTx tx
postChainTx :: PostChainTx tx
$sel:postChainTx:Observation :: PostChainTx tx
postChainTx, PostTxError tx
postTxError :: PostTxError tx
$sel:postTxError:Observation :: PostTxError tx
postTxError, $sel:failingTx:Observation :: Maybe tx
failingTx = Maybe tx
forall a. Maybe a
Nothing}
    Tracer m (HydraNodeLog tx) -> HydraNodeLog tx -> m ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer m (HydraNodeLog tx)
tracer (HydraNodeLog tx -> m ()) -> HydraNodeLog tx -> m ()
forall a b. (a -> b) -> a -> b
$ Party -> EventId -> Word32 -> HydraNodeLog tx
forall tx. Party -> EventId -> Word32 -> HydraNodeLog tx
EndEffect Party
party EventId
inputId Word32
effectId

  HydraNode
    { Network m (Message tx)
$sel:hn:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Network m (Message tx)
hn :: Network m (Message tx)
hn
    , $sel:oc:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Chain tx m
oc = Chain{MonadThrow m => PostChainTx tx -> m ()
postTx :: MonadThrow m => PostChainTx tx -> m ()
$sel:postTx:Chain :: forall tx (m :: * -> *).
Chain tx m -> MonadThrow m => PostChainTx tx -> m ()
postTx}
    , $sel:inputQueue:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> InputQueue m (Input tx)
inputQueue = InputQueue{Input tx -> m ()
$sel:enqueue:InputQueue :: forall (m :: * -> *) e. InputQueue m e -> e -> m ()
enqueue :: Input tx -> m ()
enqueue}
    , $sel:env:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Environment
env = Environment{Party
$sel:party:Environment :: Environment -> Party
party :: Party
party}
    , Server tx m
$sel:server:HydraNode :: forall tx (m :: * -> *). HydraNode tx m -> Server tx m
server :: Server tx m
server
    } = HydraNode tx m
node

-- ** Manage state

-- | Handle to access and modify the state in the Hydra Node.
data NodeStateHandler tx m = NodeStateHandler
  { forall tx (m :: * -> *).
NodeStateHandler tx m
-> forall a. (NodeState tx -> (a, NodeState tx)) -> STM m a
modifyNodeState :: forall a. (NodeState tx -> (a, NodeState tx)) -> STM m a
  , forall tx (m :: * -> *).
NodeStateHandler tx m -> STM m (NodeState tx)
queryNodeState :: STM m (NodeState tx)
  , forall tx (m :: * -> *). NodeStateHandler tx m -> STM m EventId
getNextEventId :: STM m EventId
  }

-- | Initialize a new 'NodeStateHandler'.
createNodeStateHandler ::
  MonadLabelledSTM m =>
  -- | Last seen 'EventId'.
  Maybe EventId ->
  NodeState tx ->
  m (NodeStateHandler tx m)
createNodeStateHandler :: forall (m :: * -> *) tx.
MonadLabelledSTM m =>
Maybe EventId -> NodeState tx -> m (NodeStateHandler tx m)
createNodeStateHandler Maybe EventId
lastSeenEventId NodeState tx
initialState = do
  TVar m EventId
nextEventIdV <- [Char] -> EventId -> m (TVar m EventId)
forall (m :: * -> *) a.
MonadLabelledSTM m =>
[Char] -> a -> m (TVar m a)
newLabelledTVarIO [Char]
"next-event-id" (EventId -> m (TVar m EventId)) -> EventId -> m (TVar m EventId)
forall a b. (a -> b) -> a -> b
$ EventId -> (EventId -> EventId) -> Maybe EventId -> EventId
forall b a. b -> (a -> b) -> Maybe a -> b
maybe EventId
0 (EventId -> EventId -> EventId
forall a. Num a => a -> a -> a
+ EventId
1) Maybe EventId
lastSeenEventId
  TVar m (NodeState tx)
ns <- [Char] -> NodeState tx -> m (TVar m (NodeState tx))
forall (m :: * -> *) a.
MonadLabelledSTM m =>
[Char] -> a -> m (TVar m a)
newLabelledTVarIO [Char]
"node-state" NodeState tx
initialState
  NodeStateHandler tx m -> m (NodeStateHandler tx m)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
    NodeStateHandler
      { $sel:modifyNodeState:NodeStateHandler :: forall a. (NodeState tx -> (a, NodeState tx)) -> STM m a
modifyNodeState = TVar m (NodeState tx)
-> (NodeState tx -> (a, NodeState tx)) -> STM m a
forall s a. TVar m s -> (s -> (a, s)) -> STM m a
forall (m :: * -> *) s a.
MonadSTM m =>
TVar m s -> (s -> (a, s)) -> STM m a
stateTVar TVar m (NodeState tx)
ns
      , $sel:queryNodeState:NodeStateHandler :: STM m (NodeState tx)
queryNodeState = TVar m (NodeState tx) -> STM m (NodeState tx)
forall a. TVar m a -> STM m a
forall (m :: * -> *) a. MonadSTM m => TVar m a -> STM m a
readTVar TVar m (NodeState tx)
ns
      , $sel:getNextEventId:NodeStateHandler :: STM m EventId
getNextEventId = do
          EventId
eventId <- TVar m EventId -> STM m EventId
forall a. TVar m a -> STM m a
forall (m :: * -> *) a. MonadSTM m => TVar m a -> STM m a
readTVar TVar m EventId
nextEventIdV
          TVar m EventId -> EventId -> STM m ()
forall a. TVar m a -> a -> STM m ()
forall (m :: * -> *) a. MonadSTM m => TVar m a -> a -> STM m ()
writeTVar TVar m EventId
nextEventIdV (EventId -> STM m ()) -> EventId -> STM m ()
forall a b. (a -> b) -> a -> b
$ EventId
eventId EventId -> EventId -> EventId
forall a. Num a => a -> a -> a
+ EventId
1
          EventId -> STM m EventId
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure EventId
eventId
      }

-- * Logging

data HydraNodeLog tx
  = BeginInput {forall tx. HydraNodeLog tx -> Party
by :: Party, forall tx. HydraNodeLog tx -> EventId
inputId :: Word64, forall tx. HydraNodeLog tx -> Input tx
input :: Input tx}
  | EndInput {by :: Party, inputId :: Word64}
  | BeginEffect {by :: Party, inputId :: Word64, forall tx. HydraNodeLog tx -> Word32
effectId :: Word32, forall tx. HydraNodeLog tx -> Effect tx
effect :: Effect tx}
  | EndEffect {by :: Party, inputId :: Word64, effectId :: Word32}
  | LogicOutcome {by :: Party, forall tx. HydraNodeLog tx -> Outcome tx
outcome :: Outcome tx}
  | DroppedFromQueue {inputId :: Word64, input :: Input tx}
  | LoadingState
  | LoadedState {forall tx. HydraNodeLog tx -> Last EventId
lastEventId :: Last EventId, forall tx. HydraNodeLog tx -> NodeState tx
nodeState :: NodeState tx}
  | LoadedChainState {forall tx. HydraNodeLog tx -> ChainPointType tx
lastKnownChainPoint :: ChainPointType tx}
  | ReplayingState
  | Misconfiguration {forall tx. HydraNodeLog tx -> [ParamMismatch]
misconfigurationErrors :: [ParamMismatch]}
  deriving stock ((forall x. HydraNodeLog tx -> Rep (HydraNodeLog tx) x)
-> (forall x. Rep (HydraNodeLog tx) x -> HydraNodeLog tx)
-> Generic (HydraNodeLog tx)
forall x. Rep (HydraNodeLog tx) x -> HydraNodeLog tx
forall x. HydraNodeLog tx -> Rep (HydraNodeLog tx) x
forall a.
(forall x. a -> Rep a x) -> (forall x. Rep a x -> a) -> Generic a
forall tx x. Rep (HydraNodeLog tx) x -> HydraNodeLog tx
forall tx x. HydraNodeLog tx -> Rep (HydraNodeLog tx) x
$cfrom :: forall tx x. HydraNodeLog tx -> Rep (HydraNodeLog tx) x
from :: forall x. HydraNodeLog tx -> Rep (HydraNodeLog tx) x
$cto :: forall tx x. Rep (HydraNodeLog tx) x -> HydraNodeLog tx
to :: forall x. Rep (HydraNodeLog tx) x -> HydraNodeLog tx
Generic)

deriving stock instance IsChainState tx => Eq (HydraNodeLog tx)
deriving stock instance IsChainState tx => Show (HydraNodeLog tx)
deriving anyclass instance IsChainState tx => ToJSON (HydraNodeLog tx)