- FR-11 rules editor: plain-text area at /rules, saved and versioned (SaveRules appends a version; restore re-saves an old version), parse errors surfaced, one-click re-classify. The reply model now reads the rules document too (FR-11). - FR-7 unsubscribe: parse and store List-Unsubscribe / List-Unsubscribe-Post (migration v6), detect email and web links (internal/unsubscribe), a domain-level ignore list, and an explicit record of what was marked. No code path contacts an unsubscribe link; acting on one needs Sam's approval. - Migration v6 is additive; unsubscribe gains url/updated_at. - Tests: detector, parser, store migration/records/candidates, rules editor, unsubscribe handlers, message-detail rendering. Co-Authored-By: Paperclip <noreply@paperclip.ing>
292 lines
9.6 KiB
Go
292 lines
9.6 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/mailops"
|
|
"gitea.lab.audasmedia.com.au/sam/email_agent/internal/reply"
|
|
"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
|
|
actions MailActions
|
|
replies ReplyGenerator
|
|
sender MailSender
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
|
|
mail := mailops.New(cfg, logger)
|
|
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,
|
|
),
|
|
actions: mail,
|
|
sender: mail,
|
|
replies: reply.NewGenerator(
|
|
reply.NewClient(cfg.OmniRouteBaseURL, cfg.OmniRouteAPIKey, cfg.OmniRouteModel, logger),
|
|
logger,
|
|
),
|
|
}
|
|
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))
|
|
|
|
// Phase 4 (FR-6/FR-9): bulk triage actions (delete, move, mark) and restore.
|
|
mux.HandleFunc("POST /actions/bulk", s.requireAuth(s.handleBulkAction))
|
|
|
|
// Phase 5 (FR-8/FR-15): message detail, reply draft, edit, and send.
|
|
mux.HandleFunc("GET /partials/message", s.requireAuth(s.handleMessageDetail))
|
|
mux.HandleFunc("POST /actions/draft", s.requireAuth(s.handleDraft))
|
|
mux.HandleFunc("POST /actions/draft/save", s.requireAuth(s.handleSaveDraft))
|
|
mux.HandleFunc("POST /actions/send", s.requireAuth(s.handleSendReply))
|
|
|
|
// 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))
|
|
|
|
// Phase 6 (FR-4, FR-11): rules editor, saved and versioned.
|
|
mux.HandleFunc("GET /rules", s.requireAuth(s.handleRules))
|
|
mux.HandleFunc("POST /rules/save", s.requireAuth(s.handleRulesSave))
|
|
mux.HandleFunc("POST /rules/restore", s.requireAuth(s.handleRulesRestore))
|
|
mux.HandleFunc("POST /rules/reclassify", s.requireAuth(s.handleRulesReclassify))
|
|
|
|
// Phase 6 (FR-7): unsubscribe detection, ignore list, and records.
|
|
mux.HandleFunc("GET /unsubscribe", s.requireAuth(s.handleUnsubscribe))
|
|
mux.HandleFunc("POST /unsubscribe/ignore", s.requireAuth(s.handleUnsubscribeIgnore))
|
|
mux.HandleFunc("POST /unsubscribe/unignore", s.requireAuth(s.handleUnsubscribeUnignore))
|
|
mux.HandleFunc("POST /unsubscribe/mark", s.requireAuth(s.handleUnsubscribeMark))
|
|
|
|
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)
|
|
}
|