use anyhow::{anyhow, Result};
use serde::{Deserialize, Serialize};
use std::fs;
use std::path::PathBuf;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ScanCursor {
Start,
AfterTimeId { last_time: String, last_id: i64 },
}
#[derive(Debug, Clone)]
pub struct ScanRequest {
pub table: String,
pub select: String,
pub time_column: String,
pub batch_size: usize,
pub cursor: ScanCursor,
}
impl ScanRequest {
pub fn new(
table: impl Into<String>,
select: impl Into<String>,
time_column: impl Into<String>,
) -> Self {
Self {
table: table.into(),
select: select.into(),
time_column: time_column.into(),
batch_size: 500,
cursor: ScanCursor::Start,
}
}
pub fn with_batch_size(mut self, batch_size: usize) -> Self {
self.batch_size = batch_size;
self
}
pub fn with_cursor(mut self, cursor: ScanCursor) -> Self {
self.cursor = cursor;
self
}
}
#[derive(Debug, Clone)]
pub struct ScanPage<T> {
pub rows: Vec<T>,
pub next: Option<ScanCursor>,
pub exhausted: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ScanCheckpoint {
pub table: String,
pub job_id: String,
pub cursor: ScanCursor,
pub updated_at: String,
}
pub trait ScanRow {
fn row_id(&self) -> i64;
fn scan_time(&self) -> &str;
}
pub fn page_from_rows<T: ScanRow>(mut rows: Vec<T>, batch_size: usize) -> ScanPage<T> {
if rows.is_empty() {
return ScanPage {
rows,
next: None,
exhausted: true,
};
}
let has_more = rows.len() > batch_size;
if has_more {
rows.truncate(batch_size);
}
let last = rows.last().expect("non-empty");
let next = ScanCursor::AfterTimeId {
last_time: last.scan_time().to_string(),
last_id: last.row_id(),
};
ScanPage {
exhausted: !has_more,
rows,
next: Some(next),
}
}
#[derive(Debug, Clone)]
pub struct FileCheckpointStore {
dir: PathBuf,
}
impl FileCheckpointStore {
pub fn new(dir: impl Into<PathBuf>) -> Self {
Self { dir: dir.into() }
}
fn path_for(&self, job_id: &str) -> Result<PathBuf> {
let safe = sanitize_job_id(job_id)?;
Ok(self.dir.join(format!("{safe}.json")))
}
pub fn load(&self, job_id: &str) -> Result<Option<ScanCheckpoint>> {
let path = self.path_for(job_id)?;
if !path.exists() {
return Ok(None);
}
let text = fs::read_to_string(&path)?;
serde_json::from_str(&text).map(Some).map_err(|e| {
anyhow!(
"checkpoint {} is unreadable ({e}); use a new job_id",
path.display()
)
})
}
pub fn save(&self, checkpoint: &ScanCheckpoint) -> Result<()> {
fs::create_dir_all(&self.dir)?;
let path = self.path_for(&checkpoint.job_id)?;
let tmp = path.with_extension("json.tmp");
fs::write(&tmp, serde_json::to_string_pretty(checkpoint)?)?;
fs::File::open(&tmp)?.sync_all()?;
fs::rename(&tmp, &path)?;
Ok(())
}
}
fn sanitize_job_id(job_id: &str) -> Result<String> {
if job_id.is_empty() {
return Err(anyhow!("job_id must not be empty"));
}
if job_id.contains('/') || job_id.contains('\\') || job_id.contains("..") {
return Err(anyhow!("job_id must not contain path separators: {job_id}"));
}
Ok(job_id.to_string())
}
pub(crate) fn time_id_keyset_or(time_column: &str, last_time: &str, last_id: i64) -> String {
format!(
"and({time_column}.eq.\"{last_time}\",id.gt.{last_id}),{time_column}.gt.\"{last_time}\""
)
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::tempdir;
#[derive(Clone)]
struct Row {
id: i64,
t: String,
}
impl ScanRow for Row {
fn row_id(&self) -> i64 {
self.id
}
fn scan_time(&self) -> &str {
&self.t
}
}
fn row(id: i64, t: &str) -> Row {
Row { id, t: t.into() }
}
#[test]
fn page_from_rows_exact_batch_is_terminal() {
let page = page_from_rows(
vec![
row(9, "2026-01-01T00:00:00Z"),
row(2, "2026-01-02T00:00:00Z"),
],
2,
);
assert!(page.exhausted);
assert_eq!(
page.next,
Some(ScanCursor::AfterTimeId {
last_time: "2026-01-02T00:00:00Z".into(),
last_id: 2,
})
);
}
#[test]
fn page_from_rows_drops_overfetch_and_uses_last_returned_row() {
let page = page_from_rows(
vec![
row(9, "2026-01-01T00:00:00Z"),
row(2, "2026-01-02T00:00:00Z"),
row(7, "2026-01-03T00:00:00Z"),
],
2,
);
assert!(!page.exhausted);
assert_eq!(page.rows.len(), 2);
assert_eq!(
page.next,
Some(ScanCursor::AfterTimeId {
last_time: "2026-01-02T00:00:00Z".into(),
last_id: 2,
})
);
}
#[test]
fn page_from_rows_exhausts_on_short_batch() {
let page = page_from_rows(vec![row(9, "t")], 3);
assert!(page.exhausted);
}
#[test]
fn time_id_keyset_or_quotes_timestamp() {
let or = time_id_keyset_or("updated_at", "2026-01-01T00:00:00Z", 7);
assert_eq!(
or,
"and(updated_at.eq.\"2026-01-01T00:00:00Z\",id.gt.7),updated_at.gt.\"2026-01-01T00:00:00Z\""
);
}
#[test]
fn file_checkpoint_round_trip() {
let dir = tempdir().unwrap();
let store = FileCheckpointStore::new(dir.path());
let checkpoint = ScanCheckpoint {
table: "artifacts".into(),
job_id: "media_nightly".into(),
cursor: ScanCursor::AfterTimeId {
last_time: "2026-01-01T00:00:00Z".into(),
last_id: 42,
},
updated_at: "2026-01-01T00:00:00Z".into(),
};
store.save(&checkpoint).unwrap();
assert_eq!(store.load("media_nightly").unwrap().unwrap(), checkpoint);
}
#[test]
fn file_checkpoint_rejects_unsafe_job_id() {
let dir = tempdir().unwrap();
let store = FileCheckpointStore::new(dir.path());
let checkpoint = ScanCheckpoint {
table: "artifacts".into(),
job_id: "../evil".into(),
cursor: ScanCursor::Start,
updated_at: "2026-01-01T00:00:00Z".into(),
};
assert!(store.save(&checkpoint).is_err());
}
}