//! Per-channel state for the `nomisync-eval` ssh subsystem.

//!

//! Owns the frame decoder (newline + balanced-paren), the epoch cancel

//! handle, and a tokio mpsc to a per-channel worker that drives

//! `rpc::Session::handle_form` independently of the russh handler

//! callback. Decoupling the worker from `data()` is what lets the

//! handler scan subsequent bytes (an emacs `C-g` arrives as the ETX

//! byte `0x03`) and bump the engine epoch synchronously, tripping the

//! in-flight eval and surfacing `(:error (:code interrupted ...))`

//! on the wire.

//!

//! Constructed in `subsystem_request` once the requested subsystem

//! name matches `"nomisync-eval"` AND the session has authenticated

//! (i.e. `user_id.is_some()`). Dropped in `channel_eof` /

//! `channel_close` along with the rest of the per-channel state; the

//! drop closes the mpsc, the worker answers every form already queued

//! and then finishes the SSH channel, so a client that half-closes its

//! input still gets its responses and sees the session end.

use russh::ChannelId;

use russh::server::Handle as RusshHandle;

use tokio::sync::mpsc;

use rpc::{EpochBumper, FrameDecoder, FrameError, InterruptHandle, Session};

/// ASCII End-of-Text (`Ctrl-C` / `C-g`). Clients emit this byte on

/// the same channel as form data to request a cooperative cancel of

/// the in-flight `nomi-eval`. Stripped from the byte stream before

/// the decoder sees it so it never becomes part of a frame.

const ETX: u8 = 0x03;

/// Trait for the worker's response back-channel. The russh-backed

/// impl writes to an SSH channel via the server's `Handle`; the

/// mpsc-backed impl drains into a tokio receiver for unit tests.

/// Either way, the worker emits one full response (newline-

/// terminated) per send call.

#[async_trait::async_trait]

pub trait ResponseSink: Send + Sync + 'static {

    async fn send(&self, bytes: Vec<u8>);

    /// Ends the channel after the last queued response was sent.

    async fn finish(&self);

}

pub struct RusshSink {

    handle: RusshHandle,

    channel: ChannelId,

}

impl RusshSink {

}

#[async_trait::async_trait]

impl ResponseSink for RusshSink {

        if let Err(remaining) = self.handle.data(self.channel, bytes).await {

            log::warn!(

                "eval channel response write dropped {} bytes (channel closed?)",

                remaining.len()

            );

        }

        // OpenSSH exits 255 when a channel closes without an exit status.

        let finished = self

            .handle

            .exit_status_request(self.channel, 0)

            .await

            .is_ok()

            && self.handle.eof(self.channel).await.is_ok()

            && self.handle.close(self.channel).await.is_ok();

        if !finished {

            log::debug!("eval channel session already gone at finish");

        }

}

pub struct EvalChannelState {

    decoder: FrameDecoder,

    bumper: EpochBumper,

    /// Generation-counter cancel signal, shared with the worker's

    /// `Session`. Bumped together with the epoch on ETX so an

    /// interrupt arriving in the worker-pickup / compile window (before

    /// the eval enters Wasm, where an epoch bump would be absorbed by

    /// the not-yet-set deadline) is still observed by the session's

    /// pre-start `check_interrupt`.

    interrupt: InterruptHandle,

    forms_tx: mpsc::UnboundedSender<String>,

    /// Worker task drives `Session::handle_form` sequentially per

    /// channel; held so the worker stays alive for the channel's

    /// lifetime. Dropping `EvalChannelState` closes `forms_tx`; the

    /// worker drains what is queued, finishes the sink, and exits.

    _worker: tokio::task::JoinHandle<()>,

}

impl EvalChannelState {

    /// The caller decides whether the session is guarded; the daemon always

    /// passes a guarded one, tests an unguarded one for users with no row.

            }

    /// Feeds raw bytes received on the channel. Scans for the ETX

    /// interrupt token synchronously and, on finding one, signals the

    /// cancel both ways: it advances the interrupt generation (caught

    /// by `handle_form`'s pre-start `check_interrupt` if the eval has

    /// not entered Wasm yet) and bumps the engine epoch (traps an eval

    /// already running in `call_async`). Wiring both closes the

    /// worker-pickup / compile window, where a lone epoch bump would be

    /// absorbed by the not-yet-set deadline and the cancel lost. Either

    /// path surfaces a `(:code interrupted ...)` envelope. ETX bytes are

    /// stripped before the bytes reach the frame decoder.

    ///

    /// Errors only for non-UTF8 input (the ssh layer should never

    /// emit that for an s-expression subsystem).

        let cleaned: Vec<u8>;

        } else {

        };

        loop {

                        // Worker exited (channel drop or session

                        // error). Subsequent frames silently noop;

                        // the channel will be torn down by the SSH

                        // layer when the client notices the missing

                        // responses.

                }

                }

            }

        }

}

#[cfg(test)]

mod tests;