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) }