use crate::prelude::*;

pub(crate) async fn snapshot_report(
    runtime: &Runtime,
    options: SnapshotReportOptions,
) -> Result<Value> {
    if !options.dir.exists() {
        return err(format!("{} does not exist", options.dir.display()));
    }

    let (verify, verify_ok) = if options.stats.verify {
        snapshot_report_component(verify_snapshot(&options.dir))
    } else {
        (
            json!({
                "ok": true,
                "skipped": true,
                "reason": "--skip-verify was set",
            }),
            true,
        )
    };
    let (stats, stats_ok) =
        snapshot_report_component(snapshot_stats(&options.dir, options.stats.clone()));
    let (doctor, doctor_ok) =
        snapshot_report_component(snapshot_doctor(&options.dir, options.doctor.clone()));
    let (neighbors, neighbors_ok) = snapshot_report_component(snapshot_report_neighbors(&options));
    let (drift, drift_ok) = if options.include_drift {
        snapshot_report_component(snapshot_drift(runtime, options.drift.clone()).await)
    } else {
        (
            json!({
                "ok": true,
                "included": false,
                "reason": "run with --drift or --full-contact-ids to compare live me.sh data",
            }),
            true,
        )
    };

    let summary = snapshot_report_summary(
        SnapshotReportStatus {
            verify_ok,
            stats_ok,
            doctor_ok,
            neighbors_ok,
            drift_included: options.include_drift,
            drift_ok,
        },
        &neighbors,
        &drift,
    );
    let ok = summary.get("ok").and_then(Value::as_bool).unwrap_or(false);

    Ok(json!({
        "ok": ok,
        "snapshot": options.dir.display().to_string(),
        "generated_at_unix_ms": now_millis(),
        "summary": summary,
        "verify": verify,
        "stats": stats,
        "doctor": doctor,
        "neighbors": neighbors,
        "drift": drift,
    }))
}

pub(crate) fn snapshot_report_component(result: Result<Value>) -> (Value, bool) {
    match result {
        Ok(value) => {
            let ok = value.get("ok").and_then(Value::as_bool).unwrap_or(false);
            (value, ok)
        }
        Err(error) => (
            json!({
                "ok": false,
                "error": error.to_string(),
            }),
            false,
        ),
    }
}

pub(crate) fn snapshot_report_summary(
    status: SnapshotReportStatus,
    neighbors: &Value,
    drift: &Value,
) -> Value {
    let neighbor_count = neighbors
        .get("items")
        .and_then(Value::as_array)
        .map(Vec::len)
        .unwrap_or_default();
    let neighbor_failures = neighbors
        .get("failed")
        .and_then(Value::as_u64)
        .unwrap_or_default();
    let drift_changed = drift
        .get("summary")
        .and_then(|summary| summary.get("changed"))
        .and_then(Value::as_bool)
        .unwrap_or(false);
    let drift_full_contacts_requested = drift
        .get("live")
        .and_then(|live| live.get("full_contacts_requested"))
        .and_then(Value::as_u64)
        .unwrap_or_default();
    let drift_full_contacts_compared = drift_full_contacts_requested == 0
        || drift
            .get("full_contacts")
            .and_then(|full_contacts| full_contacts.get("compared"))
            .and_then(Value::as_bool)
            .unwrap_or(false);
    let effective_drift_ok = status.drift_ok && drift_full_contacts_compared;
    let ok = status.verify_ok
        && status.stats_ok
        && status.doctor_ok
        && status.neighbors_ok
        && effective_drift_ok;
    json!({
        "ok": ok,
        "verify_ok": status.verify_ok,
        "stats_ok": status.stats_ok,
        "doctor_ok": status.doctor_ok,
        "neighbors_ok": status.neighbors_ok,
        "neighbor_count": neighbor_count,
        "neighbor_failures": neighbor_failures,
        "drift_included": status.drift_included,
        "drift_ok": effective_drift_ok,
        "drift_command_ok": status.drift_ok,
        "drift_changed": drift_changed,
        "drift_full_contacts_requested": drift_full_contacts_requested,
        "drift_full_contacts_compared": drift_full_contacts_compared,
    })
}

pub(crate) fn snapshot_report_neighbors(options: &SnapshotReportOptions) -> Result<Value> {
    if options.neighbors == 0 {
        return Ok(json!({
            "ok": true,
            "included": false,
            "reason": "--neighbors 0 was set",
            "items": [],
        }));
    }
    if !options.root.exists() {
        return Ok(json!({
            "ok": false,
            "included": true,
            "root": options.root.display().to_string(),
            "error": "neighbor root does not exist",
            "items": [],
        }));
    }

    let timeline_options = SnapshotTimelineOptions {
        root: options.root.clone(),
        recursive: options.recursive,
        max_depth: options.max_depth,
        limit: options.limit,
        changes_only: false,
        diffs: false,
        diff: options.diff,
    };
    let (snapshots, discovery_errors) = snapshot_timeline_snapshots(&timeline_options)?;
    let Some(position) = snapshot_report_find_position(&snapshots, &options.dir)? else {
        return Ok(json!({
            "ok": false,
            "included": true,
            "root": options.root.display().to_string(),
            "recursive": options.recursive,
            "max_depth": options.max_depth,
            "limit": options.limit,
            "error": "snapshot was not found under neighbor root",
            "snapshots": snapshots.len(),
            "discovery_errors": discovery_errors,
            "items": [],
        }));
    };

    let mut items = Vec::new();
    let previous_start = position.saturating_sub(options.neighbors);
    for neighbor_index in previous_start..position {
        let distance = position - neighbor_index;
        items.push(snapshot_report_neighbor_item(
            "previous",
            distance,
            &snapshots[neighbor_index],
            &snapshots[position],
            &snapshots[neighbor_index],
            options,
        )?);
    }
    let next_end = snapshots.len().min(position + options.neighbors + 1);
    for neighbor_index in (position + 1)..next_end {
        let distance = neighbor_index - position;
        items.push(snapshot_report_neighbor_item(
            "next",
            distance,
            &snapshots[position],
            &snapshots[neighbor_index],
            &snapshots[neighbor_index],
            options,
        )?);
    }

    let failed = items
        .iter()
        .filter(|item| !item.get("ok").and_then(Value::as_bool).unwrap_or(false))
        .count()
        + discovery_errors.len();
    Ok(json!({
        "ok": failed == 0,
        "included": true,
        "root": options.root.display().to_string(),
        "recursive": options.recursive,
        "max_depth": options.max_depth,
        "limit": options.limit,
        "neighbors": options.neighbors,
        "position": position,
        "snapshots": snapshots.len(),
        "discovery_errors": discovery_errors,
        "failed": failed,
        "items": items,
    }))
}

pub(crate) fn snapshot_report_find_position(
    snapshots: &[SnapshotHistorySnapshot],
    dir: &Path,
) -> Result<Option<usize>> {
    let target = fs::canonicalize(dir)
        .into_diagnostic()
        .wrap_err_with(|| format!("canonicalizing {}", dir.display()))?;
    for (index, snapshot) in snapshots.iter().enumerate() {
        let Ok(candidate) = fs::canonicalize(&snapshot.dir) else {
            continue;
        };
        if candidate == target {
            return Ok(Some(index));
        }
    }
    Ok(None)
}

pub(crate) fn snapshot_report_neighbor_item(
    relation: &str,
    distance: usize,
    old: &SnapshotHistorySnapshot,
    new: &SnapshotHistorySnapshot,
    neighbor: &SnapshotHistorySnapshot,
    options: &SnapshotReportOptions,
) -> Result<Value> {
    let diff = diff_snapshots(&old.dir, &new.dir, options.diff)?;
    let ok = diff.get("ok").and_then(Value::as_bool).unwrap_or(false);
    let summary = if ok {
        snapshot_timeline_diff_summary(&diff)
    } else {
        Value::Null
    };
    let changed = summary
        .get("changed")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    let mut item = json!({
        "relation": relation,
        "distance": distance,
        "ok": ok,
        "changed": changed,
        "neighbor": snapshot_report_snapshot_value(neighbor),
        "old": snapshot_report_snapshot_value(old),
        "new": snapshot_report_snapshot_value(new),
        "summary": summary,
    });
    if let Value::Object(object) = &mut item {
        if !ok && let Some(error) = diff.get("error") {
            object.insert("error".to_string(), error.clone());
        }
        if options.diffs {
            object.insert("diff".to_string(), diff);
        }
    }
    Ok(item)
}

pub(crate) fn snapshot_report_snapshot_value(snapshot: &SnapshotHistorySnapshot) -> Value {
    json!({
        "dir": snapshot.dir.display().to_string(),
        "created_at_unix_ms": snapshot.created_at_unix_ms,
        "rank_time_unix_ms": snapshot.rank_time_unix_ms,
        "rank_time_source": snapshot.rank_time_source,
    })
}

pub(crate) fn snapshot_report_table_rows(report: &Value) -> Value {
    let summary = report.get("summary").unwrap_or(&Value::Null);
    let snapshot = report
        .get("snapshot")
        .and_then(Value::as_str)
        .unwrap_or_default();
    let doctor_summary = report
        .get("doctor")
        .and_then(|doctor| doctor.get("summary"))
        .unwrap_or(&Value::Null);
    let stats_sections = report
        .get("stats")
        .and_then(|stats| stats.get("sections"))
        .and_then(Value::as_object)
        .map(Map::len)
        .unwrap_or_default();
    let drift_live = report
        .get("drift")
        .and_then(|drift| drift.get("live"))
        .unwrap_or(&Value::Null);

    Value::Array(vec![
        json!({
            "component": "overall",
            "status": snapshot_report_status(summary_bool(summary, "ok")),
            "detail": snapshot,
        }),
        json!({
            "component": "verify",
            "status": snapshot_report_status(summary_bool(summary, "verify_ok")),
            "detail": "manifest hashes",
        }),
        json!({
            "component": "stats",
            "status": snapshot_report_status(summary_bool(summary, "stats_ok")),
            "detail": format!("{stats_sections} sections"),
        }),
        json!({
            "component": "doctor",
            "status": snapshot_report_status(summary_bool(summary, "doctor_ok")),
            "detail": format!(
                "pass={} warn={} fail={}",
                value_u64(doctor_summary, "pass"),
                value_u64(doctor_summary, "warn"),
                value_u64(doctor_summary, "fail"),
            ),
        }),
        json!({
            "component": "neighbors",
            "status": snapshot_report_status(summary_bool(summary, "neighbors_ok")),
            "detail": format!(
                "items={} failures={}",
                value_u64(summary, "neighbor_count"),
                value_u64(summary, "neighbor_failures"),
            ),
        }),
        json!({
            "component": "drift",
            "status": if summary_bool(summary, "drift_included") {
                snapshot_report_status(summary_bool(summary, "drift_ok"))
            } else {
                "skipped"
            },
            "detail": if summary_bool(summary, "drift_included") {
                format!(
                    "changed={} contacts={} groups={} full_requested={} full_compared={}",
                    summary_bool(summary, "drift_changed"),
                    value_u64(drift_live, "contacts"),
                    value_u64(drift_live, "groups"),
                    value_u64(summary, "drift_full_contacts_requested"),
                    summary_bool(summary, "drift_full_contacts_compared"),
                )
            } else {
                "not requested".to_string()
            },
        }),
    ])
}

pub(crate) fn snapshot_report_status(ok: bool) -> &'static str {
    if ok { "ok" } else { "fail" }
}

