use crate::repository::RepositoryError;
use crate::repository::error::Result;
use serde::{Deserialize, Serialize};
use std::cmp::Ordering;
use std::path::Path;
use std::sync::Arc;
use uuid::Uuid;
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[cfg_attr(feature = "allocative", derive(allocative::Allocative))]
pub enum Status {
Active,
Archived {
#[cfg_attr(feature = "allocative", allocative(skip))]
seqnum_id: Uuid,
head_seqnum: u64,
head_realtime: u64,
},
Disposed {
timestamp: u64,
number: u64,
},
}
impl Ord for Status {
fn cmp(&self, other: &Self) -> Ordering {
match (self, other) {
(
Status::Disposed {
timestamp: t1,
number: n1,
},
Status::Disposed {
timestamp: t2,
number: n2,
},
) => t1.cmp(t2).then_with(|| n1.cmp(n2)),
(Status::Disposed { .. }, _) => Ordering::Less,
(_, Status::Disposed { .. }) => Ordering::Greater,
(
Status::Archived {
seqnum_id: lhs_seqnum_id,
head_seqnum: lhs_head_seqnum,
head_realtime: lhs_head_realtime,
},
Status::Archived {
seqnum_id: rhs_seqnum_id,
head_seqnum: rhs_head_seqnum,
head_realtime: rhs_head_realtime,
},
) => lhs_head_realtime
.cmp(rhs_head_realtime)
.then_with(|| lhs_seqnum_id.cmp(rhs_seqnum_id))
.then_with(|| lhs_head_seqnum.cmp(rhs_head_seqnum)),
(Status::Archived { .. }, Status::Active) => Ordering::Less,
(Status::Active, Status::Archived { .. }) => Ordering::Greater,
(Status::Active, Status::Active) => Ordering::Equal,
}
}
}
impl PartialOrd for Status {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
Some(self.cmp(other))
}
}
impl Status {
pub(super) fn parse(path: &str) -> Option<(Self, &str)> {
if let Some(stem) = path.strip_suffix(".journal") {
return Self::parse_journal_stem(stem);
}
let stem = path.strip_suffix(".journal~")?;
Self::parse_disposed_stem(stem)
}
fn parse_journal_stem(stem: &str) -> Option<(Self, &str)> {
if let Some((prefix, suffix)) = stem.rsplit_once('@') {
return Self::parse_archived_suffix(prefix, suffix);
}
Some((Status::Active, stem))
}
fn parse_archived_suffix<'a>(prefix: &'a str, suffix: &str) -> Option<(Self, &'a str)> {
let mut parts = suffix.split('-');
let seqnum_id = Uuid::try_parse(parts.next()?).ok()?;
let head_seqnum = u64::from_str_radix(parts.next()?, 16).ok()?;
let head_realtime = u64::from_str_radix(parts.next()?, 16).ok()?;
if parts.next().is_some() {
return None;
}
Some((
Status::Archived {
seqnum_id,
head_seqnum,
head_realtime,
},
prefix,
))
}
fn parse_disposed_stem(stem: &str) -> Option<(Self, &str)> {
let (prefix, suffix) = stem.rsplit_once('@')?;
let (timestamp, number) = suffix.rsplit_once('-')?;
let timestamp = u64::from_str_radix(timestamp, 16).ok()?;
let number = u64::from_str_radix(number, 16).ok()?;
Some((Status::Disposed { timestamp, number }, prefix))
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[cfg_attr(feature = "allocative", derive(allocative::Allocative))]
pub enum Source {
System,
User(u32),
Remote(String),
Unknown(String),
}
impl Source {
pub(super) fn parse(path: &str) -> Option<(Self, &str)> {
let (dir_path, basename) = path.rsplit_once('/')?;
let journal_type = if basename == "system" {
Source::System
} else if let Some(uid_str) = basename.strip_prefix("user-") {
if let Ok(uid) = uid_str.parse::<u32>() {
Source::User(uid)
} else {
Source::Unknown(basename.to_string())
}
} else if let Some(remote_host) = basename.strip_prefix("remote-") {
Source::Remote(remote_host.to_string())
} else {
Source::Unknown(basename.to_string())
};
Some((journal_type, dir_path))
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[cfg_attr(feature = "allocative", derive(allocative::Allocative))]
pub struct Origin {
#[cfg_attr(feature = "allocative", allocative(skip))]
pub machine_id: Option<Uuid>,
pub namespace: Option<String>,
pub source: Source,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[cfg_attr(feature = "allocative", derive(allocative::Allocative))]
pub(crate) struct FileInner {
pub(crate) path: String,
pub(crate) origin: Origin,
pub(crate) status: Status,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
#[cfg_attr(feature = "allocative", derive(allocative::Allocative))]
pub struct File {
pub(super) inner: Arc<FileInner>,
}
impl serde::Serialize for File {
fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
self.inner.as_ref().serialize(serializer)
}
}
impl<'de> serde::Deserialize<'de> for File {
fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let inner = FileInner::deserialize(deserializer)?;
Ok(File {
inner: Arc::new(inner),
})
}
}
impl File {
pub fn path(&self) -> &str {
&self.inner.path
}
pub fn origin(&self) -> &Origin {
&self.inner.origin
}
pub fn status(&self) -> &Status {
&self.inner.status
}
pub fn from_path(path: &Path) -> Option<Self> {
if !path.is_absolute() {
return None;
}
let path_str = path.to_str()?;
let filename = path.file_name()?.to_str()?;
let filename_path = format!("/{filename}");
let (status, path_after_status) = Status::parse(&filename_path)?;
let (source, _) = Source::parse(path_after_status)?;
let (machine_id, namespace) = path
.parent()
.and_then(|parent| parent.file_name())
.and_then(|dirname| dirname.to_str())
.map(parse_machine_id_namespace)
.unwrap_or((None, None));
let origin = Origin {
machine_id,
namespace,
source,
};
let inner = Arc::new(FileInner {
path: path_str.to_string(),
origin,
status,
});
Some(File { inner })
}
pub fn from_raw_path(path: &Path) -> Option<Self> {
let path = path.to_str()?;
let raw_path = Path::new(path);
if !raw_path.is_absolute() {
return None;
}
let inner = Arc::new(FileInner {
path: path.to_string(),
origin: Origin {
machine_id: None,
namespace: None,
source: Source::Unknown(
raw_path
.file_stem()
.and_then(|stem| stem.to_str())
.unwrap_or("journal")
.to_string(),
),
},
status: Status::Active,
});
Some(File { inner })
}
#[allow(clippy::should_implement_trait)]
pub fn from_str(path: &str) -> Option<Self> {
if !path.starts_with("/") {
return None;
}
let (status, path_after_status) = Status::parse(path)?;
let (source, path_after_source) = Source::parse(path_after_status)?;
let (machine_id, namespace) = if !path_after_source.is_empty() {
let dirname = if let Some((_parent, dir)) = path_after_source.rsplit_once('/') {
dir
} else {
path_after_source
};
if let Some((id_str, ns)) = dirname.split_once('.') {
let machine_id = Uuid::try_parse(id_str).ok()?;
(Some(machine_id), Some(ns.to_string()))
} else {
let machine_id = Uuid::try_parse(dirname).ok();
(machine_id, None)
}
} else {
(None, None)
};
let origin = Origin {
machine_id,
namespace,
source,
};
let inner = Arc::new(FileInner {
path: String::from(path),
origin,
status,
});
Some(File { inner })
}
pub fn dir(&self) -> Result<&str> {
Path::new(&self.inner.path)
.parent()
.and_then(|p| {
if self.inner.origin.machine_id.is_some() {
p.parent()
} else {
Some(p)
}
})
.and_then(|p| p.to_str())
.ok_or_else(|| RepositoryError::InvalidUtf8 {
path: Path::new(&self.inner.path).to_path_buf(),
})
}
pub fn is_journal_file(path: &str) -> bool {
path.ends_with(".journal") || path.ends_with(".journal~")
}
pub fn is_active(&self) -> bool {
matches!(self.inner.status, Status::Active)
}
pub fn is_archived(&self) -> bool {
matches!(self.inner.status, Status::Archived { .. })
}
pub fn is_disposed(&self) -> bool {
matches!(self.inner.status, Status::Disposed { .. })
}
pub fn is_user(&self) -> bool {
matches!(self.inner.origin.source, Source::User(_))
}
pub fn is_system(&self) -> bool {
matches!(self.inner.origin.source, Source::System)
}
pub fn is_remote(&self) -> bool {
matches!(self.inner.origin.source, Source::Remote(_))
}
pub fn user_id(&self) -> Option<u32> {
match &self.inner.origin.source {
Source::User(uid) => Some(*uid),
_ => None,
}
}
pub fn remote_host(&self) -> Option<&str> {
match &self.inner.origin.source {
Source::Remote(host) => Some(host.as_str()),
_ => None,
}
}
pub fn namespace(&self) -> Option<&str> {
self.inner.origin.namespace.as_deref()
}
}
fn parse_machine_id_namespace(dirname: &str) -> (Option<Uuid>, Option<String>) {
if let Some((id_str, ns)) = dirname.split_once('.') {
let Some(machine_id) = Uuid::try_parse(id_str).ok() else {
return (None, None);
};
(Some(machine_id), Some(ns.to_string()))
} else {
(Uuid::try_parse(dirname).ok(), None)
}
}
#[cfg(test)]
mod tests {
use super::{File, Source, Status};
use std::path::PathBuf;
#[test]
fn from_path_parses_native_absolute_paths() {
let dir = tempfile::tempdir().expect("temp dir");
let path = dir
.path()
.join("00112233445566778899aabbccddeeff")
.join("system.journal");
let file = File::from_path(&path).expect("native absolute path parses");
assert_eq!(file.path(), path.to_str().expect("utf8 path"));
assert_eq!(
file.origin()
.machine_id
.expect("machine id")
.simple()
.to_string(),
"00112233445566778899aabbccddeeff"
);
assert_eq!(file.origin().source, Source::System);
assert_eq!(file.status(), &Status::Active);
}
#[test]
fn from_path_rejects_relative_paths() {
assert!(File::from_path(&PathBuf::from("system.journal")).is_none());
}
#[test]
fn from_raw_path_accepts_native_absolute_paths() {
let dir = tempfile::tempdir().expect("temp dir");
let path = dir.path().join("raw-byte-names.journal");
let file = File::from_raw_path(&path).expect("raw native absolute path parses");
assert_eq!(file.path(), path.to_str().expect("utf8 path"));
assert_eq!(
file.origin().source,
Source::Unknown("raw-byte-names".to_string())
);
assert_eq!(file.status(), &Status::Active);
}
}
impl Ord for File {
fn cmp(&self, other: &Self) -> Ordering {
self.inner
.status
.cmp(&other.inner.status)
.then_with(|| self.inner.path.cmp(&other.inner.path))
}
}
impl PartialOrd for File {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
Some(self.cmp(other))
}
}
pub fn scan_journal_files(path: &str) -> Result<Vec<File>> {
let mut files = Vec::new();
for entry in walkdir::WalkDir::new(path).follow_links(false) {
let entry = entry?;
let path = entry.path();
if path.is_file() {
if let Some(file) = File::from_path(path) {
files.push(file);
}
}
}
Ok(files)
}