agent_can/daemon/

server.rs

use crate::can::dbc::{DbcRegistry, FrameIdentity};
use crate::can::{CanSocket, validate_frame};
use crate::daemon::config::DaemonConfig;
use crate::ipc::{self, BoxedLocalStream};
use crate::protocol::{
    ConnectRequest, ConnectResult, EventDirection, LoadedDbc, MessageEntryKind, MessageListEntry,
    MessageListRequest, MessageObservation, MessageReadRequest, MessageReadResult,
    MessageSendRequest, MessageSendResult, MessageStopRequest, Request, RequestAction, Response,
    ResponseData, Selector, SessionStatus, TraceStartRequest, payload_to_hex,
};
use crate::trace::TraceWriter;
use std::collections::{BTreeMap, HashMap, VecDeque};
use std::path::Path;
use std::path::PathBuf;
use std::time::{Duration, Instant, SystemTime};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, split};
use tokio::sync::{mpsc, oneshot, watch};

const MAX_ACTIONS_PER_TURN: usize = 16;
const POLL_INTERVAL: Duration = Duration::from_millis(10);
const RETENTION_WINDOW: Duration = Duration::from_secs(60);
const RETENTION_EVENT_CAP: usize = 4096;
const MAX_REQUEST_LINE_BYTES: usize = 64 * 1024;

pub trait Backend: Send {
    fn recv_all(&mut self) -> Result<Vec<crate::frame::CanFrame>, String>;
    fn send(&mut self, frame: &crate::frame::CanFrame) -> Result<(), String>;
}

#[derive(Debug)]
struct HardwareBackend {
    socket: CanSocket,
}

impl Backend for HardwareBackend {
    fn recv_all(&mut self) -> Result<Vec<crate::frame::CanFrame>, String> {
        self.socket.recv_all()
    }

    fn send(&mut self, frame: &crate::frame::CanFrame) -> Result<(), String> {
        self.socket.send(frame)
    }
}

#[derive(Debug, Clone)]
struct ObservedEvent {
    seq: u64,
    unix_ms: u128,
    recorded_at: Instant,
    direction: EventDirection,
    frame: crate::frame::CanFrame,
}

#[derive(Debug, Clone)]
struct LatestObservation {
    latest_any: ObservedEvent,
    latest_rx: Option<ObservedEvent>,
    latest_tx: Option<ObservedEvent>,
}

impl LatestObservation {
    fn visible(&self, include_tx: bool) -> Option<&ObservedEvent> {
        if include_tx {
            Some(&self.latest_any)
        } else {
            self.latest_rx.as_ref()
        }
    }

    fn has_rx(&self) -> bool {
        self.latest_rx.is_some()
    }

    fn has_tx(&self) -> bool {
        self.latest_tx.is_some()
    }
}

#[derive(Debug, Clone)]
struct PeriodicScheduleState {
    target: String,
    frame: crate::frame::CanFrame,
    periodicity_ms: u64,
    next_due: Instant,
}

pub struct SessionService<B: Backend> {
    connect: ConnectRequest,
    dbcs: DbcRegistry,
    backend: B,
    events: VecDeque<ObservedEvent>,
    latest: HashMap<FrameIdentity, LatestObservation>,
    schedules: BTreeMap<String, PeriodicScheduleState>,
    trace: Option<Box<dyn ActiveTrace>>,
    backend_error: Option<String>,
    next_seq: u64,
    shutdown: bool,
}

struct ActionMessage {
    request: Request,
    response_tx: oneshot::Sender<Response>,
}

trait ActiveTrace: Send {
    fn path(&self) -> &Path;
    fn write_event(
        &mut self,
        direction: EventDirection,
        frame: &crate::frame::CanFrame,
    ) -> Result<(), String>;
    fn finish(self: Box<Self>) -> Result<PathBuf, String>;
}

impl ActiveTrace for TraceWriter {
    fn path(&self) -> &Path {
        self.path()
    }

    fn write_event(
        &mut self,
        direction: EventDirection,
        frame: &crate::frame::CanFrame,
    ) -> Result<(), String> {
        self.write_event(direction, frame)
    }

    fn finish(self: Box<Self>) -> Result<PathBuf, String> {
        TraceWriter::finish(*self)
    }
}

impl<B: Backend> SessionService<B> {
    pub fn new(connect: ConnectRequest, dbcs: DbcRegistry, backend: B) -> Self {
        Self {
            connect,
            dbcs,
            backend,
            events: VecDeque::new(),
            latest: HashMap::new(),
            schedules: BTreeMap::new(),
            trace: None,
            backend_error: None,
            next_seq: 1,
            shutdown: false,
        }
    }

    pub fn handle_request(&mut self, request: Request) -> Response {
        let id = request.id;
        let result: Result<ResponseData, String> = match request.action {
            RequestAction::AdaptersList => Ok(ResponseData::AdaptersList {
                adapters: crate::can::available_adapters(),
            }),
            RequestAction::Connect(connect) => {
                self.handle_connect(connect).map(ResponseData::Connected)
            }
            RequestAction::Disconnect => self.handle_disconnect(),
            RequestAction::Status => Ok(ResponseData::Status(self.status())),
            RequestAction::Schema(schema) => schema
                .filter
                .as_deref()
                .map(Selector::parse)
                .transpose()
                .map(|filter| ResponseData::Schema {
                    messages: self.dbcs.schema(filter.as_ref()),
                }),
            RequestAction::MessageList(request) => self.handle_message_list(request),
            RequestAction::MessageRead(request) => self.handle_message_read(request),
            RequestAction::MessageSend(request) => self.handle_message_send(request),
            RequestAction::MessageStop(request) => self.handle_message_stop(request),
            RequestAction::TraceStart(request) => self.handle_trace_start(request),
            RequestAction::TraceStop => self.handle_trace_stop(),
        };
        match result {
            Ok(data) => Response::ok(id, data),
            Err(err) => Response::err(id, err),
        }
    }

