MariaDB Inventory Replenishment with DuckDB and Slack

id: mariadb-inventory-replenishment
namespace: company.team
description: |
  Monitor MariaDB for low-stock inventory, compute reorder costs in DuckDB, enforce budget approval gates, and notify Slack.

inputs:
  - id: stock_threshold
    type: INT
    displayName: Stock Threshold Quantity
    description: Items with stock quantity at or below this threshold are flagged
      for replenishment.
    defaults: 50

  - id: budget_limit
    type: FLOAT
    displayName: Replenishment Budget Limit
    description: Maximum total replenishment cost allowed for automatic purchase
      order creation.
    defaults: 2000.00

  - id: warehouse_filter
    type: STRING
    displayName: Warehouse Location Filter
    description: Optional warehouse location code filter. Use '%' to include all warehouses.
    defaults: "%"

triggers:
  - id: low_stock_trigger
    type: io.kestra.plugin.jdbc.mariadb.Trigger
    description: Poll MariaDB every 5 minutes for items requiring stock replenishment.
    url: "{{ secret('MARIADB_URL') }}"
    username: "{{ secret('MARIADB_USERNAME') }}"
    password: "{{ secret('MARIADB_PASSWORD') }}"
    interval: PT5M
    sql: |
      SELECT item_id, sku, item_name, warehouse_location, quantity_on_hand, reorder_quantity, unit_cost
      FROM inventory
      WHERE quantity_on_hand <= 50 AND reorder_status = 'OK'
      LIMIT 1;
    fetchType: FETCH_ONE
    disabled: true

tasks:
  - id: fetch_low_stock
    type: io.kestra.plugin.jdbc.mariadb.Query
    description: Extract all low-stock inventory records matching input criteria
      from MariaDB.
    url: "{{ secret('MARIADB_URL') }}"
    username: "{{ secret('MARIADB_USERNAME') }}"
    password: "{{ secret('MARIADB_PASSWORD') }}"
    sql: |
      SELECT
        item_id,
        sku,
        item_name,
        warehouse_location,
        quantity_on_hand,
        reorder_threshold,
        reorder_quantity,
        unit_cost
      FROM inventory
      WHERE quantity_on_hand <= :stock_threshold
        AND warehouse_location LIKE :warehouse_filter
        AND reorder_status = 'OK'
      ORDER BY quantity_on_hand ASC;
    parameters:
      stock_threshold: "{{ inputs.stock_threshold }}"
      warehouse_filter: "{{ inputs.warehouse_filter }}"
    fetchType: STORE

  - id: check_has_low_stock
    type: io.kestra.plugin.core.flow.If
    description: Verify if any low-stock inventory records were extracted before proceeding.
    condition: "{{ outputs.fetch_low_stock.size > 0 }}"
    then:
      - id: convert_to_csv
        type: io.kestra.plugin.serdes.csv.IonToCsv
        description: Convert stored MariaDB Ion query results into a CSV file.
        from: "{{ outputs.fetch_low_stock.uri }}"
        header: true

      - id: analyze_replenishment
        type: io.kestra.plugin.jdbc.duckdb.Query
        description: Run DuckDB SQL aggregations to calculate total items, total reorder
          volume, aggregate cost, and ordered SKU list.
        url: "jdbc:duckdb:"
        inputFiles:
          inventory.csv: "{{ outputs.convert_to_csv.uri }}"
        sql: |
          SELECT
            count(*) AS total_items,
            sum(reorder_quantity) AS total_reorder_qty,
            round(sum(reorder_quantity * unit_cost), 2) AS total_cost,
            string_agg(sku, ',' ORDER BY sku) AS sku_list
          FROM read_csv_auto('{{ workingDir }}/inventory.csv');
        fetchType: FETCH

      - id: check_budget_and_approve
        type: io.kestra.plugin.core.flow.If
        description: Check if total replenishment cost is within budget limit.
        condition: "{{ outputs.analyze_replenishment.rows[0].total_cost != null and
          outputs.analyze_replenishment.rows[0].total_cost <=
          inputs.budget_limit }}"
        then:
          - id: update_reorder_status
            type: io.kestra.plugin.jdbc.mariadb.Queries
            description: Update status of replenished SKUs in MariaDB to REORDER_PENDING.
            url: "{{ secret('MARIADB_URL') }}"
            username: "{{ secret('MARIADB_USERNAME') }}"
            password: "{{ secret('MARIADB_PASSWORD') }}"
            sql: |
              UPDATE inventory
              SET reorder_status = 'REORDER_PENDING'
              WHERE FIND_IN_SET(sku, '{{ outputs.analyze_replenishment.rows[0].sku_list }}') > 0;

          - id: send_po_slack_digest
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Post automatic purchase order approval digest to Slack.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "✅ *MariaDB Replenishment Purchase Order Approved*\nBudget Limit: *${{ inputs.budget_limit }}*\nTotal Replenishment Cost: *${{ outputs.analyze_replenishment.rows[0].total_cost }}*\nItems Reordered (*{{ outputs.analyze_replenishment.rows[0].total_items }}*): `{{ outputs.analyze_replenishment.rows[0].sku_list }}`\nStatus set to `REORDER_PENDING` in MariaDB."
              }
        else:
          - id: mark_approval_required
            type: io.kestra.plugin.jdbc.mariadb.Queries
            description: Update status of over-budget SKUs in MariaDB to APPROVAL_REQUIRED
              to prevent trigger refiring loops.
            url: "{{ secret('MARIADB_URL') }}"
            username: "{{ secret('MARIADB_USERNAME') }}"
            password: "{{ secret('MARIADB_PASSWORD') }}"
            sql: |
              UPDATE inventory
              SET reorder_status = 'APPROVAL_REQUIRED'
              WHERE FIND_IN_SET(sku, '{{ outputs.analyze_replenishment.rows[0].sku_list }}') > 0;

          - id: send_manual_approval_slack
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Notify Slack that replenishment cost exceeds budget limit and
              requires manual approval.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "⚠️ *MariaDB Replenishment Requires Manual Approval*\nTotal Replenishment Cost (*${{ outputs.analyze_replenishment.rows[0].total_cost }}*) exceeds budget limit (*${{ inputs.budget_limit }}*).\nItems flagged: `{{ outputs.analyze_replenishment.rows[0].sku_list }}`.\nNo purchase order was created. Status set to `APPROVAL_REQUIRED` in MariaDB."
              }
    else:
      - id: log_no_low_stock
        type: io.kestra.plugin.core.log.Log
        description: Log notice when no inventory items require replenishment.
        message: "No inventory items require replenishment for stock threshold {{
          inputs.stock_threshold }} and warehouse filter '{{
          inputs.warehouse_filter }}'."

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Dispatch error notification to Slack if the inventory replenishment
      workflow fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "🚨 *Flow {{ flow.namespace }}.{{ flow.id }} Failed*\nExecution ID: `{{ execution.id }}`\nError: {{ error.message }}"
      }

