Turn every Pub/Sub message into a pipeline.

id: pubsub-realtime-to-postgres
namespace: company.team
description: |
  Turn each Pub/Sub message into a governed execution. A realtime trigger starts
  one run per message, validates the payload, and upserts it into Postgres with
  safe named parameters. Microsoft Teams is alerted if a message fails to land.

tasks:
  - id: route
    type: io.kestra.plugin.core.flow.If
    description: Only process order events; skip anything else.
    condition: "{{ trigger.data.event_type == 'order' }}"
    then:
      - id: upsert_order
        type: io.kestra.plugin.jdbc.postgresql.Query
        description: Upsert the order into Postgres using named parameters.
        url: "{{ secret('POSTGRES_URL') }}"
        username: "{{ secret('POSTGRES_USERNAME') }}"
        password: "{{ secret('POSTGRES_PASSWORD') }}"
        sql: |
          INSERT INTO orders (order_id, customer, amount, received_at)
          VALUES (:order_id, :customer, :amount, now())
          ON CONFLICT (order_id) DO UPDATE SET amount = EXCLUDED.amount;
        parameters:
          order_id: "{{ trigger.data.order_id }}"
          customer: "{{ trigger.data.customer }}"
          amount: "{{ trigger.data.amount }}"

pluginDefaults:
  - type: io.kestra.plugin.gcp
    values:
      projectId: "{{ secret('GCP_PROJECT_ID') }}"
      serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT') }}"

triggers:
  - id: on_message
    type: io.kestra.plugin.gcp.pubsub.RealtimeTrigger
    description: Start one execution per Pub/Sub message. Enable once your secrets
      and subscription are set.
    topic: orders
    subscription: kestra-orders-sub
    serdeType: JSON
    disabled: true

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.microsoft365.teams.TeamsIncomingWebhook
    description: Alert Microsoft Teams if a message fails to land.
    url: "{{ secret('TEAMS_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Pub/Sub message failed to upsert into Postgres (execution {{ execution.id }})."
      }

id: pubsub-publish-pipeline-events
namespace: company.team
description: |
  Emit an event when a pipeline finishes so other systems can react. On a
  schedule, the flow runs a BigQuery aggregation, then publishes a completion
  event (run id, status, row count) to a Pub/Sub topic that downstream
  consumers and other flows subscribe to. Email confirms the publish.

tasks:
  - id: run_aggregation
    type: io.kestra.plugin.gcp.bigquery.Query
    description: Run the pipeline's work and capture a row count.
    sql: |
      SELECT COUNT(*) AS row_count
      FROM `{{ secret('GCP_PROJECT_ID') }}.analytics.daily_orders`
      WHERE order_date = CURRENT_DATE();
    fetchOne: true

  - id: publish_event
    type: io.kestra.plugin.gcp.pubsub.Publish
    description: Publish a completion event to the pipeline-events topic.
    topic: pipeline-events
    serdeType: JSON
    from:
      - data:
          run_id: "{{ execution.id }}"
          status: "completed"
          row_count: "{{ outputs.run_aggregation.row.row_count }}"
        attributes:
          source: kestra
          flow: "{{ flow.id }}"

  - id: notify
    type: io.kestra.plugin.email.MailSend
    description: Email the data team that the event was published.
    host: "{{ secret('SMTP_HOST') }}"
    port: 587
    username: "{{ secret('SMTP_USERNAME') }}"
    password: "{{ secret('SMTP_PASSWORD') }}"
    from: kestra@example.com
    to: data-team@example.com
    subject: "Pipeline completion event published"
    htmlTextContent: |
      Published a completion event for run {{ execution.id }}
      with {{ outputs.run_aggregation.row.row_count }} rows.

pluginDefaults:
  - type: io.kestra.plugin.gcp
    values:
      projectId: "{{ secret('GCP_PROJECT_ID') }}"
      serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT') }}"

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Run the aggregation and emit the event on a cadence. Adjust or
      disable as needed.
    cron: "0 7 * * *"
    disabled: true

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.email.MailSend
    description: Email the data team if the publish fails.
    host: "{{ secret('SMTP_HOST') }}"
    port: 587
    username: "{{ secret('SMTP_USERNAME') }}"
    password: "{{ secret('SMTP_PASSWORD') }}"
    from: kestra@example.com
    to: data-team@example.com
    subject: "Pipeline event publish failed"
    htmlTextContent: "The pipeline-events publish failed for execution {{ execution.id }}."