In-Memory DuckDB Customer Sentiment and Health Pipeline

id: customer-sentiment-batch-processor
namespace: company.analytics

triggers:
  - id: hourly_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 * * * *"

tasks:
  - id: query_mock_feedback
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: "Extract batch feedback in-memory"
    sql: |
      SELECT 101 AS user_id, 'App crash when clicking checkout button' AS review, 'URGENT' AS priority
      UNION ALL
      SELECT 102 AS user_id, 'Great service, loved the UI' AS review, 'NORMAL' AS priority;
    fetchType: FETCH

  - id: parallel_diagnostics
    type: io.kestra.plugin.core.flow.Parallel
    description: "Concurrent diagnostic telemetry"
    tasks:
      - id: check_api_status
        type: io.kestra.plugin.core.http.Request
        uri: "https://httpbin.org/status/200"
        method: GET

      - id: analyze_logs
        type: io.kestra.plugin.scripts.shell.Commands
        commands:
          - 'echo "Batch extracted: {{ outputs.query_mock_feedback.size }}
            records"'
          - 'echo "Parsing priority keywords from feedback stream..."'

  - id: evaluate_severity
    type: io.kestra.plugin.core.flow.If
    description: "Conditional routing based on API response"
    condition: "{{ outputs.check_api_status.code == 200 }}"
    then:
      - id: log_batch_metrics
        type: io.kestra.plugin.core.log.Log
        level: INFO
        message: |
          Feedback processing succeeded.
          Processed {{ outputs.query_mock_feedback.size }} items.
          API Status: {{ outputs.check_api_status.code }}