codit/backend/internal/notify/dispatcher.go

// Package notify is the after-commit external-delivery fan-out for events. The
// event row is persisted transactionally by the store (the in-app feed works
// with no channels configured); the Dispatcher resolves a monitor's attached
// notification channels and delivers each via its transport. It implements
// monitor.Notifier.
package notify

import "context"
import "encoding/json"
import "net/http"
import "strings"
import "time"

import "codit/internal/db"
import "codit/internal/models"
import "codit/internal/monitor"
import "codit/internal/secretcrypto"
import codit_logger "codit/logger"

const logIDNotify string = "notify"

// webhookTimeout is the per-call delivery timeout (setting, read live). Delivery
// paths derive a request context from it; the shared client has no hard Timeout
// so the setting always governs.
func (d *Dispatcher) webhookTimeout() time.Duration {
	if d.store == nil {
		return 15 * time.Second
	}
	return d.store.SettingDuration(db.SettingTimeoutWebhookSec)
}

// Dispatcher delivers events to the channels attached to their source monitor.
type Dispatcher struct {
	store    *db.Store
	logger   codit_logger.Logger
	secrets  *secretcrypto.Box
	policy   monitor.EgressPolicy
	client   *http.Client
	serverID string // this deployment's service id, stamped on outbound alerts
	siteName string // this deployment's human site name, stamped on outbound alerts
}

// NewDispatcher builds a dispatcher. The HTTP client dials through the egress
// policy so webhook delivery is subject to the same SSRF guard as the probes;
// dataDir locates the deployment key used to decrypt transport secrets. The
// policy is retained so per-transport TLS variants can reuse the same dialer.
// serverID and siteName identify this deployment and are stamped onto alerts
// sent to external channels (never onto the persisted event / in-app feed).
func NewDispatcher(store *db.Store, serverID string, siteName string, policy monitor.EgressPolicy, dataDir string, logger codit_logger.Logger) *Dispatcher {
	return &Dispatcher{
		store:    store,
		logger:   logger,
		secrets:  secretcrypto.New(dataDir),
		policy:   policy,
		serverID: strings.TrimSpace(serverID),
		siteName: strings.TrimSpace(siteName),
		client: &http.Client{
			Transport: &http.Transport{
				DialContext:       policy.DialContext,
				DisableKeepAlives: true,
			},
		},
	}
}

// originLabel identifies the sending deployment for external recipients as
// "<service-id> (<site-name>)" — the app's own self-identity convention, kept
// ASCII so it is safe in an email Subject header. Drops whichever part is empty;
// returns "" if both are empty (then no decoration is applied).
func (d *Dispatcher) originLabel() string {
	if d.serverID != "" && d.siteName != "" {
		return d.siteName + " (" + d.serverID + ")"
	}
	if d.siteName != "" {
		return d.siteName
	}
	return d.serverID
}

// withOriginTitle returns a copy of the event whose Title is prefixed with the
// sending deployment's identity, so external recipients can tell at a glance
// which system raised the alert. The persisted event is passed by value, so the
// in-app feed and other channels are unaffected.
func (d *Dispatcher) withOriginTitle(event models.Event) models.Event {
	var origin string
	origin = d.originLabel()
	if origin != "" {
		event.Title = "[" + origin + "] " + event.Title
	}
	return event
}

// secretMap decrypts a stored transport secret blob into its field map.
func (d *Dispatcher) secretMap(encrypted string) map[string]string {
	var plain string
	var m map[string]string
	m = map[string]string{}
	if encrypted == "" {
		return m
	}
	plain, _ = d.secrets.Decrypt(encrypted)
	if plain != "" {
		_ = json.Unmarshal([]byte(plain), &m)
	}
	return m
}

func (d *Dispatcher) log(level codit_logger.LogLevel, format string, args ...any) {
	if d.logger != nil {
		d.logger.Write(logIDNotify, level, format, args...)
	}
}

// Notify resolves the deliverable channels for the event's source monitor and
// delivers to each. Per-target failures are logged and do not abort the rest;
// the first error is returned so the caller can log an aggregate.
func (d *Dispatcher) Notify(ctx context.Context, event models.Event) error {
	var targets []models.DeliveryTarget
	var t models.DeliveryTarget
	var firstErr error
	var err error

	// Resolve every deliverable channel for this event: attribute-matched
	// notification rules (any producer) plus, for a monitor event, its directly
	// attached channels. Deduped by the store.
	targets, err = d.store.DeliveryTargetsForEvent(event)
	if err != nil {
		return err
	}
	if len(targets) == 0 {
		return nil
	}
	d.log(codit_logger.LOG_INFO, "dispatching event kind=%q source=%q to %d channel(s)", event.Kind, event.SourceID, len(targets))
	for _, t = range targets {
		var derr error
		switch t.TransportType {
		case models.TransportTypeWebhook:
			derr = d.deliverWebhook(ctx, t, d.secretMap(t.Secret), event)
		case models.TransportTypeSMTP:
			derr = d.deliverSMTP(ctx, t, d.secretMap(t.Secret), event)
		case models.TransportTypeMSTeams:
			derr = d.deliverTeams(ctx, t, event)
		case models.TransportTypeSlack:
			derr = d.deliverSlack(ctx, t, event)
		case models.TransportTypeMSGraph:
			derr = d.deliverMSGraph(ctx, t, d.secretMap(t.Secret), event)
		default:
			d.log(codit_logger.LOG_WARN, "unsupported transport type=%q channel=%q", t.TransportType, t.ChannelName)
			continue
		}
		if derr != nil {
			d.log(codit_logger.LOG_ERROR, "notify failed channel=%q transport=%q kind=%q source=%q err=%v", t.ChannelName, t.TransportType, event.Kind, event.SourceID, derr)
			if firstErr == nil {
				firstErr = derr
			}
			continue
		}
		d.log(codit_logger.LOG_INFO, "notified channel=%q transport=%q kind=%q source=%q", t.ChannelName, t.TransportType, event.Kind, event.SourceID)
	}
	return firstErr
}