Notification Pipeline Usage Guide

Notification Pipeline — Usage Guide

Date: 2026-07-07

This guide explains how to use the unified notification pipeline after it is implemented. It covers adding a new polled feed, implementing change-detection logic, routing through filters, and delivering via ntfy or email.

Architecture and migration details: unified-notification-pipeline-plan.md (see §7 phases, §22 optimizations, §23 playbook, §27 poll vs delivery / stale data, §24 validation).

---

1. Mental model

Every notification flow follows the same five steps:

1. POLL      — fetch raw data from an external source
2. PERSIST   — save feed state (analytics / history tables)
3. DETECT    — source-owned Core policy compares new vs known state
4. MAP       — Application maps source changes to NotificationEvent list
5. DELIVER   — ProcessNotificationEvents (match → render → deliver → ledger)
6. DONE      — feed tables updated; subscribers notified

You implement steps 1–4 per feed. Delivery is shared and rarely changes.

flowchart LR
    A[Your feed client] --> B[Your monitor use case]
    B --> C[Feed store / DB]
    B --> D[Source change detector]
    D --> E[NotificationEvent mapper]
    E --> F[ProcessNotificationEvents]
    F --> G[ntfy or email]

---

2. When to use the pipeline

Use the pipeline	Do not use the pipeline (yet)
Recurring polled feeds (outages, schedules, status pages)	One-off form submits (contact form)
Events with filterable fields (feeder, area, keywords)	Internal admin-only logs
Multiple delivery targets (global topic, user topics, email)	Real-time WebSocket push

Contact and similar workflows can adopt the pipeline later by registering a contact.messages feed.

---

3. Quick reference — existing feeds

Feed ID	Monitor	Feed store table	System ntfy topic
beneco.power-outages	SyncPowerOutagesAndPublishEvents	power_outages	beneco-power-outages
beneco.scheduled-outages	SyncScheduledOutagesAndPublishEvents	scheduled_outages	beneco-scheduled-outages

---

4. Adding a new polled feed (step by step)

Example: a hypothetical beneco.water-outages feed.

Step 1 — Register the feed in Core

File: src/Obaki.Core/Modules/Notifications/Domain/NotificationFeedCatalog.cs

public static readonly NotificationFeedDefinition WaterOutages = new()
{
    Id = new NotificationFeedId("beneco.water-outages"),
    DisplayName = "BENECO Water Outages",
    EventTypes =
    [
        NotificationEventType.NewOutage,
        NotificationEventType.Restoration
    ],
    FilterableFields =
    [
        FilterableField.Feeder,
        FilterableField.Area,
        FilterableField.Keywords
    ]
};

Add the feed to the catalog list returned by GetAll().

Step 2 — Core module for domain + contracts (if new data type)

src/Obaki.Core/Modules/WaterOutages/
  Contracts/
    IWaterOutageFeedClient.cs
    IWaterOutageStore.cs
  Domain/
    WaterOutage.cs
    WaterOutageFeedItem.cs
    WaterOutageMonitorOptions.cs

Follow the same shape as PowerOutages.

Step 3 — Infrastructure feed client (poll the provider)

public sealed class BenecoWaterOutageFeedClient(HttpClient httpClient)
    : IWaterOutageFeedClient
{
    public async Task<IReadOnlyList<WaterOutageFeedItem>> GetCurrentAsync(
        CancellationToken cancellationToken = default)
    {
        // HTTP GET → parse JSON → return normalized items
    }
}

Register in DependencyInjection.cs:

services.AddHttpClient<IWaterOutageFeedClient, BenecoWaterOutageFeedClient>((sp, client) =>
{
    var options = sp.GetRequiredService<IOptions<WaterOutageMonitorOptions>>().Value;
    client.Timeout = TimeSpan.FromSeconds(options.FeedTimeoutSeconds);
});

Step 4 — Infrastructure store (persist for analytics)

Add EF entity + migration. Table example: water_outages.

Important: This table is for history/analytics. Notification dedup lives in notification_delivery_ledger, not here.

Step 5 — Core source change detector

File: src/Obaki.Core/Modules/WaterOutages/Policies/WaterOutageChangeDetector.cs

Keep source-specific transition rules in the source module. The detector should return source-owned changes, not NotificationEvent.

public enum WaterOutageChangeKind
{
    NewOutage,
    Restoration
}

public sealed record WaterOutageChange(
    WaterOutageFeedItem Current,
    WaterOutage? Existing,
    WaterOutageChangeKind Kind);

