use chrono::{DateTime, Utc};
use sqlx::SqlitePool;
use uuid::Uuid;
use crate::{
error::{HooksmithError, Result},
model::{DeliveryAttempt, Endpoint, EventStatus, NewEndpoint, QueueStats, WebhookEvent},
retry,
worker::MAX_RESPONSE_BODY_BYTES,
};
fn now_str() -> String {
Utc::now().format("%Y-%m-%dT%H:%M:%S%.6fZ").to_string()
}
fn uuid_str() -> String {
Uuid::new_v4().to_string()
}
fn bool_to_int(b: bool) -> i64 { if b { 1 } else { 0 } }
fn int_to_bool(i: i64) -> bool { i != 0 }
fn parse_status(s: &str) -> EventStatus {
match s {
"pending" => EventStatus::Pending,
"delivering" => EventStatus::Delivering,
"delivered" => EventStatus::Delivered,
"failed" => EventStatus::Failed,
"dead" => EventStatus::Dead,
_ => EventStatus::Pending,
}
}
#[allow(dead_code)]
fn status_str(s: &EventStatus) -> &'static str {
match s {
EventStatus::Pending => "pending",
EventStatus::Delivering => "delivering",
EventStatus::Delivered => "delivered",
EventStatus::Failed => "failed",
EventStatus::Dead => "dead",
}
}
fn row_to_endpoint(row: &sqlx::sqlite::SqliteRow) -> Endpoint {
use sqlx::Row;
let enabled_int: i64 = row.get("enabled");
Endpoint {
id: row.get::<String, _>("id").parse().unwrap_or_default(),
url: row.get("url"),
signing_secret: row.get("signing_secret"),
description: row.get("description"),
enabled: int_to_bool(enabled_int),
max_attempts: row.get("max_attempts"),
initial_delay_ms: row.get("initial_delay_ms"),
created_at: row.get::<String, _>("created_at")
.parse::<DateTime<Utc>>().unwrap_or_else(|_| Utc::now()),
updated_at: row.get::<String, _>("updated_at")
.parse::<DateTime<Utc>>().unwrap_or_else(|_| Utc::now()),
}
}
fn row_to_event(row: &sqlx::sqlite::SqliteRow) -> WebhookEvent {
use sqlx::Row;
let payload_str: String = row.get("payload");
let status_str_val: String = row.get("status");
let delivering_since: Option<String> = row.get("delivering_since");
let idempotency_key: Option<String> = row.get("idempotency_key");
WebhookEvent {
id: row.get::<String, _>("id").parse().unwrap_or_default(),
endpoint_id: row.get::<String, _>("endpoint_id").parse().unwrap_or_default(),
event_type: row.get("event_type"),
payload: serde_json::from_str(&payload_str).unwrap_or_default(),
status: parse_status(&status_str_val),
attempts: row.get("attempts"),
scheduled_at: row.get::<String, _>("scheduled_at")
.parse::<DateTime<Utc>>().unwrap_or_else(|_| Utc::now()),
delivering_since: delivering_since.and_then(|s| s.parse::<DateTime<Utc>>().ok()),
idempotency_key,
created_at: row.get::<String, _>("created_at")
.parse::<DateTime<Utc>>().unwrap_or_else(|_| Utc::now()),
}
}
fn row_to_attempt(row: &sqlx::sqlite::SqliteRow) -> DeliveryAttempt {
use sqlx::Row;
let success_int: i64 = row.get("success");
DeliveryAttempt {
id: row.get::<String, _>("id").parse().unwrap_or_default(),
event_id: row.get::<String, _>("event_id").parse().unwrap_or_default(),
attempted_at: row.get::<String, _>("attempted_at")
.parse::<DateTime<Utc>>().unwrap_or_else(|_| Utc::now()),
response_status: row.get("response_status"),
response_body: row.get("response_body"),
duration_ms: row.get("duration_ms"),
error: row.get("error"),
success: int_to_bool(success_int),
}
}
pub async fn create_endpoint(pool: &SqlitePool, new: NewEndpoint) -> Result<Endpoint> {
let id = uuid_str();
let now = now_str();
sqlx::query(
"INSERT INTO webhook_endpoints
(id, url, signing_secret, description, enabled, max_attempts, initial_delay_ms, created_at, updated_at)
VALUES (?, ?, ?, ?, 1, ?, ?, ?, ?)"
)
.bind(&id)
.bind(&new.url)
.bind(&new.signing_secret)
.bind(&new.description)
.bind(new.max_attempts.unwrap_or(10))
.bind(new.initial_delay_ms.unwrap_or(1000))
.bind(&now)
.bind(&now)
.execute(pool)
.await?;
get_endpoint_required(pool, &id).await
}
pub async fn get_endpoint_by_id(pool: &SqlitePool, id: Uuid) -> Result<Option<Endpoint>> {
let id_str = id.to_string();
let row = sqlx::query("SELECT * FROM webhook_endpoints WHERE id = ?")
.bind(&id_str)
.fetch_optional(pool)
.await?;
Ok(row.as_ref().map(row_to_endpoint))
}
async fn get_endpoint_required(pool: &SqlitePool, id: &str) -> Result<Endpoint> {
let row = sqlx::query("SELECT * FROM webhook_endpoints WHERE id = ?")
.bind(id)
.fetch_optional(pool)
.await?;
row.as_ref()
.map(row_to_endpoint)
.ok_or_else(|| HooksmithError::EndpointNotFound(id.parse().unwrap_or_default()))
}
pub async fn list_endpoints(pool: &SqlitePool) -> Result<Vec<Endpoint>> {
let rows = sqlx::query("SELECT * FROM webhook_endpoints ORDER BY created_at")
.fetch_all(pool)
.await?;
Ok(rows.iter().map(row_to_endpoint).collect())
}
pub async fn list_endpoints_paged(pool: &SqlitePool, limit: i64, offset: i64) -> Result<Vec<Endpoint>> {
let limit = limit.max(0);
let offset = offset.max(0);
let rows = sqlx::query("SELECT * FROM webhook_endpoints ORDER BY created_at LIMIT ? OFFSET ?")
.bind(limit)
.bind(offset)
.fetch_all(pool)
.await?;
Ok(rows.iter().map(row_to_endpoint).collect())
}
pub async fn update_endpoint_field(
pool: &SqlitePool,
id: Uuid,
url: Option<String>,
signing_secret: Option<String>,
description_set: bool,
description: Option<String>,
enabled: Option<bool>,
max_attempts: Option<i32>,
initial_delay_ms: Option<i32>,
) -> Result<Endpoint> {
let id_str = id.to_string();
let exists: bool = sqlx::query_scalar::<_, i64>(
"SELECT COUNT(*) FROM webhook_endpoints WHERE id = ?"
)
.bind(&id_str)
.fetch_one(pool)
.await? > 0;
if !exists {
return Err(HooksmithError::EndpointNotFound(id));
}
if let Some(u) = url {
sqlx::query("UPDATE webhook_endpoints SET url = ? WHERE id = ?")
.bind(u).bind(&id_str).execute(pool).await?;
}
if let Some(s) = signing_secret {
sqlx::query("UPDATE webhook_endpoints SET signing_secret = ? WHERE id = ?")
.bind(s).bind(&id_str).execute(pool).await?;
}
if description_set {
sqlx::query("UPDATE webhook_endpoints SET description = ? WHERE id = ?")
.bind(description).bind(&id_str).execute(pool).await?;
}
if let Some(e) = enabled {
sqlx::query("UPDATE webhook_endpoints SET enabled = ? WHERE id = ?")
.bind(bool_to_int(e)).bind(&id_str).execute(pool).await?;
}
if let Some(m) = max_attempts {
sqlx::query("UPDATE webhook_endpoints SET max_attempts = ? WHERE id = ?")
.bind(m).bind(&id_str).execute(pool).await?;
}
if let Some(d) = initial_delay_ms {
sqlx::query("UPDATE webhook_endpoints SET initial_delay_ms = ? WHERE id = ?")
.bind(d).bind(&id_str).execute(pool).await?;
}
sqlx::query("UPDATE webhook_endpoints SET updated_at = ? WHERE id = ?")
.bind(now_str()).bind(&id_str).execute(pool).await?;
get_endpoint_required(pool, &id_str).await
}
pub async fn delete_endpoint(pool: &SqlitePool, id: Uuid) -> Result<()> {
let id_str = id.to_string();
let rows = sqlx::query("DELETE FROM webhook_endpoints WHERE id = ?")
.bind(&id_str)
.execute(pool)
.await?
.rows_affected();
if rows == 0 {
return Err(HooksmithError::EndpointNotFound(id));
}
Ok(())
}
pub async fn enqueue(
pool: &SqlitePool,
endpoint_id: Uuid,
event_type: &str,
payload: serde_json::Value,
) -> Result<WebhookEvent> {
let id = uuid_str();
let now = now_str();
let endpoint_id_str = endpoint_id.to_string();
let payload_str = serde_json::to_string(&payload)
.map_err(|e| HooksmithError::Config(format!("payload serialization: {e}")))?;
sqlx::query(
"INSERT INTO webhook_events
(id, endpoint_id, event_type, payload, scheduled_at, created_at)
VALUES (?, ?, ?, ?, ?, ?)"
)
.bind(&id).bind(&endpoint_id_str).bind(event_type)
.bind(&payload_str).bind(&now).bind(&now)
.execute(pool)
.await?;
get_event_required(pool, &id).await
}
pub async fn enqueue_idempotent(
pool: &SqlitePool,
endpoint_id: Uuid,
event_type: &str,
payload: serde_json::Value,
idempotency_key: &str,
) -> Result<WebhookEvent> {
let endpoint_id_str = endpoint_id.to_string();
let payload_str = serde_json::to_string(&payload)
.map_err(|e| HooksmithError::Config(format!("payload serialization: {e}")))?;
let existing = sqlx::query(
"SELECT * FROM webhook_events WHERE endpoint_id = ? AND idempotency_key = ?"
)
.bind(&endpoint_id_str)
.bind(idempotency_key)
.fetch_optional(pool)
.await?;
if let Some(row) = existing {
return Ok(row_to_event(&row));
}
let id = uuid_str();
let now = now_str();
sqlx::query(
"INSERT OR IGNORE INTO webhook_events
(id, endpoint_id, event_type, payload, idempotency_key, scheduled_at, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)"
)
.bind(&id).bind(&endpoint_id_str).bind(event_type)
.bind(&payload_str).bind(idempotency_key).bind(&now).bind(&now)
.execute(pool)
.await?;
let row = sqlx::query(
"SELECT * FROM webhook_events WHERE endpoint_id = ? AND idempotency_key = ?"
)
.bind(&endpoint_id_str)
.bind(idempotency_key)
.fetch_one(pool)
.await?;
Ok(row_to_event(&row))
}
pub async fn broadcast(
pool: &SqlitePool,
event_type: &str,
payload: serde_json::Value,
) -> Result<Vec<WebhookEvent>> {
let endpoints = list_endpoints(pool).await?;
let mut events = Vec::new();
for ep in endpoints.iter().filter(|e| e.enabled) {
let ev = enqueue(pool, ep.id, event_type, payload.clone()).await?;
events.push(ev);
}
Ok(events)
}
pub async fn broadcast_idempotent(
pool: &SqlitePool,
event_type: &str,
payload: serde_json::Value,
idempotency_key: &str,
) -> Result<Vec<WebhookEvent>> {
let endpoints = list_endpoints(pool).await?;
let mut events = Vec::new();
for ep in endpoints.iter().filter(|e| e.enabled) {
let ev = enqueue_idempotent(pool, ep.id, event_type, payload.clone(), idempotency_key).await?;
events.push(ev);
}
Ok(events)
}
pub async fn claim_due_events(pool: &SqlitePool, limit: i64) -> Result<Vec<WebhookEvent>> {
let now = now_str();
let rows = sqlx::query(
"SELECT we.* FROM webhook_events we
JOIN webhook_endpoints ep ON ep.id = we.endpoint_id
WHERE we.status IN ('pending', 'failed')
AND we.scheduled_at <= ?
AND ep.enabled = 1
ORDER BY we.scheduled_at
LIMIT ?"
)
.bind(&now)
.bind(limit)
.fetch_all(pool)
.await?;
if rows.is_empty() {
return Ok(vec![]);
}
let ids: Vec<String> = rows.iter()
.map(|r| { use sqlx::Row; r.get::<String, _>("id") })
.collect();
for id in &ids {
sqlx::query(
"UPDATE webhook_events SET status = 'delivering', delivering_since = ? WHERE id = ?"
)
.bind(&now)
.bind(id)
.execute(pool)
.await?;
}
let mut events = Vec::new();
for id in &ids {
if let Some(row) = sqlx::query("SELECT * FROM webhook_events WHERE id = ?")
.bind(id)
.fetch_optional(pool)
.await?
{
events.push(row_to_event(&row));
}
}
Ok(events)
}
pub async fn recover_stuck_deliveries(pool: &SqlitePool, stuck_after_secs: i64) -> Result<u64> {
let rows = sqlx::query(
"UPDATE webhook_events
SET status = 'pending', delivering_since = NULL
WHERE status = 'delivering'
AND delivering_since < datetime('now', ? || ' seconds')"
)
.bind(format!("-{stuck_after_secs}"))
.execute(pool)
.await?
.rows_affected();
Ok(rows)
}
pub async fn record_success(
pool: &SqlitePool,
event_id: Uuid,
response_status: i32,
response_body: Option<String>,
duration_ms: i32,
) -> Result<()> {
let event_id_str = event_id.to_string();
let attempt_id = uuid_str();
let now = now_str();
let body = response_body.map(|b| {
if b.len() > MAX_RESPONSE_BODY_BYTES { b[..MAX_RESPONSE_BODY_BYTES].to_owned() } else { b }
});
sqlx::query(
"INSERT INTO webhook_delivery_attempts
(id, event_id, attempted_at, response_status, response_body, duration_ms, success)
VALUES (?, ?, ?, ?, ?, ?, 1)"
)
.bind(&attempt_id).bind(&event_id_str).bind(&now)
.bind(response_status).bind(body).bind(duration_ms)
.execute(pool)
.await?;
sqlx::query(
"UPDATE webhook_events
SET status = 'delivered', attempts = attempts + 1, delivering_since = NULL
WHERE id = ? AND status = 'delivering'"
)
.bind(&event_id_str)
.execute(pool)
.await?;
Ok(())
}
pub async fn record_failure(
pool: &SqlitePool,
event_id: Uuid,
endpoint_max_attempts: i32,
endpoint_initial_delay_ms: i32,
error: String,
response_status: Option<i32>,
duration_ms: Option<i32>,
) -> Result<()> {
let event_id_str = event_id.to_string();
let attempt_id = uuid_str();
let now = now_str();
sqlx::query(
"INSERT INTO webhook_delivery_attempts
(id, event_id, attempted_at, response_status, duration_ms, error, success)
VALUES (?, ?, ?, ?, ?, ?, 0)"
)
.bind(&attempt_id).bind(&event_id_str).bind(&now)
.bind(response_status).bind(duration_ms).bind(&error)
.execute(pool)
.await?;
let current: i32 = sqlx::query_scalar(
"SELECT attempts FROM webhook_events WHERE id = ?"
)
.bind(&event_id_str)
.fetch_one(pool)
.await?;
let next = current + 1;
if next >= endpoint_max_attempts {
sqlx::query(
"UPDATE webhook_events SET status = 'dead', attempts = attempts + 1, delivering_since = NULL WHERE id = ? AND status = 'delivering'"
)
.bind(&event_id_str).execute(pool).await?;
} else {
let delay = retry::next_delay(next as u32, endpoint_initial_delay_ms as u32, 3_600_000);
let secs = delay.as_secs() as i64;
sqlx::query(
"UPDATE webhook_events SET status = 'failed', attempts = attempts + 1,
scheduled_at = datetime('now', ? || ' seconds'), delivering_since = NULL
WHERE id = ? AND status = 'delivering'"
)
.bind(format!("+{secs}")).bind(&event_id_str)
.execute(pool).await?;
}
Ok(())
}
pub async fn get_event(pool: &SqlitePool, id: Uuid) -> Result<Option<WebhookEvent>> {
let row = sqlx::query("SELECT * FROM webhook_events WHERE id = ?")
.bind(id.to_string())
.fetch_optional(pool)
.await?;
Ok(row.as_ref().map(row_to_event))
}
async fn get_event_required(pool: &SqlitePool, id: &str) -> Result<WebhookEvent> {
let row = sqlx::query("SELECT * FROM webhook_events WHERE id = ?")
.bind(id)
.fetch_optional(pool)
.await?;
row.as_ref()
.map(row_to_event)
.ok_or_else(|| HooksmithError::EventNotFound(id.parse().unwrap_or_default()))
}
pub async fn delivery_log(pool: &SqlitePool, event_id: Uuid) -> Result<Vec<DeliveryAttempt>> {
let rows = sqlx::query(
"SELECT * FROM webhook_delivery_attempts WHERE event_id = ? ORDER BY attempted_at"
)
.bind(event_id.to_string())
.fetch_all(pool)
.await?;
Ok(rows.iter().map(row_to_attempt).collect())
}
pub async fn list_dead_events(pool: &SqlitePool, endpoint_id: Uuid) -> Result<Vec<WebhookEvent>> {
let rows = sqlx::query(
"SELECT * FROM webhook_events WHERE endpoint_id = ? AND status = 'dead' ORDER BY created_at DESC"
)
.bind(endpoint_id.to_string())
.fetch_all(pool)
.await?;
Ok(rows.iter().map(row_to_event).collect())
}
pub async fn dead_events_paged(pool: &SqlitePool, endpoint_id: Uuid, limit: i64, offset: i64) -> Result<Vec<WebhookEvent>> {
let rows = sqlx::query(
"SELECT * FROM webhook_events WHERE endpoint_id = ? AND status = 'dead' ORDER BY created_at DESC LIMIT ? OFFSET ?"
)
.bind(endpoint_id.to_string()).bind(limit.max(0)).bind(offset.max(0))
.fetch_all(pool)
.await?;
Ok(rows.iter().map(row_to_event).collect())
}
pub async fn retry_dead_event(pool: &SqlitePool, event_id: Uuid) -> Result<()> {
let id_str = event_id.to_string();
let status: Option<String> = sqlx::query_scalar(
"SELECT status FROM webhook_events WHERE id = ?"
)
.bind(&id_str)
.fetch_optional(pool)
.await?;
match status.as_deref() {
None => return Err(HooksmithError::EventNotFound(event_id)),
Some("dead") => {}
Some(_) => return Err(HooksmithError::InvalidState(event_id)),
}
sqlx::query(
"UPDATE webhook_events SET status = 'pending', attempts = 0, scheduled_at = datetime('now'), delivering_since = NULL WHERE id = ?"
)
.bind(&id_str).execute(pool).await?;
Ok(())
}
pub async fn retry_all_dead(pool: &SqlitePool, endpoint_id: Uuid) -> Result<u64> {
let rows = sqlx::query(
"UPDATE webhook_events SET status = 'pending', attempts = 0, scheduled_at = datetime('now'), delivering_since = NULL WHERE endpoint_id = ? AND status = 'dead'"
)
.bind(endpoint_id.to_string())
.execute(pool)
.await?
.rows_affected();
Ok(rows)
}
pub async fn events_by_status(
pool: &SqlitePool,
endpoint_id: Uuid,
status: &str,
limit: i64,
offset: i64,
) -> Result<Vec<WebhookEvent>> {
let rows = sqlx::query(
"SELECT * FROM webhook_events WHERE endpoint_id = ? AND status = ? ORDER BY created_at DESC LIMIT ? OFFSET ?"
)
.bind(endpoint_id.to_string()).bind(status).bind(limit.max(0)).bind(offset.max(0))
.fetch_all(pool)
.await?;
Ok(rows.iter().map(row_to_event).collect())
}
pub async fn events_global_by_status(
pool: &SqlitePool,
status: &str,
limit: i64,
offset: i64,
) -> Result<Vec<WebhookEvent>> {
let rows = sqlx::query(
"SELECT * FROM webhook_events WHERE status = ? ORDER BY created_at DESC LIMIT ? OFFSET ?"
)
.bind(status).bind(limit.max(0)).bind(offset.max(0))
.fetch_all(pool)
.await?;
Ok(rows.iter().map(row_to_event).collect())
}
pub async fn queue_stats(pool: &SqlitePool) -> Result<QueueStats> {
let pending: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM webhook_events WHERE status='pending'").fetch_one(pool).await?;
let delivering: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM webhook_events WHERE status='delivering'").fetch_one(pool).await?;
let failed: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM webhook_events WHERE status='failed'").fetch_one(pool).await?;
let dead: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM webhook_events WHERE status='dead'").fetch_one(pool).await?;
let delivered: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM webhook_events WHERE status='delivered'").fetch_one(pool).await?;
Ok(QueueStats { pending, delivering, failed, dead, delivered })
}
pub async fn cleanup_delivered(pool: &SqlitePool, older_than_secs: i64) -> Result<u64> {
let rows = sqlx::query(
"DELETE FROM webhook_events WHERE status = 'delivered' AND created_at < datetime('now', ? || ' seconds')"
)
.bind(format!("-{older_than_secs}"))
.execute(pool).await?.rows_affected();
Ok(rows)
}
pub async fn cleanup_dead(pool: &SqlitePool, older_than_secs: i64) -> Result<u64> {
let rows = sqlx::query(
"DELETE FROM webhook_events WHERE status = 'dead' AND created_at < datetime('now', ? || ' seconds')"
)
.bind(format!("-{older_than_secs}"))
.execute(pool).await?.rows_affected();
Ok(rows)
}
pub async fn reset_to_pending(pool: &SqlitePool, event_id: Uuid) -> Result<()> {
sqlx::query(
"UPDATE webhook_events SET status = 'pending', delivering_since = NULL, scheduled_at = datetime('now') WHERE id = ? AND status = 'delivering'"
)
.bind(event_id.to_string())
.execute(pool).await?;
Ok(())
}
pub async fn record_endpoint_deleted(pool: &SqlitePool, event_id: Uuid) {
let id_str = event_id.to_string();
let attempt_id = uuid_str();
let now = now_str();
let _ = sqlx::query(
"INSERT INTO webhook_delivery_attempts (id, event_id, attempted_at, error, success) VALUES (?, ?, ?, 'endpoint deleted after claim', 0)"
)
.bind(&attempt_id).bind(&id_str).bind(&now)
.execute(pool).await;
let _ = sqlx::query(
"UPDATE webhook_events SET status = 'dead', delivering_since = NULL WHERE id = ? AND status = 'delivering'"
)
.bind(&id_str).execute(pool).await;
}