datapipeline 3.2.0: Benachrichtigungen mit Webhook, Slack und E-Mail

datapipeline 3.1.0: SQLite schreibt im Streaming in Paketen, Lizenz BSD-3-Clause
2026-10-09 22:27:00 +02:00 · 2026-10-09 22:11:24 +02:00
22 changed files with 1142 additions and 126 deletions
@@ -2,6 +2,26 @@

 Framework `datapipeline` (Gem). Neueste Version oben.

+## 3.2.0 – 2026-10-09
+
+- Benachrichtigungen: `notify notifier, on: :failure | :problems | :always` nach einem Lauf, `DataPipeline::Report`
+  (Ereignis, Text, JSON-Payload `schema: 1`), Basisklasse `DataPipeline::Notifier` für eigene Kanäle; verworfene
+  Zeilen nur mit `include_rows: true`
+- `DataPipeline::Webhook`: POST mit HMAC-SHA256-Signatur, Wiederholung bei 429/5xx (`Retry-After`), Timeout;
+  `Webhook.valid_signature?` für Empfänger
+- `DataPipeline::Email`: Bericht als Text-Mail über SMTP (STARTTLS, TLS, Anmeldung), Gem `net-smtp`
+- `DataPipeline::Slack`: `deliver(report)` mit Block Kit, Maskierung von `&`, `<`, `>`, `mention:` bei Abbruch,
+  Incoming Webhooks (`webhook:`), Timeout und Wiederholung; Fehler als `DataPipeline::Slack::Error`
+- Notifier aus der Verbindungsdatei (`notifier(:name)`, `type: webhook | slack | email`); `etl --notify=NAME
+  [--notify-on=…]` sendet einen Bericht für alle Pipelines eines Skripts
+
+## 3.1.0 – 2026-10-09
+
+- sqlite: `write_row` schreibt gesammelt je `batch_size` Zeilen (Default 1000) in einer Transaktion;
+  `batch_size: :all` = eine Transaktion für den ganzen Stream, nach einem Abbruch zurückgerollt
+- postgres, sqlite: `query` schreibt noch gesammelte Zeilen von `write_row` vorher
+- Lizenz BSD-3-Clause (`LICENSE`, gemspec)
+
 ## 3.0.0 – 2026-10-09

 - **Breaking:** `transform` (auch `processing: :prepend`/`:append`), der Parallelmodus, `chained` und `condition:` als
@@ -7,7 +7,8 @@ gemspec
 # optional plugin gems: needed for the plugin tests, loaded only when a plugin is used
 group :plugins do
  gem 'caxlsx', require: false    # excel write
-  gem 'http', require: false      # rest, slack
+  gem 'http', require: false      # rest, slack, webhook
+  gem 'net-smtp', require: false  # email
  gem 'pg', require: false        # postgres
  gem 'roo', require: false       # excel read
  gem 'sqlite3', require: false   # sqlite
@@ -1,7 +1,7 @@
 PATH
  remote: .
  specs:
-    datapipeline (3.0.0)
+    datapipeline (3.2.0)
      csv (~> 3.3)
      logger (~> 1.6)

@@ -26,6 +26,10 @@ GEM
    logger (1.7.0)
    marcel (1.2.1)
    minitest (5.27.0)
+    net-protocol (0.4.0)
+      timeout
+    net-smtp (0.5.2)
+      net-protocol
    nokogiri (1.19.4-aarch64-linux-gnu)
      racc (~> 1.4)
    nokogiri (1.19.4-aarch64-linux-musl)
@@ -66,6 +70,7 @@ GEM
    sqlite3 (2.9.6-x86_64-darwin)
    sqlite3 (2.9.6-x86_64-linux-gnu)
    sqlite3 (2.9.6-x86_64-linux-musl)
+    timeout (0.6.1)

 PLATFORMS
  aarch64-linux
@@ -84,6 +89,7 @@ DEPENDENCIES
  datapipeline!
  http
  minitest (~> 5.25)
+  net-smtp
  pg
  rake
  roo
@@ -0,0 +1,28 @@
+BSD 3-Clause License
+
+Copyright (c) 2026, Karl-Heinz Wild
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+1. Redistributions of source code must retain the above copyright notice, this
+   list of conditions and the following disclaimer.
+
+2. Redistributions in binary form must reproduce the above copyright notice,
+   this list of conditions and the following disclaimer in the documentation
+   and/or other materials provided with the distribution.
+
+3. Neither the name of the copyright holder nor the names of its
+   contributors may be used to endorse or promote products derived from
+   this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
@@ -1,6 +1,6 @@
 # DataPipeline – ETL-Framework in Ruby

-Stand: 2026-10-09 · Gem `datapipeline` 3.0.0 ([CHANGELOG.md](CHANGELOG.md)) · Ruby 3.4
+Stand: 2026-10-09 · Gem `datapipeline` 3.2.0 ([CHANGELOG.md](CHANGELOG.md)) · Ruby 3.4 · Lizenz BSD-3-Clause ([LICENSE](LICENSE))

 `DataPipeline` liest Daten aus einer Quelle, schickt jede Zeile durch eine Kette von Schritten (prüfen, filtern,
 umbauen, anreichern, abgleichen) und schreibt das Ergebnis in ein oder mehrere Ziele. Eine Pipeline ist ein kurzes
