huskies/server/src/crdt_state/read.rs

//! Read API for pipeline items, dump introspection, and dependency helpers.

#![allow(unused_imports, dead_code)]
use std::collections::HashMap;

use bft_json_crdt::json_crdt::*;

use super::state::{ALL_OPS, apply_and_persist, get_crdt, rebuild_index};
use super::types::{PipelineDoc, PipelineItemCrdt, PipelineItemView};
use bft_json_crdt::op::ROOT_ID;

// ── Debug dump ───────────────────────────────────────────────────────

/// A raw dump of a single CRDT list entry, including deleted items.
///
/// Use `content_index` (hex of the list insert `OpId`) to cross-reference
/// with rows in the `crdt_ops` SQLite table.
pub struct CrdtItemDump {
    pub story_id: Option<String>,
    pub stage: Option<String>,
    pub name: Option<String>,
    pub agent: Option<String>,
    pub retry_count: Option<i64>,
    pub blocked: Option<bool>,
    pub depends_on: Option<Vec<u32>>,
    pub claimed_by: Option<String>,
    pub claimed_at: Option<f64>,
    /// Hex-encoded OpId of the list insert op — cross-reference with `crdt_ops`.
    pub content_index: String,
    pub is_deleted: bool,
}

/// Top-level debug dump of the in-memory CRDT state.
pub struct CrdtStateDump {
    pub in_memory_state_loaded: bool,
    /// Count of non-deleted items with a valid story_id and stage.
    pub total_items: usize,
    /// Total list-level ops seen (excludes root sentinel).
    pub total_ops_in_list: usize,
    /// Highest Lamport sequence number seen across all list-level ops.
    pub max_seq_in_list: u64,
    /// Count of ops in the ALL_OPS journal (persisted ops replayed at startup).
    pub persisted_ops_count: usize,
    pub items: Vec<CrdtItemDump>,
}

/// Dump the raw in-memory CRDT state for debugging.
///
/// Unlike [`read_all_items`] this includes tombstoned (deleted) entries and
/// exposes internal op metadata (content_index, seq).  Pass a `story_id`
/// filter to restrict the output to a single item.
///
/// **This is a debug tool.**  For normal pipeline introspection use
/// [`read_all_items`] or the `get_pipeline_status` MCP tool instead.
pub fn dump_crdt_state(story_id_filter: Option<&str>) -> CrdtStateDump {
    let in_memory_state_loaded = get_crdt().is_some();

    let persisted_ops_count = ALL_OPS
        .get()
        .and_then(|m| m.lock().ok().map(|v| v.len()))
        .unwrap_or(0);

    let Some(state_mutex) = get_crdt() else {
        return CrdtStateDump {
            in_memory_state_loaded,
            total_items: 0,
            total_ops_in_list: 0,
            max_seq_in_list: 0,
            persisted_ops_count,
            items: Vec::new(),
        };
    };

    let Ok(state) = state_mutex.lock() else {
        return CrdtStateDump {
            in_memory_state_loaded,
            total_items: 0,
            total_ops_in_list: 0,
            max_seq_in_list: 0,
            persisted_ops_count,
            items: Vec::new(),
        };
    };

    let total_items = state.crdt.doc.items.iter().count();

    let max_seq_in_list = state
        .crdt
        .doc
        .items
        .ops
        .iter()
        .map(|op| op.seq)
        .max()
        .unwrap_or(0);

    // Subtract 1 for the root sentinel.
    let total_ops_in_list = state.crdt.doc.items.ops.len().saturating_sub(1);

    let mut items = Vec::new();
    for op in &state.crdt.doc.items.ops {
        // Skip root sentinel (id == [0u8; 32]).
        if op.id == ROOT_ID {
            continue;
        }
        let Some(ref item_crdt) = op.content else {
            // No content — skip (orphaned slot, should not happen in normal use).
            continue;
        };

        let story_id = match item_crdt.story_id.view() {
            JsonValue::String(s) if !s.is_empty() => Some(s),
            _ => None,
        };

        // Apply story_id filter before doing any further work.
        if let Some(filter) = story_id_filter
            && story_id.as_deref() != Some(filter)
        {
            continue;
        }

        let stage = match item_crdt.stage.view() {
            JsonValue::String(s) if !s.is_empty() => Some(s),
            _ => None,
        };
        let name = match item_crdt.name.view() {
            JsonValue::String(s) if !s.is_empty() => Some(s),
            _ => None,
        };
        let agent = match item_crdt.agent.view() {
            JsonValue::String(s) if !s.is_empty() => Some(s),
            _ => None,
        };
        let retry_count = match item_crdt.retry_count.view() {
            JsonValue::Number(n) => Some(n as i64),
            _ => None,
        };
        let blocked = match item_crdt.blocked.view() {
            JsonValue::Bool(b) => Some(b),
            _ => None,
        };
        let depends_on = match item_crdt.depends_on.view() {
            JsonValue::String(s) if !s.is_empty() => serde_json::from_str::<Vec<u32>>(&s).ok(),
            _ => None,
        };

        let claimed_by = match item_crdt.claimed_by.view() {
            JsonValue::String(s) if !s.is_empty() => Some(s),
            _ => None,
        };
        let claimed_at = match item_crdt.claimed_at.view() {
            JsonValue::Number(n) if n > 0.0 => Some(n),
            _ => None,
        };

        let content_index = op.id.iter().map(|b| format!("{b:02x}")).collect::<String>();

        items.push(CrdtItemDump {
            story_id,
            stage,
            name,
            agent,
            retry_count,
            blocked,
            depends_on,
            claimed_by,
            claimed_at,
            content_index,
            is_deleted: op.is_deleted,
        });
    }

    CrdtStateDump {
        in_memory_state_loaded,
        total_items,
        total_ops_in_list,
        max_seq_in_list,
        persisted_ops_count,
        items,
    }
}

