wasm4pm 26.6.25

//! WASM bindings for streaming discovery algorithms.
//!
//! This module provides JavaScript-accessible functions for all streaming
//! algorithms. Each algorithm has:
//!
//! - `streaming_<algorithm>_begin()` - Create new session
//! - `streaming_<algorithm>_add_event()` - Add one event
//! - `streaming_<algorithm>_add_batch()` - Add batch of events
//! - `streaming_<algorithm>_close_trace()` - Close a trace
//! - `streaming_<algorithm>_snapshot()` - Get current model
//! - `streaming_<algorithm>_finalize()` - Finalize and return model
//! - `streaming_<algorithm>_stats()` - Get statistics

use crate::state::{get_or_init_state, StoredObject};
use crate::streaming::{
    StreamingAlgorithm, StreamingDfgBuilder, StreamingHeuristicBuilder, StreamingSkeletonBuilder,
};
use crate::utilities::to_js_str;
use serde_json::json;
use wasm_bindgen::prelude::*;

// ============================================================================
// DFG Streaming (already existed, moved here)
// ============================================================================

/// Begin a new streaming DFG session.
#[wasm_bindgen]
pub fn streaming_dfg_begin() -> Result<JsValue, JsValue> {
    let handle = get_or_init_state()
        .store_object(StoredObject::StreamingDfgBuilder(StreamingDfgBuilder::new()))?;
    Ok(crate::error::js_val(&handle))
}

/// Append one event to an in-progress DFG trace.
#[wasm_bindgen]
pub fn streaming_dfg_add_event(
    handle: &str,
    case_id: &str,
    activity: &str,
) -> Result<JsValue, JsValue> {
    get_or_init_state().with_streaming_dfg_mut(handle, |b| {
        b.add_event(case_id, activity);
        to_js_str(&json!({
            "ok": true,
            "event_count": b.event_count,
            "open_traces": b.open_traces.len(),
            "activities": b.interner.len(),
        }))
    })
}

/// Add a batch of events in one call (chunked ingestion).
#[wasm_bindgen]
pub fn streaming_dfg_add_batch(handle: &str, events_json: &str) -> Result<JsValue, JsValue> {
    let batch: Vec<serde_json::Value> = serde_json::from_str(events_json)
        .map_err(|e| crate::error::js_val(&format!("Invalid events JSON: {}", e)))?;

    get_or_init_state().with_streaming_dfg_mut(handle, |b| {
        let mut added = 0usize;
        for item in &batch {
            let case_id = item["case_id"].as_str().ok_or_else(|| {
                crate::error::js_val("Each event must have a 'case_id' string field")
            })?;
            let activity = item["activity"].as_str().ok_or_else(|| {
                crate::error::js_val("Each event must have an 'activity' string field")
            })?;
            b.add_event(case_id, activity);
            added += 1;
        }
        to_js_str(&json!({
            "ok": true,
            "added": added,
            "event_count": b.event_count,
            "open_traces": b.open_traces.len(),
            "activities": b.interner.len(),
        }))
    })
}

/// Close a DFG trace and fold into model.
#[wasm_bindgen]
pub fn streaming_dfg_close_trace(handle: &str, case_id: &str) -> Result<JsValue, JsValue> {
    get_or_init_state().with_streaming_dfg_mut(handle, |b| {
        let closed = b.close_trace(case_id);
        to_js_str(&json!({
            "ok": closed,
            "trace_count": b.trace_count,
            "open_traces": b.open_traces.len(),
        }))
    })
}

/// Flush all currently-open DFG traces.
#[wasm_bindgen]
pub fn streaming_dfg_flush_open(handle: &str) -> Result<JsValue, JsValue> {
    get_or_init_state().with_streaming_dfg_mut(handle, |b| {
        let case_ids: Vec<String> = b.open_traces.keys().cloned().collect();
        let flushed = case_ids.len();
        for id in case_ids {
            b.close_trace(&id);
        }
        to_js_str(&json!({
            "ok": true,
            "flushed": flushed,
            "trace_count": b.trace_count,
        }))
    })
}

