-- | A SQLite-backed event source and sink.
--
-- This is the recommended persistence backend for new deployments.
-- See 'withSQLiteEventStore' which handles migration from the legacy
-- file-based store automatically.
--
-- == Architecture
--
-- Events are stored in a single @events@ table with an integer primary key
-- (@event_id@) and a BLOB column (@event_data@) containing CBOR-encoded event
-- data (via 'ToCBOR' / 'FromCBOR'). The database uses WAL journal mode with
-- @synchronous=NORMAL@ to avoid per-write fsyncs while still syncing at WAL
-- checkpoints.
--
-- == Schema migrations
--
-- The schema version is tracked in @PRAGMA user_version@ and migrated on open
-- (see 'applyMigrations'). Version 1 stored event data as JSON; opening a
-- version 1 database re-encodes every row to CBOR in one transaction and runs
-- @VACUUM@ afterwards to reclaim the freed space. A row that fails to decode
-- aborts the migration (and thereby node startup) with
-- 'EventDecodingException', rolling back to an intact version 1 database.
-- The legacy file-based store (JSON lines) is migrated by decoding each line
-- as JSON and inserting CBOR.
--
-- == Async write-behind
--
-- To keep persistence off the hot path, writes use an async write-behind
-- strategy. 'putEvent' and 'putEvents' encode events eagerly to strict
-- 'ByteString' and enqueue them into a bounded 'TBQueue'. A background writer
-- thread drains the queue and batch-inserts rows using 'executeMany' inside a
-- single transaction, amortising WAL frame writes across multiple events.
--
-- The last-seen event id 'TVar' is updated atomically at enqueue time (not
-- write time), so de-duplication and source-of-truth tracking remain correct
-- even though the physical write is deferred.
--
-- All SQLite writes go through the single writer thread, preventing concurrent
-- access races. Operations that need data flushed (rotation, reads) use a flush
-- marker: a 'TMVar' is enqueued and the caller blocks until the writer thread
-- has processed all preceding items and signalled it. 'sourceEvents'
-- auto-flushes before reading, so callers always see all enqueued events.
--
-- == Tradeoffs
--
-- * __Writer thread crash surfacing__: The background writer is 'link'ed to
--   the calling thread. If it dies (e.g. SQLite I/O error), the exception
--   propagates immediately rather than leaving the node silently stalled.
--   Use 'withSQLiteEventStore' which handles cleanup (flush + cancel) on exit.
--
-- * __Data loss on hard crash__: Events in the queue that have not yet been
--   flushed to SQLite are lost on SIGKILL, OOM, or power loss. This is
--   acceptable because the L1 chain is the source of truth — the node replays
--   missed events from chain on restart.
--
-- * __Rotation ordering__: 'rotate' flushes the write queue synchronously,
--   archives the current database to @old-state/hydra-<logId>.db@ via
--   @VACUUM INTO@, then performs DELETE + INSERT. This is safe because rotation
--   is only called from the single-threaded event processing loop
--   ('processStateChanges'), so no concurrent enqueues can occur between the
--   flush and the rotation write. The archive is taken before the DELETE, so a
--   backup failure aborts rotation and leaves the events intact.
--
-- * __Separate read connection__: 'sourceEvents' streams over a dedicated
--   connection. It can run concurrently on API server threads (client
--   history replay), and @VACUUM INTO@ fails with "SQL statements in
--   progress" if a streaming statement is open on the same connection as the
--   rotation. WAL mode makes readers on a separate connection safe.
module Hydra.Events.SQLiteBased where

import Hydra.Prelude

import Cardano.Binary (decodeFull', serialize')
import Conduit (ConduitT, ResourceT, bracketP, runConduitRes, sourceFile, yield, (.|))
import Control.Concurrent.Class.MonadSTM (flushTBQueue, newEmptyTMVarIO, newTBQueueIO, putTMVar, readTBQueue, takeTMVar, writeTBQueue, writeTVar)
import Control.Monad.Class.MonadAsync (async, cancel, link)
import Data.Aeson qualified as Aeson
import Data.ByteString qualified as BS
import Data.Conduit.Combinators (linesUnboundedAscii)
import Data.Conduit.Combinators qualified as C
import Database.SQLite.Simple (Connection, Only (..), Statement, close, closeStatement, execute, executeMany, execute_, nextRow, open, openStatement, query, query_, withTransaction)
import Hydra.Events (EventSink (..), EventSource (..), HasEventId (..))
import Hydra.Events.Rotation (EventStore (..))
import Hydra.Logging (Tracer, traceWith)
import System.Directory (createDirectoryIfMissing, doesFileExist, removeFile, renameFile)
import System.FilePath (takeBaseName, takeDirectory, takeExtension, (</>))

-- | Exception thrown when a persisted event cannot be decoded.
data EventDecodingException = EventDecodingException
  { EventDecodingException -> Word64
eventId :: Word64
  , EventDecodingException -> FilePath
decodeError :: String
  }
  deriving stock (Int -> EventDecodingException -> ShowS
[EventDecodingException] -> ShowS
EventDecodingException -> FilePath
(Int -> EventDecodingException -> ShowS)
-> (EventDecodingException -> FilePath)
-> ([EventDecodingException] -> ShowS)
-> Show EventDecodingException
forall a.
(Int -> a -> ShowS) -> (a -> FilePath) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> EventDecodingException -> ShowS
showsPrec :: Int -> EventDecodingException -> ShowS
$cshow :: EventDecodingException -> FilePath
show :: EventDecodingException -> FilePath
$cshowList :: [EventDecodingException] -> ShowS
showList :: [EventDecodingException] -> ShowS
Show)

instance Exception EventDecodingException

