wx-win-agent/control-plane/data_routes.go

package controlplane

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"strconv"
	"strings"
	"time"
)

const maxDataBatchBytes = 512 * 1024

type dataBatchRequest struct {
	NodeID             string                    `json:"node_id"`
	AccountID          string                    `json:"account_id"`
	BatchID            string                    `json:"batch_id"`
	SourceGeneration   string                    `json:"source_generation"`
	StreamKey          string                    `json:"stream_key"`
	Sequence           int64                     `json:"sequence"`
	CursorStart        string                    `json:"cursor_start"`
	CursorEnd          string                    `json:"cursor_end"`
	PayloadHash        string                    `json:"payload_hash"`
	CoverageState      string                    `json:"coverage_state"`
	Conversations      []dataConversationRequest `json:"conversations"`
	Messages           []dataMessageRequest      `json:"messages"`
	Contacts           []dataContactRequest      `json:"contacts"`
	ContactsSnapshotID string                    `json:"contacts_snapshot_id"`
}

type dataConversationRequest struct {
	ChatID         string     `json:"chat_id"`
	ChatType       ChatType   `json:"chat_type"`
	Title          string     `json:"title"`
	LastActivityAt *time.Time `json:"last_activity_at"`
	Source         string     `json:"source"`
	ObservedAt     time.Time  `json:"observed_at"`
	DirectoryState string     `json:"directory_state"`
}

type dataMessageRequest struct {
	MessageID       string    `json:"message_id"`
	ChatID          string    `json:"chat_id"`
	ChatType        ChatType  `json:"chat_type"`
	SourceMessageID string    `json:"source_message_id"`
	Direction       string    `json:"direction"`
	MessageType     string    `json:"message_type"`
	Text            string    `json:"text"`
	SourceTime      time.Time `json:"source_time"`
	ObservedAt      time.Time `json:"observed_at"`
	SourceVersion   string    `json:"source_version"`
	PayloadHash     string    `json:"payload_hash"`
}

type dataContactRequest struct {
	ChatID      string    `json:"chat_id"`
	ChatType    ChatType  `json:"chat_type"`
	DisplayName string    `json:"display_name"`
	Remark      string    `json:"remark"`
	ObservedAt  time.Time `json:"observed_at"`
}

type dataBatchAck struct {
	Accepted          bool   `json:"accepted"`
	Duplicate         bool   `json:"duplicate"`
	BatchID           string `json:"batch_id"`
	ConfirmedSequence int64  `json:"confirmed_sequence"`
	ConfirmedCursor   string `json:"confirmed_cursor"`
	Reason            string `json:"reason,omitempty"`
}

type dataSyncView struct {
	State             string     `json:"state"`
	Source            string     `json:"source"`
	LastSuccessAt     *time.Time `json:"last_success_at,omitempty"`
	ConfirmedSequence int64      `json:"confirmed_sequence"`
	ConfirmedCursor   string     `json:"confirmed_cursor,omitempty"`
	BacklogCount      *int64     `json:"backlog_count"`
	ErrorCode         string     `json:"error_code,omitempty"`
	ErrorMessage      string     `json:"error_message,omitempty"`
}

type dataConversationView struct {
	ChatID         string     `json:"chat_id"`
	ChatType       string     `json:"chat_type"`
	Title          string     `json:"title"`
	LastActivityAt *time.Time `json:"last_activity_at,omitempty"`
	Source         string     `json:"source"`
	ObservedAt     time.Time  `json:"observed_at"`
	DirectoryState string     `json:"directory_state"`
}

type dataContactView struct {
	ChatID      string    `json:"id"`
	ChatType    string    `json:"type"`
	DisplayName string    `json:"display_name"`
	Remark      string    `json:"remark"`
	ObservedAt  time.Time `json:"observed_at"`
}

type dataMessageView struct {
	MessageID       string    `json:"message_id"`
	ChatID          string    `json:"chat_id"`
	SourceMessageID string    `json:"source_message_id"`
	Direction       string    `json:"direction"`
	MessageType     string    `json:"message_type"`
	Text            string    `json:"text"`
	SourceTime      time.Time `json:"source_time"`
	ObservedAt      time.Time `json:"observed_at"`
	SourceVersion   string    `json:"source_version"`
}

