odidere/internal/service/service.go

// Package service orchestrates the odidere voice assistant server.
// It coordinates the HTTP server, LLM clients, and handles
// graceful shutdown.
package service

import (
	"context"
	"embed"
	"encoding/json/v2"
	"fmt"
	"html/template"
	"io/fs"
	"log/slog"
	"net/http"
	"os"
	"os/signal"
	"runtime/debug"
	"strings"
	"sync"
	"syscall"
	"time"
	"uuid"

	"code.chimeric.al/chimerical/odidere/internal/config"
	"code.chimeric.al/chimerical/odidere/internal/conversation"
	"code.chimeric.al/chimerical/odidere/internal/db"
	"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/robfig/cron/v3"
)

// Context keys for request-scoped values.
type key string

const (
	logKey key = "log"
	idKey  key = "id"
	ipKey  key = "ip"

	modelsTimeout = 10 * time.Second
)

//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
	db            *db.SQLite
	dbr           *db.Queries
	dbw           *db.Queries
	jobs          []*job.Job
	llms          map[string]*llm.Client
	log           *slog.Logger
	allowedModels map[string]map[string]struct{}
	mux           *http.ServeMux
	server        *http.Server
	tmpl          *template.Template
	tools         *tool.Registry

	ctx           context.Context
	cancel        context.CancelFunc
	wg            sync.WaitGroup
	mu            sync.Mutex
	conversations map[string]*conversation.Conversation
}

