WIP: Add blockfrost backend for chain following

Signed-off-by: Sasha Bogicevic <[email protected]>

Author: v0d1ch

hydra-node/hydra-node.cabal

@@ -98,6 +98,7 @@ library
   build-depends:
     , aeson
     , base
+    , base16-bytestring
     , blockfrost-client               >=0.9.2.0
     , bytestring
     , cardano-api
@@ -149,6 +150,7 @@ library
     , quickcheck-arbitrary-adt
     , quickcheck-instances
     , resourcet
+    , retry
     , safe-money
     , serialise
     , sop-extras

hydra-node/src/Hydra/Chain/Direct.hs

@@ -9,40 +9,55 @@ module Hydra.Chain.Direct (
 
 import Hydra.Prelude
 
+import Blockfrost.Client qualified as Blockfrost
 import Cardano.Ledger.Shelley.API qualified as Ledger
-import Cardano.Ledger.Slot (EpochInfo)
+import Cardano.Ledger.Slot (EpochInfo, SlotNo)
 import Cardano.Slotting.EpochInfo (hoistEpochInfo)
 import Control.Concurrent.Class.MonadSTM (
   newEmptyTMVar,
   newTQueueIO,
+  newTVarIO,
   putTMVar,
   readTQueue,
+  readTVarIO,
   takeTMVar,
   writeTQueue,
+  writeTVar,
  )
 import Control.Exception (IOException)
 import Control.Monad.Trans.Except (runExcept)
+import Control.Retry (constantDelay, retrying)
+import Data.ByteString.Base16 qualified as Base16
+import Data.Text qualified as T
 import Hydra.Cardano.Api (
+  BlockHeader (..),
   BlockInMode (..),
   CardanoEra (..),
-  ChainPoint,
+  ChainPoint (..),
   ChainTip,
   ConsensusModeParams (..),
   EpochSlots (..),
   EraHistory (EraHistory),
+  Hash,
   IsShelleyBasedEra (..),
   LocalChainSyncClient (..),
   LocalNodeClientProtocols (..),
   LocalNodeConnectInfo (..),
+  NetworkId (..),
+  SerialiseAsCBOR (..),
+  SlotNo (..),
   Tx,
   TxInMode (..),
   TxValidationErrorInCardanoMode,
+  UTxO,
   chainTipToChainPoint,
   connectToLocalNode,
   getBlockHeader,
   getBlockTxs,
   getTxBody,
   getTxId,
+  proxyToAsType,
+  serialiseToRawBytes,
   toLedgerUTxO,
  )
 import Hydra.Chain (
@@ -51,6 +66,7 @@ import Hydra.Chain (
   PostTxError (FailedToPostTx, failureReason),
   currentState,
  )
+import Hydra.Chain.Blockfrost.Client qualified as Blockfrost
 import Hydra.Chain.CardanoClient (
   QueryPoint (..),
  )
@@ -182,7 +198,9 @@ withDirectChain tracer config ctx wallet chainStateHistory callback action = do
               connectToLocalNode
                 (connectInfo networkId nodeSocket)
                 (clientProtocols chainPoint queue handler)
-            BlockfrostBackend{} -> undefined
+            BlockfrostBackend{projectPath} -> do
+              prj <- Blockfrost.projectFromFile projectPath
+              blockfrostChainFollow prj chainPoint handler wallet
       )
       (action chainHandle)
   case res of
@@ -191,6 +209,22 @@ withDirectChain tracer config ctx wallet chainStateHistory callback action = do
  where
   CardanoChainConfig{chainBackend, startChainFrom} = config
 
+  blockfrostChainFollow prj chainPoint handler wallet = do
+    Blockfrost.Genesis{_genesisSlotLength, _genesisActiveSlotsCoefficient} <- Blockfrost.runBlockfrostM prj Blockfrost.getLedgerGenesis
+
+    Blockfrost.Block{_blockHash = (Blockfrost.BlockHash genesisBlockHash)} <-
+      Blockfrost.runBlockfrostM prj (Blockfrost.getBlock (Left 0))
+
+    let blockTime = realToFrac _genesisSlotLength / realToFrac _genesisActiveSlotsCoefficient
+
+    let blockHash = fromChainPoint chainPoint genesisBlockHash
+
+    void $
+      retrying (retryPolicy blockTime) shouldRetry $ \_ -> do
+        loop tracer prj blockTime handler wallet 1 blockHash
+          `catch` \(ex :: APIBlockfrostError) ->
+            pure $ Left ex
+
   connectInfo networkId nodeSocket =
     LocalNodeConnectInfo
       { -- REVIEW: This was 432000 before, but all usages in the
@@ -224,6 +258,117 @@ withDirectChain tracer config ctx wallet chainStateHistory callback action = do
         { ioException
         }
 
+  shouldRetry _ = \case
+    Right{} -> pure False
+    Left err -> pure $ isRetryable err
+
+  retryPolicy blockTime = constantDelay (truncate blockTime * 1000 * 1000)
+
+loop ::
+  (MonadIO m, MonadThrow m, MonadSTM m) =>
+  Tracer IO DirectChainLog ->
+  Blockfrost.Project ->
+  DiffTime ->
+  ChainSyncHandler m ->
+  TinyWallet m ->
+  Integer ->
+  Blockfrost.BlockHash ->
+  m a
+loop tracer prj blockTime handler wallet blockConfirmations current = do
+  next <- rollForward tracer prj handler wallet blockConfirmations current
+  loop tracer prj blockTime handler wallet blockConfirmations next
+
+rollForward ::
+  (MonadIO m, MonadThrow m) =>
+  Tracer IO DirectChainLog ->
+  Blockfrost.Project ->
+  ChainSyncHandler m ->
+  TinyWallet m ->
+  Integer ->
+  Blockfrost.BlockHash ->
+  m Blockfrost.BlockHash
+rollForward tracer prj handler wallet blockConfirmations blockHash = do
+  Blockfrost.Block
+    { _blockHash
+    , _blockConfirmations
+    , _blockNextBlock
+    , _blockHeight
+    , _blockSlot
+    } <-
+    Blockfrost.runBlockfrostM prj $ Blockfrost.getBlock (Right blockHash)
+
+  -- Check if block within the safe zone to be processes
+  when (_blockConfirmations < blockConfirmations) $
+    throwIO (NotEnoughBlockConfirmations _blockHash)
+
+  -- Search block transactions
+  txHashesCBOR <- Blockfrost.runBlockfrostM prj . Blockfrost.allPages $ \p ->
+    Blockfrost.getBlockTxsCBOR' (Right _blockHash) p Blockfrost.def
+
+  -- Convert to cardano-api Tx
+  receivedTxs <- mapM (toTx . (\(Blockfrost.TxHashCBOR (_txHash, cbor)) -> cbor)) txHashesCBOR
+
+  -- Check if block contains a reference to its next
+  nextBlockHash <- maybe (throwIO $ MissingNextBlockHash _blockHash) pure _blockNextBlock
+
+  blockNo <- maybe (throwIO $ MissingBlockNo _blockHash) (pure . fromInteger) _blockHeight
+  let Blockfrost.BlockHash blockHash' = _blockHash
+  let blockHash'' = fromString $ T.unpack blockHash'
+  blockSlot <- maybe (throwIO $ MissingBlockSlot _blockSlot) (pure . fromInteger . Blockfrost.unSlot) _blockSlot
+  let header = BlockHeader (SlotNo blockSlot) blockHash'' blockNo
+  update wallet header receivedTxs
+  -- Observe Hydra transactions
+  onRollForward handler header receivedTxs
+
+  pure nextBlockHash
+
+-- * Helpers
+
+data APIBlockfrostError
+  = BlockfrostError Text
+  | DecodeError Text
+  | NotEnoughBlockConfirmations Blockfrost.BlockHash
+  | MissingBlockNo Blockfrost.BlockHash
+  | MissingBlockSlot (Maybe Blockfrost.Slot)
+  | MissingNextBlockHash Blockfrost.BlockHash
+  deriving (Show, Exception)
+
+isRetryable :: APIBlockfrostError -> Bool
+isRetryable (BlockfrostError _) = True
+isRetryable (DecodeError _) = False
+isRetryable (NotEnoughBlockConfirmations _) = True
+isRetryable (MissingBlockNo _) = True
+isRetryable (MissingBlockSlot _) = 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
+
 newtype ConnectException = ConnectException
   { ioException :: IOException
   }