netget/server/hl7/
mod.rs

1//! HL7 v2 over MLLP. Rust frames, parses and builds the acknowledgment envelope; the
2//! handler decides what each message is acknowledged with.
3pub mod actions;
4pub mod wire;
5
6use crate::llm::action_helper::call_llm;
7use crate::llm::actions::protocol_trait::ActionResult;
8use crate::logging::emit::Log;
9use crate::protocol::{Event, SpawnContext};
10use crate::server::accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS};
11use crate::server::connection::ConnectionId;
12use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
13use anyhow::{bail, Context, Result};
14use serde_json::{json, Value};
15use std::net::SocketAddr;
16use std::sync::atomic::{AtomicU64, Ordering};
17use std::sync::Arc;
18use std::time::Duration;
19
20pub const IDLE_TIMEOUT: Duration = Duration::from_secs(600);
21
22struct Shared {
23    ctx: SpawnContext,
24    idle: Duration,
25    next_control: AtomicU64,
26}
27
28pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
29    let idle = ctx
30        .startup_params
31        .as_ref()
32        .map(|p| p.get_optional_u64("idle_timeout_secs"))
33        .transpose()?
34        .flatten()
35        .unwrap_or(IDLE_TIMEOUT.as_secs());
36    anyhow::ensure!(
37        (1..=86400).contains(&idle),
38        "idle_timeout_secs must be 1..=86400"
39    );
40    let listener = tokio::net::TcpListener::bind(ctx.legacy_listen_addr()).await?;
41    let local = listener.local_addr()?;
42    Log::new(Some(&ctx.status_tx)).info(format!("HL7 MLLP listening on {local}"));
43    let shared = Arc::new(Shared {
44        ctx: ctx.clone(),
45        idle: Duration::from_secs(idle),
46        next_control: AtomicU64::new(1),
47    });
48    let server_id = ctx.server_id;
49    let accept = tokio::spawn(async move {
50        let limiter = ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
51        loop {
52            let (socket, peer, permit) =
53                match accept_bounded(&listener, &limiter, b"", "HL7", Some(&shared.ctx.status_tx))
54                    .await
55                {
56                    Ok(v) => v,
57                    Err(_) => break,
58                };
59            let id = ConnectionId::new(shared.ctx.state.get_next_unified_id().await);
60            let now = crate::utils::clock::Instant::now();
61            shared
62                .ctx
63                .state
64                .add_connection_to_server(
65                    server_id,
66                    ConnectionState {
67                        id,
68                        remote_addr: peer,
69                        local_addr: local,
70                        bytes_sent: 0,
71                        bytes_received: 0,
72                        packets_sent: 0,
73                        packets_received: 0,
74                        last_activity: now,
75                        status: ConnectionStatus::Active,
76                        status_changed_at: now,
77                        protocol_info: ProtocolConnectionInfo::empty(),
78                    },
79                )
80                .await;
81            let child = shared.clone();
82            shared
83                .ctx
84                .state
85                .spawn_server_task(server_id, async move {
86                    let _permit = permit;
87                    if let Err(e) = connection(&child, id, socket).await {
88                        Log::new(Some(&child.ctx.status_tx))
89                            .debug(format!("HL7 connection {id} ended: {e}"));
90                    }
91                    child
92                        .ctx
93                        .state
94                        .update_connection_status(server_id, id, ConnectionStatus::Closed)
95                        .await;
96                    let _ = child.ctx.status_tx.send("__UPDATE_UI__".into());
97                })
98                .await;
99        }
100    });
101    ctx.state.register_server_task(server_id, accept).await;
102    Ok(local)
103}
104
105fn outcome(ctx: &SpawnContext, id: ConnectionId, operation: &str, decision: &str) {
106    let summary = format!("HL7 connection {id} message={operation} decision={decision}");
107    let log = Log::new(Some(&ctx.status_tx));
108    if decision.starts_with("fail_closed_") || decision == "model_silent" {
109        log.error(summary);
110    } else {
111        log.info(summary);
112    }
113}
114
115async fn decide(shared: &Shared, id: ConnectionId, event: Event) -> Result<Value> {
116    let ctx = &shared.ctx;
117    let result = match call_llm(
118        &ctx.llm_client,
119        &ctx.state,
120        ctx.server_id,
121        Some(id),
122        &event,
123        &actions::Hl7Protocol,
124    )
125    .await
126    {
127        Ok(r) => r,
128        Err(e) => {
129            outcome(ctx, id, event.id(), "fail_closed_llm_error");
130            return Err(e);
131        }
132    };
133    if !result.failures.is_empty() {
134        outcome(ctx, id, event.id(), "fail_closed_invalid_reply");
135        bail!("HL7 handler supplied an invalid action");
136    }
137    let mut found = None;
138    let mut pending = result.protocol_results;
139    while let Some(r) = pending.pop() {
140        match r {
141            ActionResult::Custom { name, data } if name == "hl7_ack" => {
142                if found.replace(data).is_some() {
143                    outcome(ctx, id, event.id(), "fail_closed_invalid_reply");
144                    bail!("HL7 handler supplied more than one acknowledgment");
145                }
146            }
147            ActionResult::Multiple(items) => pending.extend(items),
148            _ => {}
149        }
150    }
151    found
152        .context("HL7 handler did not acknowledge")
153        .inspect_err(|_| outcome(ctx, id, event.id(), "model_silent"))
154}
155
156async fn connection(
157    shared: &Shared,
158    id: ConnectionId,
159    socket: tokio::net::TcpStream,
160) -> Result<()> {
161    let ctx = &shared.ctx;
162    let (mut reader, mut writer) = tokio::io::split(socket);
163    loop {
164        let Some(bytes) = wire::read_frame(&mut reader, shared.idle).await? else {
165            return Ok(());
166        };
167        ctx.state
168            .update_connection_stats(
169                ctx.server_id,
170                id,
171                Some(bytes.len() as u64 + 3),
172                None,
173                Some(1),
174                None,
175            )
176            .await;
177        // An unparseable message cannot be acknowledged: there is no MSH to answer.
178        let message = match wire::parse(&bytes) {
179            Ok(m) => m,
180            Err(e) => {
181                outcome(ctx, id, "unparseable", "protocol_refusal");
182                return Err(e.context("HL7 message could not be parsed; no MSH to acknowledge"));
183            }
184        };
185        let control = format!("NG{}", shared.next_control.fetch_add(1, Ordering::Relaxed));
186        let label = message.message_type().to_owned();
187        let decision = decide(
188            shared,
189            id,
190            Event::new(&actions::MESSAGE_EVENT, message.to_event()),
191        )
192        .await;
193        // The peer gets a category, the log gets the reason.
194        let fallback = || {
195            let error = json!({"code": "207^Application internal error^HL70357", "severity": "E", "message": "the receiver cannot process this message right now"});
196            wire::ack(
197                &message,
198                "AE",
199                "application error",
200                &control,
201                Some(&error),
202                &[],
203            )
204        };
205        let reply = match decision {
206            // `decide` has logged which way it failed.
207            Err(_) => fallback(),
208            Ok(v) => match actions::validate_ack(&v)
209                .and_then(|_| wire::segments_from(v.get("segments").unwrap_or(&Value::Null)))
210            {
211                Ok(extra) => {
212                    let code = v["code"].as_str().unwrap_or("AE");
213                    outcome(
214                        ctx,
215                        id,
216                        &label,
217                        if matches!(code, "AA" | "CA") {
218                            "model_accept"
219                        } else {
220                            "model_reject"
221                        },
222                    );
223                    wire::ack(
224                        &message,
225                        code,
226                        v["text"].as_str().unwrap_or(""),
227                        &control,
228                        v.get("error").filter(|e| !e.is_null()),
229                        &extra,
230                    )
231                }
232                Err(_) => {
233                    outcome(ctx, id, &label, "fail_closed_invalid_reply");
234                    fallback()
235                }
236            },
237        }?;
238        let n = wire::write_frame(&mut writer, &reply).await?;
239        ctx.state
240            .update_connection_stats(ctx.server_id, id, None, Some(n as u64), None, Some(1))
241            .await;
242    }
243}
netget/server/hl7/mod.rs

netget/server/hl7/
mod.rs
