registerWatcher method

@override Future<void> registerWatcher(String runId, String stepName, String topic, { DateTime? deadline, Map<String, Object?>? data, })

Registers a durable watcher for topic so the runtime can resume stepName when an event is emitted.

Implementations MUST persist the suspension metadata and watcher record atomically so an incoming payload can be recorded even if no worker is currently running. When a deadline is provided the run should surface in dueRuns once the deadline passes so timeouts can resume the workflow.

Implementation

@override
Future<void> registerWatcher(
  String runId,
  String stepName,
  String topic, {
  DateTime? deadline,
  Map<String, Object?>? data,
}) async {
  final metadata = _prepareSuspensionData(
    data,
    resumeAt: deadline,
    deadline: deadline,
    topic: topic,
  );
  await suspendOnTopic(
    runId,
    stepName,
    topic,
    deadline: deadline,
    data: metadata,
  );
  final record = _WatcherRecord(
    runId: runId,
    stepName: stepName,
    topic: topic,
    createdAt: _clock.now(),
    deadline: deadline,
    data: metadata,
  );
  final topicMap = _watchersByTopic.putIfAbsent(topic, LinkedHashMap.new);
  topicMap[runId] = record;
  _watchersByRun[runId] = record;
}