Plan: Rules Engine Implementation

use std::collections::HashMap;
use std::sync::Arc;
use std::time::Instant;
use tokio::sync::{RwLock, mpsc, oneshot};
use tracing::{info, warn};
use uuid::Uuid;
use zen_engine::Decision;
use zen_engine::model::DecisionContent;

use canopy_db::DbPool;
use canopy_mq::{EventEnvelope, Publisher};

use crate::store;

type ZenDecision = Decision;

/// Request sent to the dedicated evaluation thread.
struct EvalRequest {
    decision: Arc<ZenDecision>,
    input: serde_json::Value,
    reply: oneshot::Sender<Result<serde_json::Value, anyhow::Error>>,
}

/// In-memory rules engine backed by GoRules zen-engine.
///
/// Loads JDM rule sets from the database and caches compiled decisions.
/// Each evaluation records an audit trail in the `rule_evaluations` table.
///
/// Evaluations run on a dedicated thread with a single-threaded tokio runtime
/// because zen-engine's `evaluate()` future is `!Send` (uses `Rc` internally).
#[derive(Clone)]
pub struct RulesEngine {
    db: DbPool,
    decisions: Arc<RwLock<HashMap<String, Arc<ZenDecision>>>>,
    eval_tx: mpsc::Sender<EvalRequest>,
    publisher: Option<Publisher>,
    jurisdiction: Arc<String>,
    instance_id: Arc<String>,
}

impl RulesEngine {
    /// Create a new RulesEngine and load all active rule sets from the database.
    pub async fn new(
        db: DbPool,
        publisher: Option<Publisher>,
        jurisdiction: &str,
    ) -> anyhow::Result<Self> {
        let (eval_tx, eval_rx) = mpsc::channel::<EvalRequest>(256);

        // Spawn a dedicated OS thread running a single-threaded tokio runtime
        // for zen-engine evaluations (which produce !Send futures).
        std::thread::Builder::new()
            .name("zen-eval".into())
            .spawn(move || {
                let rt = tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                    .expect("failed to build zen-eval runtime");
                rt.block_on(eval_loop(eval_rx));
            })?;

        let engine = Self {
            db,
            decisions: Arc::new(RwLock::new(HashMap::new())),
            eval_tx,
            publisher,
            jurisdiction: Arc::new(jurisdiction.to_owned()),
            instance_id: Arc::new(Uuid::now_v7().to_string()),
        };
        engine.reload_all().await?;
        Ok(engine)
    }

    /// Reload all active rule sets from the database into the in-memory cache.
    pub async fn reload_all(&self) -> anyhow::Result<()> {
        let rule_sets = store::list_rule_sets(self.db.inner()).await?;
        let mut cache: HashMap<String, Arc<ZenDecision>> = HashMap::new();

        for rs in &rule_sets {
            match Self::compile_rule_set(&rs.content) {
                Ok(decision) => {
                    cache.insert(rs.name.clone(), Arc::new(decision));
                    info!(name = %rs.name, version = %rs.version, "rule set loaded");
                }
                Err(e) => {
                    warn!(name = %rs.name, error = %e, "failed to compile rule set, skipping");
                }
            }
        }

        info!(count = cache.len(), "rules engine initialized");
        *self.decisions.write().await = cache;
        Ok(())
    }

    /// Insert or replace a single compiled decision in the cache.
    pub async fn insert_decision(&self, name: String, decision: ZenDecision) {
        self.decisions
            .write()
            .await
            .insert(name, Arc::new(decision));
    }

    /// Remove a single decision from the cache by name.
    pub async fn remove_decision(&self, name: &str) {
        self.decisions.write().await.remove(name);
    }