data SQLiteLog
  = MigratingFromFileBased {SQLiteLog -> FilePath
legacyFile :: FilePath}
  | MigrationSkipped {legacyFile :: FilePath}
  | MigrationComplete {legacyFile :: FilePath}
  deriving stock (SQLiteLog -> SQLiteLog -> Bool
(SQLiteLog -> SQLiteLog -> Bool)
-> (SQLiteLog -> SQLiteLog -> Bool) -> Eq SQLiteLog
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: SQLiteLog -> SQLiteLog -> Bool
== :: SQLiteLog -> SQLiteLog -> Bool
$c/= :: SQLiteLog -> SQLiteLog -> Bool
/= :: SQLiteLog -> SQLiteLog -> Bool
Eq, Int -> SQLiteLog -> ShowS
[SQLiteLog] -> ShowS
SQLiteLog -> FilePath
(Int -> SQLiteLog -> ShowS)
-> (SQLiteLog -> FilePath)
-> ([SQLiteLog] -> ShowS)
-> Show SQLiteLog
forall a.
(Int -> a -> ShowS) -> (a -> FilePath) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> SQLiteLog -> ShowS
showsPrec :: Int -> SQLiteLog -> ShowS
$cshow :: SQLiteLog -> FilePath
show :: SQLiteLog -> FilePath
$cshowList :: [SQLiteLog] -> ShowS
showList :: [SQLiteLog] -> ShowS
Show, (forall x. SQLiteLog -> Rep SQLiteLog x)
-> (forall x. Rep SQLiteLog x -> SQLiteLog) -> Generic SQLiteLog
forall x. Rep SQLiteLog x -> SQLiteLog
forall x. SQLiteLog -> Rep SQLiteLog x
forall a.
(forall x. a -> Rep a x) -> (forall x. Rep a x -> a) -> Generic a
$cfrom :: forall x. SQLiteLog -> Rep SQLiteLog x
from :: forall x. SQLiteLog -> Rep SQLiteLog x
$cto :: forall x. Rep SQLiteLog x -> SQLiteLog
to :: forall x. Rep SQLiteLog x -> SQLiteLog
Generic)
  deriving anyclass ([SQLiteLog] -> Value
[SQLiteLog] -> Encoding
SQLiteLog -> Bool
SQLiteLog -> Value
SQLiteLog -> Encoding
(SQLiteLog -> Value)
-> (SQLiteLog -> Encoding)
-> ([SQLiteLog] -> Value)
-> ([SQLiteLog] -> Encoding)
-> (SQLiteLog -> Bool)
-> ToJSON SQLiteLog
forall a.
(a -> Value)
-> (a -> Encoding)
-> ([a] -> Value)
-> ([a] -> Encoding)
-> (a -> Bool)
-> ToJSON a
$ctoJSON :: SQLiteLog -> Value
toJSON :: SQLiteLog -> Value
$ctoEncoding :: SQLiteLog -> Encoding
toEncoding :: SQLiteLog -> Encoding
$ctoJSONList :: [SQLiteLog] -> Value
toJSONList :: [SQLiteLog] -> Value
$ctoEncodingList :: [SQLiteLog] -> Encoding
toEncodingList :: [SQLiteLog] -> Encoding
$comitField :: SQLiteLog -> Bool
omitField :: SQLiteLog -> Bool
ToJSON)

-- | Items in the write-behind queue: either an event to insert or a flush
-- marker that the writer thread signals after processing all preceding items.
--
-- Events are queued unencoded and CBOR-encoded on the writer thread: the
-- encoding of e.g. a SnapshotRequested event carrying a large UTxO otherwise
-- sits on the node loop between processing a ReqSn and broadcasting the
-- AckSn. The bounded queue briefly pins event values instead of compact
-- bytes, but the writer drains whole-queue batches so the window is short.
type WriteItem e = Either (TMVar IO ()) (Word64, e)

-- | Bracket-style wrapper around 'mkSQLiteEventStore'. Creates the database,
-- schema, and writer thread, runs the callback, then flushes queued writes and
-- cancels the writer thread on exit. The writer thread is 'link'ed so that
-- crashes surface immediately in the calling thread.
--
-- If a legacy state file exists at @legacyStateFile@, events are migrated into
-- SQLite automatically before the callback runs.
--
-- Flushing of the async write queue and reinitialisation of the last-seen
-- event id are handled internally: 'sourceEvents' auto-flushes before
-- reading, 'rotate' flushes before deleting, migration reinitialises the
-- event id TVar, and this bracket flushes on exit.
withSQLiteEventStore ::
  forall e a.
  (ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) =>
  Tracer IO SQLiteLog ->
  FilePath ->
  FilePath ->
  (EventStore e IO -> IO a) ->
  IO a
withSQLiteEventStore :: forall e a.
(ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) =>
Tracer IO SQLiteLog
-> FilePath -> FilePath -> (EventStore e IO -> IO a) -> IO a
withSQLiteEventStore Tracer IO SQLiteLog
tracer FilePath
dbFile FilePath
legacyStateFile EventStore e IO -> IO a
callback = do
  (Connection
conn, EventStore e IO
store, IO ()
flush, IO ()
reinitLastSeen, IO ()
cleanup) <- FilePath -> IO (Connection, EventStore e IO, IO (), IO (), IO ())
forall e.
(ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) =>
FilePath -> IO (Connection, EventStore e IO, IO (), IO (), IO ())
mkSQLiteEventStore FilePath
dbFile
  Proxy e
-> Tracer IO SQLiteLog -> FilePath -> Connection -> IO () -> IO ()
forall e.
(FromJSON e, ToCBOR e, HasEventId e) =>
Proxy e
-> Tracer IO SQLiteLog -> FilePath -> Connection -> IO () -> IO ()
migrateFromFileBased (forall t. Proxy t
forall {k} (t :: k). Proxy t
Proxy @e) Tracer IO SQLiteLog
tracer FilePath
legacyStateFile Connection
conn IO ()
reinitLastSeen
  EventStore e IO -> IO a
callback EventStore e IO
store
    IO a -> IO () -> IO a
forall a b. IO a -> IO b -> IO a
forall (m :: * -> *) a b. MonadThrow m => m a -> m b -> m a
`finally` (IO ()
flush IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> IO ()
cleanup IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Connection -> IO ()
close Connection
conn)

-- | Create an 'EventStore' backed by a SQLite database at the given file path.
-- The database and schema are created on first use if they do not exist.
-- Returns @(conn, store, flush, reinitLastSeen, cleanup)@. Internal —
-- prefer 'withSQLiteEventStore' which handles cleanup, migration, and flushing
-- automatically.
mkSQLiteEventStore ::
  forall e.
  (ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) =>
  FilePath ->
  IO (Connection, EventStore e IO, IO (), IO (), IO ())
mkSQLiteEventStore :: forall e.
(ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) =>
FilePath -> IO (Connection, EventStore e IO, IO (), IO (), IO ())
mkSQLiteEventStore FilePath
dbFile = do
  Bool -> FilePath -> IO ()
createDirectoryIfMissing Bool
True (ShowS
takeDirectory FilePath
dbFile)
  Connection
conn <- FilePath -> IO Connection
open FilePath
dbFile
  -- Rows of a version 1 database are JSON-encoded and get re-encoded to CBOR
  -- by the schema migration; a row that fails to decode aborts startup.
  let reencodeRow :: Word64 -> ByteString -> IO ByteString
      reencodeRow :: Word64 -> ByteString -> IO ByteString
reencodeRow Word64
eid ByteString
bytes =
        case forall a. FromJSON a => ByteString -> Either FilePath a
Aeson.eitherDecodeStrict' @e ByteString
bytes of
          Right e
evt -> ByteString -> IO ByteString
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ByteString -> IO ByteString) -> ByteString -> IO ByteString
forall a b. (a -> b) -> a -> b
$ e -> ByteString
forall a. ToCBOR a => a -> ByteString
serialize' e
evt
          Left FilePath