public static class WaterOutageChangeDetector
{
    public static IReadOnlyList<WaterOutageChange> Detect(
        IReadOnlyList<WaterOutageFeedItem> feedItems,
        IReadOnlyDictionary<int, WaterOutage> existingBySourceId)
    {
        // Compare source state only. Do not load subscriptions or deliver here.
    }
}

Step 6 — Application mapper and monitor

File: src/Obaki.Application/UseCases/WaterOutages/CheckWaterOutagesAndNotify.cs

public sealed class CheckWaterOutagesAndNotify(
    IWaterOutageFeedClient feedClient,
    IWaterOutageStore store,
    ProcessNotificationEvents pipeline,
    TimeProvider timeProvider,
    ILogger<CheckWaterOutagesAndNotify> logger)
{
    public async Task ExecuteAsync(CancellationToken cancellationToken = default)
    {
        var now = timeProvider.GetUtcNow();
        var feedItems = await feedClient.GetCurrentAsync(cancellationToken);
        var existing = await store.GetBySourceIdsAsync(...);

        await store.UpsertAsync(feedItems, now, cancellationToken);

        var changes = WaterOutageChangeDetector.Detect(feedItems, existing);
        var events = WaterOutageNotificationEvents.FromChanges(changes, now);

        if (events.Count > 0)
            await pipeline.ProcessAsync(events, cancellationToken);
    }
}

Rules:

• Always persist feed state before calling the pipeline.
• Keep source-state decisions in Core policies.
• Keep NotificationEvent mapping in Application.
• Build NotificationEvent with stable SourceKey (usually the provider’s id).
• Put filterable text in Fields using catalog field names.
• Put the typed object in SourcePayload for the renderer.

Step 7 — Application renderer (message text)

File: src/Obaki.Application/UseCases/Notifications/Renderers/WaterOutageNotificationRenderer.cs

public sealed class WaterOutageNotificationRenderer : INotificationRenderer
{
    public bool CanRender(NotificationFeedId feedId, NotificationEventType eventType) =>
        feedId.Value == "beneco.water-outages";

    public RenderedNotification Render(
        NotificationEvent notificationEvent,
        NotificationSubscription subscription)
    {
        var item = (WaterOutageFeedItem)notificationEvent.SourcePayload!;

        return notificationEvent.EventType switch
        {
            NotificationEventType.NewOutage => new RenderedNotification(
                Title: "Water outage reported",
                Body: $"Area: {item.Area}\nFeeder: {item.Feeder}",
                SequenceId: $"water-outage-{item.SourceId}",
                Tags: ["droplet", "warning"],
                Priority: NotificationPriority.High),

            NotificationEventType.Restoration => new RenderedNotification(
                Title: "Water restored",
                Body: $"Area: {item.Area}\nFeeder: {item.Feeder}",
                SequenceId: $"water-restored-{item.SourceId}",
                Tags: ["droplet"],
                Priority: NotificationPriority.Default),

            _ => throw new NotSupportedException()
        };
    }
}

Register in DI:

services.AddScoped<INotificationRenderer, WaterOutageNotificationRenderer>();

Step 8 — API background worker

File: src/Obaki.Api/BackgroundServices/WaterOutageMonitorWorker.cs

Copy the pattern from PowerOutageMonitorWorker: periodic timer → scope → CheckWaterOutagesAndNotify.ExecuteAsync.

Register in Program.cs when WaterOutages:Enabled is true.

Step 9 — System subscription (global topic)

Either seed a migration row or bind from config:

"NotificationPipeline": {
  "SystemSubscriptions": {
    "WaterOutages": {
      "FeedId": "beneco.water-outages",
      "NtfyTopic": "beneco-water-outages"
    }
  }
}

Add the topic to ntfy ACL (api_publisher write, everyone read) and NotificationTopicPolicy allowlist.

Step 10 — Tests

Test	Location
Feed client parsing	Obaki.Infrastructure or Application with mocked HTTP
Event detection logic	Obaki.Application.Tests
Renderer output	Golden string comparison
Pipeline fan-out	Mock subscription store + ledger
End-to-end monitor	Extend pattern from SyncPowerOutagesAndPublishEventsTests

---

5. Using the pipeline from existing monitors

After migration, outage monitors should look like this:

// 1. Poll + persist (unchanged responsibility)
await eventStore.AddAsync(...);

// 2. Detect source changes, then map to NotificationEvent signals
var changes = SourceChangeDetector.Detect(...);
var events = SourceNotificationEvents.FromChanges(changes, now);

