Use a vector based sqlite index.

cabal.project

@@ -304,4 +304,4 @@ source-repository-package
 source-repository-package
   type: git
   location: https://github.com/raduom/hysterical-screams
-  tag: 01fcf5cd9e8b89c1089dc4bb753bf007aca7f8d0
+  tag: e30c0f8e6e86e6f57249aa4437ac8cadc0d49420

nix/pkgs/haskell/sha256map.nix

@@ -16,5 +16,5 @@
   "https://github.com/input-output-hk/servant-purescript"."44e7cacf109f84984cd99cd3faf185d161826963" = "10pb0yfp80jhb9ryn65a4rha2lxzsn2vlhcc6xphrrkf4x5lhzqc";
   "https://github.com/input-output-hk/Win32-network"."3825d3abf75f83f406c1f7161883c438dac7277d" = "19wahfv726fa3mqajpqdqhnl9ica3xmf68i254q45iyjcpj1psqx";
   "https://github.com/Quid2/flat"."ee59880f47ab835dbd73bea0847dab7869fc20d8" = "1lrzknw765pz2j97nvv9ip3l1mcpf2zr4n56hwlz0rk7wq7ls4cm";
-  "https://github.com/raduom/hysterical-screams"."01fcf5cd9e8b89c1089dc4bb753bf007aca7f8d0" = "0rfaww51jd0a2f2h2xp87bx6yikac9jfncyp8pfdri2l3dyzbd14";
+  "https://github.com/raduom/hysterical-screams"."e30c0f8e6e86e6f57249aa4437ac8cadc0d49420" = "1ayqxfk7d99mcir7syji2404vzz4s6yg56rmw9aj56djvd6804v2";
 }

plutus-streaming/app/Main.hs

@@ -10,12 +10,10 @@ import Cardano.Api.Extras ()
 import Control.Concurrent.MVar (MVar, newMVar, putMVar, takeMVar)
 import Data.Aeson.Text qualified as Aeson
 import Data.Text.Lazy qualified as TL
-import Index.Sqlite (SqliteIndex, insert)
 import Index.TxIdStatus (TxStatusIndex, openIx)
-import Ledger.TxId (TxId)
+import Index.VSqlite (insert)
 import Options.Applicative (Alternative ((<|>)), Parser, auto, execParser, flag', help, helper, info, long, metavar,
                             option, str, strOption, (<**>))
-import Plutus.ChainIndex.Types (TxConfirmedState)
 import Plutus.Streaming (ChainSyncEvent (RollBackward, RollForward), SimpleChainSyncEvent,
                          withSimpleChainSyncEventStream)
 import Streaming.Prelude qualified as S

plutus-streaming/plutus-streaming.cabal

@@ -46,6 +46,7 @@ library
         bytestring,
         cardano-api,
         containers,
+        lens,
         hysterical-screams,
         plutus-chain-index-core,
         plutus-ledger,
@@ -60,6 +61,8 @@ executable plutus-streaming-cli
     import: lang
     hs-source-dirs:   app
     main-is:          Main.hs
+    ghc-options:
+        -rtsopts=all
     build-depends:
         plutus-streaming,
         base >=4.9 && <5,

plutus-streaming/src/Index/TxIdStatus.hs

@@ -5,6 +5,7 @@ module Index.TxIdStatus where
 
 import Cardano.Api (Block (Block), BlockHeader (BlockHeader), BlockInMode (BlockInMode), BlockNo (BlockNo), CardanoMode)
 import Cardano.Api qualified as C
+import Control.Lens.Operators
 import Control.Monad (forM_)
 import Data.Map (Map)
 import Data.Map qualified as Map
@@ -13,9 +14,9 @@ import Data.Monoid (Last (Last), Sum (Sum, getSum))
 import Data.Sequence (Seq, ViewL (..))
 import Data.Sequence qualified as Seq
 import Database.SQLite.Simple (execute, execute_)
