use std::io;
use std::path::{Path, PathBuf};
use std::time::Duration;
use chrono::{DateTime, SecondsFormat, Utc};
use futures_util::StreamExt;
use serde::{Deserialize, Serialize};
use tokio::fs;
use crate::client::Client;
use crate::error::Error;
use crate::wire::decode_segment_stream;
pub struct Config {
pub server_url: http::Uri,
pub data_dir: PathBuf,
pub segments: u8,
}
pub enum Outcome {
UpToDate,
DeltaSync { changed_count: usize },
FullSync { file_count: usize },
}
#[derive(Serialize, Deserialize, Default)]
struct LocalState {
last_updated: Option<DateTime<Utc>>,
}
#[derive(Debug, Serialize, Deserialize)]
struct Plan {
server_last_updated: DateTime<Utc>,
since: Option<String>,
segments: u8,
}
#[tracing::instrument(skip_all)]
pub async fn sync(config: &Config) -> Result<Outcome, Error> {
if config.segments == 0 {
return Err(Error::InvalidConfig("segments must be >= 1"));
}
let staging = config.data_dir.join(".staging");
let complete_marker = staging.join(".complete");
let plan_path = staging.join(".sync-plan.json");
if staging.exists() {
if complete_marker.exists() {
tracing::info!("staging/.complete exists - finishing interrupted commit");
return finish_commit(&staging, &config.data_dir).await;
} else if plan_path.exists() {
tracing::info!("resuming interrupted download");
let plan: Plan = serde_json::from_slice(&fs::read(&plan_path).await?)?;
let client = Client::new(&config.server_url)?;
let status = client.status().await?;
if status.last_updated != Some(plan.server_last_updated) {
tracing::warn!(
"server state changed since last attempt; discarding staging and starting fresh"
);
clear_staging(&staging).await?;
} else {
fetch_missing_segments(&config.server_url, &staging, &plan).await?;
fs::write(&complete_marker, b"").await?;
return finish_commit(&staging, &config.data_dir).await;
}
} else {
tracing::warn!("staging exists without .sync-plan.json; discarding");
clear_staging(&staging).await?;
}
}
let state_path = config.data_dir.join("sync-state.json");
let local: LocalState = match fs::read(&state_path).await {
Ok(bytes) => serde_json::from_slice(&bytes)?,
Err(e) if e.kind() == io::ErrorKind::NotFound => LocalState::default(),
Err(e) => return Err(e.into()),
};
let client = Client::new(&config.server_url)?;
let server_last_updated = match client.status().await?.last_updated {
Some(t) => t,
None => return Ok(Outcome::UpToDate),
};
if Some(server_last_updated) <= local.last_updated {
return Ok(Outcome::UpToDate);
}
let since_opt: Option<String> = if local.last_updated.is_none() {
None
} else {
let changed = client.changed().await?;
if changed.prev_last_updated == local.last_updated {
local.last_updated.map(|t| t.to_rfc3339_opts(SecondsFormat::Secs, true))
} else {
tracing::warn!(
"server prev_last_updated does not match local last_updated; falling back to full sync"
);
None
}
};
fs::create_dir_all(&staging).await?;
let plan = Plan { server_last_updated, since: since_opt, segments: config.segments };
fs::write(&plan_path, serde_json::to_vec_pretty(&plan)?).await?;
fetch_missing_segments(&config.server_url, &staging, &plan).await?;
fs::write(&complete_marker, b"").await?;
finish_commit(&staging, &config.data_dir).await
}
#[tracing::instrument(skip(server_url, staging), fields(segments = plan.segments, since = plan.since.as_deref()))]
async fn fetch_missing_segments(
server_url: &http::Uri,
staging: &Path,
plan: &Plan,
) -> Result<(), Error> {
let segments = plan.segments;
let client = Client::new(server_url)?;
for seg in 0..segments {
if staging.join(format!(".seg.{}.done", seg)).exists() {
continue;
}
fetch_segment_with_retry(&client, seg, segments, plan.since.as_deref(), staging).await?;
}
Ok(())
}
#[tracing::instrument(skip(client, staging))]
async fn fetch_segment_with_retry(
client: &Client,
segment: u8,
of: u8,
since: Option<&str>,
staging: &Path,
) -> Result<(), Error> {
const MAX_RETRIES: u32 = 5;
let mut delay = Duration::from_millis(500);
let mut last_result = Ok(());
for attempt in 0..MAX_RETRIES {
if attempt > 0 {
tokio::time::sleep(delay).await;
delay *= 2;
}
match client.segment_stream(segment, of, since).await {
Ok(decoder) => {
let mut stream = Box::pin(decode_segment_stream(decoder));
while let Some(entry_res) = stream.next().await {
let entry = entry_res?;
let prefix_str = std::str::from_utf8(&entry.prefix)
.map_err(|e| Error::Decode(format!("invalid prefix bytes: {e}")))?;
fs::write(staging.join(format!("{}.bin", prefix_str)), &entry.content).await?;
}
fs::write(staging.join(format!(".seg.{}.done", segment)), b"").await?;
return Ok(());
}
Err(e) => last_result = Err(e),
}
}
last_result
}
#[tracing::instrument(skip_all)]
async fn finish_commit(staging: &Path, data_dir: &Path) -> Result<Outcome, Error> {
let plan: Plan = serde_json::from_slice(&fs::read(staging.join(".sync-plan.json")).await?)?;
let mut entries = fs::read_dir(staging).await?;
let mut file_count = 0usize;
while let Some(entry) = entries.next_entry().await? {
let src = entry.path();
if src.extension().is_some_and(|e| e == "bin") {
fs::rename(&src, data_dir.join(src.file_name().unwrap())).await?;
file_count += 1;
}
}
let state_path = data_dir.join("sync-state.json");
let new_state = LocalState { last_updated: Some(plan.server_last_updated) };
let tmp = state_path.with_extension("json.tmp");
fs::write(&tmp, serde_json::to_vec_pretty(&new_state)?).await?;
fs::rename(&tmp, &state_path).await?;
clear_staging(staging).await?;
Ok(match plan.since {
Some(_) => Outcome::DeltaSync { changed_count: file_count },
None => Outcome::FullSync { file_count },
})
}
async fn clear_staging(staging: &Path) -> Result<(), Error> {
fs::remove_dir_all(staging).await?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn sync_rejects_zero_segments() {
let tmp = tempfile::tempdir().unwrap();
let cfg = Config {
server_url: "http://127.0.0.1:8765".parse().unwrap(),
data_dir: tmp.path().to_path_buf(),
segments: 0,
};
match sync(&cfg).await {
Err(Error::InvalidConfig(_)) => {}
Err(e) => panic!("expected InvalidConfig, got {e}"),
Ok(_) => panic!("expected error for zero segments"),
}
}
}