75 lines
2.3 KiB
Rust
75 lines
2.3 KiB
Rust
use crate::helpers::{pg_repo, MapDomainError};
|
|
use async_trait::async_trait;
|
|
use domain::{
|
|
errors::DomainError,
|
|
events::DomainEvent,
|
|
ports::EventStore,
|
|
value_objects::SystemId,
|
|
};
|
|
use event_payload::EventPayload;
|
|
use uuid::Uuid;
|
|
|
|
pg_repo!(PostgresEventStore);
|
|
|
|
/// Extracts the primary aggregate ID from a domain event.
|
|
fn aggregate_id(event: &DomainEvent) -> Uuid {
|
|
match event {
|
|
DomainEvent::AssetIngested { asset_id, .. }
|
|
| DomainEvent::MetadataUpdated { asset_id, .. }
|
|
| DomainEvent::AssetDeleted { asset_id, .. }
|
|
| DomainEvent::SidecarSyncRequested { asset_id, .. } => *asset_id.as_uuid(),
|
|
|
|
DomainEvent::ShareCreated { scope_id, .. }
|
|
| DomainEvent::ShareRevoked { scope_id, .. } => *scope_id.as_uuid(),
|
|
|
|
DomainEvent::JobEnqueued { job_id, .. }
|
|
| DomainEvent::JobCompleted { job_id, .. }
|
|
| DomainEvent::JobFailed { job_id, .. } => *job_id.as_uuid(),
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl EventStore for PostgresEventStore {
|
|
async fn append(&self, event: &DomainEvent) -> Result<(), DomainError> {
|
|
let payload = EventPayload::from(event);
|
|
let event_type = payload.subject().to_string();
|
|
let json = serde_json::to_value(&payload)
|
|
.map_err(|e| DomainError::Internal(e.to_string()))?;
|
|
let agg_id = aggregate_id(event);
|
|
|
|
sqlx::query(
|
|
"INSERT INTO event_log (aggregate_id, event_type, payload, occurred_at)
|
|
VALUES ($1, $2, $3, now())",
|
|
)
|
|
.bind(agg_id)
|
|
.bind(event_type)
|
|
.bind(json)
|
|
.execute(&self.pool)
|
|
.await
|
|
.map_pg()?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn query_by_aggregate(
|
|
&self,
|
|
aggregate_id: &SystemId,
|
|
) -> Result<Vec<DomainEvent>, DomainError> {
|
|
let rows: Vec<(serde_json::Value,)> = sqlx::query_as(
|
|
"SELECT payload FROM event_log WHERE aggregate_id = $1 ORDER BY event_id ASC",
|
|
)
|
|
.bind(*aggregate_id.as_uuid())
|
|
.fetch_all(&self.pool)
|
|
.await
|
|
.map_pg()?;
|
|
|
|
rows.into_iter()
|
|
.map(|(json,)| {
|
|
let payload: EventPayload = serde_json::from_value(json)
|
|
.map_err(|e| DomainError::Internal(e.to_string()))?;
|
|
DomainEvent::try_from(payload)
|
|
})
|
|
.collect()
|
|
}
|
|
}
|