func (s *Server) registerDataAccounts(nodeID string, registration NodeRegistration) error {
	for _, account := range registration.Accounts {
		if !account.Verified {
			continue
		}
		// A wildcard scope is the connection-authorized form. Keep explicit scopes for
		// legacy/test registrations that have not opted into connection-wide authorization.
		hasWildcard := false
		for _, chat := range account.AllowedChats {
			if chat.ChatID == "*" {
				hasWildcard = true
				break
			}
		}
		scopes := make([]ReportingScope, 0, len(account.AllowedChats))
		if hasWildcard {
			scopes = append(scopes, ReportingScope{ChatID: "*", DataType: "*", ConfigVersion: int(registration.ReportingConfigVersion)})
		} else {
			for _, chat := range account.AllowedChats {
				if chat.ChatID == "" || !validChatType(chat.ChatType) {
					continue
				}
				scopes = append(scopes, ReportingScope{ChatID: chat.ChatID, DataType: "read", ConfigVersion: int(registration.ReportingConfigVersion)})
			}
		}
		store, err := s.accountStores.RegisterAccount(context.Background(), AccountRegistration{
			AccountID:            account.AccountID,
			StableIdentity:       account.AccountID,
			SourceNodeID:         nodeID,
			SourceGeneration:     account.AccountID,
			Verified:             true,
			AuthorizationVersion: int(registration.ReportingConfigVersion),
			ReportingScopes:      scopes,
		})
		if err != nil {
			return err
		}
		_ = store.Close()
	}
	return nil
}

type nodeDataSyncStatusView struct {
	State             string     `json:"state"`
	SourceGeneration  string     `json:"source_generation"`
	ConfirmedSequence int64      `json:"confirmed_sequence"`
	ConfirmedCursor   string     `json:"confirmed_cursor"`
	LastSuccessAt     *time.Time `json:"last_success_at,omitempty"`
}

func (s *Server) nodeDataRoute(w http.ResponseWriter, r *http.Request, nodeID string, parts []string, _ string) error {
	if len(parts) != 3 || parts[0] != "accounts" || parts[2] != "sync-status" || r.Method != http.MethodGet || !validIdentifier(parts[1], 200) {
		return requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}
	}
	accountID := parts[1]
	var authorized bool
	if err := s.store.Read(func(state PersistedState) error {
		node, ok := state.Nodes[nodeID]
		if !ok {
			return requestError{status: http.StatusConflict, code: "NodeNotRegistered", message: "Register the node before reading sync status."}
		}
		authorized = nodeHasAccount(node, accountID)
		return nil
	}); err != nil {
		return err
	}
	if !authorized {
		return requestError{status: http.StatusForbidden, code: "AccountNotAuthorized", message: "The node is not authorized for this account."}
	}
	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()
	streamKey := r.URL.Query().Get("stream_key")
	if streamKey == "" {
		streamKey = "messages"
	}
	if streamKey != "messages" && streamKey != "contacts" {
		return requestError{status: http.StatusBadRequest, code: "InvalidStream", message: "The sync stream is invalid."}
	}
	status, err := store.GetSyncStatus(r.Context(), streamKey)
	if err != nil {
		return requestError{status: http.StatusInternalServerError, code: "DataStatusFailed", message: "The platform sync status query failed."}
	}
	writeJSON(w, http.StatusOK, nodeDataSyncStatusView{State: status.CoverageState, SourceGeneration: status.SourceGeneration, ConfirmedSequence: status.ConfirmedSequence, ConfirmedCursor: status.ConfirmedCursor, LastSuccessAt: status.LastSuccessAt})
	return nil
}

