{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}

-- | The insert writes behind
-- 'Ouroboros.Consensus.Storage.LeiosDB.API.LeiosDbWriter'. They run on the
-- single writer, which calls them from its
-- 'Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.WriteQueue.WriteJob's.
module Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Insert
  ( sqlInsertEbPoint
  , sqlInsertEbBody
  , sqlInsertTxs
  ) where

import Cardano.Crypto.Hash (hashToBytes)
import Cardano.Slotting.Slot (SlotNo (..))
import Control.Monad (foldM, forM_, when)
import Control.Tracer (Tracer)
import Data.ByteString (ByteString)
import Data.Int (Int64)
import qualified Database.SQLite3.Direct as DB
import Ouroboros.Consensus.Leios.Types
  ( BytesSize
  , EndorserBlock
  , LeiosPoint (..)
  , TxOffset
  , encodeEndorserBlockSize
  , endorserBlockReferences
  )
import Ouroboros.Consensus.Storage.LeiosDB.API
  ( CompletedEbs
  , LeiosEbNotification (..)
  )
import Ouroboros.Consensus.Storage.LeiosDB.Exception (throwLeiosDbException)
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Connection
  ( Conn (..)
  , dbWithWriteTransaction
  )
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Maintenance (bumpVolatileStats)
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Primitives
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Statements
import Ouroboros.Consensus.Storage.LeiosDB.Trace (TraceLeiosDb (..))

sqlInsertEbPoint :: Conn -> LeiosPoint -> BytesSize -> IO ()
sqlInsertEbPoint :: Conn -> LeiosPoint -> Word32 -> IO ()
sqlInsertEbPoint Conn
conn LeiosPoint
point Word32
ebBytesSize = do
  inserted <- Conn -> IO TxOffset -> IO TxOffset
forall a. HasCallStack => Conn -> IO a -> IO a
dbWithWriteTransaction Conn
conn (IO TxOffset -> IO TxOffset) -> IO TxOffset -> IO TxOffset
forall a b. (a -> b) -> a -> b
$ Statement -> IO TxOffset -> IO TxOffset
forall a. Statement -> IO a -> IO a
useStmt Statement
stmt (IO TxOffset -> IO TxOffset) -> IO TxOffset -> IO TxOffset
forall a b. (a -> b) -> a -> b
$ do
    HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
stmt ParamIndex
1 (Word64 -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Word64 -> Int64) -> Word64 -> Int64
forall a b. (a -> b) -> a -> b
$ SlotNo -> Word64
unSlotNo (LeiosPoint -> SlotNo
pointSlotNo LeiosPoint
point))
    Statement
-> ParamIndex -> Hash HASH EraIndependentEbReferences -> IO ()
forall h a.
HasCallStack =>
Statement -> ParamIndex -> Hash h a -> IO ()
dbBindHash Statement
stmt ParamIndex
2 (LeiosPoint -> Hash HASH EraIndependentEbReferences
pointEbHash LeiosPoint
point)
    HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
stmt ParamIndex
3 (Word32 -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Word32
ebBytesSize)
    HasCallStack => Statement -> IO ()
Statement -> IO ()
dbStep1 Statement
stmt
    Database -> IO TxOffset
DB.changes Database
db
  bumpVolatileStats conn inserted
 where
  Conn{conVolDb :: Conn -> Database
conVolDb = Database
db, connVolStmts :: Conn -> VolStmts
connVolStmts = VolStmts{stInsertEbPoint :: VolStmts -> Statement
stInsertEbPoint = Statement
stmt}} = Conn
conn

-- | Persist an EB body. The point MUST already be present (inserted
-- via 'sqlInsertEbPoint' on the announcement path).
sqlInsertEbBody ::
  Tracer IO TraceLeiosDb ->
  Conn ->
  (LeiosEbNotification -> IO ()) ->
  LeiosPoint ->
  EndorserBlock ->
  IO CompletedEbs
sqlInsertEbBody :: Tracer IO TraceLeiosDb
-> Conn
-> (LeiosEbNotification -> IO ())
-> LeiosPoint
-> EndorserBlock
-> IO CompletedEbs
sqlInsertEbBody Tracer IO TraceLeiosDb
tracer Conn
conn LeiosEbNotification -> IO ()
notify LeiosPoint
point EndorserBlock
eb = do
  Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when ([(TxOffset, (TxHash, Word32))] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [(TxOffset, (TxHash, Word32))]
items) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
    [Char] -> IO ()
