haskoin-store 0.20.2 → 0.21.0
raw patch · 20 files changed
+3481/−2409 lines, 20 filesdep +hedis
Dependencies added: hedis
Files
- CHANGELOG.md +10/−0
- app/Main.hs +2/−0
- haskoin-store.cabal +20/−12
- src/Haskoin/Store.hs +72/−134
- src/Network/Haskoin/Store/Block.hs +0/−511
- src/Network/Haskoin/Store/BlockStore.hs +535/−0
- src/Network/Haskoin/Store/CacheWriter.hs +563/−0
- src/Network/Haskoin/Store/Common.hs +57/−74
- src/Network/Haskoin/Store/Data/CacheReader.hs +385/−0
- src/Network/Haskoin/Store/Data/DatabaseReader.hs +257/−0
- src/Network/Haskoin/Store/Data/DatabaseWriter.hs +367/−0
- src/Network/Haskoin/Store/Data/ImportDB.hs +0/−356
- src/Network/Haskoin/Store/Data/KeyValue.hs +0/−360
- src/Network/Haskoin/Store/Data/Memory.hs +0/−392
- src/Network/Haskoin/Store/Data/MemoryDatabase.hs +391/−0
- src/Network/Haskoin/Store/Data/RocksDB.hs +0/−257
- src/Network/Haskoin/Store/Data/Types.hs +360/−0
- src/Network/Haskoin/Store/Web.hs +394/−310
- test/Haskoin/StoreSpec.hs +8/−3
- test/Network/Haskoin/Store/Data/CacheReaderSpec.hs +60/−0
CHANGELOG.md view
@@ -4,6 +4,16 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +## 0.21.0+### Fixed+- Fix output of web API calls when issuing limits with offsets.++### Changed+- Massive refactoring of entire codebase.++### Added+- Work in progress Redis caching for extended public keys.+ ## 0.20.2 ### Changed - Filter xpub address balances on web API to show only addresses that have been used.
app/Main.hs view
@@ -210,6 +210,7 @@ , storeConfDB = db , storeConfNetwork = net , storeConfListen = (`sendSTM` pub) . Event+ , storeConfCache = Nothing } in withStore scfg $ \st -> let wcfg =@@ -222,6 +223,7 @@ , webMaxLimits = limits , webReqLog = reqlog , webTimeouts = tos+ , webCache = Nothing } in runWeb wcfg where
haskoin-store.cabal view
@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: fab6c7b80d147905eae85dd49155972d155bc89744e0be0dd5c3d8fc9dabb4e2+-- hash: 289e2a7eef944ded88cf04d3614cbf5ded748193795614fd9315b8cd4e72fcb3 name: haskoin-store-version: 0.20.2+version: 0.21.0 synopsis: Storage and index for Bitcoin and Bitcoin Cash description: Store blocks, transactions, and balances for Bitcoin or Bitcoin Cash, and make that information via REST API. category: Bitcoin, Finance, Network@@ -31,12 +31,14 @@ Haskoin.Store Paths_haskoin_store other-modules:- Network.Haskoin.Store.Block+ Network.Haskoin.Store.BlockStore+ Network.Haskoin.Store.CacheWriter Network.Haskoin.Store.Common- Network.Haskoin.Store.Data.ImportDB- Network.Haskoin.Store.Data.KeyValue- Network.Haskoin.Store.Data.Memory- Network.Haskoin.Store.Data.RocksDB+ Network.Haskoin.Store.Data.CacheReader+ Network.Haskoin.Store.Data.DatabaseReader+ Network.Haskoin.Store.Data.DatabaseWriter+ Network.Haskoin.Store.Data.MemoryDatabase+ Network.Haskoin.Store.Data.Types Network.Haskoin.Store.Logic Network.Haskoin.Store.Web autogen-modules:@@ -56,6 +58,7 @@ , hashable , haskoin-core >=0.10.1 , haskoin-node >=0.9.21+ , hedis , http-types , monad-logger , mtl@@ -98,6 +101,7 @@ , haskoin-core >=0.10.1 , haskoin-node >=0.9.21 , haskoin-store+ , hedis , http-types , monad-logger , mtl@@ -125,13 +129,16 @@ other-modules: Haskoin.StoreSpec Network.Haskoin.Store.CommonSpec+ Network.Haskoin.Store.Data.CacheReaderSpec Haskoin.Store- Network.Haskoin.Store.Block+ Network.Haskoin.Store.BlockStore+ Network.Haskoin.Store.CacheWriter Network.Haskoin.Store.Common- Network.Haskoin.Store.Data.ImportDB- Network.Haskoin.Store.Data.KeyValue- Network.Haskoin.Store.Data.Memory- Network.Haskoin.Store.Data.RocksDB+ Network.Haskoin.Store.Data.CacheReader+ Network.Haskoin.Store.Data.DatabaseReader+ Network.Haskoin.Store.Data.DatabaseWriter+ Network.Haskoin.Store.Data.MemoryDatabase+ Network.Haskoin.Store.Data.Types Network.Haskoin.Store.Logic Network.Haskoin.Store.Web Paths_haskoin_store@@ -152,6 +159,7 @@ , hashable , haskoin-core >=0.10.1 , haskoin-node >=0.9.21+ , hedis , hspec , http-types , monad-logger
src/Haskoin/Store.hs view
@@ -1,137 +1,73 @@ module Haskoin.Store- ( Store(..)- , BlockStore- , StoreConfig(..)- , StoreRead(..)- , StoreWrite(..)- , StoreEvent(..)- , BlockData(..)- , Transaction(..)- , Spender(..)- , BlockRef(..)- , Unspent(..)- , BlockTx(..)- , XPubBal(..)- , XPubUnspent(..)- , Balance(..)- , PeerInformation(..)- , HealthCheck(..)- , PubExcept(..)- , Event(..)- , TxAfterHeight(..)- , JsonSerial(..)- , BinSerial(..)- , Except(..)- , TxId(..)- , UnixTime- , BlockPos- , BlockDB(..)- , WebConfig(..)- , MaxLimits(..)- , Timeouts(..)- , Offset- , Limit- , withStore- , runWeb+ ( StoreConfig (..) , store- , getTransaction- , fromTransaction- , toTransaction- , transactionData- , getAddressUnspentsLimit- , getAddressesUnspentsLimit- , getAddressTxsFull- , getAddressTxsLimit- , getAddressesTxsFull- , getAddressesTxsLimit- , getPeersInformation- , publishTx- , isCoinbase- , confirmed- , cbAfterHeight- , healthCheck- , withBlockMem- , withRocksDB- , connectRocksDB- , initRocksDB- , zeroBalance- , nullBalance- ) where+ , withStore+ , module X )+ where -import Control.Monad (unless, when)-import Control.Monad.Logger (MonadLoggerIO)-import Data.Serialize (decode)-import Haskoin (BlockHash (..), Inv (..),- InvType (..),- InvVector (..),- Message (..),- MessageCommand (..),- NetworkAddress (..),- NotFound (..), Pong (..),- Reject (..), TxHash (..),- VarString (..),- sockToHostAddress)-import Haskoin.Node (ChainEvent (..),- ChainMessage,- ManagerMessage,- NodeConfig (..),- NodeEvent (..),- PeerEvent (..), node)-import Network.Haskoin.Store.Block (blockStore)-import Network.Haskoin.Store.Common (Balance (..),- BinSerial (..),- BlockConfig (..),- BlockDB (..),- BlockData (..),- BlockMessage (..),- BlockPos, BlockRef (..),- BlockStore, BlockTx (..),- Event (..),- HealthCheck (..),- JsonSerial (..), Limit,- Offset,- PeerInformation (..),- PubExcept (..),- Spender (..), Store (..),- StoreConfig (..),- StoreEvent (..),- StoreRead (..),- StoreWrite (..),- Transaction (..),- TxAfterHeight (..),- TxId (..), UnixTime,- Unspent (..), XPubBal (..),- XPubUnspent (..),- confirmed, fromTransaction,- getTransaction, isCoinbase,- nullBalance, toTransaction,- transactionData,- zeroBalance)-import Network.Haskoin.Store.Data.Memory (withBlockMem)-import Network.Haskoin.Store.Data.RocksDB (connectRocksDB,- initRocksDB, withRocksDB)-import Network.Haskoin.Store.Web (Except (..),- MaxLimits (..),- Timeouts (..),- WebConfig (..),- cbAfterHeight,- getAddressTxsFull,- getAddressTxsLimit,- getAddressUnspentsLimit,- getAddressesTxsFull,- getAddressesTxsLimit,- getAddressesUnspentsLimit,- getPeersInformation,- healthCheck, publishTx,- runWeb)-import Network.Socket (SockAddr (..))-import NQE (Inbox, Listen,- Process (..),- inboxToMailbox, newInbox,- sendSTM, withProcess)-import UnliftIO (MonadUnliftIO, link,- withAsync)+import Control.Monad (unless, when)+import Control.Monad.Logger (MonadLoggerIO)+import Data.Serialize (decode)+import Haskoin (BlockHash (..),+ Inv (..),+ InvType (..),+ InvVector (..),+ Message (..),+ MessageCommand (..),+ Network,+ NetworkAddress (..),+ NotFound (..),+ Pong (..),+ Reject (..),+ TxHash (..),+ VarString (..),+ sockToHostAddress)+import Haskoin.Node (ChainEvent (..),+ ChainMessage,+ HostPort,+ ManagerMessage,+ NodeConfig (..),+ NodeEvent (..),+ PeerEvent (..),+ node)+import Network.Haskoin.Store.BlockStore as X+import Network.Haskoin.Store.CacheWriter as X+import Network.Haskoin.Store.Common as X+import Network.Haskoin.Store.Data.CacheReader as X+import Network.Haskoin.Store.Data.DatabaseReader as X+import Network.Haskoin.Store.Data.DatabaseWriter as X+import Network.Haskoin.Store.Data.MemoryDatabase as X+import Network.Haskoin.Store.Data.Types as X+import Network.Haskoin.Store.Logic as X+import Network.Haskoin.Store.Web as X+import Network.Socket (SockAddr (..))+import NQE (Inbox, Listen,+ Process (..),+ inboxToMailbox,+ newInbox, sendSTM,+ withProcess)+import UnliftIO (MonadUnliftIO, link,+ withAsync) +-- | Configuration for a 'Store'.+data StoreConfig =+ StoreConfig+ { storeConfMaxPeers :: !Int+ -- ^ max peers to connect to+ , storeConfInitPeers :: ![HostPort]+ -- ^ static set of peers to connect to+ , storeConfDiscover :: !Bool+ -- ^ discover new peers?+ , storeConfDB :: !DatabaseReader+ -- ^ RocksDB database handler+ , storeConfNetwork :: !Network+ -- ^ network constants+ , storeConfListen :: !(Listen StoreEvent)+ -- ^ listen to store events+ , storeConfCache :: !(Maybe CacheReaderConfig)+ -- ^ Redis cache configuration+ }+ withStore :: (MonadLoggerIO m, MonadUnliftIO m) => StoreConfig@@ -156,13 +92,13 @@ => StoreConfig -> Inbox ManagerMessage -> Inbox ChainMessage- -> Inbox BlockMessage+ -> Inbox BlockStoreMessage -> m () store cfg mgri chi bsi = do let ncfg = NodeConfig { nodeConfMaxPeers = storeConfMaxPeers cfg- , nodeConfDB = blockDB (storeConfDB cfg)+ , nodeConfDB = databaseHandle (storeConfDB cfg) , nodeConfPeers = storeConfInitPeers cfg , nodeConfDiscover = storeConfDiscover cfg , nodeConfEvents = storeDispatch b l@@ -174,14 +110,16 @@ withAsync (node ncfg mgri chi) $ \a -> do link a let bcfg =- BlockConfig+ BlockStoreConfig { blockConfChain = inboxToMailbox chi , blockConfManager = inboxToMailbox mgri , blockConfListener = l , blockConfDB = storeConfDB cfg , blockConfNet = storeConfNetwork cfg }- blockStore bcfg bsi+ case storeConfCache cfg of+ Nothing -> blockStore bcfg bsi+ Just cacheconf -> undefined where l = storeConfListen cfg b = inboxToMailbox bsi
− src/Network/Haskoin/Store/Block.hs
@@ -1,511 +0,0 @@-{-# LANGUAGE DeriveAnyClass #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE TemplateHaskell #-}-{-# LANGUAGE TupleSections #-}-module Network.Haskoin.Store.Block- ( blockStore- ) where--import Control.Applicative ((<|>))-import Control.Monad (forM, forM_, forever,- guard, mzero, unless,- void, when)-import Control.Monad.Except (ExceptT, runExceptT)-import Control.Monad.Logger (MonadLoggerIO, logDebugS,- logErrorS, logInfoS,- logWarnS)-import Control.Monad.Reader (MonadReader, ReaderT (..),- asks)-import Control.Monad.Trans (lift)-import Control.Monad.Trans.Maybe (MaybeT (MaybeT),- runMaybeT)-import Data.Maybe (catMaybes, isNothing,- listToMaybe)-import Data.String (fromString)-import Data.String.Conversions (cs)-import Data.Time.Clock.System (getSystemTime,- systemSeconds)-import Haskoin (Block (..),- BlockHash (..),- BlockHeight,- BlockNode (..),- GetData (..),- InvType (..),- InvVector (..),- Message (..),- Network (..), Tx,- TxHash (..),- blockHashToHex,- headerHash, txHash,- txHashToHex)-import Haskoin.Node (OnlinePeer (..), Peer,- PeerException (..),- chainBlockMain,- chainGetAncestor,- chainGetBest,- chainGetBlock,- chainGetParents, killPeer,- managerGetPeers,- sendMessage)-import Network.Haskoin.Store.Common (BlockConfig (..), BlockDB,- BlockMessage (..),- BlockStore,- StoreEvent (..),- StoreRead (..),- StoreWrite (..), UnixTime)-import Network.Haskoin.Store.Data.ImportDB (ImportDB, runImportDB)-import Network.Haskoin.Store.Logic (ImportException, deleteTx,- getOldMempool,- getOldOrphans,- importBlock, importOrphan,- initBest, newMempoolTx,- revertBlock)-import NQE (Inbox, Mailbox,- inboxToMailbox, query,- receive)-import System.Random (randomRIO)-import UnliftIO (Exception, MonadIO,- MonadUnliftIO, TVar,- atomically, liftIO,- newTVarIO, readTVarIO,- throwIO, withAsync,- writeTVar)-import UnliftIO.Concurrent (threadDelay)--data BlockException- = BlockNotInChain !BlockHash- | Uninitialized- | AncestorNotInChain !BlockHeight- !BlockHash- deriving (Show, Eq, Ord, Exception)--data Syncing = Syncing- { syncingPeer :: !Peer- , syncingTime :: !UnixTime- , syncingHead :: !BlockNode- }---- | Block store process state.-data BlockRead = BlockRead- { mySelf :: !BlockStore- , myConfig :: !BlockConfig- , myPeer :: !(TVar (Maybe Syncing))- }--type BlockT m = ReaderT BlockRead m--runImport ::- MonadLoggerIO m- => ReaderT ImportDB (ExceptT ImportException m) a- -> ReaderT BlockRead m (Either ImportException a)-runImport f =- ReaderT $ \r -> runExceptT (runImportDB (blockConfDB (myConfig r)) f)--runRocksDB :: ReaderT BlockDB m a -> ReaderT BlockRead m a-runRocksDB f =- ReaderT $ \BlockRead {myConfig = BlockConfig {blockConfDB = db}} ->- runReaderT f db--instance MonadIO m => StoreRead (ReaderT BlockRead m) where- getBestBlock = runRocksDB getBestBlock- getBlocksAtHeight = runRocksDB . getBlocksAtHeight- getBlock = runRocksDB . getBlock- getTxData = runRocksDB . getTxData- getSpender = runRocksDB . getSpender- getSpenders = runRocksDB . getSpenders- getOrphanTx = runRocksDB . getOrphanTx- getUnspent = runRocksDB . getUnspent- getBalance = runRocksDB . getBalance- getMempool = runRocksDB getMempool- getAddressesTxs addrs start limit =- runRocksDB (getAddressesTxs addrs start limit)- getAddressesUnspents addrs start limit =- runRocksDB (getAddressesUnspents addrs start limit)- getOrphans = runRocksDB getOrphans- getAddressUnspents a s = runRocksDB . getAddressUnspents a s- getAddressTxs a s = runRocksDB . getAddressTxs a s---- | Run block store process.-blockStore ::- (MonadUnliftIO m, MonadLoggerIO m)- => BlockConfig- -> Inbox BlockMessage- -> m ()-blockStore cfg inbox = do- pb <- newTVarIO Nothing- runReaderT- (ini >> run)- BlockRead {mySelf = inboxToMailbox inbox, myConfig = cfg, myPeer = pb}- where- ini = do- net <- asks (blockConfNet . myConfig)- runImport (initBest net) >>= \case- Left e -> do- $(logErrorS) "Block" $- "Could not initialize block store: " <> fromString (show e)- throwIO e- Right () -> return ()- run =- withAsync (pingMe (inboxToMailbox inbox)) . const . forever $ do- receive inbox >>= \x ->- ReaderT $ \r -> runReaderT (processBlockMessage x) r--isInSync ::- (MonadLoggerIO m, StoreRead m, MonadReader BlockRead m)- => m Bool-isInSync =- getBestBlock >>= \case- Nothing -> do- $(logErrorS) "Block" "Block database uninitialized"- throwIO Uninitialized- Just bb ->- asks (blockConfChain . myConfig) >>= chainGetBest >>= \cb ->- return (headerHash (nodeHeader cb) == bb)--mempool :: MonadLoggerIO m => Peer -> m ()-mempool p = do- $(logDebugS) "Block" "Requesting mempool from network peer"- MMempool `sendMessage` p--processBlock ::- (MonadUnliftIO m, MonadLoggerIO m)- => Peer- -> Block- -> ReaderT BlockRead m ()-processBlock peer block = do- void . runMaybeT $ do- checkpeer- blocknode <- getblocknode- net <- asks (blockConfNet . myConfig)- lift (runImport (importBlock net block blocknode)) >>= \case- Right deletedtxids -> do- listener <- asks (blockConfListener . myConfig)- $(logInfoS) "Block" $ "Best block indexed: " <> hexhash- atomically $ do- mapM_ (listener . StoreTxDeleted) deletedtxids- listener (StoreBestBlock blockhash)- lift (syncMe peer)- Left e -> do- $(logErrorS) "Block" $- "Error importing block: " <> hexhash <> ": " <>- fromString (show e)- killPeer (PeerMisbehaving (show e)) peer- where- header = blockHeader block- blockhash = headerHash header- hexhash = blockHashToHex blockhash- checkpeer =- getSyncingState >>= \case- Just Syncing {syncingPeer = syncingpeer}- | peer == syncingpeer -> return ()- _ -> do- $(logErrorS) "Block" $ "Peer sent unexpected block: " <> hexhash- killPeer (PeerMisbehaving "Sent unpexpected block") peer- mzero- getblocknode =- asks (blockConfChain . myConfig) >>= chainGetBlock blockhash >>= \case- Nothing -> do- $(logErrorS) "Block" $ "Block header not found: " <> hexhash- killPeer (PeerMisbehaving "Sent unknown block") peer- mzero- Just n -> return n--processNoBlocks ::- (MonadUnliftIO m, MonadLoggerIO m)- => Peer- -> [BlockHash]- -> ReaderT BlockRead m ()-processNoBlocks p _bs = do- $(logErrorS) "Block" (cs m)- killPeer (PeerMisbehaving m) p- where- m = "I do not like peers that cannot find them blocks"--processTx :: (MonadUnliftIO m, MonadLoggerIO m) => Peer -> Tx -> BlockT m ()-processTx _p tx =- isInSync >>= \sync ->- when sync $ do- now <- fromIntegral . systemSeconds <$> liftIO getSystemTime- net <- asks (blockConfNet . myConfig)- runImport (newMempoolTx net tx now) >>= \case- Right (Just deleted) -> do- l <- blockConfListener <$> asks myConfig- $(logInfoS) "Block" $- "New mempool tx: " <> txHashToHex (txHash tx)- atomically $ do- mapM_ (l . StoreTxDeleted) deleted- l (StoreMempoolNew (txHash tx))- _ -> return ()--processOrphans ::- (MonadUnliftIO m, MonadLoggerIO m) => BlockT m ()-processOrphans =- isInSync >>= \sync ->- when sync $ do- now <- fromIntegral . systemSeconds <$> liftIO getSystemTime- net <- asks (blockConfNet . myConfig)- old <- getOldOrphans now- case old of- [] -> return ()- _ -> do- $(logInfoS) "Block" $- "Removing " <> cs (show (length old)) <>- " expired orphan transactions"- void . runImport $ mapM_ deleteOrphanTx old- orphans <- getOrphans- case orphans of- [] -> return ()- _ ->- $(logInfoS) "Block" $- "Attempting to import " <> cs (show (length orphans)) <>- " orphan transactions"- ops <-- zip (map snd orphans) <$>- mapM (runImport . uncurry (importOrphan net)) orphans- let tths =- [ (txHash tx, hs)- | (tx, emths) <- ops- , let Right (Just hs) = emths- ]- ihs = map fst tths- dhs = concatMap snd tths- l <- blockConfListener <$> asks myConfig- atomically $ do- mapM_ (l . StoreTxDeleted) dhs- mapM_ (l . StoreMempoolNew) ihs---processTxs ::- (MonadUnliftIO m, MonadLoggerIO m)- => Peer- -> [TxHash]- -> ReaderT BlockRead m ()-processTxs p hs =- isInSync >>= \sync ->- when sync $ do- xs <-- fmap catMaybes . forM hs $ \h ->- runMaybeT $ do- t <- lift $ getTxData h- guard (isNothing t)- return (getTxHash h)- unless (null xs) $ do- $(logInfoS) "Block" $- "Requesting " <> fromString (show (length xs)) <>- " new transactions"- net <- blockConfNet <$> asks myConfig- let inv =- if getSegWit net- then InvWitnessTx- else InvTx- MGetData (GetData (map (InvVector inv) xs)) `sendMessage` p--checkTime :: (MonadUnliftIO m, MonadLoggerIO m) => ReaderT BlockRead m ()-checkTime =- asks myPeer >>= readTVarIO >>= \case- Nothing -> return ()- Just Syncing {syncingTime = t, syncingPeer = p} -> do- n <- fromIntegral . systemSeconds <$> liftIO getSystemTime- when (n > t + 60) $ do- $(logErrorS) "Block" "Syncing peer timeout"- resetPeer- killPeer PeerTimeout p--processDisconnect ::- (MonadUnliftIO m, MonadLoggerIO m)- => Peer- -> ReaderT BlockRead m ()-processDisconnect p =- asks myPeer >>= readTVarIO >>= \case- Nothing -> return ()- Just Syncing {syncingPeer = p'}- | p == p' -> do- resetPeer- getPeer >>= \case- Nothing ->- $(logWarnS)- "Block"- "No peers available after syncing peer disconnected"- Just peer -> do- $(logWarnS) "Block" "Selected another peer to sync"- syncMe peer- | otherwise -> return ()--pruneMempool :: (MonadUnliftIO m, MonadLoggerIO m) => BlockT m ()-pruneMempool =- isInSync >>= \sync ->- when sync $ do- now <- fromIntegral . systemSeconds <$> liftIO getSystemTime- getOldMempool now >>= \case- [] -> return ()- old -> deletetxs old- where- deletetxs old = do- $(logInfoS) "Block" $- "Removing " <> cs (show (length old)) <> " old mempool transactions"- net <- asks (blockConfNet . myConfig)- forM_ old $ \txid ->- runImport (deleteTx net True txid) >>= \case- Left _ -> return ()- Right txids -> do- listener <- asks (blockConfListener . myConfig)- atomically $ mapM_ (listener . StoreTxDeleted) txids--syncMe :: (MonadUnliftIO m, MonadLoggerIO m) => Peer -> BlockT m ()-syncMe peer =- void . runMaybeT $ do- checksyncingpeer- reverttomainchain- syncbest <- syncbestnode- bestblock <- bestblocknode- chainbest <- chainbestnode- end syncbest bestblock chainbest- blocknodes <- selectblocks chainbest syncbest- setPeer peer (last blocknodes)- net <- asks (blockConfNet . myConfig)- let inv =- if getSegWit net- then InvWitnessBlock- else InvBlock- vectors =- map- (InvVector inv . getBlockHash . headerHash . nodeHeader)- blocknodes- $(logInfoS) "Block" $- "Requesting " <> fromString (show (length vectors)) <> " blocks"- MGetData (GetData vectors) `sendMessage` peer- where- checksyncingpeer =- getSyncingState >>= \case- Nothing -> return ()- Just Syncing {syncingPeer = p}- | p == peer -> return ()- | otherwise -> do- $(logInfoS) "Block" "Already syncing against another peer"- mzero- chainbestnode = chainGetBest =<< asks (blockConfChain . myConfig)- bestblocknode = do- bb <-- lift getBestBlock >>= \case- Nothing -> do- $(logErrorS) "Block" "No best block set"- throwIO Uninitialized- Just b -> return b- ch <- asks (blockConfChain . myConfig)- chainGetBlock bb ch >>= \case- Nothing -> do- $(logErrorS) "Block" $- "Header not found for best block: " <> blockHashToHex bb- throwIO (BlockNotInChain bb)- Just x -> return x- syncbestnode =- asks myPeer >>= readTVarIO >>= \case- Just Syncing {syncingHead = b} -> return b- Nothing -> bestblocknode- end syncbest bestblock chainbest- | nodeHeader bestblock == nodeHeader chainbest = do- resetPeer >> mempool peer >> mzero- | nodeHeader syncbest == nodeHeader chainbest = do mzero- | otherwise =- when (nodeHeight syncbest > nodeHeight bestblock + 500) mzero- selectblocks chainbest syncbest = do- synctop <-- top- chainbest- (maxsyncheight (nodeHeight chainbest) (nodeHeight syncbest))- ch <- asks (blockConfChain . myConfig)- parents <- chainGetParents (nodeHeight syncbest + 1) synctop ch- return $- if length parents < 500- then parents <> [chainbest]- else parents- maxsyncheight chainheight syncbestheight- | chainheight <= syncbestheight + 501 = chainheight- | otherwise = syncbestheight + 501- top chainbest syncheight = do- ch <- asks (blockConfChain . myConfig)- if syncheight == nodeHeight chainbest- then return chainbest- else chainGetAncestor syncheight chainbest ch >>= \case- Just x -> return x- Nothing -> do- $(logErrorS) "Block" $- "Could not find header for ancestor of block: " <>- blockHashToHex (headerHash (nodeHeader chainbest))- throwIO $- AncestorNotInChain- syncheight- (headerHash (nodeHeader chainbest))- reverttomainchain = do- bestblockhash <- headerHash . nodeHeader <$> bestblocknode- ch <- asks (blockConfChain . myConfig)- chainBlockMain bestblockhash ch >>= \y ->- unless y $ do- $(logErrorS) "Block" $- "Reverting best block: " <> blockHashToHex bestblockhash- resetPeer- net <- asks (blockConfNet . myConfig)- lift (runImport (revertBlock net bestblockhash)) >>= \case- Left e -> do- $(logErrorS) "Block" $- "Could not revert best block: " <> cs (show e)- throwIO e- Right txids -> do- listener <- asks (blockConfListener . myConfig)- atomically $ do- mapM_ (listener . StoreTxDeleted) txids- listener (StoreBlockReverted bestblockhash)- reverttomainchain--resetPeer :: (MonadLoggerIO m, MonadReader BlockRead m) => m ()-resetPeer = do- box <- asks myPeer- atomically $ writeTVar box Nothing--setPeer :: (MonadIO m, MonadReader BlockRead m) => Peer -> BlockNode -> m ()-setPeer p b = do- box <- asks myPeer- now <- fromIntegral . systemSeconds <$> liftIO getSystemTime- atomically . writeTVar box $- Just Syncing {syncingPeer = p, syncingHead = b, syncingTime = now}--getPeer :: (MonadIO m, MonadReader BlockRead m) => m (Maybe Peer)-getPeer = runMaybeT $ MaybeT syncingpeer <|> MaybeT onlinepeer- where- syncingpeer = fmap syncingPeer <$> getSyncingState- onlinepeer =- listToMaybe . map onlinePeerMailbox <$>- (managerGetPeers =<< asks (blockConfManager . myConfig))--getSyncingState :: (MonadIO m, MonadReader BlockRead m) => m (Maybe Syncing)-getSyncingState = readTVarIO =<< asks myPeer--processBlockMessage ::- (MonadUnliftIO m, MonadLoggerIO m)- => BlockMessage- -> BlockT m ()-processBlockMessage (BlockNewBest _) = do- getPeer >>= \case- Nothing -> do- $(logErrorS) "Block" "New best block but no peer to sync from"- Just p -> syncMe p-processBlockMessage (BlockPeerConnect p _) = syncMe p-processBlockMessage (BlockPeerDisconnect p _sa) = processDisconnect p-processBlockMessage (BlockReceived p b) = processBlock p b-processBlockMessage (BlockNotFound p bs) = processNoBlocks p bs-processBlockMessage (BlockTxReceived p tx) = processTx p tx-processBlockMessage (BlockTxAvailable p ts) = processTxs p ts-processBlockMessage (BlockPing r) = do- processOrphans- checkTime- pruneMempool- atomically (r ())--pingMe :: MonadLoggerIO m => Mailbox BlockMessage -> m ()-pingMe mbox =- forever $ do- threadDelay =<< liftIO (randomRIO (5 * 1000 * 1000, 10 * 1000 * 1000))- BlockPing `query` mbox
+ src/Network/Haskoin/Store/BlockStore.hs view
@@ -0,0 +1,535 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TupleSections #-}+module Network.Haskoin.Store.BlockStore where++import Control.Applicative ((<|>))+import Control.Monad (forM, forM_,+ forever, guard,+ mzero, unless, void,+ when)+import Control.Monad.Except (ExceptT, runExceptT)+import Control.Monad.Logger (MonadLoggerIO,+ logDebugS,+ logErrorS, logInfoS,+ logWarnS)+import Control.Monad.Reader (MonadReader,+ ReaderT (..), asks)+import Control.Monad.Trans (lift)+import Control.Monad.Trans.Maybe (MaybeT (MaybeT),+ runMaybeT)+import Data.Maybe (catMaybes,+ isNothing,+ listToMaybe)+import Data.String (fromString)+import Data.String.Conversions (cs)+import Data.Time.Clock.System (getSystemTime,+ systemSeconds)+import Haskoin (Block (..),+ BlockHash (..),+ BlockHeight,+ BlockNode (..),+ GetData (..),+ InvType (..),+ InvVector (..),+ Message (..),+ Network (..), Tx,+ TxHash (..),+ blockHashToHex,+ headerHash, txHash,+ txHashToHex)+import Haskoin.Node (OnlinePeer (..),+ Peer,+ PeerException (..),+ chainBlockMain,+ chainGetAncestor,+ chainGetBest,+ chainGetBlock,+ chainGetParents,+ killPeer,+ managerGetPeers,+ sendMessage)+import Haskoin.Node (Chain, Manager)+import Network.Haskoin.Store.Common (BlockStore, BlockStoreMessage (..),+ StoreEvent (..),+ StoreRead (..),+ StoreWrite (..),+ UnixTime)+import Network.Haskoin.Store.Data.DatabaseReader (DatabaseReader)+import Network.Haskoin.Store.Data.DatabaseWriter (DatabaseWriter,+ runDatabaseWriter)+import Network.Haskoin.Store.Logic (ImportException,+ deleteTx,+ getOldMempool,+ getOldOrphans,+ importBlock,+ importOrphan,+ initBest,+ newMempoolTx,+ revertBlock)+import NQE (Inbox, Listen,+ inboxToMailbox,+ query, receive)+import System.Random (randomRIO)+import UnliftIO (Exception, MonadIO,+ MonadUnliftIO, TVar,+ atomically, liftIO,+ newTVarIO,+ readTVarIO, throwIO,+ withAsync,+ writeTVar)+import UnliftIO.Concurrent (threadDelay)++data BlockException+ = BlockNotInChain !BlockHash+ | Uninitialized+ | AncestorNotInChain !BlockHeight+ !BlockHash+ deriving (Show, Eq, Ord, Exception)++data Syncing = Syncing+ { syncingPeer :: !Peer+ , syncingTime :: !UnixTime+ , syncingHead :: !BlockNode+ }++-- | Block store process state.+data BlockRead = BlockRead+ { mySelf :: !BlockStore+ , myConfig :: !BlockStoreConfig+ , myPeer :: !(TVar (Maybe Syncing))+ }++-- | Configuration for a block store.+data BlockStoreConfig =+ BlockStoreConfig+ { blockConfManager :: !Manager+ -- ^ peer manager from running node+ , blockConfChain :: !Chain+ -- ^ chain from a running node+ , blockConfListener :: !(Listen StoreEvent)+ -- ^ listener for store events+ , blockConfDB :: !DatabaseReader+ -- ^ RocksDB database handle+ , blockConfNet :: !Network+ -- ^ network constants+ }++type BlockT m = ReaderT BlockRead m++runImport ::+ MonadLoggerIO m+ => ReaderT DatabaseWriter (ExceptT ImportException m) a+ -> ReaderT BlockRead m (Either ImportException a)+runImport f =+ ReaderT $ \r -> runExceptT (runDatabaseWriter (blockConfDB (myConfig r)) f)++runRocksDB :: ReaderT DatabaseReader m a -> ReaderT BlockRead m a+runRocksDB f =+ ReaderT $ \BlockRead {myConfig = BlockStoreConfig {blockConfDB = db}} ->+ runReaderT f db++instance MonadIO m => StoreRead (ReaderT BlockRead m) where+ getBestBlock = runRocksDB getBestBlock+ getBlocksAtHeight = runRocksDB . getBlocksAtHeight+ getBlock = runRocksDB . getBlock+ getTxData = runRocksDB . getTxData+ getSpender = runRocksDB . getSpender+ getSpenders = runRocksDB . getSpenders+ getOrphanTx = runRocksDB . getOrphanTx+ getUnspent = runRocksDB . getUnspent+ getBalance = runRocksDB . getBalance+ getMempool = runRocksDB getMempool+ getAddressesTxs addrs start limit =+ runRocksDB (getAddressesTxs addrs start limit)+ getAddressesUnspents addrs start limit =+ runRocksDB (getAddressesUnspents addrs start limit)+ getOrphans = runRocksDB getOrphans+ getAddressUnspents a s = runRocksDB . getAddressUnspents a s+ getAddressTxs a s = runRocksDB . getAddressTxs a s++-- | Run block store process.+blockStore ::+ (MonadUnliftIO m, MonadLoggerIO m)+ => BlockStoreConfig+ -> Inbox BlockStoreMessage+ -> m ()+blockStore cfg inbox = do+ pb <- newTVarIO Nothing+ runReaderT+ (ini >> run)+ BlockRead {mySelf = inboxToMailbox inbox, myConfig = cfg, myPeer = pb}+ where+ ini = do+ net <- asks (blockConfNet . myConfig)+ runImport (initBest net) >>= \case+ Left e -> do+ $(logErrorS) "Block" $+ "Could not initialize block store: " <> fromString (show e)+ throwIO e+ Right () -> return ()+ run =+ withAsync (pingMe (inboxToMailbox inbox)) . const . forever $ do+ receive inbox >>= \x ->+ ReaderT $ \r -> runReaderT (processBlockStoreMessage x) r++isInSync ::+ (MonadLoggerIO m, StoreRead m, MonadReader BlockRead m)+ => m Bool+isInSync =+ getBestBlock >>= \case+ Nothing -> do+ $(logErrorS) "Block" "Block database uninitialized"+ throwIO Uninitialized+ Just bb ->+ asks (blockConfChain . myConfig) >>= chainGetBest >>= \cb ->+ return (headerHash (nodeHeader cb) == bb)++mempool :: MonadLoggerIO m => Peer -> m ()+mempool p = do+ $(logDebugS) "Block" "Requesting mempool from network peer"+ MMempool `sendMessage` p++processBlock ::+ (MonadUnliftIO m, MonadLoggerIO m)+ => Peer+ -> Block+ -> ReaderT BlockRead m ()+processBlock peer block = do+ void . runMaybeT $ do+ checkpeer+ blocknode <- getblocknode+ net <- asks (blockConfNet . myConfig)+ lift (runImport (importBlock net block blocknode)) >>= \case+ Right deletedtxids -> do+ listener <- asks (blockConfListener . myConfig)+ $(logInfoS) "Block" $ "Best block indexed: " <> hexhash+ atomically $ do+ mapM_ (listener . StoreTxDeleted) deletedtxids+ listener (StoreBestBlock blockhash)+ lift (syncMe peer)+ Left e -> do+ $(logErrorS) "Block" $+ "Error importing block: " <> hexhash <> ": " <>+ fromString (show e)+ killPeer (PeerMisbehaving (show e)) peer+ where+ header = blockHeader block+ blockhash = headerHash header+ hexhash = blockHashToHex blockhash+ checkpeer =+ getSyncingState >>= \case+ Just Syncing {syncingPeer = syncingpeer}+ | peer == syncingpeer -> return ()+ _ -> do+ $(logErrorS) "Block" $ "Peer sent unexpected block: " <> hexhash+ killPeer (PeerMisbehaving "Sent unpexpected block") peer+ mzero+ getblocknode =+ asks (blockConfChain . myConfig) >>= chainGetBlock blockhash >>= \case+ Nothing -> do+ $(logErrorS) "Block" $ "Block header not found: " <> hexhash+ killPeer (PeerMisbehaving "Sent unknown block") peer+ mzero+ Just n -> return n++processNoBlocks ::+ (MonadUnliftIO m, MonadLoggerIO m)+ => Peer+ -> [BlockHash]+ -> ReaderT BlockRead m ()+processNoBlocks p _bs = do+ $(logErrorS) "Block" (cs m)+ killPeer (PeerMisbehaving m) p+ where+ m = "I do not like peers that cannot find them blocks"++processTx :: (MonadUnliftIO m, MonadLoggerIO m) => Peer -> Tx -> BlockT m ()+processTx _p tx =+ isInSync >>= \sync ->+ when sync $ do+ now <- fromIntegral . systemSeconds <$> liftIO getSystemTime+ net <- asks (blockConfNet . myConfig)+ runImport (newMempoolTx net tx now) >>= \case+ Right (Just deleted) -> do+ l <- blockConfListener <$> asks myConfig+ $(logInfoS) "Block" $+ "New mempool tx: " <> txHashToHex (txHash tx)+ atomically $ do+ mapM_ (l . StoreTxDeleted) deleted+ l (StoreMempoolNew (txHash tx))+ _ -> return ()++processOrphans ::+ (MonadUnliftIO m, MonadLoggerIO m) => BlockT m ()+processOrphans =+ isInSync >>= \sync ->+ when sync $ do+ now <- fromIntegral . systemSeconds <$> liftIO getSystemTime+ net <- asks (blockConfNet . myConfig)+ old <- getOldOrphans now+ case old of+ [] -> return ()+ _ -> do+ $(logInfoS) "Block" $+ "Removing " <> cs (show (length old)) <>+ " expired orphan transactions"+ void . runImport $ mapM_ deleteOrphanTx old+ orphans <- getOrphans+ case orphans of+ [] -> return ()+ _ ->+ $(logInfoS) "Block" $+ "Attempting to import " <> cs (show (length orphans)) <>+ " orphan transactions"+ ops <-+ zip (map snd orphans) <$>+ mapM (runImport . uncurry (importOrphan net)) orphans+ let tths =+ [ (txHash tx, hs)+ | (tx, emths) <- ops+ , let Right (Just hs) = emths+ ]+ ihs = map fst tths+ dhs = concatMap snd tths+ l <- blockConfListener <$> asks myConfig+ atomically $ do+ mapM_ (l . StoreTxDeleted) dhs+ mapM_ (l . StoreMempoolNew) ihs+++processTxs ::+ (MonadUnliftIO m, MonadLoggerIO m)+ => Peer+ -> [TxHash]+ -> ReaderT BlockRead m ()+processTxs p hs =+ isInSync >>= \sync ->+ when sync $ do+ xs <-+ fmap catMaybes . forM hs $ \h ->+ runMaybeT $ do+ t <- lift $ getTxData h+ guard (isNothing t)+ return (getTxHash h)+ unless (null xs) $ do+ $(logInfoS) "Block" $+ "Requesting " <> fromString (show (length xs)) <>+ " new transactions"+ net <- blockConfNet <$> asks myConfig+ let inv =+ if getSegWit net+ then InvWitnessTx+ else InvTx+ MGetData (GetData (map (InvVector inv) xs)) `sendMessage` p++checkTime :: (MonadUnliftIO m, MonadLoggerIO m) => ReaderT BlockRead m ()+checkTime =+ asks myPeer >>= readTVarIO >>= \case+ Nothing -> return ()+ Just Syncing {syncingTime = t, syncingPeer = p} -> do+ n <- fromIntegral . systemSeconds <$> liftIO getSystemTime+ when (n > t + 60) $ do+ $(logErrorS) "Block" "Syncing peer timeout"+ resetPeer+ killPeer PeerTimeout p++processDisconnect ::+ (MonadUnliftIO m, MonadLoggerIO m)+ => Peer+ -> ReaderT BlockRead m ()+processDisconnect p =+ asks myPeer >>= readTVarIO >>= \case+ Nothing -> return ()+ Just Syncing {syncingPeer = p'}+ | p == p' -> do+ resetPeer+ getPeer >>= \case+ Nothing ->+ $(logWarnS)+ "Block"+ "No peers available after syncing peer disconnected"+ Just peer -> do+ $(logWarnS) "Block" "Selected another peer to sync"+ syncMe peer+ | otherwise -> return ()++pruneMempool :: (MonadUnliftIO m, MonadLoggerIO m) => BlockT m ()+pruneMempool =+ isInSync >>= \sync ->+ when sync $ do+ now <- fromIntegral . systemSeconds <$> liftIO getSystemTime+ getOldMempool now >>= \case+ [] -> return ()+ old -> deletetxs old+ where+ deletetxs old = do+ $(logInfoS) "Block" $+ "Removing " <> cs (show (length old)) <> " old mempool transactions"+ net <- asks (blockConfNet . myConfig)+ forM_ old $ \txid ->+ runImport (deleteTx net True txid) >>= \case+ Left _ -> return ()+ Right txids -> do+ listener <- asks (blockConfListener . myConfig)+ atomically $ mapM_ (listener . StoreTxDeleted) txids++syncMe :: (MonadUnliftIO m, MonadLoggerIO m) => Peer -> BlockT m ()+syncMe peer =+ void . runMaybeT $ do+ checksyncingpeer+ reverttomainchain+ syncbest <- syncbestnode+ bestblock <- bestblocknode+ chainbest <- chainbestnode+ end syncbest bestblock chainbest+ blocknodes <- selectblocks chainbest syncbest+ setPeer peer (last blocknodes)+ net <- asks (blockConfNet . myConfig)+ let inv =+ if getSegWit net+ then InvWitnessBlock+ else InvBlock+ vectors =+ map+ (InvVector inv . getBlockHash . headerHash . nodeHeader)+ blocknodes+ $(logInfoS) "Block" $+ "Requesting " <> fromString (show (length vectors)) <> " blocks"+ MGetData (GetData vectors) `sendMessage` peer+ where+ checksyncingpeer =+ getSyncingState >>= \case+ Nothing -> return ()+ Just Syncing {syncingPeer = p}+ | p == peer -> return ()+ | otherwise -> do+ $(logInfoS) "Block" "Already syncing against another peer"+ mzero+ chainbestnode = chainGetBest =<< asks (blockConfChain . myConfig)+ bestblocknode = do+ bb <-+ lift getBestBlock >>= \case+ Nothing -> do+ $(logErrorS) "Block" "No best block set"+ throwIO Uninitialized+ Just b -> return b+ ch <- asks (blockConfChain . myConfig)+ chainGetBlock bb ch >>= \case+ Nothing -> do+ $(logErrorS) "Block" $+ "Header not found for best block: " <> blockHashToHex bb+ throwIO (BlockNotInChain bb)+ Just x -> return x+ syncbestnode =+ asks myPeer >>= readTVarIO >>= \case+ Just Syncing {syncingHead = b} -> return b+ Nothing -> bestblocknode+ end syncbest bestblock chainbest+ | nodeHeader bestblock == nodeHeader chainbest = do+ resetPeer >> mempool peer >> mzero+ | nodeHeader syncbest == nodeHeader chainbest = do mzero+ | otherwise =+ when (nodeHeight syncbest > nodeHeight bestblock + 500) mzero+ selectblocks chainbest syncbest = do+ synctop <-+ top+ chainbest+ (maxsyncheight (nodeHeight chainbest) (nodeHeight syncbest))+ ch <- asks (blockConfChain . myConfig)+ parents <- chainGetParents (nodeHeight syncbest + 1) synctop ch+ return $+ if length parents < 500+ then parents <> [chainbest]+ else parents+ maxsyncheight chainheight syncbestheight+ | chainheight <= syncbestheight + 501 = chainheight+ | otherwise = syncbestheight + 501+ top chainbest syncheight = do+ ch <- asks (blockConfChain . myConfig)+ if syncheight == nodeHeight chainbest+ then return chainbest+ else chainGetAncestor syncheight chainbest ch >>= \case+ Just x -> return x+ Nothing -> do+ $(logErrorS) "Block" $+ "Could not find header for ancestor of block: " <>+ blockHashToHex (headerHash (nodeHeader chainbest))+ throwIO $+ AncestorNotInChain+ syncheight+ (headerHash (nodeHeader chainbest))+ reverttomainchain = do+ bestblockhash <- headerHash . nodeHeader <$> bestblocknode+ ch <- asks (blockConfChain . myConfig)+ chainBlockMain bestblockhash ch >>= \y ->+ unless y $ do+ $(logErrorS) "Block" $+ "Reverting best block: " <> blockHashToHex bestblockhash+ resetPeer+ net <- asks (blockConfNet . myConfig)+ lift (runImport (revertBlock net bestblockhash)) >>= \case+ Left e -> do+ $(logErrorS) "Block" $+ "Could not revert best block: " <> cs (show e)+ throwIO e+ Right txids -> do+ listener <- asks (blockConfListener . myConfig)+ atomically $ do+ mapM_ (listener . StoreTxDeleted) txids+ listener (StoreBlockReverted bestblockhash)+ reverttomainchain++resetPeer :: (MonadLoggerIO m, MonadReader BlockRead m) => m ()+resetPeer = do+ box <- asks myPeer+ atomically $ writeTVar box Nothing++setPeer :: (MonadIO m, MonadReader BlockRead m) => Peer -> BlockNode -> m ()+setPeer p b = do+ box <- asks myPeer+ now <- fromIntegral . systemSeconds <$> liftIO getSystemTime+ atomically . writeTVar box $+ Just Syncing {syncingPeer = p, syncingHead = b, syncingTime = now}++getPeer :: (MonadIO m, MonadReader BlockRead m) => m (Maybe Peer)+getPeer = runMaybeT $ MaybeT syncingpeer <|> MaybeT onlinepeer+ where+ syncingpeer = fmap syncingPeer <$> getSyncingState+ onlinepeer =+ listToMaybe . map onlinePeerMailbox <$>+ (managerGetPeers =<< asks (blockConfManager . myConfig))++getSyncingState :: (MonadIO m, MonadReader BlockRead m) => m (Maybe Syncing)+getSyncingState = readTVarIO =<< asks myPeer++processBlockStoreMessage ::+ (MonadUnliftIO m, MonadLoggerIO m)+ => BlockStoreMessage+ -> BlockT m ()+processBlockStoreMessage (BlockNewBest _) = do+ getPeer >>= \case+ Nothing -> do+ $(logErrorS) "Block" "New best block but no peer to sync from"+ Just p -> syncMe p+processBlockStoreMessage (BlockPeerConnect p _) = syncMe p+processBlockStoreMessage (BlockPeerDisconnect p _sa) = processDisconnect p+processBlockStoreMessage (BlockReceived p b) = processBlock p b+processBlockStoreMessage (BlockNotFound p bs) = processNoBlocks p bs+processBlockStoreMessage (BlockTxReceived p tx) = processTx p tx+processBlockStoreMessage (BlockTxAvailable p ts) = processTxs p ts+processBlockStoreMessage (BlockPing r) = do+ processOrphans+ checkTime+ pruneMempool+ atomically (r ())++pingMe :: MonadLoggerIO m => BlockStore -> m ()+pingMe mbox =+ forever $ do+ threadDelay =<< liftIO (randomRIO (5 * 1000 * 1000, 10 * 1000 * 1000))+ BlockPing `query` mbox
+ src/Network/Haskoin/Store/CacheWriter.hs view
@@ -0,0 +1,563 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE TupleSections #-}+module Network.Haskoin.Store.CacheWriter where++import Control.Monad (forM_, forever, unless,+ void, when)+import Control.Monad.Reader (ReaderT (..), asks)+import Control.Monad.Trans (lift)+import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT)+import qualified Data.IntMap.Strict as IntMap+import Data.List (nub, partition, (\\))+import qualified Data.Map.Strict as Map+import Data.Maybe (catMaybes, mapMaybe)+import Data.Serialize (encode)+import Database.Redis (RedisCtx, runRedis,+ zadd, zrem)+import qualified Database.Redis as Redis+import Haskoin (Address, BlockHash,+ BlockHeader (..),+ BlockNode (..),+ DerivPathI (..),+ KeyIndex,+ OutPoint (..), Tx (..),+ TxHash, TxIn (..),+ TxOut (..),+ derivePubPath,+ headerHash, pathToList,+ scriptToAddressBS,+ txHash)+import Haskoin.Node (Chain,+ chainGetAncestor,+ chainGetBlock,+ chainGetSplitBlock)+import Network.Haskoin.Store.Common (BlockData (..),+ BlockRef (..),+ BlockTx (..),+ CacheWriterMessage (..),+ Prev (..),+ StoreRead (..),+ TxData (..),+ Unspent (..),+ XPubBal (..),+ XPubSpec (..),+ XPubUnspent (..),+ sortTxs,+ xPubAddrFunction,+ xPubBals, xPubTxs,+ xPubUnspents)+import Network.Haskoin.Store.Data.CacheReader (AddressXPub (..),+ CacheError (..),+ CacheReaderConfig (..),+ CacheReaderT, addrPfx,+ balancesPfx,+ bestBlockKey,+ blockRefScore,+ chgIndexPfx,+ extIndexPfx,+ mempoolSetKey,+ pathScore,+ redisGetAddrInfo,+ redisGetHead,+ redisGetMempool,+ redisGetXPubIndex,+ txSetPfx, utxoPfx,+ withCacheReader)+import NQE (Inbox, receive)+import UnliftIO (MonadIO, MonadUnliftIO,+ liftIO, throwIO)+type CacheWriterInbox = Inbox CacheWriterMessage++data CacheWriterConfig =+ CacheWriterConfig+ { cacheWriterReader :: !CacheReaderConfig+ , cacheWriterChain :: !Chain+ , cacheWriterMailbox :: !CacheWriterInbox+ }++type CacheWriterT = ReaderT CacheWriterConfig++instance (MonadIO m, StoreRead m) => StoreRead (CacheWriterT m) where+ getBestBlock = lift getBestBlock+ getBlocksAtHeight = lift . getBlocksAtHeight+ getBlock = lift . getBlock+ getTxData = lift . getTxData+ getOrphanTx = lift . getOrphanTx+ getOrphans = lift getOrphans+ getSpenders = lift . getSpenders+ getSpender = lift . getSpender+ getBalance = lift . getBalance+ getBalances = lift . getBalances+ getAddressesTxs addrs start = lift . getAddressesTxs addrs start+ getAddressTxs addr start = lift . getAddressTxs addr start+ getUnspent = lift . getUnspent+ getAddressUnspents addr start = lift . getAddressUnspents addr start+ getAddressesUnspents addrs start = lift . getAddressesUnspents addrs start+ getMempool = lift getMempool+ xPubBals = runCacheReaderT . xPubBals+ xPubSummary = runCacheReaderT . xPubSummary+ xPubUnspents xpub start offset limit =+ runCacheReaderT (xPubUnspents xpub start offset limit)+ xPubTxs xpub start offset limit =+ runCacheReaderT (xPubTxs xpub start offset limit)++runCacheReaderT :: StoreRead m => CacheReaderT m a -> CacheWriterT m a+runCacheReaderT f =+ ReaderT (\CacheWriterConfig {cacheWriterReader = r} -> withCacheReader r f)++cacheWriter :: (MonadUnliftIO m, StoreRead m) => CacheWriterConfig -> m ()+cacheWriter cfg@CacheWriterConfig {cacheWriterMailbox = inbox} =+ runReaderT (forever (receive inbox >>= cacheWriterReact)) cfg++cacheWriterReact ::+ (MonadUnliftIO m, StoreRead m) => CacheWriterMessage -> CacheWriterT m ()+cacheWriterReact CacheNewBlock = newBlockC+cacheWriterReact (CacheXPub xpub) = newXPubC xpub+cacheWriterReact (CacheNewTx txh) = newTxC txh+cacheWriterReact (CacheDelTx txh) = removeTxC txh++newXPubC ::+ (MonadUnliftIO m, StoreRead m)+ => XPubSpec+ -> CacheWriterT m ()+newXPubC xpub = do+ present <- (> 0) <$> cacheGetXPubIndex xpub False+ unless present $ do+ bals <- lift $ xPubBals xpub+ unless (null bals) $ go bals+ where+ go bals = do+ utxo <- xPubUnspents xpub Nothing 0 Nothing+ xtxs <- xPubTxs xpub Nothing 0 Nothing+ let (external, change) =+ partition (\b -> head (xPubBalPath b) == 0) bals+ extindex =+ case external of+ [] -> 0+ _ -> last (xPubBalPath (last external))+ chgindex =+ case change of+ [] -> 0+ _ -> last (xPubBalPath (last change))+ cacheAddXPubBalances xpub bals+ cacheAddXPubUnspents+ xpub+ (map ((\u -> (unspentPoint u, unspentBlock u)) . xPubUnspent) utxo)+ cacheAddXPubTxs xpub xtxs+ cacheSetXPubIndex xpub False extindex+ cacheSetXPubIndex xpub True chgindex++newBlockC :: (MonadIO m, StoreRead m) => CacheWriterT m ()+newBlockC =+ lift getBestBlock >>= \case+ Nothing -> return ()+ Just newhead ->+ cacheGetHead >>= \case+ Nothing -> importBlockC newhead+ Just cachehead -> go newhead cachehead+ where+ go newhead cachehead+ | cachehead == newhead = return ()+ | otherwise = do+ ch <- asks cacheWriterChain+ chainGetBlock newhead ch >>= \case+ Nothing -> return ()+ Just newheadnode ->+ chainGetBlock cachehead ch >>= \case+ Nothing -> return ()+ Just cacheheadnode -> go2 newheadnode cacheheadnode+ go2 newheadnode cacheheadnode+ | nodeHeight cacheheadnode > nodeHeight newheadnode = return ()+ | otherwise = do+ ch <- asks cacheWriterChain+ split <- chainGetSplitBlock cacheheadnode newheadnode ch+ if split == cacheheadnode+ then if prevBlock (nodeHeader newheadnode) ==+ headerHash (nodeHeader cacheheadnode)+ then importBlockC (headerHash (nodeHeader newheadnode))+ else go3 newheadnode cacheheadnode+ else removeHeadC >> newBlockC+ go3 newheadnode cacheheadnode = do+ ch <- asks cacheWriterChain+ ma <- chainGetAncestor (nodeHeight cacheheadnode + 1) newheadnode ch+ case ma of+ Nothing -> do+ throwIO (LogicError "Could not get expected ancestor block")+ Just a -> do+ importBlockC (headerHash (nodeHeader a))+ newBlockC++newTxC :: (MonadIO m, StoreRead m) => TxHash -> CacheWriterT m ()+newTxC th =+ lift (getTxData th) >>= \case+ Just txd -> importTxC txd+ Nothing -> return ()++removeTxC :: (MonadIO m, StoreRead m) => TxHash -> CacheWriterT m ()+removeTxC th =+ lift (getTxData th) >>= \case+ Just txd -> deleteTxC txd+ Nothing -> return ()++---------------+-- Importing --+---------------++importBlockC :: (StoreRead m, MonadIO m) => BlockHash -> CacheWriterT m ()+importBlockC bh =+ lift (getBlock bh) >>= \case+ Nothing -> return ()+ Just bd -> go bd+ where+ go bd = do+ let ths = blockDataTxs bd+ tds <- sortTxData . catMaybes <$> mapM (lift . getTxData) ths+ forM_ tds importTxC++removeHeadC :: (StoreRead m, MonadIO m) => CacheWriterT m ()+removeHeadC =+ void . runMaybeT $ do+ bh <- MaybeT cacheGetHead+ bd <- MaybeT (lift (getBlock bh))+ lift $ do+ tds <-+ sortTxData . catMaybes <$>+ mapM (lift . getTxData) (blockDataTxs bd)+ forM_ (reverse (map (txHash . txData) tds)) removeTxC+ cacheSetHead (prevBlock (blockDataHeader bd))+ syncMempoolC++importTxC :: (StoreRead m, MonadIO m) => TxData -> CacheWriterT m ()+importTxC txd = do+ updateAddressesC addrs+ is <- mapM cacheGetAddrInfo addrs+ let aim = Map.fromList (catMaybes (zipWith (\a i -> (a, ) <$> i) addrs is))+ dus = mapMaybe (\(a, p) -> (, p) <$> Map.lookup a aim) spnts+ ius = mapMaybe (\(a, p) -> (, p) <$> Map.lookup a aim) utxos+ forM_ aim $ \i -> do+ cacheAddXPubTxs+ (addressXPubSpec i)+ [ BlockTx+ { blockTxHash = txHash (txData txd)+ , blockTxBlock = txDataBlock txd+ }+ ]+ forM_ dus $ \(i, p) -> do cacheRemXPubUnspents (addressXPubSpec i) [p]+ forM_ ius $ \(i, p) ->+ cacheAddXPubUnspents (addressXPubSpec i) [(p, txDataBlock txd)]+ case txDataBlock txd of+ b@MemRef {} ->+ cacheAddToMempool+ BlockTx {blockTxHash = txHash (txData txd), blockTxBlock = b}+ _ -> cacheRemFromMempool (txHash (txData txd))+ where+ spnts = txSpent txd+ utxos = txUnspent txd+ addrs = nub (map fst spnts <> map fst utxos)++deleteTxC :: (StoreRead m, MonadIO m) => TxData -> CacheWriterT m ()+deleteTxC txd = do+ updateAddressesC addrs+ is <- mapM cacheGetAddrInfo addrs+ let aim = Map.fromList (catMaybes (zipWith (\a i -> (a, ) <$> i) addrs is))+ dus = mapMaybe (\(a, p) -> (, p) <$> Map.lookup a aim) spnts+ ius = mapMaybe (\(a, p) -> (, p) <$> Map.lookup a aim) utxos+ forM_ aim $ \i -> do+ cacheRemXPubTxs (addressXPubSpec i) [txHash (txData txd)]+ forM_ dus $ \(i, p) ->+ lift (getUnspent p) >>= \case+ Just u -> do+ cacheAddXPubUnspents (addressXPubSpec i) [(p, unspentBlock u)]+ Nothing -> return ()+ forM_ ius $ \(i, p) -> cacheRemXPubUnspents (addressXPubSpec i) [p]+ case txDataBlock txd of+ MemRef {} -> cacheRemFromMempool (txHash (txData txd))+ _ -> return ()+ where+ spnts = txSpent txd+ utxos = txUnspent txd+ addrs = nub (map fst spnts <> map fst utxos)++updateAddressesC ::+ (StoreRead m, MonadIO m) => [Address] -> CacheWriterT m ()+updateAddressesC as = do+ is <- mapM cacheGetAddrInfo as+ let ais = catMaybes (zipWith (\a i -> (a, ) <$> i) as is)+ forM_ (catMaybes is) $ \i -> updateAddressGapC i+ let as' = as \\ map fst ais+ when (length as /= length as') (updateAddressesC as')++updateAddressGapC ::+ (StoreRead m, MonadIO m)+ => AddressXPub+ -> CacheWriterT m ()+updateAddressGapC i = do+ current <- cacheGetXPubIndex (addressXPubSpec i) change+ gap <- asks (cacheReaderGap . cacheWriterReader)+ let ns = addrsToAddC (addressXPubSpec i) change current new gap+ forM_ ns (uncurry updateBalanceC)+ case ns of+ [] -> return ()+ _ ->+ cacheSetXPubIndex+ (addressXPubSpec i)+ change+ (last (addressXPubPath (snd (last ns))))+ where+ change =+ case head (addressXPubPath i) of+ 1 -> True+ 0 -> False+ _ -> undefined+ new = last (addressXPubPath i)++updateBalanceC ::+ (StoreRead m, MonadIO m) => Address -> AddressXPub -> CacheWriterT m ()+updateBalanceC a i = do+ cacheSetAddrInfo a i+ b <- lift (getBalance a)+ cacheAddXPubBalances+ (addressXPubSpec i)+ [XPubBal {xPubBalPath = addressXPubPath i, xPubBal = b}]++syncMempoolC :: (MonadIO m, StoreRead m) => CacheWriterT m ()+syncMempoolC = do+ nodepool <- map blockTxHash <$> lift getMempool+ cachepool <- map blockTxHash <$> cacheGetMempool+ let deltxs = cachepool \\ nodepool+ deltds <- reverse . sortTxData . catMaybes <$> mapM (lift . getTxData) deltxs+ forM_ deltds deleteTxC+ let addtxs = nodepool \\ cachepool+ addtds <- sortTxData . catMaybes <$> mapM (lift . getTxData) addtxs+ forM_ addtds importTxC++cacheAddXPubTxs :: MonadIO m => XPubSpec -> [BlockTx] -> CacheWriterT m ()+cacheAddXPubTxs xpub txs = do+ conn <- asks (cacheReaderConn . cacheWriterReader)+ liftIO (runRedis conn (redisAddXPubTxs xpub txs)) >>= \case+ Left e -> throwIO (RedisError e)+ Right () -> return ()++cacheRemXPubTxs :: MonadIO m => XPubSpec -> [TxHash] -> CacheWriterT m ()+cacheRemXPubTxs xpub ths = do+ conn <- asks (cacheReaderConn . cacheWriterReader)+ liftIO (runRedis conn (redisRemXPubTxs xpub ths)) >>= \case+ Left e -> throwIO (RedisError e)+ Right () -> return ()++cacheAddXPubUnspents ::+ MonadIO m => XPubSpec -> [(OutPoint, BlockRef)] -> CacheWriterT m ()+cacheAddXPubUnspents xpub ops = do+ conn <- asks (cacheReaderConn . cacheWriterReader)+ liftIO (runRedis conn (redisAddXPubUnspents xpub ops)) >>= \case+ Left e -> throwIO (RedisError e)+ Right () -> return ()++cacheRemXPubUnspents :: MonadIO m => XPubSpec -> [OutPoint] -> CacheWriterT m ()+cacheRemXPubUnspents xpub ops = do+ conn <- asks (cacheReaderConn . cacheWriterReader)+ liftIO (runRedis conn (redisRemXPubUnspents xpub ops)) >>= \case+ Left e -> throwIO (RedisError e)+ Right () -> return ()++cacheAddXPubBalances :: MonadIO m => XPubSpec -> [XPubBal] -> CacheWriterT m ()+cacheAddXPubBalances xpub bals = do+ conn <- asks (cacheReaderConn . cacheWriterReader)+ liftIO (runRedis conn (redisAddXPubBalances xpub bals)) >>= \case+ Left e -> throwIO (RedisError e)+ Right () -> return ()++cacheGetXPubIndex :: MonadIO m => XPubSpec -> Bool -> CacheWriterT m KeyIndex+cacheGetXPubIndex xpub change = do+ conn <- asks (cacheReaderConn . cacheWriterReader)+ liftIO (runRedis conn (redisGetXPubIndex xpub change)) >>= \case+ Left e -> throwIO (RedisError e)+ Right x -> return x++cacheSetXPubIndex ::+ MonadIO m => XPubSpec -> Bool -> KeyIndex -> CacheWriterT m ()+cacheSetXPubIndex xpub change index = do+ conn <- asks (cacheReaderConn . cacheWriterReader)+ liftIO (runRedis conn (redisSetXPubIndex xpub change index)) >>= \case+ Left e -> throwIO (RedisError e)+ Right () -> return ()++cacheGetMempool :: MonadIO m => CacheWriterT m [BlockTx]+cacheGetMempool = do+ conn <- asks (cacheReaderConn . cacheWriterReader)+ liftIO (runRedis conn redisGetMempool) >>= \case+ Left e -> do+ throwIO (RedisError e)+ Right mem -> return mem++cacheGetHead :: MonadIO m => CacheWriterT m (Maybe BlockHash)+cacheGetHead = do+ conn <- asks (cacheReaderConn . cacheWriterReader)+ liftIO (runRedis conn redisGetHead) >>= \case+ Left e ->+ throwIO (RedisError e)+ Right h -> return h++cacheSetHead :: MonadIO m => BlockHash -> CacheWriterT m ()+cacheSetHead bh = do+ conn <- asks (cacheReaderConn . cacheWriterReader)+ liftIO (runRedis conn (redisSetHead bh)) >>= \case+ Left e -> throwIO (RedisError e)+ Right () -> return ()++cacheAddToMempool :: MonadIO m => BlockTx -> CacheWriterT m ()+cacheAddToMempool btx = do+ conn <- asks (cacheReaderConn . cacheWriterReader)+ liftIO (runRedis conn (redisAddToMempool btx)) >>= \case+ Left e -> throwIO (RedisError e)+ Right () -> return ()++cacheRemFromMempool :: MonadIO m => TxHash -> CacheWriterT m ()+cacheRemFromMempool th = do+ conn <- asks (cacheReaderConn . cacheWriterReader)+ liftIO (runRedis conn (redisRemFromMempool th)) >>= \case+ Left e -> throwIO (RedisError e)+ Right () -> return ()++cacheGetAddrInfo :: MonadIO m => Address -> CacheWriterT m (Maybe AddressXPub)+cacheGetAddrInfo a = do+ conn <- asks (cacheReaderConn . cacheWriterReader)+ liftIO (runRedis conn (redisGetAddrInfo a)) >>= \case+ Left e -> throwIO (RedisError e)+ Right i -> return i++cacheSetAddrInfo :: MonadIO m => Address -> AddressXPub -> CacheWriterT m ()+cacheSetAddrInfo a i = do+ conn <- asks (cacheReaderConn . cacheWriterReader)+ liftIO (runRedis conn (redisSetAddrInfo a i)) >>= \case+ Left e -> throwIO (RedisError e)+ Right () -> return ()++redisAddToMempool :: (Monad m, Monad f, RedisCtx m f) => BlockTx -> m (f ())+redisAddToMempool btx = do+ f <-+ zadd+ mempoolSetKey+ [(blockRefScore (blockTxBlock btx), encode (blockTxHash btx))]+ return $ f >> return ()++redisRemFromMempool :: (Monad m, Monad f, RedisCtx m f) => TxHash -> m (f ())+redisRemFromMempool th = do+ f <- zrem mempoolSetKey [encode th]+ return $ f >> return ()++redisSetAddrInfo ::+ (Monad f, RedisCtx m f) => Address -> AddressXPub -> m (f ())+redisSetAddrInfo a i = do+ f <- Redis.set (addrPfx <> encode a) (encode i)+ return $ f >> return ()++redisAddXPubTxs :: (Monad f, RedisCtx m f) => XPubSpec -> [BlockTx] -> m (f ())+redisAddXPubTxs xpub btxs = do+ let entries =+ map+ (\t -> (blockRefScore (blockTxBlock t), encode (blockTxHash t)))+ btxs+ f <- zadd (txSetPfx <> encode xpub) entries+ return $ f >> return ()++redisRemXPubTxs :: (Monad f, RedisCtx m f) => XPubSpec -> [TxHash] -> m (f ())+redisRemXPubTxs xpub txhs = do+ f <- zrem (txSetPfx <> encode xpub) (map encode txhs)+ return $ f >> return ()++redisAddXPubUnspents ::+ (Monad f, RedisCtx m f) => XPubSpec -> [(OutPoint, BlockRef)] -> m (f ())+redisAddXPubUnspents xpub utxo = do+ let entries = map (\(p, r) -> (blockRefScore r, encode p)) utxo+ f <- zadd (utxoPfx <> encode xpub) entries+ return $ f >> return ()++redisRemXPubUnspents ::+ (Monad f, RedisCtx m f) => XPubSpec -> [OutPoint] -> m (f ())+redisRemXPubUnspents xpub ops = do+ f <- zrem (txSetPfx <> encode xpub) (map encode ops)+ return $ f >> return ()++redisAddXPubBalances ::+ (Monad f, RedisCtx m f) => XPubSpec -> [XPubBal] -> m (f ())+redisAddXPubBalances xpub bals = do+ let entries =+ map (\b -> (pathScore (xPubBalPath b), encode (xPubBal b))) bals+ f <- zadd (balancesPfx <> encode xpub) entries+ return $ f >> return ()++redisSetXPubIndex :: (Monad f, RedisCtx m f) => XPubSpec -> Bool -> KeyIndex -> m (f ())+redisSetXPubIndex xpub change index = do+ f <- Redis.set (pfx <> encode xpub) (encode index)+ return $ f >> return ()+ where+ pfx =+ if change+ then chgIndexPfx+ else extIndexPfx++redisSetHead :: (Monad m, Monad f, RedisCtx m f) => BlockHash -> m (f ())+redisSetHead bh = do+ f <- Redis.set bestBlockKey (encode bh)+ return $ f >> return ()++addrsToAddC ::+ XPubSpec+ -> Bool+ -> KeyIndex+ -> KeyIndex+ -> KeyIndex+ -> [(Address, AddressXPub)]+addrsToAddC xpub change current new gap+ | new <= current = []+ | otherwise =+ let top = new + gap+ indices = [current + 1 .. top]+ paths =+ map+ (Deriv :/+ (if change+ then 1+ else 0) :/)+ indices+ keys = map (\p -> derivePubPath p (xPubSpecKey xpub)) paths+ list = map pathToList paths+ xpubf = xPubAddrFunction (xPubDeriveType xpub)+ addrs = map xpubf keys+ in zipWith+ (\a p ->+ ( a+ , AddressXPub {addressXPubSpec = xpub, addressXPubPath = p}))+ addrs+ list++sortTxData :: [TxData] -> [TxData]+sortTxData tds =+ let txm = Map.fromList (map (\d -> (txHash (txData d), d)) tds)+ ths = map (txHash . snd) (sortTxs (map txData tds))+ in mapMaybe (\h -> Map.lookup h txm) ths++txSpent :: TxData -> [(Address, OutPoint)]+txSpent td =+ let is = txIn (txData td)+ ps = IntMap.toAscList (txDataPrevs td)+ as = map (scriptToAddressBS . prevScript . snd) ps+ f (Right a) i = Just (a, prevOutput i)+ f (Left _) _ = Nothing+ in catMaybes (zipWith f as is)++txUnspent :: TxData -> [(Address, OutPoint)]+txUnspent td =+ let ps =+ zipWith+ (\i _ ->+ OutPoint+ {outPointHash = txHash (txData td), outPointIndex = i})+ [0 ..]+ (txOut (txData td))+ as = map (scriptToAddressBS . scriptOutput) (txOut (txData td))+ f (Right a) p = Just (a, p)+ f (Left _) _ = Nothing+ in catMaybes (zipWith f as ps)
src/Network/Haskoin/Store/Common.hs view
@@ -36,7 +36,6 @@ import Data.String.Conversions (cs) import qualified Data.Text.Encoding as T import Data.Word (Word32, Word64)-import Database.RocksDB (DB, ReadOptions) import GHC.Generics (Generic) import Haskoin (Address, Block, BlockHash, BlockHeader (..), BlockHeight,@@ -56,10 +55,11 @@ txHash, wrapPubKey, xPubAddr, xPubCompatWitnessAddr, xPubWitnessAddr)-import Haskoin.Node (Chain, HostPort, Manager, Peer)+import Haskoin.Node (Chain, Manager, Peer) import Network.Socket (SockAddr)-import NQE (Listen, Mailbox)+import NQE (Listen, Mailbox, send) import qualified Paths_haskoin_store as P+import UnliftIO (MonadIO) data DeriveType = DeriveNormal@@ -67,6 +67,39 @@ | DeriveP2WPKH deriving (Show, Eq, Generic, NFData, Serialize) +-- | Messages for block store actor.+data BlockStoreMessage+ = BlockNewBest !BlockNode+ -- ^ new block header in chain+ | BlockPeerConnect !Peer !SockAddr+ -- ^ new peer connected+ | BlockPeerDisconnect !Peer !SockAddr+ -- ^ peer disconnected+ | BlockReceived !Peer !Block+ -- ^ new block received from a peer+ | BlockNotFound !Peer ![BlockHash]+ -- ^ block not found+ | BlockTxReceived !Peer !Tx+ -- ^ transaction received from peer+ | BlockTxAvailable !Peer ![TxHash]+ -- ^ peer has transactions available+ | BlockPing !(Listen ())+ -- ^ internal housekeeping ping++-- | Mailbox for block store.+type BlockStore = Mailbox BlockStoreMessage++-- | Store mailboxes.+data Store =+ Store+ { storeManager :: !Manager+ -- ^ peer manager mailbox+ , storeChain :: !Chain+ -- ^ chain header process mailbox+ , storeBlock :: !BlockStore+ -- ^ block storage mailbox+ }+ data XPubSpec = XPubSpec { xPubSpecKey :: !XPubKey@@ -100,64 +133,21 @@ type DeriveAddr = XPubKey -> KeyIndex -> Address --- | Mailbox for block store.-type BlockStore = Mailbox BlockMessage---- | Store mailboxes.-data Store =- Store- { storeManager :: !Manager- -- ^ peer manager mailbox- , storeChain :: !Chain- -- ^ chain header process mailbox- , storeBlock :: !BlockStore- -- ^ block storage mailbox- }---- | Configuration for a 'Store'.-data StoreConfig =- StoreConfig- { storeConfMaxPeers :: !Int- -- ^ max peers to connect to- , storeConfInitPeers :: ![HostPort]- -- ^ static set of peers to connect to- , storeConfDiscover :: !Bool- -- ^ discover new peers?- , storeConfDB :: !BlockDB- -- ^ RocksDB database handler- , storeConfNetwork :: !Network- -- ^ network constants- , storeConfListen :: !(Listen StoreEvent)- -- ^ listen to store events- }---- | Configuration for a block store.-data BlockConfig =- BlockConfig- { blockConfManager :: !Manager- -- ^ peer manager from running node- , blockConfChain :: !Chain- -- ^ chain from a running node- , blockConfListener :: !(Listen StoreEvent)- -- ^ listener for store events- , blockConfDB :: !BlockDB- -- ^ RocksDB database handle- , blockConfNet :: !Network- -- ^ network constants- }--data BlockDB =- BlockDB- { blockDB :: !DB- , blockDBopts :: !ReadOptions- }- type UnixTime = Word64 type BlockPos = Word32 type Offset = Word32 type Limit = Word32 +data CacheWriterMessage+ = CacheXPub !XPubSpec+ | CacheNewTx !TxHash+ | CacheDelTx !TxHash+ | CacheNewBlock+ deriving (Show, Eq, Generic, NFData)++type CacheWriter = Mailbox CacheWriterMessage+ class Monad m => StoreRead m where@@ -1386,25 +1376,6 @@ , unspentScript = s } --- | Messages that a 'BlockStore' can accept.-data BlockMessage- = BlockNewBest !BlockNode- -- ^ new block header in chain- | BlockPeerConnect !Peer !SockAddr- -- ^ new peer connected- | BlockPeerDisconnect !Peer !SockAddr- -- ^ peer disconnected- | BlockReceived !Peer !Block- -- ^ new block received from a peer- | BlockNotFound !Peer ![BlockHash]- -- ^ block not found- | BlockTxReceived !Peer !Tx- -- ^ transaction received from peer- | BlockTxAvailable !Peer ![TxHash]- -- ^ peer has transactions available- | BlockPing !(Listen ())- -- ^ internal housekeeping ping- -- | Events that the store can generate. data StoreEvent = StoreBestBlock !BlockHash@@ -1483,3 +1454,15 @@ txIn . snd) ts in is <> go ds++cacheXPub :: MonadIO m => CacheWriter -> XPubSpec -> m ()+cacheXPub cache xpub = CacheXPub xpub `send` cache++cacheNewTx :: MonadIO m => CacheWriter -> TxHash -> m ()+cacheNewTx cache tx = CacheNewTx tx `send` cache++cacheDelTx :: MonadIO m => CacheWriter -> TxHash -> m ()+cacheDelTx cache tx = CacheDelTx tx `send` cache++cacheNewBlock :: MonadIO m => CacheWriter -> m ()+cacheNewBlock = send CacheNewBlock
+ src/Network/Haskoin/Store/Data/CacheReader.hs view
@@ -0,0 +1,385 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeSynonymInstances #-}+module Network.Haskoin.Store.Data.CacheReader where++import Control.DeepSeq (NFData)+import Control.Monad.Reader (ReaderT (..), asks)+import Control.Monad.Trans (lift)+import Data.Bits (shift, (.&.), (.|.))+import Data.ByteString (ByteString)+import qualified Data.ByteString.Short as BSS+import Data.Either (rights)+import qualified Data.Map.Strict as Map+import Data.Maybe (catMaybes, mapMaybe)+import Data.Serialize (Serialize, decode, encode)+import Data.Word (Word64)+import Database.Redis (Connection, RedisCtx, Reply,+ runRedis, zrangeWithscores,+ zrangebyscoreWithscoresLimit)+import qualified Database.Redis as Redis+import GHC.Generics (Generic)+import Haskoin (Address, BlockHash, KeyIndex,+ OutPoint (..), scriptToAddressBS)+import Network.Haskoin.Store.Common (Balance (..), BlockRef (..),+ BlockTx (..), CacheWriter, Limit,+ Offset, StoreRead (..),+ Unspent (..), XPubBal (..),+ XPubSpec (..), XPubUnspent (..),+ cacheXPub)+import UnliftIO (Exception, MonadIO, liftIO,+ throwIO)++data CacheReaderConfig =+ CacheReaderConfig+ { cacheReaderConn :: !Connection+ , cacheReaderGap :: !KeyIndex+ , cacheReaderWriter :: !CacheWriter+ }+++data AddressXPub =+ AddressXPub+ { addressXPubSpec :: !XPubSpec+ , addressXPubPath :: ![KeyIndex]+ } deriving (Show, Eq, Generic, NFData, Serialize)++type CacheReaderT = ReaderT CacheReaderConfig++data CacheError+ = RedisError Reply+ | LogicError String+ deriving (Show, Eq, Generic, NFData, Exception)++instance (MonadIO m, StoreRead m) => StoreRead (CacheReaderT m) where+ getBestBlock = lift getBestBlock+ getBlocksAtHeight = lift . getBlocksAtHeight+ getBlock = lift . getBlock+ getTxData = lift . getTxData+ getOrphanTx = lift . getOrphanTx+ getOrphans = lift getOrphans+ getSpenders = lift . getSpenders+ getSpender = lift . getSpender+ getBalance = lift . getBalance+ getBalances = lift . getBalances+ getAddressesTxs addrs start = lift . getAddressesTxs addrs start+ getAddressTxs addr start = lift . getAddressTxs addr start+ getUnspent = lift . getUnspent+ getAddressUnspents addr start = lift . getAddressUnspents addr start+ getAddressesUnspents addrs start = lift . getAddressesUnspents addrs start+ getMempool = lift getMempool+ xPubBals = getXPubBalances+ xPubUnspents = getXPubUnspents+ xPubTxs = getXPubTxs++withCacheReader :: StoreRead m => CacheReaderConfig -> CacheReaderT m a -> m a+withCacheReader s f = runReaderT f s++-- Version of the cache database+cacheVerKey :: ByteString+cacheVerKey = "version"++cacheVerCurrent :: ByteString+cacheVerCurrent = "1"++-- External max index+extIndexPfx :: ByteString+extIndexPfx = "e"++-- Change max index+chgIndexPfx :: ByteString+chgIndexPfx = "c"++-- Ordered set of transaction ids in mempool+mempoolSetKey :: ByteString+mempoolSetKey = "mempool"++-- Best block indexed+bestBlockKey :: ByteString+bestBlockKey = "head"++-- Ordered set of balances for an extended public key+balancesPfx :: ByteString+balancesPfx = "b"++-- Ordered set of transactions for an extended public key+txSetPfx :: ByteString+txSetPfx = "t"++-- Ordered set of unspent outputs for an extended pulic key+utxoPfx :: ByteString+utxoPfx = "u"++-- Extended public key info for an address+addrPfx :: ByteString+addrPfx = "a"++getXPubTxs ::+ (MonadIO m, StoreRead m)+ => XPubSpec+ -> Maybe BlockRef+ -> Offset+ -> Maybe Limit+ -> CacheReaderT m [BlockTx]+getXPubTxs xpub start offset limit = do+ cacheGetXPubTxs xpub start offset limit >>= \case+ [] -> do+ cache <- asks cacheReaderWriter+ cacheXPub cache xpub+ lift (xPubTxs xpub start offset limit)+ txs -> return txs++getXPubUnspents ::+ (MonadIO m, StoreRead m)+ => XPubSpec+ -> Maybe BlockRef+ -> Offset+ -> Maybe Limit+ -> CacheReaderT m [XPubUnspent]+getXPubUnspents xpub start offset limit =+ getXPubBalances xpub >>= \case+ [] -> do+ cache <- asks cacheReaderWriter+ cacheXPub cache xpub+ lift (xPubUnspents xpub start offset limit)+ bals -> do+ ops <- map snd <$> cacheGetXPubUnspents xpub start offset limit+ uns <- catMaybes <$> mapM getUnspent ops+ let addrmap =+ Map.fromList $+ map (\b -> (balanceAddress (xPubBal b), xPubBalPath b)) bals+ addrutxo =+ mapMaybe+ (\u ->+ either+ (const Nothing)+ (\a -> Just (a, u))+ (scriptToAddressBS+ (BSS.fromShort (unspentScript u))))+ uns+ xpubutxo =+ mapMaybe+ (\(a, u) ->+ (\p -> XPubUnspent p u) <$> Map.lookup a addrmap)+ addrutxo+ return xpubutxo++getXPubBalances ::+ (MonadIO m, StoreRead m)+ => XPubSpec+ -> CacheReaderT m [XPubBal]+getXPubBalances xpub = do+ cacheGetXPubBalances xpub >>= \case+ [] -> do+ cache <- asks cacheReaderWriter+ cacheXPub cache xpub+ lift (xPubBals xpub)+ bals -> return bals++cacheGetXPubBalances :: MonadIO m => XPubSpec -> CacheReaderT m [XPubBal]+cacheGetXPubBalances xpub = do+ conn <- asks cacheReaderConn+ liftIO (runRedis conn (redisGetXPubBalances xpub)) >>= \case+ Left e -> throwIO (RedisError e)+ Right bals -> return bals++cacheGetXPubTxs ::+ MonadIO m+ => XPubSpec+ -> Maybe BlockRef+ -> Offset+ -> Maybe Limit+ -> CacheReaderT m [BlockTx]+cacheGetXPubTxs xpub start offset limit = do+ conn <- asks cacheReaderConn+ liftIO (runRedis conn (redisGetXPubTxs xpub start offset limit)) >>= \case+ Left e -> throwIO (RedisError e)+ Right bts -> return bts++cacheGetXPubUnspents ::+ MonadIO m+ => XPubSpec+ -> Maybe BlockRef+ -> Offset+ -> Maybe Limit+ -> CacheReaderT m [(BlockRef, OutPoint)]+cacheGetXPubUnspents xpub start offset limit = do+ conn <- asks cacheReaderConn+ liftIO (runRedis conn (redisGetXPubUnspents xpub start offset limit)) >>= \case+ Left e -> throwIO (RedisError e)+ Right ops -> return ops++redisGetHead :: (Monad m, Monad f, RedisCtx m f) => m (f (Maybe BlockHash))+redisGetHead = do+ f <- Redis.get bestBlockKey+ return $ do+ mbs <- f+ case mbs of+ Nothing -> return Nothing+ Just bs ->+ case decode bs of+ Left e -> error e+ Right h -> return h++redisGetMempool :: (Monad m, Monad f, RedisCtx m f) => m (f [BlockTx])+redisGetMempool = do+ f <- getFromSortedSet mempoolSetKey Nothing 0 Nothing+ return $ do+ bts <- f+ return+ (map (\(t, s) ->+ BlockTx {blockTxBlock = scoreBlockRef s, blockTxHash = t})+ bts)++redisGetAddrInfo :: (Monad f, RedisCtx m f) => Address -> m (f (Maybe AddressXPub))+redisGetAddrInfo a = do+ f <- Redis.get (addrPfx <> encode a)+ return $ do+ m <- f+ case m of+ Nothing -> return Nothing+ Just x -> case decode x of+ Left e -> error e+ Right i -> return (Just i)++redisGetXPubBalances :: (Monad f, RedisCtx m f) => XPubSpec -> m (f [XPubBal])+redisGetXPubBalances xpub = do+ xs <- getFromSortedSet (balancesPfx <> encode xpub) Nothing 0 Nothing+ return $ do+ xs' <- xs+ return (map (uncurry f) xs')+ where+ f b s = XPubBal {xPubBalPath = scorePath s, xPubBal = b}++redisGetXPubTxs ::+ (Monad f, RedisCtx m f)+ => XPubSpec+ -> Maybe BlockRef+ -> Offset+ -> Maybe Limit+ -> m (f [BlockTx])+redisGetXPubTxs xpub start offset limit = do+ xs <-+ getFromSortedSet+ (txSetPfx <> encode xpub)+ (blockRefScore <$> start)+ (fromIntegral offset)+ (fromIntegral <$> limit)+ return $ do+ xs' <- xs+ return (map (uncurry f) xs')+ where+ f t s = BlockTx {blockTxHash = t, blockTxBlock = scoreBlockRef s}++redisGetXPubUnspents ::+ (Monad f, RedisCtx m f)+ => XPubSpec+ -> Maybe BlockRef+ -> Offset+ -> Maybe Limit+ -> m (f [(BlockRef, OutPoint)])+redisGetXPubUnspents xpub start offset limit = do+ xs <-+ getFromSortedSet+ (utxoPfx <> encode xpub)+ (blockRefScore <$> start)+ (fromIntegral offset)+ (fromIntegral <$> limit)+ return $ do+ xs' <- xs+ return (map (uncurry f) xs')+ where+ f o s = (scoreBlockRef s, o)++redisGetXPubIndex :: (Monad f, RedisCtx m f) => XPubSpec -> Bool -> m (f KeyIndex)+redisGetXPubIndex xpub change = do+ f <- Redis.get (pfx <> encode xpub)+ return $ f >>= \case+ Nothing -> return 0+ Just x -> case decode x of+ Left e -> error e+ Right n -> return n+ where+ pfx =+ if change+ then chgIndexPfx+ else extIndexPfx++blockRefScore :: BlockRef -> Double+blockRefScore BlockRef {blockRefHeight = h, blockRefPos = p} =+ fromIntegral (0x001fffffffffffff - (h' .|. p'))+ where+ h' = (fromIntegral h .&. 0x07ffffff) `shift` 26 :: Word64+ p' = (fromIntegral p .&. 0x03ffffff) :: Word64+blockRefScore MemRef {memRefTime = t} = 0 - t'+ where+ t' = fromIntegral (t .&. 0x001fffffffffffff)++scoreBlockRef :: Double -> BlockRef+scoreBlockRef s+ | s < 0 = MemRef {memRefTime = n}+ | otherwise = BlockRef {blockRefHeight = h, blockRefPos = p}+ where+ n = truncate (abs s) :: Word64+ m = 0x001fffffffffffff - n+ h = fromIntegral (m `shift` (-26))+ p = fromIntegral (m .&. 0x03ffffff)++pathScore :: [KeyIndex] -> Double+pathScore [m, n]+ | m == 0 || m == 1 = fromIntegral (toInteger n .|. toInteger m `shift` 32)+ | otherwise = undefined+pathScore _ = undefined++scorePath :: Double -> [KeyIndex]+scorePath s+ | s < 0 = undefined+ | s > 0x01ffffffff = undefined+ | otherwise =+ [ fromInteger (round s `shift` (-32))+ , fromInteger (round s .&. 0xffffffff)+ ]++getFromSortedSet ::+ (Monad f, RedisCtx m f, Serialize a)+ => ByteString+ -> Maybe Double+ -> Integer+ -> Maybe Integer+ -> m (f [(a, Double)])+getFromSortedSet key Nothing offset Nothing = do+ xs <- zrangeWithscores key offset (-1)+ return $ do+ ys <- map (\(x, s) -> (, s) <$> decode x) <$> xs+ return (rights ys)+getFromSortedSet key Nothing offset (Just count) = do+ xs <- zrangeWithscores key offset (offset + count - 1)+ return $ do+ ys <- map (\(x, s) -> (, s) <$> decode x) <$> xs+ return (rights ys)+getFromSortedSet key (Just score) offset Nothing = do+ xs <-+ zrangebyscoreWithscoresLimit+ key+ score+ (2 ^ (53 :: Integer) - 1)+ offset+ (-1)+ return $ do+ ys <- map (\(x, s) -> (, s) <$> decode x) <$> xs+ return (rights ys)+getFromSortedSet key (Just score) offset (Just count) = do+ xs <-+ zrangebyscoreWithscoresLimit+ key+ score+ (2 ^ (53 :: Integer) - 1)+ offset+ count+ return $ do+ ys <- map (\(x, s) -> (, s) <$> decode x) <$> xs+ return (rights ys)
+ src/Network/Haskoin/Store/Data/DatabaseReader.hs view
@@ -0,0 +1,257 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE LambdaCase #-}+module Network.Haskoin.Store.Data.DatabaseReader where++import Conduit (ConduitT, MonadResource,+ mapC, runConduit,+ runResourceT, sinkList, (.|))+import Control.Monad.Except (runExceptT, throwError)+import Control.Monad.Reader (ReaderT, ask, runReaderT)+import Data.Function (on)+import Data.IntMap (IntMap)+import qualified Data.IntMap.Strict as I+import Data.List (nub, sortBy)+import Data.Maybe (fromMaybe)+import Data.Word (Word32)+import Database.RocksDB (Compression (..), DB,+ Options (..), ReadOptions,+ defaultOptions,+ defaultReadOptions, open)+import Database.RocksDB.Query (insert, matching,+ matchingAsList, matchingSkip,+ retrieve)+import Haskoin (Address, BlockHash,+ BlockHeight, OutPoint (..),+ Tx, TxHash)+import Network.Haskoin.Store.Common (Balance, BlockData,+ BlockRef (..), BlockTx (..),+ Limit, Spender,+ StoreRead (..), TxData,+ UnixTime, Unspent (..),+ UnspentVal (..), applyLimit,+ applyLimitC, valToBalance,+ valToUnspent, zeroBalance)+import Network.Haskoin.Store.Data.Types (AddrOutKey (..),+ AddrTxKey (..), BalKey (..),+ BestKey (..), BlockKey (..),+ HeightKey (..), MemKey (..),+ OldMemKey (..),+ OrphanKey (..),+ SpenderKey (..), TxKey (..),+ UnspentKey (..),+ VersionKey (..), toUnspent)+import UnliftIO (MonadIO, liftIO)++type DatabaseReaderT = ReaderT DatabaseReader++data DatabaseReader =+ DatabaseReader+ { databaseHandle :: !DB+ , databaseReadOptions :: !ReadOptions+ }++dataVersion :: Word32+dataVersion = 16++connectRocksDB :: MonadIO m => FilePath -> m DatabaseReader+connectRocksDB dir = do+ bdb <- open+ dir+ defaultOptions+ { createIfMissing = True+ , compression = SnappyCompression+ , maxOpenFiles = -1+ , writeBufferSize = 2 ^ (30 :: Integer)+ } >>= \db ->+ return DatabaseReader {databaseReadOptions = defaultReadOptions, databaseHandle = db}+ initRocksDB bdb+ return bdb++withDatabaseReader :: MonadIO m => DatabaseReader -> DatabaseReaderT m a -> m a+withDatabaseReader = flip runReaderT++initRocksDB :: MonadIO m => DatabaseReader -> m ()+initRocksDB bdb@DatabaseReader {databaseReadOptions = opts, databaseHandle = db} = do+ e <-+ runExceptT $+ retrieve db opts VersionKey >>= \case+ Just v+ | v == dataVersion -> return ()+ | v == 15 -> migrate15to16 bdb >> initRocksDB bdb+ | otherwise -> throwError "Incorrect RocksDB database version"+ Nothing -> setInitRocksDB db+ case e of+ Left s -> error s+ Right () -> return ()++migrate15to16 :: MonadIO m => DatabaseReader -> m ()+migrate15to16 DatabaseReader {databaseReadOptions = opts, databaseHandle = db} = do+ xs <- liftIO $ matchingAsList db opts OldMemKeyS+ let ys = map (\(OldMemKey t h, ()) -> (t, h)) xs+ insert db MemKey ys+ insert db VersionKey (16 :: Word32)++setInitRocksDB :: MonadIO m => DB -> m ()+setInitRocksDB db = insert db VersionKey dataVersion++getBestDatabaseReader :: MonadIO m => DatabaseReader -> m (Maybe BlockHash)+getBestDatabaseReader DatabaseReader {databaseReadOptions = opts, databaseHandle = db} =+ retrieve db opts BestKey++getBlocksAtHeightDB :: MonadIO m => BlockHeight -> DatabaseReader -> m [BlockHash]+getBlocksAtHeightDB h DatabaseReader {databaseReadOptions = opts, databaseHandle = db} =+ retrieve db opts (HeightKey h) >>= \case+ Nothing -> return []+ Just ls -> return ls++getDatabaseReader :: MonadIO m => BlockHash -> DatabaseReader -> m (Maybe BlockData)+getDatabaseReader h DatabaseReader {databaseReadOptions = opts, databaseHandle = db} =+ retrieve db opts (BlockKey h)++getTxDataDB ::+ MonadIO m => TxHash -> DatabaseReader -> m (Maybe TxData)+getTxDataDB th DatabaseReader {databaseReadOptions = opts, databaseHandle = db} =+ retrieve db opts (TxKey th)++getSpenderDB :: MonadIO m => OutPoint -> DatabaseReader -> m (Maybe Spender)+getSpenderDB op DatabaseReader {databaseReadOptions = opts, databaseHandle = db} =+ retrieve db opts $ SpenderKey op++getSpendersDB :: MonadIO m => TxHash -> DatabaseReader -> m (IntMap Spender)+getSpendersDB th DatabaseReader {databaseReadOptions = opts, databaseHandle = db} =+ I.fromList . map (uncurry f) <$>+ liftIO (matchingAsList db opts (SpenderKeyS th))+ where+ f (SpenderKey op) s = (fromIntegral (outPointIndex op), s)+ f _ _ = undefined++getBalanceDB :: MonadIO m => Address -> DatabaseReader -> m Balance+getBalanceDB a DatabaseReader {databaseReadOptions = opts, databaseHandle = db} =+ fromMaybe (zeroBalance a) . fmap (valToBalance a) <$>+ retrieve db opts (BalKey a)++getMempoolDB :: MonadIO m => DatabaseReader -> m [BlockTx]+getMempoolDB DatabaseReader {databaseReadOptions = opts, databaseHandle = db} =+ fmap f . fromMaybe [] <$> retrieve db opts MemKey+ where+ f (t, h) = BlockTx {blockTxBlock = MemRef t, blockTxHash = h}++getOrphansDB ::+ MonadIO m+ => DatabaseReader+ -> m [(UnixTime, Tx)]+getOrphansDB DatabaseReader {databaseReadOptions = opts, databaseHandle = db} =+ liftIO . runResourceT . runConduit $+ matching db opts OrphanKeyS .| mapC snd .| sinkList++getOrphanTxDB :: MonadIO m => TxHash -> DatabaseReader -> m (Maybe (UnixTime, Tx))+getOrphanTxDB h DatabaseReader {databaseReadOptions = opts, databaseHandle = db} =+ retrieve db opts (OrphanKey h)++getAddressesTxsDB ::+ MonadIO m+ => [Address]+ -> Maybe BlockRef+ -> Maybe Limit+ -> DatabaseReader+ -> m [BlockTx]+getAddressesTxsDB addrs start limit db = do+ ts <- concat <$> mapM (\a -> getAddressTxsDB a start limit db) addrs+ let ts' = nub $ sortBy (flip compare `on` blockTxBlock) ts+ return $ applyLimit limit ts'++getAddressTxsDB ::+ MonadIO m+ => Address+ -> Maybe BlockRef+ -> Maybe Limit+ -> DatabaseReader+ -> m [BlockTx]+getAddressTxsDB a start limit DatabaseReader {databaseReadOptions = opts, databaseHandle = db} =+ liftIO . runResourceT . runConduit $+ x .| applyLimitC limit .| mapC (uncurry f) .| sinkList+ where+ x =+ case start of+ Nothing -> matching db opts (AddrTxKeyA a)+ Just br -> matchingSkip db opts (AddrTxKeyA a) (AddrTxKeyB a br)+ f AddrTxKey {addrTxKeyT = t} () = t+ f _ _ = undefined++getAddressBalancesDB ::+ (MonadIO m, MonadResource m)+ => DatabaseReader+ -> ConduitT i Balance m ()+getAddressBalancesDB DatabaseReader {databaseReadOptions = opts, databaseHandle = db} =+ matching db opts BalKeyS .| mapC (\(BalKey a, b) -> valToBalance a b)++getUnspentsDB ::+ (MonadIO m, MonadResource m)+ => DatabaseReader+ -> ConduitT i Unspent m ()+getUnspentsDB DatabaseReader {databaseReadOptions = opts, databaseHandle = db} =+ matching db opts UnspentKeyB .|+ mapC (\(UnspentKey k, v) -> unspentFromDB k v)++getUnspentDB :: MonadIO m => OutPoint -> DatabaseReader -> m (Maybe Unspent)+getUnspentDB p DatabaseReader {databaseReadOptions = opts, databaseHandle = db} =+ fmap (valToUnspent p) <$> retrieve db opts (UnspentKey p)++getAddressesUnspentsDB ::+ MonadIO m+ => [Address]+ -> Maybe BlockRef+ -> Maybe Limit+ -> DatabaseReader+ -> m [Unspent]+getAddressesUnspentsDB addrs start limit bdb = do+ us <- concat <$> mapM (\a -> getAddressUnspentsDB a start limit bdb) addrs+ let us' = nub $ sortBy (flip compare `on` unspentBlock) us+ return $ applyLimit limit us'++getAddressUnspentsDB ::+ MonadIO m+ => Address+ -> Maybe BlockRef+ -> Maybe Limit+ -> DatabaseReader+ -> m [Unspent]+getAddressUnspentsDB a start limit DatabaseReader {databaseReadOptions = opts, databaseHandle = db} =+ liftIO . runResourceT . runConduit $+ x .| applyLimitC limit .| mapC (uncurry toUnspent) .| sinkList+ where+ x =+ case start of+ Nothing -> matching db opts (AddrOutKeyA a)+ Just br -> matchingSkip db opts (AddrOutKeyA a) (AddrOutKeyB a br)++unspentFromDB :: OutPoint -> UnspentVal -> Unspent+unspentFromDB p UnspentVal { unspentValBlock = b+ , unspentValAmount = v+ , unspentValScript = s+ } =+ Unspent+ { unspentBlock = b+ , unspentAmount = v+ , unspentPoint = p+ , unspentScript = s+ }++instance MonadIO m => StoreRead (DatabaseReaderT m) where+ getBestBlock = ask >>= getBestDatabaseReader+ getBlocksAtHeight h = ask >>= getBlocksAtHeightDB h+ getBlock b = ask >>= getDatabaseReader b+ getTxData t = ask >>= getTxDataDB t+ getSpender p = ask >>= getSpenderDB p+ getSpenders t = ask >>= getSpendersDB t+ getOrphanTx h = ask >>= getOrphanTxDB h+ getUnspent a = ask >>= getUnspentDB a+ getBalance a = ask >>= getBalanceDB a+ getMempool = ask >>= getMempoolDB+ getAddressesTxs addrs start limit =+ ask >>= getAddressesTxsDB addrs start limit+ getAddressesUnspents addrs start limit =+ ask >>= getAddressesUnspentsDB addrs start limit+ getOrphans = ask >>= getOrphansDB+ getAddressUnspents a b c = ask >>= getAddressUnspentsDB a b c+ getAddressTxs a b c = ask >>= getAddressTxsDB a b c
+ src/Network/Haskoin/Store/Data/DatabaseWriter.hs view
@@ -0,0 +1,367 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+module Network.Haskoin.Store.Data.DatabaseWriter where++import Control.Applicative ((<|>))+import Control.Monad (join)+import Control.Monad.Except (MonadError)+import Control.Monad.Reader (ReaderT)+import qualified Control.Monad.Reader as R+import Control.Monad.Trans.Maybe (MaybeT (..),+ runMaybeT)+import Data.HashMap.Strict (HashMap)+import qualified Data.HashMap.Strict as M+import Data.IntMap.Strict (IntMap)+import qualified Data.IntMap.Strict as I+import Data.List (nub)+import Data.Maybe (fromJust, fromMaybe,+ isJust, maybeToList)+import Database.RocksDB (BatchOp)+import Database.RocksDB.Query (deleteOp, insertOp,+ writeBatch)+import Haskoin (Address, BlockHash,+ BlockHeight,+ OutPoint (..), Tx,+ TxHash)+import Network.Haskoin.Store.Common (BalVal, Balance,+ BlockData,+ BlockRef (..),+ BlockTx (..),+ Spender,+ StoreRead (..),+ StoreWrite (..),+ TxData, UnixTime,+ Unspent, UnspentVal,+ nullBalance,+ zeroBalance)+import Network.Haskoin.Store.Data.DatabaseReader (DatabaseReader (..),+ withDatabaseReader)+import Network.Haskoin.Store.Data.MemoryDatabase (MemoryDatabase (..),+ emptyMemoryDatabase,+ getMempoolH,+ getOrphanTxH,+ getSpenderH,+ getSpendersH,+ getUnspentH,+ withMemoryDatabase)+import Network.Haskoin.Store.Data.Types (AddrOutKey (..),+ AddrTxKey (..),+ BalKey (..),+ BestKey (..),+ BlockKey (..),+ HeightKey (..),+ MemKey (..),+ OrphanKey (..),+ OutVal,+ SpenderKey (..),+ TxKey (..),+ UnspentKey (..))+import UnliftIO (MonadIO, TVar,+ newTVarIO,+ readTVarIO)++data DatabaseWriter = DatabaseWriter+ { databaseWriterReader :: !DatabaseReader+ , databaseWriterState :: !(TVar MemoryDatabase)+ }++runDatabaseWriter ::+ (MonadIO m, MonadError e m) => DatabaseReader -> ReaderT DatabaseWriter m a -> m a+runDatabaseWriter bdb@DatabaseReader {databaseHandle = db} f = do+ hm <- newTVarIO emptyMemoryDatabase+ x <- R.runReaderT f DatabaseWriter {databaseWriterReader = bdb, databaseWriterState = hm}+ ops <- hashMapOps <$> readTVarIO hm+ writeBatch db ops+ return x++hashMapOps :: MemoryDatabase -> [BatchOp]+hashMapOps db =+ bestBlockOp (hBest db) <>+ blockHashOps (hBlock db) <>+ blockHeightOps (hHeight db) <>+ txOps (hTx db) <>+ spenderOps (hSpender db) <>+ balOps (hBalance db) <>+ addrTxOps (hAddrTx db) <>+ addrOutOps (hAddrOut db) <>+ maybeToList (mempoolOp <$> hMempool db) <>+ orphanOps (hOrphans db) <>+ unspentOps (hUnspent db)++cacheMapOps :: MemoryDatabase -> [BatchOp]+cacheMapOps db =+ balOps (hBalance db) <> maybeToList (mempoolOp <$> hMempool db) <>+ addrTxOps (hAddrTx db) <>+ unspentOps (hUnspent db)++bestBlockOp :: Maybe BlockHash -> [BatchOp]+bestBlockOp Nothing = []+bestBlockOp (Just b) = [insertOp BestKey b]++blockHashOps :: HashMap BlockHash BlockData -> [BatchOp]+blockHashOps = map (uncurry f) . M.toList+ where+ f = insertOp . BlockKey++blockHeightOps :: HashMap BlockHeight [BlockHash] -> [BatchOp]+blockHeightOps = map (uncurry f) . M.toList+ where+ f = insertOp . HeightKey++txOps :: HashMap TxHash TxData -> [BatchOp]+txOps = map (uncurry f) . M.toList+ where+ f = insertOp . TxKey++spenderOps :: HashMap TxHash (IntMap (Maybe Spender)) -> [BatchOp]+spenderOps = concatMap (uncurry f) . M.toList+ where+ f h = map (uncurry (g h)) . I.toList+ g h i (Just s) = insertOp (SpenderKey (OutPoint h (fromIntegral i))) s+ g h i Nothing = deleteOp (SpenderKey (OutPoint h (fromIntegral i)))++balOps :: HashMap Address BalVal -> [BatchOp]+balOps = map (uncurry f) . M.toList+ where+ f = insertOp . BalKey++addrTxOps ::+ HashMap Address (HashMap BlockRef (HashMap TxHash Bool)) -> [BatchOp]+addrTxOps = concat . concatMap (uncurry f) . M.toList+ where+ f a = map (uncurry (g a)) . M.toList+ g a b = map (uncurry (h a b)) . M.toList+ h a b t True =+ insertOp+ (AddrTxKey+ { addrTxKeyA = a+ , addrTxKeyT =+ BlockTx+ { blockTxBlock = b+ , blockTxHash = t+ }+ })+ ()+ h a b t False =+ deleteOp+ AddrTxKey+ { addrTxKeyA = a+ , addrTxKeyT =+ BlockTx+ { blockTxBlock = b+ , blockTxHash = t+ }+ }++addrOutOps ::+ HashMap Address (HashMap BlockRef (HashMap OutPoint (Maybe OutVal)))+ -> [BatchOp]+addrOutOps = concat . concatMap (uncurry f) . M.toList+ where+ f a = map (uncurry (g a)) . M.toList+ g a b = map (uncurry (h a b)) . M.toList+ h a b p (Just l) =+ insertOp+ (AddrOutKey {addrOutKeyA = a, addrOutKeyB = b, addrOutKeyP = p})+ l+ h a b p Nothing =+ deleteOp AddrOutKey {addrOutKeyA = a, addrOutKeyB = b, addrOutKeyP = p}++mempoolOp :: [BlockTx] -> BatchOp+mempoolOp =+ insertOp MemKey .+ map (\BlockTx {blockTxBlock = MemRef t, blockTxHash = h} -> (t, h))++orphanOps :: HashMap TxHash (Maybe (UnixTime, Tx)) -> [BatchOp]+orphanOps = map (uncurry f) . M.toList+ where+ f h (Just x) = insertOp (OrphanKey h) x+ f h Nothing = deleteOp (OrphanKey h)++unspentOps :: HashMap TxHash (IntMap (Maybe UnspentVal)) -> [BatchOp]+unspentOps = concatMap (uncurry f) . M.toList+ where+ f h = map (uncurry (g h)) . I.toList+ g h i (Just u) = insertOp (UnspentKey (OutPoint h (fromIntegral i))) u+ g h i Nothing = deleteOp (UnspentKey (OutPoint h (fromIntegral i)))++setBestI :: MonadIO m => BlockHash -> DatabaseWriter -> m ()+setBestI bh DatabaseWriter {databaseWriterState = hm} =+ withMemoryDatabase hm $ setBest bh++insertBlockI :: MonadIO m => BlockData -> DatabaseWriter -> m ()+insertBlockI b DatabaseWriter {databaseWriterState = hm} =+ withMemoryDatabase hm $ insertBlock b++setBlocksAtHeightI :: MonadIO m => [BlockHash] -> BlockHeight -> DatabaseWriter -> m ()+setBlocksAtHeightI hs g DatabaseWriter {databaseWriterState = hm} =+ withMemoryDatabase hm $ setBlocksAtHeight hs g++insertTxI :: MonadIO m => TxData -> DatabaseWriter -> m ()+insertTxI t DatabaseWriter {databaseWriterState = hm} =+ withMemoryDatabase hm $ insertTx t++insertSpenderI :: MonadIO m => OutPoint -> Spender -> DatabaseWriter -> m ()+insertSpenderI p s DatabaseWriter {databaseWriterState = hm} =+ withMemoryDatabase hm $ insertSpender p s++deleteSpenderI :: MonadIO m => OutPoint -> DatabaseWriter -> m ()+deleteSpenderI p DatabaseWriter {databaseWriterState = hm} =+ withMemoryDatabase hm $ deleteSpender p++insertAddrTxI :: MonadIO m => Address -> BlockTx -> DatabaseWriter -> m ()+insertAddrTxI a t DatabaseWriter {databaseWriterState = hm} =+ withMemoryDatabase hm $ insertAddrTx a t++deleteAddrTxI :: MonadIO m => Address -> BlockTx -> DatabaseWriter -> m ()+deleteAddrTxI a t DatabaseWriter {databaseWriterState = hm} =+ withMemoryDatabase hm $ deleteAddrTx a t++insertAddrUnspentI :: MonadIO m => Address -> Unspent -> DatabaseWriter -> m ()+insertAddrUnspentI a u DatabaseWriter {databaseWriterState = hm} =+ withMemoryDatabase hm $ insertAddrUnspent a u++deleteAddrUnspentI :: MonadIO m => Address -> Unspent -> DatabaseWriter -> m ()+deleteAddrUnspentI a u DatabaseWriter {databaseWriterState = hm} =+ withMemoryDatabase hm $ deleteAddrUnspent a u++setMempoolI :: MonadIO m => [BlockTx] -> DatabaseWriter -> m ()+setMempoolI xs DatabaseWriter {databaseWriterState = hm} = withMemoryDatabase hm $ setMempool xs++insertOrphanTxI :: MonadIO m => Tx -> UnixTime -> DatabaseWriter -> m ()+insertOrphanTxI t p DatabaseWriter {databaseWriterState = hm} =+ withMemoryDatabase hm $ insertOrphanTx t p++deleteOrphanTxI :: MonadIO m => TxHash -> DatabaseWriter -> m ()+deleteOrphanTxI t DatabaseWriter {databaseWriterState = hm} =+ withMemoryDatabase hm $ deleteOrphanTx t++getBestBlockI :: MonadIO m => DatabaseWriter -> m (Maybe BlockHash)+getBestBlockI DatabaseWriter {databaseWriterState = hm, databaseWriterReader = db} =+ runMaybeT $ MaybeT f <|> MaybeT g+ where+ f = withMemoryDatabase hm getBestBlock+ g = withDatabaseReader db getBestBlock++getBlocksAtHeightI :: MonadIO m => BlockHeight -> DatabaseWriter -> m [BlockHash]+getBlocksAtHeightI bh DatabaseWriter {databaseWriterState = hm, databaseWriterReader = db} = do+ xs <- withMemoryDatabase hm $ getBlocksAtHeight bh+ ys <- withDatabaseReader db $ getBlocksAtHeight bh+ return . nub $ xs <> ys++getBlockI :: MonadIO m => BlockHash -> DatabaseWriter -> m (Maybe BlockData)+getBlockI bh DatabaseWriter {databaseWriterReader = db, databaseWriterState = hm} =+ runMaybeT $ MaybeT f <|> MaybeT g+ where+ f = withMemoryDatabase hm $ getBlock bh+ g = withDatabaseReader db $ getBlock bh++getTxDataI ::+ MonadIO m => TxHash -> DatabaseWriter -> m (Maybe TxData)+getTxDataI th DatabaseWriter {databaseWriterReader = db, databaseWriterState = hm} =+ runMaybeT $ MaybeT f <|> MaybeT g+ where+ f = withMemoryDatabase hm $ getTxData th+ g = withDatabaseReader db $ getTxData th++getOrphanTxI :: MonadIO m => TxHash -> DatabaseWriter -> m (Maybe (UnixTime, Tx))+getOrphanTxI h DatabaseWriter {databaseWriterReader = db, databaseWriterState = hm} =+ fmap join . runMaybeT $ MaybeT f <|> MaybeT g+ where+ f = getOrphanTxH h <$> readTVarIO hm+ g = Just <$> withDatabaseReader db (getOrphanTx h)++getSpenderI :: MonadIO m => OutPoint -> DatabaseWriter -> m (Maybe Spender)+getSpenderI op DatabaseWriter {databaseWriterReader = db, databaseWriterState = hm} =+ fmap join . runMaybeT $ MaybeT f <|> MaybeT g+ where+ f = getSpenderH op <$> readTVarIO hm+ g = Just <$> withDatabaseReader db (getSpender op)++getSpendersI :: MonadIO m => TxHash -> DatabaseWriter -> m (IntMap Spender)+getSpendersI t DatabaseWriter {databaseWriterReader = db, databaseWriterState = hm} = do+ hsm <- getSpendersH t <$> readTVarIO hm+ dsm <- I.map Just <$> withDatabaseReader db (getSpenders t)+ return . I.map fromJust . I.filter isJust $ hsm <> dsm++getBalanceI :: MonadIO m => Address -> DatabaseWriter -> m Balance+getBalanceI a DatabaseWriter {databaseWriterReader = db, databaseWriterState = hm} =+ fromMaybe (zeroBalance a) <$> runMaybeT (MaybeT f <|> MaybeT g)+ where+ f =+ withMemoryDatabase hm $+ getBalance a >>= \b ->+ return $+ if nullBalance b+ then Nothing+ else Just b+ g =+ withDatabaseReader db $+ getBalance a >>= \b ->+ return $+ if nullBalance b+ then Nothing+ else Just b++setBalanceI :: MonadIO m => Balance -> DatabaseWriter -> m ()+setBalanceI b DatabaseWriter {databaseWriterState = hm} =+ withMemoryDatabase hm $ setBalance b++getUnspentI :: MonadIO m => OutPoint -> DatabaseWriter -> m (Maybe Unspent)+getUnspentI op DatabaseWriter {databaseWriterReader = db, databaseWriterState = hm} =+ fmap join . runMaybeT $ MaybeT f <|> MaybeT g+ where+ f = getUnspentH op <$> readTVarIO hm+ g = Just <$> withDatabaseReader db (getUnspent op)++insertUnspentI :: MonadIO m => Unspent -> DatabaseWriter -> m ()+insertUnspentI u DatabaseWriter {databaseWriterState = hm} =+ withMemoryDatabase hm $ insertUnspent u++deleteUnspentI :: MonadIO m => OutPoint -> DatabaseWriter -> m ()+deleteUnspentI p DatabaseWriter {databaseWriterState = hm} =+ withMemoryDatabase hm $ deleteUnspent p++getMempoolI ::+ MonadIO m+ => DatabaseWriter+ -> m [BlockTx]+getMempoolI DatabaseWriter {databaseWriterState = hm, databaseWriterReader = db} =+ getMempoolH <$> readTVarIO hm >>= \case+ Just xs -> return xs+ Nothing -> withDatabaseReader db getMempool++instance MonadIO m => StoreRead (ReaderT DatabaseWriter m) where+ getBestBlock = R.ask >>= getBestBlockI+ getBlocksAtHeight h = R.ask >>= getBlocksAtHeightI h+ getBlock b = R.ask >>= getBlockI b+ getTxData t = R.ask >>= getTxDataI t+ getSpender p = R.ask >>= getSpenderI p+ getSpenders t = R.ask >>= getSpendersI t+ getOrphanTx h = R.ask >>= getOrphanTxI h+ getUnspent a = R.ask >>= getUnspentI a+ getBalance a = R.ask >>= getBalanceI a+ getMempool = R.ask >>= getMempoolI+ getAddressesTxs = undefined+ getAddressesUnspents = undefined+ getOrphans = undefined++instance MonadIO m => StoreWrite (ReaderT DatabaseWriter m) where+ setBest h = R.ask >>= setBestI h+ insertBlock b = R.ask >>= insertBlockI b+ setBlocksAtHeight hs g = R.ask >>= setBlocksAtHeightI hs g+ insertTx t = R.ask >>= insertTxI t+ insertSpender p s = R.ask >>= insertSpenderI p s+ deleteSpender p = R.ask >>= deleteSpenderI p+ insertAddrTx a t = R.ask >>= insertAddrTxI a t+ deleteAddrTx a t = R.ask >>= deleteAddrTxI a t+ insertAddrUnspent a u = R.ask >>= insertAddrUnspentI a u+ deleteAddrUnspent a u = R.ask >>= deleteAddrUnspentI a u+ setMempool xs = R.ask >>= setMempoolI xs+ insertOrphanTx t p = R.ask >>= insertOrphanTxI t p+ deleteOrphanTx t = R.ask >>= deleteOrphanTxI t+ insertUnspent u = R.ask >>= insertUnspentI u+ deleteUnspent p = R.ask >>= deleteUnspentI p+ setBalance b = R.ask >>= setBalanceI b
− src/Network/Haskoin/Store/Data/ImportDB.hs
@@ -1,356 +0,0 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE OverloadedStrings #-}-module Network.Haskoin.Store.Data.ImportDB where--import Control.Applicative ((<|>))-import Control.Monad (join)-import Control.Monad.Except (MonadError)-import Control.Monad.Reader (ReaderT)-import qualified Control.Monad.Reader as R-import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT)-import Data.HashMap.Strict (HashMap)-import qualified Data.HashMap.Strict as M-import Data.IntMap.Strict (IntMap)-import qualified Data.IntMap.Strict as I-import Data.List (nub)-import Data.Maybe (fromJust, fromMaybe,- isJust, maybeToList)-import Database.RocksDB (BatchOp)-import Database.RocksDB.Query (deleteOp, insertOp,- writeBatch)-import Haskoin (Address, BlockHash,- BlockHeight,- OutPoint (..), Tx, TxHash)-import Network.Haskoin.Store.Common (BalVal, Balance,- BlockDB (..), BlockData,- BlockRef (..),- BlockTx (..), Spender,- StoreRead (..),- StoreWrite (..), TxData,- UnixTime, Unspent,- UnspentVal, nullBalance,- zeroBalance)-import Network.Haskoin.Store.Data.KeyValue (AddrOutKey (..),- AddrTxKey (..),- BalKey (..), BestKey (..),- BlockKey (..),- HeightKey (..),- MemKey (..),- OrphanKey (..), OutVal,- SpenderKey (..),- TxKey (..),- UnspentKey (..))-import Network.Haskoin.Store.Data.Memory (BlockMem (..),- emptyBlockMem,- getMempoolH, getOrphanTxH,- getSpenderH, getSpendersH,- getUnspentH, withBlockMem)-import Network.Haskoin.Store.Data.RocksDB (withRocksDB)-import UnliftIO (MonadIO, TVar, newTVarIO,- readTVarIO)--data ImportDB = ImportDB- { importDB :: !BlockDB- , importHashMap :: !(TVar BlockMem)- }--runImportDB ::- (MonadIO m, MonadError e m) => BlockDB -> ReaderT ImportDB m a -> m a-runImportDB bdb@BlockDB {blockDB = db} f = do- hm <- newTVarIO emptyBlockMem- x <- R.runReaderT f ImportDB {importDB = bdb, importHashMap = hm}- ops <- hashMapOps <$> readTVarIO hm- writeBatch db ops- return x--hashMapOps :: BlockMem -> [BatchOp]-hashMapOps db =- bestBlockOp (hBest db) <>- blockHashOps (hBlock db) <>- blockHeightOps (hHeight db) <>- txOps (hTx db) <>- spenderOps (hSpender db) <>- balOps (hBalance db) <>- addrTxOps (hAddrTx db) <>- addrOutOps (hAddrOut db) <>- maybeToList (mempoolOp <$> hMempool db) <>- orphanOps (hOrphans db) <>- unspentOps (hUnspent db)--cacheMapOps :: BlockMem -> [BatchOp]-cacheMapOps db =- balOps (hBalance db) <> maybeToList (mempoolOp <$> hMempool db) <>- addrTxOps (hAddrTx db) <>- unspentOps (hUnspent db)--bestBlockOp :: Maybe BlockHash -> [BatchOp]-bestBlockOp Nothing = []-bestBlockOp (Just b) = [insertOp BestKey b]--blockHashOps :: HashMap BlockHash BlockData -> [BatchOp]-blockHashOps = map (uncurry f) . M.toList- where- f = insertOp . BlockKey--blockHeightOps :: HashMap BlockHeight [BlockHash] -> [BatchOp]-blockHeightOps = map (uncurry f) . M.toList- where- f = insertOp . HeightKey--txOps :: HashMap TxHash TxData -> [BatchOp]-txOps = map (uncurry f) . M.toList- where- f = insertOp . TxKey--spenderOps :: HashMap TxHash (IntMap (Maybe Spender)) -> [BatchOp]-spenderOps = concatMap (uncurry f) . M.toList- where- f h = map (uncurry (g h)) . I.toList- g h i (Just s) = insertOp (SpenderKey (OutPoint h (fromIntegral i))) s- g h i Nothing = deleteOp (SpenderKey (OutPoint h (fromIntegral i)))--balOps :: HashMap Address BalVal -> [BatchOp]-balOps = map (uncurry f) . M.toList- where- f = insertOp . BalKey--addrTxOps ::- HashMap Address (HashMap BlockRef (HashMap TxHash Bool)) -> [BatchOp]-addrTxOps = concat . concatMap (uncurry f) . M.toList- where- f a = map (uncurry (g a)) . M.toList- g a b = map (uncurry (h a b)) . M.toList- h a b t True =- insertOp- (AddrTxKey- { addrTxKeyA = a- , addrTxKeyT =- BlockTx- { blockTxBlock = b- , blockTxHash = t- }- })- ()- h a b t False =- deleteOp- AddrTxKey- { addrTxKeyA = a- , addrTxKeyT =- BlockTx- { blockTxBlock = b- , blockTxHash = t- }- }--addrOutOps ::- HashMap Address (HashMap BlockRef (HashMap OutPoint (Maybe OutVal)))- -> [BatchOp]-addrOutOps = concat . concatMap (uncurry f) . M.toList- where- f a = map (uncurry (g a)) . M.toList- g a b = map (uncurry (h a b)) . M.toList- h a b p (Just l) =- insertOp- (AddrOutKey {addrOutKeyA = a, addrOutKeyB = b, addrOutKeyP = p})- l- h a b p Nothing =- deleteOp AddrOutKey {addrOutKeyA = a, addrOutKeyB = b, addrOutKeyP = p}--mempoolOp :: [BlockTx] -> BatchOp-mempoolOp =- insertOp MemKey .- map (\BlockTx {blockTxBlock = MemRef t, blockTxHash = h} -> (t, h))--orphanOps :: HashMap TxHash (Maybe (UnixTime, Tx)) -> [BatchOp]-orphanOps = map (uncurry f) . M.toList- where- f h (Just x) = insertOp (OrphanKey h) x- f h Nothing = deleteOp (OrphanKey h)--unspentOps :: HashMap TxHash (IntMap (Maybe UnspentVal)) -> [BatchOp]-unspentOps = concatMap (uncurry f) . M.toList- where- f h = map (uncurry (g h)) . I.toList- g h i (Just u) = insertOp (UnspentKey (OutPoint h (fromIntegral i))) u- g h i Nothing = deleteOp (UnspentKey (OutPoint h (fromIntegral i)))--setBestI :: MonadIO m => BlockHash -> ImportDB -> m ()-setBestI bh ImportDB {importHashMap = hm} =- withBlockMem hm $ setBest bh--insertBlockI :: MonadIO m => BlockData -> ImportDB -> m ()-insertBlockI b ImportDB {importHashMap = hm} =- withBlockMem hm $ insertBlock b--setBlocksAtHeightI :: MonadIO m => [BlockHash] -> BlockHeight -> ImportDB -> m ()-setBlocksAtHeightI hs g ImportDB {importHashMap = hm} =- withBlockMem hm $ setBlocksAtHeight hs g--insertTxI :: MonadIO m => TxData -> ImportDB -> m ()-insertTxI t ImportDB {importHashMap = hm} =- withBlockMem hm $ insertTx t--insertSpenderI :: MonadIO m => OutPoint -> Spender -> ImportDB -> m ()-insertSpenderI p s ImportDB {importHashMap = hm} =- withBlockMem hm $ insertSpender p s--deleteSpenderI :: MonadIO m => OutPoint -> ImportDB -> m ()-deleteSpenderI p ImportDB {importHashMap = hm} =- withBlockMem hm $ deleteSpender p--insertAddrTxI :: MonadIO m => Address -> BlockTx -> ImportDB -> m ()-insertAddrTxI a t ImportDB {importHashMap = hm} =- withBlockMem hm $ insertAddrTx a t--deleteAddrTxI :: MonadIO m => Address -> BlockTx -> ImportDB -> m ()-deleteAddrTxI a t ImportDB {importHashMap = hm} =- withBlockMem hm $ deleteAddrTx a t--insertAddrUnspentI :: MonadIO m => Address -> Unspent -> ImportDB -> m ()-insertAddrUnspentI a u ImportDB {importHashMap = hm} =- withBlockMem hm $ insertAddrUnspent a u--deleteAddrUnspentI :: MonadIO m => Address -> Unspent -> ImportDB -> m ()-deleteAddrUnspentI a u ImportDB {importHashMap = hm} =- withBlockMem hm $ deleteAddrUnspent a u--setMempoolI :: MonadIO m => [BlockTx] -> ImportDB -> m ()-setMempoolI xs ImportDB {importHashMap = hm} = withBlockMem hm $ setMempool xs--insertOrphanTxI :: MonadIO m => Tx -> UnixTime -> ImportDB -> m ()-insertOrphanTxI t p ImportDB {importHashMap = hm} =- withBlockMem hm $ insertOrphanTx t p--deleteOrphanTxI :: MonadIO m => TxHash -> ImportDB -> m ()-deleteOrphanTxI t ImportDB {importHashMap = hm} =- withBlockMem hm $ deleteOrphanTx t--getBestBlockI :: MonadIO m => ImportDB -> m (Maybe BlockHash)-getBestBlockI ImportDB {importHashMap = hm, importDB = db} =- runMaybeT $ MaybeT f <|> MaybeT g- where- f = withBlockMem hm getBestBlock- g = withRocksDB db getBestBlock--getBlocksAtHeightI :: MonadIO m => BlockHeight -> ImportDB -> m [BlockHash]-getBlocksAtHeightI bh ImportDB {importHashMap = hm, importDB = db} = do- xs <- withBlockMem hm $ getBlocksAtHeight bh- ys <- withRocksDB db $ getBlocksAtHeight bh- return . nub $ xs <> ys--getBlockI :: MonadIO m => BlockHash -> ImportDB -> m (Maybe BlockData)-getBlockI bh ImportDB {importDB = db, importHashMap = hm} =- runMaybeT $ MaybeT f <|> MaybeT g- where- f = withBlockMem hm $ getBlock bh- g = withRocksDB db $ getBlock bh--getTxDataI ::- MonadIO m => TxHash -> ImportDB -> m (Maybe TxData)-getTxDataI th ImportDB {importDB = db, importHashMap = hm} =- runMaybeT $ MaybeT f <|> MaybeT g- where- f = withBlockMem hm $ getTxData th- g = withRocksDB db $ getTxData th--getOrphanTxI :: MonadIO m => TxHash -> ImportDB -> m (Maybe (UnixTime, Tx))-getOrphanTxI h ImportDB {importDB = db, importHashMap = hm} =- fmap join . runMaybeT $ MaybeT f <|> MaybeT g- where- f = getOrphanTxH h <$> readTVarIO hm- g = Just <$> withRocksDB db (getOrphanTx h)--getSpenderI :: MonadIO m => OutPoint -> ImportDB -> m (Maybe Spender)-getSpenderI op ImportDB {importDB = db, importHashMap = hm} =- fmap join . runMaybeT $ MaybeT f <|> MaybeT g- where- f = getSpenderH op <$> readTVarIO hm- g = Just <$> withRocksDB db (getSpender op)--getSpendersI :: MonadIO m => TxHash -> ImportDB -> m (IntMap Spender)-getSpendersI t ImportDB {importDB = db, importHashMap = hm} = do- hsm <- getSpendersH t <$> readTVarIO hm- dsm <- I.map Just <$> withRocksDB db (getSpenders t)- return . I.map fromJust . I.filter isJust $ hsm <> dsm--getBalanceI :: MonadIO m => Address -> ImportDB -> m Balance-getBalanceI a ImportDB {importDB = db, importHashMap = hm} =- fromMaybe (zeroBalance a) <$> runMaybeT (MaybeT f <|> MaybeT g)- where- f =- withBlockMem hm $- getBalance a >>= \b ->- return $- if nullBalance b- then Nothing- else Just b- g =- withRocksDB db $- getBalance a >>= \b ->- return $- if nullBalance b- then Nothing- else Just b--setBalanceI :: MonadIO m => Balance -> ImportDB -> m ()-setBalanceI b ImportDB {importHashMap = hm} =- withBlockMem hm $ setBalance b--getUnspentI :: MonadIO m => OutPoint -> ImportDB -> m (Maybe Unspent)-getUnspentI op ImportDB {importDB = db, importHashMap = hm} =- fmap join . runMaybeT $ MaybeT f <|> MaybeT g- where- f = getUnspentH op <$> readTVarIO hm- g = Just <$> withRocksDB db (getUnspent op)--insertUnspentI :: MonadIO m => Unspent -> ImportDB -> m ()-insertUnspentI u ImportDB {importHashMap = hm} =- withBlockMem hm $ insertUnspent u--deleteUnspentI :: MonadIO m => OutPoint -> ImportDB -> m ()-deleteUnspentI p ImportDB {importHashMap = hm} =- withBlockMem hm $ deleteUnspent p--getMempoolI ::- MonadIO m- => ImportDB- -> m [BlockTx]-getMempoolI ImportDB {importHashMap = hm, importDB = db} =- getMempoolH <$> readTVarIO hm >>= \case- Just xs -> return xs- Nothing -> withRocksDB db getMempool--instance MonadIO m => StoreRead (ReaderT ImportDB m) where- getBestBlock = R.ask >>= getBestBlockI- getBlocksAtHeight h = R.ask >>= getBlocksAtHeightI h- getBlock b = R.ask >>= getBlockI b- getTxData t = R.ask >>= getTxDataI t- getSpender p = R.ask >>= getSpenderI p- getSpenders t = R.ask >>= getSpendersI t- getOrphanTx h = R.ask >>= getOrphanTxI h- getUnspent a = R.ask >>= getUnspentI a- getBalance a = R.ask >>= getBalanceI a- getMempool = R.ask >>= getMempoolI- getAddressesTxs = undefined- getAddressesUnspents = undefined- getOrphans = undefined--instance MonadIO m => StoreWrite (ReaderT ImportDB m) where- setBest h = R.ask >>= setBestI h- insertBlock b = R.ask >>= insertBlockI b- setBlocksAtHeight hs g = R.ask >>= setBlocksAtHeightI hs g- insertTx t = R.ask >>= insertTxI t- insertSpender p s = R.ask >>= insertSpenderI p s- deleteSpender p = R.ask >>= deleteSpenderI p- insertAddrTx a t = R.ask >>= insertAddrTxI a t- deleteAddrTx a t = R.ask >>= deleteAddrTxI a t- insertAddrUnspent a u = R.ask >>= insertAddrUnspentI a u- deleteAddrUnspent a u = R.ask >>= deleteAddrUnspentI a u- setMempool xs = R.ask >>= setMempoolI xs- insertOrphanTx t p = R.ask >>= insertOrphanTxI t p- deleteOrphanTx t = R.ask >>= deleteOrphanTxI t- insertUnspent u = R.ask >>= insertUnspentI u- deleteUnspent p = R.ask >>= deleteUnspentI p- setBalance b = R.ask >>= setBalanceI b
− src/Network/Haskoin/Store/Data/KeyValue.hs
@@ -1,360 +0,0 @@-{-# LANGUAGE DeriveAnyClass #-}-{-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE MultiParamTypeClasses #-}-module Network.Haskoin.Store.Data.KeyValue where--import Control.Monad (guard)-import Data.ByteString (ByteString)-import qualified Data.ByteString as B-import qualified Data.ByteString.Short as B.Short-import Data.Hashable (Hashable)-import Data.Serialize (Serialize (..), getBytes,- getWord8, putWord8)-import Data.Word (Word32, Word64)-import qualified Database.RocksDB.Query as R-import GHC.Generics (Generic)-import Haskoin (Address, BlockHash, BlockHeight,- OutPoint (..), Tx, TxHash)-import Network.Haskoin.Store.Common (BalVal, BlockData, BlockRef,- BlockTx (..), Spender, TxData,- UnixTime, Unspent (..),- UnspentVal, getUnixTime,- putUnixTime)---- | Database key for an address transaction.-data AddrTxKey- = AddrTxKey { addrTxKeyA :: !Address- , addrTxKeyT :: !BlockTx- }- -- ^ key for a transaction affecting an address- | AddrTxKeyA { addrTxKeyA :: !Address }- -- ^ short key that matches all entries- | AddrTxKeyB { addrTxKeyA :: !Address- , addrTxKeyB :: !BlockRef- }- | AddrTxKeyS- deriving (Show, Eq, Ord, Generic, Hashable)--instance Serialize AddrTxKey- -- 0x05 · Address · BlockRef · TxHash- where- put AddrTxKey { addrTxKeyA = a- , addrTxKeyT = BlockTx { blockTxBlock = b- , blockTxHash = t- }- } = do- putWord8 0x05- put a- put b- put t- -- 0x05 · Address- put AddrTxKeyA {addrTxKeyA = a} = do- putWord8 0x05- put a- -- 0x05 · Address · BlockRef- put AddrTxKeyB {addrTxKeyA = a, addrTxKeyB = b} = do- putWord8 0x05- put a- put b- -- 0x05- put AddrTxKeyS = putWord8 0x05- get = do- guard . (== 0x05) =<< getWord8- a <- get- b <- get- t <- get- return- AddrTxKey- { addrTxKeyA = a- , addrTxKeyT =- BlockTx- { blockTxBlock = b- , blockTxHash = t- }- }--instance R.Key AddrTxKey-instance R.KeyValue AddrTxKey ()---- | Database key for an address output.-data AddrOutKey- = AddrOutKey { addrOutKeyA :: !Address- , addrOutKeyB :: !BlockRef- , addrOutKeyP :: !OutPoint }- -- ^ full key- | AddrOutKeyA { addrOutKeyA :: !Address }- -- ^ short key for all spent or unspent outputs- | AddrOutKeyB { addrOutKeyA :: !Address- , addrOutKeyB :: !BlockRef- }- | AddrOutKeyS- deriving (Show, Read, Eq, Ord, Generic, Hashable)--instance Serialize AddrOutKey- -- 0x06 · StoreAddr · BlockRef · OutPoint- where- put AddrOutKey {addrOutKeyA = a, addrOutKeyB = b, addrOutKeyP = p} = do- putWord8 0x06- put a- put b- put p- -- 0x06 · StoreAddr · BlockRef- put AddrOutKeyB {addrOutKeyA = a, addrOutKeyB = b} = do- putWord8 0x06- put a- put b- -- 0x06 · StoreAddr- put AddrOutKeyA {addrOutKeyA = a} = do- putWord8 0x06- put a- -- 0x06- put AddrOutKeyS = putWord8 0x06- get = do- guard . (== 0x06) =<< getWord8- AddrOutKey <$> get <*> get <*> get--instance R.Key AddrOutKey--data OutVal = OutVal- { outValAmount :: !Word64- , outValScript :: !ByteString- } deriving (Show, Read, Eq, Ord, Generic, Hashable, Serialize)--instance R.KeyValue AddrOutKey OutVal---- | Transaction database key.-newtype TxKey = TxKey- { txKey :: TxHash- } deriving (Show, Read, Eq, Ord, Generic, Hashable)--instance Serialize TxKey where- -- 0x02 · TxHash- put (TxKey h) = do- putWord8 0x02- put h- get = do- guard . (== 0x02) =<< getWord8- TxKey <$> get--instance R.Key TxKey-instance R.KeyValue TxKey TxData--data SpenderKey- = SpenderKey { outputPoint :: !OutPoint }- | SpenderKeyS { outputKeyS :: !TxHash }- deriving (Show, Read, Eq, Ord, Generic, Hashable)--instance Serialize SpenderKey where- -- 0x10 · TxHash · Index- put (SpenderKey OutPoint {outPointHash = h, outPointIndex = i}) = do- putWord8 0x10- put h- put i- put (SpenderKeyS h) = do- putWord8 0x10- put h- get = do- guard . (== 0x10) =<< getWord8- op <- OutPoint <$> get <*> get- return $ SpenderKey op--instance R.Key SpenderKey-instance R.KeyValue SpenderKey Spender---- | Unspent output database key.-data UnspentKey- = UnspentKey { unspentKey :: !OutPoint }- | UnspentKeyS { unspentKeyS :: !TxHash }- | UnspentKeyB- deriving (Show, Read, Eq, Ord, Generic, Hashable)--instance Serialize UnspentKey- -- 0x09 · TxHash · Index- where- put UnspentKey {unspentKey = OutPoint {outPointHash = h, outPointIndex = i}} = do- putWord8 0x09- put h- put i- -- 0x09 · TxHash- put UnspentKeyS {unspentKeyS = t} = do- putWord8 0x09- put t- -- 0x09- put UnspentKeyB = putWord8 0x09- get = do- guard . (== 0x09) =<< getWord8- h <- get- i <- get- return $ UnspentKey OutPoint {outPointHash = h, outPointIndex = i}--instance R.Key UnspentKey-instance R.KeyValue UnspentKey UnspentVal--toUnspent :: AddrOutKey -> OutVal -> Unspent-toUnspent AddrOutKey {addrOutKeyB = b, addrOutKeyP = p} OutVal { outValAmount = v- , outValScript = s- } =- Unspent- { unspentBlock = b- , unspentAmount = v- , unspentScript = B.Short.toShort s- , unspentPoint = p- }-toUnspent _ _ = undefined---- | Mempool transaction database key.-data MemKey =- MemKey- deriving (Show, Read)--instance Serialize MemKey where- -- 0x07- put MemKey = do- putWord8 0x07- get = do- guard . (== 0x07) =<< getWord8- return MemKey--instance R.Key MemKey-instance R.KeyValue MemKey [(UnixTime, TxHash)]---- | Orphan pool transaction database key.-data OrphanKey- = OrphanKey- { orphanKey :: !TxHash- }- | OrphanKeyS- deriving (Show, Read, Eq, Ord, Generic, Hashable)--instance Serialize OrphanKey- -- 0x08 · TxHash- where- put (OrphanKey h) = do- putWord8 0x08- put h- -- 0x08- put OrphanKeyS = putWord8 0x08- get = do- guard . (== 0x08) =<< getWord8- OrphanKey <$> get--instance R.Key OrphanKey-instance R.KeyValue OrphanKey (UnixTime, Tx)---- | Block entry database key.-newtype BlockKey = BlockKey- { blockKey :: BlockHash- } deriving (Show, Read, Eq, Ord, Generic, Hashable)--instance Serialize BlockKey where- -- 0x01 · BlockHash- put (BlockKey h) = do- putWord8 0x01- put h- get = do- guard . (== 0x01) =<< getWord8- BlockKey <$> get--instance R.KeyValue BlockKey BlockData---- | Block height database key.-newtype HeightKey = HeightKey- { heightKey :: BlockHeight- } deriving (Show, Read, Eq, Ord, Generic, Hashable)--instance Serialize HeightKey where- -- 0x03 · BlockHeight- put (HeightKey height) = do- putWord8 0x03- put height- get = do- guard . (== 0x03) =<< getWord8- HeightKey <$> get--instance R.Key HeightKey-instance R.KeyValue HeightKey [BlockHash]---- | Address balance database key.-data BalKey- = BalKey- { balanceKey :: !Address- }- | BalKeyS- deriving (Show, Read, Eq, Ord, Generic, Hashable)--instance Serialize BalKey where- -- 0x04 · Address- put BalKey {balanceKey = a} = do- putWord8 0x04- put a- -- 0x04- put BalKeyS = putWord8 0x04- get = do- guard . (== 0x04) =<< getWord8- BalKey <$> get--instance R.Key BalKey-instance R.KeyValue BalKey BalVal---- | Key for best block in database.-data BestKey =- BestKey- deriving (Show, Read, Eq, Ord, Generic, Hashable)--instance Serialize BestKey where- -- 0x00 × 32- put BestKey = put (B.replicate 32 0x00)- get = do- guard . (== B.replicate 32 0x00) =<< getBytes 32- return BestKey--instance R.Key BestKey-instance R.KeyValue BestKey BlockHash---- | Key for database version.-data VersionKey =- VersionKey- deriving (Eq, Show, Read, Ord, Generic, Hashable)--instance Serialize VersionKey where- -- 0x0a- put VersionKey = putWord8 0x0a- get = do- guard . (== 0x0a) =<< getWord8- return VersionKey--instance R.Key VersionKey-instance R.KeyValue VersionKey Word32----- | Old mempool transaction database key.-data OldMemKey- = OldMemKey- { memTime :: !UnixTime- , memKey :: !TxHash- }- | OldMemKeyT- { memTime :: !UnixTime- }- | OldMemKeyS- deriving (Show, Read, Eq, Ord, Generic, Hashable)--instance Serialize OldMemKey where- -- 0x07 · UnixTime · TxHash- put (OldMemKey t h) = do- putWord8 0x07- putUnixTime t- put h- -- 0x07 · UnixTime- put (OldMemKeyT t) = do- putWord8 0x07- putUnixTime t- -- 0x07- put OldMemKeyS = putWord8 0x07- get = do- guard . (== 0x07) =<< getWord8- OldMemKey <$> getUnixTime <*> get--instance R.Key OldMemKey-instance R.KeyValue OldMemKey ()
− src/Network/Haskoin/Store/Data/Memory.hs
@@ -1,392 +0,0 @@-{-# LANGUAGE FlexibleInstances #-}-module Network.Haskoin.Store.Data.Memory where--import Conduit (ConduitT, mapC, yieldMany,- (.|))-import Control.Monad (join)-import Control.Monad.Reader (ReaderT)-import qualified Control.Monad.Reader as R-import qualified Data.ByteString.Short as B.Short-import Data.Function (on)-import Data.HashMap.Strict (HashMap)-import qualified Data.HashMap.Strict as M-import Data.IntMap.Strict (IntMap)-import qualified Data.IntMap.Strict as I-import Data.List (nub, sortBy)-import Data.Maybe (catMaybes, fromJust,- fromMaybe, isJust,- maybeToList)-import Haskoin (Address, BlockHash,- BlockHeight,- OutPoint (..), Tx, TxHash,- headerHash, txHash)-import Network.Haskoin.Store.Common (BalVal, Balance,- BlockData (..), BlockRef,- BlockTx (..), Limit,- Spender, StoreRead (..),- StoreWrite (..),- TxData (..), UnixTime,- Unspent (..), UnspentVal,- applyLimit, balanceToVal,- unspentToVal,- valToBalance,- valToUnspent, zeroBalance)-import Network.Haskoin.Store.Data.KeyValue (OutVal (..))-import UnliftIO--withBlockMem :: MonadIO m => TVar BlockMem -> ReaderT (TVar BlockMem) m a -> m a-withBlockMem = flip R.runReaderT--data BlockMem = BlockMem- { hBest :: !(Maybe BlockHash)- , hBlock :: !(HashMap BlockHash BlockData)- , hHeight :: !(HashMap BlockHeight [BlockHash])- , hTx :: !(HashMap TxHash TxData)- , hSpender :: !(HashMap TxHash (IntMap (Maybe Spender)))- , hUnspent :: !(HashMap TxHash (IntMap (Maybe UnspentVal)))- , hBalance :: !(HashMap Address BalVal)- , hAddrTx :: !(HashMap Address (HashMap BlockRef (HashMap TxHash Bool)))- , hAddrOut :: !(HashMap Address (HashMap BlockRef (HashMap OutPoint (Maybe OutVal))))- , hMempool :: !(Maybe [BlockTx])- , hOrphans :: !(HashMap TxHash (Maybe (UnixTime, Tx)))- } deriving (Eq, Show)--emptyBlockMem :: BlockMem-emptyBlockMem =- BlockMem- { hBest = Nothing- , hBlock = M.empty- , hHeight = M.empty- , hTx = M.empty- , hSpender = M.empty- , hUnspent = M.empty- , hBalance = M.empty- , hAddrTx = M.empty- , hAddrOut = M.empty- , hMempool = Nothing- , hOrphans = M.empty- }--getBestBlockH :: BlockMem -> Maybe BlockHash-getBestBlockH = hBest--getBlocksAtHeightH :: BlockHeight -> BlockMem -> [BlockHash]-getBlocksAtHeightH h = M.lookupDefault [] h . hHeight--getBlockH :: BlockHash -> BlockMem -> Maybe BlockData-getBlockH h = M.lookup h . hBlock--getTxDataH :: TxHash -> BlockMem -> Maybe TxData-getTxDataH t = M.lookup t . hTx--getSpenderH :: OutPoint -> BlockMem -> Maybe (Maybe Spender)-getSpenderH op db = do- m <- M.lookup (outPointHash op) (hSpender db)- I.lookup (fromIntegral (outPointIndex op)) m--getSpendersH :: TxHash -> BlockMem -> IntMap (Maybe Spender)-getSpendersH t = M.lookupDefault I.empty t . hSpender--getBalanceH :: Address -> BlockMem -> Balance-getBalanceH a =- fromMaybe (zeroBalance a) . fmap (valToBalance a) . M.lookup a . hBalance--getMempoolH :: BlockMem -> Maybe [BlockTx]-getMempoolH = hMempool--getOrphansH :: BlockMem -> [(UnixTime, Tx)]-getOrphansH = catMaybes . M.elems . hOrphans--getOrphanTxH :: TxHash -> BlockMem -> Maybe (Maybe (UnixTime, Tx))-getOrphanTxH h = M.lookup h . hOrphans--getUnspentsH :: Monad m => BlockMem -> ConduitT i Unspent m ()-getUnspentsH BlockMem {hUnspent = us} =- yieldMany- [ u- | (h, m) <- M.toList us- , (i, mv) <- I.toList m- , v <- maybeToList mv- , let p = OutPoint h (fromIntegral i)- , let u = valToUnspent p v- ]--getAddressesTxsH ::- [Address] -> Maybe BlockRef -> Maybe Limit -> BlockMem -> [BlockTx]-getAddressesTxsH addrs start limit db = applyLimit limit xs- where- xs =- nub . sortBy (flip compare `on` blockTxBlock) . concat $- map (\a -> getAddressTxsH a start limit db) addrs--getAddressTxsH ::- Address -> Maybe BlockRef -> Maybe Limit -> BlockMem -> [BlockTx]-getAddressTxsH addr start limit db =- applyLimit limit .- dropWhile h .- sortBy (flip compare) . catMaybes . concatMap (uncurry f) . M.toList $- M.lookupDefault M.empty addr (hAddrTx db)- where- f b hm = map (uncurry (g b)) $ M.toList hm- g b h' True = Just BlockTx {blockTxBlock = b, blockTxHash = h'}- g _ _ False = Nothing- h BlockTx {blockTxBlock = b} =- case start of- Nothing -> False- Just br -> b > br--getAddressBalancesH :: Monad m => BlockMem -> ConduitT i Balance m ()-getAddressBalancesH BlockMem {hBalance = bm} =- yieldMany (M.toList bm) .| mapC (uncurry valToBalance)--getAddressesUnspentsH ::- [Address] -> Maybe BlockRef -> Maybe Limit -> BlockMem -> [Unspent]-getAddressesUnspentsH addrs start limit db = applyLimit limit xs- where- xs =- nub . sortBy (flip compare `on` unspentBlock) . concat $- map (\a -> getAddressUnspentsH a start limit db) addrs--getAddressUnspentsH ::- Address -> Maybe BlockRef -> Maybe Limit -> BlockMem -> [Unspent]-getAddressUnspentsH addr start limit db =- applyLimit limit .- dropWhile h .- sortBy (flip compare) . catMaybes . concatMap (uncurry f) . M.toList $- M.lookupDefault M.empty addr (hAddrOut db)- where- f b hm = map (uncurry (g b)) $ M.toList hm- g b p (Just u) =- Just- Unspent- { unspentBlock = b- , unspentAmount = outValAmount u- , unspentScript = B.Short.toShort (outValScript u)- , unspentPoint = p- }- g _ _ Nothing = Nothing- h Unspent {unspentBlock = b} =- case start of- Nothing -> False- Just br -> b > br--setBestH :: BlockHash -> BlockMem -> BlockMem-setBestH h db = db {hBest = Just h}--insertBlockH :: BlockData -> BlockMem -> BlockMem-insertBlockH bd db =- db {hBlock = M.insert (headerHash (blockDataHeader bd)) bd (hBlock db)}--setBlocksAtHeightH :: [BlockHash] -> BlockHeight -> BlockMem -> BlockMem-setBlocksAtHeightH hs g db = db {hHeight = M.insert g hs (hHeight db)}--insertTxH :: TxData -> BlockMem -> BlockMem-insertTxH tx db = db {hTx = M.insert (txHash (txData tx)) tx (hTx db)}--insertSpenderH :: OutPoint -> Spender -> BlockMem -> BlockMem-insertSpenderH op s db =- db- { hSpender =- M.insertWith- (<>)- (outPointHash op)- (I.singleton (fromIntegral (outPointIndex op)) (Just s))- (hSpender db)- }--deleteSpenderH :: OutPoint -> BlockMem -> BlockMem-deleteSpenderH op db =- db- { hSpender =- M.insertWith- (<>)- (outPointHash op)- (I.singleton (fromIntegral (outPointIndex op)) Nothing)- (hSpender db)- }--setBalanceH :: Balance -> BlockMem -> BlockMem-setBalanceH bal db = db {hBalance = M.insert a b (hBalance db)}- where- (a, b) = balanceToVal bal--insertAddrTxH :: Address -> BlockTx -> BlockMem -> BlockMem-insertAddrTxH a btx db =- let s =- M.singleton- a- (M.singleton- (blockTxBlock btx)- (M.singleton (blockTxHash btx) True))- in db {hAddrTx = M.unionWith (M.unionWith M.union) s (hAddrTx db)}--deleteAddrTxH :: Address -> BlockTx -> BlockMem -> BlockMem-deleteAddrTxH a btx db =- let s =- M.singleton- a- (M.singleton- (blockTxBlock btx)- (M.singleton (blockTxHash btx) False))- in db {hAddrTx = M.unionWith (M.unionWith M.union) s (hAddrTx db)}--insertAddrUnspentH :: Address -> Unspent -> BlockMem -> BlockMem-insertAddrUnspentH a u db =- let uns =- OutVal- { outValAmount = unspentAmount u- , outValScript = B.Short.fromShort (unspentScript u)- }- s =- M.singleton- a- (M.singleton- (unspentBlock u)- (M.singleton (unspentPoint u) (Just uns)))- in db {hAddrOut = M.unionWith (M.unionWith M.union) s (hAddrOut db)}--deleteAddrUnspentH :: Address -> Unspent -> BlockMem -> BlockMem-deleteAddrUnspentH a u db =- let s =- M.singleton- a- (M.singleton- (unspentBlock u)- (M.singleton (unspentPoint u) Nothing))- in db {hAddrOut = M.unionWith (M.unionWith M.union) s (hAddrOut db)}--setMempoolH :: [BlockTx] -> BlockMem -> BlockMem-setMempoolH xs db = db {hMempool = Just xs}--insertOrphanTxH :: Tx -> UnixTime -> BlockMem -> BlockMem-insertOrphanTxH tx u db =- db {hOrphans = M.insert (txHash tx) (Just (u, tx)) (hOrphans db)}--deleteOrphanTxH :: TxHash -> BlockMem -> BlockMem-deleteOrphanTxH h db = db {hOrphans = M.insert h Nothing (hOrphans db)}--getUnspentH :: OutPoint -> BlockMem -> Maybe (Maybe Unspent)-getUnspentH op db = do- m <- M.lookup (outPointHash op) (hUnspent db)- fmap (valToUnspent op) <$> I.lookup (fromIntegral (outPointIndex op)) m--insertUnspentH :: Unspent -> BlockMem -> BlockMem-insertUnspentH u db =- db- { hUnspent =- M.insertWith- (<>)- (outPointHash (unspentPoint u))- (I.singleton- (fromIntegral (outPointIndex (unspentPoint u)))- (Just (snd (unspentToVal u))))- (hUnspent db)- }--deleteUnspentH :: OutPoint -> BlockMem -> BlockMem-deleteUnspentH op db =- db- { hUnspent =- M.insertWith- (<>)- (outPointHash op)- (I.singleton (fromIntegral (outPointIndex op)) Nothing)- (hUnspent db)- }--instance MonadIO m => StoreRead (ReaderT (TVar BlockMem) m) where- getBestBlock = do- v <- R.ask >>= readTVarIO- return $ getBestBlockH v- getBlocksAtHeight h = do- v <- R.ask >>= readTVarIO- return $ getBlocksAtHeightH h v- getBlock b = do- v <- R.ask >>= readTVarIO- return $ getBlockH b v- getTxData t = do- v <- R.ask >>= readTVarIO- return $ getTxDataH t v- getSpender t = do- v <- R.ask >>= readTVarIO- return . join $ getSpenderH t v- getSpenders t = do- v <- R.ask >>= readTVarIO- return . I.map fromJust . I.filter isJust $ getSpendersH t v- getOrphanTx h = do- v <- R.ask >>= readTVarIO- return . join $ getOrphanTxH h v- getUnspent p = do- v <- R.ask >>= readTVarIO- return . join $ getUnspentH p v- getBalance a = do- v <- R.ask >>= readTVarIO- return $ getBalanceH a v- getMempool = do- v <- R.ask >>= readTVarIO- return . fromMaybe [] $ getMempoolH v- getAddressesTxs addr start limit = do- v <- R.ask >>= readTVarIO- return $ getAddressesTxsH addr start limit v- getAddressesUnspents addr start limit = do- v <- R.ask >>= readTVarIO- return $ getAddressesUnspentsH addr start limit v- getOrphans = do- v <- R.ask >>= readTVarIO- return $ getOrphansH v- getAddressTxs addr start limit = do- v <- R.ask >>= readTVarIO- return $ getAddressTxsH addr start limit v- getAddressUnspents addr start limit = do- v <- R.ask >>= readTVarIO- return $ getAddressUnspentsH addr start limit v--instance (MonadIO m) => StoreWrite (ReaderT (TVar BlockMem) m) where- setBest h = do- v <- R.ask- atomically $ modifyTVar v (setBestH h)- insertBlock b = do- v <- R.ask- atomically $ modifyTVar v (insertBlockH b)- setBlocksAtHeight h g = do- v <- R.ask- atomically $ modifyTVar v (setBlocksAtHeightH h g)- insertTx t = do- v <- R.ask- atomically $ modifyTVar v (insertTxH t)- insertSpender p s = do- v <- R.ask- atomically $ modifyTVar v (insertSpenderH p s)- deleteSpender p = do- v <- R.ask- atomically $ modifyTVar v (deleteSpenderH p)- insertAddrTx a t = do- v <- R.ask- atomically $ modifyTVar v (insertAddrTxH a t)- deleteAddrTx a t = do- v <- R.ask- atomically $ modifyTVar v (deleteAddrTxH a t)- insertAddrUnspent a u = do- v <- R.ask- atomically $ modifyTVar v (insertAddrUnspentH a u)- deleteAddrUnspent a u = do- v <- R.ask- atomically $ modifyTVar v (deleteAddrUnspentH a u)- setMempool xs = do- v <- R.ask- atomically $ modifyTVar v (setMempoolH xs)- insertOrphanTx t u = do- v <- R.ask- atomically $ modifyTVar v (insertOrphanTxH t u)- deleteOrphanTx h = do- v <- R.ask- atomically $ modifyTVar v (deleteOrphanTxH h)- setBalance b = do- v <- R.ask- atomically $ modifyTVar v (setBalanceH b)- insertUnspent h = do- v <- R.ask- atomically $ modifyTVar v (insertUnspentH h)- deleteUnspent p = do- v <- R.ask- atomically $ modifyTVar v (deleteUnspentH p)
+ src/Network/Haskoin/Store/Data/MemoryDatabase.hs view
@@ -0,0 +1,391 @@+{-# LANGUAGE FlexibleInstances #-}+module Network.Haskoin.Store.Data.MemoryDatabase where++import Conduit (ConduitT, mapC, yieldMany,+ (.|))+import Control.Monad (join)+import Control.Monad.Reader (ReaderT)+import qualified Control.Monad.Reader as R+import qualified Data.ByteString.Short as B.Short+import Data.Function (on)+import Data.HashMap.Strict (HashMap)+import qualified Data.HashMap.Strict as M+import Data.IntMap.Strict (IntMap)+import qualified Data.IntMap.Strict as I+import Data.List (nub, sortBy)+import Data.Maybe (catMaybes, fromJust,+ fromMaybe, isJust,+ maybeToList)+import Haskoin (Address, BlockHash,+ BlockHeight, OutPoint (..),+ Tx, TxHash, headerHash,+ txHash)+import Network.Haskoin.Store.Common (BalVal, Balance,+ BlockData (..), BlockRef,+ BlockTx (..), Limit, Spender,+ StoreRead (..),+ StoreWrite (..), TxData (..),+ UnixTime, Unspent (..),+ UnspentVal, applyLimit,+ balanceToVal, unspentToVal,+ valToBalance, valToUnspent,+ zeroBalance)+import Network.Haskoin.Store.Data.Types (OutVal (..))+import UnliftIO++withMemoryDatabase :: MonadIO m => TVar MemoryDatabase -> ReaderT (TVar MemoryDatabase) m a -> m a+withMemoryDatabase = flip R.runReaderT++data MemoryDatabase = MemoryDatabase+ { hBest :: !(Maybe BlockHash)+ , hBlock :: !(HashMap BlockHash BlockData)+ , hHeight :: !(HashMap BlockHeight [BlockHash])+ , hTx :: !(HashMap TxHash TxData)+ , hSpender :: !(HashMap TxHash (IntMap (Maybe Spender)))+ , hUnspent :: !(HashMap TxHash (IntMap (Maybe UnspentVal)))+ , hBalance :: !(HashMap Address BalVal)+ , hAddrTx :: !(HashMap Address (HashMap BlockRef (HashMap TxHash Bool)))+ , hAddrOut :: !(HashMap Address (HashMap BlockRef (HashMap OutPoint (Maybe OutVal))))+ , hMempool :: !(Maybe [BlockTx])+ , hOrphans :: !(HashMap TxHash (Maybe (UnixTime, Tx)))+ } deriving (Eq, Show)++emptyMemoryDatabase :: MemoryDatabase+emptyMemoryDatabase =+ MemoryDatabase+ { hBest = Nothing+ , hBlock = M.empty+ , hHeight = M.empty+ , hTx = M.empty+ , hSpender = M.empty+ , hUnspent = M.empty+ , hBalance = M.empty+ , hAddrTx = M.empty+ , hAddrOut = M.empty+ , hMempool = Nothing+ , hOrphans = M.empty+ }++getBestBlockH :: MemoryDatabase -> Maybe BlockHash+getBestBlockH = hBest++getBlocksAtHeightH :: BlockHeight -> MemoryDatabase -> [BlockHash]+getBlocksAtHeightH h = M.lookupDefault [] h . hHeight++getBlockH :: BlockHash -> MemoryDatabase -> Maybe BlockData+getBlockH h = M.lookup h . hBlock++getTxDataH :: TxHash -> MemoryDatabase -> Maybe TxData+getTxDataH t = M.lookup t . hTx++getSpenderH :: OutPoint -> MemoryDatabase -> Maybe (Maybe Spender)+getSpenderH op db = do+ m <- M.lookup (outPointHash op) (hSpender db)+ I.lookup (fromIntegral (outPointIndex op)) m++getSpendersH :: TxHash -> MemoryDatabase -> IntMap (Maybe Spender)+getSpendersH t = M.lookupDefault I.empty t . hSpender++getBalanceH :: Address -> MemoryDatabase -> Balance+getBalanceH a =+ fromMaybe (zeroBalance a) . fmap (valToBalance a) . M.lookup a . hBalance++getMempoolH :: MemoryDatabase -> Maybe [BlockTx]+getMempoolH = hMempool++getOrphansH :: MemoryDatabase -> [(UnixTime, Tx)]+getOrphansH = catMaybes . M.elems . hOrphans++getOrphanTxH :: TxHash -> MemoryDatabase -> Maybe (Maybe (UnixTime, Tx))+getOrphanTxH h = M.lookup h . hOrphans++getUnspentsH :: Monad m => MemoryDatabase -> ConduitT i Unspent m ()+getUnspentsH MemoryDatabase {hUnspent = us} =+ yieldMany+ [ u+ | (h, m) <- M.toList us+ , (i, mv) <- I.toList m+ , v <- maybeToList mv+ , let p = OutPoint h (fromIntegral i)+ , let u = valToUnspent p v+ ]++getAddressesTxsH ::+ [Address] -> Maybe BlockRef -> Maybe Limit -> MemoryDatabase -> [BlockTx]+getAddressesTxsH addrs start limit db = applyLimit limit xs+ where+ xs =+ nub . sortBy (flip compare `on` blockTxBlock) . concat $+ map (\a -> getAddressTxsH a start limit db) addrs++getAddressTxsH ::+ Address -> Maybe BlockRef -> Maybe Limit -> MemoryDatabase -> [BlockTx]+getAddressTxsH addr start limit db =+ applyLimit limit .+ dropWhile h .+ sortBy (flip compare) . catMaybes . concatMap (uncurry f) . M.toList $+ M.lookupDefault M.empty addr (hAddrTx db)+ where+ f b hm = map (uncurry (g b)) $ M.toList hm+ g b h' True = Just BlockTx {blockTxBlock = b, blockTxHash = h'}+ g _ _ False = Nothing+ h BlockTx {blockTxBlock = b} =+ case start of+ Nothing -> False+ Just br -> b > br++getAddressBalancesH :: Monad m => MemoryDatabase -> ConduitT i Balance m ()+getAddressBalancesH MemoryDatabase {hBalance = bm} =+ yieldMany (M.toList bm) .| mapC (uncurry valToBalance)++getAddressesUnspentsH ::+ [Address] -> Maybe BlockRef -> Maybe Limit -> MemoryDatabase -> [Unspent]+getAddressesUnspentsH addrs start limit db = applyLimit limit xs+ where+ xs =+ nub . sortBy (flip compare `on` unspentBlock) . concat $+ map (\a -> getAddressUnspentsH a start limit db) addrs++getAddressUnspentsH ::+ Address -> Maybe BlockRef -> Maybe Limit -> MemoryDatabase -> [Unspent]+getAddressUnspentsH addr start limit db =+ applyLimit limit .+ dropWhile h .+ sortBy (flip compare) . catMaybes . concatMap (uncurry f) . M.toList $+ M.lookupDefault M.empty addr (hAddrOut db)+ where+ f b hm = map (uncurry (g b)) $ M.toList hm+ g b p (Just u) =+ Just+ Unspent+ { unspentBlock = b+ , unspentAmount = outValAmount u+ , unspentScript = B.Short.toShort (outValScript u)+ , unspentPoint = p+ }+ g _ _ Nothing = Nothing+ h Unspent {unspentBlock = b} =+ case start of+ Nothing -> False+ Just br -> b > br++setBestH :: BlockHash -> MemoryDatabase -> MemoryDatabase+setBestH h db = db {hBest = Just h}++insertBlockH :: BlockData -> MemoryDatabase -> MemoryDatabase+insertBlockH bd db =+ db {hBlock = M.insert (headerHash (blockDataHeader bd)) bd (hBlock db)}++setBlocksAtHeightH :: [BlockHash] -> BlockHeight -> MemoryDatabase -> MemoryDatabase+setBlocksAtHeightH hs g db = db {hHeight = M.insert g hs (hHeight db)}++insertTxH :: TxData -> MemoryDatabase -> MemoryDatabase+insertTxH tx db = db {hTx = M.insert (txHash (txData tx)) tx (hTx db)}++insertSpenderH :: OutPoint -> Spender -> MemoryDatabase -> MemoryDatabase+insertSpenderH op s db =+ db+ { hSpender =+ M.insertWith+ (<>)+ (outPointHash op)+ (I.singleton (fromIntegral (outPointIndex op)) (Just s))+ (hSpender db)+ }++deleteSpenderH :: OutPoint -> MemoryDatabase -> MemoryDatabase+deleteSpenderH op db =+ db+ { hSpender =+ M.insertWith+ (<>)+ (outPointHash op)+ (I.singleton (fromIntegral (outPointIndex op)) Nothing)+ (hSpender db)+ }++setBalanceH :: Balance -> MemoryDatabase -> MemoryDatabase+setBalanceH bal db = db {hBalance = M.insert a b (hBalance db)}+ where+ (a, b) = balanceToVal bal++insertAddrTxH :: Address -> BlockTx -> MemoryDatabase -> MemoryDatabase+insertAddrTxH a btx db =+ let s =+ M.singleton+ a+ (M.singleton+ (blockTxBlock btx)+ (M.singleton (blockTxHash btx) True))+ in db {hAddrTx = M.unionWith (M.unionWith M.union) s (hAddrTx db)}++deleteAddrTxH :: Address -> BlockTx -> MemoryDatabase -> MemoryDatabase+deleteAddrTxH a btx db =+ let s =+ M.singleton+ a+ (M.singleton+ (blockTxBlock btx)+ (M.singleton (blockTxHash btx) False))+ in db {hAddrTx = M.unionWith (M.unionWith M.union) s (hAddrTx db)}++insertAddrUnspentH :: Address -> Unspent -> MemoryDatabase -> MemoryDatabase+insertAddrUnspentH a u db =+ let uns =+ OutVal+ { outValAmount = unspentAmount u+ , outValScript = B.Short.fromShort (unspentScript u)+ }+ s =+ M.singleton+ a+ (M.singleton+ (unspentBlock u)+ (M.singleton (unspentPoint u) (Just uns)))+ in db {hAddrOut = M.unionWith (M.unionWith M.union) s (hAddrOut db)}++deleteAddrUnspentH :: Address -> Unspent -> MemoryDatabase -> MemoryDatabase+deleteAddrUnspentH a u db =+ let s =+ M.singleton+ a+ (M.singleton+ (unspentBlock u)+ (M.singleton (unspentPoint u) Nothing))+ in db {hAddrOut = M.unionWith (M.unionWith M.union) s (hAddrOut db)}++setMempoolH :: [BlockTx] -> MemoryDatabase -> MemoryDatabase+setMempoolH xs db = db {hMempool = Just xs}++insertOrphanTxH :: Tx -> UnixTime -> MemoryDatabase -> MemoryDatabase+insertOrphanTxH tx u db =+ db {hOrphans = M.insert (txHash tx) (Just (u, tx)) (hOrphans db)}++deleteOrphanTxH :: TxHash -> MemoryDatabase -> MemoryDatabase+deleteOrphanTxH h db = db {hOrphans = M.insert h Nothing (hOrphans db)}++getUnspentH :: OutPoint -> MemoryDatabase -> Maybe (Maybe Unspent)+getUnspentH op db = do+ m <- M.lookup (outPointHash op) (hUnspent db)+ fmap (valToUnspent op) <$> I.lookup (fromIntegral (outPointIndex op)) m++insertUnspentH :: Unspent -> MemoryDatabase -> MemoryDatabase+insertUnspentH u db =+ db+ { hUnspent =+ M.insertWith+ (<>)+ (outPointHash (unspentPoint u))+ (I.singleton+ (fromIntegral (outPointIndex (unspentPoint u)))+ (Just (snd (unspentToVal u))))+ (hUnspent db)+ }++deleteUnspentH :: OutPoint -> MemoryDatabase -> MemoryDatabase+deleteUnspentH op db =+ db+ { hUnspent =+ M.insertWith+ (<>)+ (outPointHash op)+ (I.singleton (fromIntegral (outPointIndex op)) Nothing)+ (hUnspent db)+ }++instance MonadIO m => StoreRead (ReaderT (TVar MemoryDatabase) m) where+ getBestBlock = do+ v <- R.ask >>= readTVarIO+ return $ getBestBlockH v+ getBlocksAtHeight h = do+ v <- R.ask >>= readTVarIO+ return $ getBlocksAtHeightH h v+ getBlock b = do+ v <- R.ask >>= readTVarIO+ return $ getBlockH b v+ getTxData t = do+ v <- R.ask >>= readTVarIO+ return $ getTxDataH t v+ getSpender t = do+ v <- R.ask >>= readTVarIO+ return . join $ getSpenderH t v+ getSpenders t = do+ v <- R.ask >>= readTVarIO+ return . I.map fromJust . I.filter isJust $ getSpendersH t v+ getOrphanTx h = do+ v <- R.ask >>= readTVarIO+ return . join $ getOrphanTxH h v+ getUnspent p = do+ v <- R.ask >>= readTVarIO+ return . join $ getUnspentH p v+ getBalance a = do+ v <- R.ask >>= readTVarIO+ return $ getBalanceH a v+ getMempool = do+ v <- R.ask >>= readTVarIO+ return . fromMaybe [] $ getMempoolH v+ getAddressesTxs addr start limit = do+ v <- R.ask >>= readTVarIO+ return $ getAddressesTxsH addr start limit v+ getAddressesUnspents addr start limit = do+ v <- R.ask >>= readTVarIO+ return $ getAddressesUnspentsH addr start limit v+ getOrphans = do+ v <- R.ask >>= readTVarIO+ return $ getOrphansH v+ getAddressTxs addr start limit = do+ v <- R.ask >>= readTVarIO+ return $ getAddressTxsH addr start limit v+ getAddressUnspents addr start limit = do+ v <- R.ask >>= readTVarIO+ return $ getAddressUnspentsH addr start limit v++instance (MonadIO m) => StoreWrite (ReaderT (TVar MemoryDatabase) m) where+ setBest h = do+ v <- R.ask+ atomically $ modifyTVar v (setBestH h)+ insertBlock b = do+ v <- R.ask+ atomically $ modifyTVar v (insertBlockH b)+ setBlocksAtHeight h g = do+ v <- R.ask+ atomically $ modifyTVar v (setBlocksAtHeightH h g)+ insertTx t = do+ v <- R.ask+ atomically $ modifyTVar v (insertTxH t)+ insertSpender p s = do+ v <- R.ask+ atomically $ modifyTVar v (insertSpenderH p s)+ deleteSpender p = do+ v <- R.ask+ atomically $ modifyTVar v (deleteSpenderH p)+ insertAddrTx a t = do+ v <- R.ask+ atomically $ modifyTVar v (insertAddrTxH a t)+ deleteAddrTx a t = do+ v <- R.ask+ atomically $ modifyTVar v (deleteAddrTxH a t)+ insertAddrUnspent a u = do+ v <- R.ask+ atomically $ modifyTVar v (insertAddrUnspentH a u)+ deleteAddrUnspent a u = do+ v <- R.ask+ atomically $ modifyTVar v (deleteAddrUnspentH a u)+ setMempool xs = do+ v <- R.ask+ atomically $ modifyTVar v (setMempoolH xs)+ insertOrphanTx t u = do+ v <- R.ask+ atomically $ modifyTVar v (insertOrphanTxH t u)+ deleteOrphanTx h = do+ v <- R.ask+ atomically $ modifyTVar v (deleteOrphanTxH h)+ setBalance b = do+ v <- R.ask+ atomically $ modifyTVar v (setBalanceH b)+ insertUnspent h = do+ v <- R.ask+ atomically $ modifyTVar v (insertUnspentH h)+ deleteUnspent p = do+ v <- R.ask+ atomically $ modifyTVar v (deleteUnspentH p)
− src/Network/Haskoin/Store/Data/RocksDB.hs
@@ -1,257 +0,0 @@-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE LambdaCase #-}-{-# OPTIONS_GHC -Wno-orphans #-}-module Network.Haskoin.Store.Data.RocksDB where--import Conduit (ConduitT, MonadResource,- mapC, runConduit,- runResourceT, sinkList,- (.|))-import Control.Monad.Except (runExceptT, throwError)-import Control.Monad.Reader (ReaderT, ask, runReaderT)-import Data.Function (on)-import Data.IntMap (IntMap)-import qualified Data.IntMap.Strict as I-import Data.List (nub, sortBy)-import Data.Maybe (fromMaybe)-import Data.Word (Word32)-import Database.RocksDB (Compression (..), DB,- Options (..),- defaultOptions,- defaultReadOptions, open)-import Database.RocksDB.Query (insert, matching,- matchingAsList,- matchingSkip, retrieve)-import Haskoin (Address, BlockHash,- BlockHeight,- OutPoint (..), Tx, TxHash)-import Network.Haskoin.Store.Common (Balance, BlockDB (..),- BlockData, BlockRef (..),- BlockTx (..), Limit,- Spender, StoreRead (..),- TxData, UnixTime,- Unspent (..),- UnspentVal (..),- applyLimit, applyLimitC,- valToBalance,- valToUnspent, zeroBalance)-import Network.Haskoin.Store.Data.KeyValue (AddrOutKey (..),- AddrTxKey (..),- BalKey (..), BestKey (..),- BlockKey (..),- HeightKey (..),- MemKey (..),- OldMemKey (..),- OrphanKey (..),- SpenderKey (..),- TxKey (..),- UnspentKey (..),- VersionKey (..),- toUnspent)-import UnliftIO (MonadIO, liftIO)--dataVersion :: Word32-dataVersion = 16--connectRocksDB :: MonadIO m => FilePath -> m BlockDB-connectRocksDB dir = do- bdb <- open- dir- defaultOptions- { createIfMissing = True- , compression = SnappyCompression- , maxOpenFiles = -1- , writeBufferSize = 2 ^ (30 :: Integer)- } >>= \db ->- return BlockDB {blockDBopts = defaultReadOptions, blockDB = db}- initRocksDB bdb- return bdb--withRocksDB :: MonadIO m => BlockDB -> ReaderT BlockDB m a -> m a-withRocksDB = flip runReaderT--initRocksDB :: MonadIO m => BlockDB -> m ()-initRocksDB bdb@BlockDB {blockDBopts = opts, blockDB = db} = do- e <-- runExceptT $- retrieve db opts VersionKey >>= \case- Just v- | v == dataVersion -> return ()- | v == 15 -> migrate15to16 bdb >> initRocksDB bdb- | otherwise -> throwError "Incorrect RocksDB database version"- Nothing -> setInitRocksDB db- case e of- Left s -> error s- Right () -> return ()--migrate15to16 :: MonadIO m => BlockDB -> m ()-migrate15to16 BlockDB {blockDBopts = opts, blockDB = db} = do- xs <- liftIO $ matchingAsList db opts OldMemKeyS- let ys = map (\(OldMemKey t h, ()) -> (t, h)) xs- insert db MemKey ys- insert db VersionKey (16 :: Word32)--setInitRocksDB :: MonadIO m => DB -> m ()-setInitRocksDB db = insert db VersionKey dataVersion--getBestBlockDB :: MonadIO m => BlockDB -> m (Maybe BlockHash)-getBestBlockDB BlockDB {blockDBopts = opts, blockDB = db} =- retrieve db opts BestKey--getBlocksAtHeightDB :: MonadIO m => BlockHeight -> BlockDB -> m [BlockHash]-getBlocksAtHeightDB h BlockDB {blockDBopts = opts, blockDB = db} =- retrieve db opts (HeightKey h) >>= \case- Nothing -> return []- Just ls -> return ls--getBlockDB :: MonadIO m => BlockHash -> BlockDB -> m (Maybe BlockData)-getBlockDB h BlockDB {blockDBopts = opts, blockDB = db} =- retrieve db opts (BlockKey h)--getTxDataDB ::- MonadIO m => TxHash -> BlockDB -> m (Maybe TxData)-getTxDataDB th BlockDB {blockDBopts = opts, blockDB = db} =- retrieve db opts (TxKey th)--getSpenderDB :: MonadIO m => OutPoint -> BlockDB -> m (Maybe Spender)-getSpenderDB op BlockDB {blockDBopts = opts, blockDB = db} =- retrieve db opts $ SpenderKey op--getSpendersDB :: MonadIO m => TxHash -> BlockDB -> m (IntMap Spender)-getSpendersDB th BlockDB {blockDBopts = opts, blockDB = db} =- I.fromList . map (uncurry f) <$>- liftIO (matchingAsList db opts (SpenderKeyS th))- where- f (SpenderKey op) s = (fromIntegral (outPointIndex op), s)- f _ _ = undefined--getBalanceDB :: MonadIO m => Address -> BlockDB -> m Balance-getBalanceDB a BlockDB {blockDBopts = opts, blockDB = db} =- fromMaybe (zeroBalance a) . fmap (valToBalance a) <$>- retrieve db opts (BalKey a)--getMempoolDB :: MonadIO m => BlockDB -> m [BlockTx]-getMempoolDB BlockDB {blockDBopts = opts, blockDB = db} =- fmap f . fromMaybe [] <$> retrieve db opts MemKey- where- f (t, h) = BlockTx {blockTxBlock = MemRef t, blockTxHash = h}--getOrphansDB ::- MonadIO m- => BlockDB- -> m [(UnixTime, Tx)]-getOrphansDB BlockDB {blockDBopts = opts, blockDB = db} =- liftIO . runResourceT . runConduit $- matching db opts OrphanKeyS .| mapC snd .| sinkList--getOrphanTxDB :: MonadIO m => TxHash -> BlockDB -> m (Maybe (UnixTime, Tx))-getOrphanTxDB h BlockDB {blockDBopts = opts, blockDB = db} =- retrieve db opts (OrphanKey h)--getAddressesTxsDB ::- MonadIO m- => [Address]- -> Maybe BlockRef- -> Maybe Limit- -> BlockDB- -> m [BlockTx]-getAddressesTxsDB addrs start limit db = do- ts <- concat <$> mapM (\a -> getAddressTxsDB a start limit db) addrs- let ts' = nub $ sortBy (flip compare `on` blockTxBlock) ts- return $ applyLimit limit ts'--getAddressTxsDB ::- MonadIO m- => Address- -> Maybe BlockRef- -> Maybe Limit- -> BlockDB- -> m [BlockTx]-getAddressTxsDB a start limit BlockDB {blockDBopts = opts, blockDB = db} =- liftIO . runResourceT . runConduit $- x .| applyLimitC limit .| mapC (uncurry f) .| sinkList- where- x =- case start of- Nothing -> matching db opts (AddrTxKeyA a)- Just br -> matchingSkip db opts (AddrTxKeyA a) (AddrTxKeyB a br)- f AddrTxKey {addrTxKeyT = t} () = t- f _ _ = undefined--getAddressBalancesDB ::- (MonadIO m, MonadResource m)- => BlockDB- -> ConduitT i Balance m ()-getAddressBalancesDB BlockDB {blockDBopts = opts, blockDB = db} =- matching db opts BalKeyS .| mapC (\(BalKey a, b) -> valToBalance a b)--getUnspentsDB ::- (MonadIO m, MonadResource m)- => BlockDB- -> ConduitT i Unspent m ()-getUnspentsDB BlockDB {blockDBopts = opts, blockDB = db} =- matching db opts UnspentKeyB .|- mapC (\(UnspentKey k, v) -> unspentFromDB k v)--getUnspentDB :: MonadIO m => OutPoint -> BlockDB -> m (Maybe Unspent)-getUnspentDB p BlockDB {blockDBopts = opts, blockDB = db} =- fmap (valToUnspent p) <$> retrieve db opts (UnspentKey p)--getAddressesUnspentsDB ::- MonadIO m- => [Address]- -> Maybe BlockRef- -> Maybe Limit- -> BlockDB- -> m [Unspent]-getAddressesUnspentsDB addrs start limit bdb = do- us <- concat <$> mapM (\a -> getAddressUnspentsDB a start limit bdb) addrs- let us' = nub $ sortBy (flip compare `on` unspentBlock) us- return $ applyLimit limit us'--getAddressUnspentsDB ::- MonadIO m- => Address- -> Maybe BlockRef- -> Maybe Limit- -> BlockDB- -> m [Unspent]-getAddressUnspentsDB a start limit BlockDB {blockDBopts = opts, blockDB = db} =- liftIO . runResourceT . runConduit $- x .| applyLimitC limit .| mapC (uncurry toUnspent) .| sinkList- where- x =- case start of- Nothing -> matching db opts (AddrOutKeyA a)- Just br -> matchingSkip db opts (AddrOutKeyA a) (AddrOutKeyB a br)--unspentFromDB :: OutPoint -> UnspentVal -> Unspent-unspentFromDB p UnspentVal { unspentValBlock = b- , unspentValAmount = v- , unspentValScript = s- } =- Unspent- { unspentBlock = b- , unspentAmount = v- , unspentPoint = p- , unspentScript = s- }--instance MonadIO m => StoreRead (ReaderT BlockDB m) where- getBestBlock = ask >>= getBestBlockDB- getBlocksAtHeight h = ask >>= getBlocksAtHeightDB h- getBlock b = ask >>= getBlockDB b- getTxData t = ask >>= getTxDataDB t- getSpender p = ask >>= getSpenderDB p- getSpenders t = ask >>= getSpendersDB t- getOrphanTx h = ask >>= getOrphanTxDB h- getUnspent a = ask >>= getUnspentDB a- getBalance a = ask >>= getBalanceDB a- getMempool = ask >>= getMempoolDB- getAddressesTxs addrs start limit =- ask >>= getAddressesTxsDB addrs start limit- getAddressesUnspents addrs start limit =- ask >>= getAddressesUnspentsDB addrs start limit- getOrphans = ask >>= getOrphansDB- getAddressUnspents a b c = ask >>= getAddressUnspentsDB a b c- getAddressTxs a b c = ask >>= getAddressTxsDB a b c
+ src/Network/Haskoin/Store/Data/Types.hs view
@@ -0,0 +1,360 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+module Network.Haskoin.Store.Data.Types where++import Control.Monad (guard)+import Data.ByteString (ByteString)+import qualified Data.ByteString as BS+import qualified Data.ByteString.Short as BSS+import Data.Hashable (Hashable)+import Data.Serialize (Serialize (..), getBytes,+ getWord8, putWord8)+import Data.Word (Word32, Word64)+import Database.RocksDB.Query (Key, KeyValue)+import GHC.Generics (Generic)+import Haskoin (Address, BlockHash, BlockHeight,+ OutPoint (..), Tx, TxHash)+import Network.Haskoin.Store.Common (BalVal, BlockData, BlockRef,+ BlockTx (..), Spender, TxData,+ UnixTime, Unspent (..),+ UnspentVal, getUnixTime,+ putUnixTime)++-- | Database key for an address transaction.+data AddrTxKey+ = AddrTxKey { addrTxKeyA :: !Address+ , addrTxKeyT :: !BlockTx+ }+ -- ^ key for a transaction affecting an address+ | AddrTxKeyA { addrTxKeyA :: !Address }+ -- ^ short key that matches all entries+ | AddrTxKeyB { addrTxKeyA :: !Address+ , addrTxKeyB :: !BlockRef+ }+ | AddrTxKeyS+ deriving (Show, Eq, Ord, Generic, Hashable)++instance Serialize AddrTxKey+ -- 0x05 · Address · BlockRef · TxHash+ where+ put AddrTxKey { addrTxKeyA = a+ , addrTxKeyT = BlockTx { blockTxBlock = b+ , blockTxHash = t+ }+ } = do+ putWord8 0x05+ put a+ put b+ put t+ -- 0x05 · Address+ put AddrTxKeyA {addrTxKeyA = a} = do+ putWord8 0x05+ put a+ -- 0x05 · Address · BlockRef+ put AddrTxKeyB {addrTxKeyA = a, addrTxKeyB = b} = do+ putWord8 0x05+ put a+ put b+ -- 0x05+ put AddrTxKeyS = putWord8 0x05+ get = do+ guard . (== 0x05) =<< getWord8+ a <- get+ b <- get+ t <- get+ return+ AddrTxKey+ { addrTxKeyA = a+ , addrTxKeyT =+ BlockTx+ { blockTxBlock = b+ , blockTxHash = t+ }+ }++instance Key AddrTxKey+instance KeyValue AddrTxKey ()++-- | Database key for an address output.+data AddrOutKey+ = AddrOutKey { addrOutKeyA :: !Address+ , addrOutKeyB :: !BlockRef+ , addrOutKeyP :: !OutPoint }+ -- ^ full key+ | AddrOutKeyA { addrOutKeyA :: !Address }+ -- ^ short key for all spent or unspent outputs+ | AddrOutKeyB { addrOutKeyA :: !Address+ , addrOutKeyB :: !BlockRef+ }+ | AddrOutKeyS+ deriving (Show, Read, Eq, Ord, Generic, Hashable)++instance Serialize AddrOutKey+ -- 0x06 · StoreAddr · BlockRef · OutPoint+ where+ put AddrOutKey {addrOutKeyA = a, addrOutKeyB = b, addrOutKeyP = p} = do+ putWord8 0x06+ put a+ put b+ put p+ -- 0x06 · StoreAddr · BlockRef+ put AddrOutKeyB {addrOutKeyA = a, addrOutKeyB = b} = do+ putWord8 0x06+ put a+ put b+ -- 0x06 · StoreAddr+ put AddrOutKeyA {addrOutKeyA = a} = do+ putWord8 0x06+ put a+ -- 0x06+ put AddrOutKeyS = putWord8 0x06+ get = do+ guard . (== 0x06) =<< getWord8+ AddrOutKey <$> get <*> get <*> get++instance Key AddrOutKey++data OutVal = OutVal+ { outValAmount :: !Word64+ , outValScript :: !ByteString+ } deriving (Show, Read, Eq, Ord, Generic, Hashable, Serialize)++instance KeyValue AddrOutKey OutVal++-- | Transaction database key.+newtype TxKey = TxKey+ { txKey :: TxHash+ } deriving (Show, Read, Eq, Ord, Generic, Hashable)++instance Serialize TxKey where+ -- 0x02 · TxHash+ put (TxKey h) = do+ putWord8 0x02+ put h+ get = do+ guard . (== 0x02) =<< getWord8+ TxKey <$> get++instance Key TxKey+instance KeyValue TxKey TxData++data SpenderKey+ = SpenderKey { outputPoint :: !OutPoint }+ | SpenderKeyS { outputKeyS :: !TxHash }+ deriving (Show, Read, Eq, Ord, Generic, Hashable)++instance Serialize SpenderKey where+ -- 0x10 · TxHash · Index+ put (SpenderKey OutPoint {outPointHash = h, outPointIndex = i}) = do+ putWord8 0x10+ put h+ put i+ put (SpenderKeyS h) = do+ putWord8 0x10+ put h+ get = do+ guard . (== 0x10) =<< getWord8+ op <- OutPoint <$> get <*> get+ return $ SpenderKey op++instance Key SpenderKey+instance KeyValue SpenderKey Spender++-- | Unspent output database key.+data UnspentKey+ = UnspentKey { unspentKey :: !OutPoint }+ | UnspentKeyS { unspentKeyS :: !TxHash }+ | UnspentKeyB+ deriving (Show, Read, Eq, Ord, Generic, Hashable)++instance Serialize UnspentKey+ -- 0x09 · TxHash · Index+ where+ put UnspentKey {unspentKey = OutPoint {outPointHash = h, outPointIndex = i}} = do+ putWord8 0x09+ put h+ put i+ -- 0x09 · TxHash+ put UnspentKeyS {unspentKeyS = t} = do+ putWord8 0x09+ put t+ -- 0x09+ put UnspentKeyB = putWord8 0x09+ get = do+ guard . (== 0x09) =<< getWord8+ h <- get+ i <- get+ return $ UnspentKey OutPoint {outPointHash = h, outPointIndex = i}++instance Key UnspentKey+instance KeyValue UnspentKey UnspentVal++toUnspent :: AddrOutKey -> OutVal -> Unspent+toUnspent AddrOutKey {addrOutKeyB = b, addrOutKeyP = p} OutVal { outValAmount = v+ , outValScript = s+ } =+ Unspent+ { unspentBlock = b+ , unspentAmount = v+ , unspentScript = BSS.toShort s+ , unspentPoint = p+ }+toUnspent _ _ = undefined++-- | Mempool transaction database key.+data MemKey =+ MemKey+ deriving (Show, Read)++instance Serialize MemKey where+ -- 0x07+ put MemKey = do+ putWord8 0x07+ get = do+ guard . (== 0x07) =<< getWord8+ return MemKey++instance Key MemKey+instance KeyValue MemKey [(UnixTime, TxHash)]++-- | Orphan pool transaction database key.+data OrphanKey+ = OrphanKey+ { orphanKey :: !TxHash+ }+ | OrphanKeyS+ deriving (Show, Read, Eq, Ord, Generic, Hashable)++instance Serialize OrphanKey+ -- 0x08 · TxHash+ where+ put (OrphanKey h) = do+ putWord8 0x08+ put h+ -- 0x08+ put OrphanKeyS = putWord8 0x08+ get = do+ guard . (== 0x08) =<< getWord8+ OrphanKey <$> get++instance Key OrphanKey+instance KeyValue OrphanKey (UnixTime, Tx)++-- | Block entry database key.+newtype BlockKey = BlockKey+ { blockKey :: BlockHash+ } deriving (Show, Read, Eq, Ord, Generic, Hashable)++instance Serialize BlockKey where+ -- 0x01 · BlockHash+ put (BlockKey h) = do+ putWord8 0x01+ put h+ get = do+ guard . (== 0x01) =<< getWord8+ BlockKey <$> get++instance KeyValue BlockKey BlockData++-- | Block height database key.+newtype HeightKey = HeightKey+ { heightKey :: BlockHeight+ } deriving (Show, Read, Eq, Ord, Generic, Hashable)++instance Serialize HeightKey where+ -- 0x03 · BlockHeight+ put (HeightKey height) = do+ putWord8 0x03+ put height+ get = do+ guard . (== 0x03) =<< getWord8+ HeightKey <$> get++instance Key HeightKey+instance KeyValue HeightKey [BlockHash]++-- | Address balance database key.+data BalKey+ = BalKey+ { balanceKey :: !Address+ }+ | BalKeyS+ deriving (Show, Read, Eq, Ord, Generic, Hashable)++instance Serialize BalKey where+ -- 0x04 · Address+ put BalKey {balanceKey = a} = do+ putWord8 0x04+ put a+ -- 0x04+ put BalKeyS = putWord8 0x04+ get = do+ guard . (== 0x04) =<< getWord8+ BalKey <$> get++instance Key BalKey+instance KeyValue BalKey BalVal++-- | Key for best block in database.+data BestKey =+ BestKey+ deriving (Show, Read, Eq, Ord, Generic, Hashable)++instance Serialize BestKey where+ -- 0x00 × 32+ put BestKey = put (BS.replicate 32 0x00)+ get = do+ guard . (== BS.replicate 32 0x00) =<< getBytes 32+ return BestKey++instance Key BestKey+instance KeyValue BestKey BlockHash++-- | Key for database version.+data VersionKey =+ VersionKey+ deriving (Eq, Show, Read, Ord, Generic, Hashable)++instance Serialize VersionKey where+ -- 0x0a+ put VersionKey = putWord8 0x0a+ get = do+ guard . (== 0x0a) =<< getWord8+ return VersionKey++instance Key VersionKey+instance KeyValue VersionKey Word32+++-- | Old mempool transaction database key.+data OldMemKey+ = OldMemKey+ { memTime :: !UnixTime+ , memKey :: !TxHash+ }+ | OldMemKeyT+ { memTime :: !UnixTime+ }+ | OldMemKeyS+ deriving (Show, Read, Eq, Ord, Generic, Hashable)++instance Serialize OldMemKey where+ -- 0x07 · UnixTime · TxHash+ put (OldMemKey t h) = do+ putWord8 0x07+ putUnixTime t+ put h+ -- 0x07 · UnixTime+ put (OldMemKeyT t) = do+ putWord8 0x07+ putUnixTime t+ -- 0x07+ put OldMemKeyS = putWord8 0x07+ get = do+ guard . (== 0x07) =<< getWord8+ OldMemKey <$> getUnixTime <*> get++instance Key OldMemKey+instance KeyValue OldMemKey ()
src/Network/Haskoin/Store/Web.hs view
@@ -1,113 +1,141 @@ {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-} {-# LANGUAGE TemplateHaskell #-} module Network.Haskoin.Store.Web where -import Conduit ()-import Control.Applicative ((<|>))-import Control.Monad (forever, guard, mzero,- unless, when, (<=<))-import Control.Monad.Logger (Loc, LogLevel, LogSource,- LogStr, MonadLogger,- MonadLoggerIO, askLoggerIO,- logInfoS, monadLoggerLog)-import Control.Monad.Reader (ReaderT, ask)-import Control.Monad.Trans (lift)-import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT)-import Data.Aeson (ToJSON (..), object, (.=))-import Data.Aeson.Encoding (encodingToLazyByteString)-import qualified Data.ByteString as B-import Data.ByteString.Builder (lazyByteString)-import qualified Data.ByteString.Lazy as L-import qualified Data.ByteString.Lazy.Char8 as C-import Data.Char (isSpace)-import Data.List (nub)-import Data.Maybe (catMaybes, fromMaybe,- isJust, listToMaybe,- mapMaybe)-import Data.Serialize as Serialize-import Data.String.Conversions (cs)-import Data.Text (Text)-import qualified Data.Text as T-import qualified Data.Text.Encoding as T-import qualified Data.Text.Lazy as T.Lazy-import Data.Time.Clock (NominalDiffTime,- diffUTCTime,- getCurrentTime)-import Data.Time.Clock.System (getSystemTime,- systemSeconds)-import Data.Word (Word32, Word64)-import Database.RocksDB (Property (..), getProperty)-import Haskoin (Address, Block (..),- BlockHash (..),- BlockHeader (..),- BlockHeight,- BlockNode (..),- GetData (..), Hash256,- InvType (..),- InvVector (..),- Message (..), Network (..),- OutPoint (..), Tx,- TxHash (..),- VarString (..),- Version (..), decodeHex,- eitherToMaybe, headerHash,- hexToBlockHash,- hexToTxHash,- sockToHostAddress,- stringToAddr, txHash,- xPubImport)-import Haskoin.Node (Chain, Manager,- OnlinePeer (..),- chainGetBest,- managerGetPeers,- sendMessage)-import Network.Haskoin.Store.Common (BinSerial (..),- BlockDB (..),- BlockData (..),- BlockRef (..),- BlockTx (..),- DeriveType (..),- Event (..),- HealthCheck (..),- JsonSerial (..), Limit,- Offset,- PeerInformation (..),- PubExcept (..), Store (..),- StoreEvent (..),- StoreInput (..),- StoreRead (..),- Transaction (..),- TxAfterHeight (..),- TxData (..), TxId (..),- UnixTime, Unspent,- XPubBal (..),- XPubSpec (..), applyOffset,- blockAtOrBefore,- getTransaction, isCoinbase,- nullBalance,- transactionData)-import Network.Haskoin.Store.Data.RocksDB (withRocksDB)-import Network.HTTP.Types (Status (..), status400,- status403, status404,- status500, status503)-import Network.Wai (Middleware, Request (..),- responseStatus)-import NQE (Publisher, receive,- withSubscription)-import Text.Printf (printf)-import Text.Read (readMaybe)-import UnliftIO (Exception, MonadIO,- MonadUnliftIO, askRunInIO,- liftIO, timeout)-import Web.Scotty.Internal.Types (ActionT)-import Web.Scotty.Trans (Parsable, ScottyError)-import qualified Web.Scotty.Trans as S+import Conduit ()+import Control.Applicative ((<|>))+import Control.Monad (forever, guard,+ mzero, unless, when,+ (<=<))+import Control.Monad.Logger (Loc, LogLevel,+ LogSource, LogStr,+ MonadLogger,+ MonadLoggerIO,+ askLoggerIO,+ logInfoS,+ monadLoggerLog)+import Control.Monad.Reader (ReaderT, asks,+ runReaderT)+import Control.Monad.Trans (lift)+import Control.Monad.Trans.Maybe (MaybeT (..),+ runMaybeT)+import Data.Aeson (ToJSON (..), object,+ (.=))+import Data.Aeson.Encoding (encodingToLazyByteString)+import qualified Data.ByteString as B+import Data.ByteString.Builder (lazyByteString)+import qualified Data.ByteString.Lazy as L+import qualified Data.ByteString.Lazy.Char8 as C+import Data.Char (isSpace)+import Data.List (nub)+import Data.Maybe (catMaybes,+ fromMaybe, isJust,+ listToMaybe,+ mapMaybe)+import Data.Serialize as Serialize+import Data.String.Conversions (cs)+import Data.Text (Text)+import qualified Data.Text as T+import qualified Data.Text.Encoding as T+import qualified Data.Text.Lazy as T.Lazy+import Data.Time.Clock (NominalDiffTime,+ diffUTCTime,+ getCurrentTime)+import Data.Time.Clock.System (getSystemTime,+ systemSeconds)+import Data.Word (Word32, Word64)+import Database.RocksDB (Property (..),+ getProperty)+import Haskoin (Address, Block (..),+ BlockHash (..),+ BlockHeader (..),+ BlockHeight,+ BlockNode (..),+ GetData (..),+ Hash256,+ InvType (..),+ InvVector (..),+ Message (..),+ Network (..),+ OutPoint (..), Tx,+ TxHash (..),+ VarString (..),+ Version (..),+ decodeHex,+ eitherToMaybe,+ headerHash,+ hexToBlockHash,+ hexToTxHash,+ sockToHostAddress,+ stringToAddr,+ txHash, xPubImport)+import Haskoin.Node (Chain, Manager,+ OnlinePeer (..),+ chainGetBest,+ managerGetPeers,+ sendMessage)+import Network.Haskoin.Store.Common (BinSerial (..),+ BlockData (..),+ BlockRef (..),+ BlockTx (..),+ DeriveType (..),+ Event (..),+ HealthCheck (..),+ JsonSerial (..),+ Limit, Offset,+ PeerInformation (..),+ PubExcept (..),+ Store (..),+ StoreEvent (..),+ StoreInput (..),+ StoreRead (..),+ Transaction (..),+ TxAfterHeight (..),+ TxData (..),+ TxId (..), UnixTime,+ Unspent,+ XPubBal (..),+ XPubSpec (..),+ applyOffset,+ blockAtOrBefore,+ getTransaction,+ isCoinbase,+ nullBalance,+ transactionData)+import Network.Haskoin.Store.Data.CacheReader (CacheReaderConfig,+ CacheReaderT,+ withCacheReader)+import Network.Haskoin.Store.Data.DatabaseReader (DatabaseReader (..),+ DatabaseReaderT,+ withDatabaseReader)+import Network.HTTP.Types (Status (..),+ status400,+ status403,+ status404,+ status500,+ status503)+import Network.Wai (Middleware,+ Request (..),+ responseStatus)+import NQE (Publisher, receive,+ withSubscription)+import Text.Printf (printf)+import Text.Read (readMaybe)+import UnliftIO (Exception, MonadIO,+ MonadUnliftIO,+ askRunInIO, liftIO,+ timeout)+import Web.Scotty.Internal.Types (ActionT)+import Web.Scotty.Trans (Parsable,+ ScottyError)+import qualified Web.Scotty.Trans as S type LoggerIO = Loc -> LogSource -> LogLevel -> LogStr -> IO () -type WebT m = ActionT Except (ReaderT BlockDB m)+type WebT m = ActionT Except (ReaderT WebConfig m) data Except = ThingNotFound@@ -161,7 +189,8 @@ WebConfig { webPort :: !Int , webNetwork :: !Network- , webDB :: !BlockDB+ , webDB :: !DatabaseReader+ , webCache :: !(Maybe CacheReaderConfig) , webPublisher :: !(Publisher StoreEvent) , webStore :: !Store , webMaxLimits :: !MaxLimits@@ -219,7 +248,53 @@ guard $ x > 1230768000 return StartParamTime {startParamTime = x} -instance MonadLoggerIO m => StoreRead (WebT m) where+runInWebReader ::+ MonadIO m+ => DatabaseReaderT m a+ -> CacheReaderT (DatabaseReaderT m) a+ -> ReaderT WebConfig m a+runInWebReader bf cf = do+ bdb <- asks webDB+ mc <- asks webCache+ lift $ withDatabaseReader bdb $+ case mc of+ Nothing -> bf+ Just c -> withCacheReader c cf++instance MonadIO m => StoreRead (ReaderT WebConfig m) where+ getBestBlock = runInWebReader getBestBlock getBestBlock+ getBlocksAtHeight height =+ runInWebReader (getBlocksAtHeight height) (getBlocksAtHeight height)+ getBlock bh = runInWebReader (getBlock bh) (getBlock bh)+ getTxData th = runInWebReader (getTxData th) (getTxData th)+ getSpender op = runInWebReader (getSpender op) (getSpender op)+ getSpenders th = runInWebReader (getSpenders th) (getSpenders th)+ getOrphanTx th = runInWebReader (getOrphanTx th) (getOrphanTx th)+ getUnspent op = runInWebReader (getUnspent op) (getUnspent op)+ getBalance a = runInWebReader (getBalance a) (getBalance a)+ getBalances as = runInWebReader (getBalances as) (getBalances as)+ getMempool = runInWebReader getMempool getMempool+ getAddressesTxs addrs start limit =+ runInWebReader+ (getAddressesTxs addrs start limit)+ (getAddressesTxs addrs start limit)+ getAddressesUnspents addrs start limit =+ runInWebReader+ (getAddressesUnspents addrs start limit)+ (getAddressesUnspents addrs start limit)+ getOrphans = runInWebReader getOrphans getOrphans+ xPubBals xpub = runInWebReader (xPubBals xpub) (xPubBals xpub)+ xPubSummary xpub = runInWebReader (xPubSummary xpub) (xPubSummary xpub)+ xPubUnspents xpub start offset limit =+ runInWebReader+ (xPubUnspents xpub start offset limit)+ (xPubUnspents xpub start offset limit)+ xPubTxs xpub start offset limit =+ runInWebReader+ (xPubTxs xpub start offset limit)+ (xPubTxs xpub start offset limit)++instance MonadIO m => StoreRead (WebT m) where getBestBlock = lift getBestBlock getBlocksAtHeight = lift . getBlocksAtHeight getBlock = lift . getBlock@@ -231,14 +306,18 @@ getBalance = lift . getBalance getBalances = lift . getBalances getMempool = lift getMempool- getAddressesTxs addrs start limit =- lift (getAddressesTxs addrs start limit)+ getAddressesTxs addrs start limit = lift (getAddressesTxs addrs start limit) getAddressesUnspents addrs start limit = lift (getAddressesUnspents addrs start limit) getOrphans = lift getOrphans+ xPubBals = lift . xPubBals+ xPubSummary = lift . xPubSummary+ xPubUnspents xpub start offset limit =+ lift (xPubUnspents xpub start offset limit)+ xPubTxs xpub start offset limit = lift (xPubTxs xpub start offset limit) -defHandler :: Monad m => Network -> Except -> WebT m ()-defHandler net e = do+defHandler :: Monad m => Except -> WebT m ()+defHandler e = do proto <- setupBin case e of ThingNotFound -> S.status status404@@ -247,28 +326,29 @@ StringError _ -> S.status status400 ServerError -> S.status status500 BlockTooLarge -> S.status status403- protoSerial net proto e+ protoSerial proto e maybeSerial :: (Monad m, JsonSerial a, BinSerial a)- => Network- -> Bool -- ^ binary+ => Bool -- ^ binary -> Maybe a -> WebT m ()-maybeSerial _ _ Nothing = S.raise ThingNotFound-maybeSerial net proto (Just x) = S.raw $ serialAny net proto x+maybeSerial _ Nothing = S.raise ThingNotFound+maybeSerial proto (Just x) =+ lift (asks webNetwork) >>= \net -> S.raw (serialAny net proto x) protoSerial :: (Monad m, JsonSerial a, BinSerial a)- => Network- -> Bool+ => Bool -> a -> WebT m ()-protoSerial net proto = S.raw . serialAny net proto+protoSerial proto x =+ lift (asks webNetwork) >>= \net -> S.raw (serialAny net proto x) scottyBestBlock ::- (MonadLoggerIO m, MonadIO m) => Network -> MaxLimits -> Bool -> WebT m ()-scottyBestBlock net limits raw = do+ (MonadLoggerIO m, MonadIO m) => Bool -> WebT m ()+scottyBestBlock raw = do+ limits <- lift $ asks webMaxLimits setHeaders n <- parseNoTx proto <- setupBin@@ -283,11 +363,12 @@ if raw then do refuseLargeBlock limits b- rawBlock b >>= protoSerial net proto- else protoSerial net proto (pruneTx n b)+ rawBlock b >>= protoSerial proto+ else protoSerial proto (pruneTx n b) -scottyBlock :: MonadLoggerIO m => Network -> MaxLimits -> Bool -> WebT m ()-scottyBlock net limits raw = do+scottyBlock :: MonadLoggerIO m => Bool -> WebT m ()+scottyBlock raw = do+ limits <- lift $ asks webMaxLimits setHeaders block <- myBlockHash <$> S.param "block" n <- parseNoTx@@ -299,11 +380,13 @@ if raw then do refuseLargeBlock limits b- rawBlock b >>= protoSerial net proto- else protoSerial net proto (pruneTx n b)+ rawBlock b >>= protoSerial proto+ else protoSerial proto (pruneTx n b) -scottyBlockHeight :: MonadLoggerIO m => Network -> MaxLimits -> Bool -> WebT m ()-scottyBlockHeight net limits raw = do+scottyBlockHeight ::+ MonadLoggerIO m => Bool -> WebT m ()+scottyBlockHeight raw = do+ limits <- lift $ asks webMaxLimits setHeaders height <- S.param "height" n <- parseNoTx@@ -314,47 +397,48 @@ blocks <- catMaybes <$> mapM getBlock hs mapM_ (refuseLargeBlock limits) blocks rawblocks <- mapM rawBlock blocks- protoSerial net proto rawblocks+ protoSerial proto rawblocks else do blocks <- catMaybes <$> mapM getBlock hs let blocks' = map (pruneTx n) blocks- protoSerial net proto blocks'+ protoSerial proto blocks' -scottyBlockTime :: MonadLoggerIO m => Network -> MaxLimits -> Bool -> WebT m ()-scottyBlockTime net limits raw = do+scottyBlockTime :: MonadLoggerIO m => Bool -> WebT m ()+scottyBlockTime raw = do+ limits <- lift $ asks webMaxLimits setHeaders q <- S.param "time" n <- parseNoTx proto <- setupBin m <- fmap (pruneTx n) <$> blockAtOrBefore q if raw- then maybeSerial net proto =<<+ then maybeSerial proto =<< case m of Nothing -> return Nothing Just d -> do refuseLargeBlock limits d Just <$> rawBlock d- else maybeSerial net proto m+ else maybeSerial proto m -scottyBlockHeights :: MonadLoggerIO m => Network -> WebT m ()-scottyBlockHeights net = do+scottyBlockHeights :: MonadLoggerIO m => WebT m ()+scottyBlockHeights = do setHeaders heights <- S.param "heights" n <- parseNoTx proto <- setupBin bhs <- concat <$> mapM getBlocksAtHeight (heights :: [BlockHeight]) blocks <- map (pruneTx n) . catMaybes <$> mapM getBlock bhs- protoSerial net proto blocks+ protoSerial proto blocks -scottyBlockLatest :: MonadLoggerIO m => Network -> WebT m ()-scottyBlockLatest net = do+scottyBlockLatest :: MonadLoggerIO m => WebT m ()+scottyBlockLatest = do setHeaders n <- parseNoTx proto <- setupBin getBestBlock >>= \case Just h -> do blocks <- reverse <$> go 100 n h- protoSerial net proto blocks+ protoSerial proto blocks Nothing -> S.raise ThingNotFound where go 0 _ _ = return []@@ -370,57 +454,58 @@ else (b' :) <$> go i' n prev -scottyBlocks :: MonadLoggerIO m => Network -> WebT m ()-scottyBlocks net = do+scottyBlocks :: MonadLoggerIO m => WebT m ()+scottyBlocks = do setHeaders bhs <- map myBlockHash <$> S.param "blocks" n <- parseNoTx proto <- setupBin bks <- map (pruneTx n) . catMaybes <$> mapM getBlock (nub bhs)- protoSerial net proto bks+ protoSerial proto bks -scottyMempool :: MonadLoggerIO m => Network -> WebT m ()-scottyMempool net = do+scottyMempool :: MonadLoggerIO m => WebT m ()+scottyMempool = do setHeaders proto <- setupBin txs <- map blockTxHash <$> getMempool- protoSerial net proto txs+ protoSerial proto txs -scottyTransaction :: MonadLoggerIO m => Network -> WebT m ()-scottyTransaction net = do+scottyTransaction :: MonadLoggerIO m => WebT m ()+scottyTransaction = do setHeaders txid <- myTxHash <$> S.param "txid" proto <- setupBin res <- getTransaction txid- maybeSerial net proto res+ maybeSerial proto res -scottyRawTransaction :: MonadLoggerIO m => Network -> WebT m ()-scottyRawTransaction net = do+scottyRawTransaction :: MonadLoggerIO m => WebT m ()+scottyRawTransaction = do setHeaders txid <- myTxHash <$> S.param "txid" proto <- setupBin res <- fmap transactionData <$> getTransaction txid- maybeSerial net proto res+ maybeSerial proto res -scottyTxAfterHeight :: MonadLoggerIO m => Network -> WebT m ()-scottyTxAfterHeight net = do+scottyTxAfterHeight :: MonadLoggerIO m => WebT m ()+scottyTxAfterHeight = do setHeaders txid <- myTxHash <$> S.param "txid" height <- S.param "height" proto <- setupBin res <- cbAfterHeight 10000 height txid- protoSerial net proto res+ protoSerial proto res -scottyTransactions :: MonadLoggerIO m => Network -> WebT m ()-scottyTransactions net = do+scottyTransactions :: MonadLoggerIO m => WebT m ()+scottyTransactions = do setHeaders txids <- map myTxHash <$> S.param "txids" proto <- setupBin txs <- catMaybes <$> mapM getTransaction (nub txids)- protoSerial net proto txs+ protoSerial proto txs -scottyBlockTransactions :: MonadLoggerIO m => Network -> MaxLimits -> WebT m ()-scottyBlockTransactions net limits = do+scottyBlockTransactions :: MonadLoggerIO m => WebT m ()+scottyBlockTransactions = do+ limits <- lift $ asks webMaxLimits setHeaders h <- myBlockHash <$> S.param "block" proto <- setupBin@@ -429,17 +514,17 @@ refuseLargeBlock limits b let ths = blockDataTxs b txs <- catMaybes <$> mapM getTransaction ths- protoSerial net proto txs+ protoSerial proto txs Nothing -> S.raise ThingNotFound scottyRawTransactions ::- MonadLoggerIO m => Network -> WebT m ()-scottyRawTransactions net = do+ MonadLoggerIO m => WebT m ()+scottyRawTransactions = do setHeaders txids <- map myTxHash <$> S.param "txids" proto <- setupBin txs <- map transactionData . catMaybes <$> mapM getTransaction (nub txids)- protoSerial net proto txs+ protoSerial proto txs rawBlock :: (Monad m, StoreRead m) => BlockData -> m Block rawBlock b = do@@ -449,8 +534,9 @@ return Block {blockHeader = h, blockTxns = txs} scottyRawBlockTransactions ::- MonadLoggerIO m => Network -> MaxLimits -> WebT m ()-scottyRawBlockTransactions net limits = do+ MonadLoggerIO m => WebT m ()+scottyRawBlockTransactions = do+ limits <- lift $ asks webMaxLimits setHeaders h <- myBlockHash <$> S.param "block" proto <- setupBin@@ -459,119 +545,119 @@ refuseLargeBlock limits b let ths = blockDataTxs b txs <- map transactionData . catMaybes <$> mapM getTransaction ths- protoSerial net proto txs+ protoSerial proto txs Nothing -> S.raise ThingNotFound -scottyAddressTxs :: MonadLoggerIO m => Network -> MaxLimits -> Bool -> WebT m ()-scottyAddressTxs net limits full = do+scottyAddressTxs :: MonadLoggerIO m => Bool -> WebT m ()+scottyAddressTxs full = do setHeaders- a <- parseAddress net+ a <- parseAddress s <- getStart- o <- getOffset limits- l <- getLimit limits full+ o <- getOffset+ l <- getLimit full proto <- setupBin if full then do- getAddressTxsFull o l s a >>= protoSerial net proto+ getAddressTxsFull o l s a >>= protoSerial proto else do- getAddressTxsLimit o l s a >>= protoSerial net proto+ getAddressTxsLimit o l s a >>= protoSerial proto scottyAddressesTxs ::- MonadLoggerIO m => Network -> MaxLimits -> Bool -> WebT m ()-scottyAddressesTxs net limits full = do+ MonadLoggerIO m => Bool -> WebT m ()+scottyAddressesTxs full = do setHeaders- as <- parseAddresses net+ as <- parseAddresses s <- getStart- l <- getLimit limits full+ l <- getLimit full proto <- setupBin if full- then getAddressesTxsFull l s as >>= protoSerial net proto- else getAddressesTxsLimit l s as >>= protoSerial net proto+ then getAddressesTxsFull l s as >>= protoSerial proto+ else getAddressesTxsLimit l s as >>= protoSerial proto -scottyAddressUnspent :: MonadLoggerIO m => Network -> MaxLimits -> WebT m ()-scottyAddressUnspent net limits = do+scottyAddressUnspent :: MonadLoggerIO m => WebT m ()+scottyAddressUnspent = do setHeaders- a <- parseAddress net+ a <- parseAddress s <- getStart- o <- getOffset limits- l <- getLimit limits False+ o <- getOffset+ l <- getLimit False proto <- setupBin uns <- getAddressUnspentsLimit o l s a- protoSerial net proto uns+ protoSerial proto uns -scottyAddressesUnspent :: MonadLoggerIO m => Network -> MaxLimits -> WebT m ()-scottyAddressesUnspent net limits = do+scottyAddressesUnspent :: MonadLoggerIO m => WebT m ()+scottyAddressesUnspent = do setHeaders- as <- parseAddresses net+ as <- parseAddresses s <- getStart- l <- getLimit limits False+ l <- getLimit False proto <- setupBin uns <- getAddressesUnspentsLimit l s as- protoSerial net proto uns+ protoSerial proto uns -scottyAddressBalance :: MonadLoggerIO m => Network -> WebT m ()-scottyAddressBalance net = do+scottyAddressBalance :: MonadLoggerIO m => WebT m ()+scottyAddressBalance = do setHeaders- a <- parseAddress net+ a <- parseAddress proto <- setupBin res <- getBalance a- protoSerial net proto res+ protoSerial proto res -scottyAddressesBalances :: MonadLoggerIO m => Network -> WebT m ()-scottyAddressesBalances net = do+scottyAddressesBalances :: MonadLoggerIO m => WebT m ()+scottyAddressesBalances = do setHeaders- as <- parseAddresses net+ as <- parseAddresses proto <- setupBin res <- getBalances as- protoSerial net proto res+ protoSerial proto res -scottyXpubBalances :: MonadLoggerIO m => Network -> WebT m ()-scottyXpubBalances net = do+scottyXpubBalances :: MonadLoggerIO m => WebT m ()+scottyXpubBalances = do setHeaders- xpub <- parseXpub net+ xpub <- parseXpub proto <- setupBin res <- filter (not . nullBalance . xPubBal) <$> xPubBals xpub- protoSerial net proto res+ protoSerial proto res -scottyXpubTxs :: MonadLoggerIO m => Network -> MaxLimits -> Bool -> WebT m ()-scottyXpubTxs net limits full = do+scottyXpubTxs :: MonadLoggerIO m => Bool -> WebT m ()+scottyXpubTxs full = do setHeaders- xpub <- parseXpub net+ xpub <- parseXpub start <- getStart- limit <- getLimit limits full+ limit <- getLimit full proto <- setupBin txs <- xPubTxs xpub start 0 limit if full then do txs' <- catMaybes <$> mapM (getTransaction . blockTxHash) txs- protoSerial net proto txs'- else protoSerial net proto txs+ protoSerial proto txs'+ else protoSerial proto txs -scottyXpubUnspents :: MonadLoggerIO m => Network -> MaxLimits -> WebT m ()-scottyXpubUnspents net limits = do+scottyXpubUnspents :: MonadLoggerIO m => WebT m ()+scottyXpubUnspents = do setHeaders- xpub <- parseXpub net+ xpub <- parseXpub proto <- setupBin start <- getStart- limit <- getLimit limits False+ limit <- getLimit False uns <- xPubUnspents xpub start 0 limit- protoSerial net proto uns+ protoSerial proto uns -scottyXpubSummary :: MonadLoggerIO m => Network -> WebT m ()-scottyXpubSummary net = do+scottyXpubSummary :: MonadLoggerIO m => WebT m ()+scottyXpubSummary = do setHeaders- xpub <- parseXpub net+ xpub <- parseXpub proto <- setupBin res <- xPubSummary xpub- protoSerial net proto res+ protoSerial proto res scottyPostTx :: (MonadUnliftIO m, MonadLoggerIO m)- => Network- -> Store- -> Publisher StoreEvent- -> WebT m ()-scottyPostTx net st pub = do+ => WebT m ()+scottyPostTx = do+ net <- lift $ asks webNetwork+ pub <- lift $ asks webPublisher+ st <- lift $ asks webStore setHeaders proto <- setupBin b <- S.body@@ -583,20 +669,20 @@ Just x -> return x lift (publishTx net pub st tx) >>= \case Right () -> do- protoSerial net proto (TxId (txHash tx))+ protoSerial proto (TxId (txHash tx)) Left e -> do case e of PubNoPeers -> S.status status500 PubTimeout -> S.status status500 PubPeerDisconnected -> S.status status500 PubReject _ -> S.status status400- protoSerial net proto (UserError (show e))+ protoSerial proto (UserError (show e)) S.finish scottyDbStats :: MonadLoggerIO m => WebT m () scottyDbStats = do setHeaders- BlockDB {blockDB = db} <- lift ask+ DatabaseReader {databaseHandle = db} <- lift $ asks webDB stats <- lift (getProperty db Stats) case stats of Nothing -> do@@ -604,8 +690,10 @@ Just txt -> do S.text $ cs txt -scottyEvents :: MonadLoggerIO m => Network -> Publisher StoreEvent -> WebT m ()-scottyEvents net pub = do+scottyEvents :: MonadLoggerIO m => WebT m ()+scottyEvents = do+ net <- lift $ asks webNetwork+ pub <- lift $ asks webPublisher setHeaders proto <- setupBin S.stream $ \io flush' ->@@ -629,86 +717,76 @@ else "\n" in io (lazyByteString bs) -scottyPeers :: MonadLoggerIO m => Network -> Store -> WebT m ()-scottyPeers net st = do+scottyPeers :: MonadLoggerIO m => WebT m ()+scottyPeers = do+ mgr <- lift $ asks (storeManager . webStore) setHeaders proto <- setupBin- ps <- getPeersInformation (storeManager st)- protoSerial net proto ps+ ps <- getPeersInformation mgr+ protoSerial proto ps scottyHealth :: (MonadUnliftIO m, MonadLoggerIO m)- => Network- -> Store- -> Timeouts- -> WebT m ()-scottyHealth net st tos = do+ => WebT m ()+scottyHealth = do+ net <- lift $ asks webNetwork+ st <- lift $ asks webStore+ tos <- lift $ asks webTimeouts setHeaders proto <- setupBin h <- lift $ healthCheck net (storeManager st) (storeChain st) tos when (not (healthOK h) || not (healthSynced h)) $ S.status status503- protoSerial net proto h+ protoSerial proto h runWeb :: (MonadLoggerIO m, MonadUnliftIO m) => WebConfig -> m ()-runWeb WebConfig { webDB = bdb- , webPort = port- , webNetwork = net- , webStore = st- , webPublisher = pub- , webMaxLimits = limits- , webReqLog = reqlog- , webTimeouts = tos- } = do+runWeb cfg@WebConfig {webPort = port, webReqLog = reqlog} = do req_logger <- if reqlog then Just <$> logIt else return Nothing runner <- askRunInIO- S.scottyT port (runner . withRocksDB bdb) $ do+ S.scottyT port (runner . (`runReaderT` cfg)) $ do case req_logger of Just m -> S.middleware m Nothing -> return ()- S.defaultHandler (defHandler net)- S.get "/block/best" $ scottyBestBlock net limits False- S.get "/block/best/raw" $ scottyBestBlock net limits True- S.get "/block/:block" $ scottyBlock net limits False- S.get "/block/:block/raw" $ scottyBlock net limits True- S.get "/block/height/:height" $ scottyBlockHeight net limits False- S.get "/block/height/:height/raw" $ scottyBlockHeight net limits True- S.get "/block/time/:time" $ scottyBlockTime net limits False- S.get "/block/time/:time/raw" $ scottyBlockTime net limits True- S.get "/block/heights" $ scottyBlockHeights net- S.get "/block/latest" $ scottyBlockLatest net- S.get "/blocks" $ scottyBlocks net- S.get "/mempool" $ scottyMempool net- S.get "/transaction/:txid" $ scottyTransaction net- S.get "/transaction/:txid/raw" $ scottyRawTransaction net- S.get "/transaction/:txid/after/:height" $ scottyTxAfterHeight net- S.get "/transactions" $ scottyTransactions net- S.get "/transactions/raw" $ scottyRawTransactions net- S.get "/transactions/block/:block" $ scottyBlockTransactions net limits- S.get "/transactions/block/:block/raw" $- scottyRawBlockTransactions net limits- S.get "/address/:address/transactions" $- scottyAddressTxs net limits False- S.get "/address/:address/transactions/full" $- scottyAddressTxs net limits True- S.get "/address/transactions" $ scottyAddressesTxs net limits False- S.get "/address/transactions/full" $ scottyAddressesTxs net limits True- S.get "/address/:address/unspent" $ scottyAddressUnspent net limits- S.get "/address/unspent" $ scottyAddressesUnspent net limits- S.get "/address/:address/balance" $ scottyAddressBalance net- S.get "/address/balances" $ scottyAddressesBalances net- S.get "/xpub/:xpub/balances" $ scottyXpubBalances net- S.get "/xpub/:xpub/transactions" $ scottyXpubTxs net limits False- S.get "/xpub/:xpub/transactions/full" $ scottyXpubTxs net limits True- S.get "/xpub/:xpub/unspent" $ scottyXpubUnspents net limits- S.get "/xpub/:xpub" $ scottyXpubSummary net- S.post "/transactions" $ scottyPostTx net st pub+ S.defaultHandler defHandler+ S.get "/block/best" $ scottyBestBlock False+ S.get "/block/best/raw" $ scottyBestBlock True+ S.get "/block/:block" $ scottyBlock False+ S.get "/block/:block/raw" $ scottyBlock True+ S.get "/block/height/:height" $ scottyBlockHeight False+ S.get "/block/height/:height/raw" $ scottyBlockHeight True+ S.get "/block/time/:time" $ scottyBlockTime False+ S.get "/block/time/:time/raw" $ scottyBlockTime True+ S.get "/block/heights" scottyBlockHeights+ S.get "/block/latest" scottyBlockLatest+ S.get "/blocks" scottyBlocks+ S.get "/mempool" scottyMempool+ S.get "/transaction/:txid" scottyTransaction+ S.get "/transaction/:txid/raw" scottyRawTransaction+ S.get "/transaction/:txid/after/:height" scottyTxAfterHeight+ S.get "/transactions" scottyTransactions+ S.get "/transactions/raw" scottyRawTransactions+ S.get "/transactions/block/:block" scottyBlockTransactions+ S.get "/transactions/block/:block/raw" scottyRawBlockTransactions+ S.get "/address/:address/transactions" $ scottyAddressTxs False+ S.get "/address/:address/transactions/full" $ scottyAddressTxs True+ S.get "/address/transactions" $ scottyAddressesTxs False+ S.get "/address/transactions/full" $ scottyAddressesTxs True+ S.get "/address/:address/unspent" scottyAddressUnspent+ S.get "/address/unspent" scottyAddressesUnspent+ S.get "/address/:address/balance" scottyAddressBalance+ S.get "/address/balances" scottyAddressesBalances+ S.get "/xpub/:xpub/balances" scottyXpubBalances+ S.get "/xpub/:xpub/transactions" $ scottyXpubTxs False+ S.get "/xpub/:xpub/transactions/full" $ scottyXpubTxs True+ S.get "/xpub/:xpub/unspent" scottyXpubUnspents+ S.get "/xpub/:xpub" scottyXpubSummary+ S.post "/transactions" scottyPostTx S.get "/dbstats" scottyDbStats- S.get "/events" $ scottyEvents net pub- S.get "/peers" $ scottyPeers net st- S.get "/health" $ scottyHealth net st tos+ S.get "/events" scottyEvents+ S.get "/peers" scottyPeers+ S.get "/health" scottyHealth S.notFound $ S.raise ThingNotFound getStart :: MonadLoggerIO m => WebT m (Maybe BlockRef)@@ -738,8 +816,9 @@ return $ BlockRef g maxBound else return $ MemRef q -getOffset :: Monad m => MaxLimits -> ActionT Except m Offset-getOffset limits = do+getOffset :: Monad m => WebT m Offset+getOffset = do+ limits <- lift $ asks webMaxLimits o <- S.param "offset" `S.rescue` const (return 0) when (maxLimitOffset limits > 0 && o > maxLimitOffset limits) . S.raise . UserError $@@ -748,10 +827,10 @@ getLimit :: Monad m- => MaxLimits- -> Bool- -> ActionT Except m (Maybe Limit)-getLimit limits full = do+ => Bool+ -> WebT m (Maybe Limit)+getLimit full = do+ limits <- lift $ asks webMaxLimits l <- (Just <$> S.param "limit") `S.rescue` const (return Nothing) let m = if full@@ -771,39 +850,44 @@ then Just (min m n) else Just n -parseAddress :: (Monad m, ScottyError e) => Network -> ActionT e m Address-parseAddress net = do+parseAddress :: Monad m => WebT m Address+parseAddress = do+ net <- lift $ asks webNetwork address <- S.param "address" case stringToAddr net address of Nothing -> S.next Just a -> return a -parseAddresses :: (Monad m, ScottyError e) => Network -> ActionT e m [Address]-parseAddresses net = do+parseAddresses :: Monad m => WebT m [Address]+parseAddresses = do+ net <- lift $ asks webNetwork addresses <- S.param "addresses" let as = mapMaybe (stringToAddr net) addresses unless (length as == length addresses) S.next return as -parseXpub :: (Monad m, ScottyError e) => Network -> ActionT e m XPubSpec-parseXpub net = do+parseXpub :: Monad m => WebT m XPubSpec+parseXpub = do+ net <- lift $ asks webNetwork t <- S.param "xpub"- d <- parseDeriveAddrs net+ d <- parseDeriveAddrs case xPubImport net t of Nothing -> S.next Just x -> return XPubSpec {xPubSpecKey = x, xPubDeriveType = d} parseDeriveAddrs ::- (Monad m, ScottyError e) => Network -> ActionT e m DeriveType-parseDeriveAddrs net- | getSegWit net = do- t <- S.param "derive" `S.rescue` const (return "standard")- return $- case (t :: Text) of- "segwit" -> DeriveP2WPKH- "compat" -> DeriveP2SH- _ -> DeriveNormal- | otherwise = return DeriveNormal+ Monad m => WebT m DeriveType+parseDeriveAddrs =+ lift (asks webNetwork) >>= \case+ net+ | getSegWit net -> do+ t <- S.param "derive" `S.rescue` const (return "standard")+ return $+ case (t :: Text) of+ "segwit" -> DeriveP2WPKH+ "compat" -> DeriveP2SH+ _ -> DeriveNormal+ | otherwise -> return DeriveNormal parseNoTx :: (Monad m, ScottyError e) => ActionT e m Bool parseNoTx = S.param "notx" `S.rescue` const (return False)@@ -987,7 +1071,7 @@ -> Address -> m [BlockTx] getAddressTxsLimit offset limit start addr =- applyOffset offset <$> getAddressTxs addr start limit+ applyOffset offset <$> getAddressTxs addr start ((offset +) <$> limit) getAddressTxsFull :: (Monad m, StoreRead m)@@ -1028,7 +1112,7 @@ -> Address -> m [Unspent] getAddressUnspentsLimit offset limit start addr =- applyOffset offset <$> getAddressUnspents addr start limit+ applyOffset offset <$> getAddressUnspents addr start ((offset +) <$> limit) getAddressesUnspentsLimit :: (Monad m, StoreRead m)
test/Haskoin/StoreSpec.hs view
@@ -17,7 +17,7 @@ import UnliftIO data TestStore = TestStore- { testStoreDB :: !BlockDB+ { testStoreDB :: !DatabaseReader , testStoreBlockStore :: !BlockStore , testStoreChain :: !Chain , testStoreEvents :: !(Inbox StoreEvent)@@ -41,7 +41,7 @@ bestHeight `shouldBe` 8 it "get a block and its transactions" $ withTestStore net "get-block-txs" $ \TestStore {..} ->- withRocksDB testStoreDB $ do+ withDatabaseReader testStoreDB $ do let h1 = "e8588129e146eeb0aa7abdc3590f8c5920cc5ff42daf05c23b29d4ae5b51fc22" h2 =@@ -82,7 +82,11 @@ , errorIfExists = True , compression = SnappyCompression }- let bdb = BlockDB {blockDB = db, blockDBopts = defaultReadOptions}+ let bdb =+ DatabaseReader+ { databaseHandle = db+ , databaseReadOptions = defaultReadOptions+ } let cfg = StoreConfig { storeConfMaxPeers = 20@@ -91,6 +95,7 @@ , storeConfDB = bdb , storeConfNetwork = net , storeConfListen = (`sendSTM` x)+ , storeConfCache = Nothing } withStore cfg $ \Store {..} -> lift $
+ test/Network/Haskoin/Store/Data/CacheReaderSpec.hs view
@@ -0,0 +1,60 @@+module Network.Haskoin.Store.Data.CacheReaderSpec (spec) where++import Data.List (sort)+import Haskoin (KeyIndex)+import Network.Haskoin.Store.Common (BlockRef (..))+import Network.Haskoin.Store.Data.CacheReader (blockRefScore,+ pathScore,+ scoreBlockRef,+ scorePath)+import Test.Hspec (Spec, describe)+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck (Gen, arbitrary, choose,+ elements, forAll,+ listOf, oneof)++spec :: Spec+spec = do+ describe "Score for block reference" $ do+ prop "sorts correctly" $+ forAll arbitraryBlockRefs $ \ts ->+ let scores = map blockRefScore (sort ts)+ in sort scores == reverse scores+ prop "respects identity" $+ forAll arbitraryBlockRef $ \b ->+ let score = blockRefScore b+ ref = scoreBlockRef score+ in ref == b+ describe "Score for derivation path" $ do+ prop "sorts correctly" $+ forAll arbitraryDerivationPaths $ \ds ->+ let scores = map pathScore (sort ds)+ in sort scores == scores+ prop "respects identity" $+ forAll arbitraryDerivationPath $ \d ->+ let score = pathScore d+ d' = scorePath score+ in d' == d++arbitraryDerivationPaths :: Gen [[KeyIndex]]+arbitraryDerivationPaths = listOf arbitraryDerivationPath++arbitraryDerivationPath :: Gen [KeyIndex]+arbitraryDerivationPath = do+ x <- elements [0, 1]+ y <- arbitrary+ return [x, y]++arbitraryBlockRefs :: Gen [BlockRef]+arbitraryBlockRefs = listOf arbitraryBlockRef++arbitraryBlockRef :: Gen BlockRef+arbitraryBlockRef = oneof [b, m]+ where+ b = do+ h <- choose (0, 0x07ffffff)+ p <- choose (0, 0x03ffffff)+ return BlockRef {blockRefHeight = h, blockRefPos = p}+ m = do+ t <- choose (0, 0x001fffffffffffff)+ return MemRef {memRefTime = t}