Unified Notification Pipeline Plan

Unified Notification Pipeline — Implementation Plan

Date: 2026-07-07

This plan proposes a common notification abstraction for Obaki: ingest from a feed pool, apply filter logic, and deliver through pluggable providers (ntfy, email). The same pipeline powers today’s global outage topics, scheduled outage topics, and tomorrow’s per-user custom filters — and can accept any new source without duplicating publish/dedup/matching code.

Related docs:

• NotificationFiltersPlan.md — per-user filter feature (built on top of this pipeline)
• notification-pipeline-usage.md — how to add feeds, logic, and delivery providers
• notification-poll-vs-delivery.md — poll vs slow delivery: spam and stale data mitigation
• ../deployment/ntfy-secure-publishing-vps.md — ntfy ACL and topic security
• Architecture.md — layer boundaries

For implementers: Start at §7 (phases), §22 (optimizations), §23 (step-by-step playbook), §27 (poll vs delivery / stale data), and §24 (validation commands).

---

1. Problem: Duplicated Patterns Today

Power outages, scheduled outages, and contact flows all repeat the same shape with small variations:

Poll feed → detect change → build message → publish to ntfy topic → track "already sent"

Current use case	Feed	Filter today	Delivery	Dedup
Power outages	BENECO wballoutages.php	None (all items)	Single ntfy topic	OutageNotifiedAt on PowerOutage
Scheduled outages	BENECO scheduled feed	None	Single ntfy topic	Digest/reminder run tables
Contact submit	User form	N/A	ntfy + email outbox	Outbox idempotency
Manual announcement	API request	N/A	Single ntfy topic	None

Pain points:

• Two nearly identical *NotificationFactory classes with duplicated message trimming.
• Two Check*AndNotify classes each calling INtfyNotificationService directly with copy-pasted TryPublishAsync.
• Filter logic (WatchedAreas) is config-only and unused — no path to per-user rules.
• Adding email to outage alerts means touching every monitor separately.
• Adding a third feed (e.g. water outages, Meralco) means cloning the whole pattern again.

---

2. Target: One Pipeline, Many Sources and Sinks

flowchart TB
    subgraph sources [Feed Pool - Sources]
        PO[PowerOutageFeed]
        SO[ScheduledOutageFeed]
        Future[Future feeds...]
    end

    subgraph pipeline [Notification Pipeline - Core + Application]
        Ingest[Ingest / Normalize]
        Events[NotificationEvent stream]
        Subs[Subscription Registry]
        Match[Filter / Match Engine]
        Render[Message Renderer]
        Route[Delivery Router]
        Dedup[Delivery Ledger]
    end

    subgraph sinks [Providers - Infrastructure]
        Ntfy[INtfyNotificationService]
        Email[IEmailNotificationService / Outbox]
    end

    subgraph targets [Delivery Targets]
        GlobalPO[beneco-power-outages]
        GlobalSO[beneco-scheduled-outages]
        UserTopics[beneco-outages-alerts-*]
        UserEmail[[email protected]]
    end

    PO --> Ingest
    SO --> Ingest
    Future --> Ingest
    Ingest --> Events
    Events --> Match
    Subs --> Match
    Match --> Render
    Render --> Route
    Route --> Dedup
    Dedup --> Ntfy
    Dedup --> Email
    Ntfy --> GlobalPO
    Ntfy --> GlobalSO
    Ntfy --> UserTopics
    Email --> UserEmail

Key idea: A subscription is not a user account — it is a stored rule that says “when events from feed X match criteria Y, deliver via provider Z to destination D.” Global topics are just built-in system subscriptions with empty criteria.

---

3. Core Concepts

3.1 NotificationFeed (feed pool entry)

A registered source of events. Feeds are identified by a stable string key, not by implementation class.

public sealed record NotificationFeedId(string Value); // e.g. "beneco.power-outages"

public sealed class NotificationFeedDefinition
{
    public NotificationFeedId Id { get; init; }
    public string DisplayName { get; init; }           // "BENECO Power Outages"
    public IReadOnlyList<NotificationEventType> EventTypes { get; init; }
    public IReadOnlyList<FilterableField> FilterableFields { get; init; }
}

Initial feeds:

Feed ID	Event types	Filterable fields
beneco.power-outages	NewOutage, Restoration, RestoredAnnouncement	feeder, area, cause, status, keywords
beneco.scheduled-outages	NewEntry, Updated, Cancelled, WeeklyDigest, DailyUpdate, DayBeforeReminder	feeder, areas, interruptionType, purpose, keywords

New feeds register in a feed catalog — no pipeline code changes required beyond an ingest adapter.

3.2 NotificationEvent (normalized event)

Every source adapter maps its domain object to a common event:

public sealed record NotificationEvent(
    NotificationFeedId FeedId,
    NotificationEventType EventType,
    string SourceKey,              // e.g. BENECO source id or digest key
    DateTimeOffset OccurredAt,
    IReadOnlyDictionary<string, string> Fields,  // normalized filterable payload
    object? SourcePayload = null);  // optional typed ref for renderers — may be stale; see §27

Signal vs snapshot: NotificationEvent is a delivery signal (what happened + stable SourceKey), not a frozen message body. Final title/body are built at send time from the latest feed row (§27).

Example — power outage new event:

new NotificationEvent(
    FeedId: new("beneco.power-outages"),
    EventType: NotificationEventType.NewOutage,
    SourceKey: "42",
    OccurredAt: now,
    Fields: new Dictionary<string, string>
    {
        ["feeder"] = "Feeder 14",
        ["area"] = "Bakakeng Central, Marcos Highway",
        ["cause"] = "Line fault",
        ["status"] = "Ongoing"
    },
    SourcePayload: outageFeedItem);

Example — scheduled weekly digest:

new NotificationEvent(
    FeedId: new("beneco.scheduled-outages"),
    EventType: NotificationEventType.WeeklyDigest,
    SourceKey: "weekly-2026-07-07",
    OccurredAt: now,
    Fields: new Dictionary<string, string>
    {
        ["itemCount"] = "12",
        ["weekOf"] = "2026-07-07"
    },
    SourcePayload: outageList);

3.3 NotificationSubscription (filter + delivery target)

Replaces both “global topic config” and “user notification filter” with one model:

public sealed class NotificationSubscription
{
    public long Id { get; set; }
    public string? PublicId { get; set; }              // null for system subscriptions
    public string? UserId { get; set; }                // null until auth exists; then FK to user
    public string DisplayName { get; set; }
    public NotificationFeedId FeedId { get; set; }       // which feed pool
    public NotificationFilter Filter { get; set; }     // criteria (empty = match all)
    public NotificationDeliveryTarget Target { get; set; }
    public bool IsActive { get; set; }
    public int CooldownMinutes { get; set; }
    public DateTimeOffset? LastSentAt { get; set; }
    public SubscriptionScope Scope { get; set; }         // System | User
    public DateTimeOffset CreatedAt { get; set; }
    public string? CreatedByIpHash { get; set; }         // abuse control before auth
}

3.4 NotificationFilter (reusable criteria)

public sealed class NotificationFilter
{
    public IReadOnlyList<string> Feeders { get; init; } = [];
    public IReadOnlyList<string> LocationKeywords { get; init; } = [];
    public IReadOnlyList<string> Keywords { get; init; } = [];
    public IReadOnlyList<NotificationEventType> EventTypes { get; init; } = [];
    public FilterMatchMode MatchMode { get; init; } = FilterMatchMode.Any;
}

Empty filter = match all events from the subscription’s feed (used for global topics).

Pure matching logic lives in Core: NotificationFilterMatcher.IsMatch(filter, event).

3.5 NotificationDeliveryTarget (where to send)

public sealed record NotificationDeliveryTarget
{
    public NotificationProvider Provider { get; init; }  // Ntfy | Email
    public string Destination { get; init; }           // topic name or email address
    public NtfyDeliveryOptions? NtfyOptions { get; init; }
    public EmailDeliveryOptions? EmailOptions { get; init; }
}

Examples:

