From b8f1d88d6e8546e8edbb5bca86d017452a2ea919 Mon Sep 17 00:00:00 2001 From: codegirl-007 Date: Sat, 29 Aug 2026 01:06:26 -0700 Subject: [PATCH 1/5] Add an in-process post event bus. --- cmd/server/main.go | 7 +- internal/events/bus.go | 79 ++++++++++++ internal/events/bus_test.go | 104 +++++++++++++++ internal/events/event.go | 47 +++++++ internal/events/nop.go | 9 ++ internal/events/recording.go | 44 +++++++ internal/web/events.go | 65 ++++++++++ internal/web/events_test.go | 238 +++++++++++++++++++++++++++++++++++ internal/web/posts.go | 2 + internal/web/server.go | 7 ++ 10 files changed, 600 insertions(+), 2 deletions(-) create mode 100644 internal/events/bus.go create mode 100644 internal/events/bus_test.go create mode 100644 internal/events/event.go create mode 100644 internal/events/nop.go create mode 100644 internal/events/recording.go create mode 100644 internal/web/events.go create mode 100644 internal/web/events_test.go diff --git a/cmd/server/main.go b/cmd/server/main.go index 28e6e17..37f986e 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -17,6 +17,7 @@ import ( "plumber" "plumber/internal/blob" + "plumber/internal/events" "plumber/internal/mail" "plumber/internal/store" "plumber/internal/web" @@ -34,7 +35,7 @@ func main() { if err != nil { log.Fatalf("mail: %v", err) } - handler := newHandler(db, sessions, uploader, notifier) + handler := newHandler(db, sessions, uploader, notifier, events.New()) run(&http.Server{ Addr: listenAddr(), Handler: handler, @@ -58,13 +59,15 @@ func openDB() (*sql.DB, *store.SessionStore) { return db, sessions } -func newHandler(db *sql.DB, sessions *store.SessionStore, uploader blob.Uploader, notifier mail.Notifier) http.Handler { +func newHandler(db *sql.DB, sessions *store.SessionStore, uploader blob.Uploader, notifier mail.Notifier, 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")), "/"), }) if err != nil { log.Fatalf("server: %v", err) diff --git a/internal/events/bus.go b/internal/events/bus.go new file mode 100644 index 0000000..7f1dad2 --- /dev/null +++ b/internal/events/bus.go @@ -0,0 +1,79 @@ +package events + +import ( + "context" + "log" + "sync" +) + +const defaultBuffer = 64 + +// Publisher is the site-facing write side of the bus. +type Publisher interface { + Publish(ctx context.Context, ev any) +} + +// Bus is an in-process pub/sub with one worker and a bounded queue. +type Bus struct { + ch chan any + mu sync.Mutex + subs []func(context.Context, any) + closed sync.Once +} + +// New starts a worker that delivers events to subscribers in publish order. +func New() *Bus { + return newBus(defaultBuffer, true) +} + +func newBus(buffer int, start bool) *Bus { + if buffer < 1 { + buffer = 1 + } + b := &Bus{ch: make(chan any, buffer)} + if start { + go b.loop() + } + return b +} + +// Publish enqueues ev. It never blocks the caller; a full buffer is dropped. +func (b *Bus) Publish(_ context.Context, ev any) { + if b == nil { + return + } + select { + case b.ch <- ev: + default: + log.Printf("events: dropped %T", ev) + } +} + +// Subscribe adds a handler. Handlers run serially on the worker. +func (b *Bus) Subscribe(fn func(context.Context, any)) { + if b == nil || fn == nil { + return + } + b.mu.Lock() + b.subs = append(b.subs, fn) + b.mu.Unlock() +} + +func (b *Bus) loop() { + for ev := range b.ch { + b.mu.Lock() + subs := append([]func(context.Context, any){}, b.subs...) + b.mu.Unlock() + for _, fn := range subs { + fn(context.Background(), ev) + } + } +} + +// Close stops the worker. Safe to call more than once. +func (b *Bus) Close() { + if b == nil { + return + } + b.closed.Do(func() { close(b.ch) }) +} diff --git a/internal/events/bus_test.go b/internal/events/bus_test.go new file mode 100644 index 0000000..80c9275 --- /dev/null +++ b/internal/events/bus_test.go @@ -0,0 +1,104 @@ +package events + +import ( + "context" + "sync" + "testing" + "time" +) + +func TestPermalink(t *testing.T) { + t.Parallel() + + tests := []struct { + base, root, post, want string + }{ + {"", "root-1", "post-2", "/questions/root-1#post-post-2"}, + {"https://www.askaplumberfirst.com/", "root-1", "post-2", "https://www.askaplumberfirst.com/questions/root-1#post-post-2"}, + {"https://www.askaplumberfirst.com", "a b", "c/d", "https://www.askaplumberfirst.com/questions/a%20b#post-c%2Fd"}, + } + for _, tc := range tests { + if got := Permalink(tc.base, tc.root, tc.post); got != tc.want { + t.Fatalf("Permalink(%q, %q, %q) = %q, want %q", tc.base, tc.root, tc.post, got, tc.want) + } + } +} + +func TestNopAndRecording(t *testing.T) { + t.Parallel() + + Nop{}.Publish(context.Background(), PostCreated{}) + + rec := &Recording{} + rec.Publish(context.Background(), PostCreated{PostEvent: PostEvent{PostID: "a"}}) + rec.Publish(context.Background(), PostUpdated{PostEvent: PostEvent{PostID: "b"}}) + if rec.Len() != 2 { + t.Fatalf("len = %d", rec.Len()) + } + got := rec.Snapshot() + created, ok := got[0].(PostCreated) + if !ok || created.PostID != "a" { + t.Fatalf("first = %#v", got[0]) + } + updated, ok := got[1].(PostUpdated) + if !ok || updated.PostID != "b" { + t.Fatalf("second = %#v", got[1]) + } +} + +func TestBusDeliversInOrder(t *testing.T) { + t.Parallel() + + bus := New() + defer bus.Close() + + var mu sync.Mutex + var got []string + done := make(chan struct{}) + bus.Subscribe(func(_ context.Context, ev any) { + mu.Lock() + got = append(got, ev.(string)) + if len(got) == 3 { + close(done) + } + mu.Unlock() + }) + + ctx := context.Background() + bus.Publish(ctx, "one") + bus.Publish(ctx, "two") + bus.Publish(ctx, "three") + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("timed out waiting for events") + } + mu.Lock() + defer mu.Unlock() + if len(got) != 3 || got[0] != "one" || got[1] != "two" || got[2] != "three" { + t.Fatalf("got %v", got) + } +} + +func TestBusDropsWhenFull(t *testing.T) { + t.Parallel() + + bus := newBus(1, false) + bus.Publish(context.Background(), "kept") + bus.Publish(context.Background(), "dropped") + + select { + case ev := <-bus.ch: + if ev != "kept" { + t.Fatalf("got %v", ev) + } + default: + t.Fatal("expected buffered event") + } + select { + case ev := <-bus.ch: + t.Fatalf("unexpected extra event %v", ev) + default: + } +} diff --git a/internal/events/event.go b/internal/events/event.go new file mode 100644 index 0000000..ce5d1f2 --- /dev/null +++ b/internal/events/event.go @@ -0,0 +1,47 @@ +package events + +import ( + "net/url" + "strings" +) + +// Image is a public photo already attached to a site post. +type Image struct { + URL string + Description string +} + +// PostEvent is a Discord-free snapshot of a site post after a successful write. +type PostEvent struct { + PostID string + RootID string + ParentID string + Title string + Body string + City string + AuthorID string + AuthorName string + AuthorRole string + Images []Image + Permalink string +} + +// PostCreated is emitted after a successful site create. +type PostCreated struct { + PostEvent +} + +// PostUpdated is emitted after a successful site edit. +type PostUpdated struct { + PostEvent +} + +// Permalink builds /questions/{root}#post-{id}, prefixed by baseURL when set. +func Permalink(baseURL, rootID, postID string) string { + path := "/questions/" + url.PathEscape(rootID) + "#post-" + url.PathEscape(postID) + base := strings.TrimRight(strings.TrimSpace(baseURL), "/") + if base == "" { + return path + } + return base + path +} diff --git a/internal/events/nop.go b/internal/events/nop.go new file mode 100644 index 0000000..c71b595 --- /dev/null +++ b/internal/events/nop.go @@ -0,0 +1,9 @@ +package events + +import "context" + +// Nop is a Publisher used when nothing is subscribed. +type Nop struct{} + +// Publish discards ev. +func (Nop) Publish(context.Context, any) {} diff --git a/internal/events/recording.go b/internal/events/recording.go new file mode 100644 index 0000000..a3086bc --- /dev/null +++ b/internal/events/recording.go @@ -0,0 +1,44 @@ +package events + +import ( + "context" + "sync" +) + +// Recording is a test Publisher that records events synchronously. +type Recording struct { + mu sync.Mutex + evs []any +} + +// Publish appends ev. +func (r *Recording) Publish(_ context.Context, ev any) { + if r == nil { + return + } + r.mu.Lock() + defer r.mu.Unlock() + r.evs = append(r.evs, ev) +} + +// Len returns the number of recorded events. +func (r *Recording) Len() int { + if r == nil { + return 0 + } + r.mu.Lock() + defer r.mu.Unlock() + return len(r.evs) +} + +// Snapshot returns a copy of recorded events. +func (r *Recording) Snapshot() []any { + if r == nil { + return nil + } + r.mu.Lock() + defer r.mu.Unlock() + out := make([]any, len(r.evs)) + copy(out, r.evs) + return out +} diff --git a/internal/web/events.go b/internal/web/events.go new file mode 100644 index 0000000..e6846bb --- /dev/null +++ b/internal/web/events.go @@ -0,0 +1,65 @@ +package web + +import ( + "context" + + "plumber/internal/events" + "plumber/internal/store" +) + +func (s *Server) publishPostCreated(post, root *store.Post, author *store.User) { + s.publishPost(events.PostCreated{PostEvent: s.postEvent(post, root, author)}, root) +} + +func (s *Server) publishPostUpdated(post, root *store.Post, author *store.User) { + s.publishPost(events.PostUpdated{PostEvent: s.postEvent(post, root, author)}, root) +} + +func (s *Server) publishPost(ev any, root *store.Post) { + if root != nil && root.PostState == store.PostStateHidden { + return + } + s.cfg.Events.Publish(context.Background(), ev) +} + +func (s *Server) postEvent(post, root *store.Post, author *store.User) events.PostEvent { + if post == nil { + return events.PostEvent{} + } + rootID := post.ID + if root != nil { + rootID = root.ID + } + ev := events.PostEvent{ + PostID: post.ID, + RootID: rootID, + Title: post.Title, + Body: post.Body, + City: post.City, + AuthorID: post.AuthorID, + AuthorName: post.AuthorName, + AuthorRole: string(post.AuthorRole), + Permalink: events.Permalink(s.cfg.BaseURL, rootID, post.ID), + } + if post.ParentID != nil { + ev.ParentID = *post.ParentID + } + if author != nil { + if ev.AuthorName == "" { + ev.AuthorName = author.Name + } + if ev.AuthorRole == "" { + ev.AuthorRole = string(author.Role) + } + } + if n := len(post.Images); n > 0 { + ev.Images = make([]events.Image, 0, n) + for _, img := range post.Images { + ev.Images = append(ev.Images, events.Image{ + URL: img.PublicURL, + Description: img.Description, + }) + } + } + return ev +} diff --git a/internal/web/events_test.go b/internal/web/events_test.go new file mode 100644 index 0000000..dedfaf9 --- /dev/null +++ b/internal/web/events_test.go @@ -0,0 +1,238 @@ +package web + +import ( + "context" + "encoding/json" + "net/http" + "net/url" + "strings" + "testing" + + "plumber/internal/events" + "plumber/internal/pacific" + "plumber/internal/store" +) + +func TestPostHandlersPublishEvents(t *testing.T) { + t.Parallel() + + rec := &events.Recording{} + srv, mem := newTestServer(t, Config{ + Events: rec, + BaseURL: "https://www.askaplumberfirst.com", + }) + handler := srv.Handler() + homeowner := seedUser(t, mem, uniq("homeowner"), "hunter22", store.RoleUser) + admin := seedUser(t, mem, uniq("admin"), "hunter22", store.RoleAdmin) + homeownerCookies := loginUser(t, handler, homeowner.Username, "hunter22") + adminCookies := loginUser(t, handler, admin.Username, "hunter22") + homeownerCSRF := csrfForCookies(t, handler, homeownerCookies) + adminCSRF := csrfForCookies(t, handler, adminCookies) + + submit := postForm(handler, "/submit", url.Values{ + "_csrf": {homeownerCSRF}, + "title": {"Leaky sink"}, + "body": {"Water under the cabinet."}, + "city": {"Oakland"}, + }, homeownerCookies) + if submit.Code != http.StatusSeeOther { + t.Fatalf("submit status = %d: %s", submit.Code, submit.Body.String()) + } + + create := postForm(handler, "/posts", url.Values{ + "_csrf": {homeownerCSRF}, + "title": {"Second question"}, + "body": {"Another leak."}, + "city": {"Berkeley"}, + }, homeownerCookies) + if create.Code != http.StatusSeeOther { + t.Fatalf("create status = %d: %s", create.Code, create.Body.String()) + } + + roots, err := mem.ListRootPosts(context.Background(), pacific.Today(), homeowner.ID) + if err != nil || len(roots) != 2 { + t.Fatalf("roots = %+v, %v", roots, err) + } + var submitRoot, createRoot store.Post + for _, root := range roots { + switch root.Title { + case "Leaky sink": + submitRoot = root + case "Second question": + createRoot = root + } + } + if submitRoot.ID == "" || createRoot.ID == "" { + t.Fatalf("missing created roots: %+v", roots) + } + + reply := postForm(handler, "/posts", url.Values{ + "_csrf": {adminCSRF}, + "parent_id": {createRoot.ID}, + "body": {"Replace the cartridge."}, + }, adminCookies) + if reply.Code != http.StatusSeeOther { + t.Fatalf("reply status = %d: %s", reply.Code, reply.Body.String()) + } + thread, err := mem.GetPostThread(context.Background(), createRoot.ID) + if err != nil || len(thread.Replies) != 1 { + t.Fatalf("thread = %+v, %v", thread, err) + } + adminReply := thread.Replies[0] + + edit := postForm(handler, "/posts/"+createRoot.ID+"/edit", url.Values{ + "_csrf": {homeownerCSRF}, + "body": {"Updated leak description."}, + }, homeownerCookies) + if edit.Code != http.StatusSeeOther { + t.Fatalf("edit status = %d: %s", edit.Code, edit.Body.String()) + } + + empty := postForm(handler, "/posts", url.Values{ + "_csrf": {homeownerCSRF}, + "title": {"Missing body"}, + }, homeownerCookies) + if empty.Code != http.StatusBadRequest { + t.Fatalf("empty body status = %d, want 400", empty.Code) + } + + hidden := &store.Post{ + AuthorID: homeowner.ID, + Title: "Hidden thread", + Body: "Not public.", + PostDate: pacific.Today(), + PostState: store.PostStateHidden, + } + if err := mem.CreatePost(context.Background(), hidden); err != nil { + t.Fatal(err) + } + hiddenReply := postForm(handler, "/posts", url.Values{ + "_csrf": {homeownerCSRF}, + "parent_id": {hidden.ID}, + "body": {"Should not publish."}, + }, homeownerCookies) + if hiddenReply.Code != http.StatusNotFound { + t.Fatalf("hidden reply status = %d, want 404", hiddenReply.Code) + } + hiddenEdit := postForm(handler, "/posts/"+hidden.ID+"/edit", url.Values{ + "_csrf": {homeownerCSRF}, + "body": {"Still hidden."}, + }, homeownerCookies) + if hiddenEdit.Code != http.StatusSeeOther { + t.Fatalf("hidden edit status = %d: %s", hiddenEdit.Code, hiddenEdit.Body.String()) + } + + got := rec.Snapshot() + if len(got) != 4 { + t.Fatalf("published %d events, want 4: %#v", len(got), got) + } + + submitEv, ok := got[0].(events.PostCreated) + if !ok { + t.Fatalf("first event %T, want PostCreated", got[0]) + } + assertPostEvent(t, submitEv.PostEvent, events.PostEvent{ + PostID: submitRoot.ID, + RootID: submitRoot.ID, + Title: "Leaky sink", + Body: "Water under the cabinet.", + City: "Oakland", + AuthorID: homeowner.ID, + AuthorName: homeowner.Name, + AuthorRole: string(store.RoleUser), + Permalink: "https://www.askaplumberfirst.com/questions/" + submitRoot.ID + "#post-" + submitRoot.ID, + }) + + createEv, ok := got[1].(events.PostCreated) + if !ok { + t.Fatalf("second event %T, want PostCreated", got[1]) + } + assertPostEvent(t, createEv.PostEvent, events.PostEvent{ + PostID: createRoot.ID, + RootID: createRoot.ID, + Title: "Second question", + Body: "Another leak.", + City: "Berkeley", + AuthorID: homeowner.ID, + AuthorName: homeowner.Name, + AuthorRole: string(store.RoleUser), + Permalink: "https://www.askaplumberfirst.com/questions/" + createRoot.ID + "#post-" + createRoot.ID, + }) + + replyEv, ok := got[2].(events.PostCreated) + if !ok { + t.Fatalf("third event %T, want PostCreated", got[2]) + } + assertPostEvent(t, replyEv.PostEvent, events.PostEvent{ + PostID: adminReply.ID, + RootID: createRoot.ID, + ParentID: createRoot.ID, + Body: "Replace the cartridge.", + AuthorID: admin.ID, + AuthorName: admin.Name, + AuthorRole: string(store.RoleAdmin), + Permalink: "https://www.askaplumberfirst.com/questions/" + createRoot.ID + "#post-" + adminReply.ID, + }) + + editEv, ok := got[3].(events.PostUpdated) + if !ok { + t.Fatalf("fourth event %T, want PostUpdated", got[3]) + } + assertPostEvent(t, editEv.PostEvent, events.PostEvent{ + PostID: createRoot.ID, + RootID: createRoot.ID, + Title: "Second question", + Body: "Updated leak description.", + City: "Berkeley", + AuthorID: homeowner.ID, + AuthorName: homeowner.Name, + AuthorRole: string(store.RoleUser), + Permalink: "https://www.askaplumberfirst.com/questions/" + createRoot.ID + "#post-" + createRoot.ID, + }) + + for i, ev := range got { + raw, err := json.Marshal(ev) + if err != nil { + t.Fatal(err) + } + if strings.Contains(strings.ToLower(string(raw)), "discord") { + t.Fatalf("event %d contains discord fields: %s", i, raw) + } + } +} + +func TestStoreCreateDoesNotPublish(t *testing.T) { + t.Parallel() + + rec := &events.Recording{} + _, mem := newTestServer(t, Config{Events: rec}) + homeowner := seedUser(t, mem, uniq("homeowner"), "hunter22", store.RoleUser) + if err := mem.CreatePost(context.Background(), &store.Post{ + AuthorID: homeowner.ID, + Title: "Direct write", + Body: "No handler.", + PostDate: pacific.Today(), + }); err != nil { + t.Fatal(err) + } + if rec.Len() != 0 { + t.Fatalf("store.CreatePost published %d events", rec.Len()) + } +} + +func assertPostEvent(t *testing.T, got, want events.PostEvent) { + t.Helper() + if got.PostID != want.PostID || + got.RootID != want.RootID || + got.ParentID != want.ParentID || + got.Title != want.Title || + got.Body != want.Body || + got.City != want.City || + got.AuthorID != want.AuthorID || + got.AuthorName != want.AuthorName || + got.AuthorRole != want.AuthorRole || + got.Permalink != want.Permalink || + len(got.Images) != 0 { + t.Fatalf("event = %+v, want %+v", got, want) + } +} diff --git a/internal/web/posts.go b/internal/web/posts.go index 35d2f2e..a8812c5 100644 --- a/internal/web/posts.go +++ b/internal/web/posts.go @@ -85,6 +85,7 @@ func (s *Server) handleCreatePost(w http.ResponseWriter, r *http.Request) { if parent != nil { s.notifyPostReply(parent, root, post, user) } + s.publishPostCreated(post, root, user) http.Redirect( w, r, @@ -190,6 +191,7 @@ func (s *Server) handleEditPost(w http.ResponseWriter, r *http.Request) { http.Error(w, "could not save post", http.StatusInternalServerError) return } + s.publishPostUpdated(post, root, nil) http.Redirect( w, r, diff --git a/internal/web/server.go b/internal/web/server.go index 0998575..b6f509a 100644 --- a/internal/web/server.go +++ b/internal/web/server.go @@ -20,6 +20,7 @@ import ( "github.com/go-chi/chi/v5/middleware" "plumber/internal/blob" + "plumber/internal/events" "plumber/internal/geo" "plumber/internal/mail" "plumber/internal/pacific" @@ -35,6 +36,8 @@ type Config struct { TrustedProxies []*net.IPNet Blob blob.Uploader Mail mail.Notifier + Events events.Publisher + BaseURL string } type Server struct { @@ -110,6 +113,9 @@ func New(st store.Store, sessionStore scs.Store, templateFS fs.FS, staticFS fs.F if cfg.Mail == nil { cfg.Mail = mail.Nop{} } + if cfg.Events == nil { + cfg.Events = events.Nop{} + } funcMap := template.FuncMap{ "voteCtx": func(user *store.User, csrf, view, date string, post *store.Post) voteCtx { return voteCtx{User: user, CSRF: csrf, View: view, Date: date, Post: post} @@ -371,6 +377,7 @@ func (s *Server) handleSubmit(w http.ResponseWriter, r *http.Request) { http.Error(w, "could not save question", http.StatusInternalServerError) return } + s.publishPostCreated(post, post, u) http.Redirect(w, r, "/questions/"+url.PathEscape(post.ID), http.StatusSeeOther) } From 06fcc16c023c86d6fa408c0665a96a074856aae2 Mon Sep 17 00:00:00 2001 From: codegirl-007 Date: Sat, 29 Aug 2026 01:13:44 -0700 Subject: [PATCH 2/5] Add Discord outbound posting. --- .env.example | 4 + cmd/server/main.go | 11 +- db/queries/discord_links.sql | 32 +++ go.mod | 5 +- go.sum | 12 ++ internal/discord/api.go | 102 ++++++++++ internal/discord/bot.go | 178 ++++++++++++++++ internal/discord/bot_test.go | 247 +++++++++++++++++++++++ internal/discord/format.go | 75 +++++++ internal/discord/memory_links.go | 88 ++++++++ internal/store/discord_link.go | 105 ++++++++++ internal/store/migrate.go | 20 ++ internal/store/migrate_posts_test.go | 42 ++++ internal/store/sqlc/discord_links.sql.go | 100 +++++++++ internal/store/sqlc/models.go | 7 + schema.sql | 11 + 16 files changed, 1037 insertions(+), 2 deletions(-) create mode 100644 db/queries/discord_links.sql create mode 100644 internal/discord/api.go create mode 100644 internal/discord/bot.go create mode 100644 internal/discord/bot_test.go create mode 100644 internal/discord/format.go create mode 100644 internal/discord/memory_links.go create mode 100644 internal/store/discord_link.go create mode 100644 internal/store/sqlc/discord_links.sql.go diff --git a/.env.example b/.env.example index 0a4207b..2f35705 100644 --- a/.env.example +++ b/.env.example @@ -26,3 +26,7 @@ SECURE_COOKIE=0 # SPACES_BUCKET=your-bucket # SPACES_ENDPOINT=https://nyc3.digitaloceanspaces.com # SPACES_CDN_BASE=https://your-bucket.nyc3.cdn.digitaloceanspaces.com +# Discord bot subscriber. Leave unset to disable. +# DISCORD_BOT_TOKEN= +# DISCORD_CHANNEL_ID= +# DISCORD_ADMIN_MAP=123456789012345678:plumber,234567890123456789:otheradmin diff --git a/cmd/server/main.go b/cmd/server/main.go index 37f986e..83e6da0 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -17,6 +17,7 @@ import ( "plumber" "plumber/internal/blob" + "plumber/internal/discord" "plumber/internal/events" "plumber/internal/mail" "plumber/internal/store" @@ -35,7 +36,15 @@ func main() { if err != nil { log.Fatalf("mail: %v", err) } - handler := newHandler(db, sessions, uploader, notifier, events.New()) + bus := events.New() + bot, err := discord.FromEnv(store.NewDiscordLinks(db), bus) + if err != nil { + log.Fatalf("discord: %v", err) + } + if bot != nil { + defer bot.Close() + } + handler := newHandler(db, sessions, uploader, notifier, bus) run(&http.Server{ Addr: listenAddr(), Handler: handler, diff --git a/db/queries/discord_links.sql b/db/queries/discord_links.sql new file mode 100644 index 0000000..9ac2607 --- /dev/null +++ b/db/queries/discord_links.sql @@ -0,0 +1,32 @@ +-- name: GetDiscordPostLinkByPostID :one +SELECT post_id, discord_message_id, discord_thread_id, created_at +FROM discord_post_links +WHERE post_id = sqlc.arg(post_id); + +-- name: GetDiscordPostLinkByMessageID :one +SELECT post_id, discord_message_id, discord_thread_id, created_at +FROM discord_post_links +WHERE discord_message_id = sqlc.arg(discord_message_id); + +-- name: GetDiscordPostLinkByThreadID :one +SELECT post_id, discord_message_id, discord_thread_id, created_at +FROM discord_post_links +WHERE discord_thread_id = sqlc.arg(discord_thread_id) + AND discord_thread_id <> ''; + +-- name: UpsertDiscordPostLink :exec +INSERT INTO discord_post_links ( + post_id, discord_message_id, discord_thread_id, created_at +) +VALUES ( + sqlc.arg(post_id), + sqlc.arg(discord_message_id), + sqlc.arg(discord_thread_id), + sqlc.arg(created_at) +) +ON CONFLICT (post_id) DO UPDATE SET + discord_message_id = EXCLUDED.discord_message_id, + discord_thread_id = CASE + WHEN EXCLUDED.discord_thread_id <> '' THEN EXCLUDED.discord_thread_id + ELSE discord_post_links.discord_thread_id + END; diff --git a/go.mod b/go.mod index 2ff2b4b..7bb789c 100644 --- a/go.mod +++ b/go.mod @@ -7,10 +7,12 @@ require ( github.com/aws/aws-sdk-go-v2 v1.43.7 github.com/aws/aws-sdk-go-v2/credentials v1.19.37 github.com/aws/aws-sdk-go-v2/service/s3 v1.107.3 + github.com/bwmarrin/discordgo v0.29.0 github.com/go-chi/chi/v5 v5.3.1 github.com/google/uuid v1.6.0 github.com/jackc/pgx/v5 v5.10.0 github.com/joho/godotenv v1.5.1 + github.com/resend/resend-go/v3 v3.16.0 golang.org/x/crypto v0.55.0 golang.org/x/image v0.45.0 ) @@ -25,10 +27,11 @@ require ( github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.38 // indirect github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.39 // indirect github.com/aws/smithy-go v1.27.8 // indirect + github.com/gorilla/websocket v1.4.2 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect - github.com/resend/resend-go/v3 v3.16.0 // indirect golang.org/x/sync v0.22.0 // indirect + golang.org/x/sys v0.47.0 // indirect golang.org/x/text v0.41.0 // indirect ) diff --git a/go.sum b/go.sum index 479e538..1ce8c5f 100644 --- a/go.sum +++ b/go.sum @@ -24,6 +24,8 @@ github.com/aws/aws-sdk-go-v2/service/s3 v1.107.3 h1:IKoCZqfWfZzSBi16QFQ+QcbQ3LRQ github.com/aws/aws-sdk-go-v2/service/s3 v1.107.3/go.mod h1:RBpRcXiM4s2pOInVs32GsBonnje+fiAj4mcrStRmlCA= github.com/aws/smithy-go v1.27.8 h1:FR0dxZfIlV7Z8eh2iHfIofdunw382XsDV3Mxt9nUvRY= github.com/aws/smithy-go v1.27.8/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= +github.com/bwmarrin/discordgo v0.29.0 h1:FmWeXFaKUwrcL3Cx65c20bTRW+vOb6k8AnaP+EgjDno= +github.com/bwmarrin/discordgo v0.29.0/go.mod h1:NJZpH+1AfhIcyQsPeuBKsUtYrRnjkyu0kIVMCHkZtRY= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= @@ -31,6 +33,8 @@ github.com/go-chi/chi/v5 v5.3.1 h1:3j4HZLGZQ3JpMCrPJF/Jl3mYJfWLKBfNJ6quurUGCf8= github.com/go-chi/chi/v5 v5.3.1/go.mod h1:R+tYY2hNuVUUjxoPtqUdgBqevM9s9njzkTLutVsOCto= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc= +github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= @@ -50,14 +54,22 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4= golang.org/x/crypto v0.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M= golang.org/x/crypto v0.55.0/go.mod h1:uq0V9dE/fzQuJtbnL+2EhWOE63vo164FY8xqEnV9xis= golang.org/x/image v0.45.0 h1:FMb1nTbH5H9vF55SriQHgFw5GnNL9Jg6L25BwXKzhB0= golang.org/x/image v0.45.0/go.mod h1:n62x/7RqlwXDvGsSU4u6IUTUf6KghUZ9Bt7cG/T9Fx4= +golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +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/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= diff --git a/internal/discord/api.go b/internal/discord/api.go new file mode 100644 index 0000000..58d0bda --- /dev/null +++ b/internal/discord/api.go @@ -0,0 +1,102 @@ +package discord + +import ( + "context" + + "github.com/bwmarrin/discordgo" +) + +// API is the Discord REST surface used by the outbound subscriber. +type API interface { + SendToChannel(ctx context.Context, channelID string, msg Message) (messageID string, err error) + StartThread(ctx context.Context, channelID, messageID, name string) (threadID string, err error) + SendToThread(ctx context.Context, threadID string, msg Message) (messageID string, err error) + Edit(ctx context.Context, channelID, messageID string, msg Message) error + Close() error +} + +type sessionAPI struct { + session *discordgo.Session +} + +func (s *sessionAPI) SendToChannel(_ context.Context, channelID string, msg Message) (string, error) { + sent, err := s.session.ChannelMessageSendComplex(channelID, toMessageSend(msg)) + if err != nil { + return "", err + } + return sent.ID, nil +} + +func (s *sessionAPI) StartThread(_ context.Context, channelID, messageID, name string) (string, error) { + thread, err := s.session.MessageThreadStartComplex(channelID, messageID, &discordgo.ThreadStart{ + Name: name, + AutoArchiveDuration: 10080, + }) + if err != nil { + return "", err + } + return thread.ID, nil +} + +func (s *sessionAPI) SendToThread(ctx context.Context, threadID string, msg Message) (string, error) { + return s.SendToChannel(ctx, threadID, msg) +} + +func (s *sessionAPI) Edit(_ context.Context, channelID, messageID string, msg Message) error { + embeds := toEmbeds(msg) + _, err := s.session.ChannelMessageEditComplex(&discordgo.MessageEdit{ + ID: messageID, + Channel: channelID, + Embeds: &embeds, + }) + return err +} + +func (s *sessionAPI) Close() error { + if s == nil || s.session == nil { + return nil + } + return s.session.Close() +} + +func toMessageSend(msg Message) *discordgo.MessageSend { + return &discordgo.MessageSend{ + Embeds: toEmbeds(msg), + AllowedMentions: &discordgo.MessageAllowedMentions{}, + } +} + +func toEmbeds(msg Message) []*discordgo.MessageEmbed { + main := &discordgo.MessageEmbed{ + Title: msg.Title, + URL: msg.URL, + Description: msg.Description, + Color: embedColor, + } + if msg.City != "" { + main.Fields = append(main.Fields, &discordgo.MessageEmbedField{ + Name: "City", + Value: msg.City, + Inline: true, + }) + } + if msg.Author != "" { + main.Fields = append(main.Fields, &discordgo.MessageEmbedField{ + Name: "Author", + Value: msg.Author, + Inline: true, + }) + } + embeds := []*discordgo.MessageEmbed{main} + for i, url := range msg.ImageURLs { + if i == 0 { + main.Image = &discordgo.MessageEmbedImage{URL: url} + continue + } + embeds = append(embeds, &discordgo.MessageEmbed{ + Color: embedColor, + Image: &discordgo.MessageEmbedImage{URL: url}, + }) + } + return embeds +} diff --git a/internal/discord/bot.go b/internal/discord/bot.go new file mode 100644 index 0000000..cc4bc19 --- /dev/null +++ b/internal/discord/bot.go @@ -0,0 +1,178 @@ +package discord + +import ( + "context" + "database/sql" + "errors" + "fmt" + "log" + "os" + "strings" + "time" + + "github.com/bwmarrin/discordgo" + + "plumber/internal/events" + "plumber/internal/store" +) + +const discordTimeout = 15 * time.Second + +// Bot posts site events to a Discord channel and owns post-to-message links. +type Bot struct { + channelID string + links store.DiscordLinkStore + api API +} + +// New constructs an outbound subscriber. Tests inject a fake API. +func New(channelID string, links store.DiscordLinkStore, api API) *Bot { + return &Bot{channelID: strings.TrimSpace(channelID), links: links, api: api} +} + +// FromEnv builds a bot when Discord env is set. Missing config is a no-op. +func FromEnv(links store.DiscordLinkStore, bus *events.Bus) (*Bot, error) { + token := strings.TrimSpace(os.Getenv("DISCORD_BOT_TOKEN")) + channelID := strings.TrimSpace(os.Getenv("DISCORD_CHANNEL_ID")) + if token == "" && channelID == "" { + return nil, nil + } + if token == "" { + return nil, fmt.Errorf("DISCORD_BOT_TOKEN is required when DISCORD_CHANNEL_ID is set") + } + if channelID == "" { + return nil, fmt.Errorf("DISCORD_CHANNEL_ID is required when DISCORD_BOT_TOKEN is set") + } + if links == nil { + return nil, fmt.Errorf("discord links store is required") + } + session, err := discordgo.New("Bot " + token) + if err != nil { + return nil, err + } + bot := New(channelID, links, &sessionAPI{session: session}) + if bus != nil { + bus.Subscribe(bot.Handle) + } + log.Printf("discord: outbound subscriber enabled") + return bot, nil +} + +// Close releases the Discord session. +func (b *Bot) Close() error { + if b == nil || b.api == nil { + return nil + } + return b.api.Close() +} + +// Handle processes one site event. Failures are logged and do not fail the request. +func (b *Bot) Handle(_ context.Context, ev any) { + if b == nil { + return + } + ctx, cancel := context.WithTimeout(context.Background(), discordTimeout) + defer cancel() + switch e := ev.(type) { + case events.PostCreated: + b.onCreated(ctx, e.PostEvent) + case events.PostUpdated: + b.onUpdated(ctx, e.PostEvent) + } +} + +func (b *Bot) onCreated(ctx context.Context, ev events.PostEvent) { + if isRoot(ev) { + b.createRoot(ctx, ev) + return + } + b.createReply(ctx, ev) +} + +func (b *Bot) onUpdated(ctx context.Context, ev events.PostEvent) { + link, err := b.links.GetByPostID(ctx, ev.PostID) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + b.onCreated(ctx, ev) + return + } + log.Printf("discord: load link %s: %v", ev.PostID, err) + return + } + channelID, err := b.editChannel(ctx, ev, link) + if err != nil { + log.Printf("discord: edit channel %s: %v", ev.PostID, err) + return + } + if err := b.api.Edit(ctx, channelID, link.MessageID, formatMessage(ev)); err != nil { + log.Printf("discord: edit %s: %v", ev.PostID, err) + return + } + log.Printf("discord: edited %s", ev.PostID) +} + +func (b *Bot) createRoot(ctx context.Context, ev events.PostEvent) { + msg := formatMessage(ev) + messageID, err := b.api.SendToChannel(ctx, b.channelID, msg) + if err != nil { + log.Printf("discord: send root %s: %v", ev.PostID, err) + return + } + threadID, err := b.api.StartThread(ctx, b.channelID, messageID, msg.ThreadName) + if err != nil { + log.Printf("discord: start thread %s: %v", ev.PostID, err) + return + } + if err := b.links.Upsert(ctx, store.DiscordLink{ + PostID: ev.PostID, + MessageID: messageID, + ThreadID: threadID, + }); err != nil { + log.Printf("discord: save root link %s: %v", ev.PostID, err) + return + } + log.Printf("discord: posted root %s", ev.PostID) +} + +func (b *Bot) createReply(ctx context.Context, ev events.PostEvent) { + root, err := b.links.GetByPostID(ctx, ev.RootID) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + log.Printf("discord: skip reply %s: no root thread", ev.PostID) + return + } + log.Printf("discord: load root link %s: %v", ev.RootID, err) + return + } + if strings.TrimSpace(root.ThreadID) == "" { + log.Printf("discord: skip reply %s: no root thread", ev.PostID) + return + } + messageID, err := b.api.SendToThread(ctx, root.ThreadID, formatMessage(ev)) + if err != nil { + log.Printf("discord: send reply %s: %v", ev.PostID, err) + return + } + if err := b.links.Upsert(ctx, store.DiscordLink{ + PostID: ev.PostID, + MessageID: messageID, + }); err != nil { + log.Printf("discord: save reply link %s: %v", ev.PostID, err) + return + } + log.Printf("discord: posted reply %s", ev.PostID) +} + +func (b *Bot) editChannel(ctx context.Context, ev events.PostEvent, link *store.DiscordLink) (string, error) { + if strings.TrimSpace(link.ThreadID) != "" { + return b.channelID, nil + } + root, err := b.links.GetByPostID(ctx, ev.RootID) + if err != nil { + return "", err + } + if strings.TrimSpace(root.ThreadID) == "" { + return "", fmt.Errorf("root %s has no thread", ev.RootID) + } + return root.ThreadID, nil +} diff --git a/internal/discord/bot_test.go b/internal/discord/bot_test.go new file mode 100644 index 0000000..2c5e178 --- /dev/null +++ b/internal/discord/bot_test.go @@ -0,0 +1,247 @@ +package discord + +import ( + "context" + "sync" + "testing" + + "strconv" + + "plumber/internal/events" + "plumber/internal/store" +) + +type recordedSend struct { + Kind string + ChannelID string + Name string + Msg Message +} + +type fakeAPI struct { + mu sync.Mutex + sends []recordedSend + edits []recordedSend + next int + failSend error +} + +func (f *fakeAPI) SendToChannel(_ context.Context, channelID string, msg Message) (string, error) { + return f.record("channel", channelID, "", msg) +} + +func (f *fakeAPI) StartThread(_ context.Context, channelID, messageID, name string) (string, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.next++ + f.sends = append(f.sends, recordedSend{ + Kind: "thread", + ChannelID: channelID, + Name: name, + Msg: Message{ThreadName: name, URL: messageID}, + }) + return "thread-" + messageID, nil +} + +func (f *fakeAPI) SendToThread(_ context.Context, threadID string, msg Message) (string, error) { + return f.record("thread-msg", threadID, "", msg) +} + +func (f *fakeAPI) Edit(_ context.Context, channelID, messageID string, msg Message) error { + f.mu.Lock() + defer f.mu.Unlock() + f.edits = append(f.edits, recordedSend{ + Kind: "edit", + ChannelID: channelID, + Name: messageID, + Msg: msg, + }) + return nil +} + +func (f *fakeAPI) Close() error { return nil } + +func (f *fakeAPI) record(kind, channelID, name string, msg Message) (string, error) { + f.mu.Lock() + defer f.mu.Unlock() + if f.failSend != nil { + return "", f.failSend + } + f.next++ + id := "msg-" + strconv.Itoa(f.next) + f.sends = append(f.sends, recordedSend{Kind: kind, ChannelID: channelID, Name: name, Msg: msg}) + return id, nil +} + +func TestOutboundRootReplyAndEdit(t *testing.T) { + t.Parallel() + + links := newMemoryLinks() + api := &fakeAPI{} + bot := New("channel-1", links, api) + ctx := context.Background() + + root := events.PostEvent{ + PostID: "root-1", + RootID: "root-1", + Title: "Leaky sink", + Body: "Water under the cabinet.", + City: "Oakland", + AuthorName: "sam", + Permalink: "https://www.askaplumberfirst.com/questions/root-1#post-root-1", + Images: []events.Image{{URL: "https://cdn.example/a.jpg"}, {URL: "https://cdn.example/b.jpg"}}, + } + bot.Handle(ctx, events.PostCreated{PostEvent: root}) + + if len(api.sends) != 2 || api.sends[0].Kind != "channel" || api.sends[1].Kind != "thread" { + t.Fatalf("root sends = %+v", api.sends) + } + if api.sends[0].ChannelID != "channel-1" || api.sends[1].Name != "Leaky sink" { + t.Fatalf("root routing = %+v", api.sends) + } + if got := api.sends[0].Msg.ImageURLs; len(got) != 2 || got[0] != "https://cdn.example/a.jpg" { + t.Fatalf("root images = %v", got) + } + link, err := links.GetByPostID(ctx, "root-1") + if err != nil || link.MessageID != "msg-1" || link.ThreadID != "thread-msg-1" { + t.Fatalf("root link = %+v, %v", link, err) + } + + reply := events.PostEvent{ + PostID: "reply-1", + RootID: "root-1", + ParentID: "root-1", + Body: "Replace the cartridge.", + AuthorName: "plumber", + Permalink: "https://www.askaplumberfirst.com/questions/root-1#post-reply-1", + } + bot.Handle(ctx, events.PostCreated{PostEvent: reply}) + if len(api.sends) != 3 || api.sends[2].Kind != "thread-msg" || api.sends[2].ChannelID != "thread-msg-1" { + t.Fatalf("reply sends = %+v", api.sends) + } + replyLink, err := links.GetByPostID(ctx, "reply-1") + if err != nil || replyLink.MessageID != "msg-3" || replyLink.ThreadID != "" { + t.Fatalf("reply link = %+v, %v", replyLink, err) + } + + root.Body = "Updated leak." + bot.Handle(ctx, events.PostUpdated{PostEvent: root}) + if len(api.edits) != 1 || api.edits[0].ChannelID != "channel-1" || api.edits[0].Name != "msg-1" { + t.Fatalf("root edit = %+v", api.edits) + } + if api.edits[0].Msg.Description != "Updated leak." { + t.Fatalf("root edit body = %+v", api.edits[0].Msg) + } + + reply.Body = "Use a ceramic cartridge." + bot.Handle(ctx, events.PostUpdated{PostEvent: reply}) + if len(api.edits) != 2 || api.edits[1].ChannelID != "thread-msg-1" || api.edits[1].Name != "msg-3" { + t.Fatalf("reply edit = %+v", api.edits) + } +} + +func TestOutboundSkipsReplyWithoutRootLink(t *testing.T) { + t.Parallel() + + api := &fakeAPI{} + bot := New("channel-1", newMemoryLinks(), api) + bot.Handle(context.Background(), events.PostCreated{PostEvent: events.PostEvent{ + PostID: "reply-1", + RootID: "missing", + ParentID: "missing", + Body: "Orphan reply", + }}) + if len(api.sends) != 0 { + t.Fatalf("unexpected sends %+v", api.sends) + } +} + +func TestOutboundUpdateWithoutLinkCreates(t *testing.T) { + t.Parallel() + + links := newMemoryLinks() + api := &fakeAPI{} + bot := New("channel-1", links, api) + bot.Handle(context.Background(), events.PostUpdated{PostEvent: events.PostEvent{ + PostID: "root-2", + RootID: "root-2", + Title: "Late question", + Body: "Created while Discord was down.", + }}) + link, err := links.GetByPostID(context.Background(), "root-2") + if err != nil || link.ThreadID == "" || len(api.sends) != 2 { + t.Fatalf("late create link=%+v sends=%+v err=%v", link, api.sends, err) + } +} + +func TestFormatMessage(t *testing.T) { + t.Parallel() + + got := formatMessage(events.PostEvent{ + Title: "Leaky sink", + Body: "It drips.", + City: "Oakland", + AuthorName: "sam", + Permalink: "https://example.com/q", + Images: []events.Image{{URL: "https://cdn.example/a.jpg", Description: "ignored"}}, + }) + if got.Title != "Leaky sink" || + got.Description != "It drips." || + got.City != "Oakland" || + got.Author != "sam" || + got.URL != "https://example.com/q" || + got.ThreadName != "Leaky sink" || + len(got.ImageURLs) != 1 { + t.Fatalf("format = %+v", got) + } + + reply := formatMessage(events.PostEvent{Body: "Thanks", AuthorName: ""}) + if reply.Title != "Reply" || reply.Author != "Someone" || reply.ThreadName != "Question" { + t.Fatalf("reply format = %+v", reply) + } +} + +func TestFromEnvDisabled(t *testing.T) { + t.Setenv("DISCORD_BOT_TOKEN", "") + t.Setenv("DISCORD_CHANNEL_ID", "") + bot, err := FromEnv(newMemoryLinks(), nil) + if err != nil || bot != nil { + t.Fatalf("disabled FromEnv = (%v, %v)", bot, err) + } +} + +func TestFromEnvRequiresBoth(t *testing.T) { + t.Setenv("DISCORD_BOT_TOKEN", "token") + t.Setenv("DISCORD_CHANNEL_ID", "") + if _, err := FromEnv(newMemoryLinks(), nil); err == nil { + t.Fatal("expected error when channel is missing") + } + t.Setenv("DISCORD_BOT_TOKEN", "") + t.Setenv("DISCORD_CHANNEL_ID", "channel") + if _, err := FromEnv(newMemoryLinks(), nil); err == nil { + t.Fatal("expected error when token is missing") + } +} + +func TestMemoryLinkUpsertKeepsThread(t *testing.T) { + t.Parallel() + + links := newMemoryLinks() + ctx := context.Background() + if err := links.Upsert(ctx, store.DiscordLink{PostID: "p", MessageID: "m1", ThreadID: "t1"}); err != nil { + t.Fatal(err) + } + if err := links.Upsert(ctx, store.DiscordLink{PostID: "p", MessageID: "m2"}); err != nil { + t.Fatal(err) + } + got, err := links.GetByPostID(ctx, "p") + if err != nil || got.MessageID != "m2" || got.ThreadID != "t1" { + t.Fatalf("upsert keep thread = %+v, %v", got, err) + } + if _, err := links.GetByMessageID(ctx, "m2"); err != nil { + t.Fatal(err) + } + if _, err := links.GetRootByThreadID(ctx, "t1"); err != nil { + t.Fatal(err) + } +} diff --git a/internal/discord/format.go b/internal/discord/format.go new file mode 100644 index 0000000..f552a19 --- /dev/null +++ b/internal/discord/format.go @@ -0,0 +1,75 @@ +package discord + +import ( + "strings" + + "plumber/internal/events" +) + +const ( + embedTitleLimit = 256 + embedDescriptionLimit = 4096 + threadNameLimit = 100 + embedColor = 0xe96a26 +) + +// Message is a Discord-ready snapshot of a site post event. +type Message struct { + Title string + URL string + Description string + City string + Author string + ImageURLs []string + ThreadName string +} + +func formatMessage(ev events.PostEvent) Message { + title := strings.TrimSpace(ev.Title) + if title == "" { + title = "Reply" + } + author := strings.TrimSpace(ev.AuthorName) + if author == "" { + author = "Someone" + } + msg := Message{ + Title: truncateRunes(title, embedTitleLimit), + URL: strings.TrimSpace(ev.Permalink), + Description: truncateRunes(strings.TrimSpace(ev.Body), embedDescriptionLimit), + City: strings.TrimSpace(ev.City), + Author: author, + ThreadName: threadName(ev.Title), + } + for _, img := range ev.Images { + url := strings.TrimSpace(img.URL) + if url == "" { + continue + } + msg.ImageURLs = append(msg.ImageURLs, url) + } + return msg +} + +func threadName(title string) string { + title = strings.TrimSpace(title) + if title == "" { + return "Question" + } + return truncateRunes(title, threadNameLimit) +} + +func truncateRunes(s string, max int) string { + if max <= 0 { + return "" + } + runes := []rune(s) + if len(runes) <= max { + return s + } + return string(runes[:max]) +} + +func isRoot(ev events.PostEvent) bool { + return strings.TrimSpace(ev.ParentID) == "" +} diff --git a/internal/discord/memory_links.go b/internal/discord/memory_links.go new file mode 100644 index 0000000..8555f3d --- /dev/null +++ b/internal/discord/memory_links.go @@ -0,0 +1,88 @@ +package discord + +import ( + "context" + "database/sql" + "strings" + "sync" + "time" + + "plumber/internal/store" +) + +// memoryLinks is an in-process DiscordLinkStore for tests. +type memoryLinks struct { + mu sync.Mutex + byPost map[string]store.DiscordLink + byMessage map[string]string + byThread map[string]string +} + +func newMemoryLinks() *memoryLinks { + return &memoryLinks{ + byPost: map[string]store.DiscordLink{}, + byMessage: map[string]string{}, + byThread: map[string]string{}, + } +} + +func (m *memoryLinks) GetByPostID(_ context.Context, postID string) (*store.DiscordLink, error) { + m.mu.Lock() + defer m.mu.Unlock() + link, ok := m.byPost[strings.TrimSpace(postID)] + if !ok { + return nil, sql.ErrNoRows + } + cp := link + return &cp, nil +} + +func (m *memoryLinks) GetByMessageID(_ context.Context, messageID string) (*store.DiscordLink, error) { + m.mu.Lock() + defer m.mu.Unlock() + postID, ok := m.byMessage[strings.TrimSpace(messageID)] + if !ok { + return nil, sql.ErrNoRows + } + link := m.byPost[postID] + cp := link + return &cp, nil +} + +func (m *memoryLinks) GetRootByThreadID(_ context.Context, threadID string) (*store.DiscordLink, error) { + m.mu.Lock() + defer m.mu.Unlock() + postID, ok := m.byThread[strings.TrimSpace(threadID)] + if !ok { + return nil, sql.ErrNoRows + } + link := m.byPost[postID] + cp := link + return &cp, nil +} + +func (m *memoryLinks) Upsert(_ context.Context, link store.DiscordLink) error { + m.mu.Lock() + defer m.mu.Unlock() + link.PostID = strings.TrimSpace(link.PostID) + link.MessageID = strings.TrimSpace(link.MessageID) + link.ThreadID = strings.TrimSpace(link.ThreadID) + if link.CreatedAt == "" { + link.CreatedAt = time.Now().UTC().Format(time.RFC3339Nano) + } + if prev, ok := m.byPost[link.PostID]; ok { + delete(m.byMessage, prev.MessageID) + if prev.ThreadID != "" { + delete(m.byThread, prev.ThreadID) + } + if link.ThreadID == "" { + link.ThreadID = prev.ThreadID + } + } + m.byPost[link.PostID] = link + m.byMessage[link.MessageID] = link.PostID + if link.ThreadID != "" { + m.byThread[link.ThreadID] = link.PostID + } + return nil +} diff --git a/internal/store/discord_link.go b/internal/store/discord_link.go new file mode 100644 index 0000000..ee68521 --- /dev/null +++ b/internal/store/discord_link.go @@ -0,0 +1,105 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + "strings" + "time" + + "plumber/internal/store/sqlc" +) + +// DiscordLink is the bot-owned mapping from a site post to a Discord message. +type DiscordLink struct { + PostID string + MessageID string + ThreadID string + CreatedAt string +} + +// DiscordLinkStore is the mapping table used only by the Discord subscriber. +// It is not part of Store. +type DiscordLinkStore interface { + GetByPostID(ctx context.Context, postID string) (*DiscordLink, error) + GetByMessageID(ctx context.Context, messageID string) (*DiscordLink, error) + GetRootByThreadID(ctx context.Context, threadID string) (*DiscordLink, error) + Upsert(ctx context.Context, link DiscordLink) error +} + +// DiscordLinks implements DiscordLinkStore against Postgres. +type DiscordLinks struct { + db *sql.DB +} + +// NewDiscordLinks wraps db. It is independent of Store. +func NewDiscordLinks(db *sql.DB) *DiscordLinks { + return &DiscordLinks{db: db} +} + +// GetByPostID returns the link for a site post. +func (d *DiscordLinks) GetByPostID(ctx context.Context, postID string) (*DiscordLink, error) { + if d == nil || d.db == nil { + return nil, fmt.Errorf("discord links: no database") + } + row, err := sqlc.New(d.db).GetDiscordPostLinkByPostID(ctx, strings.TrimSpace(postID)) + if err != nil { + return nil, err + } + return discordLinkFromRow(row), nil +} + +// GetByMessageID returns the link for a Discord message. +func (d *DiscordLinks) GetByMessageID(ctx context.Context, messageID string) (*DiscordLink, error) { + if d == nil || d.db == nil { + return nil, fmt.Errorf("discord links: no database") + } + row, err := sqlc.New(d.db).GetDiscordPostLinkByMessageID(ctx, strings.TrimSpace(messageID)) + if err != nil { + return nil, err + } + return discordLinkFromRow(row), nil +} + +// GetRootByThreadID returns the root link for a Discord thread. +func (d *DiscordLinks) GetRootByThreadID(ctx context.Context, threadID string) (*DiscordLink, error) { + if d == nil || d.db == nil { + return nil, fmt.Errorf("discord links: no database") + } + row, err := sqlc.New(d.db).GetDiscordPostLinkByThreadID(ctx, strings.TrimSpace(threadID)) + if err != nil { + return nil, err + } + return discordLinkFromRow(row), nil +} + +// Upsert inserts or replaces the Discord IDs for a post. +func (d *DiscordLinks) Upsert(ctx context.Context, link DiscordLink) error { + if d == nil || d.db == nil { + return fmt.Errorf("discord links: no database") + } + link.PostID = strings.TrimSpace(link.PostID) + link.MessageID = strings.TrimSpace(link.MessageID) + link.ThreadID = strings.TrimSpace(link.ThreadID) + if link.PostID == "" || link.MessageID == "" { + return fmt.Errorf("discord links: post and message ids are required") + } + if link.CreatedAt == "" { + link.CreatedAt = time.Now().UTC().Format(time.RFC3339Nano) + } + return sqlc.New(d.db).UpsertDiscordPostLink(ctx, sqlc.UpsertDiscordPostLinkParams{ + PostID: link.PostID, + DiscordMessageID: link.MessageID, + DiscordThreadID: link.ThreadID, + CreatedAt: link.CreatedAt, + }) +} + +func discordLinkFromRow(row sqlc.DiscordPostLink) *DiscordLink { + return &DiscordLink{ + PostID: row.PostID, + MessageID: row.DiscordMessageID, + ThreadID: row.DiscordThreadID, + CreatedAt: row.CreatedAt, + } +} diff --git a/internal/store/migrate.go b/internal/store/migrate.go index 4d12c8a..28638e9 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -139,6 +139,25 @@ CREATE TABLE IF NOT EXISTS post_images ( return nil } +func migrateDiscordPostLinks(ctx context.Context, exec execContext) error { + if _, err := exec.ExecContext(ctx, ` +CREATE TABLE IF NOT EXISTS discord_post_links ( + post_id TEXT PRIMARY KEY REFERENCES posts(id) ON DELETE CASCADE, + discord_message_id TEXT NOT NULL UNIQUE, + discord_thread_id TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL +)`); err != nil { + return fmt.Errorf("create discord_post_links: %w", err) + } + if _, err := exec.ExecContext(ctx, ` +CREATE UNIQUE INDEX IF NOT EXISTS discord_post_links_thread_uidx + ON discord_post_links (discord_thread_id) + WHERE discord_thread_id <> ''`); err != nil { + return fmt.Errorf("discord_post_links_thread_uidx: %w", err) + } + return nil +} + func migratePostDate(ctx context.Context, exec execContext) error { steps := []struct { name string @@ -331,6 +350,7 @@ CREATE TABLE IF NOT EXISTS schema_migrations ( {"008_post_author_index", migratePostAuthorIndex}, {"009_drop_legacy_post_tables", migrateDropLegacyPostTables}, {"010_post_images", migratePostImages}, + {"011_discord_post_links", migrateDiscordPostLinks}, } for _, m := range migrations { if applied[m.version] { diff --git a/internal/store/migrate_posts_test.go b/internal/store/migrate_posts_test.go index 5db4972..d2dba7c 100644 --- a/internal/store/migrate_posts_test.go +++ b/internal/store/migrate_posts_test.go @@ -81,6 +81,12 @@ CREATE TABLE users ( if err := migratePostImages(ctx, conn); err != nil { t.Fatalf("post images migration is not idempotent: %v", err) } + if err := migrateDiscordPostLinks(ctx, conn); err != nil { + t.Fatal(err) + } + if err := migrateDiscordPostLinks(ctx, conn); err != nil { + t.Fatalf("discord post links migration is not idempotent: %v", err) + } if _, err := conn.ExecContext(ctx, ` INSERT INTO users (id, name, role) VALUES ('homeowner', 'Home Owner', 'user'), ('plumber', 'The Plumber', 'admin'); @@ -143,6 +149,42 @@ VALUES ('homeowner', 'root-1', 1);`); err != nil { }); err == nil { t.Fatal("fifth image position unexpectedly succeeded") } + if err := imageQueries.UpsertDiscordPostLink(ctx, sqlc.UpsertDiscordPostLinkParams{ + PostID: "root-1", + DiscordMessageID: "msg-root", + DiscordThreadID: "thread-root", + CreatedAt: "2026-08-26T08:00:00Z", + }); err != nil { + t.Fatal(err) + } + if err := imageQueries.UpsertDiscordPostLink(ctx, sqlc.UpsertDiscordPostLinkParams{ + PostID: "reply-1", + DiscordMessageID: "msg-reply", + DiscordThreadID: "", + CreatedAt: "2026-08-26T09:00:00Z", + }); err != nil { + t.Fatal(err) + } + rootLink, err := imageQueries.GetDiscordPostLinkByPostID(ctx, "root-1") + if err != nil || rootLink.DiscordMessageID != "msg-root" || rootLink.DiscordThreadID != "thread-root" { + t.Fatalf("root discord link = %+v, %v", rootLink, err) + } + threadLink, err := imageQueries.GetDiscordPostLinkByThreadID(ctx, "thread-root") + if err != nil || threadLink.PostID != "root-1" { + t.Fatalf("thread discord link = %+v, %v", threadLink, err) + } + if err := imageQueries.UpsertDiscordPostLink(ctx, sqlc.UpsertDiscordPostLinkParams{ + PostID: "reply-1", + DiscordMessageID: "msg-reply-2", + DiscordThreadID: "", + CreatedAt: "2026-08-26T09:01:00Z", + }); err != nil { + t.Fatal(err) + } + replyLink, err := imageQueries.GetDiscordPostLinkByPostID(ctx, "reply-1") + if err != nil || replyLink.DiscordMessageID != "msg-reply-2" || replyLink.DiscordThreadID != "" { + t.Fatalf("reply upsert = %+v, %v", replyLink, err) + } var postVoteIndexCount int if err := conn.QueryRowContext(ctx, ` SELECT count(*) diff --git a/internal/store/sqlc/discord_links.sql.go b/internal/store/sqlc/discord_links.sql.go new file mode 100644 index 0000000..79d1680 --- /dev/null +++ b/internal/store/sqlc/discord_links.sql.go @@ -0,0 +1,100 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: discord_links.sql + +package sqlc + +import ( + "context" +) + +const getDiscordPostLinkByMessageID = `-- name: GetDiscordPostLinkByMessageID :one +SELECT post_id, discord_message_id, discord_thread_id, created_at +FROM discord_post_links +WHERE discord_message_id = $1 +` + +func (q *Queries) GetDiscordPostLinkByMessageID(ctx context.Context, discordMessageID string) (DiscordPostLink, error) { + row := q.db.QueryRowContext(ctx, getDiscordPostLinkByMessageID, discordMessageID) + var i DiscordPostLink + err := row.Scan( + &i.PostID, + &i.DiscordMessageID, + &i.DiscordThreadID, + &i.CreatedAt, + ) + return i, err +} + +const getDiscordPostLinkByPostID = `-- name: GetDiscordPostLinkByPostID :one +SELECT post_id, discord_message_id, discord_thread_id, created_at +FROM discord_post_links +WHERE post_id = $1 +` + +func (q *Queries) GetDiscordPostLinkByPostID(ctx context.Context, postID string) (DiscordPostLink, error) { + row := q.db.QueryRowContext(ctx, getDiscordPostLinkByPostID, postID) + var i DiscordPostLink + err := row.Scan( + &i.PostID, + &i.DiscordMessageID, + &i.DiscordThreadID, + &i.CreatedAt, + ) + return i, err +} + +const getDiscordPostLinkByThreadID = `-- name: GetDiscordPostLinkByThreadID :one +SELECT post_id, discord_message_id, discord_thread_id, created_at +FROM discord_post_links +WHERE discord_thread_id = $1 + AND discord_thread_id <> '' +` + +func (q *Queries) GetDiscordPostLinkByThreadID(ctx context.Context, discordThreadID string) (DiscordPostLink, error) { + row := q.db.QueryRowContext(ctx, getDiscordPostLinkByThreadID, discordThreadID) + var i DiscordPostLink + err := row.Scan( + &i.PostID, + &i.DiscordMessageID, + &i.DiscordThreadID, + &i.CreatedAt, + ) + return i, err +} + +const upsertDiscordPostLink = `-- name: UpsertDiscordPostLink :exec +INSERT INTO discord_post_links ( + post_id, discord_message_id, discord_thread_id, created_at +) +VALUES ( + $1, + $2, + $3, + $4 +) +ON CONFLICT (post_id) DO UPDATE SET + discord_message_id = EXCLUDED.discord_message_id, + discord_thread_id = CASE + WHEN EXCLUDED.discord_thread_id <> '' THEN EXCLUDED.discord_thread_id + ELSE discord_post_links.discord_thread_id + END +` + +type UpsertDiscordPostLinkParams struct { + PostID string + DiscordMessageID string + DiscordThreadID string + CreatedAt string +} + +func (q *Queries) UpsertDiscordPostLink(ctx context.Context, arg UpsertDiscordPostLinkParams) error { + _, err := q.db.ExecContext(ctx, upsertDiscordPostLink, + arg.PostID, + arg.DiscordMessageID, + arg.DiscordThreadID, + arg.CreatedAt, + ) + return err +} diff --git a/internal/store/sqlc/models.go b/internal/store/sqlc/models.go index a9584f4..560d0ce 100644 --- a/internal/store/sqlc/models.go +++ b/internal/store/sqlc/models.go @@ -9,6 +9,13 @@ import ( "time" ) +type DiscordPostLink struct { + PostID string + DiscordMessageID string + DiscordThreadID string + CreatedAt string +} + type Post struct { ID string ParentID sql.NullString diff --git a/schema.sql b/schema.sql index a163643..4caac27 100644 --- a/schema.sql +++ b/schema.sql @@ -65,6 +65,17 @@ CREATE TABLE IF NOT EXISTS post_votes ( CREATE INDEX IF NOT EXISTS idx_post_votes_post_id ON post_votes(post_id); +CREATE TABLE IF NOT EXISTS discord_post_links ( + post_id TEXT PRIMARY KEY REFERENCES posts(id) ON DELETE CASCADE, + discord_message_id TEXT NOT NULL UNIQUE, + discord_thread_id TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL +); + +CREATE UNIQUE INDEX IF NOT EXISTS discord_post_links_thread_uidx + ON discord_post_links (discord_thread_id) + WHERE discord_thread_id <> ''; + CREATE TABLE IF NOT EXISTS sessions ( token TEXT PRIMARY KEY, data BYTEA NOT NULL, From 1acc296541336c3873ce2c92a377095ea78fca4f Mon Sep 17 00:00:00 2001 From: codegirl-007 Date: Sat, 29 Aug 2026 01:20:40 -0700 Subject: [PATCH 3/5] Add Discord inbound replies. --- cmd/server/main.go | 2 +- internal/discord/bot.go | 24 +++- internal/discord/bot_test.go | 6 +- internal/discord/inbound.go | 227 +++++++++++++++++++++++++++++++ internal/discord/inbound_test.go | 218 +++++++++++++++++++++++++++++ 5 files changed, 471 insertions(+), 6 deletions(-) create mode 100644 internal/discord/inbound.go create mode 100644 internal/discord/inbound_test.go diff --git a/cmd/server/main.go b/cmd/server/main.go index 83e6da0..7dc3658 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -37,7 +37,7 @@ func main() { log.Fatalf("mail: %v", err) } bus := events.New() - bot, err := discord.FromEnv(store.NewDiscordLinks(db), bus) + bot, err := discord.FromEnv(store.NewDiscordLinks(db), bus, store.NewPostgres(db), notifier) if err != nil { log.Fatalf("discord: %v", err) } diff --git a/internal/discord/bot.go b/internal/discord/bot.go index cc4bc19..9b04415 100644 --- a/internal/discord/bot.go +++ b/internal/discord/bot.go @@ -13,6 +13,7 @@ import ( "github.com/bwmarrin/discordgo" "plumber/internal/events" + "plumber/internal/mail" "plumber/internal/store" ) @@ -23,6 +24,10 @@ type Bot struct { channelID string links store.DiscordLinkStore api API + store store.Store + mail mail.Notifier + admins map[string]string + botUserID string } // New constructs an outbound subscriber. Tests inject a fake API. @@ -31,7 +36,7 @@ func New(channelID string, links store.DiscordLinkStore, api API) *Bot { } // FromEnv builds a bot when Discord env is set. Missing config is a no-op. -func FromEnv(links store.DiscordLinkStore, bus *events.Bus) (*Bot, error) { +func FromEnv(links store.DiscordLinkStore, bus *events.Bus, st store.Store, mailer mail.Notifier) (*Bot, error) { token := strings.TrimSpace(os.Getenv("DISCORD_BOT_TOKEN")) channelID := strings.TrimSpace(os.Getenv("DISCORD_CHANNEL_ID")) if token == "" && channelID == "" { @@ -50,11 +55,26 @@ func FromEnv(links store.DiscordLinkStore, bus *events.Bus) (*Bot, error) { if err != nil { return nil, err } + session.Identify.Intents = discordgo.IntentsGuilds | discordgo.IntentsGuildMessages | discordgo.IntentsMessageContent + if mailer == nil { + mailer = mail.Nop{} + } bot := New(channelID, links, &sessionAPI{session: session}) + bot.store = st + bot.mail = mailer + bot.admins = parseAdminMap(os.Getenv("DISCORD_ADMIN_MAP")) + session.AddHandler(bot.onMessageCreate) if bus != nil { bus.Subscribe(bot.Handle) } - log.Printf("discord: outbound subscriber enabled") + if err := session.Open(); err != nil { + _ = session.Close() + return nil, fmt.Errorf("discord gateway: %w", err) + } + if session.State != nil && session.State.User != nil { + bot.botUserID = session.State.User.ID + } + log.Printf("discord: subscriber enabled") return bot, nil } diff --git a/internal/discord/bot_test.go b/internal/discord/bot_test.go index 2c5e178..f6381db 100644 --- a/internal/discord/bot_test.go +++ b/internal/discord/bot_test.go @@ -204,7 +204,7 @@ func TestFormatMessage(t *testing.T) { func TestFromEnvDisabled(t *testing.T) { t.Setenv("DISCORD_BOT_TOKEN", "") t.Setenv("DISCORD_CHANNEL_ID", "") - bot, err := FromEnv(newMemoryLinks(), nil) + bot, err := FromEnv(newMemoryLinks(), nil, nil, nil) if err != nil || bot != nil { t.Fatalf("disabled FromEnv = (%v, %v)", bot, err) } @@ -213,12 +213,12 @@ func TestFromEnvDisabled(t *testing.T) { func TestFromEnvRequiresBoth(t *testing.T) { t.Setenv("DISCORD_BOT_TOKEN", "token") t.Setenv("DISCORD_CHANNEL_ID", "") - if _, err := FromEnv(newMemoryLinks(), nil); err == nil { + if _, err := FromEnv(newMemoryLinks(), nil, nil, nil); err == nil { t.Fatal("expected error when channel is missing") } t.Setenv("DISCORD_BOT_TOKEN", "") t.Setenv("DISCORD_CHANNEL_ID", "channel") - if _, err := FromEnv(newMemoryLinks(), nil); err == nil { + if _, err := FromEnv(newMemoryLinks(), nil, nil, nil); err == nil { t.Fatal("expected error when token is missing") } } diff --git a/internal/discord/inbound.go b/internal/discord/inbound.go new file mode 100644 index 0000000..c7414a7 --- /dev/null +++ b/internal/discord/inbound.go @@ -0,0 +1,227 @@ +package discord + +import ( + "context" + "database/sql" + "errors" + "fmt" + "log" + "strings" + "time" + + "github.com/bwmarrin/discordgo" + + "plumber/internal/mail" + "plumber/internal/store" +) + +const inboundBodyLimit = 12000 + +type inboundMessage struct { + ID string + ChannelID string + GuildID string + AuthorID string + Content string + ReferencedMessageID string + Bot bool + Attachments int +} + +func parseAdminMap(raw string) map[string]string { + out := map[string]string{} + for _, part := range strings.Split(raw, ",") { + part = strings.TrimSpace(part) + if part == "" { + continue + } + id, username, ok := strings.Cut(part, ":") + id = strings.TrimSpace(id) + username = store.NormalizeUsername(username) + if !ok || id == "" || username == "" { + log.Printf("discord: skip invalid DISCORD_ADMIN_MAP entry %q", part) + continue + } + out[id] = username + } + return out +} + +func (b *Bot) onMessageCreate(_ *discordgo.Session, m *discordgo.MessageCreate) { + if b == nil || m == nil || m.Author == nil { + return + } + in := inboundMessage{ + ID: m.ID, + ChannelID: m.ChannelID, + GuildID: m.GuildID, + AuthorID: m.Author.ID, + Content: m.Content, + Bot: m.Author.Bot, + Attachments: len(m.Attachments), + } + if m.MessageReference != nil { + in.ReferencedMessageID = m.MessageReference.MessageID + } + b.handleInbound(in) +} + +func (b *Bot) handleInbound(in inboundMessage) { + if b == nil || b.store == nil { + return + } + if in.Bot || strings.TrimSpace(in.GuildID) == "" { + return + } + if b.botUserID != "" && in.AuthorID == b.botUserID { + return + } + body := strings.TrimSpace(in.Content) + if in.Attachments > 0 { + log.Printf("discord: ignoring %d attachment(s) on %s", in.Attachments, in.ID) + } + if body == "" { + return + } + ctx, cancel := context.WithTimeout(context.Background(), discordTimeout) + defer cancel() + if !b.knownChannel(ctx, in.ChannelID) { + return + } + username := b.admins[in.AuthorID] + if username == "" { + return + } + author, err := b.store.UserByUsername(ctx, username) + if err != nil { + if !errors.Is(err, sql.ErrNoRows) { + log.Printf("discord: inbound author %s: %v", username, err) + } + return + } + if !author.Admin() { + log.Printf("discord: inbound %s is not an admin", username) + return + } + parent, root, err := b.inboundParent(ctx, in) + if err != nil { + if !errors.Is(err, sql.ErrNoRows) { + log.Printf("discord: inbound parent %s: %v", in.ID, err) + } + return + } + if root.PostState == store.PostStateHidden { + return + } + reply := &store.Post{ + AuthorID: author.ID, + Body: truncateRunes(body, inboundBodyLimit), + ParentID: &parent.ID, + } + if err := b.store.CreatePost(ctx, reply); err != nil { + log.Printf("discord: create inbound %s: %v", in.ID, err) + return + } + if err := b.links.Upsert(ctx, store.DiscordLink{ + PostID: reply.ID, + MessageID: in.ID, + }); err != nil { + log.Printf("discord: save inbound link %s: %v", reply.ID, err) + } + b.notifyInboundReply(parent, root, reply, author) + log.Printf("discord: inbound reply %s -> post %s", in.ID, reply.ID) +} + +func (b *Bot) knownChannel(ctx context.Context, channelID string) bool { + if strings.TrimSpace(channelID) == "" { + return false + } + if channelID == b.channelID { + return true + } + _, err := b.links.GetRootByThreadID(ctx, channelID) + return err == nil +} + +func (b *Bot) inboundParent(ctx context.Context, in inboundMessage) (*store.Post, *store.Post, error) { + if ref := strings.TrimSpace(in.ReferencedMessageID); ref != "" { + link, err := b.links.GetByMessageID(ctx, ref) + if err == nil { + return b.postAndRoot(ctx, link.PostID) + } + if !errors.Is(err, sql.ErrNoRows) { + return nil, nil, err + } + } + link, err := b.links.GetRootByThreadID(ctx, in.ChannelID) + if err != nil { + return nil, nil, err + } + return b.postAndRoot(ctx, link.PostID) +} + +func (b *Bot) postAndRoot(ctx context.Context, postID string) (*store.Post, *store.Post, error) { + postID = strings.TrimSpace(postID) + if postID == "" { + return nil, nil, sql.ErrNoRows + } + post, err := b.store.GetPost(ctx, postID) + if err != nil { + return nil, nil, err + } + current := post + seen := map[string]bool{} + for current.ParentID != nil { + if seen[current.ID] { + return nil, nil, fmt.Errorf("post ancestry cycle at %s", current.ID) + } + seen[current.ID] = true + current, err = b.store.GetPost(ctx, *current.ParentID) + if err != nil { + return nil, nil, err + } + } + return post, current, nil +} + +func (b *Bot) notifyInboundReply(parent, root, reply *store.Post, author *store.User) { + if parent == nil || root == nil || reply == nil || author == nil || b.mail == nil { + return + } + if _, disabled := b.mail.(mail.Nop); disabled { + return + } + recipientID := parent.AuthorID + if author.Admin() { + recipientID = root.AuthorID + } + if recipientID == author.ID { + return + } + msg := mail.PostReply{ + RootID: root.ID, + RootTitle: root.Title, + ReplyID: reply.ID, + ReplyBody: reply.Body, + ReplyAuthorName: author.Name, + } + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + recipient, err := b.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 := b.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/discord/inbound_test.go b/internal/discord/inbound_test.go new file mode 100644 index 0000000..53d925b --- /dev/null +++ b/internal/discord/inbound_test.go @@ -0,0 +1,218 @@ +package discord + +import ( + "context" + "testing" + "time" + + "plumber/internal/events" + "plumber/internal/mail" + "plumber/internal/pacific" + "plumber/internal/store" +) + +func TestParseAdminMap(t *testing.T) { + t.Parallel() + + got := parseAdminMap(" 123:Plumber ,456:other,bad, :empty,789: ") + if got["123"] != "plumber" || got["456"] != "other" || len(got) != 2 { + t.Fatalf("parseAdminMap = %#v", got) + } +} + +func TestInboundCreatesSiteReply(t *testing.T) { + t.Parallel() + + mem, homeowner, admin := seedInboundUsers(t) + links := newMemoryLinks() + api := &fakeAPI{} + mailer := &mail.Recording{} + bot := inboundTestBot(mem, links, api, mailer, admin.Username) + bus := events.New() + defer bus.Close() + bus.Subscribe(bot.Handle) + root := seedLinkedRoot(t, mem, links, homeowner.ID, "thread-1") + + bot.handleInbound(inboundMessage{ + ID: "d-reply-1", + ChannelID: "thread-1", + GuildID: "guild-1", + AuthorID: "snow-admin", + Content: "Replace the cartridge.", + }) + + thread, err := mem.GetPostThread(context.Background(), root.ID) + if err != nil || len(thread.Replies) != 1 { + t.Fatalf("thread = %+v, %v", thread, err) + } + reply := thread.Replies[0] + if reply.AuthorID != admin.ID || reply.Body != "Replace the cartridge." || reply.ParentID == nil || *reply.ParentID != root.ID { + t.Fatalf("reply = %+v", reply) + } + link, err := links.GetByPostID(context.Background(), reply.ID) + if err != nil || link.MessageID != "d-reply-1" || link.ThreadID != "" { + t.Fatalf("inbound link = %+v, %v", link, err) + } + time.Sleep(20 * time.Millisecond) + if len(api.sends) != 0 || len(api.edits) != 0 { + t.Fatalf("inbound echoed to Discord: sends=%+v edits=%+v", api.sends, api.edits) + } + msgs := waitForMail(t, mailer, 1) + if msgs[0].ToEmail != homeowner.Email || msgs[0].ReplyID != reply.ID || msgs[0].RootID != root.ID { + t.Fatalf("mail = %+v", msgs[0]) + } +} + +func TestInboundParentsFromReference(t *testing.T) { + t.Parallel() + + mem, homeowner, admin := seedInboundUsers(t) + links := newMemoryLinks() + bot := inboundTestBot(mem, links, &fakeAPI{}, &mail.Recording{}, admin.Username) + root := seedLinkedRoot(t, mem, links, homeowner.ID, "thread-1") + plumberReply := &store.Post{ParentID: &root.ID, AuthorID: admin.ID, Body: "First look."} + if err := mem.CreatePost(context.Background(), plumberReply); err != nil { + t.Fatal(err) + } + if err := links.Upsert(context.Background(), store.DiscordLink{ + PostID: plumberReply.ID, + MessageID: "d-plumber-1", + }); err != nil { + t.Fatal(err) + } + + bot.handleInbound(inboundMessage{ + ID: "d-nested", + ChannelID: "thread-1", + GuildID: "guild-1", + AuthorID: "snow-admin", + Content: "More detail.", + ReferencedMessageID: "d-plumber-1", + }) + + thread, err := mem.GetPostThread(context.Background(), root.ID) + if err != nil || len(thread.Replies) != 1 || len(thread.Replies[0].Replies) != 1 { + t.Fatalf("thread = %+v, %v", thread, err) + } + nested := thread.Replies[0].Replies[0] + if nested.ParentID == nil || *nested.ParentID != plumberReply.ID { + t.Fatalf("nested parent = %+v", nested) + } +} + +func TestInboundIgnoresAllowlistHiddenAndEchoSources(t *testing.T) { + t.Parallel() + + mem, homeowner, admin := seedInboundUsers(t) + links := newMemoryLinks() + api := &fakeAPI{} + bot := inboundTestBot(mem, links, api, mail.Nop{}, admin.Username) + root := seedLinkedRoot(t, mem, links, homeowner.ID, "thread-1") + hidden := &store.Post{ + AuthorID: homeowner.ID, + Title: "Hidden", + Body: "No.", + PostDate: pacific.Today(), + PostState: store.PostStateHidden, + } + if err := mem.CreatePost(context.Background(), hidden); err != nil { + t.Fatal(err) + } + if err := links.Upsert(context.Background(), store.DiscordLink{ + PostID: hidden.ID, + MessageID: "d-hidden", + ThreadID: "thread-hidden", + }); err != nil { + t.Fatal(err) + } + + cases := []inboundMessage{ + {ID: "bot", ChannelID: "thread-1", GuildID: "g", AuthorID: "snow-admin", Content: "x", Bot: true}, + {ID: "dm", ChannelID: "thread-1", AuthorID: "snow-admin", Content: "x"}, + {ID: "self", ChannelID: "thread-1", GuildID: "g", AuthorID: "bot-1", Content: "x"}, + {ID: "stranger", ChannelID: "thread-1", GuildID: "g", AuthorID: "snow-other", Content: "x"}, + {ID: "elsewhere", ChannelID: "other-thread", GuildID: "g", AuthorID: "snow-admin", Content: "x"}, + {ID: "empty", ChannelID: "thread-1", GuildID: "g", AuthorID: "snow-admin", Content: " ", Attachments: 1}, + {ID: "hidden", ChannelID: "thread-hidden", GuildID: "g", AuthorID: "snow-admin", Content: "x"}, + {ID: "channel-root", ChannelID: "channel-1", GuildID: "g", AuthorID: "snow-admin", Content: "new question"}, + } + for _, in := range cases { + bot.handleInbound(in) + } + thread, err := mem.GetPostThread(context.Background(), root.ID) + if err != nil || len(thread.Replies) != 0 { + t.Fatalf("unexpected replies: %+v, %v", thread, err) + } + if len(api.sends) != 0 { + t.Fatalf("unexpected discord sends %+v", api.sends) + } +} + +func inboundTestBot(mem *store.Memory, links *memoryLinks, api *fakeAPI, mailer mail.Notifier, adminUsername string) *Bot { + bot := New("channel-1", links, api) + bot.store = mem + bot.mail = mailer + bot.admins = map[string]string{"snow-admin": adminUsername} + bot.botUserID = "bot-1" + return bot +} + +func seedInboundUsers(t *testing.T) (*store.Memory, *store.User, *store.User) { + t.Helper() + mem := store.NewMemory() + homeowner := &store.User{ + Username: "homeowner", + Name: "Sam", + Email: "sam@example.com", + PasswordHash: "x", + Role: store.RoleUser, + } + if err := mem.CreateUser(context.Background(), homeowner); err != nil { + t.Fatal(err) + } + admin := &store.User{ + Username: "plumber", + Name: "Pat", + Email: "pat@example.com", + PasswordHash: "x", + Role: store.RoleAdmin, + } + if err := mem.CreateUser(context.Background(), admin); err != nil { + t.Fatal(err) + } + return mem, homeowner, admin +} + +func seedLinkedRoot(t *testing.T, mem *store.Memory, links *memoryLinks, authorID, threadID string) *store.Post { + t.Helper() + root := &store.Post{ + AuthorID: authorID, + Title: "Leaky sink", + Body: "It drips.", + PostDate: pacific.Today(), + } + if err := mem.CreatePost(context.Background(), root); err != nil { + t.Fatal(err) + } + if err := links.Upsert(context.Background(), store.DiscordLink{ + PostID: root.ID, + MessageID: "d-root", + ThreadID: threadID, + }); err != nil { + t.Fatal(err) + } + return root +} + +func waitForMail(t *testing.T, recording *mail.Recording, want int) []mail.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 +} From 573f54afd2d6101a54af05d550275f17f04beebc Mon Sep 17 00:00:00 2001 From: codegirl-007 Date: Sat, 29 Aug 2026 06:19:44 -0700 Subject: [PATCH 4/5] 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{} } From 8eddbfe4387ea6cb138140416b8669bbac44bf88 Mon Sep 17 00:00:00 2001 From: codegirl-007 Date: Sat, 29 Aug 2026 18:25:43 +0000 Subject: [PATCH 5/5] Post Discord questions as named threads (#17) Stacks on #16. Opens a public thread named "{author} asks: {title}" and puts the post body in the first thread message. Reviewed-on: https://git.codegirl007.com/codegirl007/plumber/pulls/17 Co-authored-by: codegirl-007 --- internal/discord/api.go | 12 ++++++---- internal/discord/bot.go | 14 ++++++------ internal/discord/bot_test.go | 42 ++++++++++++++++++++++------------ internal/discord/format.go | 44 ++++++++++++++++++++++++++++++++---- 4 files changed, 83 insertions(+), 29 deletions(-) diff --git a/internal/discord/api.go b/internal/discord/api.go index 58d0bda..ab964ef 100644 --- a/internal/discord/api.go +++ b/internal/discord/api.go @@ -9,7 +9,7 @@ import ( // API is the Discord REST surface used by the outbound subscriber. type API interface { SendToChannel(ctx context.Context, channelID string, msg Message) (messageID string, err error) - StartThread(ctx context.Context, channelID, messageID, name string) (threadID string, err error) + StartThread(ctx context.Context, channelID, name string) (threadID string, err error) SendToThread(ctx context.Context, threadID string, msg Message) (messageID string, err error) Edit(ctx context.Context, channelID, messageID string, msg Message) error Close() error @@ -27,9 +27,10 @@ func (s *sessionAPI) SendToChannel(_ context.Context, channelID string, msg Mess return sent.ID, nil } -func (s *sessionAPI) StartThread(_ context.Context, channelID, messageID, name string) (string, error) { - thread, err := s.session.MessageThreadStartComplex(channelID, messageID, &discordgo.ThreadStart{ +func (s *sessionAPI) StartThread(_ context.Context, channelID, name string) (string, error) { + thread, err := s.session.ThreadStartComplex(channelID, &discordgo.ThreadStart{ Name: name, + Type: discordgo.ChannelTypeGuildPublicThread, AutoArchiveDuration: 10080, }) if err != nil { @@ -43,10 +44,12 @@ func (s *sessionAPI) SendToThread(ctx context.Context, threadID string, msg Mess } func (s *sessionAPI) Edit(_ context.Context, channelID, messageID string, msg Message) error { + content := messageContent(msg) embeds := toEmbeds(msg) _, err := s.session.ChannelMessageEditComplex(&discordgo.MessageEdit{ ID: messageID, Channel: channelID, + Content: &content, Embeds: &embeds, }) return err @@ -61,6 +64,7 @@ func (s *sessionAPI) Close() error { func toMessageSend(msg Message) *discordgo.MessageSend { return &discordgo.MessageSend{ + Content: messageContent(msg), Embeds: toEmbeds(msg), AllowedMentions: &discordgo.MessageAllowedMentions{}, } @@ -69,7 +73,7 @@ func toMessageSend(msg Message) *discordgo.MessageSend { func toEmbeds(msg Message) []*discordgo.MessageEmbed { main := &discordgo.MessageEmbed{ Title: msg.Title, - URL: msg.URL, + URL: publicURL(msg.URL), Description: msg.Description, Color: embedColor, } diff --git a/internal/discord/bot.go b/internal/discord/bot.go index 9b04415..b127969 100644 --- a/internal/discord/bot.go +++ b/internal/discord/bot.go @@ -133,16 +133,16 @@ func (b *Bot) onUpdated(ctx context.Context, ev events.PostEvent) { func (b *Bot) createRoot(ctx context.Context, ev events.PostEvent) { msg := formatMessage(ev) - messageID, err := b.api.SendToChannel(ctx, b.channelID, msg) - if err != nil { - log.Printf("discord: send root %s: %v", ev.PostID, err) - return - } - threadID, err := b.api.StartThread(ctx, b.channelID, messageID, msg.ThreadName) + threadID, err := b.api.StartThread(ctx, b.channelID, msg.ThreadName) if err != nil { log.Printf("discord: start thread %s: %v", ev.PostID, err) return } + messageID, err := b.api.SendToThread(ctx, threadID, msg) + if err != nil { + log.Printf("discord: send root %s: %v", ev.PostID, err) + return + } if err := b.links.Upsert(ctx, store.DiscordLink{ PostID: ev.PostID, MessageID: messageID, @@ -185,7 +185,7 @@ func (b *Bot) createReply(ctx context.Context, ev events.PostEvent) { func (b *Bot) editChannel(ctx context.Context, ev events.PostEvent, link *store.DiscordLink) (string, error) { if strings.TrimSpace(link.ThreadID) != "" { - return b.channelID, nil + return link.ThreadID, nil } root, err := b.links.GetByPostID(ctx, ev.RootID) if err != nil { diff --git a/internal/discord/bot_test.go b/internal/discord/bot_test.go index f6381db..fde963b 100644 --- a/internal/discord/bot_test.go +++ b/internal/discord/bot_test.go @@ -2,11 +2,11 @@ package discord import ( "context" + "strconv" + "strings" "sync" "testing" - "strconv" - "plumber/internal/events" "plumber/internal/store" ) @@ -30,17 +30,18 @@ func (f *fakeAPI) SendToChannel(_ context.Context, channelID string, msg Message return f.record("channel", channelID, "", msg) } -func (f *fakeAPI) StartThread(_ context.Context, channelID, messageID, name string) (string, error) { +func (f *fakeAPI) StartThread(_ context.Context, channelID, name string) (string, error) { f.mu.Lock() defer f.mu.Unlock() f.next++ + id := "thread-" + strconv.Itoa(f.next) f.sends = append(f.sends, recordedSend{ Kind: "thread", ChannelID: channelID, Name: name, - Msg: Message{ThreadName: name, URL: messageID}, + Msg: Message{ThreadName: name}, }) - return "thread-" + messageID, nil + return id, nil } func (f *fakeAPI) SendToThread(_ context.Context, threadID string, msg Message) (string, error) { @@ -93,17 +94,21 @@ func TestOutboundRootReplyAndEdit(t *testing.T) { } bot.Handle(ctx, events.PostCreated{PostEvent: root}) - if len(api.sends) != 2 || api.sends[0].Kind != "channel" || api.sends[1].Kind != "thread" { + if len(api.sends) != 2 || + api.sends[0].Kind != "thread" || + api.sends[1].Kind != "thread-msg" { t.Fatalf("root sends = %+v", api.sends) } - if api.sends[0].ChannelID != "channel-1" || api.sends[1].Name != "Leaky sink" { + if api.sends[0].ChannelID != "channel-1" || + api.sends[0].Name != "sam asks: Leaky sink" || + api.sends[1].ChannelID != "thread-1" { t.Fatalf("root routing = %+v", api.sends) } - if got := api.sends[0].Msg.ImageURLs; len(got) != 2 || got[0] != "https://cdn.example/a.jpg" { + if got := api.sends[1].Msg.ImageURLs; len(got) != 2 || got[0] != "https://cdn.example/a.jpg" { t.Fatalf("root images = %v", got) } link, err := links.GetByPostID(ctx, "root-1") - if err != nil || link.MessageID != "msg-1" || link.ThreadID != "thread-msg-1" { + if err != nil || link.MessageID != "msg-2" || link.ThreadID != "thread-1" { t.Fatalf("root link = %+v, %v", link, err) } @@ -116,7 +121,7 @@ func TestOutboundRootReplyAndEdit(t *testing.T) { Permalink: "https://www.askaplumberfirst.com/questions/root-1#post-reply-1", } bot.Handle(ctx, events.PostCreated{PostEvent: reply}) - if len(api.sends) != 3 || api.sends[2].Kind != "thread-msg" || api.sends[2].ChannelID != "thread-msg-1" { + if len(api.sends) != 3 || api.sends[2].Kind != "thread-msg" || api.sends[2].ChannelID != "thread-1" { t.Fatalf("reply sends = %+v", api.sends) } replyLink, err := links.GetByPostID(ctx, "reply-1") @@ -126,7 +131,7 @@ func TestOutboundRootReplyAndEdit(t *testing.T) { root.Body = "Updated leak." bot.Handle(ctx, events.PostUpdated{PostEvent: root}) - if len(api.edits) != 1 || api.edits[0].ChannelID != "channel-1" || api.edits[0].Name != "msg-1" { + if len(api.edits) != 1 || api.edits[0].ChannelID != "thread-1" || api.edits[0].Name != "msg-2" { t.Fatalf("root edit = %+v", api.edits) } if api.edits[0].Msg.Description != "Updated leak." { @@ -135,7 +140,7 @@ func TestOutboundRootReplyAndEdit(t *testing.T) { reply.Body = "Use a ceramic cartridge." bot.Handle(ctx, events.PostUpdated{PostEvent: reply}) - if len(api.edits) != 2 || api.edits[1].ChannelID != "thread-msg-1" || api.edits[1].Name != "msg-3" { + if len(api.edits) != 2 || api.edits[1].ChannelID != "thread-1" || api.edits[1].Name != "msg-3" { t.Fatalf("reply edit = %+v", api.edits) } } @@ -190,15 +195,24 @@ func TestFormatMessage(t *testing.T) { got.City != "Oakland" || got.Author != "sam" || got.URL != "https://example.com/q" || - got.ThreadName != "Leaky sink" || + got.ThreadName != "sam asks: Leaky sink" || len(got.ImageURLs) != 1 { t.Fatalf("format = %+v", got) } reply := formatMessage(events.PostEvent{Body: "Thanks", AuthorName: ""}) - if reply.Title != "Reply" || reply.Author != "Someone" || reply.ThreadName != "Question" { + if reply.Title != "Reply" || reply.Author != "Someone" || reply.ThreadName != "Someone asks: Question" { t.Fatalf("reply format = %+v", reply) } + content := messageContent(got) + if strings.Contains(content, "Leaky sink") || + !strings.Contains(content, "It drips.") || + !strings.Contains(content, "Oakland") { + t.Fatalf("content = %q", content) + } + if publicURL("/questions/x") != "" || publicURL("http://localhost:8080/q") != "" { + t.Fatal("localhost or relative permalink should not be an embed URL") + } } func TestFromEnvDisabled(t *testing.T) { diff --git a/internal/discord/format.go b/internal/discord/format.go index f552a19..c14617a 100644 --- a/internal/discord/format.go +++ b/internal/discord/format.go @@ -39,7 +39,7 @@ func formatMessage(ev events.PostEvent) Message { Description: truncateRunes(strings.TrimSpace(ev.Body), embedDescriptionLimit), City: strings.TrimSpace(ev.City), Author: author, - ThreadName: threadName(ev.Title), + ThreadName: threadName(author, ev.Title), } for _, img := range ev.Images { url := strings.TrimSpace(img.URL) @@ -51,12 +51,16 @@ func formatMessage(ev events.PostEvent) Message { return msg } -func threadName(title string) string { +func threadName(author, title string) string { + author = strings.TrimSpace(author) + if author == "" { + author = "Someone" + } title = strings.TrimSpace(title) if title == "" { - return "Question" + title = "Question" } - return truncateRunes(title, threadNameLimit) + return truncateRunes(author+" asks: "+title, threadNameLimit) } func truncateRunes(s string, max int) string { @@ -73,3 +77,35 @@ func truncateRunes(s string, max int) string { func isRoot(ev events.PostEvent) bool { return strings.TrimSpace(ev.ParentID) == "" } + +func messageContent(msg Message) string { + var parts []string + if body := strings.TrimSpace(msg.Description); body != "" { + parts = append(parts, body) + } + var meta []string + if msg.City != "" { + meta = append(meta, msg.City) + } + if msg.Author != "" { + meta = append(meta, msg.Author) + } + if len(meta) > 0 { + parts = append(parts, strings.Join(meta, " ยท ")) + } + if u := publicURL(msg.URL); u != "" { + parts = append(parts, u) + } + return truncateRunes(strings.Join(parts, "\n"), 2000) +} + +func publicURL(raw string) string { + raw = strings.TrimSpace(raw) + if !strings.HasPrefix(raw, "https://") { + return "" + } + if strings.Contains(raw, "localhost") || strings.Contains(raw, "127.0.0.1") { + return "" + } + return raw +}