fix: use stable cursors for platform pagination
This commit is contained in:
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
+59
@@ -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"
|
||||
@@ -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": ""
|
||||
}
|
||||
@@ -28,6 +28,17 @@ public sealed class ServiceBoundaryTests
|
||||
public Task<IReadOnlyList<AccountInfo>> AccountsAsync(CancellationToken ct) => Task.FromResult<IReadOnlyList<AccountInfo>>([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()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user