    /// Evaluate an input against a named rule set.
    ///
    /// Records an audit trail in rule_evaluations and publishes a
    /// rules.evaluated event (best-effort).
    pub async fn evaluate(
        &self,
        rule_set_name: &str,
        context_type: Option<&str>,
        context_id: Option<Uuid>,
        input: serde_json::Value,
    ) -> Result<serde_json::Value, anyhow::Error> {
        let decision: Arc<ZenDecision> = self
            .decisions
            .read()
            .await
            .get(rule_set_name)
            .ok_or_else(|| anyhow::anyhow!("rule set not found: {rule_set_name}"))?
            .clone();

        let start = Instant::now();

        let (reply_tx, reply_rx) = oneshot::channel();
        self.eval_tx
            .send(EvalRequest {
                decision,
                input: input.clone(),
                reply: reply_tx,
            })
            .await
            .map_err(|_| anyhow::anyhow!("evaluation thread unavailable"))?;

        let output = reply_rx
            .await
            .map_err(|_| anyhow::anyhow!("evaluation thread dropped response"))??;

        let duration_ms = start.elapsed().as_millis() as i32;

        // Record audit trail.
        store::record_evaluation(
            self.db.inner(),
            rule_set_name,
            &input,
            &output,
            duration_ms,
            context_type,
            context_id,
        )
        .await?;

        // Publish rules.evaluated event (best-effort -- don't fail the evaluation).
        if let Some(ref publisher) = self.publisher {
            let payload = serde_json::json!({
                "rule_set_name": rule_set_name,
                "context_type": context_type,
                "context_id": context_id,
                "duration_ms": duration_ms,
            });
            let envelope = EventEnvelope::new("canopy-rules", "rules.evaluated", payload);
            if let Err(e) = publisher.publish(&envelope).await {
                warn!(error = %e, "failed to publish rules.evaluated event");
            }
        }

        Ok(output)
    }

    /// Compile a JDM rule set JSON value into a zen-engine Decision.
    pub fn compile_rule_set(content: &serde_json::Value) -> anyhow::Result<ZenDecision> {
        let decision_content: DecisionContent = serde_json::from_value(content.clone())?;
        Ok(ZenDecision::from(decision_content))
    }

    /// Unique identifier for this engine instance.
    pub fn instance_id(&self) -> &str {
        &self.instance_id
    }

    /// Notify other instances that the rule set cache has changed.
    ///
    /// Publishes a `rules.cache_invalidated` event so that horizontally-scaled
    /// instances can reload their in-memory caches from the database.
    pub async fn notify_cache_invalidated(&self, rule_set_name: &str, action: &str) {
        if let Some(ref publisher) = self.publisher {
            let envelope = EventEnvelope::new(
                "canopy-rules",
                "rules.cache_invalidated",
                serde_json::json!({
                    "instance_id": *self.instance_id,
                    "rule_set_name": rule_set_name,
                    "action": action,
                }),
            );
            if let Err(e) = publisher.publish(&envelope).await {
                warn!(error = %e, "failed to publish rules.cache_invalidated event");
            }
        }
    }
}

/// Event loop running on the dedicated zen-engine thread.
async fn eval_loop(mut rx: mpsc::Receiver<EvalRequest>) {
    while let Some(req) = rx.recv().await {
        let result = req
            .decision
            .evaluate((&req.input).into())
            .await
            .map(|resp| serde_json::to_value(&resp.result).unwrap_or(serde_json::Value::Null))
            .map_err(|e| anyhow::anyhow!("evaluation failed: {e}"));
        let _ = req.reply.send(result);
    }
}

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use uuid::Uuid;

/// A persisted JDM rule set.
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct RuleSet {
    pub id: Uuid,
    pub name: String,
    pub description: Option<String>,
    pub version: String,
    pub content: serde_json::Value,
    pub active: bool,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
}

/// Audit record for a rule evaluation.
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct RuleEvaluation {
    pub id: Uuid,
    pub rule_set_name: String,
    pub context_type: Option<String>,
    pub context_id: Option<Uuid>,
    pub input: serde_json::Value,
    pub output: serde_json::Value,
    pub duration_ms: i32,
    pub created_at: DateTime<Utc>,
}

/// List all active rule sets (used internally by the engine cache reload).
pub async fn list_rule_sets(pool: &PgPool) -> Result<Vec<RuleSet>, sqlx::Error> {
    sqlx::query_as::<_, RuleSet>(
        "SELECT * FROM rule_sets WHERE active = true ORDER BY name"
    )
    .fetch_all(pool)
    .await
}