pub(crate) async fn snapshot_drift(
    runtime: &Runtime,
    options: SnapshotDriftOptions,
) -> Result<Value> {
    if !options.dir.exists() {
        return err(format!("{} does not exist", options.dir.display()));
    }

    let snapshot_verify = if options.verify {
        let verify = verify_snapshot(&options.dir)?;
        if !verify.get("ok").and_then(Value::as_bool).unwrap_or(false) {
            return Ok(json!({
                "ok": false,
                "error": "snapshot failed manifest verification",
                "snapshot": options.dir.display().to_string(),
                "snapshot_verify": verify,
            }));
        }
        verify
    } else {
        json!({
            "skipped": true,
            "reason": "--skip-verify was set",
        })
    };

    let manifest = read_snapshot_manifest(&options.dir)?;
    let old_contacts_path = snapshot_manifest_file_path(&options.dir, "contacts")?;
    let old_contacts = read_snapshot_jsonl_records_at_path(&old_contacts_path)?;
    let (live_contacts, live_contact_values, live_contact_count) =
        live_contact_records(runtime, options.page_size, options.diff.details).await?;
    let mut contacts = record_diff(&old_contacts, &live_contacts);
    if options.diff.details {
        let old_contact_values = read_snapshot_jsonl_record_values_at_path(&old_contacts_path)?;
        contacts = add_record_diff_details(
            contacts,
            &old_contact_values,
            &live_contact_values,
            options.diff.detail_limit,
        )?;
    }

    let (groups, live_group_count) = if options.compare_groups {
        let old_groups_path = snapshot_manifest_file_path(&options.dir, "groups")?;
        let old_groups = read_snapshot_array_records_at_path(&old_groups_path)?;
        let (live_groups, live_group_values, live_group_count) =
            live_group_records(runtime, options.diff.details).await?;
        let mut groups = record_diff(&old_groups, &live_groups);
        if options.diff.details {
            let old_group_values = read_snapshot_array_record_values_at_path(&old_groups_path)?;
            groups = add_record_diff_details(
                groups,
                &old_group_values,
                &live_group_values,
                options.diff.detail_limit,
            )?;
        }
        (groups, Some(live_group_count))
    } else {
        (
            json!({
                "compared": false,
                "reason": "--skip-groups was set",
            }),
            None,
        )
    };

    let full_contacts = snapshot_drift_full_contacts(runtime, &options).await?;
    let summary_source = json!({
        "contacts": contacts,
        "groups": groups,
        "full_contacts": full_contacts,
    });
    let summary = snapshot_timeline_diff_summary(&summary_source);

    Ok(json!({
        "ok": true,
        "snapshot": options.dir.display().to_string(),
        "snapshot_verify": snapshot_verify,
        "snapshot_counts": manifest.get("counts").cloned().unwrap_or(Value::Null),
        "live": {
            "contacts": live_contact_count,
            "groups": live_group_count,
            "full_contacts_requested": options.full_contact_ids.len(),
        },
        "page_size": options.page_size,
        "details": options.diff.details,
        "detail_limit": options.diff.detail_limit,
        "summary": summary,
        "contacts": summary_source.get("contacts").cloned().unwrap_or(Value::Null),
        "groups": summary_source.get("groups").cloned().unwrap_or(Value::Null),
        "full_contacts": summary_source.get("full_contacts").cloned().unwrap_or(Value::Null),
    }))
}

pub(crate) async fn snapshot_drift_full_contacts(
    runtime: &Runtime,
    options: &SnapshotDriftOptions,
) -> Result<Value> {
    if options.full_contact_ids.is_empty() {
        return Ok(json!({
            "compared": false,
            "reason": "no --full-contact-ids were requested",
        }));
    }

    if !snapshot_manifest_has_file(&options.dir, "full_contacts")? {
        return Ok(json!({
            "compared": false,
            "snapshot_available": false,
            "reason": "snapshot does not contain full_contacts. Recreate it with --full-contact-ids or --full-contacts.",
            "requested_ids": options.full_contact_ids,
        }));
    }

    let old_path = snapshot_manifest_file_path(&options.dir, "full_contacts")?;
    let old_records = read_snapshot_jsonl_records_at_path(&old_path)?;
    let old_records = filter_records_by_ids(&old_records, &options.full_contact_ids);
    let (live_records, live_values) = live_full_contact_records(
        runtime,
        &options.full_contact_ids,
        options.full_concurrency,
        options.diff.details,
    )
    .await?;
    let mut diff = record_diff(&old_records, &live_records);
    if options.diff.details {
        let old_values = read_snapshot_jsonl_record_values_at_path(&old_path)?;
        let old_values = filter_records_by_ids(&old_values, &options.full_contact_ids);
        diff = add_record_diff_details(diff, &old_values, &live_values, options.diff.detail_limit)?;
    }
    if let Value::Object(object) = &mut diff {
        object.set("compared", true);
        object.set("snapshot_available", true);
        object.insert(
            "requested_ids".to_string(),
            Value::Array(
                options
                    .full_contact_ids
                    .iter()
                    .copied()
                    .map(|id| Value::Number(Number::from(id)))
                    .collect(),
            ),
        );
    }
    Ok(diff)
}

pub(crate) fn snapshot_catalog(options: SnapshotCatalogOptions) -> Result<Value> {
    if !options.root.exists() {
        return err(format!("{} does not exist", options.root.display()));
    }

    let mut candidates = Vec::new();
    let mut seen = BTreeSet::new();
    let mut discovery_errors = Vec::new();
    snapshot_catalog_discover(
        &options.root,
        0,
        &options,
        &mut seen,
        &mut candidates,
        &mut discovery_errors,
    )?;
    candidates.sort_by(|left, right| {
        snapshot_catalog_candidate_key(left).cmp(&snapshot_catalog_candidate_key(right))
    });

    let items = candidates
        .iter()
        .map(|candidate| match candidate.kind {
            SnapshotCatalogCandidateKind::Snapshot => {
                snapshot_catalog_snapshot_item(&candidate.path, &options)
            }
            SnapshotCatalogCandidateKind::Archive => {
                snapshot_catalog_archive_item(&candidate.path, &options)
            }
        })
        .collect::<Vec<_>>();
    let summary = snapshot_catalog_summary(&items, &discovery_errors);
    let ok = summary
        .get("failures")
        .and_then(Value::as_u64)
        .unwrap_or_default()
        == 0
        && summary
            .get("errors")
            .and_then(Value::as_u64)
            .unwrap_or_default()
            == 0;

    Ok(json!({
        "root": options.root.display().to_string(),
        "recursive": options.recursive,
        "max_depth": options.max_depth,
        "limit": options.limit,
        "include_snapshots": options.include_snapshots,
        "include_archives": options.include_archives,
        "verified": options.verify || options.doctor,
        "doctor": options.doctor,
        "require_index": options.require_index,
        "ok": ok,
        "summary": summary,
        "errors": discovery_errors,
        "items": items,
    }))
}

pub(crate) fn snapshot_catalog_discover(
    path: &Path,
    depth: usize,
    options: &SnapshotCatalogOptions,
    seen: &mut BTreeSet<String>,
    candidates: &mut Vec<SnapshotCatalogCandidate>,
    errors: &mut Vec<Value>,
) -> Result<()> {
    if snapshot_catalog_limit_reached(candidates.len(), options.limit) {
        return Ok(());
    }

    let metadata = match fs::symlink_metadata(path) {
        Ok(metadata) => metadata,
        Err(error) => {
            errors.push(snapshot_catalog_discovery_error(path, error));
            return Ok(());
        }
    };
    let file_type = metadata.file_type();
    if file_type.is_symlink() {
        return Ok(());
    }
    if file_type.is_file() {
        if options.include_archives && snapshot_catalog_is_archive(path) {
            snapshot_catalog_push_candidate(
                SnapshotCatalogCandidateKind::Archive,
                path,
                seen,
                candidates,
                options,
            );
        }
        return Ok(());
    }
    if !file_type.is_dir() {
        return Ok(());
    }

    if path.join("manifest.json").is_file() {
        if options.include_snapshots {
            snapshot_catalog_push_candidate(
                SnapshotCatalogCandidateKind::Snapshot,
                path,
                seen,
                candidates,
                options,
            );
        }
        return Ok(());
    }
    if depth > 0 && !options.recursive {
        return Ok(());
    }
    if options
        .max_depth
        .is_some_and(|max_depth| depth >= max_depth)
    {
        return Ok(());
    }

    let mut children = match fs::read_dir(path) {
        Ok(children) => children
            .collect::<std::result::Result<Vec<_>, _>>()
            .into_diagnostic()
            .wrap_err_with(|| format!("reading {}", path.display()))?,
        Err(error) => {
            errors.push(snapshot_catalog_discovery_error(path, error));
            return Ok(());
        }
    };
    children.sort_by_key(|entry| entry.path());
    for child in children {
        if snapshot_catalog_limit_reached(candidates.len(), options.limit) {
            break;
        }
        snapshot_catalog_discover(&child.path(), depth + 1, options, seen, candidates, errors)?;
    }
    Ok(())
}

pub(crate) fn snapshot_catalog_limit_reached(count: usize, limit: Option<usize>) -> bool {
    limit.is_some_and(|limit| count >= limit)
}

pub(crate) fn snapshot_catalog_push_candidate(
    kind: SnapshotCatalogCandidateKind,
    path: &Path,
    seen: &mut BTreeSet<String>,
    candidates: &mut Vec<SnapshotCatalogCandidate>,
    options: &SnapshotCatalogOptions,
) {
    if snapshot_catalog_limit_reached(candidates.len(), options.limit) {
        return;
    }
    let key = format!("{}:{}", kind.as_str(), path.display());
    if seen.insert(key) {
        candidates.push(SnapshotCatalogCandidate {
            kind,
            path: path.to_path_buf(),
        });
    }
}

pub(crate) fn snapshot_catalog_candidate_key(candidate: &SnapshotCatalogCandidate) -> String {
    format!("{}:{}", candidate.kind.as_str(), candidate.path.display())
}

pub(crate) fn snapshot_catalog_discovery_error(path: &Path, error: impl ToString) -> Value {
    json!({
        "path": path.display().to_string(),
        "error": error.to_string(),
    })
}

pub(crate) fn snapshot_catalog_snapshot_item(
    dir: &Path,
    options: &SnapshotCatalogOptions,
) -> Value {
    let manifest = match read_snapshot_manifest(dir) {
        Ok(manifest) => manifest,
        Err(error) => {
            return snapshot_catalog_error_item(
                SnapshotCatalogCandidateKind::Snapshot,
                dir,
                "snapshot manifest could not be read",
                error,
            );
        }
    };
    let mut item = snapshot_catalog_base_item(SnapshotCatalogCandidateKind::Snapshot, dir);
    snapshot_catalog_add_manifest_summary(&mut item, &manifest);

    if options.doctor {
        match snapshot_doctor(
            dir,
            SnapshotDoctorOptions {
                top: SNAPSHOT_STATS_TOP_DEFAULT,
                verify: true,
                require_index: options.require_index,
            },
        ) {
            Ok(report) => snapshot_catalog_add_report(&mut item, "doctor", report),
            Err(error) => {
                return snapshot_catalog_error_item(
                    SnapshotCatalogCandidateKind::Snapshot,
                    dir,
                    "snapshot doctor could not complete",
                    error,
                );
            }
        }
    } else if options.verify {
        match verify_snapshot(dir) {
            Ok(report) => snapshot_catalog_add_report(&mut item, "verify", report),
            Err(error) => {
                return snapshot_catalog_error_item(
                    SnapshotCatalogCandidateKind::Snapshot,
                    dir,
                    "snapshot verification could not complete",
                    error,
                );
            }
        }
    } else {
        item.set("status", json!("found"));
        item.set("ok", Value::Null);
    }

    Value::Object(item)
}

pub(crate) fn snapshot_catalog_base_item(
    kind: SnapshotCatalogCandidateKind,
    path: &Path,
) -> Map<String, Value> {
    let mut item = Map::new();
    item.set("kind", json!(kind.as_str()));
    item.set("path", json!(path.display().to_string()));
    item
}

pub(crate) fn snapshot_catalog_error_item(
    kind: SnapshotCatalogCandidateKind,
    path: &Path,
    message: &str,
    error: impl ToString,
) -> Value {
    let mut item = snapshot_catalog_base_item(kind, path);
    item.set("status", json!("error"));
    item.set("ok", json!(false));
    item.set("message", json!(message));
    item.set("error", json!(error.to_string()));
    Value::Object(item)
}

