haskoin-store-0.64.1: src/Haskoin/Store/Database/Reader.hs

{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE FlexibleInstances #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE RecordWildCards #-}

module Haskoin.Store.Database.Reader (
    -- * RocksDB Database Access
    DatabaseReader (..),
    DatabaseReaderT,
    withDatabaseReader,
    addrTxCF,
    addrOutCF,
    txCF,
    spenderCF,
    unspentCF,
    blockCF,
    heightCF,
    balanceCF,
) where

import Conduit (
    ConduitT,
    dropWhileC,
    lift,
    mapC,
    runConduit,
    sinkList,
    (.|),
 )
import Control.Monad.Except (runExceptT, throwError)
import Control.Monad.Reader (ReaderT, ask, asks, runReaderT)
import Data.Bits ((.&.))
import qualified Data.ByteString as BS
import Data.Default (def)
import Data.Function (on)
import Data.List (sortOn)
import Data.Maybe (fromMaybe)
import Data.Ord (Down (..))
import Data.Serialize (encode)
import Data.Word (Word32, Word64)
import Database.RocksDB (
    ColumnFamily,
    Config (..),
    DB (..),
    Iterator,
    withDBCF,
    withIterCF,
 )
import Database.RocksDB.Query (
    insert,
    matching,
    matchingAsListCF,
    matchingSkip,
    retrieve,
    retrieveCF,
 )
import Haskoin (
    Address,
    BlockHash,
    BlockHeight,
    Network,
    OutPoint (..),
    TxHash,
    pubSubKey,
    txHash,
 )
import Haskoin.Store.Common
import Haskoin.Store.Data
import Haskoin.Store.Database.Types
import qualified System.Metrics as Metrics
import System.Metrics.Counter (Counter)
import qualified System.Metrics.Counter as Counter
import UnliftIO (MonadIO, MonadUnliftIO, liftIO)

type DatabaseReaderT = ReaderT DatabaseReader

data DatabaseReader = DatabaseReader
    { databaseHandle :: !DB
    , databaseMaxGap :: !Word32
    , databaseInitialGap :: !Word32
    , databaseNetwork :: !Network
    , databaseMetrics :: !(Maybe DataMetrics)
    }

incrementCounter ::
    MonadIO m =>
    (DataMetrics -> Counter) ->
    Int ->
    ReaderT DatabaseReader m ()
incrementCounter f i =
    asks databaseMetrics >>= \case
        Just s -> liftIO $ Counter.add (f s) (fromIntegral i)
        Nothing -> return ()

dataVersion :: Word32
dataVersion = 17

withDatabaseReader ::
    MonadUnliftIO m =>
    Network ->
    Word32 ->
    Word32 ->
    FilePath ->
    Maybe DataMetrics ->
    DatabaseReaderT m a ->
    m a
withDatabaseReader net igap gap dir stats f =
    withDBCF dir cfg columnFamilyConfig $ \db -> do
        let bdb =
                DatabaseReader
                    { databaseHandle = db
                    , databaseMaxGap = gap
                    , databaseNetwork = net
                    , databaseInitialGap = igap
                    , databaseMetrics = stats
                    }
        initRocksDB bdb
        runReaderT f bdb
  where
    cfg = def{createIfMissing = True, maxFiles = Just (-1)}

columnFamilyConfig :: [(String, Config)]
columnFamilyConfig =
    [ ("addr-tx", def{prefixLength = Just 22, bloomFilter = True})
    , ("addr-out", def{prefixLength = Just 22, bloomFilter = True})
    , ("tx", def{prefixLength = Just 33, bloomFilter = True})
    , ("spender", def{prefixLength = Just 33, bloomFilter = True})
    , ("unspent", def{prefixLength = Just 37, bloomFilter = True})
    , ("block", def{prefixLength = Just 33, bloomFilter = True})
    , ("height", def{prefixLength = Nothing, bloomFilter = True})
    , ("balance", def{prefixLength = Just 22, bloomFilter = True})
    ]

addrTxCF :: DB -> ColumnFamily
addrTxCF = head . columnFamilies

addrOutCF :: DB -> ColumnFamily
addrOutCF db = columnFamilies db !! 1

txCF :: DB -> ColumnFamily
txCF db = columnFamilies db !! 2

spenderCF :: DB -> ColumnFamily
spenderCF db = columnFamilies db !! 3

unspentCF :: DB -> ColumnFamily
unspentCF db = columnFamilies db !! 4

blockCF :: DB -> ColumnFamily
blockCF db = columnFamilies db !! 5

heightCF :: DB -> ColumnFamily
heightCF db = columnFamilies db !! 6

balanceCF :: DB -> ColumnFamily
balanceCF db = columnFamilies db !! 7

initRocksDB :: MonadIO m => DatabaseReader -> m ()
initRocksDB DatabaseReader{databaseHandle = db} = do
    e <-
        runExceptT $
            retrieve db VersionKey >>= \case
                Just v
                    | v == dataVersion -> return ()
                    | otherwise -> throwError "Incorrect RocksDB database version"
                Nothing -> setInitRocksDB db
    case e of
        Left s -> error s
        Right () -> return ()

setInitRocksDB :: MonadIO m => DB -> m ()
setInitRocksDB db = insert db VersionKey dataVersion

addressConduit ::
    MonadUnliftIO m =>
    Address ->
    Maybe Start ->
    Iterator ->
    ConduitT i TxRef (DatabaseReaderT m) ()
addressConduit a s it =
    x .| mapC (uncurry f)
  where
    f (AddrTxKey _ t) () = t
    f _ _ = undefined
    x = case s of
        Nothing ->
            matching it (AddrTxKeyA a)
        Just (AtBlock bh) ->
            matchingSkip
                it
                (AddrTxKeyA a)
                (AddrTxKeyB a (BlockRef bh maxBound))
        Just (AtTx txh) ->
            lift (getTxData txh) >>= \case
                Just TxData{txDataBlock = b@BlockRef{}} ->
                    matchingSkip it (AddrTxKeyA a) (AddrTxKeyB a b)
                Just TxData{txDataBlock = MemRef{}} ->
                    let cond (AddrTxKey _a (TxRef MemRef{} th)) =
                            th /= txh
                        cond (AddrTxKey _a (TxRef BlockRef{} _th)) =
                            False
                     in matching it (AddrTxKeyA a)
                            .| (dropWhileC (cond . fst) >> mapC id)
                Nothing -> return ()

unspentConduit ::
    MonadUnliftIO m =>
    Address ->
    Maybe Start ->
    Iterator ->
    ConduitT i Unspent (DatabaseReaderT m) ()
unspentConduit a s it =
    x .| mapC (uncurry toUnspent)
  where
    x = case s of
        Nothing ->
            matching it (AddrOutKeyA a)
        Just (AtBlock h) ->
            matchingSkip
                it
                (AddrOutKeyA a)
                (AddrOutKeyB a (BlockRef h maxBound))
        Just (AtTx txh) ->
            lift (getTxData txh) >>= \case
                Just TxData{txDataBlock = b@BlockRef{}} ->
                    matchingSkip it (AddrOutKeyA a) (AddrOutKeyB a b)
                Just TxData{txDataBlock = MemRef{}} ->
                    let cond (AddrOutKey _a MemRef{} p) =
                            outPointHash p /= txh
                        cond (AddrOutKey _a BlockRef{} _p) =
                            False
                     in matching it (AddrOutKeyA a)
                            .| (dropWhileC (cond . fst) >> mapC id)
                Nothing -> return ()

instance MonadIO m => StoreReadBase (DatabaseReaderT m) where
    getNetwork = asks databaseNetwork

    getTxData th = do
        db <- asks databaseHandle
        retrieveCF db (txCF db) (TxKey th) >>= \case
            Nothing -> return Nothing
            Just t -> do
                incrementCounter dataTxCount 1
                return (Just t)

    getSpender op = do
        db <- asks databaseHandle
        retrieveCF db (spenderCF db) (SpenderKey op) >>= \case
            Nothing -> return Nothing
            Just s -> do
                incrementCounter dataSpenderCount 1
                return (Just s)

    getUnspent p = do
        db <- asks databaseHandle
        fmap (valToUnspent p) <$> retrieveCF db (unspentCF db) (UnspentKey p) >>= \case
            Nothing -> return Nothing
            Just u -> do
                incrementCounter dataUnspentCount 1
                return (Just u)

    getBalance a = do
        db <- asks databaseHandle
        incrementCounter dataBalanceCount 1
        fmap (valToBalance a) <$> retrieveCF db (balanceCF db) (BalKey a)

    getMempool = do
        db <- asks databaseHandle
        incrementCounter dataMempoolCount 1
        fromMaybe [] <$> retrieve db MemKey

    getBestBlock = do
        incrementCounter dataBestCount 1
        asks databaseHandle >>= (`retrieve` BestKey)

    getBlocksAtHeight h = do
        db <- asks databaseHandle
        retrieveCF db (heightCF db) (HeightKey h) >>= \case
            Nothing -> return []
            Just ls -> do
                incrementCounter dataBlockCount (length ls)
                return ls

    getBlock h = do
        db <- asks databaseHandle
        retrieveCF db (blockCF db) (BlockKey h) >>= \case
            Nothing -> return Nothing
            Just b -> do
                incrementCounter dataBlockCount 1
                return (Just b)

instance MonadUnliftIO m => StoreReadExtra (DatabaseReaderT m) where
    getAddressesTxs addrs limits = do
        txs <- applyLimits limits . sortOn Down . concat <$> mapM f addrs
        incrementCounter dataAddrTxCount (length txs)
        return txs
      where
        l = deOffset limits
        f a = do
            db <- asks databaseHandle
            withIterCF db (addrTxCF db) $ \it ->
                runConduit $
                    addressConduit a (start l) it
                        .| applyLimitC (limit l)
                        .| sinkList

    getAddressesUnspents addrs limits = do
        us <- applyLimits limits . sortOn Down . concat <$> mapM f addrs
        incrementCounter dataUnspentCount (length us)
        return us
      where
        l = deOffset limits
        f a = do
            db <- asks databaseHandle
            withIterCF db (addrOutCF db) $ \it ->
                runConduit $
                    unspentConduit a (start l) it
                        .| applyLimitC (limit l)
                        .| sinkList

    getAddressUnspents a limits = do
        db <- asks databaseHandle
        us <- withIterCF db (addrOutCF db) $ \it ->
            runConduit $
                x it .| applyLimitsC limits .| mapC (uncurry toUnspent) .| sinkList
        incrementCounter dataUnspentCount (length us)
        return us
      where
        x it = case start limits of
            Nothing ->
                matching it (AddrOutKeyA a)
            Just (AtBlock h) ->
                matchingSkip
                    it
                    (AddrOutKeyA a)
                    (AddrOutKeyB a (BlockRef h maxBound))
            Just (AtTx txh) ->
                lift (getTxData txh) >>= \case
                    Just TxData{txDataBlock = b@BlockRef{}} ->
                        matchingSkip it (AddrOutKeyA a) (AddrOutKeyB a b)
                    Just TxData{txDataBlock = MemRef{}} ->
                        let cond (AddrOutKey _a MemRef{} p) =
                                outPointHash p /= txh
                            cond (AddrOutKey _a BlockRef{} _p) =
                                False
                         in matching it (AddrOutKeyA a)
                                .| (dropWhileC (cond . fst) >> mapC id)
                    _ -> matching it (AddrOutKeyA a)

    getAddressTxs a limits = do
        db <- asks databaseHandle
        txs <- withIterCF db (addrTxCF db) $ \it ->
            runConduit $
                addressConduit a (start limits) it
                    .| applyLimitsC limits
                    .| sinkList
        incrementCounter dataAddrTxCount (length txs)
        return txs

    getMaxGap = asks databaseMaxGap

    getInitialGap = asks databaseInitialGap

    getNumTxData i = do
        db <- asks databaseHandle
        let (sk, w) = decodeTxKey i
        ls <- liftIO $ matchingAsListCF db (txCF db) (TxKeyS sk)
        let f t =
                let bs = encode $ txHash (txData t)
                    b = BS.head (BS.drop 6 bs)
                    w' = b .&. 0xf8
                 in w == w'
            txs = filter f $ map snd ls
        incrementCounter dataTxCount (length txs)
        return txs

    getBalances as = do
        zipWith f as <$> mapM getBalance as
      where
        f a Nothing = zeroBalance a
        f _ (Just b) = b

    xPubBals xpub = do
        igap <- getInitialGap
        gap <- getMaxGap
        ext1 <- derive_until_gap gap 0 (take (fromIntegral igap) (aderiv 0 0))
        if all (nullBalance . xPubBal) ext1
            then do
                incrementCounter dataXPubBals (length ext1)
                return ext1
            else do
                ext2 <- derive_until_gap gap 0 (aderiv 0 igap)
                chg <- derive_until_gap gap 1 (aderiv 1 0)
                let bals = ext1 <> ext2 <> chg
                incrementCounter dataXPubBals (length bals)
                return bals
      where
        aderiv m =
            deriveAddresses
                (deriveFunction (xPubDeriveType xpub))
                (pubSubKey (xPubSpecKey xpub) m)
        xbalance m b n = XPubBal{xPubBalPath = [m, n], xPubBal = b}
        derive_until_gap _ _ [] = return []
        derive_until_gap gap m as = do
            let (as1, as2) = splitAt (fromIntegral gap) as
            bs <- getBalances (map snd as1)
            let xbs = zipWith (xbalance m) bs (map fst as1)
            if all nullBalance bs
                then return xbs
                else (xbs <>) <$> derive_until_gap gap m as2

    xPubUnspents _xspec xbals limits = do
        us <- concat <$> mapM h cs
        incrementCounter dataXPubUnspents (length us)
        return . applyLimits limits $ sortOn Down us
      where
        l = deOffset limits
        cs = filter ((> 0) . balanceUnspentCount . xPubBal) xbals
        i b = do
            us <- getAddressUnspents (balanceAddress (xPubBal b)) l
            return us
        f b t = XPubUnspent{xPubUnspentPath = xPubBalPath b, xPubUnspent = t}
        h b = map (f b) <$> i b

    xPubTxs _xspec xbals limits = do
        let as =
                map balanceAddress $
                    filter (not . nullBalance) $
                        map xPubBal xbals
        txs <- getAddressesTxs as limits
        incrementCounter dataXPubTxs (length txs)
        return txs

    xPubTxCount xspec xbals = do
        incrementCounter dataXPubTxCount 1
        fromIntegral . length <$> xPubTxs xspec xbals def