Files
email_agent/internal/server/server.go
Sam Rolfe af3b5e5f0e Phase 3: JEV classification, rules table, editable categories (AUD-69)
Implements FR-3, FR-4, FR-5 and FR-14.

- internal/classify: TypeSafe System One (JEV) HTTP client with the confirmed
  request/response shape, retry with backoff on 429/529/5xx, a plain-text rule
  engine (first match wins, overrides only the named fields), and bounded
  orchestration with a per-run cap.
- internal/store: migration v3 adds an editable `categories` table (name =
  JEV Choice option and classifications key; rename cascades, remove reassigns)
  and seeds a starter rules version once. Classification persistence, coverage,
  and pending/all message queries.
- internal/server: POST /actions/classify and /actions/reclassify, plus the
  FR-14 category editor (GET /categories and add/rename/remove). The triage list
  now groups by the live category list and shows action + confidence.
- CLI: -classify and -classify-all for n8n.
- Docs: docs/phase3-classify.md with the confirmed JEV shape, a live read-only
  classification, and the DoD evidence.

DoD: all 100 stored messages have a non-null category, importance, and suggested
action (0 nulls); JEV picked a newly added/renamed category for 6 real messages
over HTTP. One live JEV call recorded in the phase doc.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
2026-10-11 21:37:38 +11:00

259 lines
8.0 KiB
Go