pub(crate) fn snapshot_catalog_add_manifest_summary(
    item: &mut Map<String, Value>,
    manifest: &Value,
) {
    item.insert(
        "schema".to_string(),
        manifest.get("schema").cloned().unwrap_or(Value::Null),
    );
    item.insert(
        "meshx_version".to_string(),
        manifest
            .get("meshx_version")
            .cloned()
            .unwrap_or(Value::Null),
    );
    item.insert(
        "created_at_unix_ms".to_string(),
        manifest
            .get("created_at_unix_ms")
            .cloned()
            .unwrap_or(Value::Null),
    );
    item.insert(
        "counts".to_string(),
        manifest.get("counts").cloned().unwrap_or(Value::Null),
    );
    let files = manifest.get("files").and_then(Value::as_object);
    item.insert(
        "file_count".to_string(),
        json!(files.map(Map::len).unwrap_or_default()),
    );
    item.insert(
        "snapshot_bytes".to_string(),
        json!(snapshot_catalog_manifest_bytes(files)),
    );
}

pub(crate) fn snapshot_catalog_manifest_bytes(files: Option<&Map<String, Value>>) -> u64 {
    files
        .into_iter()
        .flat_map(Map::values)
        .filter_map(|entry| entry.get("bytes").and_then(Value::as_u64))
        .sum()
}

pub(crate) fn snapshot_catalog_add_report(item: &mut Map<String, Value>, key: &str, report: Value) {
    let (status, ok) = snapshot_catalog_report_status(&report);
    item.set("status", json!(status));
    item.set("ok", json!(ok));
    item.insert(key.to_string(), report);
}

pub(crate) fn snapshot_catalog_report_status(report: &Value) -> (&'static str, bool) {
    let ok = report.get("ok").and_then(Value::as_bool).unwrap_or(false);
    if !ok {
        return ("fail", false);
    }
    let warnings = report
        .get("summary")
        .and_then(|summary| summary.get("warnings").or_else(|| summary.get("warn")))
        .and_then(Value::as_u64)
        .unwrap_or_default();
    if warnings > 0 {
        ("warn", true)
    } else {
        ("pass", true)
    }
}

pub(crate) fn snapshot_catalog_summary(items: &[Value], discovery_errors: &[Value]) -> Value {
    let mut snapshots = 0_u64;
    let mut archives = 0_u64;
    let mut pass = 0_u64;
    let mut warn = 0_u64;
    let mut fail = 0_u64;
    let mut found = 0_u64;
    let mut errors = discovery_errors.len() as u64;

    for item in items {
        match item.get("kind").and_then(Value::as_str) {
            Some("snapshot") => snapshots += 1,
            Some("archive") => archives += 1,
            _ => {}
        }
        match item.get("status").and_then(Value::as_str) {
            Some("pass") => pass += 1,
            Some("warn") => warn += 1,
            Some("fail") => fail += 1,
            Some("error") => {
                fail += 1;
                errors += 1;
            }
            Some("found") | None => found += 1,
            Some(_) => warn += 1,
        }
    }

    json!({
        "total": items.len(),
        "snapshots": snapshots,
        "archives": archives,
        "found": found,
        "pass": pass,
        "warn": warn,
        "fail": fail,
        "errors": errors,
        "failures": fail,
        "warnings": warn,
    })
}

pub(crate) struct SnapshotPrunePlan {
    pub(crate) actions: Vec<SnapshotPruneAction>,
    pub(crate) discovery_errors: Vec<Value>,
}

pub(crate) fn snapshot_prune(options: SnapshotPruneOptions) -> Result<Value> {
    if !options.root.exists() {
        return err(format!("{} does not exist", options.root.display()));
    }

    let plan = snapshot_prune_plan_actions(&options)?;
    let mut actions = plan.actions;
    let discovery_errors = plan.discovery_errors;
    if !discovery_errors.is_empty() && !options.dry_run {
        return err("snapshot:prune discovery failed. Re-run with --dry-run to inspect errors.");
    }
    let planned_deletes = actions.iter().filter(|action| action.delete).count();
    if planned_deletes > 0 && !options.dry_run && !options.yes {
        return err("snapshot:prune is destructive. Re-run with --yes, or use --dry-run.");
    }

    let mut delete_errors = Vec::new();
    let mut deleted = BTreeSet::new();
    if !options.dry_run {
        for action in actions.iter().filter(|action| action.delete) {
            match snapshot_prune_delete_action(action) {
                Ok(()) => {
                    deleted.insert(action.path.display().to_string());
                }
                Err(error) => delete_errors.push(json!({
                    "kind": action.kind.as_str(),
                    "path": action.path.display().to_string(),
                    "error": error.to_string(),
                })),
            }
        }
    }

    let action_values = actions
        .drain(..)
        .map(|action| snapshot_prune_action_value(action, &deleted, options.dry_run))
        .collect::<Vec<_>>();
    let summary = snapshot_prune_summary(&action_values, &delete_errors, &discovery_errors);
    let action_errors = summary
        .get("action_errors")
        .and_then(Value::as_u64)
        .unwrap_or_default();
    let ok = delete_errors.is_empty() && discovery_errors.is_empty() && action_errors == 0;

    Ok(json!({
        "root": options.root.display().to_string(),
        "recursive": options.recursive,
        "max_depth": options.max_depth,
        "limit": options.limit,
        "include_snapshots": options.include_snapshots,
        "include_archives": options.include_archives,
        "criteria": snapshot_prune_criteria_value(&options),
        "dry_run": options.dry_run,
        "confirmed": options.yes,
        "ok": ok,
        "summary": summary,
        "delete_errors": delete_errors,
        "discovery_errors": discovery_errors,
        "actions": action_values,
    }))
}

pub(crate) fn snapshot_prune_plan_actions(
    options: &SnapshotPruneOptions,
) -> Result<SnapshotPrunePlan> {
    let catalog_options = SnapshotCatalogOptions {
        root: options.root.clone(),
        recursive: options.recursive,
        max_depth: options.max_depth,
        limit: options.limit,
        include_snapshots: options.include_snapshots,
        include_archives: options.include_archives,
        verify: false,
        doctor: false,
        require_index: false,
    };
    let mut candidates = Vec::new();
    let mut seen = BTreeSet::new();
    let mut discovery_errors = Vec::new();
    snapshot_catalog_discover(
        &catalog_options.root,
        0,
        &catalog_options,
        &mut seen,
        &mut candidates,
        &mut discovery_errors,
    )?;
    candidates.sort_by(|left, right| {
        snapshot_catalog_candidate_key(left).cmp(&snapshot_catalog_candidate_key(right))
    });

    let cutoff = snapshot_prune_cutoff(options.older_than_days)?;
    let mut actions = candidates
        .into_iter()
        .map(|candidate| snapshot_prune_action(candidate, options, cutoff))
        .collect::<Result<Vec<_>>>()?;
    snapshot_prune_apply_keep_latest(&mut actions, options.keep_latest);
    Ok(SnapshotPrunePlan {
        actions,
        discovery_errors,
    })
}

pub(crate) fn snapshot_prune_action(
    candidate: SnapshotCatalogCandidate,
    options: &SnapshotPruneOptions,
    cutoff: Option<u64>,
) -> Result<SnapshotPruneAction> {
    let (rank_time_unix_ms, rank_time_source) =
        snapshot_prune_rank_time(candidate.kind, &candidate.path)?;
    let mut error = None;
    let bytes = match snapshot_prune_candidate_bytes(candidate.kind, &candidate.path) {
        Ok(bytes) => bytes,
        Err(byte_error) => {
            error = Some(byte_error.to_string());
            None
        }
    };
    let mut reasons = Vec::new();
    if let Some(cutoff) = cutoff
        && rank_time_unix_ms.is_some_and(|rank_time| rank_time < cutoff)
    {
        reasons.push("older-than-days".to_string());
    }

    let mut health = None;
    if options.failed {
        match snapshot_prune_health(candidate.kind, &candidate.path, options.require_index) {
            Ok(report) => {
                if !report.get("ok").and_then(Value::as_bool).unwrap_or(false) {
                    reasons.push("failed".to_string());
                }
                health = Some(report);
            }
            Err(health_error) => {
                reasons.push("failed".to_string());
                error = Some(health_error.to_string());
            }
        }
    }

    Ok(SnapshotPruneAction {
        kind: candidate.kind,
        path: candidate.path,
        delete: !reasons.is_empty(),
        reasons,
        rank_time_unix_ms,
        rank_time_source,
        bytes,
        health,
        error,
    })
}

pub(crate) fn snapshot_prune_apply_keep_latest(
    actions: &mut [SnapshotPruneAction],
    keep_latest: Option<usize>,
) {
    let Some(keep_latest) = keep_latest else {
        return;
    };
    for kind in [
        SnapshotCatalogCandidateKind::Snapshot,
        SnapshotCatalogCandidateKind::Archive,
    ] {
        let mut indexes = actions
            .iter()
            .enumerate()
            .filter(|(_, action)| action.kind == kind)
            .map(|(index, action)| {
                (
                    action.rank_time_unix_ms.unwrap_or_default(),
                    action.path.display().to_string(),
                    index,
                )
            })
            .collect::<Vec<_>>();
        indexes.sort_by(|left, right| right.0.cmp(&left.0).then_with(|| left.1.cmp(&right.1)));
        for (_, _, index) in indexes.into_iter().skip(keep_latest) {
            actions[index].reasons.push("keep-latest".to_string());
            actions[index].delete = true;
        }
    }
}

pub(crate) fn snapshot_prune_cutoff(older_than_days: Option<usize>) -> Result<Option<u64>> {
    let Some(days) = older_than_days else {
        return Ok(None);
    };
    let millis = (days as u64)
        .checked_mul(24)
        .and_then(|value| value.checked_mul(60))
        .and_then(|value| value.checked_mul(60))
        .and_then(|value| value.checked_mul(1000))
        .ok_or_else(|| miette!("--older-than-days is too large"))?;
    Ok(Some(now_millis().saturating_sub(millis)))
}

pub(crate) fn snapshot_prune_rank_time(
    kind: SnapshotCatalogCandidateKind,
    path: &Path,
) -> Result<(Option<u64>, &'static str)> {
    if kind == SnapshotCatalogCandidateKind::Snapshot
        && let Ok(manifest) = read_snapshot_manifest(path)
        && let Some(created_at) = manifest.get("created_at_unix_ms").and_then(Value::as_u64)
    {
        return Ok((Some(created_at), "manifest.created_at_unix_ms"));
    }
    let metadata = fs::metadata(path)
        .into_diagnostic()
        .wrap_err_with(|| format!("reading {}", path.display()))?;
    let modified = metadata
        .modified()
        .into_diagnostic()
        .wrap_err_with(|| format!("reading mtime for {}", path.display()))?;
    Ok((Some(system_time_unix_ms(modified)?), "filesystem.modified"))
}

pub(crate) fn snapshot_prune_candidate_bytes(
    kind: SnapshotCatalogCandidateKind,
    path: &Path,
) -> Result<Option<u64>> {
    match kind {
        SnapshotCatalogCandidateKind::Archive => Ok(Some(
            fs::metadata(path)
                .into_diagnostic()
                .wrap_err_with(|| format!("reading {}", path.display()))?
                .len(),
        )),
        SnapshotCatalogCandidateKind::Snapshot => {
            let manifest = read_snapshot_manifest(path)?;
            let files = manifest.get("files").and_then(Value::as_object);
            Ok(Some(snapshot_catalog_manifest_bytes(files)))
        }
    }
}

pub(crate) fn snapshot_prune_health(
    kind: SnapshotCatalogCandidateKind,
    path: &Path,
    require_index: bool,
) -> Result<Value> {
    match kind {
        SnapshotCatalogCandidateKind::Snapshot if require_index => snapshot_doctor(
            path,
            SnapshotDoctorOptions {
                top: SNAPSHOT_STATS_TOP_DEFAULT,
                verify: true,
                require_index: true,
            },
        ),
        SnapshotCatalogCandidateKind::Snapshot => verify_snapshot(path),
        SnapshotCatalogCandidateKind::Archive => {
            snapshot_verify_archive(path, SnapshotVerifyArchiveOptions { require_index })
        }
    }
}

