DoGaMa-serv/internal/notification/service.go

// Package notification manages encrypted channels and bounded asynchronous delivery.
package notification

import (
	"bytes"
	"context"
	"crypto/aes"
	"crypto/cipher"
	"crypto/hmac"
	"crypto/rand"
	"crypto/sha256"
	"crypto/tls"
	"database/sql"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/netip"
	"net/smtp"
	"net/url"
	"strconv"
	"strings"
	"time"
)

type Channel struct {
	ID, Name, Type string
	Enabled        bool
	Events         []string
	Configured     bool
}
type Input struct {
	Name, Type string
	Enabled    bool
	Events     []string
	Config     map[string]string
}
type Event struct{ Type, Title, Message, InstanceName, OperationID string }
type Service struct {
	db       *sql.DB
	aead     cipher.AEAD
	client   *http.Client
	resolver *net.Resolver
	now      func() time.Time
}

func New(db *sql.DB, key []byte) (*Service, error) {
	if len(key) != 32 {
		return nil, errors.New("notification encryption key must be exactly 32 bytes")
	}
	block, err := aes.NewCipher(key)
	if err != nil {
		return nil, err
	}
	aead, err := cipher.NewGCM(block)
	if err != nil {
		return nil, err
	}
	s := &Service{db: db, aead: aead, resolver: net.DefaultResolver, now: time.Now}
	transport := http.DefaultTransport.(*http.Transport).Clone()
	transport.DialContext = s.dialSafe
	s.client = &http.Client{Transport: transport, Timeout: 10 * time.Second, CheckRedirect: func(req *http.Request, via []*http.Request) error {
		if len(via) >= 3 {
			return errors.New("too many redirects")
		}
		return s.validateURL(req.Context(), req.URL)
	}}
	return s, nil
}

