haskoin-store 0.17.2 → 0.18.0
raw patch · 12 files changed
+545/−350 lines, 12 filesdep +vector
Dependencies added: vector
Files
- CHANGELOG.md +21/−0
- app/Main.hs +50/−21
- haskoin-store.cabal +5/−2
- src/Haskoin/Store.hs +4/−2
- src/Network/Haskoin/Store/Block.hs +1/−1
- src/Network/Haskoin/Store/Data.hs +56/−37
- src/Network/Haskoin/Store/Data/Cached.hs +5/−6
- src/Network/Haskoin/Store/Data/ImportDB.hs +5/−11
- src/Network/Haskoin/Store/Data/Memory.hs +5/−14
- src/Network/Haskoin/Store/Data/RocksDB.hs +3/−7
- src/Network/Haskoin/Store/Logic.hs +1/−1
- src/Network/Haskoin/Store/Web.hs +389/−248
CHANGELOG.md view
@@ -4,6 +4,27 @@ 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.18.0+### Changed+- Simplified limits and start point handling on web API.+- Made transaction streaming algorithm faster for xpub transactions.+- Extended public key summary output contains all addresses that have received transactions.++### Added+- Fine-grained control for maximum limits via command line options.+- Transaction hash as starting point.+- Block hash as starting point.+- Timestamp as starting point.+- Configurable xpub gap limit.+- Transaction count added to XPub Summary.+- UTXO count added to XPub Summary.++### Removed+- Mempool endpoint now has no limits or offsets and always returns full list.+- Extended public key summary output no longer includes any transactions.+- Offsets not allowed for transaction lists involving multiple addresses or xpubs.+- Confusing block position parameter no longer part of web API.+ ## 0.17.2 ### Changed - Stream address balances and unspent outputs.
app/Main.hs view
@@ -45,15 +45,15 @@ import Web.Scotty.Trans data Config = Config- { configDir :: !FilePath- , configPort :: !Int- , configNetwork :: !Network- , configDiscover :: !Bool- , configPeers :: ![(Host, Maybe Port)]- , configVersion :: !Bool- , configCache :: !FilePath- , configDebug :: !Bool- , configMaxCount :: !Word32+ { configDir :: !FilePath+ , configPort :: !Int+ , configNetwork :: !Network+ , configDiscover :: !Bool+ , configPeers :: ![(Host, Maybe Port)]+ , configVersion :: !Bool+ , configCache :: !FilePath+ , configDebug :: !Bool+ , configMaxLimits :: !MaxLimits } defPort :: Int@@ -65,8 +65,15 @@ netNames :: String netNames = intercalate "|" (map getNetworkName allNets) -defMaxCount :: Word32-defMaxCount = 10000+defMaxLimits :: MaxLimits+defMaxLimits =+ MaxLimits+ { maxLimitCount = 10000+ , maxLimitFull = 500+ , maxLimitOffset = 50000+ , maxLimitDefault = 100+ , maxLimitGap = 20+ } config :: Parser Config config = do@@ -93,17 +100,39 @@ help "Network peer (as many as required)" configCache <- option str $- long "cache" <> short 'c' <> help "Memory mapped disk for cache" <>+ long "cache" <> short 'c' <> help "RAM drive directory for acceleration" <> value "" configVersion <- switch $ long "version" <> short 'v' <> help "Show version" configDebug <- switch $ long "debug" <> help "Show debug messages"- configMaxCount <-+ maxLimitCount <- option auto $- metavar "INT" <> long "max" <> short 'x' <>- help "Max offset or length (0 for no limit)" <>+ metavar "INT" <> long "maxlimit" <>+ help "Max limit for listings (0 for no limit)" <> showDefault <>- value defMaxCount- return Config {..}+ value (maxLimitCount defMaxLimits)+ maxLimitFull <-+ option auto $+ metavar "INT" <> long "maxfull" <>+ help "Max limit for full listings (0 for no limit)" <>+ showDefault <>+ value (maxLimitFull defMaxLimits)+ maxLimitOffset <-+ option auto $+ metavar "INT" <> long "maxoffset" <>+ help "Max offset (0 for no limit)" <>+ showDefault <>+ value (maxLimitOffset defMaxLimits)+ maxLimitDefault <-+ option auto $+ metavar "INT" <> long "deflimit" <> help "Default limit (0 for max)" <>+ showDefault <>+ value (maxLimitDefault defMaxLimits)+ maxLimitGap <-+ option auto $+ metavar "INT" <> long "gap" <> help "Extended public key gap" <>+ showDefault <>+ value (maxLimitGap defMaxLimits)+ pure Config {configMaxLimits = MaxLimits {..}, ..} networkReader :: String -> Either String Network networkReader s@@ -141,7 +170,7 @@ exitSuccess when (null (configPeers conf) && not (configDiscover conf)) . liftIO $ die "ERROR: Specify peers to connect or enable peer discovery."- run conf+ runResourceT $ run conf where opts = info (helper <*> config) $@@ -153,7 +182,7 @@ cacheDir net "" = Nothing cacheDir net ch = Just (ch </> getNetworkName net </> "cache") -run :: MonadUnliftIO m => Config -> m ()+run :: (MonadUnliftIO m, MonadResource m) => Config -> m () run Config { configPort = port , configNetwork = net , configDiscover = disc@@ -161,7 +190,7 @@ , configCache = cache_path , configDir = db_dir , configDebug = deb- , configMaxCount = max_count+ , configMaxLimits = limits } = runStderrLoggingT . filterLogger l . flip finally clear $ do $(logInfoS) "Main" $@@ -217,7 +246,7 @@ , webDB = ldb , webPublisher = pub , webStore = str- , webMaxCount = max_count+ , webMaxLimits = limits } in runWeb wcfg where
haskoin-store.cabal view
@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: fe384b3ea5c3ecda43c35136aaabfb52c82bf6b08e13da5e1d98852039cdd315+-- hash: d74abcf656d9e3a1435569934046c4b9ad0b5ee8113bf6ec820f5553a7070ad7 name: haskoin-store-version: 0.17.2+version: 0.18.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@@ -72,6 +72,7 @@ , transformers , unliftio , unordered-containers+ , vector , wai , warp default-language: Haskell2010@@ -113,6 +114,7 @@ , transformers , unliftio , unordered-containers+ , vector , wai , warp default-language: Haskell2010@@ -156,6 +158,7 @@ , transformers , unliftio , unordered-containers+ , vector , wai , warp default-language: Haskell2010
src/Haskoin/Store.hs view
@@ -32,12 +32,14 @@ , BinSerial(..) , Except(..) , TxId(..)- , StartFrom(..) , UnixTime , BlockPos , BlockDB(..) , LayeredDB(..) , WebConfig(..)+ , MaxLimits(..)+ , Offset+ , Limit , newLayeredDB , withStore , runWeb@@ -53,7 +55,6 @@ , toTransaction , getBalance , getMempool- , getMempoolLimit , getAddressUnspents , getAddressUnspentsLimit , getAddressesUnspentsLimit@@ -75,6 +76,7 @@ , healthCheck , withBlockMem , withLayeredDB+ , insertNubInSortedBy ) where import Conduit
src/Network/Haskoin/Store/Block.hs view
@@ -81,7 +81,7 @@ instance (MonadResource m, MonadUnliftIO m) => StoreStream (ReaderT BlockRead m) where- getMempool = transPipe runLayered . getMempool+ getMempool = transPipe runLayered getMempool getOrphans = transPipe runLayered getOrphans getAddressUnspents a x = transPipe runLayered $ getAddressUnspents a x getAddressTxs a x = transPipe runLayered $ getAddressTxs a x
src/Network/Haskoin/Store/Data.hs view
@@ -106,8 +106,26 @@ sm <- lift $ getSpenders h return $ toTransaction d sm +blockAtOrBefore :: (Monad m, StoreRead m) => UnixTime -> m (Maybe BlockData)+blockAtOrBefore q = runMaybeT $ do+ a <- g 0+ b <- MaybeT getBestBlock >>= MaybeT . getBlock+ f a b+ where+ f a b+ | t b <= q = return b+ | t a > q = mzero+ | h b - h a == 1 = return a+ | otherwise = do+ let x = h a + (h b - h a) `div` 2+ m <- g x+ if t m > q then f a m else f m b+ g x = MaybeT (listToMaybe <$> getBlocksAtHeight x) >>= MaybeT . getBlock+ h = blockDataHeight+ t = fromIntegral . blockTimestamp . blockDataHeader+ class StoreStream m where- getMempool :: Maybe UnixTime -> ConduitT i (UnixTime, TxHash) m ()+ getMempool :: ConduitT i (UnixTime, TxHash) m () getOrphans :: ConduitT i (UnixTime, Tx) m () getAddressUnspents :: Address -> Maybe BlockRef -> ConduitT i Unspent m () getAddressTxs :: Address -> Maybe BlockRef -> ConduitT i BlockTx m ()@@ -896,30 +914,37 @@ data XPubSummary = XPubSummary- { xPubSummaryReceived :: !Word64- , xPubSummaryConfirmed :: !Word64+ { xPubSummaryConfirmed :: !Word64 , xPubSummaryZero :: !Word64+ , xPubSummaryReceived :: !Word64+ , xPubUnspentCount :: !Word64+ , xPubTxCount :: !Word64 , xPubExternalIndex :: !Word32 , xPubChangeIndex :: !Word32 , xPubSummaryPaths :: !(HashMap Address [KeyIndex])- , xPubSummaryTxs :: ![Transaction] } deriving (Eq, Show, Generic) xPubSummaryPairs :: A.KeyValue kv => Network -> XPubSummary -> [kv]-xPubSummaryPairs net XPubSummary { xPubSummaryReceived = r- , xPubSummaryConfirmed = c+xPubSummaryPairs net XPubSummary { xPubSummaryConfirmed = c , xPubSummaryZero = z+ , xPubSummaryReceived = r+ , xPubUnspentCount = u+ , xPubTxCount = t , xPubSummaryPaths = ps- , xPubSummaryTxs = ts , xPubExternalIndex = ext , xPubChangeIndex = ch } = [ "balance" .=- object ["received" .= r, "confirmed" .= c, "unconfirmed" .= z]+ object+ [ "confirmed" .= c+ , "unconfirmed" .= z+ , "received" .= r+ , "utxo" .= u+ , "txs" .= t+ ] , "indices" .= object ["change" .= ch, "external" .= ext] , "paths" .= object (mapMaybe (uncurry f) (M.toList ps))- , "txs" .= map (transactionToJSON net) ts ] where f a p = (.= p) <$> addrToString net a@@ -935,39 +960,39 @@ jsonValue = xPubSummaryToJSON instance BinSerial XPubSummary where- binSerial net XPubSummary { xPubSummaryReceived = r- , xPubSummaryConfirmed = c+ binSerial net XPubSummary { xPubSummaryConfirmed = c , xPubSummaryZero = z+ , xPubSummaryReceived = r+ , xPubUnspentCount = u+ , xPubTxCount = t , xPubExternalIndex = ext , xPubChangeIndex = ch , xPubSummaryPaths = ps- , xPubSummaryTxs = ts } = do- put r put c put z+ put r+ put u+ put t put ext put ch put (map (first (runPut . binSerial net)) (M.toList ps))- put $ map (runPut . binSerial net) ts- binDeserial net = do- r <- get- c <- get- z <- get- ext <- get- ch <- get- ps <- get- let xs = map (first (runGet (binDeserial net))) ps- ys <- forM xs $ \(k, v) -> case k of- Right a -> return (a, v)- Left _ -> mzero- ts <- get- let txs = map (runGet (binDeserial net)) ts- tys <- forM txs $ \case- Right a -> return a- Left _ -> mzero- return $ XPubSummary r c z ext ch (M.fromList ys) tys+ c <- get+ z <- get+ r <- get+ u <- get+ t <- get+ ext <- get+ ch <- get+ ps <- get+ let xs = map (first (runGet (binDeserial net))) ps+ ys <-+ forM xs $ \(k, v) ->+ case k of+ Right a -> return (a, v)+ Left _ -> mzero+ return $ XPubSummary c z r u t ext ch (M.fromList ys) data HealthCheck = HealthCheck { healthHeaderBest :: !(Maybe BlockHash)@@ -1073,12 +1098,6 @@ instance BinSerial TxId where binSerial _ (TxId th) = put th binDeserial _ = TxId <$> get--data StartFrom- = StartBlock !BlockHeight !BlockPos- | StartMem !UnixTime- | StartOffset !Word32- deriving (Show, Eq, Generic) data BalVal = BalVal { balValAmount :: !Word64
src/Network/Haskoin/Store/Data/Cached.hs view
@@ -79,11 +79,10 @@ getMempoolC :: (MonadResource m, MonadUnliftIO m)- => Maybe UnixTime- -> LayeredDB+ => LayeredDB -> ConduitT i (UnixTime, TxHash) m ()-getMempoolC mpu LayeredDB {layeredCache = Just db} = getMempoolDB mpu db-getMempoolC mpu LayeredDB {layeredDB = db} = getMempoolDB mpu db+getMempoolC LayeredDB {layeredCache = Just db} = getMempoolDB db+getMempoolC LayeredDB {layeredDB = db} = getMempoolDB db getOrphansC :: (MonadUnliftIO m, MonadResource m)@@ -148,9 +147,9 @@ instance (MonadUnliftIO m, MonadResource m) => StoreStream (ReaderT LayeredDB m) where- getMempool x = do+ getMempool = do c <- R.ask- getMempoolC x c+ getMempoolC c getOrphans = do c <- R.ask getOrphansC c
src/Network/Haskoin/Store/Data/ImportDB.hs view
@@ -316,26 +316,20 @@ getMempoolI :: MonadIO m- => Maybe UnixTime- -> ImportDB+ => ImportDB -> ConduitT i (UnixTime, TxHash) m ()-getMempoolI mpu ImportDB {importHashMap = hm, importLayeredDB = db} = do+getMempoolI ImportDB {importHashMap = hm, importLayeredDB = db} = do h <- hMempool <$> readTVarIO hm let hmap =- M.fromList . filter tfilter $+ M.fromList $ concatMap (\(u, l) -> map (\(t, b) -> ((u, t), b)) (M.toList l)) (M.toList h) dmap <- fmap M.fromList . liftIO . runResourceT . withLayeredDB db . runConduit $- getMempool mpu .| mapC (, True) .| sinkList+ getMempool .| mapC (, True) .| sinkList let rmap = M.filter id (M.union hmap dmap) yieldMany $ sortBy (flip compare) (M.keys rmap)- where- tfilter =- case mpu of- Just x -> (<= x) . fst . fst- Nothing -> const True instance MonadIO m => StoreRead (ReaderT ImportDB m) where isInitialized = R.ask >>= isInitializedI@@ -370,7 +364,7 @@ setBalance b = R.ask >>= setBalanceI b instance MonadIO m => StoreStream (ReaderT ImportDB m) where- getMempool m = R.ask >>= getMempoolI m+ getMempool = R.ask >>= getMempoolI getOrphans = undefined getAddressUnspents a m = undefined getAddressTxs a m = undefined
src/Network/Haskoin/Store/Data/Memory.hs view
@@ -83,18 +83,9 @@ getBalanceH :: Address -> BlockMem -> Maybe Balance getBalanceH a = fmap (balValToBalance a) . M.lookup a . hBalance -getMempoolH ::- Monad m- => Maybe UnixTime- -> BlockMem- -> ConduitT i (UnixTime, TxHash) m ()-getMempoolH mpu db =- let f ts =- case mpu of- Nothing -> False- Just pu -> ts > pu- ls =- dropWhile (f . fst) .+getMempoolH :: Monad m => BlockMem -> ConduitT i (UnixTime, TxHash) m ()+getMempoolH db =+ let ls = sortBy (flip compare) . M.toList . M.map (M.keys . M.filter id) $ hMempool db in yieldMany [(u, h) | (u, hs) <- ls, h <- hs]@@ -328,9 +319,9 @@ return $ getBalanceH a v instance MonadIO m => StoreStream (ReaderT (TVar BlockMem) m) where- getMempool m = do+ getMempool = do v <- R.ask >>= readTVarIO- getMempoolH m v+ getMempoolH v getOrphans = do v <- R.ask >>= readTVarIO getOrphansH v
src/Network/Haskoin/Store/Data/RocksDB.hs view
@@ -68,16 +68,12 @@ getMempoolDB :: (MonadIO m, MonadResource m)- => Maybe UnixTime- -> BlockDB+ => BlockDB -> ConduitT i (UnixTime, TxHash) m ()-getMempoolDB mpu BlockDB {blockDBopts = opts, blockDB = db} =+getMempoolDB BlockDB {blockDBopts = opts, blockDB = db} = x .| mapC (uncurry f) where- x =- case mpu of- Nothing -> matching db opts MemKeyS- Just pu -> matchingSkip db opts MemKeyS (MemKeyT pu)+ x = matching db opts MemKeyS f (MemKey u t) () = (u, t) f _ _ = undefined
src/Network/Haskoin/Store/Logic.hs view
@@ -232,7 +232,7 @@ importBlock net b n = do mp <- runConduit $- getMempool Nothing .| mapC snd .| filterC (`elem` bths) .| sinkList+ getMempool .| mapC snd .| filterC (`elem` bths) .| sinkList getBestBlock >>= \case Nothing | isGenesis n -> do
src/Network/Haskoin/Store/Web.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE LambdaCase #-}@@ -20,6 +21,7 @@ import Data.Aeson.Encoding (encodingToLazyByteString, fromEncoding) import Data.Bits+import qualified Data.ByteString as B import Data.ByteString.Builder import qualified Data.ByteString.Lazy as L import qualified Data.ByteString.Lazy.Char8 as C@@ -37,9 +39,12 @@ import qualified Data.Text.Encoding as T import qualified Data.Text.Lazy as T.Lazy import Data.Time.Clock+import Data.Vector (Vector, cons, (!))+import qualified Data.Vector as V import Data.Version-import Data.Word (Word32)+import Data.Word import Database.RocksDB as R+import GHC.Generics import Haskoin import Haskoin.Node import Network.Haskoin.Store.Data@@ -59,6 +64,9 @@ type WebT m = ActionT Except (ReaderT LayeredDB m) +type Offset = Word32+type Limit = Word32+ data Except = ThingNotFound | ServerError@@ -110,9 +118,19 @@ , webDB :: !LayeredDB , webPublisher :: !(Publisher StoreEvent) , webStore :: !Store- , webMaxCount :: !Word32+ , webMaxLimits :: !MaxLimits } +data MaxLimits =+ MaxLimits+ { maxLimitCount :: !Word32+ , maxLimitFull :: !Word32+ , maxLimitOffset :: !Word32+ , maxLimitDefault :: !Word32+ , maxLimitGap :: !Word32+ }+ deriving (Eq, Show)+ instance Parsable BlockHash where parseParam = maybe (Left "could not decode block hash") Right . hexToBlockHash . cs@@ -121,6 +139,29 @@ parseParam = maybe (Left "could not decode tx hash") Right . hexToTxHash . cs +data StartParam+ = StartParamHash+ { startParamHash :: !Hash256}+ | StartParamHeight+ { startParamHeight :: !Word32}+ | StartParamTime+ { startParamTime :: !UnixTime}++instance Parsable StartParam where+ parseParam s = maybe (Left "could not decode start") Right (h <|> g <|> t)+ where+ h = do+ x <- fmap B.reverse (decodeHex (cs s)) >>= eitherToMaybe . decode+ return StartParamHash {startParamHash = x}+ g = do+ x <- readMaybe (cs s) :: Maybe Integer+ guard $ 0 <= x && x <= 1230768000+ return StartParamHeight {startParamHeight = fromIntegral x}+ t = do+ x <- readMaybe (cs s)+ guard $ x > 1230768000+ return StartParamTime {startParamTime = x}+ instance MonadIO m => StoreRead (WebT m) where isInitialized = lift isInitialized getBestBlock = lift getBestBlock@@ -134,8 +175,8 @@ getBalance = lift . getBalance instance (MonadResource m, MonadUnliftIO m) =>- StoreStream (WebT (ReaderT LayeredDB m)) where- getMempool = transPipe lift . getMempool+ StoreStream (WebT m) where+ getMempool = transPipe lift getMempool getOrphans = transPipe lift getOrphans getAddressUnspents a x = transPipe lift $ getAddressUnspents a x getAddressTxs a x = transPipe lift $ getAddressTxs a x@@ -197,47 +238,55 @@ return $ pruneTx n b maybeSerial net proto res -scottyBlockHeight :: MonadLoggerIO m => Network -> WebT m ()+scottyBlockHeight ::+ (MonadLoggerIO m, MonadResource m, MonadUnliftIO m)+ => Network+ -> WebT m () scottyBlockHeight net = do cors height <- param "height" n <- parseNoTx proto <- setupBin- db <- askDB hs <- getBlocksAtHeight height+ runner <- lift askRunInIO stream $ \io flush' -> do- runResourceT . withLayeredDB db . runConduit $+ runner . runConduit $ yieldMany hs .| concatMapMC getBlock .| mapC (pruneTx n) .| streamAny net proto io flush' -scottyBlockHeights :: MonadLoggerIO m => Network -> WebT m ()+scottyBlockHeights ::+ (MonadLoggerIO m, MonadResource m, MonadUnliftIO m)+ => Network+ -> WebT m () scottyBlockHeights net = do cors heights <- param "heights" n <- parseNoTx proto <- setupBin- db <- askDB bs <- concat <$> mapM getBlocksAtHeight (nub heights)+ runner <- lift askRunInIO stream $ \io flush' -> do- runResourceT . withLayeredDB db . runConduit $+ runner . runConduit $ yieldMany (nub heights) .| concatMapMC getBlocksAtHeight .| concatMapMC getBlock .| mapC (pruneTx n) .| streamAny net proto io flush' -scottyBlockLatest :: MonadLoggerIO m => Network -> WebT m ()+scottyBlockLatest ::+ (MonadLoggerIO m, MonadResource m, MonadUnliftIO m)+ => Network+ -> WebT m () scottyBlockLatest net = do cors n <- parseNoTx proto <- setupBin- db <- askDB+ runner <- lift askRunInIO getBestBlock >>= \case Just h -> stream $ \io flush' -> do- runResourceT . withLayeredDB db . runConduit $- f n h 100 .| streamAny net proto io+ runner . runConduit $ f n h 100 .| streamAny net proto io flush' Nothing -> raise ThingNotFound where@@ -252,29 +301,32 @@ else f n (prevBlock (blockDataHeader b)) (i - 1) -scottyBlocks :: MonadLoggerIO m => Network -> WebT m ()+scottyBlocks ::+ (MonadLoggerIO m, MonadResource m, MonadUnliftIO m)+ => Network+ -> WebT m () scottyBlocks net = do cors blocks <- param "blocks" n <- parseNoTx proto <- setupBin- db <- askDB+ runner <- lift askRunInIO stream $ \io flush' -> do- runResourceT . withLayeredDB db . runConduit $+ runner . runConduit $ yieldMany (nub blocks) .| concatMapMC getBlock .| mapC (pruneTx n) .| streamAny net proto io flush' scottyMempool ::- (MonadLoggerIO m, MonadUnliftIO m) => Network -> Word32 -> WebT m ()-scottyMempool net max_count = do+ (MonadLoggerIO m, MonadResource m, MonadUnliftIO m)+ => Network+ -> WebT m ()+scottyMempool net = do cors- (l, s) <- parseLimits max_count proto <- setupBin- db <- askDB+ runner <- lift askRunInIO stream $ \io flush' -> do- runResourceT . withLayeredDB db . runConduit $- getMempoolLimit l s .| streamAny net proto io+ runner . runConduit $ getMempoolStream .| streamAny net proto io flush' scottyTransaction :: MonadLoggerIO m => Network -> WebT m ()@@ -302,88 +354,111 @@ res <- cbAfterHeight 10000 height txid protoSerial net proto res -scottyTransactions :: MonadLoggerIO m => Network -> WebT m ()+scottyTransactions ::+ (MonadLoggerIO m, MonadResource m, MonadUnliftIO m)+ => Network+ -> WebT m () scottyTransactions net = do cors txids <- param "txids" proto <- setupBin- db <- askDB+ runner <- lift askRunInIO stream $ \io flush' -> do- runResourceT . withLayeredDB db . runConduit $+ runner . runConduit $ yieldMany (nub txids) .| concatMapMC getTransaction .| streamAny net proto io flush' -scottyBlockTransactions :: MonadLoggerIO m => Network -> WebT m ()+scottyBlockTransactions ::+ (MonadLoggerIO m, MonadResource m, MonadUnliftIO m)+ => Network+ -> WebT m () scottyBlockTransactions net = do cors h <- param "block" proto <- setupBin- db <- askDB+ runner <- lift askRunInIO getBlock h >>= \case- Just b -> stream $ \io flush' -> do- runResourceT . withLayeredDB db . runConduit $- yieldMany (blockDataTxs b) .| concatMapMC getTransaction .|- streamAny net proto io- flush'- Nothing ->- raise ThingNotFound+ Just b ->+ stream $ \io flush' -> do+ runner . runConduit $+ yieldMany (blockDataTxs b) .| concatMapMC getTransaction .|+ streamAny net proto io+ flush'+ Nothing -> raise ThingNotFound -scottyRawTransactions :: MonadLoggerIO m => Network -> WebT m ()+scottyRawTransactions ::+ (MonadLoggerIO m, MonadResource m, MonadUnliftIO m)+ => Network+ -> WebT m () scottyRawTransactions net = do cors txids <- param "txids" proto <- setupBin- db <- askDB+ runner <- lift askRunInIO stream $ \io flush' -> do- runResourceT . withLayeredDB db . runConduit $+ runner . runConduit $ yieldMany (nub txids) .| concatMapMC getTransaction .| mapC transactionData .| streamAny net proto io flush' -scottyRawBlockTransactions :: MonadLoggerIO m => Network -> WebT m ()+scottyRawBlockTransactions ::+ (MonadLoggerIO m, MonadResource m, MonadUnliftIO m)+ => Network+ -> WebT m () scottyRawBlockTransactions net = do cors h <- param "block" proto <- setupBin- db <- askDB+ runner <- lift askRunInIO getBlock h >>= \case- Just b -> stream $ \io flush' -> do- runResourceT . withLayeredDB db . runConduit $- yieldMany (blockDataTxs b) .| concatMapMC getTransaction .|- mapC transactionData .|- streamAny net proto io- flush'- Nothing ->- raise ThingNotFound+ Just b ->+ stream $ \io flush' -> do+ runner . runConduit $+ yieldMany (blockDataTxs b) .| concatMapMC getTransaction .|+ mapC transactionData .|+ streamAny net proto io+ flush'+ Nothing -> raise ThingNotFound scottyAddressTxs ::- (MonadLoggerIO m, MonadUnliftIO m) => Network -> Word32 -> Bool -> WebT m ()-scottyAddressTxs net max_count full = do+ (MonadLoggerIO m, MonadUnliftIO m, MonadResource m)+ => Network+ -> MaxLimits+ -> Bool+ -> WebT m ()+scottyAddressTxs net limits full = do cors a <- parseAddress net- (l, s) <- parseLimits max_count+ s <- getStart+ o <- getOffset limits+ l <- getLimit limits full proto <- setupBin- db <- askDB+ runner <- lift askRunInIO stream $ \io flush' -> do- runResourceT . withLayeredDB db . runConduit $ f proto l s a io+ runner . runConduit $ f proto o l s a io flush' where- f proto l s a io- | full = getAddressTxsFull l s a .| streamAny net proto io- | otherwise = getAddressTxsLimit l s a .| streamAny net proto io+ f proto o l s a io+ | full = getAddressTxsFull o l s a .| streamAny net proto io+ | otherwise = getAddressTxsLimit o l s a .| streamAny net proto io scottyAddressesTxs ::- (MonadLoggerIO m, MonadUnliftIO m) => Network -> Word32 -> Bool -> WebT m ()-scottyAddressesTxs net max_count full = do+ (MonadLoggerIO m, MonadUnliftIO m, MonadResource m)+ => Network+ -> MaxLimits+ -> Bool+ -> WebT m ()+scottyAddressesTxs net limits full = do cors as <- parseAddresses net- (l, s) <- parseLimits max_count+ s <- getStart+ l <- getLimit limits full proto <- setupBin- db <- askDB+ runner <- lift askRunInIO stream $ \io flush' -> do- runResourceT . withLayeredDB db . runConduit $ f proto l s as io+ runner . runConduit $ f proto l s as io flush' where f proto l s as io@@ -391,28 +466,37 @@ | otherwise = getAddressesTxsLimit l s as .| streamAny net proto io scottyAddressUnspent ::- (MonadLoggerIO m, MonadUnliftIO m) => Network -> Word32 -> WebT m ()-scottyAddressUnspent net max_count = do+ (MonadLoggerIO m, MonadUnliftIO m, MonadResource m)+ => Network+ -> MaxLimits+ -> WebT m ()+scottyAddressUnspent net limits = do cors a <- parseAddress net- (l, s) <- parseLimits max_count+ s <- getStart+ o <- getOffset limits+ l <- getLimit limits False proto <- setupBin- db <- askDB+ runner <- lift askRunInIO stream $ \io flush' -> do- runResourceT . withLayeredDB db . runConduit $- getAddressUnspentsLimit l s a .| streamAny net proto io+ runner . runConduit $+ getAddressUnspentsLimit o l s a .| streamAny net proto io flush' scottyAddressesUnspent ::- (MonadLoggerIO m, MonadUnliftIO m) => Network -> Word32 -> WebT m ()-scottyAddressesUnspent net max_count = do+ (MonadLoggerIO m, MonadUnliftIO m, MonadResource m)+ => Network+ -> MaxLimits+ -> WebT m ()+scottyAddressesUnspent net limits = do cors as <- parseAddresses net- (l, s) <- parseLimits max_count+ s <- getStart+ l <- getLimit limits False proto <- setupBin- db <- askDB+ runner <- lift askRunInIO stream $ \io flush' -> do- runResourceT . withLayeredDB db . runConduit $+ runner . runConduit $ getAddressesUnspentsLimit l s as .| streamAny net proto io flush' @@ -436,12 +520,14 @@ } protoSerial net proto res -scottyAddressesBalances :: MonadLoggerIO m => Network -> WebT m ()+scottyAddressesBalances ::+ (MonadLoggerIO m, MonadResource m, MonadUnliftIO m)+ => Network+ -> WebT m () scottyAddressesBalances net = do cors as <- parseAddresses net proto <- setupBin- db <- askDB let f a Nothing = Balance { balanceAddress = a@@ -452,71 +538,79 @@ , balanceTotalReceived = 0 } f _ (Just b) = b+ runner <- lift askRunInIO stream $ \io flush' -> do- runResourceT . withLayeredDB db . runConduit $+ runner . runConduit $ yieldMany as .| mapMC (\a -> f a <$> getBalance a) .| streamAny net proto io flush' -scottyXpubBalances :: (MonadUnliftIO m, MonadLoggerIO m) => Network -> WebT m ()-scottyXpubBalances net = do+scottyXpubBalances ::+ (MonadUnliftIO m, MonadLoggerIO m, MonadResource m)+ => Network+ -> MaxLimits+ -> WebT m ()+scottyXpubBalances net max_limits = do cors xpub <- parseXpub net proto <- setupBin- db <- askDB+ runner <- lift askRunInIO stream $ \io flush' -> do- runResourceT . withLayeredDB db . runConduit $- xpubBals xpub .| streamAny net proto io+ runner . runConduit $+ xpubBals max_limits xpub .| streamAny net proto io flush' scottyXpubTxs ::- (MonadLoggerIO m, MonadUnliftIO m)+ (MonadLoggerIO m, MonadUnliftIO m, MonadResource m) => Network- -> Word32+ -> MaxLimits -> Bool -> WebT m ()-scottyXpubTxs net max_count full = do+scottyXpubTxs net limits full = do cors x <- parseXpub net- (l, s) <- parseLimits max_count+ s <- getStart+ l <- getLimit limits full proto <- setupBin- db <- askDB- bs <-- liftIO . runResourceT . withLayeredDB db $- runConduit $ xpubBals x .| sinkList+ as <-+ lift . runConduit $+ xpubBals limits x .| mapC (balanceAddress . xPubBal) .| sinkList+ runner <- lift askRunInIO stream $ \io flush' -> do- runResourceT . withLayeredDB db . runConduit $ f proto l s bs io+ runner . runConduit $ f proto l s as io flush' where- f proto l s bs io- | full =- getAddressesTxsFull l s (map (balanceAddress . xPubBal) bs) .|- streamAny net proto io- | otherwise =- getAddressesTxsLimit l s (map (balanceAddress . xPubBal) bs) .|- streamAny net proto io+ f proto l s as io+ | full = getAddressesTxsFull l s as .| streamAny net proto io+ | otherwise = getAddressesTxsLimit l s as .| streamAny net proto io -scottyXpubUnspents :: MonadLoggerIO m => Network -> Word32 -> WebT m ()-scottyXpubUnspents net max_count = do+scottyXpubUnspents ::+ (MonadLoggerIO m, MonadResource m, MonadUnliftIO m)+ => Network+ -> MaxLimits+ -> WebT m ()+scottyXpubUnspents net limits = do cors x <- parseXpub net proto <- setupBin- (l, s) <- parseLimits max_count- db <- askDB+ s <- getStart+ l <- getLimit limits False+ runner <- lift askRunInIO stream $ \io flush' -> do- runResourceT . withLayeredDB db . runConduit $- xpubUnspentLimit net l s x .| streamAny net proto io+ runner . runConduit $+ xpubUnspentLimit net limits l s x .| streamAny net proto io flush' scottyXpubSummary ::- (MonadLoggerIO m, MonadUnliftIO m) => Network -> Word32 -> WebT m ()-scottyXpubSummary net max_count = do+ (MonadLoggerIO m, MonadUnliftIO m, MonadResource m)+ => Network+ -> MaxLimits+ -> WebT m ()+scottyXpubSummary net max_limits = do cors x <- parseXpub net- (l, s) <- parseLimits max_count proto <- setupBin- db <- askDB- res <- liftIO . runResourceT . withLayeredDB db $ xpubSummary l s x+ res <- lift $ xpubSummary max_limits x protoSerial net proto res scottyPostTx ::@@ -602,18 +696,18 @@ when (not (healthOK h) || not (healthSynced h)) $ status status503 protoSerial net proto h -runWeb :: (MonadLoggerIO m, MonadUnliftIO m) => WebConfig -> m ()+runWeb ::+ (MonadLoggerIO m, MonadUnliftIO m, MonadResource m) => WebConfig -> m () runWeb WebConfig { webDB = db , webPort = port , webNetwork = net , webStore = st , webPublisher = pub- , webMaxCount = max_count+ , webMaxLimits = limits } = do- runner <- askRunInIO- logger <- askRunInIO l <- logIt- scottyOptsT (opts logger) (runner . withLayeredDB db) $ do+ runner <- askRunInIO+ scottyT port (runner . withLayeredDB db) $ do middleware l defaultHandler (defHandler net) S.get "/block/best" $ scottyBestBlock net@@ -622,7 +716,7 @@ S.get "/block/heights" $ scottyBlockHeights net S.get "/block/latest" $ scottyBlockLatest net S.get "/blocks" $ scottyBlocks net- S.get "/mempool" $ scottyMempool net max_count+ 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@@ -631,57 +725,87 @@ S.get "/transactions/block/:block" $ scottyBlockTransactions net S.get "/transactions/block/:block/raw" $ scottyRawBlockTransactions net S.get "/address/:address/transactions" $- scottyAddressTxs net max_count False+ scottyAddressTxs net limits False S.get "/address/:address/transactions/full" $- scottyAddressTxs net max_count True- S.get "/address/transactions" $ scottyAddressesTxs net max_count False- S.get "/address/transactions/full" $- scottyAddressesTxs net max_count True- S.get "/address/:address/unspent" $ scottyAddressUnspent net max_count- S.get "/address/unspent" $ scottyAddressesUnspent net max_count+ 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 max_count False- S.get "/xpub/:xpub/transactions/full" $ scottyXpubTxs net max_count True- S.get "/xpub/:xpub/unspent" $ scottyXpubUnspents net max_count- S.get "/xpub/:xpub" $ scottyXpubSummary net max_count+ S.get "/xpub/:xpub/balances" $ scottyXpubBalances net limits+ 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 limits S.post "/transactions" $ scottyPostTx net st pub S.get "/dbstats" scottyDbStats S.get "/events" $ scottyEvents net pub S.get "/peers" $ scottyPeers net st S.get "/health" $ scottyHealth net st notFound $ raise ThingNotFound++getStart :: (MonadResource m, MonadUnliftIO m) => WebT m (Maybe BlockRef)+getStart =+ runMaybeT $ do+ s <- MaybeT $ (Just <$> param "height") `rescue` const (return Nothing)+ do case s of+ StartParamHash {startParamHash = h} ->+ start_tx h <|> start_block h+ StartParamHeight {startParamHeight = h} -> start_height h+ StartParamTime {startParamTime = q} -> start_time q where- opts runner = def {settings = setPort port (setOnOpen f defaultSettings)}- where- f s =- runner $ do- $(logDebugS) "Web" $ "Incoming connection: " <> cs (show s)- return True+ start_height h = return $ BlockRef h maxBound+ start_block h = do+ b <- MaybeT $ getBlock (BlockHash h)+ let g = blockDataHeight b+ return $ BlockRef g maxBound+ start_tx h = do+ t <- MaybeT $ getTxData (TxHash h)+ return $ txDataBlock t+ start_time q = do+ b <- MaybeT getBestBlock >>= MaybeT . getBlock+ if q <= fromIntegral (blockTimestamp (blockDataHeader b))+ then do+ b <- MaybeT $ blockAtOrBefore q+ let g = blockDataHeight b+ return $ BlockRef g maxBound+ else return $ MemRef q -parseLimits ::- (ScottyError e, Monad m)- => Word32- -> ActionT e m (Maybe Word32, StartFrom)-parseLimits max_count = do- let b = do- height <- param "height"- pos <- param "pos" `rescue` const (return maxBound)- return $ StartBlock height pos- m = do- time <- param "time"- return $ StartMem time- o = do- o <- param "offset" `rescue` const (return 0)- let o' = if max_count == 0 then o else min o max_count- return $ StartOffset o'- l <- runMaybeT $ do- l <- MaybeT $ (Just <$> param "limit") `rescue` const (return Nothing)- return $ if max_count == 0 then l else min l max_count- s <- b <|> m <|> o- return (l, s)+getOffset :: Monad m => MaxLimits -> ActionT Except m Offset+getOffset limits = do+ o <- param "offset" `rescue` const (return 0)+ when (maxLimitOffset limits > 0 && o > maxLimitOffset limits) .+ raise . UserError $+ "offset exceeded: " <> show o <> " > " <> show (maxLimitOffset limits)+ return o +getLimit ::+ Monad m+ => MaxLimits+ -> Bool+ -> ActionT Except m (Maybe Limit)+getLimit limits full = do+ l <- (Just <$> param "limit") `rescue` const (return Nothing)+ let m =+ if full+ then if maxLimitFull limits > 0+ then maxLimitFull limits+ else maxLimitCount limits+ else maxLimitCount limits+ let d = maxLimitDefault limits+ return $+ case l of+ Nothing ->+ if d > 0 || m > 0+ then Just (min m d)+ else Nothing+ Just n ->+ if m > 0+ then Just (min m n)+ else Just n+ parseAddress net = do address <- param "address" case stringToAddr net address of@@ -814,11 +938,13 @@ xpubBals :: (MonadResource m, MonadUnliftIO m, StoreRead m)- => XPubKey+ => MaxLimits+ -> XPubKey -> ConduitT i XPubBal m ()-xpubBals xpub = go 0 >> go 1+xpubBals limits xpub = go 0 >> go 1 where- go m = yieldMany (addrs m) .| mapMC (uncurry bal) .| gap 20+ go m =+ yieldMany (addrs m) .| mapMC (uncurry bal) .| gap (maxLimitGap limits) bal a p = getBalance a >>= \case Nothing -> return Nothing@@ -841,14 +967,15 @@ , StoreRead m ) => Network+ -> MaxLimits -> Maybe BlockRef -> XPubKey -> ConduitT i XPubUnspent m ()-xpubUnspent net mbr xpub = xpubBals xpub .| go+xpubUnspent net max_limits start xpub = xpubBals max_limits xpub .| go where go = awaitForever $ \XPubBal {xPubBalPath = p, xPubBal = b} ->- getAddressUnspents (balanceAddress b) mbr .|+ getAddressUnspents (balanceAddress b) start .| mapC (\t -> XPubUnspent {xPubUnspentPath = p, xPubUnspent = t}) xpubUnspentLimit ::@@ -858,54 +985,46 @@ , StoreRead m ) => Network- -> Maybe Word32- -> StartFrom+ -> MaxLimits+ -> Maybe Limit+ -> Maybe BlockRef -> XPubKey -> ConduitT i XPubUnspent m ()-xpubUnspentLimit net l s x =- xpubUnspent net (mbr s) x .| (offset s >> limit l)+xpubUnspentLimit net max_limits limit start xpub =+ xpubUnspent net max_limits start xpub .| applyLimit limit xpubSummary :: (MonadResource m, MonadUnliftIO m, StoreStream m, StoreRead m)- => Maybe Word32- -> StartFrom+ => MaxLimits -> XPubKey -> m XPubSummary-xpubSummary l s x = do- bs <- runConduit $ xpubBals x .| sinkList+xpubSummary max_limits x = do+ bs <- runConduit $ xpubBals max_limits x .| sinkList let f XPubBal {xPubBalPath = p, xPubBal = Balance {balanceAddress = a}} = (a, p) pm = H.fromList $ map f bs- txs <-- runConduit $- getAddressesTxsFull l s (map (balanceAddress . xPubBal) bs) .| sinkList- let as =- nub- [ a- | t <- txs- , let is = transactionInputs t- , let os = transactionOutputs t- , let ais =- mapMaybe- (eitherToMaybe . scriptToAddressBS . inputPkScript)- is- , let aos =- mapMaybe- (eitherToMaybe . scriptToAddressBS . outputScript)- os- , a <- ais ++ aos+ ex = foldl max 0 [i | XPubBal {xPubBalPath = [0, i]} <- bs]+ ch = foldl max 0 [i | XPubBal {xPubBalPath = [1, i]} <- bs]+ uc =+ sum+ [ c+ | XPubBal {xPubBal = Balance {balanceUnspentCount = c}} <- bs ]- ps = H.fromList $ mapMaybe (\a -> (a, ) <$> H.lookup a pm) as- ex = foldl max 0 [i | XPubBal {xPubBalPath = [x, i]} <- bs, x == 0]- ch = foldl max 0 [i | XPubBal {xPubBalPath = [x, i]} <- bs, x == 1]+ ct = sum [c | XPubBal {xPubBal = Balance {balanceTxCount = c}} <- bs]+ xt = [b | b@XPubBal {xPubBalPath = [0, _]} <- bs]+ rx =+ sum+ [ r+ | XPubBal {xPubBal = Balance {balanceTotalReceived = r}} <- xt+ ] return XPubSummary- { xPubSummaryReceived =- sum (map (balanceTotalReceived . xPubBal) bs)- , xPubSummaryConfirmed = sum (map (balanceAmount . xPubBal) bs)+ { xPubSummaryConfirmed = sum (map (balanceAmount . xPubBal) bs) , xPubSummaryZero = sum (map (balanceZero . xPubBal) bs)- , xPubSummaryPaths = ps- , xPubSummaryTxs = txs+ , xPubSummaryReceived = rx+ , xPubUnspentCount = uc+ , xPubTxCount = ct+ , xPubSummaryPaths = pm , xPubChangeIndex = ch , xPubExternalIndex = ex }@@ -963,101 +1082,123 @@ where mergeSealed sources = do prefetchedSources <- lift $ traverse ($$++ await) sources- go [(a, s) | (s, Just a) <- prefetchedSources]- go [] = pure ()- go sources = do- let (a, src1):sources1 = sortBy (f `on` fst) sources- yield a- (src2, mb) <- lift $ src1 $$++ await- let sources2 =- case mb of- Nothing -> sources1- Just b -> (b, src2) : sources1- go sources2+ go . V.fromList . nubBy (\a b -> f (fst a) (fst b) == EQ) $+ sortBy (f `on` fst) [(a, s) | (s, Just a) <- prefetchedSources]+ go sources+ | V.null sources = pure ()+ | otherwise = do+ let (a, src1) = V.head sources+ sources1 = V.tail sources+ yield a+ (src2, mb) <- lift $ src1 $$++ await+ let sources2 =+ case mb of+ Nothing -> sources1+ Just b ->+ insertNubInSortedBy (f `on` fst) (b, src2) sources1+ go sources2 -getMempoolLimit ::+insertNubInSortedBy :: (a -> a -> Ordering) -> a -> Vector a -> Vector a+insertNubInSortedBy f x xs+ | null xs = xs+ | otherwise =+ case find_idx 0 (length xs - 1) of+ Nothing -> xs+ Just i ->+ let (xs1, xs2) = V.splitAt i xs+ in xs1 <> x `cons` xs2+ where+ find_idx a b+ | f (xs ! a) x == EQ = Nothing+ | f (xs ! b) x == EQ = Nothing+ | f (xs ! b) x == LT = Just (b + 1)+ | f (xs ! a) x == GT = Just a+ | b - a == 1 = Just b+ | otherwise =+ let c = a + (b - a) `div` 2+ z = xs ! c+ in if f z x == GT+ then find_idx a c+ else find_idx c b++getMempoolStream :: (Monad m, StoreStream m)- => Maybe Word32- -> StartFrom- -> ConduitT i TxHash m ()-getMempoolLimit _ StartBlock {} = return ()-getMempoolLimit l (StartMem t) =- getMempool (Just t) .| mapC snd .| limit l-getMempoolLimit l s =- getMempool Nothing .| mapC snd .| (offset s >> limit l)+ => ConduitT i TxHash m ()+getMempoolStream = getMempool .| mapC snd getAddressTxsLimit :: (Monad m, StoreStream m)- => Maybe Word32- -> StartFrom+ => Offset+ -> Maybe Limit+ -> Maybe BlockRef -> Address -> ConduitT i BlockTx m ()-getAddressTxsLimit l s a =- getAddressTxs a (mbr s) .| (offset s >> limit l)+getAddressTxsLimit offset limit start addr =+ getAddressTxs addr start .| applyOffsetLimit offset limit getAddressTxsFull :: (Monad m, StoreStream m, StoreRead m)- => Maybe Word32- -> StartFrom+ => Offset+ -> Maybe Limit+ -> Maybe BlockRef -> Address -> ConduitT i Transaction m ()-getAddressTxsFull l s a =- getAddressTxsLimit l s a .| concatMapMC (getTransaction . blockTxHash)+getAddressTxsFull offset limit start addr =+ getAddressTxsLimit offset limit start addr .|+ concatMapMC (getTransaction . blockTxHash) getAddressesTxsLimit :: (MonadResource m, MonadUnliftIO m, StoreStream m)- => Maybe Word32- -> StartFrom+ => Maybe Limit+ -> Maybe BlockRef -> [Address] -> ConduitT i BlockTx m ()-getAddressesTxsLimit l s as =- mergeSourcesBy (flip compare `on` blockTxBlock) xs .| dedup .|- (offset s >> limit l)+getAddressesTxsLimit limit start addrs =+ mergeSourcesBy (flip compare `on` blockTxBlock) xs .| applyLimit limit where- xs = map (\a -> getAddressTxs a (mbr s)) as+ xs = map (`getAddressTxs` start) addrs getAddressesTxsFull :: (MonadResource m, MonadUnliftIO m, StoreStream m, StoreRead m)- => Maybe Word32- -> StartFrom+ => Maybe Limit+ -> Maybe BlockRef -> [Address] -> ConduitT i Transaction m ()-getAddressesTxsFull l s as =- getAddressesTxsLimit l s as .| concatMapMC (getTransaction . blockTxHash)+getAddressesTxsFull limit start addrs =+ getAddressesTxsLimit limit start addrs .|+ concatMapMC (getTransaction . blockTxHash) getAddressUnspentsLimit :: (Monad m, StoreStream m)- => Maybe Word32- -> StartFrom+ => Offset+ -> Maybe Limit+ -> Maybe BlockRef -> Address -> ConduitT i Unspent m ()-getAddressUnspentsLimit l s a =- getAddressUnspents a (mbr s) .| (offset s >> limit l)+getAddressUnspentsLimit offset limit start addr =+ getAddressUnspents addr start .| applyOffsetLimit offset limit getAddressesUnspentsLimit :: (Monad m, StoreStream m)- => Maybe Word32- -> StartFrom+ => Maybe Limit+ -> Maybe BlockRef -> [Address] -> ConduitT i Unspent m ()-getAddressesUnspentsLimit l s as =+getAddressesUnspentsLimit limit start addrs = mergeSourcesBy (flip compare `on` unspentBlock)- (map (`getAddressUnspents` mbr s) as) .|- (offset s >> limit l)+ (map (`getAddressUnspents` start) addrs) .|+ applyLimit limit -offset :: Monad m => StartFrom -> ConduitT i i m ()-offset (StartOffset o) = dropC (fromIntegral o)-offset _ = return ()+applyOffsetLimit :: Monad m => Offset -> Maybe Limit -> ConduitT i i m ()+applyOffsetLimit offset limit = applyOffset offset >> applyLimit limit -limit :: Monad m => Maybe Word32 -> ConduitT i i m ()-limit Nothing = mapC id-limit (Just n) = takeC (fromIntegral n)+applyOffset :: Monad m => Offset -> ConduitT i i m ()+applyOffset = dropC . fromIntegral -mbr :: StartFrom -> Maybe BlockRef-mbr (StartBlock h p) = Just (BlockRef h p)-mbr (StartMem t) = Just (MemRef t)-mbr (StartOffset _) = Nothing+applyLimit :: Monad m => Maybe Limit -> ConduitT i i m ()+applyLimit Nothing = mapC id+applyLimit (Just l) = takeC (fromIntegral l) conduitToQueue :: MonadIO m => TBQueue (Maybe a) -> ConduitT a Void m () conduitToQueue q =