pub(crate) fn snapshot_prune_delete_action(action: &SnapshotPruneAction) -> Result<()> {
    match action.kind {
        SnapshotCatalogCandidateKind::Snapshot => {
            let metadata = fs::symlink_metadata(&action.path)
                .into_diagnostic()
                .wrap_err_with(|| format!("reading {}", action.path.display()))?;
            if metadata.file_type().is_symlink()
                || !metadata.is_dir()
                || !action.path.join("manifest.json").is_file()
            {
                return err(format!(
                    "{} is no longer a snapshot directory",
                    action.path.display()
                ));
            }
            fs::remove_dir_all(&action.path)
                .into_diagnostic()
                .wrap_err_with(|| format!("deleting {}", action.path.display()))
        }
        SnapshotCatalogCandidateKind::Archive => {
            let metadata = fs::symlink_metadata(&action.path)
                .into_diagnostic()
                .wrap_err_with(|| format!("reading {}", action.path.display()))?;
            if metadata.file_type().is_symlink()
                || !metadata.is_file()
                || !snapshot_catalog_is_archive(&action.path)
            {
                return err(format!(
                    "{} is no longer a snapshot archive",
                    action.path.display()
                ));
            }
            fs::remove_file(&action.path)
                .into_diagnostic()
                .wrap_err_with(|| format!("deleting {}", action.path.display()))
        }
    }
}

pub(crate) fn snapshot_prune_action_value(
    action: SnapshotPruneAction,
    deleted: &BTreeSet<String>,
    dry_run: bool,
) -> Value {
    let path = action.path.display().to_string();
    let status = if action.delete {
        if dry_run {
            "would-delete"
        } else if deleted.contains(&path) {
            "deleted"
        } else {
            "delete-failed"
        }
    } else {
        "keep"
    };
    let mut value = json!({
        "kind": action.kind.as_str(),
        "path": path,
        "action": if action.delete { "delete" } else { "keep" },
        "status": status,
        "reasons": action.reasons,
        "rank_time_unix_ms": action.rank_time_unix_ms,
        "rank_time_source": action.rank_time_source,
        "bytes": action.bytes,
    });
    if let Value::Object(object) = &mut value {
        if let Some(health) = action.health {
            object.insert("health".to_string(), snapshot_prune_health_value(&health));
        }
        if let Some(error) = action.error {
            object.set("error", json!(error));
        }
    }
    value
}

pub(crate) fn snapshot_prune_health_value(report: &Value) -> Value {
    let mut value = Map::new();
    value.insert(
        "ok".to_string(),
        report.get("ok").cloned().unwrap_or(Value::Null),
    );
    if let Some(summary) = report.get("summary") {
        value.insert("summary".to_string(), summary.clone());
    }
    if let Some(failures) = report.get("failures") {
        value.insert("failures".to_string(), failures.clone());
    }
    let failing_checks = report
        .get("checks")
        .and_then(Value::as_array)
        .map(|checks| {
            checks
                .iter()
                .filter(|check| check.get("status").and_then(Value::as_str) == Some("fail"))
                .filter_map(|check| check.get("name").and_then(Value::as_str))
                .map(|name| Value::String(name.to_string()))
                .collect::<Vec<_>>()
        })
        .unwrap_or_default();
    if !failing_checks.is_empty() {
        value.set("failing_checks", Value::Array(failing_checks));
    }
    Value::Object(value)
}

pub(crate) fn snapshot_prune_criteria_value(options: &SnapshotPruneOptions) -> Value {
    json!({
        "keep_latest": options.keep_latest,
        "older_than_days": options.older_than_days,
        "failed": options.failed,
        "require_index": options.require_index,
    })
}

pub(crate) fn snapshot_prune_summary(
    actions: &[Value],
    delete_errors: &[Value],
    discovery_errors: &[Value],
) -> Value {
    let mut snapshots = 0_u64;
    let mut archives = 0_u64;
    let mut keep = 0_u64;
    let mut planned_delete = 0_u64;
    let mut deleted = 0_u64;
    let mut action_errors = 0_u64;
    let mut bytes = 0_u64;

    for action in actions {
        match action.get("kind").and_then(Value::as_str) {
            Some("snapshot") => snapshots += 1,
            Some("archive") => archives += 1,
            _ => {}
        }
        if action.get("action").and_then(Value::as_str) == Some("delete") {
            planned_delete += 1;
            bytes = bytes.saturating_add(
                action
                    .get("bytes")
                    .and_then(Value::as_u64)
                    .unwrap_or_default(),
            );
        } else {
            keep += 1;
        }
        if action.get("status").and_then(Value::as_str) == Some("deleted") {
            deleted += 1;
        }
        if action.get("error").is_some() {
            action_errors += 1;
        }
    }

    json!({
        "total": actions.len(),
        "snapshots": snapshots,
        "archives": archives,
        "keep": keep,
        "planned_delete": planned_delete,
        "deleted": deleted,
        "action_errors": action_errors,
        "delete_errors": delete_errors.len(),
        "discovery_errors": discovery_errors.len(),
        "planned_bytes": bytes,
    })
}

pub(crate) fn snapshot_history(options: SnapshotHistoryOptions) -> Result<Value> {
    if !options.root.exists() {
        return err(format!("{} does not exist", options.root.display()));
    }

    let (snapshots, discovery_errors) = snapshot_history_snapshots(&options)?;
    let ids = options
        .ids
        .iter()
        .map(ToString::to_string)
        .collect::<Vec<_>>();
    let mut states = ids
        .iter()
        .map(|id| {
            (
                id.clone(),
                SnapshotHistoryState {
                    seen: false,
                    present: false,
                    hash: None,
                    record: None,
                },
            )
        })
        .collect::<BTreeMap<_, _>>();
    let mut items = Vec::new();

    for snapshot in &snapshots {
        match snapshot_history_records(&snapshot.dir, &options) {
            Ok(records) => {
                for id in &ids {
                    let record = records.get(id);
                    let state = states.get_mut(id).expect("initialized above");
                    items.push(snapshot_history_observation(
                        id, snapshot, record, state, &options,
                    )?);
                }
            }
            Err(error) => {
                for id in &ids {
                    items.push(json!({
                        "id": id,
                        "section": options.section.label,
                        "snapshot": snapshot.dir.display().to_string(),
                        "created_at_unix_ms": snapshot.created_at_unix_ms,
                        "rank_time_unix_ms": snapshot.rank_time_unix_ms,
                        "rank_time_source": snapshot.rank_time_source,
                        "status": "error",
                        "error": error.to_string(),
                    }));
                }
            }
        }
    }

    let summary = snapshot_history_summary(&items, snapshots.len(), ids.len());
    let ok = summary
        .get("error")
        .and_then(Value::as_u64)
        .unwrap_or_default()
        == 0
        && discovery_errors.is_empty();
    Ok(json!({
        "root": options.root.display().to_string(),
        "recursive": options.recursive,
        "max_depth": options.max_depth,
        "limit": options.limit,
        "section": options.section.label,
        "ids": ids,
        "verified": options.verify,
        "index": options.index.as_str(),
        "details": options.details,
        "detail_limit": options.detail_limit,
        "records": options.records,
        "ok": ok,
        "summary": summary,
        "discovery_errors": discovery_errors,
        "items": items,
    }))
}

/// Shared discovery for `snapshot:history`, `snapshot:find`, and
/// `snapshot:timeline`: walk `root` for snapshot directories, rank them by
/// manifest time, and return them oldest-first alongside discovery errors.
pub(crate) fn snapshot_discover_ranked_snapshots(
    root: &Path,
    recursive: bool,
    max_depth: Option<usize>,
    limit: Option<usize>,
) -> Result<(Vec<SnapshotHistorySnapshot>, Vec<Value>)> {
    let catalog_options = SnapshotCatalogOptions {
        root: root.to_path_buf(),
        recursive,
        max_depth,
        limit,
        include_snapshots: true,
        include_archives: false,
        verify: false,
        doctor: false,
        require_index: false,
    };
    let mut candidates = Vec::new();
    let mut seen = BTreeSet::new();
    let mut discovery_errors = Vec::new();
    snapshot_catalog_discover(
        &catalog_options.root,
        0,
        &catalog_options,
        &mut seen,
        &mut candidates,
        &mut discovery_errors,
    )?;
    let mut snapshots = candidates
        .into_iter()
        .filter(|candidate| candidate.kind == SnapshotCatalogCandidateKind::Snapshot)
        .map(|candidate| snapshot_history_snapshot(candidate.path))
        .collect::<Result<Vec<_>>>()?;
    snapshots.sort_by(|left, right| {
        left.rank_time_unix_ms
            .unwrap_or_default()
            .cmp(&right.rank_time_unix_ms.unwrap_or_default())
            .then_with(|| left.dir.cmp(&right.dir))
    });
    Ok((snapshots, discovery_errors))
}

pub(crate) fn snapshot_history_snapshots(
    options: &SnapshotHistoryOptions,
) -> Result<(Vec<SnapshotHistorySnapshot>, Vec<Value>)> {
    snapshot_discover_ranked_snapshots(
        &options.root,
        options.recursive,
        options.max_depth,
        options.limit,
    )
}

pub(crate) fn snapshot_history_snapshot(dir: PathBuf) -> Result<SnapshotHistorySnapshot> {
    let manifest = read_snapshot_manifest(&dir)?;
    let created_at_unix_ms = manifest.get("created_at_unix_ms").and_then(Value::as_u64);
    if let Some(created_at_unix_ms) = created_at_unix_ms {
        return Ok(SnapshotHistorySnapshot {
            dir,
            created_at_unix_ms: Some(created_at_unix_ms),
            rank_time_unix_ms: Some(created_at_unix_ms),
            rank_time_source: "manifest.created_at_unix_ms",
        });
    }
    let metadata = fs::metadata(&dir)
        .into_diagnostic()
        .wrap_err_with(|| format!("reading {}", dir.display()))?;
    let modified = metadata
        .modified()
        .into_diagnostic()
        .wrap_err_with(|| format!("reading mtime for {}", dir.display()))?;
    Ok(SnapshotHistorySnapshot {
        dir,
        created_at_unix_ms: None,
        rank_time_unix_ms: Some(system_time_unix_ms(modified)?),
        rank_time_source: "filesystem.modified",
    })
}

pub(crate) fn snapshot_history_records(
    dir: &Path,
    options: &SnapshotHistoryOptions,
) -> Result<BTreeMap<String, Value>> {
    let query_options = SnapshotQueryOptions {
        section: options.section,
        ids: options.ids.clone(),
        contains: None,
        limit: None,
        verify: options.verify,
        index: options.index,
    };
    match options.section.kind {
        SnapshotQuerySectionKind::Jsonl => {
            prepare_snapshot_query(dir, &query_options)?;
            let mut records = BTreeMap::new();
            let path = snapshot_manifest_file_path(dir, options.section.file_label)?;
            snapshot_query_jsonl_each(dir, &query_options, |row| {
                if let Some(id) = record_id(&row) {
                    insert_snapshot_record(
                        &mut records,
                        id,
                        row,
                        &path,
                        "selected row".to_string(),
                    )?;
                }
                Ok(())
            })?;
            Ok(records)
        }
        SnapshotQuerySectionKind::JsonArray => {
            let rows = query_snapshot(dir, query_options)?;
            let path = snapshot_manifest_file_path(dir, options.section.file_label)?;
            let mut records = BTreeMap::new();
            for (index, row) in rows.as_array().into_iter().flatten().enumerate() {
                if let Some(id) = record_id(row) {
                    insert_snapshot_record(
                        &mut records,
                        id,
                        row.clone(),
                        &path,
                        format!("selected row {}", index + 1),
                    )?;
                }
            }
            Ok(records)
        }
    }
}