// ── Read path ────────────────────────────────────────────────────────

/// Read the full pipeline state from the CRDT document.
///
/// Returns items grouped by stage, or `None` if the CRDT layer is not
/// initialised.
pub fn read_all_items() -> Option<Vec<PipelineItemView>> {
    let state_mutex = get_crdt()?;
    let state = state_mutex.lock().ok()?;

    // Only return items that appear in the deduplicated index.
    // The index maps story_id → visible_index and represents the
    // latest-wins view of each story.  Iterating raw CRDT entries
    // would return stale duplicates from earlier stage writes.
    let mut items = Vec::with_capacity(state.index.len());
    for &idx in state.index.values() {
        if let Some(view) = extract_item_view(&state.crdt.doc.items[idx]) {
            items.push(view);
        }
    }
    Some(items)
}

/// Read a single pipeline item by story_id.
pub fn read_item(story_id: &str) -> Option<PipelineItemView> {
    let state_mutex = get_crdt()?;
    let state = state_mutex.lock().ok()?;
    let &idx = state.index.get(story_id)?;
    extract_item_view(&state.crdt.doc.items[idx])
}

/// Mark a story as deleted in the in-memory CRDT and persist a tombstone op.
///
/// This is the eviction primitive for story 521 — it lets external callers
/// (e.g. the `purge_story` MCP tool, or operator scripts during incident
/// response) clear an item from the running server's in-memory state
/// without needing a full process restart.
///
/// Specifically:
/// 1. Looks up the item's CRDT `OpId` via the visible-index map.
/// 2. Constructs a delete op via the bft-json-crdt list `delete()` primitive.
/// 3. Signs it with the local node's keypair and applies it to the in-memory
///    CRDT (marking the item `is_deleted = true` so subsequent
///    `read_all_items` / `read_item` calls skip it).
/// 4. Persists the signed delete op to `crdt_ops` via the existing
///    `apply_and_persist` channel — so the eviction survives a restart.
/// 5. Rebuilds the `story_id → visible_index` map (visible indices shift
///    when an item is marked deleted).
/// 6. Drops the in-memory content-store entry for the story so the cached
///    body doesn't outlive the CRDT entry.
///
/// Returns `Ok(())` if the item was found and a tombstone op was queued,
/// or an `Err` if the CRDT layer isn't initialised or the story_id is
/// unknown to the in-memory state.
pub fn evict_item(story_id: &str) -> Result<(), String> {
    let state_mutex = get_crdt().ok_or_else(|| "CRDT layer not initialised".to_string())?;
    let mut state = state_mutex
        .lock()
        .map_err(|e| format!("CRDT lock poisoned: {e}"))?;

    let idx = state
        .index
        .get(story_id)
        .copied()
        .ok_or_else(|| format!("Story '{story_id}' not found in in-memory CRDT"))?;

    // Resolve the item's OpId before the closure (the closure will mutably
    // borrow `state`, so we can't access `state.crdt.doc.items` from inside).
    let item_id =
        state.crdt.doc.items.id_at(idx).ok_or_else(|| {
            format!("Item index {idx} for '{story_id}' did not resolve to an OpId")
        })?;

    // Write the delete op via the existing apply_and_persist machinery.
    // This signs the op, applies it to the in-memory CRDT (marking the item
    // is_deleted), and sends it to the persistence task.
    apply_and_persist(&mut state, |s| s.crdt.doc.items.delete(item_id));

    // Rebuild the story_id → visible_index map; the deleted item is no
    // longer counted by the iter that rebuild_index uses.
    state.index = rebuild_index(&state.crdt);

    // Drop the content-store entry so the cached body doesn't outlive the
    // CRDT entry. (Bug 521 follow-up: when CONTENT_STORE becomes a true
    // lazy cache, this explicit eviction can go away.)
    crate::db::delete_content(story_id);

    Ok(())
}

