felhom.eu/hub/internal/notify/dispatcher.go

package notify

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"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"},
}

// 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
	}

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

func (d *Dispatcher) processOperator(customerID, eventType, severity, message, detailsJSON, source string) {
	if !d.operatorOn || d.operatorEmail == "" {
		return
	}

	cooldownKey := customerID + ":" + eventType
	d.mu.Lock()
	if last, ok := d.opCooldowns[cooldownKey]; ok && time.Since(last) < 1*time.Hour {
		d.mu.Unlock()
		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")
}

func (d *Dispatcher) processCustomer(customerID, eventType, severity, message, detailsJSON, source string) {
	// 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
}