feat: cache contacts and add forced sync

node-agent/WxAgent.Service/RemoteAgentHostedService.cs

@@ -136,7 +136,7 @@ public sealed class RemoteAgentHostedService(
                     var tasks = await client.PollTasksAsync(accountId, stoppingToken, waitSeconds: 5);
                     foreach (var task in tasks)
                     {
-                        await ProcessTaskAsync(client, remote, reporting, ledger, task, stoppingToken);
+                        await ProcessTaskAsync(client, remote, reporting, ledger, task, dataQueue, syncState, stoppingToken);
                     }
                 }
 
@@ -284,40 +284,47 @@ public sealed class RemoteAgentHostedService(
     {
         foreach (var account in reporting.Accounts.Where(item => item.Enabled))
         {
-            var local = state.GetLatest(account.AccountId, "messages");
-            RemoteDataSyncStatus remote;
-            try
+            var revoked = false;
+            foreach (var streamKey in new[] { "messages", "contacts" })
             {
-                remote = await client.GetDataSyncStatusAsync(account.AccountId, cancellationToken).ConfigureAwait(false);
-            }
-            catch (RemoteClientException exception) when (exception.StatusCode == 404 && exception.Code == "NotFound")
-            {
-                // Older control planes do not expose the reconciliation endpoint; retain the legacy ACK path.
-                continue;
-            }
-            catch (RemoteClientException exception) when (exception.StatusCode == 403 && exception.Code == "AccountNotAuthorized")
-            {
-                var droppedOnRevoke = queue.DropAllForAccount(account.AccountId);
-                blockedAccountIds.Add(account.AccountId);
-                logger.LogWarning(
-                    "Data sync authorization is revoked; accountId={AccountId}; pending content was discarded and collection is paused; droppedPendingBatches={DroppedPendingBatches}.",
-                    account.AccountId, droppedOnRevoke);
-                continue;
-            }
-            catch (HttpRequestException)
-            {
-                // A disconnected platform must not prevent local DB collection.
-                return;
-            }
+                var local = state.GetLatest(account.AccountId, streamKey);
+                RemoteDataSyncStatus remote;
+                try
+                {
+                    remote = await client.GetDataSyncStatusAsync(account.AccountId, cancellationToken, streamKey).ConfigureAwait(false);
+                }
+                catch (RemoteClientException exception) when (exception.StatusCode == 404 && exception.Code == "NotFound")
+                {
+                    // Older control planes do not expose the reconciliation endpoint; retain the legacy ACK path.
+                    continue;
+                }
+                catch (RemoteClientException exception) when (exception.StatusCode == 403 && exception.Code == "AccountNotAuthorized")
+                {
+                    var droppedOnRevoke = queue.DropAllForAccount(account.AccountId);
+                    blockedAccountIds.Add(account.AccountId);
+                    logger.LogWarning(
+                        "Data sync authorization is revoked; accountId={AccountId}; pending content was discarded and collection is paused; droppedPendingBatches={DroppedPendingBatches}.",
+                        account.AccountId, droppedOnRevoke);
+                    revoked = true;
+                    break;
+                }
+                catch (HttpRequestException)
+                {
+                    // A disconnected platform must not prevent local DB collection.
+                    return;
+                }
 
-            if (blockedAccountIds.Remove(account.AccountId))
-                logger.LogInformation("Data sync authorization was restored; accountId={AccountId}; collection is resumed.", account.AccountId);
-            if (!state.Reconcile(remote, account.AccountId, "messages"))
+                if (blockedAccountIds.Remove(account.AccountId))
+                    logger.LogInformation("Data sync authorization was restored; accountId={AccountId}; collection is resumed.", account.AccountId);
+                if (!state.Reconcile(remote, account.AccountId, streamKey))
+                    continue;
+                var dropped = queue.DropForAccount(account.AccountId, streamKey);
+                logger.LogWarning(
+                    "Data sync checkpoint reconciled with the platform; accountId={AccountId}; stream={StreamKey}; localSequence={LocalSequence}; platformSequence={PlatformSequence}; droppedPendingBatches={DroppedPendingBatches}.",
+                    account.AccountId, streamKey, local.ConfirmedSequence, remote.ConfirmedSequence, dropped);
+            }
+            if (revoked)
                 continue;
-            var dropped = queue.DropForAccount(account.AccountId, "messages");
-            logger.LogWarning(
-                "Data sync checkpoint reconciled with the platform; accountId={AccountId}; localSequence={LocalSequence}; platformSequence={PlatformSequence}; droppedPendingBatches={DroppedPendingBatches}.",
-                account.AccountId, local.ConfirmedSequence, remote.ConfirmedSequence, dropped);
         }
     }
 