// 3. Hand off to pipeline
if (events.Count > 0)
{
    var result = await processNotificationEvents.ProcessAsync(events, cancellationToken);
    logger.LogInformation(
        "Pipeline delivered {Delivered} notifications across {Subscriptions} subscriptions.",
        result.DeliveredCount,
        result.MatchedSubscriptionCount);
}

Monitors must not call INtfyNotificationService or IEmailNotificationService directly.

---

6. Delivery providers

6.1 ntfy (push)

When: Real-time mobile alerts, public or per-user topics.

Subscription row:

{
  "feedId": "beneco.power-outages",
  "filter": { "locationKeywords": ["Bakakeng"] },
  "delivery": {
    "provider": "Ntfy",
    "destination": "beneco-outages-alerts-k8f3p92mxq"
  }
}

System topic (match all):

{
  "feedId": "beneco.power-outages",
  "filter": {},
  "delivery": {
    "provider": "Ntfy",
    "destination": "beneco-power-outages"
  },
  "scope": "System"
}

Flow:

RenderedNotification → NtfyNotificationDeliveryRouter → INtfyNotificationService → ntfy server

Topic must pass NotificationTopicPolicy (allowlisted prefixes).

6.2 Email (reliable async)

When: User wants inbox delivery, digests, or no ntfy app.

Subscription row:

{
  "feedId": "beneco.scheduled-outages",
  "filter": { "eventTypes": ["WeeklyDigest"] },
  "delivery": {
    "provider": "Email",
    "destination": "[email protected]"
  }
}

Flow:

RenderedNotification → EmailNotificationDeliveryRouter → EmailOutboxItem → ProcessContactEmailOutbox worker → Resend/MailKit

Email uses the outbox for retries. User email addresses should be verified before is_active = true.

6.3 Both ntfy and email for the same filter

Create two subscriptions with identical filter_json:

Sub A: provider=Ntfy,  destination=beneco-outages-alerts-abc123
Sub B: provider=Email, [email protected]

Each has independent cooldown and ledger entries.

---

7. Filter logic

Filters are evaluated in NotificationFilterMatcher (Core, pure functions).

User subscription limits

Phase	Limit
Before auth	MaxPerIp / MaxPerIpPerDay (default 3) — see unified-notification-pipeline-plan.md §11.3
After auth	MaxSubscriptionsPerUser = 2 by default — one live outage alert and one scheduled outage alert per user

7.1 Field matching

Filter property	Matches against
Feeders	event.Fields["feeder"] (case-insensitive)
LocationKeywords	event.Fields["area"] or ["areas"] (substring)
Keywords	All string values in Fields combined
EventTypes	event.EventType (empty = all types)

7.2 Empty filter

System subscriptions use an empty filter = match every event from that feed.

7.3 Example — user subscription via API

POST /api/notification-subscriptions
Content-Type: application/json

{
  "displayName": "Home feeder",
  "feedId": "beneco.power-outages",
  "filter": {
    "feeders": ["Feeder 14"],
    "locationKeywords": ["Bakakeng"],
    "eventTypes": ["NewOutage", "Restoration"]
  },
  "delivery": {
    "provider": "Ntfy"
  }
}

Response includes subscribeUrl for the generated topic.

7.4 Programmatic system subscription (config seed)

On startup or migration, seed:

INSERT INTO notification_subscriptions
  (display_name, feed_id, filter_json, provider, destination, scope, is_active)
VALUES
  ('BENECO Power Outages (system)', 'beneco.power-outages', '{}', 'Ntfy', 'beneco-power-outages', 'System', true);

---

8. Event types reference

Power outages (beneco.power-outages)

Event type	When emitted
NewOutage	New ongoing outage in feed
Restoration	Existing outage gets timerestored
RestoredAnnouncement	First seen already restored
ManualAnnouncement	Admin API announcement

Scheduled outages (beneco.scheduled-outages)

Event type	When emitted
NewEntry	New scheduled item
Updated	Time/area/feeder changed
Cancelled	Item cancelled
WeeklyDigest	Weekly summary hour
DailyUpdate	Daily change summary
DayBeforeReminder	Reminder for tomorrow

Digest/reminder events use SourceKey like weekly-2026-07-07 (not a BENECO row id).

---

9. Configuration checklist

When adding a feed, update:

File	What to add
manifest.json	Monitor options + pipeline system subscription
.env.template	Same
Obaki.ConfigGenerator	Option classes if new
ntfy server.yml ACL	Topic read/write rules
NotificationTopicPolicy	Allowlisted topic or prefix

Example env vars:

