From 573f54afd2d6101a54af05d550275f17f04beebc Mon Sep 17 00:00:00 2001 From: codegirl-007 Date: Sat, 29 Aug 2026 06:19:44 -0700 Subject: [PATCH] Send reply mail from the post event bus. --- cmd/server/main.go | 6 +- internal/mail/subscriber.go | 89 ++++++++++++++ internal/mail/subscriber_test.go | 200 +++++++++++++++++++++++++++++++ internal/web/posts.go | 59 --------- internal/web/posts_test.go | 7 +- internal/web/server.go | 5 - 6 files changed, 298 insertions(+), 68 deletions(-) create mode 100644 internal/mail/subscriber.go create mode 100644 internal/mail/subscriber_test.go diff --git a/cmd/server/main.go b/cmd/server/main.go index 7dc3658..90a3c2c 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -37,6 +37,7 @@ func main() { log.Fatalf("mail: %v", err) } bus := events.New() + mail.Subscribe(bus, store.NewPostgres(db), notifier) bot, err := discord.FromEnv(store.NewDiscordLinks(db), bus, store.NewPostgres(db), notifier) if err != nil { log.Fatalf("discord: %v", err) @@ -44,7 +45,7 @@ func main() { if bot != nil { defer bot.Close() } - handler := newHandler(db, sessions, uploader, notifier, bus) + handler := newHandler(db, sessions, uploader, bus) run(&http.Server{ Addr: listenAddr(), Handler: handler, @@ -68,13 +69,12 @@ func openDB() (*sql.DB, *store.SessionStore) { return db, sessions } -func newHandler(db *sql.DB, sessions *store.SessionStore, uploader blob.Uploader, notifier mail.Notifier, bus events.Publisher) http.Handler { +func newHandler(db *sql.DB, sessions *store.SessionStore, uploader blob.Uploader, bus events.Publisher) http.Handler { srv, err := web.New(store.NewPostgres(db), sessions.Store(), plumber.TemplateFS, plumber.StaticFS, web.Config{ AdminSetupSecret: strings.TrimSpace(os.Getenv("ADMIN_SETUP_SECRET")), SecureCookie: secureCookieFromEnv(), TrustedProxies: parseTrustedProxies(os.Getenv("TRUSTED_PROXY_CIDRS")), Blob: uploader, - Mail: notifier, Events: bus, BaseURL: strings.TrimRight(strings.TrimSpace(os.Getenv("APP_BASE_URL")), "/"), }) diff --git a/internal/mail/subscriber.go b/internal/mail/subscriber.go new file mode 100644 index 0000000..d38fdc0 --- /dev/null +++ b/internal/mail/subscriber.go @@ -0,0 +1,89 @@ +package mail + +import ( + "context" + "log" + "strings" + "time" + + "plumber/internal/events" + "plumber/internal/store" +) + +// Subscribe sends reply emails from PostCreated events. Nop or nil is a no-op. +func Subscribe(bus *events.Bus, st store.Store, n Notifier) { + if bus == nil || st == nil || n == nil { + return + } + if _, disabled := n.(Nop); disabled { + return + } + s := subscriber{store: st, mail: n} + bus.Subscribe(s.handle) +} + +type subscriber struct { + store store.Store + mail Notifier +} + +func (s subscriber) handle(_ context.Context, ev any) { + created, ok := ev.(events.PostCreated) + if !ok { + return + } + if strings.TrimSpace(created.ParentID) == "" { + return + } + go s.notifyReply(created.PostEvent) +} + +func (s subscriber) notifyReply(ev events.PostEvent) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + parent, err := s.store.GetPost(ctx, ev.ParentID) + if err != nil { + log.Printf("notify reply %s: load parent: %v", ev.PostID, err) + return + } + root, err := s.store.GetPost(ctx, ev.RootID) + if err != nil { + log.Printf("notify reply %s: load root: %v", ev.PostID, err) + return + } + author, err := s.store.UserByID(ctx, ev.AuthorID) + if err != nil { + log.Printf("notify reply %s: load author: %v", ev.PostID, err) + return + } + recipientID := parent.AuthorID + if author.Admin() { + recipientID = root.AuthorID + } + if recipientID == author.ID { + return + } + msg := PostReply{ + RootID: root.ID, + RootTitle: root.Title, + ReplyID: ev.PostID, + ReplyBody: ev.Body, + ReplyAuthorName: author.Name, + } + recipient, err := s.store.UserByID(ctx, recipientID) + if err != nil { + log.Printf("notify reply %s: load recipient: %v", msg.ReplyID, err) + return + } + if recipient == nil || strings.TrimSpace(recipient.Email) == "" { + return + } + msg.ToEmail = recipient.Email + msg.ToName = recipient.Name + if err := s.mail.NotifyPostReply(ctx, msg); err != nil { + log.Printf("notify reply %s: %v", msg.ReplyID, err) + return + } + log.Printf("notify reply %s: accepted", msg.ReplyID) +} diff --git a/internal/mail/subscriber_test.go b/internal/mail/subscriber_test.go new file mode 100644 index 0000000..b875339 --- /dev/null +++ b/internal/mail/subscriber_test.go @@ -0,0 +1,200 @@ +package mail + +import ( + "context" + "testing" + "time" + + "plumber/internal/events" + "plumber/internal/pacific" + "plumber/internal/store" +) + +func TestSubscribeReplyNotifications(t *testing.T) { + t.Parallel() + + mem := store.NewMemory() + homeowner := seedMailUser(t, mem, "homeowner", store.RoleUser, "sam@example.com") + admin := seedMailUser(t, mem, "plumber", store.RoleAdmin, "pat@example.com") + root := seedMailRoot(t, mem, homeowner.ID, "Leaky sink", "It drips.") + adminReply := seedMailReply(t, mem, admin.ID, root.ID) + homeownerReply := seedMailReply(t, mem, homeowner.ID, adminReply.ID) + + bus := events.New() + defer bus.Close() + recording := &Recording{} + Subscribe(bus, mem, recording) + + ctx := context.Background() + bus.Publish(ctx, events.PostCreated{PostEvent: events.PostEvent{ + PostID: root.ID, + RootID: root.ID, + Title: root.Title, + Body: root.Body, + AuthorID: homeowner.ID, + }}) + bus.Publish(ctx, events.PostUpdated{PostEvent: events.PostEvent{ + PostID: adminReply.ID, + RootID: root.ID, + ParentID: root.ID, + Body: "Edited", + AuthorID: admin.ID, + }}) + + bus.Publish(ctx, events.PostCreated{PostEvent: events.PostEvent{ + PostID: adminReply.ID, + RootID: root.ID, + ParentID: root.ID, + Body: adminReply.Body, + AuthorID: admin.ID, + }}) + msgs := waitForMail(t, recording, 1) + if msg := msgs[0]; msg.ToEmail != homeowner.Email || + msg.RootID != root.ID || + msg.RootTitle != root.Title || + msg.ReplyID != adminReply.ID || + msg.ReplyBody != adminReply.Body || + msg.ReplyAuthorName != admin.Name { + t.Fatalf("admin reply = %+v", msg) + } + + bus.Publish(ctx, events.PostCreated{PostEvent: events.PostEvent{ + PostID: homeownerReply.ID, + RootID: root.ID, + ParentID: adminReply.ID, + Body: homeownerReply.Body, + AuthorID: homeowner.ID, + }}) + msgs = waitForMail(t, recording, 2) + if msg := msgs[1]; msg.ToEmail != admin.Email || + msg.RootID != root.ID || + msg.ReplyID != homeownerReply.ID || + msg.ReplyAuthorName != homeowner.Name { + t.Fatalf("homeowner reply = %+v", msg) + } + + nestedAdmin := seedMailReply(t, mem, admin.ID, adminReply.ID) + bus.Publish(ctx, events.PostCreated{PostEvent: events.PostEvent{ + PostID: nestedAdmin.ID, + RootID: root.ID, + ParentID: adminReply.ID, + Body: nestedAdmin.Body, + AuthorID: admin.ID, + }}) + msgs = waitForMail(t, recording, 3) + if msg := msgs[2]; msg.ToEmail != homeowner.Email || + msg.RootID != root.ID || + msg.ReplyBody != nestedAdmin.Body || + msg.ReplyAuthorName != admin.Name { + t.Fatalf("nested admin reply = %+v", msg) + } + + self := seedMailReply(t, mem, homeowner.ID, root.ID) + bus.Publish(ctx, events.PostCreated{PostEvent: events.PostEvent{ + PostID: self.ID, + RootID: root.ID, + ParentID: root.ID, + Body: self.Body, + AuthorID: homeowner.ID, + }}) + + noEmail := seedMailUser(t, mem, "quiet", store.RoleUser, "") + quietRoot := seedMailRoot(t, mem, noEmail.ID, "Quiet thread", "No email.") + quietReply := seedMailReply(t, mem, admin.ID, quietRoot.ID) + bus.Publish(ctx, events.PostCreated{PostEvent: events.PostEvent{ + PostID: quietReply.ID, + RootID: quietRoot.ID, + ParentID: quietRoot.ID, + Body: quietReply.Body, + AuthorID: admin.ID, + }}) + + time.Sleep(50 * time.Millisecond) + if recording.Len() != 3 { + t.Fatalf("self, root, edit, or no-email sent mail: %+v", recording.Snapshot()) + } +} + +func TestSubscribeNopIgnoresReplies(t *testing.T) { + t.Parallel() + + mem := store.NewMemory() + homeowner := seedMailUser(t, mem, "homeowner", store.RoleUser, "sam@example.com") + admin := seedMailUser(t, mem, "plumber", store.RoleAdmin, "pat@example.com") + root := seedMailRoot(t, mem, homeowner.ID, "Leaky sink", "It drips.") + reply := seedMailReply(t, mem, admin.ID, root.ID) + + bus := events.New() + defer bus.Close() + recording := &Recording{} + Subscribe(bus, mem, Nop{}) + Subscribe(nil, mem, recording) + Subscribe(bus, mem, nil) + + bus.Publish(context.Background(), events.PostCreated{PostEvent: events.PostEvent{ + PostID: reply.ID, + RootID: root.ID, + ParentID: root.ID, + Body: reply.Body, + AuthorID: admin.ID, + }}) + time.Sleep(50 * time.Millisecond) + if recording.Len() != 0 { + t.Fatalf("Nop or nil subscribe sent mail: %+v", recording.Snapshot()) + } +} + +func seedMailUser(t *testing.T, mem *store.Memory, username string, role store.Role, email string) *store.User { + t.Helper() + u := &store.User{ + Username: username, + Name: username, + Email: email, + PasswordHash: "x", + Role: role, + } + if err := mem.CreateUser(context.Background(), u); err != nil { + t.Fatal(err) + } + return u +} + +func seedMailRoot(t *testing.T, mem *store.Memory, authorID, title, body string) *store.Post { + t.Helper() + root := &store.Post{ + AuthorID: authorID, + Title: title, + Body: body, + PostDate: pacific.Today(), + } + if err := mem.CreatePost(context.Background(), root); err != nil { + t.Fatal(err) + } + return root +} + +func seedMailReply(t *testing.T, mem *store.Memory, authorID, parentID string) *store.Post { + t.Helper() + reply := &store.Post{ + ParentID: &parentID, + AuthorID: authorID, + Body: "Reply from " + authorID, + } + if err := mem.CreatePost(context.Background(), reply); err != nil { + t.Fatal(err) + } + return reply +} + +func waitForMail(t *testing.T, recording *Recording, want int) []PostReply { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if recording.Len() >= want { + return recording.Snapshot() + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("recorded %d notifications, want %d", recording.Len(), want) + return nil +} diff --git a/internal/web/posts.go b/internal/web/posts.go index a8812c5..dbe1a21 100644 --- a/internal/web/posts.go +++ b/internal/web/posts.go @@ -5,15 +5,12 @@ import ( "database/sql" "errors" "fmt" - "log" "net/http" "net/url" "strings" - "time" "github.com/go-chi/chi/v5" - "plumber/internal/mail" "plumber/internal/store" ) @@ -82,9 +79,6 @@ func (s *Server) handleCreatePost(w http.ResponseWriter, r *http.Request) { if root == nil { root = post } - if parent != nil { - s.notifyPostReply(parent, root, post, user) - } s.publishPostCreated(post, root, user) http.Redirect( w, @@ -94,59 +88,6 @@ func (s *Server) handleCreatePost(w http.ResponseWriter, r *http.Request) { ) } -// notifyPostReply emails the root homeowner for admin replies and the direct -// parent author for homeowner replies. -func (s *Server) notifyPostReply( - parent *store.Post, - root *store.Post, - reply *store.Post, - replyAuthor *store.User, -) { - if parent == nil || - root == nil || - reply == nil || - replyAuthor == nil || - s.cfg.Mail == nil { - return - } - if _, disabled := s.cfg.Mail.(mail.Nop); disabled { - return - } - recipientID := parent.AuthorID - if replyAuthor.Admin() { - recipientID = root.AuthorID - } - if recipientID == replyAuthor.ID { - return - } - msg := mail.PostReply{ - RootID: root.ID, - RootTitle: root.Title, - ReplyID: reply.ID, - ReplyBody: reply.Body, - ReplyAuthorName: replyAuthor.Name, - } - go func() { - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - recipient, err := s.store.UserByID(ctx, recipientID) - if err != nil { - log.Printf("notify reply %s: load recipient: %v", msg.ReplyID, err) - return - } - if recipient == nil || strings.TrimSpace(recipient.Email) == "" { - return - } - msg.ToEmail = recipient.Email - msg.ToName = recipient.Name - if err := s.cfg.Mail.NotifyPostReply(ctx, msg); err != nil { - log.Printf("notify reply %s: %v", msg.ReplyID, err) - return - } - log.Printf("notify reply %s: accepted", msg.ReplyID) - }() -} - // handleEditPost updates only a post's body after verifying that the current // homeowner owns it or that an admin is editing an admin-authored post. func (s *Server) handleEditPost(w http.ResponseWriter, r *http.Request) { diff --git a/internal/web/posts_test.go b/internal/web/posts_test.go index b61b1fb..cd7b847 100644 --- a/internal/web/posts_test.go +++ b/internal/web/posts_test.go @@ -9,6 +9,7 @@ import ( "testing" "time" + "plumber/internal/events" "plumber/internal/mail" "plumber/internal/pacific" "plumber/internal/store" @@ -263,8 +264,12 @@ func TestEditPostRoutePermissions(t *testing.T) { func TestPostReplyNotifications(t *testing.T) { t.Parallel() + mem := store.NewMemory() + bus := events.New() + defer bus.Close() recording := &mail.Recording{} - srv, mem := newTestServer(t, Config{Mail: recording}) + mail.Subscribe(bus, mem, recording) + srv := newTestServerStore(t, mem, Config{Events: bus}) handler := srv.Handler() homeowner := seedUser(t, mem, uniq("homeowner"), "hunter22", store.RoleUser) admin := seedUser(t, mem, uniq("admin"), "hunter22", store.RoleAdmin) diff --git a/internal/web/server.go b/internal/web/server.go index b6f509a..17153d1 100644 --- a/internal/web/server.go +++ b/internal/web/server.go @@ -22,7 +22,6 @@ import ( "plumber/internal/blob" "plumber/internal/events" "plumber/internal/geo" - "plumber/internal/mail" "plumber/internal/pacific" "plumber/internal/store" ) @@ -35,7 +34,6 @@ type Config struct { // TrustedProxies are CIDRs allowed to set X-Forwarded-For (direct peer). TrustedProxies []*net.IPNet Blob blob.Uploader - Mail mail.Notifier Events events.Publisher BaseURL string } @@ -110,9 +108,6 @@ func New(st store.Store, sessionStore scs.Store, templateFS fs.FS, staticFS fs.F if cfg.Blob == nil { cfg.Blob = blob.Disabled{} } - if cfg.Mail == nil { - cfg.Mail = mail.Nop{} - } if cfg.Events == nil { cfg.Events = events.Nop{} }