forall (m :: * -> *) a.
(HasCallStack, MonadThrow m) =>
[Char] -> m a
throwLeiosDbException [Char]
"writeEbBody: empty EB body (programmer error)"
  completedNow <- Conn -> IO CompletedEbs -> IO CompletedEbs
forall a. HasCallStack => Conn -> IO a -> IO a
dbWithWriteTransaction Conn
conn (IO CompletedEbs -> IO CompletedEbs)
-> IO CompletedEbs -> IO CompletedEbs
forall a b. (a -> b) -> a -> b
$ do
    [(TxOffset, (TxHash, Word32))]
-> ((TxOffset, (TxHash, Word32)) -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [(TxOffset, (TxHash, Word32))]
items (((TxOffset, (TxHash, Word32)) -> IO ()) -> IO ())
-> ((TxOffset, (TxHash, Word32)) -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \(TxOffset
txOffset, (TxHash
txHash, Word32
txBytesSize)) -> Statement -> IO () -> IO ()
forall a. Statement -> IO a -> IO a
useStmt Statement
stInsertEbTxsRow (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
      Statement
-> ParamIndex -> Hash HASH EraIndependentEbReferences -> IO ()
forall h a.
HasCallStack =>
Statement -> ParamIndex -> Hash h a -> IO ()
dbBindHash Statement
stInsertEbTxsRow ParamIndex
1 (LeiosPoint -> Hash HASH EraIndependentEbReferences
pointEbHash LeiosPoint
point)
      HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
stInsertEbTxsRow ParamIndex
2 (TxOffset -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral TxOffset
txOffset)
      Statement -> ParamIndex -> TxHash -> IO ()
forall h a.
HasCallStack =>
Statement -> ParamIndex -> Hash h a -> IO ()
dbBindHash Statement
stInsertEbTxsRow ParamIndex
3 TxHash
txHash
      HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
stInsertEbTxsRow ParamIndex
4 (Word32 -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Word32
txBytesSize)
      HasCallStack =>
Tracer IO TraceLeiosDb -> [Char] -> [Char] -> Statement -> IO ()
Tracer IO TraceLeiosDb -> [Char] -> [Char] -> Statement -> IO ()
dbStepInsertOrTrace
        Tracer IO TraceLeiosDb
tracer
        [Char]
"ebTxs"
        (Hash HASH EraIndependentEbReferences -> [Char]
forall a. Show a => a -> [Char]
show (LeiosPoint -> Hash HASH EraIndependentEbReferences
pointEbHash LeiosPoint
point) [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> [Char]
"@" [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> TxOffset -> [Char]
forall a. Show a => a -> [Char]
show TxOffset
txOffset)
        Statement
stInsertEbTxsRow
    -- Allocate this body's tx-bytes rows, then count the unfilled ones. Both
    -- in this transaction, so a fill can never see the rows without the count
    -- or the other way round.
    Statement -> IO () -> IO ()
forall a. Statement -> IO a -> IO a
useStmt Statement
stPreallocEbTxBytes (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
      Statement
-> ParamIndex -> Hash HASH EraIndependentEbReferences -> IO ()
forall h a.
HasCallStack =>
Statement -> ParamIndex -> Hash h a -> IO ()
dbBindHash Statement
stPreallocEbTxBytes ParamIndex
1 (LeiosPoint -> Hash HASH EraIndependentEbReferences
pointEbHash LeiosPoint
point)
      HasCallStack => Statement -> IO ()
Statement -> IO ()
dbStep1 Statement
stPreallocEbTxBytes
    -- Initialize missingTxCount and read the resulting value via
    -- @RETURNING missingTxCount@. Only /this/ point's row can have
    -- transitioned to 0 as a consequence of the insert above. If this EB's
    -- body was already written for another point, then the inserts above
    -- changed nothing, and the new count accounts for every tx already
    -- written for this EB hash, including those written while this row's
    -- count was NULL.
    missingCount <- Statement -> IO Int64 -> IO Int64
forall a. Statement -> IO a -> IO a
useStmt Statement
stInitMissingCount (IO Int64 -> IO Int64) -> IO Int64 -> IO Int64
forall a b. (a -> b) -> a -> b
$ do
      Statement
-> ParamIndex -> Hash HASH EraIndependentEbReferences -> IO ()
forall h a.
HasCallStack =>
Statement -> ParamIndex -> Hash h a -> IO ()
dbBindHash Statement
stInitMissingCount ParamIndex
1 (LeiosPoint -> Hash HASH EraIndependentEbReferences
pointEbHash LeiosPoint
point)
      Statement
-> ParamIndex -> Hash HASH EraIndependentEbReferences -> IO ()
forall h a.
HasCallStack =>
Statement -> ParamIndex -> Hash h a -> IO ()
dbBindHash Statement
stInitMissingCount ParamIndex
2 (LeiosPoint -> Hash HASH EraIndependentEbReferences
pointEbHash LeiosPoint
point)
      HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
stInitMissingCount ParamIndex
3 (Word64 -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Word64 -> Int64) -> Word64 -> Int64
forall a b. (a -> b) -> a -> b
$ SlotNo -> Word64
unSlotNo (LeiosPoint -> SlotNo
pointSlotNo LeiosPoint
point))
      Statement -> IO Int64
readReturningInt64 Statement
stInitMissingCount
    if missingCount == 0
      then do
        useStmt stMarkPointNotified $ do
          dbBindInt64 stMarkPointNotified 1 (fromIntegral $ unSlotNo (pointSlotNo point))
          dbBindHash stMarkPointNotified 2 (pointEbHash point)
          dbStep1 stMarkPointNotified
        pure [point]
      else pure []
  notify $ AcquiredEb point ebBytesSize
  forM_ completedNow $ \LeiosPoint
p -> LeiosEbNotification -> IO ()
notify (LeiosPoint -> LeiosEbNotification
AcquiredEbTxs LeiosPoint
p)
  pure completedNow
 where
  items :: [(TxOffset, (TxHash, Word32))]
items = [TxOffset] -> [(TxHash, Word32)] -> [(TxOffset, (TxHash, Word32))]
forall a b. [a] -> [b] -> [(a, b)]
zip [TxOffset
0 :: Int ..] (EndorserBlock -> [(TxHash, Word32)]
endorserBlockReferences EndorserBlock
eb)
  ebBytesSize :: Word32
ebBytesSize = EndorserBlock -> Word32
encodeEndorserBlockSize EndorserBlock
eb
  Conn{VolStmts
connVolStmts :: Conn -> VolStmts
connVolStmts :: VolStmts
connVolStmts} = Conn
conn
  VolStmts
    { Statement
stInsertEbTxsRow :: Statement
stInsertEbTxsRow :: VolStmts -> Statement
stInsertEbTxsRow
    , Statement
stPreallocEbTxBytes :: Statement
stPreallocEbTxBytes :: VolStmts -> Statement
stPreallocEbTxBytes
    , Statement
stInitMissingCount :: Statement
stInitMissingCount :: VolStmts -> Statement
stInitMissingCount
    , Statement
stMarkPointNotified :: Statement
stMarkPointNotified :: VolStmts -> Statement
stMarkPointNotified
    } = VolStmts
connVolStmts

-- | Persist tx bytes for one EB by filling the rows 'sqlInsertEbBody'
-- pre-allocated.
--
-- PREREQ: each tx's bytes hash to the tx hash the body lists at its offset.
--
-- A transaction is dropped if:
--
-- * its offset is not in the body;
-- * its offset is already filled (@filled = 1@);
-- * or its size is not the declared one.
sqlInsertTxs ::
  Tracer IO TraceLeiosDb ->
  Conn ->
  (LeiosEbNotification -> IO ()) ->
  -- | The EB the txs belong to
  LeiosPoint ->
  -- | Tx bytes by offset into the EB's body
  [(TxOffset, ByteString)] ->
  IO CompletedEbs
sqlInsertTxs :: Tracer IO TraceLeiosDb
-> Conn
-> (LeiosEbNotification -> IO ())
-> LeiosPoint
-> [(TxOffset, ByteString)]
-> IO CompletedEbs
sqlInsertTxs Tracer IO TraceLeiosDb
_tracer Conn
conn LeiosEbNotification -> IO ()
notify LeiosPoint
point [(TxOffset, ByteString)]
txBytesWithOffsets = do
  completed <- Conn -> IO CompletedEbs -> IO CompletedEbs
forall a. HasCallStack => Conn -> IO a -> IO a
dbWithWriteTransaction Conn
conn (IO CompletedEbs -> IO CompletedEbs)
-> IO CompletedEbs -> IO CompletedEbs
forall a b. (a -> b) -> a -> b
$ do
    -- Fill the pre-allocated rows of the ebTxBytes table.
    -- Count how many txs were successfully filled (i.e. not dropped).
    nFilled <-
      (Int64 -> (TxOffset, ByteString) -> IO Int64)
-> Int64 -> [(TxOffset, ByteString)] -> IO Int64
forall (t :: * -> *) (m :: * -> *) b a.
(Foldable t, Monad m) =>
(b -> a -> m b) -> b -> t a -> m b
foldM
        ( \Int64
acc (TxOffset
txOffset, ByteString
txBytes) -> do
            Statement -> IO () -> IO ()
forall a. Statement -> IO a -> IO a
useStmt Statement
stFillEbTxBytes (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
              HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindBlob Statement
stFillEbTxBytes ParamIndex
1 ByteString
ebHash
              HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
stFillEbTxBytes ParamIndex
2 (TxOffset -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral TxOffset
txOffset)
              HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindBlob Statement
stFillEbTxBytes ParamIndex
3 ByteString
txBytes
              HasCallStack => Statement -> IO ()
Statement -> IO ()
dbStep1 Statement
stFillEbTxBytes
            changed <- Database -> IO TxOffset
DB.changes Database
db
            pure (acc + fromIntegral changed)
        )
        (Int64
0 :: Int64)
        [(TxOffset, ByteString)]
txBytesWithOffsets
    if nFilled == 0
      then pure []
      else do
        -- Decrement the missingTxCount of every ebs row with this EB hash and
        -- collect the ones this batch completed.
        completedSlots <- useStmt stDecrMissingCount $ do
          dbBindBlob stDecrMissingCount 1 ebHash
          dbBindInt64 stDecrMissingCount 2 nFilled
          let loop [SlotNo]
acc =
                HasCallStack => Statement -> IO StepResult
Statement -> IO StepResult
dbStep Statement
stDecrMissingCount IO StepResult -> (StepResult -> IO [SlotNo]) -> IO [SlotNo]
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
                  StepResult
DB.Done -> [SlotNo] -> IO [SlotNo]
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([SlotNo] -> [SlotNo]
forall a. [a] -> [a]
reverse [SlotNo]
acc)
                  StepResult
DB.Row -> do
                    slot <- Word64 -> SlotNo
SlotNo (Word64 -> SlotNo) -> (Int64 -> Word64) -> Int64 -> SlotNo
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Int64 -> Word64
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int64 -> SlotNo) -> IO Int64 -> IO SlotNo
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Statement -> ColumnIndex -> IO Int64
DB.columnInt64 Statement
stDecrMissingCount ColumnIndex
0
                    left <- DB.columnInt64 stDecrMissingCount 1
                    loop (if left == 0 then slot : acc else acc)
          loop []
        -- Mark them notified so they are not completed twice.
        forM_ completedSlots $ \SlotNo
slot -> Statement -> IO () -> IO ()
forall a. Statement -> IO a -> IO a
useStmt Statement
stMarkPointNotified (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
stMarkPointNotified ParamIndex
1 (Word64 -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Word64 -> Int64) -> Word64 -> Int64
forall a b. (a -> b) -> a -> b
$ SlotNo -> Word64
unSlotNo SlotNo
slot)
          HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindBlob Statement
stMarkPointNotified ParamIndex
2 ByteString
ebHash
          HasCallStack => Statement -> IO ()
Statement -> IO ()
dbStep1 Statement
stMarkPointNotified
        pure [MkLeiosPoint slot (pointEbHash point) | slot <- completedSlots]
  -- Emit a closure-completion notification for each completed EB
  forM_ completed $ \LeiosPoint
p -> LeiosEbNotification -> IO ()
notify (LeiosPoint -> LeiosEbNotification
AcquiredEbTxs LeiosPoint
p)
  pure completed
 where
  ebHash :: ByteString
ebHash = Hash HASH EraIndependentEbReferences -> ByteString
forall h a. Hash h a -> ByteString
hashToBytes (LeiosPoint -> Hash HASH EraIndependentEbReferences
pointEbHash LeiosPoint
point)
  Conn{conVolDb :: Conn -> Database
conVolDb = Database
db, VolStmts
connVolStmts :: Conn -> VolStmts
connVolStmts :: VolStmts
connVolStmts} = Conn
conn
  VolStmts
    { Statement
stFillEbTxBytes :: Statement
stFillEbTxBytes :: VolStmts -> Statement
stFillEbTxBytes
    , Statement
stDecrMissingCount :: Statement
stDecrMissingCount :: VolStmts -> Statement
stDecrMissingCount
    , Statement
stMarkPointNotified :: VolStmts -> Statement
stMarkPointNotified :: Statement
stMarkPointNotified
    } = VolStmts
connVolStmts