neiam /devendra

//! Adoption: scan an existing podman/systemd directory and publish the files
//! found as a *proposal* for the controller to review.
//!
//! This is the onboarding path for a host that was configured by hand before
//! devendra existed. It deliberately writes only into the host's `adopt/`
//! keyspace — never the controller-owned manifest — so a proposal can never be
//! mistaken for desired state. The operator reviews it in the web UI and
//! accepts the files they want, at which point they become managed config.

use std::path::{Path, PathBuf};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use anyhow::{Context, Result};
use devendra_node::{DocTicket, LiveEvent, Node, StreamExt};
use devendra_proto::{AdoptedFile, AdoptionReport, HostId, PROTOCOL_VERSION, content_hash, keys};
use tracing::{info, warn};

/// Inputs for an adoption scan-and-publish.
pub struct AdoptConfig {
    pub host_id: HostId,
    pub data_dir: PathBuf,
    pub ticket: DocTicket,
    /// Directory to scan (e.g. `/etc/containers/systemd`).
    pub scan_dir: PathBuf,
}

struct Discovered {
    /// Absolute path on this machine.
    path: String,
    /// Path relative to the scan directory.
    rel: String,
    bytes: Vec<u8>,
    mode: u32,
}

/// Scan the directory, publish the discovered files as an adoption proposal,
/// wait briefly for it to sync to the controller, then shut down.
pub async fn adopt(config: AdoptConfig) -> Result<()> {
    let found = scan(&config.scan_dir)
        .with_context(|| format!("scanning {}", config.scan_dir.display()))?;
    if found.is_empty() {
        warn!(dir = %config.scan_dir.display(), "no files found to adopt");
    } else {
        info!(count = found.len(), dir = %config.scan_dir.display(), "discovered files");
    }

    let node = Node::spawn(&config.data_dir, None).await?;
    info!(node_id = %node.node_id(), host = %config.host_id, "adoption node up");
    let (doc, mut events) = node.import_doc(config.ticket.clone()).await?;

    // Replace any prior proposal wholesale: clear the whole adopt/ prefix, then
    // write the current scan. Newer entries supersede the tombstones.
    node.delete_prefix(&doc, &keys::adopt_prefix(&config.host_id))
        .await
        .context("clearing previous proposal")?;

    let mut files = Vec::with_capacity(found.len());
    for f in &found {
        node.write(
            &doc,
            &keys::adopt_file(&config.host_id, &f.path),
            f.bytes.clone(),
        )
        .await
        .with_context(|| format!("publishing body for {}", f.path))?;
        files.push(AdoptedFile {
            path: f.path.clone(),
            rel_path: f.rel.clone(),
            content_blake3: content_hash(&f.bytes),
            mode: f.mode,
            size: f.bytes.len() as u64,
        });
    }

    let report = AdoptionReport {
        protocol_version: PROTOCOL_VERSION,
        scan_dir: config.scan_dir.display().to_string(),
        discovered_at_ms: now_ms(),
        files,
    };
    node.write(
        &doc,
        &keys::adopt_manifest(&config.host_id),
        serde_json::to_vec(&report).context("serializing adoption report")?,
    )
    .await
    .context("publishing adoption report")?;

    info!(
        count = report.files.len(),
        "published adoption proposal; waiting for it to sync to the controller"
    );

    // Give the proposal time to reach the controller before we exit. Return as
    // soon as a sync round completes; otherwise time out (the entries are
    // persisted locally and will sync on the controller's next connection).
    let synced = tokio::time::timeout(Duration::from_secs(20), async {
        while let Some(event) = events.next().await {
            if matches!(event, Ok(LiveEvent::SyncFinished(_))) {
                return true;
            }
        }
        false
    })
    .await
    .unwrap_or(false);

    if synced {
        info!("proposal synced to a peer");
    } else {
        warn!(
            "no sync confirmation within timeout; proposal is stored locally and will sync later"
        );
    }

    node.shutdown().await?;
    info!(host = %config.host_id, "adoption complete — review it in the controller UI");
    Ok(())
}

/// Recursively collect regular files under `dir`. Symlinks are skipped (we
/// adopt real files, not links). A missing directory yields an empty list.
fn scan(dir: &Path) -> Result<Vec<Discovered>> {
    let mut out = Vec::new();
    if !dir.exists() {
        return Ok(out);
    }
    walk(dir, dir, &mut out)?;
    out.sort_by(|a, b| a.path.cmp(&b.path));
    Ok(out)
}

fn walk(root: &Path, dir: &Path, out: &mut Vec<Discovered>) -> Result<()> {
    use std::os::unix::fs::MetadataExt;
    for entry in std::fs::read_dir(dir).with_context(|| format!("reading {}", dir.display()))? {
        let entry = entry?;
        let path = entry.path();
        let file_type = entry.file_type()?;
        if file_type.is_symlink() {
            continue;
        }
        if file_type.is_dir() {
            walk(root, &path, out)?;
        } else if file_type.is_file() {
            let bytes =
                std::fs::read(&path).with_context(|| format!("reading {}", path.display()))?;
            let mode = std::fs::symlink_metadata(&path)?.mode() & 0o7777;
            let rel = path
                .strip_prefix(root)
                .unwrap_or(&path)
                .to_string_lossy()
                .into_owned();
            out.push(Discovered {
                path: path.to_string_lossy().into_owned(),
                rel,
                bytes,
                mode,
            });
        }
    }
    Ok(())
}

fn now_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}