/// List active rule sets with pagination and optional search.
pub async fn list_rule_sets_paged(
    pool: &PgPool,
    limit: i64,
    offset: i64,
    search: Option<&str>,
    sort_by: Option<&str>,
    sort_dir: Option<&str>,
) -> Result<Vec<RuleSet>, sqlx::Error> {
    let col = match sort_by {
        Some("name") => "name",
        Some("version") => "version",
        _ => "created_at",
    };
    let dir = match sort_dir {
        Some("asc") | Some("ASC") => "ASC",
        _ => "DESC",
    };
    let sql = format!(
        r#"SELECT * FROM rule_sets
           WHERE active = true
             AND ($1::TEXT IS NULL
                  OR name ILIKE '%' || $1 || '%'
                  OR COALESCE(description, '') ILIKE '%' || $1 || '%')
           ORDER BY {col} {dir}
           LIMIT $2 OFFSET $3"#
    );
    sqlx::query_as::<_, RuleSet>(&sql)
        .bind(search)
        .bind(limit)
        .bind(offset)
        .fetch_all(pool)
        .await
}

/// Count active rule sets, optionally filtered by search term.
pub async fn count_rule_sets(
    pool: &PgPool,
    search: Option<&str>,
) -> Result<i64, sqlx::Error> {
    let row: (i64,) = sqlx::query_as(
        r#"SELECT COUNT(*) FROM rule_sets
           WHERE active = true
             AND ($1::TEXT IS NULL
                  OR name ILIKE '%' || $1 || '%'
                  OR COALESCE(description, '') ILIKE '%' || $1 || '%')"#,
    )
    .bind(search)
    .fetch_one(pool)
    .await?;
    Ok(row.0)
}

/// Get a rule set by ID.
pub async fn get_rule_set(
    pool: &PgPool,
    id: Uuid,
) -> Result<Option<RuleSet>, sqlx::Error> {
    sqlx::query_as::<_, RuleSet>("SELECT * FROM rule_sets WHERE id = $1")
        .bind(id)
        .fetch_optional(pool)
        .await
}

/// Get a rule set by name (active only).
pub async fn get_rule_set_by_name(
    pool: &PgPool,
    name: &str,
) -> Result<Option<RuleSet>, sqlx::Error> {
    sqlx::query_as::<_, RuleSet>(
        "SELECT * FROM rule_sets WHERE name = $1 AND active = true"
    )
    .bind(name)
    .fetch_optional(pool)
    .await
}

/// Insert a new rule set.
pub async fn create_rule_set(
    pool: &PgPool,
    name: &str,
    description: Option<&str>,
    content: &serde_json::Value,
) -> Result<RuleSet, sqlx::Error> {
    sqlx::query_as::<_, RuleSet>(
        r#"INSERT INTO rule_sets (id, name, description, content)
           VALUES (gen_random_uuid(), $1, $2, $3)
           RETURNING *"#,
    )
    .bind(name)
    .bind(description)
    .bind(content)
    .fetch_one(pool)
    .await
}

/// Update an existing active rule set (partial update via COALESCE).
pub async fn update_rule_set(
    pool: &PgPool,
    id: Uuid,
    name: Option<&str>,
    description: Option<&str>,
    content: Option<&serde_json::Value>,
    version: Option<&str>,
) -> Result<Option<RuleSet>, sqlx::Error> {
    sqlx::query_as::<_, RuleSet>(
        r#"UPDATE rule_sets
           SET name        = COALESCE($2, name),
               description = COALESCE($3, description),
               content     = COALESCE($4, content),
               version     = COALESCE($5, version),
               updated_at  = now()
           WHERE id = $1 AND active = true
           RETURNING *"#,
    )
    .bind(id)
    .bind(name)
    .bind(description)
    .bind(content)
    .bind(version)
    .fetch_optional(pool)
    .await
}

/// Soft-delete a rule set by setting active = false.
/// Returns the name of the deactivated rule set, or None if not found.
pub async fn deactivate_rule_set(
    pool: &PgPool,
    id: Uuid,
) -> Result<Option<String>, sqlx::Error> {
    sqlx::query_scalar::<_, String>(
        r#"UPDATE rule_sets
           SET active = false, updated_at = now()
           WHERE id = $1 AND active = true
           RETURNING name"#,
    )
    .bind(id)
    .fetch_optional(pool)
    .await
}

/// Record a rule evaluation for audit purposes.
pub async fn record_evaluation(
    pool: &PgPool,
    rule_set_name: &str,
    input: &serde_json::Value,
    output: &serde_json::Value,
    duration_ms: i32,
    context_type: Option<&str>,
    context_id: Option<Uuid>,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"INSERT INTO rule_evaluations
           (id, rule_set_name, input, output, duration_ms, context_type, context_id)
           VALUES (gen_random_uuid(), $1, $2, $3, $4, $5, $6)"#,
    )
    .bind(rule_set_name)
    .bind(input)
    .bind(output)
    .bind(duration_ms)
    .bind(context_type)
    .bind(context_id)
    .execute(pool)
    .await?;
    Ok(())
}