    pub fn tick(&mut self) {
        self.poll_backend();
        self.tick_schedules();
        self.trim_events();
    }

    pub fn should_shutdown(&self) -> bool {
        self.shutdown
    }

    fn handle_connect(&self, connect: ConnectRequest) -> Result<ConnectResult, String> {
        if connect_identity(&connect) == connect_identity(&self.connect) {
            Ok(ConnectResult {
                created: false,
                already_connected: true,
                status: self.status(),
            })
        } else {
            Err(
                "session already exists with different open parameters; disconnect first"
                    .to_string(),
            )
        }
    }

    fn handle_disconnect(&mut self) -> Result<ResponseData, String> {
        if let Err(err) = self.stop_trace() {
            self.backend_error = Some(err);
        }
        self.schedules.clear();
        self.shutdown = true;
        Ok(ResponseData::Disconnected)
    }

    fn handle_message_list(&self, request: MessageListRequest) -> Result<ResponseData, String> {
        let filter = request.filter.as_deref().map(Selector::parse).transpose()?;
        let mut out = Vec::new();
        for latest in self.latest.values() {
            let Some(last) = latest.visible(request.include_tx) else {
                continue;
            };
            let matches = self.dbcs.matches_for_frame(&last.frame);
            if matches.is_empty() {
                if (self.dbcs.is_empty() || request.allow_raw)
                    && filter
                        .as_ref()
                        .is_none_or(|selector| selector.matches_arb_id(last.frame.arb_id))
                {
                    out.push(MessageListEntry {
                        label: format!("0x{:X}", last.frame.arb_id),
                        kind: MessageEntryKind::Raw,
                        arb_id: last.frame.arb_id,
                        extended: FrameIdentity::from_frame(&last.frame).extended,
                        fd: (last.frame.flags & crate::frame::CAN_FLAG_FD) != 0,
                        len: last.frame.len,
                        last_seen_unix_ms: last.unix_ms,
                        has_rx: latest.has_rx(),
                        has_tx: latest.has_tx(),
                    });
                }
                continue;
            }

            for message in matches {
                if filter.as_ref().is_some_and(|selector| match selector {
                    Selector::ArbId(arb_id) => *arb_id != message.arb_id,
                    _ => !selector.matches_qualified_name(&message.qualified_name),
                }) {
                    continue;
                }
                out.push(MessageListEntry {
                    label: message.qualified_name.clone(),
                    kind: MessageEntryKind::Semantic,
                    arb_id: message.arb_id,
                    extended: message.extended,
                    fd: message.len > 8,
                    len: last.frame.len,
                    last_seen_unix_ms: last.unix_ms,
                    has_rx: latest.has_rx(),
                    has_tx: latest.has_tx(),
                });
            }
        }
        out.sort_by(|lhs, rhs| lhs.label.cmp(&rhs.label));
        Ok(ResponseData::MessageList { messages: out })
    }

    fn handle_message_read(&self, request: MessageReadRequest) -> Result<ResponseData, String> {
        let selector = Selector::parse(&request.select)?;
        let count = request.count.unwrap_or(1);
        let observations = match selector {
            Selector::ArbId(arb_id) => self
                .events
                .iter()
                .rev()
                .filter(|event| {
                    event.frame.arb_id == arb_id && include_direction(event, request.include_tx)
                })
                .take(count)
                .map(raw_observation)
                .collect::<Vec<_>>(),
            Selector::SemanticPattern(_) => {
                let message = self.dbcs.resolve_selector(&selector)?;
                self.events
                    .iter()
                    .rev()
                    .filter(|event| {
                        FrameIdentity::from_frame(&event.frame)
                            == FrameIdentity::from_message(message)
                            && include_direction(event, request.include_tx)
                    })
                    .take(count)
                    .map(|event| {
                        semantic_observation(
                            event,
                            self.dbcs
                                .decode_selected(&message.qualified_name, &event.frame),
                        )
                    })
                    .collect::<Result<Vec<_>, String>>()?
            }
        };
        if observations.is_empty() {
            return Err(format!(
                "selector '{}' matched no observed traffic",
                request.select
            ));
        }
        Ok(ResponseData::MessageRead(MessageReadResult {
            selector: request.select,
            count: observations.len(),
            observations,
        }))
    }

    fn handle_message_send(&mut self, request: MessageSendRequest) -> Result<ResponseData, String> {
        let selector = Selector::parse(&request.target)?;
        let (target, frame) = self.dbcs.encode_payload(&selector, &request.data)?;
        validate_frame(self.connect.fd, &frame)?;
        if let Some(periodicity_ms) = request.periodicity_ms {
            self.schedules.insert(
                target.clone(),
                PeriodicScheduleState {
                    target: target.clone(),
                    frame: frame.clone(),
                    periodicity_ms,
                    next_due: next_due_after(Instant::now(), periodicity_ms),
                },
            );
        } else {
            self.backend.send(&frame)?;
            self.record_event(EventDirection::Tx, frame.clone())?;
            self.backend_error = None;
        }
        Ok(ResponseData::MessageSent(MessageSendResult {
            target,
            arb_id: frame.arb_id,
            extended: (frame.flags & crate::frame::CAN_FLAG_EXTENDED) != 0,
            fd: (frame.flags & crate::frame::CAN_FLAG_FD) != 0,
            len: frame.len,
            periodicity_ms: request.periodicity_ms,
        }))
    }

