Post a daily data pipeline digest to Zulip

This flow turns per-run pipeline noise into a single readable morning summary. A Schedule trigger fires once a day, a PostgreSQL Query rolls up the previous day's runs from a pipeline_runs audit table (run count, rows processed, failed records, average runtime, and an overall status), and ZulipIncomingWebhook posts a formatted markdown message to a Zulip channel for the data team. If the rollup or the post fails, a flow-level errors handler posts a loud "digest FAILED" notice to the same channel so a missing summary is never silent.

The query targets a generic analytics.pipeline_runs audit table with columns pipeline_name, run_date, rows_processed, failed_records, and runtime_seconds. Point it at whatever runs/observability table you already populate (or a dbt run_results export); the digest payload references the aggregated row, so only the SQL needs to change to fit your schema.

Reliability and idempotency

• compute_summary carries a retry (constant, three attempts) so a transient database or network blip does not kill the morning report on the first try.
• The flow-level errors block posts a distinct failure message that includes the run date and execution id, so the data team can always tell "pipeline healthy and quiet" apart from "the digest flow broke."
• concurrency: { limit: 1 } ensures a backfill or a catch-up after downtime cannot post several conflicting digests for the same window at once.
• Note that ZulipIncomingWebhook does not raise on a non-200 response, so even with the retry, delivery of the digest itself is best-effort; check the task logs if a message does not appear.

Backfill-safe date window

The data window is derived from {{ (trigger.date ?? execution.startDate) | date('yyyy-MM-dd') }} and bound into the SQL as the :run_date prepared-statement parameter, and the SAME value is rendered into the message. A backfilled or manually started run therefore reports the historical day it is replaying, not "today," and the partition shown in the message always matches the data queried.

Conditional formatting

The payload prefixes a :white_check_mark: when status is healthy and a :x: when status is degraded, so an unhealthy day is visually distinct from a healthy one rather than blending in. All interpolated values pass through | toJson so a stray quote or newline in any field cannot break the JSON payload.

How it works

• io.kestra.plugin.core.trigger.Schedule runs the flow daily at 07:00 in the Europe/Paris timezone.
• compute_summary runs the rollup with fetchType: FETCH_ONE, exposing the aggregated figures as {{ outputs.compute_summary.row.* }}.
• post_digest renders those values into a markdown payload and POSTs it to the Zulip incoming webhook.

Prerequisites

• A Kestra instance.
• A PostgreSQL database with a runs/audit table (the example expects analytics.pipeline_runs).
• A Zulip organization with an Incoming Webhook integration configured for the destination channel.
• The database connection and the Zulip webhook URL stored as Kestra secrets.

Secrets

• POSTGRES_URL, POSTGRES_USERNAME, POSTGRES_PASSWORD: connection to the audit database, set once via pluginDefaults for the io.kestra.plugin.jdbc.postgresql group.
• ZULIP_WEBHOOK: the full Zulip incoming webhook URL, including the integration path and API key, in the form https://yourZulipDomain.zulipchat.com/api/v1/external/INTEGRATION_NAME?api_key=API_KEY. Used by both ZulipIncomingWebhook tasks.

Quick start

1. Create an Incoming Webhook integration in Zulip and copy its URL; store it as the ZULIP_WEBHOOK secret.
2. Add the PostgreSQL connection secrets, or change the pluginDefaults block to your warehouse plugin (for example io.kestra.plugin.jdbc.snowflake).
3. Adapt the sql to your audit table and the cron/timezone to your reporting window, keeping the :run_date and :pipeline_name bindings.
4. Deploy and run the flow once manually to confirm the digest renders in your Zulip channel, then force a query error to confirm the failure path posts the "digest FAILED" notice.

Expected outputs

• {{ outputs.compute_summary.row.* }} holds the rollup figures (run_count, rows_processed, failed_records, runtime_min, status) injected into the message.
• ZulipIncomingWebhook is a notification task and produces no internal-storage artifact, so there is no {{ outputs.post_digest.uri }}. Success is a message in the Zulip channel plus a SUCCESS state on post_digest.

Links

• Plugin docs: https://kestra.io/plugins/plugin-zulip/io.kestra.plugin.zulip.zulipincomingwebhook
• PostgreSQL Query docs: https://kestra.io/plugins/plugin-jdbc-postgres/io.kestra.plugin.jdbc.postgresql.query
• Schedule trigger docs: https://kestra.io/docs/workflow-components/triggers/schedule-trigger
• Errors and failure handling: https://kestra.io/docs/workflow-components/errors
• Zulip incoming webhooks: https://zulip.com/api/incoming-webhooks-overview