pub(crate) fn snapshot_history_observation(
    id: &str,
    snapshot: &SnapshotHistorySnapshot,
    record: Option<&Value>,
    state: &mut SnapshotHistoryState,
    options: &SnapshotHistoryOptions,
) -> Result<Value> {
    let current_hash = record.map(record_hash).transpose()?;
    let previous_hash = state.hash.clone();
    let status = match (
        state.seen,
        state.present,
        record.is_some(),
        &previous_hash,
        &current_hash,
    ) {
        (false, _, true, _, _) => "present",
        (false, _, false, _, _) => "missing",
        (true, true, false, _, _) => "removed",
        (true, false, true, _, _) => "added",
        (true, true, true, Some(previous), Some(current)) if previous == current => "unchanged",
        (true, true, true, _, _) => "changed",
        (true, false, false, _, _) => "missing",
    };
    let previous_record = state.record.clone();
    let mut value = json!({
        "id": id,
        "section": options.section.label,
        "snapshot": snapshot.dir.display().to_string(),
        "created_at_unix_ms": snapshot.created_at_unix_ms,
        "rank_time_unix_ms": snapshot.rank_time_unix_ms,
        "rank_time_source": snapshot.rank_time_source,
        "status": status,
        "hash": current_hash,
        "previous_hash": previous_hash,
    });
    if let Value::Object(object) = &mut value {
        if let Some(record) = record {
            object.insert(
                "summary".to_string(),
                snapshot_history_record_summary(options.section, record),
            );
            if options.records {
                object.insert("record".to_string(), record.clone());
            }
        }
        if options.details
            && status == "changed"
            && let (Some(previous), Some(current)) = (previous_record.as_ref(), record)
        {
            let mut changes = Vec::new();
            collect_value_changes(
                previous,
                current,
                "",
                &mut changes,
                options.detail_limit.saturating_add(1),
            );
            let changes_truncated = changes.len() > options.detail_limit;
            changes.truncate(options.detail_limit);
            object.set("changes", Value::Array(changes));
            object.set("changes_truncated", json!(changes_truncated));
        }
    }

    state.seen = true;
    state.present = record.is_some();
    state.hash = current_hash;
    state.record = record.cloned();
    Ok(value)
}

pub(crate) fn snapshot_history_record_summary(
    section: SnapshotQuerySection,
    record: &Value,
) -> Value {
    if matches!(section.label, "contacts" | "full_contacts") {
        return dedupe_contact_summary(record);
    }
    let mut summary = Map::new();
    if let Some(id) = record_id(record) {
        summary.set("id", id);
    }
    if let Some(name) =
        first_contact_string(record, &["name", "displayName", "display_name", "title"])
    {
        summary.set("name", name);
    }
    Value::Object(summary)
}

pub(crate) fn snapshot_history_summary(items: &[Value], snapshots: usize, ids: usize) -> Value {
    let mut counts = BTreeMap::<String, u64>::new();
    for item in items {
        let status = item
            .get("status")
            .and_then(Value::as_str)
            .unwrap_or("unknown")
            .to_string();
        *counts.entry(status).or_default() += 1;
    }
    let mut summary = Map::new();
    summary.set("snapshots", json!(snapshots));
    summary.set("ids", json!(ids));
    summary.set("observations", json!(items.len()));
    for status in [
        "present",
        "added",
        "removed",
        "changed",
        "unchanged",
        "missing",
        "error",
    ] {
        summary.insert(
            status.to_string(),
            json!(counts.remove(status).unwrap_or_default()),
        );
    }
    for (status, count) in counts {
        summary.insert(status, json!(count));
    }
    Value::Object(summary)
}

pub(crate) fn snapshot_find(options: SnapshotFindOptions) -> Result<Value> {
    if !options.root.exists() {
        return err(format!("{} does not exist", options.root.display()));
    }

    let (snapshots, discovery_errors) = snapshot_find_snapshots(&options)?;
    let mut items = Vec::new();
    let mut errors = Vec::new();
    let mut searched_sections = 0_u64;
    let mut skipped_sections = 0_u64;

    'snapshots: for snapshot in &snapshots {
        if options.verify {
            match verify_snapshot(&snapshot.dir) {
                Ok(verify) if verify.get("ok").and_then(Value::as_bool).unwrap_or(false) => {}
                Ok(_) => {
                    errors.push(snapshot_find_error(
                        snapshot,
                        None,
                        "snapshot failed manifest verification",
                    ));
                    continue;
                }
                Err(error) => {
                    errors.push(snapshot_find_error(snapshot, None, error.to_string()));
                    continue;
                }
            }
        }

        for section in &options.sections {
            if options.limit.is_some_and(|limit| items.len() >= limit) {
                break 'snapshots;
            }
            let has_section = match snapshot_manifest_has_file(&snapshot.dir, section.file_label) {
                Ok(value) => value,
                Err(error) => {
                    errors.push(snapshot_find_error(
                        snapshot,
                        Some(*section),
                        error.to_string(),
                    ));
                    continue;
                }
            };
            if !has_section {
                skipped_sections += 1;
                continue;
            }
            searched_sections += 1;
            let remaining = options
                .limit
                .map(|limit| limit.saturating_sub(items.len()))
                .filter(|remaining| *remaining > 0);
            let query_options = SnapshotQueryOptions {
                section: *section,
                ids: options.ids.clone(),
                contains: options.contains.clone(),
                limit: remaining,
                verify: false,
                index: options.index,
            };
            match snapshot_find_section(snapshot, &query_options, &options) {
                Ok(mut matches) => items.append(&mut matches),
                Err(error) => errors.push(snapshot_find_error(
                    snapshot,
                    Some(*section),
                    error.to_string(),
                )),
            }
        }
    }

    let summary = snapshot_find_summary(
        &items,
        snapshots.len(),
        searched_sections,
        skipped_sections,
        errors.len(),
        discovery_errors.len(),
    );
    let section_labels = options
        .sections
        .iter()
        .map(|section| section.label)
        .collect::<Vec<_>>();
    let ids = options
        .ids
        .iter()
        .map(ToString::to_string)
        .collect::<Vec<_>>();
    let ok = errors.is_empty() && discovery_errors.is_empty();
    Ok(json!({
        "root": options.root.display().to_string(),
        "recursive": options.recursive,
        "max_depth": options.max_depth,
        "snapshot_limit": options.snapshot_limit,
        "sections": section_labels,
        "ids": ids,
        "contains": options.contains,
        "limit": options.limit,
        "verified": options.verify,
        "index": options.index.as_str(),
        "records": options.records,
        "ok": ok,
        "summary": summary,
        "discovery_errors": discovery_errors,
        "errors": errors,
        "items": items,
    }))
}

pub(crate) fn snapshot_find_snapshots(
    options: &SnapshotFindOptions,
) -> Result<(Vec<SnapshotHistorySnapshot>, Vec<Value>)> {
    snapshot_discover_ranked_snapshots(
        &options.root,
        options.recursive,
        options.max_depth,
        options.snapshot_limit,
    )
}

pub(crate) fn snapshot_find_section(
    snapshot: &SnapshotHistorySnapshot,
    query_options: &SnapshotQueryOptions,
    options: &SnapshotFindOptions,
) -> Result<Vec<Value>> {
    let mut matches = Vec::new();
    match query_options.section.kind {
        SnapshotQuerySectionKind::Jsonl => {
            snapshot_query_jsonl_each(&snapshot.dir, query_options, |row| {
                matches.push(snapshot_find_item(
                    snapshot,
                    query_options.section,
                    &row,
                    options,
                )?);
                Ok(())
            })?;
        }
        SnapshotQuerySectionKind::JsonArray => {
            let rows = read_snapshot_section_values(&snapshot.dir, query_options.section)?;
            for row in filter_snapshot_query_rows(rows, query_options)? {
                matches.push(snapshot_find_item(
                    snapshot,
                    query_options.section,
                    &row,
                    options,
                )?);
            }
        }
    }
    Ok(matches)
}

pub(crate) fn snapshot_find_item(
    snapshot: &SnapshotHistorySnapshot,
    section: SnapshotQuerySection,
    record: &Value,
    options: &SnapshotFindOptions,
) -> Result<Value> {
    let mut item = Map::new();
    item.insert(
        "snapshot".to_string(),
        Value::String(snapshot.dir.display().to_string()),
    );
    item.insert(
        "section".to_string(),
        Value::String(section.label.to_string()),
    );
    item.insert(
        "created_at_unix_ms".to_string(),
        snapshot
            .created_at_unix_ms
            .map(|value| Value::Number(Number::from(value)))
            .unwrap_or(Value::Null),
    );
    item.insert(
        "rank_time_unix_ms".to_string(),
        snapshot
            .rank_time_unix_ms
            .map(|value| Value::Number(Number::from(value)))
            .unwrap_or(Value::Null),
    );
    item.insert(
        "rank_time_source".to_string(),
        Value::String(snapshot.rank_time_source.to_string()),
    );
    item.insert(
        "id".to_string(),
        record_id(record).map(Value::String).unwrap_or(Value::Null),
    );
    item.set("hash", record_hash(record)?);
    item.insert(
        "summary".to_string(),
        snapshot_history_record_summary(section, record),
    );
    if let Some(contains) = &options.contains {
        item.insert(
            "field_matches".to_string(),
            Value::Array(snapshot_find_field_matches(record, contains, 5)),
        );
    }
    if options.records {
        item.insert("record".to_string(), record.clone());
    }
    Ok(Value::Object(item))
}

pub(crate) fn snapshot_find_field_matches(
    record: &Value,
    contains: &str,
    limit: usize,
) -> Vec<Value> {
    let needle = contains.to_lowercase();
    let mut matches = Vec::new();
    collect_snapshot_find_field_matches(record, "", &needle, &mut matches, limit);
    matches
}

pub(crate) fn collect_snapshot_find_field_matches(
    value: &Value,
    path: &str,
    needle: &str,
    matches: &mut Vec<Value>,
    limit: usize,
) {
    if matches.len() >= limit {
        return;
    }
    match value {
        Value::Object(object) => {
            for (key, child) in object {
                if matches.len() >= limit {
                    break;
                }
                collect_snapshot_find_field_matches(
                    child,
                    &json_pointer_child(path, key),
                    needle,
                    matches,
                    limit,
                );
            }
        }
        Value::Array(items) => {
            for (index, child) in items.iter().enumerate() {
                if matches.len() >= limit {
                    break;
                }
                collect_snapshot_find_field_matches(
                    child,
                    &json_pointer_child(path, &index.to_string()),
                    needle,
                    matches,
                    limit,
                );
            }
        }
        Value::Null => {}
        Value::String(text) => {
            if text.to_lowercase().contains(needle) {
                matches.push(json!({
                    "path": if path.is_empty() { "/" } else { path },
                    "value": diff_preview_value(value),
                }));
            }
        }
        Value::Bool(_) | Value::Number(_) => {
            let text = cell_string(value);
            if text.to_lowercase().contains(needle) {
                matches.push(json!({
                    "path": if path.is_empty() { "/" } else { path },
                    "value": diff_preview_value(value),
                }));
            }
        }
    }
}

pub(crate) fn snapshot_find_error(
    snapshot: &SnapshotHistorySnapshot,
    section: Option<SnapshotQuerySection>,
    error: impl ToString,
) -> Value {
    json!({
        "snapshot": snapshot.dir.display().to_string(),
        "created_at_unix_ms": snapshot.created_at_unix_ms,
        "rank_time_unix_ms": snapshot.rank_time_unix_ms,
        "rank_time_source": snapshot.rank_time_source,
        "section": section.map(|section| section.label),
        "error": error.to_string(),
    })
}

pub(crate) fn snapshot_find_summary(
    items: &[Value],
    snapshots: usize,
    searched_sections: u64,
    skipped_sections: u64,
    errors: usize,
    discovery_errors: usize,
) -> Value {
    let mut snapshots_with_matches = BTreeSet::new();
    let mut section_counts = BTreeMap::<String, u64>::new();
    for item in items {
        if let Some(snapshot) = item.get("snapshot").and_then(Value::as_str) {
            snapshots_with_matches.insert(snapshot.to_string());
        }
        if let Some(section) = item.get("section").and_then(Value::as_str) {
            *section_counts.entry(section.to_string()).or_default() += 1;
        }
    }
    json!({
        "snapshots": snapshots,
        "snapshots_with_matches": snapshots_with_matches.len(),
        "searched_sections": searched_sections,
        "skipped_sections": skipped_sections,
        "matches": items.len(),
        "errors": errors,
        "discovery_errors": discovery_errors,
        "sections": section_counts,
    })
}