-import Index.Split (SplitIndex (SplitIndex, siBuffered, siHandle))
-import Index.Sqlite (SqliteIndex)
-import Index.Sqlite qualified as Ix
+import Index.VSplit (SplitIndex, getBuffer, handle, storage)
+import Index.VSqlite (SqliteIndex)
+import Index.VSqlite qualified as Ix
 import Ledger.TxId (TxId)
 import Plutus.ChainIndex.Tx (ChainIndexTx (_citxTxId))
 import Plutus.ChainIndex.Types (BlockNumber (BlockNumber),
@@ -28,27 +29,29 @@ type TxStatusIndex = SqliteIndex SimpleChainSyncEvent () TxId (Maybe TxConfirmed
 
 openIx :: FilePath -> IO TxStatusIndex
 openIx path =
-  fromJust <$> Ix.new query onInsert store 2000 path
+  fromJust <$> Ix.newBoxed query store onInsert 2000 2000 path
 
 -- Ignore notifications for now
-onInsert :: SimpleChainSyncEvent -> TxStatusIndex -> IO [()]
+onInsert :: TxStatusIndex -> SimpleChainSyncEvent -> IO [()]
 onInsert _ _ = pure []
 
 -- No one will query this for now.
-query :: TxStatusIndex -> TxId -> Seq SimpleChainSyncEvent -> IO (Maybe TxConfirmedState)
+query :: TxStatusIndex -> TxId -> [SimpleChainSyncEvent] -> IO (Maybe TxConfirmedState)
 query _ _ _ = pure Nothing
 
 store :: TxStatusIndex -> IO ()
-store SplitIndex{siHandle, siBuffered} = do
-  let bufferedTxs = foldTxs $ getTxs . getBlocks <$> siBuffered
-  execute_ siHandle "CREATE TABLE IF NOT EXISTS tx_state (txid TEXT PRIMARY KEY, confirmations INTEGER)"
-  -- execute_ siHandle "BEGIN"
+store ix = do
+  -- bufferedEvents <- getBuffer (ix ^. storage)
+  let c           = ix ^. handle
+      -- bufferedTxs = foldTxs $ getTxs . getBlocks <$> bufferedEvents
+  execute_ c "CREATE TABLE IF NOT EXISTS tx_state (txid TEXT PRIMARY KEY, confirmations INTEGER)"
+  -- execute_ handle "BEGIN"
   -- forM_ (Map.assocs bufferedTxs) $
-  --   \(txid, v) -> execute siHandle "INSERT INTO tx_state (txid, confirmations) VALUES (?, ?)" (show txid, getSum $ timesConfirmed v)
+  --   \(txid, v) -> execute handle "INSERT INTO tx_state (txid, confirmations) VALUES (?, ?)" (show txid, getSum $ timesConfirmed v)
   -- This will really work your SSD to death, and it is not very useful, since
   -- all txs that are persisted are settled.
-  -- execute siHandle "UPDATE tx_state SET confirmations = confirmations + ?" (Only $ Map.size bufferedTxs)
-  -- execute_ siHandle "COMMIT"
+  -- execute handle "UPDATE tx_state SET confirmations = confirmations + ?" (Only $ Map.size bufferedTxs)
+  -- execute_ handle "COMMIT"
 
 getBlocks :: SimpleChainSyncEvent -> BlockInMode CardanoMode
 getBlocks (RollForward block _tip) = block
@@ -73,7 +76,7 @@ getTxs (BlockInMode (Block header transactions) era) =
     go (BlockHeader _ _ blockNo) txs era' =
       (blockNo, _citxTxId <$> catMaybes (either (const Nothing) Just . fromCardanoTx era' <$> txs))
 
-foldTxs :: Seq (BlockNo, [TxId]) -> Map TxId TxConfirmedState
+foldTxs :: [(BlockNo, [TxId])] -> Map TxId TxConfirmedState
 foldTxs bs = snd $ foldl go (0, Map.empty) bs
   where
     go :: (Int, Map TxId TxConfirmedState)