forgejo-mcp-broker/internal/bridge/bridge.go

// Package bridge connects an HTTP-side MCP client (e.g. Claude.ai over
// streamable HTTP) to a stdio-side MCP server (e.g. forgejo-mcp managed
// by supervisor) by piping JSON-RPC messages.
//
// The bridge is protocol-opaque: it parses each line only enough to extract
// the JSON-RPC `id` field for response routing. Everything else passes
// through verbatim. This is the design.md §5.3 "raw bytes through"
// approach — MCP is JSON-RPC 2.0 on both transports, so we can pipe
// without understanding semantics.
//
// One Bridge wraps one child. Multiple HTTP requests can be in flight
// concurrently; responses are routed back to the right HTTP handler by
// matching their `id` field against pending waiters.
package bridge

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"sync"
	"sync/atomic"
)

// Bridge owns the message routing for a single supervised child.
type Bridge struct {
	stdin  io.Writer
	stdout *bufio.Reader
	done   <-chan struct{}
	log    *slog.Logger

	writeMu sync.Mutex // serializes writes to stdin
	pending sync.Map   // string(id JSON) -> chan []byte
	started atomic.Bool
	closed  atomic.Bool
}

// New constructs a Bridge over the supplied pipes. The caller is
// responsible for keeping stdin/stdout alive for the bridge's lifetime
// (typically by holding the supervisor.Child).
func New(stdin io.Writer, stdout *bufio.Reader, done <-chan struct{}, log *slog.Logger) *Bridge {
	if log == nil {
		log = slog.New(slog.DiscardHandler)
	}
	return &Bridge{stdin: stdin, stdout: stdout, done: done, log: log}
}

// Start kicks off the reader goroutine. Must be called exactly once
// before HandleSSE is used; calling twice is a programmer error and is
// silently ignored.
func (b *Bridge) Start() {
	if !b.started.CompareAndSwap(false, true) {
		return
	}
	go b.readLoop()
}

// HandleSSE forwards the request body to the child as one newline-framed
// JSON-RPC line, then streams the matching response back as a single SSE
// event. Returns when the response is delivered, the client disconnects,
// or the child exits.
//
// Errors before SSE headers go out: written as plain HTTP errors with the
// appropriate status. Errors after headers: best-effort write of an SSE
// `error` event, then return.
func (b *Bridge) HandleSSE(w http.ResponseWriter, r *http.Request) {
	body, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, "read body: "+err.Error(), http.StatusBadRequest)
		return
	}
	if len(body) == 0 {
		http.Error(w, "empty body", http.StatusBadRequest)
		return
	}

	idKey, ok := extractIDKey(body)
	if !ok {
		// No id → notification. Forward without waiting; reply 204.
		// (MCP notifications don't expect a response.)
		if err := b.send(body); err != nil {
			http.Error(w, "send: "+err.Error(), http.StatusBadGateway)
			return
		}
		w.WriteHeader(http.StatusNoContent)
		return
	}

	// Register a waiter before sending so a fast child can't reply before
	// we're ready to receive.
	respCh := make(chan []byte, 1)
	if _, loaded := b.pending.LoadOrStore(idKey, respCh); loaded {
		http.Error(w, "duplicate in-flight id", http.StatusConflict)
		return
	}
	defer b.pending.Delete(idKey)

	if err := b.send(body); err != nil {
		http.Error(w, "send: "+err.Error(), http.StatusBadGateway)
		return
	}

	// SSE headers must go out before the first event.
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	w.WriteHeader(http.StatusOK)
	flusher, _ := w.(http.Flusher)
	if flusher != nil {
		flusher.Flush()
	}

	select {
	case resp := <-respCh:
		writeSSEEvent(w, "message", resp)
		if flusher != nil {
			flusher.Flush()
		}
	case <-r.Context().Done():
		// Client disconnected. Just return — child keeps running, future
		// requests on the same bridge are unaffected.
		b.log.Debug("client disconnected mid-stream", slog.String("id", idKey))
	case <-b.done:
		writeSSEEvent(w, "error", []byte(`{"reason":"child_exited"}`))
		if flusher != nil {
			flusher.Flush()
		}
	}
}