    fn handle_message_stop(&mut self, request: MessageStopRequest) -> Result<ResponseData, String> {
        let selector = Selector::parse(&request.target)?;
        let target = match selector {
            Selector::ArbId(arb_id) => format!("0x{arb_id:X}"),
            Selector::SemanticPattern(_) => self
                .dbcs
                .resolve_selector(&selector)?
                .qualified_name
                .clone(),
        };
        let stopped = self.schedules.remove(&target).is_some();
        Ok(ResponseData::MessageStopped { target, stopped })
    }

    fn handle_trace_start(&mut self, request: TraceStartRequest) -> Result<ResponseData, String> {
        if self.trace.is_some() {
            return Err("trace export already active; stop it before starting another".to_string());
        }
        let writer = TraceWriter::start(PathBuf::from(&request.path).as_path())?;
        let path = writer.path().display().to_string();
        self.trace = Some(Box::new(writer));
        Ok(ResponseData::TraceStarted { path })
    }

    fn handle_trace_stop(&mut self) -> Result<ResponseData, String> {
        let path = self.stop_trace()?;
        Ok(ResponseData::TraceStopped { path })
    }

    fn stop_trace(&mut self) -> Result<Option<String>, String> {
        if let Some(writer) = self.trace.take() {
            return writer.finish().map(|path| Some(path.display().to_string()));
        }
        Ok(None)
    }

    fn status(&self) -> SessionStatus {
        let mut periodic_schedules = self
            .schedules
            .values()
            .map(|schedule| crate::protocol::PeriodicSchedule {
                target: schedule.target.clone(),
                arb_id: schedule.frame.arb_id,
                extended: (schedule.frame.flags & crate::frame::CAN_FLAG_EXTENDED) != 0,
                fd: (schedule.frame.flags & crate::frame::CAN_FLAG_FD) != 0,
                len: schedule.frame.len,
                periodicity_ms: schedule.periodicity_ms,
            })
            .collect::<Vec<_>>();
        periodic_schedules.sort_by(|lhs, rhs| lhs.target.cmp(&rhs.target));

        SessionStatus {
            connection_state: if self.backend_error.is_some() {
                "degraded".to_string()
            } else {
                "connected".to_string()
            },
            adapter: self.connect.adapter.clone(),
            bitrate: self.connect.bitrate,
            bitrate_data: self.connect.bitrate_data,
            fd: self.connect.fd,
            dbcs: self
                .dbcs
                .loaded()
                .iter()
                .map(|dbc| LoadedDbc {
                    alias: dbc.alias.clone(),
                    path: dbc.path.clone(),
                })
                .collect(),
            trace_path: self
                .trace
                .as_ref()
                .map(|writer| writer.path().display().to_string()),
            periodic_schedules,
            backend_error: self.backend_error.clone(),
            retention_window_secs: RETENTION_WINDOW.as_secs(),
            retention_event_cap: RETENTION_EVENT_CAP,
        }
    }

    fn poll_backend(&mut self) {
        let mut error = None;
        match self.backend.recv_all() {
            Ok(frames) => {
                for frame in frames {
                    if let Err(err) = validate_frame(self.connect.fd, &frame)
                        .and_then(|_| self.record_event(EventDirection::Rx, frame))
                    {
                        error.get_or_insert(err);
                    }
                }
            }
            Err(err) => error = Some(err),
        }
        self.backend_error = error;
    }

    fn tick_schedules(&mut self) {
        let now = Instant::now();
        let due = self
            .schedules
            .values()
            .filter(|schedule| schedule.next_due <= now)
            .map(|schedule| schedule.target.clone())
            .collect::<Vec<_>>();
        let had_due = !due.is_empty();
        let mut error = None;
        for target in due {
            let Some((frame, periodicity_ms)) = self
                .schedules
                .get(&target)
                .map(|schedule| (schedule.frame.clone(), schedule.periodicity_ms))
            else {
                continue;
            };
            let next_due = next_due_after(now, periodicity_ms);
            match self.backend.send(&frame) {
                Ok(()) => {
                    if let Err(err) = self.record_event(EventDirection::Tx, frame) {
                        error.get_or_insert(err);
                    }
                }
                Err(err) => {
                    error.get_or_insert(err);
                }
            }
            if let Some(schedule) = self.schedules.get_mut(&target) {
                schedule.next_due = next_due;
            }
        }
        if had_due {
            self.backend_error = error;
        }
    }

    fn record_event(
        &mut self,
        direction: EventDirection,
        frame: crate::frame::CanFrame,
    ) -> Result<(), String> {
        let event = ObservedEvent {
            seq: self.next_seq,
            unix_ms: unix_ms_now()?,
            recorded_at: Instant::now(),
            direction: direction.clone(),
            frame,
        };
        self.next_seq += 1;
        let trace_result = if let Some(trace) = self.trace.as_mut() {
            trace.write_event(direction.clone(), &event.frame)
        } else {
            Ok(())
        };
        self.events.push_back(event.clone());
        self.update_latest(&event);
        self.trim_events();
        trace_result
    }

    fn trim_events(&mut self) {
        let cutoff = Instant::now()
            .checked_sub(RETENTION_WINDOW)
            .unwrap_or_else(Instant::now);
        while self.events.len() > RETENTION_EVENT_CAP
            || self
                .events
                .front()
                .is_some_and(|event| event.recorded_at < cutoff)
        {
            if let Some(event) = self.events.pop_front() {
                self.reconcile_trimmed_event(&event);
            }
        }
    }