/// List evaluations with optional filters and pagination.
pub async fn list_evaluations_paged(
    pool: &PgPool,
    limit: i64,
    offset: i64,
    rule_set_name: Option<&str>,
    context_type: Option<&str>,
    context_id: Option<Uuid>,
) -> Result<Vec<RuleEvaluation>, sqlx::Error> {
    sqlx::query_as::<_, RuleEvaluation>(
        r#"SELECT * FROM rule_evaluations
           WHERE ($1::TEXT IS NULL OR rule_set_name = $1)
             AND ($2::TEXT IS NULL OR context_type = $2)
             AND ($3::UUID IS NULL OR context_id = $3)
           ORDER BY created_at DESC
           LIMIT $4 OFFSET $5"#,
    )
    .bind(rule_set_name)
    .bind(context_type)
    .bind(context_id)
    .bind(limit)
    .bind(offset)
    .fetch_all(pool)
    .await
}

/// Count evaluations matching optional filters.
pub async fn count_evaluations(
    pool: &PgPool,
    rule_set_name: Option<&str>,
    context_type: Option<&str>,
    context_id: Option<Uuid>,
) -> Result<i64, sqlx::Error> {
    let row: (i64,) = sqlx::query_as(
        r#"SELECT COUNT(*) FROM rule_evaluations
           WHERE ($1::TEXT IS NULL OR rule_set_name = $1)
             AND ($2::TEXT IS NULL OR context_type = $2)
             AND ($3::UUID IS NULL OR context_id = $3)"#,
    )
    .bind(rule_set_name)
    .bind(context_type)
    .bind(context_id)
    .fetch_one(pool)
    .await?;
    Ok(row.0)
}

use axum::extract::{Path, Query, State};
use axum::http::header;
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Extension, Json, Router};
use canopy_api::{ApiError, AppState, PageResponse};
use serde::Deserialize;
use uuid::Uuid;

use crate::engine::RulesEngine;
use crate::store;

/// Build the rules service route tree.
pub fn routes(engine: RulesEngine) -> Router<AppState> {
    Router::new()
        .route("/v1/rule-sets", get(list_rule_sets).post(create_rule_set))
        .route(
            "/v1/rule-sets/{id}",
            get(get_rule_set).put(update_rule_set).delete(delete_rule_set),
        )
        .route("/v1/rule-sets/{id}/import", post(import_rule_set))
        .route("/v1/rule-sets/{id}/export", get(export_rule_set))
        .route("/v1/evaluate", post(evaluate))
        .route("/v1/evaluations", get(list_evaluations))
        .layer(Extension(engine))
}

async fn list_rule_sets(
    State(app): State<AppState>,
    Query(query): Query<RuleSetQuery>,
) -> Result<Json<PageResponse<store::RuleSet>>, ApiError>

async fn get_rule_set(
    State(app): State<AppState>,
    Path(id): Path<Uuid>,
) -> Result<Json<store::RuleSet>, ApiError>

async fn create_rule_set(
    State(app): State<AppState>,
    Extension(engine): Extension<RulesEngine>,
    Json(body): Json<CreateRuleSetRequest>,
) -> Result<Json<store::RuleSet>, ApiError>

async fn update_rule_set(
    State(app): State<AppState>,
    Extension(engine): Extension<RulesEngine>,
    Path(id): Path<Uuid>,
    Json(body): Json<UpdateRuleSetRequest>,
) -> Result<Json<store::RuleSet>, ApiError>

async fn delete_rule_set(
    State(app): State<AppState>,
    Extension(engine): Extension<RulesEngine>,
    Path(id): Path<Uuid>,
) -> Result<Response, ApiError>

async fn import_rule_set(
    State(app): State<AppState>,
    Extension(engine): Extension<RulesEngine>,
    Path(id): Path<Uuid>,
    Json(content): Json<serde_json::Value>,
) -> Result<Json<store::RuleSet>, ApiError>

async fn export_rule_set(
    State(app): State<AppState>,
    Path(id): Path<Uuid>,
) -> Result<Response, ApiError>

async fn evaluate(
    Extension(engine): Extension<RulesEngine>,
    Json(body): Json<EvaluateRequest>,
) -> Result<Json<EvaluateResponse>, ApiError>

