Skip to content

refactor!: collapse postgres/sqlite repositories into a shared generic layer #111

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
patrickleet merged 1 commit into from
Jul 3, 2026
Merged

refactor!: collapse postgres/sqlite repositories into a shared generic layer #111

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,8 @@ default = []
emitter = ["dep:event-emitter-rs"]
http = ["dep:axum", "dep:reqwest", "dep:tokio"]
grpc = ["dep:tonic", "dep:tonic-prost", "dep:prost", "dep:tokio"]
postgres = ["dep:sqlx", "dep:tokio", "sqlx/postgres", "sqlx/runtime-tokio"]
sqlite = ["dep:sqlx", "dep:tokio", "sqlx/runtime-tokio", "sqlx/sqlite"]
postgres = ["dep:sqlx", "dep:tokio", "sqlx/postgres", "sqlx/migrate", "sqlx/runtime-tokio"]
sqlite = ["dep:sqlx", "dep:tokio", "sqlx/migrate", "sqlx/runtime-tokio", "sqlx/sqlite"]
nats = ["dep:async-nats", "dep:futures", "dep:tokio"]
rabbitmq = ["dep:lapin", "dep:futures", "dep:tokio"]
kafka = ["dep:rdkafka", "dep:tokio"]
Expand Down
12 changes: 6 additions & 6 deletions migrations/postgres/0001_initial.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,10 @@ CREATE TABLE IF NOT EXISTS aggregate_events (
aggregate_id text NOT NULL,
sequence bigint NOT NULL,
event_name text NOT NULL,
event_version integer NOT NULL DEFAULT 1,
event_version bigint NOT NULL DEFAULT 1,
payload bytea NOT NULL,
payload_codec text NOT NULL,
payload_codec_version integer NOT NULL,
payload_codec_version bigint NOT NULL,
Comment on lines +6 to +9

@coderabbitai coderabbitai Bot Jul 3, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🌐 Web query:

sqlx postgres try_get i64 on INT4 column type mismatch error

💡 Result:

The error occurs because sqlx strictly maps the PostgreSQL INT4 (INTEGER) type to the Rust i32 type, while i64 is exclusively reserved for the BIGINT (INT8) type [1][2][3]. When you attempt to decode an INT4 column into an i64 using try_get, sqlx detects a type mismatch between the database's reported type and the requested Rust type, resulting in a decoding error [4][5]. To resolve this issue, use one of the following approaches: 1. Use i32 in Rust: If the data in your database column is guaranteed to fit within a 32-bit integer, change your Rust code to use i32 instead of i64 [2][3]. 2. Explicitly cast in SQL: If you must use i64 in Rust, you can cast the column value in your SQL query to bigint. This promotes the type at the database level before it reaches sqlx [5]. For example: SELECT your_column::bigint FROM your_table; 3. Change the database schema: If the values in the column may exceed the range of an i32, consider altering your PostgreSQL table schema to use BIGINT instead of INT4/INTEGER [2][3].

Citations:

🏁 Script executed:

#!/bin/bash
set -euo pipefail
printf 'Repo root: '; pwd

# Find relevant migration and decoding code.
git ls-files 'migrations/postgres/*' 'src/*' 'crates/*' | sed -n '1,200p'

printf '\n--- search for widened columns and sqlx decoders ---\n'
rg -n "event_version|payload_codec_version|snapshot_version|attempts|try_get::<i64>|repository_u64_from_i64|repository_u16_from_i64|u32::try_from|Migrator|sqlx::migrate|CREATE TABLE IF NOT EXISTS" . --glob '!target' --glob '!node_modules' | sed -n '1,240p'

Repository: hops-ops/distributed

Length of output: 22782

🏁 Script executed:

#!/bin/bash
set -euo pipefail
# Inspect the initial migration and the migration runner.
for f in migrations/postgres/0001_initial.sql $(git ls-files | rg 'migrate|migration|sqlx'); do
  [ -f "$f" ] || continue
  printf '\n### %s (lines, if small enough)\n' "$f"
  wc -l "$f"
done

# If migration runner exists, show structure.
cand=$(rg -l "Migrator|migrate!|sqlx::migrate" . --glob '!target' --glob '!node_modules' | head -n 20)
printf '\nCandidates:\n%s\n' "$cand"

Repository: hops-ops/distributed

Length of output: 895

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Read the relevant SQL migration in a bounded way.
nl -ba migrations/postgres/0001_initial.sql | sed -n '1,180p'

Repository: hops-ops/distributed

Length of output: 198

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Probe for the migration implementation and runtime decoding paths.
rg -n "try_get::<i64>|repository_u64_from_i64|repository_u16_from_i64|u32::try_from|CREATE TABLE IF NOT EXISTS|ALTER TABLE .*ALTER COLUMN .*TYPE bigint|sqlx::migrate|Migrator" . --glob '!target' --glob '!node_modules' | sed -n '1,260p'

Repository: hops-ops/distributed

Length of output: 4525

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Show the initial migration and the migration runner file paths only.
printf 'Initial migration:\n'
nl -ba migrations/postgres/0001_initial.sql | sed -n '1,120p'

printf '\nMigration-related files:\n'
rg -l "sqlx::migrate|Migrator|migrate!" . --glob '!target' --glob '!node_modules' | sed -n '1,50p'

Repository: hops-ops/distributed

Length of output: 217

🏁 Script executed:

#!/bin/bash
set -euo pipefail

printf 'Postgres migration files:\n'
git ls-files 'migrations/postgres/*' | sort

printf '\n--- migrations/postgres/0001_initial.sql (selected lines) ---\n'
sed -n '1,120p' migrations/postgres/0001_initial.sql

printf '\n--- src/postgres_repo/mod.rs ---\n'
sed -n '1,140p' src/postgres_repo/mod.rs

printf '\n--- src/sqlx_repo/repo.rs (migrator-related) ---\n'
sed -n '1,120p' src/sqlx_repo/repo.rs

Repository: hops-ops/distributed

Length of output: 15165

🏁 Script executed:

#!/bin/bash
set -euo pipefail

printf 'Postgres migration files:\n'
git ls-files 'migrations/postgres/*' | sort

printf '\n--- migrations/postgres/0001_initial.sql (selected lines) ---\n'
sed -n '1,120p' migrations/postgres/0001_initial.sql

printf '\n--- src/postgres_repo/mod.rs ---\n'
sed -n '1,140p' src/postgres_repo/mod.rs

printf '\n--- src/sqlx_repo/repo.rs (migrator-related) ---\n'
sed -n '1,140p' src/sqlx_repo/repo.rs

Repository: hops-ops/distributed

Length of output: 16403

Add a follow-up migration for the widened columns

Changing 0001_initial.sql only affects fresh installs. SQLx records applied migrations in _sqlx_migrations, so already-deployed Postgres databases will keep the old integer columns and fail to decode these fields as i64. Add a new ALTER TABLE ... ALTER COLUMN ... TYPE bigint migration for event_version, payload_codec_version, snapshot_version, and attempts instead of editing 0001.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@migrations/postgres/0001_initial.sql` around lines 6 - 9, The widened
Postgres column types were changed in the initial migration, but existing
databases will not pick that up because SQLx tracks applied migrations in
_sqlx_migrations. Add a new follow-up migration that ALTER TABLEs the existing
tables to change event_version, payload_codec_version, snapshot_version, and
attempts to bigint, and leave 0001_initial.sql unchanged so fresh installs and
already-deployed databases stay compatible.

metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
recorded_at timestamptz NOT NULL DEFAULT now(),
PRIMARY KEY (aggregate_type, aggregate_id, sequence),
Expand All @@ -28,10 +28,10 @@ CREATE TABLE IF NOT EXISTS aggregate_snapshots (
aggregate_type text NOT NULL,
aggregate_id text NOT NULL,
version bigint NOT NULL,
snapshot_version integer NOT NULL,
snapshot_version bigint NOT NULL,
payload bytea NOT NULL,
payload_codec text NOT NULL,
payload_codec_version integer NOT NULL,
payload_codec_version bigint NOT NULL,
metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
recorded_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now(),
Expand All @@ -49,15 +49,15 @@ CREATE TABLE IF NOT EXISTS outbox_messages (
event_type text NOT NULL,
payload bytea NOT NULL,
payload_codec text NOT NULL,
payload_codec_version integer NOT NULL,
payload_codec_version bigint NOT NULL,
destination text,
metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
status text NOT NULL,
created_at timestamptz NOT NULL,
next_available_at timestamptz NOT NULL,
claimed_by text,
claimed_until timestamptz,
attempts integer NOT NULL DEFAULT 0,
attempts bigint NOT NULL DEFAULT 0,
last_error text,
published_at timestamptz,
failed_at timestamptz,
Expand Down
33 changes: 25 additions & 8 deletions src/aggregate/repository.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,8 +42,11 @@ pub(crate) struct SnapshotPolicy<R, A> {
/// Build a snapshot cache record for the aggregate when one is due.
record: fn(&A, u64) -> Result<Option<SnapshotRecord>, RepositoryError>,
/// Hydrate an already-loaded entity from the cache record (if any). Used on
/// load paths that have the full stream in hand (batch loads, locked reads).
/// load paths that have the full stream in hand (locked reads).
hydrate: HydrateFn<R, A>,
/// Hydrate a batch of already-loaded entities, reading all cache records in
/// one round trip. Used by batch loads (`get_all`).
hydrate_all: HydrateAllFn<R, A>,
/// Own the whole load: read the snapshot first, then fetch only the tail of
/// the stream (skipping already-snapshotted I/O), falling back to a full
/// load on a cache miss. Used by the single-aggregate `get` hot path.
Expand All @@ -57,6 +60,12 @@ type HydrateFn<R, A> =
Entity,
) -> Pin<Box<dyn Future<Output = Result<A, RepositoryError>> + Send + 'a>>;

type HydrateAllFn<R, A> =
for<'a> fn(
&'a R,
Vec<(StreamIdentity, Entity)>,
) -> Pin<Box<dyn Future<Output = Result<Vec<A>, RepositoryError>> + Send + 'a>>;

type LoadFn<R, A> = for<'a> fn(
&'a R,
&'a StreamIdentity,
Expand All @@ -71,12 +80,14 @@ impl<R, A> SnapshotPolicy<R, A> {
frequency: u64,
record: fn(&A, u64) -> Result<Option<SnapshotRecord>, RepositoryError>,
hydrate: HydrateFn<R, A>,
hydrate_all: HydrateAllFn<R, A>,
load: LoadFn<R, A>,
) -> Self {
Self {
frequency,
record,
hydrate,
hydrate_all,
load,
}
}
Expand Down Expand Up @@ -223,15 +234,21 @@ impl<R, A> AggregateRepository<R, A>
where
A: Aggregate + Send,
{
/// Hydrate a batch of entities, deriving each identity from the entity id so
/// the snapshot cache can be consulted per aggregate.
/// Hydrate a batch of entities, deriving each identity from the entity id.
/// With a snapshot policy the cache records for the whole batch are read in
/// one round trip; without one, a plain per-entity replay.
async fn hydrate_entities(&self, entities: Vec<Entity>) -> Result<Vec<A>, RepositoryError> {
let mut aggregates = Vec::with_capacity(entities.len());
for entity in entities {
let identity = stream_identity_for::<A>(entity.id())?;
aggregates.push(self.hydrate_entity(&identity, entity).await?);
match &self.snapshot {
Some(policy) => {
let mut pairs = Vec::with_capacity(entities.len());
for entity in entities {
let identity = stream_identity_for::<A>(entity.id())?;
pairs.push((identity, entity));
}
(policy.hydrate_all)(&self.repo, pairs).await
}
None => entities.into_iter().map(hydrate::<A>).collect(),
}
Ok(aggregates)
}
}

Expand Down
18 changes: 11 additions & 7 deletions src/entity/event_record.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
use std::borrow::Cow;
use std::collections::HashMap;
use std::fmt;
use std::time::SystemTime;
Expand Down Expand Up @@ -90,8 +91,8 @@ fn default_event_version() -> u64 {
fn is_version_one(v: &u64) -> bool {
*v == 1
}
fn default_payload_codec() -> String {
BITCODE_PAYLOAD_CODEC.to_string()
fn default_payload_codec() -> Cow<'static, str> {
Cow::Borrowed(BITCODE_PAYLOAD_CODEC)
}
fn default_payload_codec_version() -> u16 {
BITCODE_PAYLOAD_CODEC_VERSION
Expand All @@ -105,8 +106,11 @@ fn default_payload_codec_version() -> u16 {
#[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
pub struct EventRecord {
pub event_name: String,
/// Payload codec name. `Cow` because virtually every event carries the
/// crate's own codec constant — the borrowed variant avoids one heap
/// allocation per event created and per row decoded.
#[serde(default = "default_payload_codec")]
pub payload_codec: String,
pub payload_codec: Cow<'static, str>,
#[serde(default = "default_payload_codec_version")]
pub payload_codec_version: u16,
#[serde(with = "payload_serde")]
Expand Down Expand Up @@ -146,7 +150,7 @@ impl EventRecord {
pub fn new(event_name: impl Into<String>, payload: Vec<u8>, sequence: u64) -> Self {
EventRecord {
event_name: event_name.into(),
payload_codec: BITCODE_PAYLOAD_CODEC.to_string(),
payload_codec: Cow::Borrowed(BITCODE_PAYLOAD_CODEC),
payload_codec_version: BITCODE_PAYLOAD_CODEC_VERSION,
payload,
event_version: 1,
Expand All @@ -165,7 +169,7 @@ impl EventRecord {
) -> Self {
EventRecord {
event_name: event_name.into(),
payload_codec: BITCODE_PAYLOAD_CODEC.to_string(),
payload_codec: Cow::Borrowed(BITCODE_PAYLOAD_CODEC),
payload_codec_version: BITCODE_PAYLOAD_CODEC_VERSION,
payload,
event_version: version,
Expand All @@ -184,7 +188,7 @@ impl EventRecord {
) -> Self {
EventRecord {
event_name: event_name.into(),
payload_codec: BITCODE_PAYLOAD_CODEC.to_string(),
payload_codec: Cow::Borrowed(BITCODE_PAYLOAD_CODEC),
payload_codec_version: BITCODE_PAYLOAD_CODEC_VERSION,
payload,
event_version: 1,
Expand Down Expand Up @@ -292,7 +296,7 @@ mod tests {
#[test]
fn decode_unknown_codec_returns_error() {
let mut event_record = EventRecord::new("test_event", vec![], 1);
event_record.payload_codec = "json".to_string();
event_record.payload_codec = "json".into();

let err = event_record.decode::<()>().unwrap_err();
assert!(err.message.contains("unsupported payload codec `json`"));
Expand Down
65 changes: 25 additions & 40 deletions src/hashmap_repo/repository.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,9 @@ use crate::read_model::{
InMemoryReadModelStore, ReadModelLoadGraph, ReadModelLoadRequest, ReadModelQueryCapabilities,
};
use crate::repository::{
reject_duplicate_outbox_messages, reject_duplicate_streams,
validate_entity_id_matches_identity, validate_prepared_appends, validate_snapshot_identity,
CommitBatch, GetStream, InboxStore, PreparedEventAppend, ReadModelWritePlanStore,
RelationalReadModelQueryStore, RepositoryError, SnapshotStore, SnapshotWrite, StreamIdentity,
TransactionalCommit,
validate_commit_batch, validate_snapshot_identity, CommitBatch, GetStream, InboxStore,
ReadModelWritePlanStore, RelationalReadModelQueryStore, RepositoryError, SnapshotStore,
SnapshotWrite, StreamIdentity, TransactionalCommit,
};
use crate::snapshot::{InMemorySnapshotStore, SnapshotRecord};
use crate::table::{TableAdapterCapabilities, TableCommitOutcome, TableStoreError, TableWritePlan};
Expand Down Expand Up @@ -140,21 +138,6 @@ impl GetStream for HashMapRepository {
}
}
}

fn get_streams<'a>(
&'a self,
identities: &'a [StreamIdentity],
) -> impl Future<Output = Result<Vec<Entity>, RepositoryError>> + Send + 'a {
async move {
let mut entities = Vec::with_capacity(identities.len());
for identity in identities {
if let Some(entity) = self.get_stream(identity).await? {
entities.push(entity);
}
}
Ok(entities)
}
}
}

impl TransactionalCommit for HashMapRepository {
Expand All @@ -163,18 +146,7 @@ impl TransactionalCommit for HashMapRepository {
batch: CommitBatch<'a>,
) -> impl Future<Output = Result<(), RepositoryError>> + Send + 'a {
async move {
reject_duplicate_streams(&batch.streams)?;
validate_entity_id_matches_identity(&batch.streams)?;
let prepared = batch
.streams
.iter()
.map(PreparedEventAppend::from_stream_write)
.collect::<Vec<_>>();
validate_prepared_appends(&prepared)?;
for write in &batch.snapshots {
validate_snapshot_write(write)?;
}
reject_duplicate_outbox_messages(&batch.outbox_messages)?;
let prepared = validate_commit_batch(&batch)?;

// All-or-nothing without cloning the stores: every fallible check
// below runs against the live maps (reads) or a staging copy scoped
Expand Down Expand Up @@ -232,7 +204,9 @@ impl TransactionalCommit for HashMapRepository {
staged_rows.insert(key.clone(), row.clone());
}
}
for plan in batch.read_model_plans {
// Clone plans: `prepared` borrows `batch` through the append loop
// below, so `batch.read_model_plans` cannot be moved out here.
for plan in batch.read_model_plans.iter().cloned() {
apply_read_model_write_plan(plan, &mut staged_rows)?;
}
debug_assert!(
Expand Down Expand Up @@ -273,7 +247,7 @@ impl TransactionalCommit for HashMapRepository {
storage
.entry(append.identity.storage_key())
.or_insert_with(Vec::new)
.extend(append.events);
.extend_from_slice(append.events);
}

for key in touched_rows {
Expand Down Expand Up @@ -331,12 +305,6 @@ impl InboxStore for HashMapRepository {
}
}

fn validate_snapshot_write(write: &SnapshotWrite) -> Result<(), RepositoryError> {
match write {
SnapshotWrite::Save { identity, record } => validate_snapshot_identity(identity, record),
}
}

fn stored_stream_version(events: Option<&Vec<EventRecord>>) -> u64 {
// A missing stream has committed version 0; the first appended event will
// occupy sequence 1.
Expand Down Expand Up @@ -384,6 +352,23 @@ impl SnapshotStore for HashMapRepository {
}
}

fn get_snapshots<'a>(
&'a self,
identities: &'a [StreamIdentity],
) -> impl Future<Output = Result<Vec<SnapshotRecord>, RepositoryError>> + Send + 'a {
async move {
let storage = self
.snapshot_store
.storage
.read()
.map_err(|_| RepositoryError::LockPoisoned("async snapshot read"))?;
Ok(identities
.iter()
.filter_map(|identity| storage.get(&identity.storage_key()).cloned())
.collect())
}
}

fn save_snapshot<'a>(
&'a self,
identity: &'a StreamIdentity,
Expand Down
16 changes: 8 additions & 8 deletions src/microsvc/runtime.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -196,7 +196,7 @@ mod tests {
.unwrap();

let published_a = store_a
.messages_by_status(OutboxMessageStatus::Published)
.messages_by_status(OutboxMessageStatus::Published, usize::MAX)
.await
.unwrap();
assert_eq!(
Expand All @@ -205,10 +205,10 @@ mod tests {
"first route bundle should publish at commit time"
);
assert_eq!(published_a[0].id(), "evt-1");
assert!(store_a.pending().await.unwrap().is_empty());
assert!(store_a.pending(usize::MAX).await.unwrap().is_empty());

let published_b = store_b
.messages_by_status(OutboxMessageStatus::Published)
.messages_by_status(OutboxMessageStatus::Published, usize::MAX)
.await
.unwrap();
assert_eq!(
Expand All @@ -217,7 +217,7 @@ mod tests {
"second route bundle should publish at commit time"
);
assert_eq!(published_b[0].id(), "evt-1");
assert!(store_b.pending().await.unwrap().is_empty());
assert!(store_b.pending(usize::MAX).await.unwrap().is_empty());
}

#[tokio::test]
Expand Down Expand Up @@ -275,12 +275,12 @@ mod tests {
.unwrap();

let published = store
.messages_by_status(OutboxMessageStatus::Published)
.messages_by_status(OutboxMessageStatus::Published, usize::MAX)
.await
.unwrap();
assert_eq!(published.len(), 1, "row should be published immediately");
assert_eq!(published[0].id(), "evt-1");
assert!(store.pending().await.unwrap().is_empty());
assert!(store.pending(usize::MAX).await.unwrap().is_empty());
}

#[tokio::test]
Expand All @@ -304,7 +304,7 @@ mod tests {
service.run(RunOptions::idempotent()).await.unwrap();

let published = store
.messages_by_status(OutboxMessageStatus::Published)
.messages_by_status(OutboxMessageStatus::Published, usize::MAX)
.await
.unwrap();
assert_eq!(
Expand Down Expand Up @@ -362,7 +362,7 @@ mod tests {
.unwrap();

let published = store
.messages_by_status(OutboxMessageStatus::Published)
.messages_by_status(OutboxMessageStatus::Published, usize::MAX)
.await
.unwrap();
assert_eq!(
Expand Down
14 changes: 12 additions & 2 deletions src/outbox/commit.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -267,7 +267,12 @@ mod tests {
assert!(receipt.has_outbox_messages());
assert_eq!(receipt.outbox_message_ids(), ["msg-1".to_string()]);

let pending = repo.repo().outbox_store().pending().await.unwrap();
let pending = repo
.repo()
.outbox_store()
.pending(usize::MAX)
.await
.unwrap();
assert_eq!(pending.len(), 1);
assert_eq!(pending[0].id(), "msg-1");
}
Expand Down Expand Up @@ -404,7 +409,12 @@ mod tests {
);

// 2) outbox row present (pending — no bus attached here)
let pending = repo.repo().outbox_store().pending().await.unwrap();
let pending = repo
.repo()
.outbox_store()
.pending(usize::MAX)
.await
.unwrap();
assert_eq!(pending.len(), 1);
assert_eq!(pending[0].id(), "evt-c1");

Expand Down
Loading
Loading