    fn update_latest(&mut self, event: &ObservedEvent) {
        let entry = self
            .latest
            .entry(FrameIdentity::from_frame(&event.frame))
            .or_insert_with(|| LatestObservation {
                latest_any: event.clone(),
                latest_rx: None,
                latest_tx: None,
            });
        entry.latest_any = event.clone();
        match event.direction {
            EventDirection::Rx => entry.latest_rx = Some(event.clone()),
            EventDirection::Tx => entry.latest_tx = Some(event.clone()),
        }
    }

    fn reconcile_trimmed_event(&mut self, event: &ObservedEvent) {
        let identity = FrameIdentity::from_frame(&event.frame);
        let Some(latest) = self.latest.get(&identity) else {
            return;
        };
        let touched_latest = latest.latest_any.seq == event.seq
            || latest
                .latest_rx
                .as_ref()
                .is_some_and(|latest_rx| latest_rx.seq == event.seq)
            || latest
                .latest_tx
                .as_ref()
                .is_some_and(|latest_tx| latest_tx.seq == event.seq);
        if touched_latest {
            self.rebuild_latest_for(identity);
        }
    }

    fn rebuild_latest_for(&mut self, identity: FrameIdentity) {
        let mut latest_any = None;
        let mut latest_rx = None;
        let mut latest_tx = None;

        for event in self.events.iter().rev() {
            if FrameIdentity::from_frame(&event.frame) != identity {
                continue;
            }
            if latest_any.is_none() {
                latest_any = Some(event.clone());
            }
            match event.direction {
                EventDirection::Rx if latest_rx.is_none() => latest_rx = Some(event.clone()),
                EventDirection::Tx if latest_tx.is_none() => latest_tx = Some(event.clone()),
                _ => {}
            }
            if latest_any.is_some() && latest_rx.is_some() && latest_tx.is_some() {
                break;
            }
        }

        if let Some(latest_any) = latest_any {
            self.latest.insert(
                identity,
                LatestObservation {
                    latest_any,
                    latest_rx,
                    latest_tx,
                },
            );
        } else {
            self.latest.remove(&identity);
        }
    }
}

pub async fn run_listener(
    socket_path: PathBuf,
    config: DaemonConfig,
) -> Result<(), std::io::Error> {
    if let Some(parent) = socket_path.parent() {
        std::fs::create_dir_all(parent)?;
    }

    let socket = CanSocket::open(
        &config.connect.adapter,
        config.connect.bitrate,
        config.connect.bitrate_data.unwrap_or(0),
        config.connect.fd,
    )
    .map_err(std::io::Error::other)?;
    let dbcs = DbcRegistry::load(&config.connect.dbcs).map_err(std::io::Error::other)?;
    let mut listener = ipc::bind_listener(&socket_path).await?;
    ipc::create_endpoint_marker(&socket_path)?;
    std::fs::write(
        crate::daemon::lifecycle::pid_path(),
        std::process::id().to_string(),
    )?;

    let state = SessionService::new(config.connect, dbcs, HardwareBackend { socket });

    let (action_tx, action_rx) = mpsc::channel::<ActionMessage>(256);
    let (shutdown_tx, mut shutdown_rx) = watch::channel(false);
    let actor_task = tokio::spawn(run_actor_task(state, action_rx, shutdown_tx));

    let mut listener_error = None;
    loop {
        tokio::select! {
            changed = shutdown_rx.changed() => {
                match changed {
                    Ok(()) if *shutdown_rx.borrow() => break,
                    Ok(()) => {}
                    Err(_) => break,
                }
            }
            accepted = listener.accept() => {
                match accepted {
                    Ok(stream) => {
                        let action_tx = action_tx.clone();
                        tokio::spawn(async move {
                            let _ = handle_connection(stream, action_tx).await;
                        });
                    }
                    Err(err) => {
                        listener_error = Some(err);
                        break;
                    }
                }
            }
        }
    }

    drop(action_tx);
    let _ = actor_task.await;
    ipc::cleanup_endpoint(&socket_path);
    let pid_path = crate::daemon::lifecycle::pid_path();
    if pid_path.exists() {
        let _ = std::fs::remove_file(pid_path);
    }

    if let Some(err) = listener_error {
        return Err(err);
    }
    Ok(())
}

async fn handle_connection(
    stream: BoxedLocalStream,
    action_tx: mpsc::Sender<ActionMessage>,
) -> Result<(), std::io::Error> {
    let (read_half, mut write_half) = split(stream);
    let mut reader = BufReader::new(read_half);
    let mut line = Vec::new();

    loop {
        let read = read_request_line(&mut reader, &mut line).await?;
        if read == 0 {
            return Ok(());
        }
        let line = std::str::from_utf8(&line)
            .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?;
        let response = match serde_json::from_str::<Request>(line.trim_end()) {
            Ok(request) => {
                let request_id = request.id;
                let (response_tx, response_rx) = oneshot::channel();
                if action_tx
                    .send(ActionMessage {
                        request,
                        response_tx,
                    })
                    .await
                    .is_err()
                {
                    Response::err(request_id, "daemon unavailable")
                } else {
                    match response_rx.await {
                        Ok(response) => response,
                        Err(_) => Response::err(request_id, "daemon unavailable"),
                    }
                }
            }
            Err(err) => Response::err(uuid::Uuid::new_v4(), format!("invalid request json: {err}")),
        };
        let mut payload = serde_json::to_string(&response).unwrap_or_else(|err| {
            format!("{{\"success\":false,\"error\":\"response serialization failed: {err}\"}}")
        });
        payload.push('\n');
        write_half.write_all(payload.as_bytes()).await?;
    }
}

