//! Notification-stream and server-request assertions for the compliance corpus
//! runner.
//!
//! The corpus `expect` DSL evaluated a single request/response pair. Three of
//! the 2025-06-18 rules need more than that: progress notifications arrive out
//! of band on the notification stream, elicitation is a server-initiated
//! request, and the completion cap is a `maxItems` bound. These functions are
//! the deterministic building blocks a runner uses to evaluate those rules over
//! a captured stream of JSON-RPC messages. They take already-captured messages
//! and assert over them; capturing the stream off the wire is the runner's job,
//! which keeps this crate transport-free.

use serde_json::Value;

use crate::matchers::{schema, AssertionDiff, AssertionResult, MatcherKind};

/// True when `msg` is a `notifications/progress` notification.
fn is_progress_notification(msg: &Value) -> bool {
    msg.get("method").and_then(|v| v.as_str()) == Some("notifications/progress")
}

/// The `progressToken` carried in a notification's params, if any.
fn progress_token(msg: &Value) -> Option<&Value> {
    msg.get("params").and_then(|p| p.get("progressToken"))
}

/// The numeric `progress` value carried in a notification's params, if any.
fn progress_value(msg: &Value) -> Option<f64> {
    msg.get("params")
        .and_then(|p| p.get("progress"))
        .and_then(Value::as_f64)
}

/// Collect the `notifications/progress` messages whose `progressToken` equals
/// `token`. This is the correlation PROG-002 needs: a request attaches a token
/// and every progress notification for that request must repeat it.
pub fn progress_notifications_for_token<'a>(stream: &'a [Value], token: &Value) -> Vec<&'a Value> {
    stream
        .iter()
        .filter(|m| is_progress_notification(m) && progress_token(m) == Some(token))
        .collect()
}

/// True when the `progress` values across these notifications never decrease.
///
/// An empty or single notification is trivially non-decreasing. A notification
/// that is missing a numeric `progress` breaks the contract and returns false,
/// because the spec requires every progress notification to carry the number.
pub fn progress_is_non_decreasing(notifications: &[&Value]) -> bool {
    let mut last: Option<f64> = None;
    for notification in notifications {
        let Some(current) = progress_value(notification) else {
            return false;
        };
        if let Some(previous) = last {
            if current < previous {
                return false;
            }
        }
        last = Some(current);
    }
    true
}

/// True when the stream carries no progress notifications at all.
///
/// PROG-001: a request without a `progressToken` must not trigger any
/// `notifications/progress`.
pub fn has_no_progress_notifications(stream: &[Value]) -> bool {
    !stream.iter().any(is_progress_notification)
}

/// True when `msg` is a JSON-RPC request: it carries both an `id` and a
/// `method`. A server-initiated request reaches the client this way; a
/// notification has a `method` but no `id`.
fn is_request(msg: &Value) -> bool {
    msg.get("id").is_some() && msg.get("method").is_some()
}

/// Collect server-initiated requests for `method`. Elicitation, sampling, and
/// roots all arrive as server-to-client requests, so a runner that captured the
/// stream can pull them out by method.
pub fn server_requests_for_method<'a>(stream: &'a [Value], method: &str) -> Vec<&'a Value> {
    stream
        .iter()
        .filter(|m| is_request(m) && m.get("method").and_then(|v| v.as_str()) == Some(method))
        .collect()
}

