From d39e8e0da3d30ae84e7f7f393cec6ad058695eec Mon Sep 17 00:00:00 2001 From: whoisasx Date: Mon, 1 Jun 2026 03:09:55 +0530 Subject: [PATCH] fix: address notifier review cleanup --- backend/internal/notification/retry.go | 18 ++- backend/internal/notification/settings.go | 9 +- .../internal/notification/settings_test.go | 14 ++- .../sqlite/notification_delivery_store.go | 103 +++++++++++------- backend/notifier_wiring.go | 4 +- 5 files changed, 93 insertions(+), 55 deletions(-) diff --git a/backend/internal/notification/retry.go b/backend/internal/notification/retry.go index cf62798c7..72c54775a 100644 --- a/backend/internal/notification/retry.go +++ b/backend/internal/notification/retry.go @@ -1,8 +1,9 @@ package notification import ( + crand "crypto/rand" + "encoding/binary" "math" - "math/rand" "strings" "time" @@ -30,7 +31,7 @@ func RetryPolicyFromConfig(cfg config.NotificationRetryConfig) RetryPolicy { LeaseTTL: settings.Retry.LeaseTTL, BatchSize: settings.Retry.BatchSize, Jitter: retryJitterFraction, - RandFloat64: rand.Float64, + RandFloat64: cryptoRandFloat64, } } @@ -69,13 +70,24 @@ func (p RetryPolicy) BackoffDelay(attempt int) time.Duration { } randFloat := p.RandFloat64 if randFloat == nil { - randFloat = rand.Float64 + randFloat = cryptoRandFloat64 } // rand in [0,1) -> factor in [1-jitter, 1+jitter) factor := 1 - p.Jitter + (2 * p.Jitter * randFloat()) return time.Duration(float64(delay) * factor) } +func cryptoRandFloat64() float64 { + var b [8]byte + if _, err := crand.Read(b[:]); err != nil { + // Fall back to a time-derived value only if the OS CSPRNG fails. The + // fallback still avoids math/rand's deterministic process seed. + return float64(time.Now().UnixNano()&((1<<53)-1)) / float64(1<<53) + } + // Match math/rand.Float64's 53 bits of precision in [0,1). + return float64(binary.BigEndian.Uint64(b[:])>>11) / float64(1<<53) +} + func (p RetryPolicy) NextAttemptAt(now time.Time, attempt int) time.Time { return now.Add(p.BackoffDelay(attempt)) } diff --git a/backend/internal/notification/settings.go b/backend/internal/notification/settings.go index 623e4a716..abb8b6531 100644 --- a/backend/internal/notification/settings.go +++ b/backend/internal/notification/settings.go @@ -16,11 +16,7 @@ type staticSettings struct { } func SettingsFromConfig(cfg config.Config) SettingsProvider { - settings := cfg.Notifications - if isZeroNotificationConfig(settings) { - settings = config.DefaultNotificationConfig() - } - return staticSettings{cfg: NormalizeSettings(settings)} + return staticSettings{cfg: NormalizeSettings(cfg.Notifications)} } func StaticSettings(cfg config.NotificationConfig) SettingsProvider { @@ -35,9 +31,6 @@ func (s staticSettings) Settings(context.Context) config.NotificationConfig { // explicit route overrides, including an explicit empty route list. func NormalizeSettings(in config.NotificationConfig) config.NotificationConfig { def := config.DefaultNotificationConfig() - if isZeroNotificationConfig(in) { - return cloneSettings(def) - } out := in if isZeroDashboardConfig(in.Dashboard) { diff --git a/backend/internal/notification/settings_test.go b/backend/internal/notification/settings_test.go index f3a6884e3..bd854e988 100644 --- a/backend/internal/notification/settings_test.go +++ b/backend/internal/notification/settings_test.go @@ -9,15 +9,25 @@ import ( ) func TestSettingsFromConfigDefaultsWhenUnset(t *testing.T) { - got := SettingsFromConfig(config.Config{}).Settings(context.Background()) + got := SettingsFromConfig(config.Config{Notifications: config.DefaultNotificationConfig()}).Settings(context.Background()) if !got.Enabled || !got.Desktop.Enabled || !got.Dashboard.Enabled { - t.Fatalf("zero config should resolve safe enabled defaults: %+v", got) + t.Fatalf("default config should resolve safe enabled defaults: %+v", got) } if got.Retry.MaxAttempts != 5 || got.Retry.BatchSize != 50 { t.Fatalf("retry defaults = %+v", got.Retry) } } +func TestSettingsFromConfigPreservesExplicitGlobalDisable(t *testing.T) { + got := SettingsFromConfig(config.Config{Notifications: config.NotificationConfig{Enabled: false}}).Settings(context.Background()) + if got.Enabled { + t.Fatalf("explicit disabled notifications should stay disabled: %+v", got) + } + if got.Retry.MaxAttempts != 5 || got.Routing.Priorities == nil { + t.Fatalf("disabled config should still receive non-global defaults: %+v", got) + } +} + func TestNormalizeSettingsPreservesExplicitEmptyRoute(t *testing.T) { cfg := config.DefaultNotificationConfig() cfg.Routing.Priorities[ports.PriorityUrgent] = []string{} diff --git a/backend/internal/storage/sqlite/notification_delivery_store.go b/backend/internal/storage/sqlite/notification_delivery_store.go index 7a7503b6a..9cdddc743 100644 --- a/backend/internal/storage/sqlite/notification_delivery_store.go +++ b/backend/internal/storage/sqlite/notification_delivery_store.go @@ -70,39 +70,29 @@ func (s *Store) EnqueueDelivery(ctx context.Context, row notification.DeliveryRo s.writeMu.Lock() defer s.writeMu.Unlock() - insert := `INSERT INTO notification_deliveries ( - id, notification_id, notification_seq, project_id, session_id, - route_name, sink, destination_key, request_json, - status, attempts, max_attempts, next_attempt_at, lease_owner, lease_expires_at, - last_error_code, last_error, external_id, - created_at, updated_at, delivered_at -) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) -ON CONFLICT(notification_id, route_name, destination_key) DO NOTHING -RETURNING ` + deliveryColumns - - got, err := scanDelivery(s.writeDB.QueryRowContext(ctx, insert, - row.ID, - string(row.NotificationID), - row.NotificationSeq, - string(row.ProjectID), - string(row.SessionID), - row.RouteName, - row.Sink, - row.DestinationKey, - string(row.RequestJSON), - string(row.Status), - row.Attempts, - row.MaxAttempts, - row.NextAttemptAt, - row.LeaseOwner, - nullTime(row.LeaseExpiresAt), - row.LastErrorCode, - row.LastError, - row.ExternalID, - row.CreatedAt, - row.UpdatedAt, - nullTime(row.DeliveredAt), - )) + got, err := s.qw.InsertNotificationDelivery(ctx, gen.InsertNotificationDeliveryParams{ + ID: row.ID, + NotificationID: string(row.NotificationID), + NotificationSeq: row.NotificationSeq, + ProjectID: string(row.ProjectID), + SessionID: string(row.SessionID), + RouteName: row.RouteName, + Sink: row.Sink, + DestinationKey: row.DestinationKey, + RequestJson: string(row.RequestJSON), + Status: string(row.Status), + Attempts: int64(row.Attempts), + MaxAttempts: int64(row.MaxAttempts), + NextAttemptAt: row.NextAttemptAt, + LeaseOwner: row.LeaseOwner, + LeaseExpiresAt: nullTime(row.LeaseExpiresAt), + LastErrorCode: row.LastErrorCode, + LastError: row.LastError, + ExternalID: row.ExternalID, + CreatedAt: row.CreatedAt, + UpdatedAt: row.UpdatedAt, + DeliveredAt: nullTime(row.DeliveredAt), + }) if errors.Is(err, sql.ErrNoRows) { existing, readErr := s.getDeliveryByUniqueLocked(ctx, row.NotificationID, row.RouteName, row.DestinationKey) if readErr != nil { @@ -113,7 +103,7 @@ RETURNING ` + deliveryColumns if err != nil { return notification.DeliveryRow{}, false, fmt.Errorf("insert notification delivery: %w", err) } - return got, true, nil + return deliveryFromGen(got), true, nil } func (s *Store) ClaimDueDeliveries(ctx context.Context, sink string, owner string, now time.Time, limit int, lease time.Duration) ([]notification.DeliveryRow, error) { @@ -298,14 +288,14 @@ WHERE id = ? AND status NOT IN ('sent','failed','skipped','cancelled')`, reason, } func (s *Store) GetDelivery(ctx context.Context, id string) (notification.DeliveryRow, bool, error) { - row, err := scanDelivery(s.readDB.QueryRowContext(ctx, `SELECT `+deliveryColumns+` FROM notification_deliveries WHERE id = ?`, id)) + row, err := s.qr.GetNotificationDelivery(ctx, id) if errors.Is(err, sql.ErrNoRows) { return notification.DeliveryRow{}, false, nil } if err != nil { return notification.DeliveryRow{}, false, fmt.Errorf("get notification delivery %s: %w", id, err) } - return row, true, nil + return deliveryFromGen(row), true, nil } func (s *Store) ListDeliveries(ctx context.Context, filter DeliveryFilter) ([]notification.DeliveryRow, error) { @@ -375,13 +365,15 @@ func (s *Store) updateDelivery(ctx context.Context, what string, query string, a } func (s *Store) getDeliveryByUniqueLocked(ctx context.Context, id domain.NotificationID, routeName, destinationKey string) (notification.DeliveryRow, error) { - row, err := scanDelivery(s.writeDB.QueryRowContext(ctx, `SELECT `+deliveryColumns+` -FROM notification_deliveries -WHERE notification_id = ? AND route_name = ? AND destination_key = ?`, string(id), routeName, destinationKey)) + row, err := s.qw.GetNotificationDeliveryByUnique(ctx, gen.GetNotificationDeliveryByUniqueParams{ + NotificationID: string(id), + RouteName: routeName, + DestinationKey: destinationKey, + }) if err != nil { return notification.DeliveryRow{}, fmt.Errorf("get notification delivery by unique key: %w", err) } - return row, nil + return deliveryFromGen(row), nil } type rowScanner interface { @@ -437,3 +429,34 @@ func scanDelivery(scanner rowScanner) (notification.DeliveryRow, error) { } return row, nil } + +func deliveryFromGen(r gen.NotificationDelivery) notification.DeliveryRow { + row := notification.DeliveryRow{ + ID: r.ID, + NotificationID: domain.NotificationID(r.NotificationID), + NotificationSeq: r.NotificationSeq, + ProjectID: domain.ProjectID(r.ProjectID), + SessionID: domain.SessionID(r.SessionID), + RouteName: r.RouteName, + Sink: r.Sink, + DestinationKey: r.DestinationKey, + RequestJSON: []byte(r.RequestJson), + Status: notification.DeliveryStatus(r.Status), + Attempts: int(r.Attempts), + MaxAttempts: int(r.MaxAttempts), + NextAttemptAt: r.NextAttemptAt, + LeaseOwner: r.LeaseOwner, + LastErrorCode: r.LastErrorCode, + LastError: r.LastError, + ExternalID: r.ExternalID, + CreatedAt: r.CreatedAt, + UpdatedAt: r.UpdatedAt, + } + if r.LeaseExpiresAt.Valid { + row.LeaseExpiresAt = r.LeaseExpiresAt.Time + } + if r.DeliveredAt.Valid { + row.DeliveredAt = r.DeliveredAt.Time + } + return row +} diff --git a/backend/notifier_wiring.go b/backend/notifier_wiring.go index 06ed53b84..5463859dd 100644 --- a/backend/notifier_wiring.go +++ b/backend/notifier_wiring.go @@ -14,10 +14,10 @@ type notifierStack struct { done <-chan struct{} } -func startNotifier(ctx context.Context, cfg config.Config, store *sqlite.Store, log *slog.Logger) (*notifierStack, error) { +func startNotifier(ctx context.Context, cfg config.Config, store *sqlite.Store, log *slog.Logger) *notifierStack { mgr := notification.NewManager(store, notification.SettingsFromConfig(cfg), log) done := mgr.Start(ctx) - return ¬ifierStack{Manager: mgr, done: done}, nil + return ¬ifierStack{Manager: mgr, done: done} } func (s *notifierStack) Stop() {