WaterOutages__Enabled=true
WaterOutages__PollingIntervalSeconds=300
NotificationPipeline__SystemSubscriptions__WaterOutages__FeedId=beneco.water-outages
NotificationPipeline__SystemSubscriptions__WaterOutages__NtfyTopic=beneco-water-outages
Notifications__Ntfy__BaseUrl=https://ntfy.example.com
Notifications__Ntfy__PublishToken=tk_...

---

10. Testing a new feed locally

10.1 Disable live global topics (optional)

Point system subscription at a test topic:

NotificationPipeline__SystemSubscriptions__WaterOutages__NtfyTopic=beneco-water-outages-test

10.2 Send a test event through the pipeline

POST /api/notification-subscriptions/{publicId}/test

Or call SendNotificationSubscriptionTest from a test endpoint.

10.3 Unit test the detector without ntfy

var pipeline = new FakeProcessNotificationEvents();
var monitor = new CheckWaterOutagesAndNotify(feedClient, store, pipeline, time, logger);

await monitor.ExecuteAsync();

Assert.Single(pipeline.CapturedEvents);
Assert.Equal(NotificationEventType.NewOutage, pipeline.CapturedEvents[0].EventType);

---

11. Digest and scheduled events

Some events are time-gated before they enter the pipeline:

Feed	Gating table	Purpose
Scheduled weekly digest	scheduled_outage_digest_runs	Emit at most one WeeklyDigest per week
Scheduled daily update	scheduled_outage_digest_runs	Emit at most one DailyUpdate per day
Day-before reminder	scheduled_outage_reminder_runs	Emit at most one reminder per outage date

Pattern:

if (ScheduledOutageDigestPolicy.IsDigestHour(localNow, options) &&
    !await store.HasDigestBeenSentAsync(today, DigestType.Daily, ct))
{
    var notificationEvent = BuildDailyUpdateEvent(...);
    var result = await pipeline.ProcessAsync([notificationEvent], ct);

    if (WasSystemDelivery(result, notificationEvent))
        await store.MarkDigestSentAsync(today, DigestType.Daily, count, now, ct);
}

Gating tables control whether an event exists. The ledger controls whether a subscription already received it.

Fresh data when delivery is slow (required)

Poll may run every 60s while fan-out to many subscriptions takes longer. See notification-poll-vs-delivery.md and §27 in unified-notification-pipeline-plan.md.

---

12. Analytics — what to query

Question	Query from
How many outages this month?	power_outages
Outages per feeder	power_outages
Average restoration time	power_outages (time_off, time_restored)
How many alerts sent to a user topic?	notification_delivery_ledger JOIN notification_subscriptions
When was first alert for outage 42?	notification_delivery_ledger or power_outages.outage_notified_at

Do not use notification tables as the primary outage history source.

---

13. Troubleshooting

Symptom	Check
No notifications at all	NotificationPipeline:Enabled, monitor Enabled, ntfy token
Global topic works, user topic does not	Subscription is_active, filter criteria, ntfy ACL for beneco-outages-alerts-*
Duplicate notifications	Ledger unique constraint; stable SourceKey + EventType
Missing notifications after restart	Ledger should prevent dupes; detector must re-emit only when state warrants
Email never arrives	Outbox worker enabled, ProcessContactEmailOutbox, Resend/MailKit config
403 from ntfy	PublishToken, topic allowlist in NotificationTopicPolicy

---

14. File checklist (new feed)

[ ] Core: NotificationFeedCatalog entry
[ ] Core: Module contracts + domain (if new data type)
[ ] Core: Source change detector / digest policy if the feed needs one
[ ] Infrastructure: Feed client + store + migration
[ ] Application: Check*AndNotify (poll + persist + map source changes to events)
[ ] Application: *NotificationEvents mapper
[ ] Application: *NotificationRenderer
[ ] Application: Register renderer in DI
[ ] Api: BackgroundService worker
[ ] Api: Program.cs registration
[ ] Config: manifest + env template
[ ] ntfy: ACL for new topic
[ ] Tests: detector, renderer, pipeline integration
[ ] Docs: PowerOutageNotifications.md or feed-specific doc

---

15. Related documents

• unified-notification-pipeline-plan.md — architecture, phases, cleanup
• NotificationFiltersPlan.md — user-facing filter product details
• ../deployment/ntfy-secure-publishing-vps.md — ntfy server security
• PowerOutageNotifications.md — current BENECO power outage behavior
• Architecture.md — layer boundaries
• ModuleRules.md — module intake checklist