1 files changed, 213 insertions, 0 deletions
diff --git a/internal/controlplane/mail_operations.go b/internal/controlplane/mail_operations.go
new file mode 100644
index 0000000..aaaf2c5
--- /dev/null
+++ b/internal/controlplane/mail_operations.go
@@ -0,0 +1,213 @@
+package controlplane
+
+import (
+	"context"
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+const maxMailFeedbackBytes = 64 << 10
+
+type MailNotificationConfig struct {
+	LowBalanceMicros       int64
+	SpendAnomalyMultiplier int64
+	SpendAnomalyMinMicros  int64
+	Interval               time.Duration
+}
+
+type mailFeedback struct {
+	EventID   string `json:"event_id"`
+	EventType string `json:"event_type"`
+	Recipient string `json:"recipient"`
+	Provider  string `json:"provider"`
+	Detail    string `json:"detail"`
+}
+
+// MailFeedbackHandler accepts a provider-neutral normalized callback. An edge
+// adapter maps the provider's native event into this payload and signs
+// "<unix timestamp>.<raw body>" with HMAC-SHA256.
+func (s *Store) MailFeedbackHandler(secret string) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost {
+			w.Header().Set("Allow", http.MethodPost)
+			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+			return
+		}
+		body, err := io.ReadAll(io.LimitReader(r.Body, maxMailFeedbackBytes+1))
+		if err != nil || len(body) > maxMailFeedbackBytes || !validMailFeedbackSignature(secret, r.Header.Get("X-AIGW-Mail-Timestamp"), r.Header.Get("X-AIGW-Mail-Signature"), body, time.Now().UTC()) {
+			http.Error(w, "invalid feedback signature", http.StatusBadRequest)
+			return
+		}
+		var event mailFeedback
+		decoder := json.NewDecoder(strings.NewReader(string(body)))
+		decoder.DisallowUnknownFields()
+		if decoder.Decode(&event) != nil || strings.TrimSpace(event.EventID) == "" ||
+			!strings.Contains(event.Recipient, "@") || (event.EventType != "delivered" && event.EventType != "bounce" && event.EventType != "complaint") {
+			http.Error(w, "invalid feedback event", http.StatusBadRequest)
+			return
+		}
+		if err := s.applyMailFeedback(r.Context(), event); err != nil {
+			http.Error(w, "feedback persistence failed", http.StatusInternalServerError)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = io.WriteString(w, "{\"received\":true}\n")
+	})
+}
+
+func validMailFeedbackSignature(secret, timestamp, signature string, body []byte, now time.Time) bool {
+	if secret == "" || timestamp == "" || !strings.HasPrefix(signature, "sha256=") {
+		return false
+	}
+	seconds, err := strconv.ParseInt(timestamp, 10, 64)
+	if err != nil || now.Sub(time.Unix(seconds, 0)).Abs() > 5*time.Minute {
+		return false
+	}
+	provided, err := hex.DecodeString(strings.TrimPrefix(signature, "sha256="))
+	if err != nil {
+		return false
+	}
+	mac := hmac.New(sha256.New, []byte(secret))
+	_, _ = mac.Write([]byte(timestamp))
+	_, _ = mac.Write([]byte("."))
+	_, _ = mac.Write(body)
+	return hmac.Equal(provided, mac.Sum(nil))
+}
+
+func (s *Store) applyMailFeedback(ctx context.Context, event mailFeedback) error {
+	tx, err := s.db.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted})
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback(ctx)
+	tag, err := tx.Exec(ctx, `INSERT INTO mail_feedback_events (event_id,event_type,recipient,provider)
+		VALUES ($1,$2,lower($3),$4) ON CONFLICT DO NOTHING`, event.EventID, event.EventType, event.Recipient, event.Provider)
+	if err != nil || tag.RowsAffected() == 0 {
+		if err != nil {
+			return err
+		}
+		return tx.Commit(ctx)
+	}
+	if event.EventType == "bounce" || event.EventType == "complaint" {
+		detail := event.Detail
+		if len(detail) > 1000 {
+			detail = detail[:1000]
+		}
+		if _, err := tx.Exec(ctx, `INSERT INTO mail_suppressions
+			(recipient,reason,provider,provider_event_id,detail) VALUES (lower($1),$2,$3,$4,$5)
+			ON CONFLICT (recipient) DO UPDATE SET reason=EXCLUDED.reason,provider=EXCLUDED.provider,
+			provider_event_id=EXCLUDED.provider_event_id,detail=EXCLUDED.detail,updated_at=now()`,
+			event.Recipient, event.EventType, event.Provider, event.EventID, detail); err != nil {
+			return err
+		}
+		if _, err := tx.Exec(ctx, `UPDATE console_mail_outbox SET status='suppressed',last_error=$2,claimed_at=NULL
+			WHERE lower(recipient)=lower($1) AND status IN ('pending','retry','sending')`, event.Recipient, "recipient suppressed after "+event.EventType); err != nil {
+			return err
+		}
+	}
+	return tx.Commit(ctx)
+}
+
+func (s *Store) RunMailNotificationWorker(ctx context.Context, config MailNotificationConfig, logger *slog.Logger) {
+	if config.Interval <= 0 {
+		config.Interval = 5 * time.Minute
+	}
+	if logger == nil {
+		logger = slog.Default()
+	}
+	ticker := time.NewTicker(config.Interval)
+	defer ticker.Stop()
+	for {
+		if err := s.queueBillingNotifications(ctx, config); err != nil && !errors.Is(err, context.Canceled) {
+			logger.Warn("billing_notification_scan_failed", "error", err)
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+		}
+	}
+}
+
+func (s *Store) queueBillingNotifications(ctx context.Context, config MailNotificationConfig) error {
+	rows, err := s.db.Query(ctx, `WITH spend AS (
+		SELECT tenant_id,
+			COALESCE(sum(-amount_micros) FILTER (WHERE kind='usage' AND created_at>=date_trunc('day',now())),0)::bigint today,
+			(COALESCE(sum(-amount_micros) FILTER (WHERE kind='usage' AND created_at>=date_trunc('day',now())-interval '7 days' AND created_at<date_trunc('day',now())),0)/7)::bigint baseline
+		FROM billing_ledger GROUP BY tenant_id)
+		SELECT w.tenant_id::text,w.currency,w.balance_micros-w.reserved_micros,u.email,u.display_name,
+			COALESCE(spend.today,0),COALESCE(spend.baseline,0)
+		FROM tenant_wallets w JOIN console_users u ON u.tenant_id=w.tenant_id
+		LEFT JOIN spend ON spend.tenant_id=w.tenant_id
+		WHERE u.status='active' AND u.email_verified_at IS NOT NULL AND u.role IN ('tenant_admin','tenant_billing')
+		AND NOT EXISTS (SELECT 1 FROM mail_suppressions s WHERE lower(s.recipient)=lower(u.email))`)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var tenantID, currency, email, name string
+		var available, today, baseline int64
+		if err := rows.Scan(&tenantID, &currency, &available, &email, &name, &today, &baseline); err != nil {
+			return err
+		}
+		day := time.Now().UTC().Format("2006-01-02")
+		if available <= config.LowBalanceMicros {
+			body := fmt.Sprintf("Hi %s,\n\nYour AIGW prepaid balance is low: %.6f %s remains available. Add funds to avoid interrupted API access.\n", displayName(name), float64(available)/1_000_000, strings.ToUpper(currency))
+			if err := s.queueNotification(ctx, tenantID, email, "low_balance", day, "AIGW balance is low", body); err != nil {
+				return err
+			}
+		}
+		if baseline > 0 && today >= config.SpendAnomalyMinMicros && today >= baseline*config.SpendAnomalyMultiplier {
+			body := fmt.Sprintf("Hi %s,\n\nAIGW detected unusual API spend today: %.6f %s versus a seven-day daily baseline of %.6f %s. Review API keys and usage in the console.\n", displayName(name), float64(today)/1_000_000, strings.ToUpper(currency), float64(baseline)/1_000_000, strings.ToUpper(currency))
+			if err := s.queueNotification(ctx, tenantID, email, "spend_anomaly", day, "Unusual AIGW API spend detected", body); err != nil {
+				return err
+			}
+		}
+	}
+	return rows.Err()
+}
+
+func (s *Store) queueNotification(ctx context.Context, tenantID, recipient, kind, dedupe, subject, body string) error {
+	tx, err := s.db.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted})
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback(ctx)
+	tag, err := tx.Exec(ctx, `INSERT INTO mail_notification_events (tenant_id,recipient,notification_type,dedupe_key)
+		VALUES ($1,lower($2),$3,$4) ON CONFLICT DO NOTHING`, tenantID, recipient, kind, dedupe)
+	if err != nil || tag.RowsAffected() == 0 {
+		if err != nil {
+			return err
+		}
+		return tx.Commit(ctx)
+	}
+	ciphertext, err := s.cipher.Encrypt(body)
+	if err != nil {
+		return err
+	}
+	if _, err := tx.Exec(ctx, `INSERT INTO console_mail_outbox (recipient,template,subject,body_ciphertext)
+		VALUES (lower($1),$2,$3,$4)`, recipient, kind, subject, ciphertext); err != nil {
+		return err
+	}
+	return tx.Commit(ctx)
+}
+
+func displayName(value string) string {
+	if value = strings.TrimSpace(value); value != "" {
+		return value
+	}
+	return "there"
+}