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,
AfterId { last_id: i64 },
AfterTimeId { last_time: String, last_id: i64 },
}
#[derive(Debug, Clone)]
pub struct ScanRequest {
pub table: String,
pub select: String,
pub batch_size: usize,
pub cursor: ScanCursor,
pub time_column: Option<String>,
}
impl ScanRequest {
pub fn new(table: impl Into<String>, select: impl Into<String>) -> Self {
Self {
table: table.into(),
select: select.into(),
batch_size: 500,
cursor: ScanCursor::Start,
time_column: None,
}
}
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
}
pub fn with_time_column(mut self, time_column: impl Into<String>) -> Self {
self.time_column = Some(time_column.into());
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 HasRowId {
fn row_id(&self) -> Option<i64>;
}
pub trait HasScanTime {
fn scan_time(&self, time_column: &str) -> Option<&str>;
}
pub fn page_from_rows<T: HasRowId + HasScanTime>(
rows: Vec<T>,
batch_size: usize,
time_column: Option<&str>,
) -> Result<ScanPage<T>> {
if rows.is_empty() {
return Ok(ScanPage {
rows,
next: None,
exhausted: true,
});
}
let last = rows.last().expect("non-empty");
let last_id = last
.row_id()
.ok_or_else(|| anyhow!("scan page last row missing id"))?;
let exhausted = rows.len() < batch_size;
let next = if let Some(col) = time_column {
let last_time = last
.scan_time(col)
.ok_or_else(|| anyhow!("scan page last row missing time column '{col}'"))?
.to_string();
ScanCursor::AfterTimeId { last_time, last_id }
} else {
ScanCursor::AfterId { last_id }
};
Ok(ScanPage {
rows,
next: Some(next),
exhausted,
})
}
#[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)?;
Ok(Some(serde_json::from_str(&text)?))
}
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: Option<i64>,
t: Option<String>,
}
impl HasRowId for Row {
fn row_id(&self) -> Option<i64> {
self.id
}
}
impl HasScanTime for Row {
fn scan_time(&self, _time_column: &str) -> Option<&str> {
self.t.as_deref()
}
}
fn row(id: i64, t: &str) -> Row {
Row {
id: Some(id),
t: Some(t.into()),
}
}
#[test]
fn page_from_rows_advances_on_full_batch() {
let page = page_from_rows(vec![row(1, "a"), row(2, "b"), row(3, "c")], 3, None).unwrap();
assert!(!page.exhausted);
assert_eq!(page.next, Some(ScanCursor::AfterId { last_id: 3 }));
}
#[test]
fn page_from_rows_time_keyset_uses_last_row_not_max_id() {
let page = page_from_rows(
vec![
row(9, "2026-01-01T00:00:00Z"),
row(2, "2026-01-02T00:00:00Z"),
],
2,
Some("updated_at"),
)
.unwrap();
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, None).unwrap();
assert!(page.exhausted);
assert_eq!(page.next, Some(ScanCursor::AfterId { last_id: 9 }));
}
#[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());
}
}