/// Take a non-destructive DFG snapshot.
///
/// Returns a JSON string (not a JS object) — callers must `JSON.parse()` the result.
/// This uses `serde_json::to_string` + `JsValue::from_str` to avoid the known
/// `serde_wasm_bindgen::to_value` bug that silently returns `{}` on wasm32 for
/// `serde_json::Value` payloads.
#[wasm_bindgen]
pub fn streaming_dfg_snapshot(handle: &str) -> Result<JsValue, JsValue> {
    get_or_init_state().with_streaming_dfg(handle, |b| {
        let dfg = b.snapshot();
        to_js_str(&dfg)
    })
}

/// Finalize the stream and return DFG handle.
#[wasm_bindgen]
pub fn streaming_dfg_finalize(handle: &str) -> Result<JsValue, JsValue> {
    let dfg = get_or_init_state().with_streaming_dfg_mut(handle, |b| {
        let case_ids: Vec<String> = b.open_traces.keys().cloned().collect();
        for id in case_ids {
            b.close_trace(&id);
        }
        Ok(b.snapshot())
    })?;

    let n_nodes = dfg.nodes.len();
    let n_edges = dfg.edges.len();
    let dfg_handle = get_or_init_state()
        .store_object(StoredObject::DFG(dfg))
        .map_err(|_| crate::error::js_val("Failed to store DFG"))?;

    get_or_init_state().delete_object(handle)?;

    to_js_str(&json!({
        "dfg_handle": dfg_handle,
        "nodes": n_nodes,
        "edges": n_edges,
    }))
}

/// Report memory/progress statistics.
///
/// Returns a JSON string — callers must `JSON.parse()` the result.
#[wasm_bindgen]
pub fn streaming_dfg_stats(handle: &str) -> Result<JsValue, JsValue> {
    get_or_init_state().with_streaming_dfg(handle, |b| {
        let stats = b.stats();
        to_js_str(&stats)
    })
}

// ============================================================================
// Skeleton Streaming
// ============================================================================

/// Begin a new streaming Skeleton session.
#[wasm_bindgen]
pub fn streaming_skeleton_begin(min_frequency: usize) -> Result<JsValue, JsValue> {
    let builder = StreamingSkeletonBuilder::with_min_frequency(min_frequency);
    let handle =
        get_or_init_state().store_object(StoredObject::StreamingSkeletonBuilder(builder))?;
    Ok(crate::error::js_val(&handle))
}

/// Append one event to an in-progress Skeleton trace.
#[wasm_bindgen]
pub fn streaming_skeleton_add_event(
    handle: &str,
    case_id: &str,
    activity: &str,
) -> Result<JsValue, JsValue> {
    get_or_init_state().with_streaming_skeleton_mut(handle, |b| {
        b.add_event(case_id, activity);
        let stats = b.stats();
        to_js_str(&json!({
            "ok": true,
            "event_count": stats.event_count,
            "open_traces": stats.open_traces,
        }))
    })
}

/// Close a Skeleton trace.
#[wasm_bindgen]
pub fn streaming_skeleton_close_trace(handle: &str, case_id: &str) -> Result<JsValue, JsValue> {
    get_or_init_state().with_streaming_skeleton_mut(handle, |b| {
        let closed = b.close_trace(case_id);
        to_js_str(&json!({
            "ok": closed,
            "trace_count": b.trace_count,
        }))
    })
}

/// Take a non-destructive Skeleton snapshot.
///
/// Returns a JSON string — callers must `JSON.parse()` the result.
#[wasm_bindgen]
pub fn streaming_skeleton_snapshot(handle: &str) -> Result<JsValue, JsValue> {
    get_or_init_state().with_streaming_skeleton(handle, |b| {
        let dfg = b.snapshot();
        to_js_str(&dfg)
    })
}