pub(crate) fn snapshot_timeline(options: SnapshotTimelineOptions) -> Result<Value> {
    if !options.root.exists() {
        return err(format!("{} does not exist", options.root.display()));
    }

    let (snapshots, discovery_errors) = snapshot_timeline_snapshots(&options)?;
    let total_pairs = snapshots.len().saturating_sub(1);
    let mut items = Vec::new();

    for (index, pair) in snapshots.windows(2).enumerate() {
        let old = &pair[0];
        let new = &pair[1];
        let item = snapshot_timeline_pair(index, old, new, &options)?;
        let changed = item
            .get("changed")
            .and_then(Value::as_bool)
            .unwrap_or(false);
        let ok = item.get("ok").and_then(Value::as_bool).unwrap_or(false);
        if options.changes_only && ok && !changed {
            continue;
        }
        items.push(item);
    }

    let summary = snapshot_timeline_summary(&items, snapshots.len(), total_pairs);
    let ok = summary
        .get("failed_pairs")
        .and_then(Value::as_u64)
        .unwrap_or_default()
        == 0
        && discovery_errors.is_empty();

    Ok(json!({
        "root": options.root.display().to_string(),
        "recursive": options.recursive,
        "max_depth": options.max_depth,
        "limit": options.limit,
        "changes_only": options.changes_only,
        "diffs": options.diffs,
        "details": options.diff.details,
        "detail_limit": options.diff.detail_limit,
        "ok": ok,
        "summary": summary,
        "discovery_errors": discovery_errors,
        "items": items,
    }))
}

pub(crate) fn snapshot_timeline_snapshots(
    options: &SnapshotTimelineOptions,
) -> Result<(Vec<SnapshotHistorySnapshot>, Vec<Value>)> {
    snapshot_discover_ranked_snapshots(
        &options.root,
        options.recursive,
        options.max_depth,
        options.limit,
    )
}

pub(crate) fn snapshot_timeline_pair(
    index: usize,
    old: &SnapshotHistorySnapshot,
    new: &SnapshotHistorySnapshot,
    options: &SnapshotTimelineOptions,
) -> Result<Value> {
    let diff = diff_snapshots(&old.dir, &new.dir, options.diff)?;
    let ok = diff.get("ok").and_then(Value::as_bool).unwrap_or(false);
    let summary = if ok {
        snapshot_timeline_diff_summary(&diff)
    } else {
        Value::Null
    };
    let changed = summary
        .get("changed")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    let mut item = json!({
        "pair": index,
        "ok": ok,
        "changed": changed,
        "old": old.dir.display().to_string(),
        "new": new.dir.display().to_string(),
        "old_created_at_unix_ms": old.created_at_unix_ms,
        "new_created_at_unix_ms": new.created_at_unix_ms,
        "old_rank_time_unix_ms": old.rank_time_unix_ms,
        "new_rank_time_unix_ms": new.rank_time_unix_ms,
        "old_rank_time_source": old.rank_time_source,
        "new_rank_time_source": new.rank_time_source,
        "summary": summary,
    });
    if let Value::Object(object) = &mut item {
        if !ok && let Some(error) = diff.get("error") {
            object.insert("error".to_string(), error.clone());
        }
        if options.diffs {
            object.insert("diff".to_string(), diff);
        }
    }
    Ok(item)
}

pub(crate) fn snapshot_timeline_diff_summary(diff: &Value) -> Value {
    let contacts_added = diff_section_count(diff, "contacts", "added_count");
    let contacts_removed = diff_section_count(diff, "contacts", "removed_count");
    let contacts_changed = diff_section_count(diff, "contacts", "changed_count");
    let groups_added = diff_section_count(diff, "groups", "added_count");
    let groups_removed = diff_section_count(diff, "groups", "removed_count");
    let groups_changed = diff_section_count(diff, "groups", "changed_count");
    let full_contacts_added = diff_section_count(diff, "full_contacts", "added_count");
    let full_contacts_removed = diff_section_count(diff, "full_contacts", "removed_count");
    let full_contacts_changed = diff_section_count(diff, "full_contacts", "changed_count");
    let moments_changed = snapshot_timeline_moments_changed(diff);
    let added = contacts_added + groups_added + full_contacts_added;
    let removed = contacts_removed + groups_removed + full_contacts_removed;
    let changed_records = contacts_changed + groups_changed + full_contacts_changed;
    json!({
        "changed": added > 0 || removed > 0 || changed_records > 0 || moments_changed > 0,
        "added": added,
        "removed": removed,
        "changed_records": changed_records,
        "contacts_added": contacts_added,
        "contacts_removed": contacts_removed,
        "contacts_changed": contacts_changed,
        "groups_added": groups_added,
        "groups_removed": groups_removed,
        "groups_changed": groups_changed,
        "full_contacts_added": full_contacts_added,
        "full_contacts_removed": full_contacts_removed,
        "full_contacts_changed": full_contacts_changed,
        "moments_changed": moments_changed,
    })
}

pub(crate) fn snapshot_timeline_summary(
    items: &[Value],
    snapshots: usize,
    total_pairs: usize,
) -> Value {
    let emitted_pairs = items.len();
    let failed_pairs = items
        .iter()
        .filter(|item| !item.get("ok").and_then(Value::as_bool).unwrap_or(false))
        .count();
    let changed_pairs = items
        .iter()
        .filter(|item| {
            item.get("changed")
                .and_then(Value::as_bool)
                .unwrap_or(false)
        })
        .count();
    let emitted_unchanged_pairs = emitted_pairs
        .saturating_sub(failed_pairs)
        .saturating_sub(changed_pairs);
    let mut added = 0_u64;
    let mut removed = 0_u64;
    let mut changed_records = 0_u64;
    let mut moments_changed = 0_u64;
    for item in items {
        let Some(summary) = item.get("summary") else {
            continue;
        };
        added += summary
            .get("added")
            .and_then(Value::as_u64)
            .unwrap_or_default();
        removed += summary
            .get("removed")
            .and_then(Value::as_u64)
            .unwrap_or_default();
        changed_records += summary
            .get("changed_records")
            .and_then(Value::as_u64)
            .unwrap_or_default();
        moments_changed += summary
            .get("moments_changed")
            .and_then(Value::as_u64)
            .unwrap_or_default();
    }
    json!({
        "snapshots": snapshots,
        "total_pairs": total_pairs,
        "emitted_pairs": emitted_pairs,
        "changed_pairs": changed_pairs,
        "unchanged_pairs": total_pairs.saturating_sub(changed_pairs).saturating_sub(failed_pairs),
        "emitted_unchanged_pairs": emitted_unchanged_pairs,
        "failed_pairs": failed_pairs,
        "added": added,
        "removed": removed,
        "changed_records": changed_records,
        "moments_changed": moments_changed,
    })
}

pub(crate) fn snapshot_doctor(dir: &Path, options: SnapshotDoctorOptions) -> Result<Value> {
    let mut checks = Vec::new();
    let mut verified = false;
    let mut verify = Value::Null;

    if options.verify {
        match verify_snapshot(dir) {
            Ok(value) => {
                let ok = value.get("ok").and_then(Value::as_bool).unwrap_or(false);
                verified = ok;
                let checked = value
                    .get("checked")
                    .and_then(Value::as_array)
                    .map(Vec::len)
                    .unwrap_or_default();
                let failures = value.get("failures").cloned().unwrap_or_else(|| json!([]));
                checks.push(snapshot_doctor_check(
                    if ok { "pass" } else { "fail" },
                    "manifest_hashes",
                    if ok {
                        "all manifest file hashes match"
                    } else {
                        "one or more manifest file hashes do not match"
                    },
                    json!({
                        "checked": checked,
                        "failures": failures,
                    }),
                ));
                verify = value;
            }
            Err(error) => checks.push(snapshot_doctor_check(
                "fail",
                "manifest_hashes",
                "manifest verification could not complete",
                json!({
                    "error": error.to_string(),
                }),
            )),
        }
    } else {
        checks.push(snapshot_doctor_check(
            "skipped",
            "manifest_hashes",
            "manifest verification was skipped",
            json!({}),
        ));
    }

    let stats = match snapshot_stats(
        dir,
        SnapshotStatsOptions {
            top: options.top,
            verify: false,
        },
    ) {
        Ok(value) => {
            checks.push(snapshot_doctor_check(
                "pass",
                "stats:read",
                "snapshot records were parsed",
                json!({}),
            ));
            snapshot_doctor_add_stats_checks(&mut checks, &value);
            snapshot_doctor_add_index_checks(dir, &mut checks, &value, options.require_index);
            value
        }
        Err(error) => {
            checks.push(snapshot_doctor_check(
                "fail",
                "stats:read",
                "snapshot records could not be parsed",
                json!({
                    "error": error.to_string(),
                }),
            ));
            Value::Null
        }
    };

    let summary = snapshot_doctor_summary(&checks);
    let failures = summary
        .get("failures")
        .and_then(Value::as_u64)
        .unwrap_or_default();
    Ok(json!({
        "dir": dir.display().to_string(),
        "ok": failures == 0,
        "verified": verified,
        "require_index": options.require_index,
        "top": options.top,
        "summary": summary,
        "checks": checks,
        "verify": verify,
        "stats": stats,
    }))
}

pub(crate) fn snapshot_doctor_check(
    status: &str,
    name: impl Into<String>,
    message: impl Into<String>,
    details: Value,
) -> Value {
    json!({
        "name": name.into(),
        "status": status,
        "message": message.into(),
        "details": details,
    })
}

pub(crate) fn snapshot_doctor_summary(checks: &[Value]) -> Value {
    let mut pass = 0_u64;
    let mut warn = 0_u64;
    let mut fail = 0_u64;
    let mut skipped = 0_u64;
    for check in checks {
        match check.get("status").and_then(Value::as_str) {
            Some("pass") => pass += 1,
            Some("warn") => warn += 1,
            Some("fail") => fail += 1,
            Some("skipped") => skipped += 1,
            _ => warn += 1,
        }
    }
    json!({
        "total": checks.len(),
        "pass": pass,
        "warn": warn,
        "fail": fail,
        "skipped": skipped,
        "failures": fail,
        "warnings": warn,
    })
}

pub(crate) fn snapshot_doctor_add_stats_checks(checks: &mut Vec<Value>, stats: &Value) {
    let Some(sections) = stats.get("sections").and_then(Value::as_object) else {
        checks.push(snapshot_doctor_check(
            "fail",
            "stats:sections",
            "snapshot stats did not contain sections",
            json!({}),
        ));
        return;
    };

    for (label, section_stats) in sections {
        snapshot_doctor_add_count_check(checks, stats, label, section_stats);
        snapshot_doctor_add_id_check(checks, label, section_stats);
    }
}

pub(crate) fn snapshot_doctor_add_count_check(
    checks: &mut Vec<Value>,
    stats: &Value,
    label: &str,
    section_stats: &Value,
) {
    let actual = section_stats.get("rows").and_then(Value::as_u64);
    let expected = snapshot_doctor_manifest_count(stats, label);
    let (status, message) = match (expected, actual) {
        (Some(expected), Some(actual)) if expected == actual => {
            ("pass", "section row count matches manifest count")
        }
        (Some(_), Some(_)) => ("fail", "section row count does not match manifest count"),
        (None, Some(_)) => ("warn", "manifest does not contain a count for this section"),
        (_, None) => ("fail", "section stats did not contain a row count"),
    };
    checks.push(snapshot_doctor_check(
        status,
        format!("count:{label}"),
        message,
        json!({
            "section": label,
            "expected": expected,
            "actual": actual,
        }),
    ));
}