async fn read_request_line<R>(
    reader: &mut BufReader<R>,
    line: &mut Vec<u8>,
) -> Result<usize, std::io::Error>
where
    R: tokio::io::AsyncRead + Unpin,
{
    line.clear();
    loop {
        let available = reader.fill_buf().await?;
        if available.is_empty() {
            return Ok(line.len());
        }

        let take = available
            .iter()
            .position(|byte| *byte == b'\n')
            .map(|index| index + 1)
            .unwrap_or(available.len());
        if line.len() + take > MAX_REQUEST_LINE_BYTES {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!("request line exceeds {MAX_REQUEST_LINE_BYTES} bytes"),
            ));
        }
        line.extend_from_slice(&available[..take]);
        reader.consume(take);
        if line.last() == Some(&b'\n') {
            return Ok(line.len());
        }
    }
}

async fn run_actor_task<B: Backend>(
    mut state: SessionService<B>,
    mut action_rx: mpsc::Receiver<ActionMessage>,
    shutdown_tx: watch::Sender<bool>,
) {
    let mut poller = tokio::time::interval(POLL_INTERVAL);
    poller.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

    loop {
        tokio::select! {
            maybe_message = action_rx.recv() => match maybe_message {
                Some(message) => {
                    handle_action_message(&mut state, message);
                    process_action_batch(&mut action_rx, &mut state, MAX_ACTIONS_PER_TURN.saturating_sub(1));
                }
                None => break,
            },
            _ = poller.tick() => state.tick(),
        }

        if state.should_shutdown() {
            let _ = shutdown_tx.send(true);
            break;
        }
    }
}

fn handle_action_message<B: Backend>(state: &mut SessionService<B>, message: ActionMessage) {
    let response = state.handle_request(message.request);
    let _ = message.response_tx.send(response);
}

fn process_action_batch<B: Backend>(
    action_rx: &mut mpsc::Receiver<ActionMessage>,
    state: &mut SessionService<B>,
    limit: usize,
) {
    for _ in 0..limit {
        let Some(message) = action_rx.try_recv().ok() else {
            break;
        };
        handle_action_message(state, message);
    }
}

fn next_due_after(now: Instant, periodicity_ms: u64) -> Instant {
    now + Duration::from_millis(periodicity_ms.max(1))
}

fn connect_identity(connect: &ConnectRequest) -> ConnectRequest {
    let mut normalized = connect.clone();
    normalized.dbcs.sort();
    normalized
}

fn include_direction(event: &ObservedEvent, include_tx: bool) -> bool {
    include_tx || matches!(event.direction, EventDirection::Rx)
}

fn raw_observation(event: &ObservedEvent) -> MessageObservation {
    MessageObservation::Raw {
        seq: event.seq,
        direction: event.direction.clone(),
        unix_ms: event.unix_ms,
        arb_id: event.frame.arb_id,
        extended: (event.frame.flags & crate::frame::CAN_FLAG_EXTENDED) != 0,
        fd: (event.frame.flags & crate::frame::CAN_FLAG_FD) != 0,
        len: event.frame.len,
        payload_hex: payload_to_hex(event.frame.payload()),
    }
}

fn semantic_observation(
    event: &ObservedEvent,
    decoded: Result<crate::can::dbc::DecodedSemanticMessage, String>,
) -> Result<MessageObservation, String> {
    let decoded = decoded?;
    Ok(MessageObservation::Semantic {
        seq: event.seq,
        direction: event.direction.clone(),
        unix_ms: event.unix_ms,
        qualified_name: decoded.qualified_name,
        arb_id: decoded.arb_id,
        extended: decoded.extended,
        fd: decoded.fd,
        len: decoded.len,
        payload_hex: payload_to_hex(event.frame.payload()),
        signals: decoded.signals,
    })
}

fn unix_ms_now() -> Result<u128, String> {
    Ok(SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .map_err(|err| format!("system clock error: {err}"))?
        .as_millis())
}

#[cfg(test)]
mod tests {
    use super::{
        ActiveTrace, Backend, EventDirection, MAX_REQUEST_LINE_BYTES, MessageListRequest,
        RETENTION_WINDOW, Request, RequestAction, ResponseData, SessionService, TraceStartRequest,
        read_request_line,
    };
    use crate::can::dbc::DbcRegistry;
    use crate::frame::CanFrame;
    use crate::protocol::{
        ConnectRequest, DbcSpec, MessagePayload, MessageReadRequest, MessageSendRequest,
        MessageStopRequest,
    };
    use std::collections::{BTreeMap, VecDeque};
    use std::io::Write;
    use std::path::{Path, PathBuf};
    use std::time::{Duration, Instant};
    use tokio::io::{AsyncWriteExt, BufReader, duplex};
    use uuid::Uuid;

    #[derive(Debug, Default)]
    struct MockBackend {
        recv_results: VecDeque<Result<Vec<CanFrame>, String>>,
        send_results: VecDeque<Result<(), String>>,
        sent: Vec<CanFrame>,
    }

    impl Backend for MockBackend {
        fn recv_all(&mut self) -> Result<Vec<CanFrame>, String> {
            self.recv_results
                .pop_front()
                .unwrap_or_else(|| Ok(Vec::new()))
        }

        fn send(&mut self, frame: &CanFrame) -> Result<(), String> {
            if let Some(result) = self.send_results.pop_front() {
                return result.map(|()| self.sent.push(frame.clone()));
            }
            self.sent.push(frame.clone());
            Ok(())
        }
    }

    struct FailingTrace {
        path: PathBuf,
        finish_error: Option<String>,
        write_error: Option<String>,
    }

    impl FailingTrace {
        fn on_finish(message: &str) -> Self {
            Self {
                path: PathBuf::from("/tmp/failing.asc"),
                finish_error: Some(message.to_string()),
                write_error: None,
            }
        }

