From 6ebb6f56d64dad98093997601c086590fb9b95d1 Mon Sep 17 00:00:00 2001 From: Mikhail Yenuchenko Date: Mon, 13 Jul 2026 22:58:34 +0300 Subject: [PATCH] Phase 6: journal-milter + send-log status tailer + retention Implement the structured send log (spec 7.3), the project's highest-risk component since a milter bug can break the relay itself. - internal/milter: go-milter v0.4.1 journal-milter. Per-connection session collects SASL login, From, recipients and Subject across callbacks and writes one send_log "queued" row per (queue-id, recipient) at EOM (spec 7.3.3). Monitoring only: callbacks return Continue/Accept, recorder errors are logged never propagated, so it can never block mail. - internal/logtail: polling mail.log tailer with rotation handling (inode change / truncation), parses sent/deferred/bounced/expired by queue-id + recipient and advances rows; background retention sweep prunes rows past SEND_LOG_RETENTION_DAYS (default 90) at startup and every 6h. - internal/store/sendlog.go: InsertQueued, UpdateStatus (case-insensitive recipient match), DeleteSendLogBefore + status constants. - cmd/panel: open the store once and share it across http/milter/tailer; replace the journal/logtail stubs with the real roles. - build/postfix-config.sh: bounded milter timeouts (15/15/30s) so a hung milter also fails open in seconds, not the 300s default. Fix found in-container: SASL login (app_login) was empty because go-milter keys macros exactly as Postfix sends them, and multi-character macro names arrive brace-wrapped ({auth_authen}); the SASL-less Phase 0 spike could not observe this. Added a brace-tolerant macro lookup. Verified on selfpost.example.com: gofmt/vet/unit tests green; container e2e records rows with correct fields and advances status via the tailer; fail-open confirmed for both an unreachable and a hung milter; retention prunes at start. Co-Authored-By: Claude Opus 4.8 --- build/postfix-config.sh | 12 ++ cmd/panel/httpserver.go | 14 +- cmd/panel/{journalstub.go => journal.go} | 34 ++-- cmd/panel/logtailstub.go | 16 -- cmd/panel/main.go | 34 +++- go.mod | 2 + go.sum | 35 ++++ internal/logtail/logtail.go | 209 +++++++++++++++++++++++ internal/logtail/logtail_test.go | 176 +++++++++++++++++++ internal/milter/milter.go | 174 +++++++++++++++++++ internal/milter/milter_test.go | 156 +++++++++++++++++ internal/store/sendlog.go | 82 +++++++++ internal/store/sendlog_test.go | 140 +++++++++++++++ 13 files changed, 1033 insertions(+), 51 deletions(-) rename cmd/panel/{journalstub.go => journal.go} (53%) delete mode 100644 cmd/panel/logtailstub.go create mode 100644 internal/logtail/logtail.go create mode 100644 internal/logtail/logtail_test.go create mode 100644 internal/milter/milter.go create mode 100644 internal/milter/milter_test.go create mode 100644 internal/store/sendlog.go create mode 100644 internal/store/sendlog_test.go diff --git a/build/postfix-config.sh b/build/postfix-config.sh index 0b64d1c..1ee6b83 100644 --- a/build/postfix-config.sh +++ b/build/postfix-config.sh @@ -122,6 +122,18 @@ postconf -e \ "smtpd_milters={ unix:${OPENDKIM_SOCK}, default_action=tempfail }, { unix:${JOURNAL_SOCK}, default_action=accept }" \ "non_smtpd_milters=" +# Bounded milter timeouts (spec 7.3): a *hung* milter (socket accepts but never +# replies) must fail open just like a crash, not stall mail acceptance until the +# Postfix defaults (300s content) elapse. With default_action per milter, a +# journal-milter hang then resolves to accept and an OpenDKIM hang to tempfail, +# but within seconds rather than minutes. Values are well above any healthy +# response time (signing/DB insert are sub-second), so they never fire in normal +# operation. +postconf -e \ + "milter_connect_timeout=${MILTER_CONNECT_TIMEOUT:-15s}" \ + "milter_command_timeout=${MILTER_COMMAND_TIMEOUT:-15s}" \ + "milter_content_timeout=${MILTER_CONTENT_TIMEOUT:-30s}" + # --- master.cf: inbound submission services ---------------------------------- # smtps (465, implicit/wrapper TLS) — the primary, always-on submission service # (spec 5 p.1). chroot=n so smtpd can read the sasldb2 and sender map under /data diff --git a/cmd/panel/httpserver.go b/cmd/panel/httpserver.go index a2a6764..2de8993 100644 --- a/cmd/panel/httpserver.go +++ b/cmd/panel/httpserver.go @@ -14,16 +14,10 @@ import ( "codeberg.org/mix/selfpost/internal/web" ) -// serveHTTP opens the panel database and runs the control-panel HTTP server -// until ctx is cancelled. From Phase 2 this serves the real setup, login and -// authenticated panel surface (spec 7.6). -func serveHTTP(ctx context.Context, cfg config) error { - st, err := store.Open(cfg.dbPath) - if err != nil { - return err - } - defer st.Close() - +// serveHTTP runs the control-panel HTTP server until ctx is cancelled, using +// the database handle shared by all roles. From Phase 2 this serves the real +// setup, login and authenticated panel surface (spec 7.6). +func serveHTTP(ctx context.Context, cfg config, st *store.Store) error { // Applications own the SASL accounts and the Postfix sender map; the domain // service delegates to them when a domain (and its applications) is deleted. pf := postfix.New(cfg.postfixDir) diff --git a/cmd/panel/journalstub.go b/cmd/panel/journal.go similarity index 53% rename from cmd/panel/journalstub.go rename to cmd/panel/journal.go index 724c7c9..e15dbdc 100644 --- a/cmd/panel/journalstub.go +++ b/cmd/panel/journal.go @@ -7,13 +7,17 @@ import ( "net" "os" "path/filepath" + + "codeberg.org/mix/selfpost/internal/milter" + "codeberg.org/mix/selfpost/internal/store" ) -// serveJournalStub opens the journal-milter Unix socket so the Postfix start -// wrapper's readiness probe (test -S) succeeds and the cold-start ordering -// (spec 4) can be exercised end to end. The real milter protocol handler is -// implemented in Phase 6; here connections are simply accepted and closed. -func serveJournalStub(ctx context.Context, socketPath string) error { +// serveJournal opens the journal-milter Unix socket and runs the real milter +// (spec 7.3), recording accepted messages into the send log. Socket lifecycle +// (creation, stale cleanup, group permissions) lives here; the protocol handler +// lives in internal/milter. +func serveJournal(ctx context.Context, cfg config, st *store.Store) error { + socketPath := cfg.journalSocket if err := os.MkdirAll(filepath.Dir(socketPath), 0o755); err != nil { return err } @@ -32,24 +36,10 @@ func serveJournalStub(ctx context.Context, socketPath string) error { // dir entrypoint.sh prepares; make it group read/write so postfix can reach // it (connecting to a Unix socket needs write permission on the node). if err := os.Chmod(socketPath, 0o660); err != nil { + ln.Close() return err } - // Closing the listener unblocks Accept and unlinks the socket file. - go func() { - <-ctx.Done() - _ = ln.Close() - }() - - log.Printf("journal-milter stub listening on %s", socketPath) - for { - conn, err := ln.Accept() - if err != nil { - if ctx.Err() != nil { - return nil // expected during shutdown - } - return err - } - _ = conn.Close() - } + log.Printf("journal-milter listening on %s", socketPath) + return milter.Serve(ctx, ln, st) } diff --git a/cmd/panel/logtailstub.go b/cmd/panel/logtailstub.go deleted file mode 100644 index 480e604..0000000 --- a/cmd/panel/logtailstub.go +++ /dev/null @@ -1,16 +0,0 @@ -package main - -import ( - "context" - "log" -) - -// tailMailLog is the Phase 1 placeholder for the log-tailer role. In Phase 6 it -// will follow mail.log and reconcile send-log delivery statuses by queue-id; -// for now it just idles until shutdown so the role is present in the process -// tree and its wiring is exercised. -func tailMailLog(ctx context.Context, path string) error { - log.Printf("log-tailer stub active (will follow %s in Phase 6)", path) - <-ctx.Done() - return nil -} diff --git a/cmd/panel/main.go b/cmd/panel/main.go index 36166f6..8a34170 100644 --- a/cmd/panel/main.go +++ b/cmd/panel/main.go @@ -15,10 +15,13 @@ import ( "os" "os/signal" "path/filepath" + "strconv" "sync" "syscall" "codeberg.org/mix/selfpost/internal/buildinfo" + "codeberg.org/mix/selfpost/internal/logtail" + "codeberg.org/mix/selfpost/internal/store" ) func main() { @@ -44,6 +47,7 @@ type config struct { httpAddr string journalSocket string mailLog string + retentionDays int dataDir string dbPath string @@ -65,6 +69,9 @@ func loadConfig() config { httpAddr: envDefault("PANEL_HTTP_ADDR", ":8080"), journalSocket: envDefault("JOURNAL_MILTER_SOCKET", "/run/selfpost/journal.sock"), mailLog: envDefault("MAIL_LOG", "/var/log/mail.log"), + // Send-log retention window (spec 7.3). Non-positive/invalid falls back + // to the 90-day default inside the log-tailer. + retentionDays: envInt("SEND_LOG_RETENTION_DAYS", 90), dataDir: dataDir, dbPath: envDefault("SELFPOST_DB_PATH", filepath.Join(dataDir, "selfpost.db")), @@ -109,6 +116,18 @@ func envDefault(key, def string) string { return def } +// envInt reads an integer environment variable, returning def if it is unset or +// not a valid integer. +func envInt(key string, def int) int { + if v := os.Getenv(key); v != "" { + if n, err := strconv.Atoi(v); err == nil { + return n + } + log.Printf("ignoring invalid %s=%q, using %d", key, v, def) + } + return def +} + // run starts the panel's three roles and blocks until a shutdown signal or the // first fatal error from any role. A signal triggers a clean stop of all roles; // a role error cancels the others and is returned so the process exits non-zero @@ -121,6 +140,15 @@ func run() error { log.Printf("starting selfpost panel %s", buildinfo.Version) + // One database handle shared by every role. The store serialises writes + // (MaxOpenConns(1)), so the HTTP panel, the journal-milter and the tailer + // can all use it without stepping on each other under WAL. + st, err := store.Open(cfg.dbPath) + if err != nil { + return err + } + defer st.Close() + var wg sync.WaitGroup errc := make(chan error, 3) @@ -128,9 +156,9 @@ func run() error { name string fn func(context.Context) error }{ - {"http", func(ctx context.Context) error { return serveHTTP(ctx, cfg) }}, - {"journal-milter", func(ctx context.Context) error { return serveJournalStub(ctx, cfg.journalSocket) }}, - {"log-tailer", func(ctx context.Context) error { return tailMailLog(ctx, cfg.mailLog) }}, + {"http", func(ctx context.Context) error { return serveHTTP(ctx, cfg, st) }}, + {"journal-milter", func(ctx context.Context) error { return serveJournal(ctx, cfg, st) }}, + {"log-tailer", func(ctx context.Context) error { return logtail.Run(ctx, cfg.mailLog, st, cfg.retentionDays) }}, } for _, r := range roles { diff --git a/go.mod b/go.mod index ed5d954..5d4e1e8 100644 --- a/go.mod +++ b/go.mod @@ -3,12 +3,14 @@ module codeberg.org/mix/selfpost go 1.26 require ( + github.com/emersion/go-milter v0.4.1 golang.org/x/crypto v0.54.0 modernc.org/sqlite v1.53.0 ) require ( github.com/dustin/go-humanize v1.0.1 // indirect + github.com/emersion/go-message v0.18.1 // indirect github.com/google/uuid v1.6.0 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/ncruces/go-strftime v1.0.0 // indirect diff --git a/go.sum b/go.sum index 5456856..f73d4ef 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,9 @@ github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= +github.com/emersion/go-message v0.18.1 h1:tfTxIoXFSFRwWaZsgnqS1DSZuGpYGzSmCZD8SK3QA2E= +github.com/emersion/go-message v0.18.1/go.mod h1:XpJyL70LwRvq2a8rVbHXikPgKj8+aI0kGdHlg16ibYA= +github.com/emersion/go-milter v0.4.1 h1:gLs9QD0zEHF8omgEw8M+aGz6iwBNpWLAcwgSur0ra4M= +github.com/emersion/go-milter v0.4.1/go.mod h1:erCQVl0mH4SX9jEvwe+wyndit0rQtmvMLH86V6NGtkI= github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs= github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= @@ -12,17 +16,48 @@ github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOF github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw= golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk= +golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= +golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/mod v0.36.0 h1:JJjpVx6myfUsUdAzZuOSTTmRE0PfZeNWzzvKrP7amb4= golang.org/x/mod v0.36.0/go.mod h1:moc6ELqsWcOw5Ef3xVprK5ul/MvtVvkIXLziUOICjUQ= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= +golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= +golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= +golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= +golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= +golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU= golang.org/x/tools v0.45.0 h1:18qN3FAooORvApf5XjCXgsuayZOEtXf6JK18I3+ONa8= golang.org/x/tools v0.45.0/go.mod h1:LuUGqqaXcXMEFEruIVJVm5mgDD8vww/z/SR1gQ4uE/0= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= modernc.org/cc/v4 v4.28.4 h1:Hd/4Es+MBj+/7hSdZaisNyu6bv3V0Dp2MdllyfqaH+c= modernc.org/cc/v4 v4.28.4/go.mod h1:OnovgIhbbMXMu1aISnJ0wvVD1KnW+cAUJkIrAWh+kVI= modernc.org/ccgo/v4 v4.34.4 h1:OVnSOWQjVKOYkFxoHYB+qQmSHK5gqMqARM+K9DpR/Ws= diff --git a/internal/logtail/logtail.go b/internal/logtail/logtail.go new file mode 100644 index 0000000..b0cc9e7 --- /dev/null +++ b/internal/logtail/logtail.go @@ -0,0 +1,209 @@ +// Package logtail follows Postfix's mail.log and reconciles the send-log +// delivery statuses the journal-milter could not know at receive time (spec +// 7.3). A milter row starts life as "queued"; Postfix only decides sent / +// deferred / bounced later, per recipient, and reports it in mail.log. This +// package parses those lines by queue-id + recipient and advances the matching +// rows, and prunes rows past the retention window. +package logtail + +import ( + "bufio" + "context" + "io" + "log" + "os" + "regexp" + "strings" + "time" + + "codeberg.org/mix/selfpost/internal/store" +) + +// StatusStore is the slice of the store the log-tailer needs. *store.Store +// satisfies it. +type StatusStore interface { + UpdateStatus(queueID, recipient, status string) (int64, error) + DeleteSendLogBefore(cutoff time.Time) (int64, error) +} + +// pollInterval is how often the tail loop checks for new bytes / rotation. It +// is a var so tests can shorten it. +var pollInterval = time.Second + +const ( + // retentionInterval is how often the retention sweep runs (also once at + // startup). The window itself is configurable; the cadence need not be. + retentionInterval = 6 * time.Hour + // defaultRetentionDays applies when the configured value is unset/invalid + // (spec 7.3). + defaultRetentionDays = 90 +) + +// deliveryRe matches a Postfix delivery line and captures queue-id, recipient +// and status, e.g. +// +// postfix/smtp[26]: 41E862C00D9E: to=, relay=…, dsn=2.0.0, status=sent (250 OK) +// +// The ": to=, …, status=" shape is specific to the +// delivery agents; qmgr/smtpd/cleanup lines do not match. +var deliveryRe = regexp.MustCompile(`\b([0-9A-Za-z]+): to=<([^>]*)>,.*\bstatus=(\w+)`) + +// parseDelivery extracts (queue-id, recipient, status) from a mail.log line. +// ok is false for lines that are not recognised delivery results. +func parseDelivery(line string) (queueID, recipient, status string, ok bool) { + m := deliveryRe.FindStringSubmatch(line) + if m == nil { + return "", "", "", false + } + switch m[3] { + case "sent": + status = store.StatusSent + case "deferred": + status = store.StatusDeferred + case "bounced": + status = store.StatusBounced + case "expired": + // Postfix gave up after the queue lifetime; a final failure for us. + status = store.StatusBounced + default: + return "", "", "", false + } + return m[1], m[2], status, true +} + +// Run follows path and updates send-log statuses until ctx is cancelled, while +// a background sweep prunes rows older than retentionDays. It returns nil on a +// clean shutdown. +func Run(ctx context.Context, path string, st StatusStore, retentionDays int) error { + go retentionLoop(ctx, st, retentionDays) + + return follow(ctx, path, func(line string) { + queueID, recipient, status, ok := parseDelivery(line) + if !ok { + return + } + if _, err := st.UpdateStatus(queueID, recipient, status); err != nil { + log.Printf("log-tailer: update %s/%s -> %s: %v", queueID, recipient, status, err) + } + }) +} + +// retentionLoop prunes expired send-log rows immediately and then periodically. +func retentionLoop(ctx context.Context, st StatusStore, retentionDays int) { + if retentionDays <= 0 { + retentionDays = defaultRetentionDays + } + prune := func() { + cutoff := time.Now().UTC().AddDate(0, 0, -retentionDays) + n, err := st.DeleteSendLogBefore(cutoff) + if err != nil { + log.Printf("log-tailer: retention prune: %v", err) + return + } + if n > 0 { + log.Printf("log-tailer: pruned %d send-log rows older than %d days", n, retentionDays) + } + } + + prune() + t := time.NewTicker(retentionInterval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + prune() + } + } +} + +// follow tails path line by line, calling handle for each complete line, until +// ctx is cancelled. It starts at end-of-file (so a restart does not reprocess +// history) and reopens the file when it is rotated (inode change from +// logrotate's create, or truncation from copytruncate) so nothing is missed. +func follow(ctx context.Context, path string, handle func(string)) error { + var ( + f *os.File + r *bufio.Reader + info os.FileInfo + pending string + ) + openAt := func(offset int64, whence int) error { + nf, err := os.Open(path) + if err != nil { + return err + } + if _, err := nf.Seek(offset, whence); err != nil { + nf.Close() + return err + } + ni, err := nf.Stat() + if err != nil { + nf.Close() + return err + } + if f != nil { + f.Close() + } + f, r, info, pending = nf, bufio.NewReader(nf), ni, "" + return nil + } + + // The container may start before Postfix has created mail.log; wait for it. + for { + if err := openAt(0, io.SeekEnd); err == nil { + break + } + select { + case <-ctx.Done(): + return nil + case <-time.After(pollInterval): + } + } + defer func() { + if f != nil { + f.Close() + } + }() + + drain := func() { + for { + line, err := r.ReadString('\n') + if err == io.EOF { + pending += line // hold the partial line until it completes + return + } + if err != nil { + log.Printf("log-tailer: read %s: %v", path, err) + return + } + full := pending + line + pending = "" + handle(strings.TrimRight(full, "\r\n")) + } + } + + ticker := time.NewTicker(pollInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + drain() + ni, err := os.Stat(path) + if err != nil { + continue // file briefly gone mid-rotation; try again next tick + } + pos, _ := f.Seek(0, io.SeekCurrent) + if !os.SameFile(info, ni) || ni.Size() < pos { + // Rotated away or truncated: reopen from the start of the new + // file. Any tail of the old file was already drained above. + if err := openAt(0, io.SeekStart); err != nil { + log.Printf("log-tailer: reopen %s: %v", path, err) + } + } + } + } +} diff --git a/internal/logtail/logtail_test.go b/internal/logtail/logtail_test.go new file mode 100644 index 0000000..6abaf1f --- /dev/null +++ b/internal/logtail/logtail_test.go @@ -0,0 +1,176 @@ +package logtail + +import ( + "context" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "codeberg.org/mix/selfpost/internal/store" +) + +func TestParseDelivery(t *testing.T) { + cases := []struct { + name string + line string + wantOK bool + queueID, recipient, status string + }{ + { + name: "sent", + line: "2026-07-11T11:55:34 host postfix/smtp[26]: 41E862C00D9E: to=, relay=mx.example.net[203.0.113.9]:25, delay=0.5, dsn=2.0.0, status=sent (250 OK)", + wantOK: true, + queueID: "41E862C00D9E", + recipient: "a@example.net", + status: store.StatusSent, + }, + { + name: "deferred", + line: "host postfix/smtp[26]: 5900C2C00D9E: to=, relay=none, delay=30, dsn=4.4.1, status=deferred (connect timed out)", + wantOK: true, + queueID: "5900C2C00D9E", + recipient: "y@example.net", + status: store.StatusDeferred, + }, + { + name: "bounced", + line: "host postfix/smtp[26]: ABC: to=, relay=…, dsn=5.1.1, status=bounced (user unknown)", + wantOK: true, + queueID: "ABC", + recipient: "no@example.net", + status: store.StatusBounced, + }, + { + name: "expired maps to bounced", + line: "host postfix/smtp[26]: DEF: to=, relay=none, status=expired (delivery temporarily suspended)", + wantOK: true, + queueID: "DEF", + recipient: "slow@example.net", + status: store.StatusBounced, + }, + { + name: "qmgr from-line ignored", + line: "host postfix/qmgr[10]: 41E862C00D9E: from=, size=500, nrcpt=1 (queue active)", + wantOK: false, + }, + { + name: "smtpd client-line ignored", + line: "host postfix/smtpd[10]: 41E862C00D9E: client=unknown[203.0.113.7]", + wantOK: false, + }, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + q, r, s, ok := parseDelivery(c.line) + if ok != c.wantOK { + t.Fatalf("ok = %v, want %v", ok, c.wantOK) + } + if !ok { + return + } + if q != c.queueID || r != c.recipient || s != c.status { + t.Fatalf("got (%q,%q,%q), want (%q,%q,%q)", q, r, s, c.queueID, c.recipient, c.status) + } + }) + } +} + +// captureStore records UpdateStatus calls for the follow integration test. +type captureStore struct { + mu sync.Mutex + calls []string +} + +func (c *captureStore) UpdateStatus(queueID, recipient, status string) (int64, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.calls = append(c.calls, queueID+"|"+recipient+"|"+status) + return 1, nil +} + +func (c *captureStore) DeleteSendLogBefore(time.Time) (int64, error) { return 0, nil } + +func (c *captureStore) snapshot() []string { + c.mu.Lock() + defer c.mu.Unlock() + return append([]string(nil), c.calls...) +} + +// TestFollowTailsAndRotates writes delivery lines to a log file, then rotates +// it (rename + fresh create, as logrotate does) and writes more, asserting the +// tailer picks up lines from both the original and rotated file. +func TestFollowTailsAndRotates(t *testing.T) { + old := pollInterval + pollInterval = 10 * time.Millisecond + t.Cleanup(func() { pollInterval = old }) + + dir := t.TempDir() + path := filepath.Join(dir, "mail.log") + if err := os.WriteFile(path, []byte("preexisting line, ignored on start\n"), 0o644); err != nil { + t.Fatalf("seed log: %v", err) + } + + cs := &captureStore{} + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- Run(ctx, path, cs, 90) }() + + // Give follow() time to open at EOF (it seeks to end immediately on start, + // so the seed line above is ignored), then append a delivery line. + time.Sleep(50 * time.Millisecond) + appendLine(t, path, "host postfix/smtp[1]: Q1: to=, dsn=2.0.0, status=sent (ok)") + waitFor(t, func() bool { return contains(cs.snapshot(), "Q1|a@example.net|sent") }) + + // Rotate: move the current file aside and create a fresh one (logrotate + // "create"), then append to the new file. + if err := os.Rename(path, path+".1"); err != nil { + t.Fatalf("rotate: %v", err) + } + if err := os.WriteFile(path, nil, 0o644); err != nil { + t.Fatalf("recreate: %v", err) + } + appendLine(t, path, "host postfix/smtp[1]: Q2: to=, dsn=5.1.1, status=bounced (nope)") + waitFor(t, func() bool { return contains(cs.snapshot(), "Q2|b@example.net|bounced") }) + + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Run did not return after cancel") + } +} + +func appendLine(t *testing.T, path, line string) { + t.Helper() + f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o644) + if err != nil { + t.Fatalf("open for append: %v", err) + } + defer f.Close() + if _, err := f.WriteString(line + "\n"); err != nil { + t.Fatalf("append: %v", err) + } +} + +func contains(ss []string, want string) bool { + for _, s := range ss { + if s == want { + return true + } + } + return false +} + +func waitFor(t *testing.T, cond func() bool) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("condition not met within timeout") +} diff --git a/internal/milter/milter.go b/internal/milter/milter.go new file mode 100644 index 0000000..728c9a2 --- /dev/null +++ b/internal/milter/milter.go @@ -0,0 +1,174 @@ +// Package milter implements the SelfPost journal-milter: a lightweight milter +// (spec 7.3) attached to Postfix's smtpd_milters alongside OpenDKIM. On the +// receive path it reads the SASL login, From, recipients and Subject of each +// accepted message and records one send-log row per (queue-id, recipient), +// giving the panel a structured, filterable history that raw mail.log cannot. +// +// It is monitoring only: it never rejects, and every callback returns Continue +// or Accept so a failure of this milter can never block the relay. Postfix is +// configured with default_action=accept for this milter's socket, so even a +// crash or hang fails open (spec 7.3). +package milter + +import ( + "context" + "log" + "net" + "net/textproto" + "strings" + + "github.com/emersion/go-milter" + + "codeberg.org/mix/selfpost/internal/store" +) + +// Recorder persists queued send-log entries. *store.Store satisfies it; tests +// substitute a fake. +type Recorder interface { + InsertQueued(e store.SendLogEntry) error +} + +// session accumulates the fields of one message as the milter callbacks fire. +// Milter macros arrive per-stage and do not accumulate, so each value is +// captured at the stage that carries it (spec 7.3 / Phase 0 spike): SASL login +// and From at MAIL, each recipient at RCPT, Subject in the headers, and the +// queue-id at end-of-message. go-milter creates one session per connection; a +// connection may carry several messages, so per-message fields are reset at +// MailFrom (the start of every transaction). +type session struct { + milter.NoOpMilter + rec Recorder + + clientIP string // captured once per connection + + login string + from string + rcpts []string + subject string +} + +// Connect captures the client IP, which comes from the addr parameter rather +// than a macro (the {client_addr} macro was empty in the spike). It is the +// rate-limit key for Phase 8; here it is recorded for completeness. +func (s *session) Connect(host, family string, port uint16, addr net.IP, m *milter.Modifier) (milter.Response, error) { + if addr != nil { + s.clientIP = addr.String() + } + return milter.RespContinue, nil +} + +// MailFrom starts a new message: reset per-message state, then capture the +// envelope sender and the SASL login ({auth_authen}, carried by the MAIL-stage +// macros). +func (s *session) MailFrom(from string, m *milter.Modifier) (milter.Response, error) { + s.from = cleanAddress(from) + s.login = macro(m, "auth_authen") + s.rcpts = nil + s.subject = "" + return milter.RespContinue, nil +} + +// RcptTo records each recipient. Postfix calls this once per recipient, which +// is what lets the journal keep a separate row per (queue-id, recipient). +func (s *session) RcptTo(rcpt string, m *milter.Modifier) (milter.Response, error) { + s.rcpts = append(s.rcpts, cleanAddress(rcpt)) + return milter.RespContinue, nil +} + +// Header captures the Subject. Only the first Subject header is kept. +func (s *session) Header(name, value string, m *milter.Modifier) (milter.Response, error) { + if s.subject == "" && textproto.CanonicalMIMEHeaderKey(name) == "Subject" { + s.subject = value + } + return milter.RespContinue, nil +} + +// Body fires at end-of-message, when the queue-id macro {i} is set and the +// message is about to be committed to the queue. This is where the "queued" +// rows are written. We accept (this milter is done) without ever rejecting. +func (s *session) Body(m *milter.Modifier) (milter.Response, error) { + s.record(macro(m, "i")) + return milter.RespAccept, nil +} + +// macro reads a milter macro, tolerating Postfix's convention of wrapping +// multi-character macro names in curly braces (e.g. {auth_authen}) while +// single-character names (e.g. i) arrive bare. go-milter stores whatever name +// Postfix sends verbatim, so a lookup must try both forms — this is exactly the +// distinction the SASL-less Phase 0 spike could not observe. +func macro(m *milter.Modifier, name string) string { + if v, ok := m.Macros[name]; ok { + return v + } + return m.Macros["{"+name+"}"] +} + +// record writes one send-log row per recipient. Failures are logged, never +// propagated: journalling must not affect mail acceptance (spec 7.3). +func (s *session) record(queueID string) { + domain := domainOf(s.from) + rcpts := s.rcpts + if len(rcpts) == 0 { + // No recipient seen (unusual) — still record the message so it is + // visible in the log rather than silently dropped. + rcpts = []string{""} + } + for _, to := range rcpts { + err := s.rec.InsertQueued(store.SendLogEntry{ + QueueID: queueID, + Domain: domain, + AppLogin: s.login, + From: s.from, + To: to, + Subject: s.subject, + }) + if err != nil { + log.Printf("journal-milter: record %s -> %s: %v", queueID, to, err) + } + } +} + +// cleanAddress strips the angle brackets and any ESMTP parameters Postfix may +// pass with an address, leaving the bare mailbox. +func cleanAddress(a string) string { + a = strings.TrimSpace(a) + if i := strings.IndexByte(a, ' '); i >= 0 { // drop "addr SIZE=… BODY=…" params + a = a[:i] + } + a = strings.TrimPrefix(a, "<") + a = strings.TrimSuffix(a, ">") + return a +} + +// domainOf returns the lower-cased domain of an email address, or "" if there +// is no domain part. Sender binding (Phase 4) guarantees the From domain equals +// the application's domain, so this is the sending domain (spec 7.3). +func domainOf(addr string) string { + if i := strings.LastIndexByte(addr, '@'); i >= 0 { + return strings.ToLower(addr[i+1:]) + } + return "" +} + +// Serve runs the journal-milter on ln until ctx is cancelled. Each connection +// gets a fresh session bound to rec. It returns nil on a clean shutdown. +func Serve(ctx context.Context, ln net.Listener, rec Recorder) error { + srv := &milter.Server{ + NewMilter: func() milter.Milter { return &session{rec: rec} }, + Actions: 0, // read-only: we make no message modifications + Protocol: milter.OptNoBody, // the journal needs headers/EOM, not the body + } + + go func() { + <-ctx.Done() + _ = srv.Close() + }() + + if err := srv.Serve(ln); err != nil { + if ctx.Err() != nil { + return nil // expected: Close() during shutdown unblocks Serve + } + return err + } + return nil +} diff --git a/internal/milter/milter_test.go b/internal/milter/milter_test.go new file mode 100644 index 0000000..d7086e1 --- /dev/null +++ b/internal/milter/milter_test.go @@ -0,0 +1,156 @@ +package milter + +import ( + "errors" + "net" + "testing" + + "github.com/emersion/go-milter" + + "codeberg.org/mix/selfpost/internal/store" +) + +// fakeRecorder captures inserts and can be made to fail, to prove the milter +// swallows recorder errors and still accepts the message. +type fakeRecorder struct { + entries []store.SendLogEntry + fail bool +} + +func (f *fakeRecorder) InsertQueued(e store.SendLogEntry) error { + if f.fail { + return errors.New("boom") + } + f.entries = append(f.entries, e) + return nil +} + +func mods(kv map[string]string) *milter.Modifier { + return &milter.Modifier{Macros: kv} +} + +// drive replays a typical message through one session and returns the recorder. +func drive(t *testing.T, rec Recorder) *session { + t.Helper() + s := &session{rec: rec} + if _, err := s.Connect("localhost", "tcp4", 0, net.ParseIP("203.0.113.7"), mods(nil)); err != nil { + t.Fatalf("Connect: %v", err) + } + if _, err := s.MailFrom("noreply@example.com", mods(map[string]string{"auth_authen": "app1"})); err != nil { + t.Fatalf("MailFrom: %v", err) + } + if _, err := s.RcptTo("", mods(nil)); err != nil { + t.Fatalf("RcptTo: %v", err) + } + if _, err := s.RcptTo("b@example.net", mods(nil)); err != nil { + t.Fatalf("RcptTo: %v", err) + } + if _, err := s.Header("Subject", "Hello there", mods(nil)); err != nil { + t.Fatalf("Header: %v", err) + } + if _, err := s.Body(mods(map[string]string{"i": "ABC123"})); err != nil { + t.Fatalf("Body: %v", err) + } + return s +} + +func TestSessionRecordsRowPerRecipient(t *testing.T) { + rec := &fakeRecorder{} + s := drive(t, rec) + + if s.clientIP != "203.0.113.7" { + t.Fatalf("clientIP = %q, want 203.0.113.7", s.clientIP) + } + if len(rec.entries) != 2 { + t.Fatalf("want 2 entries, got %d: %+v", len(rec.entries), rec.entries) + } + got := rec.entries[0] + want := store.SendLogEntry{ + QueueID: "ABC123", + Domain: "example.com", + AppLogin: "app1", + From: "noreply@example.com", + To: "a@example.net", // angle brackets stripped + Subject: "Hello there", + } + if got != want { + t.Fatalf("entry[0]\n got %+v\nwant %+v", got, want) + } + if rec.entries[1].To != "b@example.net" { + t.Fatalf("entry[1].To = %q", rec.entries[1].To) + } +} + +func TestBodyAcceptsEvenWhenRecorderFails(t *testing.T) { + rec := &fakeRecorder{fail: true} + s := &session{rec: rec} + _, _ = s.MailFrom("x@example.com", mods(map[string]string{"auth_authen": "app1"})) + _, _ = s.RcptTo("y@example.net", mods(nil)) + resp, err := s.Body(mods(map[string]string{"i": "Q9"})) + if err != nil { + t.Fatalf("Body returned error, must fail open: %v", err) + } + if resp != milter.RespAccept { + t.Fatalf("Body response = %v, want Accept", resp) + } +} + +// A single connection may carry several messages; the second must not inherit +// the first's recipients or subject. +func TestSessionResetsBetweenMessages(t *testing.T) { + rec := &fakeRecorder{} + s := &session{rec: rec} + + _, _ = s.MailFrom("a@example.com", mods(map[string]string{"auth_authen": "app1"})) + _, _ = s.RcptTo("one@example.net", mods(nil)) + _, _ = s.Header("Subject", "first", mods(nil)) + _, _ = s.Body(mods(map[string]string{"i": "Q1"})) + + _, _ = s.MailFrom("b@example.com", mods(map[string]string{"auth_authen": "app2"})) + _, _ = s.RcptTo("two@example.net", mods(nil)) + _, _ = s.Body(mods(map[string]string{"i": "Q2"})) + + if len(rec.entries) != 2 { + t.Fatalf("want 2 entries, got %d", len(rec.entries)) + } + second := rec.entries[1] + if second.QueueID != "Q2" || second.To != "two@example.net" || second.Subject != "" || second.AppLogin != "app2" { + t.Fatalf("second message leaked state: %+v", second) + } +} + +// Postfix sends multi-character macro names wrapped in braces ({auth_authen}, +// {i} for some versions), so the milter must resolve those too — this is the +// case the SASL-less spike missed and that produced empty app_login at first. +func TestBracedMacros(t *testing.T) { + rec := &fakeRecorder{} + s := &session{rec: rec} + _, _ = s.MailFrom("app@example.com", mods(map[string]string{"{auth_authen}": "app1"})) + _, _ = s.RcptTo("to@example.net", mods(nil)) + _, _ = s.Body(mods(map[string]string{"{i}": "QBRACE"})) + + if len(rec.entries) != 1 { + t.Fatalf("want 1 entry, got %d", len(rec.entries)) + } + e := rec.entries[0] + if e.AppLogin != "app1" { + t.Fatalf("AppLogin = %q, want app1 (braced {auth_authen} not resolved)", e.AppLogin) + } + if e.QueueID != "QBRACE" { + t.Fatalf("QueueID = %q, want QBRACE (braced {i} not resolved)", e.QueueID) + } +} + +func TestDomainOf(t *testing.T) { + cases := map[string]string{ + "user@Example.COM": "example.com", + "no-domain": "", + "": "", + "a@b@c.com": "c.com", + } + for in, want := range cases { + if got := domainOf(in); got != want { + t.Fatalf("domainOf(%q) = %q, want %q", in, got, want) + } + } +} diff --git a/internal/store/sendlog.go b/internal/store/sendlog.go new file mode 100644 index 0000000..ff289a3 --- /dev/null +++ b/internal/store/sendlog.go @@ -0,0 +1,82 @@ +package store + +import ( + "fmt" + "time" +) + +// Send-log status values (spec 7.3). "queued" is written by the journal-milter +// when a message is accepted; the log-tailer advances it to one of the final +// states as Postfix reports delivery per recipient. +const ( + StatusQueued = "queued" + StatusSent = "sent" + StatusDeferred = "deferred" + StatusBounced = "bounced" +) + +// SendLogEntry is a single queued send-log row. The journal-milter creates one +// per (queue-id, recipient) pair at end-of-message (spec 7.3.3); every field +// except the status/timestamps comes from the accepted message. +type SendLogEntry struct { + QueueID string + Domain string + AppLogin string + From string + To string + Subject string +} + +// InsertQueued records an accepted message in the send log with status +// "queued". It is called from the journal-milter hot path, so it returns any +// error for the caller to log rather than deciding policy here; the milter must +// stay fail-open regardless (spec 7.3). +func (s *Store) InsertQueued(e SendLogEntry) error { + now := time.Now().UTC().Format(time.RFC3339) + _, err := s.db.Exec( + `INSERT INTO send_log + (queue_id, domain, app_login, from_addr, to_addr, subject, status, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + e.QueueID, e.Domain, e.AppLogin, e.From, e.To, e.Subject, StatusQueued, now, now, + ) + if err != nil { + return fmt.Errorf("insert send_log: %w", err) + } + return nil +} + +// UpdateStatus advances the delivery status of the send-log rows matching a +// (queue-id, recipient) pair, which the log-tailer parses out of mail.log. +// Recipient matching is case-insensitive because Postfix may normalise address +// case between the milter (envelope) and the delivery log. It returns the +// number of rows updated so the caller can tell whether the line matched a +// journal entry. +func (s *Store) UpdateStatus(queueID, recipient, status string) (int64, error) { + now := time.Now().UTC().Format(time.RFC3339) + res, err := s.db.Exec( + `UPDATE send_log SET status = ?, updated_at = ? + WHERE queue_id = ? AND to_addr = ? COLLATE NOCASE`, + status, now, queueID, recipient, + ) + if err != nil { + return 0, fmt.Errorf("update send_log status: %w", err) + } + n, _ := res.RowsAffected() + return n, nil +} + +// DeleteSendLogBefore removes send-log rows created before cutoff, implementing +// the configurable retention window (spec 7.3, SEND_LOG_RETENTION_DAYS). It +// returns the number of rows pruned. created_at is stored as RFC3339 UTC, so a +// lexical comparison against the same format is chronologically correct. +func (s *Store) DeleteSendLogBefore(cutoff time.Time) (int64, error) { + res, err := s.db.Exec( + `DELETE FROM send_log WHERE created_at < ?`, + cutoff.UTC().Format(time.RFC3339), + ) + if err != nil { + return 0, fmt.Errorf("prune send_log: %w", err) + } + n, _ := res.RowsAffected() + return n, nil +} diff --git a/internal/store/sendlog_test.go b/internal/store/sendlog_test.go new file mode 100644 index 0000000..a042781 --- /dev/null +++ b/internal/store/sendlog_test.go @@ -0,0 +1,140 @@ +package store + +import ( + "testing" + "time" +) + +// readSendLog returns every send_log row ordered by id. Phase 6 has no read +// query yet (the monitoring UI is Phase 7), so tests read the table directly. +type sendLogRow struct { + QueueID string + Domain string + AppLogin string + From string + To string + Subject string + Status string +} + +func readSendLog(t *testing.T, s *Store) []sendLogRow { + t.Helper() + rows, err := s.db.Query( + `SELECT queue_id, domain, app_login, from_addr, to_addr, subject, status + FROM send_log ORDER BY id`) + if err != nil { + t.Fatalf("query send_log: %v", err) + } + defer rows.Close() + var out []sendLogRow + for rows.Next() { + var r sendLogRow + if err := rows.Scan(&r.QueueID, &r.Domain, &r.AppLogin, &r.From, &r.To, &r.Subject, &r.Status); err != nil { + t.Fatalf("scan: %v", err) + } + out = append(out, r) + } + return out +} + +func TestInsertQueuedAndUpdateStatus(t *testing.T) { + st := openTestStore(t) + + // Two recipients on the same queue-id → two independent rows (spec 7.3.3). + for _, to := range []string{"a@example.net", "b@example.net"} { + if err := st.InsertQueued(SendLogEntry{ + QueueID: "ABC123", + Domain: "example.com", + AppLogin: "app1", + From: "noreply@example.com", + To: to, + Subject: "Hello", + }); err != nil { + t.Fatalf("InsertQueued: %v", err) + } + } + + rows := readSendLog(t, st) + if len(rows) != 2 { + t.Fatalf("want 2 rows, got %d: %+v", len(rows), rows) + } + for _, r := range rows { + if r.Status != StatusQueued { + t.Fatalf("new row should be queued, got %q", r.Status) + } + } + + // One recipient goes to sent; the other stays queued. + n, err := st.UpdateStatus("ABC123", "a@example.net", StatusSent) + if err != nil { + t.Fatalf("UpdateStatus: %v", err) + } + if n != 1 { + t.Fatalf("want 1 row updated, got %d", n) + } + + rows = readSendLog(t, st) + if rows[0].Status != StatusSent || rows[1].Status != StatusQueued { + t.Fatalf("unexpected statuses: %+v", rows) + } +} + +func TestUpdateStatusRecipientCaseInsensitive(t *testing.T) { + st := openTestStore(t) + if err := st.InsertQueued(SendLogEntry{QueueID: "Q1", To: "User@Example.NET"}); err != nil { + t.Fatalf("InsertQueued: %v", err) + } + // mail.log may report a differently-cased recipient; matching must still hit. + n, err := st.UpdateStatus("Q1", "user@example.net", StatusBounced) + if err != nil { + t.Fatalf("UpdateStatus: %v", err) + } + if n != 1 { + t.Fatalf("case-insensitive match failed, updated %d rows", n) + } +} + +func TestUpdateStatusNoMatch(t *testing.T) { + st := openTestStore(t) + if err := st.InsertQueued(SendLogEntry{QueueID: "Q1", To: "a@example.net"}); err != nil { + t.Fatalf("InsertQueued: %v", err) + } + // A queue-id/recipient the milter never recorded must be a no-op, not an error. + n, err := st.UpdateStatus("Q1", "unknown@example.net", StatusSent) + if err != nil { + t.Fatalf("UpdateStatus: %v", err) + } + if n != 0 { + t.Fatalf("want 0 rows updated, got %d", n) + } +} + +func TestDeleteSendLogBefore(t *testing.T) { + st := openTestStore(t) + + // Insert one row, then backdate it beyond the retention window by rewriting + // created_at directly (InsertQueued always stamps "now"). + if err := st.InsertQueued(SendLogEntry{QueueID: "OLD", To: "a@example.net"}); err != nil { + t.Fatalf("InsertQueued: %v", err) + } + old := time.Now().UTC().AddDate(0, 0, -100).Format(time.RFC3339) + if _, err := st.db.Exec(`UPDATE send_log SET created_at = ? WHERE queue_id = 'OLD'`, old); err != nil { + t.Fatalf("backdate: %v", err) + } + if err := st.InsertQueued(SendLogEntry{QueueID: "NEW", To: "b@example.net"}); err != nil { + t.Fatalf("InsertQueued: %v", err) + } + + cutoff := time.Now().UTC().AddDate(0, 0, -90) + n, err := st.DeleteSendLogBefore(cutoff) + if err != nil { + t.Fatalf("DeleteSendLogBefore: %v", err) + } + if n != 1 { + t.Fatalf("want 1 row pruned, got %d", n) + } + rows := readSendLog(t, st) + if len(rows) != 1 || rows[0].QueueID != "NEW" { + t.Fatalf("retention kept wrong rows: %+v", rows) + } +}