Subscription	Provider	Destination
System: power outages	Ntfy	beneco-power-outages
System: scheduled outages	Ntfy	beneco-scheduled-outages
User: home filter	Ntfy	beneco-outages-alerts-k8f3p92mxq
User: email backup	Email	[email protected]

3.6 NotificationDeliveryLedger (dedup)

One table replaces per-entity *NotifiedAt columns for notification purposes (feed state tables keep their own ingest dedup):

public sealed record DeliveryLedgerKey(
    long SubscriptionId,
    NotificationFeedId FeedId,
    string SourceKey,
    NotificationEventType EventType);

System subscriptions and user subscriptions share the same dedup mechanism.

---

4. Delivery Pipeline Stages (Application)

Single delivery orchestrator: ProcessNotificationEvents.

Feed polling, source-state comparison, and source-owned event decisions happen before this class. The source modules decide what changed; Application maps those changes to NotificationEvent; ProcessNotificationEvents decides who receives each event and records delivery.

1. MATCH     — load active subscriptions for FeedId; run FilterMatcher
2. RENDER    — INotificationRenderer builds provider-specific payload
3. DELIVER   — INotificationDeliveryRouter sends via ntfy/email
4. RECORD    — write DeliveryLedger entry on success

public sealed class ProcessNotificationEvents(
    INotificationSubscriptionStore subscriptionStore,
    INotificationDeliveryLedgerStore ledgerStore,
    INotificationRendererRegistry rendererRegistry,
    INotificationDeliveryRouter deliveryRouter,
    INotificationEventSourceResolver sourceResolver,  // re-fetch latest row at send time — §27
    IOptions<NotificationPipelineOptions> options,
    ILogger<ProcessNotificationEvents> logger)
{
    public async Task<ProcessNotificationEventsResult> ProcessAsync(
        IReadOnlyList<NotificationEvent> events,
        CancellationToken cancellationToken = default);
}

Monitors become thin orchestration shells:

// SyncPowerOutagesAndPublishEvents (simplified target shape)
await eventStore.PersistChangesAsync(...);           // feed state unchanged
var changes = PowerOutageChangeDetector.Detect(feedItems, existingBySourceId);
var events = PowerOutageNotificationEvents.FromChanges(changes, todaysScheduled, now);
await processNotificationEvents.ProcessAsync(events, ct);  // delivery pipeline handles notify

4.1 ProcessNotificationEvents algorithm

public async Task<ProcessNotificationEventsResult> ProcessAsync(
    IReadOnlyList<NotificationEvent> events,
    CancellationToken cancellationToken)
{
    if (!_options.Enabled || events.Count == 0)
        return ProcessNotificationEventsResult.Empty;

    var result = new ProcessNotificationEventsResult();

    foreach (var feedGroup in events.GroupBy(e => e.FeedId.Value))
    {
        var feedId = new NotificationFeedId(feedGroup.Key);
        var subscriptions = await _subscriptionStore.GetActiveByFeedAsync(feedId, cancellationToken);
        if (subscriptions.Count == 0)
            continue;

        foreach (var signal in feedGroup)
        {
            var renderer = _rendererRegistry.Resolve(signal.FeedId, signal.EventType);
            if (renderer is null)
            {
                _logger.LogWarning("No renderer for {FeedId} {EventType}", feedId.Value, signal.EventType);
                continue;
            }

            foreach (var subscription in subscriptions)
            {
                if (!NotificationFilterMatcher.IsMatch(subscription.Filter, signal))
                    continue;

                if (IsInCooldown(subscription, signal))
                    continue;

                var ledgerKey = new DeliveryLedgerKey(
                    subscription.Id, feedId, signal.SourceKey, signal.EventType);

                if (await _ledgerStore.ExistsAsync(ledgerKey, cancellationToken))
                    continue;

                if (await IsOverDailyCapAsync(subscription, cancellationToken))
                    continue;

                // §27: Re-fetch latest feed row before render — polls may have updated DB during slow fan-out
                var freshEvent = await _sourceResolver.ResolveAsync(signal, cancellationToken);
                if (freshEvent is null)
                {
                    _logger.LogDebug(
                        "Skipping delivery for missing source {SourceKey} on feed {FeedId}",
                        signal.SourceKey, feedId.Value);
                    continue;
                }

                var rendered = renderer.Render(freshEvent, subscription);

                var delivered = await _deliveryRouter.DeliverAsync(
                    rendered, subscription.Target, cancellationToken);

                if (!delivered.Success)
                {
                    result.FailedCount++;
                    continue; // no ledger row — retry on next poll if signal still valid
                }

                await _ledgerStore.RecordAsync(ledgerKey, delivered.SentAt, cancellationToken);
                await _subscriptionStore.UpdateLastSentAtAsync(subscription.Id, delivered.SentAt, cancellationToken);
                result.DeliveredCount++;
            }
        }
    }

    return result;
}

Pragmatic rules baked into this loop:

• Early exit when pipeline disabled or no events.
• Subscriptions loaded once per feed per ExecuteAsync call (not per event).
• Renderer resolved once per event (not per subscription).
• Fresh render: INotificationEventSourceResolver reloads latest DB/feed state immediately before Render (§27).
• Failed delivery → no ledger write → retry on next poll only if detector still emits signal and ledger still empty.
• No parallel fan-out to ntfy initially (sequential is fine on a small VPS; optimize only if profiling shows need).
• Ledger check before render — skip expensive work for already-delivered (subscription, source, event type).

---

5. Layer Responsibilities

5.1 Core (Obaki.Core/Modules/Notifications)

Modules/Notifications/
  Contracts/
    INotificationSubscriptionStore.cs
    INotificationDeliveryLedgerStore.cs
    INotificationRenderer.cs
    INotificationDeliveryRouter.cs
    INotificationEventSourceResolver.cs   // latest row for signal at send time — §27
  Domain/
    NotificationEvent.cs
    NotificationFeedId.cs
    NotificationEventType.cs
    NotificationFilter.cs
    NotificationFilterMatcher.cs      // pure
    NotificationSubscription.cs
    NotificationDeliveryTarget.cs
    NotificationFeedCatalog.cs        // registered feeds + filterable fields
  Policies/
    NotificationTopicPolicy.cs        // allowlist validation rules

Core stays BCL-only. No ntfy/email types in matchers or filters.

5.2 Source modules (Obaki.Core/Modules/PowerOutages, ScheduledOutages)

Source modules own the rules for deciding what changed in their own feed data.

Modules/PowerOutages/
  Domain/
    PowerOutageChange.cs
  Policies/
    PowerOutageChangeDetector.cs

Modules/ScheduledOutages/
  Domain/
    ScheduledOutageChange.cs
    ScheduledOutageReminderCandidate.cs
  Policies/
    ScheduledOutageChangeDetector.cs
    ScheduledOutageDigestPolicy.cs
    ScheduledOutageReminderPolicy.cs

These classes do not deliver notifications. They return source-owned changes such as “new outage”, “restored”, “scheduled item updated”, or “tomorrow reminder due”. Application maps those changes to NotificationEvent.

5.3 Application (Obaki.Application/UseCases/Notifications)

UseCases/Notifications/
  ProcessNotificationEvents.cs        // delivery pipeline orchestrator
  CreateNotificationSubscription.cs   // user subscriptions
  UpdateNotificationSubscription.cs
  DeleteNotificationSubscription.cs
  GetNotificationSubscription.cs
  SendNotificationSubscriptionTest.cs
  GetNotificationFeedMetadata.cs

UseCases/PowerOutages/
  SyncPowerOutagesAndPublishEvents.cs       // poll + persist + map source changes to events
  PowerOutageNotificationEvents.cs          // maps PowerOutageChange -> NotificationEvent

UseCases/ScheduledOutages/
  SyncScheduledOutagesAndPublishEvents.cs   // poll + persist + map source changes to events
  ScheduledOutageNotificationEvents.cs      // maps ScheduledOutageChange -> NotificationEvent

Renderers (can live in Application as feed-specific classes):

  Renderers/
    PowerOutageNotificationRenderer.cs
    ScheduledOutageNotificationRenderer.cs