pub(crate) fn snapshot_doctor_add_id_check(
    checks: &mut Vec<Value>,
    label: &str,
    section_stats: &Value,
) {
    let Some(ids) = section_stats.get("ids").and_then(Value::as_object) else {
        checks.push(snapshot_doctor_check(
            "fail",
            format!("ids:{label}"),
            "section stats did not contain ID metrics",
            json!({
                "section": label,
            }),
        ));
        return;
    };

    let rows = section_stats
        .get("rows")
        .and_then(Value::as_u64)
        .unwrap_or_default();
    let rows_with_id = ids
        .get("rows_with_id")
        .and_then(Value::as_u64)
        .unwrap_or_default();
    let missing_id = ids
        .get("missing_id")
        .and_then(Value::as_u64)
        .unwrap_or_default();
    let unique_ids = ids
        .get("unique_ids")
        .and_then(Value::as_u64)
        .unwrap_or_default();
    let duplicate_id_count = ids
        .get("duplicate_id_count")
        .and_then(Value::as_u64)
        .unwrap_or_default();
    let strict = snapshot_doctor_requires_top_level_ids(label);
    let has_id_problem = missing_id > 0 || duplicate_id_count > 0;
    let status = if has_id_problem && strict {
        "fail"
    } else if has_id_problem {
        "warn"
    } else {
        "pass"
    };
    let message = if !has_id_problem {
        "top-level ID coverage is consistent"
    } else if strict {
        "required top-level IDs are missing or duplicated"
    } else {
        "moment rows do not always expose stable top-level IDs"
    };

    checks.push(snapshot_doctor_check(
        status,
        format!("ids:{label}"),
        message,
        json!({
            "section": label,
            "strict": strict,
            "rows": rows,
            "rows_with_id": rows_with_id,
            "missing_id": missing_id,
            "unique_ids": unique_ids,
            "duplicate_id_count": duplicate_id_count,
            "duplicate_ids": ids.get("duplicate_ids").cloned().unwrap_or_else(|| json!([])),
        }),
    ));
}

pub(crate) fn snapshot_doctor_add_index_checks(
    dir: &Path,
    checks: &mut Vec<Value>,
    stats: &Value,
    require_index: bool,
) {
    let Some(sections) = stats.get("sections").and_then(Value::as_object) else {
        return;
    };

    for (label, section_stats) in sections {
        let Ok(section) = snapshot_query_section(label) else {
            continue;
        };
        if section.kind != SnapshotQuerySectionKind::Jsonl {
            continue;
        }
        snapshot_doctor_add_index_check(dir, checks, section, section_stats, require_index);
    }
}

pub(crate) fn snapshot_doctor_add_index_check(
    dir: &Path,
    checks: &mut Vec<Value>,
    section: SnapshotQuerySection,
    section_stats: &Value,
    require_index: bool,
) {
    let rows = section_stats
        .get("rows")
        .and_then(Value::as_u64)
        .unwrap_or_default();
    let rows_with_id = snapshot_doctor_id_metric(section_stats, "rows_with_id").unwrap_or_default();
    let missing_id = snapshot_doctor_id_metric(section_stats, "missing_id").unwrap_or_default();
    let index_path = snapshot_index_path(dir, section);
    let index_path_text = index_path.display().to_string();

    if rows == 0 {
        checks.push(snapshot_doctor_check(
            "pass",
            format!("index:{}", section.label),
            "empty JSONL section does not need an index",
            json!({
                "section": section.label,
                "file": section.file_name,
                "index_path": index_path_text,
                "rows": rows,
            }),
        ));
        return;
    }

    let source = match snapshot_index_source_file(dir, section) {
        Ok(source) => source,
        Err(error) => {
            checks.push(snapshot_doctor_check(
                if require_index { "fail" } else { "warn" },
                format!("index:{}", section.label),
                "index source file could not be read from manifest",
                json!({
                    "section": section.label,
                    "file": section.file_name,
                    "index_path": index_path_text,
                    "error": error.to_string(),
                }),
            ));
            return;
        }
    };

    match read_snapshot_index_if_present(&index_path) {
        Ok(Some(index)) if snapshot_index_matches_source(&index, section, &source) => {
            let counts_match = index.record_count == rows
                && index.indexed_count == rows_with_id
                && index.skipped_without_id == missing_id;
            checks.push(snapshot_doctor_check(
                if counts_match { "pass" } else { "fail" },
                format!("index:{}", section.label),
                if counts_match {
                    "index is fresh for this snapshot file"
                } else {
                    "index is fresh by fingerprint but its counts do not match parsed stats"
                },
                json!({
                    "section": section.label,
                    "file": section.file_name,
                    "index_path": index_path_text,
                    "record_count": index.record_count,
                    "indexed_count": index.indexed_count,
                    "skipped_without_id": index.skipped_without_id,
                    "expected_record_count": rows,
                    "expected_indexed_count": rows_with_id,
                    "expected_skipped_without_id": missing_id,
                }),
            ));
        }
        Ok(Some(index)) => checks.push(snapshot_doctor_check(
            if require_index { "fail" } else { "warn" },
            format!("index:{}", section.label),
            "index exists but is stale or belongs to a different section",
            json!({
                "section": section.label,
                "file": section.file_name,
                "index_path": index_path_text,
                "expected_file": source,
                "indexed_section": index.section,
                "indexed_file": index.file,
            }),
        )),
        Ok(None) => checks.push(snapshot_doctor_check(
            if require_index { "fail" } else { "warn" },
            format!("index:{}", section.label),
            "JSONL section does not have an index",
            json!({
                "section": section.label,
                "file": section.file_name,
                "index_path": index_path_text,
                "rows": rows,
            }),
        )),
        Err(error) => checks.push(snapshot_doctor_check(
            if require_index { "fail" } else { "warn" },
            format!("index:{}", section.label),
            "index could not be read",
            json!({
                "section": section.label,
                "file": section.file_name,
                "index_path": index_path_text,
                "error": error.to_string(),
            }),
        )),
    }
}

pub(crate) fn snapshot_doctor_manifest_count(stats: &Value, label: &str) -> Option<u64> {
    let counts = stats.get("manifest")?.get("counts")?;
    counts
        .get(label)
        .and_then(Value::as_u64)
        .or_else(|| counts.get("moments")?.get(label)?.as_u64())
}

pub(crate) fn snapshot_doctor_id_metric(section_stats: &Value, key: &str) -> Option<u64> {
    section_stats.get("ids")?.get(key).and_then(Value::as_u64)
}

pub(crate) fn snapshot_doctor_requires_top_level_ids(label: &str) -> bool {
    matches!(label, "contacts" | "full_contacts" | "groups")
}

pub(crate) fn snapshot_stats(dir: &Path, options: SnapshotStatsOptions) -> Result<Value> {
    if options.verify {
        let verify = verify_snapshot(dir)?;
        if !verify.get("ok").and_then(Value::as_bool).unwrap_or(false) {
            return err("snapshot failed manifest verification");
        }
    }
    let manifest = read_snapshot_manifest(dir)?;
    let Some(files) = manifest.get("files").and_then(Value::as_object) else {
        return err("snapshot manifest does not contain files object");
    };

    let mut sections = Map::new();
    let mut non_record_files = Vec::new();
    for label in snapshot_stats_section_order(files) {
        let Some(entry) = files.get(&label) else {
            continue;
        };
        match snapshot_query_section(&label) {
            Ok(section) => {
                let stats = match section.kind {
                    SnapshotQuerySectionKind::Jsonl => {
                        snapshot_jsonl_section_stats(dir, section, entry, options.top)?
                    }
                    SnapshotQuerySectionKind::JsonArray => {
                        snapshot_array_section_stats(dir, section, entry, options.top)?
                    }
                };
                sections.insert(label, stats);
            }
            Err(_) => {
                non_record_files.push(json!({
                    "label": label,
                    "file": snapshot_file_fingerprint(entry)?,
                }));
            }
        }
    }

    Ok(json!({
        "ok": true,
        "dir": dir.display().to_string(),
        "verified": options.verify,
        "top": options.top,
        "manifest": {
            "schema": manifest.get("schema").cloned().unwrap_or(Value::Null),
            "meshx_version": manifest.get("meshx_version").cloned().unwrap_or(Value::Null),
            "created_at_unix_ms": manifest.get("created_at_unix_ms").cloned().unwrap_or(Value::Null),
            "counts": manifest.get("counts").cloned().unwrap_or(Value::Null),
            "file_count": files.len(),
        },
        "sections": sections,
        "non_record_files": non_record_files,
    }))
}

pub(crate) fn snapshot_stats_section_order(files: &Map<String, Value>) -> Vec<String> {
    let mut labels = Vec::new();
    for label in [
        "contacts",
        "full_contacts",
        "groups",
        "notes",
        "events",
        "emails",
        "events_upcoming",
        "emails_recent",
        "reminders_recent",
        "reminders_upcoming",
    ] {
        if files.contains_key(label) {
            labels.push(label.to_string());
        }
    }
    for label in files.keys() {
        if !labels.iter().any(|value| value == label) {
            labels.push(label.clone());
        }
    }
    labels
}

pub(crate) fn snapshot_jsonl_section_stats(
    dir: &Path,
    section: SnapshotQuerySection,
    entry: &Value,
    top: usize,
) -> Result<Value> {
    let path = snapshot_manifest_file_path(dir, section.file_label)?;
    let file = fs::File::open(&path)
        .into_diagnostic()
        .wrap_err_with(|| format!("opening {}", path.display()))?;
    let reader = StdBufReader::new(file);
    let mut stats = SnapshotStatsAccumulator::default();
    for (index, line) in reader.lines().enumerate() {
        let line = line
            .into_diagnostic()
            .wrap_err_with(|| format!("reading {} line {}", path.display(), index + 1))?;
        if line.trim().is_empty() {
            continue;
        }
        let row = serde_json::from_str::<Value>(&line)
            .into_diagnostic()
            .wrap_err_with(|| format!("parsing {} line {}", path.display(), index + 1))?;
        stats.add_row(&row);
    }
    stats.finish(section, "jsonl", entry, top)
}

pub(crate) fn snapshot_array_section_stats(
    dir: &Path,
    section: SnapshotQuerySection,
    entry: &Value,
    top: usize,
) -> Result<Value> {
    let mut stats = SnapshotStatsAccumulator::default();
    let path = snapshot_manifest_file_path(dir, section.file_label)?;
    for row in read_snapshot_array_values_at_path(&path)? {
        stats.add_row(&row);
    }
    stats.finish(section, "json-array", entry, top)
}

#[cfg(test)]
mod tests {
    use super::*;

    fn prune_action(
        kind: SnapshotCatalogCandidateKind,
        path: &str,
        rank_time_unix_ms: u64,
        bytes: u64,
    ) -> SnapshotPruneAction {
        SnapshotPruneAction {
            kind,
            path: PathBuf::from(path),
            delete: false,
            reasons: Vec::new(),
            rank_time_unix_ms: Some(rank_time_unix_ms),
            rank_time_source: "test",
            bytes: Some(bytes),
            health: None,
            error: None,
        }
    }

    #[test]
    fn snapshot_prune_apply_keep_latest_prunes_older_items_per_kind() {
        let mut actions = vec![
            prune_action(SnapshotCatalogCandidateKind::Snapshot, "snap-old", 10, 100),
            prune_action(
                SnapshotCatalogCandidateKind::Archive,
                "archive-new.tar.zst",
                40,
                400,
            ),
            prune_action(SnapshotCatalogCandidateKind::Snapshot, "snap-new", 20, 200),
            prune_action(
                SnapshotCatalogCandidateKind::Archive,
                "archive-old.tar.zst",
                30,
                300,
            ),
        ];

        snapshot_prune_apply_keep_latest(&mut actions, Some(1));

        assert!(actions[0].delete);
        assert_eq!(actions[0].reasons, vec!["keep-latest"]);
        assert!(!actions[1].delete);
        assert!(!actions[2].delete);
        assert!(actions[3].delete);
        assert_eq!(actions[3].reasons, vec!["keep-latest"]);
    }

    #[test]
    fn snapshot_prune_action_value_reports_dry_run_delete() {
        let mut action = prune_action(SnapshotCatalogCandidateKind::Snapshot, "snap-old", 10, 100);
        action.delete = true;
        action.reasons.push("keep-latest".to_string());

        let value = snapshot_prune_action_value(action, &BTreeSet::new(), true);

        assert_eq!(value.get("action"), Some(&json!("delete")));
        assert_eq!(value.get("status"), Some(&json!("would-delete")));
        assert_eq!(value.get("reasons"), Some(&json!(["keep-latest"])));
    }