// New creates a Service from the provided configuration.
// It initializes all clients and the HTTP server.
func New(
	cfg *config.Config,
	log *slog.Logger,
	sdb *db.SQLite,
) (*Service, error) {
	if cfg == nil {
		return nil, fmt.Errorf("config is nil")
	}
	if err := cfg.Validate(); err != nil {
		return nil, fmt.Errorf("invalid config: %w", err)
	}
	if log == nil {
		return nil, fmt.Errorf("logger is nil")
	}
	if sdb == nil || sdb.Writer == nil || sdb.Reader == nil {
		return nil, fmt.Errorf("database is nil")
	}

	ctx, cancel := context.WithCancel(context.Background())
	var svc = &Service{
		cfg:           cfg,
		db:            sdb,
		dbr:           db.New(sdb.Reader),
		dbw:           db.New(sdb.Writer),
		log:           log,
		mux:           http.NewServeMux(),
		ctx:           ctx,
		cancel:        cancel,
		conversations: make(map[string]*conversation.Conversation),
	}

	// 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 clients for each provider.
	svc.llms = make(map[string]*llm.Client, len(cfg.Providers))
	for _, pc := range cfg.Providers {
		client, err := llm.NewClient(
			llm.Config{
				Key:      pc.Key,
				Provider: pc.Name,
				Timeout:  pc.Timeout,
				URL:      pc.URL,
			},
			log,
		)
		if err != nil {
			return nil, fmt.Errorf(
				"create LLM client for provider %q: %w",
				pc.Name, err,
			)
		}
		svc.llms[pc.Name] = client
		log.Info(
			"LLM provider registered",
			slog.String("provider", pc.Name),
			slog.String("url", pc.URL),
		)
	}
	if svc.llms[cfg.DefaultModel.Provider] == nil {
		return nil, fmt.Errorf("unknown default provider")
	}

	// Build model allowed list map: provider name -> set of base model IDs.
	svc.allowedModels = make(
		map[string]map[string]struct{}, len(cfg.Providers),
	)
	for _, pc := range cfg.Providers {
		if len(pc.Models) == 0 {
			continue
		}
		allowed := make(map[string]struct{}, len(pc.Models))
		for _, m := range pc.Models {
			base, _, _ := strings.Cut(m, ":")
			allowed[base] = struct{}{}
		}
		svc.allowedModels[pc.Name] = allowed
	}

	// Convert job configs to jobs.
	jobs := make([]*job.Job, 0, len(cfg.Jobs))
	for _, jc := range cfg.Jobs {
		// Resolve the provider and model for this job.
		provider, modelID := jc.Model.Provider, jc.Model.Model
		if provider == "" {
			provider = cfg.DefaultModel.Provider
		}
		if modelID == "" {
			modelID = cfg.DefaultModel.Model
		}
		llmc := svc.llms[provider]
		if llmc == nil {
			return nil, fmt.Errorf(
				"create job %q: no LLM client for provider %q",
				jc.Name,
				provider,
			)
		}
		jc.Model = config.ModelConfig{
			Provider: provider,
			Model:    modelID,
		}
		if jc.SystemMessage == "" {
			jc.SystemMessage = cfg.SystemMessage
		}
		j, err := job.New(jc, llmc, log)
		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("GET /models", svc.models)

	svc.mux.HandleFunc("GET /conversations", svc.listConversations)
	svc.mux.HandleFunc("POST /conversations", svc.createConversation)
	svc.mux.HandleFunc("GET /conversations/{id}", svc.getConversationHandler)
	svc.mux.HandleFunc("DELETE /conversations/{id}", svc.deleteConversation)

	svc.mux.HandleFunc(
		"POST /conversations/{id}/inference", svc.inference,
	)
	svc.mux.HandleFunc("POST /conversations/{id}/stop", svc.stop)
	svc.mux.HandleFunc(
		"POST /conversations/{id}/subscribe", svc.subscribe,
	)

	svc.mux.HandleFunc(
		"POST /conversations/{id}/messages", svc.insertMessage,
	)
	svc.mux.HandleFunc(
		"PUT /conversations/{id}/messages/{mid}", svc.updateMessage,
	)
	svc.mux.HandleFunc(
		"DELETE /conversations/{id}/messages/{mid}", svc.deleteMessage,
	)

	svc.mux.HandleFunc(
		"POST /conversations/{id}/messages/{mid}/calls",
		svc.addToolCall,
	)
	svc.mux.HandleFunc(
		"PUT /conversations/{id}/messages/{mid}/calls/{cid}",
		svc.updateToolCall,
	)
	svc.mux.HandleFunc(
		"DELETE /conversations/{id}/messages/{mid}/calls/{cid}",
		svc.deleteToolCall,
	)

	svc.mux.HandleFunc("GET /executions", svc.listExecutions)
	svc.mux.HandleFunc("PUT /executions/{id}", svc.updateExecution)

	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.NewV4().String()
		ip    = func() string {
			if ip := r.Header.Get("X-Forwarded-For"); ip != "" {
				before, _, found := strings.Cut(ip, ",")
				if found {
					return strings.TrimSpace(before)
				}
				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")

	// Always set request ID header.
	w.Header().Set("X-Request-ID", id)

	// 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.
//
// The return value tells the caller when it is safe to close the
// database:
//
//   - nil: all HTTP handlers returned, and all inference calls and
//     scheduled jobs finished. The caller can close the database.
//   - non-nil: shutdown did not finish within the configured
//     ShutdownTimeout, or the HTTP server failed. Handlers,
//     inference calls, or jobs can still use the database. Do not
//     close the database. Exit the process instead. The database
//     runs in WAL mode, so committed data stays safe when the
//     process exits without a close.
func (svc *Service) Run(ctx context.Context) error {
	svc.log.Info(
		"starting odidere",
		slog.Group(
			"llm",
			slog.String(
				"default_provider",
				svc.cfg.DefaultModel.Provider,
			),
			slog.String(
				"default_model", svc.cfg.DefaultModel.Model,
			),
			slog.Int("providers", len(svc.cfg.Providers)),
		),
		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.ShutdownTimeout),
	)

	// Setup signal handling for graceful shutdown.
	ctx, cancel := signal.NotifyContext(
		ctx,
		os.Interrupt, syscall.SIGTERM,
	)
	defer cancel()
	svc.ctx, svc.cancel = context.WithCancel(ctx)
	defer svc.cancel()

	// Register jobs with cron scheduler.
	svc.cron = cron.New()
	for _, j := range svc.jobs {
		entry, err := svc.cron.AddFunc(j.Schedule(), func() {
			title := j.Name()
			system := j.SystemMessage()
			_, c, err := svc.newConversation(
				svc.ctx, j.UserId(), &title, system,
			)
			if err != nil {
				svc.log.Error("failed to create job conversation",
					slog.String("name", j.Name()),
					slog.Any("error", err),
				)
				return
			}
			if err := j.Run(svc.ctx, c); err != nil {
				svc.log.Error("job failed",
					slog.String("name", j.Name()),
					slog.Any("error", err),
				)
			}
		})
		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.
	var runErr error
	select {
	case <-ctx.Done():
		svc.log.Info("shutdown signal received")
	case err := <-errs:
		if err != nil {
			runErr = fmt.Errorf("server error: %w", err)
		}
	}

	// Use one deadline for HTTP shutdown and job completion.
	// The shutdown context must remain active after svc.ctx is canceled.
	shutdownCtx, shutdownCancel := context.WithTimeout(
		context.Background(),
		svc.cfg.ShutdownTimeout,
	)
	defer shutdownCancel()

	svc.cancel()

	svc.log.Info("stopping job scheduler")
	cronCtx := svc.cron.Stop()

	var shutdownErr error
	svc.log.Info("shutting down HTTP server")
	if err := svc.server.Shutdown(shutdownCtx); err != nil {
		shutdownErr = fmt.Errorf("HTTP shutdown: %w", err)
	} else {
		// Jobs and HTTP inference can save tool results after
		// cancellation. Wait until both return before the caller
		// closes the database.
		done := make(chan struct{})
		go func() {
			<-cronCtx.Done()
			svc.wg.Wait()
			close(done)
		}()
		select {
		case <-done:
		case <-shutdownCtx.Done():
			shutdownErr = fmt.Errorf(
				"wait for background work: %w",
				shutdownCtx.Err(),
			)
		}
	}

	// On failure, main exits without closing the database because
	// handlers or jobs may still use it.
	if shutdownErr != nil {
		if runErr != nil {
			return fmt.Errorf("%w; %w", runErr, shutdownErr)
		}
		return shutdownErr
	}

	svc.log.Info("terminating")
	return runErr
}

// newConversation saves a conversation and its initial system message.
// It registers the handle only after the transaction commits.
func (svc *Service) newConversation(
	ctx context.Context,
	userId string,
	title *string,
	system string,
) (*db.Conversation, *conversation.Conversation, error) {
	msg := llm.Message{
		Role: llm.RoleSystem,
		ContentParts: []llm.ContentPart{{
			Type: llm.ContentTypeText,
			Text: system,
		}},
	}
	if err := msg.Validate(); err != nil {
		return nil, nil, fmt.Errorf("invalid system message: %w", err)
	}
	content, err := json.Marshal(msg)
	if err != nil {
		return nil, nil, fmt.Errorf("marshal system message: %w", err)
	}

	tx, err := svc.db.Writer.BeginTx(ctx, nil)
	if err != nil {
		return nil, nil, fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback()

	qtx := svc.dbw.WithTx(tx)
	saved, err := qtx.InsertConversation(
		ctx, db.InsertConversationParams{
			Title:  title,
			UserId: userId,
		},
	)
	if err != nil {
		return nil, nil, fmt.Errorf("insert conversation: %w", err)
	}
	if _, err := qtx.InsertMessage(ctx, db.InsertMessageParams{
		Content:        content,
		ConversationId: saved.Id,
		Role:           db.MessageRoleSystem,
		Sequence:       0,
	}); err != nil {
		return nil, nil, fmt.Errorf("insert system message: %w", err)
	}

	c, err := conversation.New(
		conversation.Config{Id: saved.Id},
		svc.db,
		svc.log,
		svc.tools,
	)
	if err != nil {
		return nil, nil, fmt.Errorf("create conversation handle: %w", err)
	}
	if err := tx.Commit(); err != nil {
		return nil, nil, fmt.Errorf("commit conversation tx: %w", err)
	}

	svc.mu.Lock()
	// A retrieval can register a handle after the commit.
	if existing := svc.conversations[saved.Id]; existing != nil {
		c = existing
	} else {
		svc.conversations[saved.Id] = c
	}
	svc.mu.Unlock()

	return saved, c, nil
}

// getConversation checks saved state before it returns a cached handle.
func (svc *Service) getConversation(
	ctx context.Context,
	id string,
) (*conversation.Conversation, error) {
	if _, err := svc.dbr.SelectConversationId(ctx, id); err != nil {
		return nil, fmt.Errorf("select conversation: %w", err)
	}

	svc.mu.Lock()
	defer svc.mu.Unlock()

	if c := svc.conversations[id]; c != nil {
		return c, nil
	}
	c, err := conversation.New(
		conversation.Config{Id: id},
		svc.db,
		svc.log,
		svc.tools,
	)
	if err != nil {
		return nil, fmt.Errorf("create conversation handle: %w", err)
	}
	svc.conversations[id] = c
	return c, 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)
}

// ModelStatus represents the availability status of a model.
type ModelStatus string

const (
	// ModelStatusAvailable indicates the model is available for use.
	ModelStatusAvailable ModelStatus = "available"
	// ModelStatusNotFound indicates the model is configured but not found
	// in the provider's model list.
	ModelStatusNotFound ModelStatus = "not_found"
	// ModelStatusProviderError indicates the provider could not be reached
	// to verify the model.
	ModelStatusProviderError ModelStatus = "provider_error"
)

// Model represents a model in the /models response.
type Model struct {
	Model       string      `json:"model"`
	Status      ModelStatus `json:"status"`
	ContextSize *int        `json:"context_size,omitempty"`
}

// Models is the response format for the /models endpoint.
type Models struct {
	Providers       map[string][]Model `json:"providers"`
	DefaultProvider string             `json:"default_provider"`
	DefaultModel    string             `json:"default_model"`
}

// models returns available LLM models from all configured providers.
func (svc *Service) models(w http.ResponseWriter, r *http.Request) {
	var (
		ctx = r.Context()
		log = ctx.Value(logKey).(*slog.Logger)

		mu        sync.Mutex
		wg        sync.WaitGroup
		providers = make(map[string][]Model, len(svc.llms))
	)

	cfg := make(map[string]config.ProviderConfig, len(svc.cfg.Providers))
	for _, pc := range svc.cfg.Providers {
		cfg[pc.Name] = pc
	}

	for name, client := range svc.llms {
		wg.Add(1)
		go func(name string, c *llm.Client, pc config.ProviderConfig) {
			defer wg.Done()

			var result = []Model{}

			fetchCtx, cancel := context.WithTimeout(
				ctx, modelsTimeout,
			)
			defer cancel()

			models, err := c.ListModels(fetchCtx)
			if err != nil {
				log.WarnContext(
					ctx,
					"failed to list models for provider",
					slog.String("provider", name),
					slog.Any("error", err),
				)
				for _, m := range pc.Models {
					result = append(result, Model{
						Model:  m,
						Status: ModelStatusProviderError,
					})
				}
				mu.Lock()
				providers[name] = result
				mu.Unlock()
				return
			}

			available := make(map[string]struct{}, len(models))
			contextSizes := make(map[string]int, len(models))
			for _, m := range models {
				baseID, _, _ := strings.Cut(m.ID, ":")
				available[baseID] = struct{}{}
				cs := m.ContextLength
				if cs == 0 && m.Meta != nil {
					cs = m.Meta.NCtx
				}
				if cs > 0 {
					contextSizes[baseID] = cs
				}
			}

			var toCheck []string
			if len(pc.Models) > 0 {
				toCheck = pc.Models
			} else {
				toCheck = make([]string, 0, len(models))
				for _, m := range models {
					toCheck = append(toCheck, m.ID)
				}
			}
			for _, m := range toCheck {
				base, _, _ := strings.Cut(m, ":")
				mi := Model{Model: m}
				if _, ok := available[base]; ok {
					mi.Status = ModelStatusAvailable
					if cs, ok := contextSizes[base]; ok {
						mi.ContextSize = &cs
					}
				} else {
					mi.Status = ModelStatusNotFound
				}
				result = append(result, mi)
			}

			mu.Lock()
			providers[name] = result
			mu.Unlock()
		}(name, client, cfg[name])
	}
	wg.Wait()

	w.Header().Set("Content-Type", "application/json")
	if err := json.MarshalWrite(w, Models{
		Providers:       providers,
		DefaultProvider: svc.cfg.DefaultModel.Provider,
		DefaultModel:    svc.cfg.DefaultModel.Model,
	}); err != nil {
		log.ErrorContext(ctx, "failed to encode models response",
			slog.Any("error", err))
	}
}