add logic for monitoring shelley stake pools registration certificates

Still rudimentary, and doesn't keep track of metadata information _yet_ but it's coming.

lib/shelley/src/Cardano/Wallet/Shelley/Compatibility.hs

@@ -61,9 +61,10 @@ module Cardano.Wallet.Shelley.Compatibility
     , fromRewards
     , optimumNumberOfPools
 
-
     , fromBlockNo
     , fromShelleyBlock
+    , fromShelleyBlock'
+    , toBlockHeader
     , fromShelleyHash
     , fromPrevHash
     , fromChainHash
@@ -302,33 +303,56 @@ toSlotNo :: W.EpochLength -> W.SlotId -> SlotNo
 toSlotNo epLength =
     SlotNo . W.flatSlot epLength
 
+toBlockHeader
+    :: W.Hash "Genesis"
+    -> W.EpochLength
+    -> ShelleyBlock
+    -> W.BlockHeader
+toBlockHeader genesisHash epLength blk =
+    let
+        O.ShelleyBlock (SL.Block (SL.BHeader header _) _) headerHash = blk
+    in
+    W.BlockHeader
+        { slotId =
+            fromSlotNo epLength $ SL.bheaderSlotNo header
+        , blockHeight =
+            fromBlockNo $ SL.bheaderBlockNo header
+        , headerHash =
+            fromShelleyHash headerHash
+        , parentHeaderHash =
+            fromPrevHash (coerce genesisHash) $
+                SL.bheaderPrev header
+        }
+
 fromShelleyBlock
     :: W.Hash "Genesis"
     -> W.EpochLength
     -> ShelleyBlock
     -> W.Block
 fromShelleyBlock genesisHash epLength blk =
     let
-       O.ShelleyBlock (SL.Block (SL.BHeader header _) txSeq) headerHash = blk
+       O.ShelleyBlock (SL.Block _ txSeq) _ = blk
        SL.TxSeq txs' = txSeq
        (txs, certs, _) = unzip3 $ map fromShelleyTx $ toList txs'
 
     in W.Block