/// Validate that an accepted elicitation response carries content matching the
/// schema the server requested.
///
/// `request` is the server's `elicitation/create` request, whose
/// `params.requestedSchema` is the schema to satisfy. `result` is the client's
/// result envelope (`{ action, content }`). Returns:
///
/// - `None` when the action is not `accept` (a decline or cancel carries no
///   content, so the schema check does not apply), or when the request declared
///   no `requestedSchema` (nothing to validate against);
/// - `Some(result)` from the JSON Schema matcher otherwise, validating the
///   `content` object. A `requestedSchema` that is not valid JSON Schema is
///   reported as a failing result rather than a panic, since that is the
///   server's contract to get right.
pub fn elicited_content_matches_requested_schema(
    request: &Value,
    result: &Value,
) -> Option<AssertionResult> {
    if result.get("action").and_then(|v| v.as_str()) != Some("accept") {
        return None;
    }
    let requested = request
        .get("params")
        .and_then(|p| p.get("requestedSchema"))?;
    let content = result.get("content").unwrap_or(&Value::Null);
    match schema::evaluate(requested, content) {
        Ok(assertion) => Some(assertion),
        Err(err) => Some(AssertionResult::failed(
            MatcherKind::Schema,
            AssertionDiff {
                path: String::new(),
                expected: Some(requested.clone()),
                actual: Some(content.clone()),
                note: Some(format!("requestedSchema is not valid JSON Schema: {err}")),
            },
        )),
    }
}

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

    fn progress(token: &str, value: f64) -> Value {
        json!({
            "jsonrpc": "2.0",
            "method": "notifications/progress",
            "params": { "progressToken": token, "progress": value }
        })
    }

    #[test]
    fn correlates_progress_notifications_by_token() {
        let stream = vec![
            progress("job-a", 0.0),
            progress("job-b", 0.0),
            progress("job-a", 0.5),
            json!({ "jsonrpc": "2.0", "id": 1, "result": {} }),
        ];
        let matched = progress_notifications_for_token(&stream, &json!("job-a"));
        assert_eq!(matched.len(), 2);
    }

    #[test]
    fn non_decreasing_progress_passes_and_a_drop_fails() {
        let rising = [progress("a", 0.0), progress("a", 0.5), progress("a", 1.0)];
        let rising_refs: Vec<&Value> = rising.iter().collect();
        assert!(progress_is_non_decreasing(&rising_refs));

        let dropping = [progress("a", 0.6), progress("a", 0.4)];
        let dropping_refs: Vec<&Value> = dropping.iter().collect();
        assert!(!progress_is_non_decreasing(&dropping_refs));
    }

    #[test]
    fn missing_progress_number_breaks_non_decreasing() {
        let bad = json!({ "method": "notifications/progress", "params": { "progressToken": "a" } });
        let refs = vec![&bad];
        assert!(!progress_is_non_decreasing(&refs));
    }

    #[test]
    fn detects_absence_of_progress_notifications() {
        let quiet = vec![
            json!({ "jsonrpc": "2.0", "id": 1, "result": {} }),
            json!({ "jsonrpc": "2.0", "method": "notifications/message", "params": {} }),
        ];
        assert!(has_no_progress_notifications(&quiet));

        let noisy = vec![progress("a", 0.0)];
        assert!(!has_no_progress_notifications(&noisy));
    }

    #[test]
    fn pulls_server_initiated_requests_by_method() {
        let stream = vec![
            json!({ "jsonrpc": "2.0", "id": 7, "method": "elicitation/create", "params": {} }),
            json!({ "jsonrpc": "2.0", "method": "notifications/progress", "params": {} }),
            json!({ "jsonrpc": "2.0", "id": 8, "method": "sampling/createMessage", "params": {} }),
        ];
        let elicit = server_requests_for_method(&stream, "elicitation/create");
        assert_eq!(elicit.len(), 1);
        assert_eq!(elicit[0]["id"], 7);
    }

    fn elicit_request() -> Value {
        json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "elicitation/create",
            "params": {
                "message": "confirm?",
                "requestedSchema": {
                    "type": "object",
                    "required": ["confirm"],
                    "properties": { "confirm": { "type": "boolean" } }
                }
            }
        })
    }

    #[test]
    fn accepted_content_matching_schema_passes() {
        let result = json!({ "action": "accept", "content": { "confirm": true } });
        let assertion = elicited_content_matches_requested_schema(&elicit_request(), &result)
            .expect("accept action is checked");
        assert!(assertion.pass, "diffs: {:?}", assertion.diffs);
    }

    #[test]
    fn accepted_content_violating_schema_fails() {
        let result = json!({ "action": "accept", "content": { "confirm": "yes" } });
        let assertion = elicited_content_matches_requested_schema(&elicit_request(), &result)
            .expect("accept action is checked");
        assert!(!assertion.pass);
        assert!(!assertion.diffs.is_empty());
    }

    #[test]
    fn declined_elicitation_is_not_applicable() {
        let result = json!({ "action": "decline" });
        assert!(elicited_content_matches_requested_schema(&elicit_request(), &result).is_none());
    }

    #[test]
    fn completion_values_cap_is_enforceable_by_the_schema_matcher() {
        // COMPL-001 tightening: the schema matcher enforces maxItems, so a
        // capped array schema rejects an oversized completion list. This proves
        // the corpus `expect.schema` can pin the spec's 100-item cap.
        let capped = json!({
            "type": "object",
            "properties": {
                "values": { "type": "array", "items": { "type": "string" }, "maxItems": 100 }
            }
        });
        let oversize: Vec<Value> = (0..101).map(|i| json!(format!("c{i}"))).collect();
        let result = json!({ "values": oversize });
        let assertion = schema::evaluate(&capped, &result).expect("schema compiles");
        assert!(
            !assertion.pass,
            "a 101-item array must violate maxItems: 100"
        );
    }
}
