// 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 // " ()" — 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 }