/// Extract a `PipelineItemView` from a `PipelineItemCrdt`.
pub(super) fn extract_item_view(item: &PipelineItemCrdt) -> Option<PipelineItemView> {
    let story_id = match item.story_id.view() {
        JsonValue::String(s) if !s.is_empty() => s,
        _ => return None,
    };
    let stage = match item.stage.view() {
        JsonValue::String(s) if !s.is_empty() => s,
        _ => return None,
    };
    let name = match item.name.view() {
        JsonValue::String(s) if !s.is_empty() => Some(s),
        _ => None,
    };
    let agent = match item.agent.view() {
        JsonValue::String(s) if !s.is_empty() => Some(s),
        _ => None,
    };
    let retry_count = match item.retry_count.view() {
        JsonValue::Number(n) => Some(n as i64),
        _ => None,
    };
    let blocked = match item.blocked.view() {
        JsonValue::Bool(b) => Some(b),
        _ => None,
    };
    let depends_on = match item.depends_on.view() {
        JsonValue::String(s) if !s.is_empty() => serde_json::from_str::<Vec<u32>>(&s).ok(),
        _ => None,
    };

    let claimed_by = match item.claimed_by.view() {
        JsonValue::String(s) if !s.is_empty() => Some(s),
        _ => None,
    };
    let claimed_at = match item.claimed_at.view() {
        JsonValue::Number(n) if n > 0.0 => Some(n),
        _ => None,
    };
    let merged_at = match item.merged_at.view() {
        JsonValue::Number(n) if n > 0.0 => Some(n),
        _ => None,
    };

    let qa_mode = match item.qa_mode.view() {
        JsonValue::String(s) if !s.is_empty() => Some(s),
        _ => None,
    };

    let mergemaster_attempted = match item.mergemaster_attempted.view() {
        JsonValue::Bool(b) => Some(b),
        _ => None,
    };

    Some(PipelineItemView {
        story_id,
        stage,
        name,
        agent,
        retry_count,
        blocked,
        depends_on,
        claimed_by,
        claimed_at,
        merged_at,
        qa_mode,
        mergemaster_attempted,
    })
}

/// Check whether a dependency (by numeric ID prefix) is in `5_done` or `6_archived`
/// according to CRDT state.
///
/// Returns `true` if the dependency is satisfied (item found in a done stage).
/// Matches both legacy slug-form IDs (`"664_story_foo"`) and numeric-only IDs
/// (`"664"`) so the check remains correct after the slug→numeric migration.
/// See `dep_is_archived_crdt` to distinguish archive-satisfied from cleanly-done.
pub fn dep_is_done_crdt(dep_number: u32) -> bool {
    use crate::pipeline_state::{Stage, read_all_typed};
    let exact = dep_number.to_string();
    let prefix = format!("{dep_number}_");
    read_all_typed().into_iter().any(|item| {
        (item.story_id.0 == exact || item.story_id.0.starts_with(&prefix))
            && matches!(item.stage, Stage::Done { .. } | Stage::Archived { .. })
    })
}

