// Command email-agent runs the Email Agent HTTP service. package main import ( "context" "flag" "fmt" "log/slog" "net/http" "os" "os/signal" "strings" "syscall" "time" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/apprise" "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/hub" "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/server" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/store" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/summary" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/version" ) func main() { var ( showVersion = flag.Bool("version", false, "print version information and exit") healthcheck = flag.Bool("healthcheck", false, "probe the local /health endpoint and exit") fetchNow = flag.Bool("fetch", false, "run one Gmail IMAP scan and exit") classifyNow = flag.Bool("classify", false, "classify pending messages with JEV and exit") classifyAll = flag.Bool("classify-all", false, "re-classify every stored message with JEV and exit") checkSMTP = flag.Bool("check-smtp", false, "verify the Gmail SMTP connection (IPv4 + STARTTLS + login) and exit; sends no mail") checkThread = flag.Int64("check-thread", 0, "verify that a stored message's Gmail thread has a sent reply (read-only), then exit") summary = flag.Bool("summary", false, "build today's inbox summary, store it, and print it; sends nothing") summarySend = flag.Bool("summary-send", false, "build today's inbox summary and deliver it by email via Apprise") notify = flag.Bool("notify", false, "send one test notification to the Hub (POST /notify) and exit") retain = flag.Bool("retain", false, "enforce retention (AGENTS.md §7): clear old message bodies and cap actions_log, then exit") ) flag.Parse() if *showVersion { fmt.Printf("email-agent %s (commit %s)\n", version.Version, version.Commit) return } cfg, err := config.Load() if err != nil { fmt.Fprintf(os.Stderr, "config error: %v\n", err) os.Exit(1) } logger := newLogger(cfg.LogLevel) slog.SetDefault(logger) if *healthcheck { if err := probeHealth(cfg.Port); err != nil { fmt.Fprintf(os.Stderr, "healthcheck failed: %v\n", err) os.Exit(1) } return } if *fetchNow { if err := runFetch(cfg, logger); err != nil { fmt.Fprintf(os.Stderr, "fetch failed: %v\n", err) os.Exit(1) } return } if *classifyNow || *classifyAll { if err := runClassify(cfg, logger, *classifyAll); err != nil { fmt.Fprintf(os.Stderr, "classify failed: %v\n", err) os.Exit(1) } return } if *checkSMTP { if err := runCheckSMTP(cfg, logger); err != nil { fmt.Fprintf(os.Stderr, "smtp check failed: %v\n", err) os.Exit(1) } fmt.Println("smtp check ok: ipv4 + starttls + auth (no mail sent)") return } if *checkThread > 0 { if err := runCheckThread(cfg, logger, *checkThread); err != nil { fmt.Fprintf(os.Stderr, "thread check failed: %v\n", err) os.Exit(1) } return } if *summary || *summarySend { if err := runSummary(cfg, logger, *summarySend); err != nil { fmt.Fprintf(os.Stderr, "summary failed: %v\n", err) os.Exit(1) } return } if *notify { if err := runNotify(cfg, logger); err != nil { fmt.Fprintf(os.Stderr, "notify failed: %v\n", err) os.Exit(1) } return } if *retain { if err := runRetain(cfg, logger); err != nil { fmt.Fprintf(os.Stderr, "retain failed: %v\n", err) os.Exit(1) } return } if err := run(cfg, logger); err != nil { logger.Error("fatal", "error", err) os.Exit(1) } } func run(cfg *config.Config, logger *slog.Logger) error { logger.Info("starting email-agent", "version", version.Version, "commit", version.Commit, "env", cfg.Env, ) if cfg.SessionSecret == "" && cfg.Password == "" { logger.Warn("no EMAIL_AGENT_PASSWORD or EMAIL_AGENT_SESSION_SECRET set; login is disabled until one is configured") } ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() st, err := store.Open(ctx, cfg.DBPath) if err != nil { return fmt.Errorf("open store: %w", err) } defer func() { if cerr := st.Close(); cerr != nil { logger.Error("close store", "error", cerr) } }() schemaVersion, err := st.SchemaVersion(ctx) if err != nil { return fmt.Errorf("read schema version: %w", err) } logger.Info("database ready", "path", st.Path(), "schema_version", schemaVersion) srv, err := server.New(cfg, st, logger) if err != nil { return fmt.Errorf("build server: %w", err) } return srv.Run(ctx) } // runFetch performs a single IMAP scan and exits. It is the entry point n8n // (FR-1) and manual operations use to trigger a sync without the HTTP UI. func runFetch(cfg *config.Config, logger *slog.Logger) error { if !cfg.GmailConfigured() { return mailfetch.ErrNoCredentials } ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() st, err := store.Open(ctx, cfg.DBPath) if err != nil { return fmt.Errorf("open store: %w", err) } defer func() { _ = st.Close() }() sum, err := mailfetch.NewFetcher(cfg, st, logger).Fetch(ctx) if err != nil { return err } fmt.Printf("fetched=%d stored=%d uidvalidity=%d since=%s\n", sum.Fetched, sum.Stored, sum.UIDValidity, sum.Since.Format("2006-01-02")) return nil } // runClassify classifies messages and exits. It is the n8n/CLI entry point for // FR-3, separate from the HTTP service so a scheduled scan can be bounded and // observed. all=true re-classifies every message, for use after the category // list or rules change. func runClassify(cfg *config.Config, logger *slog.Logger, all bool) error { if !cfg.JEVConfigured() { return classify.ErrNotConfigured } ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() st, err := store.Open(ctx, cfg.DBPath) if err != nil { return fmt.Errorf("open store: %w", err) } defer func() { _ = st.Close() }() classifier := classify.NewClassifier( classify.NewClient(cfg.JEVBaseURL, cfg.JEVAPIKey, cfg.JEVModel, logger), st, logger, cfg.ClassifyConcurrency, cfg.ClassifyMaxMessages, ) var sum classify.Summary if all { sum, err = classifier.RunAll(ctx) } else { sum, err = classifier.Run(ctx) } if err != nil { return err } fmt.Printf("considered=%d classified=%d failed=%d remaining=%d model=%s rules_version=%d\n", sum.Considered, sum.Classified, sum.Failed, sum.Remaining, sum.Model, sum.RulesVersion) return nil } // runCheckSMTP verifies the Gmail submission path (IPv4 + STARTTLS + app // password) without sending a message. It is the safe part of the Phase 5 send // verification; the real send needs Sam's explicit approval. func runCheckSMTP(cfg *config.Config, logger *slog.Logger) error { if !cfg.GmailConfigured() { return mailops.ErrNotConfigured } ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() return mailops.New(cfg, logger).CheckSMTP(ctx) } // runCheckThread reports whether a stored message's Gmail thread already has a // message in the Sent mailbox. It is read-only and exists so the Phase 5 thread // check can be exercised against real Gmail without sending anything. func runCheckThread(cfg *config.Config, logger *slog.Logger, messageID int64) error { if !cfg.GmailConfigured() { return mailops.ErrNotConfigured } ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() st, err := store.Open(ctx, cfg.DBPath) if err != nil { return fmt.Errorf("open store: %w", err) } defer func() { _ = st.Close() }() msg, err := st.MessageByID(ctx, messageID) if err != nil { return err } if strings.TrimSpace(msg.ThreadID) == "" { return fmt.Errorf("message %d has no Gmail thread id", messageID) } ok, err := mailops.New(cfg, logger).VerifyThreadUpdated(ctx, msg.ThreadID) if err != nil { return err } fmt.Printf("message=%d thread=%s sent_reply_in_thread=%v\n", messageID, msg.ThreadID, ok) return nil } // runSummary builds today's inbox summary (FR-10). With send=false it stores and // prints the summary and sends nothing; with send=true it also delivers it by // email through Apprise. The n8n schedule runs the send form once a day. func runSummary(cfg *config.Config, logger *slog.Logger, send bool) error { ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() st, err := store.Open(ctx, cfg.DBPath) if err != nil { return fmt.Errorf("open store: %w", err) } defer func() { _ = st.Close() }() data, err := summary.New(st).Build(ctx) if err != nil { return fmt.Errorf("build summary: %w", err) } id, err := st.SaveSummary(ctx, data.Date, data.Text(), "") if err != nil { return err } fmt.Print(data.Text()) if !send { fmt.Printf("\nsummary stored id=%d (not sent)\n", id) return nil } if !cfg.AppriseConfigured() { return fmt.Errorf("apprise is not configured: set APPRISE_URL and APPRISE_KEY (or APPRISE_URLS)") } if err := apprise.New(cfg.AppriseURL, cfg.AppriseKey, cfg.AppriseURLs, cfg.AppriseToken, 0). Send(ctx, data.Subject(), data.Text()); err != nil { return fmt.Errorf("deliver summary: %w", err) } if err := st.MarkSummaryDelivered(ctx, id, cfg.SummaryTo); err != nil { logger.Error("mark summary delivered", "error", err) } fmt.Printf("\nsummary delivered id=%d to=%s\n", id, cfg.SummaryTo) return nil } // runNotify sends one test notification to the Hub's POST /notify. It exists so // the outbound notify path (ask-contract.md) can be exercised without the UI. func runNotify(cfg *config.Config, logger *slog.Logger) error { ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() n := hub.NewNotifier(cfg.HubBaseURL, cfg.HubToken, cfg.HubNotifyTimeout, nil) if err := n.Notify(ctx, hub.NotifyRequest{ From: hub.ProjectName, Level: "info", Message: "Email Agent notify path check.", Person: "sam", }); err != nil { return err } fmt.Printf("notify ok: %s/notify\n", cfg.HubBaseURL) return nil } // runRetain enforces the AGENTS.md §7 data rules: message bodies older than the // configured window are cleared (metadata stays), and the append-only // actions_log is capped. It is a bounded, explicit job for the n8n schedule; // the HTTP service never runs it on a request. Running it twice is safe. func runRetain(cfg *config.Config, logger *slog.Logger) error { ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() st, err := store.Open(ctx, cfg.DBPath) if err != nil { return fmt.Errorf("open store: %w", err) } defer func() { _ = st.Close() }() policy := store.RetentionPolicy{ BodyMaxAge: time.Duration(cfg.RetentionDays) * 24 * time.Hour, ActionsLogMaxRows: cfg.ActionsLogMaxRows, } res, err := st.Retain(ctx, policy) if err != nil { return err } if err := st.LogAction(ctx, "retain", "scheduler", fmt.Sprintf("bodies_cleared=%d actions_deleted=%d retention_days=%d max_rows=%d", res.BodiesCleared, res.ActionsDeleted, cfg.RetentionDays, cfg.ActionsLogMaxRows)); err != nil { // Retention already committed; a failed audit row must not fail the job. logger.Error("log retention action", "error", err) } cutoff := "disabled" if !res.BodyCutoff.IsZero() { cutoff = res.BodyCutoff.Format("2006-01-02") } fmt.Printf("bodies_cleared=%d body_cutoff=%s actions_deleted=%d actions_max_rows=%d\n", res.BodiesCleared, cutoff, res.ActionsDeleted, cfg.ActionsLogMaxRows) return nil } // probeHealth requests the local health endpoint. It exists so the container // healthcheck does not need an external HTTP client in the image. func probeHealth(port int) error { client := &http.Client{Timeout: 3 * time.Second} url := fmt.Sprintf("http://127.0.0.1:%d/health", port) resp, err := client.Get(url) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { return fmt.Errorf("unexpected status %d from %s", resp.StatusCode, url) } return nil } func newLogger(level string) *slog.Logger { var lvl slog.Level if err := lvl.UnmarshalText([]byte(level)); err != nil { lvl = slog.LevelInfo } handler := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: lvl}) return slog.New(handler) }