flowmarkup:
  title: Cross-Storage ETL Pipeline
  version: 1
  description: "ETL: download CSV from S3, parse, transform in CEL, upload to SFTP, send email notification."
  requires:
    STORAGE:
      s3_source: [get]
      sftp_dest: [put, mkdir]
    MAIL: ["@ops.example.com"]
    ENV: [PIPELINE_ID]

  input:
    source_path:
      $kind: TEXT
      $format: "^[a-zA-Z0-9/_.-]+$"
    dest_dir:
      $kind: TEXT
      $format: "^[a-zA-Z0-9/_.-]+$"

  output:
    records_processed: INTEGER
    output_path: TEXT

  vars:
    records_processed: 0

  do:

    # Step 1: Download CSV from S3 and auto-parse
  - storage:
      _label_: Fetch source data
      url: s3_source
      operation: get
      path: =source_path
      parseAs: CSV
      timeout: 5m
      result:
        raw_data: =RESULT.data
        file_name: =RESULT.name
        file_size: =RESULT.size

  - log: ="Downloaded " + file_name + " (" + string(file_size) + " bytes, " + string(size(raw_data)) + " rows)"

    # Step 2: Transform in CEL — filter and enrich
  - set:
      active_records: =raw_data.filter(r, r.status == "active" && has(r.email))
      records_processed: =size(active_records)

  - assert:
      condition: =records_processed > 0
      message: "No active records found in source data"

    # Step 3: Ensure destination directory exists
  - storage:
      _label_: Create destination directory
      url: sftp_dest
      operation: mkdir
      path: =dest_dir
      timeout: 30s

    # Step 4: Upload transformed data to SFTP
  - storage:
      _label_: Upload processed data
      url: sftp_dest
      operation: put
      path: =dest_dir + "/active-" + file_name
      data: =active_records
      overwrite: true
      timeout: 5m
      result:
        output_path: =RESULT.path

  - log: ="Uploaded " + string(records_processed) + " records to " + output_path

    # Step 5: Send notification
  - mail:
      to: ['ops-pipeline@ops.example.com']
      subject: ="ETL Complete: " + ENV.PIPELINE_ID
      body:
        text: >-
          Pipeline {{ENV.PIPELINE_ID}} completed successfully.
          Source: {{source_path}} ({{file_size}} bytes)
          Records processed: {{records_processed}}
          Output: {{output_path}}
      timeout: 30s

  - return: {records_processed: =records_processed, output_path: =output_path}
