dnswatcher/internal/notify/notify.go

// Package notify provides notification delivery to Slack,
// Mattermost, and ntfy.
package notify

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/url"
	"sync"
	"sync/atomic"
	"time"

	"go.uber.org/fx"

	"sneak.berlin/go/dnswatcher/internal/config"
	"sneak.berlin/go/dnswatcher/internal/logger"
)

// HTTP client timeout.
const httpClientTimeout = 10 * time.Second

// HTTP status code thresholds.
const httpStatusClientError = 400

// Sentinel errors for notification failures.
var (
	// ErrNtfyFailed indicates the ntfy request failed.
	ErrNtfyFailed = errors.New("ntfy notification failed")
	// ErrSlackFailed indicates the Slack request failed.
	ErrSlackFailed = errors.New("slack notification failed")
	// ErrMattermostFailed indicates the Mattermost request failed.
	ErrMattermostFailed = errors.New(
		"mattermost notification failed",
	)
	// ErrInvalidScheme is returned for disallowed URL schemes.
	ErrInvalidScheme = errors.New("URL scheme not allowed")
	// ErrMissingHost is returned when a URL has no host.
	ErrMissingHost = errors.New("URL must have a host")
)

// IsAllowedScheme checks if the URL scheme is permitted.
func IsAllowedScheme(scheme string) bool {
	return scheme == "https" || scheme == "http"
}

// ValidateWebhookURL validates and sanitizes a webhook URL.
// It ensures the URL has an allowed scheme (http/https),
// a non-empty host, and returns a pre-parsed *url.URL
// reconstructed from validated components.
func ValidateWebhookURL(raw string) (*url.URL, error) {
	u, err := url.ParseRequestURI(raw)
	if err != nil {
		// A *url.Error quotes the whole URL, and this error is
		// logged at startup. A webhook URL's path or query lets
		// anyone post to the channel or topic, so keep only the
		// reason.
		var urlErr *url.Error
		if errors.As(err, &urlErr) {
			err = urlErr.Err
		}

		return nil, fmt.Errorf("invalid URL: %w", err)
	}

	if !IsAllowedScheme(u.Scheme) {
		return nil, fmt.Errorf(
			"%w: %s", ErrInvalidScheme, u.Scheme,
		)
	}

	if u.Host == "" {
		return nil, fmt.Errorf("%w", ErrMissingHost)
	}

	// Reconstruct from parsed components.
	clean := &url.URL{
		Scheme:   u.Scheme,
		Host:     u.Host,
		Path:     u.Path,
		RawQuery: u.RawQuery,
	}

	return clean, nil
}

// newRequest creates an http.Request from a pre-validated *url.URL.
// This avoids passing URL strings to http.NewRequestWithContext,
// which gosec flags as a potential SSRF vector.
func newRequest(
	ctx context.Context,
	method string,
	target *url.URL,
	body io.Reader,
) *http.Request {
	return (&http.Request{
		Method: method,
		URL:    target,
		Host:   target.Host,
		Header: make(http.Header),
		Body:   io.NopCloser(body),
	}).WithContext(ctx)
}

// Params contains dependencies for Service.
type Params struct {
	fx.In

	Logger *logger.Logger
	Config *config.Config
}

// Service provides notification functionality.
type Service struct {
	log                  *slog.Logger
	transport            http.RoundTripper
	config               *config.Config
	ntfyURL              *url.URL
	slackWebhookURL      *url.URL
	mattermostWebhookURL *url.URL
	history              *AlertHistory
	retryConfig          RetryConfig
	sleepFn              func(time.Duration) <-chan time.Time

	// Shutdown draining state.  drainMu guards draining and
	// serialises it against the counter increment in
	// startDelivery; inFlight tracks the delivery goroutines
	// themselves and outstanding mirrors its count so a timed
	// out drain can report how many were abandoned.
	drainMu     sync.Mutex
	draining    bool
	inFlight    sync.WaitGroup
	outstanding atomic.Int64
	abandon     chan struct{}
	abandonOnce sync.Once
}

// newService builds a Service with the fields every Service
// needs regardless of how it was constructed.
func newService(
	log *slog.Logger,
	transport http.RoundTripper,
) *Service {
	return &Service{
		log:       log,
		transport: transport,
		history:   NewAlertHistory(),
		abandon:   make(chan struct{}),
	}
}

// New creates a new notify Service.
func New(
	lifecycle fx.Lifecycle,
	params Params,
) (*Service, error) {
	svc := newService(params.Logger.Get(), http.DefaultTransport)
	svc.config = params.Config

	if params.Config.NtfyTopic != "" {
		u, err := ValidateWebhookURL(
			params.Config.NtfyTopic,
		)
		if err != nil {
			return nil, fmt.Errorf(
				"invalid ntfy topic URL: %w", err,
			)
		}

		svc.ntfyURL = u
	}

	if params.Config.SlackWebhook != "" {
		u, err := ValidateWebhookURL(
			params.Config.SlackWebhook,
		)
		if err != nil {
			return nil, fmt.Errorf(
				"invalid slack webhook URL: %w", err,
			)
		}

		svc.slackWebhookURL = u
	}

	if params.Config.MattermostWebhook != "" {
		u, err := ValidateWebhookURL(
			params.Config.MattermostWebhook,
		)
		if err != nil {
			return nil, fmt.Errorf(
				"invalid mattermost webhook URL: %w", err,
			)
		}

		svc.mattermostWebhookURL = u
	}

	lifecycle.Append(fx.Hook{
		OnStop: func(ctx context.Context) error {
			svc.drain(ctx)

			return nil
		},
	})

	return svc, nil
}

// History returns the alert history for reading recent alerts.
func (svc *Service) History() *AlertHistory {
	return svc.history
}