err -> EventDecodingException -> IO ByteString
forall e a. Exception e => e -> IO a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO EventDecodingException{$sel:eventId:EventDecodingException :: Word64
eventId = Word64
eid, $sel:decodeError:EventDecodingException :: FilePath
decodeError = FilePath
err}
  Connection -> (Word64 -> ByteString -> IO ByteString) -> IO ()
initSchema Connection
conn Word64 -> ByteString -> IO ByteString
reencodeRow
  -- Dedicated connection for 'sourceEvents' streams, so concurrent client
  -- history replay cannot hold statements open on the connection rotation
  -- runs VACUUM INTO on (see module header).
  Connection
readConn <- FilePath -> IO Connection
open FilePath
dbFile
  Connection -> IO ()
configurePragmas Connection
readConn
  TVar (Maybe Word64)
eventIdV <- FilePath -> Maybe Word64 -> IO (TVar IO (Maybe Word64))
forall (m :: * -> *) a.
MonadLabelledSTM m =>
FilePath -> a -> m (TVar m a)
newLabelledTVarIO FilePath
"sqlite-event-store-event-id" Maybe Word64
forall a. Maybe a
Nothing
  -- Initialise last-seen event id from existing rows.
  [Only Word64]
rows <- Connection -> IO [Only Word64]
selectLastEventId Connection
conn
  case [Only Word64]
rows of
    [Only Word64
lastId] -> STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar IO (Maybe Word64) -> Maybe Word64 -> STM IO ()
forall a. TVar IO a -> a -> STM IO ()
forall (m :: * -> *) a. MonadSTM m => TVar m a -> a -> STM m ()
writeTVar TVar (Maybe Word64)
TVar IO (Maybe Word64)
eventIdV (Word64 -> Maybe Word64
forall a. a -> Maybe a
Just Word64
lastId)
    [Only Word64]
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()

  TBQueue (Either (TMVar ()) (Word64, e))
writeQueue <- Natural -> IO (TBQueue IO (Either (TMVar ()) (Word64, e)))
forall a. Natural -> IO (TBQueue IO a)
forall (m :: * -> *) a. MonadSTM m => Natural -> m (TBQueue m a)
newTBQueueIO Natural
1000
  Async ()
writerThread <- IO () -> IO (Async IO ())
forall a. IO a -> IO (Async IO a)
forall (m :: * -> *) a. MonadAsync m => m a -> m (Async m a)
async (IO () -> IO (Async IO ())) -> IO () -> IO (Async IO ())
forall a b. (a -> b) -> a -> b
$ Connection -> TBQueue IO (WriteItem e) -> IO ()
forall e.
ToCBOR e =>
Connection -> TBQueue IO (WriteItem e) -> IO ()
writerLoop Connection
conn TBQueue (Either (TMVar ()) (Word64, e))
TBQueue IO (WriteItem e)
writeQueue
  Async IO () -> IO ()
forall (m :: * -> *) a.
(MonadAsync m, MonadFork m, MonadMask m) =>
Async m a -> m ()
link Async IO ()
Async ()
writerThread
  let
    getLastSeenEventId :: STM IO (Maybe Word64)
getLastSeenEventId = TVar IO (Maybe Word64) -> STM IO (Maybe Word64)
forall a. TVar IO a -> STM IO a
forall (m :: * -> *) a. MonadSTM m => TVar m a -> STM m a
readTVar TVar (Maybe Word64)
TVar IO (Maybe Word64)
eventIdV

    setLastSeenEventId :: e -> STM IO ()
setLastSeenEventId e
evt =
      TVar IO (Maybe Word64) -> Maybe Word64 -> STM IO ()
forall a. TVar IO a -> a -> STM IO ()
forall (m :: * -> *) a. MonadSTM m => TVar m a -> a -> STM m ()
writeTVar TVar (Maybe Word64)
TVar IO (Maybe Word64)
eventIdV (Word64 -> Maybe Word64
forall a. a -> Maybe a
Just (Word64 -> Maybe Word64) -> Word64 -> Maybe Word64
forall a b. (a -> b) -> a -> b
$ e -> Word64
forall a. HasEventId a => a -> Word64
getEventId e
evt)

    decodeRow :: (Word64, ByteString) -> IO e
    decodeRow :: (Word64, ByteString) -> IO e
decodeRow (Word64
eid, ByteString
evData) =
      case ByteString -> Either DecoderError e
forall a. FromCBOR a => ByteString -> Either DecoderError a
decodeFull' ByteString
evData of
        Right e
evt -> e -> IO e
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure e
evt
        -- NOTE: This will prevent the node from starting, which is intentional —
        -- starting with missing events would silently corrupt the head state.
        Left DecoderError
err -> EventDecodingException -> IO e
forall e a. Exception e => e -> IO a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO EventDecodingException{$sel:eventId:EventDecodingException :: Word64
eventId = Word64
eid, $sel:decodeError:EventDecodingException :: FilePath
decodeError = DecoderError -> FilePath
forall b a. (Show a, IsString b) => a -> b
show DecoderError
err}

    sourceEvents :: ConduitT () e (ResourceT IO) ()
    sourceEvents :: ConduitT () e (ResourceT IO) ()
sourceEvents = do
      -- Flush queued writes so reads see all enqueued events.
      IO () -> ConduitT () e (ResourceT IO) ()
forall a. IO a -> ConduitT () e (ResourceT IO) a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> ConduitT () e (ResourceT IO) ())
-> IO () -> ConduitT () e (ResourceT IO) ()
forall a b. (a -> b) -> a -> b
$ TBQueue IO (WriteItem e) -> IO ()
forall e. TBQueue IO (WriteItem e) -> IO ()
flushWriteQueue TBQueue (Either (TMVar ()) (Word64, e))
TBQueue IO (WriteItem e)
writeQueue
      IO Statement
-> (Statement -> IO ())
-> (Statement -> ConduitT () e (ResourceT IO) ())
-> ConduitT () e (ResourceT IO) ()
forall (m :: * -> *) a i o r.
MonadResource m =>
IO a -> (a -> IO ()) -> (a -> ConduitT i o m r) -> ConduitT i o m r
bracketP IO Statement
openStmt Statement -> IO ()
closeStatement Statement -> ConduitT () e (ResourceT IO) ()
yieldRows
     where
      openStmt :: IO Statement
      openStmt :: IO Statement
openStmt = Connection -> IO Statement
getEventsASC Connection
readConn

      yieldRows :: Statement -> ConduitT () e (ResourceT IO) ()
      yieldRows :: Statement -> ConduitT () e (ResourceT IO) ()
yieldRows Statement
stmt = do
        Maybe (Word64, ByteString)
mRow <- IO (Maybe (Word64, ByteString))
-> ConduitT () e (ResourceT IO) (Maybe (Word64, ByteString))
forall a. IO a -> ConduitT () e (ResourceT IO) a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (Statement -> IO (Maybe (Word64, ByteString))
forall r. FromRow r => Statement -> IO (Maybe r)
nextRow Statement
stmt)
        case Maybe (Word64, ByteString)
