From 41f2a849796ead11357642efd2e73dc621e4c2e2 Mon Sep 17 00:00:00 2001
From: Jerry Yan <792602257@qq.com>
Date: Wed, 10 Jun 2026 15:02:58 +0800
Subject: [PATCH] =?UTF-8?q?SSE=20=E8=B0=83=E6=95=B4?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 internal/chat/domain.go               |  2 +-
 internal/chat/service_message.go      |  6 +-
 internal/chat/service_message_test.go | 83 +++++++++++++++++++++++++++
 internal/chat/streamer.go             | 10 ++--
 4 files changed, 92 insertions(+), 9 deletions(-)

diff --git a/internal/chat/domain.go b/internal/chat/domain.go
index 41307b9..98974ca 100644
--- a/internal/chat/domain.go
+++ b/internal/chat/domain.go
@@ -327,5 +327,5 @@ type StreamParams struct {
 
 // Subscription 是订阅者读取 ring buffer 的句柄。
 type Subscription interface {
-	Next(ctx context.Context, lastEventID int64) (events []SSEEvent, closed bool, err error)
+	Next(ctx context.Context, nextEventID int64) (events []SSEEvent, closed bool, err error)
 }
diff --git a/internal/chat/service_message.go b/internal/chat/service_message.go
index b678f81..8d96791 100644
--- a/internal/chat/service_message.go
+++ b/internal/chat/service_message.go
@@ -258,15 +258,15 @@ func (s *messageService) Stream(
 					out <- SSEEvent{Event: "error", Data: map[string]string{"code": "server_restart", "message": e}}
 					return
 				}
-				lastID := int64(-1)
+				nextEventID := int64(0)
 				for {
-					events, closed, err := sub.Next(ctx, lastID)
+					events, closed, err := sub.Next(ctx, nextEventID)
 					if err != nil {
 						return
 					}
 					for _, ev := range events {
 						out <- ev
-						lastID = ev.ID
+						nextEventID = ev.ID + 1
 					}
 					if closed {
 						return
diff --git a/internal/chat/service_message_test.go b/internal/chat/service_message_test.go
index 61a6e04..b0e6aba 100644
--- a/internal/chat/service_message_test.go
+++ b/internal/chat/service_message_test.go
@@ -704,6 +704,35 @@ func (s *closedSubscription) Next(_ context.Context, lastID int64) ([]SSEEvent,
 	return nil, true, nil
 }
 
+type recordingSubscription struct {
+	mu      sync.Mutex
+	calls   []int64
+	batches []subscriptionBatch
+}
+
+type subscriptionBatch struct {
+	events []SSEEvent
+	closed bool
+}
+
+func (s *recordingSubscription) Next(_ context.Context, nextID int64) ([]SSEEvent, bool, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.calls = append(s.calls, nextID)
+	i := len(s.calls) - 1
+	if i >= len(s.batches) {
+		return nil, true, nil
+	}
+	batch := s.batches[i]
+	return append([]SSEEvent(nil), batch.events...), batch.closed, nil
+}
+
+func (s *recordingSubscription) callsSnapshot() []int64 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return append([]int64(nil), s.calls...)
+}
+
 func TestMessageService_Stream_HappyPath_OK(t *testing.T) {
 	repo := newMsgFakeRepo()
 	hub := &fakeHub{}
@@ -785,6 +814,60 @@ func TestMessageService_Stream_PendingFallbackToError(t *testing.T) {
 	repo.mu.Unlock()
 }
 
+func TestMessageService_Stream_PendingSubscriptionStartsAtFirstEvent(t *testing.T) {
+	repo := newMsgFakeRepo()
+	userID := uuid.New()
+	conv := newConv(userID, uuid.New())
+	repo.addConversation(conv)
+
+	pending := &Message{
+		ID:             1,
+		ConversationID: conv.ID,
+		Status:         StatusPending,
+		Role:           RoleAssistant,
+	}
+	repo.addMessage(pending)
+
+	sub := &recordingSubscription{
+		batches: []subscriptionBatch{
+			{
+				events: []SSEEvent{
+					{ID: 0, Event: "text_delta", Data: map[string]string{"text": "hi"}},
+				},
+				closed: false,
+			},
+			{
+				events: []SSEEvent{
+					{ID: 1, Event: "done", Data: map[string]string{"finish_reason": "stop"}},
+				},
+				closed: true,
+			},
+		},
+	}
+	hub := &fakeHub{
+		subscribeFn: func(id int64) (Subscription, bool) {
+			require.Equal(t, pending.ID, id)
+			return sub, true
+		},
+	}
+
+	svc := buildMessageService(repo, hub, defaultMsgConfig())
+	ch, err := svc.Stream(context.Background(), userID, conv.ID, 0)
+
+	require.NoError(t, err)
+
+	var events []SSEEvent
+	for ev := range ch {
+		events = append(events, ev)
+	}
+
+	require.Equal(t, []int64{0, 1}, sub.callsSnapshot())
+	require.Len(t, events, 3)
+	require.Equal(t, "message_meta", events[0].Event)
+	require.Equal(t, "text_delta", events[1].Event)
+	require.Equal(t, "done", events[2].Event)
+}
+
 func TestMessageService_Stream_NotOwner_Forbidden(t *testing.T) {
 	repo := newMsgFakeRepo()
 	hub := &fakeHub{}
diff --git a/internal/chat/streamer.go b/internal/chat/streamer.go
index 2d7a1b4..efd839c 100644
--- a/internal/chat/streamer.go
+++ b/internal/chat/streamer.go
@@ -66,14 +66,14 @@ type subscription struct {
 	st *streamer
 }
 
-// Next blocks until at least one event with ID >= lastEventID is available,
+// Next blocks until at least one event with ID >= nextEventID is available,
 // the stream closes, or ctx is cancelled.
-// It returns all events from lastEventID to the current tail.
-func (sub *subscription) Next(ctx context.Context, lastEventID int64) ([]SSEEvent, bool, error) {
+// It returns all events from nextEventID to the current tail.
+func (sub *subscription) Next(ctx context.Context, nextEventID int64) ([]SSEEvent, bool, error) {
 	sub.st.mu.Lock()
 	defer sub.st.mu.Unlock()
 
-	for int64(len(sub.st.events)) <= lastEventID && !sub.st.closed {
+	for int64(len(sub.st.events)) <= nextEventID && !sub.st.closed {
 		done := make(chan struct{})
 		go func() {
 			select {
@@ -88,7 +88,7 @@ func (sub *subscription) Next(ctx context.Context, lastEventID int64) ([]SSEEven
 			return nil, false, ctx.Err()
 		}
 	}
-	out := append([]SSEEvent(nil), sub.st.events[lastEventID:]...)
+	out := append([]SSEEvent(nil), sub.st.events[nextEventID:]...)
 	return out, sub.st.closed, nil
 }