worker/internal/distworker/notification.go

package distworker

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"log"
	"time"

	"rocketgit.ru/rsmon/worker/app/models"
	"rocketgit.ru/rsmon/worker/internal/sender"
	"rocketgit.ru/rsmon/worker/internal/wire"
)

// ExecuteNotification runs one NotificationTask end-to-end. It looks up the
// matching credential from the worker's in-memory cache (pushed via the init
// / config websocket message) and dispatches to the per-method executor.
// Returns a NotificationResultReport ready to send back to the control plane.
//
// Phase 1 of docs/plans/worker-notifier-mvp.md ships four supported methods
// (email / telegram / webhook / mattermost). sms and voice are explicit
// permanent failures with status=unsupported_method so the operator knows
// the worker received the task and chose not to deliver it (rather than
// silently dropping it as the production plan forbids).
//
//nolint:gocritic // task model is shared with the rest of the dispatcher; keep by-value
func (r *Runner) ExecuteNotification(ctx context.Context, task models.Task) (report wire.NotificationResultReport) {
	start := time.Now()
	method := ""
	report = wire.NotificationResultReport{
		JobID:      task.JobID,
		Status:     wire.NotificationResultPermanent,
		DurationMs: 0,
	}
	defer func() {
		if recover() != nil {
			report.Status = wire.NotificationResultPermanent
			report.ProviderResponse = nil
			report.RetryAfterSeconds = nil
			report.Error = stringPtr("notification executor panic")
		}
		report.DurationMs = int(time.Since(start) / time.Millisecond)
		r.recordDelegatedNotification(report.JobID, method, report)
	}()
	if r.notificationExecutor != nil {
		return r.notificationExecutor(ctx, task)
	}

	if len(task.Payload) == 0 {
		report.Status = wire.NotificationResultPermanent
		report.Error = stringPtr("notification task payload is empty")
		report.DurationMs = int(time.Since(start) / time.Millisecond)
		return report
	}

	var nt wire.NotificationTask
	if err := json.Unmarshal(task.Payload, &nt); err != nil {
		report.Error = stringPtr("invalid notification payload: " + err.Error())
		report.DurationMs = int(time.Since(start) / time.Millisecond)
		return report
	}
	if nt.JobID == "" {
		nt.JobID = task.JobID
	}
	method = nt.Method
	if nt.MessageID == 0 && task.MessageID != nil {
		nt.MessageID = *task.MessageID
	}
	report.JobID = nt.JobID
	report.LeaseToken = nt.LeaseToken
	report.MessageID = nt.MessageID
	if task.Deadline != nil && !task.Deadline.After(time.Now()) {
		report.Error = stringPtr("notification deadline expired")
		report.DurationMs = int(time.Since(start) / time.Millisecond)
		return report
	}

	creds := r.Credentials()
	if creds == nil {
		report.Error = stringPtr("worker has no notification credentials pushed yet")
		report.DurationMs = int(time.Since(start) / time.Millisecond)
		return report
	}

	status, providerResp, errStr, retry, execErr := r.deliverNotification(ctx, nt, creds)
	if task.Deadline != nil && !task.Deadline.After(time.Now()) {
		status, providerResp, errStr, retry, execErr = wire.NotificationResultPermanent, "", "notification deadline expired", nil, context.DeadlineExceeded
	}
	report.Status = status
	report.ProviderResponse = stringPtrOrNil(providerResp)
	report.Error = stringPtrOrNil(errStr)
	if retry != nil {
		report.RetryAfterSeconds = retry
	}
	report.DurationMs = int(time.Since(start) / time.Millisecond)
	if execErr != nil {
		log.Printf("worker: notification delivery job=%s message=%d method=%s status=%s error_class=delivery_failed duration_ms=%d",
			nt.JobID, nt.MessageID, nt.Method, status, report.DurationMs)
	}
	return report
}

