use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;
use serde::{Deserialize, Serialize};
use time::format_description::well_known::Rfc3339;
use time::OffsetDateTime;
use crate::event_log::{
sanitize_topic_component, AnyEventLog, EventLog, LogError, LogEvent, Topic,
};
pub const WAITPOINT_STATE_TOPIC_PREFIX: &str = "waitpoint.state.";
pub const WAITPOINT_WAITS_TOPIC: &str = "waitpoint.waits";
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WaitpointStatus {
#[default]
Open,
Completed,
Cancelled,
}
impl WaitpointStatus {
pub fn as_str(self) -> &'static str {
match self {
Self::Open => "open",
Self::Completed => "completed",
Self::Cancelled => "cancelled",
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WaitpointWaitStatus {
Completed,
Cancelled,
TimedOut,
Interrupted,
}
impl WaitpointWaitStatus {
pub fn as_str(self) -> &'static str {
match self {
Self::Completed => "completed",
Self::Cancelled => "cancelled",
Self::TimedOut => "timed_out",
Self::Interrupted => "interrupted",
}
}
fn event_kind(self) -> &'static str {
match self {
Self::Completed => "waitpoint_wait_completed",
Self::Cancelled => "waitpoint_wait_cancelled",
Self::TimedOut => "waitpoint_wait_timed_out",
Self::Interrupted => "waitpoint_wait_interrupted",
}
}
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct WaitpointRecord {
pub id: String,
pub status: WaitpointStatus,
pub created_at: String,
pub created_by: Option<String>,
pub completed_at: Option<String>,
pub completed_by: Option<String>,
pub cancelled_at: Option<String>,
pub cancelled_by: Option<String>,
pub reason: Option<String>,
#[serde(default)]
pub metadata: BTreeMap<String, serde_json::Value>,
}
impl WaitpointRecord {
pub fn open(
id: impl Into<String>,
created_by: Option<String>,
metadata: BTreeMap<String, serde_json::Value>,
) -> Self {
Self {
id: id.into(),
status: WaitpointStatus::Open,
created_at: now_rfc3339(),
created_by,
completed_at: None,
completed_by: None,
cancelled_at: None,
cancelled_by: None,
reason: None,
metadata,
}
}
pub fn is_terminal(&self) -> bool {
matches!(
self.status,
WaitpointStatus::Completed | WaitpointStatus::Cancelled
)
}
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct WaitpointWaitStartRecord {
pub wait_id: String,
pub waitpoint_ids: Vec<String>,
pub started_at: String,
pub trace_id: Option<String>,
pub replay_of_event_id: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct WaitpointWaitRecord {
pub wait_id: String,
pub waitpoint_ids: Vec<String>,
pub status: WaitpointWaitStatus,
pub started_at: String,
pub resolved_at: String,
pub waitpoints: Vec<WaitpointRecord>,
pub cancelled_waitpoint_id: Option<String>,
pub trace_id: Option<String>,
pub replay_of_event_id: Option<String>,
pub reason: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum WaitpointResolution {
Pending,
Completed,
Cancelled { waitpoint_id: String },
}
pub fn dedupe_waitpoint_ids(ids: &[String]) -> Vec<String> {
let mut seen = BTreeSet::new();
let mut out = Vec::new();
for id in ids {
let trimmed = id.trim();
if trimmed.is_empty() {
continue;
}
if seen.insert(trimmed.to_string()) {
out.push(trimmed.to_string());
}
}
out
}
pub fn waitpoint_topic(id: &str) -> Result<Topic, LogError> {
Topic::new(format!(
"{WAITPOINT_STATE_TOPIC_PREFIX}{}",
sanitize_topic_component(id)
))
}
pub fn waits_topic() -> Result<Topic, LogError> {
Topic::new(WAITPOINT_WAITS_TOPIC)
}
pub async fn load_waitpoint(
log: &Arc<AnyEventLog>,
id: &str,
) -> Result<Option<WaitpointRecord>, LogError> {
let events = log
.read_range(&waitpoint_topic(id)?, None, usize::MAX)
.await?;
let mut latest = None;
for (_, event) in events {
if !matches!(
event.kind.as_str(),
"waitpoint_created" | "waitpoint_completed" | "waitpoint_cancelled"
) {
continue;
}
let Ok(record) = serde_json::from_value::<WaitpointRecord>(event.payload) else {
continue;
};
latest = Some(record);
}
Ok(latest)
}
pub async fn load_waitpoints(
log: &Arc<AnyEventLog>,
ids: &[String],
) -> Result<Vec<WaitpointRecord>, LogError> {
let mut out = Vec::new();
for id in dedupe_waitpoint_ids(ids) {
if let Some(record) = load_waitpoint(log, &id).await? {
out.push(record);
}
}
Ok(out)
}
pub fn resolve_waitpoints(ids: &[String], waitpoints: &[WaitpointRecord]) -> WaitpointResolution {
let mut by_id = BTreeMap::new();
for waitpoint in waitpoints {
by_id.insert(waitpoint.id.as_str(), waitpoint);
}
let ids = dedupe_waitpoint_ids(ids);
if ids.is_empty() {
return WaitpointResolution::Pending;
}
let mut all_completed = true;
for id in ids {
let Some(waitpoint) = by_id.get(id.as_str()) else {
all_completed = false;
continue;
};
match waitpoint.status {
WaitpointStatus::Completed => {}
WaitpointStatus::Cancelled => {
return WaitpointResolution::Cancelled {
waitpoint_id: waitpoint.id.clone(),
};
}
WaitpointStatus::Open => {
all_completed = false;
}
}
}
if all_completed {
WaitpointResolution::Completed
} else {
WaitpointResolution::Pending
}
}
pub async fn create_waitpoint(
log: &Arc<AnyEventLog>,
id: &str,
created_by: Option<String>,
metadata: BTreeMap<String, serde_json::Value>,
) -> Result<WaitpointRecord, LogError> {
if let Some(existing) = load_waitpoint(log, id).await? {
return Ok(existing);
}
let record = WaitpointRecord::open(id, created_by, metadata);
append_waitpoint_state(log, "waitpoint_created", &record).await?;
Ok(record)
}
pub async fn complete_waitpoint(
log: &Arc<AnyEventLog>,
id: &str,
completed_by: Option<String>,
) -> Result<WaitpointRecord, LogError> {
let existing = load_waitpoint(log, id).await?;
if let Some(existing) = existing.as_ref() {
if existing.is_terminal() {
return Ok(existing.clone());
}
}
let now = now_rfc3339();
let mut record = existing.unwrap_or_else(|| WaitpointRecord {
id: id.to_string(),
status: WaitpointStatus::Open,
created_at: now.clone(),
created_by: completed_by.clone(),
completed_at: None,
completed_by: None,
cancelled_at: None,
cancelled_by: None,
reason: None,
metadata: BTreeMap::new(),
});
record.status = WaitpointStatus::Completed;
record.completed_at = Some(now);
record.completed_by = completed_by;
record.cancelled_at = None;
record.cancelled_by = None;
record.reason = None;
append_waitpoint_state(log, "waitpoint_completed", &record).await?;
Ok(record)
}
pub async fn cancel_waitpoint(
log: &Arc<AnyEventLog>,
id: &str,
cancelled_by: Option<String>,
reason: Option<String>,
) -> Result<WaitpointRecord, LogError> {
let existing = load_waitpoint(log, id).await?;
if let Some(existing) = existing.as_ref() {
if existing.is_terminal() {
return Ok(existing.clone());
}
}
let now = now_rfc3339();
let mut record = existing.unwrap_or_else(|| WaitpointRecord {
id: id.to_string(),
status: WaitpointStatus::Open,
created_at: now.clone(),
created_by: cancelled_by.clone(),
completed_at: None,
completed_by: None,
cancelled_at: None,
cancelled_by: None,
reason: None,
metadata: BTreeMap::new(),
});
record.status = WaitpointStatus::Cancelled;
record.completed_at = None;
record.completed_by = None;
record.cancelled_at = Some(now);
record.cancelled_by = cancelled_by;
record.reason = reason;
append_waitpoint_state(log, "waitpoint_cancelled", &record).await?;
Ok(record)
}
pub async fn append_wait_started(
log: &Arc<AnyEventLog>,
record: &WaitpointWaitStartRecord,
) -> Result<(), LogError> {
log.append(
&waits_topic()?,
LogEvent::new(
"waitpoint_wait_started",
serde_json::to_value(record).map_err(|error| {
LogError::Serde(format!("waitpoint wait encode error: {error}"))
})?,
)
.with_headers(wait_headers(&record.wait_id, &record.waitpoint_ids)),
)
.await
.map(|_| ())
}
pub async fn append_wait_terminal(
log: &Arc<AnyEventLog>,
record: &WaitpointWaitRecord,
) -> Result<(), LogError> {
log.append(
&waits_topic()?,
LogEvent::new(
record.status.event_kind(),
serde_json::to_value(record).map_err(|error| {
LogError::Serde(format!("waitpoint wait encode error: {error}"))
})?,
)
.with_headers(wait_headers(&record.wait_id, &record.waitpoint_ids)),
)
.await
.map(|_| ())
}
pub async fn find_wait_terminal(
log: &Arc<AnyEventLog>,
wait_id: &str,
) -> Result<Option<WaitpointWaitRecord>, LogError> {
let events = log.read_range(&waits_topic()?, None, usize::MAX).await?;
let mut latest = None;
for (_, event) in events {
if !matches!(
event.kind.as_str(),
"waitpoint_wait_completed"
| "waitpoint_wait_cancelled"
| "waitpoint_wait_timed_out"
| "waitpoint_wait_interrupted"
) {
continue;
}
if event.headers.get("wait_id").map(String::as_str) != Some(wait_id) {
continue;
}
let Ok(record) = serde_json::from_value::<WaitpointWaitRecord>(event.payload) else {
continue;
};
latest = Some(record);
}
Ok(latest)
}
async fn append_waitpoint_state(
log: &Arc<AnyEventLog>,
kind: &str,
record: &WaitpointRecord,
) -> Result<(), LogError> {
log.append(
&waitpoint_topic(&record.id)?,
LogEvent::new(
kind,
serde_json::to_value(record)
.map_err(|error| LogError::Serde(format!("waitpoint encode error: {error}")))?,
)
.with_headers(waitpoint_headers(record)),
)
.await
.map(|_| ())
}
fn wait_headers(wait_id: &str, waitpoint_ids: &[String]) -> BTreeMap<String, String> {
let mut headers = BTreeMap::new();
headers.insert("wait_id".to_string(), wait_id.to_string());
headers.insert("waitpoints".to_string(), waitpoint_ids.join(","));
headers
}
fn waitpoint_headers(record: &WaitpointRecord) -> BTreeMap<String, String> {
let mut headers = BTreeMap::new();
headers.insert("waitpoint_id".to_string(), record.id.clone());
headers.insert("status".to_string(), record.status.as_str().to_string());
if let Some(created_by) = record.created_by.as_ref() {
headers.insert("created_by".to_string(), created_by.clone());
}
if let Some(completed_by) = record.completed_by.as_ref() {
headers.insert("completed_by".to_string(), completed_by.clone());
}
if let Some(cancelled_by) = record.cancelled_by.as_ref() {
headers.insert("cancelled_by".to_string(), cancelled_by.clone());
}
headers
}
fn now_rfc3339() -> String {
OffsetDateTime::now_utc()
.format(&Rfc3339)
.unwrap_or_else(|_| OffsetDateTime::now_utc().to_string())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::event_log::{FileEventLog, MemoryEventLog};
#[tokio::test]
async fn waitpoint_state_persists_across_file_reopen() {
let dir = tempfile::tempdir().expect("tempdir");
let first = Arc::new(AnyEventLog::File(
FileEventLog::open(dir.path().to_path_buf(), 32).expect("open file log"),
));
create_waitpoint(&first, "demo", Some("creator".to_string()), BTreeMap::new())
.await
.expect("create waitpoint");
complete_waitpoint(&first, "demo", Some("completer".to_string()))
.await
.expect("complete waitpoint");
let reopened = Arc::new(AnyEventLog::File(
FileEventLog::open(dir.path().to_path_buf(), 32).expect("reopen file log"),
));
let state = load_waitpoint(&reopened, "demo")
.await
.expect("load state")
.expect("waitpoint exists");
assert_eq!(state.status, WaitpointStatus::Completed);
assert_eq!(state.completed_by.as_deref(), Some("completer"));
}
#[tokio::test]
async fn wait_terminal_lookup_returns_latest_terminal_record() {
let log = Arc::new(AnyEventLog::Memory(MemoryEventLog::new(32)));
append_wait_started(
&log,
&WaitpointWaitStartRecord {
wait_id: "wait-demo".to_string(),
waitpoint_ids: vec!["a".to_string(), "b".to_string()],
started_at: "2026-01-01T00:00:00Z".to_string(),
trace_id: Some("trace-demo".to_string()),
replay_of_event_id: None,
},
)
.await
.expect("append wait start");
append_wait_terminal(
&log,
&WaitpointWaitRecord {
wait_id: "wait-demo".to_string(),
waitpoint_ids: vec!["a".to_string(), "b".to_string()],
status: WaitpointWaitStatus::TimedOut,
started_at: "2026-01-01T00:00:00Z".to_string(),
resolved_at: "2026-01-01T00:01:00Z".to_string(),
waitpoints: Vec::new(),
cancelled_waitpoint_id: None,
trace_id: Some("trace-demo".to_string()),
replay_of_event_id: None,
reason: Some("deadline elapsed".to_string()),
},
)
.await
.expect("append wait result");
let record = find_wait_terminal(&log, "wait-demo")
.await
.expect("lookup wait result")
.expect("wait result exists");
assert_eq!(record.status, WaitpointWaitStatus::TimedOut);
assert_eq!(record.reason.as_deref(), Some("deadline elapsed"));
}
}