These replace PowerOutageNotificationFactory and ScheduledOutageNotificationFactory. Extract shared TrimNotificationMessage to one internal helper.

5.4 Infrastructure

Infrastructure/
  Persistence/Stores/Notifications/
    DatabaseNotificationSubscriptionStore.cs
    DatabaseNotificationDeliveryLedgerStore.cs
  Providers/Notifications/
    NtfyNotificationDeliveryRouter.cs   // wraps INtfyNotificationService
    EmailNotificationDeliveryRouter.cs  // wraps outbox or IEmailNotificationService
  NotificationPipelineExtensions.cs     // DI registration

System subscriptions seeded at startup or via migration:

// beneco-power-outages: match all power events → ntfy:beneco-power-outages
// beneco-scheduled-outages: match all scheduled events → ntfy:beneco-scheduled-outages

Config keys like NotificationPipeline:SystemSubscriptions:PowerOutages:NtfyTopic become subscription destinations, not direct publish calls.

---

6. Provider Abstraction

6.1 INotificationDeliveryRouter (Core contract)

public interface INotificationDeliveryRouter
{
    Task<NotificationDeliveryResult> DeliverAsync(
        RenderedNotification notification,
        NotificationDeliveryTarget target,
        CancellationToken cancellationToken = default);
}

Infrastructure implementation dispatches by target.Provider:

Provider	Implementation	Notes
Ntfy	NtfyNotificationDeliveryRouter	Uses existing INtfyNotificationService; enforces topic allowlist
Email	EmailNotificationDeliveryRouter	Queues to EmailOutboxItem for reliability (contact pattern)

6.2 INotificationRenderer (Core contract)

public interface INotificationRenderer
{
    bool CanRender(NotificationFeedId feedId, NotificationEventType eventType);
    RenderedNotification Render(NotificationEvent notificationEvent, NotificationSubscription subscription);
}

RenderedNotification is provider-agnostic:

public sealed record RenderedNotification(
    string Title,
    string Body,
    string? SequenceId,
    IReadOnlyList<string> Tags,
    NotificationPriority Priority);

Router maps RenderedNotification → NtfyNotification or email template payload.

6.3 Topic allowlist (security)

NotificationTopicPolicy validates destinations before publish:

public bool IsAllowedNtfyTopic(string topic) =>
    IsSystemTopic(topic)
    || topic.StartsWith(_options.UserTopicPrefix, StringComparison.Ordinal); // beneco-outages-alerts-

Allowed ntfy topic patterns:

Pattern	Who uses it	ACL
beneco-power-outages	System subscription	everyone read, api_publisher write
beneco-scheduled-outages	System subscription	same
beneco-power-outages-test	Test / dev	same
beneco-outages-alerts-*	User subscriptions (server-generated)	everyone read, api_publisher write

Users never choose a raw topic name. The API generates beneco-outages-alerts-{random} and returns the subscribe URL only.

Same rules as ../deployment/ntfy-secure-publishing-vps.md (update ACL entries to use beneco-outages-alerts-*).

---

7. Migration Path from Current Code

What NOT to build (YAGNI)

Skip	Why
MediatR / event bus	Explicit use-case calls match existing Obaki style
Generic INotificationFeedAdapter<T> framework	One feed client per source is enough until 3+ feeds exist
IRepository<T> base class	Use explicit stores like DatabasePowerOutageStore
Real-time WebSocket push	ntfy handles push to clients
Combining ntfy + email in one subscription row	Two rows; simpler dedup and cooldown
Removing *NotifiedAt columns in v1	Keep for analytics; dual-write is enough
User auth in Phase B	Phase D uses IP limits until accounts exist

---

Phase A — Extract without behavior change

Goal: Introduce types and renderers; monitors still publish directly. No DB changes.

Acceptance: dotnet test green; renderer golden tests match factory output byte-for-byte on title/message/tags.

Step	File(s)	Action
A1	src/Obaki.Core/Modules/Notifications/Domain/NotificationFeedId.cs	Add record NotificationFeedId(string Value)
A2	.../NotificationEventType.cs	Enum: NewOutage, Restoration, RestoredAnnouncement, ManualAnnouncement, NewEntry, Updated, Cancelled, WeeklyDigest, DailyUpdate, DayBeforeReminder
A3	.../NotificationEvent.cs	Record per §3.2
A4	.../NotificationFilter.cs, FilterMatchMode.cs	Per §3.4
A5	.../NotificationFilterMatcher.cs	Static IsMatch(filter, event) — pure, no I/O
A6	.../NotificationFeedCatalog.cs	Static list of feed definitions §3.1
A7	.../RenderedNotification.cs, NotificationPriority.cs	Provider-agnostic render output
A8	src/Obaki.Core/Modules/Notifications/Contracts/INotificationRenderer.cs	Per §6.2
A9	src/Obaki.Application/UseCases/Notifications/Internal/NotificationMessageTrimmer.cs	Extract shared trim logic from both factories
A10	.../Renderers/PowerOutageNotificationRenderer.cs	Port from PowerOutageNotificationFactory.cs
A11	.../Renderers/ScheduledOutageNotificationRenderer.cs	Port from ScheduledOutageNotificationFactory.cs
A12	tests/Obaki.Core.Tests/Modules/Notifications/NotificationFilterMatcherTests.cs	Feeder, keyword, event type cases
A13	tests/Obaki.Application.Tests/.../PowerOutageNotificationRendererTests.cs	Golden: same title/message as factory for each event type

Do not delete factories in Phase A. Monitors keep using them until Phase B + G.

---

Phase B — Pipeline for system subscriptions

Goal: Global topics flow through the pipeline; behavior unchanged for end users.

Acceptance: Existing SyncPowerOutagesAndPublishEventsTests pass; ntfy receives same topics/messages as before.