mRow of
          Maybe (Word64, ByteString)
Nothing -> () -> ConduitT () e (ResourceT IO) ()
forall a. a -> ConduitT () e (ResourceT IO) a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
          Just (Word64, ByteString)
row -> do
            e
evt <- IO e -> ConduitT () e (ResourceT IO) e
forall a. IO a -> ConduitT () e (ResourceT IO) a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO ((Word64, ByteString) -> IO e
decodeRow (Word64, ByteString)
row)
            e -> ConduitT () e (ResourceT IO) ()
forall (m :: * -> *) o i. Monad m => o -> ConduitT i o m ()
yield e
evt
            Statement -> ConduitT () e (ResourceT IO) ()
yieldRows Statement
stmt

    enqueueEvent :: e -> IO ()
enqueueEvent e
evt =
      STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
        TBQueue IO (Either (TMVar ()) (Word64, e))
-> Either (TMVar ()) (Word64, e) -> STM IO ()
forall a. TBQueue IO a -> a -> STM IO ()
forall (m :: * -> *) a. MonadSTM m => TBQueue m a -> a -> STM m ()
writeTBQueue TBQueue (Either (TMVar ()) (Word64, e))
TBQueue IO (Either (TMVar ()) (Word64, e))
writeQueue ((Word64, e) -> Either (TMVar ()) (Word64, e)
forall a b. b -> Either a b
Right (e -> Word64
forall a. HasEventId a => a -> Word64
getEventId e
evt, e
evt))
        e -> STM IO ()
setLastSeenEventId e
evt

    putEvent :: e -> IO ()
