Notification Gateway

Unified notification gateway with template-based message rendering, platform-specific overrides, Redis-backed rate limiting, time-window aggregation, and mute/DND rules to prevent notification fatigue.

Source: pkg/notify/, pkg/notify/template/, pkg/notify/rules/, pkg/ability/notify/

Overview

Homelab monitoring produces high-frequency events: disk alerts, download completions, RSS updates, agent status changes. Passively forwarding every event to Slack, Pushover, or ntfy causes notification fatigue -- users mute channels and miss critical alerts.

The Notification Gateway inserts a processing layer between event producers and notification providers to:

1. Separate data from presentation -- templates define message formatting, callers provide structured payloads
2. Rate-limit repetitive alerts -- prevent a buggy script from sending 1000 messages in one minute
3. Aggregate batched events -- collapse 20 RSS fetch events into a single digest every 15 minutes
4. Honor DND windows -- silence all notifications during night hours

[Pipeline / Cron / Webhook / Agent]
         │
         ▼
┌─────────────────────────────────────────────┐
│            Notification Gateway             │
│                                             │
│  ┌───────────────────────────────────────┐  │
│  │  Rule Engine (pkg/notify/rules/)      │  │
│  │  ┌──────────┐ ┌──────────┐ ┌────────┐ │  │
│  │  │ Mute/DND │ │ Throttle │ │Aggregate│ │  │
│  │  └──────────┘ └──────────┘ └────────┘ │  │
│  └─────────────────┬─────────────────────┘  │
│                    ▼                        │
│  ┌───────────────────────────────────────┐  │
│  │  Template Engine (pkg/notify/template/)│  │
│  │  Sprig functions + per-channel overrides│ │
│  └─────────────────┬─────────────────────┘  │
│                    ▼                        │
│  ┌───────────────────────────────────────┐  │
│  │  Channel Router (pkg/notify/)         │  │
│  │  Existing Notifyer registry           │  │
│  └───────────────────────────────────────┘  │
└─────────────────────────────────────────────┘
         │
         ▼
[Slack Webhook] [ntfy] [Pushover] [Message Pusher]

Architecture

Data Flow

Caller (module, cron, pipeline)
    │  notify.GatewaySend(ctx, uid, templateID, channels, payload)
    ▼
┌──────────────────────────────────────────────────────────┐
│ GatewaySend                                              │
│  1. Resolve template by ID (template.Engine)             │
│  2. For each channel:                                    │
│     a. Evaluate rules (rules.Engine)                     │
│        - Drop → skip                                     │
│        - Mute → skip                                     │
│        - Throttle → check Redis counter, skip if limited │
│        - Aggregate → push to Redis List, set timer       │
│     b. Render template for channel (template.Engine)     │
│     c. Look up user channel config from store            │
│     d. Send via existing notify.Send()                   │
│  3. Background Worker:                                   │
│     - Scans expired aggregate timers every 60s           │
│     - Flushes buffered items, renders digest template    │
│     - Sends single aggregated message                    │
└──────────────────────────────────────────────────────────┘

Pipeline Integration

The gateway can be invoked from pipeline steps via the notify capability:

pipelines:
  - name: bookmark-notify
    enabled: true
    trigger:
      event: bookmark.created
    steps:
      - name: send-notification
        capability: core
        operation: notify_send
        params:
          template_id: "bookmark.created"
          channels:
            - slack
            - ntfy
          payload: "{{ .Event.data }}"

This uses the Pipeline Template Engine to pass event data through to the notification template.

Direct Invocation

Non-pipeline code (cron jobs, webhook handlers, agent actions) calls GatewaySend directly:

import "github.com/flowline-io/flowbot/pkg/notify"

err := notify.GatewaySend(ctx.Context(), ctx.AsUser, "server.offline", []string{"slack", "ntfy"}, map[string]any{
    "hostname": item.Hostname,
    "hostid":   item.Hostid,
})

Template Engine

The template engine renders notification messages using Go text/template with the Sprig function library. Sprig provides 70+ template functions for string manipulation, date formatting, math, and type conversion.

