use crate::access_control::{
AccessLevel, CommandToken, DirectCreationToken, EventReadStartPosition, EventReadToken,
ExternalEventToken, PrivateFieldGrant, Role, RoleAccessMapping, RoleStatus, TokenStatus,
};
use crate::bootstrap::ContextCreator;
use crate::encryption::{self, DataKey, EncryptionMasterKey};
use crate::event_store::{
AckMode, BoundedContext, BoundedContextStatus, Command, CommandType, CursorUpdate,
EncryptionKey, EncryptionKeyStatus, Event, EventOrigin, EventType, MissedOccurrencePolicy,
ReadCursor,
};
use crate::projections::{Projection, ProjectionRebuild, ProjectionRebuildStatus};
use crate::shared::{Metadata, PrivateField, SensitiveField, Tag, TagMapping};
use chrono::{DateTime, Utc};
use opentelemetry::metrics::{Counter, Histogram, Meter};
use opentelemetry::KeyValue;
use sqlx::types::Json;
use sqlx::{Acquire, Postgres, Transaction};
use std::sync::LazyLock;
fn meter() -> &'static Meter {
static METER: LazyLock<Meter> = LazyLock::new(|| opentelemetry::global::meter("skilj-core"));
&METER
}
static COMMANDS_PROCESSED: LazyLock<Counter<u64>> = LazyLock::new(|| {
meter()
.u64_counter("skilj.commands.processed")
.with_description("Commands processed by ProcessCommand, by outcome.")
.build()
});
static EVENTS_APPENDED: LazyLock<Counter<u64>> = LazyLock::new(|| {
meter()
.u64_counter("skilj.events.appended")
.with_description("Events appended to the event store.")
.build()
});
static COMMAND_BATCH_SIZE: LazyLock<Histogram<u64>> = LazyLock::new(|| {
meter()
.u64_histogram("skilj.command_batch.size")
.with_description("Commands coalesced into one command-batch lock acquisition.")
.build()
});
fn record_event_appended(event: &Event) {
EVENTS_APPENDED.add(
1,
&[
KeyValue::new("bounded_context", event.bounded_context.name.clone()),
KeyValue::new("event_type", event.event_type.name.clone()),
],
);
}
async fn notify_event_appended(pool: &Pool, event: &Event, origin_instance_id: &str) {
let payload = serde_json::json!({
"bounded_context": event.bounded_context.name,
"sequence": event.sequence,
"origin_instance_id": origin_instance_id,
})
.to_string();
if let Err(err) = sqlx::query("SELECT pg_notify('skilj_events', $1)")
.bind(payload)
.execute(pool)
.await
{
tracing::warn!(
error = %err,
"NOTIFY skilj_events failed - other instances may miss this event's live push \
until their next poll-based read"
);
}
}
async fn notify_registration_changed(pool: &Pool) {
if let Err(err) = sqlx::query("SELECT pg_notify('skilj_registration_changed', '')")
.execute(pool)
.await
{
tracing::warn!(
error = %err,
"NOTIFY skilj_registration_changed failed - other instances' GraphQL schema may \
lag until their next registration change or restart"
);
}
}
pub async fn notify_revocation(
pool: &Pool,
mapping: &crate::access_control::RevokedMapping,
origin_instance_id: &str,
) {
let payload = serde_json::json!({
"role_id": mapping.role_id,
"bounded_context": mapping.bounded_context,
"origin_instance_id": origin_instance_id,
})
.to_string();
if let Err(err) = sqlx::query("SELECT pg_notify('skilj_revocations', $1)")
.bind(payload)
.execute(pool)
.await
{
tracing::warn!(
error = %err,
"NOTIFY skilj_revocations failed - other instances may miss this revocation's live \
push until their next direct database re-check"
);
}
}
pub type Pool = sqlx::PgPool;
pub use sqlx::postgres::PgPoolOptions;
#[tracing::instrument(skip_all)]
pub async fn connect(database_url: &str) -> Result<Pool, sqlx::Error> {
connect_with(database_url, PgPoolOptions::new()).await
}
#[tracing::instrument(skip_all)]
pub async fn connect_with(database_url: &str, options: PgPoolOptions) -> Result<Pool, sqlx::Error> {
options.connect(database_url).await
}
#[tracing::instrument(skip_all)]
pub async fn migrate(pool: &Pool) -> Result<(), sqlx::migrate::MigrateError> {
sqlx::migrate!("./migrations").run(pool).await
}
fn bounded_context_status_to_str(status: BoundedContextStatus) -> &'static str {
match status {
BoundedContextStatus::Active => "active",
BoundedContextStatus::Archived => "archived",
}
}
fn bounded_context_status_from_str(s: &str) -> BoundedContextStatus {
match s {
"archived" => BoundedContextStatus::Archived,
_ => BoundedContextStatus::Active,
}
}
fn role_status_to_str(status: RoleStatus) -> &'static str {
match status {
RoleStatus::Active => "active",
RoleStatus::Revoked => "revoked",
}
}
fn role_status_from_str(s: &str) -> RoleStatus {
match s {
"revoked" => RoleStatus::Revoked,
_ => RoleStatus::Active,
}
}
fn token_status_to_str(status: TokenStatus) -> &'static str {
match status {
TokenStatus::Active => "active",
TokenStatus::Revoked => "revoked",
}
}
fn token_status_from_str(s: &str) -> TokenStatus {
match s {
"revoked" => TokenStatus::Revoked,
_ => TokenStatus::Active,
}
}
fn ack_mode_to_str(mode: AckMode) -> &'static str {
match mode {
AckMode::AutoAdvance => "auto_advance",
AckMode::ManualAck => "manual_ack",
}
}
fn ack_mode_from_str(s: &str) -> AckMode {
match s {
"manual_ack" => AckMode::ManualAck,
_ => AckMode::AutoAdvance,
}
}
fn event_read_start_position_to_str(position: EventReadStartPosition) -> &'static str {
match position {
EventReadStartPosition::Beginning => "beginning",
EventReadStartPosition::Latest => "latest",
EventReadStartPosition::AtSequence => "at_sequence",
EventReadStartPosition::AtTime => "at_time",
}
}
fn event_read_start_position_from_str(s: &str) -> EventReadStartPosition {
match s {
"latest" => EventReadStartPosition::Latest,
"at_sequence" => EventReadStartPosition::AtSequence,
"at_time" => EventReadStartPosition::AtTime,
_ => EventReadStartPosition::Beginning,
}
}
fn access_level_to_str(level: AccessLevel) -> &'static str {
match level {
AccessLevel::Read => "read",
AccessLevel::Write => "write",
AccessLevel::Admin => "admin",
}
}
fn access_level_from_str(s: &str) -> AccessLevel {
match s {
"write" => AccessLevel::Write,
"admin" => AccessLevel::Admin,
_ => AccessLevel::Read,
}
}
fn schema_ident(bounded_context: &str) -> String {
format!("\"bc_{bounded_context}\"")
}
async fn provision_bounded_context_schema(
tx: &mut Transaction<'_, Postgres>,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!("CREATE SCHEMA {schema}")))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.event_types (
name TEXT PRIMARY KEY,
schema TEXT NOT NULL,
schema_version BIGINT NOT NULL,
tag_mappings JSONB NOT NULL DEFAULT '[]',
owner_tag_key TEXT,
sensitive_fields JSONB NOT NULL DEFAULT '[]',
private_fields JSONB NOT NULL DEFAULT '[]',
external_creation_allowed BOOLEAN NOT NULL,
direct_creation_allowed BOOLEAN NOT NULL,
system_triggered_allowed BOOLEAN NOT NULL,
system_triggered_schedule TEXT,
missed_occurrence_policy TEXT CHECK (missed_occurrence_policy IN \
('skip', 'fire_once', 'replay_backlog')),
schedule_position TIMESTAMPTZ,
last_fired_at TIMESTAMPTZ,
event_read_allowed BOOLEAN NOT NULL
)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.command_types (
name TEXT PRIMARY KEY,
schema TEXT NOT NULL,
schema_version BIGINT NOT NULL,
tag_mappings JSONB NOT NULL DEFAULT '[]',
owner_tag_key TEXT,
sensitive_fields JSONB NOT NULL DEFAULT '[]',
private_fields JSONB NOT NULL DEFAULT '[]',
rest_trigger_allowed BOOLEAN NOT NULL
)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.encryption_keys (
id BIGSERIAL PRIMARY KEY,
subject_key TEXT NOT NULL,
subject_value TEXT NOT NULL,
status TEXT NOT NULL CHECK (status IN ('active', 'destroyed')),
created_at TIMESTAMPTZ NOT NULL,
destroyed_at TIMESTAMPTZ,
wrapped_key BYTEA,
wrap_nonce BYTEA
)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE UNIQUE INDEX encryption_keys_unique_active ON {schema}.encryption_keys \
(subject_key, subject_value) WHERE status = 'active'"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.commands (
id BIGSERIAL PRIMARY KEY,
external_id TEXT NOT NULL UNIQUE,
command_type_name TEXT NOT NULL REFERENCES {schema}.command_types (name),
payload TEXT NOT NULL,
metadata_type TEXT NOT NULL,
metadata_version BIGINT NOT NULL,
metadata_client_id TEXT NOT NULL,
metadata_created_at TIMESTAMPTZ NOT NULL,
metadata_correlation_id TEXT,
metadata_causation_id TEXT,
consistency_tags JSONB NOT NULL DEFAULT '[]',
consistency_boundary BIGINT
)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE INDEX commands_by_created_at ON {schema}.commands (metadata_created_at)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE INDEX commands_by_correlation_id ON {schema}.commands (metadata_correlation_id) \
WHERE metadata_correlation_id IS NOT NULL"
)))
.execute(&mut **tx)
.await?;
ensure_idempotency_keys_table(&mut **tx, bounded_context).await?;
ensure_cross_context_route_cursors_table(&mut **tx, bounded_context).await?;
ensure_external_message_cursors_table(&mut **tx, bounded_context).await?;
ensure_deadline_cursors_table(&mut **tx, bounded_context).await?;
ensure_parked_deliveries_table(&mut **tx, bounded_context).await?;
sqlx::query(sqlx::AssertSqlSafe(deadlines_table_ddl(&schema)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE INDEX IF NOT EXISTS deadlines_due ON {schema}.deadlines (fire_at) \
WHERE status = 'pending'"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE INDEX IF NOT EXISTS deadlines_tags_gin ON {schema}.deadlines USING GIN (tags)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.command_encryption_keys (
command_id BIGINT NOT NULL REFERENCES {schema}.commands (id),
encryption_key_id BIGINT NOT NULL REFERENCES {schema}.encryption_keys (id),
PRIMARY KEY (command_id, encryption_key_id)
)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.projections (
name TEXT PRIMARY KEY,
schema TEXT NOT NULL,
schema_version BIGINT NOT NULL,
sync BOOLEAN NOT NULL,
caught_up_to BIGINT
)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.projection_consumed_event_types (
projection_name TEXT NOT NULL REFERENCES {schema}.projections (name),
event_type_name TEXT NOT NULL REFERENCES {schema}.event_types (name),
PRIMARY KEY (projection_name, event_type_name)
)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.projection_rebuilds (
projection_name TEXT NOT NULL REFERENCES {schema}.projections (name),
schema TEXT NOT NULL,
schema_version BIGINT NOT NULL,
sync BOOLEAN NOT NULL,
caught_up_to BIGINT,
status TEXT NOT NULL CHECK (status IN ('pending', 'building')),
PRIMARY KEY (projection_name, status)
)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.projection_rebuild_consumed_event_types (
projection_name TEXT NOT NULL,
status TEXT NOT NULL,
event_type_name TEXT NOT NULL REFERENCES {schema}.event_types (name),
PRIMARY KEY (projection_name, status, event_type_name),
FOREIGN KEY (projection_name, status)
REFERENCES {schema}.projection_rebuilds (projection_name, status)
)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.projection_rebuild_state (
projection_name TEXT NOT NULL,
status TEXT NOT NULL,
key TEXT NOT NULL,
state TEXT NOT NULL,
owner TEXT,
as_of_sequence BIGINT NOT NULL DEFAULT -1,
updated_at TIMESTAMPTZ NOT NULL,
PRIMARY KEY (projection_name, status, key),
FOREIGN KEY (projection_name, status)
REFERENCES {schema}.projection_rebuilds (projection_name, status)
)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.projection_state (
projection_name TEXT NOT NULL REFERENCES {schema}.projections (name),
key TEXT NOT NULL,
state TEXT NOT NULL,
owner TEXT,
as_of_sequence BIGINT NOT NULL DEFAULT -1,
updated_at TIMESTAMPTZ NOT NULL,
PRIMARY KEY (projection_name, key)
)"
)))
.execute(&mut **tx)
.await?;
ensure_projection_partition_progress_table(&mut **tx, bounded_context).await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.sequence (next_value BIGINT NOT NULL)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.sequence (next_value) VALUES (-1)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.events (
sequence BIGINT PRIMARY KEY,
event_type_name TEXT NOT NULL REFERENCES {schema}.event_types (name),
payload TEXT NOT NULL,
metadata_type TEXT NOT NULL,
metadata_version BIGINT NOT NULL,
metadata_client_id TEXT NOT NULL,
metadata_created_at TIMESTAMPTZ NOT NULL,
metadata_correlation_id TEXT,
metadata_causation_id TEXT,
tags JSONB NOT NULL DEFAULT '[]',
origin_kind TEXT NOT NULL CHECK (
origin_kind IN ('external_triggered', 'directly_created', 'command_triggered', 'system_triggered')
),
origin_source_content TEXT,
origin_source_context TEXT,
origin_command_id BIGINT REFERENCES {schema}.commands (id)
)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE INDEX events_by_type ON {schema}.events (event_type_name, sequence)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE INDEX events_by_correlation_id ON {schema}.events (metadata_correlation_id) \
WHERE metadata_correlation_id IS NOT NULL"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE INDEX events_by_tags ON {schema}.events USING GIN (tags)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.snapshots (
snapshot_name TEXT NOT NULL,
tag_key TEXT NOT NULL,
tag_value TEXT NOT NULL,
snapshot_version BIGINT NOT NULL,
as_of_sequence BIGINT NOT NULL,
state JSONB NOT NULL,
owner TEXT,
updated_at TIMESTAMPTZ NOT NULL,
PRIMARY KEY (snapshot_name, tag_key, tag_value)
)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.snapshot_progress (
snapshot_name TEXT PRIMARY KEY,
caught_up_to BIGINT NOT NULL
)"
)))
.execute(&mut **tx)
.await?;
ensure_snapshot_partition_progress_table(&mut **tx, bounded_context).await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.event_encryption_keys (
event_sequence BIGINT NOT NULL REFERENCES {schema}.events (sequence),
encryption_key_id BIGINT NOT NULL REFERENCES {schema}.encryption_keys (id),
PRIMARY KEY (event_sequence, encryption_key_id)
)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.access_tokens (
id TEXT PRIMARY KEY,
kind TEXT NOT NULL CHECK (kind IN ('external_event', 'direct_creation', 'event_read', 'command')),
secret TEXT NOT NULL, -- hash_secret's output, never the plaintext (see AccessToken.secret)
status TEXT NOT NULL CHECK (status IN ('active', 'revoked')),
created_at TIMESTAMPTZ NOT NULL,
revoked_at TIMESTAMPTZ,
event_type_name TEXT REFERENCES {schema}.event_types (name),
command_type_name TEXT REFERENCES {schema}.command_types (name),
-- EventReadToken.scope/ExternalEventToken.scope/
-- DirectCreationToken.scope/CommandToken.scope - one column
-- for all four kinds, the cross-tenant read/write fix
-- (docs/architecture.md's own write-up of these passes).
-- Null (the default before these fixes existed, and still
-- the default for a token minted with none) means
-- unrestricted, for every kind alike.
scope TEXT,
-- EventReadToken.start_from/.start_at_sequence/.start_at_time -
-- meaningful only for the 'event_read' kind (see
-- ensure_event_read_token_start_from_column's own doc comment
-- for why a fresh table still declares these with the
-- identical defaults a retrofitted one gets).
start_from TEXT NOT NULL DEFAULT 'beginning'
CHECK (start_from IN ('beginning', 'latest', 'at_sequence', 'at_time')),
start_at_sequence BIGINT,
start_at_time TIMESTAMPTZ,
CHECK (
(kind = 'command' AND command_type_name IS NOT NULL AND event_type_name IS NULL)
OR (kind != 'command' AND event_type_name IS NOT NULL AND command_type_name IS NULL)
)
)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE {schema}.read_cursors (
token_id TEXT PRIMARY KEY REFERENCES {schema}.access_tokens (id),
ack_mode TEXT NOT NULL CHECK (ack_mode IN ('auto_advance', 'manual_ack')),
sequence BIGINT NOT NULL,
updated_at TIMESTAMPTZ NOT NULL,
checked_out_at TIMESTAMPTZ
)"
)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(private_field_grants_table_ddl(&schema)))
.execute(&mut **tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE INDEX private_field_grants_by_grantee ON {schema}.private_field_grants (grantee_role_id, status)"
)))
.execute(&mut **tx)
.await?;
Ok(())
}
fn private_field_grants_table_ddl(schema: &str) -> String {
format!(
"CREATE TABLE IF NOT EXISTS {schema}.private_field_grants (
id TEXT PRIMARY KEY,
grantor_role_id TEXT NOT NULL REFERENCES public.roles (id),
grantee_role_id TEXT NOT NULL REFERENCES public.roles (id),
event_sequence BIGINT REFERENCES {schema}.events (sequence),
command_id BIGINT REFERENCES {schema}.commands (id),
status TEXT NOT NULL CHECK (status IN ('active', 'revoked')),
created_at TIMESTAMPTZ NOT NULL,
revoked_at TIMESTAMPTZ,
CHECK (event_sequence IS NULL OR command_id IS NULL)
)"
)
}
pub async fn ensure_private_field_grants_table(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(private_field_grants_table_ddl(&schema)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE INDEX IF NOT EXISTS private_field_grants_by_grantee ON {schema}.private_field_grants (grantee_role_id, status)"
)))
.execute(pool)
.await?;
Ok(())
}
async fn command_internal_id(
pool: &Pool,
bounded_context: &str,
external_id: &str,
) -> crate::error::Result<Option<i64>> {
let schema = schema_ident(bounded_context);
let row: Option<(i64,)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT id FROM {schema}.commands WHERE external_id = $1"
)))
.bind(external_id)
.fetch_optional(pool)
.await?;
Ok(row.map(|(id,)| id))
}
#[derive(sqlx::FromRow)]
struct PrivateFieldGrantRow {
id: String,
grantor_role_id: String,
grantee_role_id: String,
event_sequence: Option<i64>,
command_id: Option<i64>,
status: String,
created_at: DateTime<Utc>,
revoked_at: Option<DateTime<Utc>>,
}
impl PrivateFieldGrantRow {
async fn into_domain(
self,
pool: &Pool,
bounded_context: &BoundedContext,
) -> crate::error::Result<Option<PrivateFieldGrant>> {
let Some(grantor) = get_role(pool, &self.grantor_role_id).await? else {
return Ok(None);
};
let Some(grantee) = get_role(pool, &self.grantee_role_id).await? else {
return Ok(None);
};
let command_id = match self.command_id {
Some(internal_id) => {
match get_command_by_id(pool, &bounded_context.name, internal_id).await? {
Some(command) => Some(command.id),
None => return Ok(None),
}
}
None => None,
};
Ok(Some(PrivateFieldGrant {
id: self.id,
bounded_context: bounded_context.clone(),
grantor,
grantee,
event_sequence: self.event_sequence,
command_id,
status: token_status_from_str(&self.status),
created_at: self.created_at,
revoked_at: self.revoked_at,
}))
}
}
const PRIVATE_FIELD_GRANT_COLUMNS: &str =
"id, grantor_role_id, grantee_role_id, event_sequence, command_id, status, created_at, revoked_at";
#[tracing::instrument(skip_all)]
pub async fn insert_private_field_grant(
pool: &Pool,
grant: &PrivateFieldGrant,
) -> crate::error::Result<()> {
let schema = schema_ident(&grant.bounded_context.name);
let command_id = match grant.command_id.as_deref() {
Some(external_id) => Some(
command_internal_id(pool, &grant.bounded_context.name, external_id)
.await?
.expect(
"insert_private_field_grant: command_id names a command that doesn't exist",
),
),
None => None,
};
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.private_field_grants ({PRIVATE_FIELD_GRANT_COLUMNS}) \
VALUES ($1,$2,$3,$4,$5,$6,$7,$8)"
)))
.bind(&grant.id)
.bind(&grant.grantor.id)
.bind(&grant.grantee.id)
.bind(grant.event_sequence)
.bind(command_id)
.bind(token_status_to_str(grant.status))
.bind(grant.created_at)
.bind(grant.revoked_at)
.execute(pool)
.await?;
Ok(())
}
#[tracing::instrument(skip_all)]
pub async fn list_private_field_grants_for_context(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<Vec<PrivateFieldGrant>> {
let Some(bc) = get_bounded_context(pool, bounded_context).await? else {
return Ok(Vec::new());
};
let schema = schema_ident(bounded_context);
let rows: Vec<PrivateFieldGrantRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {PRIVATE_FIELD_GRANT_COLUMNS} FROM {schema}.private_field_grants"
)))
.fetch_all(pool)
.await?;
let mut grants = Vec::with_capacity(rows.len());
for row in rows {
if let Some(grant) = row.into_domain(pool, &bc).await? {
grants.push(grant);
}
}
Ok(grants)
}
#[tracing::instrument(skip_all)]
pub async fn get_private_field_grant(
pool: &Pool,
bounded_context: &str,
id: &str,
) -> crate::error::Result<Option<PrivateFieldGrant>> {
let Some(bc) = get_bounded_context(pool, bounded_context).await? else {
return Ok(None);
};
let schema = schema_ident(bounded_context);
let row: Option<PrivateFieldGrantRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {PRIVATE_FIELD_GRANT_COLUMNS} FROM {schema}.private_field_grants WHERE id = $1"
)))
.bind(id)
.fetch_optional(pool)
.await?;
match row {
Some(row) => row.into_domain(pool, &bc).await,
None => Ok(None),
}
}
#[tracing::instrument(skip_all)]
pub async fn update_private_field_grant(
pool: &Pool,
grant: &PrivateFieldGrant,
) -> crate::error::Result<()> {
let schema = schema_ident(&grant.bounded_context.name);
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.private_field_grants SET status = $1, revoked_at = $2 WHERE id = $3"
)))
.bind(token_status_to_str(grant.status))
.bind(grant.revoked_at)
.bind(&grant.id)
.execute(pool)
.await?;
Ok(())
}
#[tracing::instrument(skip_all)]
pub async fn ensure_idempotency_keys_table<'e>(
executor: impl sqlx::PgExecutor<'e>,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE IF NOT EXISTS {schema}.idempotency_keys (
command_type_name TEXT NOT NULL,
client_id TEXT NOT NULL,
idempotency_key TEXT NOT NULL,
triggered_event_sequences BIGINT[] NOT NULL,
created_at TIMESTAMPTZ NOT NULL,
PRIMARY KEY (command_type_name, client_id, idempotency_key)
)"
)))
.execute(executor)
.await?;
Ok(())
}
#[tracing::instrument(skip_all)]
pub async fn migrate_idempotency_keys_client_id_scoping(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
let raw_schema = format!("bc_{bounded_context}");
let mut tx = pool.begin().await?;
sqlx::query("SELECT pg_advisory_xact_lock(hashtext($1)::bigint)")
.bind(&raw_schema)
.execute(&mut *tx)
.await?;
let already_migrated: bool = sqlx::query_scalar(
"SELECT EXISTS (
SELECT 1 FROM information_schema.key_column_usage
WHERE table_schema = $1 AND table_name = 'idempotency_keys'
AND constraint_name = 'idempotency_keys_pkey'
AND column_name = 'client_id'
)",
)
.bind(&raw_schema)
.fetch_one(&mut *tx)
.await?;
if already_migrated {
tx.commit().await?;
return Ok(());
}
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.idempotency_keys \
ADD COLUMN IF NOT EXISTS client_id TEXT NOT NULL DEFAULT ''"
)))
.execute(&mut *tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.idempotency_keys DROP CONSTRAINT IF EXISTS idempotency_keys_pkey"
)))
.execute(&mut *tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.idempotency_keys \
ADD PRIMARY KEY (command_type_name, client_id, idempotency_key)"
)))
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(())
}
#[tracing::instrument(skip_all)]
pub async fn ensure_cross_context_route_cursors_table<'e>(
executor: impl sqlx::PgExecutor<'e>,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE IF NOT EXISTS {schema}.cross_context_route_cursors (
route_name TEXT PRIMARY KEY,
last_dispatched_sequence BIGINT NOT NULL DEFAULT -1,
updated_at TIMESTAMPTZ NOT NULL,
retry_attempt_count INT NOT NULL DEFAULT 0,
retry_first_failed_at TIMESTAMPTZ,
retry_next_attempt_at TIMESTAMPTZ
)"
)))
.execute(executor)
.await?;
Ok(())
}
pub async fn ensure_cross_context_route_retry_columns(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.cross_context_route_cursors \
ADD COLUMN IF NOT EXISTS retry_attempt_count INT NOT NULL DEFAULT 0"
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.cross_context_route_cursors \
ADD COLUMN IF NOT EXISTS retry_first_failed_at TIMESTAMPTZ"
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.cross_context_route_cursors \
ADD COLUMN IF NOT EXISTS retry_next_attempt_at TIMESTAMPTZ"
)))
.execute(pool)
.await?;
Ok(())
}
pub async fn ensure_deadline_cursors_table<'e>(
executor: impl sqlx::PgExecutor<'e>,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE IF NOT EXISTS {schema}.deadline_cursors (
cursor_owner TEXT PRIMARY KEY,
last_dispatched_sequence BIGINT NOT NULL DEFAULT -1,
updated_at TIMESTAMPTZ NOT NULL
)"
)))
.execute(executor)
.await?;
Ok(())
}
pub async fn ensure_projection_partition_progress_table<'e>(
executor: impl sqlx::PgExecutor<'e>,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE IF NOT EXISTS {schema}.projection_partition_progress (
projection_name TEXT NOT NULL REFERENCES {schema}.projections (name),
partition_index INT NOT NULL,
caught_up_to BIGINT NOT NULL DEFAULT -1,
updated_at TIMESTAMPTZ NOT NULL,
PRIMARY KEY (projection_name, partition_index)
)"
)))
.execute(executor)
.await?;
Ok(())
}
pub async fn ensure_snapshot_partition_progress_table<'e>(
executor: impl sqlx::PgExecutor<'e>,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE IF NOT EXISTS {schema}.snapshot_partition_progress (
snapshot_name TEXT NOT NULL,
partition_index INT NOT NULL,
caught_up_to BIGINT NOT NULL DEFAULT -1,
updated_at TIMESTAMPTZ NOT NULL,
PRIMARY KEY (snapshot_name, partition_index)
)"
)))
.execute(executor)
.await?;
Ok(())
}
fn deadlines_table_ddl(schema: &str) -> String {
format!(
"CREATE TABLE IF NOT EXISTS {schema}.deadlines (
id TEXT PRIMARY KEY,
schedule_name TEXT NOT NULL,
fire_at TIMESTAMPTZ NOT NULL,
tags JSONB NOT NULL,
correlation_id TEXT,
target_bounded_context TEXT NOT NULL,
target_command_type TEXT NOT NULL,
payload TEXT NOT NULL,
status TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL,
resolved_at TIMESTAMPTZ,
firing_at TIMESTAMPTZ
)"
)
}
pub async fn ensure_deadlines_table(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(deadlines_table_ddl(&schema)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.deadlines ADD COLUMN IF NOT EXISTS firing_at TIMESTAMPTZ"
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE INDEX IF NOT EXISTS deadlines_due ON {schema}.deadlines (fire_at) \
WHERE status = 'pending'"
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE INDEX IF NOT EXISTS deadlines_tags_gin ON {schema}.deadlines USING GIN (tags)"
)))
.execute(pool)
.await?;
Ok(())
}
#[tracing::instrument(skip_all)]
pub async fn ensure_external_message_cursors_table<'e>(
executor: impl sqlx::PgExecutor<'e>,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE IF NOT EXISTS {schema}.external_message_cursors (
adapter_id TEXT NOT NULL,
partition_key TEXT NOT NULL,
last_sequence BIGINT NOT NULL,
updated_at TIMESTAMPTZ NOT NULL,
PRIMARY KEY (adapter_id, partition_key)
)"
)))
.execute(executor)
.await?;
Ok(())
}
#[tracing::instrument(skip_all)]
pub async fn ensure_projection_state_owner_columns(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.projection_state ADD COLUMN IF NOT EXISTS owner TEXT"
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.projection_rebuild_state ADD COLUMN IF NOT EXISTS owner TEXT"
)))
.execute(pool)
.await?;
Ok(())
}
#[tracing::instrument(skip_all)]
pub async fn ensure_projection_state_as_of_sequence_columns(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.projection_state \
ADD COLUMN IF NOT EXISTS as_of_sequence BIGINT NOT NULL DEFAULT -1"
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.projection_rebuild_state \
ADD COLUMN IF NOT EXISTS as_of_sequence BIGINT NOT NULL DEFAULT -1"
)))
.execute(pool)
.await?;
Ok(())
}
pub async fn ensure_read_cursors_checkout_column(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.read_cursors ADD COLUMN IF NOT EXISTS checked_out_at TIMESTAMPTZ"
)))
.execute(pool)
.await?;
Ok(())
}
#[tracing::instrument(skip_all)]
pub async fn ensure_event_scoping_columns(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.event_types ADD COLUMN IF NOT EXISTS owner_tag_key TEXT"
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.access_tokens ADD COLUMN IF NOT EXISTS scope TEXT"
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.command_types ADD COLUMN IF NOT EXISTS owner_tag_key TEXT"
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.snapshots ADD COLUMN IF NOT EXISTS owner TEXT"
)))
.execute(pool)
.await?;
Ok(())
}
pub async fn ensure_private_field_columns(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.event_types ADD COLUMN IF NOT EXISTS private_fields JSONB NOT NULL DEFAULT '[]'"
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.command_types ADD COLUMN IF NOT EXISTS private_fields JSONB NOT NULL DEFAULT '[]'"
)))
.execute(pool)
.await?;
Ok(())
}
#[tracing::instrument(skip_all)]
pub async fn ensure_correlation_causation_columns(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.events ADD COLUMN IF NOT EXISTS metadata_correlation_id TEXT"
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.events ADD COLUMN IF NOT EXISTS metadata_causation_id TEXT"
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.commands ADD COLUMN IF NOT EXISTS metadata_correlation_id TEXT"
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.commands ADD COLUMN IF NOT EXISTS metadata_causation_id TEXT"
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE INDEX IF NOT EXISTS events_by_correlation_id ON {schema}.events \
(metadata_correlation_id) WHERE metadata_correlation_id IS NOT NULL"
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE INDEX IF NOT EXISTS commands_by_correlation_id ON {schema}.commands \
(metadata_correlation_id) WHERE metadata_correlation_id IS NOT NULL"
)))
.execute(pool)
.await?;
Ok(())
}
pub async fn ensure_event_read_token_start_from_column(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.access_tokens \
ADD COLUMN IF NOT EXISTS start_from TEXT NOT NULL DEFAULT 'beginning'"
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.access_tokens ADD COLUMN IF NOT EXISTS start_at_sequence BIGINT"
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.access_tokens ADD COLUMN IF NOT EXISTS start_at_time TIMESTAMPTZ"
)))
.execute(pool)
.await?;
Ok(())
}
#[tracing::instrument(skip_all, fields(name = %name))]
pub async fn hard_delete_bounded_context(pool: &Pool, name: &str) -> crate::error::Result<()> {
let schema = schema_ident(name);
let mut tx = pool.begin().await?;
sqlx::query(sqlx::AssertSqlSafe(format!("DROP SCHEMA {schema} CASCADE")))
.execute(&mut *tx)
.await?;
sqlx::query("DELETE FROM bounded_contexts WHERE name = $1")
.bind(name)
.execute(&mut *tx)
.await?;
tx.commit().await?;
notify_registration_changed(pool).await;
Ok(())
}
#[derive(sqlx::FromRow)]
struct RoleRow {
id: String,
external_subject: String,
name: String,
superadmin: bool,
status: String,
created_at: DateTime<Utc>,
revoked_at: Option<DateTime<Utc>>,
}
impl RoleRow {
fn into_domain(self) -> Role {
Role {
id: self.id,
external_subject: self.external_subject,
name: self.name,
superadmin: self.superadmin,
status: role_status_from_str(&self.status),
created_at: self.created_at,
revoked_at: self.revoked_at,
}
}
}
const ROLE_COLUMNS: &str = "id, external_subject, name, superadmin, status, created_at, revoked_at";
#[tracing::instrument(skip_all)]
pub async fn insert_role(pool: &Pool, role: &Role) -> crate::error::Result<()> {
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO roles ({ROLE_COLUMNS}) VALUES ($1,$2,$3,$4,$5,$6,$7)"
)))
.bind(&role.id)
.bind(&role.external_subject)
.bind(&role.name)
.bind(role.superadmin)
.bind(role_status_to_str(role.status))
.bind(role.created_at)
.bind(role.revoked_at)
.execute(pool)
.await?;
Ok(())
}
#[tracing::instrument(skip_all)]
pub async fn get_role(pool: &Pool, id: &str) -> crate::error::Result<Option<Role>> {
let row: Option<RoleRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {ROLE_COLUMNS} FROM roles WHERE id = $1"
)))
.bind(id)
.fetch_optional(pool)
.await?;
Ok(row.map(RoleRow::into_domain))
}
#[tracing::instrument(skip_all)]
pub async fn list_roles(pool: &Pool) -> crate::error::Result<Vec<Role>> {
let rows: Vec<RoleRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {ROLE_COLUMNS} FROM roles"
)))
.fetch_all(pool)
.await?;
Ok(rows.into_iter().map(RoleRow::into_domain).collect())
}
#[tracing::instrument(skip_all)]
pub async fn update_role<'e>(
executor: impl sqlx::PgExecutor<'e>,
role: &Role,
) -> crate::error::Result<()> {
sqlx::query(
"UPDATE roles SET external_subject = $1, name = $2, superadmin = $3, status = $4, \
created_at = $5, revoked_at = $6 WHERE id = $7",
)
.bind(&role.external_subject)
.bind(&role.name)
.bind(role.superadmin)
.bind(role_status_to_str(role.status))
.bind(role.created_at)
.bind(role.revoked_at)
.bind(&role.id)
.execute(executor)
.await?;
Ok(())
}
#[derive(sqlx::FromRow)]
struct BoundedContextRow {
name: String,
status: String,
created_at: DateTime<Utc>,
created_by_kind: String,
created_by_role_id: Option<String>,
template: Option<String>,
}
const BOUNDED_CONTEXT_COLUMNS: &str =
"name, status, created_at, created_by_kind, created_by_role_id, template";
#[tracing::instrument(skip_all)]
pub async fn insert_bounded_context(pool: &Pool, bc: &BoundedContext) -> crate::error::Result<()> {
let (kind, role_id) = match &bc.created_by {
ContextCreator::SystemCreator => ("system", None),
ContextCreator::SuperadminCreator { role } => ("superadmin", Some(role.id.clone())),
};
let mut tx = pool.begin().await?;
provision_bounded_context_schema(&mut tx, &bc.name).await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO bounded_contexts ({BOUNDED_CONTEXT_COLUMNS}) VALUES ($1,$2,$3,$4,$5,$6)"
)))
.bind(&bc.name)
.bind(bounded_context_status_to_str(bc.status))
.bind(bc.created_at)
.bind(kind)
.bind(role_id)
.bind(bc.template.as_ref().map(|t| t.name.clone()))
.execute(&mut *tx)
.await?;
tx.commit().await?;
notify_registration_changed(pool).await;
Ok(())
}
#[tracing::instrument(skip_all, fields(name = %name))]
pub async fn update_bounded_context_status(
pool: &Pool,
name: &str,
status: BoundedContextStatus,
) -> crate::error::Result<()> {
sqlx::query("UPDATE bounded_contexts SET status = $1 WHERE name = $2")
.bind(bounded_context_status_to_str(status))
.bind(name)
.execute(pool)
.await?;
notify_registration_changed(pool).await;
Ok(())
}
fn bounded_context_from_row(
row: BoundedContextRow,
role: Option<Role>,
template: Option<Box<BoundedContext>>,
) -> BoundedContext {
let created_by = match row.created_by_kind.as_str() {
"system" => ContextCreator::SystemCreator,
_ => ContextCreator::SuperadminCreator {
role: role.expect(
"bounded_contexts row has created_by_kind = 'superadmin' but its \
created_by_role_id doesn't resolve to a roles row - the FK should make \
this unreachable",
),
},
};
BoundedContext {
name: row.name,
status: bounded_context_status_from_str(&row.status),
created_at: row.created_at,
created_by,
template,
}
}
#[tracing::instrument(skip_all, fields(name = %name))]
pub async fn get_bounded_context(
pool: &Pool,
name: &str,
) -> crate::error::Result<Option<BoundedContext>> {
let Some(row): Option<BoundedContextRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {BOUNDED_CONTEXT_COLUMNS} FROM bounded_contexts WHERE name = $1"
)))
.bind(name)
.fetch_optional(pool)
.await?
else {
return Ok(None);
};
let role = match &row.created_by_role_id {
Some(role_id) => Some(get_role(pool, role_id).await?.expect(
"bounded_contexts.created_by_role_id references a roles row that no longer exists",
)),
None => None,
};
let template = match &row.template {
Some(template_name) => Box::pin(get_bounded_context(pool, template_name))
.await?
.map(Box::new),
None => None,
};
Ok(Some(bounded_context_from_row(row, role, template)))
}
#[tracing::instrument(skip_all)]
pub async fn list_bounded_contexts(pool: &Pool) -> crate::error::Result<Vec<BoundedContext>> {
let rows: Vec<BoundedContextRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {BOUNDED_CONTEXT_COLUMNS} FROM bounded_contexts"
)))
.fetch_all(pool)
.await?;
let roles_by_id: std::collections::HashMap<String, Role> = list_roles(pool)
.await?
.into_iter()
.map(|role| (role.id.clone(), role))
.collect();
let mut contexts_by_name = std::collections::HashMap::with_capacity(rows.len());
let mut template_names = Vec::with_capacity(rows.len());
for row in rows {
let role = row.created_by_role_id.as_ref().map(|role_id| {
roles_by_id.get(role_id).cloned().expect(
"bounded_contexts.created_by_role_id references a roles row that no longer \
exists",
)
});
let name = row.name.clone();
let template_name = row.template.clone();
contexts_by_name.insert(name.clone(), bounded_context_from_row(row, role, None));
template_names.push((name, template_name));
}
for (name, template_name) in template_names {
let Some(template_name) = template_name else {
continue;
};
let template = contexts_by_name.get(&template_name).cloned().expect(
"bounded_contexts.template references a bounded_contexts row that no longer exists",
);
contexts_by_name.get_mut(&name).unwrap().template = Some(Box::new(template));
}
Ok(contexts_by_name.into_values().collect())
}
#[tracing::instrument(skip_all, fields(name = %name))]
pub async fn set_dispatch_template(
pool: &Pool,
name: &str,
dispatch_template: &str,
) -> crate::error::Result<()> {
sqlx::query("UPDATE bounded_contexts SET dispatch_template = $1 WHERE name = $2")
.bind(dispatch_template)
.bind(name)
.execute(pool)
.await?;
Ok(())
}
#[tracing::instrument(skip_all)]
pub async fn list_dispatch_template_mappings(
pool: &Pool,
) -> crate::error::Result<Vec<(String, Option<String>)>> {
let rows: Vec<(String, Option<String>)> =
sqlx::query_as("SELECT name, dispatch_template FROM bounded_contexts")
.fetch_all(pool)
.await?;
Ok(rows)
}
fn missed_occurrence_policy_to_str(policy: MissedOccurrencePolicy) -> &'static str {
match policy {
MissedOccurrencePolicy::Skip => "skip",
MissedOccurrencePolicy::FireOnce => "fire_once",
MissedOccurrencePolicy::ReplayBacklog => "replay_backlog",
}
}
fn missed_occurrence_policy_from_str(s: &str) -> MissedOccurrencePolicy {
match s {
"fire_once" => MissedOccurrencePolicy::FireOnce,
"replay_backlog" => MissedOccurrencePolicy::ReplayBacklog,
_ => MissedOccurrencePolicy::Skip,
}
}
#[derive(sqlx::FromRow)]
struct EventTypeRow {
name: String,
schema: String,
schema_version: i64,
tag_mappings: Json<Vec<TagMapping>>,
owner_tag_key: Option<String>,
sensitive_fields: Json<Vec<SensitiveField>>,
external_creation_allowed: bool,
direct_creation_allowed: bool,
system_triggered_allowed: bool,
system_triggered_schedule: Option<String>,
missed_occurrence_policy: Option<String>,
schedule_position: Option<DateTime<Utc>>,
last_fired_at: Option<DateTime<Utc>>,
event_read_allowed: bool,
private_fields: Json<Vec<PrivateField>>,
}
impl EventTypeRow {
fn into_domain(self, bounded_context: BoundedContext) -> EventType {
EventType {
bounded_context,
name: self.name,
schema: self.schema,
schema_version: self.schema_version,
tag_mappings: self.tag_mappings.0,
owner_tag_key: self.owner_tag_key,
sensitive_fields: self.sensitive_fields.0,
private_fields: self.private_fields.0,
external_creation_allowed: self.external_creation_allowed,
direct_creation_allowed: self.direct_creation_allowed,
system_triggered_allowed: self.system_triggered_allowed,
system_triggered_schedule: self.system_triggered_schedule,
missed_occurrence_policy: self
.missed_occurrence_policy
.as_deref()
.map(missed_occurrence_policy_from_str),
schedule_position: self.schedule_position,
last_fired_at: self.last_fired_at,
event_read_allowed: self.event_read_allowed,
}
}
}
const EVENT_TYPE_COLUMNS: &str =
"name, schema, schema_version, tag_mappings, owner_tag_key, sensitive_fields, \
external_creation_allowed, direct_creation_allowed, system_triggered_allowed, \
system_triggered_schedule, missed_occurrence_policy, schedule_position, last_fired_at, \
event_read_allowed, private_fields";
#[tracing::instrument(skip_all)]
pub async fn upsert_event_type(pool: &Pool, et: &EventType) -> crate::error::Result<()> {
let schema = schema_ident(&et.bounded_context.name);
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.event_types ({EVENT_TYPE_COLUMNS}) \
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15) \
ON CONFLICT (name) DO UPDATE SET \
schema = EXCLUDED.schema, schema_version = EXCLUDED.schema_version, \
tag_mappings = EXCLUDED.tag_mappings, owner_tag_key = EXCLUDED.owner_tag_key, \
sensitive_fields = EXCLUDED.sensitive_fields, \
external_creation_allowed = EXCLUDED.external_creation_allowed, \
direct_creation_allowed = EXCLUDED.direct_creation_allowed, \
system_triggered_allowed = EXCLUDED.system_triggered_allowed, \
system_triggered_schedule = EXCLUDED.system_triggered_schedule, \
missed_occurrence_policy = EXCLUDED.missed_occurrence_policy, \
schedule_position = EXCLUDED.schedule_position, \
last_fired_at = EXCLUDED.last_fired_at, \
event_read_allowed = EXCLUDED.event_read_allowed, \
private_fields = EXCLUDED.private_fields"
)))
.bind(&et.name)
.bind(&et.schema)
.bind(et.schema_version)
.bind(Json(&et.tag_mappings))
.bind(&et.owner_tag_key)
.bind(Json(&et.sensitive_fields))
.bind(et.external_creation_allowed)
.bind(et.direct_creation_allowed)
.bind(et.system_triggered_allowed)
.bind(&et.system_triggered_schedule)
.bind(
et.missed_occurrence_policy
.map(missed_occurrence_policy_to_str),
)
.bind(et.schedule_position)
.bind(et.last_fired_at)
.bind(et.event_read_allowed)
.bind(Json(&et.private_fields))
.execute(pool)
.await?;
notify_registration_changed(pool).await;
Ok(())
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn get_event_type(
pool: &Pool,
bounded_context: &str,
name: &str,
) -> crate::error::Result<Option<EventType>> {
let Some(bc) = get_bounded_context(pool, bounded_context).await? else {
return Ok(None);
};
get_event_type_with_bc(pool, &bc, name).await
}
async fn get_event_type_with_bc(
pool: &Pool,
bc: &BoundedContext,
name: &str,
) -> crate::error::Result<Option<EventType>> {
let schema = schema_ident(&bc.name);
let row: Option<EventTypeRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {EVENT_TYPE_COLUMNS} FROM {schema}.event_types WHERE name = $1"
)))
.bind(name)
.fetch_optional(pool)
.await?;
Ok(row.map(|r| r.into_domain(bc.clone())))
}
async fn get_event_types_by_names_with_bc(
pool: &Pool,
bc: &BoundedContext,
names: &[&str],
) -> crate::error::Result<std::collections::HashMap<String, EventType>> {
if names.is_empty() {
return Ok(std::collections::HashMap::new());
}
let schema = schema_ident(&bc.name);
let rows: Vec<EventTypeRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {EVENT_TYPE_COLUMNS} FROM {schema}.event_types WHERE name = ANY($1)"
)))
.bind(names)
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|row| (row.name.clone(), row.into_domain(bc.clone())))
.collect())
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn list_scheduled_event_types(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<Vec<EventType>> {
let Some(bc) = get_bounded_context(pool, bounded_context).await? else {
return Ok(Vec::new());
};
let schema = schema_ident(bounded_context);
let rows: Vec<EventTypeRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {EVENT_TYPE_COLUMNS} FROM {schema}.event_types \
WHERE system_triggered_allowed = true ORDER BY name"
)))
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|r| r.into_domain(bc.clone()))
.collect())
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn list_event_types_for_bounded_context(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<Vec<EventType>> {
let Some(bc) = get_bounded_context(pool, bounded_context).await? else {
return Ok(Vec::new());
};
let schema = schema_ident(bounded_context);
let rows: Vec<EventTypeRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {EVENT_TYPE_COLUMNS} FROM {schema}.event_types ORDER BY name"
)))
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|r| r.into_domain(bc.clone()))
.collect())
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn list_command_types_for_bounded_context(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<Vec<CommandType>> {
let Some(bc) = get_bounded_context(pool, bounded_context).await? else {
return Ok(Vec::new());
};
let schema = schema_ident(bounded_context);
let rows: Vec<CommandTypeRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {COMMAND_TYPE_COLUMNS} FROM {schema}.command_types ORDER BY name"
)))
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|r| r.into_domain(bc.clone()))
.collect())
}
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn fire_system_event(
pool: &Pool,
projection_dispatcher: &dyn crate::plugin::ProjectionDispatcher,
event_dispatcher: &dyn crate::plugin::EventDispatcher,
broadcaster: &crate::event_store::EventBroadcaster,
event_cache: &crate::event_cache::EventCache,
bounded_context: &str,
event_type_name: &str,
occurrence_at: DateTime<Utc>,
now: DateTime<Utc>,
encryption_master_key: Option<&EncryptionMasterKey>,
) -> crate::error::Result<Option<Event>> {
let Some(unlocked_event_type) = get_event_type(pool, bounded_context, event_type_name).await?
else {
return Ok(None);
};
let Some(payload) = event_dispatcher.scheduled_payload(bounded_context, event_type_name) else {
return Ok(None);
};
let mut resolved = std::collections::HashMap::new();
resolve_encryption_keys(
pool,
bounded_context,
&unlocked_event_type.sensitive_fields,
&payload,
encryption_master_key,
&mut resolved,
)
.await?;
let schema = schema_ident(bounded_context);
let mut tx = pool.begin().await?;
let row: Option<EventTypeRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {EVENT_TYPE_COLUMNS} FROM {schema}.event_types WHERE name = $1 FOR UPDATE"
)))
.bind(event_type_name)
.fetch_optional(&mut *tx)
.await?;
let Some(row) = row else {
return Ok(None);
};
let event_type = row.into_domain(unlocked_event_type.bounded_context);
let next_seq = next_sequence(&mut *tx, bounded_context).await?;
let Some((event, new_position)) = crate::event_store::create_system_event(
&event_type,
occurrence_at,
now,
next_seq,
|| payload,
|subject_key, subject_value| {
let (key, _, data_key) = resolved
.get(&(subject_key.to_string(), subject_value.to_string()))
.expect(
"resolve_encryption_keys pre-resolved every subject sensitive_field_subjects \
named",
);
(key.clone(), data_key.clone())
},
) else {
return Ok(None);
};
let encryption_key_ids = encryption_key_ids(&event.encryption_keys, &resolved);
let sync_projections = sync_projections_for_bounded_context(pool, bounded_context).await?;
insert_event_and_update_sync_projections_in_tx(
&mut tx,
&event,
None,
projection_dispatcher,
&encryption_key_ids,
&sync_projections,
)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.event_types SET schedule_position = $1, last_fired_at = $2 WHERE name = $3"
)))
.bind(new_position)
.bind(new_position)
.bind(event_type_name)
.execute(&mut *tx)
.await?;
tx.commit().await?;
broadcaster.publish(&event);
record_event_appended(&event);
notify_event_appended(pool, &event, broadcaster.instance_id()).await;
event_cache.append(&event).await;
Ok(Some(event))
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn skip_missed_occurrences_for_event_type(
pool: &Pool,
bounded_context: &str,
event_type_name: &str,
now: DateTime<Utc>,
) -> crate::error::Result<Option<DateTime<Utc>>> {
let Some(bc) = get_bounded_context(pool, bounded_context).await? else {
return Ok(None);
};
let schema = schema_ident(bounded_context);
let mut tx = pool.begin().await?;
let row: Option<EventTypeRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {EVENT_TYPE_COLUMNS} FROM {schema}.event_types WHERE name = $1 FOR UPDATE"
)))
.bind(event_type_name)
.fetch_optional(&mut *tx)
.await?;
let Some(row) = row else {
return Ok(None);
};
let event_type = row.into_domain(bc);
let Some(new_position) = crate::event_store::skip_missed_occurrences(&event_type, now) else {
return Ok(None);
};
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.event_types SET schedule_position = $1 WHERE name = $2"
)))
.bind(new_position)
.bind(event_type_name)
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(Some(new_position))
}
#[derive(sqlx::FromRow)]
struct CommandTypeRow {
name: String,
schema: String,
schema_version: i64,
tag_mappings: Json<Vec<TagMapping>>,
owner_tag_key: Option<String>,
sensitive_fields: Json<Vec<SensitiveField>>,
rest_trigger_allowed: bool,
private_fields: Json<Vec<PrivateField>>,
}
impl CommandTypeRow {
fn into_domain(self, bounded_context: BoundedContext) -> CommandType {
CommandType {
bounded_context,
name: self.name,
schema: self.schema,
schema_version: self.schema_version,
tag_mappings: self.tag_mappings.0,
owner_tag_key: self.owner_tag_key,
sensitive_fields: self.sensitive_fields.0,
private_fields: self.private_fields.0,
rest_trigger_allowed: self.rest_trigger_allowed,
}
}
}
const COMMAND_TYPE_COLUMNS: &str = "name, schema, schema_version, tag_mappings, owner_tag_key, \
sensitive_fields, rest_trigger_allowed, private_fields";
#[tracing::instrument(skip_all)]
pub async fn upsert_command_type(pool: &Pool, ct: &CommandType) -> crate::error::Result<()> {
let schema = schema_ident(&ct.bounded_context.name);
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.command_types ({COMMAND_TYPE_COLUMNS}) \
VALUES ($1,$2,$3,$4,$5,$6,$7,$8) \
ON CONFLICT (name) DO UPDATE SET \
schema = EXCLUDED.schema, schema_version = EXCLUDED.schema_version, \
tag_mappings = EXCLUDED.tag_mappings, owner_tag_key = EXCLUDED.owner_tag_key, \
sensitive_fields = EXCLUDED.sensitive_fields, \
rest_trigger_allowed = EXCLUDED.rest_trigger_allowed, \
private_fields = EXCLUDED.private_fields"
)))
.bind(&ct.name)
.bind(&ct.schema)
.bind(ct.schema_version)
.bind(Json(&ct.tag_mappings))
.bind(&ct.owner_tag_key)
.bind(Json(&ct.sensitive_fields))
.bind(ct.rest_trigger_allowed)
.bind(Json(&ct.private_fields))
.execute(pool)
.await?;
notify_registration_changed(pool).await;
Ok(())
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn get_command_type(
pool: &Pool,
bounded_context: &str,
name: &str,
) -> crate::error::Result<Option<CommandType>> {
let Some(bc) = get_bounded_context(pool, bounded_context).await? else {
return Ok(None);
};
let schema = schema_ident(bounded_context);
let row: Option<CommandTypeRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {COMMAND_TYPE_COLUMNS} FROM {schema}.command_types WHERE name = $1"
)))
.bind(name)
.fetch_optional(pool)
.await?;
Ok(row.map(|r| r.into_domain(bc)))
}
fn encryption_key_status_from_str(s: &str) -> EncryptionKeyStatus {
match s {
"destroyed" => EncryptionKeyStatus::Destroyed,
_ => EncryptionKeyStatus::Active,
}
}
#[derive(sqlx::FromRow)]
struct EncryptionKeyRow {
id: i64,
subject_key: String,
subject_value: String,
status: String,
created_at: DateTime<Utc>,
destroyed_at: Option<DateTime<Utc>>,
wrapped_key: Option<Vec<u8>>,
wrap_nonce: Option<Vec<u8>>,
}
impl EncryptionKeyRow {
fn to_domain(&self, bounded_context: BoundedContext) -> EncryptionKey {
EncryptionKey {
bounded_context,
subject_key: self.subject_key.clone(),
subject_value: self.subject_value.clone(),
status: encryption_key_status_from_str(&self.status),
created_at: self.created_at,
destroyed_at: self.destroyed_at,
}
}
}
const ENCRYPTION_KEY_COLUMNS: &str =
"id, subject_key, subject_value, status, created_at, destroyed_at, wrapped_key, wrap_nonce";
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn get_or_create_encryption_key(
pool: &Pool,
bounded_context: &str,
subject_key: &str,
subject_value: &str,
master_key: &EncryptionMasterKey,
) -> crate::error::Result<(EncryptionKey, i64, DataKey)> {
let bc = get_bounded_context(pool, bounded_context).await?.expect(
"get_or_create_encryption_key: bounded_context row must exist for any caller reaching this",
);
let schema = schema_ident(bounded_context);
if let Some(row) =
get_active_encryption_key_row(pool, bounded_context, subject_key, subject_value).await?
{
let data_key = unwrap_row(master_key, &row)?;
return Ok((row.to_domain(bc), row.id, data_key));
}
let (data_key, wrapped, nonce) = encryption::generate_and_wrap_data_key(master_key);
let now = Utc::now();
let inserted: Option<EncryptionKeyRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.encryption_keys \
(subject_key, subject_value, status, created_at, wrapped_key, wrap_nonce) \
VALUES ($1, $2, 'active', $3, $4, $5) \
ON CONFLICT (subject_key, subject_value) WHERE status = 'active' DO NOTHING \
RETURNING {ENCRYPTION_KEY_COLUMNS}"
)))
.bind(subject_key)
.bind(subject_value)
.bind(now)
.bind(&wrapped)
.bind(&nonce)
.fetch_optional(pool)
.await?;
if let Some(row) = inserted {
return Ok((row.to_domain(bc), row.id, data_key));
}
let row = get_active_encryption_key_row(pool, bounded_context, subject_key, subject_value)
.await?
.expect(
"get_or_create_encryption_key: INSERT lost the race but no active row was found \
immediately after - a concurrent provisioner must have destroyed it in between, \
which the active-only unique index makes vanishingly unlikely within one call",
);
let data_key = unwrap_row(master_key, &row)?;
Ok((row.to_domain(bc), row.id, data_key))
}
fn unwrap_row(
master_key: &EncryptionMasterKey,
row: &EncryptionKeyRow,
) -> crate::error::Result<DataKey> {
let wrapped = row
.wrapped_key
.as_deref()
.expect("an active encryption_keys row always still has its wrapped key");
let nonce = row
.wrap_nonce
.as_deref()
.expect("an active encryption_keys row always still has its wrap nonce");
Ok(encryption::unwrap_data_key(master_key, wrapped, nonce)?)
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn resolve_encryption_keys(
pool: &Pool,
bounded_context: &str,
sensitive_fields: &[SensitiveField],
payload: &str,
master_key: Option<&EncryptionMasterKey>,
resolved: &mut std::collections::HashMap<(String, String), (EncryptionKey, i64, DataKey)>,
) -> crate::error::Result<()> {
let subjects = crate::event_store::sensitive_field_subjects(sensitive_fields, payload);
if subjects.is_empty() {
return Ok(());
}
let master_key = master_key.ok_or(encryption::Error::MasterKeyNotConfigured)?;
for (subject_key, subject_value) in subjects {
if resolved.contains_key(&(subject_key.clone(), subject_value.clone())) {
continue;
}
let provisioned = get_or_create_encryption_key(
pool,
bounded_context,
&subject_key,
&subject_value,
master_key,
)
.await?;
resolved.insert((subject_key, subject_value), provisioned);
}
Ok(())
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn resolve_data_keys_for_reading(
pool: &Pool,
bounded_context: &str,
sensitive_fields: &[SensitiveField],
payload: &str,
access_mapping: &crate::access_control::RoleAccessMapping,
master_key: Option<&EncryptionMasterKey>,
resolved: &mut std::collections::HashMap<(String, String), DataKey>,
) -> crate::error::Result<()> {
let granted_subjects: Vec<(String, String)> =
crate::event_store::sensitive_field_subjects(sensitive_fields, payload)
.into_iter()
.filter(|(_, subject_value)| {
crate::event_store::sensitive_field_is_granted(access_mapping, subject_value)
})
.collect();
if granted_subjects.is_empty() {
return Ok(());
}
let master_key = master_key.ok_or(encryption::Error::MasterKeyNotConfigured)?;
for (subject_key, subject_value) in granted_subjects {
if resolved.contains_key(&(subject_key.clone(), subject_value.clone())) {
continue;
}
if let Some(data_key) = get_active_data_key(
pool,
bounded_context,
&subject_key,
&subject_value,
master_key,
)
.await?
{
resolved.insert((subject_key, subject_value), data_key);
}
}
Ok(())
}
#[tracing::instrument(skip_all)]
pub fn encryption_key_ids(
used: &[EncryptionKey],
resolved: &std::collections::HashMap<(String, String), (EncryptionKey, i64, DataKey)>,
) -> Vec<i64> {
used.iter()
.map(|k| {
resolved
.get(&(k.subject_key.clone(), k.subject_value.clone()))
.expect("every key protect_sensitive_fields used came from resolve_encryption_keys' own resolve_key closure, so it must still be in the map that built it")
.1
})
.collect()
}
async fn get_active_encryption_key_row(
pool: &Pool,
bounded_context: &str,
subject_key: &str,
subject_value: &str,
) -> crate::error::Result<Option<EncryptionKeyRow>> {
let schema = schema_ident(bounded_context);
let row: Option<EncryptionKeyRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {ENCRYPTION_KEY_COLUMNS} FROM {schema}.encryption_keys \
WHERE subject_key = $1 AND subject_value = $2 AND status = 'active'"
)))
.bind(subject_key)
.bind(subject_value)
.fetch_optional(pool)
.await?;
Ok(row)
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn get_active_encryption_key(
pool: &Pool,
bounded_context: &str,
subject_key: &str,
subject_value: &str,
) -> crate::error::Result<Option<EncryptionKey>> {
let Some(bc) = get_bounded_context(pool, bounded_context).await? else {
return Ok(None);
};
Ok(
get_active_encryption_key_row(pool, bounded_context, subject_key, subject_value)
.await?
.map(|row| row.to_domain(bc)),
)
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn get_active_data_key(
pool: &Pool,
bounded_context: &str,
subject_key: &str,
subject_value: &str,
master_key: &EncryptionMasterKey,
) -> crate::error::Result<Option<DataKey>> {
match get_active_encryption_key_row(pool, bounded_context, subject_key, subject_value).await? {
Some(row) => Ok(Some(unwrap_row(master_key, &row)?)),
None => Ok(None),
}
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn list_active_data_keys_for_subject_value(
pool: &Pool,
bounded_context: &str,
subject_value: &str,
master_key: Option<&EncryptionMasterKey>,
) -> crate::error::Result<Vec<DataKey>> {
let schema = schema_ident(bounded_context);
let rows: Vec<EncryptionKeyRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {ENCRYPTION_KEY_COLUMNS} FROM {schema}.encryption_keys \
WHERE subject_value = $1 AND status = 'active'"
)))
.bind(subject_value)
.fetch_all(pool)
.await?;
if rows.is_empty() {
return Ok(Vec::new());
}
let master_key = master_key.ok_or(encryption::Error::MasterKeyNotConfigured)?;
rows.iter().map(|row| unwrap_row(master_key, row)).collect()
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn destroy_encryption_key(
pool: &Pool,
bounded_context: &str,
subject_key: &str,
subject_value: &str,
destroyed_at: DateTime<Utc>,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.encryption_keys SET status = 'destroyed', destroyed_at = $1, \
wrapped_key = NULL, wrap_nonce = NULL \
WHERE subject_key = $2 AND subject_value = $3 AND status = 'active'"
)))
.bind(destroyed_at)
.bind(subject_key)
.bind(subject_value)
.execute(pool)
.await?;
Ok(())
}
#[derive(sqlx::FromRow)]
struct CommandRow {
external_id: String,
command_type_name: String,
payload: String,
metadata_type: String,
metadata_version: i64,
metadata_client_id: String,
metadata_created_at: DateTime<Utc>,
metadata_correlation_id: Option<String>,
metadata_causation_id: Option<String>,
consistency_tags: Json<Vec<Tag>>,
consistency_boundary: Option<i64>,
}
impl CommandRow {
async fn into_domain(
self,
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<Command> {
let command_type = get_command_type(pool, bounded_context, &self.command_type_name)
.await?
.expect("commands row references a command_types row that no longer exists");
Ok(Command {
id: self.external_id,
bounded_context: command_type.bounded_context.clone(),
command_type,
payload: self.payload,
metadata: Metadata {
r#type: self.metadata_type,
version: self.metadata_version,
client_id: self.metadata_client_id,
created_at: self.metadata_created_at,
correlation_id: self.metadata_correlation_id,
causation_id: self.metadata_causation_id,
},
encryption_keys: Vec::new(),
consistency_tags: self.consistency_tags.0,
consistency_boundary: self.consistency_boundary,
})
}
}
const COMMAND_COLUMNS: &str = "external_id, command_type_name, payload, metadata_type, \
metadata_version, metadata_client_id, metadata_created_at, metadata_correlation_id, \
metadata_causation_id, consistency_tags, consistency_boundary";
#[tracing::instrument(skip_all)]
pub async fn insert_command(
tx: &mut Transaction<'_, Postgres>,
command: &Command,
encryption_key_ids: &[i64],
) -> crate::error::Result<i64> {
let schema = schema_ident(&command.bounded_context.name);
let (id,): (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.commands ({COMMAND_COLUMNS}) \
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) RETURNING id"
)))
.bind(&command.id)
.bind(&command.command_type.name)
.bind(&command.payload)
.bind(&command.metadata.r#type)
.bind(command.metadata.version)
.bind(&command.metadata.client_id)
.bind(command.metadata.created_at)
.bind(&command.metadata.correlation_id)
.bind(&command.metadata.causation_id)
.bind(Json(&command.consistency_tags))
.bind(command.consistency_boundary)
.fetch_one(&mut **tx)
.await?;
if !encryption_key_ids.is_empty() {
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.command_encryption_keys (command_id, encryption_key_id) \
SELECT $1, unnest($2::bigint[])"
)))
.bind(id)
.bind(encryption_key_ids)
.execute(&mut **tx)
.await?;
}
Ok(id)
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn get_command_by_id(
pool: &Pool,
bounded_context: &str,
id: i64,
) -> crate::error::Result<Option<Command>> {
let schema = schema_ident(bounded_context);
let row: Option<CommandRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {COMMAND_COLUMNS} FROM {schema}.commands WHERE id = $1"
)))
.bind(id)
.fetch_optional(pool)
.await?;
match row {
Some(row) => Ok(Some(row.into_domain(pool, bounded_context).await?)),
None => Ok(None),
}
}
async fn get_commands_by_ids_with_bc(
pool: &Pool,
bc: &BoundedContext,
ids: &[i64],
) -> crate::error::Result<std::collections::HashMap<i64, Command>> {
if ids.is_empty() {
return Ok(std::collections::HashMap::new());
}
let schema = schema_ident(&bc.name);
let rows: Vec<CommandRowWithId> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT id, {COMMAND_COLUMNS} FROM {schema}.commands WHERE id = ANY($1)"
)))
.bind(ids)
.fetch_all(pool)
.await?;
let mut distinct_type_names: Vec<&str> = Vec::new();
for row in &rows {
if !distinct_type_names.contains(&row.command_type_name.as_str()) {
distinct_type_names.push(&row.command_type_name);
}
}
let command_types = get_command_types_by_names_with_bc(pool, bc, &distinct_type_names).await?;
let mut commands = std::collections::HashMap::with_capacity(rows.len());
for row in rows {
let command_type = command_types
.get(&row.command_type_name)
.expect("commands row references a command_types row that no longer exists")
.clone();
commands.insert(
row.id,
Command {
id: row.external_id,
bounded_context: command_type.bounded_context.clone(),
command_type,
payload: row.payload,
metadata: Metadata {
r#type: row.metadata_type,
version: row.metadata_version,
client_id: row.metadata_client_id,
created_at: row.metadata_created_at,
correlation_id: row.metadata_correlation_id,
causation_id: row.metadata_causation_id,
},
encryption_keys: Vec::new(),
consistency_tags: row.consistency_tags.0,
consistency_boundary: row.consistency_boundary,
},
);
}
Ok(commands)
}
#[derive(sqlx::FromRow)]
struct CommandRowWithId {
id: i64,
external_id: String,
command_type_name: String,
payload: String,
metadata_type: String,
metadata_version: i64,
metadata_client_id: String,
metadata_created_at: DateTime<Utc>,
metadata_correlation_id: Option<String>,
metadata_causation_id: Option<String>,
consistency_tags: Json<Vec<Tag>>,
consistency_boundary: Option<i64>,
}
async fn get_command_types_by_names_with_bc(
pool: &Pool,
bc: &BoundedContext,
names: &[&str],
) -> crate::error::Result<std::collections::HashMap<String, CommandType>> {
if names.is_empty() {
return Ok(std::collections::HashMap::new());
}
let schema = schema_ident(&bc.name);
let rows: Vec<CommandTypeRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {COMMAND_TYPE_COLUMNS} FROM {schema}.command_types WHERE name = ANY($1)"
)))
.bind(names)
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|row| (row.name.clone(), row.into_domain(bc.clone())))
.collect())
}
pub async fn get_command_by_external_id(
pool: &Pool,
bounded_context: &str,
external_id: &str,
) -> crate::error::Result<Option<Command>> {
let schema = schema_ident(bounded_context);
let row: Option<CommandRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {COMMAND_COLUMNS} FROM {schema}.commands WHERE external_id = $1"
)))
.bind(external_id)
.fetch_optional(pool)
.await?;
match row {
Some(row) => Ok(Some(row.into_domain(pool, bounded_context).await?)),
None => Ok(None),
}
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn list_commands_for_bounded_context(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<Vec<Command>> {
let schema = schema_ident(bounded_context);
let rows: Vec<CommandRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {COMMAND_COLUMNS} FROM {schema}.commands"
)))
.fetch_all(pool)
.await?;
let mut commands = Vec::with_capacity(rows.len());
for row in rows {
commands.push(row.into_domain(pool, bounded_context).await?);
}
Ok(commands)
}
#[derive(sqlx::FromRow)]
struct ProjectionRow {
name: String,
schema: String,
schema_version: i64,
sync: bool,
caught_up_to: Option<i64>,
}
async fn consumed_event_types(
pool: &Pool,
bounded_context: &str,
join_table: &str,
projection_name: &str,
) -> crate::error::Result<Vec<EventType>> {
let schema = schema_ident(bounded_context);
let names: Vec<(String,)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT event_type_name FROM {schema}.{join_table} WHERE projection_name = $1"
)))
.bind(projection_name)
.fetch_all(pool)
.await?;
let mut event_types = Vec::with_capacity(names.len());
for (event_type_name,) in names {
let et = get_event_type(pool, bounded_context, &event_type_name)
.await?
.expect(
"consumed_event_types join row references an event_types row that no longer exists",
);
event_types.push(et);
}
Ok(event_types)
}
async fn replace_consumed_event_types(
executor: &mut sqlx::PgConnection,
bounded_context: &str,
join_table: &str,
projection_name: &str,
event_types: &[EventType],
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.{join_table} WHERE projection_name = $1"
)))
.bind(projection_name)
.execute(&mut *executor)
.await?;
for et in event_types {
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.{join_table} (projection_name, event_type_name) VALUES ($1,$2)"
)))
.bind(projection_name)
.bind(&et.name)
.execute(&mut *executor)
.await?;
}
Ok(())
}
const PROJECTION_COLUMNS: &str = "name, schema, schema_version, sync, caught_up_to";
#[tracing::instrument(skip_all)]
pub async fn upsert_projection(pool: &Pool, projection: &Projection) -> crate::error::Result<()> {
let schema = schema_ident(&projection.bounded_context.name);
let mut tx = pool.begin().await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.projections ({PROJECTION_COLUMNS}) \
VALUES ($1,$2,$3,$4,$5) \
ON CONFLICT (name) DO UPDATE SET \
schema = EXCLUDED.schema, schema_version = EXCLUDED.schema_version, \
sync = EXCLUDED.sync, caught_up_to = EXCLUDED.caught_up_to"
)))
.bind(&projection.name)
.bind(&projection.schema)
.bind(projection.schema_version)
.bind(projection.sync)
.bind(projection.caught_up_to)
.execute(&mut *tx)
.await?;
replace_consumed_event_types(
&mut tx,
&projection.bounded_context.name,
"projection_consumed_event_types",
&projection.name,
&projection.consumed_event_types,
)
.await?;
tx.commit().await?;
notify_registration_changed(pool).await;
Ok(())
}
async fn get_or_create_projection_state_for_update(
executor: impl sqlx::PgExecutor<'_>,
schema: &str,
projection_name: &str,
key: &str,
default_state_json: &str,
) -> crate::error::Result<(i64, String)> {
let (as_of_sequence, state): (i64, String) = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.projection_state (projection_name, key, state, updated_at) \
VALUES ($1, $2, $3, now()) \
ON CONFLICT (projection_name, key) DO UPDATE SET state = {schema}.projection_state.state \
RETURNING as_of_sequence, state"
)))
.bind(projection_name)
.bind(key)
.bind(default_state_json)
.fetch_one(executor)
.await?;
Ok((as_of_sequence, state))
}
async fn get_or_create_projection_rebuild_state_for_update(
executor: impl sqlx::PgExecutor<'_>,
schema: &str,
projection_name: &str,
key: &str,
default_state_json: &str,
) -> crate::error::Result<(i64, String)> {
let (as_of_sequence, state): (i64, String) = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.projection_rebuild_state (projection_name, status, key, state, \
updated_at) VALUES ($1, 'building', $2, $3, now()) \
ON CONFLICT (projection_name, status, key) DO UPDATE SET \
state = {schema}.projection_rebuild_state.state \
RETURNING as_of_sequence, state"
)))
.bind(projection_name)
.bind(key)
.bind(default_state_json)
.fetch_one(executor)
.await?;
Ok((as_of_sequence, state))
}
#[allow(clippy::too_many_arguments)]
async fn apply_projection_fold_update(
executor: impl sqlx::PgExecutor<'_>,
schema: &str,
table: &str,
extra_where: &str,
projection_name: &str,
key: &str,
new_state: &str,
owner_tag_key: Option<&str>,
event: &Event,
) -> crate::error::Result<()> {
let owner = owner_tag_key.and_then(|owner_tag_key| {
event
.tags
.iter()
.find(|tag| tag.key == owner_tag_key)
.and_then(|tag| tag.value.clone())
});
match owner {
Some(owner) => {
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.{table} SET state = $1, owner = $2, as_of_sequence = $3, \
updated_at = now() WHERE projection_name = $4 AND key = $5{extra_where}"
)))
.bind(new_state)
.bind(owner)
.bind(event.sequence)
.bind(projection_name)
.bind(key)
.execute(executor)
.await?;
}
None => {
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.{table} SET state = $1, as_of_sequence = $2, updated_at = now() \
WHERE projection_name = $3 AND key = $4{extra_where}"
)))
.bind(new_state)
.bind(event.sequence)
.bind(projection_name)
.bind(key)
.execute(executor)
.await?;
}
}
Ok(())
}
async fn get_or_create_snapshot_state_for_update(
executor: impl sqlx::PgExecutor<'_>,
schema: &str,
snapshot_name: &str,
tag_key: &str,
tag_value: &str,
version: u64,
default_state_json: &str,
) -> crate::error::Result<(i64, String)> {
let (as_of_sequence, state): (i64, String) = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.snapshots \
(snapshot_name, tag_key, tag_value, snapshot_version, as_of_sequence, state, updated_at) \
VALUES ($1, $2, $3, $4, -1, $5::jsonb, now()) \
ON CONFLICT (snapshot_name, tag_key, tag_value) DO UPDATE SET \
snapshot_version = CASE WHEN {schema}.snapshots.snapshot_version = $4 \
THEN {schema}.snapshots.snapshot_version ELSE $4 END, \
as_of_sequence = CASE WHEN {schema}.snapshots.snapshot_version = $4 \
THEN {schema}.snapshots.as_of_sequence ELSE -1 END, \
state = CASE WHEN {schema}.snapshots.snapshot_version = $4 \
THEN {schema}.snapshots.state ELSE $5::jsonb END \
RETURNING as_of_sequence, state::text"
)))
.bind(snapshot_name)
.bind(tag_key)
.bind(tag_value)
.bind(version as i64)
.bind(default_state_json)
.fetch_one(executor)
.await?;
Ok((as_of_sequence, state))
}
type SnapshotStateRow = (u64, i64, String, DateTime<Utc>);
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn get_snapshot_state(
pool: &Pool,
bounded_context: &str,
snapshot_name: &str,
tag_key: &str,
tag_value: &str,
) -> crate::error::Result<Option<SnapshotStateRow>> {
let schema = schema_ident(bounded_context);
let row: Option<(i64, i64, String, DateTime<Utc>)> =
sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT snapshot_version, as_of_sequence, state::text, updated_at FROM {schema}.snapshots \
WHERE snapshot_name = $1 AND tag_key = $2 AND tag_value = $3"
)))
.bind(snapshot_name)
.bind(tag_key)
.bind(tag_value)
.fetch_optional(pool)
.await?;
Ok(row.map(|(version, as_of_sequence, state, updated_at)| {
(version as u64, as_of_sequence, state, updated_at)
}))
}
pub async fn get_snapshot_state_and_owner(
pool: &Pool,
bounded_context: &str,
snapshot_name: &str,
tag_key: &str,
tag_value: &str,
) -> crate::error::Result<Option<(SnapshotStateRow, Option<String>)>> {
type RawSnapshotStateAndOwnerRow = (i64, i64, String, DateTime<Utc>, Option<String>);
let schema = schema_ident(bounded_context);
let row: Option<RawSnapshotStateAndOwnerRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT snapshot_version, as_of_sequence, state::text, updated_at, owner \
FROM {schema}.snapshots WHERE snapshot_name = $1 AND tag_key = $2 AND tag_value = $3"
)))
.bind(snapshot_name)
.bind(tag_key)
.bind(tag_value)
.fetch_optional(pool)
.await?;
Ok(
row.map(|(version, as_of_sequence, state, updated_at, owner)| {
((version as u64, as_of_sequence, state, updated_at), owner)
}),
)
}
pub struct ResolvedSnapshot {
pub state_json: String,
pub as_of_sequence: i64,
}
pub async fn resolve_snapshot_context(
pool: &Pool,
bounded_context: &str,
snapshot_dispatcher: &dyn crate::plugin::SnapshotDispatcher,
snapshot_name: &str,
consistency_tags: &[Tag],
) -> crate::error::Result<Option<ResolvedSnapshot>> {
let Some(tag_key) = snapshot_dispatcher.tag_key(bounded_context, snapshot_name) else {
return Ok(None);
};
let [only_tag] = consistency_tags else {
return Ok(None);
};
if only_tag.key != tag_key {
return Ok(None);
}
let Some(tag_value) = &only_tag.value else {
return Ok(None);
};
let version = snapshot_dispatcher
.version(bounded_context, snapshot_name)
.unwrap_or(0);
let stored =
get_snapshot_state(pool, bounded_context, snapshot_name, tag_key, tag_value).await?;
let (as_of_sequence, state_json) = match stored {
Some((stored_version, as_of_sequence, state_json, _updated_at))
if stored_version == version =>
{
(as_of_sequence, state_json)
}
_ => (
-1,
snapshot_dispatcher
.default_state(bounded_context, snapshot_name)
.unwrap_or_default(),
),
};
Ok(Some(ResolvedSnapshot {
state_json,
as_of_sequence,
}))
}
async fn list_snapshot_progress_for_bounded_context(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<std::collections::HashMap<String, i64>> {
let schema = schema_ident(bounded_context);
let rows: Vec<(String, i64)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT snapshot_name, caught_up_to FROM {schema}.snapshot_progress"
)))
.fetch_all(pool)
.await?;
Ok(rows.into_iter().collect())
}
async fn upsert_snapshot_progress(
executor: impl sqlx::PgExecutor<'_>,
schema: &str,
snapshot_name: &str,
caught_up_to: i64,
) -> crate::error::Result<()> {
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.snapshot_progress (snapshot_name, caught_up_to) VALUES ($1, $2) \
ON CONFLICT (snapshot_name) DO UPDATE SET caught_up_to = $2"
)))
.bind(snapshot_name)
.bind(caught_up_to)
.execute(executor)
.await?;
Ok(())
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn get_projection_state(
pool: &Pool,
bounded_context: &str,
projection_name: &str,
key: &str,
) -> crate::error::Result<Option<String>> {
let schema = schema_ident(bounded_context);
let row: Option<(String,)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT state FROM {schema}.projection_state WHERE projection_name = $1 AND key = $2"
)))
.bind(projection_name)
.bind(key)
.fetch_optional(pool)
.await?;
Ok(row.map(|(state,)| state))
}
pub async fn get_projection_state_and_owner(
pool: &Pool,
bounded_context: &str,
projection_name: &str,
key: &str,
) -> crate::error::Result<Option<(String, Option<String>)>> {
let schema = schema_ident(bounded_context);
let row: Option<(String, Option<String>)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT state, owner FROM {schema}.projection_state \
WHERE projection_name = $1 AND key = $2"
)))
.bind(projection_name)
.bind(key)
.fetch_optional(pool)
.await?;
Ok(row)
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn get_projection_rebuild_state(
pool: &Pool,
bounded_context: &str,
projection_name: &str,
key: &str,
) -> crate::error::Result<Option<String>> {
let schema = schema_ident(bounded_context);
let row: Option<(String,)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT state FROM {schema}.projection_rebuild_state \
WHERE projection_name = $1 AND status = 'building' AND key = $2"
)))
.bind(projection_name)
.bind(key)
.fetch_optional(pool)
.await?;
Ok(row.map(|(state,)| state))
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn get_projection(
pool: &Pool,
bounded_context: &str,
name: &str,
) -> crate::error::Result<Option<Projection>> {
let Some(bc) = get_bounded_context(pool, bounded_context).await? else {
return Ok(None);
};
let schema = schema_ident(bounded_context);
let Some(row): Option<ProjectionRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {PROJECTION_COLUMNS} FROM {schema}.projections WHERE name = $1"
)))
.bind(name)
.fetch_optional(pool)
.await?
else {
return Ok(None);
};
let consumed = consumed_event_types(
pool,
bounded_context,
"projection_consumed_event_types",
name,
)
.await?;
Ok(Some(Projection {
bounded_context: bc,
name: row.name,
schema: row.schema,
schema_version: row.schema_version,
consumed_event_types: consumed,
sync: row.sync,
caught_up_to: row.caught_up_to,
}))
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn list_projections_for_bounded_context(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<Vec<Projection>> {
let Some(bc) = get_bounded_context(pool, bounded_context).await? else {
return Ok(Vec::new());
};
let schema = schema_ident(bounded_context);
let rows: Vec<ProjectionRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {PROJECTION_COLUMNS} FROM {schema}.projections"
)))
.fetch_all(pool)
.await?;
let mut projections = Vec::with_capacity(rows.len());
for row in rows {
let consumed = consumed_event_types(
pool,
bounded_context,
"projection_consumed_event_types",
&row.name,
)
.await?;
projections.push(Projection {
bounded_context: bc.clone(),
name: row.name,
schema: row.schema,
schema_version: row.schema_version,
consumed_event_types: consumed,
sync: row.sync,
caught_up_to: row.caught_up_to,
});
}
Ok(projections)
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn sync_projections_for_bounded_context(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<Vec<Projection>> {
Ok(list_projections_for_bounded_context(pool, bounded_context)
.await?
.into_iter()
.filter(|p| p.sync)
.collect())
}
#[derive(sqlx::FromRow)]
struct ProjectionRebuildRow {
projection_name: String,
schema: String,
schema_version: i64,
sync: bool,
caught_up_to: Option<i64>,
status: String,
}
fn projection_rebuild_status_to_str(status: ProjectionRebuildStatus) -> &'static str {
match status {
ProjectionRebuildStatus::Pending => "pending",
ProjectionRebuildStatus::Building => "building",
}
}
fn projection_rebuild_status_from_str(s: &str) -> ProjectionRebuildStatus {
match s {
"building" => ProjectionRebuildStatus::Building,
_ => ProjectionRebuildStatus::Pending,
}
}
const PROJECTION_REBUILD_COLUMNS: &str =
"projection_name, schema, schema_version, sync, caught_up_to, status";
async fn rebuild_consumed_event_types(
pool: &Pool,
bounded_context: &str,
projection_name: &str,
status: ProjectionRebuildStatus,
) -> crate::error::Result<Vec<EventType>> {
let schema = schema_ident(bounded_context);
let names: Vec<(String,)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT event_type_name FROM {schema}.projection_rebuild_consumed_event_types \
WHERE projection_name = $1 AND status = $2"
)))
.bind(projection_name)
.bind(projection_rebuild_status_to_str(status))
.fetch_all(pool)
.await?;
let mut event_types = Vec::with_capacity(names.len());
for (event_type_name,) in names {
let et = get_event_type(pool, bounded_context, &event_type_name)
.await?
.expect(
"projection_rebuild_consumed_event_types join row references an event_types \
row that no longer exists",
);
event_types.push(et);
}
Ok(event_types)
}
async fn replace_rebuild_consumed_event_types(
pool: &Pool,
bounded_context: &str,
projection_name: &str,
status: ProjectionRebuildStatus,
event_types: &[EventType],
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
let status = projection_rebuild_status_to_str(status);
sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.projection_rebuild_consumed_event_types \
WHERE projection_name = $1 AND status = $2"
)))
.bind(projection_name)
.bind(status)
.execute(pool)
.await?;
for et in event_types {
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.projection_rebuild_consumed_event_types \
(projection_name, status, event_type_name) VALUES ($1,$2,$3)"
)))
.bind(projection_name)
.bind(status)
.bind(&et.name)
.execute(pool)
.await?;
}
Ok(())
}
#[tracing::instrument(skip_all)]
pub async fn upsert_projection_rebuild(
pool: &Pool,
rebuild: &ProjectionRebuild,
) -> crate::error::Result<()> {
let schema = schema_ident(&rebuild.projection.bounded_context.name);
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.projection_rebuilds ({PROJECTION_REBUILD_COLUMNS}) \
VALUES ($1,$2,$3,$4,$5,$6) \
ON CONFLICT (projection_name, status) DO UPDATE SET \
schema = EXCLUDED.schema, schema_version = EXCLUDED.schema_version, \
sync = EXCLUDED.sync, caught_up_to = EXCLUDED.caught_up_to"
)))
.bind(&rebuild.projection.name)
.bind(&rebuild.schema)
.bind(rebuild.schema_version)
.bind(rebuild.sync)
.bind(rebuild.caught_up_to)
.bind(projection_rebuild_status_to_str(rebuild.status))
.execute(pool)
.await?;
replace_rebuild_consumed_event_types(
pool,
&rebuild.projection.bounded_context.name,
&rebuild.projection.name,
rebuild.status,
&rebuild.consumed_event_types,
)
.await
}
#[tracing::instrument(skip_all)]
pub async fn transition_projection_rebuild_to_building(
pool: &Pool,
rebuild: &ProjectionRebuild,
) -> crate::error::Result<()> {
debug_assert_eq!(
rebuild.status,
ProjectionRebuildStatus::Building,
"transition_projection_rebuild_to_building's own rebuild must already carry status: \
Building - rebuild_projection's own output does"
);
let mut tx = pool.begin().await?;
let schema = schema_ident(&rebuild.projection.bounded_context.name);
let pending = projection_rebuild_status_to_str(ProjectionRebuildStatus::Pending);
sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.projection_rebuild_consumed_event_types \
WHERE projection_name = $1 AND status = $2"
)))
.bind(&rebuild.projection.name)
.bind(pending)
.execute(&mut *tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.projection_rebuilds WHERE projection_name = $1 AND status = $2"
)))
.bind(&rebuild.projection.name)
.bind(pending)
.execute(&mut *tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.projection_rebuilds ({PROJECTION_REBUILD_COLUMNS}) \
VALUES ($1,$2,$3,$4,$5,$6) \
ON CONFLICT (projection_name, status) DO UPDATE SET \
schema = EXCLUDED.schema, schema_version = EXCLUDED.schema_version, \
sync = EXCLUDED.sync, caught_up_to = EXCLUDED.caught_up_to"
)))
.bind(&rebuild.projection.name)
.bind(&rebuild.schema)
.bind(rebuild.schema_version)
.bind(rebuild.sync)
.bind(rebuild.caught_up_to)
.bind(projection_rebuild_status_to_str(rebuild.status))
.execute(&mut *tx)
.await?;
let building = projection_rebuild_status_to_str(rebuild.status);
sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.projection_rebuild_consumed_event_types \
WHERE projection_name = $1 AND status = $2"
)))
.bind(&rebuild.projection.name)
.bind(building)
.execute(&mut *tx)
.await?;
for et in &rebuild.consumed_event_types {
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.projection_rebuild_consumed_event_types \
(projection_name, status, event_type_name) VALUES ($1,$2,$3)"
)))
.bind(&rebuild.projection.name)
.bind(building)
.bind(&et.name)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
Ok(())
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn get_projection_rebuild(
pool: &Pool,
bounded_context: &str,
projection_name: &str,
status: ProjectionRebuildStatus,
) -> crate::error::Result<Option<ProjectionRebuild>> {
let Some(projection) = get_projection(pool, bounded_context, projection_name).await? else {
return Ok(None);
};
let schema = schema_ident(bounded_context);
let Some(row): Option<ProjectionRebuildRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {PROJECTION_REBUILD_COLUMNS} FROM {schema}.projection_rebuilds \
WHERE projection_name = $1 AND status = $2"
)))
.bind(projection_name)
.bind(projection_rebuild_status_to_str(status))
.fetch_optional(pool)
.await?
else {
return Ok(None);
};
debug_assert_eq!(
row.projection_name, projection_name,
"projection_rebuilds row matched the WHERE clause but its own projection_name column disagrees"
);
let consumed =
rebuild_consumed_event_types(pool, bounded_context, projection_name, status).await?;
Ok(Some(ProjectionRebuild {
projection,
schema: row.schema,
schema_version: row.schema_version,
consumed_event_types: consumed,
sync: row.sync,
caught_up_to: row.caught_up_to,
status: projection_rebuild_status_from_str(&row.status),
}))
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn delete_projection_rebuild(
pool: &Pool,
bounded_context: &str,
projection_name: &str,
status: ProjectionRebuildStatus,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
let status = projection_rebuild_status_to_str(status);
sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.projection_rebuild_consumed_event_types \
WHERE projection_name = $1 AND status = $2"
)))
.bind(projection_name)
.bind(status)
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.projection_rebuilds WHERE projection_name = $1 AND status = $2"
)))
.bind(projection_name)
.bind(status)
.execute(pool)
.await?;
Ok(())
}
#[derive(sqlx::FromRow)]
struct RoleAccessMappingRow {
role_id: String,
bounded_context: String,
level: String,
can_read_sensitive: bool,
scope: Option<String>,
status: String,
created_at: DateTime<Utc>,
revoked_at: Option<DateTime<Utc>>,
}
impl RoleAccessMappingRow {
async fn into_domain(self, pool: &Pool) -> crate::error::Result<Option<RoleAccessMapping>> {
let Some(role) = get_role(pool, &self.role_id).await? else {
return Ok(None);
};
let Some(bounded_context) = get_bounded_context(pool, &self.bounded_context).await? else {
return Ok(None);
};
Ok(Some(RoleAccessMapping {
role,
bounded_context,
level: access_level_from_str(&self.level),
can_read_sensitive: self.can_read_sensitive,
scope: self.scope,
status: role_status_from_str(&self.status),
created_at: self.created_at,
revoked_at: self.revoked_at,
}))
}
}
const ROLE_ACCESS_MAPPING_COLUMNS: &str =
"role_id, bounded_context, level, can_read_sensitive, scope, status, created_at, revoked_at";
#[tracing::instrument(skip_all)]
pub async fn insert_role_access_mapping(
pool: &Pool,
mapping: &RoleAccessMapping,
) -> crate::error::Result<()> {
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO role_access_mappings ({ROLE_ACCESS_MAPPING_COLUMNS}) \
VALUES ($1,$2,$3,$4,$5,$6,$7,$8)"
)))
.bind(&mapping.role.id)
.bind(&mapping.bounded_context.name)
.bind(access_level_to_str(mapping.level))
.bind(mapping.can_read_sensitive)
.bind(&mapping.scope)
.bind(role_status_to_str(mapping.status))
.bind(mapping.created_at)
.bind(mapping.revoked_at)
.execute(pool)
.await?;
Ok(())
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn get_active_role_access_mapping(
pool: &Pool,
role_id: &str,
bounded_context: &str,
) -> crate::error::Result<Option<RoleAccessMapping>> {
let row: Option<RoleAccessMappingRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {ROLE_ACCESS_MAPPING_COLUMNS} FROM role_access_mappings \
WHERE role_id = $1 AND bounded_context = $2 AND status = 'active'"
)))
.bind(role_id)
.bind(bounded_context)
.fetch_optional(pool)
.await?;
match row {
Some(row) => row.into_domain(pool).await,
None => Ok(None),
}
}
#[tracing::instrument(skip_all)]
pub async fn list_role_access_mappings(
pool: &Pool,
) -> crate::error::Result<Vec<RoleAccessMapping>> {
let rows: Vec<RoleAccessMappingRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {ROLE_ACCESS_MAPPING_COLUMNS} FROM role_access_mappings"
)))
.fetch_all(pool)
.await?;
let mut mappings = Vec::with_capacity(rows.len());
for row in rows {
if let Some(mapping) = row.into_domain(pool).await? {
mappings.push(mapping);
}
}
Ok(mappings)
}
#[tracing::instrument(skip_all)]
pub async fn list_active_role_access_mappings_for_role(
pool: &Pool,
role_id: &str,
) -> crate::error::Result<Vec<RoleAccessMapping>> {
let rows: Vec<RoleAccessMappingRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {ROLE_ACCESS_MAPPING_COLUMNS} FROM role_access_mappings \
WHERE role_id = $1 AND status = 'active'"
)))
.bind(role_id)
.fetch_all(pool)
.await?;
let mut mappings = Vec::with_capacity(rows.len());
for row in rows {
if let Some(mapping) = row.into_domain(pool).await? {
mappings.push(mapping);
}
}
Ok(mappings)
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn revoke_active_role_access_mapping<'e>(
executor: impl sqlx::PgExecutor<'e>,
role_id: &str,
bounded_context: &str,
revoked_at: DateTime<Utc>,
) -> crate::error::Result<()> {
sqlx::query(
"UPDATE role_access_mappings SET status = 'revoked', revoked_at = $1 \
WHERE role_id = $2 AND bounded_context = $3 AND status = 'active'",
)
.bind(revoked_at)
.bind(role_id)
.bind(bounded_context)
.execute(executor)
.await?;
Ok(())
}
#[tracing::instrument(skip_all)]
pub async fn revoke_role_and_mappings(
pool: &Pool,
role: &Role,
mappings: &[RoleAccessMapping],
) -> crate::error::Result<()> {
let mut tx = pool.begin().await?;
update_role(&mut *tx, role).await?;
for mapping in mappings {
let revoked_at = mapping
.revoked_at
.expect("revoke_role always stamps revoked_at on every mapping it returns");
revoke_active_role_access_mapping(
&mut *tx,
&role.id,
&mapping.bounded_context.name,
revoked_at,
)
.await?;
}
tx.commit().await?;
Ok(())
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn next_sequence<'e>(
executor: impl sqlx::PgExecutor<'e>,
bounded_context: &str,
) -> crate::error::Result<i64> {
let schema = schema_ident(bounded_context);
let (next,): (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.sequence SET next_value = next_value + 1 RETURNING next_value"
)))
.fetch_one(executor)
.await?;
Ok(next)
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn next_sequence_batch<'e>(
executor: impl sqlx::PgExecutor<'e>,
bounded_context: &str,
count: i64,
) -> crate::error::Result<Vec<i64>> {
if count == 0 {
return Ok(Vec::new());
}
let schema = schema_ident(bounded_context);
let (last,): (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.sequence SET next_value = next_value + $1 RETURNING next_value"
)))
.bind(count)
.fetch_one(executor)
.await?;
Ok(((last - count + 1)..=last).collect())
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn latest_sequence(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<Option<i64>> {
let schema = schema_ident(bounded_context);
let (max,): (Option<i64>,) = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT MAX(sequence) FROM {schema}.events"
)))
.fetch_one(pool)
.await?;
Ok(max)
}
#[derive(sqlx::FromRow)]
struct EventRow {
sequence: i64,
payload: String,
metadata_type: String,
metadata_version: i64,
metadata_client_id: String,
metadata_created_at: DateTime<Utc>,
metadata_correlation_id: Option<String>,
metadata_causation_id: Option<String>,
tags: Json<Vec<Tag>>,
origin_kind: String,
origin_source_content: Option<String>,
origin_source_context: Option<String>,
origin_command_id: Option<i64>,
}
async fn event_origin_from_row(
pool: &Pool,
bounded_context: &str,
kind: &str,
source_content: Option<String>,
source_context: Option<String>,
command_id: Option<i64>,
) -> crate::error::Result<EventOrigin> {
Ok(match kind {
"external_triggered" => EventOrigin::ExternalTriggered {
source_content: source_content
.expect("external_triggered event row without origin_source_content"),
source_context,
},
"directly_created" => EventOrigin::DirectlyCreated,
"system_triggered" => EventOrigin::SystemTriggered,
"command_triggered" => {
let command_id =
command_id.expect("command_triggered event row without origin_command_id");
let command = get_command_by_id(pool, bounded_context, command_id)
.await?
.expect("events row references a commands row that no longer exists");
EventOrigin::CommandTriggered {
command: Box::new(command),
}
}
other => panic!("events row has unknown origin_kind {other:?}"),
})
}
impl EventRow {
async fn into_domain(
self,
pool: &Pool,
bounded_context: BoundedContext,
event_type: EventType,
) -> crate::error::Result<Event> {
let origin = event_origin_from_row(
pool,
&bounded_context.name,
&self.origin_kind,
self.origin_source_content,
self.origin_source_context,
self.origin_command_id,
)
.await?;
Ok(Event {
bounded_context,
event_type,
payload: self.payload,
metadata: Metadata {
r#type: self.metadata_type,
version: self.metadata_version,
client_id: self.metadata_client_id,
created_at: self.metadata_created_at,
correlation_id: self.metadata_correlation_id,
causation_id: self.metadata_causation_id,
},
sequence: self.sequence,
tags: self.tags.0,
encryption_keys: Vec::new(),
origin,
})
}
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn list_events(
pool: &Pool,
bounded_context: &str,
event_type_name: &str,
) -> crate::error::Result<Vec<Event>> {
let bc = get_bounded_context(pool, bounded_context)
.await?
.expect("list_events: bounded_context row must exist for any event_type referencing it");
let et = get_event_type(pool, bounded_context, event_type_name)
.await?
.expect("list_events: event_type row must exist for any event referencing it");
let schema = schema_ident(bounded_context);
let rows: Vec<EventRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT sequence, payload, metadata_type, metadata_version, metadata_client_id, \
metadata_created_at, metadata_correlation_id, metadata_causation_id, tags, \
origin_kind, origin_source_content, origin_source_context, \
origin_command_id FROM {schema}.events WHERE event_type_name = $1 ORDER BY sequence"
)))
.bind(event_type_name)
.fetch_all(pool)
.await?;
let mut events = Vec::with_capacity(rows.len());
for row in rows {
events.push(row.into_domain(pool, bc.clone(), et.clone()).await?);
}
Ok(events)
}
#[derive(sqlx::FromRow)]
struct EventRowAnyType {
event_type_name: String,
sequence: i64,
payload: String,
metadata_type: String,
metadata_version: i64,
metadata_client_id: String,
metadata_created_at: DateTime<Utc>,
metadata_correlation_id: Option<String>,
metadata_causation_id: Option<String>,
tags: Json<Vec<Tag>>,
origin_kind: String,
origin_source_content: Option<String>,
origin_source_context: Option<String>,
origin_command_id: Option<i64>,
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn list_events_for_bounded_context(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<Vec<Event>> {
let bc = get_bounded_context(pool, bounded_context).await?.expect(
"list_events_for_bounded_context: bounded_context row must exist for any event referencing it",
);
let schema = schema_ident(bounded_context);
let rows: Vec<EventRowAnyType> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT event_type_name, sequence, payload, metadata_type, metadata_version, \
metadata_client_id, metadata_created_at, metadata_correlation_id, \
metadata_causation_id, tags, origin_kind, origin_source_content, \
origin_source_context, origin_command_id FROM {schema}.events ORDER BY sequence"
)))
.fetch_all(pool)
.await?;
let mut event_types: std::collections::HashMap<String, EventType> =
std::collections::HashMap::new();
let mut events = Vec::with_capacity(rows.len());
for row in rows {
if !event_types.contains_key(&row.event_type_name) {
let et = get_event_type(pool, bounded_context, &row.event_type_name)
.await?
.expect("events row references an event_types row that no longer exists");
event_types.insert(row.event_type_name.clone(), et);
}
let origin = event_origin_from_row(
pool,
bounded_context,
&row.origin_kind,
row.origin_source_content,
row.origin_source_context,
row.origin_command_id,
)
.await?;
events.push(Event {
bounded_context: bc.clone(),
event_type: event_types[&row.event_type_name].clone(),
payload: row.payload,
metadata: Metadata {
r#type: row.metadata_type,
version: row.metadata_version,
client_id: row.metadata_client_id,
created_at: row.metadata_created_at,
correlation_id: row.metadata_correlation_id,
causation_id: row.metadata_causation_id,
},
sequence: row.sequence,
tags: row.tags.0,
encryption_keys: Vec::new(),
origin,
});
}
Ok(events)
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn list_events_for_bounded_context_from(
pool: &Pool,
bounded_context: &str,
after_sequence: i64,
) -> crate::error::Result<Vec<Event>> {
let bc = get_bounded_context(pool, bounded_context).await?.expect(
"list_events_for_bounded_context_from: bounded_context row must exist for any event referencing it",
);
let schema = schema_ident(bounded_context);
let rows: Vec<EventRowAnyType> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT event_type_name, sequence, payload, metadata_type, metadata_version, \
metadata_client_id, metadata_created_at, metadata_correlation_id, \
metadata_causation_id, tags, origin_kind, origin_source_content, \
origin_source_context, origin_command_id FROM {schema}.events \
WHERE sequence > $1 ORDER BY sequence"
)))
.bind(after_sequence)
.fetch_all(pool)
.await?;
let mut event_types: std::collections::HashMap<String, EventType> =
std::collections::HashMap::new();
let mut events = Vec::with_capacity(rows.len());
for row in rows {
if !event_types.contains_key(&row.event_type_name) {
let et = get_event_type(pool, bounded_context, &row.event_type_name)
.await?
.expect("events row references an event_types row that no longer exists");
event_types.insert(row.event_type_name.clone(), et);
}
let origin = event_origin_from_row(
pool,
bounded_context,
&row.origin_kind,
row.origin_source_content,
row.origin_source_context,
row.origin_command_id,
)
.await?;
events.push(Event {
bounded_context: bc.clone(),
event_type: event_types[&row.event_type_name].clone(),
payload: row.payload,
metadata: Metadata {
r#type: row.metadata_type,
version: row.metadata_version,
client_id: row.metadata_client_id,
created_at: row.metadata_created_at,
correlation_id: row.metadata_correlation_id,
causation_id: row.metadata_causation_id,
},
sequence: row.sequence,
tags: row.tags.0,
encryption_keys: Vec::new(),
origin,
});
}
Ok(events)
}
pub async fn list_events_for_bounded_context_from_limited(
pool: &Pool,
bounded_context: &str,
after_sequence: i64,
limit: i64,
) -> crate::error::Result<Vec<Event>> {
let bc = get_bounded_context(pool, bounded_context).await?.expect(
"list_events_for_bounded_context_from_limited: bounded_context row must exist for any event referencing it",
);
let schema = schema_ident(bounded_context);
let rows: Vec<EventRowAnyType> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT event_type_name, sequence, payload, metadata_type, metadata_version, \
metadata_client_id, metadata_created_at, metadata_correlation_id, \
metadata_causation_id, tags, origin_kind, origin_source_content, \
origin_source_context, origin_command_id FROM {schema}.events \
WHERE sequence > $1 ORDER BY sequence LIMIT $2"
)))
.bind(after_sequence)
.bind(limit)
.fetch_all(pool)
.await?;
let mut event_types: std::collections::HashMap<String, EventType> =
std::collections::HashMap::new();
let mut events = Vec::with_capacity(rows.len());
for row in rows {
if !event_types.contains_key(&row.event_type_name) {
let et = get_event_type(pool, bounded_context, &row.event_type_name)
.await?
.expect("events row references an event_types row that no longer exists");
event_types.insert(row.event_type_name.clone(), et);
}
let origin = event_origin_from_row(
pool,
bounded_context,
&row.origin_kind,
row.origin_source_content,
row.origin_source_context,
row.origin_command_id,
)
.await?;
events.push(Event {
bounded_context: bc.clone(),
event_type: event_types[&row.event_type_name].clone(),
payload: row.payload,
metadata: Metadata {
r#type: row.metadata_type,
version: row.metadata_version,
client_id: row.metadata_client_id,
created_at: row.metadata_created_at,
correlation_id: row.metadata_correlation_id,
causation_id: row.metadata_causation_id,
},
sequence: row.sequence,
tags: row.tags.0,
encryption_keys: Vec::new(),
origin,
});
}
Ok(events)
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn list_events_for_bounded_context_matching_tags(
pool: &Pool,
bounded_context: &str,
tags: &[Tag],
after_sequence: Option<i64>,
) -> crate::error::Result<Vec<Event>> {
if tags.is_empty() {
return Ok(Vec::new());
}
let bc = get_bounded_context(pool, bounded_context).await?.expect(
"list_events_for_bounded_context_matching_tags: bounded_context row must exist for any event referencing it",
);
list_events_for_bounded_context_matching_tags_with_bc(pool, &bc, tags, after_sequence, None)
.await
}
async fn list_events_for_bounded_context_matching_tags_with_bc(
pool: &Pool,
bc: &BoundedContext,
tags: &[Tag],
after_sequence: Option<i64>,
known_event_types: Option<&std::collections::HashMap<String, EventType>>,
) -> crate::error::Result<Vec<Event>> {
if tags.is_empty() {
return Ok(Vec::new());
}
let bounded_context = bc.name.as_str();
let schema = schema_ident(bounded_context);
let tag_literals: Vec<String> = tags
.iter()
.map(|t| {
serde_json::to_string(std::slice::from_ref(t)).expect("Tag serialisation is infallible")
})
.collect();
let tag_clause = (1..=tag_literals.len())
.map(|i| format!("tags @> ${i}::jsonb"))
.collect::<Vec<_>>()
.join(" OR ");
let sequence_param = tag_literals.len() + 1;
let where_clause = match after_sequence {
Some(_) => format!("({tag_clause}) AND sequence > ${sequence_param}"),
None => tag_clause,
};
let sql = format!(
"SELECT event_type_name, sequence, payload, metadata_type, metadata_version, \
metadata_client_id, metadata_created_at, metadata_correlation_id, \
metadata_causation_id, tags, origin_kind, origin_source_content, \
origin_source_context, origin_command_id FROM {schema}.events \
WHERE {where_clause} ORDER BY sequence"
);
let mut query = sqlx::query_as::<_, EventRowAnyType>(sqlx::AssertSqlSafe(sql));
for literal in tag_literals {
query = query.bind(literal);
}
if let Some(after_sequence) = after_sequence {
query = query.bind(after_sequence);
}
let rows: Vec<EventRowAnyType> = query.fetch_all(pool).await?;
let row_count = rows.len();
let row_loop_started = std::time::Instant::now();
let mut event_types: std::collections::HashMap<String, EventType> =
std::collections::HashMap::new();
let mut missing_type_names: Vec<&str> = Vec::new();
for row in &rows {
if event_types.contains_key(&row.event_type_name) {
continue;
}
match known_event_types.and_then(|m| m.get(&row.event_type_name)) {
Some(et) => {
event_types.insert(row.event_type_name.clone(), et.clone());
}
None => {
if !missing_type_names.contains(&row.event_type_name.as_str()) {
missing_type_names.push(&row.event_type_name);
}
}
}
}
let event_type_lookup_started = std::time::Instant::now();
event_types.extend(get_event_types_by_names_with_bc(pool, bc, &missing_type_names).await?);
let event_type_lookup_elapsed = event_type_lookup_started.elapsed();
let origin_lookup_started = std::time::Instant::now();
let mut command_ids: Vec<i64> = Vec::new();
for row in &rows {
if row.origin_kind == "command_triggered" {
if let Some(id) = row.origin_command_id {
if !command_ids.contains(&id) {
command_ids.push(id);
}
}
}
}
let commands_by_id = get_commands_by_ids_with_bc(pool, bc, &command_ids).await?;
let mut origins = Vec::with_capacity(rows.len());
for row in &rows {
origins.push(match row.origin_kind.as_str() {
"external_triggered" => EventOrigin::ExternalTriggered {
source_content: row
.origin_source_content
.clone()
.expect("external_triggered event row without origin_source_content"),
source_context: row.origin_source_context.clone(),
},
"directly_created" => EventOrigin::DirectlyCreated,
"system_triggered" => EventOrigin::SystemTriggered,
"command_triggered" => {
let command_id = row
.origin_command_id
.expect("command_triggered event row without origin_command_id");
let command = commands_by_id
.get(&command_id)
.expect("events row references a commands row that no longer exists");
EventOrigin::CommandTriggered {
command: Box::new(command.clone()),
}
}
other => panic!("events row has unknown origin_kind {other:?}"),
});
}
let origin_lookup_elapsed = origin_lookup_started.elapsed();
let mut events = Vec::with_capacity(rows.len());
for (row, origin) in rows.into_iter().zip(origins) {
events.push(Event {
bounded_context: bc.clone(),
event_type: event_types[&row.event_type_name].clone(),
payload: row.payload,
metadata: Metadata {
r#type: row.metadata_type,
version: row.metadata_version,
client_id: row.metadata_client_id,
created_at: row.metadata_created_at,
correlation_id: row.metadata_correlation_id,
causation_id: row.metadata_causation_id,
},
sequence: row.sequence,
tags: row.tags.0,
encryption_keys: Vec::new(),
origin,
});
}
tracing::info!(
row_count,
row_loop_us = row_loop_started.elapsed().as_micros(),
event_type_lookup_us = event_type_lookup_elapsed.as_micros(),
origin_lookup_us = origin_lookup_elapsed.as_micros(),
"delta query row-processing loop"
);
Ok(events)
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn get_event_by_sequence(
pool: &Pool,
bounded_context: &str,
sequence: i64,
) -> crate::error::Result<Option<Event>> {
let Some(bc) = get_bounded_context(pool, bounded_context).await? else {
return Ok(None);
};
let schema = schema_ident(bounded_context);
let Some(row): Option<EventRowAnyType> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT event_type_name, sequence, payload, metadata_type, metadata_version, \
metadata_client_id, metadata_created_at, metadata_correlation_id, \
metadata_causation_id, tags, origin_kind, origin_source_content, \
origin_source_context, origin_command_id FROM {schema}.events WHERE sequence = $1"
)))
.bind(sequence)
.fetch_optional(pool)
.await?
else {
return Ok(None);
};
let et = get_event_type(pool, bounded_context, &row.event_type_name)
.await?
.expect("events row references an event_types row that no longer exists");
let origin = event_origin_from_row(
pool,
bounded_context,
&row.origin_kind,
row.origin_source_content,
row.origin_source_context,
row.origin_command_id,
)
.await?;
Ok(Some(Event {
bounded_context: bc,
event_type: et,
payload: row.payload,
metadata: Metadata {
r#type: row.metadata_type,
version: row.metadata_version,
client_id: row.metadata_client_id,
created_at: row.metadata_created_at,
correlation_id: row.metadata_correlation_id,
causation_id: row.metadata_causation_id,
},
sequence: row.sequence,
tags: row.tags.0,
encryption_keys: Vec::new(),
origin,
}))
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn list_recent_events_for_bounded_context(
pool: &Pool,
bounded_context: &str,
limit: usize,
) -> crate::error::Result<Vec<Event>> {
let bc = get_bounded_context(pool, bounded_context).await?.expect(
"list_recent_events_for_bounded_context: bounded_context row must exist for any event \
referencing it",
);
let schema = schema_ident(bounded_context);
let rows: Vec<EventRowAnyType> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT event_type_name, sequence, payload, metadata_type, metadata_version, \
metadata_client_id, metadata_created_at, metadata_correlation_id, \
metadata_causation_id, tags, origin_kind, origin_source_content, \
origin_source_context, origin_command_id FROM {schema}.events \
ORDER BY sequence DESC LIMIT $1"
)))
.bind(limit as i64)
.fetch_all(pool)
.await?;
let mut event_types: std::collections::HashMap<String, EventType> =
std::collections::HashMap::new();
let mut events = Vec::with_capacity(rows.len());
for row in rows {
if !event_types.contains_key(&row.event_type_name) {
let et = get_event_type(pool, bounded_context, &row.event_type_name)
.await?
.expect("events row references an event_types row that no longer exists");
event_types.insert(row.event_type_name.clone(), et);
}
let origin = event_origin_from_row(
pool,
bounded_context,
&row.origin_kind,
row.origin_source_content,
row.origin_source_context,
row.origin_command_id,
)
.await?;
events.push(Event {
bounded_context: bc.clone(),
event_type: event_types[&row.event_type_name].clone(),
payload: row.payload,
metadata: Metadata {
r#type: row.metadata_type,
version: row.metadata_version,
client_id: row.metadata_client_id,
created_at: row.metadata_created_at,
correlation_id: row.metadata_correlation_id,
causation_id: row.metadata_causation_id,
},
sequence: row.sequence,
tags: row.tags.0,
encryption_keys: Vec::new(),
origin,
});
}
events.reverse();
Ok(events)
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn list_events_from(
pool: &Pool,
bounded_context: &str,
event_type_name: &str,
after_sequence: i64,
) -> crate::error::Result<Vec<Event>> {
let bc = get_bounded_context(pool, bounded_context).await?.expect(
"list_events_from: bounded_context row must exist for any event_type referencing it",
);
let et = get_event_type(pool, bounded_context, event_type_name)
.await?
.expect("list_events_from: event_type row must exist for any event referencing it");
let schema = schema_ident(bounded_context);
let rows: Vec<EventRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT sequence, payload, metadata_type, metadata_version, metadata_client_id, \
metadata_created_at, metadata_correlation_id, metadata_causation_id, tags, \
origin_kind, origin_source_content, origin_source_context, \
origin_command_id FROM {schema}.events WHERE event_type_name = $1 AND sequence > $2 \
ORDER BY sequence"
)))
.bind(event_type_name)
.bind(after_sequence)
.fetch_all(pool)
.await?;
let mut events = Vec::with_capacity(rows.len());
for row in rows {
events.push(row.into_domain(pool, bc.clone(), et.clone()).await?);
}
Ok(events)
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn list_events_cached(
pool: &Pool,
cache: &crate::event_cache::EventCache,
bounded_context: &str,
event_type_name: &str,
after_sequence: i64,
) -> crate::error::Result<Vec<Event>> {
match cache
.try_events_after(pool, bounded_context, after_sequence)
.await?
{
Some(events) => Ok(events
.into_iter()
.filter(|e| e.event_type.name == event_type_name)
.collect()),
None => list_events_from(pool, bounded_context, event_type_name, after_sequence).await,
}
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn list_events_for_bounded_context_cached(
pool: &Pool,
cache: &crate::event_cache::EventCache,
bounded_context: &str,
after_sequence: i64,
) -> crate::error::Result<Vec<Event>> {
match cache
.try_events_after(pool, bounded_context, after_sequence)
.await?
{
Some(events) => Ok(events),
None if after_sequence < 0 => list_events_for_bounded_context(pool, bounded_context).await,
None => list_events_for_bounded_context_from(pool, bounded_context, after_sequence).await,
}
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn list_events_for_bounded_context_matching_tags_cached(
pool: &Pool,
cache: &crate::event_cache::EventCache,
bounded_context: &str,
tags: &[Tag],
after_sequence: Option<i64>,
) -> crate::error::Result<Vec<Event>> {
match cache
.try_events_matching_tags(pool, bounded_context, tags)
.await?
{
Some(events) => Ok(events
.into_iter()
.filter(|e| e.sequence > after_sequence.unwrap_or(-1))
.collect()),
None => {
list_events_for_bounded_context_matching_tags(
pool,
bounded_context,
tags,
after_sequence,
)
.await
}
}
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn get_event_by_sequence_cached(
pool: &Pool,
cache: &crate::event_cache::EventCache,
bounded_context: &str,
sequence: i64,
) -> crate::error::Result<Option<Event>> {
match cache
.try_event_by_sequence(pool, bounded_context, sequence)
.await?
{
Some(event) => Ok(Some(event)),
None => get_event_by_sequence(pool, bounded_context, sequence).await,
}
}
#[tracing::instrument(skip_all)]
pub async fn insert_event<'e>(
executor: impl sqlx::PgExecutor<'e>,
event: &Event,
command_id: Option<i64>,
) -> crate::error::Result<()> {
let (origin_kind, source_content, source_context) = match &event.origin {
EventOrigin::ExternalTriggered {
source_content,
source_context,
} => {
debug_assert!(command_id.is_none());
(
"external_triggered",
Some(source_content.clone()),
source_context.clone(),
)
}
EventOrigin::DirectlyCreated => {
debug_assert!(command_id.is_none());
("directly_created", None, None)
}
EventOrigin::SystemTriggered => {
debug_assert!(command_id.is_none());
("system_triggered", None, None)
}
EventOrigin::CommandTriggered { .. } => {
debug_assert!(
command_id.is_some(),
"insert_event: a CommandTriggered origin needs command_id - call \
insert_command(pool, &result.command) first and pass its returned id"
);
("command_triggered", None, None)
}
};
let schema = schema_ident(&event.bounded_context.name);
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.events (sequence, event_type_name, payload, metadata_type, \
metadata_version, metadata_client_id, metadata_created_at, metadata_correlation_id, \
metadata_causation_id, tags, \
origin_kind, origin_source_content, origin_source_context, origin_command_id) \
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)"
)))
.bind(event.sequence)
.bind(&event.event_type.name)
.bind(&event.payload)
.bind(&event.metadata.r#type)
.bind(event.metadata.version)
.bind(&event.metadata.client_id)
.bind(event.metadata.created_at)
.bind(&event.metadata.correlation_id)
.bind(&event.metadata.causation_id)
.bind(Json(&event.tags))
.bind(origin_kind)
.bind(source_content)
.bind(source_context)
.bind(command_id)
.execute(executor)
.await?;
Ok(())
}
#[tracing::instrument(skip_all)]
pub async fn insert_event_and_update_sync_projections(
pool: &Pool,
event: &Event,
command_id: Option<i64>,
dispatcher: &dyn crate::plugin::ProjectionDispatcher,
encryption_key_ids: &[i64],
broadcaster: &crate::event_store::EventBroadcaster,
event_cache: &crate::event_cache::EventCache,
) -> crate::error::Result<()> {
let sync_projections =
sync_projections_for_bounded_context(pool, &event.bounded_context.name).await?;
let mut tx = pool.begin().await?;
insert_event_and_update_sync_projections_in_tx(
&mut tx,
event,
command_id,
dispatcher,
encryption_key_ids,
&sync_projections,
)
.await?;
tx.commit().await?;
broadcaster.publish(event);
record_event_appended(event);
notify_event_appended(pool, event, broadcaster.instance_id()).await;
event_cache.append(event).await;
Ok(())
}
#[tracing::instrument(skip_all)]
pub async fn insert_event_and_update_sync_projections_in_tx(
tx: &mut Transaction<'_, Postgres>,
event: &Event,
command_id: Option<i64>,
dispatcher: &dyn crate::plugin::ProjectionDispatcher,
encryption_key_ids: &[i64],
sync_projections: &[Projection],
) -> crate::error::Result<()> {
let bounded_context = &event.bounded_context.name;
let schema = schema_ident(bounded_context);
insert_event(&mut **tx, event, command_id).await?;
if !encryption_key_ids.is_empty() {
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.event_encryption_keys (event_sequence, encryption_key_id) \
SELECT $1, unnest($2::bigint[])"
)))
.bind(event.sequence)
.bind(encryption_key_ids)
.execute(&mut **tx)
.await?;
}
for projection in sync_projections {
let keys = dispatcher
.keys(bounded_context, &projection.name, event)
.unwrap_or_default();
let default_state_json = dispatcher
.default_state(bounded_context, &projection.name)
.unwrap_or_default();
let owner_tag_key = dispatcher
.owner_tag_key(bounded_context, &projection.name)
.flatten();
for key in &keys {
let (as_of_sequence, current_state) = get_or_create_projection_state_for_update(
&mut **tx,
&schema,
&projection.name,
key,
&default_state_json,
)
.await?;
if as_of_sequence >= event.sequence {
continue;
}
let new_state = match dispatcher.project(
bounded_context,
&projection.name,
¤t_state,
event,
key,
) {
Some(result) => result?,
None => current_state,
};
apply_projection_fold_update(
&mut **tx,
&schema,
"projection_state",
"",
&projection.name,
key,
&new_state,
owner_tag_key,
event,
)
.await?;
}
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.projections SET caught_up_to = $1 WHERE name = $2"
)))
.bind(event.sequence)
.bind(&projection.name)
.execute(&mut **tx)
.await?;
}
Ok(())
}
pub struct DedupeCursor<'a> {
pub partition_key: &'a str,
pub sequence: i64,
}
#[derive(Debug)]
pub enum CreateExternalEventOutcome {
Created(Box<Event>),
Redelivered,
}
async fn highest_dedupe_sequence<'e>(
executor: impl sqlx::PgExecutor<'e>,
schema: &str,
adapter_id: &str,
partition_key: &str,
) -> crate::error::Result<Option<i64>> {
let row: Option<(i64,)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT last_sequence FROM {schema}.external_message_cursors \
WHERE adapter_id = $1 AND partition_key = $2"
)))
.bind(adapter_id)
.bind(partition_key)
.fetch_optional(executor)
.await?;
Ok(row.map(|(seq,)| seq))
}
async fn advance_dedupe_watermark<'e>(
executor: impl sqlx::PgExecutor<'e>,
schema: &str,
adapter_id: &str,
partition_key: &str,
sequence: i64,
now: DateTime<Utc>,
) -> crate::error::Result<()> {
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.external_message_cursors \
(adapter_id, partition_key, last_sequence, updated_at) \
VALUES ($1, $2, $3, $4) \
ON CONFLICT (adapter_id, partition_key) \
DO UPDATE SET last_sequence = EXCLUDED.last_sequence, updated_at = EXCLUDED.updated_at"
)))
.bind(adapter_id)
.bind(partition_key)
.bind(sequence)
.bind(now)
.execute(executor)
.await?;
Ok(())
}
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(skip_all)]
pub async fn create_and_insert_external_event(
pool: &Pool,
projection_dispatcher: &dyn crate::plugin::ProjectionDispatcher,
broadcaster: &crate::event_store::EventBroadcaster,
event_cache: &crate::event_cache::EventCache,
adapter: &ExternalEventToken,
payload: String,
source_content: String,
source_context: Option<String>,
correlation_id: Option<String>,
causation_id: Option<String>,
dedupe: Option<DedupeCursor<'_>>,
now: DateTime<Utc>,
encryption_master_key: Option<&EncryptionMasterKey>,
) -> crate::error::Result<CreateExternalEventOutcome> {
let bounded_context_name = adapter.event_type.bounded_context.name.clone();
let schema = schema_ident(&bounded_context_name);
let mut resolved = std::collections::HashMap::new();
resolve_encryption_keys(
pool,
&bounded_context_name,
&adapter.event_type.sensitive_fields,
&payload,
encryption_master_key,
&mut resolved,
)
.await?;
let mut tx = pool.begin().await?;
let next_seq = next_sequence(&mut *tx, &bounded_context_name).await?;
if let Some(cursor) = &dedupe {
let watermark =
highest_dedupe_sequence(&mut *tx, &schema, &adapter.id, cursor.partition_key).await?;
if watermark.is_some_and(|w| cursor.sequence <= w) {
return Ok(CreateExternalEventOutcome::Redelivered);
}
}
let event = crate::event_store::create_external_event(
adapter,
payload,
source_content,
source_context,
correlation_id,
causation_id,
next_seq,
now,
|subject_key, subject_value| {
let (key, _, data_key) = resolved
.get(&(subject_key.to_string(), subject_value.to_string()))
.expect(
"resolve_encryption_keys pre-resolved every subject sensitive_field_subjects \
named",
);
(key.clone(), data_key.clone())
},
)?;
let encryption_key_ids = encryption_key_ids(&event.encryption_keys, &resolved);
let sync_projections =
sync_projections_for_bounded_context(pool, &bounded_context_name).await?;
insert_event_and_update_sync_projections_in_tx(
&mut tx,
&event,
None,
projection_dispatcher,
&encryption_key_ids,
&sync_projections,
)
.await?;
if let Some(cursor) = &dedupe {
advance_dedupe_watermark(
&mut *tx,
&schema,
&adapter.id,
cursor.partition_key,
cursor.sequence,
now,
)
.await?;
}
tx.commit().await?;
broadcaster.publish(&event);
record_event_appended(&event);
notify_event_appended(pool, &event, broadcaster.instance_id()).await;
event_cache.append(&event).await;
Ok(CreateExternalEventOutcome::Created(Box::new(event)))
}
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(skip_all)]
pub async fn create_and_insert_direct_event(
pool: &Pool,
projection_dispatcher: &dyn crate::plugin::ProjectionDispatcher,
broadcaster: &crate::event_store::EventBroadcaster,
event_cache: &crate::event_cache::EventCache,
adapter: &DirectCreationToken,
payload: String,
correlation_id: Option<String>,
causation_id: Option<String>,
now: DateTime<Utc>,
encryption_master_key: Option<&EncryptionMasterKey>,
) -> crate::error::Result<Event> {
let bounded_context_name = adapter.event_type.bounded_context.name.clone();
let mut resolved = std::collections::HashMap::new();
resolve_encryption_keys(
pool,
&bounded_context_name,
&adapter.event_type.sensitive_fields,
&payload,
encryption_master_key,
&mut resolved,
)
.await?;
let mut tx = pool.begin().await?;
let next_seq = next_sequence(&mut *tx, &bounded_context_name).await?;
let event = crate::event_store::create_direct_event(
adapter,
payload,
correlation_id,
causation_id,
next_seq,
now,
|subject_key, subject_value| {
let (key, _, data_key) = resolved
.get(&(subject_key.to_string(), subject_value.to_string()))
.expect(
"resolve_encryption_keys pre-resolved every subject sensitive_field_subjects \
named",
);
(key.clone(), data_key.clone())
},
)?;
let encryption_key_ids = encryption_key_ids(&event.encryption_keys, &resolved);
let sync_projections =
sync_projections_for_bounded_context(pool, &bounded_context_name).await?;
insert_event_and_update_sync_projections_in_tx(
&mut tx,
&event,
None,
projection_dispatcher,
&encryption_key_ids,
&sync_projections,
)
.await?;
tx.commit().await?;
broadcaster.publish(&event);
record_event_appended(&event);
notify_event_appended(pool, &event, broadcaster.instance_id()).await;
event_cache.append(&event).await;
Ok(event)
}
#[derive(Debug)]
pub enum SubmitCommandOutcome {
Accepted {
command: Box<Command>,
events: Vec<Event>,
},
Rejected {
reason: String,
kind: String,
matching_events: Vec<Event>,
},
Deduplicated { triggered_event_sequences: Vec<i64> },
}
pub struct SnapshotContext<'a> {
pub state_json: &'a str,
pub as_of_sequence: i64,
}
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(
skip_all,
fields(
bounded_context = %command_type.bounded_context.name,
command_type = %command_type.name,
)
)]
pub async fn submit_command(
pool: &Pool,
dispatcher: &dyn crate::plugin::CommandDispatcher,
projection_dispatcher: &dyn crate::plugin::ProjectionDispatcher,
broadcaster: &crate::event_store::EventBroadcaster,
event_cache: &crate::event_cache::EventCache,
command_type: &CommandType,
payload: &str,
client_id: &str,
correlation_id: Option<&str>,
causation_id: Option<&str>,
bounded_context_events: &[Event],
consistency_tags: &[Tag],
matching_events: &[Event],
initial_decision: crate::shared::CommandDecision,
encryption_master_key: Option<&EncryptionMasterKey>,
now: DateTime<Utc>,
snapshot: Option<SnapshotContext<'_>>,
idempotency_key: Option<&str>,
) -> crate::error::Result<SubmitCommandOutcome> {
let bounded_context_name = command_type.bounded_context.name.clone();
let (event_types_by_name, resolved) = warm_up_event_types_and_encryption_keys(
pool,
&bounded_context_name,
command_type,
payload,
&initial_decision,
encryption_master_key,
)
.await?;
let mut tx = pool.begin().await?;
let locked_highest = lock_bounded_context_sequence(&mut tx, &bounded_context_name).await?;
let sync_projections =
sync_projections_for_bounded_context(pool, &bounded_context_name).await?;
let outcome = submit_one_command_in_tx(
&mut tx,
pool,
dispatcher,
projection_dispatcher,
command_type,
payload,
client_id,
correlation_id,
causation_id,
bounded_context_events,
consistency_tags,
matching_events,
initial_decision,
encryption_master_key,
now,
snapshot,
idempotency_key,
locked_highest,
&sync_projections,
&[],
event_types_by_name,
resolved,
)
.await?;
tx.commit().await?;
broadcast_appended_events(pool, broadcaster, event_cache, &outcome).await;
Ok(outcome)
}
pub async fn warm_up_event_types_and_encryption_keys(
pool: &Pool,
bounded_context_name: &str,
command_type: &CommandType,
payload: &str,
initial_decision: &crate::shared::CommandDecision,
encryption_master_key: Option<&EncryptionMasterKey>,
) -> crate::error::Result<(
std::collections::HashMap<String, EventType>,
std::collections::HashMap<(String, String), (EncryptionKey, i64, DataKey)>,
)> {
let mut event_types_by_name: std::collections::HashMap<String, EventType> =
std::collections::HashMap::new();
let mut resolved = std::collections::HashMap::new();
if let crate::shared::CommandDecision::Accepted { events } = initial_decision {
for spec in events {
if let std::collections::hash_map::Entry::Vacant(entry) =
event_types_by_name.entry(spec.event_type.clone())
{
if let Some(et) =
get_event_type(pool, bounded_context_name, &spec.event_type).await?
{
entry.insert(et);
}
}
}
resolve_encryption_keys(
pool,
bounded_context_name,
&command_type.sensitive_fields,
payload,
encryption_master_key,
&mut resolved,
)
.await?;
for spec in events {
if let Some(event_type) = event_types_by_name.get(&spec.event_type) {
let spec_payload = spec.payload.to_string();
resolve_encryption_keys(
pool,
bounded_context_name,
&event_type.sensitive_fields,
&spec_payload,
encryption_master_key,
&mut resolved,
)
.await?;
}
}
}
Ok((event_types_by_name, resolved))
}
async fn lock_bounded_context_sequence(
tx: &mut Transaction<'_, Postgres>,
bounded_context: &str,
) -> crate::error::Result<i64> {
let schema = schema_ident(bounded_context);
let (locked_highest,): (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT next_value FROM {schema}.sequence FOR UPDATE"
)))
.fetch_one(&mut **tx)
.await?;
Ok(locked_highest)
}
pub async fn broadcast_appended_events(
pool: &Pool,
broadcaster: &crate::event_store::EventBroadcaster,
event_cache: &crate::event_cache::EventCache,
outcome: &SubmitCommandOutcome,
) {
let SubmitCommandOutcome::Accepted { events, .. } = outcome else {
return;
};
for event in events {
broadcaster.publish(event);
record_event_appended(event);
notify_event_appended(pool, event, broadcaster.instance_id()).await;
event_cache.append(event).await;
}
}
struct AcceptedDecision {
event_specs: Vec<crate::shared::EventSpec>,
final_bounded_context_events: Vec<Event>,
event_types_by_name: std::collections::HashMap<String, EventType>,
}
enum DecideOutcome {
Terminal(SubmitCommandOutcome),
Accepted(AcceptedDecision),
}
#[allow(clippy::too_many_arguments)]
async fn decide_command_in_tx(
pool: &Pool,
dispatcher: &dyn crate::plugin::CommandDispatcher,
command_type: &CommandType,
payload: &str,
client_id: &str,
bounded_context_events: &[Event],
consistency_tags: &[Tag],
matching_events: &[Event],
initial_decision: crate::shared::CommandDecision,
snapshot: Option<SnapshotContext<'_>>,
idempotency_key: Option<&str>,
locked_highest: i64,
extra_committed_events: &[Event],
extra_committed_idempotency_keys: &std::collections::HashMap<
(String, String, String),
Vec<i64>,
>,
mut event_types_by_name: std::collections::HashMap<String, EventType>,
) -> crate::error::Result<DecideOutcome> {
let bounded_context_name = command_type.bounded_context.name.clone();
let schema = schema_ident(&bounded_context_name);
let original_highest = bounded_context_events
.iter()
.map(|e| e.sequence)
.max()
.unwrap_or_else(|| snapshot.as_ref().map(|s| s.as_of_sequence).unwrap_or(-1));
if let Some(key) = idempotency_key {
let cache_key = (
command_type.name.clone(),
client_id.to_string(),
key.to_string(),
);
let triggered_event_sequences = match extra_committed_idempotency_keys.get(&cache_key) {
Some(sequences) => Some(sequences.clone()),
None => {
lookup_idempotency_key(pool, &schema, &command_type.name, client_id, key).await?
}
};
if let Some(triggered_event_sequences) = triggered_event_sequences {
return Ok(DecideOutcome::Terminal(
SubmitCommandOutcome::Deduplicated {
triggered_event_sequences,
},
));
}
}
let mut final_decision = initial_decision;
let mut final_bounded_context_events = bounded_context_events.to_vec();
let mut final_matching_events = matching_events.to_vec();
let mut delta = if locked_highest > original_highest {
let delta_query_started = std::time::Instant::now();
let result = list_events_for_bounded_context_matching_tags_with_bc(
pool,
&command_type.bounded_context,
consistency_tags,
Some(original_highest),
Some(&event_types_by_name),
)
.await?;
tracing::info!(
bounded_context = %bounded_context_name,
delta_query_us = delta_query_started.elapsed().as_micros(),
delta_rows = result.len(),
"command decide delta query"
);
result
} else {
Vec::new()
};
delta.extend(
extra_committed_events
.iter()
.filter(|e| {
e.sequence > original_highest && consistency_tags.iter().any(|t| e.tags.contains(t))
})
.cloned(),
);
if !delta.is_empty() {
final_bounded_context_events.extend(delta);
final_bounded_context_events.sort_by_key(|e| e.sequence);
let (_boundary, redispatch_matching_events) =
crate::event_store::consistency_boundary_and_matching_events(
&final_bounded_context_events,
consistency_tags,
);
final_decision = match &snapshot {
Some(ctx) => match dispatcher.dispatch_from_snapshot(
&bounded_context_name,
&command_type.name,
payload,
ctx.state_json,
&redispatch_matching_events,
) {
None => return Err(crate::error::Error::NoDeciderRegistered),
Some(Err(e)) => return Err(e),
Some(Ok(d)) => d,
},
None => match dispatcher.dispatch(
&bounded_context_name,
&command_type.name,
payload,
&redispatch_matching_events,
) {
None => return Err(crate::error::Error::NoDeciderRegistered),
Some(Err(e)) => return Err(e),
Some(Ok(d)) => d,
},
};
final_matching_events = redispatch_matching_events;
}
let event_specs = match final_decision {
crate::shared::CommandDecision::Rejected { reason, kind } => {
COMMANDS_PROCESSED.add(
1,
&[
KeyValue::new("bounded_context", command_type.bounded_context.name.clone()),
KeyValue::new("command_type", command_type.name.clone()),
KeyValue::new("outcome", "rejected"),
],
);
return Ok(DecideOutcome::Terminal(SubmitCommandOutcome::Rejected {
reason,
kind,
matching_events: final_matching_events,
}));
}
crate::shared::CommandDecision::Accepted { events } => events,
};
for spec in &event_specs {
if let std::collections::hash_map::Entry::Vacant(entry) =
event_types_by_name.entry(spec.event_type.clone())
{
if let Some(et) = get_event_type(pool, &bounded_context_name, &spec.event_type).await? {
entry.insert(et);
}
}
}
Ok(DecideOutcome::Accepted(AcceptedDecision {
event_specs,
final_bounded_context_events,
event_types_by_name,
}))
}
#[allow(clippy::too_many_arguments)]
async fn finish_accepted_command_in_tx(
tx: &mut Transaction<'_, Postgres>,
pool: &Pool,
projection_dispatcher: &dyn crate::plugin::ProjectionDispatcher,
command_type: &CommandType,
payload: &str,
client_id: &str,
correlation_id: Option<&str>,
causation_id: Option<&str>,
encryption_master_key: Option<&EncryptionMasterKey>,
now: DateTime<Utc>,
idempotency_key: Option<&str>,
sync_projections: &[Projection],
sequences: Vec<i64>,
decided: AcceptedDecision,
mut resolved: std::collections::HashMap<(String, String), (EncryptionKey, i64, DataKey)>,
) -> crate::error::Result<SubmitCommandOutcome> {
let bounded_context_name = command_type.bounded_context.name.clone();
let schema = schema_ident(&bounded_context_name);
let AcceptedDecision {
event_specs,
final_bounded_context_events,
event_types_by_name,
} = decided;
let mut sequences = sequences.into_iter();
let mut needed_subjects: Vec<(String, String)> =
crate::event_store::sensitive_field_subjects(&command_type.sensitive_fields, payload);
for spec in &event_specs {
if let Some(event_type) = event_types_by_name.get(&spec.event_type) {
needed_subjects.extend(crate::event_store::sensitive_field_subjects(
&event_type.sensitive_fields,
&spec.payload.to_string(),
));
}
}
needed_subjects.retain(|pair| !resolved.contains_key(pair));
needed_subjects.sort();
needed_subjects.dedup();
if !needed_subjects.is_empty() {
let master_key = encryption_master_key.ok_or(encryption::Error::MasterKeyNotConfigured)?;
let provisioned = futures_util::future::try_join_all(needed_subjects.iter().map(
|(subject_key, subject_value)| {
get_or_create_encryption_key(
pool,
&bounded_context_name,
subject_key,
subject_value,
master_key,
)
},
))
.await?;
for (subject, provisioned) in needed_subjects.into_iter().zip(provisioned) {
resolved.insert(subject, provisioned);
}
}
let result = crate::event_store::process_command(
crate::shared::generate_token_id(),
command_type,
payload,
client_id,
correlation_id,
causation_id,
&final_bounded_context_events,
crate::shared::CommandDecision::Accepted {
events: event_specs,
},
|name| event_types_by_name.get(name).cloned(),
|| {
sequences.next().expect(
"process_command called next_sequence more times than there are accepted events",
)
},
now,
|subject_key, subject_value| {
let (key, _, data_key) = resolved
.get(&(subject_key.to_string(), subject_value.to_string()))
.expect(
"resolve_encryption_keys pre-resolved every subject sensitive_field_subjects \
named",
);
(key.clone(), data_key.clone())
},
)?;
let command_key_ids = encryption_key_ids(&result.command.encryption_keys, &resolved);
let command_id = insert_command(tx, &result.command, &command_key_ids).await?;
for event in &result.events {
let event_key_ids = encryption_key_ids(&event.encryption_keys, &resolved);
insert_event_and_update_sync_projections_in_tx(
tx,
event,
Some(command_id),
projection_dispatcher,
&event_key_ids,
sync_projections,
)
.await?;
}
if let Some(key) = idempotency_key {
let triggered_event_sequences: Vec<i64> =
result.events.iter().map(|e| e.sequence).collect();
insert_idempotency_key(
&mut **tx,
&schema,
&command_type.name,
client_id,
key,
&triggered_event_sequences,
now,
)
.await?;
}
COMMANDS_PROCESSED.add(
1,
&[
KeyValue::new("bounded_context", command_type.bounded_context.name.clone()),
KeyValue::new("command_type", command_type.name.clone()),
KeyValue::new("outcome", "accepted"),
],
);
Ok(SubmitCommandOutcome::Accepted {
command: Box::new(result.command),
events: result.events,
})
}
#[allow(clippy::too_many_arguments)]
async fn submit_one_command_in_tx(
tx: &mut Transaction<'_, Postgres>,
pool: &Pool,
dispatcher: &dyn crate::plugin::CommandDispatcher,
projection_dispatcher: &dyn crate::plugin::ProjectionDispatcher,
command_type: &CommandType,
payload: &str,
client_id: &str,
correlation_id: Option<&str>,
causation_id: Option<&str>,
bounded_context_events: &[Event],
consistency_tags: &[Tag],
matching_events: &[Event],
initial_decision: crate::shared::CommandDecision,
encryption_master_key: Option<&EncryptionMasterKey>,
now: DateTime<Utc>,
snapshot: Option<SnapshotContext<'_>>,
idempotency_key: Option<&str>,
locked_highest: i64,
sync_projections: &[Projection],
extra_committed_events: &[Event],
event_types_by_name: std::collections::HashMap<String, EventType>,
resolved: std::collections::HashMap<(String, String), (EncryptionKey, i64, DataKey)>,
) -> crate::error::Result<SubmitCommandOutcome> {
let bounded_context_name = command_type.bounded_context.name.clone();
let decided = match decide_command_in_tx(
pool,
dispatcher,
command_type,
payload,
client_id,
bounded_context_events,
consistency_tags,
matching_events,
initial_decision,
snapshot,
idempotency_key,
locked_highest,
extra_committed_events,
&std::collections::HashMap::new(),
event_types_by_name,
)
.await?
{
DecideOutcome::Terminal(outcome) => return Ok(outcome),
DecideOutcome::Accepted(decided) => decided,
};
let sequences = next_sequence_batch(
&mut **tx,
&bounded_context_name,
decided.event_specs.len() as i64,
)
.await?;
finish_accepted_command_in_tx(
tx,
pool,
projection_dispatcher,
command_type,
payload,
client_id,
correlation_id,
causation_id,
encryption_master_key,
now,
idempotency_key,
sync_projections,
sequences,
decided,
resolved,
)
.await
}
#[derive(Debug, Clone)]
pub struct OwnedSnapshotContext {
pub state_json: String,
pub as_of_sequence: i64,
}
pub struct BatchedCommand {
pub command_type: CommandType,
pub payload: String,
pub client_id: String,
pub correlation_id: Option<String>,
pub causation_id: Option<String>,
pub bounded_context_events: Vec<Event>,
pub consistency_tags: Vec<Tag>,
pub matching_events: Vec<Event>,
pub initial_decision: crate::shared::CommandDecision,
pub now: DateTime<Utc>,
pub snapshot: Option<OwnedSnapshotContext>,
pub idempotency_key: Option<String>,
pub event_types_by_name: std::collections::HashMap<String, EventType>,
pub resolved: std::collections::HashMap<(String, String), (EncryptionKey, i64, DataKey)>,
}
pub type BatchedCommandResult = crate::error::Result<SubmitCommandOutcome>;
pub struct CommandBatchLeaderTx {
tx: Transaction<'static, Postgres>,
bounded_context_name: String,
locked_highest: i64,
sync_projections: Vec<Projection>,
}
pub async fn begin_command_batch_leader_tx(
pool: &Pool,
bounded_context_name: &str,
idle_in_transaction_session_timeout: Option<std::time::Duration>,
) -> crate::error::Result<CommandBatchLeaderTx> {
let mut tx = pool.begin().await?;
if let Some(timeout) = idle_in_transaction_session_timeout {
sqlx::query(sqlx::AssertSqlSafe(format!(
"SET LOCAL idle_in_transaction_session_timeout = '{}ms'",
timeout.as_millis()
)))
.execute(&mut *tx)
.await?;
}
let locked_highest = lock_bounded_context_sequence(&mut tx, bounded_context_name).await?;
let sync_projections = sync_projections_for_bounded_context(pool, bounded_context_name).await?;
Ok(CommandBatchLeaderTx {
tx,
bounded_context_name: bounded_context_name.to_string(),
locked_highest,
sync_projections,
})
}
struct SequencePool {
reserved: std::collections::VecDeque<i64>,
last_draw: Vec<i64>,
}
impl SequencePool {
fn new() -> Self {
Self {
reserved: std::collections::VecDeque::new(),
last_draw: Vec::new(),
}
}
async fn take(
&mut self,
tx: &mut Transaction<'_, Postgres>,
bounded_context: &str,
count: i64,
refill_hint: i64,
) -> crate::error::Result<Vec<i64>> {
if (self.reserved.len() as i64) < count {
let refill =
next_sequence_batch(&mut **tx, bounded_context, refill_hint.max(count)).await?;
self.reserved.extend(refill);
}
let drawn: Vec<i64> = (0..count)
.map(|_| {
self.reserved
.pop_front()
.expect("just topped up the pool to at least `count`")
})
.collect();
self.last_draw = drawn.clone();
Ok(drawn)
}
fn confirm_last_draw(&mut self) {
self.last_draw.clear();
}
fn return_last_draw(&mut self) {
for value in self.last_draw.drain(..).rev() {
self.reserved.push_front(value);
}
}
async fn shrink_back(
self,
tx: &mut Transaction<'_, Postgres>,
bounded_context: &str,
) -> crate::error::Result<()> {
debug_assert!(
self.last_draw.is_empty(),
"every draw must be confirmed or returned before shrink_back runs"
);
let leftover = self.reserved.len() as i64;
if leftover > 0 {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.sequence SET next_value = next_value - $1"
)))
.bind(leftover)
.execute(&mut **tx)
.await?;
}
Ok(())
}
}
#[tracing::instrument(skip_all, fields(batch_size = batch.len()))]
pub async fn commit_command_batch(
leader_tx: CommandBatchLeaderTx,
pool: &Pool,
dispatcher: &dyn crate::plugin::CommandDispatcher,
projection_dispatcher: &dyn crate::plugin::ProjectionDispatcher,
encryption_master_key: Option<&EncryptionMasterKey>,
batch: Vec<BatchedCommand>,
) -> crate::error::Result<Vec<BatchedCommandResult>> {
let CommandBatchLeaderTx {
mut tx,
bounded_context_name,
locked_highest,
sync_projections,
} = leader_tx;
COMMAND_BATCH_SIZE.record(
batch.len() as u64,
&[KeyValue::new(
"bounded_context",
bounded_context_name.clone(),
)],
);
tracing::debug!(
bounded_context = %bounded_context_name,
batch_size = batch.len(),
"command batch committed"
);
let mut decide_total = std::time::Duration::ZERO;
let mut sequence_total = std::time::Duration::ZERO;
let mut persist_total = std::time::Duration::ZERO;
let mut results = Vec::with_capacity(batch.len());
let mut extra_committed_events: Vec<Event> = Vec::new();
let mut batch_idempotency_keys: std::collections::HashMap<(String, String, String), Vec<i64>> =
std::collections::HashMap::new();
let mut sequence_pool = SequencePool::new();
let batch_len = batch.len();
for (idx, item) in batch.into_iter().enumerate() {
let decide_started = std::time::Instant::now();
let decide_result = decide_command_in_tx(
pool,
dispatcher,
&item.command_type,
&item.payload,
&item.client_id,
&item.bounded_context_events,
&item.consistency_tags,
&item.matching_events,
item.initial_decision,
item.snapshot.as_ref().map(|s| SnapshotContext {
state_json: &s.state_json,
as_of_sequence: s.as_of_sequence,
}),
item.idempotency_key.as_deref(),
locked_highest,
&extra_committed_events,
&batch_idempotency_keys,
item.event_types_by_name,
)
.await;
decide_total += decide_started.elapsed();
let decided = match decide_result {
Ok(DecideOutcome::Terminal(outcome)) => {
results.push(Ok(outcome));
continue;
}
Ok(DecideOutcome::Accepted(decided)) => decided,
Err(e) => {
results.push(Err(e));
continue;
}
};
let refill_hint = (batch_len - idx) as i64;
let sequence_started = std::time::Instant::now();
let sequence_result = sequence_pool
.take(
&mut tx,
&bounded_context_name,
decided.event_specs.len() as i64,
refill_hint,
)
.await;
sequence_total += sequence_started.elapsed();
let sequences = match sequence_result {
Ok(sequences) => sequences,
Err(e) => {
return Err(e);
}
};
let command_type_name = item.command_type.name.clone();
let client_id = item.client_id.clone();
let idempotency_key = item.idempotency_key.clone();
let persist_started = std::time::Instant::now();
let mut nested = tx.begin().await?;
let outcome = finish_accepted_command_in_tx(
&mut nested,
pool,
projection_dispatcher,
&item.command_type,
&item.payload,
&item.client_id,
item.correlation_id.as_deref(),
item.causation_id.as_deref(),
encryption_master_key,
item.now,
item.idempotency_key.as_deref(),
&sync_projections,
sequences,
decided,
item.resolved,
)
.await;
match outcome {
Ok(outcome) => {
nested.commit().await?;
persist_total += persist_started.elapsed();
sequence_pool.confirm_last_draw();
if let SubmitCommandOutcome::Accepted { ref events, .. } = outcome {
extra_committed_events.extend(events.iter().cloned());
if let Some(key) = idempotency_key {
let triggered_event_sequences: Vec<i64> =
events.iter().map(|e| e.sequence).collect();
batch_idempotency_keys.insert(
(command_type_name, client_id, key),
triggered_event_sequences,
);
}
}
results.push(Ok(outcome));
}
Err(e) => {
nested.rollback().await?;
persist_total += persist_started.elapsed();
sequence_pool.return_last_draw();
results.push(Err(e));
}
}
}
sequence_pool
.shrink_back(&mut tx, &bounded_context_name)
.await?;
let commit_started = std::time::Instant::now();
tx.commit().await?;
let commit_elapsed = commit_started.elapsed();
tracing::debug!(
bounded_context = %bounded_context_name,
batch_size = batch_len,
decide_us = decide_total.as_micros(),
sequence_us = sequence_total.as_micros(),
persist_us = persist_total.as_micros(),
commit_us = commit_elapsed.as_micros(),
"command batch phase timing"
);
Ok(results)
}
pub async fn submit_command_batch(
pool: &Pool,
dispatcher: &dyn crate::plugin::CommandDispatcher,
projection_dispatcher: &dyn crate::plugin::ProjectionDispatcher,
encryption_master_key: Option<&EncryptionMasterKey>,
bounded_context_name: &str,
batch: Vec<BatchedCommand>,
) -> crate::error::Result<Vec<BatchedCommandResult>> {
let leader_tx = begin_command_batch_leader_tx(pool, bounded_context_name, None).await?;
commit_command_batch(
leader_tx,
pool,
dispatcher,
projection_dispatcher,
encryption_master_key,
batch,
)
.await
}
pub struct ResolvedCommandSubmission {
pub bounded_context_events: Vec<Event>,
pub consistency_tags: Vec<Tag>,
pub matching_events: Vec<Event>,
pub decision: crate::shared::CommandDecision,
pub snapshot_context: Option<ResolvedSnapshot>,
}
pub async fn resolve_command_submission(
pool: &Pool,
dispatcher: &dyn crate::plugin::CommandDispatcher,
snapshot_dispatcher: &dyn crate::plugin::SnapshotDispatcher,
event_cache: &crate::event_cache::EventCache,
command_type: &CommandType,
payload: &str,
) -> crate::error::Result<ResolvedCommandSubmission> {
let bounded_context_name = command_type.bounded_context.name.clone();
let consistency_tags = crate::event_store::derive_tags(&command_type.tag_mappings, payload);
let snapshot_context = match dispatcher.snapshot_name(&bounded_context_name, &command_type.name)
{
Some(Some(snapshot_name)) => {
resolve_snapshot_context(
pool,
&bounded_context_name,
snapshot_dispatcher,
snapshot_name,
&consistency_tags,
)
.await?
}
_ => None,
};
let bounded_context_events = list_events_for_bounded_context_matching_tags_cached(
pool,
event_cache,
&bounded_context_name,
&consistency_tags,
snapshot_context.as_ref().map(|ctx| ctx.as_of_sequence),
)
.await?;
let (_boundary, matching_events) = crate::event_store::consistency_boundary_and_matching_events(
&bounded_context_events,
&consistency_tags,
);
let decision = match &snapshot_context {
Some(ctx) => dispatcher
.dispatch_from_snapshot(
&bounded_context_name,
&command_type.name,
payload,
&ctx.state_json,
&matching_events,
)
.ok_or(crate::error::Error::NoDeciderRegistered)??,
None => dispatcher
.dispatch(
&bounded_context_name,
&command_type.name,
payload,
&matching_events,
)
.ok_or(crate::error::Error::NoDeciderRegistered)??,
};
Ok(ResolvedCommandSubmission {
bounded_context_events,
consistency_tags,
matching_events,
decision,
snapshot_context,
})
}
#[allow(clippy::too_many_arguments)]
pub async fn decide_and_submit_command(
pool: &Pool,
dispatcher: &dyn crate::plugin::CommandDispatcher,
projection_dispatcher: &dyn crate::plugin::ProjectionDispatcher,
snapshot_dispatcher: &dyn crate::plugin::SnapshotDispatcher,
broadcaster: &crate::event_store::EventBroadcaster,
event_cache: &crate::event_cache::EventCache,
command_type: &CommandType,
payload: &str,
client_id: &str,
correlation_id: Option<&str>,
causation_id: Option<&str>,
encryption_master_key: Option<&EncryptionMasterKey>,
now: DateTime<Utc>,
idempotency_key: Option<&str>,
) -> crate::error::Result<SubmitCommandOutcome> {
let resolved = resolve_command_submission(
pool,
dispatcher,
snapshot_dispatcher,
event_cache,
command_type,
payload,
)
.await?;
submit_command(
pool,
dispatcher,
projection_dispatcher,
broadcaster,
event_cache,
command_type,
payload,
client_id,
correlation_id,
causation_id,
&resolved.bounded_context_events,
&resolved.consistency_tags,
&resolved.matching_events,
resolved.decision,
encryption_master_key,
now,
resolved
.snapshot_context
.as_ref()
.map(|ctx| SnapshotContext {
state_json: &ctx.state_json,
as_of_sequence: ctx.as_of_sequence,
}),
idempotency_key,
)
.await
}
async fn get_cross_context_route_cursor(
pool: &Pool,
source_bounded_context: &str,
route_name: &str,
) -> crate::error::Result<Option<i64>> {
let schema = schema_ident(source_bounded_context);
let row: Option<(i64,)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT last_dispatched_sequence FROM {schema}.cross_context_route_cursors \
WHERE route_name = $1"
)))
.bind(route_name)
.fetch_optional(pool)
.await?;
Ok(row.map(|(seq,)| seq))
}
async fn update_cross_context_route_cursor(
pool: &Pool,
source_bounded_context: &str,
route_name: &str,
sequence: i64,
now: DateTime<Utc>,
) -> crate::error::Result<()> {
let schema = schema_ident(source_bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.cross_context_route_cursors \
(route_name, last_dispatched_sequence, updated_at) VALUES ($1, $2, $3) \
ON CONFLICT (route_name) DO UPDATE SET \
last_dispatched_sequence = EXCLUDED.last_dispatched_sequence, \
updated_at = EXCLUDED.updated_at, \
retry_attempt_count = 0, \
retry_first_failed_at = NULL, \
retry_next_attempt_at = NULL"
)))
.bind(route_name)
.bind(sequence)
.bind(now)
.execute(pool)
.await?;
Ok(())
}
type CrossContextRouteRetryState = (i32, Option<DateTime<Utc>>, Option<DateTime<Utc>>);
async fn get_cross_context_route_retry_state(
pool: &Pool,
source_bounded_context: &str,
route_name: &str,
) -> crate::error::Result<CrossContextRouteRetryState> {
let schema = schema_ident(source_bounded_context);
let row: Option<CrossContextRouteRetryState> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT retry_attempt_count, retry_first_failed_at, retry_next_attempt_at \
FROM {schema}.cross_context_route_cursors WHERE route_name = $1"
)))
.bind(route_name)
.fetch_optional(pool)
.await?;
Ok(row.unwrap_or((0, None, None)))
}
#[allow(clippy::too_many_arguments)]
async fn record_cross_context_route_retry_failure(
pool: &Pool,
source_bounded_context: &str,
route_name: &str,
current_cursor: i64,
attempt_count: i32,
first_failed_at: DateTime<Utc>,
next_attempt_at: DateTime<Utc>,
now: DateTime<Utc>,
) -> crate::error::Result<()> {
let schema = schema_ident(source_bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.cross_context_route_cursors \
(route_name, last_dispatched_sequence, updated_at, retry_attempt_count, \
retry_first_failed_at, retry_next_attempt_at) \
VALUES ($1, $2, $3, $4, $5, $6) \
ON CONFLICT (route_name) DO UPDATE SET \
retry_attempt_count = EXCLUDED.retry_attempt_count, \
retry_first_failed_at = EXCLUDED.retry_first_failed_at, \
retry_next_attempt_at = EXCLUDED.retry_next_attempt_at"
)))
.bind(route_name)
.bind(current_cursor)
.bind(now)
.bind(attempt_count)
.bind(first_failed_at)
.bind(next_attempt_at)
.execute(pool)
.await?;
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ParkedDeliveryKind {
CrossContextRoute,
ExternalEvent,
CommandTrigger,
}
impl ParkedDeliveryKind {
fn as_str(self) -> &'static str {
match self {
ParkedDeliveryKind::CrossContextRoute => "cross_context_route",
ParkedDeliveryKind::ExternalEvent => "external_event",
ParkedDeliveryKind::CommandTrigger => "command_trigger",
}
}
}
fn parked_delivery_kind_from_str(s: &str) -> ParkedDeliveryKind {
match s {
"external_event" => ParkedDeliveryKind::ExternalEvent,
"command_trigger" => ParkedDeliveryKind::CommandTrigger,
_ => ParkedDeliveryKind::CrossContextRoute,
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct ParkedDelivery {
pub id: String,
pub source: String,
pub kind: ParkedDeliveryKind,
pub identifier: String,
pub access_token_id: Option<String>,
pub target_bounded_context: Option<String>,
pub target_command_type: Option<String>,
pub request_json: serde_json::Value,
pub error: String,
pub attempt_count: i32,
pub first_failed_at: DateTime<Utc>,
pub last_failed_at: DateTime<Utc>,
}
#[derive(sqlx::FromRow)]
struct ParkedDeliveryRow {
id: String,
source: String,
kind: String,
identifier: String,
access_token_id: Option<String>,
target_bounded_context: Option<String>,
target_command_type: Option<String>,
request_json: Json<serde_json::Value>,
error: String,
attempt_count: i32,
first_failed_at: DateTime<Utc>,
last_failed_at: DateTime<Utc>,
}
impl From<ParkedDeliveryRow> for ParkedDelivery {
fn from(row: ParkedDeliveryRow) -> Self {
ParkedDelivery {
id: row.id,
source: row.source,
kind: parked_delivery_kind_from_str(&row.kind),
identifier: row.identifier,
access_token_id: row.access_token_id,
target_bounded_context: row.target_bounded_context,
target_command_type: row.target_command_type,
request_json: row.request_json.0,
error: row.error,
attempt_count: row.attempt_count,
first_failed_at: row.first_failed_at,
last_failed_at: row.last_failed_at,
}
}
}
const PARKED_DELIVERY_COLUMNS: &str = "id, source, kind, identifier, access_token_id, \
target_bounded_context, target_command_type, request_json, error, attempt_count, \
first_failed_at, last_failed_at";
#[tracing::instrument(skip_all)]
pub async fn ensure_parked_deliveries_table<'e>(
executor: impl sqlx::PgExecutor<'e>,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TABLE IF NOT EXISTS {schema}.parked_deliveries (
id TEXT PRIMARY KEY,
source TEXT NOT NULL,
kind TEXT NOT NULL,
identifier TEXT NOT NULL,
access_token_id TEXT,
target_bounded_context TEXT,
target_command_type TEXT,
request_json JSONB NOT NULL,
error TEXT NOT NULL,
attempt_count INT NOT NULL,
first_failed_at TIMESTAMPTZ NOT NULL,
last_failed_at TIMESTAMPTZ NOT NULL
)"
)))
.execute(executor)
.await?;
Ok(())
}
const PARKED_DELIVERIES_DEDUP_INDEX: &str = "parked_deliveries_source_kind_identifier_key";
pub async fn migrate_parked_deliveries_dedup_and_unique_index(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
let raw_schema = format!("bc_{bounded_context}");
let mut tx = pool.begin().await?;
sqlx::query("SELECT pg_advisory_xact_lock(hashtext($1)::bigint)")
.bind(&raw_schema)
.execute(&mut *tx)
.await?;
let already_migrated: bool = sqlx::query_scalar(
"SELECT EXISTS (
SELECT 1 FROM pg_indexes
WHERE schemaname = $1 AND indexname = $2
)",
)
.bind(&raw_schema)
.bind(PARKED_DELIVERIES_DEDUP_INDEX)
.fetch_one(&mut *tx)
.await?;
if already_migrated {
tx.commit().await?;
return Ok(());
}
sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.parked_deliveries a USING {schema}.parked_deliveries b \
WHERE a.source = b.source AND a.kind = b.kind AND a.identifier = b.identifier \
AND (a.last_failed_at, a.id) < (b.last_failed_at, b.id)"
)))
.execute(&mut *tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE UNIQUE INDEX IF NOT EXISTS {PARKED_DELIVERIES_DEDUP_INDEX} \
ON {schema}.parked_deliveries (source, kind, identifier)"
)))
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub async fn insert_parked_delivery(
pool: &Pool,
bounded_context: &str,
source: &str,
kind: ParkedDeliveryKind,
identifier: &str,
access_token_id: Option<&str>,
target_bounded_context: Option<&str>,
target_command_type: Option<&str>,
request_json: &serde_json::Value,
error: &str,
attempt_count: i32,
first_failed_at: DateTime<Utc>,
last_failed_at: DateTime<Utc>,
) -> crate::error::Result<ParkedDelivery> {
let schema = schema_ident(bounded_context);
let id = crate::shared::generate_token_id();
let row: ParkedDeliveryRow = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.parked_deliveries \
(id, source, kind, identifier, access_token_id, target_bounded_context, \
target_command_type, request_json, error, attempt_count, first_failed_at, \
last_failed_at) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) \
ON CONFLICT (source, kind, identifier) DO UPDATE SET \
request_json = EXCLUDED.request_json, \
error = EXCLUDED.error, \
attempt_count = EXCLUDED.attempt_count, \
last_failed_at = EXCLUDED.last_failed_at \
RETURNING {PARKED_DELIVERY_COLUMNS}"
)))
.bind(&id)
.bind(source)
.bind(kind.as_str())
.bind(identifier)
.bind(access_token_id)
.bind(target_bounded_context)
.bind(target_command_type)
.bind(Json(request_json))
.bind(error)
.bind(attempt_count)
.bind(first_failed_at)
.bind(last_failed_at)
.fetch_one(pool)
.await?;
Ok(row.into())
}
pub async fn list_parked_deliveries(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<Vec<ParkedDelivery>> {
let schema = schema_ident(bounded_context);
let rows: Vec<ParkedDeliveryRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {PARKED_DELIVERY_COLUMNS} FROM {schema}.parked_deliveries \
ORDER BY last_failed_at DESC"
)))
.fetch_all(pool)
.await?;
Ok(rows.into_iter().map(ParkedDelivery::from).collect())
}
pub async fn get_parked_delivery(
pool: &Pool,
bounded_context: &str,
id: &str,
) -> crate::error::Result<Option<ParkedDelivery>> {
let schema = schema_ident(bounded_context);
let row: Option<ParkedDeliveryRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {PARKED_DELIVERY_COLUMNS} FROM {schema}.parked_deliveries WHERE id = $1"
)))
.bind(id)
.fetch_optional(pool)
.await?;
Ok(row.map(ParkedDelivery::from))
}
pub async fn record_parked_delivery_retry_failure(
pool: &Pool,
bounded_context: &str,
id: &str,
error: &str,
now: DateTime<Utc>,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.parked_deliveries \
SET error = $1, attempt_count = attempt_count + 1, last_failed_at = $2 \
WHERE id = $3"
)))
.bind(error)
.bind(now)
.bind(id)
.execute(pool)
.await?;
Ok(())
}
pub async fn delete_parked_delivery(
pool: &Pool,
bounded_context: &str,
id: &str,
) -> crate::error::Result<Option<ParkedDelivery>> {
let existing = get_parked_delivery(pool, bounded_context, id).await?;
if existing.is_none() {
return Ok(None);
}
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.parked_deliveries WHERE id = $1"
)))
.bind(id)
.execute(pool)
.await?;
Ok(existing)
}
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(skip_all, fields(route = %route.name))]
pub async fn catch_up_cross_context_route(
pool: &Pool,
route: &crate::plugin::CrossContextRouteInfo,
route_dispatcher: &dyn crate::plugin::CrossContextRouteDispatcher,
command_dispatcher: &dyn crate::plugin::CommandDispatcher,
projection_dispatcher: &dyn crate::plugin::ProjectionDispatcher,
snapshot_dispatcher: &dyn crate::plugin::SnapshotDispatcher,
broadcaster: &crate::event_store::EventBroadcaster,
event_cache: &crate::event_cache::EventCache,
encryption_master_key: Option<&EncryptionMasterKey>,
retry_policy: &skilj_retry::RetryPolicy,
) -> crate::error::Result<()> {
let mut lock_tx = pool.begin().await?;
let lock_key = format!(
"cross_context_route:{}:{}",
route.source_bounded_context, route.name
);
sqlx::query("SELECT pg_advisory_xact_lock(hashtext($1)::bigint)")
.bind(&lock_key)
.execute(&mut *lock_tx)
.await?;
let result = catch_up_cross_context_route_locked(
pool,
route,
route_dispatcher,
command_dispatcher,
projection_dispatcher,
snapshot_dispatcher,
broadcaster,
event_cache,
encryption_master_key,
retry_policy,
)
.await;
lock_tx.commit().await?;
result
}
#[allow(clippy::too_many_arguments)]
async fn catch_up_cross_context_route_locked(
pool: &Pool,
route: &crate::plugin::CrossContextRouteInfo,
route_dispatcher: &dyn crate::plugin::CrossContextRouteDispatcher,
command_dispatcher: &dyn crate::plugin::CommandDispatcher,
projection_dispatcher: &dyn crate::plugin::ProjectionDispatcher,
snapshot_dispatcher: &dyn crate::plugin::SnapshotDispatcher,
broadcaster: &crate::event_store::EventBroadcaster,
event_cache: &crate::event_cache::EventCache,
encryption_master_key: Option<&EncryptionMasterKey>,
retry_policy: &skilj_retry::RetryPolicy,
) -> crate::error::Result<()> {
let existing_cursor =
get_cross_context_route_cursor(pool, route.source_bounded_context, route.name).await?;
let cursor = existing_cursor.unwrap_or(-1);
let events = list_events_cached(
pool,
event_cache,
route.source_bounded_context,
route.source_event_type,
cursor,
)
.await?;
if existing_cursor.is_none()
&& route.start_from != crate::plugin::CrossContextRouteStartFrom::Beginning
{
let seed = match route.start_from {
crate::plugin::CrossContextRouteStartFrom::Beginning => {
unreachable!("excluded by this branch's own condition above")
}
crate::plugin::CrossContextRouteStartFrom::Latest => {
events.iter().map(|e| e.sequence).max().unwrap_or(-1)
}
crate::plugin::CrossContextRouteStartFrom::AtSequence(n) => n,
crate::plugin::CrossContextRouteStartFrom::AtTime(unix_secs) => {
let threshold = DateTime::from_timestamp(unix_secs, 0).unwrap_or(Utc::now());
events
.iter()
.filter(|e| e.metadata.created_at <= threshold)
.map(|e| e.sequence)
.max()
.unwrap_or(-1)
}
};
update_cross_context_route_cursor(
pool,
route.source_bounded_context,
route.name,
seed,
Utc::now(),
)
.await?;
return Ok(());
}
let (mut retry_attempt, mut retry_first_failed_at, retry_next_attempt_at) =
get_cross_context_route_retry_state(pool, route.source_bounded_context, route.name).await?;
if retry_attempt > 0 {
if let Some(next_attempt_at) = retry_next_attempt_at {
if Utc::now() < next_attempt_at {
return Ok(());
}
}
}
for event in &events {
match route_dispatcher.route(route.name, &event.payload) {
None => {
tracing::warn!(
sequence = event.sequence,
"cross-context route not found in its own dispatcher - skipping"
);
}
Some(Err(e)) => {
tracing::warn!(
sequence = event.sequence,
error = %e,
"cross-context route: source payload did not deserialize - skipping"
);
}
Some(Ok(None)) => {
}
Some(Ok(Some(target_payload))) => {
let Some(target_command_type) = get_command_type(
pool,
route.target_bounded_context,
route.target_command_type,
)
.await?
else {
tracing::warn!(
sequence = event.sequence,
target_bounded_context = route.target_bounded_context,
target_command_type = route.target_command_type,
"cross-context route's own target CommandType isn't registered - \
skipping this occurrence, cursor still advances"
);
update_cross_context_route_cursor(
pool,
route.source_bounded_context,
route.name,
event.sequence,
Utc::now(),
)
.await?;
retry_attempt = 0;
retry_first_failed_at = None;
continue;
};
let idempotency_key = format!(
"{}{}:{}",
crate::event_store::RESERVED_IDEMPOTENCY_KEY_PREFIX,
route.name,
event.sequence
);
let outcome = decide_and_submit_command(
pool,
command_dispatcher,
projection_dispatcher,
snapshot_dispatcher,
broadcaster,
event_cache,
&target_command_type,
&target_payload,
"cross-context-route",
event.metadata.correlation_id.as_deref(),
Some(&crate::event_store::event_causation_id(event)),
encryption_master_key,
Utc::now(),
Some(&idempotency_key),
)
.await;
let outcome = match outcome {
Ok(outcome) => outcome,
Err(e) => {
retry_attempt += 1;
let first_failed_at = retry_first_failed_at.unwrap_or_else(Utc::now);
let now = Utc::now();
let elapsed = (now - first_failed_at).to_std().unwrap_or_default();
if retry_policy.is_exhausted(retry_attempt as u32, elapsed) {
tracing::error!(
sequence = event.sequence,
error = %e,
attempt = retry_attempt,
"cross-context route: target command submission failed \
repeatedly - parking and skipping this occurrence"
);
let request_json = serde_json::from_str(&target_payload)
.unwrap_or(serde_json::Value::String(target_payload));
insert_parked_delivery(
pool,
route.target_bounded_context,
&format!("cross-context-route:{}", route.name),
ParkedDeliveryKind::CrossContextRoute,
&event.sequence.to_string(),
None,
Some(route.target_bounded_context),
Some(route.target_command_type),
&request_json,
&e.to_string(),
retry_attempt,
first_failed_at,
now,
)
.await?;
update_cross_context_route_cursor(
pool,
route.source_bounded_context,
route.name,
event.sequence,
now,
)
.await?;
retry_attempt = 0;
retry_first_failed_at = None;
continue;
}
let next_attempt_at = now
+ chrono::Duration::from_std(
retry_policy.next_backoff(retry_attempt as u32),
)
.unwrap_or(chrono::Duration::zero());
tracing::warn!(
sequence = event.sequence,
error = %e,
attempt = retry_attempt,
next_attempt_at = %next_attempt_at,
"cross-context route: target command submission failed - will \
retry with backoff; route blocked until then"
);
record_cross_context_route_retry_failure(
pool,
route.source_bounded_context,
route.name,
cursor,
retry_attempt,
first_failed_at,
next_attempt_at,
now,
)
.await?;
return Ok(());
}
};
if matches!(outcome, SubmitCommandOutcome::Deduplicated { .. }) {
tracing::warn!(
sequence = event.sequence,
"cross-context route's own idempotency key was already present - \
expected after a crash/restart between a prior submission and its \
cursor advance; unexpected otherwise, since the key space is reserved"
);
}
}
}
update_cross_context_route_cursor(
pool,
route.source_bounded_context,
route.name,
event.sequence,
Utc::now(),
)
.await?;
retry_attempt = 0;
retry_first_failed_at = None;
}
Ok(())
}
async fn get_deadline_cursor(
pool: &Pool,
bounded_context: &str,
cursor_owner: &str,
) -> crate::error::Result<Option<i64>> {
let schema = schema_ident(bounded_context);
let row: Option<(i64,)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT last_dispatched_sequence FROM {schema}.deadline_cursors WHERE cursor_owner = $1"
)))
.bind(cursor_owner)
.fetch_optional(pool)
.await?;
Ok(row.map(|(seq,)| seq))
}
async fn update_deadline_cursor(
pool: &Pool,
bounded_context: &str,
cursor_owner: &str,
sequence: i64,
now: DateTime<Utc>,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.deadline_cursors \
(cursor_owner, last_dispatched_sequence, updated_at) VALUES ($1, $2, $3) \
ON CONFLICT (cursor_owner) DO UPDATE SET \
last_dispatched_sequence = EXCLUDED.last_dispatched_sequence, \
updated_at = EXCLUDED.updated_at"
)))
.bind(cursor_owner)
.bind(sequence)
.bind(now)
.execute(pool)
.await?;
Ok(())
}
pub async fn catch_up_schedule_deadline(
pool: &Pool,
schedule: &crate::plugin::ScheduleDeadlineInfo,
dispatcher: &dyn crate::plugin::ScheduleDeadlineDispatcher,
event_cache: &crate::event_cache::EventCache,
) -> crate::error::Result<()> {
let existing_cursor =
get_deadline_cursor(pool, schedule.source_bounded_context, schedule.name).await?;
let cursor = existing_cursor.unwrap_or(-1);
let events = list_events_cached(
pool,
event_cache,
schedule.source_bounded_context,
schedule.source_event_type,
cursor,
)
.await?;
if existing_cursor.is_none()
&& schedule.start_from != crate::plugin::DeadlinePollStartFrom::Beginning
{
let seed = match schedule.start_from {
crate::plugin::DeadlinePollStartFrom::Beginning => {
unreachable!("excluded by this branch's own condition above")
}
crate::plugin::DeadlinePollStartFrom::Latest => {
events.iter().map(|e| e.sequence).max().unwrap_or(-1)
}
crate::plugin::DeadlinePollStartFrom::AtSequence(n) => n,
crate::plugin::DeadlinePollStartFrom::AtTime(unix_secs) => {
let threshold = DateTime::from_timestamp(unix_secs, 0).unwrap_or(Utc::now());
events
.iter()
.filter(|e| e.metadata.created_at <= threshold)
.map(|e| e.sequence)
.max()
.unwrap_or(-1)
}
};
update_deadline_cursor(
pool,
schedule.source_bounded_context,
schedule.name,
seed,
Utc::now(),
)
.await?;
return Ok(());
}
let schema = schema_ident(schedule.source_bounded_context);
for event in &events {
match dispatcher.schedule(schedule.name, &event.payload) {
None => {
tracing::warn!(
sequence = event.sequence,
"schedule deadline not found in its own dispatcher - skipping"
);
}
Some(Err(e)) => {
tracing::warn!(
sequence = event.sequence,
error = %e,
"schedule deadline: source payload did not deserialize - skipping"
);
}
Some(Ok(None)) => {
}
Some(Ok(Some(spec))) => {
let id = format!("{}:{}", schedule.name, event.sequence);
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.deadlines \
(id, schedule_name, fire_at, tags, correlation_id, target_bounded_context, \
target_command_type, payload, status, created_at) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 'pending', $9) \
ON CONFLICT (id) DO NOTHING"
)))
.bind(&id)
.bind(schedule.name)
.bind(spec.fire_at)
.bind(Json(&spec.tags))
.bind(event.metadata.correlation_id.as_deref())
.bind(schedule.target_bounded_context)
.bind(schedule.target_command_type)
.bind(&spec.payload_json)
.bind(Utc::now())
.execute(pool)
.await?;
}
}
update_deadline_cursor(
pool,
schedule.source_bounded_context,
schedule.name,
event.sequence,
Utc::now(),
)
.await?;
}
Ok(())
}
pub async fn catch_up_cancel_deadline(
pool: &Pool,
cancel: &crate::plugin::CancelDeadlineInfo,
dispatcher: &dyn crate::plugin::CancelDeadlineDispatcher,
event_cache: &crate::event_cache::EventCache,
) -> crate::error::Result<()> {
let existing_cursor =
get_deadline_cursor(pool, cancel.source_bounded_context, cancel.name).await?;
let cursor = existing_cursor.unwrap_or(-1);
let events = list_events_cached(
pool,
event_cache,
cancel.source_bounded_context,
cancel.source_event_type,
cursor,
)
.await?;
if existing_cursor.is_none()
&& cancel.start_from != crate::plugin::DeadlinePollStartFrom::Beginning
{
let seed = match cancel.start_from {
crate::plugin::DeadlinePollStartFrom::Beginning => {
unreachable!("excluded by this branch's own condition above")
}
crate::plugin::DeadlinePollStartFrom::Latest => {
events.iter().map(|e| e.sequence).max().unwrap_or(-1)
}
crate::plugin::DeadlinePollStartFrom::AtSequence(n) => n,
crate::plugin::DeadlinePollStartFrom::AtTime(unix_secs) => {
let threshold = DateTime::from_timestamp(unix_secs, 0).unwrap_or(Utc::now());
events
.iter()
.filter(|e| e.metadata.created_at <= threshold)
.map(|e| e.sequence)
.max()
.unwrap_or(-1)
}
};
update_deadline_cursor(
pool,
cancel.source_bounded_context,
cancel.name,
seed,
Utc::now(),
)
.await?;
return Ok(());
}
let target_schema = schema_ident(cancel.deadline_schedule_bounded_context);
for event in &events {
match dispatcher.cancel_tags(cancel.name, &event.payload) {
None => {
tracing::warn!(
sequence = event.sequence,
"cancel deadline not found in its own dispatcher - skipping"
);
}
Some(Err(e)) => {
tracing::warn!(
sequence = event.sequence,
error = %e,
"cancel deadline: source payload did not deserialize - skipping"
);
}
Some(Ok(None)) => {
}
Some(Ok(Some(tags))) if tags.is_empty() => {
}
Some(Ok(Some(tags))) => {
let tag_literals: Vec<String> = tags
.iter()
.map(|t| {
serde_json::to_string(std::slice::from_ref(t))
.expect("Tag serialisation is infallible")
})
.collect();
let tag_clause = (0..tag_literals.len())
.map(|i| format!("tags @> ${}::jsonb", i + 3))
.collect::<Vec<_>>()
.join(" OR ");
let mut query = sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {target_schema}.deadlines SET status = 'cancelled', resolved_at = $1 \
WHERE schedule_name = $2 AND status = 'pending' AND ({tag_clause})"
)))
.bind(Utc::now())
.bind(cancel.deadline_schedule_name);
for literal in tag_literals {
query = query.bind(literal);
}
query.execute(pool).await?;
}
}
update_deadline_cursor(
pool,
cancel.source_bounded_context,
cancel.name,
event.sequence,
Utc::now(),
)
.await?;
}
Ok(())
}
const MAX_DUE_DEADLINES_PER_TICK: i64 = 1000;
fn deadline_firing_claim_stale_after() -> chrono::Duration {
chrono::Duration::minutes(5)
}
#[derive(sqlx::FromRow)]
struct DueDeadlineRow {
id: String,
correlation_id: Option<String>,
target_bounded_context: String,
target_command_type: String,
payload: String,
}
#[allow(clippy::too_many_arguments)]
pub async fn fire_due_deadlines(
pool: &Pool,
command_dispatcher: &dyn crate::plugin::CommandDispatcher,
projection_dispatcher: &dyn crate::plugin::ProjectionDispatcher,
snapshot_dispatcher: &dyn crate::plugin::SnapshotDispatcher,
broadcaster: &crate::event_store::EventBroadcaster,
event_cache: &crate::event_cache::EventCache,
bounded_context: &str,
now: DateTime<Utc>,
encryption_master_key: Option<&EncryptionMasterKey>,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
let stale_cutoff = now - deadline_firing_claim_stale_after();
let rows: Vec<DueDeadlineRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT id, correlation_id, target_bounded_context, target_command_type, payload \
FROM {schema}.deadlines \
WHERE (status = 'pending' OR (status = 'firing' AND firing_at <= $1)) AND fire_at <= $2 \
ORDER BY fire_at LIMIT {MAX_DUE_DEADLINES_PER_TICK}"
)))
.bind(stale_cutoff)
.bind(now)
.fetch_all(pool)
.await?;
for row in rows {
let Some(target_command_type) =
get_command_type(pool, &row.target_bounded_context, &row.target_command_type).await?
else {
tracing::warn!(
deadline_id = %row.id,
target_bounded_context = %row.target_bounded_context,
target_command_type = %row.target_command_type,
"deadline's own target CommandType isn't registered - marking fired without submitting"
);
mark_deadline_resolved(pool, &schema, &row.id, "fired", now).await?;
continue;
};
let claimed: Option<(String,)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.deadlines SET status = 'firing', firing_at = $1 \
WHERE id = $2 AND (status = 'pending' OR (status = 'firing' AND firing_at <= $3)) \
RETURNING id"
)))
.bind(now)
.bind(&row.id)
.bind(stale_cutoff)
.fetch_optional(pool)
.await?;
if claimed.is_none() {
continue;
}
let idempotency_key = format!(
"{}{}",
crate::event_store::RESERVED_DEADLINE_IDEMPOTENCY_KEY_PREFIX,
row.id
);
let outcome = decide_and_submit_command(
pool,
command_dispatcher,
projection_dispatcher,
snapshot_dispatcher,
broadcaster,
event_cache,
&target_command_type,
&row.payload,
"deadline",
row.correlation_id.as_deref(),
None,
encryption_master_key,
now,
Some(&idempotency_key),
)
.await?;
if matches!(outcome, SubmitCommandOutcome::Deduplicated { .. }) {
tracing::warn!(
deadline_id = %row.id,
"deadline's own idempotency key was already present - expected after a \
crash/restart between a prior fire attempt and marking it resolved; \
unexpected otherwise, since the key space is reserved"
);
}
mark_deadline_resolved(pool, &schema, &row.id, "fired", now).await?;
}
Ok(())
}
async fn mark_deadline_resolved(
pool: &Pool,
schema: &str,
id: &str,
status: &str,
now: DateTime<Utc>,
) -> crate::error::Result<()> {
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.deadlines SET status = $1, resolved_at = $2 \
WHERE id = $3 AND status IN ('pending', 'firing')"
)))
.bind(status)
.bind(now)
.bind(id)
.execute(pool)
.await?;
Ok(())
}
async fn lookup_idempotency_key<'e>(
executor: impl sqlx::PgExecutor<'e>,
schema: &str,
command_type_name: &str,
client_id: &str,
idempotency_key: &str,
) -> crate::error::Result<Option<Vec<i64>>> {
let row: Option<(Vec<i64>,)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT triggered_event_sequences FROM {schema}.idempotency_keys \
WHERE command_type_name = $1 AND client_id = $2 AND idempotency_key = $3"
)))
.bind(command_type_name)
.bind(client_id)
.bind(idempotency_key)
.fetch_optional(executor)
.await?;
Ok(row.map(|(sequences,)| sequences))
}
async fn insert_idempotency_key<'e>(
executor: impl sqlx::PgExecutor<'e>,
schema: &str,
command_type_name: &str,
client_id: &str,
idempotency_key: &str,
triggered_event_sequences: &[i64],
now: DateTime<Utc>,
) -> crate::error::Result<()> {
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.idempotency_keys \
(command_type_name, client_id, idempotency_key, triggered_event_sequences, created_at) \
VALUES ($1, $2, $3, $4, $5)"
)))
.bind(command_type_name)
.bind(client_id)
.bind(idempotency_key)
.bind(triggered_event_sequences)
.bind(now)
.execute(executor)
.await?;
Ok(())
}
async fn list_building_projection_rebuilds_for_bounded_context(
pool: &Pool,
bounded_context: &str,
all_projections: &[Projection],
) -> crate::error::Result<Vec<ProjectionRebuild>> {
let schema = schema_ident(bounded_context);
let rebuild_rows: Vec<ProjectionRebuildRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {PROJECTION_REBUILD_COLUMNS} FROM {schema}.projection_rebuilds \
WHERE status = 'building'"
)))
.fetch_all(pool)
.await?;
if rebuild_rows.is_empty() {
return Ok(Vec::new());
}
let consumed_rows: Vec<(String, String)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT projection_name, event_type_name \
FROM {schema}.projection_rebuild_consumed_event_types WHERE status = 'building'"
)))
.fetch_all(pool)
.await?;
let mut consumed_by_projection: std::collections::HashMap<String, Vec<String>> =
std::collections::HashMap::new();
for (projection_name, event_type_name) in consumed_rows {
consumed_by_projection
.entry(projection_name)
.or_default()
.push(event_type_name);
}
let all_event_types = list_event_types_for_bounded_context(pool, bounded_context).await?;
let event_types_by_name: std::collections::HashMap<&str, &EventType> = all_event_types
.iter()
.map(|et| (et.name.as_str(), et))
.collect();
let projections_by_name: std::collections::HashMap<&str, &Projection> = all_projections
.iter()
.map(|p| (p.name.as_str(), p))
.collect();
let mut rebuilds = Vec::with_capacity(rebuild_rows.len());
for row in rebuild_rows {
let Some(projection) = projections_by_name.get(row.projection_name.as_str()) else {
continue;
};
let consumed_event_types = consumed_by_projection
.remove(&row.projection_name)
.unwrap_or_default()
.into_iter()
.map(|name| {
event_types_by_name
.get(name.as_str())
.copied()
.cloned()
.expect(
"projection_rebuild_consumed_event_types join row references an \
event_types row that no longer exists",
)
})
.collect();
rebuilds.push(ProjectionRebuild {
projection: (*projection).clone(),
schema: row.schema,
schema_version: row.schema_version,
consumed_event_types,
sync: row.sync,
caught_up_to: row.caught_up_to,
status: projection_rebuild_status_from_str(&row.status),
});
}
Ok(rebuilds)
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn catch_up_bounded_context(
pool: &Pool,
bounded_context: &str,
dispatcher: &dyn crate::plugin::ProjectionDispatcher,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
let latest = latest_sequence(pool, bounded_context).await?.unwrap_or(-1);
let all_projections = list_projections_for_bounded_context(pool, bounded_context).await?;
let unpartitioned_async_projections: Vec<_> = all_projections
.iter()
.filter(|p| {
!p.sync
&& dispatcher
.partition_count(bounded_context, &p.name)
.unwrap_or(1)
.max(1)
<= 1
})
.collect();
let partitioned_async_projections: Vec<(&Projection, u32)> = all_projections
.iter()
.filter(|p| !p.sync)
.filter_map(|p| {
let partition_count = dispatcher
.partition_count(bounded_context, &p.name)
.unwrap_or(1)
.max(1);
(partition_count > 1).then_some((p, partition_count))
})
.collect();
let building_rebuilds = list_building_projection_rebuilds_for_bounded_context(
pool,
bounded_context,
&all_projections,
)
.await?;
if unpartitioned_async_projections.is_empty()
&& partitioned_async_projections.is_empty()
&& building_rebuilds.is_empty()
{
return Ok(());
}
for rebuild in &building_rebuilds {
if rebuild.caught_up_to.is_none() {
sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.projection_rebuild_state \
WHERE projection_name = $1 AND status = 'building'"
)))
.bind(&rebuild.projection.name)
.execute(pool)
.await?;
}
}
let min_caught_up = unpartitioned_async_projections
.iter()
.map(|p| p.caught_up_to.unwrap_or(-1))
.chain(
building_rebuilds
.iter()
.map(|r| r.caught_up_to.unwrap_or(-1)),
)
.min()
.unwrap_or(latest);
let events = if min_caught_up >= latest {
Vec::new()
} else {
list_events_for_bounded_context_from(pool, bounded_context, min_caught_up).await?
};
for event in &events {
let mut tx = pool.begin().await?;
for projection in &unpartitioned_async_projections {
if projection.caught_up_to.unwrap_or(-1) >= event.sequence {
continue;
}
let keys = dispatcher
.keys(bounded_context, &projection.name, event)
.unwrap_or_default();
let default_state_json = dispatcher
.default_state(bounded_context, &projection.name)
.unwrap_or_default();
let owner_tag_key = dispatcher
.owner_tag_key(bounded_context, &projection.name)
.flatten();
for key in &keys {
let (as_of_sequence, current_state) = get_or_create_projection_state_for_update(
&mut *tx,
&schema,
&projection.name,
key,
&default_state_json,
)
.await?;
if as_of_sequence >= event.sequence {
continue;
}
let new_state = match dispatcher.project(
bounded_context,
&projection.name,
¤t_state,
event,
key,
) {
Some(result) => result?,
None => current_state,
};
apply_projection_fold_update(
&mut *tx,
&schema,
"projection_state",
"",
&projection.name,
key,
&new_state,
owner_tag_key,
event,
)
.await?;
}
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.projections SET caught_up_to = $1 WHERE name = $2"
)))
.bind(event.sequence)
.bind(&projection.name)
.execute(&mut *tx)
.await?;
}
for rebuild in &building_rebuilds {
if rebuild.caught_up_to.unwrap_or(-1) >= event.sequence {
continue;
}
let keys = dispatcher
.keys(bounded_context, &rebuild.projection.name, event)
.unwrap_or_default();
let default_state_json = dispatcher
.default_state(bounded_context, &rebuild.projection.name)
.unwrap_or_default();
let owner_tag_key = dispatcher
.owner_tag_key(bounded_context, &rebuild.projection.name)
.flatten();
for key in &keys {
let (as_of_sequence, current_state) =
get_or_create_projection_rebuild_state_for_update(
&mut *tx,
&schema,
&rebuild.projection.name,
key,
&default_state_json,
)
.await?;
if as_of_sequence >= event.sequence {
continue;
}
let new_state = match dispatcher.project(
bounded_context,
&rebuild.projection.name,
¤t_state,
event,
key,
) {
Some(result) => result?,
None => current_state,
};
apply_projection_fold_update(
&mut *tx,
&schema,
"projection_rebuild_state",
" AND status = 'building'",
&rebuild.projection.name,
key,
&new_state,
owner_tag_key,
event,
)
.await?;
}
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.projection_rebuilds SET caught_up_to = $1 \
WHERE projection_name = $2 AND status = 'building'"
)))
.bind(event.sequence)
.bind(&rebuild.projection.name)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
}
for rebuild in &building_rebuilds {
let current = get_projection_rebuild(
pool,
bounded_context,
&rebuild.projection.name,
ProjectionRebuildStatus::Building,
)
.await?
.expect("a building rebuild this function just loaded can't have vanished mid-tick");
if current.caught_up_to.unwrap_or(-1) == latest {
let _ =
promote_projection_rebuild(pool, bounded_context, &rebuild.projection.name).await?;
}
}
for (projection, partition_count) in partitioned_async_projections {
catch_up_partitioned_projection(
pool,
bounded_context,
&schema,
dispatcher,
projection,
partition_count,
latest,
)
.await?;
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
async fn catch_up_partitioned_projection(
pool: &Pool,
bounded_context: &str,
schema: &str,
dispatcher: &dyn crate::plugin::ProjectionDispatcher,
projection: &Projection,
partition_count: u32,
latest: i64,
) -> crate::error::Result<()> {
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.projection_partition_progress \
(projection_name, partition_index, caught_up_to, updated_at) \
SELECT $1, gs, -1, now() FROM generate_series(0, $2::int - 1) AS gs \
ON CONFLICT (projection_name, partition_index) DO NOTHING"
)))
.bind(&projection.name)
.bind(partition_count as i32)
.execute(pool)
.await?;
let mut progress = list_projection_partition_progress(pool, schema, &projection.name).await?;
let min_caught_up = (0..partition_count)
.map(|p| progress.get(&p).copied().unwrap_or(-1))
.min()
.unwrap_or(latest);
let events = if min_caught_up >= latest {
Vec::new()
} else {
list_events_for_bounded_context_from_limited(
pool,
bounded_context,
min_caught_up,
MAX_EVENTS_PER_PARTITION_TICK,
)
.await?
};
if !events.is_empty() {
let batch_end = events
.last()
.expect("just checked events is non-empty")
.sequence;
let default_state_json = dispatcher
.default_state(bounded_context, &projection.name)
.unwrap_or_default();
let owner_tag_key = dispatcher
.owner_tag_key(bounded_context, &projection.name)
.flatten();
for partition_index in 0..partition_count {
if progress.get(&partition_index).copied().unwrap_or(-1) >= batch_end {
continue;
}
let mut tx = pool.begin().await?;
let locked: bool =
sqlx::query_scalar("SELECT pg_try_advisory_xact_lock(hashtext($1)::bigint)")
.bind(partition_lock_key(
bounded_context,
&projection.name,
partition_index,
))
.fetch_one(&mut *tx)
.await?;
if !locked {
continue;
}
let mut this_partition_progress = progress.get(&partition_index).copied().unwrap_or(-1);
for event in &events {
if this_partition_progress >= event.sequence {
continue;
}
let keys = dispatcher
.keys(bounded_context, &projection.name, event)
.unwrap_or_default();
for key in &keys {
if partition_for_key(key, partition_count) != partition_index {
continue;
}
let (as_of_sequence, current_state) =
get_or_create_projection_state_for_update(
&mut *tx,
schema,
&projection.name,
key,
&default_state_json,
)
.await?;
if as_of_sequence >= event.sequence {
continue;
}
let new_state = match dispatcher.project(
bounded_context,
&projection.name,
¤t_state,
event,
key,
) {
Some(result) => result?,
None => current_state,
};
apply_projection_fold_update(
&mut *tx,
schema,
"projection_state",
"",
&projection.name,
key,
&new_state,
owner_tag_key,
event,
)
.await?;
}
this_partition_progress = event.sequence;
}
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.projection_partition_progress \
(projection_name, partition_index, caught_up_to, updated_at) \
VALUES ($1, $2, $3, now()) \
ON CONFLICT (projection_name, partition_index) DO UPDATE SET \
caught_up_to = GREATEST(\
{schema}.projection_partition_progress.caught_up_to, EXCLUDED.caught_up_to), \
updated_at = now()"
)))
.bind(&projection.name)
.bind(partition_index as i32)
.bind(this_partition_progress)
.execute(&mut *tx)
.await?;
tx.commit().await?;
progress.insert(partition_index, this_partition_progress);
}
}
let rolled_up: Option<i64> = sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
"SELECT MIN(caught_up_to) FROM {schema}.projection_partition_progress \
WHERE projection_name = $1 AND partition_index < $2"
)))
.bind(&projection.name)
.bind(partition_count as i32)
.fetch_one(pool)
.await?;
if let Some(rolled_up) = rolled_up {
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.projections SET caught_up_to = $1 \
WHERE name = $2 AND (caught_up_to IS NULL OR caught_up_to < $1)"
)))
.bind(rolled_up)
.bind(&projection.name)
.execute(pool)
.await?;
}
Ok(())
}
const MAX_EVENTS_PER_PARTITION_TICK: i64 = 1000;
fn partition_for_key(key: &str, partition_count: u32) -> u32 {
const FNV_OFFSET_BASIS: u64 = 0xcbf29ce484222325;
const FNV_PRIME: u64 = 0x100000001b3;
let mut hash = FNV_OFFSET_BASIS;
for byte in key.as_bytes() {
hash ^= u64::from(*byte);
hash = hash.wrapping_mul(FNV_PRIME);
}
(hash % u64::from(partition_count)) as u32
}
fn partition_lock_key(
bounded_context: &str,
projection_name: &str,
partition_index: u32,
) -> String {
format!("projection_partition:{bounded_context}:{projection_name}:{partition_index}")
}
async fn list_projection_partition_progress(
pool: &Pool,
schema: &str,
projection_name: &str,
) -> crate::error::Result<std::collections::HashMap<u32, i64>> {
let rows: Vec<(i32, i64)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT partition_index, caught_up_to \
FROM {schema}.projection_partition_progress WHERE projection_name = $1"
)))
.bind(projection_name)
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|(idx, seq)| (idx as u32, seq))
.collect())
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn catch_up_snapshots(
pool: &Pool,
bounded_context: &str,
dispatcher: &dyn crate::plugin::SnapshotDispatcher,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
let latest = latest_sequence(pool, bounded_context).await?.unwrap_or(-1);
let all_snapshot_names = dispatcher.snapshot_names(bounded_context);
let unpartitioned_snapshot_names: Vec<&str> = all_snapshot_names
.iter()
.filter(|name| {
dispatcher
.partition_count(bounded_context, name)
.unwrap_or(1)
.max(1)
<= 1
})
.copied()
.collect();
let partitioned_snapshot_names: Vec<(&str, u32)> = all_snapshot_names
.iter()
.filter_map(|name| {
let partition_count = dispatcher
.partition_count(bounded_context, name)
.unwrap_or(1)
.max(1);
(partition_count > 1).then_some((*name, partition_count))
})
.collect();
if unpartitioned_snapshot_names.is_empty() && partitioned_snapshot_names.is_empty() {
return Ok(());
}
let progress_from_db =
list_snapshot_progress_for_bounded_context(pool, bounded_context).await?;
let mut progress: std::collections::HashMap<&str, i64> = unpartitioned_snapshot_names
.iter()
.map(|name| (*name, progress_from_db.get(*name).copied().unwrap_or(-1)))
.collect();
let min_caught_up = progress.values().copied().min().unwrap_or(latest);
let events = if unpartitioned_snapshot_names.is_empty() || min_caught_up >= latest {
Vec::new()
} else {
list_events_for_bounded_context_from(pool, bounded_context, min_caught_up).await?
};
for event in &events {
let mut tx = pool.begin().await?;
for name in &unpartitioned_snapshot_names {
if progress[name] >= event.sequence {
continue;
}
let Some(tag_key) = dispatcher.tag_key(bounded_context, name) else {
continue;
};
let Some(tag) = event.tags.iter().find(|t| t.key == tag_key) else {
continue;
};
let Some(tag_value) = &tag.value else {
continue;
};
let version = dispatcher.version(bounded_context, name).unwrap_or(0);
let default_state_json = dispatcher
.default_state(bounded_context, name)
.unwrap_or_default();
let (as_of_sequence, current_state) = get_or_create_snapshot_state_for_update(
&mut *tx,
&schema,
name,
tag_key,
tag_value,
version,
&default_state_json,
)
.await?;
if as_of_sequence >= event.sequence {
continue;
}
let new_state = match dispatcher.fold(bounded_context, name, ¤t_state, event) {
Some(result) => result?,
None => current_state,
};
let owner = dispatcher
.owner_tag_key(bounded_context, name)
.flatten()
.and_then(|owner_tag_key| {
event
.tags
.iter()
.find(|t| t.key == owner_tag_key)
.and_then(|t| t.value.clone())
});
match owner {
Some(owner) => {
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.snapshots SET snapshot_version = $1, as_of_sequence = $2, \
state = $3::jsonb, owner = $4, updated_at = now() \
WHERE snapshot_name = $5 AND tag_key = $6 AND tag_value = $7"
)))
.bind(version as i64)
.bind(event.sequence)
.bind(&new_state)
.bind(owner)
.bind(name)
.bind(tag_key)
.bind(tag_value.as_str())
.execute(&mut *tx)
.await?;
}
None => {
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.snapshots SET snapshot_version = $1, as_of_sequence = $2, \
state = $3::jsonb, updated_at = now() \
WHERE snapshot_name = $4 AND tag_key = $5 AND tag_value = $6"
)))
.bind(version as i64)
.bind(event.sequence)
.bind(&new_state)
.bind(name)
.bind(tag_key)
.bind(tag_value.as_str())
.execute(&mut *tx)
.await?;
}
}
}
for name in &unpartitioned_snapshot_names {
if progress[name] >= event.sequence {
continue;
}
upsert_snapshot_progress(&mut *tx, &schema, name, event.sequence).await?;
progress.insert(*name, event.sequence);
}
tx.commit().await?;
}
for (name, partition_count) in partitioned_snapshot_names {
catch_up_partitioned_snapshot(
pool,
bounded_context,
&schema,
dispatcher,
name,
partition_count,
latest,
)
.await?;
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
async fn catch_up_partitioned_snapshot(
pool: &Pool,
bounded_context: &str,
schema: &str,
dispatcher: &dyn crate::plugin::SnapshotDispatcher,
name: &str,
partition_count: u32,
latest: i64,
) -> crate::error::Result<()> {
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.snapshot_partition_progress \
(snapshot_name, partition_index, caught_up_to, updated_at) \
SELECT $1, gs, -1, now() FROM generate_series(0, $2::int - 1) AS gs \
ON CONFLICT (snapshot_name, partition_index) DO NOTHING"
)))
.bind(name)
.bind(partition_count as i32)
.execute(pool)
.await?;
let mut progress = list_snapshot_partition_progress(pool, schema, name).await?;
let min_caught_up = (0..partition_count)
.map(|p| progress.get(&p).copied().unwrap_or(-1))
.min()
.unwrap_or(latest);
let events = if min_caught_up >= latest {
Vec::new()
} else {
list_events_for_bounded_context_from_limited(
pool,
bounded_context,
min_caught_up,
MAX_EVENTS_PER_PARTITION_TICK,
)
.await?
};
if !events.is_empty() {
let batch_end = events
.last()
.expect("just checked events is non-empty")
.sequence;
let Some(tag_key) = dispatcher.tag_key(bounded_context, name) else {
return Ok(());
};
let version = dispatcher.version(bounded_context, name).unwrap_or(0);
let default_state_json = dispatcher
.default_state(bounded_context, name)
.unwrap_or_default();
let owner_tag_key = dispatcher.owner_tag_key(bounded_context, name).flatten();
for partition_index in 0..partition_count {
if progress.get(&partition_index).copied().unwrap_or(-1) >= batch_end {
continue;
}
let mut tx = pool.begin().await?;
let locked: bool =
sqlx::query_scalar("SELECT pg_try_advisory_xact_lock(hashtext($1)::bigint)")
.bind(snapshot_partition_lock_key(
bounded_context,
name,
partition_index,
))
.fetch_one(&mut *tx)
.await?;
if !locked {
continue;
}
let mut this_partition_progress = progress.get(&partition_index).copied().unwrap_or(-1);
for event in &events {
if this_partition_progress >= event.sequence {
continue;
}
let tag_value = event
.tags
.iter()
.find(|t| t.key == tag_key)
.and_then(|t| t.value.as_deref());
if let Some(tag_value) = tag_value {
if partition_for_key(tag_value, partition_count) == partition_index {
let (as_of_sequence, current_state) =
get_or_create_snapshot_state_for_update(
&mut *tx,
schema,
name,
tag_key,
tag_value,
version,
&default_state_json,
)
.await?;
if as_of_sequence < event.sequence {
let new_state =
match dispatcher.fold(bounded_context, name, ¤t_state, event)
{
Some(result) => result?,
None => current_state,
};
let owner = owner_tag_key.and_then(|owner_tag_key| {
event
.tags
.iter()
.find(|t| t.key == owner_tag_key)
.and_then(|t| t.value.clone())
});
match owner {
Some(owner) => {
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.snapshots SET snapshot_version = $1, \
as_of_sequence = $2, state = $3::jsonb, owner = $4, \
updated_at = now() \
WHERE snapshot_name = $5 AND tag_key = $6 AND tag_value = $7"
)))
.bind(version as i64)
.bind(event.sequence)
.bind(&new_state)
.bind(owner)
.bind(name)
.bind(tag_key)
.bind(tag_value)
.execute(&mut *tx)
.await?;
}
None => {
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.snapshots SET snapshot_version = $1, \
as_of_sequence = $2, state = $3::jsonb, updated_at = now() \
WHERE snapshot_name = $4 AND tag_key = $5 AND tag_value = $6"
)))
.bind(version as i64)
.bind(event.sequence)
.bind(&new_state)
.bind(name)
.bind(tag_key)
.bind(tag_value)
.execute(&mut *tx)
.await?;
}
}
}
}
}
this_partition_progress = event.sequence;
}
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.snapshot_partition_progress \
(snapshot_name, partition_index, caught_up_to, updated_at) \
VALUES ($1, $2, $3, now()) \
ON CONFLICT (snapshot_name, partition_index) DO UPDATE SET \
caught_up_to = GREATEST(\
{schema}.snapshot_partition_progress.caught_up_to, EXCLUDED.caught_up_to), \
updated_at = now()"
)))
.bind(name)
.bind(partition_index as i32)
.bind(this_partition_progress)
.execute(&mut *tx)
.await?;
tx.commit().await?;
progress.insert(partition_index, this_partition_progress);
}
}
let rolled_up: Option<i64> = sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
"SELECT MIN(caught_up_to) FROM {schema}.snapshot_partition_progress \
WHERE snapshot_name = $1 AND partition_index < $2"
)))
.bind(name)
.bind(partition_count as i32)
.fetch_one(pool)
.await?;
if let Some(rolled_up) = rolled_up {
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.snapshot_progress (snapshot_name, caught_up_to) \
VALUES ($1, $2) \
ON CONFLICT (snapshot_name) DO UPDATE SET \
caught_up_to = GREATEST({schema}.snapshot_progress.caught_up_to, EXCLUDED.caught_up_to)"
)))
.bind(name)
.bind(rolled_up)
.execute(pool)
.await?;
}
Ok(())
}
fn snapshot_partition_lock_key(
bounded_context: &str,
snapshot_name: &str,
partition_index: u32,
) -> String {
format!("snapshot_partition:{bounded_context}:{snapshot_name}:{partition_index}")
}
async fn list_snapshot_partition_progress(
pool: &Pool,
schema: &str,
snapshot_name: &str,
) -> crate::error::Result<std::collections::HashMap<u32, i64>> {
let rows: Vec<(i32, i64)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT partition_index, caught_up_to \
FROM {schema}.snapshot_partition_progress WHERE snapshot_name = $1"
)))
.bind(snapshot_name)
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|(idx, seq)| (idx as u32, seq))
.collect())
}
#[tracing::instrument(skip_all)]
pub async fn fold_history_into_new_sync_projection(
pool: &Pool,
projection: &Projection,
dispatcher: &dyn crate::plugin::ProjectionDispatcher,
) -> crate::error::Result<Projection> {
let bounded_context = &projection.bounded_context.name;
let schema = schema_ident(bounded_context);
let events = list_events_for_bounded_context(pool, bounded_context).await?;
let default_state_json = dispatcher
.default_state(bounded_context, &projection.name)
.unwrap_or_default();
let owner_tag_key = dispatcher
.owner_tag_key(bounded_context, &projection.name)
.flatten();
let mut caught_up_to = None;
for event in &events {
let mut tx = pool.begin().await?;
let keys = dispatcher
.keys(bounded_context, &projection.name, event)
.unwrap_or_default();
for key in &keys {
let (as_of_sequence, current_state) = get_or_create_projection_state_for_update(
&mut *tx,
&schema,
&projection.name,
key,
&default_state_json,
)
.await?;
if as_of_sequence >= event.sequence {
continue;
}
let new_state = match dispatcher.project(
bounded_context,
&projection.name,
¤t_state,
event,
key,
) {
Some(result) => result?,
None => current_state,
};
apply_projection_fold_update(
&mut *tx,
&schema,
"projection_state",
"",
&projection.name,
key,
&new_state,
owner_tag_key,
event,
)
.await?;
}
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.projections SET caught_up_to = $1 WHERE name = $2"
)))
.bind(event.sequence)
.bind(&projection.name)
.execute(&mut *tx)
.await?;
tx.commit().await?;
caught_up_to = Some(event.sequence);
}
Ok(Projection {
caught_up_to,
..projection.clone()
})
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn promote_projection_rebuild(
pool: &Pool,
bounded_context: &str,
projection_name: &str,
) -> crate::error::Result<bool> {
let schema = schema_ident(bounded_context);
let mut tx = pool.begin().await?;
let (locked_highest,): (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT next_value FROM {schema}.sequence FOR UPDATE"
)))
.fetch_one(&mut *tx)
.await?;
let building = projection_rebuild_status_to_str(ProjectionRebuildStatus::Building);
let rebuild_row: ProjectionRebuildRow = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {PROJECTION_REBUILD_COLUMNS} FROM {schema}.projection_rebuilds \
WHERE projection_name = $1 AND status = $2"
)))
.bind(projection_name)
.bind(building)
.fetch_one(&mut *tx)
.await?;
if rebuild_row.caught_up_to.unwrap_or(-1) != locked_highest {
return Ok(false);
}
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.projections SET schema = $1, schema_version = $2, sync = $3, \
caught_up_to = $4 WHERE name = $5"
)))
.bind(&rebuild_row.schema)
.bind(rebuild_row.schema_version)
.bind(rebuild_row.sync)
.bind(rebuild_row.caught_up_to)
.bind(projection_name)
.execute(&mut *tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.projection_consumed_event_types WHERE projection_name = $1"
)))
.bind(projection_name)
.execute(&mut *tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.projection_consumed_event_types (projection_name, event_type_name) \
SELECT projection_name, event_type_name \
FROM {schema}.projection_rebuild_consumed_event_types \
WHERE projection_name = $1 AND status = $2"
)))
.bind(projection_name)
.bind(building)
.execute(&mut *tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.projection_state WHERE projection_name = $1"
)))
.bind(projection_name)
.execute(&mut *tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.projection_state \
(projection_name, key, state, owner, as_of_sequence, updated_at) \
SELECT projection_name, key, state, owner, as_of_sequence, updated_at \
FROM {schema}.projection_rebuild_state WHERE projection_name = $1 AND status = $2"
)))
.bind(projection_name)
.bind(building)
.execute(&mut *tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.projection_rebuild_state WHERE projection_name = $1 AND status = $2"
)))
.bind(projection_name)
.bind(building)
.execute(&mut *tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.projection_rebuild_consumed_event_types \
WHERE projection_name = $1 AND status = $2"
)))
.bind(projection_name)
.bind(building)
.execute(&mut *tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.projection_rebuilds WHERE projection_name = $1 AND status = $2"
)))
.bind(projection_name)
.bind(building)
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(true)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AccessTokenKind {
ExternalEvent,
DirectCreation,
EventRead,
Command,
}
impl AccessTokenKind {
fn as_str(self) -> &'static str {
match self {
AccessTokenKind::ExternalEvent => "external_event",
AccessTokenKind::DirectCreation => "direct_creation",
AccessTokenKind::EventRead => "event_read",
AccessTokenKind::Command => "command",
}
}
fn from_str(s: &str) -> Self {
match s {
"direct_creation" => AccessTokenKind::DirectCreation,
"event_read" => AccessTokenKind::EventRead,
"command" => AccessTokenKind::Command,
_ => AccessTokenKind::ExternalEvent,
}
}
}
async fn resolve_token_bounded_context(
pool: &Pool,
id: &str,
) -> crate::error::Result<Option<String>> {
let row: Option<(String,)> =
sqlx::query_as("SELECT bounded_context FROM access_token_index WHERE id = $1")
.bind(id)
.fetch_optional(pool)
.await?;
Ok(row.map(|(bounded_context,)| bounded_context))
}
async fn insert_token_index(
pool: &Pool,
id: &str,
bounded_context: &str,
) -> crate::error::Result<()> {
sqlx::query("INSERT INTO access_token_index (id, bounded_context) VALUES ($1, $2)")
.bind(id)
.bind(bounded_context)
.execute(pool)
.await?;
Ok(())
}
#[derive(sqlx::FromRow)]
struct AccessTokenColumns {
id: String,
kind: String,
secret: String,
status: String,
created_at: DateTime<Utc>,
revoked_at: Option<DateTime<Utc>>,
event_type_name: Option<String>,
command_type_name: Option<String>,
scope: Option<String>,
start_from: String,
start_at_sequence: Option<i64>,
start_at_time: Option<DateTime<Utc>>,
}
struct AccessTokenRow {
columns: AccessTokenColumns,
bounded_context: String,
}
async fn fetch_access_token_row(
pool: &Pool,
id: &str,
) -> crate::error::Result<Option<AccessTokenRow>> {
let Some(bounded_context) = resolve_token_bounded_context(pool, id).await? else {
return Ok(None);
};
let schema = schema_ident(&bounded_context);
let columns: Option<AccessTokenColumns> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT id, kind, secret, status, created_at, revoked_at, event_type_name, \
command_type_name, scope, start_from, start_at_sequence, start_at_time \
FROM {schema}.access_tokens WHERE id = $1"
)))
.bind(id)
.fetch_optional(pool)
.await?;
Ok(columns.map(|columns| AccessTokenRow {
columns,
bounded_context,
}))
}
#[tracing::instrument(skip_all)]
pub async fn access_token_kind(
pool: &Pool,
id: &str,
) -> crate::error::Result<Option<AccessTokenKind>> {
Ok(fetch_access_token_row(pool, id)
.await?
.map(|r| AccessTokenKind::from_str(&r.columns.kind)))
}
#[tracing::instrument(skip_all)]
pub async fn revoke_access_token(
pool: &Pool,
id: &str,
revoked_at: DateTime<Utc>,
) -> crate::error::Result<()> {
let Some(bounded_context) = resolve_token_bounded_context(pool, id).await? else {
return Ok(());
};
let schema = schema_ident(&bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.access_tokens SET status = 'revoked', revoked_at = $1 WHERE id = $2"
)))
.bind(revoked_at)
.bind(id)
.execute(pool)
.await?;
Ok(())
}
#[allow(clippy::too_many_arguments)]
async fn insert_access_token_row(
pool: &Pool,
id: &str,
kind: AccessTokenKind,
secret: &str,
status: TokenStatus,
created_at: DateTime<Utc>,
revoked_at: Option<DateTime<Utc>>,
bounded_context: &str,
event_type_name: &str,
scope: Option<&str>,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.access_tokens (id, kind, secret, status, created_at, revoked_at, \
event_type_name, scope) VALUES ($1,$2,$3,$4,$5,$6,$7,$8)"
)))
.bind(id)
.bind(kind.as_str())
.bind(crate::shared::hash_secret(secret))
.bind(token_status_to_str(status))
.bind(created_at)
.bind(revoked_at)
.bind(event_type_name)
.bind(scope)
.execute(pool)
.await?;
insert_token_index(pool, id, bounded_context).await
}
#[tracing::instrument(skip_all)]
pub async fn insert_external_event_token(
pool: &Pool,
token: &ExternalEventToken,
) -> crate::error::Result<()> {
insert_access_token_row(
pool,
&token.id,
AccessTokenKind::ExternalEvent,
&token.secret,
token.status,
token.created_at,
token.revoked_at,
&token.event_type.bounded_context.name,
&token.event_type.name,
token.scope.as_deref(),
)
.await
}
#[tracing::instrument(skip_all)]
pub async fn insert_direct_creation_token(
pool: &Pool,
token: &DirectCreationToken,
) -> crate::error::Result<()> {
insert_access_token_row(
pool,
&token.id,
AccessTokenKind::DirectCreation,
&token.secret,
token.status,
token.created_at,
token.revoked_at,
&token.event_type.bounded_context.name,
&token.event_type.name,
token.scope.as_deref(),
)
.await
}
#[tracing::instrument(skip_all)]
pub async fn insert_event_read_token(
pool: &Pool,
token: &EventReadToken,
) -> crate::error::Result<()> {
let schema = schema_ident(&token.event_type.bounded_context.name);
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.access_tokens (id, kind, secret, status, created_at, revoked_at, \
event_type_name, scope, start_from, start_at_sequence, start_at_time) \
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)"
)))
.bind(&token.id)
.bind(AccessTokenKind::EventRead.as_str())
.bind(crate::shared::hash_secret(&token.secret))
.bind(token_status_to_str(token.status))
.bind(token.created_at)
.bind(token.revoked_at)
.bind(&token.event_type.name)
.bind(token.scope.as_deref())
.bind(event_read_start_position_to_str(token.start_from))
.bind(token.start_at_sequence)
.bind(token.start_at_time)
.execute(pool)
.await?;
insert_token_index(pool, &token.id, &token.event_type.bounded_context.name).await
}
#[tracing::instrument(skip_all)]
pub async fn insert_command_token(pool: &Pool, token: &CommandToken) -> crate::error::Result<()> {
let schema = schema_ident(&token.command_type.bounded_context.name);
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.access_tokens (id, kind, secret, status, created_at, revoked_at, \
command_type_name, scope) VALUES ($1,$2,$3,$4,$5,$6,$7,$8)"
)))
.bind(&token.id)
.bind(AccessTokenKind::Command.as_str())
.bind(crate::shared::hash_secret(&token.secret))
.bind(token_status_to_str(token.status))
.bind(token.created_at)
.bind(token.revoked_at)
.bind(&token.command_type.name)
.bind(token.scope.as_deref())
.execute(pool)
.await?;
insert_token_index(pool, &token.id, &token.command_type.bounded_context.name).await
}
macro_rules! get_event_type_access_token {
($fn_name:ident, $return_type:ident, $kind:literal) => {
#[tracing::instrument(skip_all)]
pub async fn $fn_name(pool: &Pool, id: &str) -> crate::error::Result<Option<$return_type>> {
let Some(row) = fetch_access_token_row(pool, id).await? else {
return Ok(None);
};
if row.columns.kind != $kind {
return Ok(None);
}
let event_type_name = row
.columns
.event_type_name
.expect(concat!($kind, " access_tokens row without event_type_name"));
let event_type = get_event_type(pool, &row.bounded_context, &event_type_name)
.await?
.expect("access_tokens row references an event_type that no longer exists");
Ok(Some($return_type {
id: row.columns.id,
secret: row.columns.secret,
status: token_status_from_str(&row.columns.status),
created_at: row.columns.created_at,
revoked_at: row.columns.revoked_at,
event_type,
scope: row.columns.scope,
}))
}
};
}
get_event_type_access_token!(
get_external_event_token,
ExternalEventToken,
"external_event"
);
get_event_type_access_token!(
get_direct_creation_token,
DirectCreationToken,
"direct_creation"
);
#[tracing::instrument(skip_all)]
pub async fn get_event_read_token(
pool: &Pool,
id: &str,
) -> crate::error::Result<Option<EventReadToken>> {
let Some(row) = fetch_access_token_row(pool, id).await? else {
return Ok(None);
};
if row.columns.kind != "event_read" {
return Ok(None);
}
let event_type_name = row
.columns
.event_type_name
.expect("event_read access_tokens row without event_type_name");
let event_type = get_event_type(pool, &row.bounded_context, &event_type_name)
.await?
.expect("access_tokens row references an event_type that no longer exists");
Ok(Some(EventReadToken {
id: row.columns.id,
secret: row.columns.secret,
status: token_status_from_str(&row.columns.status),
created_at: row.columns.created_at,
revoked_at: row.columns.revoked_at,
event_type,
scope: row.columns.scope,
start_from: event_read_start_position_from_str(&row.columns.start_from),
start_at_sequence: row.columns.start_at_sequence,
start_at_time: row.columns.start_at_time,
}))
}
#[tracing::instrument(skip_all)]
pub async fn get_command_token(
pool: &Pool,
id: &str,
) -> crate::error::Result<Option<CommandToken>> {
let Some(row) = fetch_access_token_row(pool, id).await? else {
return Ok(None);
};
if row.columns.kind != "command" {
return Ok(None);
}
let command_type_name = row
.columns
.command_type_name
.expect("command access_tokens row without command_type_name");
let command_type = get_command_type(pool, &row.bounded_context, &command_type_name)
.await?
.expect("access_tokens row references a command_type that no longer exists");
Ok(Some(CommandToken {
id: row.columns.id,
secret: row.columns.secret,
status: token_status_from_str(&row.columns.status),
created_at: row.columns.created_at,
revoked_at: row.columns.revoked_at,
command_type,
scope: row.columns.scope,
}))
}
#[derive(sqlx::FromRow)]
struct ReadCursorRow {
ack_mode: String,
sequence: i64,
updated_at: DateTime<Utc>,
checked_out_at: Option<DateTime<Utc>>,
}
#[tracing::instrument(skip_all)]
pub async fn get_read_cursor(
executor: impl sqlx::PgExecutor<'_>,
token: &EventReadToken,
) -> crate::error::Result<Option<ReadCursor>> {
let schema = schema_ident(&token.event_type.bounded_context.name);
let row: Option<ReadCursorRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT ack_mode, sequence, updated_at, checked_out_at \
FROM {schema}.read_cursors WHERE token_id = $1"
)))
.bind(&token.id)
.fetch_optional(executor)
.await?;
Ok(row.map(|r| ReadCursor {
token: token.clone(),
ack_mode: ack_mode_from_str(&r.ack_mode),
sequence: r.sequence,
updated_at: r.updated_at,
checked_out_at: r.checked_out_at,
}))
}
#[tracing::instrument(skip_all)]
pub async fn lock_read_cursor_for_consume(
tx: &mut Transaction<'_, Postgres>,
token: &EventReadToken,
) -> crate::error::Result<()> {
let key = format!(
"read_cursor:{}:{}",
token.event_type.bounded_context.name, token.id
);
sqlx::query("SELECT pg_advisory_xact_lock(hashtext($1)::bigint)")
.bind(key)
.execute(&mut **tx)
.await?;
Ok(())
}
async fn upsert_read_cursor(
executor: impl sqlx::PgExecutor<'_>,
cursor: &ReadCursor,
) -> crate::error::Result<()> {
let schema = schema_ident(&cursor.token.event_type.bounded_context.name);
sqlx::query(sqlx::AssertSqlSafe(format!(
"INSERT INTO {schema}.read_cursors (token_id, ack_mode, sequence, updated_at, checked_out_at) \
VALUES ($1,$2,$3,$4,$5) \
ON CONFLICT (token_id) DO UPDATE SET \
ack_mode = EXCLUDED.ack_mode, sequence = EXCLUDED.sequence, \
updated_at = EXCLUDED.updated_at, checked_out_at = EXCLUDED.checked_out_at"
)))
.bind(&cursor.token.id)
.bind(ack_mode_to_str(cursor.ack_mode))
.bind(cursor.sequence)
.bind(cursor.updated_at)
.bind(cursor.checked_out_at)
.execute(executor)
.await?;
Ok(())
}
#[tracing::instrument(skip_all)]
pub async fn apply_cursor_update(
executor: impl sqlx::PgExecutor<'_>,
token: &EventReadToken,
update: &CursorUpdate,
) -> crate::error::Result<()> {
match update {
CursorUpdate::Created(cursor) => upsert_read_cursor(executor, cursor).await,
CursorUpdate::Advanced {
sequence,
updated_at,
} => {
let schema = schema_ident(&token.event_type.bounded_context.name);
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.read_cursors SET sequence = $1, updated_at = $2 WHERE token_id = $3"
)))
.bind(sequence)
.bind(updated_at)
.bind(&token.id)
.execute(executor)
.await?;
Ok(())
}
CursorUpdate::Claimed { checked_out_at } => {
let schema = schema_ident(&token.event_type.bounded_context.name);
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.read_cursors SET checked_out_at = $1 WHERE token_id = $2"
)))
.bind(checked_out_at)
.bind(&token.id)
.execute(executor)
.await?;
Ok(())
}
CursorUpdate::Unchanged => Ok(()),
}
}
#[tracing::instrument(skip_all)]
pub async fn record_acknowledgement(
pool: &Pool,
token: &EventReadToken,
sequence: i64,
updated_at: DateTime<Utc>,
) -> crate::error::Result<()> {
let schema = schema_ident(&token.event_type.bounded_context.name);
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.read_cursors SET sequence = $1, updated_at = $2, checked_out_at = NULL \
WHERE token_id = $3"
)))
.bind(sequence)
.bind(updated_at)
.bind(&token.id)
.execute(pool)
.await?;
Ok(())
}