use kcode_k1_audio_classification_callback_proof::verify_callback;
use kcode_k1_audio_classification_projection_cache_format::{
CacheCursor, CacheRecord, decode_cursor, decode_record, encode_cursor, encode_record,
fragment_filename, parse_fragment_filename,
};
use kcode_k1_txn_ordering::K1TxnOrdering;
pub use kcode_k1_txn_ordering::TxId;
use std::collections::BTreeSet;
use std::fs::{self, File, OpenOptions};
use std::io::Write;
use std::path::{Path, PathBuf};
const LOGS: &str = "audio-classification-fragments";
const CURSOR: &str = "audio-classification-cursor.json";
const TEMP: &str = "audio-classification-cursor.json.tmp";
const SQLITE: &str = "audio-classification.sqlite3";
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CachedCallback {
pub callback_txid: TxId,
pub fragment_id: TxId,
pub payload: Vec<u8>,
}
fn cached(record: CacheRecord) -> CachedCallback {
CachedCallback {
callback_txid: record.callback_txid,
fragment_id: record.fragment_id,
payload: record.payload,
}
}
pub struct ProjectionCache {
root: PathBuf,
sequence: u64,
callbacks: BTreeSet<[u8; 12]>,
poisoned: bool,
}
type Loaded = (Vec<CachedCallback>, Option<TxId>);
impl ProjectionCache {
pub fn open(
root: &Path,
ordering: &K1TxnOrdering,
) -> Result<(Self, Vec<CachedCallback>, Option<TxId>), String> {
fs::create_dir_all(root).map_err(|error| context("create root", root, error))?;
for suffix in ["", "-wal", "-shm"] {
remove_file(&root.join(format!("{SQLITE}{suffix}")))?;
}
let (records, cursor) = match load(root, ordering)? {
Some(value) => value,
None => {
reset(root)?;
(Vec::new(), None)
}
};
let cache = Self {
root: root.into(),
sequence: records.len() as u64,
callbacks: records
.iter()
.map(|record| *record.callback_txid.as_bytes())
.collect(),
poisoned: false,
};
Ok((cache, records, cursor))
}
pub fn append(
&mut self,
callback_txid: TxId,
fragment_id: TxId,
payload: &[u8],
) -> Result<(), String> {
self.ensure_healthy()?;
if self.callbacks.contains(callback_txid.as_bytes()) {
return Err("duplicate callback transaction".into());
}
let sequence = self
.sequence
.checked_add(1)
.ok_or("cache sequence overflow")?;
let record = encode_record(&CacheRecord {
sequence,
callback_txid,
fragment_id,
payload: payload.to_vec(),
})?;
let cursor = encode_cursor(CacheCursor {
last_sequence: sequence,
last_callback_txid: callback_txid,
})?;
if let Err(error) = persist(&self.root, fragment_id, &record, &cursor) {
self.poisoned = true;
return Err(error);
}
self.sequence = sequence;
self.callbacks.insert(*callback_txid.as_bytes());
Ok(())
}
pub fn clear(&mut self) -> Result<(), String> {
self.ensure_healthy()?;
if let Err(error) = reset(&self.root) {
self.poisoned = true;
return Err(error);
}
self.sequence = 0;
self.callbacks.clear();
Ok(())
}
pub fn ensure_healthy(&self) -> Result<(), String> {
(!self.poisoned)
.then_some(())
.ok_or_else(|| "cache is poisoned; reopen required".into())
}
}
fn load(root: &Path, ordering: &K1TxnOrdering) -> Result<Option<Loaded>, String> {
let logs = root.join(LOGS);
let cursor_path = root.join(CURSOR);
if metadata(&root.join(TEMP))?.is_some() {
return Ok(None);
}
if !metadata(&logs)?.is_some_and(|value| value.file_type().is_dir()) {
return Ok(None);
}
let Some(cursor_metadata) = metadata(&cursor_path)? else {
let mut entries = fs::read_dir(&logs).map_err(|error| context("read", &logs, error))?;
return match entries.next() {
None => Ok(Some((Vec::new(), None))),
Some(Ok(_)) => Ok(None),
Some(Err(error)) => Err(context("read", &logs, error)),
};
};
if !cursor_metadata.file_type().is_file() {
return Ok(None);
}
let Ok(cursor) = decode_cursor(&read(&cursor_path)?) else {
return Ok(None);
};
let Some(mut records) = read_records(&logs)? else {
return Ok(None);
};
records.sort_by_key(|record| record.0);
if u64::try_from(records.len()).ok() != Some(cursor.last_sequence) {
return Ok(None);
}
let mut callbacks = BTreeSet::new();
if records.iter().enumerate().any(|(index, record)| {
record.0 != index as u64 + 1 || !callbacks.insert(*record.1.callback_txid.as_bytes())
}) || records.last().map(|record| record.1.callback_txid) != Some(cursor.last_callback_txid)
{
return Ok(None);
}
for (_, record) in &records {
if !verify_callback(ordering, record.callback_txid, &record.payload)? {
return Ok(None);
}
}
Ok(Some((
records.into_iter().map(|record| record.1).collect(),
Some(cursor.last_callback_txid),
)))
}
fn read_records(logs: &Path) -> Result<Option<Vec<(u64, CachedCallback)>>, String> {
let mut records = Vec::new();
for entry in fs::read_dir(logs).map_err(|error| context("read", logs, error))? {
let entry = entry.map_err(|error| context("read", logs, error))?;
let kind = entry
.file_type()
.map_err(|error| context("inspect", &entry.path(), error))?;
if !kind.is_file() {
return Ok(None);
}
let name = entry.file_name();
let Some(name) = name.to_str() else {
return Ok(None);
};
let Ok(fragment) = parse_fragment_filename(name) else {
return Ok(None);
};
let bytes = read(&entry.path())?;
if bytes.is_empty() || bytes.last() != Some(&b'\n') {
return Ok(None);
}
let mut previous = 0;
for line in bytes.split_inclusive(|byte| *byte == b'\n') {
let Ok(record) = decode_record(line) else {
return Ok(None);
};
if record.sequence <= previous || record.fragment_id != fragment {
return Ok(None);
}
previous = record.sequence;
let sequence = record.sequence;
records.push((sequence, cached(record)));
}
}
Ok(Some(records))
}
fn persist(root: &Path, fragment: TxId, record: &[u8], cursor: &[u8]) -> Result<(), String> {
let logs = root.join(LOGS);
let path = logs.join(fragment_filename(fragment));
let mut file = OpenOptions::new()
.create(true)
.append(true)
.open(&path)
.map_err(|error| context("open", &path, error))?;
file.write_all(record)
.and_then(|_| file.flush())
.and_then(|_| file.sync_all())
.map_err(|error| context("sync", &path, error))?;
sync_dir(&logs)?;
let temporary = root.join(TEMP);
let mut file =
File::create(&temporary).map_err(|error| context("create", &temporary, error))?;
file.write_all(cursor)
.and_then(|_| file.flush())
.and_then(|_| file.sync_all())
.map_err(|error| context("sync", &temporary, error))?;
fs::rename(&temporary, root.join(CURSOR))
.map_err(|error| context("rename", &temporary, error))?;
sync_dir(root)
}
fn reset(root: &Path) -> Result<(), String> {
remove_path(&root.join(LOGS))?;
remove_file(&root.join(CURSOR))?;
remove_file(&root.join(TEMP))?;
let logs = root.join(LOGS);
fs::create_dir(&logs).map_err(|error| context("create", &logs, error))?;
sync_dir(&logs)?;
sync_dir(root)
}
fn metadata(path: &Path) -> Result<Option<fs::Metadata>, String> {
match fs::symlink_metadata(path) {
Ok(value) => Ok(Some(value)),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(error) => Err(context("inspect", path, error)),
}
}
fn read(path: &Path) -> Result<Vec<u8>, String> {
fs::read(path).map_err(|error| context("read", path, error))
}
fn remove_path(path: &Path) -> Result<(), String> {
match fs::symlink_metadata(path) {
Ok(value) if value.file_type().is_dir() => {
fs::remove_dir_all(path).map_err(|error| context("remove", path, error))
}
Ok(_) => fs::remove_file(path).map_err(|error| context("remove", path, error)),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(error) => Err(context("inspect", path, error)),
}
}
fn remove_file(path: &Path) -> Result<(), String> {
match fs::remove_file(path) {
Ok(()) => Ok(()),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(error) => Err(context("remove", path, error)),
}
}
fn sync_dir(path: &Path) -> Result<(), String> {
File::open(path)
.and_then(|file| file.sync_all())
.map_err(|error| context("sync", path, error))
}
fn context(action: &str, path: &Path, error: impl std::fmt::Display) -> String {
format!("{action} {}: {error}", path.display())
}
#[cfg(test)]
mod tests;