        fn on_write(message: &str) -> Self {
            Self {
                path: PathBuf::from("/tmp/failing.asc"),
                finish_error: None,
                write_error: Some(message.to_string()),
            }
        }
    }

    impl ActiveTrace for FailingTrace {
        fn path(&self) -> &Path {
            &self.path
        }

        fn write_event(
            &mut self,
            _direction: EventDirection,
            _frame: &CanFrame,
        ) -> Result<(), String> {
            if let Some(err) = &self.write_error {
                return Err(err.clone());
            }
            Ok(())
        }

        fn finish(self: Box<Self>) -> Result<PathBuf, String> {
            if let Some(err) = self.finish_error {
                return Err(err);
            }
            Ok(self.path)
        }
    }

    fn write_temp_dbc(contents: &str) -> tempfile::NamedTempFile {
        let mut file = tempfile::NamedTempFile::new().expect("tempfile");
        write!(file, "{contents}").expect("write dbc");
        file
    }

    fn sample_service(overlap: bool) -> SessionService<MockBackend> {
        let dbc_a = write_temp_dbc(
            "VERSION \"\"\nBO_ 291 Foo: 8 ECU\n SG_ enable : 0|8@1+ (1,0) [0|1] \"\" ECU\n",
        );
        let dbc_b = write_temp_dbc(
            "VERSION \"\"\nBO_ 291 Bar: 8 ECU\n SG_ enable : 0|8@1+ (1,0) [0|255] \"\" ECU\n",
        );
        let dbcs = if overlap {
            vec![
                DbcSpec {
                    alias: "a".to_string(),
                    path: dbc_a.path().display().to_string(),
                },
                DbcSpec {
                    alias: "b".to_string(),
                    path: dbc_b.path().display().to_string(),
                },
            ]
        } else {
            vec![DbcSpec {
                alias: "a".to_string(),
                path: dbc_a.path().display().to_string(),
            }]
        };
        let registry = DbcRegistry::load(&dbcs).expect("registry");
        SessionService::new(
            ConnectRequest {
                adapter: "usb1".to_string(),
                bitrate: 500_000,
                bitrate_data: None,
                fd: false,
                dbcs,
            },
            registry,
            MockBackend::default(),
        )
    }

    fn empty_service() -> SessionService<MockBackend> {
        SessionService::new(
            ConnectRequest {
                adapter: "usb1".to_string(),
                bitrate: 500_000,
                bitrate_data: None,
                fd: false,
                dbcs: Vec::new(),
            },
            DbcRegistry::empty(),
            MockBackend::default(),
        )
    }

    fn raw_frame(arb_id: u32, first_byte: u8) -> CanFrame {
        raw_frame_with_len(arb_id, first_byte, 8)
    }

    fn raw_frame_with_len(arb_id: u32, first_byte: u8, len: u8) -> CanFrame {
        let mut data = [0_u8; 64];
        data[0] = first_byte;
        CanFrame {
            arb_id,
            len,
            flags: 0,
            data,
        }
    }

    #[test]
    fn connect_same_session_attaches_and_different_session_fails() {
        let service = sample_service(false);
        let same = service
            .handle_connect(service.connect.clone())
            .expect("same config must attach");
        assert!(same.already_connected);

        let err = service
            .handle_connect(ConnectRequest {
                adapter: "usb2".to_string(),
                ..service.connect.clone()
            })
            .expect_err("different config must fail");
        assert!(err.contains("disconnect first"));
    }

    #[test]
    fn connect_succeeds_with_zero_dbcs_loaded() {
        let service = empty_service();
        let same = service
            .handle_connect(service.connect.clone())
            .expect("same config must attach");
        assert!(same.already_connected);
        assert!(same.status.dbcs.is_empty());
    }

