datum-agent 0.10.1

use std::collections::HashMap;

use prost::Message as ProstMessage;

/// DCP protocol version implemented by this crate.
///
/// Minor-version bumps are additive within one DCP major: new request variants
/// and payload fields may be added without changing the major. Servers keep
/// rejecting unknown major versions during `Hello` negotiation.
pub const DCP_PROTOCOL_VERSION: &str = "0.10.0";

/// DCP major version accepted by this crate.
pub const DCP_PROTOCOL_MAJOR: u32 = 0;

/// Client role advertised in the initial DCP hello.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, prost::Enumeration)]
#[repr(i32)]
pub enum ClientKind {
    Unspecified = 0,
    Cli = 1,
    Tui = 2,
    Daemon = 3,
    ClusterNode = 4,
}

/// Response status for unary DCP requests and hello negotiation.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, prost::Enumeration)]
#[repr(i32)]
pub enum ResponseStatus {
    Ok = 0,
    BadRequest = 1,
    Unauthorized = 2,
    NotFound = 3,
    Conflict = 4,
    Failed = 5,
    ProtocolMismatch = 6,
    DeadlineExceeded = 7,
}

/// Authentication data carried in the hello. Remote deployments are expected to
/// rely on QUIC mTLS first; this is an optional operator token hook.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct Auth {
    #[prost(string, tag = "1")]
    pub bearer_token: String,
}

/// First client-to-server frame on every DCP stream.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct Hello {
    #[prost(string, tag = "1")]
    pub protocol_version: String,
    #[prost(string, tag = "2")]
    pub node_id: String,
    #[prost(enumeration = "ClientKind", tag = "3")]
    pub client_kind: i32,
    #[prost(message, optional, tag = "4")]
    pub auth: Option<Auth>,
    #[prost(string, repeated, tag = "5")]
    pub capabilities: Vec<String>,
}

impl Hello {
    #[must_use]
    pub fn new(node_id: impl Into<String>, client_kind: ClientKind) -> Self {
        Self {
            protocol_version: DCP_PROTOCOL_VERSION.to_owned(),
            node_id: node_id.into(),
            client_kind: client_kind as i32,
            auth: None,
            capabilities: Vec::new(),
        }
    }
}

/// Unary or subscription request.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct Request {
    #[prost(uint64, tag = "1")]
    pub request_id: u64,
    #[prost(uint64, tag = "2")]
    pub deadline_ms: u64,
    #[prost(
        oneof = "request::Command",
        tags = "10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29"
    )]
    pub command: Option<request::Command>,
}

