From f96df3222df1270a5555abd5868d03a0994769fa Mon Sep 17 00:00:00 2001 From: codegirl-007 Date: Mon, 31 Aug 2026 02:00:56 -0700 Subject: [PATCH] Add video picker and Discord video attachments. The picker accepts one MP4 or WebM, the thread lightbox zooms photos, and Discord gets the video as a file on a follow-up message so photo embeds still render. --- cmd/server/main.go | 18 ++- internal/discord/api.go | 29 +++- internal/discord/bot.go | 101 +++++++++--- internal/discord/bot_test.go | 132 +++++++++++++++- internal/discord/format.go | 76 ++++++++- internal/events/bus.go | 2 + internal/events/bus_test.go | 41 +++++ internal/events/event.go | 24 ++- internal/events/nop.go | 6 +- internal/events/release.go | 41 +++++ internal/web/events.go | 26 +++- internal/web/events_test.go | 64 +++++++- internal/web/post_images.go | 89 +++++++---- internal/web/post_images_test.go | 13 ++ internal/web/posts.go | 18 ++- internal/web/posts_test.go | 10 +- internal/web/server.go | 42 ++++- internal/web/server_test.go | 6 +- static/app.css | 123 +++++++++++++++ static/app.js | 212 +++++++++++++++++++++----- templates/base.html | 7 + templates/partials/_image_picker.html | 82 ++++++++-- 22 files changed, 1018 insertions(+), 144 deletions(-) create mode 100644 internal/events/release.go diff --git a/cmd/server/main.go b/cmd/server/main.go index ce8efdb..8de0d6f 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -38,6 +38,7 @@ func main() { } bus := events.New() mail.Subscribe(bus, store.NewPostgres(db), notifier) + events.SubscribeRelease(bus) bot, err := discord.FromEnv(store.NewDiscordLinks(db), bus, store.NewPostgres(db)) if err != nil { log.Fatalf("discord: %v", err) @@ -45,7 +46,7 @@ func main() { if bot != nil { defer bot.Close() } - handler := newHandler(db, sessions, uploader, bus) + handler := newHandler(db, sessions, uploader, bus, bot != nil) run(&http.Server{ Addr: listenAddr(), Handler: handler, @@ -69,14 +70,15 @@ func openDB() (*sql.DB, *store.SessionStore) { return db, sessions } -func newHandler(db *sql.DB, sessions *store.SessionStore, uploader blob.Uploader, bus events.Publisher) http.Handler { +func newHandler(db *sql.DB, sessions *store.SessionStore, uploader blob.Uploader, bus events.Publisher, holdUploads bool) 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, - Events: bus, - BaseURL: strings.TrimRight(strings.TrimSpace(os.Getenv("APP_BASE_URL")), "/"), + AdminSetupSecret: strings.TrimSpace(os.Getenv("ADMIN_SETUP_SECRET")), + SecureCookie: secureCookieFromEnv(), + TrustedProxies: parseTrustedProxies(os.Getenv("TRUSTED_PROXY_CIDRS")), + Blob: uploader, + Events: bus, + BaseURL: strings.TrimRight(strings.TrimSpace(os.Getenv("APP_BASE_URL")), "/"), + HoldUploadUntilDiscord: holdUploads, }) if err != nil { log.Fatalf("server: %v", err) diff --git a/internal/discord/api.go b/internal/discord/api.go index ab964ef..3b4e3cc 100644 --- a/internal/discord/api.go +++ b/internal/discord/api.go @@ -2,6 +2,7 @@ package discord import ( "context" + "strings" "github.com/bwmarrin/discordgo" ) @@ -63,14 +64,27 @@ func (s *sessionAPI) Close() error { } func toMessageSend(msg Message) *discordgo.MessageSend { + var files []*discordgo.File + for _, f := range msg.Files { + files = append(files, &discordgo.File{ + Name: f.Name, + ContentType: f.ContentType, + Reader: f.Reader, + }) + } return &discordgo.MessageSend{ Content: messageContent(msg), Embeds: toEmbeds(msg), + Files: files, AllowedMentions: &discordgo.MessageAllowedMentions{}, } } func toEmbeds(msg Message) []*discordgo.MessageEmbed { + imageURLs := embedImageURLs(msg) + if msg.Title == "" && msg.Description == "" && msg.City == "" && msg.Author == "" && len(imageURLs) == 0 { + return nil + } main := &discordgo.MessageEmbed{ Title: msg.Title, URL: publicURL(msg.URL), @@ -92,7 +106,7 @@ func toEmbeds(msg Message) []*discordgo.MessageEmbed { }) } embeds := []*discordgo.MessageEmbed{main} - for i, url := range msg.ImageURLs { + for i, url := range imageURLs { if i == 0 { main.Image = &discordgo.MessageEmbedImage{URL: url} continue @@ -104,3 +118,16 @@ func toEmbeds(msg Message) []*discordgo.MessageEmbed { } return embeds } + +func embedImageURLs(msg Message) []string { + var attached []string + for _, f := range msg.Files { + if strings.HasPrefix(f.ContentType, "image/") { + attached = append(attached, "attachment://"+f.Name) + } + } + if len(attached) > 0 { + return attached + } + return msg.ImageURLs +} diff --git a/internal/discord/bot.go b/internal/discord/bot.go index 1dd7342..b862102 100644 --- a/internal/discord/bot.go +++ b/internal/discord/bot.go @@ -16,7 +16,7 @@ import ( "plumber/internal/store" ) -const discordTimeout = 15 * time.Second +const discordTimeout = 60 * time.Second // Bot posts site events to a Discord channel and owns post-to-message links. type Bot struct { @@ -91,13 +91,27 @@ func (b *Bot) Handle(_ context.Context, ev any) { defer cancel() switch e := ev.(type) { case events.PostCreated: - b.onCreated(ctx, e.PostEvent) + defer b.publishPosted(e) + b.onCreated(ctx, e) case events.PostUpdated: b.onUpdated(ctx, e.PostEvent) } } -func (b *Bot) onCreated(ctx context.Context, ev events.PostEvent) { +func (b *Bot) publishPosted(e events.PostCreated) { + if e.Release == nil { + return + } + done := events.PostedToDiscord{PostID: e.PostID, Release: e.Release} + if b.bus == nil { + e.Release() + return + } + b.bus.Publish(context.Background(), done) +} + +func (b *Bot) onCreated(ctx context.Context, e events.PostCreated) { + ev := e.PostEvent _, err := b.links.GetByPostID(ctx, ev.PostID) if err == nil { return @@ -107,17 +121,17 @@ func (b *Bot) onCreated(ctx context.Context, ev events.PostEvent) { return } if isRoot(ev) { - b.createRoot(ctx, ev) + b.createRoot(ctx, e) return } - b.createReply(ctx, ev) + b.createReply(ctx, e) } 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) + b.onCreated(ctx, events.PostCreated{PostEvent: ev}) return } log.Printf("discord: load link %s: %v", ev.PostID, err) @@ -135,56 +149,95 @@ func (b *Bot) onUpdated(ctx context.Context, ev events.PostEvent) { log.Printf("discord: edited %s", ev.PostID) } -func (b *Bot) createRoot(ctx context.Context, ev events.PostEvent) { - msg := formatMessage(ev) +func (b *Bot) createRoot(ctx context.Context, e events.PostCreated) { + msg := formatCreated(e) threadID, err := b.api.StartThread(ctx, b.channelID, msg.ThreadName) if err != nil { - log.Printf("discord: start thread %s: %v", ev.PostID, err) + log.Printf("discord: start thread %s: %v", e.PostID, err) return } - messageID, err := b.api.SendToThread(ctx, threadID, msg) + messageID, err := b.sendCreated(ctx, threadID, msg) if err != nil { - log.Printf("discord: send root %s: %v", ev.PostID, err) + log.Printf("discord: send root %s: %v", e.PostID, err) return } if err := b.links.Upsert(ctx, store.DiscordLink{ - PostID: ev.PostID, + PostID: e.PostID, MessageID: messageID, ThreadID: threadID, }); err != nil { - log.Printf("discord: save root link %s: %v", ev.PostID, err) + log.Printf("discord: save root link %s: %v", e.PostID, err) return } - log.Printf("discord: posted root %s", ev.PostID) + log.Printf("discord: posted root %s", e.PostID) } -func (b *Bot) createReply(ctx context.Context, ev events.PostEvent) { - root, err := b.links.GetByPostID(ctx, ev.RootID) +func (b *Bot) createReply(ctx context.Context, e events.PostCreated) { + root, err := b.links.GetByPostID(ctx, e.RootID) if err != nil { if errors.Is(err, sql.ErrNoRows) { - log.Printf("discord: skip reply %s: no root thread", ev.PostID) + log.Printf("discord: skip reply %s: no root thread", e.PostID) return } - log.Printf("discord: load root link %s: %v", ev.RootID, err) + log.Printf("discord: load root link %s: %v", e.RootID, err) return } if strings.TrimSpace(root.ThreadID) == "" { - log.Printf("discord: skip reply %s: no root thread", ev.PostID) + log.Printf("discord: skip reply %s: no root thread", e.PostID) return } - messageID, err := b.api.SendToThread(ctx, root.ThreadID, formatMessage(ev)) + messageID, err := b.sendCreated(ctx, root.ThreadID, formatCreated(e)) if err != nil { - log.Printf("discord: send reply %s: %v", ev.PostID, err) + log.Printf("discord: send reply %s: %v", e.PostID, err) return } if err := b.links.Upsert(ctx, store.DiscordLink{ - PostID: ev.PostID, + PostID: e.PostID, MessageID: messageID, }); err != nil { - log.Printf("discord: save reply link %s: %v", ev.PostID, err) + log.Printf("discord: save reply link %s: %v", e.PostID, err) return } - log.Printf("discord: posted reply %s", ev.PostID) + log.Printf("discord: posted reply %s", e.PostID) +} + +func (b *Bot) sendCreated(ctx context.Context, threadID string, msg Message) (string, error) { + _, videos := splitAttachments(msg.Files) + msg.Files = nil + if len(videos) > 0 { + msg.VideoURLs = nil + } + messageID, err := b.send(ctx, threadID, msg) + if err != nil { + return "", err + } + for _, video := range videos { + if _, err := b.send(ctx, threadID, Message{Files: []Attachment{video}}); err != nil { + log.Printf("discord: send video %s: %v", threadID, err) + } + } + return messageID, nil +} + +func splitAttachments(files []Attachment) (images, videos []Attachment) { + for _, f := range files { + if strings.HasPrefix(f.ContentType, "video/") || f.Kind == "video" { + videos = append(videos, f) + continue + } + images = append(images, f) + } + return images, videos +} + +func (b *Bot) send(ctx context.Context, threadID string, msg Message) (string, error) { + messageID, err := b.api.SendToThread(ctx, threadID, msg) + if err != nil && len(msg.Files) > 0 { + log.Printf("discord: send with files %s: %v; retrying without files", threadID, err) + msg.Files = nil + return b.api.SendToThread(ctx, threadID, msg) + } + return messageID, err } func (b *Bot) editChannel(ctx context.Context, ev events.PostEvent, link *store.DiscordLink) (string, error) { diff --git a/internal/discord/bot_test.go b/internal/discord/bot_test.go index 1e34a72..b37443b 100644 --- a/internal/discord/bot_test.go +++ b/internal/discord/bot_test.go @@ -6,6 +6,7 @@ import ( "strings" "sync" "testing" + "time" "plumber/internal/events" "plumber/internal/store" @@ -209,6 +210,125 @@ func TestOutboundUpdateWithoutLinkCreates(t *testing.T) { } } +func TestOutboundAttachesVideoFile(t *testing.T) { + t.Parallel() + + bus := events.New() + defer bus.Close() + events.SubscribeRelease(bus) + links := newMemoryLinks() + api := &fakeAPI{} + bot := New("channel-1", links, api) + bot.bus = bus + + released := make(chan struct{}) + bot.Handle(context.Background(), events.PostCreated{ + PostEvent: events.PostEvent{ + PostID: "root-1", + RootID: "root-1", + Title: "Valve clip", + Body: "Watch the handle.", + AuthorName: "sam", + Permalink: "https://www.askaplumberfirst.com/questions/root-1#post-root-1", + Images: []events.Image{{URL: "https://cdn.example/clip.mp4", Kind: "video"}}, + }, + Media: []events.Media{{ + Name: "clip.mp4", + ContentType: "video/mp4", + Kind: "video", + Bytes: []byte("fake-mp4"), + }}, + Release: func() { close(released) }, + }) + + if len(api.sends) != 3 || + api.sends[1].Kind != "thread-msg" || + api.sends[2].Kind != "thread-msg" { + t.Fatalf("sends = %+v", api.sends) + } + if len(api.sends[1].Msg.Files) != 0 { + t.Fatalf("text message files = %+v", api.sends[1].Msg.Files) + } + files := api.sends[2].Msg.Files + if len(files) != 1 || files[0].Name != "clip.mp4" || files[0].ContentType != "video/mp4" { + t.Fatalf("video files = %+v", files) + } + content := messageContent(api.sends[1].Msg) + if strings.Contains(content, "https://cdn.example/clip.mp4") { + t.Fatalf("content still has video URL: %q", content) + } + select { + case <-released: + case <-time.After(time.Second): + t.Fatal("PostedToDiscord did not release") + } +} + +func TestOutboundPhotosThenVideo(t *testing.T) { + t.Parallel() + + api := &fakeAPI{} + bot := New("channel-1", newMemoryLinks(), api) + bot.Handle(context.Background(), events.PostCreated{ + PostEvent: events.PostEvent{ + PostID: "root-1", + RootID: "root-1", + Title: "Valve clip", + Body: "Photo and video.", + AuthorName: "sam", + Images: []events.Image{ + {URL: "https://cdn.example/a.jpg"}, + {URL: "https://cdn.example/clip.mp4", Kind: "video"}, + }, + }, + Media: []events.Media{ + {Name: "still.jpg", ContentType: "image/jpeg", Kind: "image", Bytes: []byte("jpeg")}, + {Name: "clip.mp4", ContentType: "video/mp4", Kind: "video", Bytes: []byte("mp4")}, + }, + }) + if len(api.sends) != 3 || + api.sends[1].Kind != "thread-msg" || + api.sends[2].Kind != "thread-msg" { + t.Fatalf("sends = %+v", api.sends) + } + photos := api.sends[1].Msg.Files + if len(photos) != 0 { + t.Fatalf("text message should not attach files: %+v", photos) + } + if got := api.sends[1].Msg.ImageURLs; len(got) != 1 || got[0] != "https://cdn.example/a.jpg" { + t.Fatalf("text message images = %v", got) + } + if embeds := toEmbeds(api.sends[1].Msg); len(embeds) == 0 || embeds[0].Image == nil || embeds[0].Image.URL != "https://cdn.example/a.jpg" { + t.Fatalf("text embeds = %+v", toEmbeds(api.sends[1].Msg)) + } + videos := api.sends[2].Msg.Files + if len(videos) != 1 || videos[0].Name != "clip.mp4" { + t.Fatalf("video files = %+v", videos) + } +} + +func TestOutboundReleasesWhenAlreadyLinked(t *testing.T) { + t.Parallel() + + links := newMemoryLinks() + if err := links.Upsert(context.Background(), store.DiscordLink{ + PostID: "root-1", + MessageID: "d-root", + ThreadID: "thread-1", + }); err != nil { + t.Fatal(err) + } + released := false + bot := New("channel-1", links, &fakeAPI{}) + bot.Handle(context.Background(), events.PostCreated{ + PostEvent: events.PostEvent{PostID: "root-1", RootID: "root-1", Title: "Already posted"}, + Release: func() { released = true }, + }) + if !released { + t.Fatal("skipped send did not release held upload") + } +} + func TestFormatMessage(t *testing.T) { t.Parallel() @@ -218,7 +338,10 @@ func TestFormatMessage(t *testing.T) { City: "Oakland", AuthorName: "sam", Permalink: "https://example.com/q", - Images: []events.Image{{URL: "https://cdn.example/a.jpg", Description: "ignored"}}, + Images: []events.Image{ + {URL: "https://cdn.example/a.jpg", Description: "ignored"}, + {URL: "https://cdn.example/clip.mp4", Kind: "video"}, + }, }) if got.Title != "Leaky sink" || got.Description != "It drips." || @@ -226,7 +349,8 @@ func TestFormatMessage(t *testing.T) { got.Author != "sam" || got.URL != "https://example.com/q" || got.ThreadName != "sam asks: Leaky sink" || - len(got.ImageURLs) != 1 { + len(got.ImageURLs) != 1 || got.ImageURLs[0] != "https://cdn.example/a.jpg" || + len(got.VideoURLs) != 1 || got.VideoURLs[0] != "https://cdn.example/clip.mp4" { t.Fatalf("format = %+v", got) } @@ -237,7 +361,9 @@ func TestFormatMessage(t *testing.T) { content := messageContent(got) if strings.Contains(content, "Leaky sink") || !strings.Contains(content, "It drips.") || - !strings.Contains(content, "Oakland") { + !strings.Contains(content, "Oakland") || + !strings.Contains(content, "https://cdn.example/a.jpg") || + !strings.Contains(content, "https://cdn.example/clip.mp4") { t.Fatalf("content = %q", content) } if publicURL("/questions/x") != "" || publicURL("http://localhost:8080/q") != "" { diff --git a/internal/discord/format.go b/internal/discord/format.go index c14617a..6fb1161 100644 --- a/internal/discord/format.go +++ b/internal/discord/format.go @@ -1,6 +1,8 @@ package discord import ( + "bytes" + "io" "strings" "plumber/internal/events" @@ -13,6 +15,14 @@ const ( embedColor = 0xe96a26 ) +// Attachment is a local file Discord should upload with the message. +type Attachment struct { + Name string + ContentType string + Kind string + Reader io.Reader +} + // Message is a Discord-ready snapshot of a site post event. type Message struct { Title string @@ -21,36 +31,76 @@ type Message struct { City string Author string ImageURLs []string + VideoURLs []string + Files []Attachment ThreadName string } func formatMessage(ev events.PostEvent) Message { - title := strings.TrimSpace(ev.Title) + return formatCreated(events.PostCreated{PostEvent: ev}) +} + +func formatCreated(e events.PostCreated) Message { + title := strings.TrimSpace(e.Title) if title == "" { title = "Reply" } - author := strings.TrimSpace(ev.AuthorName) + author := strings.TrimSpace(e.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), + URL: strings.TrimSpace(e.Permalink), + Description: truncateRunes(strings.TrimSpace(e.Body), embedDescriptionLimit), + City: strings.TrimSpace(e.City), Author: author, - ThreadName: threadName(author, ev.Title), + ThreadName: threadName(author, e.Title), + Files: filesFromMedia(e.Media), } - for _, img := range ev.Images { + for _, img := range e.Images { url := strings.TrimSpace(img.URL) if url == "" { continue } + if img.Kind == "video" { + msg.VideoURLs = append(msg.VideoURLs, url) + continue + } msg.ImageURLs = append(msg.ImageURLs, url) } return msg } +func filesFromMedia(media []events.Media) []Attachment { + if len(media) == 0 { + return nil + } + files := make([]Attachment, 0, len(media)) + for _, m := range media { + var r io.Reader + switch { + case len(m.Bytes) > 0: + r = bytes.NewReader(m.Bytes) + case m.Body != nil: + r = m.Body + default: + continue + } + name := strings.TrimSpace(m.Name) + if name == "" { + name = "upload" + } + files = append(files, Attachment{ + Name: name, + ContentType: m.ContentType, + Kind: m.Kind, + Reader: r, + }) + } + return files +} + func threadName(author, title string) string { author = strings.TrimSpace(author) if author == "" { @@ -96,6 +146,18 @@ func messageContent(msg Message) string { if u := publicURL(msg.URL); u != "" { parts = append(parts, u) } + if len(msg.Files) == 0 { + for _, mediaURL := range msg.ImageURLs { + if u := publicURL(mediaURL); u != "" { + parts = append(parts, u) + } + } + for _, mediaURL := range msg.VideoURLs { + if u := publicURL(mediaURL); u != "" { + parts = append(parts, u) + } + } + } return truncateRunes(strings.Join(parts, "\n"), 2000) } diff --git a/internal/events/bus.go b/internal/events/bus.go index 7f1dad2..b36cf06 100644 --- a/internal/events/bus.go +++ b/internal/events/bus.go @@ -40,12 +40,14 @@ func newBus(buffer int, start bool) *Bus { // Publish enqueues ev. It never blocks the caller; a full buffer is dropped. func (b *Bus) Publish(_ context.Context, ev any) { if b == nil { + CallRelease(ev) return } select { case b.ch <- ev: default: log.Printf("events: dropped %T", ev) + CallRelease(ev) } } diff --git a/internal/events/bus_test.go b/internal/events/bus_test.go index 80c9275..158a83c 100644 --- a/internal/events/bus_test.go +++ b/internal/events/bus_test.go @@ -102,3 +102,44 @@ func TestBusDropsWhenFull(t *testing.T) { default: } } + +func TestNopReleasesPostCreated(t *testing.T) { + t.Parallel() + + released := false + Nop{}.Publish(context.Background(), PostCreated{Release: func() { released = true }}) + if !released { + t.Fatal("Nop.Publish did not release held upload") + } +} + +func TestBusDropReleasesPostCreated(t *testing.T) { + t.Parallel() + + bus := newBus(1, false) + bus.Publish(context.Background(), "kept") + released := false + bus.Publish(context.Background(), PostCreated{Release: func() { released = true }}) + if !released { + t.Fatal("dropped PostCreated did not release held upload") + } +} + +func TestSubscribeReleasePostedToDiscord(t *testing.T) { + t.Parallel() + + bus := New() + defer bus.Close() + SubscribeRelease(bus) + + released := make(chan struct{}) + bus.Publish(context.Background(), PostedToDiscord{ + PostID: "post-1", + Release: func() { close(released) }, + }) + select { + case <-released: + case <-time.After(time.Second): + t.Fatal("PostedToDiscord did not release") + } +} diff --git a/internal/events/event.go b/internal/events/event.go index ce5d1f2..0ea1dde 100644 --- a/internal/events/event.go +++ b/internal/events/event.go @@ -1,14 +1,27 @@ package events import ( + "io" "net/url" "strings" ) -// Image is a public photo already attached to a site post. +// Image is a public photo or video already attached to a site post. type Image struct { URL string Description string + Kind string +} + +// Media is a local upload still held from the create request so Discord can +// attach the bytes. Photos carry re-encoded Bytes; videos carry an open Body. +type Media struct { + Name string + ContentType string + Kind string + Size int64 + Body io.ReadCloser + Bytes []byte } // PostEvent is a Discord-free snapshot of a site post after a successful write. @@ -29,6 +42,8 @@ type PostEvent struct { // PostCreated is emitted after a successful site create. type PostCreated struct { PostEvent + Media []Media `json:"-"` + Release func() `json:"-"` } // PostUpdated is emitted after a successful site edit. @@ -36,6 +51,13 @@ type PostUpdated struct { PostEvent } +// PostedToDiscord is emitted after the Discord subscriber finishes with a +// PostCreated (sent, skipped, or failed). Release deletes held upload files. +type PostedToDiscord struct { + PostID string + Release func() `json:"-"` +} + // 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) diff --git a/internal/events/nop.go b/internal/events/nop.go index c71b595..246b640 100644 --- a/internal/events/nop.go +++ b/internal/events/nop.go @@ -5,5 +5,7 @@ import "context" // Nop is a Publisher used when nothing is subscribed. type Nop struct{} -// Publish discards ev. -func (Nop) Publish(context.Context, any) {} +// Publish discards ev after releasing any held upload. +func (Nop) Publish(_ context.Context, ev any) { + CallRelease(ev) +} diff --git a/internal/events/release.go b/internal/events/release.go new file mode 100644 index 0000000..57efbe9 --- /dev/null +++ b/internal/events/release.go @@ -0,0 +1,41 @@ +package events + +import ( + "context" + "sync" +) + +// Once returns fn wrapped so it runs at most once. +func Once(fn func()) func() { + if fn == nil { + return func() {} + } + var once sync.Once + return func() { once.Do(fn) } +} + +// CallRelease runs Release on PostCreated or PostedToDiscord when set. +func CallRelease(ev any) { + switch e := ev.(type) { + case PostCreated: + if e.Release != nil { + e.Release() + } + case PostedToDiscord: + if e.Release != nil { + e.Release() + } + } +} + +// SubscribeRelease runs PostedToDiscord.Release on the worker. +func SubscribeRelease(bus *Bus) { + if bus == nil { + return + } + bus.Subscribe(func(_ context.Context, ev any) { + if _, ok := ev.(PostedToDiscord); ok { + CallRelease(ev) + } + }) +} diff --git a/internal/web/events.go b/internal/web/events.go index 0678b26..6b9cf84 100644 --- a/internal/web/events.go +++ b/internal/web/events.go @@ -7,8 +7,26 @@ import ( "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) publishPostCreated(post, root *store.Post, author *store.User, media []events.Media, cleanup func()) { + release := events.Once(func() { + closeHeldMedia(media) + if cleanup != nil { + cleanup() + } + }) + if root != nil && root.PostState == store.PostStateHidden { + release() + return + } + ev := events.PostCreated{PostEvent: s.postEvent(post, root, author)} + if s.cfg.HoldUploadUntilDiscord && len(media) > 0 { + ev.Media = media + ev.Release = release + s.cfg.Events.Publish(context.Background(), ev) + return + } + release() + s.cfg.Events.Publish(context.Background(), events.PostCreated{PostEvent: ev.PostEvent}) } func (s *Server) publishPostUpdated(post, root *store.Post, author *store.User) { @@ -55,12 +73,10 @@ func (s *Server) postEvent(post, root *store.Post, author *store.User) events.Po if n := len(post.Images); n > 0 { ev.Images = make([]events.Image, 0, n) for _, img := range post.Images { - if img.Kind == store.MediaKindVideo { - continue - } ev.Images = append(ev.Images, events.Image{ URL: img.PublicURL, Description: img.Description, + Kind: img.Kind, }) } } diff --git a/internal/web/events_test.go b/internal/web/events_test.go index 6c2d194..061cbbc 100644 --- a/internal/web/events_test.go +++ b/internal/web/events_test.go @@ -1,8 +1,10 @@ package web import ( + "bytes" "context" "encoding/json" + "io" "net/http" "net/url" "strings" @@ -237,20 +239,72 @@ func assertPostEvent(t *testing.T, got, want events.PostEvent) { } } -func TestPostEventOmitsVideos(t *testing.T) { +func TestPostEventIncludesPhotosAndVideos(t *testing.T) { t.Parallel() srv, _ := newTestServer(t, Config{}) got := srv.postEvent(&store.Post{ - ID: "root-1", - Title: "Clip", - Body: "Photo and video.", + ID: "root-1", + Title: "Clip", + Body: "Photo and video.", Images: []store.PostImage{ {PublicURL: "https://cdn.example/a.jpg", Description: "Still", Kind: store.MediaKindImage}, {PublicURL: "https://cdn.example/a.mp4", Description: "Walkthrough", Kind: store.MediaKindVideo}, }, }, nil, nil) - if len(got.Images) != 1 || got.Images[0].URL != "https://cdn.example/a.jpg" { + if len(got.Images) != 2 || + got.Images[0].URL != "https://cdn.example/a.jpg" || got.Images[0].Kind != store.MediaKindImage || + got.Images[1].URL != "https://cdn.example/a.mp4" || got.Images[1].Kind != store.MediaKindVideo { t.Fatalf("event images = %+v", got.Images) } } + +func TestCreatePostHoldsVideoUntilRelease(t *testing.T) { + t.Parallel() + + rec := &events.Recording{} + blobs := &recordingImageBlob{} + srv, mem := newTestServer(t, Config{ + Events: rec, + Blob: blobs, + HoldUploadUntilDiscord: true, + }) + handler := srv.Handler() + user := seedUser(t, mem, uniq("hold-video"), "hunter22", store.RoleUser) + cookies := loginUser(t, handler, user.Username, "hunter22") + csrf := csrfForCookies(t, handler, cookies) + clip := tinyMP4() + + res := multipartPost(t, handler, "/submit", map[string][]string{ + "_csrf": {csrf}, + "title": {"Valve clip"}, + "body": {"Watch the handle."}, + "city": {"Oakland"}, + }, []multipartTestFile{{name: "walk.mp4", body: clip}}, cookies) + if res.Code != http.StatusSeeOther { + t.Fatalf("submit status = %d: %s", res.Code, res.Body.String()) + } + + got := rec.Snapshot() + if len(got) != 1 { + t.Fatalf("published %d events, want 1: %#v", len(got), got) + } + created, ok := got[0].(events.PostCreated) + if !ok { + t.Fatalf("event %T, want PostCreated", got[0]) + } + if len(created.Media) != 1 || + created.Media[0].Kind != store.MediaKindVideo || + created.Media[0].ContentType != "video/mp4" || + created.Media[0].Body == nil { + t.Fatalf("held media = %+v", created.Media) + } + if created.Release == nil { + t.Fatal("missing Release") + } + body, err := io.ReadAll(created.Media[0].Body) + if err != nil || !bytes.Equal(body, clip) { + t.Fatalf("held video body = %d bytes err=%v", len(body), err) + } + created.Release() +} diff --git a/internal/web/post_images.go b/internal/web/post_images.go index 148a3cc..0e3bfd8 100644 --- a/internal/web/post_images.go +++ b/internal/web/post_images.go @@ -22,6 +22,7 @@ import ( _ "golang.org/x/image/webp" "plumber/internal/blob" + "plumber/internal/events" "plumber/internal/store" ) @@ -109,18 +110,18 @@ func (s *Server) postImagesFromForm( r *http.Request, postID string, existing []store.PostImage, -) ([]store.PostImage, []string, error) { +) ([]store.PostImage, []string, []events.Media, error) { if r.MultipartForm == nil { - return append([]store.PostImage(nil), existing...), nil, nil + return append([]store.PostImage(nil), existing...), nil, nil, nil } retained, err := retainedPostImages(r.MultipartForm, existing) if err != nil { - return nil, nil, err + return nil, nil, nil, err } files := r.MultipartForm.File["images"] descriptions := r.MultipartForm.Value["image_description"] if len(descriptions) > len(files) { - return nil, nil, invalidPostImage("Image descriptions do not match selected images.", nil) + return nil, nil, nil, invalidPostImage("Image descriptions do not match selected images.", nil) } kinds := make([]string, len(files)) newImages, newVideos := 0, 0 @@ -134,13 +135,13 @@ func (s *Server) postImagesFromForm( } retainedImages, retainedVideos := countPostMedia(retained) if retainedImages+newImages > store.MaxPostImages { - return nil, nil, invalidPostImage("You can attach up to 4 images.", nil) + return nil, nil, nil, invalidPostImage("You can attach up to 4 images.", nil) } if retainedVideos+newVideos > store.MaxPostVideos { - return nil, nil, invalidPostImage("You can attach one video.", nil) + return nil, nil, nil, invalidPostImage("You can attach one video.", nil) } if len(files) > 0 && !s.cfg.Blob.Enabled() { - return nil, nil, &postImageRequestError{ + return nil, nil, nil, &postImageRequestError{ status: http.StatusServiceUnavailable, message: "Image uploads are not configured on this server.", } @@ -148,24 +149,28 @@ func (s *Server) postImagesFromForm( images := append([]store.PostImage(nil), retained...) newKeys := make([]string, 0, len(files)) + held := make([]events.Media, 0, len(files)) for i, header := range files { description := "" if i < len(descriptions) { description = strings.TrimSpace(descriptions[i]) } if len([]rune(description)) > store.MaxImageDescriptionRunes { + closeHeldMedia(held) s.deletePostImageObjects(newKeys) - return nil, nil, invalidPostImage("Image descriptions must be 500 characters or fewer.", nil) + return nil, nil, nil, invalidPostImage("Image descriptions must be 500 characters or fewer.", nil) } - item, objectKey, err := s.uploadPostMedia(ctx, postID, header, kinds[i], description) + item, objectKey, media, err := s.uploadPostMedia(ctx, postID, header, kinds[i], description) if err != nil { + closeHeldMedia(held) s.deletePostImageObjects(newKeys) - return nil, nil, err + return nil, nil, nil, err } newKeys = append(newKeys, objectKey) images = append(images, item) + held = append(held, media) } - return images, newKeys, nil + return images, newKeys, held, nil } func countPostMedia(items []store.PostImage) (images, videos int) { @@ -179,18 +184,26 @@ func countPostMedia(items []store.PostImage) (images, videos int) { return images, videos } +func closeHeldMedia(media []events.Media) { + for i := range media { + if media[i].Body != nil { + _ = media[i].Body.Close() + media[i].Body = nil + } + } +} + func (s *Server) uploadPostMedia( ctx context.Context, postID string, header *multipart.FileHeader, kind, description string, -) (store.PostImage, string, error) { +) (store.PostImage, string, events.Media, error) { if kind == store.MediaKindVideo { prepared, err := preparePostVideo(header) if err != nil { - return store.PostImage{}, "", err + return store.PostImage{}, "", events.Media{}, err } - defer prepared.body.Close() mediaID := uuid.NewString() objectKey := path.Join("post-videos", postID, mediaID+prepared.extension) publicURL, err := s.cfg.Blob.Upload(ctx, blob.FileUpload{ @@ -200,12 +213,28 @@ func (s *Server) uploadPostMedia( Size: prepared.size, }) if err != nil { - return store.PostImage{}, "", &postImageRequestError{ + _ = prepared.body.Close() + return store.PostImage{}, "", events.Media{}, &postImageRequestError{ status: http.StatusServiceUnavailable, message: "Could not upload video. Try again later.", cause: err, } } + media := events.Media{ + Name: mediaID + prepared.extension, + ContentType: prepared.contentType, + Kind: store.MediaKindVideo, + Size: prepared.size, + } + if seeker, ok := prepared.body.(io.Seeker); ok { + if _, err := seeker.Seek(0, io.SeekStart); err == nil { + media.Body = prepared.body + } else { + _ = prepared.body.Close() + } + } else { + _ = prepared.body.Close() + } return store.PostImage{ ID: mediaID, PostID: postID, @@ -213,11 +242,11 @@ func (s *Server) uploadPostMedia( PublicURL: publicURL, Description: description, Kind: store.MediaKindVideo, - }, objectKey, nil + }, objectKey, media, nil } prepared, err := preparePostImage(header) if err != nil { - return store.PostImage{}, "", err + return store.PostImage{}, "", events.Media{}, err } imageID := uuid.NewString() objectKey := path.Join("post-images", postID, imageID+prepared.extension) @@ -228,22 +257,28 @@ func (s *Server) uploadPostMedia( Size: int64(len(prepared.body)), }) if err != nil { - return store.PostImage{}, "", &postImageRequestError{ + return store.PostImage{}, "", events.Media{}, &postImageRequestError{ status: http.StatusServiceUnavailable, message: "Could not upload image. Try again later.", cause: err, } } return store.PostImage{ - ID: imageID, - PostID: postID, - ObjectKey: objectKey, - PublicURL: publicURL, - Description: description, - Kind: store.MediaKindImage, - Width: prepared.width, - Height: prepared.height, - }, objectKey, nil + ID: imageID, + PostID: postID, + ObjectKey: objectKey, + PublicURL: publicURL, + Description: description, + Kind: store.MediaKindImage, + Width: prepared.width, + Height: prepared.height, + }, objectKey, events.Media{ + Name: imageID + prepared.extension, + ContentType: prepared.contentType, + Kind: store.MediaKindImage, + Size: int64(len(prepared.body)), + Bytes: prepared.body, + }, nil } func retainedPostImages(form *multipart.Form, existing []store.PostImage) ([]store.PostImage, error) { diff --git a/internal/web/post_images_test.go b/internal/web/post_images_test.go index 3cdbd7e..d4cb09d 100644 --- a/internal/web/post_images_test.go +++ b/internal/web/post_images_test.go @@ -295,6 +295,19 @@ func TestPostVideoMultipartLifecycle(t *testing.T) { t.Fatalf("streamed video upload = %+v", streamed) } + rec = httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/questions/"+root.ID, nil) + for _, cookie := range cookies { + req.AddCookie(cookie) + } + handler.ServeHTTP(rec, req) + page := rec.Body.String() + if rec.Code != http.StatusOK || + !strings.Contains(page, `