# Async Orchestration — handle, source, cancel, and RACE patterns
# Demonstrates: async sub-flows with handle:, selective waitFor with source:,
# cancellation via cancel:, and group RACE for alternative strategies.

flowmarkup:
  title: Data Pipeline Orchestrator
  version: 1
  description: >
    Launches async sub-flows for a data pipeline, monitors progress,
    and cancels slow extractors in favor of alternatives.
  requires:
    SERVICES: [validator]
    SUBFLOWS: true

  input:
    dataset_id: STRING
    source_config:
      $kind: MAP
      _notes_: Configuration for data extraction sources
  output:
    records_loaded: NUMBER
    validation_status: STRING

  events:
    extraction_complete:
      _notes_: Emitted by extractor sub-flow when done
      data:
        record_count: NUMBER
        source: STRING
    transform_complete:
      _notes_: Emitted by transformer sub-flow when done
      data:
        records_transformed: NUMBER

  examples:
    successful_pipeline:
      _notes_: Pipeline completes with primary extractor
      input:
        dataset_id: DS-100
        source_config:
          primary: s3://data/input
          fallback: gs://backup/input
      output:
        records_loaded: 5000
        validation_status: passed

  vars:
    record_count: 0
    validation_status: pending

  do:
    # Launch primary extractor as async sub-flow with handle
  - run:
      _label_: Primary extractor
      flow: pipeline/extract.flowmarkup.yaml
      handle: primary_extract
      async: true
      cap:
        STORAGE: INHERIT
      params:
        dataset_id: =dataset_id
        source: =source_config.primary

    # Launch fallback extractor in parallel — we'll race them
  - run:
      _label_: Fallback extractor
      flow: pipeline/extract.flowmarkup.yaml
      handle: fallback_extract
      async: true
      cap:
        STORAGE: INHERIT
      params:
        dataset_id: =dataset_id
        source: =source_config.fallback

    # Do independent validation work while extractors run
  - call:
      _label_: Validate schema
      service: validator
      operation: check_schema
      params:
        dataset_id: =dataset_id
      result:
        validation_status: =RESULT.status

    # Race: wait for whichever extractor finishes first
  - race:
      _label_: Race extractors
      primary_wins:
      - waitFor:
          event: extraction_complete
          source: =primary_extract
          timeout: 30m
          capture: [record_count]
      - cancel: =fallback_extract
      - log: "Primary extractor won with {{record_count}} records"
      fallback_wins:
      - waitFor:
          event: extraction_complete
          source: =fallback_extract
          timeout: 30m
          capture: [record_count]
      - cancel: =primary_extract
      - log: "Fallback extractor won with {{record_count}} records"

    # Launch transformer now that extraction is done
  - run:
      _label_: Transform data
      flow: pipeline/transform.flowmarkup.yaml
      handle: transform_proc
      async: true
      cap:
        STORAGE: INHERIT
      params:
        dataset_id: =dataset_id
        record_count: =record_count

    # Wait for transformation to complete using source filter
  - waitFor:
      event: transform_complete
      source: =transform_proc
      timeout: 1h
      capture:
        records_loaded: =EVENT.DATA.records_transformed

  - return: [records_loaded, validation_status]
