Add blockfrost mode to hydra-chain-observer

.github/workflows/explorer/docker-compose.yaml

@@ -23,7 +23,8 @@ services:
     ports:
       - "80:8080"
     command:
-      [ "--node-socket", "/data/node.socket"
+      [ "direct"
+      , "--node-socket", "/data/node.socket"
       , "--testnet-magic", "2"
       , "--api-port", "8080"
       # NOTE: Block in which current master scripts were published

CHANGELOG.md

@@ -40,6 +40,8 @@ changes.
   - Overall this results in transactions still to be submitted once per client,
     but requires signifanctly less book-keeping on the client-side.
 
+- Add **Blockfrost Mode** to `hydra-chain-observer`, to follow the chain via Blockfrost API.
+
 ## [0.19.0] - 2024-09-13
 
 - Tested with `cardano-node 9.1.1` and `cardano-cli 9.2.1.0`
@@ -54,7 +56,6 @@ changes.
 
 - Add a demo mode to hydra-cluster to facilitate network resiliance tests [#1552](https://github.com/cardano-scaling/hydra/pull/1552)
 
-
 ## [0.18.1] - 2024-08-15
 
 - New landing page and updated documentation style. [#1560](https://github.com/cardano-scaling/hydra/pull/1560)

cabal.project

@@ -12,8 +12,8 @@ repository cardano-haskell-packages
 
 -- See CONTRIBUTING.md for information about when and how to update these.
 index-state:
-  , hackage.haskell.org 2024-09-23T15:45:50Z
-  , cardano-haskell-packages 2024-09-20T19:39:13Z
+  , hackage.haskell.org 2024-09-25T13:28:12Z
+  , cardano-haskell-packages 2024-09-23T21:46:49Z
 
 packages:
   hydra-prelude

hydra-chain-observer/README.md

@@ -1,14 +1,37 @@
 # Hydra Chain Observer
 
-A small executable which connects to a chain like the `hydra-node`, but puts any
-observations as traces onto `stdout`.
+A lightweight executable designed to connect to a blockchain, such as the `hydra-node`, and streams chain observations as traces to `stdout`.
+It supports two modes of operation: **Direct** connection to a node via socket, and connection through **Blockfrost** API.
 
-To run, pass a `--node-socket`, corresponding network id and optionally
-`--start-chain-from`. For example:
+## Direct Mode
+
+To run the observer in Direct Mode, provide the following arguments:
+- `--node-socket`: path to the node socket file.
+- network id: `--testnet-magic` (with magic number) for the testnet or `--mainnet` for the mainnet.
+- (optional) `--start-chain-from`: specify a chain point (SLOT.HEADER_HASH) to start observing from.
+
+For example:
 
 ``` shell
-hydra-chain-observer \
+hydra-chain-observer direct \
   --node-socket testnets/preprod/node.socket \
   --testnet-magic 1 \
   --start-chain-from "41948777.5d34af0f42be9823ebd35c2d83d5d879c5615ac17f7158bb9aa4ef89072455a7"
 ```
+
+
+## Blockfrost Mode
+
+To run the observer in Blockfrost Mode, provide the following arguments:
+- `--project-path`: file path to your Blockfrost project API token hash.
+> expected to be prefixed with environment (e.g. testnetA3C2E...)
+- (optional) `--start-chain-from`: specify a chain point (SLOT.HEADER_HASH) to start observing from.
+
+For example:
+
+``` shell
+hydra-chain-observer blockfrost \
+  --project-path $PROJECT_TOKEN_HASH_PATH \
+  --start-chain-from "41948777.5d34af0f42be9823ebd35c2d83d5d879c5615ac17f7158bb9aa4ef89072455a7"
+```
+

hydra-chain-observer/exe/Main.hs

@@ -1,7 +1,7 @@
 module Main where
 
-import Hydra.ChainObserver (defaultObserverHandler)
 import Hydra.ChainObserver qualified
+import Hydra.ChainObserver.NodeClient (defaultObserverHandler)
 import Hydra.Prelude
 
 main :: IO ()

hydra-chain-observer/hydra-chain-observer.cabal

@@ -65,17 +65,24 @@ library
   hs-source-dirs:  src
   ghc-options:     -haddock
   build-depends:
+    , base16-bytestring
+    , blockfrost-client            >=0.9.1.0
     , hydra-cardano-api
     , hydra-node
     , hydra-plutus
     , hydra-prelude
     , hydra-tx
+    , io-classes
     , optparse-applicative
     , ouroboros-network-protocols
+    , retry
 
   exposed-modules:
+    Hydra.Blockfrost.ChainObserver
     Hydra.ChainObserver
+    Hydra.ChainObserver.NodeClient
     Hydra.ChainObserver.Options
+    Hydra.Ouroborus.ChainObserver
 
 executable hydra-chain-observer
   import:         project-config

hydra-chain-observer/src/Hydra/Blockfrost/ChainObserver.hs

@@ -0,0 +1,227 @@
+{-# LANGUAGE DuplicateRecordFields #-}
+
+module Hydra.Blockfrost.ChainObserver where
+
+import Hydra.Prelude
+
+import Blockfrost.Client (
+  BlockfrostClientT,
+  runBlockfrost,
+ )
+import Blockfrost.Client qualified as Blockfrost
+import Control.Concurrent.Class.MonadSTM (
+  MonadSTM (readTVarIO),
+  newTVarIO,
+  writeTVar,
+ )
+import Control.Retry (constantDelay, retrying)
+import Data.ByteString.Base16 qualified as Base16
+import Hydra.Cardano.Api (
+  ChainPoint (..),
+  HasTypeProxy (..),
+  Hash,
+  NetworkId (..),
+  NetworkMagic (..),
+  SerialiseAsCBOR (..),
+  SlotNo (..),
+  Tx,
+  UTxO,
+  serialiseToRawBytes,
+ )
+import Hydra.Cardano.Api.Prelude (
+  BlockHeader (..),
+ )
+import Hydra.Chain.Direct.Handlers (convertObservation)
+import Hydra.ChainObserver.NodeClient (
+  ChainObservation (..),
+  ChainObserverLog (..),
+  NodeClient (..),
+  ObserverHandler,
+  logOnChainTx,
+  observeAll,
+ )
+import Hydra.Logging (Tracer, traceWith)
+import Hydra.Tx (IsTx (..))
+
+data APIBlockfrostError
+  = BlockfrostError Text
+  | DecodeError Text
+  | NotEnoughBlockConfirmations Blockfrost.BlockHash
+  | MissingBlockNo Blockfrost.BlockHash
+  | MissingNextBlockHash Blockfrost.BlockHash
+  deriving (Show, Exception)
+
+runBlockfrostM ::
+  (MonadIO m, MonadThrow m) =>
+  Blockfrost.Project ->
+  BlockfrostClientT IO a ->
+  m a
+runBlockfrostM prj action = do
+  result <- liftIO $ runBlockfrost prj action
+  case result of
+    Left err -> throwIO (BlockfrostError $ show err)
+    Right val -> pure val
+
+blockfrostClient ::
+  Tracer IO ChainObserverLog ->
+  FilePath ->
+  Integer ->
+  NodeClient IO
+blockfrostClient tracer projectPath blockConfirmations = do
+  NodeClient
+    { follow = \startChainFrom observerHandler -> do
+        prj <- Blockfrost.projectFromFile projectPath
+
+        Blockfrost.Block{_blockHash = (Blockfrost.BlockHash genesisBlockHash)} <-
+          runBlockfrostM prj (Blockfrost.getBlock (Left 0))
+
+        Blockfrost.Genesis
+          { _genesisActiveSlotsCoefficient
+          , _genesisSlotLength
+          , _genesisNetworkMagic
+          } <-
+          runBlockfrostM prj Blockfrost.getLedgerGenesis
+
+        let networkId = fromNetworkMagic _genesisNetworkMagic
+        traceWith tracer ConnectingToExternalNode{networkId}
+
+        chainPoint <-
+          case startChainFrom of
+            Just point -> pure point
+            Nothing -> do
+              toChainPoint <$> runBlockfrostM prj Blockfrost.getLatestBlock
+
+        traceWith tracer StartObservingFrom{chainPoint}
+
+        let blockTime = realToFrac _genesisSlotLength / realToFrac _genesisActiveSlotsCoefficient
+
+        let blockHash = fromChainPoint chainPoint genesisBlockHash
+
+        stateTVar <- newTVarIO (blockHash, mempty)
+        void $
+          retrying (retryPolicy blockTime) shouldRetry $ \_ -> do
+            loop tracer prj networkId blockTime observerHandler blockConfirmations stateTVar
+              `catch` \(ex :: APIBlockfrostError) ->
+                pure $ Left ex
+    }
+ where
+  shouldRetry _ = \case
+    Right{} -> pure False
+    Left err -> pure $ isRetryable err
+
+  retryPolicy blockTime = constantDelay (truncate blockTime * 1000 * 1000)
+
+-- | Iterative process that follows the chain using a naive roll-forward approach,
+-- keeping track of the latest known current block and UTxO view.
+-- This process operates at full speed without waiting between calls,
+-- favoring the catch-up process.
+loop ::
+  (MonadIO m, MonadThrow m, MonadSTM m) =>
+  Tracer m ChainObserverLog ->
+  Blockfrost.Project ->
+  NetworkId ->
+  DiffTime ->
+  ObserverHandler m ->
+  Integer ->
+  TVar m (Blockfrost.BlockHash, UTxO) ->
+  m a
+loop tracer prj networkId blockTime observerHandler blockConfirmations stateTVar = do
+  current <- readTVarIO stateTVar
+  next <- rollForward tracer prj networkId observerHandler blockConfirmations current
+  atomically $ writeTVar stateTVar next
+  loop tracer prj networkId blockTime observerHandler blockConfirmations stateTVar
+
+-- | From the current block and UTxO view, we collect Hydra observations
+-- and yield the next block and adjusted UTxO view.
+rollForward ::
+  (MonadIO m, MonadThrow m) =>
+  Tracer m ChainObserverLog ->
+  Blockfrost.Project ->
+  NetworkId ->
+  ObserverHandler m ->
+  Integer ->
+  (Blockfrost.BlockHash, UTxO) ->
+  m (Blockfrost.BlockHash, UTxO)
+rollForward tracer prj networkId observerHandler blockConfirmations (blockHash, utxo) = do
+  [email protected]
+    { _blockHash
+    , _blockConfirmations
+    , _blockNextBlock
+    , _blockHeight
+    } <-
+    runBlockfrostM prj $ Blockfrost.getBlock (Right blockHash)
+
+  -- Check if block within the safe zone to be processes
+  when (_blockConfirmations < blockConfirmations) $
+    throwIO (NotEnoughBlockConfirmations _blockHash)
+
+  -- Check if block contains a reference to its next
+  nextBlockHash <- maybe (throwIO $ MissingNextBlockHash _blockHash) pure _blockNextBlock
+
+  -- Search block transactions
+  txHashes <- runBlockfrostM prj . Blockfrost.allPages $ \p ->
+    Blockfrost.getBlockTxs' (Right _blockHash) p Blockfrost.def
+
+  -- Collect CBOR representations
+  cborTxs <- traverse (runBlockfrostM prj . Blockfrost.getTxCBOR) txHashes
+
+  -- Convert to cardano-api Tx
+  receivedTxs <- mapM toTx cborTxs
+  let receivedTxIds = txId <$> receivedTxs
+  let point = toChainPoint block
+  traceWith tracer RollForward{point, receivedTxIds}
+
+  -- Collect head observations
+  let (adjustedUTxO, observations) = observeAll networkId utxo receivedTxs
+  let onChainTxs = mapMaybe convertObservation observations
+  forM_ onChainTxs (traceWith tracer . logOnChainTx)
+
+  blockNo <- maybe (throwIO $ MissingBlockNo _blockHash) (pure . fromInteger) _blockHeight
+  let observationsAt = HeadObservation point blockNo <$> onChainTxs
+
+  -- Call observer handler
+  observerHandler $
+    if null observationsAt
+      then [Tick point blockNo]
+      else observationsAt
+
+  -- Next
+  pure (nextBlockHash, adjustedUTxO)
+
+-- * Helpers
+
+isRetryable :: APIBlockfrostError -> Bool
+isRetryable (BlockfrostError _) = True
+isRetryable (DecodeError _) = False
+isRetryable (NotEnoughBlockConfirmations _) = True
+isRetryable (MissingBlockNo _) = True
+isRetryable (MissingNextBlockHash _) = True
+
+toChainPoint :: Blockfrost.Block -> ChainPoint
+toChainPoint Blockfrost.Block{_blockSlot, _blockHash} =
+  ChainPoint slotNo headerHash
+ where
+  slotNo :: SlotNo
+  slotNo = maybe 0 (fromInteger . Blockfrost.unSlot) _blockSlot
+
+  headerHash :: Hash BlockHeader
+  headerHash = fromString . toString $ Blockfrost.unBlockHash _blockHash
+
+fromNetworkMagic :: Integer -> NetworkId
+fromNetworkMagic = \case
+  0 -> Mainnet
+  magicNbr -> Testnet (NetworkMagic (fromInteger magicNbr))
+
+toTx :: MonadThrow m => Blockfrost.TransactionCBOR -> m Tx
+toTx (Blockfrost.TransactionCBOR txCbor) =
+  case decodeBase16 txCbor of
+    Left decodeErr -> throwIO . DecodeError $ "Bad Base16 Tx CBOR: " <> decodeErr
+    Right bytes ->
+      case deserialiseFromCBOR (proxyToAsType (Proxy @Tx)) bytes of
+        Left deserializeErr -> throwIO . DecodeError $ "Bad Tx CBOR: " <> show deserializeErr
+        Right tx -> pure tx
+
+fromChainPoint :: ChainPoint -> Text -> Blockfrost.BlockHash
+fromChainPoint chainPoint genesisBlockHash = case chainPoint of
+  ChainPoint _ headerHash -> Blockfrost.BlockHash (decodeUtf8 . Base16.encode . serialiseToRawBytes $ headerHash)
+  ChainPointAtGenesis -> Blockfrost.BlockHash genesisBlockHash