@@ -14,8 +14,8 @@ Festbreiten-Dateien und REST-APIs.
 3. [Pipelines schreiben](#3-pipelines-schreiben) – DSL, Schritte, Prüfen, Fehler, Ergebnis
 4. [Plugins](#4-plugins) – mitgelieferte Quellen und Ziele, eigene Plugins, Abhängigkeiten
 5. [Satzarten-Layouts](#5-satzarten-layouts) – Dateien mit Kopf-, Positions- und Schlusssätzen
-6. [Betrieb](#6-betrieb) – Kommandozeile, Verbindungen, Sperre, Abgleich, Historie, Benachrichtigung, Fehlersuche
-7. [Umstieg](#7-umstieg) – von 2.x auf 3.0 (`transform` entfernt), von 2.0 auf 2.1
+6. [Betrieb](#6-betrieb) – Kommandozeile, Verbindungen, Sperre, Abgleich, Historie, Benachrichtigungen, Fehlersuche
+7. [Umstieg](#7-umstieg) – auf 3.1, von 2.x auf 3.0 (`transform` entfernt), von 2.0 auf 2.1
 8. [Entwicklung](#8-entwicklung) – Aufbau, Tests, Gem bauen
 9. [Ausblick](#9-ausblick)

@@ -32,7 +32,8 @@ Festbreiten-Dateien und REST-APIs.
 | Prüfen | Regeln je Spalte (Pflicht, Länge, Format, Werteliste, Typ), eigene Regeln als Block, Satzarten-Layouts; als Fehler oder nur als Warnung |
 | Fehler | Abbruch, Zeile überspringen oder mit Grund in eine Fehlerdatei schreiben |
 | Datenmengen | Batch im Speicher oder Streaming Zeile für Zeile mit konstantem Speicherbedarf; Postgres-Cursor und Schreiben in Paketen |
-| Betrieb | Kommandozeile `etl` mit Exit-Codes, Dry-Run, Limit, inkrementelle Läufe (Wasserzeichen), Sperre gegen parallele Läufe, Upsert für wiederholbare Läufe, Summenabgleich vor dem Schreiben, Laufhistorie, Verbindungsdatei, Slack |
+| Betrieb | Kommandozeile `etl` mit Exit-Codes, Dry-Run, Limit, inkrementelle Läufe (Wasserzeichen), Sperre gegen parallele Läufe, Upsert für wiederholbare Läufe, Summenabgleich vor dem Schreiben, Laufhistorie, Verbindungsdatei |
+| Benachrichtigung | Berichte an Webhooks (signiert), Slack und E-Mail, je Pipeline oder je Skript; eigene Notifier |
 | Dateiformate | Zeichensatz (z. B. Windows-1252) für CSV und Festbreite, BOM für CSV; Satzarten-Dateien schreiben, prüfen und lesen |

 **Anwendungen:** Das Projekt `etl` nutzt das Gem für die PDS → SAP-Schnittstelle (CUDO) und für das Anlegen von
@@ -54,10 +55,13 @@ In einer anderen Anwendung:

 ```ruby
 # Gemfile
-gem 'datapipeline', path: '../datapipeline'        # oder git: '…'
+gem 'datapipeline', '~> 3.1', git: 'https://git.wicom.li/wkh/datapipeline.git', branch: 'main'
 gem 'pg'                                            # nur die Gems der benötigten Plugins (4.10)
 ```

+`'~> 3.1'` verhindert, dass `bundle update` unbemerkt eine neue Hauptversion mit Breaking Changes holt (8.3). Für die
+Entwicklung gegen einen lokalen Checkout: `bundle config set local.datapipeline ../datapipeline`.
+
 ```sh
 bundle install
 bundle exec etl pipeline.rb --dry-run               # das Kommando etl kommt mit dem Gem
@@ -103,7 +107,7 @@ der Code hier genau der Datei entspricht.
 | [2.3 Inkrementell und gestreamt](#23-inkrementell-und-gestreamt) | Wasserzeichen, Streaming |
 | [2.4 Eigenes Plugin](#24-eigenes-plugin) | eine Quelle selbst schreiben |
 | [2.5 REST-APIs](#25-rest-apis) | Paging, Wiederholung, Pakete, eine URL je Datensatz |
-| [2.6 Produktiver Lauf](#26-produktiver-lauf) | Verbindungsdatei, Sperre, Kurzschritte, Summenabgleich, Upsert, Zeichensatz, Historie, Slack |
+| [2.6 Produktiver Lauf](#26-produktiver-lauf) | Verbindungsdatei, Sperre, Kurzschritte, Summenabgleich, Upsert, Zeichensatz, Historie, Benachrichtigung |

 ### 2.1 Kundenimport

@@ -621,13 +625,11 @@ FileUtils.mkdir_p(out)
 # Zugangsdaten stehen nicht im Skript, sondern in der Verbindungsdatei (Default: config/connections.yml)
 DataPipeline.connections_file = File.expand_path('config/connections.yml', __dir__)

-# Slack nur, wenn ein Token gesetzt ist; so fällt ein fehlender Token schon vor dem Lauf auf
-slack = DataPipeline::Slack.new(channel: '#etl') if ENV['SLACK_TOKEN']
-
 DataPipeline.new do
  name 'buchungen_export'
  lock "#{out}/.buchungen.lock"              # läuft schon ein Export, bricht dieser ab (Exit-Code 3)
  history "#{out}/runs.ndjson"               # eine JSON-Zeile je Lauf, auch nach Abbruch
+  notify DataPipeline::Slack.new(channel: '#etl') if ENV['SLACK_TOKEN']   # Meldung nur bei Problemen

  read container(:csv, file: "#{DATA}/buchungen.csv", csv: { col_sep: ';' })

@@ -659,10 +661,7 @@ DataPipeline.new do
  write container(:csv, file: "#{out}/buchungen_sap.csv", encoding: 'Windows-1252', csv: { col_sep: ';' },
                        columns: %w[belnr buzei datum konto shkzg dmbtr buchungstext])

-  on_finish do |stats|
-    puts "#{stats.name}: #{stats}"
-    slack&.send_message("*#{stats.name}*: #{stats}") unless stats.ok?
-  end
+  on_finish { |stats| puts "#{stats.name}: #{stats}" }
 end.run
 ```

@@ -676,7 +675,7 @@ end.run
 | `totals`, `reconcile` | Summen werden exakt gerechnet; eine Abweichung bricht vor dem Schreiben ab (6.4) |
 | `upsert: [belnr, buzei]` | ein wiederholter Lauf aktualisiert dieselben Zeilen, statt sie doppelt anzulegen (4.2) |
 | `encoding: 'Windows-1252'` | Zeichensatz der Zieldatei (4.4) |
-| `slack&.send_message` | Meldung nur bei Problemen und nur, wenn `SLACK_TOKEN` gesetzt ist (6.8) |
+| `notify DataPipeline::Slack…` | Meldung nur bei Problemen und nur, wenn `SLACK_TOKEN` gesetzt ist (6.8) |

 ```text
 INFO start
@@ -762,6 +761,7 @@ Tippfehler fallen also sofort auf.
 | `on_error(:abort \| :skip \| :collect, to: storage)` | Umgang mit Fehlern in Schritten |
 | `totals 'spalte', …` | Summen über Quelle und Ergebnis |
 | `reconcile { \|stats\| … }` | Abgleich vor dem Schreiben; eine Meldung bricht ab |
+| `notify(notifier, on: :problems)`, `notifier(:name)` | Bericht an Webhook, Slack oder E-Mail (6.8) |
 | `on_finish { \|stats\| … }` | nach jedem Lauf, auch nach einem Abbruch |
 | `name(str)`, `version(str)` | Name und Version des Laufs für Log, Stats und Historie |

@@ -826,7 +826,7 @@ Unterschieden:
 | `header` | kennt `read` und `written` | wird zuerst geschrieben, `read` und `written` sind `nil` |
 | `ctx.read` in `map`/`filter` | Anzahl Quellzeilen | `nil` |
 | `reconcile` | vor dem Schreiben | nach der letzten Zeile; die Zeilen sind schon geschrieben |
-| Postgres als Ziel | eine Transaktion für alles | eine Transaktion je `batch_size` Zeilen, mit `batch_size: :all` eine für alles (4.2) |
+| Postgres, SQLite als Ziel | eine Transaktion für alles | eine Transaktion je `batch_size` Zeilen, mit `batch_size: :all` eine für alles (4.2, 4.3) |
 | Postgres als Quelle | eine Abfrage | serverseitiger Cursor, je `fetch_size` Zeilen |

 `limit n` beendet im Streaming das Lesen nach n Zeilen.
@@ -1013,9 +1013,13 @@ write container(:sqlite, database: 'tmp/kunden.sqlite3', table: 'kunden', create
 | `create: true` | Tabelle beim ersten Schreiben aus der ersten Zeile anlegen (Integer → `integer`, Float → `real`, sonst `text`) |
 | `truncate: true` | Tabelle vor dem ersten Schreiben leeren |
 | `upsert: %w[belnr]` | wie bei Postgres; mit `create: true` wird der nötige Unique-Index mit angelegt |
+| `batch_size` | Streaming: Zeilen je Transaktion (Default 1000) oder `:all` |

 Werte kommen mit SQLite-Typen zurück (Integer, Float, String, `nil`); `Date` und `Time` werden als ISO-Text
-geschrieben. `write` schreibt in einer Transaktion.
+geschrieben. `write` schreibt in einer Transaktion. Im Streaming gilt dasselbe wie bei Postgres (4.2): `write_row`
+schreibt je `batch_size` Zeilen in einer Transaktion, `finalize` und `close` den Rest; mit `batch_size: :all` ist der
+ganze Stream eine Transaktion, die nach einem Abbruch zurückgerollt wird. Eine Abfrage (`query`, `lookup`) auf
+demselben Storage schreibt noch gesammelte Zeilen vorher, sie sieht also alles, was schon übergeben wurde.

 ### 4.4 CSV

@@ -1130,7 +1134,8 @@ Gemfile der Anwendung, die das Plugin nutzt:
 | `:postgres` | `pg` |
 | `:sqlite` | `sqlite3` |
 | `:excel` | `roo` (lesen), `caxlsx` (schreiben) |
-| `:rest`, `DataPipeline::Slack` | `http` |
+| `:rest`, `DataPipeline::Slack`, `DataPipeline::Webhook` | `http` |
+| `DataPipeline::Email` | `net-smtp` |
 | `:csv`, `:json`, `:fixed` | – |

 Eine CSV- oder JSON-Pipeline lädt keines dieser Gems. Fehlt ein Gem, meldet das Plugin bei der ersten Nutzung, welches
@@ -1292,6 +1297,7 @@ bin/etl run cudo.rb --lock=data/.cudo.lock --history=data/runs.ndjson
 | `--dry-run`, `--limit=N` | für alle Pipelines des Skripts |
 | `--lock=FILE` | Sperre für das ganze Skript (6.3) |
 | `--history=FILE` | Laufhistorie für alle Pipelines des Skripts (6.6) |
+| `--notify=NAME`, `--notify-on=WHEN` | ein Bericht für alle Pipelines des Skripts an den Notifier `NAME` aus der Verbindungsdatei; `WHEN` = `problems` (Default), `failure`, `always` (6.8) |
 | `--verbose`, `--quiet` | Log-Level Debug bzw. Warnung (Default: Info) |
 | jede andere `--name=wert` | `params[:name]` und unverändert an das Skript, z. B. `--out=data`, `--incremental=1` |

@@ -1416,81 +1422,181 @@ Eigene Schritte mit Nebenwirkungen (API-Aufrufe in einem `map`) fragen `dry_run?

 `limit n` verarbeitet nur die ersten n Quellzeilen; im Streaming wird das Lesen dann beendet.

-### 6.8 Benachrichtigung über Slack
+### 6.8 Benachrichtigungen

-`DataPipeline::Slack` schickt Text über die Slack-API `chat.postMessage`. Einrichten:
-
-1. In Slack eine App anlegen (api.slack.com/apps), unter *OAuth & Permissions* den Bot-Scope `chat:write` vergeben,
-   die App im Workspace installieren und den *Bot User OAuth Token* (`xoxb-…`) kopieren.
-2. Die App in den Ziel-Channel einladen (`/invite @appname`), sonst antwortet Slack mit `not_in_channel`.
-3. Das Gem `http` ins Gemfile der Anwendung aufnehmen (4.10).
-4. Den Token als Umgebungsvariable `SLACK_TOKEN` setzen oder mit `token:` übergeben; nicht ins Skript schreiben.
+`notify` schickt nach einem Lauf einen Bericht an einen Notifier: einen Webhook, Slack oder E-Mail. Ein Fehler beim
+Senden wird nur geloggt; Ergebnis und Exit-Code bleiben unverändert.

 ```ruby
-slack = DataPipeline::Slack.new                                       # Token aus ENV['SLACK_TOKEN'], Channel '#general'
-slack = DataPipeline::Slack.new(channel: '#etl')                      # anderer Channel (Name oder ID wie 'C0123ABCD')
-slack = DataPipeline::Slack.new(channel: '#etl', token: ENV.fetch('SLACK_ETL_TOKEN'))  # eigener Token je Anwendung
-
-slack.send_message('Test aus der Pipeline', verbose: true)             # verbose: Antwort von Slack als JSON auf $stdout
-```
-
-| Option | Default | Bedeutung |
-| --- | --- | --- |
-| `token:` | `ENV['SLACK_TOKEN']` | Bot-Token; fehlt er, bricht schon `Slack.new` mit `ArgumentError` ab |
-| `channel:` | `'#general'` | Ziel-Channel als Name oder ID |
-
-`send_message(text, verbose: false)` liefert die Antwort von Slack als Hash. Meldet Slack einen Fehler (z. B.
-`invalid_auth`, `channel_not_found`, `not_in_channel`), wirft die Methode eine Exception mit diesem Code. In
-`on_finish` wird sie nur geloggt (`on_finish: RuntimeError: Slack#send_message failed: …`), der Lauf und sein
-Exit-Code bleiben unverändert. Es gibt keine Wiederholung bei Netzwerkfehlern.
-
-Eine ausführliche Meldung mit Status, Zählern, Laufzeit und den ersten verworfenen Zeilen. Der Notifier wird vor der
-Pipeline angelegt, damit ein fehlender Token schon beim Start auffällt. Slack setzt `*fett*`, `_kursiv_` und
-`` `code` `` um.
-
-```ruby
-slack = DataPipeline::Slack.new(channel: '#etl')
-
 DataPipeline.new do
+  name 'cudo_export'
+  notify DataPipeline::Webhook.new(url: ENV['ALERT_URL'], secret: ENV['ALERT_SECRET'])   # bei Problemen
+  notify DataPipeline::Slack.new(channel: '#etl', mention: '<!here>'), on: :failure
+  notify DataPipeline::Email.new(to: 'buchhaltung@example.com'), on: :always
+  notify notifier(:alerts)                                                               # aus der Verbindungsdatei
  # … read, Schritte, write …
-
-  on_finish do |stats|
-    next if stats.ok? && stats.warnings.zero?                  # nur melden, wenn etwas zu tun ist
-
-    status = if stats.error then ':red_circle: abgebrochen'
-             elsif stats.rejected.positive? then ':large_orange_circle: mit verworfenen Zeilen'
-             else ':large_yellow_circle: mit Warnungen'
-             end
-    lines = ["*Buchungsexport #{stats.version}* #{status}#{' (Dry-Run)' if stats.dry_run}",
-             "Zeitraum #{params[:from]} – #{params[:till]}, Start #{stats.started_at.strftime('%d.%m.%Y %H:%M')}, " \
-             "Dauer #{stats.duration}s",
-             "gelesen #{stats.read} · geschrieben #{stats.written} · verworfen #{stats.rejected} · Warnungen #{stats.warnings}"]
-    lines << "Fehler: `#{stats.error.class}: #{stats.error.message}`" if stats.error
-    stats.errors.first(5).each do |e|
-      lines << "• #{e[:step]}: #{e[:error]} – Beleg #{e[:row]['belnr']}"
-    end
-    lines << "… und #{stats.rejected - 5} weitere, siehe data/fehler.csv" if stats.rejected > 5
-
-    slack.send_message(lines.join("\n"))
-  end
 end.run
 ```

-So sieht die Meldung in Slack aus:
+| Option von `notify` | Wirkung |
+| --- | --- |
+| `on: :problems` (Default) | bei Abbruch, verworfenen Zeilen oder Warnungen |
+| `on: :failure` | nur bei Abbruch |
+| `on: :always` | nach jedem Lauf |
+| `title:` | Titel des Berichts, Default der Name der Pipeline |
+| `dry_run: true` | auch nach einem Dry-Run senden (Default: nicht) |
+| `include_rows: true` | die verworfenen Zeilen mitsenden; Default: nur Schritt und Meldung, weil Zeilen Kundendaten enthalten können |

-```text
-Buchungsexport 1.0 🟠 mit verworfenen Zeilen
-Zeitraum 2025-11-01 – 2025-11-30, Start 01.12.2025 06:00, Dauer 4.213s
-gelesen 1250 · geschrieben 1243 · verworfen 7 · Warnungen 0
-• validate#1: konto: missing – Beleg 5100000123
-• validate#1: dmbtr: not decimal – Beleg 5100000124
-• validate#2: Betrag 0 – Beleg 5100000130
-• validate#1: shkzg: not in ["S", "H"] – Beleg 5100000131
-• validate#1: konto: format /\A\d+\z/ – Beleg 5100000140
-… und 2 weitere, siehe data/fehler.csv
+**Bericht** – Jeder Notifier bekommt einen `DataPipeline::Report` mit den Stats eines oder mehrerer Läufe:
+
+| Methode | Inhalt |
+| --- | --- |
+| `event` | `run.failed` (Abbruch), `run.rejected` (verworfene Zeilen), `run.warnings`, `run.succeeded`; bei mehreren Läufen der schlimmste |
+| `title`, `status`, `subject` | z. B. `cudo_export`, `mit verworfenen Zeilen`, `cudo_export: mit verworfenen Zeilen` |
+| `lines`, `to_s` | Text je Lauf: Zähler, Start, Dauer, die ersten 5 Fehler (ohne Zeilen) |
+| `to_h(include_rows: false)` | JSON-Payload für Webhooks (unten) |
+| `id`, `host`, `params`, `runs`, `error` | eindeutige Kennung, Rechner, Parameter der Kommandozeile, Stats, Fehler ausserhalb eines Laufs |
+
+**Ein Bericht für das ganze Skript** – `notify` meldet jede Pipeline einzeln. Für einen Sammelbericht über alle
+Pipelines eines Skripts gibt es die Kommandozeile; der Notifier kommt aus der Verbindungsdatei:
+
+```sh
+bin/etl run cudo.rb --notify=alerts                     # bei Problemen (Default)
+bin/etl run cudo.rb --notify=alerts --notify-on=always
 ```

-`params` gibt es im Pipeline-Block; ausserhalb heisst es `DataPipeline.params`.
+Bricht das Skript selbst ab (z. B. ein Fehler vor der ersten Pipeline), meldet der Bericht `run.failed` mit diesem
+Fehler. Bei belegter Sperre (Exit-Code `3`) und im Dry-Run wird nichts gesendet.
+
+#### Webhook
+
+`DataPipeline::Webhook` sendet den Bericht als JSON per `POST`.
+
+| Option | Default | Bedeutung |
+| --- | --- | --- |
+| `url:` | – | Ziel, Pflicht |
+| `secret:` | – | signiert den Body (unten) |
+| `headers:` | `{}` | weitere Header, z. B. `{ 'Authorization' => "Bearer #{ENV['TOKEN']}" }` |
+| `timeout:` | 10 | Sekunden je Anfrage |
+| `retries:`, `retry_wait:` | 3, 2 | Wiederholung bei 429, 5xx und Netzwerkfehlern, Wartezeit `retry_wait` × Versuch; `Retry-After` wird beachtet; andere 4xx brechen sofort ab |
+
+```json
+{"schema":1,"id":"3f6c2a1e-…","event":"run.rejected","title":"cudo_export","host":"etl01",
+ "params":{"from":"2026-10-08"},"error":null,
+ "runs":[{"name":"cudo_export","version":"1.0","started_at":"2026-10-09T06:00:00.123+02:00","duration":4.213,
+          "read":1250,"written":1243,"rejected":7,"warnings":0,"ok":false,"error":null,
+          "totals":{"dmbtr":{"read":"98231.4","written":"97810.1"}},
+          "errors":[{"step":"validate#1","error":"konto: missing"}]}]}
+```
+
+`schema: 1` ist ein Versprechen: Felder können dazukommen, eine Änderung bestehender Felder hiesse `schema: 2`. Jede
+Anfrage trägt die Header `X-DataPipeline-Event` und `X-DataPipeline-Delivery` (die `id`, bei Wiederholungen gleich;
+damit erkennt der Empfänger Doppelte). Mit `secret:` kommen dazu:
+
+```text
+X-DataPipeline-Timestamp: 1791525600
+X-DataPipeline-Signature: sha256=<HMAC-SHA256(secret, "<timestamp>.<body>") hexadezimal>
+```
+
+Der Empfänger prüft Signatur und Alter (gegen wiederholt eingespielte Anfragen); in Ruby z. B.:
+
+```ruby
+ok = DataPipeline::Webhook.valid_signature?(secret: ENV['ALERT_SECRET'], body: request.raw_post,
+                                            timestamp: request.headers['X-DataPipeline-Timestamp'],
+                                            signature: request.headers['X-DataPipeline-Signature'],
+                                            max_age: 300)
+```
+
+#### Slack
+
+`DataPipeline::Slack` meldet einen Bericht als Statuszeile mit Symbol (🔴 abgebrochen, 🟠 verworfene Zeilen,
+🟡 Warnungen, 🟢 ok) und darunter den Text je Lauf. `&`, `<` und `>` aus Daten und Meldungen werden maskiert.
+
+Zwei Wege zu Slack:
+
+| Weg | Einrichten | Aufruf |
+| --- | --- | --- |
+| Incoming Webhook (einfacher) | Slack-App anlegen (api.slack.com/apps), *Incoming Webhooks* aktivieren, Webhook für den Channel erzeugen | `Slack.new(webhook: ENV['SLACK_WEBHOOK'])` |
+| Bot-Token | Slack-App mit Bot-Scope `chat:write` installieren, *Bot User OAuth Token* (`xoxb-…`) kopieren, App in den Channel einladen (`/invite @appname`) | `Slack.new(channel: '#etl')`, Token aus `ENV['SLACK_TOKEN']` |
+
+| Option | Default | Bedeutung |
+| --- | --- | --- |
+| `webhook:` | – | URL eines Incoming Webhook; dann braucht es weder Token noch Channel |
+| `token:` | `ENV['SLACK_TOKEN']` | Bot-Token; ohne Token und ohne Webhook bricht schon `Slack.new` mit `ArgumentError` ab |
+| `channel:` | `'#general'` | Channel als Name oder ID (nur mit Token) |
+| `mention:` | – | z. B. `'<!here>'` oder `'<!subteam^ID>'`, nur bei `run.failed` |
+| `timeout:`, `retries:`, `retry_wait:` | 10, 3, 2 | wie beim Webhook |
+
+Eigener Text geht weiter mit `send_message`, z. B. in `on_finish`:
+
+```ruby
+slack = DataPipeline::Slack.new(channel: '#etl')
+slack.send_message('Monatsabschluss exportiert', verbose: true)    # verbose: Antwort von Slack auf $stdout
+```
+
+Meldet Slack einen Fehler (`invalid_auth`, `channel_not_found`, `not_in_channel`), wirft der Aufruf
+`DataPipeline::Slack::Error` mit diesem Code.
+
+#### E-Mail
+
+`DataPipeline::Email` sendet den Bericht als Text-Mail (UTF-8) über SMTP. Betreff: `[etl] <Titel>: <Status>`. Die
+Zugangsdaten kommen aus der Umgebung oder als Option:
+
+| Option | Default | Bedeutung |
+| --- | --- | --- |
+| `to:` | – | eine Adresse oder ein Array, Pflicht |
+| `from:` | `ENV['SMTP_FROM']`, sonst `etl@<rechner>` | Absender |
+| `host:`, `port:` | `ENV['SMTP_HOST']` bzw. `localhost`; `ENV['SMTP_PORT']` bzw. 587 | SMTP-Server |
+| `username:`, `password:` | `ENV['SMTP_USERNAME']`, `ENV['SMTP_PASSWORD']` | Anmeldung (LOGIN); ohne Benutzer keine Anmeldung |
+| `starttls:` | `:auto` | `:auto` (wenn der Server es anbietet), `:always` oder `false` (dann Default-Port 25) |
+| `tls:` | `false` | `true` = TLS von Anfang an (Port 465) |
+| `subject_prefix:` | `'[etl] '` | Anfang des Betreffs |
+| `timeout:` | 15 | Sekunden für Verbindung und Antwort |
+
+Mit `include_rows: true` stehen die verworfenen Zeilen als JSON am Ende der Mail. Eigener Text:
+`DataPipeline::Email.new(to: …).send_mail('Betreff', "Text")`. Ein Fehler beim Senden wirft
+`DataPipeline::Email::Error`. Das Plugin braucht das Gem `net-smtp`.
+
+#### Notifier in der Verbindungsdatei
+
+Wie Storages lassen sich Notifier in `config/connections.yml` ablegen (6.2); `type` ist `webhook`, `slack` oder
+`email`, die anderen Schlüssel sind die Optionen:
+
+```yaml
+alerts:
+  type: webhook
+  url: <%= ENV['ALERT_URL'] %>
+  secret: <%= ENV['ALERT_SECRET'] %>
+buchhaltung_mail:
+  type: email
+  to: [buchhaltung@example.com, etl@example.com]
+  host: smtp.example.com
+  username: etl@example.com
+  password: <%= ENV['SMTP_PASSWORD'] %>
+```
+
+```ruby
+notify notifier(:buchhaltung_mail), on: :always
+```
+
+#### Eigene Notifier
+
+Ein Notifier ist eine Klasse mit Basisklasse `DataPipeline::Notifier` und einer Methode
+`deliver(report, include_rows: false)`, z. B. für Microsoft Teams oder ein Ticketsystem:
+
+```ruby
+require 'http'
+
+class TeamsNotifier < DataPipeline::Notifier
+  def initialize(url:) = @url = url
+
+  def deliver report, include_rows: false
+    HTTP.post(@url, json: { title: report.subject, text: report.lines.join("\n\n") })
+  end
+end
+
+notify TeamsNotifier.new(url: ENV['TEAMS_WEBHOOK']), on: :failure
+```

 ### 6.9 Fehlersuche

@@ -1516,6 +1622,20 @@ schnellen Blick auf das Ergebnis.

 ## 7. Umstieg

+### 7.0 Von 3.0 auf 3.1 und 3.2
+
+3.1 und 3.2 sind abwärtskompatibel. Anders als in 3.0:
+
+| Thema | bis 3.0 | jetzt | seit |
+| --- | --- | --- | --- |
+| SQLite, `write_row` im Streaming | jede Zeile einzeln | je 1000 Zeilen in einer Transaktion (4.3); eine zweite Verbindung sieht die Zeilen paketweise | 3.1 |
+| Lizenz | nicht festgelegt | BSD-3-Clause | 3.1 |
+| Slack-Fehler | `RuntimeError` | `DataPipeline::Slack::Error` (weiterhin ein `StandardError`) | 3.2 |
+| `Slack.new` | lädt `http` erst beim Senden | lädt `http` sofort; Timeout 10 s und Wiederholung bei 429/5xx | 3.2 |
+| `send_message(…, verbose: true)` | Antwort als eingerücktes JSON | Antwort wie von Slack geliefert | 3.2 |
+
+Selbstgebaute Slack-Meldungen in `on_finish` laufen weiter; einfacher ist `notify` (6.8).
+
 ### 7.1 Von 2.x auf 3.0

 3.0 entfernt das alte Ausführungsmodell: `transform`, den Parallelmodus, `chained` und `condition:` als String
@@ -1571,7 +1691,7 @@ ersten Zeilen sollten gleich sein.
 | [lib/datapipeline/lock.rb](lib/datapipeline/lock.rb) | Sperre gegen parallele Läufe |
 | [lib/datapipeline/connections.rb](lib/datapipeline/connections.rb) | Verbindungsdatei |
 | [lib/datapipeline/storage/](lib/datapipeline/storage/) | `storage_plugin.rb` (Basisklasse) und je Plugin eine Datei |
-| [lib/datapipeline/notification/slack.rb](lib/datapipeline/notification/slack.rb) | Slack |
+| [lib/datapipeline/notification/](lib/datapipeline/notification/) | `notifier.rb` (Report, Basisklasse), `webhook.rb`, `slack.rb`, `email.rb` |
 | [lib/datapipeline/version.rb](lib/datapipeline/version.rb) | `DataPipeline::VERSION` |
 | [exe/etl](exe/etl) | Kommando `etl`; [bin/etl](bin/etl) startet es im Repository mit dem Bundle |
 | [examples/](examples/) | Beispiele aus Kapitel 2, Daten in `examples/data/`, Verbindungsdatei in `examples/config/`, Demo-API |
@@ -1579,14 +1699,14 @@ ersten Zeilen sollten gleich sein.

 ### 8.2 Tests

-`bundle exec rake test` startet die Minitest-Suite in [test/](test/), rund 190 Fälle. Einzelne Datei:
+`bundle exec rake test` startet die Minitest-Suite in [test/](test/), rund 210 Fälle. Einzelne Datei:
 `bundle exec rake test TEST=test/core/steps_test.rb`.

 | Ordner | Inhalt |
 | --- | --- |
-| `test/plugins` | je Plugin das gemeinsame Testkit `test/support/storage_contract.rb` (Schreiben und Zurücklesen, zeilenweise, als Quelle und Ziel) plus eigene Fälle; REST gegen einen lokalen Testserver; Laden der Gems erst bei Nutzung |
+| `test/plugins` | je Plugin das gemeinsame Testkit `test/support/storage_contract.rb` (Schreiben und Zurücklesen, zeilenweise, als Quelle und Ziel) plus eigene Fälle; REST, Webhook und Slack gegen einen lokalen HTTP-Testserver, E-Mail gegen einen SMTP-Testserver (`test/support/`); Laden der Gems erst bei Nutzung |
 | `test/core` | DSL, Schritte und Kurzschritte, Batch und Streaming, Fehlerbehandlung, Layouts, Sperre, Verbindungen, Summen, Historie, Wasserzeichen |
-| `test/cli` | `bin/etl` als Prozess: Parameter, Dry-Run, Sperre, Historie, Exit-Codes |
+| `test/cli` | `bin/etl` als Prozess: Parameter, Dry-Run, Sperre, Historie, Sammelbericht `--notify`, Exit-Codes |
 | `test/examples` | die Beispiele aus Kapitel 2: Code in der Doku = Datei, Lauf und Ergebnis; REST gegen die Demo-API |

 Eine `.env` wird in Tests nicht geladen; kein Test erreicht eine Produktivdatenbank. Postgres-Tests legen je Test eine
@@ -1607,9 +1727,13 @@ Ein eigenes Plugin lässt sich mit `MemoryStorage` aus `test/test_helper.rb` als

 - Version in [lib/datapipeline/version.rb](lib/datapipeline/version.rb), Änderungen in [CHANGELOG.md](CHANGELOG.md)
  (neueste oben). Neue Funktionen erhöhen die mittlere Stelle, Breaking Changes die erste (wie 3.0, 7.1).
- Das `Gemfile` bindet das Gem über `gemspec` ein und ergänzt die Gruppen `plugins` (`pg`, `sqlite3`, `roo`,
+- Jede Version bekommt ein Tag `vX.Y.Z` (`git tag -a v3.2.0 -m "datapipeline 3.2.0" && git push origin v3.2.0`).
+  Anwendungen binden das Gem mit einer Versionsgrenze ein (`'~> 3.1'`, 1.2) oder fest auf ein Tag
+  (`tag: 'v3.2.0'`; dann geht `local.datapipeline` nicht).
+- Das `Gemfile` bindet das Gem über `gemspec` ein und ergänzt die Gruppen `plugins` (`pg`, `sqlite3`, `roo`, `net-smtp`,
  `caxlsx`, `http`) und `test`; `Gemfile.lock` ist versioniert. Ohne Plugin-Gems: `bundle config set without plugins`.
 - Gem bauen: `gem build datapipeline.gemspec`.
+- Lizenz: BSD-3-Clause, siehe [LICENSE](LICENSE); die Datei ist Teil des Gems.
 - Beispiele in Kapitel 2 werden mit einem Marker `<!-- beispiel: examples/datei.rb -->` vor dem Codeblock eingebunden;
  der Doku-Test vergleicht Block und Datei, Änderungen also immer in beiden machen.

@@ -1623,8 +1747,5 @@ Offene Punkte, Reihenfolge = empfohlene Umsetzung:
   schreiben), `etl list` (Pipelines, Wasserzeichen, letzter Lauf aus der Historie), `--json` für die Stats.
 2. **Qualität:** CI auf git.wicom.li mit `rake test` und Postgres-Service, RuboCop, `DataPipeline::Testing` mit
   `MemoryStorage` als öffentliche Testhilfe für Anwendungen.
-3. **SQLite im Streaming:** `write_row` wie bei Postgres in Paketen schreiben.
-4. **Slack:** fertige Meldung aus den Stats (`notify slack, on: :problems`), Timeout und Wiederholung bei `429`,
-   Maskierung von `&`, `<`, `>`, Incoming Webhooks.
-5. **Weitere Plugins:** SFTP-Ablage.
-6. **Veröffentlichung:** eigenes Remote-Repository, Lizenz festlegen, Versionen taggen.
+3. **Benachrichtigungen:** Microsoft Teams als mitgelieferter Notifier, HTML-Mail.
+4. **Weitere Plugins:** SFTP-Ablage.
@@ -12,13 +12,13 @@ Gem::Specification.new do |spec|
    Fehlerdateien, Dry-Run, inkrementellen Läufen und dem Kommando etl mit Exit-Codes.
  TEXT
  spec.homepage      = 'https://git.wicom.li/wkh/datapipeline'
-  spec.license       = 'Nonstandard'
+  spec.license       = 'BSD-3-Clause'
  spec.required_ruby_version = '>= 3.4'

  spec.metadata['changelog_uri'] = "#{spec.homepage}/src/branch/main/CHANGELOG.md"
  spec.metadata['rubygems_mfa_required'] = 'true'

-  spec.files         = Dir['lib/**/*.rb', 'exe/*', 'README.md', 'CHANGELOG.md']
+  spec.files         = Dir['lib/**/*.rb', 'exe/*', 'README.md', 'CHANGELOG.md', 'LICENSE']
  spec.bindir        = 'exe'
  spec.executables   = ['etl']
  spec.require_paths = ['lib']
@@ -28,5 +28,5 @@ Gem::Specification.new do |spec|
  spec.add_dependency 'logger', '~> 1.6'

  # optional, loaded only when the plugin is used (add them to the Gemfile of the application):
-  #   pg (postgres), sqlite3 (sqlite), roo + caxlsx (excel), http (rest, slack)
+  #   pg (postgres), sqlite3 (sqlite), roo + caxlsx (excel), http (rest, slack, webhook), net-smtp (email)
 end
@@ -14,13 +14,11 @@ FileUtils.mkdir_p(out)
 # Zugangsdaten stehen nicht im Skript, sondern in der Verbindungsdatei (Default: config/connections.yml)
 DataPipeline.connections_file = File.expand_path('config/connections.yml', __dir__)

-# Slack nur, wenn ein Token gesetzt ist; so fällt ein fehlender Token schon vor dem Lauf auf
-slack = DataPipeline::Slack.new(channel: '#etl') if ENV['SLACK_TOKEN']
-
 DataPipeline.new do
  name 'buchungen_export'
  lock "#{out}/.buchungen.lock"              # läuft schon ein Export, bricht dieser ab (Exit-Code 3)
  history "#{out}/runs.ndjson"               # eine JSON-Zeile je Lauf, auch nach Abbruch
+  notify DataPipeline::Slack.new(channel: '#etl') if ENV['SLACK_TOKEN']   # Meldung nur bei Problemen

  read container(:csv, file: "#{DATA}/buchungen.csv", csv: { col_sep: ';' })

@@ -52,8 +50,5 @@ DataPipeline.new do
  write container(:csv, file: "#{out}/buchungen_sap.csv", encoding: 'Windows-1252', csv: { col_sep: ';' },
                        columns: %w[belnr buzei datum konto shkzg dmbtr buchungstext])

-  on_finish do |stats|
-    puts "#{stats.name}: #{stats}"
-    slack&.send_message("*#{stats.name}*: #{stats}") unless stats.ok?
-  end
+  on_finish { |stats| puts "#{stats.name}: #{stats}" }
 end.run
@@ -11,6 +11,8 @@
 #   --verbose                 log level debug, --quiet only warnings (default info)
 #   --lock=FILE               exclusive lock for the whole script, exit 3 when another run holds it
 #   --history=FILE            one json line per pipeline run (DataPipeline.history)
+#   --notify=NAME             one report for all runs of the script, notifier NAME from the connections file
+#   --notify-on=WHEN          failure, problems (default) or always
 #   --name=value              any other option: DataPipeline.params[:name] and passed on to the script
 #
 # exit code: 0 ok, 1 aborted, 2 finished with rejected rows, 3 locked by another run
@@ -35,6 +37,8 @@ parser = OptionParser.new do |opts|
  opts.on('--quiet') { options[:settings][:log_level] = Logger::WARN }
  opts.on('--lock=FILE', String) { |v| options[:lock] = v }
  opts.on('--history=FILE', String) { |v| options[:settings][:history] = v }
+  opts.on('--notify=NAME', String) { |v| options[:notify] = v }
+  opts.on('--notify-on=WHEN', String) { |v| options[:notify_on] = v }
 end

 # own options for the parser, everything else (also --name=value of the script) goes to the script
@@ -42,9 +46,9 @@ own = []
 args = []
 argv = ARGV.dup
 while (arg = argv.shift)
-  if arg.match?(/\A--(from|till|param|dry-run|limit|verbose|quiet|lock|history)(=|\z)/)
+  if arg.match?(/\A--(from|till|param|dry-run|limit|verbose|quiet|lock|history|notify|notify-on)(=|\z)/)
    own << arg
-    own << argv.shift if %w[--param --limit --from --till --lock --history].include?(arg)
+    own << argv.shift if %w[--param --limit --from --till --lock --history --notify --notify-on].include?(arg)
  else
    args << arg
  end
@@ -56,7 +60,26 @@ DataPipeline.params.merge!(DataPipeline.parse_params(args)) { |_key, own_value,
 abort parser.help if script.nil?
 abort "etl: #{script} not found\n\n#{parser.help}" unless File.exist?(script)

-DataPipeline.settings(options[:settings].merge(name: File.basename(script, '.rb')))
+name = File.basename(script, '.rb')
+DataPipeline.settings(options[:settings].merge(name:))
+
+# one report for the whole script; created before the run, so a wrong name fails early
+if options[:notify]
+  begin
+    notifier = DataPipeline::Notifier.from_connection(options[:notify])
+    notify_when = DataPipeline::Notifier.condition(options.fetch(:notify_on, :problems))
+  rescue ArgumentError, LoadError => e
+    abort "etl: --notify=#{options[:notify]}: #{e.message}"
+  end
+end
+summary = lambda do |error = nil|
+  return unless notifier && !options[:settings][:dry_run]
+
+  report = DataPipeline::Report.new(DataPipeline.runs, title: name, params: DataPipeline.params, error:)
+  notifier.deliver(report) if notify_when.call(report)
+rescue StandardError => e
+  warn "#{name}: notify #{options[:notify]}: #{e.class}: #{e.message}"
+end

 begin
  script_lock = DataPipeline::Lock.new(options[:lock]).acquire if options[:lock]   # held until the process ends
@@ -72,9 +95,12 @@ begin
  load File.expand_path(script)
 rescue StandardError => e
  warn "#{File.basename(script)}: #{e.class}: #{e.message}"
+  # an aborted pipeline is already in the runs, other errors (script, lock) are reported as such
+  summary.call(DataPipeline.runs.any? { |stats| stats.error.equal?(e) } ? nil : e) unless e.is_a?(DataPipeline::LockError)
  exit(e.is_a?(DataPipeline::LockError) ? 3 : 1)
 end

+summary.call
 runs = DataPipeline.runs
 exit 3 if runs.any? { |stats| stats.error.is_a?(DataPipeline::LockError) }
 exit 1 if runs.any?(&:error)
@@ -8,10 +8,11 @@ require_relative 'datapipeline/layout'
 require_relative 'datapipeline/watermark'
 require_relative 'datapipeline/lock'
 require_relative 'datapipeline/connections'
+require_relative 'datapipeline/notification/notifier'

 class DataPipeline
  # plugins are loaded on first use, their gems too: pg (postgres), sqlite3 (sqlite), roo/caxlsx (excel),
-  # http (rest, slack)
+  # http (rest, slack, webhook), net-smtp (email)
  autoload :CsvStorage,      File.expand_path('datapipeline/storage/csv', __dir__)
  autoload :PostgresStorage, File.expand_path('datapipeline/storage/postgres', __dir__)
  autoload :JsonStorage,     File.expand_path('datapipeline/storage/json', __dir__)
@@ -20,6 +21,8 @@ class DataPipeline
  autoload :RestStorage,     File.expand_path('datapipeline/storage/rest', __dir__)
  autoload :FixedStorage,    File.expand_path('datapipeline/storage/fixed', __dir__)
  autoload :Slack,           File.expand_path('datapipeline/notification/slack', __dir__)
+  autoload :Webhook,         File.expand_path('datapipeline/notification/webhook', __dir__)
+  autoload :Email,           File.expand_path('datapipeline/notification/email', __dir__)

  # passed to header, trailer, map and filter
  #   read:    number of source rows (nil while streaming, until the trailer)
@@ -240,6 +243,26 @@ class DataPipeline
    @reject_target = to
  end

+  #
+  # notify notifier, on: :problems           -> a report after the run, see DataPipeline::Notifier
+  #   on:           :failure (abort), :problems (abort, rejected rows or warnings, default), :always
+  #   title:        default the name of the pipeline
+  #   dry_run:      true -> also after a dry run (default: not)
+  #   include_rows: true -> rejected rows in the report (default: not, they may hold customer data)
+  #
+  # an error while sending is only logged
+  #
+  def notify notifier, on: :problems, title: nil, dry_run: false, include_rows: false
+    raise ArgumentError, 'notify: notifier needs deliver(report)' unless notifier.respond_to?(:deliver)
+
+    (@notifiers ||= []) << { notifier:, when: Notifier.condition(on), title:, dry_run:, include_rows: }
+  end
+
+  # notifier(:alerts) -> Webhook, Slack or Email from the connections file (type webhook, slack, email)
+  def notifier name
+    Notifier.from_connection(name)
+  end
+
  # on_finish { |stats| ... } -> after every run, also after an abort (stats.error)
  def on_finish(&block)
    @on_finish << block
@@ -484,6 +507,7 @@ class DataPipeline
    print_preview if @stats.dry_run && @stats.error.nil?
    @logger.info "finish #{@stats}"
    append_history
+    send_notifications
    @on_finish.each do |hook|
      hook.call(@stats)
    rescue StandardError => e
@@ -504,6 +528,20 @@ class DataPipeline
    @logger.error "history: #{e.class}: #{e.message}"
  end

+  def send_notifications
+    Array(@notifiers).each do |entry|
+      next if @stats.dry_run && !entry[:dry_run]
+
+      report = Report.new([@stats], title: entry[:title], params:)
+      next unless entry[:when].call(report)
+
+      entry[:notifier].deliver(report, include_rows: entry[:include_rows])
+      @logger.info "notified #{entry[:notifier].class.name.split('::').last}: #{report.event}"
+    rescue StandardError => e
+      @logger.error "notify #{entry[:notifier].class}: #{e.class}: #{e.message}"
+    end
+  end
+
  def add_total side, row
    @stats.totals&.each do |col, sum|
      value = row.key?(col) ? row[col] : row[col.to_sym]
@@ -0,0 +1,93 @@
+require 'securerandom'
+require_relative 'notifier'
+require_relative '../storage/storage_plugin'
+
+class DataPipeline
+  #
+  # e-mail over SMTP, text/plain UTF-8
+  #
+  #   notify DataPipeline::Email.new(to: 'etl@example.com'), on: :failure
+  #
+  # options (defaults from ENV)
+  #   to:        one address or an array (required)
+  #   from:      sender, default ENV['SMTP_FROM'] or etl@<hostname>
+  #   host:      ENV['SMTP_HOST'], default localhost
+  #   port:      ENV['SMTP_PORT'], default 587 (465 with tls: true, 25 without starttls)
+  #   username:  ENV['SMTP_USERNAME'], password: ENV['SMTP_PASSWORD']; without username no login
+  #   tls:       true -> TLS from the start (port 465)
+  #   starttls:  :auto (default, if the server offers it), :always or false
+  #   subject_prefix: default '[etl] '
+  #   timeout:   seconds for open and read (default 15)
+  #
+  class Email < Notifier
+    class Error < StandardError; end
+
+    def initialize to:, from: nil, host: nil, port: nil, username: nil, password: nil, tls: false,
+                   starttls: :auto, subject_prefix: '[etl] ', timeout: 15
+      StoragePlugin.require_gem 'net/smtp', 'net-smtp'
+      @to = Array(to).map(&:to_s).reject(&:empty?)
+      raise ArgumentError, 'Email: to missing' if @to.empty?
+
+      @from = from || ENV['SMTP_FROM'] || "etl@#{Socket.gethostname}"
+      @host = host || ENV.fetch('SMTP_HOST', 'localhost')
+      @tls = tls
+      @starttls = tls || !starttls ? false : starttls.to_sym
+      @port = Integer(port || ENV['SMTP_PORT'] || default_port)
+      @username = username || ENV['SMTP_USERNAME']
+      @password = password || ENV['SMTP_PASSWORD']
+      @subject_prefix = subject_prefix
+      @timeout = timeout
+    end
+
+    def deliver report, include_rows: false
+      send_mail("#{@subject_prefix}#{report.subject}", body_of(report, include_rows))
+    end
+
+    # any text, e.g. from on_finish
+    def send_mail subject, body
+      message = build(subject, body)
+      smtp = Net::SMTP.new(@host, @port, tls: @tls, starttls: @starttls || false)
+      smtp.open_timeout = @timeout
+      smtp.read_timeout = @timeout
+      login = @username ? { user: @username, secret: @password, authtype: :login } : {}
+      smtp.start(helo: Socket.gethostname, **login) { |session| session.send_message(message, @from, *@to) }
+      message
+    rescue Net::SMTPError, SystemCallError, IOError, Timeout::Error, OpenSSL::SSL::SSLError => e
+      raise Error, "#{@host}:#{@port}: #{e.class}: #{e.message}"
+    end
+
+    private
+
+    def default_port
+      return 465 if @tls
+
+      @starttls ? 587 : 25
+    end
+
+    # rows of rejected lines only on request, they may hold customer data
+    def body_of report, include_rows
+      text = report.to_s
+      return text unless include_rows
+
+      rows = report.runs.flat_map { |stats| stats.errors.first(Report::MAX_ERRORS).map { |e| "#{e[:step]}: #{JSON.generate(e[:row])}" } }
+      rows.empty? ? text : "#{text}\n\nZeilen:\n#{rows.join("\n")}"
+    end
+
+    # RFC 5322 with a base64 body, the subject RFC 2047 encoded when it is not ascii
+    def build subject, body
+      encoded_subject = subject.ascii_only? ? subject : "=?UTF-8?B?#{[subject].pack('m0')}?="
+      [
+        "From: #{@from}",
+        "To: #{@to.join(', ')}",
+        "Subject: #{encoded_subject}",
+        "Date: #{Time.now.strftime('%a, %d %b %Y %H:%M:%S %z')}",
+        "Message-ID: <#{SecureRandom.uuid}@#{@from.split('@').last}>",
+        'MIME-Version: 1.0',
+        'Content-Type: text/plain; charset=UTF-8',
+        'Content-Transfer-Encoding: base64',
+        '',
+        [body].pack('m')
+      ].join("\r\n")
+    end
+  end
+end
@@ -0,0 +1,141 @@
+require 'json'
+require 'securerandom'
+require 'socket'
+
+class DataPipeline
+  #
+  # what a notifier gets: the stats of one or more runs (one script), with event, title and text lines
+  #
+  #   event: run.failed (an abort), run.rejected (rejected rows), run.warnings, run.succeeded
+  #   to_h:  stable payload for webhooks (schema 1); rows of stats.errors only with include_rows: true,
+  #          because they may hold customer data
+  #
+  class Report
+    EVENTS = %w[run.succeeded run.warnings run.rejected run.failed].freeze
+    MAX_ERRORS = 20
+
+    attr_reader :id, :title, :runs, :params, :host, :error
+
+    # error: an exception outside of a run (e.g. the script failed before or between pipelines)
+    def initialize runs, title: nil, params: {}, error: nil
+      @id = SecureRandom.uuid
+      @runs = Array(runs)
+      @title = title || @runs.filter_map(&:name).uniq.join(', ').then { |t| t.empty? ? 'datapipeline' : t }
+      @params = params.to_h
+      @host = Socket.gethostname
+      @error = error
+    end
+
+    # the worst event of all runs
+    def event
+      return 'run.failed' if @error
+
+      @runs.map { |stats| event_of(stats) }.max_by { |event| EVENTS.index(event) } || 'run.succeeded'
+    end
+
+    def failed? = event == 'run.failed'
+    def ok? = event == 'run.succeeded'
+
+    def status
+      { 'run.failed' => 'abgebrochen', 'run.rejected' => 'mit verworfenen Zeilen',
+        'run.warnings' => 'mit Warnungen', 'run.succeeded' => 'ok' }.fetch(event)
+    end
+
+    def subject = "#{@title}: #{status}"
+
+    # text for mail, slack and logs: one block per run, the first errors without row data
+    def lines errors: 5
+      lines = ["#{subject} (#{@host})"]
+      lines << "Parameter: #{@params.map { |key, value| "#{key}=#{value}" }.join(' ')}" if @params.any?
+      lines << "Fehler: #{@error.class}: #{@error.message}" if @error
+      @runs.each do |stats|
+        lines << ''
+        lines << "#{stats.name || 'Pipeline'}#{" #{stats.version}" if stats.version}: #{stats}"
+        lines << "Start #{stats.started_at&.strftime('%d.%m.%Y %H:%M:%S')}, Dauer #{stats.duration}s"
+        stats.errors.first(errors).each { |e| lines << "- #{e[:step]}: #{e[:error]}" }
+        lines << "- … und #{stats.rejected - errors} weitere" if stats.rejected > errors
+      end
+      lines
+    end
+
+    def to_s = lines.join("\n")
+
+    def to_h include_rows: false
+      {
+        schema: 1, id: @id, event:, title: @title, host: @host, params: @params,
+        error: @error && "#{@error.class}: #{@error.message}",
+        runs: @runs.map do |stats|
+          stats.to_h.merge(errors: stats.errors.first(MAX_ERRORS).map { |e| include_rows ? e : e.except(:row) })
+        end
+      }
+    end
+
+    private
+
+    def event_of stats
+      return 'run.failed' if stats.error
+      return 'run.rejected' if stats.rejected.positive?
+      return 'run.warnings' if stats.warnings.to_i.positive?
+
+      'run.succeeded'
+    end
+  end
+
+  #
+  # a notifier only needs deliver(report, include_rows: false); see Webhook, Slack, Email
+  #
+  #   notify DataPipeline::Webhook.new(url: ...), on: :problems
+  #
+  class Notifier
+    # when a notification is sent
+    WHEN = {
+      always: ->(_report) { true },
+      failure: ->(report) { report.failed? },
+      problems: ->(report) { !report.ok? }
+    }.freeze
+
+    TYPES = { webhook: 'DataPipeline::Webhook', slack: 'DataPipeline::Slack', email: 'DataPipeline::Email' }.freeze
+
+    # notifier from the connections file: type webhook, slack or email, the other keys are its options
+    def self.from_connection name
+      options = Connections.fetch(name)
+      type = options[:type].to_s.to_sym
+      klass = TYPES.fetch(type) { raise ArgumentError, "connection #{name.inspect}: type #{type} is no notifier (#{TYPES.keys.join(', ')})" }
+      Object.const_get(klass).new(**options.except(:type))
+    end
+
+    def self.condition on
+      WHEN.fetch(on.to_sym) { raise ArgumentError, "notify on: #{on} unknown, use #{WHEN.keys.join(', ')}" }
+    end
+
+    def deliver(_report, include_rows: false)
+      raise NotImplementedError, "#{self.class}#deliver is missing"
+    end
+
+    private
+
+    # POST with retries on 429/5xx (Retry-After honoured) and network errors; other 4xx raise at once
+    def post_with_retries url, body:, headers:, timeout:, retries:, retry_wait:, error:
+      attempt = 0
+      loop do
+        attempt += 1
+        begin
+          response = HTTP.timeout(timeout).headers(headers).post(url, body:)
+        rescue HTTP::Error, SystemCallError, IOError => e
+          raise error, "#{url}: #{e.class}: #{e.message}" if attempt > retries
+
+          sleep(retry_wait * attempt)
+          next
+        end
+        return response if response.status.success?
+
+        code = response.code
+        unless attempt <= retries && (code == 429 || code >= 500)
+          raise error, "#{url}: HTTP #{code} #{response.body.to_s[0, 200]}"
+        end
+
+        sleep(response.headers['Retry-After']&.to_f || retry_wait * attempt)
+      end
+    end
+  end
+end
@@ -1,34 +1,88 @@
-require 'http'
 require 'json'
+require_relative 'notifier'
+require_relative '../storage/storage_plugin'

 class DataPipeline
  #
  # Slack.new(channel: '#etl').send_message('posting export done')
+  # notify DataPipeline::Slack.new(channel: '#etl'), on: :problems
+  # notify DataPipeline::Slack.new(webhook: ENV['SLACK_WEBHOOK'])      # incoming webhook, no token needed
  #
  # options
-  #   token:   bot token, default ENV['SLACK_TOKEN']
-  #   channel: default '#general'
+  #   token:      bot token for chat.postMessage, default ENV['SLACK_TOKEN'] (not needed with webhook)
+  #   channel:    default '#general' (with webhook the channel of the webhook)
+  #   webhook:    url of an incoming webhook instead of token and channel
+  #   mention:    e.g. '<!here>' or '<!subteam^ID>', only in reports of failed runs
+  #   timeout:    seconds per request (default 10)
+  #   retries:    repetitions on 429, 5xx and network errors (default 3)
+  #   retry_wait: seconds, multiplied by the attempt (default 2)
+  #   api_url:    default https://slack.com/api/chat.postMessage
  #
-  class Slack
+  class Slack < Notifier
    API_URL = 'https://slack.com/api/chat.postMessage'.freeze
+    MAX_TEXT = 2900 # a section block holds 3000 characters
+
+    class Error < StandardError; end

    DEFAULTS = {
      token: nil,
-      channel: '#general'
+      channel: '#general',
+      webhook: nil,
+      mention: nil,
+      timeout: 10,
+      retries: 3,
+      retry_wait: 2,
+      api_url: API_URL
    }.freeze

+    ICONS = { 'run.failed' => ':red_circle:', 'run.rejected' => ':large_orange_circle:',
+              'run.warnings' => ':large_yellow_circle:', 'run.succeeded' => ':large_green_circle:' }.freeze
+
+    # & < > have a meaning in slack texts
+    def self.escape(text) = text.to_s.gsub('&', '&amp;').gsub('<', '&lt;').gsub('>', '&gt;')
+
    def initialize options = {}
+      StoragePlugin.require_gem 'http'
      @options = DEFAULTS.merge(token: ENV['SLACK_TOKEN']).merge(options)
-      raise ArgumentError, 'Slack: token missing (option token: or ENV SLACK_TOKEN)' unless @options[:token]
+      return if @options[:webhook] || @options[:token]
+
+      raise ArgumentError, 'Slack: token missing (option token: or ENV SLACK_TOKEN, or webhook:)'
    end

-    # returns the parsed slack response, raises when slack reports an error
+    # plain text (slack markup allowed); returns the parsed slack response, raises when slack reports an error
    def send_message(content, verbose: false)
-      response = HTTP.auth("Bearer #{@options[:token]}")
-                     .post(API_URL, json: { channel: @options[:channel], text: content })
-      body = JSON.parse(response.body.to_s)
-      puts JSON.pretty_generate(body) if verbose
-      raise "Slack#send_message failed: #{body['error']}" unless body['ok']
+      post({ text: content }, verbose:)
+    end
+
+    # the report as blocks: status line, then the runs as preformatted text
+    def deliver report, include_rows: false
+      icon = ICONS.fetch(report.event)
+      mention = report.failed? && @options[:mention] ? " #{@options[:mention]}" : ''
+      head = "#{icon} *#{self.class.escape(report.title)}*: #{report.status}#{mention}"
+      details = self.class.escape(report.lines.drop(1).join("\n").strip)
+      details = "#{details[0, MAX_TEXT]}\n…" if details.length > MAX_TEXT
+      blocks = [{ type: 'section', text: { type: 'mrkdwn', text: head } }]
+      blocks << { type: 'section', text: { type: 'mrkdwn', text: "```#{details}```" } } unless details.empty?
+      post({ text: "#{report.subject}#{mention}", blocks: })
+    end
+
+    private
+
+    def post payload, verbose: false
+      url = @options[:webhook] || @options[:api_url]
+      headers = { 'Content-Type' => 'application/json; charset=utf-8' }
+      unless @options[:webhook]
+        headers['Authorization'] = "Bearer #{@options[:token]}"
+        payload = payload.merge(channel: @options[:channel])
+      end
+      response = post_with_retries(url, body: JSON.generate(payload), headers:, timeout: @options[:timeout],
+                                        retries: @options[:retries], retry_wait: @options[:retry_wait], error: Error)
+      text = response.body.to_s
+      puts text if verbose
+      return text if @options[:webhook] # incoming webhooks answer "ok"
+
+      body = JSON.parse(text)
+      raise Error, "Slack#send_message failed: #{body['error']}" unless body['ok']

      body
    end
@@ -0,0 +1,63 @@
+require 'openssl'
+require_relative 'notifier'
+require_relative '../storage/storage_plugin'
+
+class DataPipeline
+  #
+  # POST of the report as json (Report#to_h, schema 1)
+  #
+  #   notify DataPipeline::Webhook.new(url: ENV['ALERT_URL'], secret: ENV['ALERT_SECRET'])
+  #
+  # options
+  #   url:        target (required)
+  #   secret:     signs the body: X-DataPipeline-Signature: sha256=HMAC_SHA256(secret, "<timestamp>.<body>")
+  #               with X-DataPipeline-Timestamp (unix seconds)
+  #   headers:    more headers, e.g. { 'Authorization' => 'Bearer ...' }
+  #   timeout:    seconds per request (default 10)
+  #   retries:    repetitions on 429, 5xx and network errors (default 3), Retry-After is honoured
+  #   retry_wait: seconds, multiplied by the attempt (default 2)
+  #
+  # every request carries X-DataPipeline-Event and X-DataPipeline-Delivery (the report id, the same on retries)
+  #
+  class Webhook < Notifier
+    class Error < StandardError; end
+
+    def initialize url:, secret: nil, headers: {}, timeout: 10, retries: 3, retry_wait: 2
+      StoragePlugin.require_gem 'http'
+      @url = url.to_s
+      raise ArgumentError, 'Webhook: url missing' if @url.empty?
+
+      @secret = secret
+      @headers = headers.to_h { |key, value| [key.to_s, value.to_s] }
+      @timeout = timeout
+      @retries = retries
+      @retry_wait = retry_wait
+    end
+
+    def deliver report, include_rows: false
+      body = JSON.generate(report.to_h(include_rows:))
+      post_with_retries(@url, body:, headers: headers_for(report, body), timeout: @timeout, retries: @retries,
+                              retry_wait: @retry_wait, error: Error)
+    end
+
+    # for receivers: true when signature and timestamp (max_age seconds) of a request are valid
+    def self.valid_signature? secret:, body:, timestamp:, signature:, max_age: 300
+      return false if timestamp.to_s.empty? || (Time.now.to_i - timestamp.to_i).abs > max_age
+
+      expected = "sha256=#{OpenSSL::HMAC.hexdigest('SHA256', secret, "#{timestamp}.#{body}")}"
+      expected.bytesize == signature.to_s.bytesize && OpenSSL.fixed_length_secure_compare(expected, signature.to_s)
+    end
+
+    private
+
+    def headers_for report, body
+      headers = { 'Content-Type' => 'application/json', 'User-Agent' => "datapipeline/#{VERSION}",
+                  'X-DataPipeline-Event' => report.event, 'X-DataPipeline-Delivery' => report.id }.merge(@headers)
+      return headers unless @secret
+
+      timestamp = Time.now.to_i.to_s
+      signature = OpenSSL::HMAC.hexdigest('SHA256', @secret, "#{timestamp}.#{body}")
+      headers.merge('X-DataPipeline-Timestamp' => timestamp, 'X-DataPipeline-Signature' => "sha256=#{signature}")
+    end
+  end
+end
@@ -48,9 +48,12 @@ class DataPipeline
      each_row(@options[:query], params, &block)
    end

+    # pending rows of write_row are written first, so a query sees them
    def query options = {}
      raise "Query for PostgresStorage:query missing" unless options[:query]

+      flush
+
      result = @conn.exec_params(options[:query], options.fetch(:params, []))
      return result.to_a unless block_given?

@@ -12,6 +12,8 @@ class DataPipeline
  #   truncate: true -> delete all rows of the table before the first write
  #   upsert:   %w[belnr] -> insert ... on conflict (belnr) do update; with create: true the unique index
  #             is created too, otherwise the table needs one on these columns
+  #   batch_size: write_row collects rows and writes them in one transaction per batch (default 1000);
+  #             :all -> one transaction for the whole stream, committed by finalize, rolled back by close
  #
  # rows come back as Hash<String, value> with sqlite types (Integer, Float, String, nil).
  # the database file is opened on first use: a dry run creates no file.
@@ -33,9 +35,11 @@ class DataPipeline
      query(query: @options[:query], params: @options.fetch(:params, []), &block)
    end

+    # pending rows of write_row are written first, so a query sees them
    def query options = {}
      raise 'Query for SqliteStorage:query missing' unless options[:query]

+      flush
      statement = db.prepare(options[:query])
      result = statement.execute(*bind_args(options.fetch(:params, [])))
      return result.map(&:to_h) unless block_given?
@@ -58,20 +62,47 @@ class DataPipeline

      return if data.empty?

-      db.transaction do
+      transaction do
        prepare_table table_name, data.first
        data.each { |row| insert table_name, row }
      end
    end

+    # rows are collected and written per batch_size in one transaction
    def write_row row
      raise 'Table for SqliteStorage#write_row missing' unless @options[:table]

-      prepare_table @options[:table], row
-      insert @options[:table], row
+      if batch_size == :all
+        begin_stream
+        prepare_table @options[:table], row
+        insert @options[:table], row
+      else
+        (@pending ||= []) << row
+        flush if @pending.size >= batch_size
+      end
    end

+    # end of a stream: remaining rows, commit of batch_size: :all
+    def finalize
+      flush
+      if @stream_open
+        db.commit
+        @stream_open = false
+      end
+      close
+    end
+
+    # pending rows are written; an open stream with batch_size: :all is rolled back (run aborted);
+    # an error while flushing does not hide the error that aborted the run
    def close
+      aborted_by = $!
+      begin
+        flush
+      rescue StandardError
+        raise unless aborted_by
+      end
+      @db.rollback if @stream_open && @db.transaction_active?
+      @stream_open = false
      @db.close if @db && !@db.closed?
    end

@@ -81,6 +112,38 @@ class DataPipeline
      @db ||= SQLite3::Database.new(@database).tap { |d| d.results_as_hash = true }
    end

+    def batch_size
+      size = @options.fetch(:batch_size, 1000)
+      size == :all ? :all : Integer(size)
+    end
+
+    def flush
+      return if @pending.nil? || @pending.empty?
+
+      rows = @pending
+      @pending = []
+      transaction do
+        rows.each do |row|
+          prepare_table @options[:table], row
+          insert @options[:table], row
+        end
+      end
+    end
+
+    def begin_stream
+      return if @stream_open
+
+      raise 'SqliteStorage: batch_size: :all needs its own connection (no open transaction)' if db.transaction_active?
+
+      db.transaction
+      @stream_open = true
+    end
+
+    # inside an open transaction no own transaction is opened
+    def transaction(&block)
+      db.transaction_active? ? yield : db.transaction(&block)
+    end
+
    # create / truncate once per table, before its first row
    def prepare_table table_name, row
      return if @prepared[table_name]
@@ -1,5 +1,5 @@
 class DataPipeline
-  VERSION = '3.0.0'.freeze
+  VERSION = '3.2.0'.freeze

  # kept for scripts that print the version (rake info:info); the history is in CHANGELOG.md
  class Version
@@ -1,5 +1,6 @@
 require_relative '../test_helper'
 require 'open3'
+require_relative '../support/test_http_server'

 # bin/etl as a subprocess: arguments, params, dry run, exit codes
 class EtlCliTest < PipelineTest
@@ -127,4 +128,44 @@ class EtlCliTest < PipelineTest
      assert_equal ['export_kunden', 3, true], entry.values_at('name', 'read', 'ok')
    end
  end
+
+  def test_notify_one_report_for_all_pipelines_of_the_script
+    server = TestHttpServer.new { [200, 'ok'] }
+    with_tmpdir do |dir|
+      Dir.mkdir("#{dir}/config")
+      File.write("#{dir}/config/connections.yml", "alerts:\n  type: webhook\n  url: #{server.url}/hook\n")
+      path = script(dir, 'zwei.rb', <<~RUBY)
+        2.times do |i|
+          DataPipeline.new do
+            name "teil\#{i}"
+            on_error :skip
+            read container(:csv, file: 'in.csv')
+            validate 'a' => { format: /\\A\\d+\\z/ } if i == 1
+            write container(:csv, file: "out\#{i}.csv")
+          end.run
+        end
+      RUBY
+      _out, err, status = etl(dir, 'run', path, '--notify=alerts')
+      assert_equal 2, status.exitstatus, err
+      assert_equal 1, server.requests.size
+      payload = JSON.parse(server.requests.last[:body])
+      assert_equal ['run.rejected', 'zwei', %w[teil0 teil1]], [payload['event'], payload['title'], payload['runs'].map { |r| r['name'] }]
+
+      _out, err, status = etl(dir, 'run', path, '--notify=alerts', '--notify-on=failure')
+      assert_equal 2, status.exitstatus, err
+      assert_equal 1, server.requests.size # nothing failed
+
+      broken = script(dir, 'kaputt.rb', "raise 'Skriptfehler'\n")
+      _out, _err, status = etl(dir, 'run', broken, '--notify=alerts')
+      assert_equal 1, status.exitstatus
+      payload = JSON.parse(server.requests.last[:body])
+      assert_equal ['run.failed', 'RuntimeError: Skriptfehler', []], payload.values_at('event', 'error', 'runs')
+
+      _out, err, status = etl(dir, 'run', path, '--notify=gibtsnicht')
+      assert_equal 1, status.exitstatus
+      assert_includes err, 'etl: --notify=gibtsnicht: connection "gibtsnicht" not in config/connections.yml (alerts)'
+    end
+  ensure
+    server&.stop
+  end
 end
@@ -1,4 +1,5 @@
 require_relative '../test_helper'
+require 'time'

 # connections file, totals with reconcile, run history
 class ConnectionsTotalsHistoryTest < PipelineTest
@@ -0,0 +1,206 @@
+require_relative '../test_helper'
+require_relative '../support/test_http_server'
+require_relative '../support/test_smtp_server'
+
+# Report, Webhook, Slack, Email and notify against local test servers
+class NotificationTest < PipelineTest
+  ROWS = [{ 'id' => '1', 'konto' => '1000' }, { 'id' => '2', 'konto' => '' }].freeze
+
+  # collects reports, for notify without network
+  class MemoryNotifier < DataPipeline::Notifier
+    attr_reader :reports
+
+    def initialize(fail: false) = (@reports = []; @fail = fail)
+
+    def deliver report, include_rows: false
+      raise 'kaputt' if @fail
+
+      @reports << [report, include_rows]
+    end
+  end
+
+  def teardown
+    @server&.stop
+    DataPipeline.connections_file = nil
+    super
+  end
+
+  def stats_with_rejected
+    run_pipeline(ROWS.map(&:dup)) { name 'export'; on_error :skip; validate 'konto' => { required: true } }.last
+  end
+
+  def report(stats = stats_with_rejected) = DataPipeline::Report.new([stats], params: { from: '2026-10-01' })
+
+  def test_report_event_lines_and_payload_without_rows
+    report = report()
+    assert_equal ['run.rejected', 'export: mit verworfenen Zeilen'], [report.event, report.subject]
+    assert_includes report.to_s, '- validate#1: konto: missing'
+    assert_includes report.to_s, 'Parameter: from=2026-10-01'
+    payload = report.to_h
+    assert_equal [1, 'run.rejected', 'export'], payload.values_at(:schema, :event, :title)
+    assert_equal [{ step: 'validate#1', error: 'konto: missing' }], payload[:runs].first[:errors]
+    assert_equal({ 'id' => '2', 'konto' => '' }, report.to_h(include_rows: true)[:runs].first[:errors].first[:row])
+  end
+
+  def test_report_worst_event_of_several_runs_and_script_error
+    ok = run_pipeline(ROWS.map(&:dup)).last
+    assert_equal 'run.succeeded', DataPipeline::Report.new([ok]).event
+    assert_equal 'run.rejected', DataPipeline::Report.new([ok, stats_with_rejected]).event
+    failed = DataPipeline::Report.new([ok], title: 'skript', error: RuntimeError.new('kaputt'))
+    assert_equal %w[run.failed skript], [failed.event, failed.title]
+    assert_includes failed.to_s, 'Fehler: RuntimeError: kaputt'
+  end
+
+  def test_webhook_signed_post
+    @server = TestHttpServer.new { [200, 'ok'] }
+    report = report()
+    DataPipeline::Webhook.new(url: "#{@server.url}/hook", secret: 's3cret', headers: { 'X-Team' => 'etl' }).deliver(report)
+    request = @server.requests.last
+    headers = request[:headers]
+    assert_equal ['/hook', 'run.rejected', report.id, 'etl'],
+                 [request[:path], headers['x-datapipeline-event'], headers['x-datapipeline-delivery'], headers['x-team']]
+    assert_equal 1, JSON.parse(request[:body])['schema']
+    refute_includes request[:body], '"row"'
+    assert DataPipeline::Webhook.valid_signature?(secret: 's3cret', body: request[:body],
+                                                  timestamp: headers['x-datapipeline-timestamp'],
+                                                  signature: headers['x-datapipeline-signature'])
+    refute DataPipeline::Webhook.valid_signature?(secret: 'falsch', body: request[:body],
+                                                  timestamp: headers['x-datapipeline-timestamp'],
+                                                  signature: headers['x-datapipeline-signature'])
+    refute DataPipeline::Webhook.valid_signature?(secret: 's3cret', body: request[:body], timestamp: (Time.now.to_i - 600).to_s,
+                                                  signature: headers['x-datapipeline-signature'])
+  end
+
+  def test_webhook_retries_429_and_5xx_but_not_4xx
+    answers = [[503, 'busy'], [429, 'slow', { 'Retry-After' => '0' }], [200, 'ok']]
+    @server = TestHttpServer.new { answers.shift }
+    DataPipeline::Webhook.new(url: @server.url, retry_wait: 0).deliver(report)
+    assert_equal 3, @server.requests.size
+    assert_equal 1, @server.requests.map { |r| r[:headers]['x-datapipeline-delivery'] }.uniq.size
+
+    @server.stop
+    @server = TestHttpServer.new { [400, 'bad'] }
+    error = assert_raises(DataPipeline::Webhook::Error) { DataPipeline::Webhook.new(url: @server.url, retry_wait: 0).deliver(report) }
+    assert_includes error.message, 'HTTP 400'
+    assert_equal 1, @server.requests.size
+  end
+
+  def test_webhook_network_error_after_retries
+    port = TCPServer.new('127.0.0.1', 0).then { |s| s.addr[1].tap { s.close } }
+    error = assert_raises(DataPipeline::Webhook::Error) do
+      DataPipeline::Webhook.new(url: "http://127.0.0.1:#{port}", retries: 1, retry_wait: 0, timeout: 2).deliver(report)
+    end
+    assert_includes error.message, '127.0.0.1'
+  end
+
+  def test_slack_api_blocks_escape_and_mention
+    @server = TestHttpServer.new { [200, { 'ok' => true }] }
+    slack = DataPipeline::Slack.new(token: 'xoxb-1', channel: '#etl', api_url: @server.url, mention: '<!here>')
+    failed = DataPipeline::Report.new([stats_with_rejected], title: 'A & <B>', error: RuntimeError.new('x < y'))
+    slack.deliver(failed)
+    request = @server.requests.last
+    body = JSON.parse(request[:body])
+    assert_equal ['Bearer xoxb-1', '#etl'], [request[:headers]['authorization'], body['channel']]
+    assert_equal ':red_circle: *A &amp; &lt;B&gt;*: abgebrochen <!here>', body['blocks'][0]['text']['text']
+    assert_includes body['blocks'][1]['text']['text'], 'RuntimeError: x &lt; y'
+
+    slack.deliver(report)
+    refute_includes JSON.parse(@server.requests.last[:body])['blocks'][0]['text']['text'], '<!here>' # mention only on failure
+  end
+
+  def test_slack_error_and_incoming_webhook
+    @server = TestHttpServer.new { |req| req[:path] == '/api' ? [200, { 'ok' => false, 'error' => 'not_in_channel' }] : [200, 'ok'] }
+    error = assert_raises(DataPipeline::Slack::Error) { DataPipeline::Slack.new(token: 't', api_url: "#{@server.url}/api").send_message('x') }
+    assert_includes error.message, 'not_in_channel'
+
+    token = ENV.delete('SLACK_TOKEN')
+    hook = DataPipeline::Slack.new(webhook: "#{@server.url}/hook")
+    assert_equal 'ok', hook.deliver(report)
+    request = @server.requests.last
+    assert_nil request[:headers]['authorization']
+    refute JSON.parse(request[:body]).key?('channel')
+  ensure
+    ENV['SLACK_TOKEN'] = token if token
+  end
+
+  def test_email_over_smtp_with_login
+    @server = TestSmtpServer.new
+    mail = DataPipeline::Email.new(to: %w[a@example.com b@example.com], from: 'etl@example.com', host: '127.0.0.1',
+                                   port: @server.port, starttls: false, username: 'user', password: 'geheim')
+    mail.deliver(DataPipeline::Report.new([stats_with_rejected], title: 'Grösse Ä'))
+    sent = @server.mails.last
+    assert_equal ['etl@example.com', %w[a@example.com b@example.com], %w[user geheim]], sent.values_at(:from, :to, :auth)
+    head, body = sent[:data].split("\r\n\r\n", 2)
+    assert_includes head, "Subject: =?UTF-8?B?#{['[etl] Grösse Ä: mit verworfenen Zeilen'].pack('m0')}?="
+    assert_includes head, 'Content-Type: text/plain; charset=UTF-8'
+    text = body.unpack1('m').force_encoding('UTF-8')
+    assert_includes text, '- validate#1: konto: missing'
+    refute_includes text, 'Zeilen:'
+  end
+
+  def test_email_rows_on_request_and_errors
+    @server = TestSmtpServer.new
+    mail = DataPipeline::Email.new(to: 'a@example.com', host: '127.0.0.1', port: @server.port, starttls: false)
+    mail.deliver(report, include_rows: true)
+    assert_includes @server.mails.last[:data].split("\r\n\r\n", 2).last.unpack1('m').force_encoding('UTF-8'), '{"id":"2","konto":""}'
+    assert_nil @server.mails.last[:auth]
+
+    assert_raises(ArgumentError) { DataPipeline::Email.new(to: []) }
+    port = TCPServer.new('127.0.0.1', 0).then { |s| s.addr[1].tap { s.close } }
+    assert_raises(DataPipeline::Email::Error) { DataPipeline::Email.new(to: 'a@b', host: '127.0.0.1', port:, starttls: false).send_mail('x', 'y') }
+  end
+
+  def test_notify_conditions_dry_run_and_rows
+    always = MemoryNotifier.new
+    failure = MemoryNotifier.new
+    problems = MemoryNotifier.new
+    rows = MemoryNotifier.new
+    run_pipeline(ROWS.map(&:dup)) do
+      name 'export'
+      on_error :skip
+      validate 'konto' => { required: true }
+      notify always, on: :always
+      notify failure, on: :failure
+      notify problems
+      notify rows, include_rows: true, title: 'Export Konten'
+    end
+    assert_equal [1, 0, 1], [always, failure, problems].map { |n| n.reports.size }
+    assert_equal ['Export Konten', true], [rows.reports.first[0].title, rows.reports.first[1]]
+
+    dry = MemoryNotifier.new
+    dry_too = MemoryNotifier.new
+    capture_stdout { run_pipeline(ROWS.map(&:dup), dry_run: true) { notify dry, on: :always; notify dry_too, on: :always, dry_run: true } }
+    assert_equal [0, 1], [dry.reports.size, dry_too.reports.size]
+
+    assert_raises(ArgumentError) { run_pipeline { notify Object.new } }
+    assert_raises(ArgumentError) { run_pipeline { notify MemoryNotifier.new, on: :sometimes } }
+  end
+
+  def test_notify_error_is_only_logged
+    log = StringIO.new
+    stats = DataPipeline.new do
+      logger Logger.new(log)
+      read MemoryStorage.new(rows: ROWS.map(&:dup))
+      write MemoryStorage.new
+      notify MemoryNotifier.new(fail: true), on: :always
+    end.run
+    assert stats.ok?
+    assert_includes log.string, 'notify NotificationTest::MemoryNotifier: RuntimeError: kaputt'
+  end
+
+  def test_notifier_from_connections_file
+    with_tmpdir do |dir|
+      File.write("#{dir}/c.yml", <<~YAML)
+        alerts: { type: webhook, url: "http://127.0.0.1:9/hook", secret: "<%= 'abc' %>" }
+        mail: { type: email, to: [etl@example.com], host: smtp.example.com }
+        db: { type: sqlite, database: x.sqlite3 }
+      YAML
+      DataPipeline.connections_file = "#{dir}/c.yml"
+      pipeline = DataPipeline.new { logger PipelineHelper::QUIET }
+      assert_kind_of DataPipeline::Webhook, pipeline.notifier(:alerts)
+      assert_equal 'abc', pipeline.notifier(:alerts).instance_variable_get(:@secret)
+      assert_kind_of DataPipeline::Email, pipeline.notifier(:mail)
+      assert_includes assert_raises(ArgumentError) { pipeline.notifier(:db) }.message, 'is no notifier'
+    end
+  end
+end
@@ -103,4 +103,60 @@ class SqliteStorageTest < PipelineTest
      writer.close
    end
  end
+
+  def count(file) = DataPipeline::SqliteStorage.new(database: file).then { |db| db.query(query: 'select count(*) as n from b').first['n'].tap { db.close } }
+
+  def test_write_row_in_batches_and_query_sees_pending_rows
+    with_tmpdir do |dir|
+      file = "#{dir}/b.sqlite3"
+      writer = DataPipeline::SqliteStorage.new(database: file, table: 'b', create: true, batch_size: 2)
+      3.times { |i| writer.write_row({ 'id' => i }) }
+      assert_equal 2, count(file) # third row still pending
+      assert_equal 3, writer.query(query: 'select count(*) as n from b').first['n']
+      writer.write_row({ 'id' => 3 })
+      writer.finalize
+      assert_equal 4, count(file)
+    end
+  end
+
+  def test_close_writes_pending_rows
+    with_tmpdir do |dir|
+      writer = DataPipeline::SqliteStorage.new(database: "#{dir}/b.sqlite3", table: 'b', create: true)
+      writer.write_row({ 'id' => 1 })
+      writer.close
+      assert_equal 1, count("#{dir}/b.sqlite3")
+    end
+  end
+
+  def test_batch_size_all_is_one_transaction_for_the_stream
+    with_tmpdir do |dir|
+      file = "#{dir}/b.sqlite3"
+      DataPipeline::SqliteStorage.new(database: file, table: 'b', create: true).tap { |w| w.write([{ 'id' => 99 }]); w.close }
+      aborted = DataPipeline::SqliteStorage.new(database: file, table: 'b', truncate: true, batch_size: :all)
+      2.times { |i| aborted.write_row({ 'id' => i }) }
+      aborted.close # without finalize: rolled back, truncate too
+      assert_equal 1, count(file)
+
+      done = DataPipeline::SqliteStorage.new(database: file, table: 'b', truncate: true, batch_size: :all)
+      2.times { |i| done.write_row({ 'id' => i }) }
+      done.finalize
+      assert_equal 2, count(file)
+    end
+  end
+
+  def test_streamed_pipeline_aborted_with_batch_size_all_writes_nothing
+    with_tmpdir do |dir|
+      file = "#{dir}/b.sqlite3"
+      assert_raises(RuntimeError) do
+        DataPipeline.new do
+          logger PipelineHelper::QUIET
+          streamed
+          read MemoryStorage.new(rows: Array.new(5) { |i| { 'id' => i } })
+          map { |row| row['id'] == 3 ? raise('kaputt') : row }
+          write DataPipeline::SqliteStorage.new(database: file, table: 'b', create: true, batch_size: :all)
+        end.run
+      end
+      refute DataPipeline::SqliteStorage.new(database: file).query(query: "select name from sqlite_master where name = 'b'").any?
+    end
+  end
 end
@@ -5,7 +5,7 @@ require 'json'
 #
 # minimal http server for DataPipeline::RestStorage tests, one request per connection
 #   server = TestHttpServer.new { |req| [200, { 'items' => [] }] }
-#   req: { method:, path:, query:, headers:, body: }  -> [status, hash/array or string]
+#   req: { method:, path:, query:, headers:, body: }  -> [status, hash/array or string, optional headers]
 #
 class TestHttpServer
  attr_reader :requests
@@ -40,9 +40,10 @@ class TestHttpServer
    request = { method:, path:, query: URI.decode_www_form(query.to_s).to_h, headers:, body: }
    @requests << request

-    status, payload = @handler.call(request)
+    status, payload, extra = @handler.call(request)
    text = payload.is_a?(String) ? payload : JSON.generate(payload)
-    client.write "HTTP/1.1 #{status} X\r\nContent-Type: application/json\r\n" \
+    extra = Hash(extra).map { |key, value| "#{key}: #{value}\r\n" }.join
+    client.write "HTTP/1.1 #{status} X\r\nContent-Type: application/json\r\n#{extra}" \
                 "Content-Length: #{text.bytesize}\r\nConnection: close\r\n\r\n#{text}"
  rescue StandardError => e
    warn "TestHttpServer: #{e.class}: #{e.message}"
@@ -0,0 +1,59 @@
+require 'socket'
+
+#
+# minimal smtp server for DataPipeline::Email tests: EHLO, AUTH LOGIN, MAIL, RCPT, DATA, QUIT
+#   server = TestSmtpServer.new
+#   server.mails -> [{ from:, to: [...], data:, auth: [user, password] }]
+#
+class TestSmtpServer
+  attr_reader :mails
+
+  def initialize
+    @mails = []
+    @server = TCPServer.new('127.0.0.1', 0)
+    @thread = Thread.new { loop { serve(@server.accept) } }
+  end
+
+  def port = @server.addr[1]
+
+  def stop
+    @thread.kill
+    @server.close
+  end
+
+  private
+
+  def serve client
+    mail = { to: [] }
+    client.write "220 test ESMTP\r\n"
+    while (line = client.gets)
+      case line
+      when /\AEHLO/i then client.write "250-test\r\n250 AUTH LOGIN PLAIN\r\n"
+      when /\AAUTH LOGIN/i
+        client.write "334 VXNlcm5hbWU6\r\n"
+        user = client.gets.unpack1('m')
+        client.write "334 UGFzc3dvcmQ6\r\n"
+        mail[:auth] = [user, client.gets.unpack1('m')]
+        client.write "235 ok\r\n"
+      when /\AMAIL FROM:<(.*)>/i then mail[:from] = $1; client.write "250 ok\r\n"
+      when /\ARCPT TO:<(.*)>/i then mail[:to] << $1; client.write "250 ok\r\n"
+      when /\ADATA/i
+        client.write "354 go\r\n"
+        data = +''
+        while (part = client.gets) && part != ".\r\n"
+          data << part
+        end
+        mail[:data] = data
+        @mails << mail
+        mail = { to: [] }
+        client.write "250 queued\r\n"
+      when /\AQUIT/i then client.write "221 bye\r\n"; break
+      else client.write "250 ok\r\n"
+      end
+    end
+  rescue StandardError => e
+    warn "TestSmtpServer: #{e.class}: #{e.message}"
+  ensure
+    client.close
+  end
+end
Author	SHA1	Message	Date
wkh	3d8323a701	datapipeline 3.2.0: Benachrichtigungen mit Webhook, Slack und E-Mail	2026-10-09 22:27:00 +02:00
wkh	ee47baa22c	datapipeline 3.1.0: SQLite schreibt im Streaming in Paketen, Lizenz BSD-3-Clause	2026-10-09 22:11:24 +02:00