1use crate::db::{DbPool, DbTransaction};
2use crate::error::{AppError, AppResult, ErrorCode};
3use crate::events::EventEnvelope;
4use crate::execution_logs::{
5 ExecutionLogRecord, ExecutionLogSeverity, insert_execution_log_projection,
6};
7use crate::{RuntimeSpanAttributes, record_runtime_span_attributes, trace_context_from_headers};
8use async_trait::async_trait;
9use chrono::{DateTime, Utc};
10use serde::{Deserialize, Serialize};
11use serde_json::{Value, json};
12use std::collections::BTreeMap;
13use std::fmt::Debug;
14use std::sync::Arc;
15use tracing::Instrument;
16
17const OUTBOX_RETRY_DELAY_SECONDS: i64 = 5;
18const STALE_PROCESSING_LOCK_SECONDS: i64 = 300;
19
20#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
21#[serde(rename_all = "snake_case")]
22pub enum OutboxStatus {
23 Pending,
24 Processing,
25 Published,
26 Failed,
27 Dead,
28}
29
30impl OutboxStatus {
31 fn as_str(self) -> &'static str {
32 match self {
33 Self::Pending => "pending",
34 Self::Processing => "processing",
35 Self::Published => "published",
36 Self::Failed => "failed",
37 Self::Dead => "dead",
38 }
39 }
40}
41
42#[derive(Debug, Clone, Deserialize, Serialize)]
43pub struct OutboxEvent {
44 pub id: String,
45 pub event_name: String,
46 pub event_version: u16,
47 pub source_module: String,
48 pub aggregate_type: String,
49 pub aggregate_id: String,
50 pub correlation_id: String,
51 pub causation_id: Option<String>,
52 pub occurred_at: DateTime<Utc>,
53 pub payload: Value,
54 pub headers: Value,
55}
56
57impl OutboxEvent {
58 pub fn from_envelope(aggregate_type: impl Into<String>, event: &EventEnvelope) -> Self {
59 Self {
60 id: event.event_id.clone(),
61 event_name: event.event_name.clone(),
62 event_version: event.event_version,
63 source_module: event.source_module.clone(),
64 aggregate_type: aggregate_type.into(),
65 aggregate_id: event.subject.clone(),
66 correlation_id: event.correlation_id.0.clone(),
67 causation_id: event.causation_id.clone(),
68 occurred_at: event.occurred_at,
69 payload: event.payload.clone(),
70 headers: json!({
71 "actor": event.actor,
72 "tenant_id": event.tenant_id,
73 "schema_ref": event.schema_ref,
74 "trace": event.trace,
75 }),
76 }
77 }
78}
79
80#[derive(Debug, Clone, Deserialize, Serialize)]
81pub struct ClaimedOutboxEvent {
82 pub id: String,
83 pub event_name: String,
84 pub event_version: u16,
85 pub source_module: String,
86 pub aggregate_type: String,
87 pub aggregate_id: String,
88 pub correlation_id: String,
89 pub causation_id: Option<String>,
90 pub occurred_at: DateTime<Utc>,
91 pub payload: Value,
92 pub headers: Value,
93 pub attempts: i32,
94 pub max_attempts: i32,
95}
96
97#[derive(Debug, Clone, Default)]
98pub struct OutboxPublisher;
99
100impl OutboxPublisher {
101 pub async fn publish_in_tx(
102 &self,
103 tx: &mut DbTransaction<'_>,
104 event: &OutboxEvent,
105 ) -> AppResult<()> {
106 let span = tracing::info_span!(
107 "outbox_publish",
108 lenso.correlation_id = tracing::field::Empty,
109 lenso.story_id = tracing::field::Empty,
110 lenso.outbox_event_id = tracing::field::Empty,
111 lenso.execution.kind = tracing::field::Empty,
112 lenso.execution.name = tracing::field::Empty,
113 );
114 record_runtime_span_attributes(
115 &span,
116 &RuntimeSpanAttributes::outbox(
117 event.correlation_id.clone(),
118 event.id.clone(),
119 event.event_name.clone(),
120 ),
121 );
122
123 async {
124 let inserted = sqlx::query_scalar::<_, bool>(
125 r#"
126 insert into platform.outbox (
127 id,
128 event_name,
129 event_version,
130 source_module,
131 aggregate_type,
132 aggregate_id,
133 correlation_id,
134 causation_id,
135 occurred_at,
136 payload,
137 headers
138 )
139 values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
140 on conflict (id) do nothing
141 returning true
142 "#,
143 )
144 .bind(&event.id)
145 .bind(&event.event_name)
146 .bind(i32::from(event.event_version))
147 .bind(&event.source_module)
148 .bind(&event.aggregate_type)
149 .bind(&event.aggregate_id)
150 .bind(&event.correlation_id)
151 .bind(&event.causation_id)
152 .bind(event.occurred_at)
153 .bind(&event.payload)
154 .bind(&event.headers)
155 .fetch_optional(&mut **tx)
156 .await
157 .map_err(map_outbox_error)?
158 .unwrap_or(false);
159
160 if inserted {
161 return Ok(());
162 }
163
164 let existing = sqlx::query_as::<
165 _,
166 (
167 String,
168 i32,
169 String,
170 String,
171 String,
172 String,
173 Option<String>,
174 DateTime<Utc>,
175 Value,
176 Value,
177 ),
178 >(
179 r#"
180 select
181 event_name,
182 event_version,
183 source_module,
184 aggregate_type,
185 aggregate_id,
186 correlation_id,
187 causation_id,
188 occurred_at,
189 payload,
190 headers
191 from platform.outbox
192 where id = $1
193 "#,
194 )
195 .bind(&event.id)
196 .fetch_one(&mut **tx)
197 .await
198 .map_err(map_outbox_error)?;
199
200 let same_event = existing.0 == event.event_name
201 && existing.1 == i32::from(event.event_version)
202 && existing.2 == event.source_module
203 && existing.3 == event.aggregate_type
204 && existing.4 == event.aggregate_id
205 && existing.5 == event.correlation_id
206 && existing.6 == event.causation_id
207 && existing.7.timestamp_micros() == event.occurred_at.timestamp_micros()
208 && existing.8 == event.payload
209 && existing.9 == event.headers;
210 if !same_event {
211 return Err(AppError::new(
212 ErrorCode::Conflict,
213 "outbox event identity is already bound to different content",
214 ));
215 }
216
217 Ok(())
218 }
219 .instrument(span)
220 .await
221 }
222
223 pub async fn pending_count(&self, pool: &DbPool) -> AppResult<i64> {
224 sqlx::query_scalar(
225 r#"
226 select count(*)
227 from platform.outbox
228 where status = 'pending'
229 "#,
230 )
231 .fetch_one(pool)
232 .await
233 .map_err(map_outbox_error)
234 }
235}
236
237#[async_trait]
238pub trait EventDispatcher: Debug + Send + Sync {
239 async fn dispatch(&self, event: &ClaimedOutboxEvent) -> AppResult<()>;
240}
241
242#[async_trait]
243pub trait EventHandler: Debug + Send + Sync {
244 fn handler_name(&self) -> &str {
246 self.event_name()
247 }
248
249 fn event_name(&self) -> &str;
250 async fn handle(&self, event: &ClaimedOutboxEvent) -> AppResult<()>;
251}
252
253#[derive(Debug, Clone, Default)]
254pub struct EventHandlerRegistry {
255 handlers: BTreeMap<String, Vec<Arc<dyn EventHandler>>>,
256}
257
258impl EventHandlerRegistry {
259 pub fn new() -> Self {
260 Self::default()
261 }
262
263 pub fn register(&mut self, handler: Arc<dyn EventHandler>) {
264 self.handlers
265 .entry(handler.event_name().to_owned())
266 .or_default()
267 .push(handler);
268 }
269
270 pub fn register_all(&mut self, handlers: impl IntoIterator<Item = Arc<dyn EventHandler>>) {
271 for handler in handlers {
272 self.register(handler);
273 }
274 }
275
276 pub fn handler_count(&self, event_name: &str) -> usize {
277 self.handlers.get(event_name).map_or(0, Vec::len)
278 }
279
280 pub fn registrations(&self) -> impl Iterator<Item = &Arc<dyn EventHandler>> {
281 self.handlers.values().flatten()
282 }
283}
284
285#[async_trait]
286impl EventDispatcher for EventHandlerRegistry {
287 async fn dispatch(&self, event: &ClaimedOutboxEvent) -> AppResult<()> {
288 let Some(handlers) = self.handlers.get(event.event_name.as_str()) else {
289 tracing::debug!(
290 event_name = %event.event_name,
291 outbox_id = %event.id,
292 "no in-process event handlers registered"
293 );
294 return Ok(());
295 };
296
297 for handler in handlers {
298 handler.handle(event).await?;
299 }
300
301 Ok(())
302 }
303}
304
305#[derive(Debug, Default)]
306pub struct LoggingEventDispatcher;
307
308#[async_trait]
309impl EventDispatcher for LoggingEventDispatcher {
310 async fn dispatch(&self, event: &ClaimedOutboxEvent) -> AppResult<()> {
311 tracing::info!(
312 outbox_id = %event.id,
313 event_name = %event.event_name,
314 event_version = event.event_version,
315 aggregate_id = %event.aggregate_id,
316 correlation_id = %event.correlation_id,
317 "outbox event dispatched"
318 );
319 Ok(())
320 }
321}
322
323#[derive(Debug, Clone)]
324pub struct OutboxRelay {
325 pool: DbPool,
326 worker_id: String,
327}
328
329impl OutboxRelay {
330 pub fn new(pool: DbPool, worker_id: impl Into<String>) -> Self {
331 Self {
332 pool,
333 worker_id: worker_id.into(),
334 }
335 }
336
337 pub async fn claim_batch(&self, batch_size: i64) -> AppResult<Vec<ClaimedOutboxEvent>> {
338 let span = tracing::info_span!(
339 "outbox_claim_batch",
340 worker_id = %self.worker_id,
341 lenso.execution.kind = "outbox_claim",
342 lenso.execution.name = "outbox.claim_batch",
343 );
344
345 async {
346 let events = sqlx::query_as::<_, OutboxRow>(
347 r#"
348 with claimed as (
349 select id
350 from platform.outbox
351 where (
352 status in ('pending', 'failed')
353 and available_at <= now()
354 )
355 or (
356 status = 'processing'
357 and locked_at <= now() - ($1::double precision * interval '1 second')
358 )
359 order by available_at asc, created_at asc
360 limit $2
361 for update skip locked
362 )
363 update platform.outbox outbox
364 set status = 'processing',
365 locked_at = now(),
366 locked_by = $3,
367 last_error = null
368 from claimed
369 where outbox.id = claimed.id
370 returning
371 outbox.id,
372 outbox.event_name,
373 outbox.event_version,
374 outbox.source_module,
375 outbox.aggregate_type,
376 outbox.aggregate_id,
377 outbox.correlation_id,
378 outbox.causation_id,
379 outbox.occurred_at,
380 outbox.payload,
381 outbox.headers,
382 outbox.attempts,
383 outbox.max_attempts
384 "#,
385 )
386 .bind(stale_processing_lock_seconds())
387 .bind(batch_size)
388 .bind(&self.worker_id)
389 .fetch_all(&self.pool)
390 .await
391 .map(|rows| rows.into_iter().map(Into::into).collect())
392 .map_err(map_outbox_error)?;
393
394 for event in &events {
395 self.record_outbox_execution_log(
396 event,
397 ExecutionLogSeverity::Info,
398 "Outbox event claimed",
399 json!({
400 "attempt": event.attempts + 1,
401 "max_attempts": event.max_attempts,
402 "worker_id": self.worker_id,
403 }),
404 )
405 .await;
406 }
407
408 Ok(events)
409 }
410 .instrument(span)
411 .await
412 }
413
414 pub async fn relay_once(
415 &self,
416 dispatcher: &dyn EventDispatcher,
417 batch_size: i64,
418 ) -> AppResult<usize> {
419 let span = tracing::info_span!(
420 "outbox_relay_once",
421 worker_id = %self.worker_id,
422 lenso.execution.kind = "outbox_relay",
423 lenso.execution.name = "outbox.relay_once",
424 );
425
426 async {
427 let events = self.claim_batch(batch_size).await?;
428 let count = events.len();
429
430 for event in events {
431 let event_span = tracing::info_span!(
432 "outbox_dispatch",
433 lenso.correlation_id = tracing::field::Empty,
434 lenso.story_id = tracing::field::Empty,
435 lenso.outbox_event_id = tracing::field::Empty,
436 lenso.execution.kind = tracing::field::Empty,
437 lenso.execution.name = tracing::field::Empty,
438 );
439 record_runtime_span_attributes(
440 &event_span,
441 &RuntimeSpanAttributes::outbox(
442 event.correlation_id.clone(),
443 event.id.clone(),
444 event.event_name.clone(),
445 ),
446 );
447
448 async {
449 self.record_outbox_execution_log(
450 &event,
451 ExecutionLogSeverity::Info,
452 "Outbox event dispatch started",
453 json!({
454 "event_name": event.event_name,
455 "attempt": event.attempts + 1,
456 "worker_id": self.worker_id,
457 }),
458 )
459 .await;
460 match dispatcher.dispatch(&event).await {
461 Ok(()) => self.mark_published(&event).await?,
462 Err(error) => self.mark_dispatch_failed(&event, &error).await?,
463 }
464
465 Ok::<(), AppError>(())
466 }
467 .instrument(event_span)
468 .await?;
469 }
470
471 Ok(count)
472 }
473 .instrument(span)
474 .await
475 }
476
477 pub async fn mark_published(&self, event: &ClaimedOutboxEvent) -> AppResult<()> {
478 sqlx::query(
479 r#"
480 update platform.outbox
481 set status = 'published',
482 published_at = now(),
483 locked_at = null,
484 locked_by = null,
485 last_error = null
486 where id = $1
487 "#,
488 )
489 .bind(&event.id)
490 .execute(&self.pool)
491 .await
492 .map(|_| ())
493 .map_err(map_outbox_error)?;
494
495 self.record_outbox_execution_log(
496 event,
497 ExecutionLogSeverity::Info,
498 "Outbox event published",
499 json!({
500 "event_name": event.event_name,
501 "attempt": event.attempts + 1,
502 "worker_id": self.worker_id,
503 }),
504 )
505 .await;
506
507 Ok(())
508 }
509
510 pub async fn mark_dispatch_failed(
511 &self,
512 event: &ClaimedOutboxEvent,
513 error: &AppError,
514 ) -> AppResult<()> {
515 let next_attempt = event.attempts + 1;
516 let status = if next_attempt >= event.max_attempts {
517 OutboxStatus::Dead
518 } else if error.retryable {
519 OutboxStatus::Failed
520 } else {
521 OutboxStatus::Dead
522 };
523
524 let span = tracing::info_span!(
525 "outbox_retry",
526 lenso.correlation_id = tracing::field::Empty,
527 lenso.story_id = tracing::field::Empty,
528 lenso.outbox_event_id = tracing::field::Empty,
529 lenso.execution.kind = tracing::field::Empty,
530 lenso.execution.name = tracing::field::Empty,
531 );
532 record_runtime_span_attributes(
533 &span,
534 &RuntimeSpanAttributes::outbox(
535 event.correlation_id.clone(),
536 event.id.clone(),
537 event.event_name.clone(),
538 ),
539 );
540
541 async {
542 sqlx::query(
543 r#"
544 update platform.outbox
545 set status = $2,
546 attempts = attempts + 1,
547 available_at = case
548 when $2 = 'failed' then now() + ($4::double precision * interval '1 second')
549 else available_at
550 end,
551 locked_at = null,
552 locked_by = null,
553 last_error = $3
554 where id = $1
555 "#,
556 )
557 .bind(&event.id)
558 .bind(status.as_str())
559 .bind(error.public_message.as_str())
560 .bind(outbox_retry_delay_seconds(error))
561 .execute(&self.pool)
562 .await
563 .map(|_| ())
564 .map_err(map_outbox_error)?;
565
566 self.record_outbox_execution_log(
567 event,
568 ExecutionLogSeverity::Error,
569 if status == OutboxStatus::Dead {
570 "Outbox event marked dead"
571 } else {
572 "Outbox event failed"
573 },
574 json!({
575 "attempt": next_attempt,
576 "max_attempts": event.max_attempts,
577 "status": status.as_str(),
578 "retryable": error.retryable,
579 "retry_after_ms": error.retry_after_ms,
580 "provider_trace_reference": error.provider_trace_reference,
581 "error": error.public_message,
582 "worker_id": self.worker_id,
583 }),
584 )
585 .await;
586
587 Ok(())
588 }
589 .instrument(span)
590 .await
591 }
592
593 async fn record_outbox_execution_log(
594 &self,
595 event: &ClaimedOutboxEvent,
596 severity: ExecutionLogSeverity,
597 body: &'static str,
598 attributes: Value,
599 ) {
600 emit_outbox_lifecycle_event(event, severity, body, &attributes, Some(&self.worker_id));
601 if let Err(error) = insert_execution_log_projection(
602 &self.pool,
603 outbox_log_record(event, severity, body, attributes),
604 )
605 .await
606 {
607 tracing::warn!(
608 error = ?error,
609 outbox_id = %event.id,
610 "failed to write outbox execution log"
611 );
612 }
613 }
614}
615
616type OutboxRow = (
617 String,
618 String,
619 i32,
620 String,
621 String,
622 String,
623 String,
624 Option<String>,
625 DateTime<Utc>,
626 Value,
627 Value,
628 i32,
629 i32,
630);
631
632impl From<OutboxRow> for ClaimedOutboxEvent {
633 fn from(row: OutboxRow) -> Self {
634 let (
635 id,
636 event_name,
637 event_version,
638 source_module,
639 aggregate_type,
640 aggregate_id,
641 correlation_id,
642 causation_id,
643 occurred_at,
644 payload,
645 headers,
646 attempts,
647 max_attempts,
648 ) = row;
649
650 Self {
651 id,
652 event_name,
653 event_version: event_version
654 .try_into()
655 .expect("event_version should fit into u16"),
656 source_module,
657 aggregate_type,
658 aggregate_id,
659 correlation_id,
660 causation_id,
661 occurred_at,
662 payload,
663 headers,
664 attempts,
665 max_attempts,
666 }
667 }
668}
669
670fn map_outbox_error(source: sqlx::Error) -> AppError {
671 AppError::new(ErrorCode::Internal, "Outbox operation failed").with_source(source)
672}
673
674fn outbox_retry_delay_seconds(error: &AppError) -> f64 {
675 error.retry_after_ms.map_or_else(
676 || OUTBOX_RETRY_DELAY_SECONDS as f64,
677 |milliseconds| std::time::Duration::from_millis(milliseconds).as_secs_f64(),
678 )
679}
680
681fn stale_processing_lock_seconds() -> f64 {
682 STALE_PROCESSING_LOCK_SECONDS as f64
683}
684
685fn emit_outbox_lifecycle_event(
686 event: &ClaimedOutboxEvent,
687 severity: ExecutionLogSeverity,
688 body: &'static str,
689 attributes: &Value,
690 worker_id: Option<&str>,
691) {
692 match severity {
693 ExecutionLogSeverity::Error => {
694 tracing::error!(
695 outbox_id = %event.id,
696 event_name = %event.event_name,
697 correlation_id = %event.correlation_id,
698 worker_id = worker_id.unwrap_or(""),
699 attributes = %attributes,
700 "{body}"
701 );
702 }
703 ExecutionLogSeverity::Warn => {
704 tracing::warn!(
705 outbox_id = %event.id,
706 event_name = %event.event_name,
707 correlation_id = %event.correlation_id,
708 worker_id = worker_id.unwrap_or(""),
709 attributes = %attributes,
710 "{body}"
711 );
712 }
713 _ => {
714 tracing::info!(
715 outbox_id = %event.id,
716 event_name = %event.event_name,
717 correlation_id = %event.correlation_id,
718 worker_id = worker_id.unwrap_or(""),
719 attributes = %attributes,
720 "{body}"
721 );
722 }
723 }
724}
725
726fn outbox_log_record(
727 event: &impl OutboxLogSource,
728 severity: ExecutionLogSeverity,
729 body: impl Into<String>,
730 attributes: Value,
731) -> ExecutionLogRecord {
732 ExecutionLogRecord::from_runtime_attrs(
733 RuntimeSpanAttributes::outbox(event.correlation_id(), event.id(), event.execution_name()),
734 severity,
735 body,
736 )
737 .with_attributes(attributes)
738 .with_trace(trace_context_from_headers(event.headers()))
739}
740
741trait OutboxLogSource {
742 fn id(&self) -> String;
743 fn correlation_id(&self) -> String;
744 fn execution_name(&self) -> String;
745 fn headers(&self) -> &Value;
746}
747
748impl OutboxLogSource for OutboxEvent {
749 fn id(&self) -> String {
750 self.id.clone()
751 }
752
753 fn correlation_id(&self) -> String {
754 self.correlation_id.clone()
755 }
756
757 fn execution_name(&self) -> String {
758 self.event_name.clone()
759 }
760
761 fn headers(&self) -> &Value {
762 &self.headers
763 }
764}
765
766impl OutboxLogSource for ClaimedOutboxEvent {
767 fn id(&self) -> String {
768 self.id.clone()
769 }
770
771 fn correlation_id(&self) -> String {
772 self.correlation_id.clone()
773 }
774
775 fn execution_name(&self) -> String {
776 self.event_name.clone()
777 }
778
779 fn headers(&self) -> &Value {
780 &self.headers
781 }
782}