use std::collections::BTreeMap;
use std::fs::{File, Permissions};
use std::io::Write;
use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use serde::{Deserialize, Serialize};
const SPOOL_DIR_MODE: u32 = 0o700;
const SPOOL_FILE_MODE: u32 = 0o600;
pub const SPOOL_SCHEMA_VERSION: u32 = 1;
pub const SPOOL_DIR_NAME: &str = "webhook-spool";
pub const EXHAUSTED_DIR_NAME: &str = "exhausted";
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum SpoolError {
#[error("prepare spool directory {path}: {source}")]
PrepareDir {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("serialize spool entry {delivery_id}: {source}")]
Encode {
delivery_id: String,
#[source]
source: serde_json::Error,
},
#[error("write spool entry to {path}: {source}")]
Write {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("commit spool entry {from} -> {to}: {source}")]
Commit {
from: PathBuf,
to: PathBuf,
#[source]
source: std::io::Error,
},
#[error("fsync spool directory {path}: {source}")]
SyncDir {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("a spool entry already exists at {path}; refusing to clobber it")]
AlreadyExists {
path: PathBuf,
},
#[error("read spool directory {path}: {source}")]
ReadDir {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("remove acknowledged spool entry {path}: {source}")]
Remove {
path: PathBuf,
#[source]
source: std::io::Error,
},
}
pub use trusty_common::webhook_relay::Provenance;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SpoolEntry {
pub schema_version: u32,
pub delivery_id: String,
pub source: String,
pub event: String,
pub headers: BTreeMap<String, String>,
pub body_b64: String,
pub provenance: Provenance,
pub received_at_unix_ms: u64,
pub attempts: u32,
pub last_error: Option<String>,
pub last_attempt_at_unix_ms: Option<u64>,
}
#[derive(Debug, Clone)]
pub struct PendingEntry {
pub path: PathBuf,
pub entry: SpoolEntry,
}
#[derive(Debug, Clone)]
pub struct Spool {
root: PathBuf,
opened: bool,
}
impl Spool {
pub fn at(root: impl Into<PathBuf>) -> Self {
Self {
root: root.into(),
opened: false,
}
}
pub fn open(root: impl Into<PathBuf>) -> Result<Self, SpoolError> {
let mut spool = Self::at(root);
spool.prepare_dir()?;
spool.opened = true;
Ok(spool)
}
pub fn default_root() -> anyhow::Result<PathBuf> {
Ok(trusty_common::resolve_data_dir("trusty-console")?.join(SPOOL_DIR_NAME))
}
pub fn root(&self) -> &Path {
&self.root
}
fn prepare_dir(&self) -> Result<(), SpoolError> {
std::fs::create_dir_all(&self.root).map_err(|source| SpoolError::PrepareDir {
path: self.root.clone(),
source,
})?;
std::fs::set_permissions(&self.root, Permissions::from_mode(SPOOL_DIR_MODE)).map_err(
|source| SpoolError::PrepareDir {
path: self.root.clone(),
source,
},
)
}
pub fn entry_path(&self, entry: &SpoolEntry) -> PathBuf {
self.root.join(format!(
"{:013}-{}.json",
entry.received_at_unix_ms,
sanitise_delivery_id(&entry.delivery_id)
))
}
pub fn persist_new(&self, entry: &SpoolEntry) -> Result<PathBuf, SpoolError> {
let final_path = self.entry_path(entry);
let tmp_path = self.write_temp(entry, &final_path)?;
std::fs::hard_link(&tmp_path, &final_path).map_err(|source| {
let _ = std::fs::remove_file(&tmp_path);
if source.kind() == std::io::ErrorKind::AlreadyExists {
SpoolError::AlreadyExists {
path: final_path.clone(),
}
} else {
SpoolError::Commit {
from: tmp_path.clone(),
to: final_path.clone(),
source,
}
}
})?;
let _ = std::fs::remove_file(&tmp_path);
sync_dir(&self.root)?;
Ok(final_path)
}
pub fn persist_update(&self, entry: &SpoolEntry) -> Result<PathBuf, SpoolError> {
let final_path = self.entry_path(entry);
let tmp_path = self.write_temp(entry, &final_path)?;
std::fs::rename(&tmp_path, &final_path).map_err(|source| {
let _ = std::fs::remove_file(&tmp_path);
SpoolError::Commit {
from: tmp_path.clone(),
to: final_path.clone(),
source,
}
})?;
sync_dir(&self.root)?;
Ok(final_path)
}
fn write_temp(&self, entry: &SpoolEntry, final_path: &Path) -> Result<PathBuf, SpoolError> {
let bytes = serde_json::to_vec_pretty(entry).map_err(|source| SpoolError::Encode {
delivery_id: entry.delivery_id.clone(),
source,
})?;
let stamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0);
let tmp_path =
final_path.with_extension(format!("json.{}.{stamp}.tmp", std::process::id()));
let write = || -> std::io::Result<()> {
let mut file = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(&tmp_path)?;
file.set_permissions(Permissions::from_mode(SPOOL_FILE_MODE))?;
file.write_all(&bytes)?;
file.sync_all()
};
write().map_err(|source| SpoolError::Write {
path: tmp_path.clone(),
source,
})?;
Ok(tmp_path)
}
pub fn record_attempt(
&self,
entry: &mut SpoolEntry,
reason: String,
now_unix_ms: u64,
) -> Result<PathBuf, SpoolError> {
entry.attempts = entry.attempts.saturating_add(1);
entry.last_error = Some(reason);
entry.last_attempt_at_unix_ms = Some(now_unix_ms);
self.persist_update(entry)
}
pub fn remove_acked(&self, path: &Path) -> Result<(), SpoolError> {
match std::fs::remove_file(path) {
Ok(()) => {}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()),
Err(source) => {
return Err(SpoolError::Remove {
path: path.to_path_buf(),
source,
});
}
}
sync_dir(&self.root)
}
pub fn exhausted_root(&self) -> PathBuf {
self.root.join(EXHAUSTED_DIR_NAME)
}
pub fn quarantine(&self, path: &Path) -> Result<PathBuf, SpoolError> {
let dest_dir = self.exhausted_root();
std::fs::create_dir_all(&dest_dir).map_err(|source| SpoolError::PrepareDir {
path: dest_dir.clone(),
source,
})?;
std::fs::set_permissions(&dest_dir, Permissions::from_mode(SPOOL_DIR_MODE)).map_err(
|source| SpoolError::PrepareDir {
path: dest_dir.clone(),
source,
},
)?;
let name = path.file_name().ok_or_else(|| SpoolError::Remove {
path: path.to_path_buf(),
source: std::io::Error::other("spool entry path has no file name"),
})?;
let dest = dest_dir.join(name);
std::fs::rename(path, &dest).map_err(|source| SpoolError::Commit {
from: path.to_path_buf(),
to: dest.clone(),
source,
})?;
sync_dir(&dest_dir)?;
sync_dir(&self.root)?;
Ok(dest)
}
pub fn load(&self, path: &Path) -> Result<SpoolEntry, SpoolError> {
let bytes = std::fs::read(path).map_err(|source| SpoolError::ReadDir {
path: path.to_path_buf(),
source,
})?;
serde_json::from_slice(&bytes).map_err(|source| SpoolError::Encode {
delivery_id: path.display().to_string(),
source,
})
}
pub fn scan_metadata(&self) -> Result<SpoolMetadata, SpoolError> {
let mut meta = SpoolMetadata::default();
collect_metadata(
&self.root,
self.opened,
&mut meta.live,
&mut meta.unparsable,
)?;
collect_metadata(
&self.exhausted_root(),
false,
&mut meta.exhausted,
&mut meta.unparsable,
)?;
meta.live.sort();
meta.exhausted.sort();
meta.unparsable.sort();
Ok(meta)
}
pub fn list_pending(&self) -> Result<PendingListing, SpoolError> {
let read = match std::fs::read_dir(&self.root) {
Ok(read) => read,
Err(e) if e.kind() == std::io::ErrorKind::NotFound && !self.opened => {
return Ok(PendingListing::default());
}
Err(source) => {
return Err(SpoolError::ReadDir {
path: self.root.clone(),
source,
});
}
};
let mut listing = PendingListing::default();
for dirent in read {
let dirent = dirent.map_err(|source| SpoolError::ReadDir {
path: self.root.clone(),
source,
})?;
let path = dirent.path();
if path.extension().and_then(|e| e.to_str()) != Some("json") {
continue;
}
match std::fs::read(&path)
.map_err(SpoolReadFailure::Io)
.and_then(|b| {
serde_json::from_slice::<SpoolEntry>(&b).map_err(SpoolReadFailure::Decode)
}) {
Ok(entry) => listing.pending.push(PendingEntry { path, entry }),
Err(failure) => listing.undecodable.push((path, failure.to_string())),
}
}
listing
.pending
.sort_by_key(|p| (p.entry.received_at_unix_ms, p.path.clone()));
listing.undecodable.sort();
Ok(listing)
}
}
#[derive(Debug, Default, Clone)]
pub struct PendingListing {
pub pending: Vec<PendingEntry>,
pub undecodable: Vec<(PathBuf, String)>,
}
#[derive(Debug, thiserror::Error)]
enum SpoolReadFailure {
#[error("read: {0}")]
Io(#[from] std::io::Error),
#[error("decode: {0}")]
Decode(#[from] serde_json::Error),
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub struct EntryMeta {
pub received_at_unix_ms: u64,
pub delivery_id: String,
pub path: PathBuf,
}
#[derive(Debug, Default, Clone)]
pub struct SpoolMetadata {
pub live: Vec<EntryMeta>,
pub exhausted: Vec<EntryMeta>,
pub unparsable: Vec<(PathBuf, String)>,
}
fn collect_metadata(
dir: &Path,
required: bool,
out: &mut Vec<EntryMeta>,
unparsable: &mut Vec<(PathBuf, String)>,
) -> Result<(), SpoolError> {
let read = match std::fs::read_dir(dir) {
Ok(read) => read,
Err(e) if e.kind() == std::io::ErrorKind::NotFound && !required => return Ok(()),
Err(source) => {
return Err(SpoolError::ReadDir {
path: dir.to_path_buf(),
source,
});
}
};
for dirent in read {
let dirent = dirent.map_err(|source| SpoolError::ReadDir {
path: dir.to_path_buf(),
source,
})?;
let path = dirent.path();
if path.extension().and_then(|e| e.to_str()) != Some("json") {
continue;
}
match path
.file_name()
.and_then(|n| n.to_str())
.and_then(parse_entry_filename)
{
Some((received_at_unix_ms, delivery_id)) => out.push(EntryMeta {
received_at_unix_ms,
delivery_id,
path,
}),
None => unparsable.push((path, "filename does not carry a receipt timestamp".into())),
}
}
Ok(())
}
fn parse_entry_filename(name: &str) -> Option<(u64, String)> {
let stem = name.strip_suffix(".json")?;
let (ts, id) = stem.split_once('-')?;
let received = ts.parse::<u64>().ok()?;
Some((received, id.to_string()))
}
fn sync_dir(dir: &Path) -> Result<(), SpoolError> {
File::open(dir)
.and_then(|d| d.sync_all())
.map_err(|source| SpoolError::SyncDir {
path: dir.to_path_buf(),
source,
})
}
fn sanitise_delivery_id(raw: &str) -> String {
let cleaned: String = raw
.chars()
.map(|c| {
if c.is_ascii_alphanumeric() || c == '-' || c == '_' {
c
} else {
'_'
}
})
.take(64)
.collect();
if cleaned.is_empty() {
"unknown".to_string()
} else {
cleaned
}
}