szx/NewSzxcn-Email
1
0
Fork 0
Code Issues Pull Requests Actions 57 Packages Projects Releases Wiki Activity

feat(admin): 增加 Maildir 同步健康状态

Browse Source 
- 新增后端健康检查接口与同步追踪，记录运行状态、最近结果、错误摘要和统计信息。
- 前端管理页在存储设置中展示 Maildir 同步健康卡片，并支持手动刷新。
- 补充相关类型定义与测试，覆盖未配置和同步成功两类场景。
This commit is contained in:
LanQin_
2026-06-24 16:08:48 +08:00
parent b36adfdac2
commit d28ed4adcc
8 changed files with 488 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/internal/app/app.go
+8 -7
View File
@@ -22,12 +22,13 @@ import (
)
type App struct {
cfg Config
db *sql.DB
log *slog.Logger
now func() time.Time
policy *HTMLPolicy
workerCancel context.CancelFunc
cfg Config
db *sql.DB
log *slog.Logger
now func() time.Time
policy *HTMLPolicy
workerCancel context.CancelFunc
maildirHealth *maildirSyncHealthTracker
}
func New(cfg Config, logger *slog.Logger) (*App, error) {
@@ -47,7 +48,7 @@ func New(cfg Config, logger *slog.Logger) (*App, error) {
}
db.SetMaxOpenConns(1)
a := &App{cfg: cfg, db: db, log: logger, now: time.Now, policy: NewHTMLPolicy()}
a := &App{cfg: cfg, db: db, log: logger, now: time.Now, policy: NewHTMLPolicy(), maildirHealth: newMaildirSyncHealthTracker()}
if err := a.configureSQLite(context.Background()); err != nil {
db.Close()
return nil, err
apps/api/internal/app/app_test.go
+97
View File
@@ -2323,6 +2323,103 @@ func TestMaildirSyncImportsRFC822(t *testing.T) {
}
}
func TestMaildirSyncHealthDisabled(t *testing.T) {
a := newTestApp(t)
ts := httptest.NewServer(a.Router())
defer ts.Close()
admin := &testClient{t: t, server: ts}
var login map[string]any
if code := admin.do("POST", "/api/auth/login", map[string]string{"email": "admin@lanqin.local", "password": "ChangeMe123!"}, &login); code != http.StatusOK {
t.Fatalf("login code=%d body=%v", code, login)
}
var health maildirSyncHealthResponse
if code := admin.do("GET", "/api/admin/maildir-sync/health", nil, &health); code != http.StatusOK {
t.Fatalf("health code=%d body=%+v", code, health)
}
if health.Configured || health.Enabled || health.WorkerStarted || health.Running {
t.Fatalf("unexpected disabled health: %+v", health)
}
if health.ScanSeconds != 30 {
t.Fatalf("scan seconds=%d, want default 30", health.ScanSeconds)
}
}
func TestMaildirSyncHealthAfterTrackedSync(t *testing.T) {
a := newTestApp(t)
ctx := context.Background()
root := t.TempDir()
a.cfg.MaildirRoot = root
a.cfg.MaildirScanSeconds = 45
adminUser, _, err := a.userByEmail(ctx, "admin@lanqin.local")
if err != nil {
t.Fatal(err)
}
var mailboxID string
if err := a.db.QueryRowContext(ctx, `SELECT id FROM mailboxes WHERE user_id=? AND address=?`, adminUser.ID, "admin@lanqin.local").Scan(&mailboxID); err != nil {
t.Fatal(err)
}
if _, err := a.db.ExecContext(ctx, `DELETE FROM messages WHERE mailbox_id=?`, mailboxID); err != nil {
t.Fatal(err)
}
mailboxes, err := a.maildirMailboxes(ctx)
if err != nil {
t.Fatal(err)
}
var admin maildirMailbox
for _, mb := range mailboxes {
if mb.Address == "admin@lanqin.local" {
admin = mb
break
}
}
if admin.ID == "" {
t.Fatal("admin mailbox not found")
}
dir := filepath.Join(root, admin.Domain, admin.LocalPart, "Maildir", "new")
if err := os.MkdirAll(dir, 0o755); err != nil {
t.Fatal(err)
}
raw := strings.Join([]string{
"From: sender@example.test",
"To: admin@lanqin.local",
"Subject: Maildir health import",
"Message-Id: <maildir-health@example.test>",
"Date: Sat, 13 Jun 2026 15:00:00 +0000",
"MIME-Version: 1.0",
"Content-Type: text/plain; charset=utf-8",
"",
"hello from health test",
}, "\r\n")
if err := os.WriteFile(filepath.Join(dir, "1749826800.M1P1.health"), []byte(raw), 0o600); err != nil {
t.Fatal(err)
}
counts, err := a.syncMaildirOnceTracked(ctx, time.Minute)
if err != nil {
t.Fatal(err)
}
if counts.Imported != 1 || counts.FilesScanned != 1 {
t.Fatalf("counts=%+v, want imported=1 filesScanned=1", counts)
}
health := a.maildirHealth.snapshot(a.cfg)
if !health.Configured || !health.Enabled {
t.Fatalf("configured health=%+v, want enabled", health)
}
if health.Running {
t.Fatalf("health still running: %+v", health)
}
if health.LastRun == nil || health.LastRun.Status != "success" {
t.Fatalf("last run=%+v, want success", health.LastRun)
}
if health.LastRun.Counts.Imported != 1 || health.Summary.Imported != 1 {
t.Fatalf("health counts last=%+v summary=%+v", health.LastRun.Counts, health.Summary)
}
if health.NextRunAt == nil {
t.Fatalf("next run is nil")
}
}
func TestMaildirSyncImportsSentFolder(t *testing.T) {
a := newTestApp(t)
ctx := context.Background()
apps/api/internal/app/maildir_health.go
+197
View File
@@ -0,0 +1,197 @@
package app
import (
"net/http"
"strings"
"sync"
"time"
)
const maxMaildirRecentErrors = 10
type maildirSyncCounts struct {
FilesScanned int `json:"filesScanned"`
Imported int `json:"imported"`
Backfilled int `json:"backfilled"`
Cleaned int `json:"cleaned"`
FileErrors int `json:"fileErrors"`
fileErrorDetails []string `json:"-"`
}
func (c maildirSyncCounts) total() int {
return c.Imported + c.Backfilled + c.Cleaned
}
type maildirSyncRun struct {
StartedAt time.Time `json:"startedAt"`
FinishedAt *time.Time `json:"finishedAt,omitempty"`
DurationMs int64 `json:"durationMs"`
Status string `json:"status"`
Error string `json:"error,omitempty"`
Counts maildirSyncCounts `json:"counts"`
}
type maildirSyncHealthResponse struct {
Configured bool `json:"configured"`
Enabled bool `json:"enabled"`
Root string `json:"root"`
ScanSeconds int `json:"scanSeconds"`
WorkerStarted bool `json:"workerStarted"`
Running bool `json:"running"`
LastRun *maildirSyncRun `json:"lastRun,omitempty"`
LastError string `json:"lastError,omitempty"`
NextRunAt *time.Time `json:"nextRunAt,omitempty"`
RecentErrors []string `json:"recentErrors"`
Summary maildirSyncCounts `json:"summary"`
}
type maildirSyncHealthTracker struct {
mu sync.Mutex
workerStarted bool
running bool
current *maildirSyncRun
lastRun *maildirSyncRun
lastError string
nextRunAt *time.Time
recentErrors []string
summary maildirSyncCounts
}
func newMaildirSyncHealthTracker() *maildirSyncHealthTracker {
return &maildirSyncHealthTracker{}
}
func (h *maildirSyncHealthTracker) markWorkerStarted(nextRunAt *time.Time) {
if h == nil {
return
}
h.mu.Lock()
defer h.mu.Unlock()
h.workerStarted = true
h.nextRunAt = cloneTimePtr(nextRunAt)
}
func (h *maildirSyncHealthTracker) markWorkerStopped() {
if h == nil {
return
}
h.mu.Lock()
defer h.mu.Unlock()
h.workerStarted = false
h.nextRunAt = nil
}
func (h *maildirSyncHealthTracker) markRunStarted(startedAt time.Time) {
if h == nil {
return
}
h.mu.Lock()
defer h.mu.Unlock()
run := &maildirSyncRun{StartedAt: startedAt.UTC(), Status: "running"}
h.running = true
h.current = run
h.lastRun = cloneMaildirSyncRun(run)
}
func (h *maildirSyncHealthTracker) markRunFinished(finishedAt time.Time, counts maildirSyncCounts, err error, nextRunAt *time.Time) {
if h == nil {
return
}
h.mu.Lock()
defer h.mu.Unlock()
run := h.current
if run == nil {
run = &maildirSyncRun{StartedAt: finishedAt.UTC()}
}
finished := finishedAt.UTC()
run.FinishedAt = &finished
run.DurationMs = finished.Sub(run.StartedAt).Milliseconds()
run.Counts = counts
run.Status = "success"
run.Error = ""
if err != nil {
run.Status = "error"
run.Error = err.Error()
h.lastError = run.Error
h.pushRecentError(run.Error)
} else if counts.FileErrors > 0 {
run.Status = "partial"
if len(counts.fileErrorDetails) > 0 {
run.Error = counts.fileErrorDetails[0]
h.lastError = run.Error
}
for _, detail := range counts.fileErrorDetails {
h.pushRecentError(detail)
}
} else {
h.lastError = ""
}
h.summary.FilesScanned += counts.FilesScanned
h.summary.Imported += counts.Imported
h.summary.Backfilled += counts.Backfilled
h.summary.Cleaned += counts.Cleaned
h.summary.FileErrors += counts.FileErrors
h.running = false
h.current = nil
h.lastRun = cloneMaildirSyncRun(run)
h.nextRunAt = cloneTimePtr(nextRunAt)
}
func (h *maildirSyncHealthTracker) snapshot(cfg Config) maildirSyncHealthResponse {
root := strings.TrimSpace(cfg.MaildirRoot)
scanSeconds := cfg.MaildirScanSeconds
if scanSeconds <= 0 {
scanSeconds = 30
}
out := maildirSyncHealthResponse{
Configured: root != "",
Enabled: root != "",
Root: root,
ScanSeconds: scanSeconds,
}
if h == nil {
return out
}
h.mu.Lock()
defer h.mu.Unlock()
out.WorkerStarted = h.workerStarted
out.Running = h.running
out.LastRun = cloneMaildirSyncRun(h.lastRun)
out.LastError = h.lastError
out.NextRunAt = cloneTimePtr(h.nextRunAt)
out.RecentErrors = append([]string(nil), h.recentErrors...)
out.Summary = h.summary
return out
}
func (h *maildirSyncHealthTracker) pushRecentError(value string) {
value = strings.TrimSpace(value)
if value == "" {
return
}
h.recentErrors = append([]string{value}, h.recentErrors...)
if len(h.recentErrors) > maxMaildirRecentErrors {
h.recentErrors = h.recentErrors[:maxMaildirRecentErrors]
}
}
func cloneMaildirSyncRun(in *maildirSyncRun) *maildirSyncRun {
if in == nil {
return nil
}
out := *in
out.FinishedAt = cloneTimePtr(in.FinishedAt)
return &out
}
func cloneTimePtr(in *time.Time) *time.Time {
if in == nil {
return nil
}
out := in.UTC()
return &out
}
func (a *App) handleMaildirSyncHealth(w http.ResponseWriter, r *http.Request) {
respondJSON(w, http.StatusOK, a.maildirHealth.snapshot(a.cfg))
}
apps/api/internal/app/maildir_sync.go
+60 -24
View File
@@ -49,10 +49,12 @@ func (a *App) maildirWorker(ctx context.Context) {
if interval <= 0 {
interval = 30 * time.Second
}
nextRunAt := a.now().UTC()
a.maildirHealth.markWorkerStarted(&nextRunAt)
a.log.Info("maildir sync worker started", "root", a.cfg.MaildirRoot, "interval", interval.String())
if n, err := a.syncMaildirOnce(ctx); err != nil {
if counts, err := a.syncMaildirOnceTracked(ctx, interval); err != nil {
a.log.Warn("initial maildir sync failed", "error", err)
} else if n > 0 {
} else if n := counts.total(); n > 0 {
a.log.Info("initial maildir sync processed messages", "count", n)
}
ticker := time.NewTicker(interval)
@@ -60,43 +62,66 @@ func (a *App) maildirWorker(ctx context.Context) {
for {
select {
case <-ctx.Done():
a.maildirHealth.markWorkerStopped()
a.log.Info("maildir sync worker stopped")
return
case <-ticker.C:
n, err := a.syncMaildirOnce(ctx)
counts, err := a.syncMaildirOnceTracked(ctx, interval)
if err != nil {
a.log.Warn("maildir sync failed", "error", err)
continue
}
if n > 0 {
if n := counts.total(); n > 0 {
a.log.Info("maildir sync processed messages", "count", n)
}
}
}
}
func (a *App) syncMaildirOnceTracked(ctx context.Context, interval time.Duration) (maildirSyncCounts, error) {
startedAt := a.now().UTC()
a.maildirHealth.markRunStarted(startedAt)
counts, err := a.syncMaildirOnceDetailed(ctx)
finishedAt := a.now().UTC()
var nextRunAt *time.Time
if interval > 0 && err == nil {
next := finishedAt.Add(interval)
nextRunAt = &next
}
a.maildirHealth.markRunFinished(finishedAt, counts, err, nextRunAt)
return counts, err
}
func (a *App) syncMaildirOnce(ctx context.Context) (int, error) {
counts, err := a.syncMaildirOnceDetailed(ctx)
return counts.total(), err
}
func (a *App) syncMaildirOnceDetailed(ctx context.Context) (maildirSyncCounts, error) {
root := strings.TrimSpace(a.cfg.MaildirRoot)
if root == "" {
return 0, nil
return maildirSyncCounts{}, nil
}
mailboxes, err := a.maildirMailboxes(ctx)
if err != nil {
return 0, err
return maildirSyncCounts{}, err
}
imported := 0
counts := maildirSyncCounts{}
for _, mb := range mailboxes {
if mb.Unregistered {
count, err := a.syncUnregisteredMaildir(ctx, mb)
mbCounts, err := a.syncUnregisteredMaildirDetailed(ctx, mb)
counts.FilesScanned += mbCounts.FilesScanned
counts.Imported += mbCounts.Imported
counts.FileErrors += mbCounts.FileErrors
counts.fileErrorDetails = append(counts.fileErrorDetails, mbCounts.fileErrorDetails...)
if err != nil {
return imported, err
return counts, err
}
imported += count
continue
}
folders, err := a.maildirFolders(ctx, mb.ID)
if err != nil {
return imported, err
return counts, err
}
base := filepath.Join(root, mb.Domain, mb.LocalPart, "Maildir")
for _, folder := range folders {
@@ -104,7 +129,7 @@ func (a *App) syncMaildirOnce(ctx context.Context) (int, error) {
for _, sub := range []string{"new", "cur"} {
select {
case <-ctx.Done():
return imported, ctx.Err()
return counts, ctx.Err()
default:
}
dir := filepath.Join(folderBase, sub)
@@ -113,20 +138,23 @@ func (a *App) syncMaildirOnce(ctx context.Context) (int, error) {
if errors.Is(err, os.ErrNotExist) {
continue
}
return imported, err
return counts, err
}
for _, entry := range entries {
if entry.IsDir() || strings.HasPrefix(entry.Name(), ".") {
continue
}
path := filepath.Join(dir, entry.Name())
counts.FilesScanned++
ok, err := a.syncMaildirFile(ctx, mb, folder, path)
if err != nil {
counts.FileErrors++
counts.fileErrorDetails = append(counts.fileErrorDetails, fmt.Sprintf("%s: %v", path, err))
a.log.Warn("maildir file import failed", "path", path, "error", err)
continue
}
if ok {
imported++
counts.Imported++
}
}
}
@@ -134,15 +162,15 @@ func (a *App) syncMaildirOnce(ctx context.Context) (int, error) {
}
backfilled, err := a.backfillSQLiteMessagesToMaildir(ctx)
if err != nil {
return imported, err
return counts, err
}
imported += backfilled
counts.Backfilled += backfilled
cleaned, err := a.cleanupMissingMaildirMessages(ctx)
if err != nil {
return imported, err
return counts, err
}
imported += cleaned
return imported, nil
counts.Cleaned += cleaned
return counts, nil
}
func (a *App) maildirMailboxes(ctx context.Context) ([]maildirMailbox, error) {
@@ -189,12 +217,17 @@ func (a *App) maildirMailboxes(ctx context.Context) ([]maildirMailbox, error) {
}
func (a *App) syncUnregisteredMaildir(ctx context.Context, mb maildirMailbox) (int, error) {
counts, err := a.syncUnregisteredMaildirDetailed(ctx, mb)
return counts.Imported, err
}
func (a *App) syncUnregisteredMaildirDetailed(ctx context.Context, mb maildirMailbox) (maildirSyncCounts, error) {
base := filepath.Join(strings.TrimSpace(a.cfg.MaildirRoot), mb.Domain, mb.LocalPart, "Maildir")
imported := 0
counts := maildirSyncCounts{}
for _, sub := range []string{"new", "cur"} {
select {
case <-ctx.Done():
return imported, ctx.Err()
return counts, ctx.Err()
default:
}
dir := filepath.Join(base, sub)
@@ -203,24 +236,27 @@ func (a *App) syncUnregisteredMaildir(ctx context.Context, mb maildirMailbox) (i
if errors.Is(err, os.ErrNotExist) {
continue
}
return imported, err
return counts, err
}
for _, entry := range entries {
if entry.IsDir() || strings.HasPrefix(entry.Name(), ".") {
continue
}
path := filepath.Join(dir, entry.Name())
counts.FilesScanned++
ok, err := a.syncUnregisteredMaildirFile(ctx, mb, path)
if err != nil {
counts.FileErrors++
counts.fileErrorDetails = append(counts.fileErrorDetails, fmt.Sprintf("%s: %v", path, err))
a.log.Warn("unregistered maildir file import failed", "path", path, "error", err)
continue
}
if ok {
imported++
counts.Imported++
}
}
}
return imported, nil
return counts, nil
}
func (a *App) syncUnregisteredMaildirFile(ctx context.Context, mb maildirMailbox, path string) (bool, error) {
apps/api/internal/app/router_auth.go
+1
View File
@@ -118,6 +118,7 @@ func (a *App) Router() http.Handler {
r.With(a.requirePermission(PermissionMessagesRead)).Get("/admin/messages/{id}", a.handleAdminMessage)
r.With(a.requirePermission(PermissionMessagesAttachment)).Get("/admin/attachments/{id}", a.handleAdminAttachment)
r.With(a.requirePermission(PermissionSettingsView)).Get("/admin/settings", a.handleGetSystemSettings)
r.With(a.requirePermission(PermissionSettingsView)).Get("/admin/maildir-sync/health", a.handleMaildirSyncHealth)
r.With(a.requirePermission(PermissionSettingsUpdate)).Post("/admin/settings", a.handleUpdateSystemSettings)
r.With(a.requirePermission(PermissionSettingsTestSMTP)).Post("/admin/settings/test-smtp", a.handleTestSMTP)
r.With(a.requirePermission(PermissionTemplatesView)).Get("/admin/mail-templates", a.handleListMailTemplates)