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.replace('"', "\"\""))
}
fn sql_literal(value: &str) -> String {
format!("'{}'", value.replace('\'', "''"))
}
fn add_columns_patch(schema: &str, table: &str, columns: &[&str], ddl: String) -> String {
let table = sql_literal(&format!("{schema}.{table}"));
let missing = columns
.iter()
.map(|column| {
format!(
"NOT EXISTS (SELECT 1 FROM pg_attribute WHERE attrelid = to_regclass({table}) \
AND attname = {} AND attnum > 0 AND NOT attisdropped)",
sql_literal(column)
)
})
.collect::<Vec<_>>()
.join(" OR ");
format!("DO $patch$ BEGIN IF {missing} THEN {ddl}; END IF; END $patch$")
}
fn create_index_patch(schema: &str, index: &str, ddl: String) -> String {
format!(
"DO $patch$ BEGIN IF to_regclass({}) IS NULL THEN {ddl}; END IF; END $patch$",
sql_literal(&format!("{schema}.{index}"))
)
}
fn add_constraint_patch(schema: &str, table: &str, name: &str, ddl: String) -> String {
format!(
"DO $patch$ BEGIN IF NOT EXISTS (SELECT 1 FROM pg_constraint \
WHERE conrelid = to_regclass({}) AND conname = {}) THEN {ddl}; END IF; END $patch$",
sql_literal(&format!("{schema}.{table}")),
sql_literal(name)
)
}
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(parked_deliveries_occurrence_index_ddl(
&schema,
)))
.execute(&mut **tx)
.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?;
add_registration_version_columns(&mut **tx, bounded_context).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(create_index_patch(&schema, "private_field_grants_by_grantee", 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?;
grants_from_rows(pool, &bc, rows).await
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context))]
pub async fn list_active_private_field_grants_for_grantee(
pool: &Pool,
bounded_context: &str,
grantee: &Role,
) -> 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 \
WHERE grantee_role_id = $1 AND status = 'active'"
)))
.bind(&grantee.id)
.fetch_all(pool)
.await?;
grants_from_rows(pool, &bc, rows).await
}
async fn grants_from_rows(
pool: &Pool,
bc: &BoundedContext,
rows: Vec<PrivateFieldGrantRow>,
) -> crate::error::Result<Vec<PrivateFieldGrant>> {
if rows.is_empty() {
return Ok(Vec::new());
}
let mut role_ids: Vec<&str> = rows
.iter()
.flat_map(|r| [r.grantor_role_id.as_str(), r.grantee_role_id.as_str()])
.collect();
role_ids.sort_unstable();
role_ids.dedup();
let roles: std::collections::HashMap<String, Role> =
sqlx::query_as::<_, RoleRow>(sqlx::AssertSqlSafe(format!(
"SELECT {ROLE_COLUMNS} FROM roles WHERE id = ANY($1)"
)))
.bind(&role_ids)
.fetch_all(pool)
.await?
.into_iter()
.map(|row| {
let role = row.into_domain();
(role.id.clone(), role)
})
.collect();
let mut command_ids: Vec<i64> = rows.iter().filter_map(|r| r.command_id).collect();
command_ids.sort_unstable();
command_ids.dedup();
let command_external_ids: std::collections::HashMap<i64, String> = if command_ids.is_empty() {
std::collections::HashMap::new()
} else {
let schema = schema_ident(&bc.name);
sqlx::query_as::<_, (i64, String)>(sqlx::AssertSqlSafe(format!(
"SELECT id, external_id FROM {schema}.commands WHERE id = ANY($1)"
)))
.bind(&command_ids)
.fetch_all(pool)
.await?
.into_iter()
.collect()
};
Ok(rows
.into_iter()
.filter_map(|row| {
let command_id = match row.command_id {
Some(internal_id) => Some(command_external_ids.get(&internal_id)?.clone()),
None => None,
};
Some(PrivateFieldGrant {
id: row.id,
bounded_context: bc.clone(),
grantor: roles.get(&row.grantor_role_id)?.clone(),
grantee: roles.get(&row.grantee_role_id)?.clone(),
event_sequence: row.event_sequence,
command_id,
status: token_status_from_str(&row.status),
created_at: row.created_at,
revoked_at: row.revoked_at,
})
})
.collect())
}
#[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);
let index = create_index_patch(
&schema,
"idempotency_keys_created_at",
format!(
"CREATE INDEX IF NOT EXISTS idempotency_keys_created_at \
ON {schema}.idempotency_keys (created_at)"
),
);
sqlx::raw_sql(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)
);
{index}"
)))
.execute(executor)
.await?;
Ok(())
}
pub async fn delete_expired_idempotency_keys(
pool: &Pool,
bounded_context: &str,
cutoff: DateTime<Utc>,
batch: i64,
) -> crate::error::Result<u64> {
let schema = schema_ident(bounded_context);
let result = sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.idempotency_keys WHERE ctid IN ( \
SELECT ctid FROM {schema}.idempotency_keys WHERE created_at < $1 LIMIT $2)"
)))
.bind(cutoff)
.bind(batch)
.execute(pool)
.await?;
Ok(result.rows_affected())
}
pub async fn delete_expired_deadlines(
pool: &Pool,
bounded_context: &str,
cutoff: DateTime<Utc>,
batch: i64,
) -> crate::error::Result<u64> {
let schema = schema_ident(bounded_context);
let result = sqlx::query(sqlx::AssertSqlSafe(format!(
"DELETE FROM {schema}.deadlines WHERE ctid IN ( \
SELECT ctid FROM {schema}.deadlines \
WHERE status NOT IN ('pending', 'firing') AND resolved_at < $1 LIMIT $2)"
)))
.bind(cutoff)
.bind(batch)
.execute(pool)
.await?;
Ok(result.rows_affected())
}
#[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(add_columns_patch(
&schema,
"cross_context_route_cursors",
&["retry_attempt_count"],
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(add_columns_patch(
&schema,
"cross_context_route_cursors",
&["retry_first_failed_at"],
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(add_columns_patch(
&schema,
"cross_context_route_cursors",
&["retry_next_attempt_at"],
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,
attempt_count INT NOT NULL DEFAULT 0,
first_failed_at TIMESTAMPTZ,
next_attempt_at TIMESTAMPTZ,
last_error TEXT
)"
)
}
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(add_columns_patch(
&schema,
"deadlines",
&["firing_at"],
format!("ALTER TABLE {schema}.deadlines ADD COLUMN IF NOT EXISTS firing_at TIMESTAMPTZ"),
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(add_columns_patch(
&schema,
"deadlines",
&[
"attempt_count",
"first_failed_at",
"next_attempt_at",
"last_error",
],
format!(
"ALTER TABLE {schema}.deadlines \
ADD COLUMN IF NOT EXISTS attempt_count INT NOT NULL DEFAULT 0, \
ADD COLUMN IF NOT EXISTS first_failed_at TIMESTAMPTZ, \
ADD COLUMN IF NOT EXISTS next_attempt_at TIMESTAMPTZ, \
ADD COLUMN IF NOT EXISTS last_error TEXT"
),
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(create_index_patch(
&schema,
"deadlines_due",
format!(
"CREATE INDEX IF NOT EXISTS deadlines_due ON {schema}.deadlines (fire_at) \
WHERE status = 'pending'"
),
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(create_index_patch(
&schema,
"deadlines_tags_gin",
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(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RegistrationTable {
EventTypes,
CommandTypes,
Projections,
}
impl RegistrationTable {
fn table(self) -> &'static str {
match self {
RegistrationTable::EventTypes => "event_types",
RegistrationTable::CommandTypes => "command_types",
RegistrationTable::Projections => "projections",
}
}
}
async fn add_registration_version_columns<'e>(
executor: impl sqlx::PgExecutor<'e>,
bounded_context: &str,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {schema}.event_types ADD COLUMN IF NOT EXISTS registered_by_version BIGINT;
ALTER TABLE {schema}.command_types ADD COLUMN IF NOT EXISTS registered_by_version BIGINT;
ALTER TABLE {schema}.projections ADD COLUMN IF NOT EXISTS registered_by_version BIGINT"
)))
.execute(executor)
.await?;
Ok(())
}
pub async fn ensure_registration_version_columns(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<()> {
let (present,): (i64,) = sqlx::query_as(
"SELECT count(*) FROM information_schema.columns \
WHERE table_schema = $1 AND column_name = 'registered_by_version' \
AND table_name IN ('event_types', 'command_types', 'projections')",
)
.bind(format!("bc_{bounded_context}"))
.fetch_one(pool)
.await?;
if present < 3 {
add_registration_version_columns(pool, bounded_context).await?;
}
Ok(())
}
pub async fn registration_version(
pool: &Pool,
bounded_context: &str,
table: RegistrationTable,
name: &str,
) -> crate::error::Result<Option<i64>> {
let schema = schema_ident(bounded_context);
let row: Option<(Option<i64>,)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT registered_by_version FROM {schema}.{} WHERE name = $1",
table.table()
)))
.bind(name)
.fetch_optional(pool)
.await?;
Ok(row.and_then(|(version,)| version))
}
pub async fn set_registration_version(
pool: &Pool,
bounded_context: &str,
table: RegistrationTable,
name: &str,
version: i64,
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.{} SET registered_by_version = $1 WHERE name = $2",
table.table()
)))
.bind(version)
.bind(name)
.execute(pool)
.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(add_columns_patch(
&schema,
"projection_state",
&["owner"],
format!("ALTER TABLE {schema}.projection_state ADD COLUMN IF NOT EXISTS owner TEXT"),
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(add_columns_patch(
&schema,
"projection_rebuild_state",
&["owner"],
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(add_columns_patch(
&schema,
"projection_state",
&["as_of_sequence"],
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(add_columns_patch(
&schema,
"projection_rebuild_state",
&["as_of_sequence"],
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(add_columns_patch(
&schema,
"read_cursors",
&["checked_out_at"],
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(add_columns_patch(
&schema,
"event_types",
&["owner_tag_key"],
format!("ALTER TABLE {schema}.event_types ADD COLUMN IF NOT EXISTS owner_tag_key TEXT"),
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(add_columns_patch(
&schema,
"access_tokens",
&["scope"],
format!("ALTER TABLE {schema}.access_tokens ADD COLUMN IF NOT EXISTS scope TEXT"),
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(add_columns_patch(
&schema,
"command_types",
&["owner_tag_key"],
format!("ALTER TABLE {schema}.command_types ADD COLUMN IF NOT EXISTS owner_tag_key TEXT"),
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(add_columns_patch(
&schema,
"snapshots",
&["owner"],
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(add_columns_patch(
&schema,
"event_types",
&["private_fields"],
format!("ALTER TABLE {schema}.event_types ADD COLUMN IF NOT EXISTS private_fields JSONB NOT NULL DEFAULT '[]'"),
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(add_columns_patch(
&schema,
"command_types",
&["private_fields"],
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(add_columns_patch(
&schema,
"events",
&["metadata_correlation_id"],
format!(
"ALTER TABLE {schema}.events ADD COLUMN IF NOT EXISTS metadata_correlation_id TEXT"
),
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(add_columns_patch(
&schema,
"events",
&["metadata_causation_id"],
format!("ALTER TABLE {schema}.events ADD COLUMN IF NOT EXISTS metadata_causation_id TEXT"),
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(add_columns_patch(
&schema,
"commands",
&["metadata_correlation_id"],
format!(
"ALTER TABLE {schema}.commands ADD COLUMN IF NOT EXISTS metadata_correlation_id TEXT"
),
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(add_columns_patch(
&schema,
"commands",
&["metadata_causation_id"],
format!(
"ALTER TABLE {schema}.commands ADD COLUMN IF NOT EXISTS metadata_causation_id TEXT"
),
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(create_index_patch(
&schema,
"events_by_correlation_id",
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(create_index_patch(
&schema,
"commands_by_correlation_id",
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(add_columns_patch(
&schema,
"access_tokens",
&["start_from"],
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(add_columns_patch(
&schema,
"access_tokens",
&["start_at_sequence"],
format!(
"ALTER TABLE {schema}.access_tokens ADD COLUMN IF NOT EXISTS start_at_sequence BIGINT"
),
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(add_columns_patch(
&schema,
"access_tokens",
&["start_at_time"],
format!(
"ALTER TABLE {schema}.access_tokens ADD COLUMN IF NOT EXISTS start_at_time TIMESTAMPTZ"
),
)))
.execute(pool)
.await?;
sqlx::query(sqlx::AssertSqlSafe(add_constraint_patch(
&schema,
"access_tokens",
"access_tokens_start_from_check",
format!(
"ALTER TABLE {schema}.access_tokens ADD CONSTRAINT access_tokens_start_from_check \
CHECK (start_from IN ('beginning', 'latest', 'at_sequence', 'at_time'))"
),
)))
.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(())
}
pub async fn insert_superadmin_if_none_active(
pool: &Pool,
role: &Role,
) -> crate::error::Result<bool> {
let mut tx = pool.begin().await?;
sqlx::query("SELECT pg_advisory_xact_lock(hashtext('skilj_create_superadmin')::bigint)")
.execute(&mut *tx)
.await?;
let (exists,): (bool,) = sqlx::query_as(
"SELECT EXISTS (SELECT 1 FROM roles WHERE superadmin AND status = 'active')",
)
.fetch_one(&mut *tx)
.await?;
if exists {
return Ok(false);
}
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(&mut *tx)
.await?;
tx.commit().await?;
Ok(true)
}
#[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 active_roles_by_external_subject(
pool: &Pool,
external_subject: &str,
) -> crate::error::Result<Vec<Role>> {
let rows: Vec<RoleRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT {ROLE_COLUMNS} FROM roles WHERE external_subject = $1 AND status = 'active'"
)))
.bind(external_subject)
.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,
}
}
async fn require_bounded_context(pool: &Pool, name: &str) -> crate::error::Result<BoundedContext> {
get_bounded_context(pool, name)
.await?
.ok_or_else(crate::error::Error::row_not_found)
}
async fn require_event_type(
pool: &Pool,
bounded_context: &str,
name: &str,
) -> crate::error::Result<EventType> {
get_event_type(pool, bounded_context, name)
.await?
.ok_or_else(crate::error::Error::row_not_found)
}
async fn require_command_type(
pool: &Pool,
bounded_context: &str,
name: &str,
) -> crate::error::Result<CommandType> {
get_command_type(pool, bounded_context, name)
.await?
.ok_or_else(crate::error::Error::row_not_found)
}
async fn require_command(
pool: &Pool,
bounded_context: &str,
id: i64,
) -> crate::error::Result<Command> {
get_command_by_id(pool, bounded_context, id)
.await?
.ok_or_else(crate::error::Error::row_not_found)
}
#[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 = CASE \
WHEN EXCLUDED.system_triggered_allowed \
AND NOT {schema}.event_types.system_triggered_allowed \
THEN EXCLUDED.schedule_position \
ELSE {schema}.event_types.schedule_position END, \
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(
executor: impl sqlx::PgExecutor<'_>,
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(executor)
.await?;
Ok(row.map(|r| r.into_domain(bc.clone())))
}
async fn get_event_types_by_names_with_bc(
executor: impl sqlx::PgExecutor<'_>,
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(executor)
.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_projection_names(&mut tx, 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 = require_bounded_context(pool, bounded_context).await?;
get_or_create_encryption_key_on(
&mut *pool.acquire().await?,
bc,
subject_key,
subject_value,
master_key,
)
.await
}
async fn get_or_create_encryption_key_on(
conn: &mut sqlx::PgConnection,
bc: BoundedContext,
subject_key: &str,
subject_value: &str,
master_key: &EncryptionMasterKey,
) -> crate::error::Result<(EncryptionKey, i64, DataKey)> {
let bounded_context = bc.name.clone();
let bounded_context = bounded_context.as_str();
let schema = schema_ident(bounded_context);
const ATTEMPTS: usize = 5;
for _ in 0..ATTEMPTS {
if let Some(row) =
get_active_encryption_key_row(&mut *conn, 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(&mut *conn)
.await?;
if let Some(row) = inserted {
return Ok((row.to_domain(bc), row.id, data_key));
}
}
Err(crate::error::Error::row_not_found())
}
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), Option<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;
}
let 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(
executor: impl sqlx::PgExecutor<'_>,
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(executor)
.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 =
require_command_type(pool, bounded_context, &self.command_type_name).await?;
Ok(self.with_type(command_type))
}
fn with_type(self, command_type: CommandType) -> Command {
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,
}
}
}
async fn commands_from_rows(
pool: &Pool,
bounded_context: &str,
rows: Vec<CommandRow>,
) -> crate::error::Result<Vec<Command>> {
if rows.is_empty() {
return Ok(Vec::new());
}
let bc = require_bounded_context(pool, bounded_context).await?;
let mut names: Vec<&str> = rows.iter().map(|r| r.command_type_name.as_str()).collect();
names.sort_unstable();
names.dedup();
let command_types = get_command_types_by_names_with_bc(pool, &bc, &names).await?;
rows.into_iter()
.map(|row| {
let command_type = command_types
.get(&row.command_type_name)
.cloned()
.ok_or_else(crate::error::Error::row_not_found)?;
Ok(row.with_type(command_type))
})
.collect()
}
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(
conn: &mut sqlx::PgConnection,
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(&mut *conn)
.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(conn, 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(
executor: impl sqlx::PgExecutor<'_>,
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(executor)
.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),
}
}
pub async fn command_recording_position(
pool: &Pool,
bounded_context: &str,
external_id: &str,
) -> crate::error::Result<Option<i64>> {
let schema = schema_ident(bounded_context);
Ok(sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
"SELECT id FROM {schema}.commands WHERE external_id = $1"
)))
.bind(external_id)
.fetch_optional(pool)
.await?)
}
#[derive(sqlx::FromRow)]
struct PositionedCommandRow {
id: i64,
#[sqlx(flatten)]
row: CommandRow,
}
pub async fn collect_command_page(
pool: &Pool,
bounded_context: &str,
mut after_position: i64,
max_commands: usize,
mut select: impl FnMut(&[Command], usize) -> crate::error::Result<Vec<Command>>,
) -> crate::error::Result<Vec<Command>> {
let chunk_size = max_commands.max(1);
let limit = i64::try_from(chunk_size).unwrap_or(i64::MAX);
let schema = schema_ident(bounded_context);
let mut page = Vec::new();
loop {
let rows: Vec<PositionedCommandRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT id, {COMMAND_COLUMNS} FROM {schema}.commands \
WHERE id > $1 ORDER BY id LIMIT $2"
)))
.bind(after_position)
.bind(limit)
.fetch_all(pool)
.await?;
let exhausted = rows.len() < chunk_size;
if let Some(last) = rows.last() {
after_position = last.id;
}
let chunk = commands_from_rows(
pool,
bounded_context,
rows.into_iter().map(|p| p.row).collect(),
)
.await?;
page.extend(select(&chunk, chunk_size - page.len())?);
if page.len() >= chunk_size || exhausted {
return Ok(page);
}
}
}
#[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 ORDER BY id"
)))
.fetch_all(pool)
.await?;
commands_from_rows(pool, bounded_context, rows).await
}
#[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 {
event_types.push(require_event_type(pool, bounded_context, &event_type_name).await?);
}
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 mut tx = pool.begin().await?;
upsert_projection_in_tx(&mut tx, projection).await?;
tx.commit().await?;
notify_registration_changed(pool).await;
Ok(())
}
async fn upsert_projection_in_tx(
tx: &mut Transaction<'_, Postgres>,
projection: &Projection,
) -> crate::error::Result<()> {
let schema = schema_ident(&projection.bounded_context.name);
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(
tx,
&projection.bounded_context.name,
"projection_consumed_event_types",
&projection.name,
&projection.consumed_event_types,
)
.await
}
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 projection_caught_up_to(
pool: &Pool,
bounded_context: &str,
name: &str,
) -> crate::error::Result<Option<Option<i64>>> {
let schema = schema_ident(bounded_context);
let row: Option<(Option<i64>,)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT caught_up_to FROM {schema}.projections WHERE name = $1"
)))
.bind(name)
.fetch_optional(pool)
.await?;
Ok(row.map(|(caught_up_to,)| caught_up_to))
}
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_projection_names(
conn: &mut sqlx::PgConnection,
bounded_context: &str,
) -> crate::error::Result<Vec<String>> {
let schema = schema_ident(bounded_context);
Ok(sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
"SELECT name FROM {schema}.projections WHERE sync ORDER BY name"
)))
.fetch_all(conn)
.await?)
}
#[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 {
event_types.push(require_event_type(pool, bounded_context, &event_type_name).await?);
}
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?;
mappings_from_rows(pool, rows, None).await
}
#[tracing::instrument(skip_all, fields(bounded_context = %bounded_context.name))]
pub async fn list_active_role_access_mappings_for_bounded_context(
pool: &Pool,
bounded_context: &BoundedContext,
) -> 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 bounded_context = $1 AND status = 'active' ORDER BY id"
)))
.bind(&bounded_context.name)
.fetch_all(pool)
.await?;
mappings_from_rows(pool, rows, Some(bounded_context)).await
}
async fn mappings_from_rows(
pool: &Pool,
rows: Vec<RoleAccessMappingRow>,
known: Option<&BoundedContext>,
) -> crate::error::Result<Vec<RoleAccessMapping>> {
if rows.is_empty() {
return Ok(Vec::new());
}
let mut role_ids: Vec<&str> = rows.iter().map(|r| r.role_id.as_str()).collect();
role_ids.sort_unstable();
role_ids.dedup();
let roles: std::collections::HashMap<String, Role> =
sqlx::query_as::<_, RoleRow>(sqlx::AssertSqlSafe(format!(
"SELECT {ROLE_COLUMNS} FROM roles WHERE id = ANY($1)"
)))
.bind(&role_ids)
.fetch_all(pool)
.await?
.into_iter()
.map(|row| {
let role = row.into_domain();
(role.id.clone(), role)
})
.collect();
let mut contexts: std::collections::HashMap<String, Option<BoundedContext>> =
std::collections::HashMap::new();
if let Some(bc) = known {
contexts.insert(bc.name.clone(), Some(bc.clone()));
}
for row in &rows {
if !contexts.contains_key(&row.bounded_context) {
let bc = get_bounded_context(pool, &row.bounded_context).await?;
contexts.insert(row.bounded_context.clone(), bc);
}
}
Ok(rows
.into_iter()
.filter_map(|row| {
Some(RoleAccessMapping {
role: roles.get(&row.role_id)?.clone(),
bounded_context: contexts.get(&row.bounded_context)?.clone()?,
level: access_level_from_str(&row.level),
can_read_sensitive: row.can_read_sensitive,
scope: row.scope,
status: role_status_from_str(&row.status),
created_at: row.created_at,
revoked_at: row.revoked_at,
})
})
.collect())
}
pub async fn list_accessible_bounded_context_names(
pool: &Pool,
role_id: &str,
) -> crate::error::Result<std::collections::BTreeSet<String>> {
let names: Vec<String> = sqlx::query_scalar(
"SELECT DISTINCT bounded_context FROM role_access_mappings \
WHERE role_id = $1 AND status = 'active'",
)
.bind(role_id)
.fetch_all(pool)
.await?;
Ok(names.into_iter().collect())
}
#[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?;
mappings_from_rows(pool, rows, None).await
}
#[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 events_table_identity(
pool: &Pool,
bounded_context: &str,
) -> crate::error::Result<(i64, Option<i64>)> {
let schema = schema_ident(bounded_context);
let (oid, max): (i64, Option<i64>) = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT '{schema}.events'::regclass::oid::bigint, \
(SELECT MAX(sequence) FROM {schema}.events)"
)))
.fetch_one(pool)
.await?;
Ok((oid, max))
}
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)
}
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 = require_command(pool, bounded_context, command_id).await?;
EventOrigin::CommandTriggered {
command: Box::new(command),
}
}
other => panic!("events row has unknown origin_kind {other:?}"),
})
}
#[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 = require_bounded_context(pool, bounded_context).await?;
let et = require_event_type(pool, bounded_context, event_type_name).await?;
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 event_type_name = $1 ORDER BY sequence"
)))
.bind(event_type_name)
.fetch_all(pool)
.await?;
let known = std::collections::HashMap::from([(et.name.clone(), et)]);
events_from_rows(&mut *pool.acquire().await?, &bc, rows, Some(&known)).await
}
#[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 = require_bounded_context(pool, bounded_context).await?;
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?;
events_from_rows(&mut *pool.acquire().await?, &bc, rows, None).await
}
#[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 = require_bounded_context(pool, bounded_context).await?;
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?;
events_from_rows(&mut *pool.acquire().await?, &bc, rows, None).await
}
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 = require_bounded_context(pool, bounded_context).await?;
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?;
events_from_rows(&mut *pool.acquire().await?, &bc, rows, None).await
}
#[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 = require_bounded_context(pool, bounded_context).await?;
list_events_for_bounded_context_matching_tags_with_bc(
&mut *pool.acquire().await?,
&bc,
tags,
after_sequence,
None,
None,
)
.await
}
async fn list_events_for_bounded_context_matching_tags_with_bc(
conn: &mut sqlx::PgConnection,
bc: &BoundedContext,
tags: &[Tag],
after_sequence: Option<i64>,
known_event_types: Option<&std::collections::HashMap<String, EventType>>,
limit: Option<i64>,
) -> 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 limit_clause = match limit {
Some(_) => format!(
" LIMIT ${}",
sequence_param + usize::from(after_sequence.is_some())
),
None => String::new(),
};
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{limit_clause}"
);
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);
}
if let Some(limit) = limit {
query = query.bind(limit);
}
let rows: Vec<EventRowAnyType> = query.fetch_all(&mut *conn).await?;
events_from_rows(conn, bc, rows, known_event_types).await
}
async fn events_from_rows(
conn: &mut sqlx::PgConnection,
bc: &BoundedContext,
rows: Vec<EventRowAnyType>,
known_event_types: Option<&std::collections::HashMap<String, EventType>>,
) -> crate::error::Result<Vec<Event>> {
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(&mut *conn, 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(conn, 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::debug!(
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(),
"event rows resolved"
);
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 = require_event_type(pool, bounded_context, &row.event_type_name).await?;
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 = require_bounded_context(pool, bounded_context).await?;
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 events = events_from_rows(&mut *pool.acquire().await?, &bc, rows, None).await?;
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 = require_bounded_context(pool, bounded_context).await?;
let et = require_event_type(pool, bounded_context, event_type_name).await?;
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 event_type_name = $1 AND sequence > $2 \
ORDER BY sequence"
)))
.bind(event_type_name)
.bind(after_sequence)
.fetch_all(pool)
.await?;
let known = std::collections::HashMap::from([(et.name.clone(), et)]);
events_from_rows(&mut *pool.acquire().await?, &bc, rows, Some(&known)).await
}
pub async fn witness_events_of_types(
pool: &Pool,
bounded_context: &str,
event_types: &[EventType],
) -> crate::error::Result<Vec<Event>> {
let mut witnesses = Vec::new();
for event_type in event_types {
witnesses.extend(
list_events_from_limited(pool, bounded_context, &event_type.name, -1, 1).await?,
);
}
Ok(witnesses)
}
pub const MAX_EVENTS_PER_CATCH_UP_TICK: i64 = 1000;
async fn list_events_cached_limited(
pool: &Pool,
cache: &crate::event_cache::EventCache,
bounded_context: &str,
event_type_name: &str,
after_sequence: i64,
limit: 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)
.take(usize::try_from(limit).unwrap_or(usize::MAX))
.collect()),
None => {
list_events_from_limited(
pool,
bounded_context,
event_type_name,
after_sequence,
limit,
)
.await
}
}
}
async fn highest_sequence_of_type(
pool: &Pool,
bounded_context: &str,
event_type_name: &str,
created_at_or_before: Option<DateTime<Utc>>,
) -> crate::error::Result<i64> {
let schema = schema_ident(bounded_context);
let (max,): (Option<i64>,) = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT MAX(sequence) FROM {schema}.events WHERE event_type_name = $1 \
AND ($2::timestamptz IS NULL OR metadata_created_at <= $2)"
)))
.bind(event_type_name)
.bind(created_at_or_before)
.fetch_one(pool)
.await?;
Ok(max.unwrap_or(-1))
}
pub async fn list_events_from_limited(
pool: &Pool,
bounded_context: &str,
event_type_name: &str,
after_sequence: i64,
limit: i64,
) -> crate::error::Result<Vec<Event>> {
let bc = require_bounded_context(pool, bounded_context).await?;
let et = require_event_type(pool, bounded_context, event_type_name).await?;
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 event_type_name = $1 AND sequence > $2 \
ORDER BY sequence LIMIT $3"
)))
.bind(event_type_name)
.bind(after_sequence)
.bind(limit)
.fetch_all(pool)
.await?;
let known = std::collections::HashMap::from([(et.name.clone(), et)]);
events_from_rows(&mut *pool.acquire().await?, &bc, rows, Some(&known)).await
}
#[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,
}
}
pub async fn for_each_event_chunk(
pool: &Pool,
cache: &crate::event_cache::EventCache,
bounded_context: &str,
event_type_name: Option<&str>,
mut after_sequence: i64,
chunk_size: usize,
mut visit: impl FnMut(&[Event]) -> crate::error::Result<bool>,
) -> crate::error::Result<()> {
let chunk_size = chunk_size.max(1);
let limit = i64::try_from(chunk_size).unwrap_or(i64::MAX);
loop {
let chunk = match cache
.try_events_after(pool, bounded_context, after_sequence)
.await?
{
Some(events) => {
let chunk: Vec<Event> = events
.into_iter()
.filter(|e| event_type_name.is_none_or(|name| e.event_type.name == name))
.take(chunk_size)
.collect();
let exhausted = chunk.len() < chunk_size;
(chunk, exhausted)
}
None => {
let chunk = match event_type_name {
Some(name) => {
list_events_from_limited(pool, bounded_context, name, after_sequence, limit)
.await?
}
None => {
list_events_for_bounded_context_from_limited(
pool,
bounded_context,
after_sequence,
limit,
)
.await?
}
};
let exhausted = chunk.len() < chunk_size;
(chunk, exhausted)
}
};
let (chunk, exhausted) = chunk;
if let Some(last) = chunk.last() {
after_sequence = last.sequence;
}
if !visit(&chunk)? || exhausted {
return Ok(());
}
}
}
#[allow(clippy::too_many_arguments)]
pub async fn collect_event_page(
pool: &Pool,
cache: &crate::event_cache::EventCache,
bounded_context: &str,
event_type_name: Option<&str>,
after_sequence: i64,
max_events: usize,
select: impl FnMut(&[Event], usize) -> crate::error::Result<Vec<Event>>,
) -> crate::error::Result<Vec<Event>> {
Ok(collect_scanned_event_page(
pool,
cache,
bounded_context,
event_type_name,
after_sequence,
max_events,
select,
)
.await?
.events)
}
#[derive(Debug, Clone, PartialEq)]
pub struct ScannedEventPage {
pub events: Vec<Event>,
pub scanned_through: Option<i64>,
}
#[allow(clippy::too_many_arguments)]
pub async fn collect_scanned_event_page(
pool: &Pool,
cache: &crate::event_cache::EventCache,
bounded_context: &str,
event_type_name: Option<&str>,
after_sequence: i64,
max_events: usize,
mut select: impl FnMut(&[Event], usize) -> crate::error::Result<Vec<Event>>,
) -> crate::error::Result<ScannedEventPage> {
let mut page = Vec::new();
let mut scanned_through = None;
for_each_event_chunk(
pool,
cache,
bounded_context,
event_type_name,
after_sequence,
max_events,
|chunk| {
if let Some(last) = chunk.last() {
scanned_through = Some(last.sequence);
}
page.extend(select(chunk, max_events - page.len())?);
Ok(page.len() < max_events)
},
)
.await?;
Ok(ScannedEventPage {
events: page,
scanned_through,
})
}
pub async fn for_each_tagged_event_chunk(
pool: &Pool,
bounded_context: &str,
tags: &[Tag],
after_sequence: i64,
chunk_size: usize,
mut f: impl FnMut(&[Event]) -> crate::error::Result<bool>,
) -> crate::error::Result<()> {
let bc = require_bounded_context(pool, bounded_context).await?;
let chunk_size = chunk_size.max(1);
let limit = i64::try_from(chunk_size).unwrap_or(i64::MAX);
let mut after = after_sequence;
loop {
let chunk = list_events_for_bounded_context_matching_tags_with_bc(
&mut *pool.acquire().await?,
&bc,
tags,
Some(after),
None,
Some(limit),
)
.await?;
if !f(&chunk)? || chunk.len() < chunk_size {
return Ok(());
}
after = chunk.last().map_or(after, |e| e.sequence);
}
}
pub async fn collect_tagged_event_page(
pool: &Pool,
bounded_context: &str,
tags: &[Tag],
after_sequence: i64,
max_events: usize,
mut select: impl FnMut(&[Event], usize) -> crate::error::Result<Vec<Event>>,
) -> crate::error::Result<Vec<Event>> {
let mut page = Vec::new();
for_each_tagged_event_chunk(
pool,
bounded_context,
tags,
after_sequence,
max_events,
|chunk| {
page.extend(select(chunk, max_events - page.len())?);
Ok(page.len() < max_events)
},
)
.await?;
Ok(page)
}
#[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 mut tx = pool.begin().await?;
let sync_projections = sync_projection_names(&mut tx, &event.bounded_context.name).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: &[String],
) -> 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_name in sync_projections {
if dispatcher
.default_state(bounded_context, projection_name)
.is_none()
{
let consumes: bool = sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
"SELECT EXISTS (SELECT 1 FROM {schema}.projection_consumed_event_types \
WHERE projection_name = $1 AND event_type_name = $2)"
)))
.bind(projection_name)
.bind(&event.event_type.name)
.fetch_one(&mut **tx)
.await?;
if consumes {
return Err(crate::event_store::Error::SyncProjectionNotDeclared(
projection_name.clone(),
)
.into());
}
}
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> {
crate::event_store::check_create_external_event(
adapter,
&payload,
correlation_id.as_deref(),
causation_id.as_deref(),
)?;
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_projection_names(&mut tx, &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> {
crate::event_store::check_create_direct_event(
adapter,
&payload,
correlation_id.as_deref(),
causation_id.as_deref(),
)?;
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_projection_names(&mut tx, &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_projection_names(&mut tx, &bounded_context_name).await?;
let outcome = submit_one_command_in_tx(
&mut tx,
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(
conn: &mut sqlx::PgConnection,
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],
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 {
if let Some(triggered_event_sequences) =
lookup_idempotency_key(&mut *conn, &schema, &command_type.name, client_id, key).await?
{
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 mut result = list_events_for_bounded_context_matching_tags_with_bc(
&mut *conn,
&command_type.bounded_context,
consistency_tags,
Some(original_highest),
Some(&event_types_by_name),
None,
)
.await?;
result.retain(|e| e.sequence <= locked_highest);
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_with_bc(&mut *conn, &command_type.bounded_context, &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>,
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: &[String],
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)?;
for subject in needed_subjects {
let provisioned = get_or_create_encryption_key_on(
tx,
command_type.bounded_context.clone(),
&subject.0,
&subject.1,
master_key,
)
.await?;
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>,
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: &[String],
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(
tx,
dispatcher,
command_type,
payload,
client_id,
bounded_context_events,
consistency_tags,
matching_events,
initial_decision,
snapshot,
idempotency_key,
locked_highest,
extra_committed_events,
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,
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<String>,
}
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_projection_names(&mut tx, 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,
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 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(
&mut tx,
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,
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 persist_started = std::time::Instant::now();
let mut nested = tx.begin().await?;
let outcome = finish_accepted_command_in_tx(
&mut nested,
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());
}
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,
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(())
}
async fn defer_cross_context_route(
pool: &Pool,
source_bounded_context: &str,
route_name: &str,
current_cursor: i64,
next_attempt_at: 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_next_attempt_at) \
VALUES ($1, $2, $3, $4) \
ON CONFLICT (route_name) DO UPDATE SET \
retry_next_attempt_at = EXCLUDED.retry_next_attempt_at"
)))
.bind(route_name)
.bind(current_cursor)
.bind(Utc::now())
.bind(next_attempt_at)
.execute(pool)
.await?;
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ParkedDeliveryKind {
CrossContextRoute,
ExternalEvent,
CommandTrigger,
Deadline,
}
impl ParkedDeliveryKind {
fn as_str(self) -> &'static str {
match self {
ParkedDeliveryKind::CrossContextRoute => "cross_context_route",
ParkedDeliveryKind::ExternalEvent => "external_event",
ParkedDeliveryKind::CommandTrigger => "command_trigger",
ParkedDeliveryKind::Deadline => "deadline",
}
}
}
fn parked_delivery_kind_from_str(s: &str) -> ParkedDeliveryKind {
match s {
"external_event" => ParkedDeliveryKind::ExternalEvent,
"command_trigger" => ParkedDeliveryKind::CommandTrigger,
"deadline" => ParkedDeliveryKind::Deadline,
_ => 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);
let index = create_index_patch(
&schema,
"parked_deliveries_page_order",
format!(
"CREATE INDEX IF NOT EXISTS parked_deliveries_page_order \
ON {schema}.parked_deliveries (first_failed_at DESC, id DESC)"
),
);
sqlx::raw_sql(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
);
{index}"
)))
.execute(executor)
.await?;
Ok(())
}
const PARKED_DELIVERIES_LEGACY_DEDUP_INDEX: &str = "parked_deliveries_source_kind_identifier_key";
const PARKED_DELIVERIES_OCCURRENCE_INDEX: &str = "parked_deliveries_occurrence_key";
fn parked_deliveries_occurrence_index_ddl(schema: &str) -> String {
format!(
"CREATE UNIQUE INDEX IF NOT EXISTS {PARKED_DELIVERIES_OCCURRENCE_INDEX} \
ON {schema}.parked_deliveries (source, kind, identifier, (COALESCE(access_token_id, '')))"
)
}
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 (has_current, has_legacy): (bool, bool) = sqlx::query_as(
"SELECT \
EXISTS (SELECT 1 FROM pg_indexes WHERE schemaname = $1 AND indexname = $2), \
EXISTS (SELECT 1 FROM pg_indexes WHERE schemaname = $1 AND indexname = $3)",
)
.bind(&raw_schema)
.bind(PARKED_DELIVERIES_OCCURRENCE_INDEX)
.bind(PARKED_DELIVERIES_LEGACY_DEDUP_INDEX)
.fetch_one(&mut *tx)
.await?;
if has_current && !has_legacy {
tx.commit().await?;
return Ok(());
}
if !has_current {
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.access_token_id IS NOT DISTINCT FROM b.access_token_id \
AND (a.last_failed_at, a.id) < (b.last_failed_at, b.id)"
)))
.execute(&mut *tx)
.await?;
sqlx::query(sqlx::AssertSqlSafe(parked_deliveries_occurrence_index_ddl(
&schema,
)))
.execute(&mut *tx)
.await?;
}
sqlx::query(sqlx::AssertSqlSafe(format!(
"DROP INDEX IF EXISTS {schema}.{PARKED_DELIVERIES_LEGACY_DEDUP_INDEX}"
)))
.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, (COALESCE(access_token_id, ''))) 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>> {
list_parked_deliveries_page(pool, bounded_context, None, i64::MAX).await
}
pub async fn list_parked_deliveries_page(
pool: &Pool,
bounded_context: &str,
after: Option<&ParkedDeliveryCursor>,
limit: i64,
) -> 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 \
WHERE $1::timestamptz IS NULL OR (first_failed_at, id) < ($1, $2) \
ORDER BY first_failed_at DESC, id DESC LIMIT $3"
)))
.bind(after.map(|c| c.first_failed_at))
.bind(after.map(|c| c.id.clone()))
.bind(limit)
.fetch_all(pool)
.await?;
Ok(rows.into_iter().map(ParkedDelivery::from).collect())
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ParkedDeliveryCursor {
pub first_failed_at: DateTime<Utc>,
pub id: String,
}
impl ParkedDeliveryCursor {
pub fn of(delivery: &ParkedDelivery) -> Self {
Self {
first_failed_at: delivery.first_failed_at,
id: delivery.id.clone(),
}
}
pub fn encode(&self) -> String {
format!("{}:{}", self.first_failed_at.timestamp_micros(), self.id)
}
pub fn decode(cursor: &str) -> Option<Self> {
let (micros, id) = cursor.split_once(':')?;
Some(Self {
first_failed_at: DateTime::from_timestamp_micros(micros.parse().ok()?)?,
id: id.to_string(),
})
}
}
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 const CROSS_CONTEXT_ROUTE_CLIENT_ID: &str = "cross-context-route";
const CROSS_CONTEXT_ROUTE_PARKED_SOURCE_PREFIX: &str = "cross-context-route:";
fn cross_context_route_idempotency_key(route_name: &str, source_sequence: &str) -> String {
format!(
"{}{route_name}:{source_sequence}",
crate::event_store::RESERVED_IDEMPOTENCY_KEY_PREFIX
)
}
pub fn parked_delivery_redrive_identity(
delivery: &ParkedDelivery,
caller_client_id: &str,
original_idempotency_key: Option<&str>,
) -> Option<(String, String)> {
match delivery.kind {
ParkedDeliveryKind::CrossContextRoute => {
match delivery
.source
.strip_prefix(CROSS_CONTEXT_ROUTE_PARKED_SOURCE_PREFIX)
{
Some(route_name) => Some((
CROSS_CONTEXT_ROUTE_CLIENT_ID.to_string(),
cross_context_route_idempotency_key(route_name, &delivery.identifier),
)),
None => Some((
CROSS_CONTEXT_ROUTE_CLIENT_ID.to_string(),
format!(
"{}{}",
crate::event_store::RESERVED_PARKED_DELIVERY_IDEMPOTENCY_KEY_PREFIX,
delivery.id
),
)),
}
}
ParkedDeliveryKind::CommandTrigger => Some((
caller_client_id.to_string(),
match original_idempotency_key {
Some(key) => key.to_string(),
None => format!(
"{}{}",
crate::event_store::RESERVED_PARKED_DELIVERY_IDEMPOTENCY_KEY_PREFIX,
delivery.id
),
},
)),
ParkedDeliveryKind::Deadline => Some((
DEADLINE_CLIENT_ID.to_string(),
format!(
"{}{}",
crate::event_store::RESERVED_DEADLINE_IDEMPOTENCY_KEY_PREFIX,
delivery.identifier
),
)),
ParkedDeliveryKind::ExternalEvent => None,
}
}
pub fn parked_delivery_redrive_dedupe_partition_key(delivery: &ParkedDelivery) -> String {
format!(
"{}{}",
crate::event_store::RESERVED_PARKED_DELIVERY_IDEMPOTENCY_KEY_PREFIX,
delivery.id
)
}
pub async fn try_lock_parked_delivery_for_retry(
pool: &Pool,
bounded_context: &str,
id: &str,
) -> crate::error::Result<Option<Transaction<'static, Postgres>>> {
let mut lock_tx = pool.begin().await?;
let acquired: bool =
sqlx::query_scalar("SELECT pg_try_advisory_xact_lock(hashtext($1)::bigint)")
.bind(format!("parked_delivery:{bounded_context}:{id}"))
.fetch_one(&mut *lock_tx)
.await?;
Ok(acquired.then_some(lock_tx))
}
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_deliveries_for_subject(
pool: &Pool,
bounded_context: &str,
subject_key: &str,
subject_value: &str,
) -> crate::error::Result<u64> {
let mut deleted = 0;
let mut after: Option<ParkedDeliveryCursor> = None;
let mut rules_cache = ParkedPayloadRulesCache::default();
loop {
let page = list_parked_deliveries_page(pool, bounded_context, after.as_ref(), 1000).await?;
let Some(last) = page.last() else {
return Ok(deleted);
};
after = Some(ParkedDeliveryCursor::of(last));
for delivery in &page {
if parked_delivery_references_subject(
pool,
&mut rules_cache,
delivery,
subject_key,
subject_value,
)
.await?
&& delete_parked_delivery(pool, bounded_context, &delivery.id)
.await?
.is_some()
{
deleted += 1;
}
}
}
}
async fn parked_delivery_references_subject(
pool: &Pool,
rules_cache: &mut ParkedPayloadRulesCache,
delivery: &ParkedDelivery,
subject_key: &str,
subject_value: &str,
) -> crate::error::Result<bool> {
let rules = rules_cache.get(pool, delivery).await?;
Ok(match (rules, parked_delivery_payload(delivery)) {
(Some(rules), Some(payload)) => {
let payload = serde_json::to_string(payload)
.expect("serde_json::Value serialization is infallible");
crate::event_store::sensitive_field_subjects(&rules.sensitive_fields, &payload)
.iter()
.any(|(key, value)| key == subject_key && value == subject_value)
}
_ => json_mentions(&delivery.request_json, subject_value),
})
}
#[derive(Debug, Clone)]
pub struct ParkedPayloadRules {
pub sensitive_fields: Vec<SensitiveField>,
pub private_fields: Vec<PrivateField>,
pub tag_mappings: Vec<TagMapping>,
pub owner_tag_key: Option<String>,
}
pub async fn parked_delivery_payload_rules(
pool: &Pool,
delivery: &ParkedDelivery,
) -> crate::error::Result<Option<ParkedPayloadRules>> {
let from_command = |ct: CommandType| ParkedPayloadRules {
sensitive_fields: ct.sensitive_fields,
private_fields: ct.private_fields,
tag_mappings: ct.tag_mappings,
owner_tag_key: ct.owner_tag_key,
};
Ok(match delivery.kind {
ParkedDeliveryKind::CrossContextRoute | ParkedDeliveryKind::Deadline => match (
&delivery.target_bounded_context,
&delivery.target_command_type,
) {
(Some(bc), Some(name)) => get_command_type(pool, bc, name).await?.map(from_command),
_ => None,
},
ParkedDeliveryKind::ExternalEvent => match &delivery.access_token_id {
Some(id) => get_external_event_token(pool, id).await?.map(|token| {
let et = token.event_type;
ParkedPayloadRules {
sensitive_fields: et.sensitive_fields,
private_fields: et.private_fields,
tag_mappings: et.tag_mappings,
owner_tag_key: et.owner_tag_key,
}
}),
None => None,
},
ParkedDeliveryKind::CommandTrigger => match &delivery.access_token_id {
Some(id) => get_command_token(pool, id)
.await?
.map(|token| from_command(token.command_type)),
None => None,
},
})
}
#[derive(Default)]
pub struct ParkedPayloadRulesCache {
rules: std::collections::HashMap<String, Option<ParkedPayloadRules>>,
}
impl ParkedPayloadRulesCache {
pub async fn get(
&mut self,
pool: &Pool,
delivery: &ParkedDelivery,
) -> crate::error::Result<Option<ParkedPayloadRules>> {
let key = match delivery.kind {
ParkedDeliveryKind::CrossContextRoute | ParkedDeliveryKind::Deadline => format!(
"type:{:?}:{:?}",
delivery.target_bounded_context, delivery.target_command_type
),
ParkedDeliveryKind::ExternalEvent => {
format!("event-token:{:?}", delivery.access_token_id)
}
ParkedDeliveryKind::CommandTrigger => {
format!("command-token:{:?}", delivery.access_token_id)
}
};
if let Some(rules) = self.rules.get(&key) {
return Ok(rules.clone());
}
let rules = parked_delivery_payload_rules(pool, delivery).await?;
self.rules.insert(key, rules.clone());
Ok(rules)
}
}
pub fn parked_delivery_payload(delivery: &ParkedDelivery) -> Option<&serde_json::Value> {
match delivery.kind {
ParkedDeliveryKind::CrossContextRoute | ParkedDeliveryKind::Deadline => {
Some(&delivery.request_json)
}
ParkedDeliveryKind::ExternalEvent | ParkedDeliveryKind::CommandTrigger => {
delivery.request_json.get("payload")
}
}
}
pub fn parked_delivery_visible_to(
delivery: &ParkedDelivery,
rules: Option<&ParkedPayloadRules>,
access_mapping: &RoleAccessMapping,
) -> bool {
if access_mapping.scope.is_none() {
return true;
}
let (Some(rules), Some(payload)) = (rules, parked_delivery_payload(delivery)) else {
return false;
};
let payload =
serde_json::to_string(payload).expect("serde_json::Value serialization is infallible");
crate::event_store::tag_owner_scope_satisfied(
&crate::event_store::derive_tags(&rules.tag_mappings, &payload),
rules.owner_tag_key.as_deref(),
access_mapping.scope.as_deref(),
)
}
pub fn render_parked_delivery(
delivery: &ParkedDelivery,
rules: Option<&ParkedPayloadRules>,
access_mapping: &RoleAccessMapping,
) -> ParkedDelivery {
let mut rendered = delivery.clone();
let payload = match rendered.kind {
ParkedDeliveryKind::CrossContextRoute | ParkedDeliveryKind::Deadline => {
Some(&mut rendered.request_json)
}
ParkedDeliveryKind::ExternalEvent | ParkedDeliveryKind::CommandTrigger => {
rendered.request_json.get_mut("payload")
}
};
let Some(payload) = payload else {
return rendered;
};
let Some(rules) = rules else {
*payload = serde_json::Value::Null;
return rendered;
};
let original = payload.clone();
for sf in &rules.sensitive_fields {
let subject = crate::event_store::payload_field_value(&original, &sf.subject_field)
.and_then(crate::event_store::json_scalar_to_string);
let granted = subject.is_some_and(|subject| {
crate::event_store::sensitive_field_is_granted(access_mapping, &subject)
});
if !granted {
if let Some(leaf) = crate::event_store::payload_field_value_mut(payload, &sf.field) {
*leaf = serde_json::Value::Null;
}
}
}
for pf in &rules.private_fields {
if let Some(leaf) = crate::event_store::payload_field_value_mut(payload, &pf.field) {
*leaf = serde_json::Value::Null;
}
}
rendered
}
pub async fn forget_subject_in_deadlines(
pool: &Pool,
bounded_context: &str,
subject_key: &str,
subject_value: &str,
now: DateTime<Utc>,
) -> crate::error::Result<u64> {
let mut forgotten = 0;
let mut sensitive_fields = std::collections::HashMap::new();
for holder in list_bounded_contexts(pool).await? {
match forget_subject_in_holder_deadlines(
pool,
&holder.name,
bounded_context,
subject_key,
subject_value,
now,
&mut sensitive_fields,
)
.await
{
Ok(n) => forgotten += n,
Err(_) if get_bounded_context(pool, &holder.name).await?.is_none() => {}
Err(err) => return Err(err),
}
}
Ok(forgotten)
}
async fn forget_subject_in_holder_deadlines(
pool: &Pool,
holder: &str,
bounded_context: &str,
subject_key: &str,
subject_value: &str,
now: DateTime<Utc>,
sensitive_fields: &mut std::collections::HashMap<String, Option<Vec<SensitiveField>>>,
) -> crate::error::Result<u64> {
const PAGE: i64 = 1000;
let schema = schema_ident(holder);
let mut forgotten = 0;
let mut after = String::new();
loop {
let rows: Vec<(String, String, String, String)> =
sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT id, target_command_type, payload, status FROM {schema}.deadlines \
WHERE target_bounded_context = $1 AND id > $2 \
AND (status = 'pending' OR (status <> 'firing' AND payload <> '{{}}')) \
ORDER BY id LIMIT $3"
)))
.bind(bounded_context)
.bind(&after)
.bind(PAGE)
.fetch_all(pool)
.await?;
let Some((last_id, _, _, _)) = rows.last() else {
break;
};
after = last_id.clone();
for (id, command_type, payload, status) in &rows {
if !sensitive_fields.contains_key(command_type) {
let fields = get_command_type(pool, bounded_context, command_type)
.await?
.map(|ct| ct.sensitive_fields);
sensitive_fields.insert(command_type.clone(), fields);
}
let parsed = serde_json::from_str::<serde_json::Value>(payload).ok();
let names_subject = match (&sensitive_fields[command_type], &parsed) {
(Some(fields), Some(_)) => {
crate::event_store::sensitive_field_subjects(fields, payload)
.iter()
.any(|(key, value)| key == subject_key && value == subject_value)
}
(_, Some(json)) => json_mentions(json, subject_value),
(_, None) => payload.contains(subject_value),
};
if names_subject && status == "pending" {
forgotten += sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.deadlines \
SET status = 'forgotten', payload = '{{}}', resolved_at = $1 \
WHERE id = $2 AND status = 'pending'"
)))
.bind(now)
.bind(id)
.execute(pool)
.await?
.rows_affected();
} else if names_subject {
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.deadlines SET payload = '{{}}' \
WHERE id = $1 AND status NOT IN ('pending', 'firing')"
)))
.bind(id)
.execute(pool)
.await?;
}
}
if (rows.len() as i64) < PAGE {
break;
}
}
Ok(forgotten)
}
fn json_mentions(json: &serde_json::Value, value: &str) -> bool {
match json {
serde_json::Value::String(s) => s == value,
serde_json::Value::Number(n) => n.to_string() == value,
serde_json::Value::Bool(b) => b.to_string() == value,
serde_json::Value::Array(items) => items.iter().any(|v| json_mentions(v, value)),
serde_json::Value::Object(fields) => fields.values().any(|v| json_mentions(v, value)),
serde_json::Value::Null => false,
}
}
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);
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 => {
highest_sequence_of_type(
pool,
route.source_bounded_context,
route.source_event_type,
None,
)
.await?
}
crate::plugin::CrossContextRouteStartFrom::AtSequence(n) => n,
crate::plugin::CrossContextRouteStartFrom::AtTime(unix_secs) => {
let threshold = DateTime::from_timestamp(unix_secs, 0).unwrap_or(Utc::now());
highest_sequence_of_type(
pool,
route.source_bounded_context,
route.source_event_type,
Some(threshold),
)
.await?
}
};
update_cross_context_route_cursor(
pool,
route.source_bounded_context,
route.name,
seed,
Utc::now(),
)
.await?;
return Ok(());
}
let events = list_events_cached_limited(
pool,
event_cache,
route.source_bounded_context,
route.source_event_type,
cursor,
MAX_EVENTS_PER_CATCH_UP_TICK,
)
.await?;
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 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 =
cross_context_route_idempotency_key(route.name, &event.sequence.to_string());
let outcome = decide_and_submit_command(
pool,
command_dispatcher,
projection_dispatcher,
snapshot_dispatcher,
broadcaster,
event_cache,
&target_command_type,
&target_payload,
CROSS_CONTEXT_ROUTE_CLIENT_ID,
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) if e.another_instance_can_do_it() => {
let next_attempt_at = Utc::now()
+ chrono::Duration::from_std(ANOTHER_INSTANCE_RETRY_DELAY)
.expect("a ten-second delay fits a chrono::Duration");
tracing::warn!(
sequence = event.sequence,
error = %e,
next_attempt_at = %next_attempt_at,
"cross-context route: this instance can't submit the target \
command, another can - retrying later; route blocked until then"
);
defer_cross_context_route(
pool,
route.source_bounded_context,
route.name,
cursor,
next_attempt_at,
)
.await?;
return Ok(());
}
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_PARKED_SOURCE_PREFIX}{}",
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 =
retry_policy.next_attempt_at(now, retry_attempt as u32);
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);
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 => {
highest_sequence_of_type(
pool,
schedule.source_bounded_context,
schedule.source_event_type,
None,
)
.await?
}
crate::plugin::DeadlinePollStartFrom::AtSequence(n) => n,
crate::plugin::DeadlinePollStartFrom::AtTime(unix_secs) => {
let threshold = DateTime::from_timestamp(unix_secs, 0).unwrap_or(Utc::now());
highest_sequence_of_type(
pool,
schedule.source_bounded_context,
schedule.source_event_type,
Some(threshold),
)
.await?
}
};
update_deadline_cursor(
pool,
schedule.source_bounded_context,
schedule.name,
seed,
Utc::now(),
)
.await?;
return Ok(());
}
let events = list_events_cached_limited(
pool,
event_cache,
schedule.source_bounded_context,
schedule.source_event_type,
cursor,
MAX_EVENTS_PER_CATCH_UP_TICK,
)
.await?;
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(())
}
async fn cancel_backlog_start(
pool: &Pool,
cancel: &crate::plugin::CancelDeadlineInfo,
) -> crate::error::Result<Option<DateTime<Utc>>> {
let source_exists: Option<i32> =
sqlx::query_scalar("SELECT 1 FROM bounded_contexts WHERE name = $1")
.bind(cancel.source_bounded_context)
.fetch_optional(pool)
.await?;
if source_exists.is_none() {
return Ok(None);
}
let cursor = get_deadline_cursor(pool, cancel.source_bounded_context, cancel.name)
.await?
.unwrap_or(-1);
let schema = schema_ident(cancel.source_bounded_context);
Ok(sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
"SELECT metadata_created_at FROM {schema}.events \
WHERE event_type_name = $1 AND sequence > $2 ORDER BY sequence LIMIT 1"
)))
.bind(cancel.source_event_type)
.bind(cursor)
.fetch_optional(pool)
.await?)
}
enum ScheduleBacklogStart {
Sequence(i64),
Time(DateTime<Utc>),
}
async fn deadline_schedule_backlog_start(
pool: &Pool,
cancel: &crate::plugin::CancelDeadlineInfo,
) -> crate::error::Result<Option<ScheduleBacklogStart>> {
let cursor = get_deadline_cursor(
pool,
cancel.deadline_schedule_bounded_context,
cancel.deadline_schedule_name,
)
.await?
.unwrap_or(-1);
let schema = schema_ident(cancel.deadline_schedule_bounded_context);
let first: Option<(i64, DateTime<Utc>)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT sequence, metadata_created_at FROM {schema}.events \
WHERE event_type_name = $1 AND sequence > $2 ORDER BY sequence LIMIT 1"
)))
.bind(cancel.deadline_schedule_source_event_type)
.bind(cursor)
.fetch_optional(pool)
.await?;
Ok(first.map(|(sequence, created_at)| {
if cancel.deadline_schedule_bounded_context == cancel.source_bounded_context {
ScheduleBacklogStart::Sequence(sequence)
} else {
ScheduleBacklogStart::Time(created_at)
}
}))
}
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);
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 => {
highest_sequence_of_type(
pool,
cancel.source_bounded_context,
cancel.source_event_type,
None,
)
.await?
}
crate::plugin::DeadlinePollStartFrom::AtSequence(n) => n,
crate::plugin::DeadlinePollStartFrom::AtTime(unix_secs) => {
let threshold = DateTime::from_timestamp(unix_secs, 0).unwrap_or(Utc::now());
highest_sequence_of_type(
pool,
cancel.source_bounded_context,
cancel.source_event_type,
Some(threshold),
)
.await?
}
};
update_deadline_cursor(
pool,
cancel.source_bounded_context,
cancel.name,
seed,
Utc::now(),
)
.await?;
return Ok(());
}
let events = list_events_cached_limited(
pool,
event_cache,
cancel.source_bounded_context,
cancel.source_event_type,
cursor,
MAX_EVENTS_PER_CATCH_UP_TICK,
)
.await?;
let before = deadline_schedule_backlog_start(pool, cancel).await?;
let target_schema = schema_ident(cancel.deadline_schedule_bounded_context);
for event in &events {
let scheduled_past = match &before {
None => true,
Some(ScheduleBacklogStart::Sequence(first_unscheduled)) => {
event.sequence < *first_unscheduled
}
Some(ScheduleBacklogStart::Time(first_unscheduled)) => {
event.metadata.created_at < *first_unscheduled
}
};
if !scheduled_past {
break;
}
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, \
payload = '{{}}' \
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,
schedule_name: String,
fire_at: DateTime<Utc>,
correlation_id: Option<String>,
target_bounded_context: String,
target_command_type: String,
payload: String,
attempt_count: i32,
first_failed_at: Option<DateTime<Utc>>,
}
pub const DEADLINE_PARKED_SOURCE_PREFIX: &str = "deadline:";
pub const DEADLINE_CLIENT_ID: &str = "deadline";
#[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>,
retry_policy: &skilj_retry::RetryPolicy,
cancels: &[crate::plugin::CancelDeadlineInfo],
) -> crate::error::Result<()> {
let schema = schema_ident(bounded_context);
let stale_cutoff = now - deadline_firing_claim_stale_after();
let mut cancel_backlogs: std::collections::HashMap<&str, Option<DateTime<Utc>>> =
std::collections::HashMap::new();
let rows: Vec<DueDeadlineRow> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT id, schedule_name, fire_at, correlation_id, target_bounded_context, \
target_command_type, payload, attempt_count, first_failed_at \
FROM {schema}.deadlines \
WHERE (status = 'pending' OR (status = 'firing' AND firing_at <= $1)) AND fire_at <= $2 \
AND (next_attempt_at IS NULL OR next_attempt_at <= $2) \
ORDER BY fire_at LIMIT {MAX_DUE_DEADLINES_PER_TICK}"
)))
.bind(stale_cutoff)
.bind(now)
.fetch_all(pool)
.await?;
'rows: for row in rows {
for cancel in cancels.iter().filter(|c| {
c.deadline_schedule_bounded_context == bounded_context
&& c.deadline_schedule_name == row.schedule_name
}) {
let backlog = match cancel_backlogs.get(cancel.name) {
Some(backlog) => *backlog,
None => {
let backlog = cancel_backlog_start(pool, cancel).await?;
cancel_backlogs.insert(cancel.name, backlog);
backlog
}
};
if backlog.is_some_and(|first_unprocessed| first_unprocessed <= row.fire_at) {
continue '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;
};
if target_command_type.bounded_context.status
!= crate::event_store::BoundedContextStatus::Active
{
tracing::warn!(
deadline_id = %row.id,
target_bounded_context = %row.target_bounded_context,
target_command_type = %row.target_command_type,
"deadline's own target bounded context is archived - 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)) \
AND (next_attempt_at IS NULL OR next_attempt_at <= $1) \
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_CLIENT_ID,
row.correlation_id.as_deref(),
None,
encryption_master_key,
now,
Some(&idempotency_key),
)
.await;
let outcome = match outcome {
Ok(outcome) => outcome,
Err(e) if e.another_instance_can_do_it() => {
defer_deadline_to_another_instance(pool, &schema, &row, &e, now).await?;
continue;
}
Err(e) => {
record_deadline_failure(pool, &schema, &row, &e, retry_policy, now).await?;
continue;
}
};
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(())
}
pub const ANOTHER_INSTANCE_RETRY_DELAY: std::time::Duration =
skilj_retry::ANOTHER_INSTANCE_RETRY_DELAY;
async fn defer_deadline_to_another_instance(
pool: &Pool,
schema: &str,
row: &DueDeadlineRow,
error: &crate::error::Error,
now: DateTime<Utc>,
) -> crate::error::Result<()> {
let next_attempt_at = now
+ chrono::Duration::from_std(ANOTHER_INSTANCE_RETRY_DELAY)
.expect("a ten-second delay fits a chrono::Duration");
tracing::warn!(
deadline_id = %row.id,
error = %error,
next_attempt_at = %next_attempt_at,
"deadline: this instance can't fire it, another can - retrying later"
);
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.deadlines SET status = 'pending', firing_at = NULL, \
next_attempt_at = $1 WHERE id = $2 AND status = 'firing'"
)))
.bind(next_attempt_at)
.bind(&row.id)
.execute(pool)
.await?;
Ok(())
}
async fn record_deadline_failure(
pool: &Pool,
schema: &str,
row: &DueDeadlineRow,
error: &crate::error::Error,
retry_policy: &skilj_retry::RetryPolicy,
now: DateTime<Utc>,
) -> crate::error::Result<()> {
let attempt_count = row.attempt_count + 1;
let first_failed_at = row.first_failed_at.unwrap_or(now);
let elapsed = (now - first_failed_at).to_std().unwrap_or_default();
let message = error.to_string();
if retry_policy.is_exhausted(attempt_count as u32, elapsed) {
tracing::error!(
deadline_id = %row.id,
error = %message,
attempt = attempt_count,
"deadline: target command submission failed repeatedly - parking it"
);
let request_json = serde_json::from_str(&row.payload)
.unwrap_or_else(|_| serde_json::Value::String(row.payload.clone()));
insert_parked_delivery(
pool,
&row.target_bounded_context,
&format!("{DEADLINE_PARKED_SOURCE_PREFIX}{}", row.schedule_name),
ParkedDeliveryKind::Deadline,
&row.id,
None,
Some(&row.target_bounded_context),
Some(&row.target_command_type),
&request_json,
&message,
attempt_count,
first_failed_at,
now,
)
.await?;
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.deadlines SET status = 'parked', resolved_at = $1, \
attempt_count = $2, last_error = $3, payload = '{{}}' \
WHERE id = $4 AND status = 'firing'"
)))
.bind(now)
.bind(attempt_count)
.bind(&message)
.bind(&row.id)
.execute(pool)
.await?;
return Ok(());
}
let next_attempt_at = retry_policy.next_attempt_at(now, attempt_count as u32);
tracing::warn!(
deadline_id = %row.id,
error = %message,
attempt = attempt_count,
next_attempt_at = %next_attempt_at,
"deadline: target command submission failed - will retry"
);
sqlx::query(sqlx::AssertSqlSafe(format!(
"UPDATE {schema}.deadlines SET status = 'pending', firing_at = NULL, \
attempt_count = $1, first_failed_at = $2, next_attempt_at = $3, last_error = $4 \
WHERE id = $5 AND status = 'firing'"
)))
.bind(attempt_count)
.bind(first_failed_at)
.bind(next_attempt_at)
.bind(&message)
.bind(&row.id)
.execute(pool)
.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, payload = '{{}}' \
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: Vec<Projection> =
list_projections_for_bounded_context(pool, bounded_context)
.await?
.into_iter()
.filter(|p| dispatcher.default_state(bounded_context, &p.name).is_some())
.collect();
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(());
}
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_limited(
pool,
bounded_context,
min_caught_up,
MAX_EVENTS_PER_CATCH_UP_TICK,
)
.await?
};
let mut previous = min_caught_up;
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 current: Option<(Option<i64>, bool)> =
sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT caught_up_to, sync FROM {schema}.projections \
WHERE name = $1 FOR UPDATE"
)))
.bind(&projection.name)
.fetch_optional(&mut *tx)
.await?;
match current {
Some((caught_up_to, false)) if 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 AND (caught_up_to IS NULL OR caught_up_to < $1)"
)))
.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 current: Option<(Option<i64>,)> = sqlx::query_as(sqlx::AssertSqlSafe(format!(
"SELECT caught_up_to FROM {schema}.projection_rebuilds \
WHERE projection_name = $1 AND status = 'building' FOR UPDATE"
)))
.bind(&rebuild.projection.name)
.fetch_optional(&mut *tx)
.await?;
let Some((current,)) = current else {
continue;
};
if current.unwrap_or(-1) != previous {
continue;
}
if current.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(&mut *tx)
.await?;
}
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' \
AND (caught_up_to IS NULL OR caught_up_to < $1)"
)))
.bind(event.sequence)
.bind(&rebuild.projection.name)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
previous = event.sequence;
}
for rebuild in &building_rebuilds {
let Some(current) = get_projection_rebuild(
pool,
bounded_context,
&rebuild.projection.name,
ProjectionRebuildStatus::Building,
)
.await?
else {
continue;
};
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())
}
#[allow(clippy::too_many_arguments)]
async fn fold_snapshot_tag_history(
conn: &mut sqlx::PgConnection,
bc: &BoundedContext,
dispatcher: &dyn crate::plugin::SnapshotDispatcher,
snapshot_name: &str,
tag_key: &str,
tag_value: &str,
default_state_json: &str,
up_to: i64,
) -> crate::error::Result<String> {
const CHUNK: i64 = 1000;
let tags = [Tag {
key: tag_key.to_string(),
value: Some(tag_value.to_string()),
}];
let mut state = default_state_json.to_string();
let mut after = -1;
loop {
let chunk = list_events_for_bounded_context_matching_tags_with_bc(
&mut *conn,
bc,
&tags,
Some(after),
None,
Some(CHUNK),
)
.await?;
for event in &chunk {
if event.sequence > up_to {
return Ok(state);
}
if let Some(result) = dispatcher.fold(&bc.name, snapshot_name, &state, event) {
state = result?;
}
}
match chunk.last() {
Some(last) if chunk.len() as i64 == CHUNK => after = last.sequence,
_ => return Ok(state),
}
}
}
#[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_limited(
pool,
bounded_context,
min_caught_up,
MAX_EVENTS_PER_CATCH_UP_TICK,
)
.await?
};
let bc = match events.first() {
Some(_) => Some(require_bounded_context(pool, bounded_context).await?),
None => None,
};
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 = if as_of_sequence == -1 {
fold_snapshot_tag_history(
&mut tx,
bc.as_ref().expect("fetched whenever there are events"),
dispatcher,
name,
tag_key,
tag_value,
&default_state_json,
event.sequence,
)
.await?
} else {
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();
let bc = require_bounded_context(pool, bounded_context).await?;
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 = if as_of_sequence == -1 {
fold_snapshot_tag_history(
&mut tx,
&bc,
dispatcher,
name,
tag_key,
tag_value,
&default_state_json,
event.sequence,
)
.await?
} else {
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 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;
let mut after = -1;
loop {
let events = list_events_for_bounded_context_from_limited(
pool,
bounded_context,
after,
MAX_EVENTS_PER_CATCH_UP_TICK,
)
.await?;
let exhausted = (events.len() as i64) < MAX_EVENTS_PER_CATCH_UP_TICK;
if let Some(last) = events.last() {
after = last.sequence;
}
for event in &events {
let mut tx = pool.begin().await?;
fold_event_into_new_projection(
&mut tx,
projection,
event,
dispatcher,
&default_state_json,
owner_tag_key,
)
.await?;
tx.commit().await?;
caught_up_to = Some(event.sequence);
}
if exhausted {
break;
}
}
Ok(Projection {
caught_up_to,
..projection.clone()
})
}
async fn fold_event_into_new_projection(
tx: &mut Transaction<'_, Postgres>,
projection: &Projection,
event: &Event,
dispatcher: &dyn crate::plugin::ProjectionDispatcher,
default_state_json: &str,
owner_tag_key: Option<&str>,
) -> crate::error::Result<()> {
let bounded_context = &projection.bounded_context.name;
let schema = schema_ident(bounded_context);
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 AND (caught_up_to IS NULL OR caught_up_to < $1)"
)))
.bind(event.sequence)
.bind(&projection.name)
.execute(&mut **tx)
.await?;
Ok(())
}
pub async fn create_projection(
pool: &Pool,
projection: &Projection,
needs_history_fold: bool,
dispatcher: &dyn crate::plugin::ProjectionDispatcher,
) -> crate::error::Result<Projection> {
if !projection.sync {
upsert_projection(pool, projection).await?;
return Ok(projection.clone());
}
register_new_sync_projection(pool, projection, needs_history_fold, dispatcher).await
}
pub async fn register_new_sync_projection(
pool: &Pool,
projection: &Projection,
needs_history_fold: bool,
dispatcher: &dyn crate::plugin::ProjectionDispatcher,
) -> crate::error::Result<Projection> {
let bounded_context = &projection.bounded_context.name;
let mut folded_through = None;
if needs_history_fold {
let hidden = Projection {
sync: false,
caught_up_to: None,
..projection.clone()
};
upsert_projection(pool, &hidden).await?;
folded_through = fold_history_into_new_sync_projection(pool, &hidden, dispatcher)
.await?
.caught_up_to;
}
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 tx = pool.begin().await?;
let locked_highest = lock_bounded_context_sequence(&mut tx, bounded_context).await?;
let consumed_types: std::collections::HashMap<String, EventType> = projection
.consumed_event_types
.iter()
.map(|et| (et.name.clone(), et.clone()))
.collect();
let consumed_names: Vec<&str> = consumed_types.keys().map(String::as_str).collect();
let schema = schema_ident(bounded_context);
let mut tail = Vec::new();
let mut after = folded_through.unwrap_or(-1);
loop {
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 event_type_name = ANY($1) AND sequence > $2 ORDER BY sequence LIMIT $3"
)))
.bind(&consumed_names)
.bind(after)
.bind(MAX_EVENTS_PER_CATCH_UP_TICK)
.fetch_all(&mut *tx)
.await?;
let exhausted = (rows.len() as i64) < MAX_EVENTS_PER_CATCH_UP_TICK;
let chunk = events_from_rows(
&mut tx,
&projection.bounded_context,
rows,
Some(&consumed_types),
)
.await?;
if let Some(last) = chunk.last() {
after = last.sequence;
}
tail.extend(chunk);
if exhausted {
break;
}
}
let stored = Projection {
sync: true,
caught_up_to: (folded_through.is_some() || !tail.is_empty() || needs_history_fold)
.then_some(locked_highest),
..projection.clone()
};
if !needs_history_fold {
upsert_projection_in_tx(&mut tx, &stored).await?;
}
for event in &tail {
fold_event_into_new_projection(
&mut tx,
&stored,
event,
dispatcher,
&default_state_json,
owner_tag_key,
)
.await?;
}
upsert_projection_in_tx(&mut tx, &stored).await?;
tx.commit().await?;
notify_registration_changed(pool).await;
Ok(stored)
}
#[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 access_token_kind_for_credential(
pool: &Pool,
id: &str,
hashed_secret: &str,
) -> crate::error::Result<Option<AccessTokenKind>> {
Ok(fetch_access_token_row(pool, id)
.await?
.filter(|r| crate::shared::secret_matches(hashed_secret, &r.columns.secret))
.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 =
require_event_type(pool, &row.bounded_context, &event_type_name).await?;
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 = require_event_type(pool, &row.bounded_context, &event_type_name).await?;
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 = require_command_type(pool, &row.bounded_context, &command_type_name).await?;
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(
executor: impl sqlx::PgExecutor<'_>,
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(executor)
.await?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::schema_ident;
#[test]
fn schema_ident_quotes_and_escapes() {
assert_eq!(schema_ident("banking"), "\"bc_banking\"");
assert_eq!(
schema_ident("x\"; DROP SCHEMA public; --"),
"\"bc_x\"\"; DROP SCHEMA public; --\""
);
}
}