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..90a3c2c 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,16 @@ func main() { if err != nil { log.Fatalf("mail: %v", err) } - handler := newHandler(db, sessions, uploader, notifier, events.New()) + 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) + } + if bot != nil { + defer bot.Close() + } + handler := newHandler(db, sessions, uploader, bus) run(&http.Server{ Addr: listenAddr(), Handler: handler, @@ -59,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/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..ab964ef --- /dev/null +++ b/internal/discord/api.go @@ -0,0 +1,106 @@ +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, 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, name string) (string, error) { + thread, err := s.session.ThreadStartComplex(channelID, &discordgo.ThreadStart{ + Name: name, + Type: discordgo.ChannelTypeGuildPublicThread, + 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 { + content := messageContent(msg) + embeds := toEmbeds(msg) + _, err := s.session.ChannelMessageEditComplex(&discordgo.MessageEdit{ + ID: messageID, + Channel: channelID, + Content: &content, + 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{ + Content: messageContent(msg), + Embeds: toEmbeds(msg), + AllowedMentions: &discordgo.MessageAllowedMentions{}, + } +} + +func toEmbeds(msg Message) []*discordgo.MessageEmbed { + main := &discordgo.MessageEmbed{ + Title: msg.Title, + URL: publicURL(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..b127969 --- /dev/null +++ b/internal/discord/bot.go @@ -0,0 +1,198 @@ +package discord + +import ( + "context" + "database/sql" + "errors" + "fmt" + "log" + "os" + "strings" + "time" + + "github.com/bwmarrin/discordgo" + + "plumber/internal/events" + "plumber/internal/mail" + "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 + store store.Store + mail mail.Notifier + admins map[string]string + botUserID string +} + +// 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, 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 == "" { + 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 + } + 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) + } + 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 +} + +// 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) + 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, + 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 link.ThreadID, 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..fde963b --- /dev/null +++ b/internal/discord/bot_test.go @@ -0,0 +1,261 @@ +package discord + +import ( + "context" + "strconv" + "strings" + "sync" + "testing" + + "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, 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}, + }) + return id, 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 != "thread" || + api.sends[1].Kind != "thread-msg" { + t.Fatalf("root sends = %+v", api.sends) + } + 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[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-2" || link.ThreadID != "thread-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-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 != "thread-1" || api.edits[0].Name != "msg-2" { + 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-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 != "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 != "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) { + t.Setenv("DISCORD_BOT_TOKEN", "") + t.Setenv("DISCORD_CHANNEL_ID", "") + bot, err := FromEnv(newMemoryLinks(), nil, nil, 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, 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, nil, 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..c14617a --- /dev/null +++ b/internal/discord/format.go @@ -0,0 +1,111 @@ +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(author, 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(author, title string) string { + author = strings.TrimSpace(author) + if author == "" { + author = "Someone" + } + title = strings.TrimSpace(title) + if title == "" { + title = "Question" + } + return truncateRunes(author+" asks: "+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) == "" +} + +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 +} 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 +} 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/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/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/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{} } 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,