resolveWatchers method

@override Future<List<WorkflowWatcherResolution>> resolveWatchers(String topic, Map<String, Object?> payload, { int limit = 256, })

Resolves watchers listening on topic, persisting payload and marking runs ready to resume atomically. Returns the resolved watchers so the runtime can enqueue follow-up work.

The returned WorkflowWatcherResolution objects MUST include the merged suspension metadata (resumeData) that will be exposed to FlowContext.takeResumeData/WorkflowScriptStepContext.takeResumeData.

Implementation

@override
Future<List<WorkflowWatcherResolution>> resolveWatchers(
  String topic,
  Map<String, Object?> payload, {
  int limit = 256,
}) async {
  final topicMap = _watchersByTopic[topic];
  if (topicMap == null || topicMap.isEmpty) return const [];
  final now = _clock.now();
  final ids = topicMap.keys.take(limit).toList(growable: false);
  final results = <WorkflowWatcherResolution>[];
  for (final runId in ids) {
    final record = topicMap.remove(runId);
    if (record == null) continue;
    _watchersByRun.remove(runId);
    final state = _runs[runId];
    if (state == null) {
      final topicSet = _suspendedTopics[topic];
      topicSet?.remove(runId);
      if (topicSet != null && topicSet.isEmpty) {
        _suspendedTopics.remove(topic);
      }
      continue;
    }
    final metadata = Map<String, Object?>.from(record.data);
    metadata['type'] = 'event';
    metadata['topic'] = topic;
    metadata['payload'] = payload;
    metadata
      ..putIfAbsent('step', () => record.stepName)
      ..putIfAbsent(
        'iterationStep',
        () => metadata['step'] ?? record.stepName,
      );
    metadata['deliveredAt'] = now.toIso8601String();
    _runs[runId] = state.copyWith(
      status: WorkflowStatus.running,
      waitTopic: null,
      resumeAt: null,
      suspensionData: _freeze(metadata),
      updatedAt: now,
    );
    for (final entry in _due.values) {
      entry.remove(runId);
    }
    final topicSet = _suspendedTopics[topic];
    topicSet?.remove(runId);
    if (topicSet != null && topicSet.isEmpty) {
      _suspendedTopics.remove(topic);
    }
    results.add(
      WorkflowWatcherResolution(
        runId: runId,
        stepName: record.stepName,
        topic: topic,
        resumeData: metadata,
      ),
    );
  }
  if (topicMap.isEmpty) {
    _watchersByTopic.remove(topic);
  }
  return results;
}