func (s *Server) ingestDataBatch(w http.ResponseWriter, r *http.Request, correlationID string) error {
	nodeID, err := s.authenticateNode(r)
	if err != nil {
		return err
	}
	var request dataBatchRequest
	if err := decodeJSON(r, &request, maxDataBatchBytes); err != nil {
		return err
	}
	if request.NodeID != nodeID || request.AccountID == "" || request.BatchID == "" || request.SourceGeneration == "" || (request.StreamKey != "messages" && request.StreamKey != "contacts") || request.Sequence < 1 || request.PayloadHash == "" || len(request.Messages) > 5000 || len(request.Conversations) > 1000 || len(request.Contacts) > 1000 || request.StreamKey == "contacts" && (request.ContactsSnapshotID == "" || len(request.Messages) > 0 || len(request.Conversations) > 0) || request.StreamKey == "messages" && (len(request.Contacts) > 0 || request.ContactsSnapshotID != "") {
		return requestError{status: http.StatusBadRequest, code: "InvalidDataBatch", message: "The data batch is invalid."}
	}
	var account AccountSummary
	if err := s.store.Read(func(state PersistedState) error {
		node, ok := state.Nodes[nodeID]
		if !ok {
			return requestError{status: http.StatusConflict, code: "NodeNotRegistered", message: "Register the node before sending data."}
		}
		for _, candidate := range node.Accounts {
			if candidate.AccountID == request.AccountID {
				account = candidate
				return nil
			}
		}
		return requestError{status: http.StatusForbidden, code: "AccountNotAuthorized", message: "The node is not authorized for this account."}
	}); err != nil {
		return err
	}
	if !account.Verified {
		return requestError{status: http.StatusForbidden, code: "AccountNotVerified", message: "The account identity is not verified."}
	}
	store, err := s.accountStores.OpenAccount(r.Context(), request.AccountID)
	if err != nil {
		return requestError{status: http.StatusForbidden, code: "AccountDataUnavailable", message: "The account data store is not available."}
	}
	defer store.Close()
	batch := IngestBatch{
		BatchID: request.BatchID, SourceGeneration: request.SourceGeneration, StreamKey: request.StreamKey,
		Sequence: request.Sequence, CursorStart: request.CursorStart, CursorEnd: request.CursorEnd,
		PayloadHash: request.PayloadHash, CoverageState: request.CoverageState,
		ContactsSnapshotID: request.ContactsSnapshotID,
	}
	for _, conversation := range request.Conversations {
		if !validChatType(conversation.ChatType) || conversation.ObservedAt.IsZero() {
			return requestError{status: http.StatusBadRequest, code: "InvalidDataBatch", message: "A conversation record is invalid."}
		}
		batch.Conversations = append(batch.Conversations, ConversationRecord{
			ChatID: conversation.ChatID, ChatType: string(conversation.ChatType), Title: conversation.Title,
			LastActivityAt: conversation.LastActivityAt, Source: conversation.Source,
			ObservedAt: conversation.ObservedAt, DirectoryState: conversation.DirectoryState,
		})
	}
	for _, message := range request.Messages {
		if !validChatType(message.ChatType) || message.SourceTime.IsZero() || message.ObservedAt.IsZero() || message.MessageID == "" || message.ChatID == "" || message.PayloadHash == "" {
			return requestError{status: http.StatusBadRequest, code: "InvalidDataBatch", message: "A message record is invalid."}
		}
		batch.Messages = append(batch.Messages, MessageRecord{
			MessageID: message.MessageID, ChatID: message.ChatID, SourceMessageID: message.SourceMessageID,
			Direction: message.Direction, MessageType: message.MessageType, Text: message.Text,
			SourceTime: message.SourceTime, ObservedAt: message.ObservedAt, SourceVersion: message.SourceVersion,
			PayloadHash: message.PayloadHash,
		})
	}
	for _, contact := range request.Contacts {
		if !validChatType(contact.ChatType) || strings.TrimSpace(contact.ChatID) == "" || len(contact.ChatID) > 512 || len(contact.DisplayName) > 2048 || len(contact.Remark) > 2048 || contact.ObservedAt.IsZero() {
			return requestError{status: http.StatusBadRequest, code: "InvalidDataBatch", message: "A contact record is invalid."}
		}
		batch.Contacts = append(batch.Contacts, ContactRecord{
			ChatID: contact.ChatID, ChatType: string(contact.ChatType), DisplayName: contact.DisplayName,
			Remark: contact.Remark, ObservedAt: contact.ObservedAt,
		})
	}
	result, err := store.ApplyBatch(r.Context(), batch)
	if err != nil {
		switch {
		case errors.Is(err, ErrAccountNotAuthorized):
			return requestError{status: http.StatusForbidden, code: "DataNotAuthorized", message: "The data batch contains data outside the active reporting scopes."}
		case errors.Is(err, ErrBatchConflict):
			return requestError{status: http.StatusConflict, code: "BatchConflict", message: "The batch id or payload conflicts with an existing batch."}
		case errors.Is(err, ErrBatchSequenceGap):
			return requestError{status: http.StatusConflict, code: "BatchSequenceGap", message: "The batch sequence has a gap; resend from the confirmed cursor."}
		case errors.Is(err, ErrBatchSequenceConflict):
			return requestError{status: http.StatusConflict, code: "BatchSequenceConflict", message: "The batch sequence is older than confirmed progress."}
		case errors.Is(err, ErrAccountBindingConflict):
			return requestError{status: http.StatusConflict, code: "AccountBindingConflict", message: "The source generation does not match the verified account binding."}
		case errors.Is(err, ErrAccountCapacityExceeded):
			return requestError{status: http.StatusInsufficientStorage, code: "AccountCapacityExceeded", message: "The account data capacity budget has been reached."}
		default:
			return requestError{status: http.StatusInternalServerError, code: "DataBatchFailed", message: "The platform could not persist the data batch."}
		}
	}
	writeJSON(w, http.StatusOK, dataBatchAck{Accepted: true, Duplicate: result.Duplicate, BatchID: request.BatchID, ConfirmedSequence: result.ConfirmedSequence, ConfirmedCursor: result.ConfirmedCursor, Reason: "Persisted"})
	return nil
}