async fn list_evaluations(
    State(app): State<AppState>,
    Query(query): Query<EvaluationQuery>,
) -> Result<Json<PageResponse<store::RuleEvaluation>>, ApiError>

#[derive(Deserialize)]
struct CreateRuleSetRequest {
    name: String,
    description: Option<String>,
    content: serde_json::Value,
}

#[derive(Deserialize)]
struct UpdateRuleSetRequest {
    name: Option<String>,
    description: Option<String>,
    content: Option<serde_json::Value>,
}

#[derive(Deserialize)]
struct EvaluateRequest {
    rule_set_name: String,
    input: serde_json::Value,
    context_type: Option<String>,
    context_id: Option<Uuid>,
}

#[derive(serde::Serialize)]
struct EvaluateResponse {
    rule_set_name: String,
    output: serde_json::Value,
    duration_ms: i32,
    evaluation_id: Uuid,
}

{
  "rule_set_name": "snap-eligibility",
  "context_type": "application",
  "context_id": "019513a2-b3f4-7000-8000-000000000001",
  "input": {
    "gross_monthly_income": 1800,
    "household_size": 3,
    "countable_assets": 1500
  }
}

{
  "rule_set_name": "snap-eligibility",
  "output": {
    "result": "pass",
    "eligible": true,
    "basis": "gross_income_under_130pct_fpl"
  },
  "duration_ms": 2,
  "evaluation_id": "019513b1-c4e5-7000-8000-000000000002"
}

{
  "name": "georgia-snap-eligibility",
  "description": "SNAP eligibility: gross income, net income, asset tests",
  "content": { "nodes": [...], "edges": [...] }
}

{
  "id": "019513a0-a1b2-7000-8000-000000000001",
  "name": "georgia-snap-eligibility",
  "description": "SNAP eligibility: gross income, net income, asset tests",
  "version": "v1.0",
  "content": { "nodes": [...], "edges": [...] },
  "active": true,
  "created_at": "2026-03-26T14:30:00Z",
  "updated_at": "2026-03-26T14:30:00Z"
}

// In main.rs after engine initialization:

let ruleset_dir = format!("rulesets/{}", settings.jurisdiction);
if let Ok(entries) = std::fs::read_dir(&ruleset_dir) {
    for entry in entries.flatten() {
        let path = entry.path();
        if path.extension().map_or(false, |e| e == "json") {
            let raw = std::fs::read_to_string(&path)
                .with_context(|| format!("failed to read {}", path.display()))?;
            let content: serde_json::Value = serde_json::from_str(&raw)
                .with_context(|| format!("failed to parse {}", path.display()))?;

            let name = content["name"]
                .as_str()
                .unwrap_or_else(|| {
                    path.file_stem()
                        .and_then(|s| s.to_str())
                        .unwrap_or("unknown")
                });

            // Only import if not already in the database.
            if store::get_rule_set_by_name(boot.db.inner(), name).await?.is_none() {
                let description = content["_comment"].as_str();
                store::create_rule_set(boot.db.inner(), name, description, &content).await
                    .with_context(|| format!("failed to import ruleset '{name}'"))?;
                // Compile and insert into the in-memory cache.
                let decision = engine::RulesEngine::compile_rule_set(&content)
                    .with_context(|| format!("failed to compile ruleset '{name}'"))?;
                rules_engine.insert_decision(name.to_owned(), decision).await;
                info!(name, path = %path.display(), "imported ruleset from disk");
            }
        }
    }
}

// SPDX-License-Identifier: AGPL-3.0-or-later

#![forbid(unsafe_code)]

mod api;
mod engine;
mod events;
mod store;