Template Schema

Templates are managed in the Notifications UI (/service/web/notifications → Templates tab)
and persisted in PostgreSQL (notify_templates). They are loaded at startup and hot-reloaded after CRUD.

Example fields (same shape as the UI form / JSON overrides):

# Illustrative shape only — not loaded from flowbot.yaml
id: bookmark.created
name: "New Bookmark Notification"
description: "Triggered when a bookmark is successfully created"
default_format: markdown
default_template: |
  **New Bookmark Saved**
  **URL:** {{ .url | default "N/A" }}
  {{ if .title }}**Title:** {{ .title }}{{ end }}
overrides:
  - channel: telegram
    format: html
    template: |
      <b>New Bookmark Saved</b>
      <b>URL:</b> <a href="{{ .url }}">{{ .url }}</a>

Field Reference

Override Fields

Template Data Context

Template payload data is accessed via {{ .key }} dot-notation. The payload is a map[string]any passed by the caller:

{{ .title }}                    -- string field
{{ .url | default "N/A" }}     -- with fallback
{{ .tags | join ", " }}        -- join a string slice
{{ .count | default 0 }}       -- integer with default
{{ .name | upper }}            -- uppercase transform
{{ shorten .text 80 }}         -- truncate with "..."
{{ if .urgent }}URGENT: {{ end }}{{ .title }}  -- conditional

Available Sprig Functions

All Sprig string functions, date functions, math functions, and list functions are available. Commonly used:

Custom Functions

Template Rendering Logic

1. Gateway passes templateID and channel to the engine
2. Engine looks up the channel-specific override first
3. If no override exists for the channel, uses the default template
4. Template receives the payload as . and renders via text/template.Execute()
5. Output includes Title (first line, stripped of markdown formatting), Body (full rendered output), and Format

Rule Engine

Rules are evaluated before any notification is sent. They are managed in the Notifications UI
(/service/web/notifications → Rules tab), persisted in PostgreSQL (notify_rules), and
processed in priority order (higher priority first).

Rule Schema

# Illustrative shape only — not loaded from flowbot.yaml
id: "night_mute"
action: mute
match:
  event: "*"
  channel: "*"
condition: "time.hour >= 23 || time.hour < 8"
priority: 100

Rule Fields

Match Fields

Match Examples

Rule Actions

Mute (DND)

Suppresses all matching notifications when the time condition is met. Useful for night-time silence.

- id: "night_mute"
  action: mute
  match:
    event: "*"
    channel: "*"
  condition: "time.hour >= 23 || time.hour < 8"
  priority: 100

Condition syntax: time.hour >= N, time.hour < N, time.hour == N connected with || (OR) and && (AND).

Throttle

Limits how many notifications of a specific type are sent within a time window. Uses Redis INCR with TTL for atomic counting.

- id: "infra_throttle"
  action: throttle
  match:
    event: "infra.*"
    channel: "*"
  priority: 50
  params:
    window: "5m"
    limit: 1

Throttle parameters:

Redis key pattern: notify:throttle:{ruleID}:{eventType}:{channel}

Aggregate

Buffers individual events into a Redis List and flushes them as a single digest message when the window expires. A background worker scans for expired timers every 60 seconds.

- id: "download_batch"
  action: aggregate
  match:
    event: "download.completed"
    channel: "ntfy"
  priority: 40
  params:
    window: "15m"
    digest_template_id: "download.digest"

Aggregate parameters:

Redis key pattern: notify:agg:{ruleID}:{eventType}:{channel} (List), notify:agg:timer:{ruleID}:{eventType}:{channel} (timer key with TTL)

The digest template receives a .items field containing all aggregated payloads:

**Digest: {{ len .items }} items in the last 15 minutes**
{{ range .items }}
- {{ .title }} ({{ .size }})
{{ end }}

Drop

Silently discards matching notifications. Useful for suppressing known noise.

- id: "drop_test_events"
  action: drop
  match:
    event: "test.*"
    channel: "*"
  priority: 10

