server.go

  1package server
  2
  3import (
  4	"context"
  5	"fmt"
  6	"log/slog"
  7	"net"
  8	"net/http"
  9	"os/user"
 10	"runtime"
 11	"strings"
 12
 13	"github.com/charmbracelet/crush/internal/app"
 14	"github.com/charmbracelet/crush/internal/config"
 15	"github.com/charmbracelet/crush/internal/csync"
 16)
 17
 18// ErrServerClosed is returned when the server is closed.
 19var ErrServerClosed = http.ErrServerClosed
 20
 21// InstanceState represents the state of a running [app.App] instance.
 22type InstanceState uint8
 23
 24const (
 25	// InstanceStateCreated indicates that the instance has been created but not yet started.
 26	InstanceStateCreated InstanceState = iota
 27	// InstanceStateStarted indicates that the instance is currently running.
 28	InstanceStateStarted
 29	// InstanceStateStopped indicates that the instance has been stopped.
 30	InstanceStateStopped
 31)
 32
 33// Instance represents a running [app.App] instance with its associated
 34// resources and state.
 35type Instance struct {
 36	*app.App
 37	State InstanceState
 38	ln    net.Listener
 39	cfg   *config.Config
 40	id    string
 41	path  string
 42}
 43
 44// ID returns the unique identifier of the instance.
 45func (i *Instance) ID() string {
 46	return i.id
 47}
 48
 49// Path returns the filesystem path associated with the instance.
 50func (i *Instance) Path() string {
 51	return i.path
 52}
 53
 54// DefaultHost returns the default server host.
 55func DefaultHost() string {
 56	sock := "crush.sock"
 57	usr, err := user.Current()
 58	if err == nil && usr.Uid != "" {
 59		sock = fmt.Sprintf("crush-%s.sock", usr.Uid)
 60	}
 61	if runtime.GOOS == "windows" {
 62		return fmt.Sprintf("npipe:////./pipe/%s", sock)
 63	}
 64	return fmt.Sprintf("unix:///tmp/%s", sock)
 65}
 66
 67// Server represents a Crush server instance bound to a specific address.
 68type Server struct {
 69	// Addr can be a TCP address, a Unix socket path, or a Windows named pipe.
 70	Addr string
 71
 72	h   *http.Server
 73	ln  net.Listener
 74	ctx context.Context
 75
 76	// instances is a map of running applications managed by the server.
 77	instances *csync.Map[string, *Instance]
 78	cfg       *config.Config
 79	logger    *slog.Logger
 80}
 81
 82// SetLogger sets the logger for the server.
 83func (s *Server) SetLogger(logger *slog.Logger) {
 84	s.logger = logger
 85}
 86
 87// DefaultServer returns a new [Server] instance with the default address.
 88func DefaultServer(cfg *config.Config) *Server {
 89	proto, addr, ok := strings.Cut(DefaultHost(), "://")
 90	if !ok {
 91		panic("invalid default host")
 92	}
 93	return NewServer(cfg, proto, addr)
 94}
 95
 96// NewServer is a helper to create a new [Server] instance with the given
 97// address. On Windows, if the address is not a "tcp" address, it will be
 98// converted to a named pipe format.
 99func NewServer(cfg *config.Config, network, address string) *Server {
100	s := new(Server)
101	s.Addr = address
102	s.cfg = cfg
103	s.instances = csync.NewMap[string, *Instance]()
104	s.ctx = context.Background()
105
106	var p http.Protocols
107	p.SetHTTP1(true)
108	p.SetUnencryptedHTTP2(true)
109	c := &controllerV1{Server: s}
110	mux := http.NewServeMux()
111	mux.HandleFunc("GET /v1/health", c.handleGetHealth)
112	mux.HandleFunc("GET /v1/version", c.handleGetVersion)
113	mux.HandleFunc("GET /v1/config", c.handleGetConfig)
114	mux.HandleFunc("POST /v1/control", c.handlePostControl)
115	mux.HandleFunc("GET /v1/instances", c.handleGetInstances)
116	mux.HandleFunc("POST /v1/instances", c.handlePostInstances)
117	mux.HandleFunc("DELETE /v1/instances", c.handleDeleteInstances)
118	mux.HandleFunc("GET /v1/instances/{id}/config", c.handleGetInstanceConfig)
119	mux.HandleFunc("GET /v1/instances/{id}/events", c.handleGetInstanceEvents)
120	mux.HandleFunc("GET /v1/instances/{id}/sessions", c.handleGetInstanceSessions)
121	mux.HandleFunc("POST /v1/instances/{id}/sessions", c.handlePostInstanceSessions)
122	mux.HandleFunc("GET /v1/instances/{id}/sessions/{sid}", c.handleGetInstanceSession)
123	mux.HandleFunc("GET /v1/instances/{id}/sessions/{sid}/history", c.handleGetInstanceSessionHistory)
124	mux.HandleFunc("GET /v1/instances/{id}/sessions/{sid}/messages", c.handleGetInstanceSessionMessages)
125	mux.HandleFunc("GET /v1/instances/{id}/lsps", c.handleGetInstanceLSPs)
126	mux.HandleFunc("GET /v1/instances/{id}/lsps/{lsp}/diagnostics", c.handleGetInstanceLSPDiagnostics)
127	mux.HandleFunc("GET /v1/instances/{id}/permissions/skip", c.handleGetInstancePermissionsSkip)
128	mux.HandleFunc("POST /v1/instances/{id}/permissions/skip", c.handlePostInstancePermissionsSkip)
129	mux.HandleFunc("POST /v1/instances/{id}/permissions/grant", c.handlePostInstancePermissionsGrant)
130	mux.HandleFunc("GET /v1/instances/{id}/agent", c.handleGetInstanceAgent)
131	mux.HandleFunc("POST /v1/instances/{id}/agent", c.handlePostInstanceAgent)
132	mux.HandleFunc("POST /v1/instances/{id}/agent/init", c.handlePostInstanceAgentInit)
133	mux.HandleFunc("POST /v1/instances/{id}/agent/update", c.handlePostInstanceAgentUpdate)
134	mux.HandleFunc("GET /v1/instances/{id}/agent/sessions/{sid}", c.handleGetInstanceAgentSession)
135	mux.HandleFunc("POST /v1/instances/{id}/agent/sessions/{sid}/cancel", c.handlePostInstanceAgentSessionCancel)
136	mux.HandleFunc("GET /v1/instances/{id}/agent/sessions/{sid}/prompts/queued", c.handleGetInstanceAgentSessionPromptQueued)
137	mux.HandleFunc("POST /v1/instances/{id}/agent/sessions/{sid}/prompts/clear", c.handlePostInstanceAgentSessionPromptClear)
138	mux.HandleFunc("POST /v1/instances/{id}/agent/sessions/{sid}/summarize", c.handleGetInstanceAgentSessionSummarize)
139	s.h = &http.Server{
140		Protocols: &p,
141		Handler:   s.loggingHandler(mux),
142	}
143	return s
144}
145
146// Serve accepts incoming connections on the listener.
147func (s *Server) Serve(ln net.Listener) error {
148	return s.h.Serve(ln)
149}
150
151// ListenAndServe starts the server and begins accepting connections.
152func (s *Server) ListenAndServe() error {
153	if s.ln != nil {
154		return fmt.Errorf("server already started")
155	}
156	proto := "unix"
157	if runtime.GOOS == "windows" {
158		proto = "npipe"
159	}
160	ln, err := listen(proto, s.Addr)
161	if err != nil {
162		return fmt.Errorf("failed to listen on %s: %w", s.Addr, err)
163	}
164	return s.Serve(ln)
165}
166
167func (s *Server) closeListener() {
168	if s.ln != nil {
169		s.ln.Close()
170		s.ln = nil
171	}
172}
173
174// Close force close all listeners and connections.
175func (s *Server) Close() error {
176	defer func() { s.closeListener() }()
177	return s.h.Close()
178}
179
180// Shutdown gracefully shuts down the server without interrupting active
181// connections. It stops accepting new connections and waits for existing
182// connections to finish.
183func (s *Server) Shutdown(ctx context.Context) error {
184	defer func() { s.closeListener() }()
185	return s.h.Shutdown(ctx)
186}
187
188func (s *Server) logDebug(r *http.Request, msg string, args ...any) {
189	if s.logger != nil {
190		s.logger.With(
191			slog.String("method", r.Method),
192			slog.String("url", r.URL.String()),
193			slog.String("remote_addr", r.RemoteAddr),
194		).Debug(msg, args...)
195	}
196}
197
198func (s *Server) logError(r *http.Request, msg string, args ...any) {
199	if s.logger != nil {
200		s.logger.With(
201			slog.String("method", r.Method),
202			slog.String("url", r.URL.String()),
203			slog.String("remote_addr", r.RemoteAddr),
204		).Error(msg, args...)
205	}
206}