pub mod request {
    #[derive(Clone, PartialEq, prost::Oneof)]
    pub enum Command {
        #[prost(message, tag = "10")]
        ListJobs(super::ListJobs),
        #[prost(message, tag = "11")]
        StartJob(super::StartJob),
        #[prost(message, tag = "12")]
        DrainJob(super::DrainJob),
        #[prost(message, tag = "13")]
        StopJob(super::StopJob),
        #[prost(message, tag = "14")]
        RestartJob(super::RestartJob),
        #[prost(message, tag = "15")]
        JobStatus(super::JobStatusRequest),
        #[prost(message, tag = "16")]
        SubscribeEvents(super::SubscribeEvents),
        #[prost(message, tag = "17")]
        SubscribeMetrics(super::SubscribeMetrics),
        #[prost(message, tag = "18")]
        GetConfig(super::GetConfig),
        #[prost(message, tag = "19")]
        PutConfig(super::PutConfig),
        #[prost(message, tag = "20")]
        ListClusterJobs(super::ListClusterJobs),
        #[prost(message, tag = "21")]
        ClusterNodeInfo(super::ClusterNodeInfo),
        #[prost(message, tag = "22")]
        SubmitClusterJob(super::SubmitClusterJob),
        #[prost(message, tag = "23")]
        RememberClusterAssignment(super::RememberClusterAssignment),
        #[prost(message, tag = "24")]
        AllocateShard(super::ShardAllocationRequest),
        #[prost(message, tag = "25")]
        RememberShardAllocations(super::RememberShardAllocations),
        #[prost(message, tag = "26")]
        GetShardAllocations(super::GetShardAllocations),
        #[prost(message, tag = "27")]
        ForwardShardEnvelopes(super::ForwardShardEnvelopes),
        #[prost(message, tag = "28")]
        CompleteShardingAsk(super::CompleteShardingAsk),
        #[prost(message, tag = "29")]
        OpenShardPipe(super::OpenShardPipe),
    }
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct ListJobs {}

/// StartJob uses the v0.9.1 registered-factory model.
///
/// Dynamic blueprint upload is intentionally out of scope until cluster
/// placement work in v0.10.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct StartJob {
    #[prost(string, tag = "1")]
    pub factory_name: String,
    #[prost(string, tag = "2")]
    pub instance_name: String,
    #[prost(map = "string, string", tag = "3")]
    pub params: HashMap<String, String>,
    #[prost(message, optional, tag = "4")]
    pub cluster: Option<ClusterJobStart>,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct DrainJob {
    #[prost(string, tag = "1")]
    pub name: String,
    #[prost(bool, tag = "2")]
    pub cluster: bool,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct StopJob {
    #[prost(string, tag = "1")]
    pub name: String,
    #[prost(bool, tag = "2")]
    pub cluster: bool,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct RestartJob {
    #[prost(string, tag = "1")]
    pub name: String,
    #[prost(bool, tag = "2")]
    pub cluster: bool,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct JobStatusRequest {
    #[prost(string, tag = "1")]
    pub name: String,
    #[prost(bool, tag = "2")]
    pub cluster: bool,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct SubscribeEvents {
    #[prost(uint64, tag = "1")]
    pub buffer: u64,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct SubscribeMetrics {
    #[prost(uint64, tag = "1")]
    pub interval_ms: u64,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct GetConfig {
    #[prost(string, tag = "1")]
    pub key: String,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct PutConfig {
    #[prost(string, tag = "1")]
    pub key: String,
    #[prost(string, tag = "2")]
    pub value: String,
}

/// Aggregate job snapshots across the cluster.
///
/// The server fans out to currently known peer node sessions and returns
/// partial results with per-node errors when a peer is unreachable or the
/// bounded wait expires.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct ListClusterJobs {
    /// Per-peer fan-out timeout in milliseconds. A zero value uses the server
    /// default.
    #[prost(uint64, tag = "1")]
    pub timeout_ms: u64,
}

/// Return the local node's cluster membership and DCP session view.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct ClusterNodeInfo {
    /// Snapshot collection timeout in milliseconds. A zero value uses the
    /// server default.
    #[prost(uint64, tag = "1")]
    pub timeout_ms: u64,
}

/// Submit a registered-factory job to the placement coordinator.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct SubmitClusterJob {
    #[prost(string, tag = "1")]
    pub factory_name: String,
    #[prost(string, tag = "2")]
    pub instance_name: String,
    #[prost(map = "string, string", tag = "3")]
    pub params: HashMap<String, String>,
    #[prost(message, optional, tag = "4")]
    pub placement: Option<PlacementSpec>,
    /// Coordinator/placement timeout in milliseconds. A zero value uses the
    /// server default.
    #[prost(uint64, tag = "5")]
    pub timeout_ms: u64,
}

/// Internal best-effort replication of coordinator assignment state to peers.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct RememberClusterAssignment {
    #[prost(string, tag = "1")]
    pub instance_name: String,
    #[prost(message, optional, tag = "2")]
    pub assignment: Option<ClusterJobStart>,
}

/// Request the coordinator allocation for one shard.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct ShardAllocationRequest {
    #[prost(string, tag = "1")]
    pub type_name: String,
    #[prost(string, tag = "2")]
    pub shard_id: String,
    /// Coordinator timeout in milliseconds. A zero value uses the server
    /// default.
    #[prost(uint64, tag = "3")]
    pub timeout_ms: u64,
}

/// Coordinator answer for one shard home.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct ShardAllocation {
    #[prost(string, tag = "1")]
    pub type_name: String,
    #[prost(string, tag = "2")]
    pub shard_id: String,
    #[prost(string, tag = "3")]
    pub node_id: String,
    #[prost(uint64, tag = "4")]
    pub generation: u64,
}

/// Allocation-table row replicated to peer regions.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct ShardAllocationEntry {
    #[prost(string, tag = "1")]
    pub type_name: String,
    #[prost(string, tag = "2")]
    pub shard_id: String,
    #[prost(string, tag = "3")]
    pub node_id: String,
    #[prost(uint64, tag = "4")]
    pub generation: u64,
}

/// Allocation-table snapshot. v0.10 keeps this in memory; persistence and
/// remember-entities belong to later sharding work packages.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct ShardAllocationTable {
    #[prost(message, repeated, tag = "1")]
    pub entries: Vec<ShardAllocationEntry>,
}

