// Request handlers intentionally spell out their deserialized protocol fields.
// Keeping those authority and payload values explicit at this boundary is safer
// than passing an unvalidated wire request deeper into the service.
#![allow(clippy::too_many_arguments)]

use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::time::{SystemTime, UNIX_EPOCH};

use clusterflux_core::{
    Actor, AgentId, ArtifactRegistry, CapabilityReportError, CredentialKind, Digest, LimitError,
    NativeQuicTransport, NodeDescriptor, NodeId, PanelState, Placement, ProcessId, ProjectId,
    RateLimit, TenantId, TransportError, UserId,
};
use thiserror::Error;

use crate::{Coordinator, CoordinatorError};

mod admin;
mod artifacts;
mod authenticated;
mod authorization;
mod debug;
mod debug_requests;
mod durable_runtime;
mod keys;
mod logs;
mod main_runtime;
mod nodes;
mod panels;
mod process_launch;
mod processes;
mod protocol;
mod quota;
mod relay;
mod routing;
mod signed_nodes;
mod tcp;
mod wire_protocol;
use authorization::authorize_authenticated_user_operation;
use durable_runtime::RuntimeDurableStore;
use keys::{
    artifact_id_from_path, enrollment_grant_key, EnrollmentGrantKey, PanelStopKey,
    ProcessControlKey, TaskAssignmentKey, TaskControlKey, TaskRestartKey,
};
pub use protocol::{
    ArtifactTransferAssignment, AuthenticatedCoordinatorRequest, CoordinatorRequest,
    CoordinatorResponse, DebugAcknowledgementState, DebugAuditEvent,
    DebugParticipantAcknowledgement, SourcePreparationDisposition, SourcePreparationStatus,
    TaskAssignment, TaskAttemptSnapshot, TaskAttemptState, TaskCancellationTarget,
    TaskCompletionEvent, TaskExecutor, TaskFailureResolution, TaskReplacementBundle,
    TaskTerminalState, VirtualProcessStatus, WorkflowActor,
};
pub use quota::CoordinatorQuotaConfiguration;
pub use relay::{
    ArtifactRelayConfiguration, ArtifactRelayDurableState, ArtifactRelayError, ArtifactRelayUsage,
};
pub use tcp::{bind_listener, ClientAuthorityMode};
pub use wire_protocol::CoordinatorWireRequest;