Automate inventory low-stock monitoring and budget-gated replenishment workflows for MariaDB. This flow polls MariaDB for items below threshold, converts the result set to CSV, calculates total reorder costs and SKU lists using DuckDB in-memory SQL, and evaluates aggregate cost against a configurable budget limit. If within budget, it updates MariaDB item status to REORDER_PENDING and posts a purchase order digest to Slack; otherwise, it alerts the team for manual approval.

What you get

Automated polling trigger (io.kestra.plugin.jdbc.mariadb.Trigger) for low-stock inventory events.
Zero-footprint in-memory SQL aggregation using DuckDB (io.kestra.plugin.jdbc.duckdb.Query).
Budget-gated conditional branching (io.kestra.plugin.core.flow.If).
Automated status updates (UPDATE inventory SET reorder_status = 'REORDER_PENDING') upon budget approval.
Targeted Slack notifications for both approved purchase orders and budget-exceeded manual approval requests.

Prerequisites

A MariaDB database reachable from Kestra.
A Slack incoming webhook URL for notifications.
Docker networking note: When Kestra runs inside Docker and MariaDB runs on the host machine, use host.docker.internal in MARIADB_URL (e.g., jdbc:mariadb://host.docker.internal:3306/warehouse). On Linux Docker Engine, host.docker.internal may need extra_hosts: "host.docker.internal:host-gateway".

Quick Docker MariaDB setup:

docker run --name mariadb-inventory -e MARIADB_ROOT_PASSWORD=<choose-a-password> -e MARIADB_DATABASE=warehouse -p 3306:3306 -d mariadb:latest

Secrets

MARIADB_URL: JDBC URL for MariaDB (e.g. jdbc:mariadb://db.example.com:3306/warehouse).
MARIADB_USERNAME: Database username.
MARIADB_PASSWORD: Database password.
SLACK_WEBHOOK_URL: Slack Incoming Webhook URL.

Quick start

Run the seed SQL script in your MariaDB database.
Configure the required secrets in Kestra.
Run the flow manually or enable the MariaDB trigger.

Expected output

fetch_low_stock extracts matching items as Ion storage URI {{ outputs.fetch_low_stock.uri }}.
check_has_low_stock evaluates {{ outputs.fetch_low_stock.size > 0 }}. If no items match, it logs a notice and skips downstream processing.
With the seed data, fetch_low_stock matches 4 low-stock items (SKU-1001, SKU-1002, SKU-1004, SKU-1005) for a total reorder quantity of 117 units and aggregate cost of $1,705.00 (sku_list = SKU-1001,SKU-1002,SKU-1004,SKU-1005).
convert_to_csv converts the Ion result to CSV format.
analyze_replenishment outputs total_items: 4, total_reorder_qty: 117, total_cost: 1705.00, and sku_list: "SKU-1001,SKU-1002,SKU-1004,SKU-1005".
If budget_limit is at default ($2,000.00), total_cost (1705.00) <= 2000.00 is true: update_reorder_status sets reorder_status = 'REORDER_PENDING' for the 4 SKUs and posts a PO digest to Slack.
If budget_limit is set to $1,000.00, total_cost (1705.00) <= 1000.00 is false: mark_approval_required sets reorder_status = 'APPROVAL_REQUIRED' in MariaDB and send_manual_approval_slack posts a manual approval alert to Slack stating that no purchase order was created. A manager resolves this by setting item status to 'REORDER_PENDING' to approve reordering or 'OK' to re-evaluate.
Trigger note: The trigger SQL uses fixed literal values (quantity_on_hand <= 50 AND reorder_status = 'OK') because trigger queries in Kestra cannot reference flow inputs.

Re-running / Testing Note

To reset the inventory table state back to OK for re-testing or subsequent manual runs:

UPDATE inventory SET reorder_status = 'OK';

Seed SQL

CREATE TABLE IF NOT EXISTS inventory (
    item_id INT AUTO_INCREMENT PRIMARY KEY,
    sku VARCHAR(50) NOT NULL UNIQUE,
    item_name VARCHAR(100) NOT NULL,
    warehouse_location VARCHAR(50) NOT NULL,
    quantity_on_hand INT NOT NULL,
    reorder_threshold INT NOT NULL,
    reorder_quantity INT NOT NULL,
    unit_cost DECIMAL(10,2) NOT NULL,
    reorder_status VARCHAR(50) DEFAULT 'OK',
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

INSERT INTO inventory (sku, item_name, warehouse_location, quantity_on_hand, reorder_threshold, reorder_quantity, unit_cost, reorder_status) VALUES
    ('SKU-1001', 'Wireless Scanner', 'WH-WEST', 12, 50, 10, 45.50, 'OK'),
    ('SKU-1002', 'Label Printer', 'WH-WEST', 8, 30, 5, 120.00, 'OK'),
    ('SKU-1003', 'Packing Tape Reel', 'WH-EAST', 150, 50, 50, 3.25, 'OK'),
    ('SKU-1004', 'Cardboard Box Pack', 'WH-EAST', 25, 50, 100, 1.50, 'OK'),
    ('SKU-1005', 'Pallet Jack', 'WH-SOUTH', 3, 10, 2, 250.00, 'OK');

MariaDB Inventory Replenishment & Budget Approval Pipeline with DuckDB SQL

What you get

Prerequisites

Secrets

Quick start

Expected output

Re-running / Testing Note

Seed SQL

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

MariaDB Inventory Replenishment & Budget Approval Pipeline with DuckDB SQL

What you get

Prerequisites

Secrets

Quick start

Expected output

Re-running / Testing Note

Seed SQL

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Invoice Math Discrepancy Gate with In-Memory DuckDB Reconciliation

AI Data Quality Anomaly Advisor and SQL Remediation Generator

AI Synthetic Relational Data Generator and Referential Integrity Gate

Automated PostgreSQL Backup to AWS S3 with Restore Verification

Generate and Execute SQL Queries from Natural Language Using OpenAI GPT-4o

Build a point-in-time correct training set and block feature leakage

More Business Blueprints

Autonomous Deep Research and Competitor Market Intelligence Agent

Convert Natural Language Prompts to Structured Todoist Tasks with Deepseek AI

AI Contract Clause Risk Analyzer and Legal Sign-Off Gate

Automate Customer Feedback Triage, Sentiment Analysis, and Escalation with Google Gemini and Slack

Parallel Multilingual Markdown Translation with OpenAI

AI Customer Call Compliance and Regulatory Disclosure Auditor

New to Kestra?