feat(history): 消息正向增量拉取(A5) — GET /sessions/{id}/messages?after=

- history/list_after.go: ListAfter(升序, created_at>after, 走覆盖索引, rowid兜底同毫秒) - HistoryStore 新增 ListAfter(store+fakes同步) - gateway history.go: handleListMessages解析after分流; after+before互斥400; 负数/非数字400 - 新增测试(list_after/history_after), build/vet/test绿+交叉编译 - Docs/api-client v0.26 + api-reference §3.4 同步
2026-09-08 21:49:12 +08:00
1 parent 026b27b310
commit f43751fd07
@@ -263,6 +263,9 @@ func (c *captureHistory) AppendFull(_ context.Context, msg history.Message) (his
 func (c *captureHistory) List(context.Context, string, int64, int, string) ([]history.Message, error) {
 	return nil, nil
 }
+func (c *captureHistory) ListAfter(context.Context, string, int64, int, string) ([]history.Message, error) {
+	return nil, nil
+}
 func (c *captureHistory) Recent(context.Context, string, int) ([]history.Message, error) {
 	return nil, nil
 }
@@ -111,9 +111,19 @@ func (s *Server) handleCreateSubSession(w http.ResponseWriter, r *http.Request)
 	writeJSON(w, http.StatusCreated, map[string]any{"sessionId": id})
 }

-// handleListMessages returns a page of messages for a session, newest first, with cursor
-// pagination. It is the hot-paging history endpoint (design/16 §2.2, §5.1): limit defaults
-// to 50 (clamped to 200), before is the oldest message's created_at from the previous page.
+// handleListMessages returns a page of messages for a session with cursor pagination.
+// It is the hot-paging history endpoint (design/16 §2.2, §5.1). Two mutually exclusive
+// cursors are supported:
+//   - before (millisecond createdAt): newest-first page of messages strictly older than
+//     before — the default, used to page backwards through history;
+//   - after (millisecond createdAt): oldest-first page of messages strictly newer than
+//     after — forward incremental catch-up, so a client that reconnected can pull
+//     everything it missed from the newest createdAt it already has, without the
+//     one-page limit of a backward walk.
+//
+// limit defaults to 50 (clamped to 200) for both directions. platform filters the page
+// by source platform in either direction. Supplying both before and after (both > 0)
+// answers 400, as does a non-numeric/negative after or an invalid limit/before.
 func (s *Server) handleListMessages(w http.ResponseWriter, r *http.Request) {
 	if s.history == nil {
 		writeError(w, &apperr.AppError{Code: "not_implemented", Message: "会话历史服务未配置（未注入 HistoryStore）", HTTPCode: http.StatusNotImplemented})
@@ -143,9 +153,24 @@ func (s *Server) handleListMessages(w http.ResponseWriter, r *http.Request) {
 		}
 		before = v
 	}
+	after, ok := parseAfterCursor(r)
+	if !ok {
+		writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "invalid after"))
+		return
+	}
+	if before > 0 && after > 0 {
+		writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "before and after are mutually exclusive"))
+		return
+	}
 	platform := r.URL.Query().Get("platform")

-	msgs, err := s.history.List(r.Context(), id, before, limit, platform)
+	var msgs []history.Message
+	var err error
+	if after > 0 {
+		msgs, err = s.history.ListAfter(r.Context(), id, after, limit, platform)
+	} else {
+		msgs, err = s.history.List(r.Context(), id, before, limit, platform)
+	}
 	if err != nil {
 		s.log.Error("history list messages", "err", err, "session", id)
 		writeError(w, apperr.Wrap(apperr.ErrInternal, "messages_failed", "failed to list messages"))
@@ -158,6 +183,21 @@ func (s *Server) handleListMessages(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, out)
 }

+// parseAfterCursor reads the non-negative "after" query param (a millisecond createdAt
+// cursor). An absent/empty param yields (0, true) — no forward bound. A non-numeric or
+// negative value yields (0, false) so the caller answers 400.
+func parseAfterCursor(r *http.Request) (int64, bool) {
+	raw := r.URL.Query().Get("after")
+	if raw == "" {
+		return 0, true
+	}
+	v, err := strconv.ParseInt(raw, 10, 64)
+	if err != nil || v < 0 {
+		return 0, false
+	}
+	return v, true
+}
+
 // handleChannelPlatformLog returns up to limit messages for an IM platform across its
 // isolated (group/stranger) sessions, newest first, for the admin adapter log page
 // (design/16 §4.6 §4). It excludes the fixed main web session, so only the platform's
