submit method

Future<MissionOutcome> submit(Mission mission, { void onEvent(MissionEvent event)?, })

Submits a mission. If an active coalescing group exists for the mission's key, the caller subscribes to that group's event stream and receives the same events. Otherwise, a new group is created and the mission executes once (FR-001, FR-002).

If idempotency is enabled and a cached outcome exists for the key, it is returned immediately without re-execution (FR-007).

Implementation

Future<MissionOutcome> submit(
  Mission mission, {
  void Function(MissionEvent event)? onEvent,
}) async {
  final cached = _cache.lookup(mission.key);
  if (cached != null) {
    mission.outcome = cached;
    mission.status = MissionStatus.completed;
    onEvent?.call(MissionEventCompleted(mission.id, cached));
    return cached;
  }

  final canonical = mission.key.canonical;
  final existing = _groups[canonical];
  if (existing != null) {
    existing.addSubscriber(mission.callerId);
    _missionIdToKey[mission.id] = canonical;
    final sub = existing.events.listen(onEvent ?? (_) {});
    try {
      return await existing.done;
    } finally {
      await sub.cancel();
    }
  }

  // New coalescing group — execute once.
  final group = CoalescingGroup(mission);
  group.addSubscriber(mission.callerId);
  _groups[canonical] = group;
  _missionIdToKey[mission.id] = canonical;

  final sub = group.events.listen(onEvent ?? (_) {});
  final cancelToken = CancelToken(
    gracePeriod: _config.cancellationGracePeriod,
  );
  group.cancelToken = cancelToken;

  try {
    final execOutcome = await _executor(mission, group, cancelToken);
    // If cancel() ran first, the group is already completed with the
    // salvaged outcome — use that. Otherwise, complete with the
    // executor's outcome.
    if (group.isCompleted) {
      // Cancellation completed the group; return the salvaged outcome.
      final salvaged = group.mission.outcome!;
      _cache.store(mission.key, salvaged);
      return salvaged;
    }
    group.complete(execOutcome);
    _cache.store(mission.key, execOutcome);
    return execOutcome;
  } catch (e, st) {
    final outcome = OutcomeFailed(e, st);
    group.complete(outcome);
    _cache.store(mission.key, outcome);
    return outcome;
  } finally {
    await sub.cancel();
    await group.close();
    _groups.remove(canonical);
    _missionIdToKey.remove(mission.id);
  }
}