| Safe Haskell | Safe-Inferred |
|---|---|
| Language | GHC2021 |
Data.EventSource.SQLite
Description
A SQLite-backed event source and sink.
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 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
linked to the calling thread. If it dies (e.g. SQLite I/O error), the exception propagates immediately rather than leaving the caller silently stalled. UsewithSQLiteEventStorewhich 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 when an external source of truth can replay missed events.
- Rotation ordering:
rotateflushes the write queue synchronously, archives the current database toold-state/name-logId.dbviaVACUUM INTO, then performs DELETE + INSERT. This is safe because rotation is expected to be called only from a single-threaded processing loop, 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:
sourceEventsstreams over a dedicated connection. It can run concurrently on other threads, andVACUUM INTOfails 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.
Synopsis
- data EventDecodingException = EventDecodingException {
- eventId :: Word64
- decodeError :: String
- data SQLiteLog
- = MigratingFromFileBased { }
- | MigrationSkipped { }
- | MigrationComplete { }
- type WriteItem e = Either (TMVar IO ()) (Word64, e)
- withSQLiteEventStore :: forall e a. (ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) => Tracer IO SQLiteLog -> FilePath -> FilePath -> (EventStore e IO -> IO a) -> IO a
- mkSQLiteEventStore :: forall e. (ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) => FilePath -> IO (Connection, EventStore e IO, IO (), IO (), IO ())
- writerLoop :: ToCBOR e => Connection -> TBQueue IO (WriteItem e) -> IO ()
- flushWriteQueue :: TBQueue IO (WriteItem e) -> IO ()
- migrateFromFileBased :: forall e. (FromJSON e, ToCBOR e, HasEventId e) => Proxy e -> Tracer IO SQLiteLog -> FilePath -> Connection -> IO () -> IO ()
- nextVersion :: Int
- type ReencodeRow = Word64 -> ByteString -> IO ByteString
- initSchema :: Connection -> ReencodeRow -> IO ()
- configurePragmas :: Connection -> IO ()
- getSchemaVersion :: Connection -> IO Int
- setSchemaVersion :: Connection -> Int -> IO ()
- applyMigrations :: Connection -> ReencodeRow -> Int -> IO ()
- migrateStep :: Connection -> ReencodeRow -> Int -> IO ()
- reencodeAllEvents :: Connection -> ReencodeRow -> IO ()
- createEventsTable :: Connection -> IO ()
- selectLastEventId :: Connection -> IO [Only Word64]
- getEventsASC :: Connection -> IO Statement
- insertEvent :: Connection -> (Word64, ByteString) -> IO ()
- insertEvents :: Connection -> [(Word64, ByteString)] -> IO ()
- deleteAllEvents :: Connection -> IO ()
- backupDatabase :: Connection -> FilePath -> Word64 -> IO ()
Documentation
data EventDecodingException Source #
Exception thrown when a persisted event cannot be decoded.
Constructors
| EventDecodingException | |
Fields
| |
Instances
| Exception EventDecodingException Source # | |
Defined in Data.EventSource.SQLite | |
| Show EventDecodingException Source # | |
Defined in Data.EventSource.SQLite | |
Constructors
| MigratingFromFileBased | |
Fields | |
| MigrationSkipped | |
Fields | |
| MigrationComplete | |
Fields | |
Instances
| ToJSON SQLiteLog Source # | |
Defined in Data.EventSource.SQLite Methods toEncoding :: SQLiteLog -> Encoding toJSONList :: [SQLiteLog] -> Value toEncodingList :: [SQLiteLog] -> Encoding | |
| Generic SQLiteLog Source # | |
| Show SQLiteLog Source # | |
| Eq SQLiteLog Source # | |
| type Rep SQLiteLog Source # | |
Defined in Data.EventSource.SQLite type Rep SQLiteLog = D1 ('MetaData "SQLiteLog" "Data.EventSource.SQLite" "event-sourcing-0.1.0.0-H0AiciIiaHFEdrDzec4Tbt" 'False) (C1 ('MetaCons "MigratingFromFileBased" 'PrefixI 'True) (S1 ('MetaSel ('Just "legacyFile") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 FilePath)) :+: (C1 ('MetaCons "MigrationSkipped" 'PrefixI 'True) (S1 ('MetaSel ('Just "legacyFile") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 FilePath)) :+: C1 ('MetaCons "MigrationComplete" 'PrefixI 'True) (S1 ('MetaSel ('Just "legacyFile") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 FilePath)))) | |
type WriteItem e = Either (TMVar IO ()) (Word64, e) Source #
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 an event carrying a large payload otherwise sits on the caller's thread. The bounded queue briefly pins event values instead of compact bytes, but the writer drains whole-queue batches so the window is short.
withSQLiteEventStore :: forall e a. (ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) => Tracer IO SQLiteLog -> FilePath -> FilePath -> (EventStore e IO -> IO a) -> IO a Source #
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 linked 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.
mkSQLiteEventStore :: forall e. (ToCBOR e, FromCBOR e, FromJSON e, HasEventId e) => FilePath -> IO (Connection, EventStore e IO, IO (), IO (), IO ()) Source #
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.
writerLoop :: ToCBOR e => Connection -> TBQueue IO (WriteItem e) -> IO () Source #
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
linked to the caller.
flushWriteQueue :: TBQueue IO (WriteItem e) -> IO () Source #
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.
migrateFromFileBased :: forall e. (FromJSON e, ToCBOR e, HasEventId e) => Proxy e -> Tracer IO SQLiteLog -> FilePath -> Connection -> IO () -> IO () Source #
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 inputs are processed). 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 restarts skip the migration step automatically.
nextVersion :: Int Source #
Current schema version. Bump this and add a migration step to
migrateStep whenever the schema changes.
type ReencodeRow = Word64 -> ByteString -> IO ByteString Source #
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.
initSchema :: Connection -> ReencodeRow -> IO () Source #
Initialise connection pragmas, then create or migrate the schema to
nextVersion using SQLite's built-in user_version pragma.
configurePragmas :: Connection -> IO () Source #
getSchemaVersion :: Connection -> IO Int Source #
Read the schema version from PRAGMA user_version (0 for a fresh DB).
setSchemaVersion :: Connection -> Int -> IO () Source #
applyMigrations :: Connection -> ReencodeRow -> Int -> IO () Source #
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.
migrateStep :: Connection -> ReencodeRow -> Int -> IO () Source #
Individual migration steps. Pattern-match on the source version.
reencodeAllEvents :: Connection -> ReencodeRow -> IO () Source #
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.
createEventsTable :: Connection -> IO () Source #
selectLastEventId :: Connection -> IO [Only Word64] Source #
getEventsASC :: Connection -> IO Statement Source #
insertEvent :: Connection -> (Word64, ByteString) -> IO () Source #
insertEvents :: Connection -> [(Word64, ByteString)] -> IO () Source #
deleteAllEvents :: Connection -> IO () Source #
backupDatabase :: Connection -> FilePath -> Word64 -> IO () Source #
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/state-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.