appcore_gateway/
connection.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: connection.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/07/26 08:53:09 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/08/02 12:48:56 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Connection descriptors for workers and clients.
12
13use appcore_contracts::InstallationId;
14use appcore_types::{ClusterId, CoreId, TenantId};
15use axum::extract::ws::Message;
16use std::sync::atomic::{AtomicU64, Ordering};
17use std::sync::Arc;
18use tokio::sync::mpsc::Sender;
19
20/// Maximum number of outbound WebSocket frames buffered per connection.
21pub const CONNECTION_BUFFER_CAPACITY: usize = 128;
22
23// appcore-norm: allow(global-state) reason: atomic generation distinguishes replaced gateway connections
24static CONNECTION_GENERATION: AtomicU64 = AtomicU64::new(1);
25
26/// Unique identifier for a worker connection.
27#[derive(Debug, Clone, PartialEq, Eq, Hash)]
28pub struct WorkerConnectionKey {
29    /// Tenant boundary.
30    pub tenant_id: TenantId,
31    /// Specific installation target.
32    pub installation_id: InstallationId,
33    /// Specific core/worker identity.
34    pub core_id: CoreId,
35}
36
37/// Models an active, authenticated worker WebSocket connection.
38#[derive(Debug, Clone)]
39pub struct WorkerConnection {
40    /// Authentication metadata.
41    pub key: WorkerConnectionKey,
42    /// Channel sender to push frames onto the worker's writer loop.
43    pub sender: Sender<Message>,
44    cluster_id: Option<ClusterId>,
45    generation: u64,
46    /// Last recorded heartbeat epoch in milliseconds.
47    last_heartbeat_ms: Arc<AtomicU64>,
48}
49
50impl WorkerConnection {
51    /// Creates a new worker connection handle.
52    pub fn new(key: WorkerConnectionKey, sender: Sender<Message>, now_ms: u64) -> Self {
53        Self::new_inner(key, None, sender, now_ms)
54    }
55
56    /// Creates a worker connection bound to an explicit cluster.
57    pub fn new_in_cluster(
58        key: WorkerConnectionKey,
59        cluster_id: ClusterId,
60        sender: Sender<Message>,
61        now_ms: u64,
62    ) -> Self {
63        Self::new_inner(key, Some(cluster_id), sender, now_ms)
64    }
65
66    fn new_inner(
67        key: WorkerConnectionKey,
68        cluster_id: Option<ClusterId>,
69        sender: Sender<Message>,
70        now_ms: u64,
71    ) -> Self {
72        Self {
73            key,
74            sender,
75            cluster_id,
76            generation: CONNECTION_GENERATION.fetch_add(1, Ordering::Relaxed),
77            last_heartbeat_ms: Arc::new(AtomicU64::new(now_ms)),
78        }
79    }
80
81    /// Returns the cluster bound during authenticated connection setup.
82    pub fn cluster_id(&self) -> Option<&ClusterId> {
83        self.cluster_id.as_ref()
84    }
85
86    pub(crate) fn generation(&self) -> u64 {
87        self.generation
88    }
89
90    /// Updates the heartbeat timestamp.
91    pub fn update_heartbeat(&self, now_ms: u64) {
92        self.last_heartbeat_ms.store(now_ms, Ordering::SeqCst);
93    }
94
95    /// Gets the last heartbeat timestamp.
96    pub fn last_heartbeat(&self) -> u64 {
97        self.last_heartbeat_ms.load(Ordering::SeqCst)
98    }
99
100    /// Sends a WebSocket message to the worker.
101    pub fn send(&self, message: Message) -> Result<(), crate::error::GatewayError> {
102        self.sender.try_send(message).map_err(|_| {
103            crate::error::GatewayError::Transport("worker connection closed".to_string())
104        })
105    }
106}
107
108/// Models an active client WebSocket connection.
109#[derive(Debug, Clone)]
110pub struct ClientConnection {
111    /// Unique identifier for this connection instance.
112    pub connection_id: String,
113    /// Tenant boundary.
114    pub tenant_id: TenantId,
115    /// Session identifier if authenticated.
116    pub session_id: String,
117    /// Channel sender to push frames onto the client's writer loop.
118    pub sender: Sender<Message>,
119}
120
121impl ClientConnection {
122    /// Creates a new client connection handle.
123    pub fn new(
124        connection_id: String,
125        tenant_id: TenantId,
126        session_id: String,
127        sender: Sender<Message>,
128    ) -> Self {
129        Self {
130            connection_id,
131            tenant_id,
132            session_id,
133            sender,
134        }
135    }
136
137    /// Sends a WebSocket message to the client.
138    pub fn send(&self, message: Message) -> Result<(), crate::error::GatewayError> {
139        self.sender.try_send(message).map_err(|_| {
140            crate::error::GatewayError::Transport("client connection closed".to_string())
141        })
142    }
143}
appcore_gateway/connection.rs

appcore_gateway/
connection.rs
