Files
felhom.eu/hub/internal/notify/dispatcher.go
T
admin 91cabdde1b
gates / gates (push) Successful in 7s
hub v0.93.0: the retention keeps the key it was built to keep (R-198) + three honesty fixes (R-197, R-192, R-196)
R-198 — host_escrow_superseded shipped with `blob` (the K-escrow / PBS datastore key) and
identity_blob was added to host_escrow LATER, never here. The offsite restic REPOSITORY
password lives in identity_blob. So demoteCurrentEscrowTx -- whose own comment calls it "THE
ONE escrow row-copy routine" -- retained the whole-guest key and silently dropped the off-site
data key, which is the secret the retention was built to preserve. And because the copy happens
as the new blob overwrites the old, the destroying act was the ESCROW CEREMONY: the exact thing
a rebuilt box tells its customer to run, on a card promising in Hungarian that the old backups
stay recoverable. Both demo boxes crossed that line on 2026-08-04.

  - identity_blob added to the table (CREATE + additive ALTER) and carried in the shared copy
    routine, so BOTH callers are fixed at once: re-escrow and host-delete demotion.
  - ListSupersededEscrow reads it back; store.HostEscrow gains IdentityBlob.
  - CountCurrentEscrowWithIdentity is the census of who the fix protects.
  - Nothing is backfillable: pre-v0.93.0 retained rows have no blob and their sources are gone.
  - Tests assert the CONSEQUENCE (a retained row can still yield a repo password), which is why
    the pre-existing retention test stayed green for two months asserting the mechanism.

R-197 — SaveHostEscrow returns the hash it replaced; the escrow PUT raises
offsite_repo_key_changed (warning, operator-only, edge-triggered) when both hashes are known and
differ. No hash value travels. Severity chosen for the world v0.93.0 creates: with the identity
blob retained, a changed key is "this history now depends on an older recovery code", not a loss.

R-192 (half) — the stuck alert now reports the two shapes it actually covers, burned and
regressed, each stating its own measurement; the regressed text withdraws the Re-issue
recommendation. Every self-heal refusal leaves a notification_log row with its reason. The
guard's logic is unchanged; its 500-oldest-reports scoping stays OPEN and the window is named in
the alert text so the limitation travels with the number. offsite_delivery_stuck and
offsite_credential_restaged are added to operatorOnlyEvents -- neither was registered and neither
has a customerMessages entry, which is not a block.

R-196 — five comments (not the three the spec expected) claimed ReissueCredentials rotates the
restic repo password. It resets the PROVIDER password and cannot touch the repo password, which
is generated on the box. All five corrected; the staleness mark documented as precautionary. The
BEHAVIOUR stays open.

Not in this release: R-199, R-200, R-201 remain open -- the chain that hands the key back is
still unassembled. Part 5 hit its gate; the orphan card is untouched (R-202).
2026-08-04 12:56:58 +02:00

590 lines
29 KiB
Go

