use std::sync::Arc;

use rocketmq_client_rust::{ClientRuntime, ClientRuntimeConfig};
use rocketmq_observability::TelemetryRuntimeGuard;
use rocketmq_runtime::{RuntimeConfig, RuntimeOwner};

use crate::error::{QueueError, QueueResult};

pub struct QueueRuntime {
    owner: RuntimeOwner,
    client: Arc<ClientRuntime>,
    _telemetry: TelemetryRuntimeGuard,
}

impl QueueRuntime {
    pub fn connect(scope: &str) -> QueueResult<Self> {
        let owner = RuntimeOwner::plan(RuntimeConfig {
            thread_name: format!("rocketmq-{scope}"),
            ..RuntimeConfig::default()
        })
        .map_err(|error| QueueError::config(format!("invalid rocketmq runtime config: {error}")))?
        .build()
        .map_err(|error| QueueError::connect(format!("creating the rocketmq runtime: {error}")))?;
        let telemetry = TelemetryRuntimeGuard::noop();
        let client = ClientRuntime::try_new(
            owner.root_context().component("client"),
            ClientRuntimeConfig::default(),
            telemetry.handle(),
        )
        .map_err(|error| {
            QueueError::connect(format!("creating the rocketmq client runtime: {error}"))
        })?;
        Ok(Self {
            owner,
            client,
            _telemetry: telemetry,
        })
    }

    pub fn client(&self) -> Arc<ClientRuntime> {
        Arc::clone(&self.client)
    }

    pub async fn shutdown(self) {
        let report = self.client.shutdown().await;
        if !report.is_healthy() {
            tracing::warn!(report = %report.to_json(), "rocketmq client runtime shut down uncleanly");
        }
        let report = self.owner.shutdown_tasks().await;
        if !report.is_healthy() {
            tracing::warn!(report = %report.to_json(), "rocketmq runtime tasks shut down uncleanly");
        }
        let report = self.owner.shutdown_background();
        if !report.is_healthy() {
            tracing::warn!(report = %report.to_json(), "rocketmq runtime did not shut down cleanly");
        }
    }
}