diff --git a/control-plane/account_store.go b/control-plane/account_store.go index 555860c..89cfbde 100644 --- a/control-plane/account_store.go +++ b/control-plane/account_store.go @@ -730,18 +730,43 @@ func (m *AccountStoreManager) activeScopes(ctx context.Context, accountID string return scopes, generation, nil } -func (s *AccountStore) QueryConversations(ctx context.Context, limit, offset int) ([]ConversationRecord, error) { - if limit < 1 || limit > 200 || offset < 0 { +func (s *AccountStore) QueryConversations(ctx context.Context, limit int, cursor *ConversationPageCursor) ([]ConversationRecord, error) { + if limit < 1 || limit > 200 { return nil, errors.New("invalid conversation pagination") } scopes, _, err := s.manager.activeScopes(ctx, s.accountID) if err != nil { return nil, err } - if len(scopes) == 0 { + allowedChats := make([]string, 0, len(scopes)) + seenChats := make(map[string]struct{}, len(scopes)) + for _, scope := range scopes { + if !scopeAllows(scopes, scope.ChatID, "conversations") { + continue + } + if _, exists := seenChats[scope.ChatID]; exists { + continue + } + seenChats[scope.ChatID] = struct{}{} + allowedChats = append(allowedChats, scope.ChatID) + } + if len(allowedChats) == 0 { return nil, ErrAccountNotAuthorized } - rows, err := s.db.QueryContext(ctx, `SELECT chat_id, chat_type, title, last_activity_at, source, observed_at, directory_state FROM conversations ORDER BY COALESCE(last_activity_at, observed_at) DESC, chat_id`) + + const sortExpression = "COALESCE(last_activity_at, observed_at)" + query := `SELECT chat_id, chat_type, title, last_activity_at, source, observed_at, directory_state FROM conversations WHERE chat_id IN (` + strings.TrimSuffix(strings.Repeat("?,", len(allowedChats)), ",") + ")" + args := make([]any, 0, len(allowedChats)+4) + for _, chatID := range allowedChats { + args = append(args, chatID) + } + if cursor != nil { + query += " AND (" + sortExpression + " < ? OR (" + sortExpression + " = ? AND chat_id > ?))" + args = append(args, cursor.SortTime, cursor.SortTime, cursor.ChatID) + } + query += " ORDER BY " + sortExpression + " DESC, chat_id ASC LIMIT ?" + args = append(args, limit+1) + rows, err := s.db.QueryContext(ctx, query, args...) if err != nil { return nil, fmt.Errorf("query conversations: %w", err) } @@ -758,21 +783,12 @@ func (s *AccountStore) QueryConversations(ctx context.Context, limit, offset int item.LastActivityAt = &value } item.ObservedAt = parseTime(observed) - if scopeAllows(scopes, item.ChatID, "conversations") { - records = append(records, item) - } + records = append(records, item) } if err := rows.Err(); err != nil { return nil, err } - if offset >= len(records) { - return []ConversationRecord{}, nil - } - end := offset + limit - if end > len(records) { - end = len(records) - } - return records[offset:end], nil + return records, nil } type AccountSyncStatus struct { @@ -839,8 +855,8 @@ func (s *AccountStore) GetSyncStatus(ctx context.Context, streamKey string) (Acc return status, nil } -func (s *AccountStore) QueryMessages(ctx context.Context, chatID string, limit, offset int) ([]MessageRecord, error) { - if strings.TrimSpace(chatID) == "" || limit < 1 || limit > 200 || offset < 0 { +func (s *AccountStore) QueryMessages(ctx context.Context, chatID string, limit int, cursor *MessagePageCursor) ([]MessageRecord, error) { + if strings.TrimSpace(chatID) == "" || limit < 1 || limit > 200 { return nil, errors.New("invalid message query") } allowed, err := s.AuthorizeChat(ctx, chatID, "messages") @@ -850,7 +866,15 @@ func (s *AccountStore) QueryMessages(ctx context.Context, chatID string, limit, if !allowed { return nil, ErrAccountNotAuthorized } - rows, err := s.db.QueryContext(ctx, `SELECT message_id, chat_id, source_message_id, direction, message_type, text, source_time, observed_at, source_version, payload_hash FROM messages WHERE chat_id = ? ORDER BY source_time DESC, message_id DESC LIMIT ? OFFSET ?`, chatID, limit, offset) + query := `SELECT message_id, chat_id, source_message_id, direction, message_type, text, source_time, observed_at, source_version, payload_hash FROM messages WHERE chat_id = ?` + args := []any{chatID} + if cursor != nil { + query += " AND (source_time < ? OR (source_time = ? AND message_id < ?))" + args = append(args, cursor.SourceTime, cursor.SourceTime, cursor.MessageID) + } + query += " ORDER BY source_time DESC, message_id DESC LIMIT ?" + args = append(args, limit+1) + rows, err := s.db.QueryContext(ctx, query, args...) if err != nil { return nil, fmt.Errorf("query messages: %w", err) } diff --git a/control-plane/account_store_test.go b/control-plane/account_store_test.go index 3d75e3b..1fb9a64 100644 --- a/control-plane/account_store_test.go +++ b/control-plane/account_store_test.go @@ -92,16 +92,59 @@ func TestAccountStoreIsolatesAccountsAndAppliesIdempotentBatches(t *testing.T) { t.Fatalf("expected unauthorized batch, got %v", err) } - messages, err := accountA.QueryMessages(ctx, "chat-a", 20, 0) + messages, err := accountA.QueryMessages(ctx, "chat-a", 20, nil) if err != nil || len(messages) != 1 || messages[0].Text != "hello" { t.Fatalf("unexpected stored messages: %v %+v", err, messages) } - conversations, err := accountA.QueryConversations(ctx, 20, 0) + conversations, err := accountA.QueryConversations(ctx, 20, nil) if err != nil || len(conversations) != 1 || conversations[0].ChatID != "chat-a" { t.Fatalf("unexpected stored conversations: %v %+v", err, conversations) } } +func TestAccountStoreCursorPaginationKeepsStableMessageBoundary(t *testing.T) { + ctx := context.Background() + manager, err := OpenAccountStoreManager(t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer manager.Close() + store, err := manager.RegisterAccount(ctx, AccountRegistration{ + AccountID: "account-cursor", StableIdentity: "wechat-cursor", SourceNodeID: "node-cursor", SourceGeneration: "generation-cursor", Verified: true, + ReportingScopes: []ReportingScope{{ChatID: "chat-a", DataType: "read", ConfigVersion: 1}}, + }) + if err != nil { + t.Fatal(err) + } + defer store.Close() + now := time.Now().UTC().Truncate(time.Microsecond) + batch := IngestBatch{ + BatchID: "cursor-batch-1", SourceGeneration: "generation-cursor", StreamKey: "messages", Sequence: 1, + CursorStart: "0", CursorEnd: "1", PayloadHash: "cursor-hash-1", CoverageState: "complete", + Messages: []MessageRecord{ + {MessageID: "message-2", ChatID: "chat-a", SourceMessageID: "source-2", Direction: "incoming", MessageType: "text", Text: "two", SourceTime: now, ObservedAt: now, SourceVersion: "wx", PayloadHash: "message-hash-2"}, + {MessageID: "message-1", ChatID: "chat-a", SourceMessageID: "source-1", Direction: "incoming", MessageType: "text", Text: "one", SourceTime: now.Add(-time.Second), ObservedAt: now, SourceVersion: "wx", PayloadHash: "message-hash-1"}, + }, + } + if _, err := store.ApplyBatch(ctx, batch); err != nil { + t.Fatal(err) + } + firstPage, err := store.QueryMessages(ctx, "chat-a", 1, nil) + if err != nil || len(firstPage) != 2 || firstPage[0].MessageID != "message-2" { + t.Fatalf("unexpected first cursor page: %v %+v", err, firstPage) + } + cursor := &MessagePageCursor{ChatID: firstPage[0].ChatID, SourceTime: firstPage[0].SourceTime.UTC().Format(time.RFC3339Nano), MessageID: firstPage[0].MessageID} + batch.BatchID, batch.Sequence, batch.CursorStart, batch.CursorEnd, batch.PayloadHash = "cursor-batch-2", 2, "1", "2", "cursor-hash-2" + batch.Messages = []MessageRecord{{MessageID: "message-3", ChatID: "chat-a", SourceMessageID: "source-3", Direction: "incoming", MessageType: "text", Text: "three", SourceTime: now.Add(time.Second), ObservedAt: now.Add(time.Second), SourceVersion: "wx", PayloadHash: "message-hash-3"}} + if _, err := store.ApplyBatch(ctx, batch); err != nil { + t.Fatal(err) + } + nextPage, err := store.QueryMessages(ctx, "chat-a", 20, cursor) + if err != nil || len(nextPage) != 1 || nextPage[0].MessageID != "message-1" { + t.Fatalf("cursor page repeated or skipped a row: %v %+v", err, nextPage) + } +} + func TestAccountStoreBackupRestoreAndSchemaFailureAreSafe(t *testing.T) { ctx := context.Background() root := t.TempDir() @@ -205,13 +248,13 @@ func TestAccountStoreMaintenanceCapacityAndRevocation(t *testing.T) { if report.DeletedMessages != 1 || report.DeletedConversations != 1 || report.Checkpointed != 1 { t.Fatalf("unexpected maintenance report: %+v", report) } - if _, err := store.QueryMessages(ctx, "chat-a", 20, 0); err != nil { + if _, err := store.QueryMessages(ctx, "chat-a", 20, nil); err != nil { t.Fatal(err) } if err := manager.SetScopes(ctx, "account-a", nil); err != nil { t.Fatal(err) } - if _, err := store.QueryMessages(ctx, "chat-a", 20, 0); !errors.Is(err, ErrAccountNotAuthorized) { + if _, err := store.QueryMessages(ctx, "chat-a", 20, nil); !errors.Is(err, ErrAccountNotAuthorized) { t.Fatalf("expected revoked query to be rejected, got %v", err) } if err := manager.SetScopes(ctx, "account-a", []ReportingScope{{ChatID: "chat-a", DataType: "read", ConfigVersion: 2}}); err != nil { @@ -281,7 +324,7 @@ func TestAccountStoreSyntheticConcurrentReadWrite(t *testing.T) { defer store.Close() localStart := time.Now() for i := 0; i < 20; i++ { - items, err := store.QueryMessages(ctx, chatID, 100, 0) + items, err := store.QueryMessages(ctx, chatID, 100, nil) if err != nil { t.Errorf("query %s: %v", accountID, err) return diff --git a/control-plane/data_routes.go b/control-plane/data_routes.go index dea3dbd..472c39b 100644 --- a/control-plane/data_routes.go +++ b/control-plane/data_routes.go @@ -296,19 +296,31 @@ func (s *Server) dataRoute(w http.ResponseWriter, r *http.Request, correlationID } func (s *Server) listStoredConversations(w http.ResponseWriter, r *http.Request, accountID string) error { - limit, offset := queryLimit(r.URL.Query().Get("limit")), queryOffset(r.URL.Query().Get("offset")) + limit := queryLimit(r.URL.Query().Get("limit")) + var cursor *ConversationPageCursor + if raw := r.URL.Query().Get("cursor"); raw != "" { + parsed, err := DecodeConversationPageCursor(raw) + if err != nil { + return requestError{status: http.StatusBadRequest, code: "InvalidCursor", message: "The conversation cursor is invalid."} + } + cursor = parsed + } store, err := s.accountStores.OpenAccount(r.Context(), accountID) if err != nil { return requestError{status: http.StatusNotFound, code: "AccountDataNotFound", message: "No platform data is available for this account."} } defer store.Close() - items, err := store.QueryConversations(r.Context(), limit, offset) + items, err := store.QueryConversations(r.Context(), limit, cursor) if err != nil { if errors.Is(err, ErrAccountNotAuthorized) { return requestError{status: http.StatusForbidden, code: "DataNotAuthorized", message: "The account has no active conversation reporting scope."} } return requestError{status: http.StatusInternalServerError, code: "DataQueryFailed", message: "The platform data query failed."} } + hasMore := len(items) > limit + if hasMore { + items = items[:limit] + } status, err := store.GetSyncStatus(r.Context(), "messages") if err != nil { return requestError{status: http.StatusInternalServerError, code: "DataStatusFailed", message: "The platform sync status query failed."} @@ -317,7 +329,16 @@ func (s *Server) listStoredConversations(w http.ResponseWriter, r *http.Request, for _, item := range items { views = append(views, dataConversationView{item.ChatID, item.ChatType, item.Title, item.LastActivityAt, item.Source, item.ObservedAt, item.DirectoryState}) } - writeJSON(w, http.StatusOK, map[string]any{"items": views, "limit": limit, "offset": offset, "has_more": len(views) == limit, "next_offset": offset + len(views), "sync": syncView(status)}) + nextCursor := "" + if hasMore && len(items) > 0 { + last := items[len(items)-1] + sortTime := last.ObservedAt + if last.LastActivityAt != nil { + sortTime = *last.LastActivityAt + } + nextCursor = EncodeConversationPageCursor(sortTime, last.ChatID) + } + writeJSON(w, http.StatusOK, map[string]any{"items": views, "limit": limit, "has_more": hasMore, "next_cursor": nextCursor, "sync": syncView(status)}) return nil } @@ -326,7 +347,15 @@ func (s *Server) listStoredMessages(w http.ResponseWriter, r *http.Request, acco if !validIdentifier(chatID, 512) { return requestError{status: http.StatusBadRequest, code: "InvalidChat", message: "chat_id is required."} } - limit, offset := queryLimit(r.URL.Query().Get("limit")), queryOffset(r.URL.Query().Get("offset")) + limit := queryLimit(r.URL.Query().Get("limit")) + var cursor *MessagePageCursor + if raw := r.URL.Query().Get("cursor"); raw != "" { + parsed, err := DecodeMessagePageCursor(raw, chatID) + if err != nil { + return requestError{status: http.StatusBadRequest, code: "InvalidCursor", message: "The message cursor is invalid."} + } + cursor = parsed + } store, err := s.accountStores.OpenAccount(r.Context(), accountID) if err != nil { return requestError{status: http.StatusNotFound, code: "AccountDataNotFound", message: "No platform data is available for this account."} @@ -336,10 +365,14 @@ func (s *Server) listStoredMessages(w http.ResponseWriter, r *http.Request, acco if err != nil || !allowed { return requestError{status: http.StatusForbidden, code: "DataNotAuthorized", message: "The chat is not in the active reporting scope."} } - items, err := store.QueryMessages(r.Context(), chatID, limit, offset) + items, err := store.QueryMessages(r.Context(), chatID, limit, cursor) if err != nil { return requestError{status: http.StatusInternalServerError, code: "DataQueryFailed", message: "The platform data query failed."} } + hasMore := len(items) > limit + if hasMore { + items = items[:limit] + } status, err := store.GetSyncStatus(r.Context(), "messages") if err != nil { return requestError{status: http.StatusInternalServerError, code: "DataStatusFailed", message: "The platform sync status query failed."} @@ -348,7 +381,12 @@ func (s *Server) listStoredMessages(w http.ResponseWriter, r *http.Request, acco for _, item := range items { views = append(views, dataMessageView{item.MessageID, item.ChatID, item.SourceMessageID, item.Direction, item.MessageType, item.Text, item.SourceTime, item.ObservedAt, item.SourceVersion}) } - writeJSON(w, http.StatusOK, map[string]any{"items": views, "limit": limit, "offset": offset, "has_more": len(views) == limit, "next_offset": offset + len(views), "sync": syncView(status)}) + nextCursor := "" + if hasMore && len(items) > 0 { + last := items[len(items)-1] + nextCursor = EncodeMessagePageCursor(last.ChatID, last.SourceTime, last.MessageID) + } + writeJSON(w, http.StatusOK, map[string]any{"items": views, "limit": limit, "has_more": hasMore, "next_cursor": nextCursor, "sync": syncView(status)}) return nil } @@ -408,14 +446,3 @@ func (s *Server) requestDataRefresh(w http.ResponseWriter, r *http.Request, acco payload := json.RawMessage(`{"stream_key":"messages","reason":"web-refresh"}`) return s.createTaskSubmission(w, TaskSubmission{NodeID: nodeID, AccountID: accountID, Kind: "sync-data", IdempotencyKey: "sync:" + accountID + ":" + strconv.FormatInt(time.Now().UTC().Unix()/5, 10), Payload: payload}, username, correlationID) } - -func queryOffset(raw string) int { - if raw == "" { - return 0 - } - value, err := strconv.Atoi(raw) - if err != nil || value < 0 || value > 10_000_000 { - return 0 - } - return value -} diff --git a/control-plane/pagination.go b/control-plane/pagination.go new file mode 100644 index 0000000..93005c8 --- /dev/null +++ b/control-plane/pagination.go @@ -0,0 +1,86 @@ +package controlplane + +import ( + "encoding/base64" + "encoding/json" + "errors" + "strings" + "time" +) + +const pageCursorVersion = 1 + +type pageCursorPayload struct { + Version int `json:"v"` + Kind string `json:"kind"` + ChatID string `json:"chat_id,omitempty"` + SortTime string `json:"sort_time"` + MessageID string `json:"message_id,omitempty"` +} + +type MessagePageCursor struct { + ChatID string + SourceTime string + MessageID string +} + +type ConversationPageCursor struct { + SortTime string + ChatID string +} + +func EncodeMessagePageCursor(chatID string, sourceTime time.Time, messageID string) string { + return encodePageCursor(pageCursorPayload{ + Version: pageCursorVersion, Kind: "messages", ChatID: chatID, + SortTime: sourceTime.UTC().Format(time.RFC3339Nano), MessageID: messageID, + }) +} + +func DecodeMessagePageCursor(raw, chatID string) (*MessagePageCursor, error) { + payload, err := decodePageCursor(raw, "messages") + if err != nil || payload.ChatID != chatID || payload.MessageID == "" { + return nil, errors.New("invalid message cursor") + } + if _, err := time.Parse(time.RFC3339Nano, payload.SortTime); err != nil { + return nil, errors.New("invalid message cursor") + } + return &MessagePageCursor{ChatID: payload.ChatID, SourceTime: payload.SortTime, MessageID: payload.MessageID}, nil +} + +func EncodeConversationPageCursor(sortTime time.Time, chatID string) string { + return encodePageCursor(pageCursorPayload{ + Version: pageCursorVersion, Kind: "conversations", ChatID: chatID, + SortTime: sortTime.UTC().Format(time.RFC3339Nano), + }) +} + +func DecodeConversationPageCursor(raw string) (*ConversationPageCursor, error) { + payload, err := decodePageCursor(raw, "conversations") + if err != nil || payload.ChatID == "" { + return nil, errors.New("invalid conversation cursor") + } + if _, err := time.Parse(time.RFC3339Nano, payload.SortTime); err != nil { + return nil, errors.New("invalid conversation cursor") + } + return &ConversationPageCursor{SortTime: payload.SortTime, ChatID: payload.ChatID}, nil +} + +func encodePageCursor(payload pageCursorPayload) string { + encoded, _ := json.Marshal(payload) + return base64.RawURLEncoding.EncodeToString(encoded) +} + +func decodePageCursor(raw, kind string) (pageCursorPayload, error) { + if strings.TrimSpace(raw) == "" { + return pageCursorPayload{}, errors.New("cursor is empty") + } + decoded, err := base64.RawURLEncoding.DecodeString(raw) + if err != nil { + return pageCursorPayload{}, errors.New("cursor is not base64url") + } + var payload pageCursorPayload + if err := json.Unmarshal(decoded, &payload); err != nil || payload.Version != pageCursorVersion || payload.Kind != kind || payload.SortTime == "" { + return pageCursorPayload{}, errors.New("cursor payload is invalid") + } + return payload, nil +} diff --git a/control-plane/web/src/main.jsx b/control-plane/web/src/main.jsx index 4799b8f..502da44 100644 --- a/control-plane/web/src/main.jsx +++ b/control-plane/web/src/main.jsx @@ -395,7 +395,7 @@ function MessagesView({ token, client, onSettings }) { setSessionsState((current) => ({ ...current, loading: true, error: "" })); try { const content = await readStoredOrFallback({ - storedPath: dataPath(accountId, "conversations", "?limit=100&offset=0"), + storedPath: dataPath(accountId, "conversations", "?limit=100"), legacyPath: "/v1/reads/sessions", legacyBody: { node_id: clientId, account_id: accountId, limit: 100 }, token, @@ -432,7 +432,7 @@ function MessagesView({ token, client, onSettings }) { const generation = contextGenerationRef.current; setMessagesState((current) => ({ ...current, loading: true, error: "" })); readStoredOrFallback({ - storedPath: dataPath(accountId, "messages", `?chat_id=${encodeURIComponent(selectedChat.id)}&limit=100&offset=0`), + storedPath: dataPath(accountId, "messages", `?chat_id=${encodeURIComponent(selectedChat.id)}&limit=100`), legacyPath: "/v1/reads/messages", legacyBody: { node_id: clientId, account_id: accountId, chat_id: selectedChat.id, limit: 100, include_content: true }, token, diff --git a/docs/validation/raw/cursor-http-transcript.sh b/docs/validation/raw/cursor-http-transcript.sh new file mode 100755 index 0000000..d84ee2f --- /dev/null +++ b/docs/validation/raw/cursor-http-transcript.sh @@ -0,0 +1,59 @@ +#!/usr/bin/env bash +set -euo pipefail + +binary=${CONTROL_PLANE_BIN:-./control-plane} +work=$(mktemp -d) +port=${P4_CURSOR_PORT:-18091} +addr="127.0.0.1:${port}" +base="http://${addr}" +cleanup() { [[ -n "${pid:-}" ]] && kill "$pid" 2>/dev/null || true; rm -rf "$work"; } +trap cleanup EXIT + +WXAGENT_CONTROL_PLANE_ADDR="$addr" \ +WXAGENT_CONTROL_PLANE_DATA="$work/control-plane.json" \ +WXAGENT_CONTROL_PLANE_BACKUP_DIR="$work/backups" \ +WXAGENT_NODE_ID=node-cursor \ +WXAGENT_NODE_TOKEN=node-cursor-token \ +WXAGENT_WEB_USER=admin \ +WXAGENT_WEB_PASSWORD=fixture-password \ +"$binary" >"$work/server.log" 2>&1 & +pid=$! +for _ in $(seq 1 40); do code=$(curl -sS -o /dev/null -w '%{http_code}' "$base/v1/auth/login" || true); [[ "$code" != "000" ]] && break; sleep .1; done + +curl -fsS -X POST "$base/v1/nodes/register" \ + -H 'Authorization: Bearer node-cursor-token' -H 'Content-Type: application/json' \ + -d '{"node_id":"node-cursor","connection_id":"connection-cursor","agent_version":"fixture","protocol_version":"v1","capabilities":["sync-data"],"accounts":[{"account_id":"account-cursor","active":true,"verified":true,"allowed_chats":[{"chat_id":"chat-cursor","chat_type":"Private"}]}]}' >/dev/null + +token=$(curl -fsS -X POST "$base/v1/auth/login" -H 'Content-Type: application/json' \ + -d '{"username":"admin","password":"fixture-password"}' | jq -r .access_token) + +post_batch() { + local batch_id=$1 sequence=$2 payload=$3 messages + if [[ $sequence == 1 ]]; then + messages='[{"message_id":"message-2","chat_id":"chat-cursor","chat_type":"Private","source_message_id":"source-2","direction":"incoming","message_type":"text","text":"fixture-two","source_time":"2026-01-01T00:00:02Z","observed_at":"2026-01-01T00:00:02Z","source_version":"fixture","payload_hash":"message-hash-2"},{"message_id":"message-1","chat_id":"chat-cursor","chat_type":"Private","source_message_id":"source-1","direction":"incoming","message_type":"text","text":"fixture-one","source_time":"2026-01-01T00:00:01Z","observed_at":"2026-01-01T00:00:01Z","source_version":"fixture","payload_hash":"message-hash-1"}]' + else + messages='[{"message_id":"message-3","chat_id":"chat-cursor","chat_type":"Private","source_message_id":"source-3","direction":"incoming","message_type":"text","text":"fixture-newer","source_time":"2026-01-01T00:00:03Z","observed_at":"2026-01-01T00:00:03Z","source_version":"fixture","payload_hash":"message-hash-3"}]' + fi + jq -n --arg batch "$batch_id" --arg payload "$payload" --argjson messages "$messages" \ + --argjson sequence "$sequence" --argjson cursor_end "$((sequence+1))" \ + '{node_id:"node-cursor",account_id:"account-cursor",batch_id:$batch,source_generation:"account-cursor",stream_key:"messages",sequence:$sequence,cursor_start:($sequence|tostring),cursor_end:($cursor_end|tostring),payload_hash:$payload,coverage_state:"complete",conversations:[],messages:$messages}' | + curl -fsS -X POST "$base/v1/data/batches" \ + -H 'Authorization: Bearer node-cursor-token' -H 'Content-Type: application/json' --data-binary @- >/dev/null +} + +request_page() { + local label=$1 url=$2 + local body="$work/$label.json" status + status=$(curl -fsS -o "$body" -w '%{http_code}' "$url" -H "Authorization: Bearer $token") + echo "REQUEST $label GET $url" + echo "HTTP $status" + jq '{items:[.items[]|{message_id,source_time,text}],limit,has_more,next_cursor}' "$body" +} + +post_batch batch-cursor-1 1 payload-cursor-1 +first_url="$base/v1/data/accounts/account-cursor/messages?chat_id=chat-cursor&limit=1" +request_page first "$first_url" +cursor=$(jq -r .next_cursor "$work/first.json") +post_batch batch-cursor-2 2 payload-cursor-2 +second_url="$base/v1/data/accounts/account-cursor/messages?chat_id=chat-cursor&limit=10&cursor=$(python3 -c 'import urllib.parse,sys; print(urllib.parse.quote(sys.argv[1], safe=""))' "$cursor")" +request_page second "$second_url" diff --git a/docs/validation/raw/cursor-http-transcript.txt b/docs/validation/raw/cursor-http-transcript.txt new file mode 100644 index 0000000..2df0cb6 --- /dev/null +++ b/docs/validation/raw/cursor-http-transcript.txt @@ -0,0 +1,28 @@ +REQUEST first GET http://127.0.0.1:18091/v1/data/accounts/account-cursor/messages?chat_id=chat-cursor&limit=1 +HTTP 200 +{ + "items": [ + { + "message_id": "message-2", + "source_time": "2026-01-01T00:00:02Z", + "text": "fixture-two" + } + ], + "limit": 1, + "has_more": true, + "next_cursor": "eyJ2IjoxLCJraW5kIjoibWVzc2FnZXMiLCJjaGF0X2lkIjoiY2hhdC1jdXJzb3IiLCJzb3J0X3RpbWUiOiIyMDI2LTAxLTAxVDAwOjAwOjAyWiIsIm1lc3NhZ2VfaWQiOiJtZXNzYWdlLTIifQ" +} +REQUEST second GET http://127.0.0.1:18091/v1/data/accounts/account-cursor/messages?chat_id=chat-cursor&limit=10&cursor=eyJ2IjoxLCJraW5kIjoibWVzc2FnZXMiLCJjaGF0X2lkIjoiY2hhdC1jdXJzb3IiLCJzb3J0X3RpbWUiOiIyMDI2LTAxLTAxVDAwOjAwOjAyWiIsIm1lc3NhZ2VfaWQiOiJtZXNzYWdlLTIifQ +HTTP 200 +{ + "items": [ + { + "message_id": "message-1", + "source_time": "2026-01-01T00:00:01Z", + "text": "fixture-one" + } + ], + "limit": 10, + "has_more": false, + "next_cursor": "" +} diff --git a/tests/node-agent/WxAgent.Service.Tests/ServiceBoundaryTests.cs b/tests/node-agent/WxAgent.Service.Tests/ServiceBoundaryTests.cs index 06d9042..dded17d 100644 --- a/tests/node-agent/WxAgent.Service.Tests/ServiceBoundaryTests.cs +++ b/tests/node-agent/WxAgent.Service.Tests/ServiceBoundaryTests.cs @@ -28,6 +28,17 @@ public sealed class ServiceBoundaryTests public Task> AccountsAsync(CancellationToken ct) => Task.FromResult>([new("account-1", null, null, null, "fingerprint", false)]); } + [Fact] + public void DatabaseSyncIsExplicitOptInByDefault() + { + var options = new ServiceOptions + { + CredentialFile = Path.Combine(Path.GetTempPath(), Guid.NewGuid().ToString("N")), + DataDirectory = Path.GetTempPath() + }; + Assert.False(options.EnableDataSync); + } + [Fact] public async Task StaticUiAndBoundedReadOnlyPagesAreRealProductionRoutes() {