const MAX_TASK_LOG_TAIL_BYTES: usize = 256 * 1024;
const DEBUG_CONTROL_READ_BYTES: u64 = 1024;
const MAX_REPLAY_NONCES_PER_AUTHORITY: usize = 1_024;
const NODE_SIGNATURE_WINDOW_SECONDS: u64 = 30;
const MAX_NODE_REPLAY_NONCES_PER_AUTHORITY: usize = 4_096;
const MAX_ENROLLMENT_GRANTS_PER_PROJECT: usize = 64;
const MAX_TASK_EVENTS_PER_PROCESS: usize = 128;
const MAX_DEBUG_AUDIT_EVENTS_PER_PROCESS: usize = 256;
const MAX_RESTART_CHECKPOINTS_PER_PROCESS: usize = 128;
const MAX_TASK_EVENTS_TOTAL: usize = 8_192;
const MAX_DEBUG_AUDIT_EVENTS_TOTAL: usize = 8_192;
const MAX_RESTART_CHECKPOINTS_TOTAL: usize = 4_096;
const MAX_TASK_ATTEMPT_HISTORIES: usize = 4_096;
const MAX_IN_FLIGHT_TASKS_PER_PROCESS: usize = 256;
const MAX_NODE_REPORTED_OBJECTS_PER_KIND: usize = 1_024;
const DEFAULT_NODE_STALE_AFTER_SECONDS: u64 = 30;
fn bounded_ttl(requested: u64, maximum: u64) -> u64 {
    requested.clamp(1, maximum)
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CoordinatorMainRuntimeConfiguration {
    pub fuel_units_per_second: u64,
    pub fuel_burst_seconds: u64,
    pub memory_bytes: usize,
    pub nested_join_timeout_ms: u64,
    pub max_active_mains: usize,
    pub max_wakeups_per_minute: u64,
    pub max_output_bytes: usize,
    pub max_state_bytes: usize,
}

impl Default for CoordinatorMainRuntimeConfiguration {
    fn default() -> Self {
        Self {
            fuel_units_per_second: 10_000_000,
            fuel_burst_seconds: 60,
            memory_bytes: 256 * 1024 * 1024,
            nested_join_timeout_ms: 24 * 60 * 60 * 1_000,
            max_active_mains: usize::MAX,
            max_wakeups_per_minute: 6_000,
            max_output_bytes: MAX_TASK_LOG_TAIL_BYTES,
            max_state_bytes: clusterflux_core::MAX_WASM_TASK_ENVELOPE_BYTES,
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CoordinatorAdmission {
    pub workflow_placement_allowed: bool,
    pub max_node_enrollment_ttl_seconds: u64,
    pub max_artifact_download_ttl_seconds: u64,
}

impl Default for CoordinatorAdmission {
    fn default() -> Self {
        Self {
            workflow_placement_allowed: true,
            max_node_enrollment_ttl_seconds: 15 * 60,
            max_artifact_download_ttl_seconds: 15 * 60,
        }
    }
}

#[derive(Debug, Error)]
pub enum CoordinatorServiceError {
    #[error("coordinator protocol I/O error: {0}")]
    Io(#[from] std::io::Error),
    #[error("coordinator protocol JSON error: {0}")]
    Json(#[from] serde_json::Error),
    #[error("coordinator protocol error: {0}")]
    Protocol(String),
    #[error("coordinator request failed: {0}")]
    Coordinator(#[from] CoordinatorError),
    #[error("artifact download request failed: {0}")]
    Download(#[from] clusterflux_core::DownloadError),
    #[error("scheduler placement failed: {0}")]
    Scheduler(#[from] clusterflux_core::PlacementError),
    #[error("transport request failed: {0}")]
    Transport(#[from] TransportError),
    #[error("resource limit failed: {0}")]
    Resource(#[from] LimitError),
    #[error("operator panel request failed: {0}")]
    Panel(#[from] clusterflux_core::PanelError),
    #[error("invalid node capability report: {0}")]
    CapabilityReport(#[from] CapabilityReportError),
    #[error("invalid VFS artifact path reported by node: {0}")]
    InvalidArtifactPath(String),
    #[error("invalid task log tail reported by node: {0}")]
    InvalidTaskLogTail(String),
    #[error("durable coordinator state failed: {0}")]
    Durable(String),
}

pub struct CoordinatorService {
    coordinator: Coordinator,
    store: RuntimeDurableStore,
    node_descriptors: BTreeMap<NodeId, NodeDescriptor>,
    node_last_seen_epoch_seconds: BTreeMap<NodeId, u64>,
    node_stale_after_seconds: u64,
    debug_freeze_timeout: std::time::Duration,
    enrollment_grants: BTreeMap<EnrollmentGrantKey, clusterflux_core::EnrollmentGrant>,
    task_events: VecDeque<TaskCompletionEvent>,
    process_scope_history: VecDeque<ProcessControlKey>,
    debug_audit_events: VecDeque<DebugAuditEvent>,
    debug_epochs: BTreeMap<ProcessControlKey, u64>,
    debug_epoch_runtime: BTreeMap<ProcessControlKey, debug::DebugEpochRuntime>,
    debug_breakpoints: BTreeMap<ProcessControlKey, debug::DebugBreakpointPlan>,
    debug_commands: BTreeMap<TaskControlKey, debug::DebugPendingCommand>,
    task_assignments: BTreeMap<TaskAssignmentKey, VecDeque<TaskAssignment>>,
    task_restart_checkpoints: BTreeMap<TaskRestartKey, processes::TaskRestartCheckpoint>,
    task_restart_checkpoint_order: VecDeque<TaskRestartKey>,
    task_attempts: BTreeMap<TaskRestartKey, Vec<protocol::TaskAttemptSnapshot>>,
    restart_launches: BTreeSet<TaskRestartKey>,
    main_runtime: main_runtime::CoordinatorMainRuntime,
    pending_task_launches: VecDeque<processes::PendingTaskLaunch>,
    task_placements: BTreeMap<TaskControlKey, Placement>,
    active_tasks: BTreeSet<TaskControlKey>,
    task_cancellations: BTreeSet<TaskControlKey>,
    task_aborts: BTreeSet<TaskControlKey>,
    process_cancellations: BTreeSet<ProcessControlKey>,
    process_aborts: BTreeSet<ProcessControlKey>,
    agent_replay_nonces: BTreeMap<(TenantId, ProjectId, AgentId, String), u64>,
    node_replay_nonces: BTreeMap<(NodeId, String), u64>,
    panel_snapshots: BTreeMap<PanelStopKey, PanelState>,
    stopped_panels: BTreeSet<PanelStopKey>,
    panel_event_limits: BTreeMap<(TenantId, ProjectId, ProcessId, String), RateLimit>,
    artifact_registry: ArtifactRegistry,
    artifact_reverse_transfers: BTreeMap<String, artifacts::ArtifactReverseTransfer>,
    artifact_transfer_by_token: BTreeMap<Digest, String>,
    artifact_relay: relay::ArtifactRelayLedger,
    transport: NativeQuicTransport,
    quota: quota::CoordinatorQuota,
    admission: CoordinatorAdmission,
    #[cfg(test)]
    server_time_override: Option<u64>,
    admin_token_digest: Option<Digest>,
    admin_replay_nonces: BTreeMap<String, u64>,
}

impl CoordinatorService {
    pub fn configure_artifact_relay(
        &mut self,
        configuration: ArtifactRelayConfiguration,
    ) -> Result<(), CoordinatorServiceError> {
        let now_epoch_seconds = self.current_epoch_seconds()?;
        let mut candidate = self.artifact_relay.clone();
        candidate
            .configure(configuration)
            .map_err(|error| CoordinatorServiceError::Protocol(error.to_string()))?;
        candidate.reconcile_after_restart(now_epoch_seconds);
        self.commit_artifact_relay(candidate)
    }

    pub fn artifact_relay_usage(&self) -> ArtifactRelayUsage {
        self.artifact_relay.usage()
    }

    fn commit_artifact_relay(
        &mut self,
        candidate: relay::ArtifactRelayLedger,
    ) -> Result<(), CoordinatorServiceError> {
        let previous = self.artifact_relay.clone();
        let previous_state = previous.durable_state();
        self.artifact_relay = candidate;
        self.coordinator
            .set_artifact_relay_state(self.artifact_relay.durable_state());
        if let Err(error) = self.persist_durable_state() {
            self.artifact_relay = previous;
            self.coordinator.set_artifact_relay_state(previous_state);
            return Err(error);
        }
        Ok(())
    }

    fn mutate_artifact_relay<T>(
        &mut self,
        mutation: impl FnOnce(&mut relay::ArtifactRelayLedger) -> Result<T, ArtifactRelayError>,
    ) -> Result<T, CoordinatorServiceError> {
        let mut candidate = self.artifact_relay.clone();
        let result = mutation(&mut candidate)
            .map_err(|error| CoordinatorServiceError::Protocol(error.to_string()))?;
        self.commit_artifact_relay(candidate)?;
        Ok(result)
    }

    pub fn set_debug_freeze_timeout(&mut self, timeout: std::time::Duration) {
        self.debug_freeze_timeout = timeout.max(std::time::Duration::from_millis(1));
    }

    pub fn configure_coordinator_main_runtime(
        &mut self,
        configuration: CoordinatorMainRuntimeConfiguration,
    ) -> Result<(), CoordinatorServiceError> {
        self.main_runtime.configure(configuration)
    }

    pub fn record_service_policy(
        &mut self,
        tenant: TenantId,
        name: impl Into<String>,
        digest: Digest,
    ) -> Result<crate::ServicePolicyRecord, CoordinatorServiceError> {
        let name = name.into();
        self.coordinator
            .upsert_service_policy_record(tenant.clone(), name.clone(), digest);
        self.persist_durable_state()?;
        Ok(self
            .coordinator
            .service_policy_record(&tenant, &name)
            .expect("service policy record was persisted immediately after insertion")
            .clone())
    }

    pub(super) fn authorize_node_for_process_or_termination(
        &self,
        node: &NodeId,
        tenant: &TenantId,
        project: &ProjectId,
        process: &ProcessId,
    ) -> Result<(), CoordinatorServiceError> {
        let process_key = keys::process_control_key(tenant, project, process);
        if self.process_cancellations.contains(&process_key)
            || self.process_aborts.contains(&process_key)
        {
            let identity = self
                .coordinator
                .node_identity(node)
                .ok_or(CoordinatorError::UnknownNode)?;
            if &identity.tenant != tenant || &identity.project != project {
                return Err(CoordinatorError::Unauthorized(
                    "node process-control request is outside its enrolled tenant/project scope"
                        .to_owned(),
                )
                .into());
            }
            return Ok(());
        }
        self.coordinator
            .authorize_node_for_process(node, tenant, project, process)?;
        Ok(())
    }

    pub fn new(coordinator_epoch: u64) -> Self {
        Self::new_with_optional_admin_token_and_admission(
            coordinator_epoch,
            std::env::var("CLUSTERFLUX_ADMIN_TOKEN").ok(),
            CoordinatorAdmission::default(),
        )
    }

    pub fn new_with_admin_token(coordinator_epoch: u64, admin_token: impl Into<String>) -> Self {
        Self::new_with_optional_admin_token_and_admission(
            coordinator_epoch,
            Some(admin_token.into()),
            CoordinatorAdmission::default(),
        )
    }

    pub fn new_with_admission(coordinator_epoch: u64, admission: CoordinatorAdmission) -> Self {
        Self::new_with_optional_admin_token_and_admission(
            coordinator_epoch,
            std::env::var("CLUSTERFLUX_ADMIN_TOKEN").ok(),
            admission,
        )
    }

    fn new_with_optional_admin_token_and_admission(
        coordinator_epoch: u64,
        admin_token: Option<String>,
        admission: CoordinatorAdmission,
    ) -> Self {
        Self::try_new_with_optional_admin_token_admission_and_database_url(
            coordinator_epoch,
            admin_token,
            admission,
            None,
            CoordinatorQuotaConfiguration::default(),
        )
        .expect("in-memory durable coordinator store initialization cannot fail")
    }

    pub fn new_with_database_url(
        coordinator_epoch: u64,
        database_url: Option<&str>,
    ) -> Result<Self, CoordinatorServiceError> {
        Self::try_new_with_optional_admin_token_admission_and_database_url(
            coordinator_epoch,
            std::env::var("CLUSTERFLUX_ADMIN_TOKEN").ok(),
            CoordinatorAdmission::default(),
            database_url,
            CoordinatorQuotaConfiguration::default(),
        )
    }

    pub fn new_with_admin_token_and_database_url(
        coordinator_epoch: u64,
        admin_token: impl Into<String>,
        database_url: Option<&str>,
    ) -> Result<Self, CoordinatorServiceError> {
        Self::new_with_admin_token_database_url_and_quota(
            coordinator_epoch,
            admin_token,
            database_url,
            CoordinatorQuotaConfiguration::default(),
        )
    }

    pub fn new_with_admin_token_database_url_and_quota(
        coordinator_epoch: u64,
        admin_token: impl Into<String>,
        database_url: Option<&str>,
        quota_configuration: CoordinatorQuotaConfiguration,
    ) -> Result<Self, CoordinatorServiceError> {
        Self::try_new_with_optional_admin_token_admission_and_database_url(
            coordinator_epoch,
            Some(admin_token.into()),
            CoordinatorAdmission::default(),
            database_url,
            quota_configuration,
        )
    }

    fn try_new_with_optional_admin_token_admission_and_database_url(
        coordinator_epoch: u64,
        admin_token: Option<String>,
        admission: CoordinatorAdmission,
        database_url: Option<&str>,
        quota_configuration: CoordinatorQuotaConfiguration,
    ) -> Result<Self, CoordinatorServiceError> {
        let mut store = RuntimeDurableStore::from_database_url(database_url)
            .map_err(CoordinatorServiceError::Durable)?;
        let coordinator = Coordinator::try_boot(&mut store, coordinator_epoch)
            .map_err(CoordinatorServiceError::Durable)?;
        let artifact_relay_state = coordinator.artifact_relay_state().clone();
        let admin_token_digest = admin_token
            .filter(|token| !token.trim().is_empty())
            .map(Digest::sha256);
        Ok(Self {
            coordinator,
            store,
            node_descriptors: BTreeMap::new(),
            node_last_seen_epoch_seconds: BTreeMap::new(),
            node_stale_after_seconds: std::env::var("CLUSTERFLUX_NODE_STALE_AFTER_SECONDS")
                .ok()
                .and_then(|value| value.parse::<u64>().ok())
                .filter(|seconds| *seconds > 0)
                .unwrap_or(DEFAULT_NODE_STALE_AFTER_SECONDS),
            debug_freeze_timeout: std::time::Duration::from_secs(5),
            enrollment_grants: BTreeMap::new(),
            task_events: VecDeque::new(),
            process_scope_history: VecDeque::new(),
            debug_audit_events: VecDeque::new(),
            debug_epochs: BTreeMap::new(),
            debug_epoch_runtime: BTreeMap::new(),
            debug_breakpoints: BTreeMap::new(),
            debug_commands: BTreeMap::new(),
            task_assignments: BTreeMap::new(),
            task_restart_checkpoints: BTreeMap::new(),
            task_restart_checkpoint_order: VecDeque::new(),
            task_attempts: BTreeMap::new(),
            restart_launches: BTreeSet::new(),
            main_runtime: main_runtime::CoordinatorMainRuntime::default(),
            pending_task_launches: VecDeque::new(),
            task_placements: BTreeMap::new(),
            active_tasks: BTreeSet::new(),
            task_cancellations: BTreeSet::new(),
            task_aborts: BTreeSet::new(),
            process_cancellations: BTreeSet::new(),
            process_aborts: BTreeSet::new(),
            agent_replay_nonces: BTreeMap::new(),
            node_replay_nonces: BTreeMap::new(),
            panel_snapshots: BTreeMap::new(),
            stopped_panels: BTreeSet::new(),
            panel_event_limits: BTreeMap::new(),
            artifact_registry: ArtifactRegistry::default(),
            artifact_reverse_transfers: BTreeMap::new(),
            artifact_transfer_by_token: BTreeMap::new(),
            artifact_relay: relay::ArtifactRelayLedger::from_durable(
                ArtifactRelayConfiguration::default(),
                artifact_relay_state,
            ),
            transport: NativeQuicTransport,
            quota: quota::CoordinatorQuota::new(quota_configuration),
            admission,
            #[cfg(test)]
            server_time_override: None,
            admin_token_digest,
            admin_replay_nonces: BTreeMap::new(),
        })
    }

    pub fn durable_store_kind(&self) -> &'static str {
        self.store.kind()
    }

    fn persist_durable_state(&mut self) -> Result<(), CoordinatorServiceError> {
        self.coordinator
            .try_persist(&mut self.store)
            .map_err(CoordinatorServiceError::Durable)
    }

    fn current_epoch_seconds(&self) -> Result<u64, CoordinatorServiceError> {
        #[cfg(test)]
        if let Some(now) = self.server_time_override {
            return Ok(now);
        }
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|duration| duration.as_secs())
            .map_err(|err| CoordinatorServiceError::Protocol(format!("system clock error: {err}")))
    }

    fn handle_quota_status(
        &self,
        tenant: String,
        project: String,
        actor_user: String,
    ) -> Result<CoordinatorResponse, CoordinatorServiceError> {
        let tenant = TenantId::new(tenant);
        let project = ProjectId::new(project);
        let actor = UserId::new(actor_user);
        let durable_project = self.coordinator.project(&project).ok_or_else(|| {
            CoordinatorError::Unauthorized("quota status requires an existing project".to_owned())
        })?;
        if durable_project.tenant != tenant {
            return Err(CoordinatorError::Unauthorized(
                "quota status project is outside the tenant scope".to_owned(),
            )
            .into());
        }
        let now_epoch_seconds = self.current_epoch_seconds()?;
        let status = self
            .quota
            .project_status(&tenant, &project, now_epoch_seconds);
        Ok(CoordinatorResponse::QuotaStatus {
            tenant,
            project,
            actor,
            policy_label: status.policy_label,
            limits: status.limits,
            window_seconds: status.window_seconds,
            usage: status.usage,
            window_started_epoch_seconds: status.window_started_epoch_seconds,
        })
    }

    #[cfg(test)]
    fn set_server_time(&mut self, now_epoch_seconds: u64) {
        self.server_time_override = Some(now_epoch_seconds);
    }

    pub fn issue_cli_session(
        &mut self,
        tenant: TenantId,
        project: ProjectId,
        user: UserId,
        session_secret: &str,
        expires_at_epoch_seconds: Option<u64>,
    ) -> Result<crate::CliSessionRecord, CoordinatorServiceError> {
        if let Some(existing) = self.coordinator.project(&project) {
            if existing.tenant != tenant {
                return Err(CoordinatorError::Unauthorized(
                    "CLI session project belongs to a different tenant".to_owned(),
                )
                .into());
            }
        } else {
            self.coordinator
                .upsert_project(tenant.clone(), project.clone(), "Session project");
        }
        self.coordinator
            .grant_project_debug(tenant.clone(), project.clone(), user.clone());
        let record = self.coordinator.issue_cli_session(
            tenant,
            project,
            user,
            session_secret,
            expires_at_epoch_seconds,
        );
        self.persist_durable_state()?;
        Ok(record)
    }

    pub fn authenticate_cli_session_context(
        &self,
        session_secret: &str,
    ) -> Result<clusterflux_core::AuthContext, CoordinatorServiceError> {
        Ok(self.coordinator.authenticate_cli_session(session_secret)?)
    }

    pub fn authenticate_cli_session_status_context(
        &self,
        session_secret: &str,
    ) -> Result<clusterflux_core::AuthContext, CoordinatorServiceError> {
        Ok(self
            .coordinator
            .authenticate_cli_session_for_status(session_secret)?)
    }

    pub fn revoke_cli_session(
        &mut self,
        session_secret: &str,
    ) -> Result<crate::CliSessionRecord, CoordinatorServiceError> {
        let record = self.coordinator.revoke_cli_session(session_secret)?;
        self.persist_durable_state()?;
        Ok(record)
    }
}

#[cfg(test)]
mod tests;