    #[test]
    fn snapshot_prune_summary_counts_planned_deletes_and_bytes() {
        let actions = vec![
            json!({
                "kind": "snapshot",
                "action": "delete",
                "status": "would-delete",
                "bytes": 100,
            }),
            json!({
                "kind": "archive",
                "action": "keep",
                "status": "keep",
                "bytes": 300,
            }),
            json!({
                "kind": "archive",
                "action": "delete",
                "status": "deleted",
                "bytes": 400,
            }),
        ];

        let summary = snapshot_prune_summary(&actions, &[json!({"path": "broken"})], &[]);

        assert_eq!(summary.get("total"), Some(&json!(3)));
        assert_eq!(summary.get("snapshots"), Some(&json!(1)));
        assert_eq!(summary.get("archives"), Some(&json!(2)));
        assert_eq!(summary.get("keep"), Some(&json!(1)));
        assert_eq!(summary.get("planned_delete"), Some(&json!(2)));
        assert_eq!(summary.get("deleted"), Some(&json!(1)));
        assert_eq!(summary.get("action_errors"), Some(&json!(0)));
        assert_eq!(summary.get("delete_errors"), Some(&json!(1)));
        assert_eq!(summary.get("planned_bytes"), Some(&json!(500)));
    }

    #[test]
    fn snapshot_report_component_requires_explicit_ok() {
        let (_, ok) = snapshot_report_component(Ok(json!({"items": []})));

        assert!(!ok);
    }

    #[test]
    fn snapshot_stats_reports_explicit_ok() -> Result<()> {
        let dir = std::env::temp_dir().join(format!(
            "meshx-stats-ok-{}-{}",
            std::process::id(),
            now_millis()
        ));
        fs::create_dir_all(&dir).into_diagnostic()?;
        fs::write(dir.join("manifest.json"), r#"{"files":{}}"#).into_diagnostic()?;

        let value = snapshot_stats(
            &dir,
            SnapshotStatsOptions {
                top: 5,
                verify: false,
            },
        )?;

        fs::remove_dir_all(&dir).ok();
        assert_eq!(value.get("ok"), Some(&json!(true)));
        Ok(())
    }

    #[test]
    fn snapshot_stats_reads_sections_from_manifest_paths() -> Result<()> {
        let dir = std::env::temp_dir().join(format!(
            "meshx-stats-manifest-path-{}-{}",
            std::process::id(),
            now_millis()
        ));
        fs::create_dir_all(dir.join("data")).into_diagnostic()?;
        let contacts = "{\"id\":1,\"name\":\"Ada\"}\n";
        let groups = r#"[{"id":10,"name":"Operators"}]"#;
        fs::write(dir.join("data/contacts.jsonl"), contacts).into_diagnostic()?;
        fs::write(dir.join("data/groups.json"), groups).into_diagnostic()?;
        fs::write(
            dir.join("manifest.json"),
            serde_json::to_string(&json!({
                "files": {
                    "contacts": {
                        "path": "data/contacts.jsonl",
                        "bytes": contacts.len() as u64,
                        "sha256": sha256_hex(contacts.as_bytes()),
                    },
                    "groups": {
                        "path": "data/groups.json",
                        "bytes": groups.len() as u64,
                        "sha256": sha256_hex(groups.as_bytes()),
                    }
                }
            }))
            .into_diagnostic()?,
        )
        .into_diagnostic()?;

        let value = snapshot_stats(
            &dir,
            SnapshotStatsOptions {
                top: 5,
                verify: false,
            },
        )?;

        fs::remove_dir_all(&dir).ok();
        assert_eq!(
            value.pointer("/sections/contacts/file/path"),
            Some(&json!("data/contacts.jsonl"))
        );
        assert_eq!(value.pointer("/sections/contacts/rows"), Some(&json!(1)));
        assert_eq!(
            value.pointer("/sections/groups/file/path"),
            Some(&json!("data/groups.json"))
        );
        assert_eq!(value.pointer("/sections/groups/rows"), Some(&json!(1)));
        Ok(())
    }

    #[test]
    fn snapshot_history_snapshot_rejects_malformed_manifest() -> Result<()> {
        let dir = std::env::temp_dir().join(format!(
            "meshx-history-malformed-manifest-{}-{}",
            std::process::id(),
            now_millis()
        ));
        fs::create_dir_all(&dir).into_diagnostic()?;
        fs::write(dir.join("manifest.json"), "{").into_diagnostic()?;

        let error = snapshot_history_snapshot(dir.clone())
            .expect_err("malformed manifests should not rank as valid snapshots");

        fs::remove_dir_all(&dir).ok();
        assert!(error.to_string().contains("parsing"));
        Ok(())
    }

    #[test]
    fn snapshot_history_snapshot_falls_back_to_mtime_without_created_at() -> Result<()> {
        let dir = std::env::temp_dir().join(format!(
            "meshx-history-mtime-fallback-{}-{}",
            std::process::id(),
            now_millis()
        ));
        fs::create_dir_all(&dir).into_diagnostic()?;
        fs::write(dir.join("manifest.json"), r#"{"files":{}}"#).into_diagnostic()?;

        let snapshot = snapshot_history_snapshot(dir.clone())?;

        fs::remove_dir_all(&dir).ok();
        assert_eq!(snapshot.created_at_unix_ms, None);
        assert!(snapshot.rank_time_unix_ms.is_some());
        assert_eq!(snapshot.rank_time_source, "filesystem.modified");
        Ok(())
    }

    #[test]
    fn snapshot_history_records_rejects_duplicate_jsonl_ids() -> Result<()> {
        let dir = std::env::temp_dir().join(format!(
            "meshx-history-duplicate-jsonl-{}-{}",
            std::process::id(),
            now_millis()
        ));
        fs::create_dir_all(dir.join("data")).into_diagnostic()?;
        let content = "{\"id\":1,\"name\":\"first\"}\n{\"id\":1,\"name\":\"second\"}\n";
        fs::write(dir.join("data/contacts.jsonl"), content).into_diagnostic()?;
        fs::write(
            dir.join("manifest.json"),
            serde_json::to_string(&json!({
                "files": {
                    "contacts": {
                        "path": "data/contacts.jsonl",
                        "bytes": content.len() as u64,
                        "sha256": sha256_hex(content.as_bytes()),
                    }
                }
            }))
            .into_diagnostic()?,
        )
        .into_diagnostic()?;

        let options = SnapshotHistoryOptions {
            root: dir.clone(),
            recursive: false,
            max_depth: None,
            limit: None,
            section: snapshot_query_section("contacts")?,
            ids: vec![1],
            verify: false,
            index: SnapshotIndexMode::Off,
            details: false,
            detail_limit: 0,
            records: false,
        };
        let error = snapshot_history_records(&dir, &options)
            .expect_err("duplicate IDs should not be collapsed");

        fs::remove_dir_all(&dir).ok();
        assert!(error.to_string().contains("duplicate record ID 1"));
        assert!(error.to_string().contains("data/contacts.jsonl"));
        Ok(())
    }

    #[test]
    fn snapshot_history_records_rejects_duplicate_array_ids() -> Result<()> {
        let dir = std::env::temp_dir().join(format!(
            "meshx-history-duplicate-array-{}-{}",
            std::process::id(),
            now_millis()
        ));
        fs::create_dir_all(dir.join("data")).into_diagnostic()?;
        let content = r#"[{"id":1,"name":"first"},{"id":1,"name":"second"}]"#;
        fs::write(dir.join("data/groups.json"), content).into_diagnostic()?;
        fs::write(
            dir.join("manifest.json"),
            serde_json::to_string(&json!({
                "files": {
                    "groups": {
                        "path": "data/groups.json",
                        "bytes": content.len() as u64,
                        "sha256": sha256_hex(content.as_bytes()),
                    }
                }
            }))
            .into_diagnostic()?,
        )
        .into_diagnostic()?;

        let options = SnapshotHistoryOptions {
            root: dir.clone(),
            recursive: false,
            max_depth: None,
            limit: None,
            section: snapshot_query_section("groups")?,
            ids: vec![1],
            verify: false,
            index: SnapshotIndexMode::Off,
            details: false,
            detail_limit: 0,
            records: false,
        };
        let error = snapshot_history_records(&dir, &options)
            .expect_err("duplicate IDs should not be collapsed");

        fs::remove_dir_all(&dir).ok();
        assert!(error.to_string().contains("duplicate record ID 1"));
        assert!(error.to_string().contains("data/groups.json"));
        Ok(())
    }

    #[test]
    fn snapshot_prune_reports_snapshot_byte_errors() -> Result<()> {
        let root = std::env::temp_dir().join(format!(
            "meshx-prune-byte-error-{}-{}",
            std::process::id(),
            now_millis()
        ));
        let snapshot = root.join("broken-snapshot");
        fs::create_dir_all(&snapshot).into_diagnostic()?;
        fs::write(snapshot.join("manifest.json"), "{").into_diagnostic()?;

        let result = snapshot_prune(SnapshotPruneOptions {
            root: root.clone(),
            recursive: true,
            max_depth: None,
            limit: None,
            include_snapshots: true,
            include_archives: true,
            keep_latest: Some(0),
            older_than_days: None,
            failed: false,
            require_index: false,
            dry_run: true,
            yes: false,
        });

        fs::remove_dir_all(&root).ok();

        let value = result?;
        assert_eq!(value.get("ok"), Some(&json!(false)));
        assert_eq!(value.pointer("/summary/action_errors"), Some(&json!(1)));
        let action = value
            .get("actions")
            .and_then(Value::as_array)
            .and_then(|actions| actions.first())
            .expect("expected prune action");
        assert_eq!(action.get("status"), Some(&json!("would-delete")));
        assert_eq!(action.get("bytes"), Some(&Value::Null));
        assert!(
            action
                .get("error")
                .and_then(Value::as_str)
                .is_some_and(|error| error.contains("parsing"))
        );
        Ok(())
    }

    #[cfg(unix)]
    #[test]
    fn snapshot_prune_reports_discovery_errors() -> Result<()> {
        use std::os::unix::fs::PermissionsExt;

        let root = std::env::temp_dir().join(format!(
            "meshx-prune-discovery-error-{}-{}",
            std::process::id(),
            now_millis()
        ));
        let blocked = root.join("blocked");
        fs::create_dir_all(&blocked).into_diagnostic()?;
        fs::set_permissions(&blocked, fs::Permissions::from_mode(0o000)).into_diagnostic()?;

        let result = snapshot_prune(SnapshotPruneOptions {
            root: root.clone(),
            recursive: true,
            max_depth: None,
            limit: None,
            include_snapshots: true,
            include_archives: true,
            keep_latest: Some(0),
            older_than_days: None,
            failed: false,
            require_index: false,
            dry_run: true,
            yes: false,
        });

        fs::set_permissions(&blocked, fs::Permissions::from_mode(0o700)).ok();
        fs::remove_dir_all(&root).ok();

        let value = result?;
        assert_eq!(value.get("ok"), Some(&json!(false)));
        assert_eq!(value.pointer("/summary/discovery_errors"), Some(&json!(1)));
        assert_eq!(
            value
                .get("discovery_errors")
                .and_then(Value::as_array)
                .map(Vec::len),
            Some(1)
        );
        Ok(())
    }

    #[cfg(unix)]
    #[test]
    fn snapshot_prune_refuses_live_delete_after_discovery_errors() -> Result<()> {
        use std::os::unix::fs::PermissionsExt;

        let root = std::env::temp_dir().join(format!(
            "meshx-prune-live-discovery-error-{}-{}",
            std::process::id(),
            now_millis()
        ));
        let blocked = root.join("blocked");
        fs::create_dir_all(&blocked).into_diagnostic()?;
        fs::set_permissions(&blocked, fs::Permissions::from_mode(0o000)).into_diagnostic()?;

        let result = snapshot_prune(SnapshotPruneOptions {
            root: root.clone(),
            recursive: true,
            max_depth: None,
            limit: None,
            include_snapshots: true,
            include_archives: true,
            keep_latest: Some(0),
            older_than_days: None,
            failed: false,
            require_index: false,
            dry_run: false,
            yes: true,
        });

        fs::set_permissions(&blocked, fs::Permissions::from_mode(0o700)).ok();
        fs::remove_dir_all(&root).ok();

        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("snapshot:prune discovery failed")
        );
        Ok(())
    }
}