/// Check whether a dependency (by numeric ID prefix) is specifically in `6_archived`
/// according to CRDT state.
///
/// Used to detect when a dependency is satisfied via archive rather than via a clean
/// completion through `5_done`.  Returns `false` when the CRDT layer is not initialised.
/// Matches both legacy slug-form IDs (`"664_story_foo"`) and numeric-only IDs (`"664"`).
pub fn dep_is_archived_crdt(dep_number: u32) -> bool {
    use crate::pipeline_state::{Stage, read_all_typed};
    let exact = dep_number.to_string();
    let prefix = format!("{dep_number}_");
    read_all_typed().into_iter().any(|item| {
        (item.story_id.0 == exact || item.story_id.0.starts_with(&prefix))
            && matches!(item.stage, Stage::Archived { .. })
    })
}

/// Check unmet dependencies for a story by reading its `depends_on` from the
/// CRDT document and checking each dependency against CRDT state.
///
/// Returns the list of dependency numbers that are NOT in `5_done` or `6_archived`.
pub fn check_unmet_deps_crdt(story_id: &str) -> Vec<u32> {
    let item = match read_item(story_id) {
        Some(i) => i,
        None => return Vec::new(),
    };
    let deps = match item.depends_on {
        Some(d) => d,
        None => return Vec::new(),
    };
    deps.into_iter()
        .filter(|&dep| !dep_is_done_crdt(dep))
        .collect()
}

/// Return the list of dependency numbers from `story_id`'s `depends_on` that are
/// specifically in `6_archived` according to CRDT state.
///
/// Used to emit a warning when promotion fires because a dep is archived rather than
/// cleanly completed.  Returns an empty `Vec` when no deps are archived.
pub fn check_archived_deps_crdt(story_id: &str) -> Vec<u32> {
    let item = match read_item(story_id) {
        Some(i) => i,
        None => return Vec::new(),
    };
    let deps = match item.depends_on {
        Some(d) => d,
        None => return Vec::new(),
    };
    deps.into_iter()
        .filter(|&dep| dep_is_archived_crdt(dep))
        .collect()
}

#[cfg(test)]
mod tests {
    use super::super::state::init_for_test;
    use super::super::state::rebuild_index;
    use super::super::types::PipelineItemCrdt;
    use super::super::write::write_item;
    use super::*;
    use bft_json_crdt::json_crdt::OpState;
    use bft_json_crdt::keypair::make_keypair;
    use bft_json_crdt::op::ROOT_ID;
    use serde_json::json;

    #[test]
    fn extract_item_view_parses_crdt_item() {
        let kp = make_keypair();
        let mut crdt = BaseCrdt::<PipelineDoc>::new(&kp);

        let item_json: JsonValue = json!({
            "story_id": "40_story_view",
            "stage": "3_qa",
            "name": "View Test",
            "agent": "coder-1",
            "retry_count": 2.0,
            "blocked": true,
            "depends_on": "[10,20]",
            "claimed_by": "",
            "claimed_at": 0.0,
        })
        .into();

        let op = crdt.doc.items.insert(ROOT_ID, item_json).sign(&kp);
        crdt.apply(op);

        let view = extract_item_view(&crdt.doc.items[0]).unwrap();
        assert_eq!(view.story_id, "40_story_view");
        assert_eq!(view.stage, "3_qa");
        assert_eq!(view.name.as_deref(), Some("View Test"));
        assert_eq!(view.agent.as_deref(), Some("coder-1"));
        assert_eq!(view.retry_count, Some(2));
        assert_eq!(view.blocked, Some(true));
        assert_eq!(view.depends_on, Some(vec![10, 20]));
    }

    #[test]
    fn dep_is_done_crdt_returns_false_when_no_crdt_state() {
        // When the global CRDT state is not initialised (or in a test environment),
        // dep_is_done_crdt should return false rather than panicking.
        // Note: in the test binary the global may or may not be initialised,
        // but the function should never panic either way.
        let _ = dep_is_done_crdt(9999);
    }

    #[test]
    fn check_unmet_deps_crdt_returns_empty_when_item_not_found() {
        // Non-existent story should return empty deps.
        let result = check_unmet_deps_crdt("nonexistent_story");
        assert!(result.is_empty());
    }

    #[test]
    fn dep_is_archived_crdt_returns_false_when_no_crdt_state() {
        // When the global CRDT state is not initialised, must not panic.
        let _ = dep_is_archived_crdt(9998);
    }

    #[test]
    fn check_archived_deps_crdt_returns_empty_when_item_not_found() {
        // Non-existent story should return empty archived deps.
        let result = check_archived_deps_crdt("nonexistent_story_archived");
        assert!(result.is_empty());
    }
}