func validChatType(chatType ChatType) bool {
	return chatType == ChatGroup || chatType == ChatPrivate
}

func (s *Server) dataRoute(w http.ResponseWriter, r *http.Request, correlationID string) error {
	username, err := s.authenticateWeb(r)
	if err != nil {
		return err
	}
	parts := pathParts(r.URL.Path)
	if len(parts) < 5 || parts[0] != "v1" || parts[1] != "data" || parts[2] != "accounts" || !validIdentifier(parts[3], 200) {
		return requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}
	}
	accountID := parts[3]
	if len(parts) != 5 {
		return requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}
	}
	switch parts[4] {
	case "conversations":
		if r.Method != http.MethodGet {
			return requestError{status: http.StatusMethodNotAllowed, code: "MethodNotAllowed", message: "Method is not allowed."}
		}
		return s.listStoredConversations(w, r, accountID)
	case "contacts":
		if r.Method != http.MethodGet {
			return requestError{status: http.StatusMethodNotAllowed, code: "MethodNotAllowed", message: "Method is not allowed."}
		}
		return s.listStoredContacts(w, r, accountID)
	case "messages":
		if r.Method != http.MethodGet {
			return requestError{status: http.StatusMethodNotAllowed, code: "MethodNotAllowed", message: "Method is not allowed."}
		}
		return s.listStoredMessages(w, r, accountID)
	case "sync-status":
		if r.Method != http.MethodGet {
			return requestError{status: http.StatusMethodNotAllowed, code: "MethodNotAllowed", message: "Method is not allowed."}
		}
		return s.storedSyncStatus(w, r, accountID)
	case "refresh":
		if r.Method != http.MethodPost {
			return requestError{status: http.StatusMethodNotAllowed, code: "MethodNotAllowed", message: "Method is not allowed."}
		}
		return s.requestDataRefresh(w, r, accountID, username, correlationID)
	case "revoke":
		if r.Method != http.MethodPost {
			return requestError{status: http.StatusMethodNotAllowed, code: "MethodNotAllowed", message: "Method is not allowed."}
		}
		return s.revokeDataAuthorization(w, r, accountID, username, correlationID)
	default:
		return requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}
	}
}

func (s *Server) listStoredConversations(w http.ResponseWriter, r *http.Request, accountID string) error {
	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, 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."}
	}
	views := make([]dataConversationView, 0, len(items))
	for _, item := range items {
		views = append(views, dataConversationView{item.ChatID, item.ChatType, item.Title, item.LastActivityAt, item.Source, item.ObservedAt, item.DirectoryState})
	}
	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
}

func (s *Server) listStoredContacts(w http.ResponseWriter, r *http.Request, accountID string) error {
	limit := queryLimit(r.URL.Query().Get("limit"))
	offset := 0
	if raw := r.URL.Query().Get("offset"); raw != "" {
		parsed, err := strconv.Atoi(raw)
		if err != nil || parsed < 0 || parsed > 1_000_000 {
			return requestError{status: http.StatusBadRequest, code: "InvalidOffset", message: "The contact offset is invalid."}
		}
		offset = parsed
	}
	groupsOnly := false
	if raw := r.URL.Query().Get("groups_only"); raw != "" {
		parsed, err := strconv.ParseBool(raw)
		if err != nil {
			return requestError{status: http.StatusBadRequest, code: "InvalidGroupFilter", message: "groups_only must be a boolean."}
		}
		groupsOnly = parsed
	}
	contains := r.URL.Query().Get("contains")
	if len(contains) > 200 {
		return requestError{status: http.StatusBadRequest, code: "InvalidQuery", message: "The contact query is too long."}
	}
	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, hasMore, err := store.QueryContacts(r.Context(), contains, groupsOnly, limit, offset)
	if err != nil {
		if errors.Is(err, ErrAccountNotAuthorized) {
			return requestError{status: http.StatusForbidden, code: "DataNotAuthorized", message: "The account has no active contact reporting scope."}
		}
		return requestError{status: http.StatusInternalServerError, code: "DataQueryFailed", message: "The platform contact query failed."}
	}
	status, err := store.GetSyncStatus(r.Context(), "contacts")
	if err != nil {
		return requestError{status: http.StatusInternalServerError, code: "DataStatusFailed", message: "The platform sync status query failed."}
	}
	views := make([]dataContactView, 0, len(items))
	for _, item := range items {
		views = append(views, dataContactView{item.ChatID, item.ChatType, item.DisplayName, item.Remark, item.ObservedAt})
	}
	writeJSON(w, http.StatusOK, map[string]any{"items": views, "limit": limit, "offset": offset, "has_more": hasMore, "sync": syncView(status)})
	return nil
}