@@ -0,0 +1,160 @@
+package gateway
+
+import (
+	"context"
+	"net/http"
+	"testing"
+
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
+)
+
+// TestHistoryListMessagesAfterCursor covers the forward cursor (A5): after returns the
+// strictly-newer messages oldest-first, before keeps its existing newest-first behavior,
+// and the two cursors are mutually exclusive.
+func TestHistoryListMessagesAfterCursor(t *testing.T) {
+	t.Parallel()
+	_, ts, jwt, st := newHistoryEnv(t)
+	ctx := context.Background()
+	for _, ms := range []int64{1000, 2000, 3000, 4000} {
+		if err := st.Append(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "x", CreatedAt: ms}); err != nil {
+			t.Fatalf("seed %d: %v", ms, err)
+		}
+	}
+	tok := signAny(t, jwt, "acct", auth.RoleOwner)
+	base := ts.URL + "/api/v1/sessions/" + history.MainSessionID + "/messages"
+
+	// after=2000 → strictly newer, oldest first.
+	resp, body := doReq(t, http.MethodGet, base+"?after=2000", nil, tok)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("after status = %d, want 200 (body=%s)", resp.StatusCode, body)
+	}
+	var forward []historyMessage
+	decodeDataSlice(t, body, &forward)
+	if len(forward) != 2 || forward[0].CreatedAt != 3000 || forward[1].CreatedAt != 4000 {
+		t.Fatalf("after=2000 = %+v, want ascending [3000 4000]", forward)
+	}
+
+	// after at the newest message → empty page, still 200.
+	resp, body = doReq(t, http.MethodGet, base+"?after=4000", nil, tok)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("after newest status = %d, want 200 (body=%s)", resp.StatusCode, body)
+	}
+	var empty []historyMessage
+	decodeDataSlice(t, body, &empty)
+	if len(empty) != 0 {
+		t.Fatalf("after=4000 = %+v, want empty", empty)
+	}
+
+	// before behavior is unchanged: newest first, strictly older.
+	resp, body = doReq(t, http.MethodGet, base+"?before=3000&limit=2", nil, tok)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("before status = %d, want 200 (body=%s)", resp.StatusCode, body)
+	}
+	var backward []historyMessage
+	decodeDataSlice(t, body, &backward)
+	if len(backward) != 2 || backward[0].CreatedAt != 2000 || backward[1].CreatedAt != 1000 {
+		t.Fatalf("before=3000 = %+v, want descending [2000 1000]", backward)
+	}
+
+	// after=0 is not a forward bound → falls back to the default newest-first page.
+	resp, body = doReq(t, http.MethodGet, base+"?after=0&limit=2", nil, tok)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("after=0 status = %d, want 200 (body=%s)", resp.StatusCode, body)
+	}
+	var zero []historyMessage
+	decodeDataSlice(t, body, &zero)
+	if len(zero) != 2 || zero[0].CreatedAt != 4000 || zero[1].CreatedAt != 3000 {
+		t.Fatalf("after=0 = %+v, want default descending [4000 3000]", zero)
+	}
+}
+
+// TestHistoryListMessagesAfterRejectsBadInput covers the 400s: both cursors together,
+// and a negative or non-numeric after.
+func TestHistoryListMessagesAfterRejectsBadInput(t *testing.T) {
+	t.Parallel()
+	_, ts, jwt, _ := newHistoryEnv(t)
+	tok := signAny(t, jwt, "acct", auth.RoleOwner)
+	base := ts.URL + "/api/v1/sessions/" + history.MainSessionID + "/messages"
+
+	for _, q := range []string{
+		"?after=1000&before=3000", // both cursors
+		"?after=-1",               // negative
+		"?after=abc",              // non-numeric
+	} {
+		resp, body := doReq(t, http.MethodGet, base+q, nil, tok)
+		if resp.StatusCode != http.StatusBadRequest {
+			t.Fatalf("query %s status = %d, want 400 (body=%s)", q, resp.StatusCode, body)
+		}
+	}
+}
+
+// TestHistoryListMessagesAfterPlatformFilter asserts the platform filter composes with
+// the forward cursor.
+func TestHistoryListMessagesAfterPlatformFilter(t *testing.T) {
+	t.Parallel()
+	_, ts, jwt, st := newHistoryEnv(t)
+	ctx := context.Background()
+	seeds := []history.Message{
+		{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "web", Platform: history.PlatformWeb, CreatedAt: 1000},
+		{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "tg1", Platform: "telegram", CreatedAt: 2000},
+		{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "tg2", Platform: "telegram", CreatedAt: 3000},
+	}
+	for _, m := range seeds {
+		if err := st.Append(ctx, m); err != nil {
+			t.Fatalf("seed %s: %v", m.Content, err)
+		}
+	}
+	tok := signAny(t, jwt, "acct", auth.RoleOwner)
+
+	resp, body := doReq(t, http.MethodGet,
+		ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages?after=1000&platform=telegram", nil, tok)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("after+platform status = %d, want 200 (body=%s)", resp.StatusCode, body)
+	}
+	var tg []historyMessage
+	decodeDataSlice(t, body, &tg)
+	if len(tg) != 2 || tg[0].Content != "tg1" || tg[1].Content != "tg2" {
+		t.Fatalf("after=1000&platform=telegram = %+v, want [tg1 tg2]", tg)
+	}
+}
+
+// TestHistoryListMessagesAfterLimitClamp asserts the forward page is capped at 200 like
+// the backward one, and that advancing the returned cursor pages forward without a gap.
+func TestHistoryListMessagesAfterLimitClamp(t *testing.T) {
+	t.Parallel()
+	_, ts, jwt, st := newHistoryEnv(t)
+	ctx := context.Background()
+	const total = 205
+	for i := 0; i < total; i++ {
+		tsMs := int64(1000 + i)
+		if err := st.Append(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "m", CreatedAt: tsMs}); err != nil {
+			t.Fatalf("seed %d: %v", tsMs, err)
+		}
+	}
+	tok := signAny(t, jwt, "acct", auth.RoleOwner)
+	base := ts.URL + "/api/v1/sessions/" + history.MainSessionID + "/messages"
+
+	resp, body := doReq(t, http.MethodGet, base+"?after=999&limit=999", nil, tok)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("clamp status = %d, want 200 (body=%s)", resp.StatusCode, body)
+	}
+	var page []historyMessage
+	decodeDataSlice(t, body, &page)
+	if len(page) != 200 {
+		t.Fatalf("clamped page len = %d, want 200", len(page))
+	}
+	if page[0].CreatedAt != 1000 || page[len(page)-1].CreatedAt != 1199 {
+		t.Fatalf("clamped page bounds = [%d..%d], want [1000..1199]", page[0].CreatedAt, page[len(page)-1].CreatedAt)
+	}
+
+	resp, body = doReq(t, http.MethodGet, base+"?after=1199", nil, tok)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("next page status = %d, want 200 (body=%s)", resp.StatusCode, body)
+	}
+	var next []historyMessage
+	decodeDataSlice(t, body, &next)
+	if len(next) != total-200 || next[0].CreatedAt != 1200 {
+		t.Fatalf("next page = len %d first %d, want len 5 first 1200", len(next), next[0].CreatedAt)
+	}
+}
@@ -114,6 +114,10 @@ type HistoryStore interface {
 	// cross-end realtime sync (design/16 §4.2).
 	AppendFull(ctx context.Context, msg history.Message) (history.Message, error)
 	List(ctx context.Context, sessionID string, before int64, limit int, platform string) ([]history.Message, error)
+	// ListAfter returns up to limit messages strictly newer than after in
+	// oldest-first order, for forward incremental catch-up after a reconnect
+	// (the "before" cursor can only walk backwards). platform filters as in List.
+	ListAfter(ctx context.Context, sessionID string, after int64, limit int, platform string) ([]history.Message, error)
 	Recent(ctx context.Context, sessionID string, n int) ([]history.Message, error)
 	ListPlatformLog(ctx context.Context, platform string, before int64, limit int) ([]history.Message, error)
 	ListAdapterLog(ctx context.Context, adapterID string, before int64, limit int) ([]history.Message, error)
@@ -179,6 +179,10 @@ func (f *fakeHistory) List(context.Context, string, int64, int, string) ([]histo
 	return nil, nil
 }

+func (f *fakeHistory) ListAfter(context.Context, string, int64, int, string) ([]history.Message, error) {
+	return nil, nil
+}
+
 func (f *fakeHistory) Recent(context.Context, string, int) ([]history.Message, error) {
 	return nil, nil
 }