// Package service orchestrates the odidere voice assistant server. // It coordinates the HTTP server, LLM client, and handles // graceful shutdown. package service import ( "context" "embed" "encoding/json" "fmt" "html/template" "io/fs" "log/slog" "net/http" "os" "os/signal" "runtime/debug" "strings" "syscall" "time" "code.chimeric.al/chimerical/odidere/internal/config" "code.chimeric.al/chimerical/odidere/internal/job" "code.chimeric.al/chimerical/odidere/internal/llm" "code.chimeric.al/chimerical/odidere/internal/service/templates" "code.chimeric.al/chimerical/odidere/internal/tool" "github.com/google/uuid" "github.com/robfig/cron/v3" openai "github.com/sashabaranov/go-openai" ) // Context keys for request-scoped values. type key string const ( logKey key = "log" idKey key = "id" ipKey key = "ip" ) //go:embed all:static/* var static embed.FS // Service is the main application coordinator. // It owns the HTTP server and all processing clients. type Service struct { cfg *config.Config cron *cron.Cron jobs []*job.Job llm *llm.Client log *slog.Logger mux *http.ServeMux server *http.Server tmpl *template.Template tools *tool.Registry } // New creates a Service from the provided configuration. // It initializes all clients and the HTTP server. func New(cfg *config.Config, log *slog.Logger) (*Service, error) { var svc = &Service{ cfg: cfg, log: log, mux: http.NewServeMux(), } // Setup tool registry. registry, err := tool.NewRegistry(cfg.Tools) if err != nil { return nil, fmt.Errorf("load tools: %v", err) } svc.tools = registry // Create LLM client. llmClient, err := llm.NewClient(cfg.LLM, registry, log) if err != nil { return nil, fmt.Errorf("create LLM client: %v", err) } svc.llm = llmClient // Convert job configs to jobs. jobs := make([]*job.Job, 0, len(cfg.Jobs)) for _, jc := range cfg.Jobs { j, err := job.New(jc, log, svc.llm) if err != nil { return nil, fmt.Errorf("create job %q: %w", jc.Name, err) } jobs = append(jobs, j) } svc.jobs = jobs // Parse templates. tmpl, err := templates.Parse() if err != nil { return nil, fmt.Errorf("parse templates: %v", err) } svc.tmpl = tmpl // Setup static file server. staticFS, err := fs.Sub(static, "static") if err != nil { return nil, fmt.Errorf("setup static fs: %v", err) } // Register routes. svc.mux.HandleFunc("GET /", svc.home) svc.mux.HandleFunc("GET /status", svc.status) svc.mux.Handle( "GET /static/", http.StripPrefix( "/static/", http.FileServer(http.FS(staticFS)), ), ) svc.mux.HandleFunc("POST /v1/chat/voice", svc.chat) svc.mux.HandleFunc("POST /v1/chat/voice/stream", svc.chatStream) svc.mux.HandleFunc("GET /v1/models", svc.models) svc.server = &http.Server{ Addr: cfg.Address, Handler: svc, } return svc, nil } // ServeHTTP implements http.Handler. It logs requests, assigns a UUID, // sets context values, handles panics, and delegates to the mux. func (svc *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) { var ( start = time.Now() id = uuid.NewString() ip = func() string { if ip := r.Header.Get("X-Forwarded-For"); ip != "" { if idx := strings.Index(ip, ","); idx != -1 { return strings.TrimSpace(ip[:idx]) } return ip } return r.RemoteAddr }() log = svc.log.With(slog.Group( "request", slog.String("id", id), slog.String("ip", ip), slog.String("method", r.Method), slog.String("path", r.URL.Path), )) ) // Log completion time. defer func() { log.InfoContext( r.Context(), "completed", slog.Duration("duration", time.Since(start)), ) }() // Panic recovery. defer func() { if err := recover(); err != nil { log.ErrorContext( r.Context(), "panic recovered", slog.Any("error", err), slog.String("stack", string(debug.Stack())), ) http.Error( w, http.StatusText( http.StatusInternalServerError, ), http.StatusInternalServerError, ) } }() // Enrich context with request-scoped values. ctx := r.Context() ctx = context.WithValue(ctx, logKey, log) ctx = context.WithValue(ctx, idKey, id) ctx = context.WithValue(ctx, ipKey, ip) r = r.WithContext(ctx) log.InfoContext(ctx, "handling") // Pass the request on to the multiplexer. svc.mux.ServeHTTP(w, r) } // Run starts the service and blocks until shutdown. // Shutdown is triggered by SIGINT or SIGTERM. func (svc *Service) Run(ctx context.Context) error { svc.log.Info( "starting odidere", slog.Group( "llm", slog.String("url", svc.cfg.LLM.URL), slog.String("model", svc.cfg.LLM.Model), ), slog.Group( "server", slog.String("address", svc.cfg.Address), ), slog.Group( "tools", slog.Int("count", len(svc.tools.List())), slog.Any( "names", strings.Join(svc.tools.List(), ","), ), ), slog.Group( "tts_url", slog.String("url", svc.cfg.TTS.URL), slog.String("default_voice", svc.cfg.TTS.Voice), ), slog.Group( "stt_url", slog.String("url", svc.cfg.STT.URL), ), slog.Group( "jobs", slog.Int("count", svc.JobCount()), ), slog.Any("shutdown_timeout", svc.cfg.GetShutdownTimeout()), ) // Setup signal handling for graceful shutdown. ctx, cancel := signal.NotifyContext( ctx, os.Interrupt, syscall.SIGTERM, ) defer cancel() // Register jobs with cron scheduler. svc.cron = cron.New() for _, j := range svc.jobs { if !j.IsEnabled() { svc.log.Info("job disabled", slog.String("name", j.Name())) continue } entry, err := svc.cron.AddFunc(j.Schedule(), func() { result := j.Run(ctx) if result.Error != nil { svc.log.Error("job failed", slog.String("name", j.Name()), slog.Any("error", result.Error), slog.Duration( "duration", result.Duration, ), ) } else { svc.log.Info("job completed", slog.String("name", j.Name()), slog.Duration( "duration", result.Duration, ), ) } }) if err != nil { svc.log.Error("failed to schedule job", slog.String("name", j.Name()), slog.Any("error", err), ) continue } svc.log.Info("job scheduled", slog.String("name", j.Name()), slog.String("schedule", j.Schedule()), slog.Int("entry_id", int(entry)), ) } svc.cron.Start() // Start HTTP server in background. var errs = make(chan error, 1) go func() { svc.log.Info( "HTTP server listening", slog.String("address", svc.cfg.Address), ) if err := svc.server.ListenAndServe(); err != nil && err != http.ErrServerClosed { errs <- err } close(errs) }() // Wait for shutdown signal or server error. select { case <-ctx.Done(): svc.log.Info("shutdown signal received") case err := <-errs: if err != nil { return fmt.Errorf("server error: %w", err) } } // Graceful shutdown with timeout. shutdownCtx, shutdownCancel := context.WithTimeout( context.Background(), svc.cfg.GetShutdownTimeout(), ) defer shutdownCancel() // Stop cron scheduler — in-flight jobs continue to completion. svc.log.Info("stopping job scheduler") svc.cron.Stop() svc.log.Info("shutting down HTTP server") if err := svc.server.Shutdown(shutdownCtx); err != nil { svc.log.Warn( "shutdown timeout reached", slog.Any("error", err), ) } svc.log.Info("terminating") return nil } // ListJobs returns all registered jobs. func (svc *Service) ListJobs() []*job.Job { return svc.jobs } // JobCount returns the number of registered jobs. func (svc *Service) JobCount() int { return len(svc.jobs) } func (svc *Service) home(w http.ResponseWriter, r *http.Request) { var ( ctx = r.Context() log = ctx.Value(logKey).(*slog.Logger) ) if r.URL.Path != "/" { http.NotFound(w, r) return } w.Header().Set("Content-Type", "text/html; charset=utf-8") if err := svc.tmpl.ExecuteTemplate( w, "index.gohtml", struct { TTSURL string TTSDefaultVoice string STTURL string }{ TTSURL: svc.cfg.TTS.URL, TTSDefaultVoice: svc.cfg.TTS.Voice, STTURL: svc.cfg.STT.URL, }, ); err != nil { log.ErrorContext( ctx, "template error", slog.Any("error", err), ) http.Error( w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError, ) } } // status returns server status. func (svc *Service) status(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } // Request is the incoming request format for the chat endpoints. type Request struct { // Messages is the conversation history. Messages []openai.ChatCompletionMessage `json:"messages"` // Model is the LLM model ID. If empty, the default model is used. Model string `json:"model,omitempty"` // SystemMessage overrides the configured system message for this request. SystemMessage string `json:"system_message,omitempty"` // Voice is the voice ID for TTS. Voice string `json:"voice,omitempty"` } // Response is the response format for chat and voice endpoints. type Response struct { // Messages is the full list of messages generated during the query, // including tool calls and tool results. Messages []openai.ChatCompletionMessage `json:"messages,omitempty"` // Model is the LLM model used for the response. Model string `json:"used_model,omitempty"` // Voice is the voice used for TTS synthesis. Voice string `json:"used_voice,omitempty"` } // chat processes text chat requests. func (svc *Service) chat(w http.ResponseWriter, r *http.Request) { var ( ctx = r.Context() log = ctx.Value(logKey).(*slog.Logger) ) // Parse request. r.Body = http.MaxBytesReader(w, r.Body, 32<<20) var req Request if err := json.NewDecoder(r.Body).Decode(&req); err != nil { log.ErrorContext( ctx, "failed to decode request", slog.Any("error", err), ) http.Error(w, "invalid request", http.StatusBadRequest) return } // Validate messages. if len(req.Messages) == 0 { http.Error(w, "messages required", http.StatusBadRequest) return } log.InfoContext(ctx, "messages", slog.Any("data", req.Messages), ) // Get LLM response. var model = req.Model if model == "" { model = svc.llm.DefaultModel() } msgs, err := svc.llm.Query(ctx, req.Messages, model, req.SystemMessage, 0) if err != nil { log.ErrorContext( ctx, "LLM request failed", slog.Any("error", err), ) http.Error(w, "LLM error", http.StatusInternalServerError) return } if len(msgs) == 0 { http.Error( w, "no response from LLM", http.StatusInternalServerError, ) return } final := msgs[len(msgs)-1] log.InfoContext( ctx, "LLM response", slog.String("text", final.Content), slog.String("model", model), ) w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(Response{ Messages: msgs, Model: model, Voice: req.Voice, }); err != nil { log.ErrorContext( ctx, "failed to json encode response", slog.Any("error", err), ) } } // StreamMessage is the SSE event payload for the streaming chat endpoint. type StreamMessage struct { // Error is an error message, if any. Error string `json:"error,omitempty"` // Message is the chat completion message. Message openai.ChatCompletionMessage `json:"message"` // Model is the LLM model used for the response. Model string `json:"model,omitempty"` // Voice is the voice used for TTS synthesis. Voice string `json:"voice,omitempty"` } // chatStream processes chat requests with streaming SSE output. func (svc *Service) chatStream(w http.ResponseWriter, r *http.Request) { var ( ctx = r.Context() log = ctx.Value(logKey).(*slog.Logger) ) // Check that the response writer supports flushing. flusher, ok := w.(http.Flusher) if !ok { http.Error( w, "streaming not supported", http.StatusInternalServerError, ) return } // Parse request. r.Body = http.MaxBytesReader(w, r.Body, 32<<20) var req Request if err := json.NewDecoder(r.Body).Decode(&req); err != nil { log.ErrorContext( ctx, "failed to decode request", slog.Any("error", err), ) http.Error(w, "invalid request", http.StatusBadRequest) return } // Validate messages. if len(req.Messages) == 0 { http.Error(w, "messages required", http.StatusBadRequest) return } // Set SSE headers. w.Header().Set("Cache-Control", "no-cache") w.Header().Set("Connection", "keep-alive") w.Header().Set("Content-Type", "text/event-stream") // Helper to send an SSE event. send := func(msg StreamMessage) { data, err := json.Marshal(msg) if err != nil { log.ErrorContext(ctx, "failed to marshal SSE event", slog.Any("error", err), ) return } fmt.Fprintf(w, "event: message\ndata: %s\n\n", data) flusher.Flush() } // Get model. var model = req.Model if model == "" { model = svc.llm.DefaultModel() } // Start streaming LLM query. var ( events = make(chan llm.StreamEvent) llmErr error ) go func() { llmErr = svc.llm.QueryStream(ctx, req.Messages, model, req.SystemMessage, 0, events) }() // Consume events and send as SSE. var last StreamMessage for evt := range events { msg := StreamMessage{Message: evt.Message} // Track the last assistant message for TTS. if evt.Message.Role == openai.ChatMessageRoleAssistant && len(evt.Message.ToolCalls) == 0 { last = msg continue } send(msg) } // Check for LLM errors. if llmErr != nil { log.ErrorContext( ctx, "LLM stream failed", slog.Any("error", llmErr), ) send(StreamMessage{ Message: openai.ChatCompletionMessage{ Role: openai.ChatMessageRoleAssistant, }, Error: fmt.Sprintf("LLM error: %v", llmErr), }) return } last.Model = model last.Voice = req.Voice send(last) } // models returns available LLM models. func (svc *Service) models(w http.ResponseWriter, r *http.Request) { var ( ctx = r.Context() log = ctx.Value(logKey).(*slog.Logger) ) models, err := svc.llm.ListModels(ctx) if err != nil { log.ErrorContext( ctx, "failed to list models", slog.Any("error", err), ) http.Error( w, "failed to list models", http.StatusInternalServerError, ) return } w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(struct { Models []openai.Model `json:"models"` DefaultModel string `json:"default_model"` }{ Models: models, DefaultModel: svc.llm.DefaultModel(), }); err != nil { log.ErrorContext( ctx, "failed to encode models response", slog.Any("error", err), ) } }