// send writes one newline-framed message to the child. Concurrent writes
// to the underlying pipe are not safe (interleaved bytes), so we serialize
// here.
func (b *Bridge) send(msg []byte) error {
	b.writeMu.Lock()
	defer b.writeMu.Unlock()
	if _, err := b.stdin.Write(msg); err != nil {
		return err
	}
	if len(msg) == 0 || msg[len(msg)-1] != '\n' {
		if _, err := b.stdin.Write([]byte{'\n'}); err != nil {
			return err
		}
	}
	return nil
}

// readLoop reads one JSON-RPC message per line from the child's stdout
// and routes each to the waiting handler (if any). Lines without an id
// (server-initiated notifications) are dropped on the floor for now —
// phase 5 introduces the GET-stream channel for those.
func (b *Bridge) readLoop() {
	for {
		line, err := b.stdout.ReadString('\n')
		if line != "" {
			b.routeResponse(line)
		}
		if err != nil {
			if errors.Is(err, io.EOF) {
				b.log.Debug("child stdout closed; bridge reader exiting")
			} else {
				b.log.Warn("bridge reader error", slog.String("err", err.Error()))
			}
			b.closed.Store(true)
			// Wake any pending waiters so handlers don't hang forever after
			// the child dies.
			b.pending.Range(func(k, v any) bool {
				if ch, ok := v.(chan []byte); ok {
					select {
					case <-ch:
					default:
						close(ch)
					}
				}
				return true
			})
			return
		}
	}
}

func (b *Bridge) routeResponse(line string) {
	idKey, ok := extractIDKey([]byte(line))
	if !ok {
		// Server-initiated notification or malformed message — log and skip.
		b.log.Debug("bridge: dropping un-routable line", slog.Int("len", len(line)))
		return
	}
	v, ok := b.pending.LoadAndDelete(idKey)
	if !ok {
		// No waiter — request was likely abandoned (client disconnected).
		// Logging at debug; this is normal.
		b.log.Debug("bridge: response with no waiter", slog.String("id", idKey))
		return
	}
	ch, _ := v.(chan []byte)
	if ch == nil {
		return
	}
	// Non-blocking send: the caller created the channel with buffer 1, so
	// this is guaranteed to succeed unless we're racing a Close.
	select {
	case ch <- []byte(line):
	default:
	}
}

// extractIDKey parses the JSON-RPC `id` field and returns its raw JSON
// representation as a routing key. Returns ok=false when the message has
// no id (notification) or fails to parse.
//
// JSON-RPC ids are string, number, or null. Comparing the raw JSON token
// avoids quirks like distinguishing 1 from "1".
func extractIDKey(data []byte) (string, bool) {
	var probe struct {
		ID json.RawMessage `json:"id"`
	}
	if err := json.Unmarshal(data, &probe); err != nil {
		return "", false
	}
	if len(probe.ID) == 0 || string(probe.ID) == "null" {
		return "", false
	}
	return string(probe.ID), true
}

// writeSSEEvent writes one Server-Sent Event to w. The data may be
// multi-line — we re-frame it per the SSE spec (each \n becomes a
// separate `data:` line). For our typical single-line JSON, this is a
// no-op.
func writeSSEEvent(w io.Writer, event string, data []byte) {
	fmt.Fprintf(w, "event: %s\n", event)
	for _, line := range splitLines(data) {
		fmt.Fprintf(w, "data: %s\n", line)
	}
	_, _ = w.Write([]byte("\n"))
}

func splitLines(data []byte) []string {
	// Strip a single trailing newline (typical for child output).
	if n := len(data); n > 0 && data[n-1] == '\n' {
		data = data[:n-1]
	}
	if len(data) == 0 {
		return []string{""}
	}
	out := []string{}
	start := 0
	for i, b := range data {
		if b == '\n' {
			out = append(out, string(data[start:i]))
			start = i + 1
		}
	}
	out = append(out, string(data[start:]))
	return out
}