Rule Evaluation Order

Rules are sorted by priority descending. The first matching rule wins. If a higher-priority mute rule matches, lower-priority throttle or aggregate rules are never evaluated.

Example with two rules:

- id: "night_mute"
  action: mute
  match: { event: "*", channel: "*" }
  condition: "time.hour >= 23 || time.hour < 8"
  priority: 100 # Evaluated first

- id: "infra_throttle"
  action: throttle
  match: { event: "infra.*", channel: "*" }
  params: { window: "5m", limit: 1 }
  priority: 50 # Only evaluated if mute doesn't match

At 2 PM: night_mute condition is false, so infra_throttle applies to infra.host.down events.
At 1 AM: night_mute condition is true, all notifications are silenced regardless of other rules.

Configuration

Minimum Setup

1. Create templates in the Notifications UI (/service/web/notifications → Templates), for example:

   • ID: bookmark.created
   • Format: markdown
   • Body: **New Bookmark**\n{{ .url }}

2. Optionally add rules on the Rules tab (mute / throttle / aggregate / drop).

3. Configure at least one notification channel on the Channels tab (or via the notify config chat command):

/notify config

name:    slack
template: slack://tokenA/tokenB/tokenC

Full Configuration Reference

See reference/notify.yaml (deprecated YAML note) and this guide for template/rule field shapes.
Templates and rules live in the database, not in flowbot.yaml.

Predefined Templates

The following template IDs are commonly used by modules and pipelines.
Create matching rows in the Notifications UI (or seed notify_templates) to enable them.

Usage Patterns

From Pipeline Steps

pipelines:
  - name: notify-on-bookmark
    enabled: true
    trigger:
      event: bookmark.created
    steps:
      - name: send-notify
        capability: core
        operation: notify_send
        params:
          template_id: "bookmark.created"
          channels: ["slack", "ntfy"]
          payload: "{{ .Event.data }}"

From Cron Jobs

func(ctx types.Context) []types.MsgPayload {
    // ... fetch data ...
    err := notify.GatewaySend(ctx.Context(), ctx.AsUser, "reader.news.summary",
        []string{"slack", "ntfy"}, map[string]any{
            "body":        summaryText,
            "entry_count": count,
        })
    if err != nil {
        flog.Error(err)
    }
    return nil
}

From Webhook Handlers

err := notify.GatewaySend(ctx.Context(), ctx.AsUser, "finance.transaction",
    []string{"slack", "ntfy"}, map[string]any{
        "amount":   payload.Amount,
        "currency": payload.Currency,
        "category": payload.Category,
    })

From Agent Actions

err := notify.GatewaySend(ctx.Context(), uid, "agent.status",
    []string{"slack", "ntfy"}, map[string]any{
        "hostid":   hostid,
        "hostname": hostname,
        "status":   "online",
    })

Integration Points

GatewaySend Entry Point

notify.GatewaySend() is the single entry point into the gateway. It is invoked by CapCore notify_send (pkg/capability/core), which is reachable from every ruleset that can dispatch a capability:

Interactive chat messages (command responses, form interactions) continue to use event.SendMessage() directly.

Redis Usage

The rule engine uses Redis for three state-tracking patterns:

Throttle counters use atomic INCR with EXPIRE on first increment, avoiding TOCTOU race conditions. Aggregate lists are flushed by a background worker that scans for expired timer keys.

Error Handling

• Missing template: returns types.ErrNotFound, gateway logs warning and skips
• Template parse error: caught at startup during notifytmpl.Init(), server fails to start
• Render failure: logged per-channel, other channels continue
• Channel not configured: logged and skipped, no error returned
• Redis unavailable: throttle/aggregate operations fail-open (notifications allowed through)
• Invalid rule window: logged and rule is skipped (fails-open)

Testing

# Template engine tests (11 test cases)
go test ./pkg/notify/template/...

# Rule engine tests (5 test cases)
go test ./pkg/notify/rules/...

# Existing notify tests (13 test cases)
go test ./pkg/notify/...

# Full suite
go tool task test