// deliverNotification dispatches one notification task to the per-method
// sub-executor. Returns (status, providerResponse, error, retryAfterSeconds,
// internalError). The internalError is non-nil only for unexpected panics or
// credential-resolution failures the result report should log.
//
//nolint:gocritic // wire payload is shared; keep by-value
func (r *Runner) deliverNotification(
	ctx context.Context,
	nt wire.NotificationTask,
	creds *wire.NotificationCredentials,
) (string, string, string, *int, error) {
	switch nt.Method {
	case systemContactKindEmail:
		if len(creds.SMTP) == 0 {
			return wire.NotificationResultPermanent,
				"", "no SMTP credential authorized for this worker", nil,
				errors.New("smtp: no credential")
		}
		cred, ok := selectSMTPCredential(creds.SMTP, nt.CredentialID)
		if !ok {
			return wire.NotificationResultPermanent, "", "requested SMTP credential is not authorized for this worker",
				nil, errors.New("smtp: credential not found")
		}
		err := sender.SendEmailWithCredentialContext(ctx,
			nt.Contact.Value, nt.Subject, nt.BodyText, nt.BodyHTML, &cred,
		)
		return classifyResult(err, "")

	case "telegram":
		if len(creds.Telegram) == 0 {
			return wire.NotificationResultPermanent, "", "no Telegram credential authorized for this worker",
				nil, errors.New("telegram: no credential")
		}
		// Convert wire.TelegramCredential -> models.NotificationCredential
		// for the existing tg.SendMessageWithToken entry point. We do not
		// persist this; it lives only on the goroutine stack.
		cred, ok := selectTelegramCredential(creds.Telegram, nt.CredentialID)
		if !ok {
			return wire.NotificationResultPermanent, "", "requested Telegram credential is not authorized for this worker",
				nil, errors.New("telegram: credential not found")
		}
		nc := wireTelegramToModel(&cred)
		err := sender.SendTelegramWithCredentialContext(ctx,
			nt.Contact.Value, nt.Subject, nt.BodyMarkdown, nc,
		)
		return classifyResult(err, "")

	case "webhook":
		signingSecret := ""
		if creds.Webhook != nil {
			signingSecret = creds.Webhook.SigningSecret
		}
		body, mErr := json.Marshal(nt)
		if mErr != nil {
			return wire.NotificationResultPermanent, "", "marshal webhook payload: " + mErr.Error(), nil, mErr
		}
		resp, err := sender.SendWebhookWithCredentialContext(ctx, body, nt.Contact.Value, signingSecret)
		return classifyResult(err, stringOrEmpty(resp))

	case "mattermost":
		var mc *wire.MattermostCredential
		if creds.Mattermost != nil {
			mc = creds.Mattermost
		}
		resp, err := sender.SendMattermostWithCredentialContext(ctx,
			nt.Contact.Value, nt.MessageKind, nt.Subject, nt.BodyMarkdown, mc,
		)
		return classifyResult(err, stringOrEmpty(resp))

	case "sms", "voice":
		return wire.NotificationResultPermanent, "", "unsupported_method: " + nt.Method,
			nil, errors.New("unsupported method: " + nt.Method)
	}

	if err := ctx.Err(); err != nil {
		return wire.NotificationResultRetryable, "", "context canceled: " + err.Error(), intPtr(5), err
	}
	return wire.NotificationResultPermanent, "", "unknown notification method: " + nt.Method,
		nil, errors.New("unknown method: " + nt.Method)
}

func selectSMTPCredential(creds []wire.SMTPCredential, id *int64) (wire.SMTPCredential, bool) {
	if id != nil {
		for _, cred := range creds {
			if cred.ID == *id {
				return cred, true
			}
		}
		return wire.SMTPCredential{}, false
	}
	return creds[0], true
}

func selectTelegramCredential(creds []wire.TelegramCredential, id *int64) (wire.TelegramCredential, bool) {
	if id != nil {
		for _, cred := range creds {
			if cred.ID == *id {
				return cred, true
			}
		}
		return wire.TelegramCredential{}, false
	}
	return creds[0], true
}

// classifyResult translates a delivery error into the wire status enum.
// SMTP 421 / 4xx with a Retry-After hint becomes retryable + suggested
// delay. Network errors become retryable with a fixed 30s backoff. Anything
// else is permanent.
func classifyResult(err error, providerResp string) (string, string, string, *int, error) {
	if err == nil {
		return wire.NotificationResultDelivered, providerResp, "", nil, nil
	}
	errStr := err.Error()

	// Heuristic: any 4xx SMTP response is retryable. The legacy sender does
	// not parse this, so the worker does a substring match on the error.
	if isRetryableSMTP(errStr) {
		retry := 30
		return wire.NotificationResultRetryable, providerResp, errStr, &retry, err
	}
	// Network / DNS / TLS / context errors are typically transient.
	if isTransientTransport(errStr) {
		retry := 15
		return wire.NotificationResultRetryable, providerResp, errStr, &retry, err
	}
	// Telegram "chat not found", webhook 4xx response (non-5xx) -> permanent.
	return wire.NotificationResultPermanent, providerResp, errStr, nil, err
}

func isRetryableSMTP(s string) bool {
	prefixes := []string{"smtp: 4", "421", "450", "451", "452"}
	for _, p := range prefixes {
		if len(s) >= len(p) && s[:len(p)] == p {
			return true
		}
	}
	return false
}

func isTransientTransport(s string) bool {
	markers := []string{"timeout", "tempor", "connection refused", "no such host", "i/o timeout", "tls"}
	for _, m := range markers {
		if bytes.Contains([]byte(s), []byte(m)) {
			return true
		}
	}
	return false
}

func stringOrEmpty(s *string) string {
	if s == nil {
		return ""
	}
	return *s
}

func stringPtr(s string) *string { return &s }

func stringPtrOrNil(s string) *string {
	if s == "" {
		return nil
	}
	return &s
}

func intPtr(i int) *int { return &i }

// wireTelegramToModel translates a wire.TelegramCredential (init push shape)
// into a models.NotificationCredential (DB shape) so it can be passed to the
// existing tg.SendMessageWithToken entry point. The BotName/APIURL fields
// are preserved; the secret is the bot token.
func wireTelegramToModel(wt *wire.TelegramCredential) *models.NotificationCredential {
	nc := &models.NotificationCredential{
		Kind: models.CredentialKindTelegram,
		Name: wt.Name,
	}
	if wt.BotName != "" {
		nc.BotName = &wt.BotName
	}
	if wt.APIURL != "" {
		nc.APIURL = &wt.APIURL
	}
	// Store the token in SecretEnc without encryption; SendMessageWithToken
	// does not read SecretEnc directly — it expects the helper to decrypt
	// via GetSecret. We go around that by writing the plaintext into the
	// struct field via the SetSecret path. Plain prefix lets the legacy
	// decrypt path return the raw value.
	nc.SetSecret(wt.Token) //nolint:errcheck // best-effort; failure becomes runtime error in SendMessageWithToken
	return nc
}