Plan: Eligibility Orchestrator

-- Tracks applications received for eligibility determination.
-- canopy-eligibility does not own the application itself (canopy-applications does);
-- this table records the orchestration lifecycle.
CREATE TABLE eligibility_requests (
    id UUID PRIMARY KEY,
    application_id UUID NOT NULL,
    household_id UUID NOT NULL,
    programs_requested TEXT[] NOT NULL,  -- e.g., {'snap', 'tanf', 'medicaid'}
    status TEXT NOT NULL DEFAULT 'pending',  -- pending, in_progress, completed, failed
    requested_by TEXT NOT NULL,
    requested_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    completed_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Individual program determinations received from program services.
-- Each row is the signed determination object from one program service.
CREATE TABLE program_determinations (
    id UUID PRIMARY KEY,
    eligibility_request_id UUID NOT NULL REFERENCES eligibility_requests(id),
    program TEXT NOT NULL,
    application_id UUID NOT NULL,
    household_id UUID NOT NULL,
    status TEXT NOT NULL,  -- approved, denied, pending_verification
    benefit_amount NUMERIC(10,2),
    benefit_unit TEXT,
    effective_date DATE,
    expiration_date DATE,
    renewal_date DATE,
    basis TEXT,
    program_service_version TEXT NOT NULL,
    determined_at TIMESTAMPTZ NOT NULL,
    signature TEXT NOT NULL,  -- the detached JWS, stored verbatim
    signature_verified BOOLEAN NOT NULL DEFAULT false,
    received_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Combined result after all program determinations are assembled
-- and the eligibility hierarchy has been applied.
CREATE TABLE combined_results (
    id UUID PRIMARY KEY,
    eligibility_request_id UUID NOT NULL REFERENCES eligibility_requests(id),
    application_id UUID NOT NULL,
    household_id UUID NOT NULL,
    programs_approved TEXT[] NOT NULL DEFAULT '{}',
    programs_denied TEXT[] NOT NULL DEFAULT '{}',
    programs_pending TEXT[] NOT NULL DEFAULT '{}',
    medicaid_assigned_group TEXT,  -- EE15 most advantageous group, if applicable
    total_monthly_benefit NUMERIC(10,2),
    assembled_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX idx_eligibility_requests_application ON eligibility_requests(application_id);
CREATE INDEX idx_program_determinations_request ON program_determinations(eligibility_request_id);
CREATE INDEX idx_combined_results_request ON combined_results(eligibility_request_id);
CREATE INDEX idx_combined_results_application ON combined_results(application_id);

/// Request body for POST /v1/eligibility/determine.
/// Submitted by canopy-applications after application intake is complete.
#[derive(Debug, Deserialize)]
pub struct DetermineRequest {
    pub application_id: Uuid,
    pub household_id: Uuid,
    pub programs: Vec<Program>,
}

/// Response for POST /v1/eligibility/determine.
#[derive(Debug, Serialize)]
pub struct DetermineResponse {
    pub request_id: Uuid,
    pub application_id: Uuid,
    pub programs_approved: Vec<ProgramResult>,
    pub programs_denied: Vec<ProgramResult>,
    pub programs_pending: Vec<ProgramResult>,
    pub medicaid_assigned_group: Option<String>,
    pub total_monthly_benefit: Decimal,
    pub assembled_at: DateTime<Utc>,
}

#[derive(Debug, Serialize)]
pub struct ProgramResult {
    pub program: Program,
    pub status: DeterminationStatus,
    pub benefit_amount: Option<Decimal>,
    pub basis: Option<String>,
    pub effective_date: Option<NaiveDate>,
}

/// Sent by canopy-eligibility to each program service.
#[derive(Debug, Serialize)]
pub struct ApplicationContext {
    pub application_id: Uuid,
    pub household_id: Uuid,
    pub applicant_person_id: Uuid,
    pub household_member_ids: Vec<Uuid>,
    /// Self-reported income, assets, expenses — as references.
    /// The program service fetches full records from canopy-persons.
    pub income_ids: Vec<Uuid>,
    pub asset_ids: Vec<Uuid>,
    pub expense_ids: Vec<Uuid>,
}

POST /v1/eligibility/determine
│
├── 1. Validate request, create eligibility_request row (status: pending)
│
├── 2. Build ApplicationContext from canopy-applications data
│      (GET /v1/applications/{id} to resolve household/person references)
│
├── 3. Determine target program services from request.programs
│      Look up base URLs in ProgramServiceRegistry
│
├── 4. Dispatch to program services in parallel
│      For each program:
│        POST {base_url}/v1/determine  with ApplicationContext body
│        Timeout: 30 seconds per service (configurable)
│        Circuit breaker: 5 failures in 60 seconds trips the breaker
│
├── 5. Collect Determination responses
│      For each response:
│        Verify JWS signature via DeterminationVerifier
│        If verification fails: log error, mark program as failed
│        If verification passes: persist to program_determinations
│
├── 6. Apply eligibility hierarchy (EE15) if Medicaid is among results
│      Calls canopy-rules with medicaid-eligibility-hierarchy ruleset
│      Input: all Medicaid-eligible categories from the determination
│      Output: most advantageous group assignment
│
├── 7. Assemble combined result
│      Categorize programs into approved/denied/pending
│      Sum total monthly benefit across approved programs
│      Persist to combined_results
│      Update eligibility_request status to completed
│
├── 8. Publish determination.completed event
│      Payload: { request_id, application_id, household_id,
│                 programs_approved[], programs_denied[], timestamp }
│      NO benefit amounts, NO determination bases in event payload (ADR-004)
│
└── 9. Return DetermineResponse

/// Maps programs to their service base URLs.
/// Loaded from environment variables at startup.
pub struct ProgramServiceRegistry {
    services: HashMap<Program, ProgramServiceConfig>,
}

pub struct ProgramServiceConfig {
    pub base_url: String,
    pub timeout: Duration,
}

impl ProgramServiceRegistry {
    /// Load from environment.
    /// CANOPY_PROGRAM_URL_SNAP=http://canopy-snap:8003
    /// CANOPY_PROGRAM_URL_TANF=http://canopy-tanf:8004
    /// CANOPY_PROGRAM_URL_MEDICAID=http://canopy-medicaid:8005
    pub fn from_env() -> Result<Self, anyhow::Error>;
}

pub struct CircuitBreaker {
    failure_count: AtomicU32,
    last_failure: AtomicI64,  -- unix timestamp
    state: AtomicU8,          -- 0=closed, 1=open, 2=half-open
}

impl CircuitBreaker {
    pub fn new(failure_threshold: u32, recovery_timeout: Duration) -> Self;
    pub fn can_call(&self) -> bool;
    pub fn record_success(&self);
    pub fn record_failure(&self);
}

canopy-applications    canopy-eligibility       canopy-snap       canopy-tanf       canopy-rules
       │                      │                     │                  │                 │
       │  POST /determine     │                     │                  │                 │
       │─────────────────────>│                     │                  │                 │
       │                      │  POST /v1/determine │                  │                 │
       │                      │────────────────────>│                  │                 │
       │                      │  POST /v1/determine │                  │                 │
       │                      │───────────────────────────────────────>│                 │
       │                      │                     │                  │                 │
       │                      │                     │  POST /evaluate  │                 │
       │                      │                     │────────────────────────────────────>│
       │                      │                     │  ruleset result  │                 │
       │                      │                     │<───────────────────────────────────│
       │                      │                     │                  │                 │
       │                      │                     │                  │  POST /evaluate │
       │                      │                     │                  │────────────────>│
       │                      │                     │                  │  ruleset result │
       │                      │                     │                  │<───────────────│
       │                      │                     │                  │                 │
       │                      │  Determination(JWS) │                  │                 │
       │                      │<────────────────────│                  │                 │
       │                      │  Determination(JWS) │                  │                 │
       │                      │<──────────────────────────────────────│                 │
       │                      │                     │                  │                 │
       │                      │  verify signatures  │                  │                 │
       │                      │  apply EE15 hierarchy                  │                 │
       │                      │  persist combined result               │                 │
       │                      │  publish determination.completed       │                 │
       │                      │                     │                  │                 │
       │  DetermineResponse   │                     │                  │                 │
       │<─────────────────────│                     │                  │                 │

-- Migration: 20260326000000_create_eligibility_tables.sql
-- Creates the three core tables for the eligibility orchestrator.

CREATE TABLE eligibility_requests (
    id UUID PRIMARY KEY,
    application_id UUID NOT NULL,
    household_id UUID NOT NULL,
    programs_requested TEXT[] NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending',
    requested_by TEXT NOT NULL,
    requested_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    completed_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE program_determinations (
    id UUID PRIMARY KEY,
    eligibility_request_id UUID NOT NULL REFERENCES eligibility_requests(id),
    program TEXT NOT NULL,
    application_id UUID NOT NULL,
    household_id UUID NOT NULL,
    status TEXT NOT NULL,
    benefit_amount NUMERIC(10,2),
    benefit_unit TEXT,
    effective_date DATE,
    expiration_date DATE,
    renewal_date DATE,
    basis TEXT,
    program_service_version TEXT NOT NULL,
    determined_at TIMESTAMPTZ NOT NULL,
    signature TEXT NOT NULL,
    signature_verified BOOLEAN NOT NULL DEFAULT false,
    received_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE combined_results (
    id UUID PRIMARY KEY,
    eligibility_request_id UUID NOT NULL REFERENCES eligibility_requests(id),
    application_id UUID NOT NULL,
    household_id UUID NOT NULL,
    programs_approved TEXT[] NOT NULL DEFAULT '{}',
    programs_denied TEXT[] NOT NULL DEFAULT '{}',
    programs_pending TEXT[] NOT NULL DEFAULT '{}',
    medicaid_assigned_group TEXT,
    total_monthly_benefit NUMERIC(10,2),
    assembled_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Indexes for query performance
CREATE INDEX idx_eligibility_requests_application ON eligibility_requests(application_id);
CREATE INDEX idx_eligibility_requests_household ON eligibility_requests(household_id);
CREATE INDEX idx_eligibility_requests_status ON eligibility_requests(status);
CREATE INDEX idx_program_determinations_request ON program_determinations(eligibility_request_id);
CREATE INDEX idx_program_determinations_program ON program_determinations(program);
CREATE INDEX idx_program_determinations_application ON program_determinations(application_id);
CREATE INDEX idx_combined_results_request ON combined_results(eligibility_request_id);
CREATE INDEX idx_combined_results_application ON combined_results(application_id);
CREATE INDEX idx_combined_results_household ON combined_results(household_id);

// services/canopy-eligibility/src/store/models.rs

use chrono::{DateTime, NaiveDate, Utc};
use rust_decimal::Decimal;
use serde::{Deserialize, Serialize};
use uuid::Uuid;

#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct EligibilityRequest {
    pub id: Uuid,
    pub application_id: Uuid,
    pub household_id: Uuid,
    pub programs_requested: Vec<String>,
    pub status: String,
    pub requested_by: String,
    pub requested_at: DateTime<Utc>,
    pub completed_at: Option<DateTime<Utc>>,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
}

#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct ProgramDetermination {
    pub id: Uuid,
    pub eligibility_request_id: Uuid,
    pub program: String,
    pub application_id: Uuid,
    pub household_id: Uuid,
    pub status: String,
    pub benefit_amount: Option<Decimal>,
    pub benefit_unit: Option<String>,
    pub effective_date: Option<NaiveDate>,
    pub expiration_date: Option<NaiveDate>,
    pub renewal_date: Option<NaiveDate>,
    pub basis: Option<String>,
    pub program_service_version: String,
    pub determined_at: DateTime<Utc>,
    pub signature: String,
    pub signature_verified: bool,
    pub received_at: DateTime<Utc>,
    pub created_at: DateTime<Utc>,
}

#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct CombinedResult {
    pub id: Uuid,
    pub eligibility_request_id: Uuid,
    pub application_id: Uuid,
    pub household_id: Uuid,
    pub programs_approved: Vec<String>,
    pub programs_denied: Vec<String>,
    pub programs_pending: Vec<String>,
    pub medicaid_assigned_group: Option<String>,
    pub total_monthly_benefit: Option<Decimal>,
    pub assembled_at: DateTime<Utc>,
    pub created_at: DateTime<Utc>,
}

// services/canopy-eligibility/src/store/eligibility.rs

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

use super::models::{CombinedResult, EligibilityRequest, ProgramDetermination};

pub async fn create_eligibility_request(
    pool: &PgPool,
    id: Uuid,
    application_id: Uuid,
    household_id: Uuid,
    programs_requested: &[String],
    requested_by: &str,
) -> Result<EligibilityRequest, sqlx::Error> {
    sqlx::query_as::<_, EligibilityRequest>(
        r#"INSERT INTO eligibility_requests
           (id, application_id, household_id, programs_requested, status, requested_by)
           VALUES ($1, $2, $3, $4, 'pending', $5)
           RETURNING *"#,
    )
    .bind(id)
    .bind(application_id)
    .bind(household_id)
    .bind(programs_requested)
    .bind(requested_by)
    .fetch_one(pool)
    .await
}

pub async fn update_request_status(
    pool: &PgPool,
    id: Uuid,
    status: &str,
    completed_at: Option<DateTime<Utc>>,
) -> Result<EligibilityRequest, sqlx::Error> {
    sqlx::query_as::<_, EligibilityRequest>(
        r#"UPDATE eligibility_requests
           SET status = $2, completed_at = $3, updated_at = now()
           WHERE id = $1
           RETURNING *"#,
    )
    .bind(id)
    .bind(status)
    .bind(completed_at)
    .fetch_one(pool)
    .await
}

pub async fn insert_program_determination(
    pool: &PgPool,
    det: &ProgramDetermination,
) -> Result<ProgramDetermination, sqlx::Error> {
    sqlx::query_as::<_, ProgramDetermination>(
        r#"INSERT INTO program_determinations
           (id, eligibility_request_id, program, application_id, household_id,
            status, benefit_amount, benefit_unit, effective_date, expiration_date,
            renewal_date, basis, program_service_version, determined_at,
            signature, signature_verified)
           VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
           RETURNING *"#,
    )
    .bind(det.id)
    .bind(det.eligibility_request_id)
    .bind(&det.program)
    .bind(det.application_id)
    .bind(det.household_id)
    .bind(&det.status)
    .bind(det.benefit_amount)
    .bind(&det.benefit_unit)
    .bind(det.effective_date)
    .bind(det.expiration_date)
    .bind(det.renewal_date)
    .bind(&det.basis)
    .bind(&det.program_service_version)
    .bind(det.determined_at)
    .bind(&det.signature)
    .bind(det.signature_verified)
    .fetch_one(pool)
    .await
}

pub async fn insert_combined_result(
    pool: &PgPool,
    result: &CombinedResult,
) -> Result<CombinedResult, sqlx::Error> {
    sqlx::query_as::<_, CombinedResult>(
        r#"INSERT INTO combined_results
           (id, eligibility_request_id, application_id, household_id,
            programs_approved, programs_denied, programs_pending,
            medicaid_assigned_group, total_monthly_benefit)
           VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
           RETURNING *"#,
    )
    .bind(result.id)
    .bind(result.eligibility_request_id)
    .bind(result.application_id)
    .bind(result.household_id)
    .bind(&result.programs_approved)
    .bind(&result.programs_denied)
    .bind(&result.programs_pending)
    .bind(&result.medicaid_assigned_group)
    .bind(result.total_monthly_benefit)
    .fetch_one(pool)
    .await
}

pub async fn get_eligibility_request(
    pool: &PgPool,
    id: Uuid,
) -> Result<Option<EligibilityRequest>, sqlx::Error> {
    sqlx::query_as::<_, EligibilityRequest>(
        "SELECT * FROM eligibility_requests WHERE id = $1",
    )
    .bind(id)
    .fetch_optional(pool)
    .await
}

pub async fn list_determinations_for_request(
    pool: &PgPool,
    eligibility_request_id: Uuid,
) -> Result<Vec<ProgramDetermination>, sqlx::Error> {
    sqlx::query_as::<_, ProgramDetermination>(
        "SELECT * FROM program_determinations WHERE eligibility_request_id = $1 ORDER BY created_at",
    )
    .bind(eligibility_request_id)
    .fetch_all(pool)
    .await
}

pub async fn get_combined_result_for_application(
    pool: &PgPool,
    application_id: Uuid,
) -> Result<Option<CombinedResult>, sqlx::Error> {
    sqlx::query_as::<_, CombinedResult>(
        "SELECT * FROM combined_results WHERE application_id = $1 ORDER BY created_at DESC LIMIT 1",
    )
    .bind(application_id)
    .fetch_optional(pool)
    .await
}

// services/canopy-eligibility/src/registry.rs

use std::collections::HashMap;
use std::time::Duration;

use canopy_reference::Program;

pub struct ProgramServiceRegistry {
    services: HashMap<Program, ProgramServiceConfig>,
}

pub struct ProgramServiceConfig {
    pub program: Program,
    pub base_url: String,
    pub timeout: Duration,
    pub client: reqwest::Client,
}

impl ProgramServiceRegistry {
    /// Load from environment variables.
    /// Reads CANOPY_PROGRAM_URL_{PROGRAM} for each known program.
    /// Programs without a configured URL are silently skipped (not all
    /// programs may be deployed in every environment).
    pub fn from_env() -> Result<Self, anyhow::Error> {
        let mut services = HashMap::new();

        let programs = [
            (Program::Snap, "CANOPY_PROGRAM_URL_SNAP"),
            (Program::Tanf, "CANOPY_PROGRAM_URL_TANF"),
            (Program::Medicaid, "CANOPY_PROGRAM_URL_MEDICAID"),
            (Program::Chip, "CANOPY_PROGRAM_URL_CHIP"),
            (Program::Caps, "CANOPY_PROGRAM_URL_CAPS"),
            (Program::Wic, "CANOPY_PROGRAM_URL_WIC"),
        ];

        for (program, env_var) in programs {
            if let Ok(base_url) = std::env::var(env_var) {
                let timeout = Duration::from_secs(
                    std::env::var(format!("{env_var}_TIMEOUT_SECS"))
                        .ok()
                        .and_then(|s| s.parse().ok())
                        .unwrap_or(30),
                );
                let client = reqwest::Client::builder()
                    .timeout(timeout)
                    .build()?;
                services.insert(program, ProgramServiceConfig {
                    program,
                    base_url,
                    timeout,
                    client,
                });
            }
        }

        if services.is_empty() {
            anyhow::bail!("no program service URLs configured; set at least one CANOPY_PROGRAM_URL_* env var");
        }

        Ok(Self { services })
    }

    /// Look up the configuration for a program.
    /// Returns None if the program service is not registered.
    pub fn get(&self, program: &Program) -> Option<&ProgramServiceConfig> {
        self.services.get(program)
    }

    /// Dispatch a determination request to a specific program service.
    pub async fn determine(
        &self,
        program: Program,
        context: &crate::orchestrator::ApplicationContext,
    ) -> Result<crate::determination::Determination, crate::errors::ApiError> {
        let config = self.get(&program).ok_or_else(|| {
            crate::errors::ApiError::Internal(format!(
                "no service registered for program: {program:?}"
            ))
        })?;

        let url = format!("{}/v1/determine", config.base_url);
        let response = config
            .client
            .post(&url)
            .json(context)
            .send()
            .await
            .map_err(|e| crate::errors::ApiError::ProgramService(format!(
                "{program:?} request failed: {e}"
            )))?;

        if !response.status().is_success() {
            let status = response.status();
            let body = response.text().await.unwrap_or_default();
            return Err(crate::errors::ApiError::ProgramService(format!(
                "{program:?} returned {status}: {body}"
            )));
        }

        response
            .json()
            .await
            .map_err(|e| crate::errors::ApiError::ProgramService(format!(
                "{program:?} response parse failed: {e}"
            )))
    }
}

// services/canopy-eligibility/src/client.rs

use std::sync::atomic::{AtomicI64, AtomicU32, AtomicU8, Ordering};
use std::time::Duration;

pub struct CircuitBreaker {
    failure_count: AtomicU32,
    last_failure: AtomicI64,
    state: AtomicU8,  // 0=closed, 1=open, 2=half-open
    failure_threshold: u32,
    recovery_timeout: Duration,
}

impl CircuitBreaker {
    pub fn new(failure_threshold: u32, recovery_timeout: Duration) -> Self {
        Self {
            failure_count: AtomicU32::new(0),
            last_failure: AtomicI64::new(0),
            state: AtomicU8::new(0),
            failure_threshold,
            recovery_timeout,
        }
    }

    pub fn can_call(&self) -> bool {
        match self.state.load(Ordering::Acquire) {
            0 => true,  // closed — allow calls
            1 => {
                // open — check if recovery timeout has elapsed
                let last = self.last_failure.load(Ordering::Acquire);
                let now = chrono::Utc::now().timestamp();
                if now - last > self.recovery_timeout.as_secs() as i64 {
                    self.state.store(2, Ordering::Release);  // transition to half-open
                    true
                } else {
                    false
                }
            }
            2 => true,  // half-open — allow one probe call
            _ => false,
        }
    }

    pub fn record_success(&self) {
        self.failure_count.store(0, Ordering::Release);
        self.state.store(0, Ordering::Release);
    }

    pub fn record_failure(&self) {
        let count = self.failure_count.fetch_add(1, Ordering::AcqRel) + 1;
        self.last_failure.store(
            chrono::Utc::now().timestamp(),
            Ordering::Release,
        );
        if count >= self.failure_threshold {
            self.state.store(1, Ordering::Release);  // open the breaker
        }
    }
}

// services/canopy-eligibility/src/orchestrator.rs

use std::sync::Arc;
use chrono::Utc;
use rust_decimal::Decimal;
use sqlx::PgPool;
use uuid::Uuid;

use crate::determination::{Determination, DeterminationVerifier};
use crate::errors::{ApiError, OrchestratorError};
use crate::events::Publisher;
use crate::registry::ProgramServiceRegistry;
use crate::store;

pub async fn orchestrate(
    request: DetermineRequest,
    registry: &ProgramServiceRegistry,
    verifier: &dyn DeterminationVerifier,
    db: &PgPool,
    publisher: &Publisher,
) -> Result<DetermineResponse, OrchestratorError> {
    // 1. Create eligibility request row
    let request_id = Uuid::new_v4();
    let programs_str: Vec<String> = request.programs.iter()
        .map(|p| format!("{p:?}").to_lowercase())
        .collect();
    let elig_request = store::eligibility::create_eligibility_request(
        db,
        request_id,
        request.application_id,
        request.household_id,
        &programs_str,
        "canopy-applications",
    ).await.map_err(OrchestratorError::Store)?;

    // 2. Build application context
    let context = ApplicationContext {
        application_id: request.application_id,
        household_id: request.household_id,
        applicant_person_id: request.applicant_person_id,
        household_member_ids: request.household_member_ids.clone(),
        income_ids: request.income_ids.clone(),
        asset_ids: request.asset_ids.clone(),
        expense_ids: request.expense_ids.clone(),
    };

    // 3. Update status to in_progress
    store::eligibility::update_request_status(db, request_id, "in_progress", None)
        .await
        .map_err(OrchestratorError::Store)?;

    // 4. Dispatch to program services in parallel using JoinSet
    let mut join_set = tokio::task::JoinSet::new();
    for program in &request.programs {
        let registry = registry.clone();
        let context = context.clone();
        let program = *program;
        join_set.spawn(async move {
            (program, registry.determine(program, &context).await)
        });
    }

    // 5. Collect results, verify signatures
    let mut determinations = Vec::new();
    let mut failed_programs = Vec::new();
    while let Some(result) = join_set.join_next().await {
        match result {
            Ok((program, Ok(determination))) => {
                // Verify JWS signature
                match verifier.verify(&determination) {
                    Ok(true) => {
                        determinations.push(determination);
                    }
                    Ok(false) => {
                        tracing::error!(
                            program = ?program,
                            "signature verification failed for determination"
                        );
                        return Err(OrchestratorError::SignatureVerification(format!(
                            "signature verification failed for {program:?} determination"
                        )));
                    }
                    Err(e) => {
                        tracing::error!(
                            program = ?program,
                            error = %e,
                            "signature verification error"
                        );
                        return Err(OrchestratorError::SignatureVerification(format!(
                            "signature verification error for {program:?}: {e}"
                        )));
                    }
                }
            }
            Ok((program, Err(e))) => {
                tracing::error!(
                    program = ?program,
                    error = %e,
                    "program determination failed"
                );
                failed_programs.push(program);
            }
            Err(e) => {
                tracing::error!(error = %e, "join error in program dispatch");
            }
        }
    }

    // 6. Persist program determinations
    for det in &determinations {
        let pd = store::models::ProgramDetermination::from_determination(
            det,
            elig_request.id,
            true, // signature_verified
        );
        store::eligibility::insert_program_determination(db, &pd)
            .await
            .map_err(OrchestratorError::Store)?;
    }

    // 7. Apply EE15 hierarchy (placeholder for Phase 2 — SNAP only)
    let medicaid_assigned_group: Option<String> = None;
    // See Step 4 for full EE15 implementation in Phase 4.

    // 8. Assemble combined result
    let mut programs_approved = Vec::new();
    let mut programs_denied = Vec::new();
    let mut programs_pending = Vec::new();
    let mut total_benefit = Decimal::ZERO;

    for det in &determinations {
        let result = ProgramResult {
            program: det.program,
            status: det.status.clone(),
            benefit_amount: det.benefit_amount,
            basis: det.basis.clone(),
            effective_date: det.effective_date,
        };
        match det.status.as_str() {
            "approved" => {
                if let Some(amount) = det.benefit_amount {
                    total_benefit += amount;
                }
                programs_approved.push(result);
            }
            "denied" => programs_denied.push(result),
            _ => programs_pending.push(result),
        }
    }

    // Add failed programs as pending
    for program in &failed_programs {
        programs_pending.push(ProgramResult {
            program: *program,
            status: "pending_verification".to_string(),
            benefit_amount: None,
            basis: None,
            effective_date: None,
        });
    }

    let now = Utc::now();

    // Persist combined result
    let combined = store::models::CombinedResult {
        id: Uuid::new_v4(),
        eligibility_request_id: elig_request.id,
        application_id: request.application_id,
        household_id: request.household_id,
        programs_approved: programs_approved.iter().map(|r| format!("{:?}", r.program).to_lowercase()).collect(),
        programs_denied: programs_denied.iter().map(|r| format!("{:?}", r.program).to_lowercase()).collect(),
        programs_pending: programs_pending.iter().map(|r| format!("{:?}", r.program).to_lowercase()).collect(),
        medicaid_assigned_group: medicaid_assigned_group.clone(),
        total_monthly_benefit: Some(total_benefit),
        assembled_at: now,
        created_at: now,
    };
    store::eligibility::insert_combined_result(db, &combined)
        .await
        .map_err(OrchestratorError::Store)?;

    // Update request status
    store::eligibility::update_request_status(db, request_id, "completed", Some(now))
        .await
        .map_err(OrchestratorError::Store)?;

    // 9. Publish determination.completed event
    crate::events::publish_determination_completed(
        publisher,
        request_id,
        request.application_id,
        request.household_id,
        &programs_approved.iter().map(|r| r.program).collect::<Vec<_>>(),
        &programs_denied.iter().map(|r| r.program).collect::<Vec<_>>(),
    ).await.map_err(OrchestratorError::EventPublish)?;

    // 10. Return response
    Ok(DetermineResponse {
        request_id,
        application_id: request.application_id,
        programs_approved,
        programs_denied,
        programs_pending,
        medicaid_assigned_group,
        total_monthly_benefit: total_benefit,
        assembled_at: now,
    })
}

// Within the orchestrator, after collecting all determinations:

for determination in &determinations {
    if !verifier.verify(determination)? {
        return Err(OrchestratorError::SignatureVerification(format!(
            "signature verification failed for {} determination",
            determination.program
        )));
    }
}

#[derive(Debug, thiserror::Error)]
pub enum OrchestratorError {
    #[error("store error: {0}")]
    Store(#[from] sqlx::Error),

    #[error("signature verification failed: {0}")]
    SignatureVerification(String),

    #[error("event publishing failed: {0}")]
    EventPublish(#[source] lapin::Error),

    #[error("program service error: {0}")]
    ProgramService(String),
}

impl From<OrchestratorError> for ApiError {
    fn from(e: OrchestratorError) -> Self {
        match e {
            OrchestratorError::Store(e) => ApiError::Internal(format!("database error: {e}")),
            OrchestratorError::SignatureVerification(msg) => ApiError::Internal(msg),
            OrchestratorError::EventPublish(e) => {
                tracing::error!(error = %e, "event publishing failed — determination was persisted");
                // Do NOT fail the request — the determination is already stored.
                // Event will be retried via outbox pattern in a future plan.
                ApiError::Internal(format!("event publish failed: {e}"))
            }
            OrchestratorError::ProgramService(msg) => ApiError::Internal(msg),
        }
    }
}

{
  "application_id": "b7e2f310-1234-4abc-9def-abcdef123456",
  "household_id": "c8f3a421-5678-4def-abcd-fedcba654321",
  "applicant_person_id": "d9a4b532-9abc-4012-3456-789abcdef012",
  "household_member_ids": [
    "d9a4b532-9abc-4012-3456-789abcdef012",
    "e0b5c643-bcde-4123-4567-890abcdef345"
  ],
  "programs": ["snap"],
  "income_ids": ["f1c6d754-cdef-4234-5678-901bcdef0456"],
  "asset_ids": ["a2d7e865-def0-4345-6789-012cdef01567"],
  "expense_ids": ["b3e8f976-ef01-4456-789a-123def012678"]
}

{
  "request_id": "12345678-aaaa-bbbb-cccc-ddddeeee0001",
  "application_id": "b7e2f310-1234-4abc-9def-abcdef123456",
  "programs_approved": [
    {
      "program": "snap",
      "status": "approved",
      "benefit_amount": 847.00,
      "basis": "Household passes gross income test (130% FPL), net income test (100% FPL), and asset test.",
      "effective_date": "2026-03-26"
    }
  ],
  "programs_denied": [],
  "programs_pending": [],
  "medicaid_assigned_group": null,
  "total_monthly_benefit": 847.00,
  "assembled_at": "2026-03-26T14:30:02Z"
}

// services/canopy-eligibility/src/hierarchy.rs

use crate::errors::HierarchyError;

/// Medicaid eligibility categories, ordered by advantageousness
/// (higher index = more advantageous for the applicant).
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum MedicaidCategory {
    ChipExpansion,      // least advantageous
    MagiPregnant,
    MagiChild,
    MagiAdultExpansion,
    MagiAdultParent,
    MedicallyNeedy,
    SsiRelated,         // most advantageous (aged/blind/disabled)
}

pub async fn apply_hierarchy(
    medicaid_determination: &Determination,
    eligible_categories: &[MedicaidCategory],
    rules_client: &RulesClient,
) -> Result<String, HierarchyError> {
    if eligible_categories.is_empty() {
        return Err(HierarchyError::NoEligibleCategories);
    }

    // If only one category, no hierarchy needed
    if eligible_categories.len() == 1 {
        return Ok(format!("{:?}", eligible_categories[0]));
    }

    // Call canopy-rules for hierarchy assignment
    let input = serde_json::json!({
        "eligible_categories": eligible_categories,
        "household_id": medicaid_determination.household_id,
        "applicant_age": medicaid_determination.applicant_age,
    });

    let output = rules_client.evaluate(
        "medicaid-eligibility-hierarchy",
        "determination",
        medicaid_determination.id,
        input,
    ).await.map_err(HierarchyError::RulesEngine)?;

    output.get("assigned_group")
        .and_then(|v| v.as_str())
        .map(String::from)
        .ok_or(HierarchyError::MissingAssignedGroup)
}

#[derive(Debug, thiserror::Error)]
pub enum HierarchyError {
    #[error("no eligible Medicaid categories")]
    NoEligibleCategories,

    #[error("rules engine error: {0}")]
    RulesEngine(#[source] crate::errors::ApiError),

    #[error("rules engine did not return assigned_group")]
    MissingAssignedGroup,
}

// In orchestrator.rs, after collecting determinations:
let medicaid_assigned_group = if determinations.iter().any(|d| d.program == Program::Medicaid) {
    let medicaid_det = determinations.iter()
        .find(|d| d.program == Program::Medicaid)
        .unwrap();
    // Extract eligible categories from the determination basis
    let categories = parse_eligible_categories(&medicaid_det.basis)?;
    Some(hierarchy::apply_hierarchy(medicaid_det, &categories, rules_client).await?)
} else {
    None  // Phase 2: always takes this branch (SNAP only)
};

// services/canopy-eligibility/src/api/determine.rs

use axum::{extract::State, Json};
use crate::errors::ApiError;
use crate::orchestrator::{self, DetermineRequest, DetermineResponse};
use crate::state::EligibilityState;

/// POST /v1/eligibility/determine
pub async fn determine(
    State(state): State<EligibilityState>,
    Json(request): Json<DetermineRequest>,
) -> Result<Json<DetermineResponse>, ApiError> {
    // Validate request
    if request.programs.is_empty() {
        return Err(ApiError::Validation("programs must not be empty".into()));
    }

    let response = orchestrator::orchestrate(
        request,
        &state.registry,
        state.verifier.as_ref(),
        &state.db,
        &state.publisher,
    ).await?;

    Ok(Json(response))
}

use axum::extract::Path;
use uuid::Uuid;
use crate::store;

/// GET /v1/eligibility/requests/{id}
pub async fn get_request(
    State(state): State<EligibilityState>,
    Path(id): Path<Uuid>,
) -> Result<Json<store::models::EligibilityRequest>, ApiError> {
    let request = store::eligibility::get_eligibility_request(&state.db, id)
        .await
        .map_err(|e| ApiError::Internal(format!("query failed: {e}")))?
        .ok_or(ApiError::NotFound(format!("request {id} not found")))?;
    Ok(Json(request))
}

/// GET /v1/eligibility/requests/{id}/determinations
pub async fn get_request_determinations(
    State(state): State<EligibilityState>,
    Path(id): Path<Uuid>,
) -> Result<Json<Vec<store::models::ProgramDetermination>>, ApiError> {
    let dets = store::eligibility::list_determinations_for_request(&state.db, id)
        .await
        .map_err(|e| ApiError::Internal(format!("query failed: {e}")))?;
    Ok(Json(dets))
}

/// GET /v1/eligibility/results/{application_id}
pub async fn get_combined_result(
    State(state): State<EligibilityState>,
    Path(application_id): Path<Uuid>,
) -> Result<Json<store::models::CombinedResult>, ApiError> {
    let result = store::eligibility::get_combined_result_for_application(&state.db, application_id)
        .await
        .map_err(|e| ApiError::Internal(format!("query failed: {e}")))?
        .ok_or(ApiError::NotFound(format!(
            "no combined result for application {application_id}"
        )))?;
    Ok(Json(result))
}

// services/canopy-eligibility/src/api/mod.rs

use axum::{routing::{get, post}, Router};
use crate::state::EligibilityState;

pub mod determine;

pub fn routes() -> Router<EligibilityState> {
    Router::new()
        .route("/v1/eligibility/determine", post(determine::determine))
        .route("/v1/eligibility/requests/:id", get(determine::get_request))
        .route(
            "/v1/eligibility/requests/:id/determinations",
            get(determine::get_request_determinations),
        )
        .route(
            "/v1/eligibility/results/:application_id",
            get(determine::get_combined_result),
        )
}

// services/canopy-eligibility/src/state.rs

use std::sync::Arc;
use sqlx::PgPool;
use crate::determination::DeterminationVerifier;
use crate::events::Publisher;
use crate::registry::ProgramServiceRegistry;

#[derive(Clone)]
pub struct EligibilityState {
    pub db: PgPool,
    pub registry: Arc<ProgramServiceRegistry>,
    pub verifier: Arc<dyn DeterminationVerifier>,
    pub publisher: Arc<Publisher>,
}

// services/canopy-eligibility/src/events.rs

use canopy_reference::Program;
use uuid::Uuid;

pub struct Publisher {
    channel: lapin::Channel,
    exchange: String,
}

impl Publisher {
    pub fn new(channel: lapin::Channel, exchange: String) -> Self {
        Self { channel, exchange }
    }
}

/// Publish a determination.completed event.
///
/// Event payload contains ONLY IDs, status codes, and timestamps —
/// NO benefit amounts, NO bases, NO restricted data per ADR-004.
pub async fn publish_determination_completed(
    publisher: &Publisher,
    request_id: Uuid,
    application_id: Uuid,
    household_id: Uuid,
    programs_approved: &[Program],
    programs_denied: &[Program],
) -> Result<(), lapin::Error> {
    let payload = serde_json::json!({
        "event_type": "determination.completed",
        "request_id": request_id,
        "application_id": application_id,
        "household_id": household_id,
        "programs_approved": programs_approved.iter()
            .map(|p| format!("{p:?}").to_lowercase())
            .collect::<Vec<_>>(),
        "programs_denied": programs_denied.iter()
            .map(|p| format!("{p:?}").to_lowercase())
            .collect::<Vec<_>>(),
        "timestamp": chrono::Utc::now().to_rfc3339(),
    });

    let bytes = serde_json::to_vec(&payload)
        .expect("event serialization should not fail");

    publisher.channel.basic_publish(
        &publisher.exchange,
        "determination.completed",
        lapin::options::BasicPublishOptions::default(),
        &bytes,
        lapin::BasicProperties::default()
            .with_content_type("application/json".into())
            .with_delivery_mode(2), // persistent
    ).await?
    .await?;

    Ok(())
}

{
  "event_type": "determination.completed",
  "request_id": "12345678-aaaa-bbbb-cccc-ddddeeee0001",
  "application_id": "b7e2f310-1234-4abc-9def-abcdef123456",
  "household_id": "c8f3a421-5678-4def-abcd-fedcba654321",
  "programs_approved": ["snap"],
  "programs_denied": [],
  "timestamp": "2026-03-26T14:30:02Z"
}

// services/canopy-eligibility/tests/orchestrator.rs

use canopy_eligibility::orchestrator::{DetermineRequest, DetermineResponse};
use canopy_eligibility::determination::Determination;
use wiremock::{MockServer, Mock, ResponseTemplate};
use wiremock::matchers::{method, path};

/// Happy path: single program (SNAP) requested, returns valid signed
/// determination, combined result assembled correctly.
#[tokio::test]
async fn orchestrate_single_program_snap() {
    // Arrange:
    //   - Start wiremock server for canopy-snap
    //   - Configure mock to return a signed SNAP determination
    //   - Set up test database, registry, verifier
    // Act:
    //   let response = orchestrator::orchestrate(request, ...).await.unwrap();
    // Assert:
    //   assert_eq!(response.programs_approved.len(), 1);
    //   assert_eq!(response.programs_approved[0].program, Program::Snap);
    //   assert_eq!(response.programs_approved[0].status, "approved");
    //   assert!(response.total_monthly_benefit > Decimal::ZERO);
    //   assert!(response.programs_denied.is_empty());
    //   assert!(response.programs_pending.is_empty());
    //   // Verify database state
    //   let request_row = store::get_eligibility_request(&db, response.request_id).await.unwrap().unwrap();
    //   assert_eq!(request_row.status, "completed");
    //   assert!(request_row.completed_at.is_some());
}

/// Signature verification gate: mock returns a determination with an
/// invalid (tampered) signature. Orchestrator must reject it entirely.
#[tokio::test]
async fn reject_unsigned_determination() {
    // Arrange: mock returns determination with empty signature
    // Act:
    //   let result = orchestrator::orchestrate(request, ...).await;
    // Assert:
    //   assert!(result.is_err());
    //   match result.unwrap_err() {
    //       OrchestratorError::SignatureVerification(msg) => {
    //           assert!(msg.contains("signature verification failed"));
    //       }
    //       other => panic!("expected SignatureVerification error, got: {other:?}"),
    //   }
}

/// Mock returns a determination that was signed correctly but then
/// the benefit_amount was tampered with. Verification must fail.
#[tokio::test]
async fn reject_tampered_determination() {
    // Arrange:
    //   - Generate key pair, create a valid signed determination
    //   - Modify benefit_amount after signing
    //   - Configure mock to return the tampered determination
    // Act:
    //   let result = orchestrator::orchestrate(request, ...).await;
    // Assert:
    //   assert!(result.is_err());
    //   match result.unwrap_err() {
    //       OrchestratorError::SignatureVerification(msg) => {
    //           assert!(msg.contains("signature verification failed"));
    //       }
    //       other => panic!("expected SignatureVerification error, got: {other:?}"),
    //   }
}

/// Multiple programs dispatched in parallel. Both return valid signed
/// determinations. Verify that both are collected and the combined
/// result includes both.
#[tokio::test]
async fn parallel_multi_program_dispatch() {
    // Arrange:
    //   - Start two wiremock servers (snap, tanf)
    //   - Configure both to return valid signed determinations
    //   - Register both in ProgramServiceRegistry
    // Act:
    //   let response = orchestrator::orchestrate(request, ...).await.unwrap();
    // Assert:
    //   assert_eq!(response.programs_approved.len(), 2);
    //   let programs: Vec<_> = response.programs_approved.iter().map(|r| r.program).collect();
    //   assert!(programs.contains(&Program::Snap));
    //   assert!(programs.contains(&Program::Tanf));
    //   assert!(response.total_monthly_benefit > Decimal::ZERO);
}

/// Verify that determination.completed event is published to RabbitMQ
/// after successful orchestration, and that it contains only IDs and
/// status (no benefit amounts per ADR-004).
#[tokio::test]
async fn determination_completed_event_published() {
    // Arrange: set up mock program service and a test RabbitMQ consumer
    // Act:
    //   let response = orchestrator::orchestrate(request, ...).await.unwrap();
    // Assert:
    //   let event = consumer.next_event().await;
    //   assert_eq!(event["event_type"], "determination.completed");
    //   assert_eq!(event["request_id"], response.request_id.to_string());
    //   assert_eq!(event["application_id"], request.application_id.to_string());
    //   assert!(event.get("benefit_amount").is_none(), "ADR-004: no benefit amounts in events");
    //   assert!(event.get("basis").is_none(), "ADR-004: no bases in events");
}

/// One program succeeds, another fails (timeout). Combined result
/// reflects the success and marks the failed program as pending.
#[tokio::test]
async fn partial_failure_one_success_one_timeout() {
    // Arrange:
    //   - Mock snap to return valid determination
    //   - Mock tanf to delay 60 seconds (beyond 30s timeout)
    // Act:
    //   let response = orchestrator::orchestrate(request, ...).await.unwrap();
    // Assert:
    //   assert_eq!(response.programs_approved.len(), 1);
    //   assert_eq!(response.programs_approved[0].program, Program::Snap);
    //   assert_eq!(response.programs_pending.len(), 1);
    //   assert_eq!(response.programs_pending[0].program, Program::Tanf);
    //   assert_eq!(response.programs_pending[0].status, "pending_verification");
}

/// Circuit breaker: mock fails repeatedly, breaker opens, subsequent
/// calls are short-circuited without making HTTP requests.
#[tokio::test]
async fn circuit_breaker_opens_after_failures() {
    // Arrange: mock snap to return 500 errors
    // Act: call orchestrate 6 times (threshold = 5)
    // Assert:
    //   - First 5 calls result in program failures (HTTP errors)
    //   - 6th call short-circuits: mock receives no request
    //   let snap_mock = mock_server.received_requests().await.unwrap();
    //   assert_eq!(snap_mock.len(), 5);  // breaker prevented 6th call
}