zebflow 0.10.10

//! Office registry orchestration service.
//!
//! This service is the product-facing owner of office inventory. It persists runtime-host records
//! in the catalog so cluster-aware UI, placement decisions, and runtime dispatch can all resolve
//! the same source of truth.

use std::sync::Arc;

use crate::infra::cluster::registry::{
    WorkerHeartbeat, WorkerRegistryRecord, WorkerRegistrySnapshot,
};
use crate::platform::adapters::data::DataAdapter;
use crate::platform::error::PlatformError;
use crate::platform::model::{
    ClusterWorkerHeartbeatRequest, ClusterWorkerHeartbeatResponse, ClusterWorkerRegisterRequest,
    PlatformOffice, PlatformOfficeNode, now_ts, slug_segment,
};

/// Product-facing office registry service.
#[derive(Clone)]
pub struct ClusterRegistryService {
    data: Arc<dyn DataAdapter>,
}

impl ClusterRegistryService {
    /// Build a registry service backed by the platform catalog adapter.
    pub fn new(data: Arc<dyn DataAdapter>) -> Self {
        Self { data }
    }

    /// Register or refresh one office/runtime host in the catalog.
    pub fn register_worker(
        &self,
        request: &ClusterWorkerRegisterRequest,
    ) -> Result<WorkerRegistryRecord, PlatformError> {
        let now = now_ts();
        // An office row is written below; it needs an address
        // (`kinds/office-topology/README.md`, Rejections).
        let base_url = super::normalize_office_base_url(&request.base_url)?;
        let existing = self.data.get_worker_registry_record(&request.node_id)?;
        let office_id = {
            let value = slug_segment(&request.node_id);
            if value.is_empty() {
                "local".to_string()
            } else {
                value
            }
        };
        let record = WorkerRegistryRecord {
            office_id: office_id.clone(),
            office_slug: office_id.clone(),
            node_id: request.node_id.trim().to_string(),
            label: if request.label.trim().is_empty() {
                request.node_id.trim().to_string()
            } else {
                request.label.trim().to_string()
            },
            base_url,
            status: existing
                .as_ref()
                .map(|value| value.status.clone())
                .filter(|value| !value.trim().is_empty())
                .unwrap_or_else(|| "online".to_string()),
            capabilities: request.capabilities.clone(),
            registered_at: existing
                .as_ref()
                .map(|value| value.registered_at)
                .unwrap_or(now),
            last_heartbeat_at: now,
            // Registration says who the office is, not what it holds; the
            // list arrives with the heartbeat that follows.
            projects: existing
                .as_ref()
                .map(|value| value.projects.clone())
                .unwrap_or_default(),
        };
        // The office row may predate this registration: minting a join token
        // creates it, planned and unreachable, so the controller knows who
        // holds the token before anything presents it (`offices.md` §8).
        // Registering fills it in rather than replacing it, so the minted
        // record's own creation time survives first contact.
        let minted_office = self.data.get_platform_office(&office_id)?;
        self.data.put_platform_office(&PlatformOffice {
            office_id: office_id.clone(),
            office_slug: minted_office
                .as_ref()
                .map(|value| value.office_slug.clone())
                .filter(|value| !value.trim().is_empty())
                .unwrap_or_else(|| office_id.clone()),
            label: record.label.clone(),
            office_kind: "office".to_string(),
            base_url: record.base_url.clone(),
            status: record.status.clone(),
            created_at: minted_office
                .as_ref()
                .map(|value| value.created_at)
                .or_else(|| existing.as_ref().map(|value| value.registered_at))
                .unwrap_or(now),
            updated_at: now,
        })?;
        self.data.put_platform_office_node(&PlatformOfficeNode {
            office_id: office_id.clone(),
            node_id: record.node_id.clone(),
            label: record.label.clone(),
            base_url: record.base_url.clone(),
            status: record.status.clone(),
            capabilities: record.capabilities.clone(),
            registered_at: record.registered_at,
            last_heartbeat_at: record.last_heartbeat_at,
        })?;
        self.data.put_worker_registry_record(&record)?;
        Ok(record)
    }