@@ -347,6 +354,8 @@ public sealed class RemoteAgentHostedService(
         ReportingConfig reporting,
         RemoteTaskLedger ledger,
         RemoteTaskEnvelope task,
+        RemoteDataBatchQueue? dataQueue,
+        RemoteDataSyncStateStore? syncState,
         CancellationToken cancellationToken)
     {
         if (task.Status is RemoteTaskStatus.Accepted or RemoteTaskStatus.Running)
@@ -415,10 +424,56 @@ public sealed class RemoteAgentHostedService(
         var renewed = await client.RenewTaskAsync(started, cancellationToken);
         if (task.Kind == "sync-data")
         {
-            var result = new RemoteTaskResult(task.TaskId, task.AccountId, renewed.LeaseGeneration,
+            var streamKey = task.Payload.ValueKind == JsonValueKind.Object
+                && task.Payload.TryGetProperty("stream_key", out var streamElement)
+                && streamElement.ValueKind == JsonValueKind.String
+                    ? streamElement.GetString() ?? "messages"
+                    : "messages";
+            if (streamKey == "contacts")
+            {
+                try
+                {
+                    if (dataQueue is null || syncState is null)
+                        throw new ServiceException("SyncUnavailable", 503, "The contacts sync queue is not available.");
+                    var queued = await QueueContactsSnapshotAsync(remote.NodeId!, reporting, task.AccountId, dataQueue, syncState, cancellationToken);
+                    var message = queued ? "通讯录同步已加入节点本地队列。" : "已有通讯录同步待上传，本次请求未重复排队。";
+                    var result = new RemoteTaskResult(task.TaskId, task.AccountId, renewed.LeaseGeneration,
+                        RemoteTaskStatus.Succeeded, null, message, false, null, Guid.NewGuid().ToString("N"));
+                    ledger.Complete(result);
+                    await ReportResultAsync(client, ledger, result, reporting, cancellationToken);
+                }
+                catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+                {
+                    throw;
+                }
+                catch (ServiceException exception)
+                {
+                    var result = new RemoteTaskResult(task.TaskId, task.AccountId, renewed.LeaseGeneration,
+                        RemoteTaskStatus.Failed, exception.Code, "通讯录同步失败。", false, null, Guid.NewGuid().ToString("N"));
+                    ledger.Complete(result);
+                    await ReportResultAsync(client, ledger, result, reporting, cancellationToken);
+                }
+                catch (WxAgentException exception)
+                {
+                    var result = new RemoteTaskResult(task.TaskId, task.AccountId, renewed.LeaseGeneration,
+                        RemoteTaskStatus.Failed, exception.Code.ToString(), "通讯录同步失败。", false, null, Guid.NewGuid().ToString("N"));
+                    ledger.Complete(result);
+                    await ReportResultAsync(client, ledger, result, reporting, cancellationToken);
+                }
+                return;
+            }
+            if (streamKey != "messages")
+            {
+                var result = new RemoteTaskResult(task.TaskId, task.AccountId, renewed.LeaseGeneration,
+                    RemoteTaskStatus.Failed, "InvalidTaskPayload", "The sync stream is unsupported.", false, null, Guid.NewGuid().ToString("N"));
+                ledger.Complete(result);
+                await ReportResultAsync(client, ledger, result, reporting, cancellationToken);
+                return;
+            }
+            var messagesResult = new RemoteTaskResult(task.TaskId, task.AccountId, renewed.LeaseGeneration,
                 RemoteTaskStatus.Succeeded, null, "后台数据同步已受理；采集器将在下一轮同步周期执行。", false, null, Guid.NewGuid().ToString("N"));
-            ledger.Complete(result);
-            await ReportResultAsync(client, ledger, result, reporting, cancellationToken);
+            ledger.Complete(messagesResult);
+            await ReportResultAsync(client, ledger, messagesResult, reporting, cancellationToken);
             return;
         }
 
@@ -600,6 +655,105 @@ public sealed class RemoteAgentHostedService(
         }
     }
 
+    private async Task<bool> QueueContactsSnapshotAsync(
+        string nodeId,
+        ReportingConfig reporting,
+        string accountId,
+        RemoteDataBatchQueue dataQueue,
+        RemoteDataSyncStateStore syncState,
+        CancellationToken cancellationToken)
+    {
+        if (dataQueue.Pending().Any(batch => batch.AccountId == accountId && batch.StreamKey == "contacts"))
+            return false;
+
+        var scopes = AuthorizedChatScopes(reporting, accountId);
+        RequireScopes(reporting, accountId, scopes);
+        var authorizedContacts = await ReadContactsForScopesAsync(accountId, scopes, cancellationToken);
+        var authorizedById = authorizedContacts.ToDictionary(contact => contact.Id, StringComparer.Ordinal);
+        var groupIds = new HashSet<string>(StringComparer.Ordinal);
+        for (var offset = 0; ;)
+        {
+            var page = await backend.ContactsAsync(accountId, null, true, 200, offset, cancellationToken);
+            foreach (var group in page.Items)
+            {
+                if (authorizedById.ContainsKey(group.Id))
+                    groupIds.Add(group.Id);
+            }
+            if (!page.HasMore)
+                break;
+            var nextOffset = page.NextOffset ?? offset + page.Items.Count;
+            if (nextOffset <= offset)
+                throw new ServiceException("InvalidPagination", 502, "The contacts page did not advance.");
+            offset = nextOffset;
+        }
+
+        var observedAt = DateTimeOffset.UtcNow;
+        var contacts = authorizedContacts
+            .Select(contact => new RemoteSyncContact(
+                contact.Id,
+                groupIds.Contains(contact.Id) ? ReportingChatType.Group : ReportingChatType.Private,
+                contact.DisplayName ?? string.Empty,
+                contact.Remark ?? string.Empty,
+                observedAt))
+            .Where(contact => scopes.Any(scope => scope.ChatType == contact.ChatType
+                && (scope.ChatId == "*" || scope.ChatId == contact.ChatId)))
+            .GroupBy(contact => contact.ChatId, StringComparer.Ordinal)
+            .Select(group => group.First())
+            .OrderBy(contact => contact.ChatType)
+            .ThenBy(contact => contact.ChatId, StringComparer.Ordinal)
+            .ToArray();
+
+        const string streamKey = "contacts";
+        var checkpoint = syncState.GetLatest(accountId, streamKey);
+        var sourceGeneration = syncState.GetLatest(accountId, "messages").SourceGeneration;
+        if (string.IsNullOrWhiteSpace(sourceGeneration))
+            sourceGeneration = checkpoint.SourceGeneration;
+        if (string.IsNullOrWhiteSpace(sourceGeneration))
+            throw new ServiceException("SyncUnavailable", 503, "The account data source has not been initialized.");
+        var sequence = checkpoint.SourceGeneration == sourceGeneration ? checkpoint.ConfirmedSequence + 1 : 1;
+        var snapshotId = Guid.NewGuid().ToString("N");
+        var pendingContacts = new List<RemoteSyncContact>();
+
+        RemoteSyncBatch CreateBatch(IReadOnlyList<RemoteSyncContact> items, bool complete)
+        {
+            var batch = new RemoteSyncBatch(
+                nodeId, accountId, Guid.NewGuid().ToString("N"), sourceGeneration, streamKey, sequence,
+                "{}", "{}", string.Empty, complete ? "complete" : "partial", [], [])
+            {
+                Contacts = items,
+                ContactsSnapshotId = snapshotId
+            };
+            return batch with { PayloadHash = RemoteDataBatchAuthorization.ComputePayloadHash(batch) };
+        }
+
+        void Enqueue(bool complete)
+        {
+            var batch = CreateBatch(pendingContacts.ToArray(), complete);
+            if (JsonSerializer.SerializeToUtf8Bytes(batch, RemoteJson.Options).Length > RemoteDataProtocol.MaxBatchBytes - 2048)
+                throw new WxAgentException(WxAgentErrorCode.InvalidArgument, "A contacts sync batch exceeds the size limit.");
+            var result = dataQueue.Enqueue(reporting, batch);
+            if (!result.Accepted)
+                throw new ServiceException(result.Reason ?? "SyncQueueFull", 503, "The contacts sync batch was not queued.");
+            sequence++;
+            pendingContacts.Clear();
+        }
+
+        foreach (var contact in contacts)
+        {
+            pendingContacts.Add(contact);
+            if (pendingContacts.Count <= RemoteDataProtocol.MaxBatchItems
+                && JsonSerializer.SerializeToUtf8Bytes(CreateBatch(pendingContacts, false), RemoteJson.Options).Length <= RemoteDataProtocol.MaxBatchBytes - 2048)
+                continue;
+            pendingContacts.RemoveAt(pendingContacts.Count - 1);
+            if (pendingContacts.Count == 0)
+                throw new WxAgentException(WxAgentErrorCode.InvalidArgument, "A contact record exceeds the sync batch size limit.");
+            Enqueue(false);
+            pendingContacts.Add(contact);
+        }
+        Enqueue(true);
+        return true;
+    }
+
     private async Task<IReadOnlyList<ContactInfo>> ReadContactsForScopesAsync(
         string accountId, IReadOnlyList<RemoteReportingScope> scopes, CancellationToken cancellationToken)
     {