silo-server/internal/notifications/operational_dispatch.go

package notifications

import (
	"context"
	"fmt"

	"github.com/oklog/ulid/v2"
)

// OperationalDispatch describes how one operational delivery (a non-fanout
// notice such as webhook.auto_disabled or request.fulfilled) reaches the
// per-target channels. WebhookFilter selects which of the profile's enabled
// webhooks receive it; nil means the type must not reach webhooks at all
// (e.g. the auto-disable notice's loop guard). Web push has no per-type
// filter: profile-level gating happens before dispatch.
type OperationalDispatch struct {
	WebhookFilter func(Webhook) bool
}

// DispatchOperational durably creates one operational delivery. The inbox row
// and the per-target webhook / web push / Apple push outbox rows commit in a
// single transaction — a crash afterwards delays channel sends instead of dropping
// them, because the retry workers recover pending outbox rows — then realtime
// and channel dispatch run post-commit. Returns nil when the delivery deduped
// away (the partial unique indexes make operational notices idempotent).
func (s *System) DispatchOperational(ctx context.Context, delivery Delivery, opts OperationalDispatch) (*InsertedDelivery, error) {
	if s == nil {
		return nil, nil
	}
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return nil, fmt.Errorf("begin operational dispatch tx: %w", err)
	}
	defer func() { _ = tx.Rollback(ctx) }()

	inserted, err := s.Deliveries.BulkInsert(ctx, tx, []Delivery{delivery})
	if err != nil {
		return nil, err
	}
	if len(inserted) == 0 {
		return nil, nil
	}
	row := inserted[0]

	if opts.WebhookFilter != nil && s.webhookRepo != nil && s.Settings.WebhooksEnabled(ctx) {
		hooksByProfile, err := s.webhookRepo.ListEnabledByProfiles(ctx, tx, []string{delivery.ProfileID})
		if err != nil {
			return nil, err
		}
		attempts := make([]DeliveryAttempt, 0, 2)
		for _, hook := range hooksByProfile[delivery.ProfileID] {
			if !opts.WebhookFilter(hook) {
				continue
			}
			attempts = append(attempts, DeliveryAttempt{
				ID:                     ulid.Make().String(),
				NotificationDeliveryID: row.ID,
				TargetID:               hook.ID,
			})
		}
		if err := s.webhookRepo.EnqueueAttempts(ctx, tx, attempts); err != nil {
			return nil, err
		}
	}
	if s.webPushRepo != nil && s.Settings.WebPushEnabled(ctx) {
		subsByProfile, err := s.webPushRepo.ListEnabledByProfiles(ctx, tx, []string{delivery.ProfileID})
		if err != nil {
			return nil, err
		}
		attempts := make([]DeliveryAttempt, 0, 2)
		for _, sub := range subsByProfile[delivery.ProfileID] {
			attempts = append(attempts, DeliveryAttempt{
				ID:                     ulid.Make().String(),
				NotificationDeliveryID: row.ID,
				TargetID:               sub.ID,
			})
		}
		if err := s.webPushRepo.EnqueueAttempts(ctx, tx, attempts); err != nil {
			return nil, err
		}
	}
	if s.pushDeviceRepo != nil {
		if platforms := s.Settings.EnabledPushPlatforms(ctx); len(platforms) > 0 {
			devicesByProfile, err := s.pushDeviceRepo.ListEnabledPushByProfiles(ctx, tx, []string{delivery.ProfileID}, platforms)
			if err != nil {
				return nil, err
			}
			attempts := newPushDeliveryAttempts(row.ID, devicesByProfile[delivery.ProfileID])
			if err := s.pushDeviceRepo.EnqueuePushAttempts(ctx, tx, attempts); err != nil {
				return nil, err
			}
		}
	}
	if err := tx.Commit(ctx); err != nil {
		return nil, fmt.Errorf("commit operational dispatch: %w", err)
	}

	// Post-commit dispatch is best-effort: the durable inbox row covers
	// websocket reconnect, and the retry workers recover the outbox rows.
	full, err := s.Deliveries.GetRowByID(ctx, row.ID)
	if err != nil || full == nil {
		s.logger.WarnContext(ctx, "operational delivery reload failed",
			"delivery_id", row.ID, "error", err)
		return &row, nil
	}
	if err := s.dispatcher.Dispatch(ctx, *full); err != nil {
		s.logger.WarnContext(ctx, "operational delivery dispatch failed",
			"delivery_id", row.ID, "error", err)
	}
	return &row, nil
}