    /// Persist one office heartbeat update.
    pub fn heartbeat(
        &self,
        request: &ClusterWorkerHeartbeatRequest,
    ) -> Result<ClusterWorkerHeartbeatResponse, PlatformError> {
        let now = now_ts();
        let existing = self
            .data
            .get_worker_registry_record(&request.node_id)?
            .ok_or_else(|| {
                PlatformError::new(
                    "CLUSTER_WORKER_UNKNOWN",
                    format!("office '{}' is not registered", request.node_id),
                )
            })?;
        let mut record = existing.clone();
        if !request.status.trim().is_empty() {
            record.status = request.status.trim().to_string();
        }
        // A heartbeat may leave the address as it is, but it may not leave the
        // office without one: this call rewrites the office row, and a row with
        // no address is not an office (`kinds/office-topology/README.md`).
        let requested = request.base_url.trim();
        record.base_url = super::normalize_office_base_url(if requested.is_empty() {
            &record.base_url
        } else {
            requested
        })?;
        if !request.capabilities.tags.is_empty()
            || request.capabilities.supports_resident
            || request.capabilities.supports_k8s_job
            || request.capabilities.supports_spark_submit
        {
            record.capabilities = request.capabilities.clone();
        }
        record.last_heartbeat_at = now;
        // The office reports its whole list every time, so an empty list is
        // an office that holds nothing, not a heartbeat that forgot.
        record.projects = request.projects.clone();
        let office_id = record
            .office_id
            .trim()
            .to_string()
            .if_empty_then(record.node_id.clone());
        self.data.put_platform_office(&PlatformOffice {
            office_id: office_id.clone(),
            office_slug: record.office_slug.clone().if_empty_then(office_id.clone()),
            label: record.label.clone(),
            office_kind: "office".to_string(),
            base_url: record.base_url.clone(),
            status: record.status.clone(),
            created_at: record.registered_at,
            updated_at: now,
        })?;
        self.data.put_platform_office_node(&PlatformOfficeNode {
            office_id,
            node_id: record.node_id.clone(),
            label: record.label.clone(),
            base_url: record.base_url.clone(),
            status: record.status.clone(),
            capabilities: record.capabilities.clone(),
            registered_at: record.registered_at,
            last_heartbeat_at: record.last_heartbeat_at,
        })?;
        self.data.put_worker_registry_record(&record)?;
        Ok(ClusterWorkerHeartbeatResponse {
            ok: true,
            heartbeat: WorkerHeartbeat {
                node_id: record.node_id,
                at: now,
                status: record.status,
                base_url: record.base_url,
                capabilities: record.capabilities,
            },
        })
    }

    /// Return one office/runtime host by id.
    pub fn get_worker(&self, node_id: &str) -> Result<Option<WorkerRegistryRecord>, PlatformError> {
        self.data.get_worker_registry_record(node_id)
    }

    /// Return the full office snapshot.
    pub fn snapshot(&self) -> Result<WorkerRegistrySnapshot, PlatformError> {
        let workers = self.data.list_worker_registry_records()?;
        Ok(WorkerRegistrySnapshot { workers })
    }

    /// Build runtime target options for project create/clone UI.
    pub fn runtime_target_options(
        &self,
    ) -> Result<Vec<crate::platform::model::ClusterRuntimeTargetOption>, PlatformError> {
        let mut options = vec![crate::platform::model::ClusterRuntimeTargetOption {
            value: "local".to_string(),
            label: "Local office".to_string(),
            description: "Run inside the current self-controlled office.".to_string(),
        }];
        for worker in self.data.list_worker_registry_records()? {
            options.push(crate::platform::model::ClusterRuntimeTargetOption {
                value: worker.node_id.clone(),
                label: worker.label.clone(),
                description: if worker.base_url.is_empty() {
                    worker.status.clone()
                } else {
                    format!("{} · {}", worker.base_url, worker.status)
                },
            });
        }
        Ok(options)
    }
}

trait StringFallbackExt {
    fn if_empty_then(self, fallback: String) -> String;
}

impl StringFallbackExt for String {
    fn if_empty_then(self, fallback: String) -> String {
        if self.trim().is_empty() {
            fallback
        } else {
            self
        }
    }
}