package notify
import (
"bytes"
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"strings"
"sync"
"time"
"gitea.dooplex.hu/admin/felhom-hub/internal/store"
)
// Dispatcher routes events to operator and/or customer email channels.
// Cooldowns are in-memory (lost on restart, acceptable).
type Dispatcher struct {
store *store.Store
resendAPIKey string
fromEmail string
operatorEmail string
operatorOn bool
httpClient *http.Client
logger *log.Logger
mu sync.Mutex
opCooldowns map[string]time.Time // "customerID:eventType" → last operator notify
custCooldowns map[string]time.Time // "customerID:eventType" → last customer notify
// sendEmailFn is the email sender, seam-injected so tests exercise routing without real HTTP.
// Defaults to (*Dispatcher).sendEmail (Resend) in NewDispatcher. headers (nil = none) become
// Resend custom headers — used for the high-priority nudge on error/critical mails (v0.71.0,
// audit F14-light).
sendEmailFn func(to, subject, textBody string, headers map[string]string) error
}
// NewDispatcher creates a new notification dispatcher.
func NewDispatcher(s *store.Store, resendAPIKey, fromEmail, operatorEmail string, operatorOn bool, logger *log.Logger) *Dispatcher {
d := &Dispatcher{
store: s,
resendAPIKey: resendAPIKey,
fromEmail: fromEmail,
operatorEmail: operatorEmail,
operatorOn: operatorOn,
httpClient: &http.Client{Timeout: 10 * time.Second},
logger: logger,
opCooldowns: make(map[string]time.Time),
custCooldowns: make(map[string]time.Time),
}
d.sendEmailFn = d.sendEmail
return d
}
// priorityHeaders returns the Resend custom headers that nudge mail clients toward attention for
// error/critical mails (X-Priority + Importance; v0.71.0, audit F14-light: delivered ≠ noticed).
// Everything else gets nil — a warning or info mail must NOT masquerade as urgent. Pure → tested.
func priorityHeaders(severity string) map[string]string {
switch severity {
case "error", "critical":
return map[string]string{"X-Priority": "1", "Importance": "high"}
default:
return nil
}
}
// recoveredPairedDownTypes maps a *_recovered eventType to the stale/down set whose customer-channel
// "sent" evidence licenses the customer recovery mail (v0.71.0, audit F11): recovery notifies
// exactly whoever the down notified.
var recoveredPairedDownTypes = map[string][]string{
"node_recovered": {"node_stale", "node_down"},
"host_recovered": {"host_stale", "host_down"},
// R-97a. This branch runs BEFORE the severity gate, which is exactly why the recovery belongs
// here: `whole_guest_backup_recovered` is severity "info", and severityNotifies drops "info", so
// routing it normally would store the event and never mail it — the operator would be told the
// tier broke and never told it healed, which is the half of Scenario B that matters.
//
// The customer leg needs no special handling: it is PAIRING-gated on a customer-channel "sent"
// row for the down type, and `whole_guest_backup_failed` has no customerMessages entry and is in
// nobody's enabled_events — so no such row can exist, and the customer correctly hears neither
// edge. Operator hears both.
"whole_guest_backup_recovered": {"whole_guest_backup_failed"},
}
// severityNotifies reports whether a severity triggers email notifications. warning / error / critical
// notify; everything else (info, recovery/status, or an unrecognized value) does not. Pure → unit-tested.
// (Before v0.24.0 a "critical" severity was silently dropped here — the host_disk-class bug.)
func severityNotifies(severity string) bool {
switch severity {
case "warning", "error", "critical":
return true
default:
return false
}
}
// ProcessEvent evaluates an event and sends notifications as appropriate.
// Safe to call from goroutines.
func (d *Dispatcher) ProcessEvent(customerID, eventType, severity, message, detailsJSON, source string) {
if d.resendAPIKey == "" {
return
}
// "test" bypass — send directly to customer email, skip prefs/cooldown
if eventType == "test" {
d.sendTestEmail(customerID)
return
}
// Recovery branch (v0.71.0, audit F11) — BEFORE the severity gate, as an explicit eventType
// branch: *_recovered stays severity "info" (semantics frozen), but is no longer silent.
// Operator always hears both edges; the customer hears recovery iff they heard the down.
if _, isRecovery := recoveredPairedDownTypes[eventType]; isRecovery {
d.processRecovery(customerID, eventType, severity, message, detailsJSON, source)
return
}
// R-182: record-only types are written down and never mailed. Placed BEFORE the severity gate
// so the row is written whatever the severity — the record must not inherit the notification's
// conditions, which is the coupling this whole finding is about.
if recordOnlyEvents[eventType] {
if err := d.store.LogNotification(customerID, eventType, severity, message, "recorded",
"record-only: the per-run digest (backup_run_failures) carries the notification", "operator"); err != nil {
d.logger.Printf("[WARN] Failed to record %s for %s: %v", eventType, customerID, err)
}
d.logger.Printf("[INFO] Recorded (not mailed) %s for %s — the run digest is the notification", eventType, customerID)
return
}
// warning / error / critical trigger notifications. "info" is an intentional non-notify (status/
// recovery events). Anything else is UNRECOGNIZED — log it (don't silently drop), so a bad severity
// surfaces instead of vanishing (the felhom-pve-class lesson: a critical event must never be lost).
if !severityNotifies(severity) {
if severity != "info" {
d.logger.Printf("[WARN] Dispatcher: unrecognized severity %q for %s/%s — not routing", severity, customerID, eventType)
}
return
}
// Operator channel
d.processOperator(customerID, eventType, severity, message, detailsJSON, source)
// Customer channel
d.processCustomer(customerID, eventType, severity, message, detailsJSON, source)
}
func (d *Dispatcher) sendTestEmail(customerID string) {
// nil-prefs guard (v0.71.0): GetNotificationPrefs returns (nil, nil) for a customer with no
// notification row — dereferencing prefs.Email here panicked the dispatcher goroutine for such
// a customer (latent since the test leg shipped; found while adding the operator copy).
prefs, err := d.store.GetNotificationPrefs(customerID)
if err != nil || prefs == nil || prefs.Email == "" {
d.logger.Printf("[WARN] Test email: no email configured for %s", customerID)
} else {
subject := "[Felhom] Teszt értesítés"
body := "Kedves Ügyfél!\n\nEz egy teszt értesítés a Felhom monitoring rendszerből.\nAz értesítések megfelelően működnek.\n\nÜdvözlettel,\nFelhom.eu monitoring"
if err := d.sendEmailFn(prefs.Email, subject, body, nil); err != nil {
d.logger.Printf("[ERROR] Test email to %s failed: %v", prefs.Email, err)
d.store.LogNotification(customerID, "test", "info", "Teszt értesítés", "failed", err.Error(), "customer")
} else {
d.logger.Printf("[INFO] Test email sent to %s for %s", prefs.Email, customerID)
d.store.LogNotification(customerID, "test", "info", "Teszt értesítés", "sent", "", "customer")
}
}
// Operator copy (v0.71.0, audit F14-light): one test click proves the customer channel, the
// operator channel AND the high-priority header rendering in a single shot.
if !d.operatorOn || d.operatorEmail == "" {
return
}
opSubject := fmt.Sprintf("[Felhom] ✅ %s: teszt / operator channel OK", customerID)
opBody := fmt.Sprintf(`Operator copy of the customer notification test for %s.
If this mail shows as high priority in your client, the X-Priority/Importance
headers render correctly. The customer test mail result is recorded in the
notification log.
Dashboard: https://hub.felhom.eu/customers/%s`, customerID, customerID)
if err := d.sendEmailFn(d.operatorEmail, opSubject, opBody, priorityHeaders("critical")); err != nil {
d.logger.Printf("[ERROR] Operator test email failed for %s: %v", customerID, err)
d.store.LogNotification(customerID, "test", "info", "operator test copy", "failed", err.Error(), "operator")
return
}
d.logger.Printf("[INFO] Operator test email sent for %s", customerID)
d.store.LogNotification(customerID, "test", "info", "operator test copy", "sent", "", "operator")
}
// processRecovery routes a *_recovered event (v0.71.0, audit F11). Severity semantics stay frozen
// ("info" everywhere else remains non-notify) — this is an explicit eventType branch.
// - Operator leg: always wanted (both edges), gated only by operatorOn + the 1h per-type
// cooldown — exactly processOperator.
// - Customer leg: gated by the PAIRING rule, not enabled_events — "recovery notifies exactly
// whoever the down notified." Evidence = a customer-channel status=sent row for the paired
// stale/down set newer than the last customer-channel sent recovery of this type.
func (d *Dispatcher) processRecovery(customerID, eventType, severity, message, detailsJSON, source string) {
d.processOperator(customerID, eventType, severity, message, detailsJSON, source)
if d.store.IsCustomerBlocked(customerID) {
return
}
prefs, err := d.store.GetNotificationPrefs(customerID)
if err != nil || prefs == nil || prefs.Email == "" {
return
}
// Pairing check — the customer gate. enabled_events is deliberately ignored here: a customer
// who was told "down" must be told "recovered", and one who wasn't must not be.
lastDown, downOk, err := d.store.LastCustomerSentAt(customerID, recoveredPairedDownTypes[eventType])
if err != nil {
d.logger.Printf("[ERROR] Recovery pairing query failed for %s/%s: %v", customerID, eventType, err)
return
}
lastRecovered, recOk, err := d.store.LastCustomerSentAt(customerID, []string{eventType})
if err != nil {
d.logger.Printf("[ERROR] Recovery pairing query failed for %s/%s: %v", customerID, eventType, err)
return
}
// Second-granularity ties resolve to NOT-after → no mail (flap-safe direction).
if !downOk || (recOk && !lastDown.After(lastRecovered)) {
d.logger.Printf("[INFO] Recovery %s for %s: customer mail skipped — no unanswered customer down mail (pairing miss)", eventType, customerID)
return
}
// Prefs cooldown keyed on the recovered eventType (belt over the pairing braces).
cooldownHours := prefs.CooldownHours
if cooldownHours <= 0 {
cooldownHours = 6
}
cooldownKey := customerID + ":" + eventType
d.mu.Lock()
if last, ok := d.custCooldowns[cooldownKey]; ok && time.Since(last) < time.Duration(cooldownHours)*time.Hour {
d.mu.Unlock()
d.logger.Printf("[INFO] Recovery %s for %s: customer mail skipped — cooldown", eventType, customerID)
return
}
d.custCooldowns[cooldownKey] = time.Now()
d.mu.Unlock()
subject, body := FormatCustomerEmail(customerID, eventType, severity, message, detailsJSON)
if err := d.sendEmailFn(prefs.Email, subject, body, priorityHeaders(severity)); err != nil {
d.logger.Printf("[ERROR] Customer recovery email failed for %s/%s: %v", customerID, eventType, err)
d.store.LogNotification(customerID, eventType, severity, message, "failed", err.Error(), "customer")
return
}
d.logger.Printf("[INFO] Customer recovery email sent to %s for %s/%s", prefs.Email, customerID, eventType)
d.store.LogNotification(customerID, eventType, severity, message, "sent", "", "customer")
}
// cooldownTierSuffix returns ":"+tier when the event's details carry a non-empty `tier`, else "".
//
// R-97a. The operator cooldown was keyed `customerID + ":" + eventType` alone, which is correct for
// every event that describes ONE thing — but a whole-guest backup failure describes ONE TIER, and a
// box has two. `felhom-pbs` failing at 09:00 would swallow `local` failing at 09:20 for the whole
// hour, so the operator would be told about the offsite tier and never about the local one. That is
// precisely the masking the per-tier signal exists to prevent.
//
// NARROW ON PURPOSE: the suffix is empty unless the producer opts in by sending a `tier`, so no
// existing event type's cooldown behaviour changes. Widening the key for everything would, e.g.,
// turn one hourly `app_start_failed` into one per app — a flood, not a fix.
func cooldownTierSuffix(detailsJSON string) string {
if detailsJSON == "" || !strings.Contains(detailsJSON, "\"tier\"") {
return ""
}
var d struct {
Tier string `json:"tier"`
}
if err := json.Unmarshal([]byte(detailsJSON), &d); err != nil || d.Tier == "" {
return ""
}
return ":" + d.Tier
}
// cooldownRunSuffix returns ":"+run_id when the event's details carry a non-empty `run_id`, else "".
//
// R-182. `cooldownTierSuffix`'s sibling, and deliberately a SEPARATE function rather than an extra
// branch inside it: `tier` keeps byte-identical semantics for every type that uses it, so R-97a's
// behaviour and its tests are untouched by this.
//
// WHY A BACKUP RUN NEEDS ONE. The run digest describes ONE RUN, and a box can have two in a day —
// the nightly one and a manual one the operator triggered *because* something looked wrong. With no
// run-scoped discriminator the 1-hour cooldown would swallow the second, which is the failure this
// row exists to fix, reappearing one level up: the operator presses the button, the run fails, and
// they are told nothing because the machine already wrote that hour.
//
// IT MAKES THE COOLDOWN EFFECTIVELY INERT FOR THIS TYPE, AND THAT IS THE INTENT, NOT AN OVERSIGHT.
// A digest is already rate-limited by construction — one per run, emitted only when something
// failed — so there is nothing for a timer to collapse. The cooldown protects against a repeating
// identical alert; a digest cannot repeat, because each run is a different run.
//
// NARROW, LIKE ITS SIBLING: empty unless the producer opts in by sending a `run_id`, so no existing
// event type's cooldown behaviour changes.
func cooldownRunSuffix(detailsJSON string) string {
if detailsJSON == "" || !strings.Contains(detailsJSON, "\"run_id\"") {
return ""
}
var d struct {
RunID string `json:"run_id"`
}
if err := json.Unmarshal([]byte(detailsJSON), &d); err != nil || d.RunID == "" {
return ""
}
return ":" + d.RunID
}
func (d *Dispatcher) processOperator(customerID, eventType, severity, message, detailsJSON, source string) {
if !d.operatorOn || d.operatorEmail == "" {
return
}
cooldownKey := customerID + ":" + eventType + cooldownTierSuffix(detailsJSON) + cooldownRunSuffix(detailsJSON)
d.mu.Lock()
if last, ok := d.opCooldowns[cooldownKey]; ok && time.Since(last) < 1*time.Hour {
d.mu.Unlock()
// R-182: RECORD THE SUPPRESSION. This used to be a bare `return` — the event was dropped
// before any LogNotification, so a cooldown drop and an event that never happened were
// indistinguishable from the operator's side AND from the hub's own records.
//
// Measured 2026-08-03: nine `recovery_unit_capture_failed` events arrived, two emails were
// sent, and the other seven left NO ROW ON ANY CHANNEL. The defect that hid was serious —
// the cooldown key carries no app identifier, so the first refused app took the slot and
// every other app's failure that hour was discarded — but the reason it took a day to find
// the right way round is this line: there was nothing to read.
//
// "We chose not to e-mail you" and "nothing happened" must never look identical. This
// applies to EVERY operator event, not only the one that exposed it. It makes the drop
// visible; it deliberately does NOT change the cooldown's duration or semantics.
if err := d.store.LogNotification(customerID, eventType, severity, message,
"suppressed", "operator cooldown 1h, key="+cooldownKey, "operator"); err != nil {
d.logger.Printf("[WARN] Failed to record suppressed operator notification for %s/%s: %v",
customerID, eventType, err)
}
d.logger.Printf("[INFO] Operator email suppressed for %s/%s — cooldown (key=%s)",
customerID, eventType, cooldownKey)
return
}
d.opCooldowns[cooldownKey] = time.Now()
d.mu.Unlock()
subject, body := FormatOperatorEmail(customerID, eventType, severity, message, detailsJSON)
if err := d.sendEmailFn(d.operatorEmail, subject, body, priorityHeaders(severity)); err != nil {
d.logger.Printf("[ERROR] Operator email failed for %s/%s: %v", customerID, eventType, err)
d.store.LogNotification(customerID, eventType, severity, message, "failed", err.Error(), "operator")
return
}
d.logger.Printf("[INFO] Operator email sent for %s/%s", customerID, eventType)
d.store.LogNotification(customerID, eventType, severity, message, "sent", "", "operator")
}
// recordOnlyEvents are STORED and RECORDED but never e-mailed, on either channel.
//
// R-182. The distinction this register exists to make is the whole of that finding: **the record and
// the notification are different things.** A per-app backup failure must always be written down —
// every time, unconditionally, regardless of cooldowns, preferences or whether any mail went out —
// and it must NOT compete for an e-mail slot, because the per-run digest
// (`backup_run_failures`) is what a person is meant to read.
//
// Before this, `recovery_unit_capture_failed` was both at once, and it did neither well: on
// 2026-08-03 nine of them arrived, two were e-mailed, and the other seven were dropped by the
// 1-hour cooldown BEFORE anything was written down. So the operator was told about one app, the
// other apps' failures were discarded, and nothing anywhere recorded that a choice had been made.
//
// WHY A REGISTER AND NOT severity "info". Downgrading the severity would have the same routing
// effect — `severityNotifies` drops info — but it would also relabel a genuine failure as
// informational in the events table, the operator UI and every historical query, and it would
// silently drop the X-Priority handling if the type were ever promoted back. This says what it
// means: not silent, not urgent, RECORDED.
//
// IT IS NOT A WAY TO MUTE THINGS. A type belongs here only when something else carries its
// notification. Adding one with no digest behind it rebuilds the silence R-182 was filed against.
var recordOnlyEvents = map[string]bool{
// The per-app Tier-1 capture/refusal failure. Its notification is the run digest, which lists
// every failed app in one mail; this row is the durable per-failure record behind it.
"recovery_unit_capture_failed": true,
}
// operatorOnlyEvents are event types that must NEVER reach a customer, whatever their preferences say.
//
// R-97c. This register exists because the guarantee it provides was previously ASSERTED IN A COMMENT
// and not implemented. The claim was that a type with no `customerMessages` entry "structurally
// cannot" be routed to a customer. It cannot: `FormatCustomerEmail` (templates.go) treats a missing
// entry as a **fallback to the raw message**, not a block —
//
// hunMessage := customerMessages[eventType]
// if hunMessage == "" { hunMessage = message }
//
// — and the only customer gate is `isEventEnabled(prefs.EnabledEvents, ...)`, i.e. CONFIGURATION.
// So a customer with `whole_guest_backup_failed` in their enabled list and an email set would have
// received the raw English operator text about a backup they can take no action on.
//
// That is the `EffectiveProtected` shape: a doc comment claiming a property the code stopped
// providing, which is how the samba false alarm survived. The register makes the claim true.
//
// NOT implemented as "a missing customerMessages entry blocks delivery" — several existing types rely
// on the raw-message fallback deliberately (e.g. offbox_enlarge_blocked, whose dynamic Hungarian text
// is customer-grade and would be DISCARDED by a template). Turning the fallback into a gate would
// change behaviour well outside this concern.
var operatorOnlyEvents = map[string]bool{
// R-97a. A customer can take no action on a failed whole-guest backup, and being told it failed
// while it is still retrying behind the R-88 breaker is alarming without being actionable.
"whole_guest_backup_failed": true,
// The recovery is ALSO listed, even though its customer leg is pairing-gated on a customer-channel
// "sent" row that can never exist for the line above. Relying on that would make this type's safety
// a consequence of another type's routing — true today, and silently untrue the moment the failed
// event becomes customer-visible. Belt, not inference.
"whole_guest_backup_recovered": true,
// R-158 / R-167 (D-c). A per-app Tier-1 recovery-unit capture failure. A customer can take no
// action on it — the causes are a full filesystem, a permission fault or a broken dump, all of
// which the operator resolves — and the alert carries operator-grade detail (target path, byte
// figures, the raw error). The customer's half of D-c is the FILL WARNING, which fires BEFORE
// this and is actionable: free space, delete files, add a drive.
"recovery_unit_capture_failed": true,
// R-182. The per-run backup digest. It is the same class as the line above and for the same
// reason — a customer can act on a full disk (that is the fill warning, which fires first and
// IS customer-facing) but not on a list of which apps' backups failed and why. It also carries
// operator-grade detail: per-app leg names, raw refusal reasons and byte figures.
//
// Listed here rather than relying on the absence of a `customerMessages` entry, which is NOT a
// block — `FormatCustomerEmail` falls back to the raw English message. That mistake shipped
// once (v0.78.0) and the comment above records it.
"backup_run_failures": true,
// R-197 (v0.93.0). "The sealed offsite repository key changed" is a custody fact about escrow
// blobs. A customer can take no action on it — the remedy is the operator's inspection of the
// off-site tier — and the text is operator-grade English naming host ids and retained-blob
// counts. Listed here in the SAME commit that mints the type: an operator-tier type that is not
// registered here reaches customers as raw English, because a missing customerMessages entry is
// NOT a block (the v0.78.0 defect recorded above).
"offsite_repo_key_changed": true,
// R-192 (v0.93.0). These two predate the register and were never added to it — a real gap, not a
// tidy-up. `offsite_delivery_stuck` is severity warning, has no customerMessages entry, and
// therefore fell through to FormatCustomerEmail's raw-English fallback: a customer whose box hit
// the stuck shape was in line for an English e-mail about one-time passwords being "likely
// burned". Measured on the live hub: notification_log holds operator rows for demo-hp and no
// customer rows — which is NOT evidence the leg is blocked (it is equally consistent with no
// configured recipient), so the register makes it structural instead of incidental. Narrowing
// only: the operator channel is untouched.
"offsite_delivery_stuck": true,
"offsite_credential_restaged": true,
}
// IsOperatorOnly reports whether an event type is barred from customer dispatch. Exported so the
// api package can pin BOTH registers of a new event type in one test — allowlisted-but-not-
// operator-only is the v0.78.0 defect, and it is only visible when the two are checked together.
// Read-only: the register itself stays unexported so nothing can widen it at runtime.
func IsOperatorOnly(eventType string) bool { return operatorOnlyEvents[eventType] }
func (d *Dispatcher) processCustomer(customerID, eventType, severity, message, detailsJSON, source string) {
// R-97c: operator-tier events stop here, BEFORE prefs are consulted — the point is that no
// customer configuration can opt in. Logged rather than dropped, so the skip is visible in
// notification_log instead of looking like a delivery that never happened.
if operatorOnlyEvents[eventType] {
d.store.LogNotification(customerID, eventType, severity, message, "skipped", "operator_only", "customer")
return
}
// Check if customer is blocked
if d.store.IsCustomerBlocked(customerID) {
return
}
// Load preferences. GetNotificationPrefs returns (nil, nil) for a customer with no notification row —
// guard the nil BEFORE dereferencing (else an event for such a customer panics the dispatcher
// goroutine and crashes the hub). No prefs / no email → no customer notification.
prefs, err := d.store.GetNotificationPrefs(customerID)
if err != nil || prefs == nil || prefs.Email == "" {
return
}
// Check if event type is enabled
if !isEventEnabled(prefs.EnabledEvents, eventType) {
return
}
// Customer cooldown (from prefs, default 6h)
cooldownHours := prefs.CooldownHours
if cooldownHours <= 0 {
cooldownHours = 6
}
cooldownDur := time.Duration(cooldownHours) * time.Hour
cooldownKey := customerID + ":" + eventType
d.mu.Lock()
if last, ok := d.custCooldowns[cooldownKey]; ok && time.Since(last) < cooldownDur {
d.mu.Unlock()
return
}
d.custCooldowns[cooldownKey] = time.Now()
d.mu.Unlock()
subject, body := FormatCustomerEmail(customerID, eventType, severity, message, detailsJSON)
if err := d.sendEmailFn(prefs.Email, subject, body, priorityHeaders(severity)); err != nil {
d.logger.Printf("[ERROR] Customer email failed for %s/%s: %v", customerID, eventType, err)
d.store.LogNotification(customerID, eventType, severity, message, "failed", err.Error(), "customer")
return
}
d.logger.Printf("[INFO] Customer email sent to %s for %s/%s", prefs.Email, customerID, eventType)
d.store.LogNotification(customerID, eventType, severity, message, "sent", "", "customer")
}
func (d *Dispatcher) sendEmail(to, subject, textBody string, headers map[string]string) error {
payload := map[string]interface{}{
"from": d.fromEmail,
"to": []string{to},
"subject": subject,
"text": textBody,
}
if len(headers) > 0 {
payload["headers"] = headers
}
jsonData, err := json.Marshal(payload)
if err != nil {
return fmt.Errorf("marshaling email payload: %w", err)
}
req, err := http.NewRequest("POST", "https://api.resend.com/emails", bytes.NewReader(jsonData))
if err != nil {
return fmt.Errorf("creating request: %w", err)
}
req.Header.Set("Authorization", "Bearer "+d.resendAPIKey)
req.Header.Set("Content-Type", "application/json")
resp, err := d.httpClient.Do(req)
if err != nil {
return fmt.Errorf("sending request: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
return fmt.Errorf("resend API returned %d: %s", resp.StatusCode, string(respBody))
}
return nil
}
func isEventEnabled(enabledEvents []string, eventType string) bool {
for _, e := range enabledEvents {
if e == eventType {
return true
}
}
return false
}
// SendClaimEmail delivers a customer-claim arc email (claim / reset / claimed confirmation) to
// the REGISTERED customer address — the claim.Mailer implementation. Every send result is
// logged + recorded in notification_log; the code itself never is (rule: plaintext exists only
// inside the send).
func (d *Dispatcher) SendClaimEmail(kind, customerID, email, domain, code string) error {
if d.resendAPIKey == "" {
d.logger.Printf("[ERROR] claim %s email for %s NOT sent: no Resend API key configured", kind, customerID)
return fmt.Errorf("notify: no resend api key")
}
subject, body := FormatClaimEmail(kind, customerID, domain, code)
eventType := "claim_" + kind
if err := d.sendEmailFn(email, subject, body, nil); err != nil {
d.logger.Printf("[ERROR] claim %s email to customer %s failed: %v", kind, customerID, err)
d.store.LogNotification(customerID, eventType, "info", subject, "failed", err.Error(), "customer")
return err
}
d.logger.Printf("[INFO] claim %s email sent to the registered address of %s", kind, customerID)
d.store.LogNotification(customerID, eventType, "info", subject, "sent", "", "customer")
return nil
}
// SendSelfBindEmail delivers the customer self-bind capability link (v0.66.0, R-27 slice 1) to the
// REGISTERED customer address. Sibling of SendClaimEmail — NOT routed through the claim engine. The
// link is the capability; it is logged only via the notification_log subject (which carries no
// token), never the raw link. On failure the caller (web) invalidates the just-minted token so it is
// not left silently live.
func (d *Dispatcher) SendSelfBindEmail(customerID, email, link string) error {
if d.resendAPIKey == "" {
d.logger.Printf("[ERROR] self-bind link email for %s NOT sent: no Resend API key configured", customerID)
return fmt.Errorf("notify: no resend api key")
}
subject, body := FormatSelfBindEmail(customerID, link)
if err := d.sendEmailFn(email, subject, body, nil); err != nil {
d.logger.Printf("[ERROR] self-bind link email to customer %s failed: %v", customerID, err)
d.store.LogNotification(customerID, "selfbind_link", "info", subject, "failed", err.Error(), "customer")
return err
}
d.logger.Printf("[INFO] self-bind link emailed to the registered address of %s", customerID)
d.store.LogNotification(customerID, "selfbind_link", "info", subject, "sent", "", "customer")
return nil
}