/// Best-effort allocation-table replication from the active coordinator.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct RememberShardAllocations {
    #[prost(message, optional, tag = "1")]
    pub table: Option<ShardAllocationTable>,
}

/// Ask a peer region for its cached allocation table.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct GetShardAllocations {
    #[prost(string, tag = "1")]
    pub type_name: String,
}

/// One serialized entity envelope carried between peer regions.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct ShardEnvelopeWire {
    #[prost(string, tag = "1")]
    pub entity_id: String,
    #[prost(string, tag = "2")]
    pub shard_id: String,
    #[prost(bytes = "vec", tag = "3")]
    pub payload: Vec<u8>,
}

/// Batched peer-region forwarding on the existing C2 session.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct ForwardShardEnvelopes {
    #[prost(string, tag = "1")]
    pub type_name: String,
    #[prost(message, repeated, tag = "2")]
    pub envelopes: Vec<ShardEnvelopeWire>,
}

/// Per-envelope enqueue result for a forwarded batch.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct ShardEnvelopeAck {
    #[prost(string, tag = "1")]
    pub entity_id: String,
    #[prost(string, tag = "2")]
    pub shard_id: String,
    #[prost(bool, tag = "3")]
    pub ok: bool,
    #[prost(string, tag = "4")]
    pub message: String,
}

/// Result for a peer-region forwarding batch.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct ShardEnvelopeBatchResult {
    #[prost(message, repeated, tag = "1")]
    pub acks: Vec<ShardEnvelopeAck>,
}

/// Complete an ask whose reply port originated on another region.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct CompleteShardingAsk {
    #[prost(uint64, tag = "1")]
    pub request_id: u64,
    #[prost(bool, tag = "2")]
    pub ok: bool,
    #[prost(bytes = "vec", tag = "3")]
    pub payload: Vec<u8>,
    #[prost(string, tag = "4")]
    pub message: String,
}

/// Switch this DCP connection to the sharding hot-path pipe after hello.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct OpenShardPipe {}

/// One sharding hot-path pipe frame.
///
/// Frames are one-way and ordered per peer connection. The sender batches only
/// when its local pipe queue already has backlog; idle sends are flushed
/// immediately so p50 does not pay a linger tax.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct ShardPipeFrame {
    #[prost(message, repeated, tag = "1")]
    pub forwards: Vec<ForwardShardEnvelopes>,
    #[prost(message, repeated, tag = "2")]
    pub replies: Vec<CompleteShardingAsk>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, prost::Enumeration)]