// SendNotification sends a notification to all configured
// endpoints and records it in the alert history.
func (svc *Service) SendNotification(
	ctx context.Context,
	title, message, priority string,
) {
	svc.history.Add(AlertEntry{
		Timestamp: time.Now().UTC(),
		Title:     title,
		Message:   message,
		Priority:  priority,
	})

	svc.dispatchNtfy(ctx, title, message, priority)
	svc.dispatchSlack(ctx, title, message, priority)
	svc.dispatchMattermost(ctx, title, message, priority)
}

// dispatch delivers a notification to one endpoint on a
// tracked background goroutine.
//
// The delivery context is detached from ctx with
// context.WithoutCancel so that a cancelled caller does not
// kill a delivery already under way; the shutdown drain, not
// the caller, decides how long deliveries may keep running.
func (svc *Service) dispatch(
	ctx context.Context,
	endpoint string,
	send func(context.Context) error,
) {
	notifyCtx := context.WithoutCancel(ctx)

	svc.startDelivery(endpoint, func() {
		err := svc.deliverWithRetry(notifyCtx, endpoint, send)
		if err != nil {
			svc.log.Error(
				"failed to send notification after retries",
				"endpoint", endpoint,
				"error", err,
			)
		}
	})
}

func (svc *Service) dispatchNtfy(
	ctx context.Context,
	title, message, priority string,
) {
	if svc.ntfyURL == nil {
		return
	}

	svc.dispatch(ctx, "ntfy", func(c context.Context) error {
		return svc.sendNtfy(
			c, svc.ntfyURL, title, message, priority,
		)
	})
}

func (svc *Service) dispatchSlack(
	ctx context.Context,
	title, message, priority string,
) {
	if svc.slackWebhookURL == nil {
		return
	}

	svc.dispatch(ctx, "slack", func(c context.Context) error {
		return svc.sendSlack(
			c, svc.slackWebhookURL, "slack", ErrSlackFailed,
			title, message, priority,
		)
	})
}

func (svc *Service) dispatchMattermost(
	ctx context.Context,
	title, message, priority string,
) {
	if svc.mattermostWebhookURL == nil {
		return
	}

	svc.dispatch(
		ctx, "mattermost",
		func(c context.Context) error {
			return svc.sendSlack(
				c, svc.mattermostWebhookURL, "mattermost",
				ErrMattermostFailed, title, message, priority,
			)
		},
	)
}

func (svc *Service) sendNtfy(
	ctx context.Context,
	topicURL *url.URL,
	title, message, priority string,
) error {
	// The topic URL is left out of the log: it lets anyone who has
	// it read and post to the topic.
	svc.log.Debug(
		"sending ntfy notification",
		"endpoint", "ntfy",
		"title", title,
	)

	ctx, cancel := context.WithTimeout(
		ctx, httpClientTimeout,
	)
	defer cancel()

	body := bytes.NewBufferString(message)
	request := newRequest(
		ctx, http.MethodPost, topicURL, body,
	)

	request.Header.Set("Title", title)
	request.Header.Set("Priority", ntfyPriority(priority))

	resp, err := svc.transport.RoundTrip(request)
	if err != nil {
		return fmt.Errorf("sending ntfy request: %w", err)
	}

	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode >= httpStatusClientError {
		return fmt.Errorf(
			"%w: status %d",
			ErrNtfyFailed, resp.StatusCode,
		)
	}

	return nil
}

func ntfyPriority(priority string) string {
	switch priority {
	case "error":
		return "urgent"
	case "warning":
		return "high"
	case "success":
		return "default"
	case "info":
		return "low"
	default:
		return "default"
	}
}

// SlackPayload represents a Slack/Mattermost webhook payload.
type SlackPayload struct {
	Text        string            `json:"text"`
	Attachments []SlackAttachment `json:"attachments,omitempty"`
}

// SlackAttachment represents a Slack/Mattermost attachment.
type SlackAttachment struct {
	Color string `json:"color"`
	Title string `json:"title"`
	Text  string `json:"text"`
}

// sendSlack posts to a Slack or Mattermost incoming webhook, which
// take the same payload. endpoint, "slack" or "mattermost", names
// the webhook in the log in place of its URL, which lets anyone who
// has it post to the channel. An HTTP error status is returned
// wrapped in failed, ErrSlackFailed or ErrMattermostFailed, so the
// error names the endpoint.
func (svc *Service) sendSlack(
	ctx context.Context,
	webhookURL *url.URL,
	endpoint string,
	failed error,
	title, message, priority string,
) error {
	ctx, cancel := context.WithTimeout(
		ctx, httpClientTimeout,
	)
	defer cancel()

	svc.log.Debug(
		"sending webhook notification",
		"endpoint", endpoint,
		"title", title,
	)

	payload := SlackPayload{
		Attachments: []SlackAttachment{
			{
				Color: slackColor(priority),
				Title: title,
				Text:  message,
			},
		},
	}

	body, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf(
			"marshaling webhook payload: %w", err,
		)
	}

	request := newRequest(
		ctx, http.MethodPost, webhookURL,
		bytes.NewBuffer(body),
	)

	request.Header.Set("Content-Type", "application/json")

	resp, err := svc.transport.RoundTrip(request)
	if err != nil {
		return fmt.Errorf("sending webhook request: %w", err)
	}

	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode >= httpStatusClientError {
		return fmt.Errorf(
			"%w: status %d",
			failed, resp.StatusCode,
		)
	}

	return nil
}

func slackColor(priority string) string {
	switch priority {
	case "error":
		return "#dc3545"
	case "warning":
		return "#ffc107"
	case "success":
		return "#28a745"
	case "info":
		return "#17a2b8"
	default:
		return "#6c757d"
	}
}