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 }