/// Finalize Skeleton stream.
#[wasm_bindgen]
pub fn streaming_skeleton_finalize(handle: &str) -> Result<JsValue, JsValue> {
    let dfg = get_or_init_state().with_streaming_skeleton_mut(handle, |b| {
        let case_ids: Vec<String> = b.open_trace_ids();
        for id in case_ids {
            b.close_trace(&id);
        }
        Ok(b.snapshot())
    })?;

    let n_nodes = dfg.nodes.len();
    let n_edges = dfg.edges.len();
    let dfg_handle = get_or_init_state()
        .store_object(StoredObject::DFG(dfg))
        .map_err(|_| crate::error::js_val("Failed to store DFG"))?;

    get_or_init_state().delete_object(handle)?;

    to_js_str(&json!({
        "dfg_handle": dfg_handle,
        "nodes": n_nodes,
        "edges": n_edges,
    }))
}

// ============================================================================
// Heuristic Streaming
// ============================================================================

/// Begin a new streaming Heuristic Miner session.
#[wasm_bindgen]
pub fn streaming_heuristic_begin(threshold: f64) -> Result<JsValue, JsValue> {
    let builder = StreamingHeuristicBuilder::with_dependency_threshold(threshold);
    let handle =
        get_or_init_state().store_object(StoredObject::StreamingHeuristicBuilder(builder))?;
    Ok(crate::error::js_val(&handle))
}

/// Append one event to an in-progress Heuristic trace.
#[wasm_bindgen]
pub fn streaming_heuristic_add_event(
    handle: &str,
    case_id: &str,
    activity: &str,
) -> Result<JsValue, JsValue> {
    get_or_init_state().with_streaming_heuristic_mut(handle, |b| {
        b.add_event(case_id, activity);
        let stats = b.stats();
        to_js_str(&json!({
            "ok": true,
            "event_count": stats.event_count,
            "open_traces": stats.open_traces,
        }))
    })
}

/// Close a Heuristic trace.
#[wasm_bindgen]
pub fn streaming_heuristic_close_trace(handle: &str, case_id: &str) -> Result<JsValue, JsValue> {
    get_or_init_state().with_streaming_heuristic_mut(handle, |b| {
        let closed = b.close_trace(case_id);
        to_js_str(&json!({
            "ok": closed,
            "trace_count": b.trace_count,
        }))
    })
}

/// Take a non-destructive Heuristic snapshot.
///
/// Returns a JSON string — callers must `JSON.parse()` the result.
#[wasm_bindgen]
pub fn streaming_heuristic_snapshot(handle: &str) -> Result<JsValue, JsValue> {
    get_or_init_state().with_streaming_heuristic(handle, |b| {
        let dfg = b.snapshot();
        to_js_str(&dfg)
    })
}

/// Finalize Heuristic stream.
#[wasm_bindgen]
pub fn streaming_heuristic_finalize(handle: &str) -> Result<JsValue, JsValue> {
    let dfg = get_or_init_state().with_streaming_heuristic_mut(handle, |b| {
        let case_ids: Vec<String> = b.open_trace_ids();
        for id in case_ids {
            b.close_trace(&id);
        }
        Ok(b.snapshot())
    })?;

    let n_nodes = dfg.nodes.len();
    let n_edges = dfg.edges.len();
    let dfg_handle = get_or_init_state()
        .store_object(StoredObject::DFG(dfg))
        .map_err(|_| crate::error::js_val("Failed to store DFG"))?;

    get_or_init_state().delete_object(handle)?;

    to_js_str(&json!({
        "dfg_handle": dfg_handle,
        "nodes": n_nodes,
        "edges": n_edges,
    }))
}

/// Streaming module info.
#[wasm_bindgen]
pub fn streaming_info() -> String {
    serde_json::json!({
        "status": "streaming_api_available",
        "description": "Streaming discovery algorithms for infinite event streams",
        "algorithms": [
            {"name": "dfg", "status": "implemented"},
            {"name": "skeleton", "status": "implemented"},
            {"name": "heuristic", "status": "implemented"},
            {"name": "alpha_plus_plus", "status": "implemented"},
            {"name": "declare", "status": "implemented"},
            {"name": "inductive_miner", "status": "implemented"},
            {"name": "hill_climbing", "status": "implemented"},
            {"name": "noise_filtered_dfg", "status": "implemented"},
            {"name": "astar", "status": "implemented"},
        ]
    })
    .to_string()
}