Add Discord outbound posting.

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

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

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

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) == ""
+}

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