Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 10 additions & 13 deletions .agents/skills/notification/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -115,7 +115,8 @@ type Service interface {
Every event mutating operation:
1. Validates input (rule exists, rule is not disabled, payload matches type)
2. Persists event + per-channel delivery statuses atomically via adapter
3. Dispatches asynchronously via `EventHandler.Dispatch()`

The service does NOT trigger delivery: events are written as `PENDING` and picked up by the background `Reconcile()` loop. `EventHandler.Dispatch` exists on the interface but has no call sites.

Reference: `openmeter/notification/service.go`, `openmeter/notification/service/event.go`

Expand All @@ -130,16 +131,12 @@ Consumer handler (consumer/)
Service.CreateEvent()
├── Validates rule + payload
├── adapter.CreateEvent() → writes notification_event + delivery_status rows
└── EventHandler.Dispatch() → async goroutine (30s timeout)
reconcileEvent() → reconcileWebhookEvent()
├── webhook.SendMessage() → Svix API
└── Updates delivery status based on Svix response
└── adapter.CreateEvent() → writes notification_event + PENDING delivery_status rows
Background Reconcile() loop (every 15s)
Background Reconcile() loop (every 15s, started only in cmd/server)
├── Lists PENDING/SENDING/RESENDING delivery statuses
├── Fetches status from Svix for each
├── PENDING → webhook.SendMessage() → Svix API
├── SENDING → fetches message/attempt status from Svix
└── Updates to SUCCESS/FAILED or retries
```

Expand Down Expand Up @@ -296,23 +293,23 @@ Reference: `openmeter/notification/defaults.go`

Two wire sets in `app/common/notification.go`:

- `Notification` — full production wiring (Svix webhook handler + real event handler with reconciliation loop)
- `NotificationService` — service-only wiring with no-op webhook and event handler (used by non-notification services like `cmd/server`)
- `Notification` — full production wiring (Svix webhook handler + event handler with reconciliation loop). Used by `cmd/server`, the only binary that starts the reconcile loop and the only one whose call paths (channel/rule CRUD over HTTP) reach the webhook handler. `cmd/balance-worker` and `cmd/jobs` also use this set: wire prunes the event handler there, but they still construct the real Svix webhook handler.
- `NotificationService` — service-only wiring with a no-op webhook handler, for binaries that only need `notification.Service` without Svix. Used by `cmd/notification-service`: its consumer only calls `CreateEvent` (persist pending events), so its startup must not depend on Svix availability.

```go
func NewNotificationService(
logger *slog.Logger,
adapter notification.Repository,
webhook notificationwebhook.Handler,
eventHandler notification.EventHandler,
featureConnector feature.FeatureConnector,
) (notification.Service, error)
```

The `cmd/notification-service/` standalone worker wires the full consumer + Kafka subscriber + production Svix handler.
The `cmd/notification-service/` standalone worker runs only the Kafka consumer + telemetry server; webhook delivery and reconciliation are `cmd/server`'s job.

## Non-Obvious Pitfalls

- **Constructing the Svix webhook handler is not side-effect-free.** `webhooksvix.New` registers `NotificationEventTypes` with the Svix API at construction time and fails startup on error unless `notification.webhook.skipEventTypeRegistrationOnError` is set. Any binary wired with the real handler therefore requires Svix reachability at boot, even if it never sends a message.
- **Rules and channels are NOT auto-disabled after delivery failures.** Svix retries for up to 48h; after that the delivery status is `FAILED` but the rule stays active for future events.
- **Balance threshold dedup is rule-scoped.** The dedup hash includes `ruleID` — if a second rule is added for the same event type, it will independently trigger (no cross-rule dedup).
- **Invoice events skip `gathering` status.** The consumer explicitly checks `event.Invoice.Status` and skips if the invoice is still being assembled.
Expand Down
24 changes: 13 additions & 11 deletions cmd/notification-service/wire.go
Original file line number Diff line number Diff line change
Expand Up @@ -27,15 +27,15 @@ type Application struct {
common.GlobalInitializer
common.Migrator

BrokerOptions watermillkafka.BrokerOptions
EventPublisher eventbus.Publisher
EntClient *db.Client
FeatureConnector feature.FeatureConnector
Logger *slog.Logger
MessagePublisher message.Publisher
Meter metric.Meter
Tracer trace.Tracer
Metadata common.Metadata
BrokerOptions watermillkafka.BrokerOptions
EventPublisher eventbus.Publisher
EntClient *db.Client
FeatureConnector feature.FeatureConnector
Logger *slog.Logger
MessagePublisher message.Publisher
Meter metric.Meter
Tracer trace.Tracer
Metadata common.Metadata
MeterService meter.Service
Notification notification.Service
RuntimeMetricsCollector common.RuntimeMetricsCollector
Expand All @@ -55,11 +55,13 @@ func initializeApplication(ctx context.Context, conf config.Configuration) (Appl
common.Namespace,
common.NewDefaultTextMapPropagator,
common.NewKafkaTopicProvisioner,
common.Notification,
// This worker only persists pending notification events (consumer -> Service.CreateEvent);
// webhook delivery and reconciliation run in cmd/server. Use the noop-webhook service set
// so startup does not construct a Svix client or depend on Svix availability.
common.NotificationService,
common.NotificationServiceProvisionTopics,
common.ProgressManager,
common.Streaming,
common.NewSvixAPIClient,
common.Telemetry,
common.TelemetryLoggerNoAdditionalMiddlewares,
common.Watermill,
Expand Down
22 changes: 5 additions & 17 deletions cmd/notification-service/wire_gen.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading