diff --git a/.agents/skills/notification/SKILL.md b/.agents/skills/notification/SKILL.md index 4255770471..855ee8ce75 100644 --- a/.agents/skills/notification/SKILL.md +++ b/.agents/skills/notification/SKILL.md @@ -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` @@ -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 ``` @@ -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. diff --git a/cmd/notification-service/wire.go b/cmd/notification-service/wire.go index d709ac6b12..a838a692b0 100644 --- a/cmd/notification-service/wire.go +++ b/cmd/notification-service/wire.go @@ -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 @@ -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, diff --git a/cmd/notification-service/wire_gen.go b/cmd/notification-service/wire_gen.go index ec56d42785..61ffb08227 100644 --- a/cmd/notification-service/wire_gen.go +++ b/cmd/notification-service/wire_gen.go @@ -137,19 +137,7 @@ func initializeApplication(ctx context.Context, conf config.Configuration) (Appl cleanup() return Application{}, nil, err } - webhookConfiguration := notificationConfiguration.Webhook - v3 := conf.Svix - svix, err := common.NewSvixAPIClient(v3, meterProvider, tracerProvider) - if err != nil { - cleanup6() - cleanup5() - cleanup4() - cleanup3() - cleanup2() - cleanup() - return Application{}, nil, err - } - handler, err := common.NewNotificationWebhookHandler(logger, tracer, webhookConfiguration, svix) + handler, err := common.NewNoopNotificationWebhookHandler(logger) if err != nil { cleanup6() cleanup5() @@ -181,7 +169,7 @@ func initializeApplication(ctx context.Context, conf config.Configuration) (Appl } aggregationConfiguration := conf.Aggregation clickHouseAggregationConfiguration := aggregationConfiguration.ClickHouse - v4, cleanup7, err := common.NewClickHouse(ctx, clickHouseAggregationConfiguration, tracer, meter, logger) + v3, cleanup7, err := common.NewClickHouse(ctx, clickHouseAggregationConfiguration, tracer, meter, logger) if err != nil { cleanup6() cleanup5() @@ -215,7 +203,7 @@ func initializeApplication(ctx context.Context, conf config.Configuration) (Appl cleanup() return Application{}, nil, err } - connector, err := common.NewStreamingConnector(ctx, aggregationConfiguration, v4, logger, progressmanagerService, manager) + connector, err := common.NewStreamingConnector(ctx, aggregationConfiguration, v3, logger, progressmanagerService, manager) if err != nil { cleanup7() cleanup6() @@ -228,7 +216,7 @@ func initializeApplication(ctx context.Context, conf config.Configuration) (Appl } health := common.NewHealthChecker(logger) telemetryHandler := common.NewTelemetryHandler(metricsTelemetryConfig, health, logger) - v5, cleanup8 := common.NewTelemetryServer(telemetryConfig, telemetryHandler) + v4, cleanup8 := common.NewTelemetryServer(telemetryConfig, telemetryHandler) application := Application{ GlobalInitializer: globalInitializer, Migrator: migrator, @@ -245,7 +233,7 @@ func initializeApplication(ctx context.Context, conf config.Configuration) (Appl Notification: notificationService, RuntimeMetricsCollector: runtimeMetricsCollector, StreamingConnector: connector, - TelemetryServer: v5, + TelemetryServer: v4, } return application, func() { cleanup8()