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
datapipeline 3.0.0: transform, Parallelmodus, chained und String-Bedingungen entfernt
2026-10-09 22:27:00 +02:00 · 2026-10-09 22:11:24 +02:00 · 2026-10-09 22:04:12 +02:00
29 changed files with 1253 additions and 467 deletions
@@ -2,6 +2,34 @@

 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
+  String (`eval`) entfernt; alle Schritte laufen verkettet. Umstellung: `map`, `header`, `trailer`,
+  `condition: ->(row) { … }` (README 7.1)
+- `condition:` muss ein Lambda sein, sonst `ArgumentError` beim Aufbau der Pipeline
+- Hinweise `WARN deprecated` und `test/core/compat_test.rb` entfernt
+
 ## 2.1.0 – 2026-10-09

 - Sperre gegen parallele Läufe: `lock 'datei'` in der Pipeline, `etl --lock=datei` für das ganze Skript; ein zweiter
@@ -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 (2.1.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` 2.1.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. [Veraltetes und Umstieg](#7-veraltetes-und-umstieg) – von 2.0 auf 2.1, `transform` und Parallelmodus
+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 |

@@ -783,7 +783,6 @@ Tippfehler fallen also sofort auf.
 | --- | --- |
 | `streamed(flag = true)` | Zeile für Zeile statt Batch (3.3) |
 | `verbose(flag = true)` | Quelle, Ergebnis und Ziel im Batch als JSON auf `$stdout` |
-| `chained(flag = true)` | nur für `transform` nötig (7.2) |
 | `configure(opts)` | mehrere Einstellungen als Hash, z. B. `configure(streamed: true, version: '1.0')` |
 | `run` | führt die Pipeline aus und liefert `DataPipeline::Stats` |
 | `DataPipeline.new(**settings) { … }` | Einstellungen nur für diese Pipeline, z. B. `new(limit: 10)` |
@@ -827,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.
@@ -918,8 +917,8 @@ Dry-Run wird die Fehlerdatei nicht geschrieben.
 | `to_s` | `read=9 written=4 rejected=4 0.097s`; mit Warnungen `warnings=n`, mit `totals` `spalte=quelle/ergebnis` |
 | `to_h` | alles als Hash, wie in der Laufhistorie |

-Der Logger schreibt `start` und `finish` (Info), jede verworfene Zeile und Warnung (Warnung), veraltete Aufrufe
-einmal je Prozess (Warnung, 7.2) und einen Abbruch (Fehler). `on_finish` läuft nach jedem Lauf, auch nach einem
+Der Logger schreibt `start` und `finish` (Info), jede verworfene Zeile und Warnung (Warnung) und einen
+Abbruch (Fehler). `on_finish` läuft nach jedem Lauf, auch nach einem
 Abbruch; ein Fehler im Hook wird nur geloggt und ändert weder Ergebnis noch Exit-Code.

 ```ruby
@@ -1014,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

@@ -1131,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
@@ -1293,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` |

@@ -1417,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

@@ -1507,53 +1612,70 @@ gelesen 1250 · geschrieben 1243 · verworfen 7 · Warnungen 0
 | `LayoutError: TEXT: "…" longer than 30` | Wert passt nicht in das Feld einer Festbreiten-Datei | vorher `validate LAYOUT` mit `on_error`, oder den Wert kürzen |
 | `Encoding::UndefinedConversionError: U+20AC from UTF-8 to ISO-8859-1` | Zeichen gibt es im Zielzeichensatz nicht | `Windows-1252` statt ISO-8859-1, oder `replace: '?'` |
 | `connection :x not in config/connections.yml (a, b)` | Name falsch oder Datei nicht gefunden | Namen prüfen; Pfad relativ zum Arbeitsverzeichnis, sonst `connections_file` setzen |
-| `streamed: several transformations need chained (or map/filter)` | mehrere `transform` im Streaming | `map` verwenden (7.2) |
-| `WARN deprecated: …` | veralteter Aufruf | umstellen, siehe 7.2 |
+| `NoMethodError: undefined method 'transform'` (oder `chained`) | Skript für 2.x | auf `map`, `header`, `trailer` umstellen (7.1) |
+| `map: condition must be a lambda ->(row) { ... }` | `condition:` als String aus 2.x | `condition: ->(row) { … }` (7.1) |

 Mehr sehen: `--verbose` (Log-Level Debug), `verbose` in der Pipeline (Daten als JSON), `--dry-run --limit=20` für einen
 schnellen Blick auf das Ergebnis.

 ---

-## 7. Veraltetes und Umstieg
+## 7. Umstieg

-### 7.1 Von 2.0 auf 2.1
+### 7.0 Von 3.0 auf 3.1 und 3.2

-2.1 ist abwärtskompatibel; bestehende Skripte laufen ohne Änderung. Anders als bisher:
+3.1 und 3.2 sind abwärtskompatibel. Anders als in 3.0:

-| Thema | 2.0 | 2.1 |
+| 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
+(`eval`). Alle Schritte laufen jetzt verkettet, jeder bekommt die Ausgabe des vorigen. Alles andere ist unverändert;
+ein Skript, das unter 2.1 ohne `WARN deprecated` lief, läuft auch unter 3.0.
+
+| 2.x | 3.0 |
+| --- | --- |
+| `transform { \|row, index\| … }` | `map { \|row, ctx\| … }`, Index als `ctx.index` |
+| `transform(processing: :prepend) { \|_, _\| kopf }` | `header { \|ctx\| kopf }` |
+| `transform(processing: :append) { \|_, n\| schluss }` | `trailer { \|ctx\| schluss }`, Anzahl als `ctx.written` |
+| `transform(condition: "row['a'] == 1") { … }` | `map(condition: ->(row) { row['a'] == 1 }) { … }` |
+| `chained`, `configure(chained: true)` | weglassen |
+
+Worauf beim Umstellen zu achten ist:
+
+- **Parallelmodus:** Ohne `map`, `filter`, `validate` und `chained` bekam unter 2.x jeder `transform` eine eigene
+  Kopie der Quellzeilen, und die Ergebnisse wurden hintereinander angehängt. Zwei `transform` auf 2 Quellzeilen
+  ergaben 4 Zeilen, zwei `map` ergeben 2 Zeilen mit beiden Änderungen. Wer wirklich jede Zeile mehrfach ausgeben will,
+  gibt in einem `map` ein Array zurück: `map { |r| [mit_b(r), mit_c(r)] }`.
+- **Kopfsatz:** `transform(processing: :prepend)` lief vor den Daten, `header` läuft im Batch erst danach (steht aber
+  vorne). Das zählt nur, wenn der Block etwas anderes tut, als eine Zeile zu liefern.
+- **Schlusssatz:** `processing: :append` bekam im Parallelmodus die Anzahl der Quellzeilen; `ctx.written` ist die
+  Anzahl der Datenzeilen nach allen Schritten, `ctx.read` die der Quellzeilen.
+- **Bedingung:** Ein String in `condition:` bricht schon beim Aufbau der Pipeline mit `ArgumentError` ab.
+
+Prüfen lässt sich die Umstellung mit `etl run skript.rb --dry-run` vorher und nachher: Die Zusammenfassung und die
+ersten Zeilen sollten gleich sein.
+
+### 7.2 Von 2.0 auf 2.1
+
+2.1 war abwärtskompatibel. Anders als in 2.0:
+
+| Thema | 2.0 | ab 2.1 |
 | --- | --- | --- |
 | Postgres, `write_row` im Streaming | jede Zeile sofort, ohne Transaktion | je 1000 Zeilen in einer Transaktion; der Rest in `finalize`/`close`, auch nach einem Abbruch |
 | CSV lesen mit UTF-8-BOM | BOM steht im ersten Spaltennamen | BOM wird übersprungen |
 | `Stats#to_s` | – | mit `totals` zusätzlich `spalte=quelle/ergebnis` |
-| `transform`, Parallelmodus, `condition:` als String | ohne Hinweis | `WARN deprecated: …` einmal je Prozess |
 | `etl` ohne `name` | – | `stats.name` = Skriptname |

-### 7.2 `transform` und Parallelmodus
-
-`transform`, der Parallelmodus und `condition:` als String (`eval`) sind veraltet und entfallen in 3.0. Ab 3.0
-laufen alle Schritte verkettet, wie heute schon, sobald `map`, `filter` oder `validate` vorkommt.
-
-| Alt | Neu |
-| --- | --- |
-| `transform { \|row, index\| … }` | `map { \|row, ctx\| … }`, Index als `ctx.index` |
-| `transform(processing: :prepend) { \|_, _\| kopf }` | `header { \|ctx\| kopf }` |
-| `transform(processing: :append) { \|_, n\| schluss }` | `trailer { \|ctx\| schluss }` mit `ctx.written` |
-| `transform(condition: "row['a'] == 1")` | `map(condition: ->(row) { row['a'] == 1 })` |
-| mehrere `transform` ohne `chained` (Parallelmodus) | mehrere `map`; wer wirklich jede Zeile mehrfach ausgeben will, gibt in einem `map` ein Array zurück |
-
-Bis dahin gilt für alte Skripte:
-
- **Parallelmodus** (Default ohne `map`/`filter`/`validate` und ohne `chained`): Jeder `transform` bekommt eine eigene
-  Kopie der Quellzeilen, die Ergebnisse aller Schritte werden hintereinander angehängt. Beispiel mit 2 Quellzeilen und
-  `transform { |r| r.merge('b' => 1) }` + `transform { |r| r.merge('c' => 2) }`: parallel 4 Zeilen (2× mit `b`, dann
-  2× mit `c`), verkettet 2 Zeilen mit `b` und `c`. Im Streaming sind mehrere `transform` nur verkettet erlaubt.
- **`transform(processing: :prepend)`** wird mit `(nil, 0)` vor den Daten aufgerufen, **`:append`** mit `(nil, n)`
-  nach den Daten (parallel n = Quellzeilen, verkettet n = Datenzeilen).
- **`condition:` als String** wird mit `eval` ausgewertet, die Variable `row` ist sichtbar.
-
-`test/core/compat_test.rb` hält die Ausgabe des alten Modells fest, damit es sich bis 3.0 nicht ändert.
-
 ---

 ## 8. Entwicklung
@@ -1569,7 +1691,7 @@ Bis dahin gilt für alte Skripte:
 | [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 |
@@ -1577,14 +1699,14 @@ Bis dahin gilt für alte Skripte:

 ### 8.2 Tests

-`bundle exec rake test` startet die Minitest-Suite in [test/](test/), rund 205 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/core` | DSL, Schritte und Kurzschritte, Batch und Streaming, Fehlerbehandlung, Layouts, Sperre, Verbindungen, Summen, Historie, Wasserzeichen; `compat_test.rb` hält das alte `transform`-Modell fest |
-| `test/cli` | `bin/etl` als Prozess: Parameter, Dry-Run, Sperre, Historie, Exit-Codes |
+| `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, 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
@@ -1604,10 +1726,14 @@ Ein eigenes Plugin lässt sich mit `MemoryStorage` aus `test/test_helper.rb` als
 ### 8.3 Gem und Versionen

 - 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 (3.0 entfernt `transform`).
- Das `Gemfile` bindet das Gem über `gemspec` ein und ergänzt die Gruppen `plugins` (`pg`, `sqlite3`, `roo`,
+  (neueste oben). Neue Funktionen erhöhen die mittlere Stelle, Breaking Changes die erste (wie 3.0, 7.1).
+- 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.

@@ -1621,9 +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. **3.0:** `transform`, Parallelmodus und `condition:` als String entfernen.
-7. **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)
@@ -64,28 +67,15 @@ class DataPipeline
  # raised when a reconcile check fails
  class ReconcileError < StandardError; end

-  EDGE_BEFORE = %i[prepend header].freeze
-  EDGE_AFTER  = %i[append trailer].freeze
-  CHAINING    = %i[map filter validate].freeze
  ON_ERROR    = %i[abort skip collect].freeze
  MAX_ERRORS  = 100

-  # removed in 3.0, logged once per process as a warning
-  DEPRECATIONS = {
-    transform: 'transform is deprecated and will be removed in 3.0: use map/filter, ' \
-               'header/trailer instead of processing: :prepend/:append',
-    parallel: 'parallel mode (several transform steps without chained) is deprecated and will be removed in 3.0: ' \
-              'set chained or use map',
-    condition: 'condition: as string (eval) is deprecated and will be removed in 3.0: use a lambda ->(row) { ... }'
-  }.freeze
-
  # value conversions for cast, empty values become nil
  CASTS = Validation::TYPES.merge(string: ->(v) { v.to_s }).freeze

  DEFAULT_SETTINGS = {
    version: nil,
    streamed: false,
-    chained: false,
    verbose: false,
    dry_run: false,
    limit: nil,
@@ -100,7 +90,6 @@ class DataPipeline
  SETTINGS_LOCK = Mutex.new
  @defaults = DEFAULT_SETTINGS
  @plugins = {}
-  @deprecations_logged = []

  #
  # defaults for every following DataPipeline.new, e.g. from bin/etl
@@ -166,18 +155,6 @@ class DataPipeline
    @runs ||= []
  end

-  # true the first time a deprecation is reported in this process
-  def self.deprecation_once key
-    SETTINGS_LOCK.synchronize do
-      next false if @deprecations_logged.include?(key)
-
-      @deprecations_logged << key
-      true
-    end
-  end
-
-  def self.reset_deprecations! = SETTINGS_LOCK.synchronize { @deprecations_logged = [] }
-
  # see DataPipeline::Connections
  def self.connections_file = Connections.file
  def self.connections_file=(path)
@@ -204,7 +181,7 @@ class DataPipeline
  # DataPipeline.new(verbose: true) { ... } overrides the defaults for this pipeline only
  def initialize(**settings, &block)
    @settings = DataPipeline.settings.merge(settings)
-    @transformations = []
+    @steps = []
    @targets = []
    @on_finish = []
    @logger = Logger.new($stderr, level: @settings[:log_level], progname: 'etl')
@@ -232,11 +209,6 @@ class DataPipeline
    @settings[:streamed] = flag
  end

-  # each body step gets the output of the previous step instead of the source rows
-  def chained flag = true
-    @settings[:chained] = flag
-  end
-
  # run everything, write nothing (target, rejects, watermark); prints the first rows
  def dry_run flag = true
    @settings[:dry_run] = flag
@@ -271,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
@@ -343,29 +335,19 @@ class DataPipeline
    klass.new(**options, &block)
  end

-  #
-  # transform(options) { |row, index| ... }
-  #   processing: :prepend | :normal (default) | :append
-  #   condition:  ruby expression as string (row is visible) or lambda { |row| ... }
-  #
-  def transform(options = {}, &block)
-    kind = options.fetch(:processing, :normal)
-    kind = :normal unless %i[prepend append].include?(kind)
-    add_step kind, options, &block
-  end
-
-  # map { |row, ctx| ... }    -> hash, array of hashes or nil; turns chaining on
+  # map { |row, ctx| ... }    -> hash, array of hashes or nil
+  # condition: ->(row) { ... } on map and filter: the step only for matching rows, the others are dropped
  def map(options = {}, &block)
    add_step :map, options, &block
  end

-  # filter { |row, ctx| ... } -> keeps the row when the block is truthy; turns chaining on
+  # filter { |row, ctx| ... } -> keeps the row when the block is truthy
  def filter(options = {}, &block)
    add_step :filter, options, &block
  end

  #
-  # short steps, each one a map step (turns chaining on)
+  # short steps, each one a map step
  #
  #   rename 'kunnr' => 'kundennr'               -> new key at the same position
  #   select %w[belnr betrag]                    -> only these keys in this order, missing ones nil
@@ -419,7 +401,7 @@ class DataPipeline
  # validate layout                                   -> fields of a DataPipeline::Layout
  # validate { |row| 'soll ungleich haben' if ... }   -> nil, message or array of messages
  #
-  # a failing row raises ValidationError, handled by on_error; turns chaining on
+  # a failing row raises ValidationError, handled by on_error
  # warn: true      -> only a warning in the log and stats.warnings, the row is kept
  # ignore: %w[A B] -> messages for these fields are left out (known gaps)
  #
@@ -427,7 +409,7 @@ class DataPipeline
    warn = options.delete(:warn) || false
    ignore = Array(options.delete(:ignore)).map(&:to_s)
    rules ||= options # validate 'konto' => { ... } arrives as keywords
-    @transformations << { kind: :validate, block:, options: { rules:, warn:, ignore: } }
+    @steps << { kind: :validate, block:, options: { rules:, warn:, ignore: } }
  end

  def read storage, options = {}
@@ -450,7 +432,7 @@ class DataPipeline
  end

  #
-  # loads a table once, for lookups in map/transform instead of one query per row
+  # loads a table once, for lookups in map instead of one query per row
  #
  #   lookup pg, query: 'select ...', params: [...], key: 'rechnung_dbid'  -> { key => [rows] }
  #   lookup pg, query: 'select ...', key: 'id', unique: true              -> { key => row }
@@ -472,10 +454,9 @@ class DataPipeline
    (@lock = Lock.new(@settings[:lock])).acquire if @settings[:lock]
    raise ArgumentError, 'write storage missing' if @targets.empty?

-    before = @transformations.select { |t| EDGE_BEFORE.include?(t[:kind]) }
-    after  = @transformations.select { |t| EDGE_AFTER.include?(t[:kind]) }
-    body   = @transformations - before - after
-    log_deprecations body
+    before = @steps.select { |t| t[:kind] == :header }
+    after  = @steps.select { |t| t[:kind] == :trailer }
+    body   = @steps - before - after

    @settings[:streamed] ? run_streamed(before, body, after) : run_batch(before, body, after)

@@ -493,8 +474,11 @@ class DataPipeline

  def add_step kind, options, &block
    raise ArgumentError, "#{kind}: block missing" unless block
+    if options.key?(:condition) && !options[:condition].respond_to?(:call)
+      raise ArgumentError, "#{kind}: condition must be a lambda ->(row) { ... }"
+    end

-    @transformations << { kind:, block:, options: }
+    @steps << { kind:, block:, options: }
  end

  def value_of row, key
@@ -511,21 +495,6 @@ class DataPipeline
    raise ArgumentError, "#{key}: #{value.inspect} is not #{type}"
  end

-  # each deprecated feature once per process, with the logger of the pipeline
-  def log_deprecations body
-    used = []
-    used << :transform if @transformations.any? { |t| %i[prepend normal append].include?(t[:kind]) }
-    used << :parallel if !chained? && body.size > 1
-    used << :condition if @transformations.any? { |t| t[:options].is_a?(Hash) && t[:options][:condition].is_a?(String) }
-    used.each { |key| @logger.warn "deprecated: #{DEPRECATIONS[key]}" if DataPipeline.deprecation_once(key) }
-  end
-
-  # without any transformation the source is copied unchanged
-  def chained?
-    @settings[:chained] || @transformations.empty? ||
-      @transformations.any? { |t| CHAINING.include?(t[:kind]) }
-  end
-
  def write_row row
    @targets.each { |t| t[:storage].write_row(row) if t[:only].nil? || t[:only].call(row) }
  end
@@ -538,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
@@ -558,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]
@@ -607,33 +591,21 @@ class DataPipeline
    data.each { |row| add_total(:read, row) } if @totals
    dump 'Source#data', data

-    # prepend keeps its old timing: called before the body
-    results = {}.compare_by_identity
-    before.select { |t| t[:kind] == :prepend }.each { |t| results[t] = edge(t, nil, 0) }
-
    rows = []
-    if chained?
-      counters = Array.new(body.size, 0)
-      data.each { |row| push(row, body, counters, data.length) { |out| rows << out } }
-    else
-      body.each { |step| rows.concat apply(step, data) }
-    end
+    counters = Array.new(body.size, 0)
+    data.each { |row| push(row, body, counters, data.length) { |out| rows << out } }
    dump 'Transformation#data', rows
    @stats.written = rows.length
    rows.each { |row| add_total(:written, row) } if @totals
    check_reconcile

+    # header is computed after the body, so it knows the counters
    ctx = Context.new(read: data.length, written: rows.length)
-    before.select { |t| t[:kind] == :header }.each { |t| results[t] = edge(t, ctx, 0) }
-
-    # parallel mode keeps the old count of source rows for append
-    count = chained? ? rows.length : data.length
    new_data = []
-    before.each { |t| add_rows new_data, results[t] }
+    before.each { |t| add_rows new_data, t[:block].call(ctx) }
    new_data.concat rows
-    after.each { |t| add_rows new_data, edge(t, ctx, count) }
+    after.each { |t| add_rows new_data, t[:block].call(ctx) }

-    @stats.written = rows.length
    dump 'Target#data', new_data
    output(new_data) do |all|
      @targets.each { |t| t[:storage].write(t[:only] ? all.select(&t[:only]) : all) }
@@ -642,13 +614,9 @@ class DataPipeline

  # row by row from source.read { } to target.write_row, header only knows nothing yet
  def run_streamed before, body, after
-    if !chained? && body.size > 1
-      raise ArgumentError, 'streamed: several transformations need chained (or map/filter)'
-    end
-
    emit = ->(result) { output(add_rows([], result)) { |rows| rows.each { |row| write_row(row) } } }

-    before.each { |t| emit.call edge(t, Context.new, 0) }
+    before.each { |t| emit.call t[:block].call(Context.new) }

    read = written = 0
    counters = Array.new(body.size, 0)
@@ -673,7 +641,7 @@ class DataPipeline
    check_reconcile

    ctx = Context.new(read:, written:)
-    after.each { |t| emit.call edge(t, ctx, chained? ? written : read) }
+    after.each { |t| emit.call t[:block].call(ctx) }
    @targets.each { |t| t[:storage].finalize }
  end

@@ -688,11 +656,6 @@ class DataPipeline
    end
  end

-  # parallel mode: every step reads its own copy of the source rows
-  def apply step, rows
-    rows.each_with_index.flat_map { |row, index| safe_step_rows(step, row.dup, index, rows.length) }
-  end
-
  # errors of a step for one row go to on_error
  def safe_step_rows step, row, index, read
    step_rows(step, row, index, read)
@@ -715,7 +678,7 @@ class DataPipeline

  def step_name step
    name = step[:options].is_a?(Hash) && step[:options][:name] || step[:kind]
-    "#{name}##{@transformations.index(step) + 1}"
+    "#{name}##{@steps.index(step) + 1}"
  end

  # rows produced by one step for one input row
@@ -743,26 +706,13 @@ class DataPipeline
      step[:block].call(row, Context.new(read:, index:)) ? [row] : []
    when :map
      add_rows [], step[:block].call(row, Context.new(read:, index:))
-    else
-      add_rows [], step[:block].call(row, index)
    end
  end

-  def edge step, ctx, count
-    case step[:kind]
-    when :prepend then step[:block].call(nil, 0)
-    when :append  then step[:block].call(nil, count)
-    else step[:block].call(ctx)
-    end
-  end
-
-  # string conditions are evaluated with the local variable row
+  # condition: ->(row) { ... } on map and filter
  def condition? step, row
-    condition = step[:options].fetch(:condition, true)
-    return true if condition == true
-    return condition.call(row) if condition.respond_to?(:call)
-
-    eval(condition)
+    condition = step[:options][:condition]
+    condition.nil? || condition.call(row)
  end

  # a hash is one row, an array are several rows, nil is no row
@@ -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 = '2.1.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
@@ -14,8 +15,8 @@ class EtlCliTest < PipelineTest
    "#{dir}/#{name}"
  end

-  # like import.*.rb: own OptionParser for --from/--till, transform api
-  def test_legacy_script_gets_from_till_and_limit
+  # scripts with an own OptionParser for --from/--till
+  def test_script_with_own_option_parser_gets_from_till_and_limit
    with_tmpdir do |dir|
      path = script(dir, 'legacy.rb', <<~RUBY)
        require 'optparse/date'
@@ -24,8 +25,8 @@ class EtlCliTest < PipelineTest
        File.write('args.txt', "\#{opts[:from]} \#{opts[:till]}")
        DataPipeline.new do
          read container(:csv, file: 'in.csv')
-          transform(processing: :prepend) { |_, _| { 'h' => 'HEAD' } }
-          transform { |r, i| r.merge('i' => i) }
+          header { { 'h' => 'HEAD' } }
+          map { |r, ctx| r.merge('i' => ctx.index) }
          write container(:csv, file: 'out.csv', csv: { headers: false, col_sep: '|' })
        end.run
      RUBY
@@ -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
@@ -4,16 +4,22 @@ require_relative '../test_helper'
 class ChainedTest < PipelineTest
  ROWS = [{ 'a' => 1 }, { 'a' => 2 }, { 'a' => 3 }].freeze

-  def test_chained_transform_gets_previous_output
-    out = output(ROWS) { chained; transform { |r| r.merge('b' => 1) }; transform { |r| r.merge('c' => 2) } }
+  def test_each_step_gets_the_previous_output
+    out = output(ROWS) { map { |r| r.merge('b' => 1) }; map { |r| r.merge('c' => 2) } }
    assert_equal ROWS.map { |r| r.merge('b' => 1, 'c' => 2) }, out
  end

  def test_one_to_many_then_index
-    out = output(ROWS) { chained; transform { |r| [r, r] }; transform { |r, i| r.merge('i' => i) } }
+    out = output(ROWS) { map { |r| [r, r] }; map { |r, ctx| r.merge('i' => ctx.index) } }
    assert_equal (0...6).to_a, out.map { |r| r['i'] }
  end

+  def test_nil_and_empty_array_drop_rows
+    assert_equal [{ 'a' => 1 }], output(ROWS) { map { |r| r['a'] == 1 ? r : nil } }
+    assert_equal ROWS, output(ROWS) { map { |r| [r, nil] } }
+    assert_equal [], output(ROWS) { map { [] } }
+  end
+
  def test_map_and_filter
    out = output(ROWS) do
      map { |r| r.merge('d' => r['a'] * 2) }
@@ -23,12 +29,6 @@ class ChainedTest < PipelineTest
    assert_equal [{ 'a' => 2, 'd' => 4, 'i' => 0 }, { 'a' => 3, 'd' => 6, 'i' => 1 }], out
  end

-  def test_map_turns_chaining_on_for_transform
-    out = output(ROWS) { transform { |r| r.merge('b' => 1) }; map { |r| r.merge('c' => 1) } }
-    assert_equal 3, out.size
-    assert(out.all? { |r| r['b'] && r['c'] })
-  end
-
  def test_header_and_trailer_with_context
    out = output(ROWS) do
      header { |c| { 'H' => c.read, 'W' => c.written } }
@@ -41,9 +41,14 @@ class ChainedTest < PipelineTest
    assert_equal 6, out.size
  end

-  def test_edge_order_follows_registration
-    out = output(ROWS) { header { { 'h' => 1 } }; transform(processing: :prepend) { { 'p' => 1 } }; map { |r| r } }
-    assert_equal [{ 'h' => 1 }, { 'p' => 1 }], out.first(2)
+  def test_several_headers_and_trailers_in_registration_order
+    out = output(ROWS) { header { { 'h' => 1 } }; trailer { { 't' => 1 } }; header { { 'h' => 2 } }; trailer { { 't' => 2 } } }
+    assert_equal [{ 'h' => 1 }, { 'h' => 2 }, *ROWS, { 't' => 1 }, { 't' => 2 }], out
+  end
+
+  def test_stable_order_of_many_steps
+    out = output(ROWS) { 20.times { |k| map { |r| r.merge("k#{k}" => k) } } }
+    assert_equal (0...20).to_a, out.first.values.drop(1)
  end

  def test_header_and_trailer_nil_and_arrays
@@ -51,12 +56,13 @@ class ChainedTest < PipelineTest
    assert_equal ROWS + [{ 'x' => 1 }, { 'y' => 2 }], out
  end

-  def test_condition_on_map
+  def test_condition_on_map_and_filter
    assert_equal [{ 'a' => 1 }], output(ROWS) { map(condition: ->(r) { r['a'] == 1 }) { |r| r } }
+    assert_equal [{ 'a' => 3 }], output(ROWS) { filter(condition: ->(r) { r['a'] > 1 }) { |r| r['a'] == 3 } }
  end

-  def test_chained_false
-    assert_equal 6, output(ROWS) { chained false; transform { |r| r }; transform { |r| r } }.size
+  def test_condition_must_be_a_lambda
+    assert_includes assert_raises(ArgumentError) { output(ROWS) { map(condition: "row['a'] == 1") { |r| r } } }.message, 'lambda'
  end

  def test_csv_end_to_end
@@ -1,81 +0,0 @@
-require_relative '../test_helper'
-
-#
-# parallel mode (transform without chained/map/filter) must keep the output of v1.2.1,
-# the version the CUDO importers were built on. expected values were recorded with v1.2.1.
-#
-class CompatTest < PipelineTest
-  ROWS = (1..6).map { |i| { 'id' => i, 'land' => i.even? ? 'DE' : 'AT' } }
-
-  def assert_output expected_json, &block
-    assert_equal expected_json, JSON.generate(output(ROWS, &block))
-  end
-
-  # pattern of import.customer.rb
-  def test_customer_pattern
-    assert_output '[{"SATYP":"FIRC","LFDNR":0},{"id":1,"land":"AT","lfdnr":"0000000001"},' \
-                  '{"id":2,"land":"DE","lfdnr":"0000000002"},{"id":3,"land":"AT","lfdnr":"0000000003"},' \
-                  '{"id":4,"land":"DE","lfdnr":"0000000004"},{"id":5,"land":"AT","lfdnr":"0000000005"},' \
-                  '{"id":6,"land":"DE","lfdnr":"0000000006"},{"topic":"LARC","counter":"0000000006"}]' do
-      transform(processing: :prepend) { |_row, n| { 'SATYP': 'FIRC', 'LFDNR': n } }
-      transform(processing: :append) { |_row, n| { 'topic': 'LARC', 'counter': format('%010d', n) } }
-      transform { |row, _i| row.merge('lfdnr' => format('%010d', row['id'])) }
-    end
-  end
-
-  # pattern of import.posting.rb: closure counter, 1:n
-  def test_posting_pattern
-    expected = [{ SATYP: 'FIRC', LFDNR: 0 }]
-    lfdnr = 0
-    ROWS.each do |row|
-      expected << row.merge('lfdnr' => lfdnr += 1)
-      row['id'].times { |k| expected << { 'pos' => k, 'lfdnr' => lfdnr += 1 } }
-    end
-    expected << { SATYPE: 'LARC', SUMME: '0000000027' }
-
-    assert_output JSON.generate(expected) do
-      cnt = 1
-      transform(processing: :prepend) { |_row, _n| { 'SATYP': 'FIRC', 'LFDNR': 0 } }
-      transform do |row|
-        out = [row.merge('lfdnr' => cnt)]
-        cnt += 1
-        row['id'].times do |k|
-          out << { 'pos' => k, 'lfdnr' => cnt }
-          cnt += 1
-        end
-        out
-      end
-      transform(processing: :append) { |_row| { 'SATYPE': 'LARC', 'SUMME': format('%010d', cnt - 1) } }
-    end
-  end
-
-  def test_append_registered_before_prepend
-    assert_output '[{"h":0},{"id":1,"land":"AT"},{"id":2,"land":"DE"},{"id":3,"land":"AT"},' \
-                  '{"id":4,"land":"DE"},{"id":5,"land":"AT"},{"id":6,"land":"DE"},{"t":6}]' do
-      transform(processing: :append) { |_, n| { 't' => n } }
-      transform { |r| r }
-      transform(processing: :prepend) { |_, n| { 'h' => n } }
-    end
-  end
-
-  def test_condition_string
-    assert_output '[{"id":2,"land":"DE","i":1},{"id":4,"land":"DE","i":3},{"id":6,"land":"DE","i":5}]' do
-      transform(condition: "row['land'] == 'DE'") { |r, i| r.merge('i' => i) }
-    end
-  end
-
-  def test_two_parallel_transforms
-    expected = ROWS.map { |r| r.merge('a' => 1) } + ROWS.map { |r| r.merge('b' => 1) }
-    assert_output JSON.generate(expected) do
-      transform { |r| r.merge('a' => 1) }
-      transform { |r| [r.merge('b' => 1), nil] }
-    end
-  end
-
-  def test_nil_and_empty_array
-    assert_output '[{"id":1,"land":"AT"}]' do
-      transform { |r| r['id'] > 3 ? nil : [] }
-      transform { |r| r['id'] == 1 ? r : nil }
-    end
-  end
-end
@@ -1,4 +1,5 @@
 require_relative '../test_helper'
+require 'time'

 # connections file, totals with reconcile, run history
 class ConnectionsTotalsHistoryTest < PipelineTest
@@ -41,8 +41,8 @@ class OperationsTest < PipelineTest
    assert_raises(RuntimeError) { run_pipeline(ROWS) { on_error :skip; header { raise 'h' }; map { |r| r } } }
  end

-  def test_skip_in_parallel_mode
-    target, stats = run_pipeline(ROWS) { on_error :skip; transform { |r| r['id'] == '1' ? raise('x') : r } }
+  def test_skip_in_map
+    target, stats = run_pipeline(ROWS) { on_error :skip; map { |r| r['id'] == '1' ? raise('x') : r } }
    assert_equal 4, target.output.size
    assert_equal 1, stats.rejected
  end
@@ -67,8 +67,8 @@ class OperationsTest < PipelineTest
    assert_empty DataPipeline::Validation.errors({ 'a' => '' }, 'a' => { max: 1, type: :integer })
  end

-  def test_validate_turns_chaining_on
-    target, = run_pipeline(ROWS) { on_error :skip; transform { |r| r.merge('x' => 1) }; validate RULES }
+  def test_validate_gets_the_output_of_the_previous_step
+    target, = run_pipeline(ROWS) { on_error :skip; map { |r| r.merge('x' => 1) }; validate RULES }
    assert_equal 3, target.output.size
    assert(target.output.all? { |r| r['x'] })
  end
@@ -1,17 +1,11 @@
 require_relative '../test_helper'

-# dsl, settings, plugin lookup, parallel mode
+# dsl, settings, plugin lookup
 class PipelineCoreTest < PipelineTest
  ROWS = [{ 'a' => 1 }, { 'a' => 2 }, { 'a' => 3 }].freeze

  def settings_of(pipeline) = pipeline.instance_variable_get(:@settings)

-  def test_nil_filters_rows
-    assert_equal [{ 'a' => 1 }], output(ROWS) { transform { |r| r['a'] == 1 ? r : nil } }
-    assert_equal ROWS, output(ROWS) { transform { |r| [r, nil] } }
-    assert_equal ROWS, output(ROWS) { transform(processing: :prepend) { nil }; transform { |r| r }; transform(processing: :append) { nil } }
-  end
-
  def test_dsl_switches
    assert_equal false, settings_of(DataPipeline.new { verbose false })[:verbose]
    assert_equal true, settings_of(DataPipeline.new { verbose })[:verbose]
@@ -73,39 +67,8 @@ class PipelineCoreTest < PipelineTest
    assert_equal Array.new(8) { |i| (i % 3) + 1 }, reads
  end

-  def test_parallel_mode_is_side_by_side
-    out = output(ROWS) { transform { |r| r.merge('b' => 1) }; transform { |r| r.merge('c' => 2) } }
-    assert_equal 6, out.size
-  end
-
-  def test_parallel_mode_copies_rows_per_step
-    out = output(ROWS) { transform { |r| r['x'] = 1; r }; transform { |r| r.key?('x') ? nil : r } }
-    assert_equal 6, out.size
-    assert_equal [{ 'a' => 10 }, { 'a' => 20 }, { 'a' => 30 }], output(ROWS) { transform { |r| [r.transform_values! { |v| v * 10 }] } }
-  end
-
-  def test_stable_order_and_all_prepends_appends
-    out = output(ROWS) { 20.times { |k| transform { |r| r['a'] == 1 ? { 'k' => k } : nil } } }
-    assert_equal (0...20).to_a, out.map { |r| r['k'] }
-
-    out = output(ROWS) do
-      transform(processing: :prepend) { { 'h' => 1 } }
-      transform(processing: :prepend) { { 'h' => 2 } }
-      transform { |r| r }
-      transform(processing: :append) { { 't' => 1 } }
-      transform(processing: :append) { { 't' => 2 } }
-    end
-    assert_equal [{ 'h' => 1 }, { 'h' => 2 }, *ROWS, { 't' => 1 }, { 't' => 2 }], out
-  end
-
-  def test_append_count_parallel_vs_chained
-    assert_equal({ 'n' => 3 }, output(ROWS) { transform { |r| [r, r] }; transform(processing: :append) { |_, n| { 'n' => n } } }.last)
-    assert_equal({ 'n' => 6 }, output(ROWS) { chained; transform { |r| [r, r] }; transform(processing: :append) { |_, n| { 'n' => n } } }.last)
-  end
-
-  def test_conditions
-    assert_equal [{ 'a' => 1 }, { 'a' => 3 }], output(ROWS) { transform(condition: ->(r) { r['a'] != 2 }) { |r| r } }
-    assert_equal [{ 'a' => 2 }], output(ROWS) { transform(condition: "row['a'] == 2") { |r| r } }
+  def test_removed_dsl_of_version_2
+    %i[transform chained].each { |name| assert_raises(NoMethodError) { DataPipeline.new { send(name) { |r| r } } } }
  end

  def test_copy_without_steps
@@ -1,6 +1,6 @@
 require_relative '../test_helper'

-# rename, select, cast, default and the deprecation warnings
+# rename, select, cast, default
 class StepsTest < PipelineTest
  ROWS = [{ 'kunnr' => '1', 'betrag' => '12.50', 'datum' => '2026-10-09', 'land' => nil },
          { 'kunnr' => '2', 'betrag' => '', 'datum' => ' ', 'land' => 'DE' }].freeze
@@ -44,33 +44,4 @@ class StepsTest < PipelineTest
    end
    assert_equal [{ 'id' => '1', 'betrag' => 12.5 }, { 'id' => '2', 'betrag' => 0.0 }], rows
  end
-
-  def deprecation_log(&block)
-    log = StringIO.new
-    DataPipeline.new do
-      logger Logger.new(log)
-      read MemoryStorage.new(rows: [{ 'a' => 1 }])
-      write MemoryStorage.new
-      instance_eval(&block)
-    end.run
-    log.string.lines.grep(/deprecated/)
-  end
-
-  def test_deprecations_once_per_process
-    lines = deprecation_log do
-      transform(condition: 'true') { |r| r }
-      transform { |r| r }
-    end
-    assert_equal 3, lines.size
-    assert lines.any? { |l| l.include?('transform is deprecated') }
-    assert lines.any? { |l| l.include?('parallel mode') }
-    assert lines.any? { |l| l.include?('condition: as string') }
-    assert_empty deprecation_log { transform { |r| r } }
-  end
-
-  def test_no_deprecation_for_map_filter_and_single_transform_in_chained_mode
-    assert_empty deprecation_log { map { |r| r }; filter(condition: ->(_r) { true }) { true } }
-    lines = deprecation_log { chained; transform { |r| r }; transform { |r| r } }
-    assert_equal 1, lines.size # only transform, no parallel mode
-  end
 end
@@ -21,23 +21,14 @@ class StreamingTest < PipelineTest
    assert_equal({ 'T' => 4, 'R' => 3 }, stream.last)
  end

-  def test_stream_prepend_append
+  def test_stream_several_steps_one_to_many
    out = output(ROWS, streamed: true) do
-      chained
-      transform(processing: :prepend) { |_, n| { 'p' => n } }
-      transform { |r| [r, r] }
-      transform(processing: :append) { |_, n| { 'n' => n } }
+      map { |r| [r, r] }
+      map { |r, ctx| r.merge('i' => ctx.index) }
+      trailer { |ctx| { 'n' => ctx.written } }
    end
-    assert_equal({ 'p' => 0 }, out.first)
    assert_equal({ 'n' => 6 }, out.last)
-    assert_equal 8, out.size
-  end
-
-  def test_stream_parallel_single_step_and_error_for_several
-    out = output(ROWS, streamed: true) { transform { |r| [r, r] }; transform(processing: :append) { |_, n| { 'n' => n } } }
-    assert_equal({ 'n' => 3 }, out.last)
-    error = assert_raises(ArgumentError) { output(ROWS, streamed: true) { transform { |r| r }; transform { |r| r } } }
-    assert_includes error.message, 'chained'
+    assert_equal (0...6).to_a, out.first(6).map { |r| r['i'] }
  end

  def test_stream_condition_and_copy
@@ -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
@@ -55,7 +55,6 @@ module PipelineHelper
    DataPipeline.reset_settings!
    DataPipeline.runs.clear
    DataPipeline.params.clear
-    DataPipeline.reset_deprecations!
  end

  # runs a pipeline from rows into a MemoryStorage, returns [target, stats]
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
wkh	a8b8381293	datapipeline 3.0.0: transform, Parallelmodus, chained und String-Bedingungen entfernt	2026-10-09 22:04:12 +02:00