clanker-saddle/src/Api/TerminalSse.vala

using Invercargill.Expressions;
using InvercargillSql.Orm;
using Inversion;
using Astralis;
using ClankerSaddle.Data;

namespace ClankerSaddle.Api {

    /**
     * `GET /terminal-stream/{session_id}` — a Server-Sent Events stream of
     * one session's terminal log. Every existing `session_commands` row is
     * pushed as an initial event; the connection then stays open, polling
     * the database each second and streaming entries newer than the id
     * cursor until the client disconnects.
     *
     * Each event carries `id: <entry id>` (so EventSource reconnects
     * resume from the last seen entry) and a `data` payload of
     * `{"id", "command", "output", "type", "input"?}` — the entry types
     * the terminal page renders with per-type styling (command, comment,
     * block, user-comment, progress, review, plan, shell-reset, failed,
     * query).
     *
     * Sessions with raw PTY rows (`session_raw_log`, the PTY-terminal
     * feature) stream those instead: `{"seq", "type": "raw", "text"}`
     * events in seq order, polled the same way. The client renders them
     * as one unstyled `<pre>` of concatenated chunks.
     */
    public class TerminalSseEndpoint : SseEndpoint {

        private OrmSession db = inject<OrmSession>();

        /** Reconnect hint for browsers (EventSource retries sooner). */
        public override uint retry_interval { get { return 3000; } }

        public override async void new_connection(HttpContext http_context,
                RouteContext route_context, SseStream stream) {
            string id_text = "";
            route_context.mapped_parameters.try_get("session_id", out id_text);
            int64 session_id = int64.parse(id_text);

            bool has_raw = false;
            try {
                var raw_probe = yield db.query<SessionRawLog>()
                    .where(expr("session_id == $0", elem<int64?>(session_id)))
                    .first_async();
                has_raw = raw_probe != null;
            } catch (Error e) {
                warning(@"[ClankerSaddle] Terminal stream probe for session $(session_id) failed: $(e.message)");
                return;
            }
            if (has_raw) {
                yield stream_raw_log(session_id, stream);
                return;
            }

            int64 cursor = 0;
            try {
                while (!stream.is_closed) {
                    bool pushed = false;
                    foreach (var entry in yield db.query<SessionCommand>()
                            .where(expr("session_id == $0 && id > $1",
                                elem<int64?>(session_id), elem<int64?>(cursor)))
                            .order_by(expr("id"))
                            .materialise_async()) {
                        yield stream.send_event(new SseEvent.with_id(
                            entry.id.to_string(), payload(entry)));
                        cursor = entry.id;
                        pushed = true;
                    }
                    if (!pushed) {
                        // Heartbeat: detects disconnected clients even when
                        // the session is quiet.
                        yield stream.send_event(new SseEvent("ping"));
                    }
                    yield nap_seconds(1);
                }
            } catch (Error e) {
                // A closed stream ends the loop; anything else is logged.
                if (!stream.is_closed) {
                    warning(@"[ClankerSaddle] Terminal stream for session $(session_id) ended: $(e.message)");
                }
            }
        }

        /** The raw-log variant of the stream loop above. */
        private async void stream_raw_log(int64 session_id, SseStream stream) {
            int64 cursor = 0;
            try {
                while (!stream.is_closed) {
                    bool pushed = false;
                    foreach (var chunk in yield db.query<SessionRawLog>()
                            .where(expr("session_id == $0 && seq > $1",
                                elem<int64?>(session_id), elem<int64?>(cursor)))
                            .order_by(expr("seq"))
                            .materialise_async()) {
                        yield stream.send_event(new SseEvent.with_id(
                            chunk.seq.to_string(), raw_payload(chunk)));
                        cursor = chunk.seq;
                        pushed = true;
                    }
                    if (!pushed) {
                        yield stream.send_event(new SseEvent("ping"));
                    }
                    yield nap_seconds(1);
                }
            } catch (Error e) {
                if (!stream.is_closed) {
                    warning(@"[ClankerSaddle] Raw terminal stream for session $(session_id) ended: $(e.message)");
                }
            }
        }

        /** The entry as a one-line JSON object payload. */
        private static string payload(SessionCommand entry) {
            var builder = new Json.Builder();
            builder.begin_object();
            builder.set_member_name("id");
            builder.add_int_value(entry.id);
            builder.set_member_name("command");
            builder.add_string_value(entry.command_text);
            builder.set_member_name("output");
            builder.add_string_value(entry.output_text ?? "");
            builder.set_member_name("type");
            builder.add_string_value(entry.entry_type);
            if (entry.piped_input != null && ((!)entry.piped_input).length > 0) {
                builder.set_member_name("input");
                builder.add_string_value(((!)entry.piped_input));
            }
            builder.end_object();
            var generator = new Json.Generator();
            generator.set_root(builder.get_root());
            return generator.to_data(null);
        }

        /** A raw chunk as a one-line JSON object payload. */
        private static string raw_payload(SessionRawLog chunk) {
            var builder = new Json.Builder();
            builder.begin_object();
            builder.set_member_name("seq");
            builder.add_int_value(chunk.seq);
            builder.set_member_name("type");
            builder.add_string_value("raw");
            builder.set_member_name("text");
            builder.add_string_value(chunk.chunk);
            builder.end_object();
            var generator = new Json.Generator();
            generator.set_root(builder.get_root());
            return generator.to_data(null);
        }

        private static async void nap_seconds(uint seconds) {
            GLib.Timeout.add_seconds(seconds, () => {
                nap_seconds.callback();
                return false;
            });
            yield;
        }
    }
}