func (s *Server) listStoredMessages(w http.ResponseWriter, r *http.Request, accountID string) error {
	chatID := r.URL.Query().Get("chat_id")
	if !validIdentifier(chatID, 512) {
		return requestError{status: http.StatusBadRequest, code: "InvalidChat", message: "chat_id is required."}
	}
	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."}
	}
	defer store.Close()
	allowed, err := store.AuthorizeChat(r.Context(), chatID, "messages")
	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, 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."}
	}
	views := make([]dataMessageView, 0, len(items))
	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})
	}
	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
}

func (s *Server) storedSyncStatus(w http.ResponseWriter, r *http.Request, accountID string) error {
	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()
	status, err := store.GetSyncStatus(r.Context(), "messages")
	if err != nil {
		return requestError{status: http.StatusInternalServerError, code: "DataStatusFailed", message: "The platform sync status query failed."}
	}
	writeJSON(w, http.StatusOK, syncView(status))
	return nil
}

func syncView(status AccountSyncStatus) dataSyncView {
	return dataSyncView{State: status.CoverageState, Source: "platform-cache", LastSuccessAt: status.LastSuccessAt, ConfirmedSequence: status.ConfirmedSequence, ConfirmedCursor: status.ConfirmedCursor, BacklogCount: nil, ErrorCode: status.ErrorCode, ErrorMessage: status.ErrorMessage}
}

func (s *Server) revokeDataAuthorization(w http.ResponseWriter, r *http.Request, accountID, username, correlationID string) error {
	if err := s.accountStores.SetScopes(r.Context(), accountID, nil); err != nil {
		if errors.Is(err, ErrAccountStoreNotFound) {
			return requestError{status: http.StatusNotFound, code: "AccountDataNotFound", message: "No platform data is available for this account."}
		}
		return requestError{status: http.StatusInternalServerError, code: "AuthorizationRevokeFailed", message: "The account authorization could not be revoked."}
	}
	if err := s.appendAudit("user:"+username, "data.authorization.revoke", accountID, correlationID, "success"); err != nil {
		return err
	}
	writeJSON(w, http.StatusOK, map[string]any{"account_id": accountID, "revoked": true})
	return nil
}

func (s *Server) requestDataRefresh(w http.ResponseWriter, r *http.Request, accountID, username, correlationID string) error {
	streamKey := "messages"
	if r.ContentLength > 0 || strings.HasPrefix(r.Header.Get("Content-Type"), "application/json") {
		var request struct {
			StreamKey string `json:"stream_key"`
		}
		if err := decodeJSON(r, &request, 8*1024); err != nil {
			return err
		}
		if request.StreamKey != "" {
			streamKey = request.StreamKey
		}
	}
	if streamKey != "messages" && streamKey != "contacts" {
		return requestError{status: http.StatusBadRequest, code: "InvalidStream", message: "The sync stream is invalid."}
	}
	var nodeID string
	if err := s.store.Read(func(state PersistedState) error {
		for id, node := range state.Nodes {
			for _, account := range node.Accounts {
				if account.AccountID == accountID && account.Verified {
					nodeID = id
					break
				}
			}
			if nodeID != "" {
				break
			}
		}
		return nil
	}); err != nil {
		return err
	}
	if nodeID == "" {
		return requestError{status: http.StatusConflict, code: "NoAuthorizedNode", message: "No verified node is available for this account."}
	}
	payload, _ := json.Marshal(map[string]string{"stream_key": streamKey, "reason": "web-refresh"})
	return s.createTaskSubmission(w, TaskSubmission{NodeID: nodeID, AccountID: accountID, Kind: "sync-data", IdempotencyKey: "sync:" + accountID + ":" + streamKey + ":" + strconv.FormatInt(time.Now().UTC().Unix()/5, 10), Payload: payload}, username, correlationID)
}