-        { header = W.BlockHeader
-            { slotId =
-                fromSlotNo epLength $ SL.bheaderSlotNo header
-            , blockHeight =
-                fromBlockNo $ SL.bheaderBlockNo header
-            , headerHash =
-                fromShelleyHash headerHash
-            , parentHeaderHash =
-                fromPrevHash (coerce genesisHash) $
-                    SL.bheaderPrev header
-            }
+        { header = toBlockHeader genesisHash epLength blk
         , transactions = txs
         , delegations  = mconcat certs
         }
 
+fromShelleyBlock'
+    :: W.EpochLength
+    -> ShelleyBlock
+    -> (W.SlotId, [W.PoolRegistrationCertificate])
+fromShelleyBlock' epLength blk =
+    let
+        O.ShelleyBlock (SL.Block (SL.BHeader header _) txSeq) _ = blk
+        SL.TxSeq txs' = txSeq
+        (_, _, certs) = unzip3 $ map fromShelleyTx $ toList txs'
+    in
+        (fromSlotNo epLength $ SL.bheaderSlotNo header, mconcat certs)
+
 fromShelleyHash :: ShelleyHash c -> W.Hash "BlockHeader"
 fromShelleyHash (ShelleyHash (SL.HashHeader h)) = W.Hash (getHash h)
 
@@ -397,7 +421,7 @@ fromMaxTxSize :: Natural -> Quantity "byte" Word16
 fromMaxTxSize =
     Quantity . fromIntegral
 
-fromPParams :: HasCallStack => SL.PParams -> W.ProtocolParameters
+fromPParams :: SL.PParams -> W.ProtocolParameters
 fromPParams pp = W.ProtocolParameters
     { decentralizationLevel =
         decentralizationLevelFromPParams pp
@@ -450,8 +474,7 @@ txParametersFromPParams pp = W.TxParameters
 
 -- | Convert genesis data into blockchain params and an initial set of UTxO
 fromGenesisData
-    :: HasCallStack
-    => ShelleyGenesis TPraosStandardCrypto
+    :: ShelleyGenesis TPraosStandardCrypto
     -> (W.NetworkParameters, W.Block)
 fromGenesisData g =
     ( W.NetworkParameters

lib/shelley/src/Cardano/Wallet/Shelley/Pools.hs

@@ -2,8 +2,10 @@
 {-# LANGUAGE DeriveGeneric #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE PolyKinds #-}
+{-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE TypeFamilies #-}
 {-# LANGUAGE TypeOperators #-}
 -- |
@@ -16,33 +18,59 @@ module Cardano.Wallet.Shelley.Pools where
 
 import Prelude
 
+import Cardano.BM.Data.Severity
+    ( Severity (..) )
+import Cardano.BM.Data.Tracer
+    ( HasPrivacyAnnotation (..), HasSeverityAnnotation (..) )
+import Cardano.Pool.DB
+    ( DBLayer (..) )
 import Cardano.Wallet.Api.Server
     ( LiftHandler (..), apiError )
 import Cardano.Wallet.Api.Types
     ( ApiErrorCode (..), ApiT (..) )
 import Cardano.Wallet.Network
-    ( NetworkLayer (..) )
+    ( FollowAction (..)
+    , FollowExit (..)
+    , FollowLog
+    , NetworkLayer (..)
+    , follow
+    )
 import Cardano.Wallet.Primitive.Types
-    ( Coin (..), GenesisParameters (..), PoolId )
+    ( BlockHeader
+    , Coin (..)
+    , GenesisParameters (..)
+    , PoolId
+    , PoolRegistrationCertificate
+    , ProtocolParameters
+    , SlotId
+    )
 import Cardano.Wallet.Shelley.Compatibility
     ( Shelley
     , ShelleyBlock
     , fromPoolDistr
     , fromRewards
+    , fromShelleyBlock'
     , optimumNumberOfPools
+    , toBlockHeader
     , toPoint
     , toShelleyCoin
     )
 import Cardano.Wallet.Shelley.Network
     ( pattern Cursor )
 import Cardano.Wallet.Unsafe
     ( unsafeMkPercentage, unsafeRunExceptT )
+import Control.Monad
+    ( forM_ )
 import Control.Monad.Class.MonadSTM
     ( MonadSTM, TQueue )
 import Control.Monad.IO.Class
     ( liftIO )
 import Control.Monad.Trans.Except
     ( ExceptT (..), runExceptT, withExceptT )
+import Control.Tracer
+    ( Tracer, contramap, traceWith )
+import Data.List.NonEmpty
+    ( NonEmpty (..) )
 import Data.Map
     ( Map )
 import Data.Map.Merge.Strict
@@ -53,8 +81,12 @@ import Data.Quantity
     ( Percentage (..), Quantity (..) )
 import Data.Sort
     ( sortOn )
+import Data.Text.Class
+    ( ToText (..) )
 import Data.Word
     ( Word64 )
+import Fmt
+    ( pretty )
 import GHC.Generics
     ( Generic )
 import Ouroboros.Network.Block
@@ -200,3 +232,89 @@ newStakePoolLayer gp nl = StakePoolLayer
             }
 
         mapQ f (Quantity x) = Quantity $ f x
+
+
+--
+-- Monitoring stake pool
+--
+
+monitorStakePools
+    :: Tracer IO StakePoolLog
+    -> GenesisParameters
+    -> NetworkLayer IO t ShelleyBlock
+    -> DBLayer IO
+    -> IO ()
+monitorStakePools tr gp nl db@DBLayer{..} = do
+    cursor <- initCursor
+    traceWith tr $ MsgStartMonitoring cursor
+    follow nl (contramap MsgFollow tr) cursor forward getHeader >>= \case
+        FollowInterrupted -> traceWith tr MsgHaltMonitoring
+        FollowFailure -> traceWith tr MsgCrashMonitoring
+        FollowRollback point -> do
+            traceWith tr $ MsgRollingBackTo point
+            liftIO . atomically $ rollbackTo point
+            monitorStakePools tr gp nl db
+  where
+    GenesisParameters
+        { getGenesisBlockHash
+        , getEpochLength
+        , getEpochStability
+        } = gp
+
+    initCursor :: IO [BlockHeader]
+    initCursor = atomically $ readPoolProductionCursor (max 100 k)
+      where k = fromIntegral $ getQuantity getEpochStability
+
+    getHeader :: ShelleyBlock -> BlockHeader
+    getHeader = toBlockHeader getGenesisBlockHash getEpochLength
+
+    forward
+        :: NonEmpty ShelleyBlock
+        -> (BlockHeader, ProtocolParameters)
+        -> IO (FollowAction ())
+    forward blocks (_nodeTip, _pparams) = do
+        atomically $ forM_ blocks $ \blk -> do
+            -- FIXME: Also extract & store metadata information from the block
+            let (slot, registrations) = fromShelleyBlock' getEpochLength blk
+            forM_ registrations $ \pool -> do
+                liftIO $ traceWith tr $ MsgStakePoolRegistration pool
+                putPoolRegistration slot pool
+        pure Continue
+
+data StakePoolLog
+    = MsgFollow FollowLog
+    | MsgStartMonitoring [BlockHeader]
+    | MsgHaltMonitoring
+    | MsgCrashMonitoring
+    | MsgRollingBackTo SlotId
+    | MsgStakePoolRegistration PoolRegistrationCertificate
+    deriving (Show, Eq)
+
+instance HasPrivacyAnnotation StakePoolLog
+instance HasSeverityAnnotation StakePoolLog where
+    getSeverityAnnotation = \case
+        MsgFollow e -> getSeverityAnnotation e
+        MsgStartMonitoring{} -> Info
+        MsgHaltMonitoring{} -> Info
+        MsgCrashMonitoring{} -> Error
+        MsgRollingBackTo{} -> Info
+        MsgStakePoolRegistration{} -> Info
+
+instance ToText StakePoolLog where
+    toText = \case
+        MsgFollow e ->
+            toText e
+        MsgStartMonitoring cursor -> mconcat
+            [ "Monitoring stake pools. Currently at "
+            , case cursor of
+                [] -> "genesis"
+                _  -> pretty (last cursor)
+            ]
+        MsgHaltMonitoring ->
+            "Stopping stake pool monitoring as requested."
+        MsgCrashMonitoring ->
+            "Chain follower exited with error. Worker will no longer monitor stake pools."
+        MsgRollingBackTo point ->
+            "Rolling back to " <> pretty point
+        MsgStakePoolRegistration pool ->
+            "Discovered stake pool registration: " <> pretty pool