Step	File(s)	Action
B1	src/Obaki.Core/Modules/Notifications/Domain/NotificationSubscription.cs	Domain model §3.3
B2	.../NotificationDeliveryTarget.cs, NotificationProvider.cs, SubscriptionScope.cs	Enums + target record
B3	.../Contracts/INotificationSubscriptionStore.cs	GetActiveByFeedAsync, CRUD for later phases
B4	.../Contracts/INotificationDeliveryLedgerStore.cs	ExistsAsync, RecordAsync
B5	.../Contracts/INotificationDeliveryRouter.cs	Per §6.1
B6	.../Policies/NotificationTopicPolicy.cs	Allowlist §6.3
B7	src/Obaki.Infrastructure/Persistence/Migrations/YYYYMMDDHHMMSS_AddNotificationPipeline.cs	Tables §8
B8	.../Stores/Notifications/DatabaseNotificationSubscriptionStore.cs	EF implementation
B9	.../Stores/Notifications/DatabaseNotificationDeliveryLedgerStore.cs	EF + unique constraint
B10	.../Providers/Notifications/NtfyNotificationDeliveryRouter.cs	Maps RenderedNotification → NtfyNotification → INtfyNotificationService
B11	.../NotificationPipelineExtensions.cs	AddNotificationPipeline() — bind options, register stores, routers
B12	src/Obaki.Infrastructure/Persistence/Seeding/NotificationSubscriptionSeeder.cs	Seed system rows from config on startup or in migration
B13	src/Obaki.Application/UseCases/Notifications/ProcessNotificationEvents.cs	Algorithm §4.1 + §27 freshness
B14	.../NotificationEventSourceResolver.cs	Per-feed re-fetch before render (§27)
B15	.../NotificationRendererRegistry.cs	IEnumerable<INotificationRenderer> → resolve by feed + event type
B16	src/Obaki.Core/Modules/PowerOutages/Policies/PowerOutageChangeDetector.cs	Source-owned transition rules — see §23.1, §27.2
B17	src/Obaki.Application/UseCases/PowerOutages/PowerOutageNotificationEvents.cs	Map PowerOutageChange to NotificationEvent
B18	SyncPowerOutagesAndPublishEvents.cs	Remove INtfyNotificationService; persist → detect source changes → map signals → pipeline; dual-write
B19	src/Obaki.Core/Modules/ScheduledOutages/Policies/ScheduledOutageChangeDetector.cs	Source-owned scheduled item diff rules
B20	src/Obaki.Core/Modules/ScheduledOutages/Policies/ScheduledOutageDigestPolicy.cs, ScheduledOutageReminderPolicy.cs	Source-owned digest/reminder timing rules
B21	.../ScheduledOutages/ScheduledOutageNotificationEvents.cs	Map scheduled source changes to NotificationEvent
B22	SyncScheduledOutagesAndPublishEvents.cs	Same refactor as B18
B23	src/Obaki.Application/DependencyInjection.cs	Register pipeline, resolver, renderers, registry
B24	src/Obaki.Infrastructure/DependencyInjection.cs	Call AddNotificationPipeline()
B25	src/Obaki.Infrastructure/Providers/Ntfy/NtfyNotificationService.cs	Call NotificationTopicPolicy before publish
B26	manifest.json, Obaki.ConfigGenerator	Add NotificationPipeline section
B27	tests/Obaki.Core.Tests/.../*ChangeDetectorTests.cs	Source transition rules
B28	tests/Obaki.Application.Tests/.../ProcessNotificationEventsTests.cs	Match, dedup, cooldown, fan-out, fresh render
B29	tests/.../NotificationEventSourceResolverTests.cs	Updated area text after DB change between signal and send

Migration command:

dotnet ef migrations add AddNotificationPipeline `
  --project src/Obaki.Infrastructure `
  --startup-project src/Obaki.Api

Seed SQL (migration or seeder):

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

Use topic values from config, not hard-coded strings, when seeding from NotificationPipelineOptions.

Phase C — Ledger backfill and dual-write (required before cleanup)

Before removing any legacy notification columns or code paths:

1. Backfill notification_delivery_ledger from existing state:
   • For each power_outages row with outage_notified_at, insert a ledger row for the system power subscription (NewOutage or equivalent).
   • For each row with restored_notified_at, insert a ledger row for restoration / restored-announcement event types.
2. Run dual-write during a transition window: pipeline writes ledger and legacy *NotifiedAt columns (or vice versa) until counts match.
3. Add a one-off verification query comparing ledger rows to legacy columns; do not proceed to Phase G until it passes.

Phase G — Post-implementation cleanup (required, not optional)

Cleanup is a deliverable, not an afterthought. Complete this checklist before marking the pipeline done.

G1. Code removal

Remove / replace	After
PowerOutageNotificationFactory.cs	PowerOutageNotificationRenderer + shared NotificationMessageTrimmer
ScheduledOutageNotificationFactory.cs	ScheduledOutageNotificationRenderer
TryPublishAsync in SyncPowerOutagesAndPublishEvents	ProcessNotificationEvents
TryPublishAsync in SyncScheduledOutagesAndPublishEvents	ProcessNotificationEvents
Direct INtfyNotificationService injection in outage monitors	Pipeline router only
Former watched-area placeholder	User subscriptions with location keywords
Duplicate TrimNotificationMessage helpers	One shared internal helper

Keep but refactor (do not delete):

File	Change
SendPowerOutageTestNotifications.cs	Call pipeline with test destination override or ephemeral subscription
SendPowerOutageAnnouncement.cs	Emit a NotificationEvent with type ManualAnnouncement or call pipeline directly
SendTestNotification.cs	Stays for raw ntfy smoke tests; document as low-level escape hatch

G2. Config cleanup

Legacy key	Replacement
PowerOutages:NtfyTopic	NotificationPipeline:SystemSubscriptions:PowerOutages:NtfyTopic
ScheduledOutages:NtfyTopic	NotificationPipeline:SystemSubscriptions:ScheduledOutages:NtfyTopic
Former watched-area placeholder	Removed
PowerOutages:TestNtfyTopic	NotificationPipeline:TestNtfyTopic or test subscription row

Update manifest.json, .env.template, and Obaki.ConfigGenerator in the same PR as config removal.

G3. Table and column policy

Object	Action
power_outages	Keep. Analytics source of truth. Do not drop columns used for reporting.
scheduled_outages	Keep. Same.
scheduled_outage_digest_runs	Keep. Gates whether digest events are emitted (ingest logic), not delivery.
scheduled_outage_reminder_runs	Keep. Same.
power_outages.outage_notified_at	Keep for analytics (“time to notify”). Optional to stop writing after ledger is trusted; do not drop without migration plan.
power_outages.restored_notified_at	Keep for analytics. Same policy.
notification_subscriptions	Keep. Active runtime data.
notification_delivery_ledger	Keep. Add retention job (see G5).
power_outage_digest_runs	Already removed in migration AddScheduledOutages. Do not reintroduce.

Do not drop OutageNotifiedAt / RestoredNotifiedAt unless analytics queries are migrated to the ledger and stakeholders sign off.

G4. Scripts and docs

Item	Action
scripts/ImportHistoricalPowerOutages.cs	Fix table name power_outage_events → power_outages
docs/PowerOutageNotifications.md	Update to describe pipeline + subscriptions
docs/NotificationFiltersPlan.md	Mark superseded by subscription model where overlapping
docs/MigrationStatus.md	Record pipeline completion
README.md	Link to usage guide

G5. Operational cleanup

• Ledger retention: Scheduled job deletes notification_delivery_ledger rows older than N days (default 180). Feed history tables are never trimmed by this job.
• Expired user subscriptions: Background job sets is_active = false where expires_at < now.
• Inactive subscription purge (optional): Hard-delete user subscriptions inactive > 1 year with no ledger entries in 90 days.

G6. Test cleanup

• Delete golden tests that reference deleted factories once renderers replace them.
• Keep regression tests for SyncPowerOutagesAndPublishEvents and SyncScheduledOutagesAndPublishEvents — update assertions to verify events emitted, not direct ntfy calls.
• Add architecture test: outage Application use cases must not reference INtfyNotificationService directly (except test/announcement escape hatches if documented).

G7. Definition of done

The pipeline implementation is not complete until:

• No outage monitor calls INtfyNotificationService directly
• Factory classes deleted; renderers are the only message builders for feeds
• System subscriptions seeded; global topics work through pipeline
• Ledger backfill verified against legacy *NotifiedAt columns
• manifest.json / ConfigGenerator aligned with new options
• Usage guide published and linked
• Analytics queries on power_outages / scheduled_outages still pass unchanged
• INotificationEventSourceResolver re-fetches latest row before every render (§27)
• PowerOutageChangeDetector does not re-emit NewOutage on PayloadHash-only poll updates (§27.6)

Phase D — User subscriptions (customizable outage notifications)

Goal: per-user custom filters on BENECO outage feeds (NotificationFiltersPlan.md).

1. Add CreateNotificationSubscription and related Application use cases under UseCases/Notifications/.
2. Generate beneco-outages-alerts-* topics on create (server-side only).
3. ProcessNotificationEvents already handles multiple subscriptions — user subs are just more rows.
4. Enforce security policy (see §11 below).
5. Rate limits on public API routes; IP hash until user auth exists.

Deferred until user accounts exist: MaxSubscriptionsPerUser = 2 (see §11.4).

Phase E — Email provider for outage events

1. EmailNotificationDeliveryRouter queues rendered messages.
2. Require verified email on user subscriptions.
3. System subscriptions stay ntfy-only unless explicitly configured.

Phase F — New feed sources

Adding a new source (example: beneco.water-outages):

1. Core: register feed in NotificationFeedCatalog.
2. Infrastructure: implement IWaterOutageFeedClient (or generic INotificationFeedAdapter).
3. Application: CheckWaterOutagesAndNotify emits events.
4. API: optional worker registration.
5. Optionally seed a system subscription + allow user filters on the new feed.

No changes to ProcessNotificationEvents, matchers, or delivery routers.

---

8. Database Schema

notification_subscriptions

Column	Type	Notes
id	bigint PK
public_id	varchar, unique, nullable	User-facing opaque id; null for system
user_id	varchar, nullable	Set when auth exists; enforces per-user limits
display_name	varchar
feed_id	varchar	e.g. beneco.power-outages
filter_json	jsonb	NotificationFilter serialized
provider	varchar	Ntfy, Email
destination	varchar	Topic or email
scope	varchar	System, User
is_active	bool
cooldown_minutes	int
last_sent_at	timestamptz, nullable
created_at	timestamptz
created_by_ip_hash	varchar, nullable	User subs only
expires_at	timestamptz, nullable

Indexes: (feed_id, is_active), (scope), (public_id), (user_id) where user_id is not null.

Unique (when auth enabled): one active user subscription per user per feed — (user_id, feed_id) where scope = 'User' and is_active = true.

notification_delivery_ledger

Column	Type	Notes
id	bigint PK
subscription_id	bigint FK
feed_id	varchar
source_key	varchar
event_type	varchar
sent_at	timestamptz

Unique: (subscription_id, feed_id, source_key, event_type).

---

9. Configuration

"NotificationPipeline": {
  "Enabled": true,
  "SystemSubscriptions": {
    "PowerOutages": {
      "FeedId": "beneco.power-outages",
      "NtfyTopic": "beneco-power-outages"
    },
    "ScheduledOutages": {
      "FeedId": "beneco.scheduled-outages",
      "NtfyTopic": "beneco-scheduled-outages"
    }
  },
  "UserTopicPrefix": "beneco-outages-alerts-",
  "DefaultCooldownMinutes": 30,
  "MaxNotificationsPerSubscriptionPerDay": 50
},
"NotificationSubscriptions": {
  "MaxPerIp": 3,
  "MaxPerIpPerDay": 3,
  "MaxKeywordsPerField": 5,
  "AnonymousTtlDays": 90,
  "MaxSubscriptionsPerUser": 2
}

MaxSubscriptionsPerUser is enforced for authenticated user subscriptions. IP hashes remain useful for auditing, but the active subscription cap is account-owned.

Legacy PowerOutages:NtfyTopic and ScheduledOutages:NtfyTopic are removed once Phase B is complete; ConfigGenerator emits the NotificationPipeline:SystemSubscriptions:* keys.

---

10. API Surface (User Subscriptions)

Same as filters plan, renamed to match unified model:

Method	Route	Purpose
GET	/api/notification-subscriptions	List current user’s subscriptions
POST	/api/notification-subscriptions	Create user subscription
GET	/api/notification-subscriptions/{publicId}	Get subscription
PUT	/api/notification-subscriptions/{publicId}	Update filter/target
DELETE	/api/notification-subscriptions/{publicId}	Delete
POST	/api/notification-subscriptions/{publicId}/test	Test delivery
GET	/api/notification-feeds	List feeds, event types, filterable fields

Request ties explicitly to a feed:

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

Response includes generated topic and subscribe URL when provider is ntfy:

{
  "publicId": "ns_k8f3p92m",
  "displayName": "Home - Bakakeng",
  "subscribeUrl": "https://ntfy.example.com/beneco-outages-alerts-k8f3p92mxq",
  "feedId": "beneco.power-outages",
  "isActive": true
}

---

11. User customizable outage notifications (architecture & security)

User subscriptions let someone define what outage events they care about and how they receive them (ntfy topic or email). They are implemented as notification_subscriptions rows with scope = User — not a separate filters system.

11.1 Architecture alignment

Follow Architecture.md and ModuleRules.md:

Layer	Responsibility	User subscription touchpoints
Core (Modules/Notifications)	NotificationFilter, NotificationFilterMatcher, NotificationTopicPolicy, subscription domain	Pure match rules; topic allowlist; no EF, no HTTP
Application (UseCases/Notifications)	CreateNotificationSubscription, Update…, Get…, Delete…, ProcessNotificationEvents	Owns validation, limits, topic generation orchestration
Infrastructure	DatabaseNotificationSubscriptionStore, topic generator, delivery routers	Persists subscriptions; generates beneco-outages-alerts-*; never exposes ntfy publish token
Obaki.Api	NotificationSubscriptionEndpoints	Maps HTTP → Application only; rate limits; no Core references

Dependency rules:

• API → Application → Core contracts. API does not call stores or ntfy directly.
• Monitors (SyncPowerOutagesAndPublishEvents, etc.) persist feed state, ask source-owned policies what changed, map those changes to NotificationEvents, and do not know about user subscriptions.
• User subscription CRUD does not live under UseCases/PowerOutages/ — Notifications owns the callable surface.

Allowed feeds for user subscriptions (MVP):

Feed ID	User can subscribe?
beneco.power-outages	Yes
beneco.scheduled-outages	Yes
Future feeds	Opt-in per feed in NotificationFeedCatalog.AllowsUserSubscriptions
Contact / internal	No

11.2 Security policy

Rule	Enforcement
Users cannot publish to ntfy	ntfy ACL: anonymous read-only on beneco-outages-alerts-*; only api_publisher writes
Users cannot choose topic names	API generates beneco-outages-alerts-{random}; reject client-supplied destination for ntfy
API cannot publish to arbitrary topics	NotificationTopicPolicy + NtfyNotificationService allowlist
Topic names are subscription secrets	No barangay, address, or real name in topic string
Authenticated create endpoint is rate-limited	ASP.NET rate limiter on POST /api/notification-subscriptions
Opaque publicId for owned subscription routes	Non-guessable id; not the ntfy topic

Topic generation (Infrastructure):

public static string GenerateUserOutageAlertTopic(string prefix)
{
    var id = Convert.ToHexString(RandomNumberGenerator.GetBytes(8)).ToLowerInvariant();
    return $"{prefix}{id}"; // beneco-outages-alerts-k8f3p92mxq
}

11.3 Subscription limits — two phases

Phase D1 — Before user accounts (current)

No user_id yet. Abuse controls use IP hash (salted, no raw IP stored):

Limit	Default	Config key
Max active subscriptions per IP	3	NotificationSubscriptions:MaxPerIp
Max creates per IP per day	3	NotificationSubscriptions:MaxPerIpPerDay
Anonymous subscription TTL	90 days	NotificationSubscriptions:AnonymousTtlDays

Manage subscription via publicId returned at create (link-in-email / bookmark model until auth).

Phase D2 — After user accounts (TODO)

When auth exists, attach user_id to subscriptions and enforce:

Limit	Default	Config key
Max active subscriptions per user	2	NotificationSubscriptions:MaxSubscriptionsPerUser
Max per feed per user	1	Enforced via unique index (user_id, feed_id) for active user subs

Policy: One authenticated user gets two active outage notification subscriptions by default, enough for one live outage alert and one scheduled outage alert. Raising the limit remains a config or admin override, not a client choice.

IP hashes remain stored for abuse investigation, but the active subscription cap is enforced per authenticated user.

11.4 End-to-end user flow

1. Authenticated user POST /api/notification-subscriptions { feedId, filter, delivery }
2. Application validates filter + per-user limits
3. Infrastructure generates beneco-outages-alerts-{random}
4. Row saved: scope=User, filter_json, destination=topic, user_id={authenticated key}
5. Response: publicId + subscribeUrl
6. User subscribes in ntfy app (read-only)
7. Monitor maps source change to NotificationEvent → ProcessNotificationEvents
8. Matcher runs user + system subscriptions → deliver → ledger

User updates filter via PUT /api/notification-subscriptions/{publicId}. The API requires authentication and matches the subscription to the authenticated user key.

11.5 What user subscriptions must not do

• Call INtfyNotificationService from the API or frontend
• Accept destination for ntfy from the client (server-generated only)
• Bypass ProcessNotificationEvents for delivery
• Store ntfy PublishToken on subscription rows
• Create subscriptions for feeds not marked AllowsUserSubscriptions

11.6 Tests required for user subscriptions

Test	Layer
Topic always matches beneco-outages-alerts-*	Application
Client-supplied ntfy topic rejected	Application
MaxPerIp blocks excess creates	Application
MaxSubscriptionsPerUser = 2 when user_id set	Application + Infrastructure
NotificationTopicPolicy rejects evil-topic	Core
Architecture: API does not reference INtfyNotificationService for subscription CRUD	ArchitectureTests

---

12. How Existing Topics Map

Today	After unified pipeline
NotificationPipeline:SystemSubscriptions:PowerOutages:NtfyTopic	System subscription: feed beneco.power-outages, empty filter, ntfy destination
NotificationPipeline:SystemSubscriptions:ScheduledOutages:NtfyTopic	System subscription: feed beneco.scheduled-outages, empty filter, ntfy destination
NotificationPipeline:TestNtfyTopic	Test endpoint creates ephemeral subscription or overrides destination
User custom filter (future)	User subscription: custom filter, generated beneco-outages-alerts-* topic
Contact ntfy ping	Out of scope for v1 pipeline — contact owns its workflow; revisit if contact becomes a registered feed

Contact stays separate initially. The pipeline is optimized for recurring polled feeds with filterable structured events, not one-off form submissions.

---

13. Testing Strategy

Layer	Tests
Core	NotificationFilterMatcher across feeds and event types
Core	NotificationTopicPolicy allowlist
Application	ProcessNotificationEvents: match, skip dedup, cooldown, multi-subscription fan-out
Application	Renderers match legacy factory output (golden files)
Infrastructure	Ledger unique constraint; subscription CRUD
API	Create subscription, test endpoint, rate limits

Critical regression test: Run existing SyncPowerOutagesAndPublishEventsTests and SyncScheduledOutagesAndPublishEventsTests unchanged after Phase B — same notifications, same topics, same dedup behavior.

---

14. Data ownership (analytics-safe)

Two layers must stay separate after implementation:

FEED / ANALYTICS LAYER          NOTIFICATION LAYER
──────────────────────          ──────────────────
power_outages                   notification_subscriptions
scheduled_outages               notification_delivery_ledger
scheduled_outage_digest_runs
scheduled_outage_reminder_runs

"What happened?"                "Who was told, when, via what channel?"

Monitors always persist feed state first, then emit NotificationEvents. Analytics dashboards should query feed tables only. Notification tables are for delivery audit and user subscription management.

---

15. Review gaps addressed (second pass)

Items missing from the first draft, now explicit:

Gap	Resolution
No mandatory cleanup phase	Phase G checklist (code, config, tables, scripts, tests)
Ledger vs legacy dedup transition	Phase C backfill + dual-write before column removal
Analytics impact	Feed tables untouched; *NotifiedAt kept for reporting
Digest run tables role	Stay as event gating (emit or skip), not delivery dedup
WatchedAreas dead config	Removed in Phase G; replaced by subscriptions
Test / announcement endpoints	Refactored to use pipeline, not deleted blindly
ImportHistoricalPowerOutages stale table name	Fixed in Phase G
Ledger growth	Retention job in G5
INotificationRenderer registration	INotificationRendererRegistry resolves all INotificationRenderer implementations from DI
Multi-provider per subscription	One subscription = one provider + destination; multiple rows for ntfy + email on same filter
Failure handling	Failed delivery does not write ledger row; monitor can retry on next poll
NotificationFiltersPlan overlap	User filters are user-scoped notification_subscriptions; topic prefix beneco-outages-alerts-*
Fast poll vs slow delivery	Signal at detect, fresh render at send via INotificationEventSourceResolver (§27)
Spam from minute-by-minute polls	Emit rules §27.2 + ledger; PayloadHash alone does not re-emit NewOutage
Stale notification body	Never queue rendered text; re-read feed row before Render

---

16. Suggested Implementation Order

Phase	Effort	Outcome
A Extract events + renderers	2–3 days	Shared types; no behavior change
B Pipeline + system subscriptions	4–5 days	Global topics via pipeline
C Ledger backfill + dual-write	1–2 days	Safe transition from legacy dedup
D User subscriptions API	3–4 days	Custom per-user filters
E Email delivery	2–3 days	Email as a provider
F New feed template	1–2 days per feed	Prove extensibility
G Cleanup	2–3 days	Remove dead code, align config, verify analytics

Recommended first milestone: Phase A + B + G (partial). Ship the pipeline for system topics and delete factories in the same release to avoid two message-building paths.

Phase D adds the product feature from NotificationFiltersPlan.md without a second dispatch system.

---

17. Design Decisions

Decision	Choice	Rationale
Pipeline location	Application orchestrator, Core types	Matches Architecture.md
Subscriptions vs filters	Subscription = filter + target	One model for system and user
Feed ID as string key	beneco.power-outages	Extensible without enum explosion
Renderers per feed	Application classes	Feed-specific message formatting
Routers per provider	Infrastructure	Provider details stay in Infrastructure
Contact in pipeline?	Defer	Different trigger model (form submit)
Keep *NotifiedAt columns?	Keep for analytics	Optional stop writing after ledger trusted
User topic prefix	beneco-outages-alerts-*	Matches BENECO branding; distinct from system topics
Subscriptions per user	1 (default, post-auth)	Configurable; enforce when user_id exists
Digest events	First-class NotificationEvent	Same filter/dedup path as real-time
Render timing	At delivery, not at detect	Latest DB row when fan-out is slow (§27)
ntfy SequenceId	{event}-{sourceKey} only	No poll counter; optional ntfy dedup

---

18. Success Criteria

• Power and scheduled outage monitors emit events; they do not call ntfy directly.
• System subscriptions reproduce today’s global topic behavior exactly.
• Adding a user subscription requires no monitor code changes.
• Adding a new feed requires: adapter + renderer + catalog entry — not a new delivery stack.
• ntfy and email are interchangeable per subscription via NotificationDeliveryTarget.
• User subscriptions use beneco-outages-alerts-* topics; clients cannot set ntfy destinations.
• Post-auth: two active subscriptions per user by default (MaxSubscriptionsPerUser = 2), covering live and scheduled outage alert types.
• Slow delivery still sends latest area/status text (fresh resolver), not detect-time snapshot.
• Minute-by-minute polls do not spam: one ledger row per (subscription, source, event type).
• All matching and dedup logic is testable without HTTP or database (Core + Application unit tests).

---

19. Files to Create / Modify (Summary)

Layer	Action
Core	Extend Modules/Notifications with event, filter, subscription, matcher, contracts; add source-owned change policies in feed modules
Application	ProcessNotificationEvents, renderers, notification-event mappers, refactor both Check*AndNotify
Infrastructure	Subscription + ledger stores, delivery routers, migration, seed system subs
API	NotificationSubscriptionEndpoints, register pipeline options
Tests	Matcher, pipeline fan-out, renderer golden tests, existing monitor regression
Docs	Update PowerOutageNotifications.md; add notification-pipeline-usage.md

This unified pipeline is the foundation. NotificationFiltersPlan.md describes the user-facing feature; notification-pipeline-usage.md is the developer playbook for new feeds and providers.

---

20. Failure modes and retries

Scenario	Behavior
ntfy publish fails	Log warning; no ledger row; retry on next poll if signal still valid and ledger empty
Email outbox enqueue fails	Same; outbox worker retries independently once queued
No matching subscriptions	Event dropped silently (debug log)
All subscriptions in cooldown	Skip; do not write ledger
Pipeline disabled (NotificationPipeline:Enabled=false)	Monitors still persist feed state; no delivery
Poll updates area while delivery in progress	INotificationEventSourceResolver renders latest row at send time (§27.3)
Same outage every poll	Detector emits signal once; ledger blocks repeat delivery (§27.2)
PayloadHash changes on ongoing outage	DB updated each poll; no new NewOutage signal (§27.2)
Source row deleted before send	Resolver returns null; skip without ledger write

---

21. Multi-channel example

One user wants the same filter on ntfy and email — create two subscriptions with the same filter_json:

Subscription 1: feed=beneco.power-outages, filter={...}, provider=Ntfy,  destination=beneco-outages-alerts-abc
Subscription 2: feed=beneco.power-outages, filter={...}, provider=Email, [email protected]

Each has its own ledger rows and cooldown. Do not combine providers into one subscription row.

---

22. Pragmatic optimizations

These are defaults for v1 — enough to avoid obvious waste on a small VPS without over-engineering.

22.1 Pipeline hot path

Optimization	Implementation
Skip pipeline when empty	if (events.Count == 0) return at top of ProcessAsync
Skip when disabled	NotificationPipeline:Enabled=false still persists feed state
Load subscriptions once per feed	GetActiveByFeedAsync(feedId) inside feed group loop, not per event
Index-backed subscription query	DB index on (feed_id, is_active) — see §8
Ledger existence: single-key lookup	Sufficient at current scale; batch IN query only if profiling shows need
Renderer resolved once per event	Outside the foreach (subscription) loop
Sequential ntfy publishes	No Task.WhenAll storm; typical poll emits 0–3 events
No cross-poll subscription cache	Simplicity wins until > 500 active user subscriptions

22.2 When to optimize further

Signal	Action
> 500 active user subscriptions	In-memory cache of GetActiveByFeedAsync with 30–60s TTL
Ledger table > 1M rows	Run retention job (§G5); index on sent_at
ntfy latency blocks poll	Background delivery channel — not in v1
Same event matches 100+ subs	Render once, fan-out delivery (unlikely for BENECO MVP)

22.3 Required indexes (include in Phase B migration)

CREATE INDEX ix_notification_subscriptions_feed_id_is_active
  ON notification_subscriptions (feed_id, is_active);

CREATE UNIQUE INDEX ux_notification_delivery_ledger_dedup
  ON notification_delivery_ledger (subscription_id, feed_id, source_key, event_type);

22.4 Config validation at startup

Bind NotificationPipelineOptions and NotificationSubscriptionsOptions with ValidateOnStart():

• UserTopicPrefix non-empty, ends with -, valid ntfy charset
• MaxPerIp > 0, DefaultCooldownMinutes >= 0
• System subscription topics pass NotificationTopicPolicy

22.5 Dual-write during transition (Phase B–C)

Keep writing OutageNotifiedAt / RestoredNotifiedAt in monitors until ledger backfill is verified. Stops duplicate notifications if pipeline regresses. Stop dual-write in Phase G only.

---

23. Implementation playbook (detailed)

23.1 Power outage event detection

Create source-owned transition rules in src/Obaki.Core/Modules/PowerOutages/Policies/PowerOutageChangeDetector.cs. Application mapping lives in src/Obaki.Application/UseCases/PowerOutages/PowerOutageNotificationEvents.cs.

Condition	PowerOutageChangeKind	NotificationEventType	SourceKey
New row, outage.IsOngoing	NewOutage	NewOutage	outage.SourceId.ToString()
New row, outage.HasTimeRestored	RestoredAnnouncement	RestoredAnnouncement	same
Existing, ongoing, existing.OutageNotifiedAt is null	NewOutage	NewOutage	same
Existing, restored, FirstSeenRestored && OutageNotifiedAt is null	RestoredAnnouncement	RestoredAnnouncement	same
Existing, restored, else	Restoration	Restoration	same

Pass linkedScheduledOutageId through notification fields so the renderer can set “Scheduled Maintenance Has Begun” when non-null.

Only emit when current code would call Publish*Async — not on every poll.

Do not emit on PayloadHash / field-only changes for an already-notified ongoing outage (§27.2).

// 1. Persist feed state first (every poll)
await PersistFeedFromPollAsync(...);

// 2. Detect source changes, then map to notification signals
var changes = PowerOutageChangeDetector.Detect(feedItems, existingBySourceId);
var events = PowerOutageNotificationEvents.FromChanges(changes, todaysScheduled, now);

// 3. Pipeline: ledger + fresh render at send time
if (events.Count > 0)
{
    var pipelineResult = await processNotificationEvents.ProcessAsync(events, cancellationToken);
    await ApplyLegacyNotifiedMarkersAsync(events, pipelineResult, now, cancellationToken);
}

23.2 Scheduled outage event detection

Create source-owned scheduled rules in Core:

ScheduledOutageChangeDetector.cs
ScheduledOutageDigestPolicy.cs
ScheduledOutageReminderPolicy.cs

Application maps those source decisions to NotificationEvent in ScheduledOutageNotificationEvents.cs.

Condition	Source rule	NotificationEventType	SourceKey
New item in diff	ScheduledOutageChangeKind.New	NewEntry	item.Id.ToString()
Changed item	ScheduledOutageChangeKind.Updated	Updated	same
Newly cancelled	ScheduledOutageChangeKind.Cancelled	Cancelled	same
Weekly digest hour + not sent	ScheduledOutageDigestPolicy	WeeklyDigest	weekly-{yyyy-MM-dd}
Daily digest hour + not sent	ScheduledOutageDigestPolicy	DailyUpdate	daily-{yyyy-MM-dd}
Day-before reminder + not sent	ScheduledOutageReminderPolicy	DayBeforeReminder	reminder-{yyyy-MM-dd}

Digest/reminder date selection is source-owned, but persistence gates still use scheduled_outage_digest_runs / reminder_runs before final emit/marking.

23.3 Store contracts (copy into Core)

INotificationSubscriptionStore.cs

Task<IReadOnlyList<NotificationSubscription>> GetActiveByFeedAsync(
    NotificationFeedId feedId, CancellationToken cancellationToken = default);
Task<NotificationSubscription?> GetByPublicIdAsync(string publicId, CancellationToken cancellationToken = default);
Task<IReadOnlyList<NotificationSubscription>> GetByUserIdAsync(string userId, CancellationToken cancellationToken = default);
Task<NotificationSubscription> CreateAsync(NotificationSubscription subscription, CancellationToken cancellationToken = default);
Task UpdateAsync(NotificationSubscription subscription, CancellationToken cancellationToken = default);
Task<int> CountActiveByIpHashAsync(string ipHash, CancellationToken cancellationToken = default);
Task<int> CountActiveByUserIdAsync(string userId, CancellationToken cancellationToken = default);
Task UpdateLastSentAtAsync(long id, DateTimeOffset sentAt, CancellationToken cancellationToken = default);

// INotificationDeliveryLedgerStore.cs
Task<bool> ExistsAsync(DeliveryLedgerKey key, CancellationToken cancellationToken = default);
Task RecordAsync(DeliveryLedgerKey key, DateTimeOffset sentAt, CancellationToken cancellationToken = default);
Task<int> CountSinceAsync(long subscriptionId, DateTimeOffset since, CancellationToken cancellationToken = default);

23.4 DI registration checklist

src/Obaki.Application/DependencyInjection.cs:

services.AddScoped<ProcessNotificationEvents>();
services.AddScoped<INotificationRenderer, PowerOutageNotificationRenderer>();
services.AddScoped<INotificationRenderer, ScheduledOutageNotificationRenderer>();
services.AddScoped<NotificationRendererRegistry>();
// Phase D+: CreateNotificationSubscription, etc.

src/Obaki.Infrastructure/NotificationPipelineExtensions.cs:

services.AddScoped<INotificationSubscriptionStore, DatabaseNotificationSubscriptionStore>();
services.AddScoped<INotificationDeliveryLedgerStore, DatabaseNotificationDeliveryLedgerStore>();
services.AddScoped<INotificationDeliveryRouter, NtfyNotificationDeliveryRouter>();
services.AddSingleton<NotificationTopicPolicy>();
services.AddOptions<NotificationPipelineOptions>().Bind(...).ValidateOnStart();

Call from Infrastructure/DependencyInjection.cs: services.AddNotificationPipeline(configuration);

23.5 Phase D steps (user subscriptions)

Step	File	Action
D1	CreateNotificationSubscription.cs	Validate filter, feed AllowsUserSubscriptions, IP limits
D2	NotificationTopicGenerator.cs	beneco-outages-alerts-{16 hex chars}
D3	CreateNotificationSubscription.cs	Reject client delivery.destination when provider is Ntfy
D4	NotificationSubscriptionEndpoints.cs	Routes §10; public path + rate limiter
D5	Program.cs	MapNotificationSubscriptionEndpoints()

publicId format: ns_ + 12 url-safe chars (e.g. ns_k8f3p92mxq).

23.6 Phase E steps (email)

Step	Action
E1	EmailNotificationDeliveryRouter → queue EmailOutboxItem, template notification-pipeline/default
E2	Infrastructure/Templates/Email/NotificationPipelineDefault.html
E3	Email subs require verification before is_active=true (when auth exists)

23.7 Test/announcement refactor (Phase G)

• SendPowerOutageTestNotifications → build test events, call ProcessNotificationEvents with test topic from options
• SendPowerOutageAnnouncement → single ManualAnnouncement event; add handler in PowerOutageNotificationRenderer

---

24. Validation commands (run after every phase)

dotnet build Obaki.slnx -c Debug
dotnet test tests/Obaki.Core.Tests/Obaki.Core.Tests.csproj -c Debug --no-restore
dotnet test tests/Obaki.Application.Tests/Obaki.Application.Tests.csproj -c Debug --no-restore
dotnet test tests/Obaki.ArchitectureTests/Obaki.ArchitectureTests.csproj -c Debug --no-restore
dotnet test tests/Obaki.Api.Tests/Obaki.Api.Tests.csproj -c Debug --no-restore

With Docker (after Phase B persistence):

$env:RUN_DOCKER_TESTS = "1"
dotnet test tests/Obaki.Infrastructure.IntegrationTests/Obaki.Infrastructure.IntegrationTests.csproj -c Debug --no-restore

Manual smoke:

dotnet run --project src/Obaki.Api
Invoke-RestMethod -Method Post http://localhost:5195/api/power-outages/test-notifications

---

25. PR checklist (copy per phase)

- [ ] Core: BCL only, no EF/HTTP
- [ ] API maps HTTP → Application only (no Core reference)
- [ ] Flat UseCases/Notifications/ layout
- [ ] Subscriptions loaded per feed, not per event
- [ ] Failed delivery → no ledger row
- [ ] ntfy topics pass NotificationTopicPolicy
- [ ] User topics: beneco-outages-alerts-* only (server-generated)
- [ ] Feed state saved before ProcessNotificationEvents
- [ ] Analytics tables (power_outages, scheduled_outages) unchanged
- [ ] Tests at correct layer; fast dotnet test path green
- [ ] manifest.json + ConfigGenerator if new options
- [ ] §27: fresh render via INotificationEventSourceResolver before delivery
- [ ] §27: no NewOutage re-emit on PayloadHash-only updates

---

26. Troubleshooting during implementation

Problem	Likely cause	Fix
Duplicate notifications	Ledger empty; legacy markers removed early	Phase C backfill; restore dual-write
No notifications	Pipeline disabled or missing system subscription rows	Check seed + NotificationPipeline:Enabled
ntfy 403	ACL or policy	Update ntfy + NotificationTopicPolicy
Golden test failure	Renderer drift	Fix renderer to match factory output
Architecture test fail	Monitor still uses INtfyNotificationService	Route through pipeline
User sub never fires	Wrong feed_id or filter	Debug-log subscription match count
Subscriber got old area text	Render used stale SourcePayload	Ensure INotificationEventSourceResolver runs before Render (§27)
Five pings for same outage	Detector re-emits each poll	Fix detect rules; confirm ledger writes on success

---

27. Fast poll, slow delivery — fresh data and anti-spam

This section is required behavior for BENECO monitors (e.g. poll every 60s, fan-out to many subscriptions may take minutes).

Developer doc: notification-poll-vs-delivery.md — standalone explanation of this concern and code hooks.

27.1 Problem

T+0:00  Poll — new outage 42 detected, signal emitted
T+0:00  Pipeline starts (many subscriptions; slow ntfy)
T+1:00  Poll — area/cause updated in power_outages
T+2:00  Poll — status refreshed
T+3:00  Sub 1 receives notification

Without rules, subscribers could get spam (one push per poll) or stale text (body built at T+0:00).

27.2 Three layers (all required)

flowchart TB
    Poll[Poll every 1 min] --> Persist[(power_outages updated)]
    Persist --> Detect[Emit signal once per state change]
    Detect --> Pipeline[ProcessNotificationEvents]
    Pipeline --> Ledger{Ledger exists?}
    Ledger -->|yes| Skip[Skip — no spam]
    Ledger -->|no| Fresh[INotificationEventSourceResolver]
    Fresh --> Render[Render title/body from latest row]
    Render --> Deliver[ntfy / email]
    Deliver --> WriteLedger[Write ledger]

Layer	Where	Rule
1. Source change rules	PowerOutageChangeDetector, ScheduledOutageChangeDetector, digest/reminder policies, then *NotificationEvents mappers	Signal only on real transitions (new ongoing, restoration, etc.). Never re-emit NewOutage because PayloadHash, area, or status changed on an already-notified ongoing row.
2. Ledger dedup	notification_delivery_ledger	Unique (subscription_id, feed_id, source_key, event_type). Check before render.
3. Fresh render	INotificationEventSourceResolver	Immediately before Render, reload latest row from feed store by SourceKey.

27.3 Signal vs snapshot

	Signal (NotificationEvent)	Snapshot (rendered message)
Contains	FeedId, EventType, SourceKey	Title, body, tags
Created	Source change detect + Application mapping (after persist)	At delivery, from latest DB row
Queued?	Pass signal list to pipeline only	Never queue rendered text async

// INotificationEventSourceResolver.cs (Application)
public interface INotificationEventSourceResolver
{
    Task<NotificationEvent?> ResolveAsync(
        NotificationEvent signal,
        CancellationToken cancellationToken = default);
}

// PowerOutageNotificationEventSourceResolver
public async Task<NotificationEvent?> ResolveAsync(NotificationEvent signal, CancellationToken ct)
{
    if (signal.FeedId.Value != "beneco.power-outages")
        return signal;

    if (!int.TryParse(signal.SourceKey, out var sourceId))
        return signal; // digest keys — use signal as-is

    var rows = await _powerOutageStore.GetBySourceIdsAsync([sourceId], ct);
    if (!rows.TryGetValue(sourceId, out var row))
        return null;

    return signal with
    {
        Fields = PowerOutageNotificationFields.From(row),
        SourcePayload = row
    };
}

27.4 Poll vs delivery timing

Activity	Interval	Notes
Feed poll + DB persist	e.g. 60s (PollingIntervalSeconds)	Always runs; keeps analytics current
Signal detect	Same as poll	Cheap; no ntfy call
Pipeline / deliver	Same call as detect in v1	Fan-out may take minutes; freshness handled by §27.3

Optional later: separate DeliveryIntervalSeconds to batch signals — if added, on delivery tick re-scan DB for pending candidates (ledger miss), do not rely on in-memory events only. Not required for v1.

27.5 ntfy SequenceId rules

Stable per logical event (already used in factories):

power-outage-{sourceId}
power-restored-{sourceId}
power-restored-announced-{sourceId}

• Do not include poll count, timestamp, or PayloadHash in sequence id.
• ntfy may collapse duplicate sequence publishes; ledger remains source of truth.

27.6 What does not trigger a new notification

DB change on existing source_id	New signal?
area / cause / status text updated (ongoing)	No
PayloadHash changed (ongoing, already notified)	No
last_seen_at refreshed	No
time_restored set	Yes → Restoration or RestoredAnnouncement (once each)
New source_id	Yes → NewOutage

To notify on field updates later, add explicit Updated event type + separate ledger key (not in MVP).

27.7 Tests required (§27)

Test	Assert
PowerOutageChangeDetector_does_not_emit_when_only_payload_hash_changes	No source change when ongoing + already notified
ProcessNotificationEvents_uses_latest_row_at_render	DB area updated between signal and send → body matches new area
ProcessNotificationEvents_skips_when_ledger_exists	Second call does not deliver
Resolver_returns_null_when_row_deleted	No ledger write

27.8 PR checklist additions

- [ ] `PowerOutageChangeDetector` does not re-emit NewOutage on PayloadHash-only poll updates
- [ ] INotificationEventSourceResolver registered and called before every Render
- [ ] Ledger checked before Render (not after)
- [ ] SequenceId stable per source + event type (no poll suffix)