func (s *Service) Upsert(ctx context.Context, id string, in Input) (Channel, error) {
	if strings.TrimSpace(in.Name) == "" || !validType(in.Type) {
		return Channel{}, errors.New("invalid notification channel")
	}
	if err := validateConfigShape(in.Type, in.Config); err != nil {
		return Channel{}, err
	}
	encrypted, err := s.seal(in.Config)
	if err != nil {
		return Channel{}, err
	}
	events, _ := json.Marshal(normalizeEvents(in.Events))
	now := s.now().UTC().Format(time.RFC3339Nano)
	if id == "" {
		id = randomID()
	}
	_, err = s.db.ExecContext(ctx, `INSERT INTO notification_channels(id,name,type,enabled,encrypted_config,event_filter_json,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(id) DO UPDATE SET name=excluded.name,type=excluded.type,enabled=excluded.enabled,encrypted_config=excluded.encrypted_config,event_filter_json=excluded.event_filter_json,updated_at=excluded.updated_at`, id, strings.TrimSpace(in.Name), in.Type, in.Enabled, encrypted, string(events), now, now)
	if err != nil {
		return Channel{}, fmt.Errorf("save notification channel: %w", err)
	}
	return Channel{ID: id, Name: strings.TrimSpace(in.Name), Type: in.Type, Enabled: in.Enabled, Events: normalizeEvents(in.Events), Configured: true}, nil
}
func (s *Service) List(ctx context.Context) ([]Channel, error) {
	rows, err := s.db.QueryContext(ctx, `SELECT id,name,type,enabled,event_filter_json,length(encrypted_config)>0 FROM notification_channels ORDER BY name COLLATE NOCASE`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Channel
	for rows.Next() {
		var c Channel
		var body string
		if err := rows.Scan(&c.ID, &c.Name, &c.Type, &c.Enabled, &body, &c.Configured); err != nil {
			return nil, err
		}
		_ = json.Unmarshal([]byte(body), &c.Events)
		out = append(out, c)
	}
	return out, rows.Err()
}
func (s *Service) Delete(ctx context.Context, id string) error {
	_, err := s.db.ExecContext(ctx, `DELETE FROM notification_channels WHERE id=?`, id)
	return err
}
func (s *Service) Queue(ctx context.Context, event Event) error {
	if !validEvent(event.Type) || len(event.Message) > 1000 {
		return errors.New("invalid notification event")
	}
	payload, _ := json.Marshal(event)
	rows, err := s.db.QueryContext(ctx, `SELECT id,event_filter_json FROM notification_channels WHERE enabled=1`)
	if err != nil {
		return err
	}
	var channelIDs []string
	for rows.Next() {
		var id, filter string
		if err := rows.Scan(&id, &filter); err != nil {
			rows.Close()
			return err
		}
		var events []string
		_ = json.Unmarshal([]byte(filter), &events)
		if matches(events, event.Type) {
			channelIDs = append(channelIDs, id)
		}
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return err
	}
	rows.Close()
	now := s.now().UTC().Format(time.RFC3339Nano)
	for _, id := range channelIDs {
		if _, err := s.db.ExecContext(ctx, `INSERT INTO notification_deliveries(id,channel_id,event_type,payload_redacted,next_attempt_at,created_at) VALUES(?,?,?,?,?,?)`, randomID(), id, event.Type, string(payload), now, now); err != nil {
			return err
		}
	}
	return nil
}
func (s *Service) Test(ctx context.Context, id string) error {
	return s.queueForChannel(ctx, id, Event{Type: "notification.test", Title: "DoGaMa test notification", Message: "This is a test notification from DoGaMa."})
}
func (s *Service) queueForChannel(ctx context.Context, id string, event Event) error {
	var enabled bool
	if err := s.db.QueryRowContext(ctx, `SELECT enabled FROM notification_channels WHERE id=?`, id).Scan(&enabled); err != nil {
		return err
	}
	payload, _ := json.Marshal(event)
	now := s.now().UTC().Format(time.RFC3339Nano)
	_, err := s.db.ExecContext(ctx, `INSERT INTO notification_deliveries(id,channel_id,event_type,payload_redacted,next_attempt_at,created_at) VALUES(?,?,?,?,?,?)`, randomID(), id, event.Type, string(payload), now, now)
	return err
}

func (s *Service) RunDue(ctx context.Context) error {
	rows, err := s.db.QueryContext(ctx, `SELECT d.id,d.attempt,d.payload_redacted,c.type,c.encrypted_config FROM notification_deliveries d JOIN notification_channels c ON c.id=d.channel_id WHERE d.status IN ('queued','retrying') AND d.next_attempt_at<=? ORDER BY d.next_attempt_at LIMIT 20`, s.now().UTC().Format(time.RFC3339Nano))
	if err != nil {
		return err
	}
	type job struct {
		id, typ, payload string
		attempt          int
		encrypted        []byte
	}
	var jobs []job
	for rows.Next() {
		var j job
		if err := rows.Scan(&j.id, &j.attempt, &j.payload, &j.typ, &j.encrypted); err != nil {
			rows.Close()
			return err
		}
		jobs = append(jobs, j)
	}
	rows.Close()
	for _, j := range jobs {
		config, e := s.open(j.encrypted)
		if e == nil {
			e = s.deliver(ctx, j.id, j.typ, config, []byte(j.payload))
		}
		attempt := j.attempt + 1
		if e == nil {
			if _, updateErr := s.db.ExecContext(ctx, `UPDATE notification_deliveries SET status='succeeded',attempt=?,completed_at=?,last_error_code='' WHERE id=?`, attempt, s.now().UTC().Format(time.RFC3339Nano), j.id); updateErr != nil {
				return updateErr
			}
		} else {
			status := "retrying"
			if attempt >= 5 {
				status = "failed"
			}
			delay := time.Duration(1<<min(attempt, 6)) * time.Minute
			_, _ = s.db.ExecContext(ctx, `UPDATE notification_deliveries SET status=?,attempt=?,next_attempt_at=?,last_error_code=? WHERE id=?`, status, attempt, s.now().UTC().Add(delay).Format(time.RFC3339Nano), errorCode(e), j.id)
		}
	}
	return nil
}
func (s *Service) deliver(ctx context.Context, id, typ string, c map[string]string, payload []byte) error {
	if typ == "email" {
		host := c["host"]
		port := c["port"]
		if port == "" {
			port = "587"
		}
		addr := net.JoinHostPort(host, port)
		var auth smtp.Auth
		if c["username"] != "" {
			auth = smtp.PlainAuth("", c["username"], c["password"], host)
		}
		msg := []byte("To: " + c["to"] + "\r\nSubject: DoGaMa notification\r\nContent-Type: application/json\r\n\r\n" + string(payload))
		return sendSMTP(ctx, addr, host, auth, c["from"], strings.Split(c["to"], ","), msg)
	}
	u, err := url.Parse(c["url"])
	if err != nil {
		return err
	}
	if err = s.validateURL(ctx, u); err != nil {
		return err
	}
	body := payload
	if typ == "discord" {
		var event Event
		_ = json.Unmarshal(payload, &event)
		body, _ = json.Marshal(map[string]string{"content": event.Title + "\n" + event.Message})
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, u.String(), bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("X-DoGaMa-Event-ID", id)
	timestamp := strconv.FormatInt(s.now().Unix(), 10)
	req.Header.Set("X-DoGaMa-Timestamp", timestamp)
	if secret := c["signing_secret"]; secret != "" {
		mac := hmac.New(sha256.New, []byte(secret))
		_, _ = mac.Write([]byte(timestamp + "." + string(body)))
		req.Header.Set("X-DoGaMa-Signature", "sha256="+hex.EncodeToString(mac.Sum(nil)))
	}
	resp, err := s.client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4096))
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return fmt.Errorf("remote_status_%d", resp.StatusCode)
	}
	return nil
}
func (s *Service) validateURL(ctx context.Context, u *url.URL) error {
	if u.Scheme != "https" || u.User != nil || u.Hostname() == "" {
		return errors.New("unsafe_destination")
	}
	ips, err := s.resolver.LookupNetIP(ctx, "ip", u.Hostname())
	if err != nil {
		return errors.New("destination_resolution_failed")
	}
	for _, ip := range ips {
		if unsafeIP(ip) {
			return errors.New("unsafe_destination")
		}
	}
	return nil
}