    #[test]
    fn schema_reports_semantic_inventory_while_message_list_stays_empty_without_traffic() {
        let mut service = sample_service(false);

        let ResponseData::Schema { messages } = service
            .handle_request(Request {
                id: Uuid::new_v4(),
                action: RequestAction::Schema(crate::protocol::SchemaRequest { filter: None }),
            })
            .data
            .expect("schema data")
        else {
            panic!("wrong schema response");
        };
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].qualified_name, "a.Foo");

        let ResponseData::MessageList { messages } = service
            .handle_message_list(MessageListRequest {
                filter: None,
                allow_raw: false,
                include_tx: false,
            })
            .expect("message list")
        else {
            panic!("wrong message list response");
        };
        assert!(messages.is_empty());
    }

    #[test]
    fn message_list_splits_overlap_entries() {
        let mut service = sample_service(true);
        service
            .record_event(EventDirection::Rx, raw_frame(291, 1))
            .expect("record");
        let data = service
            .handle_message_list(MessageListRequest {
                filter: None,
                allow_raw: false,
                include_tx: false,
            })
            .expect("message list");
        let ResponseData::MessageList { messages } = data else {
            panic!("wrong response");
        };
        assert_eq!(messages.len(), 2);
        assert_eq!(messages[0].label, "a.Foo");
        assert_eq!(messages[1].label, "b.Bar");
    }

    #[test]
    fn message_list_excluding_tx_uses_latest_rx_observation() {
        let mut service = sample_service(false);
        service
            .record_event(EventDirection::Rx, raw_frame_with_len(0x123, 1, 8))
            .expect("record rx");
        service
            .record_event(EventDirection::Tx, raw_frame_with_len(0x123, 2, 1))
            .expect("record tx");

        let ResponseData::MessageList { messages } = service
            .handle_message_list(MessageListRequest {
                filter: None,
                allow_raw: false,
                include_tx: false,
            })
            .expect("message list")
        else {
            panic!("wrong response");
        };

        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].len, 8);
        assert!(messages[0].has_rx);
        assert!(messages[0].has_tx);
    }

    #[test]
    fn message_read_decodes_selected_alias() {
        let mut service = sample_service(true);
        service
            .record_event(EventDirection::Rx, raw_frame(291, 7))
            .expect("record");
        let data = service
            .handle_message_read(MessageReadRequest {
                select: "b.Bar".to_string(),
                count: None,
                include_tx: false,
            })
            .expect("message read");
        let ResponseData::MessageRead(result) = data else {
            panic!("wrong response");
        };
        let crate::protocol::MessageObservation::Semantic { qualified_name, .. } =
            &result.observations[0]
        else {
            panic!("expected semantic observation");
        };
        assert_eq!(qualified_name, "b.Bar");
    }

    #[test]
    fn semantic_send_requires_full_signal_map_and_periodic_stop_works() {
        let dbc = write_temp_dbc(
            "VERSION \"\"\nBO_ 291 Foo: 8 ECU\n SG_ enable : 0|8@1+ (1,0) [0|1] \"\" ECU\n SG_ torque : 8|8@1+ (1,0) [0|255] \"\" ECU\n",
        );
        let dbcs = vec![DbcSpec {
            alias: "a".to_string(),
            path: dbc.path().display().to_string(),
        }];
        let registry = DbcRegistry::load(&dbcs).expect("registry");
        let mut service = SessionService::new(
            ConnectRequest {
                adapter: "usb1".to_string(),
                bitrate: 500_000,
                bitrate_data: None,
                fd: false,
                dbcs,
            },
            registry,
            MockBackend::default(),
        );

        let mut incomplete = BTreeMap::new();
        incomplete.insert("enable".to_string(), 1.0);
        let err = service
            .handle_message_send(MessageSendRequest {
                target: "a.Foo".to_string(),
                data: MessagePayload::Signals(incomplete),
                periodicity_ms: None,
            })
            .expect_err("missing signal");
        assert!(err.contains("missing required signal 'torque'"));

        let data = service
            .handle_message_send(MessageSendRequest {
                target: "0x123".to_string(),
                data: MessagePayload::RawHex("DEADBEEF".to_string()),
                periodicity_ms: Some(100),
            })
            .expect("schedule");
        let ResponseData::MessageSent(sent) = data else {
            panic!("wrong response");
        };
        assert_eq!(sent.target, "0x123");

        let stopped = service
            .handle_message_stop(MessageStopRequest {
                target: "0x123".to_string(),
            })
            .expect("stop");
        let ResponseData::MessageStopped { stopped, .. } = stopped else {
            panic!("wrong response");
        };
        assert!(stopped);
    }

    #[test]
    fn periodic_send_overwrites_existing_schedule_for_same_target() {
        let mut service = sample_service(false);
        service
            .handle_message_send(MessageSendRequest {
                target: "0x123".to_string(),
                data: MessagePayload::RawHex("AAAA".to_string()),
                periodicity_ms: Some(100),
            })
            .expect("initial schedule");

        service
            .handle_message_send(MessageSendRequest {
                target: "0x123".to_string(),
                data: MessagePayload::RawHex("BBBB".to_string()),
                periodicity_ms: Some(250),
            })
            .expect("overwrite schedule");

        assert_eq!(service.schedules.len(), 1);
        let schedule = service.schedules.get("0x123").expect("schedule");
        assert_eq!(schedule.periodicity_ms, 250);
        assert_eq!(
            crate::protocol::payload_to_hex(schedule.frame.payload()),
            "BBBB"
        );
    }

    #[test]
    fn semantic_periodic_send_stores_raw_encoded_schedule_state() {
        let dbc = write_temp_dbc(
            "VERSION \"\"\nBO_ 291 Foo: 8 ECU\n SG_ enable : 0|8@1+ (1,0) [0|1] \"\" ECU\n SG_ torque : 8|8@1+ (1,0) [0|255] \"\" ECU\n",
        );
        let dbcs = vec![DbcSpec {
            alias: "a".to_string(),
            path: dbc.path().display().to_string(),
        }];
        let registry = DbcRegistry::load(&dbcs).expect("registry");
        let mut service = SessionService::new(
            ConnectRequest {
                adapter: "usb1".to_string(),
                bitrate: 500_000,
                bitrate_data: None,
                fd: false,
                dbcs,
            },
            registry,
            MockBackend::default(),
        );

        let mut data = BTreeMap::new();
        data.insert("enable".to_string(), 1.0);
        data.insert("torque".to_string(), 12.0);
        service
            .handle_message_send(MessageSendRequest {
                target: "a.Foo".to_string(),
                data: MessagePayload::Signals(data),
                periodicity_ms: Some(100),
            })
            .expect("semantic schedule");

        let schedule = service.schedules.get("a.Foo").expect("schedule");
        assert_eq!(schedule.frame.arb_id, 291);
        assert_eq!(schedule.frame.len, 8);
        assert_eq!(
            crate::protocol::payload_to_hex(schedule.frame.payload()),
            "010C000000000000"
        );
    }

    #[test]
    fn message_read_returns_error_when_no_matching_traffic_exists() {
        let service = sample_service(false);
        let err = service
            .handle_message_read(MessageReadRequest {
                select: "a.Foo".to_string(),
                count: None,
                include_tx: false,
            })
            .expect_err("empty traffic must fail");
        assert!(err.contains("matched no observed traffic"));
    }

    #[test]
    fn disconnect_still_shuts_down_when_trace_finish_fails() {
        let mut service = sample_service(false);
        service.trace = Some(Box::new(FailingTrace::on_finish("flush failed")));
        service.schedules.insert(
            "0x123".to_string(),
            super::PeriodicScheduleState {
                target: "0x123".to_string(),
                frame: raw_frame(0x123, 1),
                periodicity_ms: 100,
                next_due: Instant::now() + Duration::from_millis(100),
            },
        );

        let response = service.handle_disconnect().expect("disconnect");
        assert!(matches!(response, ResponseData::Disconnected));
        assert!(service.should_shutdown());
        assert!(service.schedules.is_empty());
        assert_eq!(service.backend_error.as_deref(), Some("flush failed"));
        assert!(service.trace.is_none());
    }

    #[test]
    fn periodic_send_failure_reschedules_and_clears_degraded_after_recovery() {
        let mut service = sample_service(false);
        let data = service
            .handle_message_send(MessageSendRequest {
                target: "0x123".to_string(),
                data: MessagePayload::RawHex("DEADBEEF".to_string()),
                periodicity_ms: Some(50),
            })
            .expect("schedule");
        assert!(matches!(data, ResponseData::MessageSent(_)));

        let schedule = service.schedules.get_mut("0x123").expect("schedule state");
        schedule.next_due = Instant::now() - Duration::from_millis(1);
        service
            .backend
            .send_results
            .push_back(Err("tx failed".to_string()));

        service.tick_schedules();
        assert_eq!(service.backend.sent.len(), 0);
        assert_eq!(service.backend_error.as_deref(), Some("tx failed"));
        let failed_next_due = service.schedules["0x123"].next_due;
        assert!(failed_next_due > Instant::now() - Duration::from_millis(1));

        service.tick_schedules();
        assert_eq!(service.backend.sent.len(), 0);
        assert_eq!(service.backend_error.as_deref(), Some("tx failed"));

        service
            .schedules
            .get_mut("0x123")
            .expect("schedule")
            .next_due = Instant::now() - Duration::from_millis(1);
        service.backend.send_results.push_back(Ok(()));
        service.tick_schedules();

        assert_eq!(service.backend.sent.len(), 1);
        assert_eq!(service.backend_error, None);
    }

    #[test]
    fn periodic_send_trace_errors_surface_without_dropping_event_state() {
        let mut service = sample_service(false);
        service.trace = Some(Box::new(FailingTrace::on_write("trace write failed")));
        service.schedules.insert(
            "0x123".to_string(),
            super::PeriodicScheduleState {
                target: "0x123".to_string(),
                frame: raw_frame(0x123, 1),
                periodicity_ms: 25,
                next_due: Instant::now() - Duration::from_millis(1),
            },
        );

        service.tick_schedules();

        assert_eq!(service.backend.sent.len(), 1);
        assert_eq!(service.backend_error.as_deref(), Some("trace write failed"));
        assert_eq!(service.events.len(), 1);
        assert!(
            service
                .latest
                .contains_key(&crate::can::dbc::FrameIdentity {
                    arb_id: 0x123,
                    extended: false,
                })
        );
    }

    #[test]
    fn tick_trims_expired_events_and_rebuilds_latest_index() {
        let mut service = sample_service(false);
        service
            .record_event(EventDirection::Rx, raw_frame(0x123, 1))
            .expect("record");
        service.events[0].recorded_at = Instant::now() - RETENTION_WINDOW - Duration::from_secs(1);
        service.tick();
        assert!(service.events.is_empty());
        assert!(service.latest.is_empty());
    }

    #[test]
    fn degraded_state_clears_after_successful_backend_poll() {
        let mut service = sample_service(false);
        service
            .backend
            .recv_results
            .push_back(Err("rx failed".to_string()));
        service.backend.recv_results.push_back(Ok(Vec::new()));
        service.tick();
        assert_eq!(service.status().connection_state, "degraded");
        service.tick();
        assert_eq!(service.status().connection_state, "connected");
    }

    #[test]
    fn trace_start_stop_and_degraded_backend_preserve_status() {
        let mut service = sample_service(false);
        let tempdir = tempfile::tempdir().expect("trace dir");
        let path = tempdir.path().join("trace.asc").display().to_string();
        let started = service
            .handle_trace_start(TraceStartRequest { path: path.clone() })
            .expect("trace start");
        assert!(matches!(started, ResponseData::TraceStarted { .. }));
        let stopped = service.handle_trace_stop().expect("trace stop");
        assert!(matches!(stopped, ResponseData::TraceStopped { .. }));
    }

    #[test]
    fn request_round_trip_through_service_serializes_current_verbs() {
        let mut service = sample_service(false);
        let response = service.handle_request(Request {
            id: Uuid::new_v4(),
            action: RequestAction::Status,
        });
        assert!(response.success);
    }

    #[tokio::test]
    async fn read_request_line_rejects_oversized_requests() {
        let (mut writer, reader) = duplex(1024);
        let writer_task = tokio::spawn(async move {
            let oversized = vec![b'a'; MAX_REQUEST_LINE_BYTES + 1];
            let _ = writer.write_all(&oversized).await;
        });

        let mut reader = BufReader::new(reader);
        let mut line = Vec::new();
        let err = read_request_line(&mut reader, &mut line)
            .await
            .expect_err("oversized line must fail");
        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
        drop(reader);
        writer_task.await.expect("writer task");
    }

    #[test]
    fn next_due_after_never_reuses_due_now_deadline() {
        let now = Instant::now();
        assert!(super::next_due_after(now, 0) >= now + Duration::from_millis(1));
        assert!(super::next_due_after(now, 25) >= now + Duration::from_millis(25));
    }
}