825 lines
44 KiB
C#
825 lines
44 KiB
C#
using System.Text.Json;
|
|
using System.Text.Json.Serialization;
|
|
using Microsoft.Extensions.Hosting;
|
|
using Microsoft.Extensions.Logging;
|
|
using WxAgent.Core;
|
|
|
|
namespace WxAgent.Service;
|
|
|
|
public sealed class RemoteAgentHostedService(
|
|
ServiceOptions options,
|
|
IAgentBackend backend,
|
|
ILogger<RemoteAgentHostedService> logger,
|
|
RemoteEventQueue? remoteQueue = null,
|
|
IRemoteDataCollector? dataCollector = null) : BackgroundService
|
|
{
|
|
private static readonly TimeSpan HeartbeatInterval = TimeSpan.FromSeconds(10);
|
|
private static readonly TimeSpan RetryDelay = TimeSpan.FromSeconds(2);
|
|
private readonly RemoteAccountContext accountContext = new();
|
|
|
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
|
{
|
|
RemoteAgentOptions? configuredRemote = options.Remote;
|
|
if (!string.IsNullOrWhiteSpace(options.RemoteConfigurationFile))
|
|
{
|
|
try
|
|
{
|
|
configuredRemote = (await RemoteNodeConfigurationStore.LoadAsync(options.RemoteConfigurationFile, stoppingToken)).Remote;
|
|
}
|
|
catch (Exception exception) when (exception is IOException or JsonException or WxAgentException)
|
|
{
|
|
logger.LogWarning("Remote agent is disabled because the external configuration could not be loaded; type={ExceptionType}.", exception.GetType().Name);
|
|
return;
|
|
}
|
|
}
|
|
if (configuredRemote is null || !configuredRemote.IsConfigured)
|
|
{
|
|
logger.LogInformation("Remote agent is disabled because authAddress, token or nodeId is not configured.");
|
|
return;
|
|
}
|
|
var remote = configuredRemote;
|
|
|
|
using var client = new RemoteControlClient(remote);
|
|
var ledger = new RemoteTaskLedger(Path.Combine(options.DataDirectory, "remote-task-ledger.json"));
|
|
var eventQueue = remoteQueue ?? new RemoteEventQueue(Path.Combine(options.DataDirectory, "remote-event-queue.json"));
|
|
// A configured Agent connection is the data-sync authorization boundary.
|
|
// EnableDataSync remains only as a read-compatibility property for old service.json files.
|
|
var dataQueue = new RemoteDataBatchQueue(
|
|
Path.Combine(options.DataDirectory, "remote-data-queue.json"), options.DataSyncQueueMaxItems, options.DataSyncQueueMaxBytes);
|
|
var syncState = new RemoteDataSyncStateStore(Path.Combine(options.DataDirectory, "remote-data-sync-state.json"));
|
|
var blockedDataSyncAccounts = new HashSet<string>(StringComparer.Ordinal);
|
|
var lastDataSyncAt = DateTimeOffset.MinValue;
|
|
string? registeredActiveAccountId = null;
|
|
bool? registeredActiveAccountVerified = null;
|
|
long registeredReportingConfigVersion = -1;
|
|
ReportingConfig? activeReporting = null;
|
|
var retry = RetryDelay;
|
|
while (!stoppingToken.IsCancellationRequested)
|
|
{
|
|
try
|
|
{
|
|
var runtime = await LoadRuntimeConfigurationAsync(options, stoppingToken);
|
|
if (!runtime.Remote.IsConfigured || !runtime.Remote.Equals(remote))
|
|
{
|
|
logger.LogWarning("Remote configuration changed or is not valid; remote work is paused until the service is restarted.");
|
|
await DelayAsync(RetryDelay, stoppingToken);
|
|
continue;
|
|
}
|
|
var snapshot = await ReadSnapshotAsync(remote, stoppingToken);
|
|
var reporting = ConnectionAuthorizedReporting(runtime.Reporting, snapshot);
|
|
activeReporting = reporting;
|
|
var registrationNeedsRefresh = client.AuthState != RemoteAuthState.Authenticated
|
|
|| !string.Equals(registeredActiveAccountId, snapshot.ActiveAccountId, StringComparison.Ordinal)
|
|
|| registeredActiveAccountVerified != snapshot.ActiveAccountVerified
|
|
|| registeredReportingConfigVersion != reporting.ConfigVersion;
|
|
if (registrationNeedsRefresh)
|
|
{
|
|
await client.RegisterAsync(CreateRegistration(remote, reporting, snapshot), stoppingToken);
|
|
registeredActiveAccountId = snapshot.ActiveAccountId;
|
|
registeredActiveAccountVerified = snapshot.ActiveAccountVerified;
|
|
registeredReportingConfigVersion = reporting.ConfigVersion;
|
|
blockedDataSyncAccounts.Clear();
|
|
retry = RetryDelay;
|
|
}
|
|
|
|
await client.HeartbeatAsync(CreateHeartbeat(remote, reporting, snapshot, null), stoppingToken);
|
|
await ReplayUnreportedResultsAsync(client, ledger, reporting, stoppingToken);
|
|
await client.FlushEventsAsync(eventQueue, reporting, stoppingToken);
|
|
if (dataQueue is not null && syncState is not null && dataCollector is not null
|
|
&& DateTimeOffset.UtcNow - lastDataSyncAt >= TimeSpan.FromSeconds(options.DataSyncIntervalSeconds))
|
|
{
|
|
await ReconcileDataStatusAsync(client, reporting, dataQueue, syncState, blockedDataSyncAccounts, stoppingToken);
|
|
var blockedBeforeFlush = blockedDataSyncAccounts.ToHashSet(StringComparer.Ordinal);
|
|
try
|
|
{
|
|
var confirmedBatches = await client.FlushDataBatchesAsync(dataQueue, reporting, blockedDataSyncAccounts, stoppingToken);
|
|
foreach (var confirmedBatch in confirmedBatches)
|
|
syncState.MarkConfirmed(confirmedBatch);
|
|
foreach (var accountId in blockedDataSyncAccounts.Except(blockedBeforeFlush, StringComparer.Ordinal))
|
|
logger.LogWarning("Data sync authorization was revoked; accountId={AccountId}; pending content was discarded and collection is paused.", accountId);
|
|
}
|
|
catch (Exception exception) when (exception is HttpRequestException or RemoteClientException)
|
|
{
|
|
logger.LogWarning("Data sync delivery deferred; type={ExceptionType}.", exception.GetType().Name);
|
|
}
|
|
// Collection must continue while the platform is unreachable; the durable queue is
|
|
// the offline buffer and will be flushed on the next successful connection. Revoked
|
|
// accounts are the exception: their pending content is discarded and collection stops.
|
|
await CollectDataBatchesAsync(remote, reporting, dataCollector, dataQueue, syncState, blockedDataSyncAccounts, stoppingToken);
|
|
lastDataSyncAt = DateTimeOffset.UtcNow;
|
|
}
|
|
var pollAccountIds = reporting.Accounts
|
|
.Where(account => account.Enabled)
|
|
.Select(account => account.AccountId)
|
|
.ToList();
|
|
if (snapshot.ActiveAccountId is { Length: > 0 } activeAccountId &&
|
|
!pollAccountIds.Contains(activeAccountId, StringComparer.Ordinal))
|
|
{
|
|
// Poll the verified active account even before a Reporting whitelist is configured.
|
|
// The task then returns an explicit authorization result instead of staying Pending.
|
|
pollAccountIds.Add(activeAccountId);
|
|
}
|
|
foreach (var accountId in pollAccountIds)
|
|
{
|
|
var tasks = await client.PollTasksAsync(accountId, stoppingToken, waitSeconds: 5);
|
|
foreach (var task in tasks)
|
|
{
|
|
await ProcessTaskAsync(client, remote, reporting, ledger, task, stoppingToken);
|
|
}
|
|
}
|
|
|
|
await Task.Delay(HeartbeatInterval, stoppingToken);
|
|
}
|
|
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
|
{
|
|
break;
|
|
}
|
|
catch (RemoteClientException exception)
|
|
{
|
|
logger.LogWarning("Remote control-plane request failed; code={Code}; status={StatusCode}; correlationId={CorrelationId}.",
|
|
exception.Code, exception.StatusCode, exception.CorrelationId);
|
|
await TryCollectOfflineDataBatchesAsync(remote, activeReporting, dataCollector, dataQueue, syncState, blockedDataSyncAccounts, stoppingToken);
|
|
await DelayAsync(retry, stoppingToken);
|
|
retry = TimeSpan.FromSeconds(Math.Min(retry.TotalSeconds * 2, 30));
|
|
}
|
|
catch (Exception exception)
|
|
{
|
|
logger.LogWarning("Remote agent cycle failed; type={ExceptionType}.", exception.GetType().Name);
|
|
await TryCollectOfflineDataBatchesAsync(remote, activeReporting, dataCollector, dataQueue, syncState, blockedDataSyncAccounts, stoppingToken);
|
|
await DelayAsync(retry, stoppingToken);
|
|
retry = TimeSpan.FromSeconds(Math.Min(retry.TotalSeconds * 2, 30));
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task CollectDataBatchesAsync(
|
|
RemoteAgentOptions remote,
|
|
ReportingConfig reporting,
|
|
IRemoteDataCollector collector,
|
|
RemoteDataBatchQueue queue,
|
|
RemoteDataSyncStateStore state,
|
|
ISet<string> blockedAccountIds,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
foreach (var account in reporting.Accounts.Where(item => item.Enabled))
|
|
{
|
|
cancellationToken.ThrowIfCancellationRequested();
|
|
if (blockedAccountIds.Contains(account.AccountId))
|
|
continue;
|
|
var scopes = AuthorizedChatScopes(reporting, account.AccountId);
|
|
if (scopes.Count == 0 || queue.Pending().Any(batch => string.Equals(batch.AccountId, account.AccountId, StringComparison.Ordinal)
|
|
&& string.Equals(batch.StreamKey, "messages", StringComparison.Ordinal)))
|
|
continue;
|
|
|
|
var checkpoint = state.GetLatest(account.AccountId, "messages");
|
|
var collection = await collector.CollectAsync(account.AccountId, scopes, checkpoint, cancellationToken).ConfigureAwait(false);
|
|
var sourceChanged = !string.Equals(checkpoint.SourceGeneration, collection.SourceGeneration, StringComparison.Ordinal);
|
|
var cursorStart = sourceChanged ? new Dictionary<string, long>(StringComparer.Ordinal) : checkpoint.ConfirmedCursors;
|
|
var cursorChanged = !cursorStart.OrderBy(item => item.Key).SequenceEqual(collection.NextCursors.OrderBy(item => item.Key));
|
|
if (!sourceChanged && !cursorChanged && !collection.HasNewItems && checkpoint.ConfirmedSequence > 0)
|
|
continue;
|
|
|
|
var sequence = sourceChanged ? 1 : checkpoint.ConfirmedSequence + 1;
|
|
var batch = new RemoteSyncBatch(
|
|
remote.NodeId!, account.AccountId, Guid.NewGuid().ToString("N"), collection.SourceGeneration,
|
|
"messages", sequence, SerializeCursors(cursorStart), SerializeCursors(collection.NextCursors),
|
|
string.Empty, collection.IsComplete ? "complete" : "partial", collection.Conversations, collection.Messages);
|
|
batch = batch with { PayloadHash = RemoteDataBatchAuthorization.ComputePayloadHash(batch) };
|
|
var queued = queue.Enqueue(reporting, batch);
|
|
if (queued.Accepted && queued.Batch is not null)
|
|
state.MarkCollected(queued.Batch, collection.UnavailableSources);
|
|
if (!collection.IsComplete)
|
|
logger.LogWarning("Data sync is partial; accountId={AccountId}; unavailableSources={UnavailableSourceCount}.", account.AccountId, collection.UnavailableSources.Count);
|
|
}
|
|
}
|
|
|
|
private async Task TryCollectOfflineDataBatchesAsync(
|
|
RemoteAgentOptions remote,
|
|
ReportingConfig? reporting,
|
|
IRemoteDataCollector? collector,
|
|
RemoteDataBatchQueue? queue,
|
|
RemoteDataSyncStateStore? state,
|
|
ISet<string> blockedAccountIds,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
if (reporting is null || collector is null || queue is null || state is null)
|
|
return;
|
|
try
|
|
{
|
|
await CollectDataBatchesAsync(remote, reporting, collector, queue, state, blockedAccountIds, cancellationToken).ConfigureAwait(false);
|
|
}
|
|
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
|
{
|
|
throw;
|
|
}
|
|
catch (Exception exception)
|
|
{
|
|
logger.LogWarning("Offline data collection deferred; type={ExceptionType}.", exception.GetType().Name);
|
|
}
|
|
}
|
|
|
|
private async Task ReconcileDataStatusAsync(
|
|
RemoteControlClient client,
|
|
ReportingConfig reporting,
|
|
RemoteDataBatchQueue queue,
|
|
RemoteDataSyncStateStore state,
|
|
ISet<string> blockedAccountIds,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
foreach (var account in reporting.Accounts.Where(item => item.Enabled))
|
|
{
|
|
var local = state.GetLatest(account.AccountId, "messages");
|
|
RemoteDataSyncStatus remote;
|
|
try
|
|
{
|
|
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;
|
|
}
|
|
|
|
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"))
|
|
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);
|
|
}
|
|
}
|
|
|
|
private static string SerializeCursors(IReadOnlyDictionary<string, long> cursors) =>
|
|
JsonSerializer.Serialize(cursors.OrderBy(item => item.Key).ToDictionary(item => item.Key, item => item.Value), RemoteJson.Options);
|
|
|
|
private async Task ProcessTaskAsync(
|
|
RemoteControlClient client,
|
|
RemoteAgentOptions remote,
|
|
ReportingConfig reporting,
|
|
RemoteTaskLedger ledger,
|
|
RemoteTaskEnvelope task,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
if (task.Status is RemoteTaskStatus.Accepted or RemoteTaskStatus.Running)
|
|
{
|
|
if (ledger.TryGet(task.TaskId, out var existing) && existing?.Result is not null)
|
|
{
|
|
await ReportResultAsync(client, ledger, existing.Result, reporting, cancellationToken);
|
|
}
|
|
else
|
|
{
|
|
var result = new RemoteTaskResult(task.TaskId, task.AccountId, task.LeaseGeneration,
|
|
RemoteTaskStatus.ResultUnconfirmed, "AgentRestartedWithIncompleteTask", "Execution was not replayed.", true, null, Guid.NewGuid().ToString("N"));
|
|
ledger.Complete(result);
|
|
await ReportResultAsync(client, ledger, result, reporting, cancellationToken);
|
|
}
|
|
return;
|
|
}
|
|
if (task.Status != RemoteTaskStatus.Pending)
|
|
return;
|
|
if (!ledger.Accept(task))
|
|
{
|
|
if (ledger.TryGet(task.TaskId, out var existing) && existing?.Result is not null)
|
|
await ReportResultAsync(client, ledger, existing.Result, reporting, cancellationToken);
|
|
return;
|
|
}
|
|
|
|
var accepted = await client.AcknowledgeTaskAsync(task, cancellationToken);
|
|
ledger.MarkAccepted(accepted);
|
|
if (accepted.Status == RemoteTaskStatus.Cancelled || accepted.CancelRequestedAt is not null)
|
|
{
|
|
var result = new RemoteTaskResult(accepted.TaskId, accepted.AccountId, accepted.LeaseGeneration,
|
|
RemoteTaskStatus.Cancelled, "CancelRequestedBeforeExecution", "Execution was not started.", false, null, Guid.NewGuid().ToString("N"));
|
|
ledger.Complete(result);
|
|
if (accepted.Status != RemoteTaskStatus.Cancelled)
|
|
await ReportResultAsync(client, ledger, result, reporting, cancellationToken);
|
|
else
|
|
ledger.MarkReported(result.TaskId);
|
|
return;
|
|
}
|
|
|
|
var currentSnapshot = await ReadSnapshotAsync(remote, cancellationToken);
|
|
if (!string.Equals(currentSnapshot.ActiveAccountId, task.AccountId, StringComparison.Ordinal) || !currentSnapshot.ActiveAccountVerified)
|
|
{
|
|
var result = new RemoteTaskResult(task.TaskId, task.AccountId, task.LeaseGeneration,
|
|
RemoteTaskStatus.Failed, "AccountContextUnconfirmed", "The active account binding could not be confirmed.", false, null, Guid.NewGuid().ToString("N"));
|
|
ledger.Complete(result);
|
|
await ReportResultAsync(client, ledger, result, reporting, cancellationToken);
|
|
return;
|
|
}
|
|
|
|
// Persist the local running barrier before asking the center to mark the lease Running.
|
|
ledger.MarkRunning(accepted);
|
|
var started = await client.StartTaskAsync(accepted, cancellationToken);
|
|
if (started.Status != RemoteTaskStatus.Running)
|
|
{
|
|
var terminalStatus = started.Status is RemoteTaskStatus.Cancelled or RemoteTaskStatus.Expired
|
|
? started.Status
|
|
: RemoteTaskStatus.ResultUnconfirmed;
|
|
var result = new RemoteTaskResult(task.TaskId, task.AccountId, task.LeaseGeneration,
|
|
terminalStatus, "ExecutionStartUnconfirmed", "Execution was not started after lease validation.", false, null, Guid.NewGuid().ToString("N"));
|
|
ledger.Complete(result);
|
|
ledger.MarkReported(result.TaskId);
|
|
return;
|
|
}
|
|
|
|
var renewed = await client.RenewTaskAsync(started, cancellationToken);
|
|
if (task.Kind == "sync-data")
|
|
{
|
|
var result = 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);
|
|
return;
|
|
}
|
|
|
|
if (task.Kind == "send-text")
|
|
{
|
|
if (!TryReadSendTextPayload(task.Payload, out var targetId, out var text))
|
|
{
|
|
var result = new RemoteTaskResult(task.TaskId, task.AccountId, renewed.LeaseGeneration,
|
|
RemoteTaskStatus.Failed, "InvalidTaskPayload", "The task payload did not pass the node command allowlist.", false, null, Guid.NewGuid().ToString("N"));
|
|
ledger.Complete(result);
|
|
await ReportResultAsync(client, ledger, result, reporting, cancellationToken);
|
|
return;
|
|
}
|
|
|
|
try
|
|
{
|
|
await backend.SendTextAsync(task.AccountId, targetId, text, cancellationToken);
|
|
var result = new RemoteTaskResult(task.TaskId, task.AccountId, renewed.LeaseGeneration,
|
|
RemoteTaskStatus.Succeeded, null, null, true, 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, "The node operation failed.", true, 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(), "The node operation failed.", true, null, Guid.NewGuid().ToString("N"));
|
|
ledger.Complete(result);
|
|
await ReportResultAsync(client, ledger, result, reporting, cancellationToken);
|
|
}
|
|
return;
|
|
}
|
|
|
|
if (!IsReadTaskKind(task.Kind))
|
|
{
|
|
var result = new RemoteTaskResult(task.TaskId, task.AccountId, renewed.LeaseGeneration,
|
|
RemoteTaskStatus.Failed, "UnsupportedTask", "The task kind is not enabled on this node.", false, null, Guid.NewGuid().ToString("N"));
|
|
ledger.Complete(result);
|
|
await ReportResultAsync(client, ledger, result, reporting, cancellationToken);
|
|
return;
|
|
}
|
|
|
|
try
|
|
{
|
|
var read = await ExecuteReadTaskAsync(task, reporting, cancellationToken);
|
|
var result = new RemoteTaskResult(task.TaskId, task.AccountId, renewed.LeaseGeneration,
|
|
RemoteTaskStatus.Succeeded, null, null, false, read.Content, Guid.NewGuid().ToString("N"));
|
|
ledger.Complete(result, read.ChatScopes);
|
|
await ReportResultAsync(client, ledger, result, reporting, cancellationToken, chatScopes: read.ChatScopes);
|
|
}
|
|
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
|
{
|
|
throw;
|
|
}
|
|
catch (ServiceException exception)
|
|
{
|
|
var result = new RemoteTaskResult(task.TaskId, task.AccountId, renewed.LeaseGeneration,
|
|
RemoteTaskStatus.Failed, exception.Code, "The node read operation failed.", 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(), "The node read operation failed.", false, null, Guid.NewGuid().ToString("N"));
|
|
ledger.Complete(result);
|
|
await ReportResultAsync(client, ledger, result, reporting, cancellationToken);
|
|
}
|
|
}
|
|
|
|
private async Task<RemoteReadExecution> ExecuteReadTaskAsync(
|
|
RemoteTaskEnvelope task, ReportingConfig reporting, CancellationToken cancellationToken)
|
|
{
|
|
switch (task.Kind)
|
|
{
|
|
case "read-sessions":
|
|
{
|
|
var payload = ReadPayload<ReadSessionsPayload>(task.Payload);
|
|
ValidatePage(payload.Limit, payload.Offset);
|
|
var scopes = AuthorizedChatScopes(reporting, task.AccountId);
|
|
RequireScopes(reporting, task.AccountId, scopes);
|
|
var sessions = await backend.SessionsAsync(task.AccountId, cancellationToken);
|
|
var contacts = await ReadContactsForScopesAsync(task.AccountId, scopes, cancellationToken);
|
|
var visibleSessions = sessions
|
|
.Where(session => scopes.Any(scope => SessionMatchesScope(session, scope, contacts)))
|
|
.ToArray();
|
|
var matchedScopeCount = scopes.Count(scope => sessions.Any(session => SessionMatchesScope(session, scope, contacts)));
|
|
var coverage = CreateCoverage(
|
|
matchedScopeCount == scopes.Count
|
|
? ReadCoverageStates.Complete
|
|
: matchedScopeCount == 0 ? ReadCoverageStates.Unknown : ReadCoverageStates.Partial,
|
|
"uia-session-list",
|
|
sessions.Count,
|
|
scopes.Count,
|
|
matchedScopeCount,
|
|
matchedScopeCount == 0 ? "ScopeIdentityNotObserved" : null);
|
|
var page = visibleSessions.ToPage(payload.Limit, payload.Offset) with { Coverage = coverage };
|
|
return new RemoteReadExecution(JsonSerializer.SerializeToElement(page, RemoteJson.Options), scopes);
|
|
}
|
|
case "read-contacts":
|
|
{
|
|
var payload = ReadPayload<ReadContactsPayload>(task.Payload);
|
|
ValidatePage(payload.Limit, payload.Offset);
|
|
var chatType = payload.GroupsOnly ? ReportingChatType.Group : ReportingChatType.Private;
|
|
var scopes = AuthorizedChatScopes(reporting, task.AccountId, chatType);
|
|
RequireScopes(reporting, task.AccountId, scopes);
|
|
var all = new List<ContactInfo>();
|
|
for (var offset = 0; ;)
|
|
{
|
|
var page = await backend.ContactsAsync(task.AccountId, payload.Contains, payload.GroupsOnly, 200, offset, cancellationToken);
|
|
all.AddRange(page.Items);
|
|
if (!page.HasMore) break;
|
|
var next = page.NextOffset ?? offset + page.Items.Count;
|
|
if (next <= offset)
|
|
throw new ServiceException("InvalidPage", 500, "The node returned a non-advancing contact page.");
|
|
offset = next;
|
|
}
|
|
var allChats = scopes.Any(scope => scope.ChatId == "*");
|
|
var allowed = scopes.Select(scope => scope.ChatId).ToHashSet(StringComparer.Ordinal);
|
|
var items = allChats ? all.ToArray() : all.Where(contact => allowed.Contains(contact.Id)).ToArray();
|
|
var matchedScopeCount = scopes.Count(scope => all.Any(contact => string.Equals(contact.Id, scope.ChatId, StringComparison.Ordinal)));
|
|
var coverage = CreateCoverage(
|
|
matchedScopeCount == scopes.Count
|
|
? ReadCoverageStates.Complete
|
|
: matchedScopeCount == 0 ? ReadCoverageStates.Unknown : ReadCoverageStates.Partial,
|
|
"uia-contact-list",
|
|
all.Count,
|
|
scopes.Count,
|
|
matchedScopeCount,
|
|
matchedScopeCount == 0 ? "ScopeIdentityNotObserved" : null);
|
|
var resultPage = items.ToPage(payload.Limit, payload.Offset) with { Coverage = coverage };
|
|
return new RemoteReadExecution(JsonSerializer.SerializeToElement(resultPage, RemoteJson.Options), scopes);
|
|
}
|
|
case "read-messages":
|
|
{
|
|
var payload = ReadPayload<ReadMessagesPayload>(task.Payload);
|
|
ValidatePage(payload.Limit, payload.Offset);
|
|
var scopes = AuthorizedChatScopes(reporting, task.AccountId);
|
|
RequireScopes(reporting, task.AccountId, scopes);
|
|
var sessions = await backend.SessionsAsync(task.AccountId, cancellationToken);
|
|
var contacts = await ReadContactsForScopesAsync(task.AccountId, scopes, cancellationToken);
|
|
var candidateSessions = sessions
|
|
.Where(session => string.Equals(session.AutomationId, payload.ChatId, StringComparison.Ordinal))
|
|
.ToArray();
|
|
var scope = scopes.FirstOrDefault(allowed => string.Equals(allowed.ChatId, payload.ChatId, StringComparison.Ordinal));
|
|
if (scope is null)
|
|
{
|
|
var mapped = sessions
|
|
.Where(session => scopes.Any(allowed => SessionMatchesScope(session, allowed, contacts)))
|
|
.Where(session => string.Equals(session.AutomationId, payload.ChatId, StringComparison.Ordinal))
|
|
.ToArray();
|
|
if (mapped.Length == 1)
|
|
{
|
|
candidateSessions = mapped;
|
|
scope = scopes.First(allowed => SessionMatchesScope(mapped[0], allowed, contacts));
|
|
}
|
|
}
|
|
if (scope is null)
|
|
throw new ServiceException("ChatNotAuthorized", 403, "The requested chat is not enabled in the local whitelist.");
|
|
if (candidateSessions.Length != 1)
|
|
throw new ServiceException("ChatIdentityUnconfirmed", 409, "The requested chat identity is not uniquely visible.");
|
|
var messages = await backend.MessagesAsync(task.AccountId, candidateSessions[0].AutomationId, payload.IncludeContent, cancellationToken);
|
|
var coverage = CreateCoverage(ReadCoverageStates.Complete, "uia-session-messages", messages.Count, 1, 1);
|
|
var page = messages.ToPage(payload.Limit, payload.Offset) with { Coverage = coverage };
|
|
return new RemoteReadExecution(JsonSerializer.SerializeToElement(page, RemoteJson.Options), scopes);
|
|
}
|
|
default:
|
|
throw new ServiceException("UnsupportedTask", 400, "The task kind is not a read operation.");
|
|
}
|
|
}
|
|
|
|
private async Task<IReadOnlyList<ContactInfo>> ReadContactsForScopesAsync(
|
|
string accountId, IReadOnlyList<RemoteReportingScope> scopes, CancellationToken cancellationToken)
|
|
{
|
|
var contacts = new Dictionary<string, ContactInfo>(StringComparer.Ordinal);
|
|
if (scopes.Any(scope => scope.ChatId == "*"))
|
|
{
|
|
for (var offset = 0; ;)
|
|
{
|
|
var page = await backend.ContactsAsync(accountId, null, null, 200, offset, cancellationToken);
|
|
foreach (var contact in page.Items)
|
|
contacts[contact.Id] = contact;
|
|
if (!page.HasMore) break;
|
|
var next = page.NextOffset ?? offset + page.Items.Count;
|
|
if (next <= offset)
|
|
throw new ServiceException("InvalidPage", 500, "The node returned a non-advancing contact page.");
|
|
offset = next;
|
|
}
|
|
return contacts.Values.ToArray();
|
|
}
|
|
foreach (var scope in scopes)
|
|
{
|
|
for (var offset = 0; ;)
|
|
{
|
|
var page = await backend.ContactsAsync(
|
|
accountId,
|
|
contains: scope.ChatId,
|
|
groupsOnly: scope.ChatType == ReportingChatType.Group,
|
|
200,
|
|
offset,
|
|
cancellationToken);
|
|
foreach (var contact in page.Items.Where(contact => string.Equals(contact.Id, scope.ChatId, StringComparison.Ordinal)))
|
|
contacts[contact.Id] = contact;
|
|
if (!page.HasMore) break;
|
|
var next = page.NextOffset ?? offset + page.Items.Count;
|
|
if (next <= offset)
|
|
throw new ServiceException("InvalidPage", 500, "The node returned a non-advancing contact page.");
|
|
offset = next;
|
|
}
|
|
}
|
|
return contacts.Values.ToArray();
|
|
}
|
|
|
|
private static bool SessionMatchesScope(SessionInfo session, RemoteReportingScope scope, IReadOnlyList<ContactInfo> contacts)
|
|
{
|
|
if (scope.ChatId == "*")
|
|
return true;
|
|
if (string.Equals(scope.ChatId, session.AutomationId, StringComparison.Ordinal) ||
|
|
string.Equals(scope.ChatId, session.Name, StringComparison.Ordinal))
|
|
return true;
|
|
var matches = contacts.Where(contact => string.Equals(contact.Id, scope.ChatId, StringComparison.Ordinal)).ToArray();
|
|
return matches.Length == 1 && matches[0] is { } contact &&
|
|
(string.Equals(contact.DisplayName, session.Name, StringComparison.Ordinal) ||
|
|
string.Equals(contact.Remark, session.Name, StringComparison.Ordinal));
|
|
}
|
|
|
|
private static IReadOnlyList<RemoteReportingScope> AuthorizedChatScopes(
|
|
ReportingConfig reporting, string accountId, ReportingChatType? chatType = null) =>
|
|
(reporting.FindAccount(accountId)?.AllowedChats ?? [])
|
|
.Where(chat => chat.Enabled && chat.IdentityVerified && (chatType is null || chat.Type == chatType))
|
|
.Select(chat => new RemoteReportingScope(chat.ChatId, chat.Type))
|
|
.Where(scope => ReportingAuthorization.IsAllowed(reporting, accountId, scope.ChatId, scope.ChatType, ReportingDataType.TaskResult))
|
|
.Distinct()
|
|
.ToArray();
|
|
|
|
private static void RequireScopes(ReportingConfig reporting, string accountId, IReadOnlyList<RemoteReportingScope> scopes)
|
|
{
|
|
if (!reporting.Enabled)
|
|
throw new ServiceException("ReportingDisabled", 403, "Reporting is disabled on the node.");
|
|
var account = reporting.FindAccount(accountId);
|
|
if (account is null || !account.Enabled)
|
|
throw new ServiceException("AccountNotAuthorized", 403, "The target account is not enabled in the local Reporting configuration.");
|
|
if (scopes.Count == 0)
|
|
throw new ServiceException("ChatNotAuthorized", 403, "The requested data scope is not enabled in the local whitelist.");
|
|
}
|
|
|
|
private static T ReadPayload<T>(JsonElement payload)
|
|
{
|
|
try
|
|
{
|
|
return payload.Deserialize<T>(RemoteJson.Options)
|
|
?? throw new ServiceException("InvalidTaskPayload", 400, "The read task payload is empty.");
|
|
}
|
|
catch (JsonException)
|
|
{
|
|
throw new ServiceException("InvalidTaskPayload", 400, "The read task payload is invalid.");
|
|
}
|
|
}
|
|
|
|
private static ReadCoverage CreateCoverage(
|
|
string state,
|
|
string source,
|
|
int observedCount,
|
|
int authorizedScopeCount,
|
|
int matchedScopeCount,
|
|
string? errorCode = null) =>
|
|
new(state, source, observedCount, authorizedScopeCount, matchedScopeCount, DateTimeOffset.UtcNow, errorCode);
|
|
|
|
private static void ValidatePage(int limit, int offset)
|
|
{
|
|
if (limit is < 1 or > 200 || offset < 0)
|
|
throw new ServiceException("InvalidPagination", 400, "limit must be 1..200 and offset must be non-negative.");
|
|
}
|
|
|
|
private static bool IsReadTaskKind(string kind) => kind is "read-sessions" or "read-contacts" or "read-messages";
|
|
|
|
private sealed record ReadSessionsPayload(
|
|
[property: JsonPropertyName("limit")] int Limit,
|
|
[property: JsonPropertyName("offset")] int Offset);
|
|
|
|
private sealed record ReadContactsPayload(
|
|
[property: JsonPropertyName("limit")] int Limit,
|
|
[property: JsonPropertyName("offset")] int Offset,
|
|
[property: JsonPropertyName("contains")] string? Contains,
|
|
[property: JsonPropertyName("groups_only")] bool GroupsOnly);
|
|
|
|
private sealed record ReadMessagesPayload(
|
|
[property: JsonPropertyName("limit")] int Limit,
|
|
[property: JsonPropertyName("offset")] int Offset,
|
|
[property: JsonPropertyName("chat_id")] string ChatId,
|
|
[property: JsonPropertyName("include_content")] bool IncludeContent);
|
|
|
|
private sealed record RemoteReadExecution(
|
|
JsonElement Content,
|
|
IReadOnlyList<RemoteReportingScope> ChatScopes);
|
|
|
|
private static async Task<RemoteNodeConfiguration> LoadRuntimeConfigurationAsync(ServiceOptions options, CancellationToken cancellationToken)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(options.RemoteConfigurationFile))
|
|
return new RemoteNodeConfiguration { Remote = options.Remote ?? new RemoteAgentOptions(), Reporting = options.Reporting };
|
|
return await RemoteNodeConfigurationStore.LoadAsync(options.RemoteConfigurationFile, cancellationToken);
|
|
}
|
|
|
|
private static async Task ReplayUnreportedResultsAsync(
|
|
RemoteControlClient client,
|
|
RemoteTaskLedger ledger,
|
|
ReportingConfig reporting,
|
|
CancellationToken cancellationToken)
|
|
{
|
|
foreach (var pending in ledger.UnreportedResultsWithScopes())
|
|
await ReportResultAsync(client, ledger, pending.Result, reporting, cancellationToken, pending.ReportingScopes);
|
|
}
|
|
|
|
private static async Task ReportResultAsync(
|
|
RemoteControlClient client,
|
|
RemoteTaskLedger ledger,
|
|
RemoteTaskResult result,
|
|
ReportingConfig reporting,
|
|
CancellationToken cancellationToken,
|
|
IReadOnlyList<RemoteReportingScope>? chatScopes = null)
|
|
{
|
|
await client.SendTaskResultAsync(result, reporting, chatScopes: chatScopes, cancellationToken: cancellationToken);
|
|
ledger.MarkReported(result.TaskId);
|
|
}
|
|
|
|
private async Task<BackendSnapshot> ReadSnapshotAsync(RemoteAgentOptions remote, CancellationToken cancellationToken)
|
|
{
|
|
try
|
|
{
|
|
var status = await backend.StatusAsync(cancellationToken);
|
|
var element = JsonSerializer.SerializeToElement(status, RemoteJson.Options);
|
|
var wechatRunning = GetBoolean(element, "wechatAvailable");
|
|
var sessionAvailable = GetBoolean(element, "sessionAvailable");
|
|
var sessionLocked = GetBoolean(element, "sessionLocked");
|
|
// Heartbeats must not refresh UI identity. Binding already contains the verified identity;
|
|
// the explicit accounts API remains the only path that may inspect the profile.
|
|
var accounts = await backend.AccountsAsync(cancellationToken, refreshUiIdentity: false);
|
|
var identities = accounts.Select(account => new RemoteAccountIdentity(
|
|
account.AccountId,
|
|
account.IsUiBindingKnown && account.Binding is not null && string.Equals(account.BindingStatus, "Bound", StringComparison.Ordinal))).ToArray();
|
|
var activeAccountId = remote.ActiveAccountId;
|
|
var boundAccounts = accounts
|
|
.Where(account => account.IsUiBindingKnown && account.Binding is not null && string.Equals(account.BindingStatus, "Bound", StringComparison.Ordinal))
|
|
.Select(account => account.AccountId)
|
|
.Distinct(StringComparer.OrdinalIgnoreCase)
|
|
.ToArray();
|
|
if ((string.IsNullOrWhiteSpace(activeAccountId) || !identities.Any(identity => identity.Verified && string.Equals(identity.AccountId, activeAccountId, StringComparison.Ordinal)))
|
|
&& boundAccounts.Length == 1)
|
|
{
|
|
// A single verified binding is safe to use when the optional GUI value is empty or a display name.
|
|
activeAccountId = boundAccounts[0];
|
|
}
|
|
var activeAccountVerified = false;
|
|
if (activeAccountId is { Length: > 0 })
|
|
{
|
|
try
|
|
{
|
|
accountContext.SwitchTo(activeAccountId, identities);
|
|
activeAccountVerified = accountContext.IsConfirmedFor(activeAccountId);
|
|
}
|
|
catch (WxAgentException)
|
|
{
|
|
accountContext.Invalidate();
|
|
}
|
|
}
|
|
else
|
|
{
|
|
accountContext.Invalidate();
|
|
}
|
|
var nodeStatus = sessionLocked ? RemoteNodeStatus.SessionLocked
|
|
: !wechatRunning ? RemoteNodeStatus.WechatNotRunning
|
|
: !sessionAvailable ? RemoteNodeStatus.WechatNotLoggedIn
|
|
: !activeAccountVerified ? RemoteNodeStatus.Degraded
|
|
: RemoteNodeStatus.Online;
|
|
return new BackendSnapshot(nodeStatus, wechatRunning, sessionAvailable, sessionLocked, activeAccountId, 0, activeAccountVerified, identities);
|
|
}
|
|
catch
|
|
{
|
|
return new BackendSnapshot(RemoteNodeStatus.Degraded, false, false, false, remote.ActiveAccountId, 0, false, []);
|
|
}
|
|
}
|
|
|
|
private static ReportingConfig ConnectionAuthorizedReporting(ReportingConfig configured, BackendSnapshot snapshot)
|
|
{
|
|
var accounts = snapshot.Accounts
|
|
.Where(identity => identity.Verified)
|
|
.Select(identity => new AccountReportingConfig
|
|
{
|
|
AccountId = identity.AccountId,
|
|
Enabled = true,
|
|
AllowedChats =
|
|
[
|
|
new AllowedChat { Type = ReportingChatType.Group, ChatId = "*", Enabled = true, IdentityVerified = true },
|
|
new AllowedChat { Type = ReportingChatType.Private, ChatId = "*", Enabled = true, IdentityVerified = true }
|
|
]
|
|
})
|
|
.ToArray();
|
|
return configured with
|
|
{
|
|
Enabled = true,
|
|
ConfigVersion = Math.Max(1, configured.ConfigVersion),
|
|
Accounts = accounts
|
|
};
|
|
}
|
|
|
|
private static RemoteNodeRegistration CreateRegistration(RemoteAgentOptions remote, ReportingConfig reporting, BackendSnapshot snapshot) =>
|
|
new(remote.NodeId!, typeof(RemoteAgentHostedService).Assembly.GetName().Version?.ToString() ?? "dev",
|
|
RemoteProtocol.Version, ["heartbeat", "poll-tasks", "send-text", "read-sessions", "read-contacts", "read-messages", "db-messages", "db-merged", "report-message"], reporting.ConfigVersion,
|
|
snapshot.Accounts
|
|
.Where(identity => identity.Verified)
|
|
.Select(identity =>
|
|
{
|
|
var account = reporting.FindAccount(identity.AccountId);
|
|
var summary = new RemoteAccountSummary(
|
|
identity.AccountId,
|
|
string.Equals(identity.AccountId, snapshot.ActiveAccountId, StringComparison.Ordinal),
|
|
identity.Verified,
|
|
account?.AllowedChats.Count(chat => chat.Type == ReportingChatType.Group && chat.Enabled && chat.IdentityVerified) ?? 0,
|
|
account?.AllowedChats.Count(chat => chat.Type == ReportingChatType.Private && chat.Enabled && chat.IdentityVerified) ?? 0);
|
|
return summary with
|
|
{
|
|
AllowedChats = account?.AllowedChats
|
|
.Where(chat => chat.Enabled && chat.IdentityVerified)
|
|
.Select(chat => new RemoteAllowedChatSummary(chat.ChatId, chat.Type))
|
|
.ToArray() ?? []
|
|
};
|
|
}).ToArray());
|
|
|
|
private static RemoteHeartbeat CreateHeartbeat(RemoteAgentOptions remote, ReportingConfig reporting, BackendSnapshot snapshot, string? errorCode) =>
|
|
new(remote.NodeId!, typeof(RemoteAgentHostedService).Assembly.GetName().Version?.ToString() ?? "dev",
|
|
RemoteProtocol.Version, snapshot.Status, snapshot.WechatRunning, snapshot.WechatLoggedIn,
|
|
snapshot.SessionLocked, snapshot.ActiveAccountVerified ? snapshot.ActiveAccountId : null, snapshot.QueueLength, reporting.ConfigVersion,
|
|
Guid.NewGuid().ToString("N"), errorCode);
|
|
|
|
private static bool TryReadSendTextPayload(JsonElement payload, out string targetId, out string text)
|
|
{
|
|
targetId = "";
|
|
text = "";
|
|
if (payload.ValueKind != JsonValueKind.Object
|
|
|| !payload.TryGetProperty("target_id", out var target)
|
|
|| !payload.TryGetProperty("text", out var content)
|
|
|| !payload.TryGetProperty("confirmed", out var confirmed)
|
|
|| target.ValueKind != JsonValueKind.String || content.ValueKind != JsonValueKind.String
|
|
|| confirmed.ValueKind != JsonValueKind.True)
|
|
return false;
|
|
targetId = target.GetString() ?? "";
|
|
text = content.GetString() ?? "";
|
|
return targetId.Length is > 0 and <= 512 && text.Length is > 0 and <= 4000;
|
|
}
|
|
|
|
private static bool GetBoolean(JsonElement value, string propertyName) =>
|
|
value.TryGetProperty(propertyName, out var property) && property.ValueKind == JsonValueKind.True;
|
|
|
|
private static async Task DelayAsync(TimeSpan delay, CancellationToken cancellationToken) =>
|
|
await Task.Delay(delay, cancellationToken);
|
|
|
|
private sealed record BackendSnapshot(
|
|
RemoteNodeStatus Status,
|
|
bool WechatRunning,
|
|
bool WechatLoggedIn,
|
|
bool SessionLocked,
|
|
string? ActiveAccountId,
|
|
int QueueLength,
|
|
bool ActiveAccountVerified,
|
|
IReadOnlyList<RemoteAccountIdentity> Accounts);
|
|
}
|