//! listRepos backfill walk progress — persists the pagination cursor and
//! completion timestamp for the `com.atproto.sync.listRepos` walk.
//!
//! Key: `"lsr"<host>`
//! Value: `<cursor>` or `<cursor>\0<completed_at>`
use jacquard_common::url::Host;

use crate::storage::{
    DbRef, PREFIX_LIST_REPOS,
    error::{StorageError, StorageResult},
};

fn key(host: &Host) -> Vec<u8> {
    let host = host.to_string();
    let mut k = Vec::with_capacity(PREFIX_LIST_REPOS.len() + host.len());
    k.extend_from_slice(&PREFIX_LIST_REPOS);
    k.extend_from_slice(host.as_bytes());
    k
}

/// Progress record for the listRepos backfill walk.
#[derive(Debug, Clone)]
pub struct BackfillProgress {
    /// The current pagination cursor (empty string = start from beginning).
    pub cursor: String,
    /// Set when the full walk has completed at least once.
    pub completed_at: Option<String>,
}

/// Read the listRepos backfill progress for `host`.
///
/// Returns `None` if no walk has been started yet.
pub fn get(db: &DbRef, host: &Host) -> StorageResult<Option<BackfillProgress>> {
    let k = key(host);
    let key_str = String::from_utf8_lossy(&k).to_string();
    match db.ks.get(k)? {
        None => Ok(None),
        Some(v) => {
            let s = std::str::from_utf8(v.as_ref()).map_err(|_| StorageError::Corrupt {
                key: key_str,
                reason: "listRepos progress not valid UTF-8",
            })?;
            // Wire format: `<cursor>` or `<cursor>\0<completed_at>`
            let (cursor, completed_at) = match s.split_once('\0') {
                Some((c, ca)) => (c.to_owned(), Some(ca.to_owned())),
                None => (s.to_owned(), None),
            };
            Ok(Some(BackfillProgress {
                cursor,
                completed_at,
            }))
        }
    }
}

/// Persist the listRepos backfill progress for `host`.
pub fn set(db: &DbRef, host: &Host, progress: &BackfillProgress) -> StorageResult<()> {
    let value = match &progress.completed_at {
        None => progress.cursor.clone(),
        Some(ca) => format!("{}\0{}", progress.cursor, ca),
    };
    db.ks.insert(key(host), value.as_bytes())?;
    Ok(())
}