#[repr(i32)]
pub enum PlacementStrategy {
    LeastJobs = 0,
    Pinned = 1,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct PlacementSpec {
    #[prost(string, tag = "1")]
    pub role_constraint: String,
    #[prost(enumeration = "PlacementStrategy", tag = "2")]
    pub strategy: i32,
    #[prost(string, tag = "3")]
    pub pinned_node_id: String,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct ClusterPlacementHistory {
    #[prost(uint64, tag = "1")]
    pub generation: u64,
    #[prost(string, tag = "2")]
    pub from_node_id: String,
    #[prost(string, tag = "3")]
    pub to_node_id: String,
    #[prost(string, tag = "4")]
    pub reason: String,
    #[prost(uint64, tag = "5")]
    pub timestamp_ms: u64,
}

/// Cluster metadata carried on coordinator-issued StartJob.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct ClusterJobStart {
    #[prost(string, tag = "1")]
    pub factory_name: String,
    #[prost(map = "string, string", tag = "2")]
    pub params: HashMap<String, String>,
    #[prost(message, optional, tag = "3")]
    pub placement: Option<PlacementSpec>,
    #[prost(string, tag = "4")]
    pub coordinator_node_id: String,
    #[prost(string, tag = "5")]
    pub assigned_node_id: String,
    #[prost(uint64, tag = "6")]
    pub placement_generation: u64,
    #[prost(message, repeated, tag = "7")]
    pub history: Vec<ClusterPlacementHistory>,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct Response {
    #[prost(uint64, tag = "1")]
    pub request_id: u64,
    #[prost(enumeration = "ResponseStatus", tag = "2")]
    pub status: i32,
    #[prost(string, tag = "3")]
    pub message: String,
    #[prost(bytes = "vec", tag = "4")]
    pub payload: Vec<u8>,
}

impl Response {
    #[must_use]
    pub fn ok(request_id: u64, payload: Vec<u8>) -> Self {
        Self {
            request_id,
            status: ResponseStatus::Ok as i32,
            message: String::new(),
            payload,
        }
    }

    #[must_use]
    pub fn error(request_id: u64, status: ResponseStatus, message: impl Into<String>) -> Self {
        Self {
            request_id,
            status: status as i32,
            message: message.into(),
            payload: Vec::new(),
        }
    }

    #[must_use]
    pub fn response_status(&self) -> ResponseStatus {
        ResponseStatus::try_from(self.status).unwrap_or(ResponseStatus::Failed)
    }
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct JobList {
    #[prost(message, repeated, tag = "1")]
    pub jobs: Vec<JobStatus>,
}

/// Cluster-wide job list response.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct ClusterJobList {
    #[prost(message, repeated, tag = "1")]
    pub nodes: Vec<ClusterJobNode>,
    #[prost(message, repeated, tag = "2")]
    pub errors: Vec<ClusterNodeError>,
    #[prost(bool, tag = "3")]
    pub partial: bool,
}

/// Jobs reported by one cluster node.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct ClusterJobNode {
    #[prost(string, tag = "1")]
    pub node_id: String,
    #[prost(string, tag = "2")]
    pub address: String,
    #[prost(bool, tag = "3")]
    pub local: bool,
    #[prost(message, repeated, tag = "4")]
    pub jobs: Vec<JobStatus>,
}

/// Cluster node information response.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct ClusterNodeList {
    #[prost(message, repeated, tag = "1")]
    pub nodes: Vec<ClusterNodeStatus>,
    #[prost(message, repeated, tag = "2")]
    pub errors: Vec<ClusterNodeError>,
    #[prost(bool, tag = "3")]
    pub partial: bool,
}

/// Membership and DCP session state for one node.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct ClusterNodeStatus {
    #[prost(string, tag = "1")]
    pub node_id: String,
    #[prost(string, tag = "2")]
    pub member_state: String,
    #[prost(string, tag = "3")]
    pub address: String,
    #[prost(string, tag = "4")]
    pub agent_addr: String,
    #[prost(string, repeated, tag = "5")]
    pub roles: Vec<String>,
    #[prost(bool, tag = "6")]
    pub unreachable: bool,
    #[prost(bool, tag = "7")]
    pub local: bool,
    #[prost(string, tag = "8")]
    pub session_state: String,
}

