mirror of https://github.com/usememos/memos
fix(api): improve SSE hub design and fix double-broadcast on comments
- Fix duplicate SSE event on comment creation: CreateMemoComment now suppresses the redundant memo.created broadcast from the inner CreateMemo call, emitting only memo.comment.created - Extract reaction event-building IIFEs into buildMemoReactionSSEEvent helper, removing duplicated inline DB-fetch logic - Promote resolveSSEAudienceCreatorID from method to free function (resolveSSECreatorID) since it never used the receiver - Add userID to SSE connect/disconnect log lines for traceability - Change canReceive default from permissive (return true) to deny-with-warning for unknown visibility types - Add comprehensive tests covering all new helpers, visibility edge cases, slow-client drop behavior, and the double-broadcast fix Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>pull/5795/head
memoclaw
2 months ago
parent
d720efb6e6
commit
c53677fcba
@ -0,0 +1,40 @@
|
||||
package v1
|
||||
|
||||
import "github.com/usememos/memos/store"
|
||||
|
||||
func buildMemoName(uid string) string {
|
||||
return MemoNamePrefix + uid
|
||||
}
|
||||
|
||||
// resolveSSECreatorID returns the CreatorID used for SSE delivery filtering.
|
||||
// For a comment memo, it returns the parent memo's CreatorID so that private
|
||||
// parent-memo events are scoped to the parent owner.
|
||||
func resolveSSECreatorID(memo *store.Memo, parentMemo *store.Memo) int32 {
|
||||
if memo == nil {
|
||||
return 0
|
||||
}
|
||||
if parentMemo != nil {
|
||||
return parentMemo.CreatorID
|
||||
}
|
||||
return memo.CreatorID
|
||||
}
|
||||
|
||||
// buildMemoReactionSSEEvent constructs an SSEEvent for a reaction on a memo.
|
||||
// Pass parentMemo when the memo is a comment (memo.ParentUID != nil).
|
||||
func buildMemoReactionSSEEvent(eventType SSEEventType, contentID string, memo *store.Memo, parentMemo *store.Memo) *SSEEvent {
|
||||
parent := ""
|
||||
if memo != nil && memo.ParentUID != nil {
|
||||
parent = buildMemoName(*memo.ParentUID)
|
||||
}
|
||||
visibility := store.Visibility("")
|
||||
if memo != nil {
|
||||
visibility = memo.Visibility
|
||||
}
|
||||
return &SSEEvent{
|
||||
Type: eventType,
|
||||
Name: contentID,
|
||||
Parent: parent,
|
||||
Visibility: visibility,
|
||||
CreatorID: resolveSSECreatorID(memo, parentMemo),
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,203 @@
|
||||
package v1
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/usememos/memos/internal/profile"
|
||||
v1pb "github.com/usememos/memos/proto/gen/api/v1"
|
||||
"github.com/usememos/memos/server/auth"
|
||||
"github.com/usememos/memos/store"
|
||||
teststore "github.com/usememos/memos/store/test"
|
||||
)
|
||||
|
||||
// newIntegrationService builds a minimal APIV1Service backed by an in-memory
|
||||
// SQLite database. The store is closed automatically via t.Cleanup.
|
||||
func newIntegrationService(t *testing.T) *APIV1Service {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
st := teststore.NewTestingStore(ctx, t)
|
||||
t.Cleanup(func() { st.Close() })
|
||||
p := &profile.Profile{Demo: true, Data: t.TempDir(), Driver: "sqlite", DSN: ":memory:"}
|
||||
return NewAPIV1Service("test-secret", p, st)
|
||||
}
|
||||
|
||||
// userCtx returns a context that authenticates as the given user.
|
||||
func userCtx(ctx context.Context, userID int32) context.Context {
|
||||
return context.WithValue(ctx, auth.UserIDContextKey, userID)
|
||||
}
|
||||
|
||||
// drainEvents reads all events currently buffered in the channel and returns
|
||||
// them as a string slice. It stops as soon as the channel is empty (non-blocking).
|
||||
func drainEvents(ch <-chan []byte) []string {
|
||||
var out []string
|
||||
for {
|
||||
select {
|
||||
case data := <-ch:
|
||||
out = append(out, string(data))
|
||||
default:
|
||||
return out
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// collectEventsFor reads events from ch for the given duration and returns them.
|
||||
func collectEventsFor(ch <-chan []byte, d time.Duration) []string {
|
||||
var out []string
|
||||
deadline := time.After(d)
|
||||
for {
|
||||
select {
|
||||
case data := <-ch:
|
||||
out = append(out, string(data))
|
||||
case <-deadline:
|
||||
return out
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- context suppression ----
|
||||
|
||||
func TestSuppressSSEContext(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
t.Run("default context is not suppressed", func(t *testing.T) {
|
||||
assert.False(t, isSSESuppressed(ctx))
|
||||
})
|
||||
|
||||
t.Run("withSuppressSSE marks context as suppressed", func(t *testing.T) {
|
||||
assert.True(t, isSSESuppressed(withSuppressSSE(ctx)))
|
||||
})
|
||||
|
||||
t.Run("suppression does not bleed into parent context", func(t *testing.T) {
|
||||
suppressed := withSuppressSSE(ctx)
|
||||
_ = suppressed
|
||||
assert.False(t, isSSESuppressed(ctx))
|
||||
})
|
||||
}
|
||||
|
||||
// ---- CreateMemoComment double-broadcast fix ----
|
||||
|
||||
func TestCreateMemoComment_NoDuplicateSSEBroadcast(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
svc := newIntegrationService(t)
|
||||
|
||||
// Create an admin so the store is initialised, then a regular commenter.
|
||||
author, err := svc.Store.CreateUser(ctx, &store.User{
|
||||
Username: "author", Role: store.RoleAdmin, Email: "author@example.com",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
commenter, err := svc.Store.CreateUser(ctx, &store.User{
|
||||
Username: "commenter", Role: store.RoleUser, Email: "commenter@example.com",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
authorCtx := userCtx(ctx, author.ID)
|
||||
commenterCtx := userCtx(ctx, commenter.ID)
|
||||
|
||||
// Create a public memo so the commenter can react.
|
||||
parent, err := svc.CreateMemo(authorCtx, &v1pb.CreateMemoRequest{
|
||||
Memo: &v1pb.Memo{Content: "parent memo", Visibility: v1pb.Visibility_PUBLIC},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Subscribe after the parent memo is created so the memo.created event
|
||||
// for the parent does not pollute the assertion window.
|
||||
client := svc.SSEHub.Subscribe(author.ID, store.RoleAdmin)
|
||||
defer svc.SSEHub.Unsubscribe(client)
|
||||
|
||||
// Create a comment. Before the fix, this fired both memo.created (for the
|
||||
// comment memo) and memo.comment.created (for the parent).
|
||||
_, err = svc.CreateMemoComment(commenterCtx, &v1pb.CreateMemoCommentRequest{
|
||||
Name: parent.Name,
|
||||
Comment: &v1pb.Memo{Content: "a comment", Visibility: v1pb.Visibility_PUBLIC},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Give the synchronous broadcast a moment to land in the buffer, then
|
||||
// collect everything that arrived.
|
||||
events := collectEventsFor(client.events, 150*time.Millisecond)
|
||||
|
||||
require.Len(t, events, 1, "expected exactly one SSE event for a comment creation, got: %v", events)
|
||||
assert.True(t, strings.Contains(events[0], `"memo.comment.created"`),
|
||||
"expected memo.comment.created, got: %s", events[0])
|
||||
}
|
||||
|
||||
// ---- Reaction SSE events carry correct visibility / parent fields ----
|
||||
|
||||
func TestUpsertMemoReaction_SSEEvent(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
svc := newIntegrationService(t)
|
||||
|
||||
user, err := svc.Store.CreateUser(ctx, &store.User{
|
||||
Username: "user", Role: store.RoleAdmin, Email: "user@example.com",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
uctx := userCtx(ctx, user.ID)
|
||||
|
||||
memo, err := svc.CreateMemo(uctx, &v1pb.CreateMemoRequest{
|
||||
Memo: &v1pb.Memo{Content: "reacted memo", Visibility: v1pb.Visibility_PUBLIC},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
client := svc.SSEHub.Subscribe(user.ID, store.RoleAdmin)
|
||||
defer svc.SSEHub.Unsubscribe(client)
|
||||
|
||||
_, err = svc.UpsertMemoReaction(uctx, &v1pb.UpsertMemoReactionRequest{
|
||||
Name: memo.Name,
|
||||
Reaction: &v1pb.Reaction{
|
||||
ContentId: memo.Name,
|
||||
ReactionType: "👍",
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
data := mustReceive(t, client.events, time.Second)
|
||||
payload := string(data)
|
||||
assert.Contains(t, payload, `"reaction.upserted"`)
|
||||
assert.Contains(t, payload, memo.Name)
|
||||
mustNotReceive(t, client.events, 100*time.Millisecond)
|
||||
}
|
||||
|
||||
func TestDeleteMemoReaction_SSEEvent(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
svc := newIntegrationService(t)
|
||||
|
||||
user, err := svc.Store.CreateUser(ctx, &store.User{
|
||||
Username: "user", Role: store.RoleAdmin, Email: "user@example.com",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
uctx := userCtx(ctx, user.ID)
|
||||
|
||||
memo, err := svc.CreateMemo(uctx, &v1pb.CreateMemoRequest{
|
||||
Memo: &v1pb.Memo{Content: "reacted memo", Visibility: v1pb.Visibility_PUBLIC},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
reaction, err := svc.UpsertMemoReaction(uctx, &v1pb.UpsertMemoReactionRequest{
|
||||
Name: memo.Name,
|
||||
Reaction: &v1pb.Reaction{
|
||||
ContentId: memo.Name,
|
||||
ReactionType: "❤️",
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
client := svc.SSEHub.Subscribe(user.ID, store.RoleAdmin)
|
||||
defer svc.SSEHub.Unsubscribe(client)
|
||||
|
||||
_, err = svc.DeleteMemoReaction(uctx, &v1pb.DeleteMemoReactionRequest{
|
||||
Name: reaction.Name,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
data := mustReceive(t, client.events, time.Second)
|
||||
payload := string(data)
|
||||
assert.Contains(t, payload, `"reaction.deleted"`)
|
||||
assert.Contains(t, payload, memo.Name)
|
||||
mustNotReceive(t, client.events, 100*time.Millisecond)
|
||||
}
|
||||
Loading…
Reference in New Issue