putEvent e
evt =
      STM IO (Maybe Word64) -> IO (Maybe Word64)
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically STM IO (Maybe Word64)
getLastSeenEventId IO (Maybe Word64) -> (Maybe Word64 -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
        Maybe Word64
Nothing -> e -> IO ()
enqueueEvent e
evt
        Just Word64
lastSeenEventId
          | e -> Word64
forall a. HasEventId a => a -> Word64
getEventId e
evt Word64 -> Word64 -> Bool
forall a. Ord a => a -> a -> Bool
> Word64
lastSeenEventId -> e -> IO ()
enqueueEvent e
evt
          | Bool
otherwise -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()

    putEvents :: [e] -> IO ()
putEvents [e]
evts = do
      Maybe Word64
lastSeen <- STM IO (Maybe Word64) -> IO (Maybe Word64)
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically STM IO (Maybe Word64)
getLastSeenEventId
      let newEvts :: [e]
newEvts = case Maybe Word64
lastSeen of
            Maybe Word64
Nothing -> [e]
evts
            Just Word64
lastId -> (e -> Bool) -> [e] -> [e]
forall a. (a -> Bool) -> [a] -> [a]
filter (\e
e -> e -> Word64
forall a. HasEventId a => a -> Word64
getEventId e
e Word64 -> Word64 -> Bool
forall a. Ord a => a -> a -> Bool
> Word64
lastId) [e]
evts
      Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless ([e] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [e]
newEvts) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
        STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          [e] -> (e -> STM ()) -> STM ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [e]
newEvts ((e -> STM ()) -> STM ()) -> (e -> STM ()) -> STM ()
forall a b. (a -> b) -> a -> b
$ \e
evt -> TBQueue IO (Either (TMVar ()) (Word64, e))
-> Either (TMVar ()) (Word64, e) -> STM IO ()
forall a. TBQueue IO a -> a -> STM IO ()
forall (m :: * -> *) a. MonadSTM m => TBQueue m a -> a -> STM m ()
writeTBQueue TBQueue (Either (TMVar ()) (Word64, e))
TBQueue IO (Either (TMVar ()) (Word64, e))
writeQueue ((Word64, e) -> Either (TMVar ()) (Word64, e)
forall a b. b -> Either a b
Right (e -> Word64
forall a. HasEventId a => a -> Word64
getEventId e
evt, e
evt))
          case [e] -> Maybe (NonEmpty e)
forall a. [a] -> Maybe (NonEmpty a)
nonEmpty [e]
newEvts of
            Just NonEmpty e
ne -> e -> STM IO ()
setLastSeenEventId (NonEmpty e -> e
forall (f :: * -> *) a. IsNonEmpty f a a "last" => f a -> a
last NonEmpty e
ne)
            Maybe (NonEmpty e)
Nothing -> () -> STM ()
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()

    rotate :: Word64 -> e -> IO ()
rotate Word64
logId e
checkpointEvent = do
      TBQueue IO (WriteItem e) -> IO ()
forall e. TBQueue IO (WriteItem e) -> IO ()
flushWriteQueue TBQueue (Either (TMVar ()) (Word64, e))
TBQueue IO (WriteItem e)
writeQueue
      -- Archive the current database before removing events, so the
      -- pre-rotation log is retained (mirrors the old file-based backup).
      Connection -> FilePath -> Word64 -> IO ()
backupDatabase Connection
conn FilePath
dbFile Word64
logId
      let evData :: ByteString
evData = e -> ByteString
forall a. ToCBOR a => a -> ByteString
serialize' e
checkpointEvent
      Connection -> IO () -> IO ()
forall a. Connection -> IO a -> IO a
withTransaction Connection
conn (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
        Connection -> IO ()
deleteAllEvents Connection
conn
        Connection -> (Word64, ByteString) -> IO ()
insertEvent Connection
conn (e -> Word64
forall a. HasEventId a => a -> Word64
getEventId e
checkpointEvent, ByteString
evData)
      STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ e -> STM IO ()
setLastSeenEventId e
checkpointEvent

  let reinitLastSeen :: IO ()
reinitLastSeen = do
        [Only Word64]
latestRows <- Connection -> IO [Only Word64]
selectLastEventId Connection
conn
        case [Only Word64]
latestRows of
          [Only Word64
lastId] -> STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar IO (Maybe Word64) -> Maybe Word64 -> STM IO ()
forall a. TVar IO a -> a -> STM IO ()
forall (m :: * -> *) a. MonadSTM m => TVar m a -> a -> STM m ()
writeTVar TVar (Maybe Word64)
TVar IO (Maybe Word64)
eventIdV (Word64 -> Maybe Word64
forall a. a -> Maybe a
Just Word64
lastId)
          [Only Word64]
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()

  (Connection, EventStore e IO, IO (), IO (), IO ())
-> IO (Connection, EventStore e IO, IO (), IO (), IO ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
    ( Connection
conn
    , EventStore
        { $sel:eventSource:EventStore :: EventSource e IO
eventSource = EventSource{ConduitT () e (ResourceT IO) ()
HasEventId e => ConduitT () e (ResourceT IO) ()
sourceEvents :: HasEventId e => ConduitT () e (ResourceT IO) ()
sourceEvents :: ConduitT () e (ResourceT IO) ()
sourceEvents}
        , $sel:eventSink:EventStore :: EventSink e IO
eventSink = EventSink{e -> IO ()
HasEventId e => e -> IO ()
putEvent :: HasEventId e => e -> IO ()
putEvent :: e -> IO ()
putEvent, [e] -> IO ()
HasEventId e => [e] -> IO ()
putEvents :: HasEventId e => [e] -> IO ()
putEvents :: [e] -> IO ()
putEvents}
        , Word64 -> e -> IO ()
rotate :: Word64 -> e -> IO ()
rotate :: Word64 -> e -> IO ()
rotate
        }
    , TBQueue IO (WriteItem e) -> IO ()
forall e. TBQueue IO (WriteItem e) -> IO ()
flushWriteQueue TBQueue (Either (TMVar ()) (Word64, e))
TBQueue IO (WriteItem e)
writeQueue
    , IO ()
reinitLastSeen
    , Async IO () -> IO ()
forall a. Async IO a -> IO ()
forall (m :: * -> *) a. MonadAsync m => Async m a -> m ()
cancel Async IO ()
Async ()
writerThread IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Connection -> IO ()
close Connection
readConn
    )

-- | Background writer that drains the queue and batch-inserts into SQLite.
-- Each iteration blocks for at least one item, then flushes everything
-- available. Events are CBOR-encoded here, off the caller's thread, then
-- batch-inserted in a single transaction, and any flush markers in the batch
-- are signalled. Encode errors surface as writer thread crashes, which are
-- 'link'ed to the node.
writerLoop :: ToCBOR e => Connection -> TBQueue IO (WriteItem e) -> IO ()
writerLoop :: forall e.
ToCBOR e =>
Connection -> TBQueue IO (WriteItem e) -> IO ()
writerLoop Connection
conn TBQueue IO (WriteItem e)
queue = IO () -> IO ()
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
  Either (TMVar ()) (Word64, e)
first' <- STM IO (Either (TMVar ()) (Word64, e))
-> IO (Either (TMVar ()) (Word64, e))
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO (Either (TMVar ()) (Word64, e))
 -> IO (Either (TMVar ()) (Word64, e)))
-> STM IO (Either (TMVar ()) (Word64, e))
-> IO (Either (TMVar ()) (Word64, e))
forall a b. (a -> b) -> a -> b
$ TBQueue IO (Either (TMVar ()) (Word64, e))
-> STM IO (Either (TMVar ()) (Word64, e))
forall a. TBQueue IO a -> STM IO a
forall (m :: * -> *) a. MonadSTM m => TBQueue m a -> STM m a
readTBQueue TBQueue IO (Either (TMVar ()) (Word64, e))
TBQueue IO (WriteItem e)
queue
  [Either (TMVar ()) (Word64, e)]
rest <- STM IO [Either (TMVar ()) (Word64, e)]
-> IO [Either (TMVar ()) (Word64, e)]
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO [Either (TMVar ()) (Word64, e)]
 -> IO [Either (TMVar ()) (Word64, e)])
-> STM IO [Either (TMVar ()) (Word64, e)]
-> IO [Either (TMVar ()) (Word64, e)]
forall a b. (a -> b) -> a -> b
$ TBQueue IO (Either (TMVar ()) (Word64, e))
-> STM IO [Either (TMVar ()) (Word64, e)]
forall a. TBQueue IO a -> STM IO [a]
forall (m :: * -> *) a. MonadSTM m => TBQueue m a -> STM m [a]
flushTBQueue TBQueue IO (Either (TMVar ()) (Word64, e))
TBQueue IO (WriteItem e)
queue
  let allItems :: [Either (TMVar ()) (Word64, e)]
allItems = Either (TMVar ()) (Word64, e)
first' Either (TMVar ()) (Word64, e)
-> [Either (TMVar ()) (Word64, e)]
-> [Either (TMVar ()) (Word64, e)]
forall a. a -> [a] -> [a]
: [Either (TMVar ()) (Word64, e)]
rest
      ([TMVar ()]
flushSignals, [(Word64, e)]
events) = [Either (TMVar ()) (Word64, e)] -> ([TMVar ()], [(Word64, e)])
forall a b. [Either a b] -> ([a], [b])
partitionEithers [Either (TMVar ()) (Word64, e)]
allItems
      eventRows :: [(Word64, ByteString)]
eventRows = ((Word64, e) -> (Word64, ByteString))
-> [(Word64, e)] -> [(Word64, ByteString)]
forall a b. (a -> b) -> [a] -> [b]
map ((e -> ByteString) -> (Word64, e) -> (Word64, ByteString)
forall b c a. (b -> c) -> (a, b) -> (a, c)
forall (p :: * -> * -> *) b c a.
Bifunctor p =>
(b -> c) -> p a b -> p a c
second e -> ByteString
forall a. ToCBOR a => a -> ByteString
serialize') [(Word64, e)]
events
  Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless ([(Word64, ByteString)] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [(Word64, ByteString)]
eventRows) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
    Connection -> IO () -> IO ()
forall a. Connection -> IO a -> IO a
withTransaction Connection
conn (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
      Connection -> [(Word64, ByteString)] -> IO ()
insertEvents Connection
conn [(Word64, ByteString)]
eventRows
  [TMVar ()] -> (TMVar () -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [TMVar ()]
flushSignals ((TMVar () -> IO ()) -> IO ()) -> (TMVar () -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \TMVar ()
mv -> STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ TMVar IO () -> () -> STM IO ()
forall a. TMVar IO a -> a -> STM IO ()
forall (m :: * -> *) a. MonadSTM m => TMVar m a -> a -> STM m ()
putTMVar TMVar ()
TMVar IO ()
mv ()

-- | Block until all items currently in the write queue have been flushed to
-- SQLite. Sends a flush marker through the queue and waits for the writer thread
-- to signal completion.
flushWriteQueue :: TBQueue IO (WriteItem e) -> IO ()
flushWriteQueue :: forall e. TBQueue IO (WriteItem e) -> IO ()
flushWriteQueue TBQueue IO (WriteItem e)
queue = do
  TMVar ()
mv <- IO (TMVar ())
IO (TMVar IO ())
forall a. IO (TMVar IO a)
forall (m :: * -> *) a. MonadSTM m => m (TMVar m a)
newEmptyTMVarIO
  STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ TBQueue IO (Either (TMVar ()) (Word64, e))
-> Either (TMVar ()) (Word64, e) -> STM IO ()
forall a. TBQueue IO a -> a -> STM IO ()
forall (m :: * -> *) a. MonadSTM m => TBQueue m a -> a -> STM m ()
writeTBQueue TBQueue IO (Either (TMVar ()) (Word64, e))
TBQueue IO (WriteItem e)
queue (TMVar () -> Either (TMVar ()) (Word64, e)
forall a b. a -> Either a b
Left TMVar ()
mv)
  STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ TMVar IO () -> STM IO ()
forall a. TMVar IO a -> STM IO a
forall (m :: * -> *) a. MonadSTM m => TMVar m a -> STM m a
takeTMVar TMVar ()
TMVar IO ()
mv

-- | Migrate events from a legacy newline-delimited JSON file into SQLite.
-- Writes directly to the database, bypassing the async write queue (migration
-- runs at startup before the node processes inputs). After inserting, calls
-- @reinitLastSeen@ to sync the in-memory event id TVar with the database.
--
-- Safe to call when the legacy file does not exist (no-op). Not safe to re-run:
-- duplicate event ids will cause a primary key constraint violation.
--
-- On success the legacy file is renamed to @<path>.migrated@ so that
-- subsequent node restarts skip the migration step automatically.
migrateFromFileBased ::
  forall e.
  (FromJSON e, ToCBOR e, HasEventId e) =>
  Proxy e ->
  Tracer IO SQLiteLog ->
  FilePath ->
  Connection ->
  IO () ->
  IO ()
migrateFromFileBased :: forall e.
(FromJSON e, ToCBOR e, HasEventId e) =>
Proxy e
-> Tracer IO SQLiteLog -> FilePath -> Connection -> IO () -> IO ()
migrateFromFileBased Proxy e
_proxy Tracer IO SQLiteLog
tracer FilePath
legacyFile Connection
conn IO ()
reinitLastSeen = do
  Bool
exists <- FilePath -> IO Bool
doesFileExist FilePath
legacyFile
  if Bool -> Bool
not Bool
exists
    then Tracer IO SQLiteLog -> SQLiteLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO SQLiteLog
tracer MigrationSkipped{FilePath
$sel:legacyFile:MigratingFromFileBased :: FilePath
legacyFile :: FilePath
legacyFile}
    else do
      Tracer IO SQLiteLog -> SQLiteLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO SQLiteLog
tracer MigratingFromFileBased{FilePath
$sel:legacyFile:MigratingFromFileBased :: FilePath
legacyFile :: FilePath
legacyFile}
      [ByteString]
rawLines <-
        ConduitT () Void (ResourceT IO) [ByteString] -> IO [ByteString]
forall (m :: * -> *) r.
MonadUnliftIO m =>
ConduitT () Void (ResourceT m) r -> m r
runConduitRes (ConduitT () Void (ResourceT IO) [ByteString] -> IO [ByteString])
-> ConduitT () Void (ResourceT IO) [ByteString] -> IO [ByteString]
forall a b. (a -> b) -> a -> b
$
          FilePath -> ConduitT () ByteString (ResourceT IO) ()
forall (m :: * -> *) i.
MonadResource m =>
FilePath -> ConduitT i ByteString m ()
sourceFile FilePath
legacyFile
            ConduitT () ByteString (ResourceT IO) ()
-> ConduitT ByteString Void (ResourceT IO) [ByteString]
-> ConduitT () Void (ResourceT IO) [ByteString]
forall (m :: * -> *) a b c r.
Monad m =>
ConduitT a b m () -> ConduitT b c m r -> ConduitT a c m r
.| ConduitT ByteString ByteString (ResourceT IO) ()
forall (m :: * -> *) seq.
(Monad m, IsSequence seq, Element seq ~ Word8) =>
ConduitT seq seq m ()
linesUnboundedAscii
            ConduitT ByteString ByteString (ResourceT IO) ()
-> ConduitT ByteString Void (ResourceT IO) [ByteString]
-> ConduitT ByteString Void (ResourceT IO) [ByteString]
forall (m :: * -> *) a b c r.
Monad m =>
ConduitT a b m () -> ConduitT b c m r -> ConduitT a c m r
.| (ByteString -> Bool)
-> ConduitT ByteString ByteString (ResourceT IO) ()
forall (m :: * -> *) a. Monad m => (a -> Bool) -> ConduitT a a m ()
C.filter (Bool -> Bool
not (Bool -> Bool) -> (ByteString -> Bool) -> ByteString -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ByteString -> Bool
BS.null)
            ConduitT ByteString ByteString (ResourceT IO) ()
-> ConduitT ByteString Void (ResourceT IO) [ByteString]
-> ConduitT ByteString Void (ResourceT IO) [ByteString]
forall (m :: * -> *) a b c r.
Monad m =>
ConduitT a b m () -> ConduitT b c m r -> ConduitT a c m r
.| ConduitT ByteString Void (ResourceT IO) [ByteString]
forall (m :: * -> *) a o. Monad m => ConduitT a o m [a]
C.sinkList
      -- Decode each JSON line (legacy files are always JSON) and store the
      -- event re-encoded as CBOR. Invalid JSON is caught here so corrupt
      -- files fail at migration.
      [(Word64, ByteString)]
rowParams <- [(Int, ByteString)]
-> ((Int, ByteString) -> IO (Word64, ByteString))
-> IO [(Word64, ByteString)]
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
t a -> (a -> m b) -> m (t b)
forM ([Int] -> [ByteString] -> [(Int, ByteString)]
forall a b. [a] -> [b] -> [(a, b)]
zip [Int
1 ..] [ByteString]
rawLines) (((Int, ByteString) -> IO (Word64, ByteString))
 -> IO [(Word64, ByteString)])
-> ((Int, ByteString) -> IO (Word64, ByteString))
-> IO [(Word64, ByteString)]
forall a b. (a -> b) -> a -> b
$ \(Int
lineNo :: Int, ByteString
line) ->
        case forall a. FromJSON a => ByteString -> Either FilePath a
Aeson.eitherDecodeStrict' @e ByteString
line of
          Right e
evt -> (Word64, ByteString) -> IO (Word64, ByteString)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (e -> Word64
forall a. HasEventId a => a -> Word64
getEventId e
evt, e -> ByteString
forall a. ToCBOR a => a -> ByteString
serialize' e
evt)
          Left FilePath
err -> EventDecodingException -> IO (Word64, ByteString)
forall e a. Exception e => e -> IO a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO EventDecodingException{$sel:eventId:EventDecodingException :: Word64
eventId = Int -> Word64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
lineNo, $sel:decodeError:EventDecodingException :: FilePath
decodeError = FilePath
err}
      Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless ([(Word64, ByteString)] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [(Word64, ByteString)]
rowParams) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
        Connection -> IO () -> IO ()
forall a. Connection -> IO a -> IO a
withTransaction Connection
conn (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
          Connection -> [(Word64, ByteString)] -> IO ()
insertEvents Connection
conn [(Word64, ByteString)]
rowParams
      -- Re-read the last event id from the database so the in-memory
      -- de-duplication TVar is consistent with the migrated rows.
      IO ()
reinitLastSeen
      FilePath -> FilePath -> IO ()
renameFile FilePath
legacyFile (FilePath
legacyFile FilePath -> ShowS
forall a. Semigroup a => a -> a -> a
<> FilePath
".migrated")
      Tracer IO SQLiteLog -> SQLiteLog -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO SQLiteLog
tracer MigrationComplete{FilePath
$sel:legacyFile:MigratingFromFileBased :: FilePath
legacyFile :: FilePath
legacyFile}

-- Internal

-- | Current schema version. Bump this and add a migration step to
-- 'migrateStep' whenever the schema changes.
nextVersion :: Int
nextVersion :: Int
nextVersion = Int
2

-- | Re-encode a single event row given its event id and stored bytes, used by
-- the version 1 (JSON) to version 2 (CBOR) migration. Must throw when the row
-- cannot be decoded.
type ReencodeRow = Word64 -> ByteString -> IO ByteString

-- | Initialise connection pragmas, then create or migrate the schema to
-- 'nextVersion' using SQLite's built-in @user_version@ pragma.
initSchema :: Connection -> ReencodeRow -> IO ()
initSchema :: Connection -> (Word64 -> ByteString -> IO ByteString) -> IO ()
initSchema Connection
conn Word64 -> ByteString -> IO ByteString
reencodeRow = do
  Connection -> IO ()
configurePragmas Connection
conn
  Int
v <- Connection -> IO Int
getSchemaVersion Connection
conn
  Connection
-> (Word64 -> ByteString -> IO ByteString) -> Int -> IO ()
applyMigrations Connection
conn Word64 -> ByteString -> IO ByteString
reencodeRow Int
v
  -- Reclaim the space freed by the JSON -> CBOR re-encode. VACUUM cannot run
  -- inside a transaction and is a space optimization only: a crash between
  -- the migration commit and here costs disk space, not correctness.
  Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
v Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
1) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ Connection -> Query -> IO ()
execute_ Connection
conn Query
"VACUUM"

configurePragmas :: Connection -> IO ()
configurePragmas :: Connection -> IO ()
configurePragmas Connection
conn =
  (Query -> IO ()) -> [Query] -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_
    (Connection -> Query -> IO ()
execute_ Connection
conn)
    [ Query
"PRAGMA journal_mode=WAL"
    , Query
"PRAGMA busy_timeout=5000"
    , -- With WAL, NORMAL skips per-write fsyncs and only syncs during
      -- checkpoints — safe because the chain is the source of truth.
      Query
"PRAGMA synchronous=NORMAL"
    , Query
"PRAGMA cache_size=-65536" -- 64 MB page cache
    , Query
"PRAGMA temp_store=MEMORY"
    ]

-- | Read the schema version from @PRAGMA user_version@ (0 for a fresh DB).
getSchemaVersion :: Connection -> IO Int
getSchemaVersion :: Connection -> IO Int
getSchemaVersion Connection
conn = do
  [[Int
v]] <- Connection -> Query -> IO [[Int]]
forall r. FromRow r => Connection -> Query -> IO [r]
query_ Connection
conn Query
"PRAGMA user_version"
  Int -> IO Int
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Int
v

setSchemaVersion :: Connection -> Int -> IO ()
setSchemaVersion :: Connection -> Int -> IO ()
setSchemaVersion Connection
conn Int
v =
  -- PRAGMA doesn't support parameter binding, so we use show directly.
  -- The value is an Int we control, not user input.
  Connection -> Query -> IO ()
execute_ Connection
conn (Query -> IO ()) -> Query -> IO ()
forall a b. (a -> b) -> a -> b
$ FilePath -> Query
forall a. IsString a => FilePath -> a
fromString (FilePath -> Query) -> FilePath -> Query
forall a b. (a -> b) -> a -> b
$ FilePath
"PRAGMA user_version = " FilePath -> ShowS
forall a. Semigroup a => a -> a -> a
<> Int -> FilePath
forall b a. (Show a, IsString b) => a -> b
show Int
v

-- | Apply all pending migrations from version @v@ up to 'nextVersion'.
-- Each step runs together with its version bump in one transaction
-- (@PRAGMA user_version@ is transactional), so a crash or decoding failure
-- mid-migration rolls back to a well-defined version.
applyMigrations :: Connection -> ReencodeRow -> Int -> IO ()
applyMigrations :: Connection
-> (Word64 -> ByteString -> IO ByteString) -> Int -> IO ()
applyMigrations Connection
conn Word64 -> ByteString -> IO ByteString
reencodeRow Int
v
  | Int
v Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
nextVersion =
      Text -> IO ()
forall a t. (HasCallStack, IsText t) => t -> a
error (Text -> IO ()) -> Text -> IO ()
forall a b. (a -> b) -> a -> b
$ Text
"Database schema version " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Int -> Text
forall b a. (Show a, IsString b) => a -> b
show Int
v Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" is newer than supported " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Int -> Text
forall b a. (Show a, IsString b) => a -> b
show Int
nextVersion Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
", cannot downgrade"
  | Int
v Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
nextVersion = () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
  | Bool
otherwise = do
      Connection -> IO () -> IO ()
forall a. Connection -> IO a -> IO a
withTransaction Connection
conn (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
        Connection
-> (Word64 -> ByteString -> IO ByteString) -> Int -> IO ()
migrateStep Connection
conn Word64 -> ByteString -> IO ByteString
reencodeRow Int
v
        Connection -> Int -> IO ()
setSchemaVersion Connection
conn (Int
v Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1)
      Connection
-> (Word64 -> ByteString -> IO ByteString) -> Int -> IO ()
applyMigrations Connection
conn Word64 -> ByteString -> IO ByteString
reencodeRow (Int
v Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1)

-- | Individual migration steps. Pattern-match on the /source/ version.
migrateStep :: Connection -> ReencodeRow -> Int -> IO ()
migrateStep :: Connection
-> (Word64 -> ByteString -> IO ByteString) -> Int -> IO ()
migrateStep Connection
conn Word64 -> ByteString -> IO ByteString
reencodeRow = \case
  Int
0 -> Connection -> IO ()
createEventsTable Connection
conn
  Int
1 -> Connection -> (Word64 -> ByteString -> IO ByteString) -> IO ()
reencodeAllEvents Connection
conn Word64 -> ByteString -> IO ByteString
reencodeRow
  Int
unknown ->
    Text -> IO ()
forall a t. (HasCallStack, IsText t) => t -> a
error (Text -> IO ()) -> Text -> IO ()
forall a b. (a -> b) -> a -> b
$ Text
"Unknown schema version " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Int -> Text
forall b a. (Show a, IsString b) => a -> b
show Int
unknown Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
", cannot migrate"

-- | Re-encode all event rows using the given 'ReencodeRow' function (the
-- version 1 JSON to version 2 CBOR migration). Rows are processed in batches
-- of ascending event id so memory stays bounded for large databases. Runs
-- inside the caller's transaction.
reencodeAllEvents :: Connection -> ReencodeRow -> IO ()
reencodeAllEvents :: Connection -> (Word64 -> ByteString -> IO ByteString) -> IO ()
reencodeAllEvents Connection
conn Word64 -> ByteString -> IO ByteString
reencodeRow = Word64 -> IO ()
go Word64
0
 where
  batchSize :: Int
batchSize = Int
1000 :: Int

  go :: Word64 -> IO ()
  go :: Word64 -> IO ()
go Word64
startId = do
    [(Word64, ByteString)]
rows :: [(Word64, ByteString)] <-
      Connection -> Query -> (Word64, Int) -> IO [(Word64, ByteString)]
forall q r.
(ToRow q, FromRow r) =>
Connection -> Query -> q -> IO [r]
query Connection
conn Query
"SELECT event_id, event_data FROM events WHERE event_id >= ? ORDER BY event_id LIMIT ?" (Word64
startId, Int
batchSize)
    case [(Word64, ByteString)] -> Maybe (NonEmpty (Word64, ByteString))
forall a. [a] -> Maybe (NonEmpty a)
nonEmpty [(Word64, ByteString)]
rows of
      Maybe (NonEmpty (Word64, ByteString))
Nothing -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      Just NonEmpty (Word64, ByteString)
neRows -> do
        [(ByteString, Word64)]
updates <- [(Word64, ByteString)]
-> ((Word64, ByteString) -> IO (ByteString, Word64))
-> IO [(ByteString, Word64)]
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
t a -> (a -> m b) -> m (t b)
forM [(Word64, ByteString)]
rows (((Word64, ByteString) -> IO (ByteString, Word64))
 -> IO [(ByteString, Word64)])
-> ((Word64, ByteString) -> IO (ByteString, Word64))
-> IO [(ByteString, Word64)]
forall a b. (a -> b) -> a -> b
$ \(Word64
eid, ByteString
evData) -> do
          ByteString
encoded <- Word64 -> ByteString -> IO ByteString
reencodeRow Word64
eid ByteString
evData
          (ByteString, Word64) -> IO (ByteString, Word64)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ByteString
encoded, Word64
eid)
        Connection -> Query -> [(ByteString, Word64)] -> IO ()
forall q. ToRow q => Connection -> Query -> [q] -> IO ()
executeMany Connection
conn Query
"UPDATE events SET event_data = ? WHERE event_id = ?" [(ByteString, Word64)]
updates
        Word64 -> IO ()
go ((Word64, ByteString) -> Word64
forall a b. (a, b) -> a
fst (NonEmpty (Word64, ByteString) -> (Word64, ByteString)
forall (f :: * -> *) a. IsNonEmpty f a a "last" => f a -> a
last NonEmpty (Word64, ByteString)
neRows) Word64 -> Word64 -> Word64
forall a. Num a => a -> a -> a
+ Word64
1)

-- SQL queries

createEventsTable :: Connection -> IO ()
createEventsTable :: Connection -> IO ()
createEventsTable Connection
conn =
  Connection -> Query -> IO ()
execute_
    Connection
conn
    Query
"CREATE TABLE IF NOT EXISTS events \
    \(event_id INTEGER NOT NULL PRIMARY KEY, event_data BLOB NOT NULL)"

selectLastEventId :: Connection -> IO [Only Word64]
selectLastEventId :: Connection -> IO [Only Word64]
selectLastEventId Connection
conn =
  Connection -> Query -> IO [Only Word64]
forall r. FromRow r => Connection -> Query -> IO [r]
query_ Connection
conn Query
"SELECT event_id FROM events ORDER BY event_id DESC LIMIT 1"

getEventsASC :: Connection -> IO Statement
getEventsASC :: Connection -> IO Statement
getEventsASC Connection
conn =
  Connection -> Query -> IO Statement
openStatement Connection
conn Query
"SELECT event_id, event_data FROM events ORDER BY event_id ASC"

insertEvent :: Connection -> (Word64, ByteString) -> IO ()
insertEvent :: Connection -> (Word64, ByteString) -> IO ()
insertEvent Connection
conn =
  Connection -> Query -> (Word64, ByteString) -> IO ()
forall q. ToRow q => Connection -> Query -> q -> IO ()
execute Connection
conn Query
"INSERT INTO events (event_id, event_data) VALUES (?, ?)"

insertEvents :: Connection -> [(Word64, ByteString)] -> IO ()
insertEvents :: Connection -> [(Word64, ByteString)] -> IO ()
insertEvents Connection
conn =
  Connection -> Query -> [(Word64, ByteString)] -> IO ()
forall q. ToRow q => Connection -> Query -> [q] -> IO ()
executeMany Connection
conn Query
"INSERT INTO events (event_id, event_data) VALUES (?, ?)"

deleteAllEvents :: Connection -> IO ()
deleteAllEvents :: Connection -> IO ()
deleteAllEvents Connection
conn =
  Connection -> Query -> IO ()
execute_ Connection
conn Query
"DELETE FROM events"

-- | Archive the current database before rotation removes the events, into an
-- @old-state@ subdirectory next to the database, with the log id inserted
-- before the extension (e.g. @old-state/hydra-42.db@). Uses @VACUUM INTO@ so the
-- snapshot reflects all committed (WAL) data in a single self-contained file,
-- regardless of WAL checkpoint state. The destination is removed first if
-- present (e.g. a re-rotation at the same log id), since @VACUUM INTO@ requires
-- it not to exist.
backupDatabase :: Connection -> FilePath -> Word64 -> IO ()
backupDatabase :: Connection -> FilePath -> Word64 -> IO ()
backupDatabase Connection
conn FilePath
dbFile Word64
logId = do
  let backupDir :: FilePath
backupDir = ShowS
takeDirectory FilePath
dbFile FilePath -> ShowS
</> FilePath
"old-state"
      backupPath :: FilePath
backupPath = FilePath
backupDir FilePath -> ShowS
</> (ShowS
takeBaseName FilePath
dbFile FilePath -> ShowS
forall a. Semigroup a => a -> a -> a
<> FilePath
"-" FilePath -> ShowS
forall a. Semigroup a => a -> a -> a
<> Word64 -> FilePath
forall b a. (Show a, IsString b) => a -> b
show Word64
logId FilePath -> ShowS
forall a. Semigroup a => a -> a -> a
<> ShowS
takeExtension FilePath
dbFile)
  Bool -> FilePath -> IO ()
createDirectoryIfMissing Bool
True FilePath
backupDir
  IO Bool -> IO () -> IO ()
forall (m :: * -> *). Monad m => m Bool -> m () -> m ()
whenM (FilePath -> IO Bool
doesFileExist FilePath
backupPath) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ FilePath -> IO ()
removeFile FilePath
backupPath
  Connection -> Query -> Only FilePath -> IO ()
forall q. ToRow q => Connection -> Query -> q -> IO ()
execute Connection
conn Query
"VACUUM INTO ?" (FilePath -> Only FilePath
forall a. a -> Only a
Only FilePath
backupPath)