/// Per-node partial-result error for cluster fan-out requests.
#[derive(Clone, PartialEq, ProstMessage)]
pub struct ClusterNodeError {
    #[prost(string, tag = "1")]
    pub node_id: String,
    #[prost(string, tag = "2")]
    pub message: String,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct ConfigValue {
    #[prost(string, tag = "1")]
    pub key: String,
    #[prost(string, tag = "2")]
    pub value: String,
    #[prost(bool, tag = "3")]
    pub existed: bool,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct JobStatus {
    #[prost(string, tag = "1")]
    pub name: String,
    #[prost(uint64, tag = "2")]
    pub job_id: u64,
    #[prost(string, tag = "3")]
    pub state: String,
    #[prost(string, tag = "4")]
    pub desired_state: String,
    #[prost(uint64, tag = "5")]
    pub generation: u64,
    #[prost(uint64, tag = "6")]
    pub starts_total: u64,
    #[prost(uint64, tag = "7")]
    pub restarts_total: u64,
    #[prost(uint64, optional, tag = "8")]
    pub last_start_at_ms: Option<u64>,
    #[prost(uint64, optional, tag = "9")]
    pub last_exit_at_ms: Option<u64>,
    #[prost(string, tag = "10")]
    pub last_exit_reason: String,
    #[prost(uint64, optional, tag = "11")]
    pub backoff_remaining_ms: Option<u64>,
    #[prost(uint64, optional, tag = "12")]
    pub drain_remaining_ms: Option<u64>,
    #[prost(bool, tag = "13")]
    pub drain_supported: bool,
    #[prost(uint64, optional, tag = "14")]
    pub active_streams: Option<u64>,
    #[prost(bool, tag = "15")]
    pub cluster_job: bool,
    #[prost(string, tag = "16")]
    pub factory_name: String,
    #[prost(message, optional, tag = "17")]
    pub placement: Option<PlacementSpec>,
    #[prost(string, tag = "18")]
    pub coordinator_node_id: String,
    #[prost(string, tag = "19")]
    pub placement_node_id: String,
    #[prost(uint64, tag = "20")]
    pub placement_generation: u64,
    #[prost(message, repeated, tag = "21")]
    pub placement_history: Vec<ClusterPlacementHistory>,
    #[prost(map = "string, string", tag = "22")]
    pub params: HashMap<String, String>,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct EventFrame {
    #[prost(uint64, tag = "1")]
    pub subscription_id: u64,
    #[prost(message, optional, tag = "2")]
    pub event: Option<Event>,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct Event {
    #[prost(uint64, tag = "1")]
    pub sequence: u64,
    #[prost(uint64, tag = "2")]
    pub timestamp_ms: u64,
    #[prost(string, tag = "3")]
    pub name: String,
    #[prost(uint64, tag = "4")]
    pub job_id: u64,
    #[prost(uint64, tag = "5")]
    pub generation: u64,
    #[prost(string, tag = "6")]
    pub kind: String,
    #[prost(string, tag = "7")]
    pub detail: String,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct MetricFrame {
    #[prost(uint64, tag = "1")]
    pub subscription_id: u64,
    #[prost(message, optional, tag = "2")]
    pub sample: Option<MetricSample>,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct MetricSample {
    #[prost(uint64, tag = "1")]
    pub timestamp_ms: u64,
    #[prost(message, repeated, tag = "2")]
    pub streams: Vec<StreamMetric>,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct StreamMetric {
    #[prost(uint64, tag = "1")]
    pub id: u64,
    #[prost(string, tag = "2")]
    pub name: String,
    #[prost(uint64, tag = "3")]
    pub elements_through: u64,
    #[prost(uint64, tag = "4")]
    pub restarts: u64,
    #[prost(string, tag = "5")]
    pub state: String,
    #[prost(uint64, tag = "6")]
    pub started_at_ms: u64,
    #[prost(uint64, tag = "7")]
    pub state_changed_at_ms: u64,
    #[prost(uint64, optional, tag = "8")]
    pub finished_at_ms: Option<u64>,
    #[prost(uint64, tag = "9")]
    pub uptime_ms: u64,
}

#[derive(Clone, PartialEq, ProstMessage)]
pub struct DcpFrame {
    #[prost(oneof = "dcp_frame::Frame", tags = "1, 2, 3, 4, 5, 6")]
    pub frame: Option<dcp_frame::Frame>,
}

impl DcpFrame {
    #[must_use]
    pub fn hello(hello: Hello) -> Self {
        Self {
            frame: Some(dcp_frame::Frame::Hello(hello)),
        }
    }

    #[must_use]
    pub fn request(request: Request) -> Self {
        Self {
            frame: Some(dcp_frame::Frame::Request(request)),
        }
    }

    #[must_use]
    pub fn response(response: Response) -> Self {
        Self {
            frame: Some(dcp_frame::Frame::Response(response)),
        }
    }

    #[must_use]
    pub fn event(subscription_id: u64, event: Event) -> Self {
        Self {
            frame: Some(dcp_frame::Frame::Event(EventFrame {
                subscription_id,
                event: Some(event),
            })),
        }
    }

    #[must_use]
    pub fn metric(subscription_id: u64, sample: MetricSample) -> Self {
        Self {
            frame: Some(dcp_frame::Frame::Metric(MetricFrame {
                subscription_id,
                sample: Some(sample),
            })),
        }
    }

    #[must_use]
    pub fn shard_pipe(pipe: ShardPipeFrame) -> Self {
        Self {
            frame: Some(dcp_frame::Frame::ShardPipe(pipe)),
        }
    }
}

pub mod dcp_frame {
    #[allow(clippy::large_enum_variant)]
    #[derive(Clone, PartialEq, prost::Oneof)]
    pub enum Frame {
        #[prost(message, tag = "1")]
        Hello(super::Hello),
        #[prost(message, tag = "2")]
        Request(super::Request),
        #[prost(message, tag = "3")]
        Response(super::Response),
        #[prost(message, tag = "4")]
        Event(super::EventFrame),
        #[prost(message, tag = "5")]
        Metric(super::MetricFrame),
        #[prost(message, tag = "6")]
        ShardPipe(super::ShardPipeFrame),
    }
}