// Package server wires the HTTP surface of the Email Agent.
//
// Phase 1.2 exposes a session login and an empty HTMX dashboard shell in the
// Family Home Lab design system. Mail fetch arrives in Phase 2.
package server
import (
"context"
"crypto/rand"
"encoding/json"
"errors"
"fmt"
"html/template"
"io/fs"
"log/slog"
"mime"
"net/http"
"time"
"gitea.lab.audasmedia.com.au/sam/email_agent/internal/classify"
"gitea.lab.audasmedia.com.au/sam/email_agent/internal/config"
"gitea.lab.audasmedia.com.au/sam/email_agent/internal/mailfetch"
"gitea.lab.audasmedia.com.au/sam/email_agent/internal/session"
"gitea.lab.audasmedia.com.au/sam/email_agent/internal/store"
"gitea.lab.audasmedia.com.au/sam/email_agent/internal/version"
"gitea.lab.audasmedia.com.au/sam/email_agent/web"
)
func init() {
// Go's mime table has no woff2 entry; the self-hosted Inter fonts need it.
_ = mime.AddExtensionType(".woff2", "font/woff2")
}
// Server holds the HTTP dependencies.
type Server struct {
cfg *config.Config
store *store.Store
logger *slog.Logger
http *http.Server
tmpl *template.Template
sessions *session.Manager
fetcher *mailfetch.Fetcher
classifier *classify.Classifier
}
// New builds a Server. It does not start listening.
func New(cfg *config.Config, st *store.Store, logger *slog.Logger) (*Server, error) {
tmpl, err := template.ParseFS(web.TemplateFS, "templates/*.html")
if err != nil {
return nil, fmt.Errorf("parse templates: %w", err)
}
key, err := sessionKey(cfg, logger)
if err != nil {
return nil, fmt.Errorf("session key: %w", err)
}
sessions, err := session.NewManager(key, session.DefaultTTL)
if err != nil {
return nil, fmt.Errorf("session manager: %w", err)
}
s := &Server{
cfg: cfg,
store: st,
logger: logger,
tmpl: tmpl,
sessions: sessions,
fetcher: mailfetch.NewFetcher(cfg, st, logger),
classifier: classify.NewClassifier(
classify.NewClient(cfg.JEVBaseURL, cfg.JEVAPIKey, cfg.JEVModel, logger),
st, logger, cfg.ClassifyConcurrency, cfg.ClassifyMaxMessages,
),
}
s.http = &http.Server{
Addr: cfg.Addr,
Handler: s.Handler(),
ReadHeaderTimeout: 10 * time.Second,
ReadTimeout: 30 * time.Second,
WriteTimeout: 30 * time.Second,
IdleTimeout: 60 * time.Second,
}
return s, nil
}
// sessionKey returns the cookie signing key. In production it comes from
// EMAIL_AGENT_SESSION_SECRET; with no secret configured the service still
// starts with an ephemeral key so it is usable, but sessions do not survive a
// restart and the operator sees a warning.
func sessionKey(cfg *config.Config, logger *slog.Logger) ([]byte, error) {
if cfg.SessionSecret != "" {
return []byte(cfg.SessionSecret), nil
}
key := make([]byte, 32)
if _, err := rand.Read(key); err != nil {
return nil, err
}
logger.Warn("EMAIL_AGENT_SESSION_SECRET is not set; using an ephemeral session key (sessions reset on restart)")
return key, nil
}
// Handler returns the root HTTP handler with middleware applied.
func (s *Server) Handler() http.Handler {
staticFS, err := fs.Sub(web.StaticFS, "static")
if err != nil {
// web.StaticFS is a compile-time constant; this cannot fail at runtime.
panic(fmt.Sprintf("static assets: %v", err))
}
static := http.StripPrefix("/static/", http.FileServer(http.FS(staticFS)))
mux := http.NewServeMux()
mux.HandleFunc("GET /health", s.handleHealth)
mux.HandleFunc("GET /readyz", s.handleReady)
mux.Handle("GET /static/", static)
mux.HandleFunc("GET /login", s.handleLoginGet)
mux.HandleFunc("POST /login", s.handleLoginPost)
mux.HandleFunc("POST /logout", s.handleLogout)
mux.HandleFunc("GET /{$}", s.requireAuth(s.handleDashboard))
mux.HandleFunc("GET /partials/inbox", s.requireAuth(s.handleInboxPartial))
mux.HandleFunc("POST /actions/fetch", s.requireAuth(s.handleFetchMail))
mux.HandleFunc("POST /actions/classify", s.requireAuth(s.handleClassify))
mux.HandleFunc("POST /actions/reclassify", s.requireAuth(s.handleReclassify))
// FR-14: editable category list.
mux.HandleFunc("GET /categories", s.requireAuth(s.handleCategories))
mux.HandleFunc("POST /categories/add", s.requireAuth(s.handleCategoryAdd))
mux.HandleFunc("POST /categories/rename", s.requireAuth(s.handleCategoryRename))
mux.HandleFunc("POST /categories/remove", s.requireAuth(s.handleCategoryRemove))
return s.recoverPanic(s.logRequests(mux))
}
// Run starts the HTTP server and blocks until ctx is cancelled.
func (s *Server) Run(ctx context.Context) error {
errCh := make(chan error, 1)
go func() {
s.logger.Info("http server listening",
"addr", s.cfg.Addr,
"env", s.cfg.Env,
"db", s.store.Path(),
"version", version.Version,
)
if err := s.http.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
errCh <- err
return
}
errCh <- nil
}()
select {
case err := <-errCh:
return err
case <-ctx.Done():
shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
s.logger.Info("shutting down http server")
return s.http.Shutdown(shutdownCtx)
}
}
type healthResponse struct {
Status string `json:"status"`
Version string `json:"version"`
Commit string `json:"commit"`
Env string `json:"env"`
UptimeSeconds int64 `json:"uptime_seconds"`
Database string `json:"database"`
Time string `json:"time"`
}
var startedAt = time.Now()
// handleHealth reports liveness plus a database connectivity check.
func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
defer cancel()
resp := healthResponse{
Status: "ok",
Version: version.Version,
Commit: version.Commit,
Env: s.cfg.Env,
UptimeSeconds: int64(time.Since(startedAt).Seconds()),
Database: "ok",
Time: time.Now().UTC().Format(time.RFC3339),
}
code := http.StatusOK
if err := s.store.Ping(ctx); err != nil {
s.logger.Error("health check: database unreachable", "error", err)
resp.Status = "degraded"
resp.Database = "unreachable"
code = http.StatusServiceUnavailable
}
writeJSON(w, code, resp)
}
// handleReady reports readiness for load balancers (same signal as health).
func (s *Server) handleReady(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
defer cancel()
if err := s.store.Ping(ctx); err != nil {
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"status": "not ready"})
return
}
writeJSON(w, http.StatusOK, map[string]string{"status": "ready"})
}
func writeJSON(w http.ResponseWriter, code int, v any) {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(code)
_ = json.NewEncoder(w).Encode(v)
}
// logRequests logs one line per request at debug level.
func (s *Server) logRequests(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
start := time.Now()
rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK}
next.ServeHTTP(rec, r)
s.logger.Debug("request",
"method", r.Method,
"path", r.URL.Path,
"status", rec.status,
"duration_ms", time.Since(start).Milliseconds(),
"remote", r.RemoteAddr,
)
})
}
// recoverPanic keeps one bad request from taking down the process.
func (s *Server) recoverPanic(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
defer func() {
if rec := recover(); rec != nil {
s.logger.Error("panic recovered", "error", rec, "path", r.URL.Path)
http.Error(w, "internal server error", http.StatusInternalServerError)
}
}()
next.ServeHTTP(w, r)
})
}
type statusRecorder struct {
http.ResponseWriter
status int
}
// Unwrap lets http.ResponseController reach the underlying writer (the classify
// handler extends its write deadline for long runs).
func (r *statusRecorder) Unwrap() http.ResponseWriter { return r.ResponseWriter }
func (r *statusRecorder) WriteHeader(code int) {
r.status = code
r.ResponseWriter.WriteHeader(code)
}