use anyhow::Context;
use canopy_api::{ApiServer, AppState, ServerOptions, shutdown_signal};
use canopy_mq::EventEnvelope;
use tracing::info;
use uuid::Uuid;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let (settings, boot) = canopy_api::bootstrap("CANOPY_RULES", "canopy-rules").await?;

    // Run database migrations.
    boot.db
        .run_migrations(&sqlx::migrate!())
        .await
        .context("failed to run database migrations")?;

    // Initialize rules engine (spawns dedicated zen-eval thread).
    let rules_engine = engine::RulesEngine::new(
        boot.db.clone(),
        Some(boot.publisher),
        &settings.jurisdiction,
    )
    .await
    .context("failed to initialize rules engine")?;

    // Import rulesets from disk (rulesets/{jurisdiction}/*.json).
    let ruleset_dir = format!("rulesets/{}", settings.jurisdiction);
    if let Ok(entries) = std::fs::read_dir(&ruleset_dir) {
        for entry in entries.flatten() {
            let path = entry.path();
            if path.extension().map_or(false, |e| e == "json") {
                let raw = std::fs::read_to_string(&path)
                    .with_context(|| format!("failed to read {}", path.display()))?;
                let content: serde_json::Value = serde_json::from_str(&raw)
                    .with_context(|| format!("failed to parse {}", path.display()))?;

                let name = content["name"]
                    .as_str()
                    .unwrap_or_else(|| {
                        path.file_stem()
                            .and_then(|s| s.to_str())
                            .unwrap_or("unknown")
                    });

                if store::get_rule_set_by_name(boot.db.inner(), name)
                    .await?
                    .is_none()
                {
                    let description = content["_comment"].as_str();
                    store::create_rule_set(boot.db.inner(), name, description, &content)
                        .await
                        .with_context(|| format!("failed to import ruleset '{name}'"))?;
                    let decision = engine::RulesEngine::compile_rule_set(&content)
                        .with_context(|| format!("failed to compile ruleset '{name}'"))?;
                    rules_engine
                        .insert_decision(name.to_owned(), decision)
                        .await;
                    info!(name, path = %path.display(), "imported ruleset from disk");
                }
            }
        }
    }

    // Subscribe to cache invalidation events on a per-instance exclusive queue.
    let cache_queue = format!("canopy-rules.cache.{}", Uuid::now_v7());
    let engine_for_cache = rules_engine.clone();
    let my_instance_id = rules_engine.instance_id().to_owned();
    let _cache_sub_handle = boot
        .subscriber
        .subscribe_exclusive(
            &cache_queue,
            &["rules.cache_invalidated"],
            move |envelope: EventEnvelope| {
                let engine = engine_for_cache.clone();
                let my_id = my_instance_id.clone();
                async move {
                    let from_self = envelope
                        .payload
                        .get("instance_id")
                        .and_then(|v| v.as_str())
                        .is_some_and(|id| id == my_id);
                    if from_self {
                        return Ok(());
                    }
                    info!("received cache invalidation event, reloading rule sets");
                    engine.reload_all().await?;
                    Ok(())
                }
            },
        )
        .await
        .context("failed to start cache invalidation subscriber")?;

    // Build Axum router.
    let state = AppState {
        db: boot.db.clone(),
        auth: boot.auth,
    };
    let service_routes = api::routes(rules_engine);
    let router = ApiServer::router(
        state,
        service_routes,
        ServerOptions {
            cors_origins: settings.cors_origins,
            body_limit: settings.body_limit,
        },
        None,
    )
    .layer(axum::Extension(boot.mq_health));

    ApiServer::serve(router, settings.port, shutdown_signal()).await?;
    Ok(())
}

#[tokio::test]
async fn evaluate_pass_through_ruleset() {
    let (db, _) = canopy_test::setup_db().await;
    let engine = RulesEngine::new(db.clone(), None, "georgia")
        .await
        .expect("engine init");

    // Load snap-eligibility.json stub into the engine.
    let content: serde_json::Value = serde_json::from_str(
        &std::fs::read_to_string("../../rulesets/georgia/snap-eligibility.json").unwrap(),
    )
    .unwrap();
    let name = content["name"].as_str().unwrap();
    store::create_rule_set(db.inner(), name, None, &content)
        .await
        .unwrap();
    let decision = RulesEngine::compile_rule_set(&content).unwrap();
    engine.insert_decision(name.to_owned(), decision).await;

    // Evaluate with empty input -- stub returns {"result": "pass"}.
    let output = engine
        .evaluate(name, None, None, serde_json::json!({}))
        .await
        .expect("evaluation should succeed");

    assert_eq!(output["result"], "pass");
}