func (s *Service) dialSafe(ctx context.Context, network, address string) (net.Conn, error) {
	host, port, err := net.SplitHostPort(address)
	if err != nil {
		return nil, errors.New("unsafe_destination")
	}
	ips, err := s.resolver.LookupNetIP(ctx, "ip", host)
	if err != nil || len(ips) == 0 {
		return nil, errors.New("destination_resolution_failed")
	}
	dialer := net.Dialer{Timeout: 10 * time.Second}
	for _, ip := range ips {
		if unsafeIP(ip) {
			return nil, errors.New("unsafe_destination")
		}
		connection, dialErr := dialer.DialContext(ctx, network, net.JoinHostPort(ip.String(), port))
		if dialErr == nil {
			return connection, nil
		}
	}
	return nil, errors.New("delivery_failed")
}

func sendSMTP(ctx context.Context, address, host string, auth smtp.Auth, from string, recipients []string, message []byte) error {
	dialer := net.Dialer{Timeout: 10 * time.Second}
	connection, err := dialer.DialContext(ctx, "tcp", address)
	if err != nil {
		return err
	}
	client, err := smtp.NewClient(connection, host)
	if err != nil {
		_ = connection.Close()
		return err
	}
	defer client.Close()
	if ok, _ := client.Extension("STARTTLS"); !ok {
		return errors.New("smtp_tls_required")
	}
	if err := client.StartTLS(&tls.Config{ServerName: host, MinVersion: tls.VersionTLS12}); err != nil {
		return err
	}
	if auth != nil {
		if err := client.Auth(auth); err != nil {
			return err
		}
	}
	if err := client.Mail(from); err != nil {
		return err
	}
	for _, recipient := range recipients {
		if err := client.Rcpt(strings.TrimSpace(recipient)); err != nil {
			return err
		}
	}
	w, err := client.Data()
	if err != nil {
		return err
	}
	if _, err = w.Write(message); err != nil {
		_ = w.Close()
		return err
	}
	if err = w.Close(); err != nil {
		return err
	}
	return client.Quit()
}
func unsafeIP(ip netip.Addr) bool {
	return !ip.IsValid() || ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || ip.IsMulticast() || ip.IsUnspecified()
}
func (s *Service) seal(config map[string]string) ([]byte, error) {
	body, _ := json.Marshal(config)
	nonce := make([]byte, s.aead.NonceSize())
	if _, err := rand.Read(nonce); err != nil {
		return nil, err
	}
	return s.aead.Seal(nonce, nonce, body, nil), nil
}
func (s *Service) open(body []byte) (map[string]string, error) {
	n := s.aead.NonceSize()
	if len(body) < n {
		return nil, errors.New("invalid encrypted channel")
	}
	plain, err := s.aead.Open(nil, body[:n], body[n:], nil)
	if err != nil {
		return nil, err
	}
	var out map[string]string
	err = json.Unmarshal(plain, &out)
	return out, err
}
func validType(v string) bool { return v == "email" || v == "webhook" || v == "discord" }
func validEvent(v string) bool {
	return v == "notification.test" || strings.HasSuffix(v, ".failed") || strings.HasSuffix(v, ".completed") || strings.HasSuffix(v, ".required")
}
func validateConfigShape(typ string, c map[string]string) error {
	if typ == "email" {
		if c["host"] == "" || c["from"] == "" || c["to"] == "" {
			return errors.New("email host, from and to are required")
		}
		return nil
	}
	u, err := url.Parse(c["url"])
	if err != nil || u.Scheme != "https" || u.Hostname() == "" || u.User != nil {
		return errors.New("an HTTPS webhook URL is required")
	}
	return nil
}
func normalizeEvents(in []string) []string {
	seen := map[string]bool{}
	out := []string{}
	for _, v := range in {
		v = strings.TrimSpace(v)
		if validEvent(v) && !seen[v] {
			seen[v] = true
			out = append(out, v)
		}
	}
	return out
}
func matches(filter []string, event string) bool {
	for _, v := range filter {
		if v == event || v == "*" {
			return true
		}
	}
	return false
}
func errorCode(err error) string {
	v := err.Error()
	if strings.HasPrefix(v, "remote_status_") {
		return v
	}
	switch v {
	case "unsafe_destination", "destination_resolution_failed":
		return v
	}
	return "delivery_failed"
}
func randomID() string {
	b := make([]byte, 18)
	_, _ = rand.Read(b)
	return base64.RawURLEncoding.EncodeToString(b)
}
func min(a, b int) int {
	if a < b {
		return a
	}
	return b
}