#[tokio::test]
async fn evaluation_recorded_in_audit_trail() {
    let (db, _) = canopy_test::setup_db().await;
    let engine = RulesEngine::new(db.clone(), None, "georgia")
        .await
        .expect("engine init");

    // Load and evaluate a stub ruleset.
    let content: serde_json::Value = serde_json::from_str(
        &std::fs::read_to_string("../../rulesets/georgia/snap-eligibility.json").unwrap(),
    )
    .unwrap();
    let name = content["name"].as_str().unwrap();
    store::create_rule_set(db.inner(), name, None, &content)
        .await
        .unwrap();
    let decision = RulesEngine::compile_rule_set(&content).unwrap();
    engine.insert_decision(name.to_owned(), decision).await;

    engine
        .evaluate(
            name,
            Some("application"),
            Some(Uuid::now_v7()),
            serde_json::json!({"gross_monthly_income": 1800}),
        )
        .await
        .unwrap();

    // Verify the evaluation was recorded.
    let evals = store::list_evaluations_paged(
        db.inner(), 10, 0, Some(name), None, None,
    )
    .await
    .unwrap();

    assert_eq!(evals.len(), 1);
    assert_eq!(evals[0].rule_set_name, name);
    assert!(evals[0].duration_ms >= 0);
    assert_eq!(evals[0].context_type.as_deref(), Some("application"));
}

#[tokio::test]
async fn crud_rule_set_lifecycle() {
    let (db, _) = canopy_test::setup_db().await;
    let pool = db.inner();

    // Create
    let content = serde_json::json!({"nodes": [], "edges": []});
    let created = store::create_rule_set(pool, "test-ruleset", Some("A test"), &content)
        .await
        .unwrap();
    assert_eq!(created.name, "test-ruleset");
    assert!(created.active);

    // Get
    let fetched = store::get_rule_set(pool, created.id).await.unwrap().unwrap();
    assert_eq!(fetched.name, "test-ruleset");

    // Update
    let updated = store::update_rule_set(
        pool,
        created.id,
        Some("renamed-ruleset"),
        Some("Updated desc"),
        None,
        Some("v2.0"),
    )
    .await
    .unwrap()
    .unwrap();
    assert_eq!(updated.name, "renamed-ruleset");
    assert_eq!(updated.version, "v2.0");

    // List -- should find the updated rule set.
    let list = store::list_rule_sets_paged(pool, 10, 0, None, None, None)
        .await
        .unwrap();
    assert_eq!(list.len(), 1);
    assert_eq!(list[0].name, "renamed-ruleset");

    // Soft-delete
    let deleted_name = store::deactivate_rule_set(pool, created.id)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(deleted_name, "renamed-ruleset");

    // Verify gone from active list.
    let list_after = store::list_rule_sets_paged(pool, 10, 0, None, None, None)
        .await
        .unwrap();
    assert!(list_after.is_empty());
}

#[tokio::test]
async fn import_replaces_content_and_bumps_version() {
    let (db, _) = canopy_test::setup_db().await;
    let pool = db.inner();

    // Create initial rule set.
    let v1_content = serde_json::json!({"nodes": [], "edges": []});
    let created = store::create_rule_set(pool, "import-test", None, &v1_content)
        .await
        .unwrap();
    assert_eq!(created.version, "v1.0");

    // Import new content with a new version (simulates the import handler).
    let v2_content = serde_json::json!({"nodes": [{"id": "new"}], "edges": []});
    let version = chrono::Utc::now().format("%Y%m%d%H%M%S").to_string();
    let updated = store::update_rule_set(
        pool,
        created.id,
        None,
        None,
        Some(&v2_content),
        Some(&version),
    )
    .await
    .unwrap()
    .unwrap();

    assert_eq!(updated.version, version);
    assert_eq!(updated.content["nodes"][0]["id"], "new");
}

#[tokio::test]
async fn cache_invalidation_event_published() {
    let (db, _) = canopy_test::setup_db().await;

    // Use a mock publisher that captures published events.
    let (mock_pub, rx) = canopy_test::mock_publisher();

    let engine = RulesEngine::new(db, Some(mock_pub), "georgia")
        .await
        .expect("engine init");

    engine
        .notify_cache_invalidated("snap-eligibility", "created")
        .await;

    // Verify the event was published.
    let envelope = rx.try_recv().expect("should have received an event");
    assert_eq!(envelope.event_type, "rules.cache_invalidated");
    assert_eq!(
        envelope.payload["rule_set_name"].as_str().unwrap(),
        "snap-eligibility"
    );
    assert_eq!(
        envelope.payload["action"].as_str().unwrap(),
        "created"
    );
    assert_eq!(
        envelope.payload["instance_id"].as_str().unwrap(),
        engine.instance_id()
    );
}