feat: make notification delivery durable

This commit is contained in:
benya
2026-07-23 00:17:38 +03:00
parent 61b1c06205
commit 5d6fa00e94
24 changed files with 710 additions and 136 deletions
+2
View File
@@ -17,6 +17,8 @@ RMM_COOKIE_SECURE=true
RMM_PUBLIC_SCHEME=https
RMM_PUBLIC_URL=https://rmm.example.com
RMM_METRIC_RETENTION_DAYS=30
RMM_NOTIFICATION_RETENTION_DAYS=90
RMM_NOTIFICATION_MAX_ATTEMPTS=5
# Optional password recovery and alert notifications. Leave RMM_SMTP_HOST empty to disable e-mail.
# Use starttls for port 587 or tls for implicit TLS on port 465.
+1 -2
View File
@@ -13,6 +13,7 @@
- [x] Alerts: offline, WAN, memory, disk, packet loss, latency, WAN IP, commands.
- [x] Acknowledge/resolved lifecycle и audit events.
- [x] E-mail/Telegram notifications, пользовательские пороги, repeat и delivery history.
- [x] Durable notification queue: leased claims, restart recovery, exponential retry, dead-letter и retention.
- [x] Reverse tunnel, cloud LuCI и wildcard router domain.
- [x] DirectDNS удалён из продукта; адресация роутеров работает только через cloud/wildcard-домен и исходящий туннель.
- [x] UCI preview/apply/commit/revert/backup/restore operations.
@@ -30,8 +31,6 @@
## Следующая разработка
- [ ] Восстанавливать зависшие `queued`-доставки после перезапуска сервера; добавить retry/backoff, лимит попыток и dead-letter состояние.
- [ ] Добавить retention и плановую очистку `notification_deliveries`, а также определить политику защиты e-mail и Telegram Chat ID в SQLite.
- [ ] Добавить подтверждение e-mail и безопасную привязку Telegram-чата перед включением канала.
- [ ] Сделать встроенный центр уведомлений: unread/read, счётчик, переход к роутеру и обновление через SSE.
- [ ] Настройки по типам событий, maintenance/snooze и временное подавление алертов на период работ.
+1
View File
@@ -56,6 +56,7 @@ OpenWrt через исходящее соединение агента. Пол
- [x] Пользовательские пороги памяти, диска, потерь и latency.
- [x] Lifecycle deduplication, повтор открытой проблемы и журнал доставки.
- [x] Тестовая отправка из профиля.
- [x] Перезапускаемая очередь с lease, exponential retry, dead-letter и retention терминальной истории.
- [ ] Webhook-канал с подписанным payload и ротацией секрета.
- [ ] Per-device override и расписание тишины.
- [ ] Группировка нескольких событий в incident.
+2 -2
View File
@@ -57,12 +57,12 @@ Docker-сборке с фиктивным роутером и длинными
- [x] Персональные настройки e-mail/Telegram и порогов алертов.
- [x] Warning, critical, resolved, repeat и тестовая отправка.
- [x] История queued/sent/failed использует только masked destination.
- [x] История queued/sending/retry/sent/dead-letter использует только masked destination.
- [x] Повторить browser matrix нового profile-раздела перед production deploy.
- [x] При отсутствии SMTP и Telegram token каналы недоступны и объясняют причину.
- [x] Ошибка тестовой отправки показана по-русски и сохраняет request ID.
- [ ] Реальная доставка через production SMTP и Telegram проверена из пользовательского профиля.
- [ ] Failed-доставка и длинный текст ошибки проверены в истории без раскрытия внутренних деталей.
- [ ] Retry/dead-letter доставка и длинный текст ошибки проверены в истории без раскрытия внутренних деталей.
## Ограничения production-проверки
+2 -1
View File
@@ -14,6 +14,7 @@
- [x] Landing motion и reduced-motion fallback.
- [x] Desktop, 4:3, tablet и mobile без page-level overflow.
- [x] Notification settings, thresholds, test send и delivery history.
- [x] История различает queued/sending/retry/sent/dead-letter и показывает номер попытки и время повтора.
## Проверить перед выпуском notification UI
@@ -21,7 +22,7 @@
- [x] SMTP отсутствует: e-mail channel недоступен, остальные настройки сохраняются.
- [x] Telegram token отсутствует: Telegram channel недоступен.
- [ ] Неверный Chat ID и пороги показывают ошибку формы.
- [ ] Test send показывает sent и failed без раскрытия секретов.
- [ ] Test send показывает sent/retry/dead-letter без раскрытия секретов.
- [ ] История корректна при пустом списке и длинной ошибке доставки.
- [ ] Profile dialog прокручивается внутри 390×844 и 360×800.
- [ ] Keyboard navigation и zoom 200%.
+2 -1
View File
@@ -20,7 +20,8 @@
- [x] Настройки каналов e-mail/Telegram в личном кабинете.
- [x] Пороги memory/disk/packet loss/latency.
- [x] Warning/critical/resolved и период повтора.
- [x] Тестовая отправка и история sent/failed/queued.
- [x] Тестовая отправка и история queued/sending/retry/sent/dead-letter.
- [x] Retry/dead-letter состояния, номер попытки и время следующей доставки.
- [x] Недоступный серверный канал объясняется и блокируется.
- [ ] Quiet hours и выбор timezone.
- [ ] Per-device overrides в карточке объекта.
+2
View File
@@ -22,6 +22,8 @@ services:
RMM_PUBLIC_SCHEME: "${RMM_PUBLIC_SCHEME:-https}"
RMM_PUBLIC_URL: "${RMM_PUBLIC_URL:-}"
RMM_METRIC_RETENTION_DAYS: "${RMM_METRIC_RETENTION_DAYS:-30}"
RMM_NOTIFICATION_RETENTION_DAYS: "${RMM_NOTIFICATION_RETENTION_DAYS:-90}"
RMM_NOTIFICATION_MAX_ATTEMPTS: "${RMM_NOTIFICATION_MAX_ATTEMPTS:-5}"
RMM_SMTP_HOST: "${RMM_SMTP_HOST:-}"
RMM_SMTP_PORT: "${RMM_SMTP_PORT:-587}"
RMM_SMTP_USERNAME: "${RMM_SMTP_USERNAME:-}"
+12 -4
View File
@@ -350,10 +350,18 @@ The update request accepts:
E-mail requires SMTP on the server and an e-mail address in the user profile. Telegram
requires `RMM_TELEGRAM_BOT_TOKEN` on the server and a numeric Chat ID in the profile.
Deliveries are persisted with `queued`, `sent`, or `failed` status. Raw destinations and
server-side credentials are not returned by the API. Active and resolved events use
lifecycle deduplication; an optional repeat interval can remind the user about an open
problem.
Deliveries are persisted with `queued`, `sending`, `retry`, `sent`, or `dead_letter`
status. History also includes `attempt_count`, `max_attempts`, `last_attempt_at`, and
`next_attempt_at`. Raw destinations and server-side credentials are not returned by the
API. Active and resolved events use lifecycle deduplication; an optional repeat interval
can remind the user about an open problem.
The worker claims a delivery with a time-limited lease before contacting a provider. A
temporary failure uses exponential backoff starting at 30 seconds and capped at 30
minutes. After `RMM_NOTIFICATION_MAX_ATTEMPTS`, the delivery becomes `dead_letter`.
Expired worker leases are recovered after restart, so delivery is at-least-once and a
provider may rarely receive a duplicate if the server stopped after the provider accepted
the message but before SQLite recorded success.
## Create Command
+5
View File
@@ -39,6 +39,11 @@ Core tables:
- `notification_settings`
- `notification_deliveries`
Notification delivery uses a persistent SQLite queue. Workers atomically claim ready rows
with a lease, increment the attempt counter, and finish them as `sent`, `retry`, or
`dead_letter`. Expired leases are claimable after a process restart. Terminal history has
an independent retention period; queued, retrying, and in-flight rows are preserved.
## Agent
The agent is a small Go binary and uses tools normally available on OpenWrt for platform
+5
View File
@@ -46,10 +46,15 @@ Telegram notifications use one server-side bot token and a per-user numeric Chat
```dotenv
RMM_TELEGRAM_BOT_TOKEN=replace-with-the-token-from-botfather
RMM_NOTIFICATION_MAX_ATTEMPTS=5
RMM_NOTIFICATION_RETENTION_DAYS=90
```
Do not commit this token. After deployment, each user enables Telegram and enters their
Chat ID in **Profile → Notifications and thresholds**, then uses **Send test**.
Failed deliveries are retried with exponential backoff. Terminal `sent` and `dead_letter`
history is removed after `RMM_NOTIFICATION_RETENTION_DAYS`; pending work is never removed
by retention maintenance.
## 2. Start The Stack
+2
View File
@@ -4,6 +4,8 @@
- [x] Per-user notification settings and alert thresholds.
- [x] E-mail and Telegram alert delivery with lifecycle deduplication and optional repeats.
- [x] Persistent delivery history, masked destinations, test send and profile UI.
- [x] Durable notification queue with restart recovery, leased claims, exponential retry,
dead-letter state and terminal-history retention.
- [x] Persistent alert state with `active`, `acknowledged`, and `resolved`.
- [x] Alert acknowledge API and UI action.
+8
View File
@@ -29,6 +29,12 @@ operators.
- SMTP credentials and the Telegram bot token are read only from the deployment
environment. The browser API returns masked notification destinations and never returns
channel credentials or full delivery destinations.
- Notification provider errors are logged server-side, while users receive a generic
delivery error. Terminal notification history is removed by configurable retention;
pending rows are preserved.
- The pending queue stores the full destination because the worker needs it for delivery.
Treat the SQLite volume and its backups as sensitive data; the browser receives only a
masked destination. Storage encryption and access control remain deployment concerns.
## Operational guidance
@@ -42,6 +48,8 @@ operators.
- Back up SQLite without deleting or recreating the live volume.
- Treat notification delivery errors as sensitive operational data and restrict database
backups to trusted administrators.
- Delivery is at-least-once. A rare duplicate is possible if a provider accepted a message
immediately before the server stopped and its lease later expired.
## Remaining hardening work
+7 -1
View File
@@ -17,9 +17,12 @@ Environment variables:
- `RMM_OPERATOR_USERNAME` - web UI username, default `admin`
- `RMM_PUBLIC_URL` - trusted external base URL used in password recovery links
- `RMM_SMTP_HOST`, `RMM_SMTP_PORT`, `RMM_SMTP_USERNAME`, `RMM_SMTP_PASSWORD`,
`RMM_SMTP_FROM` - optional SMTP delivery for password recovery
`RMM_SMTP_FROM` - optional SMTP delivery for password recovery and alert notifications
- `RMM_SMTP_TLS_MODE` - `starttls` (default), `tls`, or `none` in insecure local mode only
- `RMM_SMTP_SERVER_NAME` - optional TLS server name override
- `RMM_TELEGRAM_BOT_TOKEN` - optional Telegram bot used for per-user alert delivery
- `RMM_NOTIFICATION_MAX_ATTEMPTS` - delivery attempt limit, default `5`
- `RMM_NOTIFICATION_RETENTION_DAYS` - terminal delivery history retention, default `90`
- `RMM_COOKIE_SECURE` - defaults to `true` outside explicit development mode
- `RMM_DEVICE_DOMAIN` - wildcard device domain, for example `routers.example.com`
- `RMM_ALLOW_LEGACY_ENROLLMENT` - opt-in shared enrollment compatibility mode
@@ -55,6 +58,9 @@ Operator API:
- `PATCH /api/auth/profile`
- `POST /api/auth/change-password`
- `POST /api/auth/logout-all`
- `GET|PUT /api/notifications/settings`
- `GET /api/notifications`
- `POST /api/notifications/test`
- `GET|POST /api/users` (administrator only)
- `PATCH /api/users/{id}` (administrator only)
- `POST /api/enrollment-grants`
+28 -18
View File
@@ -93,24 +93,25 @@ func main() {
go maintenanceLoop(st)
handler := httpapi.NewHandler(st, httpapi.Config{
EnrollmentToken: enrollmentToken,
AllowLegacyEnrollment: allowLegacyEnrollment,
AllowLegacyLuCIProxy: envBool("RMM_ALLOW_LEGACY_LUCI_PROXY", false),
OperatorToken: operatorToken,
OperatorUsername: env("RMM_OPERATOR_USERNAME", "admin"),
OperatorPassword: operatorPassword,
CookieSecure: envBool("RMM_COOKIE_SECURE", !insecureDevMode),
TunnelHTTPHost: env("RMM_TUNNEL_HTTP_HOST", "tunnel-ssh"),
TunnelPublicHost: strings.TrimSpace(os.Getenv("RMM_TUNNEL_PUBLIC_HOST")),
TunnelPublicPort: envInt("RMM_TUNNEL_PUBLIC_PORT", 2222, 1, 65535),
DeviceDomain: strings.TrimSpace(os.Getenv("RMM_DEVICE_DOMAIN")),
PublicScheme: env("RMM_PUBLIC_SCHEME", "https"),
PublicURL: publicURL,
PasswordResetSender: passwordResetSender,
AlertEmailSender: alertEmailSender,
TelegramSender: telegramSender,
BackgroundTasks: true,
StaticDir: env("RMM_WEB_DIR", "web"),
EnrollmentToken: enrollmentToken,
AllowLegacyEnrollment: allowLegacyEnrollment,
AllowLegacyLuCIProxy: envBool("RMM_ALLOW_LEGACY_LUCI_PROXY", false),
OperatorToken: operatorToken,
OperatorUsername: env("RMM_OPERATOR_USERNAME", "admin"),
OperatorPassword: operatorPassword,
CookieSecure: envBool("RMM_COOKIE_SECURE", !insecureDevMode),
TunnelHTTPHost: env("RMM_TUNNEL_HTTP_HOST", "tunnel-ssh"),
TunnelPublicHost: strings.TrimSpace(os.Getenv("RMM_TUNNEL_PUBLIC_HOST")),
TunnelPublicPort: envInt("RMM_TUNNEL_PUBLIC_PORT", 2222, 1, 65535),
DeviceDomain: strings.TrimSpace(os.Getenv("RMM_DEVICE_DOMAIN")),
PublicScheme: env("RMM_PUBLIC_SCHEME", "https"),
PublicURL: publicURL,
PasswordResetSender: passwordResetSender,
AlertEmailSender: alertEmailSender,
TelegramSender: telegramSender,
BackgroundTasks: true,
NotificationMaxAttempts: envInt("RMM_NOTIFICATION_MAX_ATTEMPTS", 5, 1, 20),
StaticDir: env("RMM_WEB_DIR", "web"),
})
srv := &http.Server{
@@ -135,6 +136,7 @@ func insecurePlaceholder(value string) bool {
type maintenanceStore interface {
PurgeExpiredSecurityData(ctx context.Context) error
PurgeMetricSamplesBefore(ctx context.Context, cutoff time.Time) (int64, error)
PurgeNotificationDeliveriesBefore(ctx context.Context, cutoff time.Time) (int64, error)
}
func maintenanceLoop(st maintenanceStore) {
@@ -159,6 +161,14 @@ func runMaintenance(ctx context.Context, st maintenanceStore) error {
if deleted > 0 {
log.Printf("maintenance removed %d expired metric samples", deleted)
}
notificationRetentionDays := envInt("RMM_NOTIFICATION_RETENTION_DAYS", 90, 1, 3650)
deleted, err = st.PurgeNotificationDeliveriesBefore(ctx, time.Now().UTC().Add(-time.Duration(notificationRetentionDays)*24*time.Hour))
if err != nil {
return err
}
if deleted > 0 {
log.Printf("maintenance removed %d expired notification deliveries", deleted)
}
return nil
}
+58 -19
View File
@@ -18,6 +18,8 @@ type NotificationSender interface {
SendNotification(ctx context.Context, destination, title, body string) error
}
const notificationDeliveryLease = 5 * time.Minute
type notificationSettingsRequest struct {
EmailEnabled bool `json:"email_enabled"`
TelegramEnabled bool `json:"telegram_enabled"`
@@ -113,7 +115,12 @@ func (a *App) handleTestNotifications(w http.ResponseWriter, r *http.Request) {
writeError(w, http.StatusInternalServerError, "failed to create test notification")
return
}
results = append(results, a.deliverNotification(r.Context(), created))
claimed, found, claimErr := a.store.ClaimNotificationDelivery(r.Context(), created.ID, time.Now().UTC(), notificationDeliveryLease)
if claimErr != nil || !found {
writeError(w, http.StatusInternalServerError, "failed to claim test notification")
return
}
results = append(results, a.deliverNotification(r.Context(), claimed))
}
_, _ = a.store.AddAuditEvent(r.Context(), principal.User.Username, "notifications.test", "", "", mustJSON(map[string]string{"request_id": requestID(r.Context())}))
writeJSON(w, http.StatusOK, map[string]any{"notifications": results})
@@ -170,7 +177,7 @@ func validTelegramChatID(value string) bool {
}
func (a *App) alertNotificationLoop() {
ticker := time.NewTicker(30 * time.Second)
ticker := time.NewTicker(a.notificationWorkerInterval)
defer ticker.Stop()
for {
if err := a.runAlertNotificationCycle(context.Background()); err != nil {
@@ -190,16 +197,13 @@ func (a *App) runAlertNotificationCycle(ctx context.Context) error {
log.Printf("refresh alerts for %s: %v", device.ID, refreshErr)
continue
}
deliveries, queueErr := a.queueDeviceNotifications(ctx, device.ID)
_, queueErr := a.queueDeviceNotifications(ctx, device.ID)
if queueErr != nil {
log.Printf("queue notifications for %s: %v", device.ID, queueErr)
continue
}
if len(deliveries) > 0 {
go a.deliverNotifications(deliveries)
}
}
return nil
return a.processNotificationQueue(ctx)
}
func (a *App) queueDeviceNotifications(ctx context.Context, deviceID string) ([]model.NotificationDelivery, error) {
@@ -258,9 +262,6 @@ func (a *App) queueDeviceNotifications(ctx context.Context, deviceID string) ([]
return nil, lastErr
}
retryAfter := time.Duration(settings.RepeatMinutes) * time.Minute
if lastFound && last.Status == "failed" {
retryAfter = 5 * time.Minute
}
if !lastFound || last.CreatedAt.Add(retryAfter).After(now) {
continue
}
@@ -287,7 +288,10 @@ func notificationSeverityEnabled(settings model.NotificationSettings, severity s
func (a *App) notificationDeliveriesForMessage(user model.User, settings model.NotificationSettings, event, deviceID, alertID, title, body string) []model.NotificationDelivery {
deliveries := make([]model.NotificationDelivery, 0, 2)
base := model.NotificationDelivery{UserID: user.ID, DeviceID: deviceID, AlertID: alertID, Event: event, Title: title, Body: body}
base := model.NotificationDelivery{
UserID: user.ID, DeviceID: deviceID, AlertID: alertID, Event: event,
Title: title, Body: body, MaxAttempts: a.notificationMaxAttempts,
}
if settings.EmailEnabled && a.alertEmailSender != nil && user.Email != "" {
delivery := base
delivery.Channel = "email"
@@ -305,12 +309,20 @@ func (a *App) notificationDeliveriesForMessage(user model.User, settings model.N
return deliveries
}
func (a *App) deliverNotifications(deliveries []model.NotificationDelivery) {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
for _, delivery := range deliveries {
a.deliverNotification(ctx, delivery)
func (a *App) processNotificationQueue(ctx context.Context) error {
deliveries, err := a.store.ClaimNotificationDeliveries(ctx, time.Now().UTC(), notificationDeliveryLease, 10)
if err != nil {
return err
}
for _, delivery := range deliveries {
deliveryCtx, cancel := context.WithTimeout(ctx, 20*time.Second)
a.deliverNotification(deliveryCtx, delivery)
cancel()
}
if len(deliveries) > 0 {
a.events.publish("notifications")
}
return nil
}
func (a *App) deliverNotification(ctx context.Context, delivery model.NotificationDelivery) model.NotificationDelivery {
@@ -326,18 +338,45 @@ func (a *App) deliverNotification(ctx context.Context, delivery model.Notificati
}
if err != nil {
log.Printf("notification delivery %s via %s failed: %v", delivery.ID, delivery.Channel, err)
delivery.Status = "failed"
delivery.Status = "retry"
delivery.Error = "Канал не подтвердил доставку. Проверьте его настройки и повторите тест."
_ = a.store.CompleteNotificationDelivery(context.Background(), delivery.ID, "failed", delivery.Error)
var nextAttemptAt *time.Time
if delivery.AttemptCount >= delivery.MaxAttempts {
delivery.Status = "dead_letter"
} else {
next := time.Now().UTC().Add(notificationRetryDelay(delivery.AttemptCount))
nextAttemptAt = &next
delivery.NextAttemptAt = &next
}
if completeErr := a.store.CompleteNotificationDelivery(context.Background(), delivery.ID, delivery.Status, delivery.Error, nextAttemptAt); completeErr != nil {
log.Printf("notification delivery %s completion failed: %v", delivery.ID, completeErr)
}
return delivery
}
now := time.Now().UTC()
delivery.Status = "sent"
delivery.SentAt = &now
_ = a.store.CompleteNotificationDelivery(context.Background(), delivery.ID, "sent", "")
delivery.NextAttemptAt = nil
if completeErr := a.store.CompleteNotificationDelivery(context.Background(), delivery.ID, "sent", "", nil); completeErr != nil {
log.Printf("notification delivery %s completion failed: %v", delivery.ID, completeErr)
}
return delivery
}
func notificationRetryDelay(attempt int) time.Duration {
if attempt < 1 {
attempt = 1
}
delay := 30 * time.Second
for current := 1; current < attempt && delay < 30*time.Minute; current++ {
delay *= 2
}
if delay > 30*time.Minute {
return 30 * time.Minute
}
return delay
}
func notificationCopy(device model.Device, alert model.Alert, event, publicURL string) (string, string) {
name := strings.TrimSpace(device.Hostname)
if name == "" {
@@ -0,0 +1,18 @@
package httpapi
import (
"testing"
"time"
)
func TestNotificationRetryDelay(t *testing.T) {
want := []time.Duration{30 * time.Second, time.Minute, 2 * time.Minute, 4 * time.Minute, 8 * time.Minute}
for index, expected := range want {
if got := notificationRetryDelay(index + 1); got != expected {
t.Fatalf("attempt %d: got %s want %s", index+1, got, expected)
}
}
if got := notificationRetryDelay(20); got != 30*time.Minute {
t.Fatalf("retry cap: got %s want 30m", got)
}
}
+94 -4
View File
@@ -2,15 +2,20 @@ package httpapi_test
import (
"context"
"database/sql"
"errors"
"net/http"
"net/http/httptest"
"path/filepath"
"strings"
"testing"
"time"
"rmm-openwrt/server/internal/httpapi"
"rmm-openwrt/server/internal/model"
"rmm-openwrt/server/internal/store"
_ "modernc.org/sqlite"
)
type capturedNotification struct {
@@ -23,13 +28,20 @@ type captureNotificationSender struct {
messages chan capturedNotification
}
type failingNotificationSender struct{}
func (failingNotificationSender) SendNotification(context.Context, string, string, string) error {
return errors.New("smtp.internal.example: authentication secret rejected")
}
func (s *captureNotificationSender) SendNotification(_ context.Context, destination, title, body string) error {
s.messages <- capturedNotification{destination: destination, title: title, body: body}
return nil
}
func TestNotificationSettingsTestAndAlertLifecycle(t *testing.T) {
st, err := store.OpenSQLite(context.Background(), filepath.Join(t.TempDir(), "notifications.db"))
dbPath := filepath.Join(t.TempDir(), "notifications.db")
st, err := store.OpenSQLite(context.Background(), dbPath)
if err != nil {
t.Fatal(err)
}
@@ -62,7 +74,7 @@ func TestNotificationSettingsTestAndAlertLifecycle(t *testing.T) {
"email_enabled": true, "telegram_enabled": false, "notify_warning": true,
"notify_critical": true, "notify_resolved": true, "memory_threshold_percent": 80,
"disk_threshold_percent": 90, "packet_loss_percent": 25, "latency_threshold_ms": 250,
"repeat_minutes": 0,
"repeat_minutes": 15,
}, http.StatusOK, &saved)
if !saved.Settings.Configured || !saved.Settings.EmailEnabled || saved.Settings.MemoryThresholdPercent != 80 {
t.Fatalf("settings were not saved: %#v", saved.Settings)
@@ -83,6 +95,29 @@ func TestNotificationSettingsTestAndAlertLifecycle(t *testing.T) {
if activeMessage.title == "" {
t.Fatal("active alert notification was empty")
}
waitNotificationDelivery(t, client, srv.URL, "active", "sent")
db, err := sql.Open("sqlite", dbPath)
if err != nil {
t.Fatal(err)
}
old := time.Now().UTC().Add(-16 * time.Minute).Format(time.RFC3339Nano)
if _, err := db.Exec(`UPDATE notification_deliveries SET created_at = ?, updated_at = ? WHERE event = 'active'`, old, old); err != nil {
db.Close()
t.Fatal(err)
}
if err := db.Close(); err != nil {
t.Fatal(err)
}
requestJSON(t, http.MethodPost, srv.URL+"/api/agent/heartbeat", enrolled.DeviceToken, map[string]any{
"device_id": enrolled.DeviceID,
"inventory": map[string]any{"hostname": "notify-router"},
"metrics": map[string]any{"memory": map[string]any{"used_kb": 900, "total_kb": 1000}},
}, http.StatusOK, nil)
repeatMessage := waitNotification(t, sender.messages)
if repeatMessage.title == "" {
t.Fatal("repeat alert notification was empty")
}
waitNotificationDelivery(t, client, srv.URL, "repeat", "sent")
requestJSON(t, http.MethodPost, srv.URL+"/api/agent/heartbeat", enrolled.DeviceToken, map[string]any{
"device_id": enrolled.DeviceID,
@@ -98,8 +133,8 @@ func TestNotificationSettingsTestAndAlertLifecycle(t *testing.T) {
Notifications []model.NotificationDelivery `json:"notifications"`
}
authRequestJSON(t, client, http.MethodGet, srv.URL+"/api/notifications", nil, http.StatusOK, &history)
if len(history.Notifications) < 3 {
t.Fatalf("expected test, active and resolved deliveries, got %#v", history.Notifications)
if len(history.Notifications) < 4 {
t.Fatalf("expected test, active, repeat and resolved deliveries, got %#v", history.Notifications)
}
for _, delivery := range history.Notifications {
if delivery.Destination != "" {
@@ -108,6 +143,25 @@ func TestNotificationSettingsTestAndAlertLifecycle(t *testing.T) {
}
}
func waitNotificationDelivery(t *testing.T, client *http.Client, serverURL, event, status string) model.NotificationDelivery {
t.Helper()
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
var history struct {
Notifications []model.NotificationDelivery `json:"notifications"`
}
authRequestJSON(t, client, http.MethodGet, serverURL+"/api/notifications", nil, http.StatusOK, &history)
for _, delivery := range history.Notifications {
if delivery.Event == event && delivery.Status == status {
return delivery
}
}
time.Sleep(10 * time.Millisecond)
}
t.Fatalf("notification event %q did not reach status %q", event, status)
return model.NotificationDelivery{}
}
func TestNotificationSettingsValidation(t *testing.T) {
st, err := store.OpenSQLite(context.Background(), filepath.Join(t.TempDir(), "notification-validation.db"))
if err != nil {
@@ -130,6 +184,42 @@ func TestNotificationSettingsValidation(t *testing.T) {
}, http.StatusBadRequest, nil)
}
func TestNotificationFailureIsRetriedWithoutLeakingProviderError(t *testing.T) {
st, err := store.OpenSQLite(context.Background(), filepath.Join(t.TempDir(), "notification-failure.db"))
if err != nil {
t.Fatal(err)
}
defer st.Close()
srv := httptest.NewServer(httpapi.NewHandler(st, httpapi.Config{
OperatorUsername: "admin", OperatorPassword: "notification-password-123",
AlertEmailSender: failingNotificationSender{}, NotificationMaxAttempts: 3,
}))
defer srv.Close()
client := authenticatedClient(t, srv.URL, "admin", "notification-password-123")
authRequestJSON(t, client, http.MethodPatch, srv.URL+"/api/auth/profile", map[string]any{
"display_name": "Owner", "email": "owner@example.test",
}, http.StatusOK, nil)
authRequestJSON(t, client, http.MethodPut, srv.URL+"/api/notifications/settings", map[string]any{
"email_enabled": true, "notify_warning": true, "notify_critical": true, "notify_resolved": true,
"memory_threshold_percent": 85, "disk_threshold_percent": 85, "packet_loss_percent": 20,
"latency_threshold_ms": 200, "repeat_minutes": 0,
}, http.StatusOK, nil)
var result struct {
Notifications []model.NotificationDelivery `json:"notifications"`
}
authRequestJSON(t, client, http.MethodPost, srv.URL+"/api/notifications/test", nil, http.StatusOK, &result)
if len(result.Notifications) != 1 {
t.Fatalf("unexpected test result: %#v", result.Notifications)
}
delivery := result.Notifications[0]
if delivery.Status != "retry" || delivery.AttemptCount != 1 || delivery.MaxAttempts != 3 || delivery.NextAttemptAt == nil {
t.Fatalf("unexpected retry state: %#v", delivery)
}
if strings.Contains(delivery.Error, "smtp.internal") || strings.Contains(delivery.Error, "secret") {
t.Fatalf("provider error leaked to API: %q", delivery.Error)
}
}
func waitNotification(t *testing.T, messages <-chan capturedNotification) capturedNotification {
t.Helper()
select {
+80 -61
View File
@@ -69,9 +69,12 @@ type Store interface {
GetNotificationSettings(ctx context.Context, userID string) (model.NotificationSettings, bool, error)
UpsertNotificationSettings(ctx context.Context, settings model.NotificationSettings) (model.NotificationSettings, error)
CreateNotificationDelivery(ctx context.Context, delivery model.NotificationDelivery, dedupeKey string) (model.NotificationDelivery, bool, error)
CompleteNotificationDelivery(ctx context.Context, id, status, errorMessage string) error
ClaimNotificationDelivery(ctx context.Context, id string, now time.Time, leaseDuration time.Duration) (model.NotificationDelivery, bool, error)
ClaimNotificationDeliveries(ctx context.Context, now time.Time, leaseDuration time.Duration, limit int) ([]model.NotificationDelivery, error)
CompleteNotificationDelivery(ctx context.Context, id, status, errorMessage string, nextAttemptAt *time.Time) error
LatestNotificationDelivery(ctx context.Context, userID, alertID, channel string) (model.NotificationDelivery, bool, error)
ListNotificationDeliveries(ctx context.Context, opts store.NotificationListOptions) ([]model.NotificationDelivery, error)
PurgeNotificationDeliveriesBefore(ctx context.Context, cutoff time.Time) (int64, error)
CreateCommand(ctx context.Context, deviceID, commandType string, args json.RawMessage) (model.Command, bool, error)
ListCommands(ctx context.Context, deviceID string, opts store.CommandListOptions) ([]model.Command, bool, error)
GetCommand(ctx context.Context, deviceID, commandID string) (model.Command, bool, error)
@@ -89,48 +92,52 @@ type Store interface {
}
type Config struct {
EnrollmentToken string
AllowLegacyEnrollment bool
AllowLegacyLuCIProxy bool
OperatorToken string
OperatorUsername string
OperatorPassword string
SessionSecret string
CookieSecure bool
TunnelHTTPHost string
TunnelPublicHost string
TunnelPublicPort int
DeviceDomain string
PublicScheme string
PublicURL string
PasswordResetSender PasswordResetSender
AlertEmailSender NotificationSender
TelegramSender NotificationSender
BackgroundTasks bool
StaticDir string
EnrollmentToken string
AllowLegacyEnrollment bool
AllowLegacyLuCIProxy bool
OperatorToken string
OperatorUsername string
OperatorPassword string
SessionSecret string
CookieSecure bool
TunnelHTTPHost string
TunnelPublicHost string
TunnelPublicPort int
DeviceDomain string
PublicScheme string
PublicURL string
PasswordResetSender PasswordResetSender
AlertEmailSender NotificationSender
TelegramSender NotificationSender
BackgroundTasks bool
NotificationMaxAttempts int
NotificationWorkerInterval time.Duration
StaticDir string
}
type App struct {
store Store
enrollmentToken string
allowLegacyEnrollment bool
operatorToken string
operatorUsername string
dummyPasswordHash string
cookieSecure bool
tunnelHTTPHost string
tunnelPublicHost string
tunnelPublicPort int
deviceDomain string
publicScheme string
publicURL string
passwordResetSender PasswordResetSender
alertEmailSender NotificationSender
telegramSender NotificationSender
loginLimiter *loginRateLimiter
passwordResetLimiter *loginRateLimiter
loginSlots chan struct{}
events *eventHub
store Store
enrollmentToken string
allowLegacyEnrollment bool
operatorToken string
operatorUsername string
dummyPasswordHash string
cookieSecure bool
tunnelHTTPHost string
tunnelPublicHost string
tunnelPublicPort int
deviceDomain string
publicScheme string
publicURL string
passwordResetSender PasswordResetSender
alertEmailSender NotificationSender
telegramSender NotificationSender
notificationMaxAttempts int
notificationWorkerInterval time.Duration
loginLimiter *loginRateLimiter
passwordResetLimiter *loginRateLimiter
loginSlots chan struct{}
events *eventHub
}
type contextKey string
@@ -278,26 +285,34 @@ func NewHandler(s Store, cfg Config) http.Handler {
panic("initialize password verifier: " + err.Error())
}
a := &App{
store: s,
enrollmentToken: cfg.EnrollmentToken,
allowLegacyEnrollment: cfg.AllowLegacyEnrollment || strings.TrimSpace(cfg.EnrollmentToken) != "",
operatorToken: cfg.OperatorToken,
operatorUsername: cfg.OperatorUsername,
dummyPasswordHash: dummyPasswordHash,
cookieSecure: cfg.CookieSecure,
tunnelHTTPHost: strings.TrimSpace(cfg.TunnelHTTPHost),
tunnelPublicHost: strings.TrimSpace(cfg.TunnelPublicHost),
tunnelPublicPort: cfg.TunnelPublicPort,
deviceDomain: strings.Trim(strings.ToLower(strings.TrimSpace(cfg.DeviceDomain)), "."),
publicScheme: strings.ToLower(strings.TrimSpace(cfg.PublicScheme)),
publicURL: strings.TrimRight(strings.TrimSpace(cfg.PublicURL), "/"),
passwordResetSender: cfg.PasswordResetSender,
alertEmailSender: cfg.AlertEmailSender,
telegramSender: cfg.TelegramSender,
loginLimiter: newLoginRateLimiter(5, 5*time.Minute),
passwordResetLimiter: newLoginRateLimiter(3, time.Hour),
loginSlots: make(chan struct{}, 4),
events: newEventHub(),
store: s,
enrollmentToken: cfg.EnrollmentToken,
allowLegacyEnrollment: cfg.AllowLegacyEnrollment || strings.TrimSpace(cfg.EnrollmentToken) != "",
operatorToken: cfg.OperatorToken,
operatorUsername: cfg.OperatorUsername,
dummyPasswordHash: dummyPasswordHash,
cookieSecure: cfg.CookieSecure,
tunnelHTTPHost: strings.TrimSpace(cfg.TunnelHTTPHost),
tunnelPublicHost: strings.TrimSpace(cfg.TunnelPublicHost),
tunnelPublicPort: cfg.TunnelPublicPort,
deviceDomain: strings.Trim(strings.ToLower(strings.TrimSpace(cfg.DeviceDomain)), "."),
publicScheme: strings.ToLower(strings.TrimSpace(cfg.PublicScheme)),
publicURL: strings.TrimRight(strings.TrimSpace(cfg.PublicURL), "/"),
passwordResetSender: cfg.PasswordResetSender,
alertEmailSender: cfg.AlertEmailSender,
telegramSender: cfg.TelegramSender,
notificationMaxAttempts: cfg.NotificationMaxAttempts,
notificationWorkerInterval: cfg.NotificationWorkerInterval,
loginLimiter: newLoginRateLimiter(5, 5*time.Minute),
passwordResetLimiter: newLoginRateLimiter(3, time.Hour),
loginSlots: make(chan struct{}, 4),
events: newEventHub(),
}
if a.notificationMaxAttempts <= 0 {
a.notificationMaxAttempts = 5
}
if a.notificationWorkerInterval <= 0 {
a.notificationWorkerInterval = 30 * time.Second
}
if a.tunnelHTTPHost == "" {
a.tunnelHTTPHost = "tunnel-ssh"
@@ -420,7 +435,11 @@ func (a *App) handleHeartbeat(w http.ResponseWriter, r *http.Request) {
} else if deliveries, queueErr := a.queueDeviceNotifications(r.Context(), req.DeviceID); queueErr != nil {
log.Printf("queue notifications for %s: %v", req.DeviceID, queueErr)
} else if len(deliveries) > 0 {
go a.deliverNotifications(deliveries)
go func() {
if err := a.processNotificationQueue(context.Background()); err != nil {
log.Printf("process notification queue: %v", err)
}
}()
}
a.events.publish("devices")
+4
View File
@@ -126,7 +126,11 @@ type NotificationDelivery struct {
Destination string `json:"-"`
DestinationMasked string `json:"destination"`
Error string `json:"error,omitempty"`
AttemptCount int `json:"attempt_count"`
MaxAttempts int `json:"max_attempts"`
CreatedAt time.Time `json:"created_at"`
LastAttemptAt *time.Time `json:"last_attempt_at,omitempty"`
NextAttemptAt *time.Time `json:"next_attempt_at,omitempty"`
SentAt *time.Time `json:"sent_at,omitempty"`
}
+170 -20
View File
@@ -5,6 +5,7 @@ import (
"database/sql"
"errors"
"strings"
"time"
"rmm-openwrt/server/internal/model"
)
@@ -91,6 +92,10 @@ func (s *Store) CreateNotificationDelivery(ctx context.Context, delivery model.N
return model.NotificationDelivery{}, false, err
}
now := nowText()
maxAttempts := delivery.MaxAttempts
if maxAttempts <= 0 {
maxAttempts = 5
}
var deviceID any
if strings.TrimSpace(delivery.DeviceID) != "" {
deviceID = delivery.DeviceID
@@ -98,10 +103,11 @@ func (s *Store) CreateNotificationDelivery(ctx context.Context, delivery model.N
res, err := s.db.ExecContext(ctx, `
INSERT OR IGNORE INTO notification_deliveries (
id, user_id, device_id, alert_id, dedupe_key, event, channel, status, title, body,
destination, destination_masked, error, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, 'queued', ?, ?, ?, ?, '', ?, ?)
destination, destination_masked, error, attempt_count, max_attempts, created_at,
next_attempt_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, 'queued', ?, ?, ?, ?, '', 0, ?, ?, ?, ?)
`, id, delivery.UserID, deviceID, delivery.AlertID, dedupeKey, delivery.Event, delivery.Channel,
delivery.Title, delivery.Body, delivery.Destination, delivery.DestinationMasked, now, now)
delivery.Title, delivery.Body, delivery.Destination, delivery.DestinationMasked, maxAttempts, now, now, now)
if err != nil {
return model.NotificationDelivery{}, false, err
}
@@ -111,33 +117,167 @@ INSERT OR IGNORE INTO notification_deliveries (
}
delivery.ID = id
delivery.Status = "queued"
delivery.MaxAttempts = maxAttempts
delivery.CreatedAt = parseTime(now)
nextAttemptAt := parseTime(now)
delivery.NextAttemptAt = &nextAttemptAt
return delivery, true, nil
}
func (s *Store) CompleteNotificationDelivery(ctx context.Context, id, status, errorMessage string) error {
func (s *Store) CompleteNotificationDelivery(ctx context.Context, id, status, errorMessage string, nextAttemptAt *time.Time) error {
status = strings.TrimSpace(status)
if status != "sent" && status != "failed" {
if status != "sent" && status != "retry" && status != "dead_letter" {
return errors.New("invalid notification delivery status")
}
if status == "retry" && nextAttemptAt == nil {
return errors.New("retry notification delivery requires next attempt time")
}
now := nowText()
var sentAt any
var nextAttempt any
if status == "sent" {
sentAt = now
} else if nextAttemptAt != nil {
nextAttempt = notificationTimeText(nextAttemptAt.UTC())
}
_, err := s.db.ExecContext(ctx, `
res, err := s.db.ExecContext(ctx, `
UPDATE notification_deliveries
SET status = ?, error = ?, sent_at = ?, updated_at = ?
WHERE id = ? AND status = 'queued'
`, status, strings.TrimSpace(errorMessage), sentAt, now, id)
return err
SET status = ?, error = ?, sent_at = ?, next_attempt_at = COALESCE(?, ''),
lease_expires_at = NULL, updated_at = ?
WHERE id = ? AND status = 'sending'
`, status, strings.TrimSpace(errorMessage), sentAt, nextAttempt, now, id)
if err != nil {
return err
}
updated, err := res.RowsAffected()
if err != nil {
return err
}
if updated == 0 {
return errors.New("notification delivery is not claimed")
}
return nil
}
func (s *Store) LatestNotificationDelivery(ctx context.Context, userID, alertID, channel string) (model.NotificationDelivery, bool, error) {
row := s.db.QueryRowContext(ctx, `
SELECT id, user_id, COALESCE(device_id, ''), alert_id, event, channel, status, title, body,
destination, destination_masked, error, created_at, sent_at
func (s *Store) ClaimNotificationDelivery(ctx context.Context, id string, now time.Time, leaseDuration time.Duration) (model.NotificationDelivery, bool, error) {
deliveries, err := s.claimNotificationDeliveries(ctx, strings.TrimSpace(id), now, leaseDuration, 1)
if err != nil || len(deliveries) == 0 {
return model.NotificationDelivery{}, false, err
}
return deliveries[0], true, nil
}
func (s *Store) ClaimNotificationDeliveries(ctx context.Context, now time.Time, leaseDuration time.Duration, limit int) ([]model.NotificationDelivery, error) {
return s.claimNotificationDeliveries(ctx, "", now, leaseDuration, limit)
}
func (s *Store) claimNotificationDeliveries(ctx context.Context, onlyID string, now time.Time, leaseDuration time.Duration, limit int) ([]model.NotificationDelivery, error) {
if limit <= 0 || limit > 100 {
limit = 10
}
if leaseDuration <= 0 {
leaseDuration = 3 * time.Minute
}
now = now.UTC()
nowValue := notificationTimeText(now)
leaseValue := notificationTimeText(now.Add(leaseDuration))
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return nil, err
}
defer tx.Rollback()
if _, err := tx.ExecContext(ctx, `
UPDATE notification_deliveries
SET status = 'dead_letter', error = CASE WHEN error = '' THEN 'Попытки доставки исчерпаны. Проверьте настройки канала.' ELSE error END,
next_attempt_at = '', lease_expires_at = NULL, updated_at = ?
WHERE status IN ('retry', 'sending') AND attempt_count >= max_attempts
AND (status = 'retry' OR lease_expires_at IS NULL OR lease_expires_at <= ?)
`, nowValue, nowValue); err != nil {
return nil, err
}
query := `
SELECT id
FROM notification_deliveries
WHERE attempt_count < max_attempts
AND ((status IN ('queued', 'retry') AND (next_attempt_at = '' OR next_attempt_at <= ?))
OR (status = 'sending' AND lease_expires_at IS NOT NULL AND lease_expires_at <= ?))`
args := []any{nowValue, nowValue}
if onlyID != "" {
query += ` AND id = ?`
args = append(args, onlyID)
}
query += ` ORDER BY created_at ASC LIMIT ?`
args = append(args, limit)
rows, err := tx.QueryContext(ctx, query, args...)
if err != nil {
return nil, err
}
ids := make([]string, 0, limit)
for rows.Next() {
var id string
if err := rows.Scan(&id); err != nil {
rows.Close()
return nil, err
}
ids = append(ids, id)
}
if err := rows.Close(); err != nil {
return nil, err
}
if err := rows.Err(); err != nil {
return nil, err
}
claimed := make([]model.NotificationDelivery, 0, len(ids))
for _, id := range ids {
res, err := tx.ExecContext(ctx, `
UPDATE notification_deliveries
SET status = 'sending', attempt_count = attempt_count + 1, last_attempt_at = ?,
lease_expires_at = ?, next_attempt_at = '', updated_at = ?
WHERE id = ? AND attempt_count < max_attempts
AND ((status IN ('queued', 'retry') AND (next_attempt_at = '' OR next_attempt_at <= ?))
OR (status = 'sending' AND lease_expires_at IS NOT NULL AND lease_expires_at <= ?))
`, nowValue, leaseValue, nowValue, id, nowValue, nowValue)
if err != nil {
return nil, err
}
updated, err := res.RowsAffected()
if err != nil {
return nil, err
}
if updated == 0 {
continue
}
delivery, err := scanNotificationDelivery(tx.QueryRowContext(ctx, notificationDeliverySelect+` WHERE id = ?`, id))
if err != nil {
return nil, err
}
claimed = append(claimed, delivery)
}
if err := tx.Commit(); err != nil {
return nil, err
}
return claimed, nil
}
func (s *Store) PurgeNotificationDeliveriesBefore(ctx context.Context, cutoff time.Time) (int64, error) {
res, err := s.db.ExecContext(ctx, `
DELETE FROM notification_deliveries
WHERE status IN ('sent', 'dead_letter', 'failed') AND updated_at < ?
`, notificationTimeText(cutoff.UTC()))
if err != nil {
return 0, err
}
return res.RowsAffected()
}
const notificationDeliverySelect = `
SELECT id, user_id, COALESCE(device_id, ''), alert_id, event, channel, status, title, body,
destination, destination_masked, error, attempt_count, max_attempts, created_at,
last_attempt_at, next_attempt_at, sent_at
FROM notification_deliveries`
func (s *Store) LatestNotificationDelivery(ctx context.Context, userID, alertID, channel string) (model.NotificationDelivery, bool, error) {
row := s.db.QueryRowContext(ctx, notificationDeliverySelect+`
WHERE user_id = ? AND alert_id = ? AND channel = ? AND event IN ('active', 'repeat')
ORDER BY created_at DESC LIMIT 1
`, userID, alertID, channel)
@@ -153,10 +293,7 @@ func (s *Store) ListNotificationDeliveries(ctx context.Context, opts Notificatio
if limit <= 0 || limit > 200 {
limit = 50
}
rows, err := s.db.QueryContext(ctx, `
SELECT id, user_id, COALESCE(device_id, ''), alert_id, event, channel, status, title, body,
destination, destination_masked, error, created_at, sent_at
FROM notification_deliveries
rows, err := s.db.QueryContext(ctx, notificationDeliverySelect+`
WHERE user_id = ?
ORDER BY created_at DESC LIMIT ?
`, opts.UserID, limit)
@@ -182,16 +319,25 @@ type notificationScanner interface {
func scanNotificationDelivery(scanner notificationScanner) (model.NotificationDelivery, error) {
var delivery model.NotificationDelivery
var createdAt string
var sentAt sql.NullString
var lastAttemptAt, nextAttemptAt, sentAt sql.NullString
err := scanner.Scan(
&delivery.ID, &delivery.UserID, &delivery.DeviceID, &delivery.AlertID, &delivery.Event,
&delivery.Channel, &delivery.Status, &delivery.Title, &delivery.Body, &delivery.Destination,
&delivery.DestinationMasked, &delivery.Error, &createdAt, &sentAt,
&delivery.DestinationMasked, &delivery.Error, &delivery.AttemptCount, &delivery.MaxAttempts,
&createdAt, &lastAttemptAt, &nextAttemptAt, &sentAt,
)
if err != nil {
return model.NotificationDelivery{}, err
}
delivery.CreatedAt = parseTime(createdAt)
if lastAttemptAt.Valid && lastAttemptAt.String != "" {
value := parseTime(lastAttemptAt.String)
delivery.LastAttemptAt = &value
}
if nextAttemptAt.Valid && nextAttemptAt.String != "" {
value := parseTime(nextAttemptAt.String)
delivery.NextAttemptAt = &value
}
if sentAt.Valid {
value := parseTime(sentAt.String)
delivery.SentAt = &value
@@ -205,3 +351,7 @@ func boolInt(value bool) int {
}
return 0
}
func notificationTimeText(value time.Time) string {
return value.UTC().Format(time.RFC3339Nano)
}
+167
View File
@@ -0,0 +1,167 @@
package store
import (
"context"
"database/sql"
"path/filepath"
"testing"
"time"
"rmm-openwrt/server/internal/model"
)
func TestMigrateLegacyNotificationDeliveries(t *testing.T) {
ctx := context.Background()
path := filepath.Join(t.TempDir(), "legacy-notifications.db")
db, err := sql.Open("sqlite", path)
if err != nil {
t.Fatal(err)
}
_, err = db.Exec(`
CREATE TABLE notification_deliveries (
id TEXT PRIMARY KEY, user_id TEXT NOT NULL, device_id TEXT, alert_id TEXT NOT NULL DEFAULT '',
dedupe_key TEXT NOT NULL UNIQUE, event TEXT NOT NULL, channel TEXT NOT NULL, status TEXT NOT NULL,
title TEXT NOT NULL, body TEXT NOT NULL, destination TEXT NOT NULL, destination_masked TEXT NOT NULL,
error TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL, sent_at TEXT, updated_at TEXT NOT NULL
);
INSERT INTO notification_deliveries VALUES
('ntf_sent', 'usr_legacy', NULL, '', 'sent', 'test', 'email', 'sent', 'sent', 'body', 'a@example.test', 'a***@example.test', '', '2026-01-01T00:00:00Z', '2026-01-01T00:00:01Z', '2026-01-01T00:00:01Z'),
('ntf_failed', 'usr_legacy', NULL, '', 'failed', 'test', 'email', 'failed', 'failed', 'body', 'a@example.test', 'a***@example.test', 'old error', '2026-01-01T00:00:00Z', NULL, '2026-01-01T00:00:01Z'),
('ntf_queued', 'usr_legacy', NULL, '', 'queued', 'test', 'email', 'queued', 'queued', 'body', 'a@example.test', 'a***@example.test', '', '2026-01-01T00:00:00Z', NULL, '2026-01-01T00:00:01Z');
`)
if err != nil {
db.Close()
t.Fatal(err)
}
if err := db.Close(); err != nil {
t.Fatal(err)
}
st, err := OpenSQLite(ctx, path)
if err != nil {
t.Fatal(err)
}
defer st.Close()
history, err := st.ListNotificationDeliveries(ctx, NotificationListOptions{UserID: "usr_legacy"})
if err != nil || len(history) != 3 {
t.Fatalf("migrated history: %#v err=%v", history, err)
}
byID := make(map[string]model.NotificationDelivery, len(history))
for _, delivery := range history {
byID[delivery.ID] = delivery
}
if byID["ntf_sent"].NextAttemptAt != nil {
t.Fatalf("sent delivery became retryable: %#v", byID["ntf_sent"])
}
if byID["ntf_failed"].Status != "dead_letter" || byID["ntf_failed"].AttemptCount != 1 {
t.Fatalf("failed delivery was not migrated safely: %#v", byID["ntf_failed"])
}
if byID["ntf_queued"].NextAttemptAt == nil {
t.Fatalf("queued delivery did not become claimable: %#v", byID["ntf_queued"])
}
}
func TestNotificationQueueSurvivesRestartAndRetries(t *testing.T) {
ctx := context.Background()
path := filepath.Join(t.TempDir(), "notification-queue.db")
st, err := OpenSQLite(ctx, path)
if err != nil {
t.Fatal(err)
}
user, err := st.EnsureBootstrapUser(ctx, "admin", "test-password-hash")
if err != nil {
t.Fatal(err)
}
created, inserted, err := st.CreateNotificationDelivery(ctx, model.NotificationDelivery{
UserID: user.ID, Event: "test", Channel: "email", Title: "test", Body: "body",
Destination: "owner@example.test", DestinationMasked: "o***@example.test", MaxAttempts: 3,
}, "restart-retry")
if err != nil || !inserted {
t.Fatalf("create delivery: inserted=%v err=%v", inserted, err)
}
if err := st.Close(); err != nil {
t.Fatal(err)
}
st, err = OpenSQLite(ctx, path)
if err != nil {
t.Fatal(err)
}
defer st.Close()
now := time.Now().UTC().Add(time.Second)
claimed, found, err := st.ClaimNotificationDelivery(ctx, created.ID, now, time.Minute)
if err != nil || !found {
t.Fatalf("claim after restart: found=%v err=%v", found, err)
}
if claimed.Status != "sending" || claimed.AttemptCount != 1 || claimed.MaxAttempts != 3 {
t.Fatalf("unexpected first claim: %#v", claimed)
}
next := now.Add(30 * time.Second)
if err := st.CompleteNotificationDelivery(ctx, claimed.ID, "retry", "temporary failure", &next); err != nil {
t.Fatal(err)
}
if deliveries, err := st.ClaimNotificationDeliveries(ctx, next.Add(-time.Second), time.Minute, 10); err != nil || len(deliveries) != 0 {
t.Fatalf("delivery became ready too early: deliveries=%#v err=%v", deliveries, err)
}
claimed, found, err = st.ClaimNotificationDelivery(ctx, created.ID, next, time.Minute)
if err != nil || !found || claimed.AttemptCount != 2 {
t.Fatalf("claim retry: delivery=%#v found=%v err=%v", claimed, found, err)
}
if err := st.CompleteNotificationDelivery(ctx, claimed.ID, "sent", "", nil); err != nil {
t.Fatal(err)
}
history, err := st.ListNotificationDeliveries(ctx, NotificationListOptions{UserID: user.ID})
if err != nil || len(history) != 1 {
t.Fatalf("list history: %#v err=%v", history, err)
}
if history[0].Status != "sent" || history[0].AttemptCount != 2 || history[0].SentAt == nil {
t.Fatalf("unexpected completed delivery: %#v", history[0])
}
}
func TestNotificationQueueDeadLetterLeaseRecoveryAndRetention(t *testing.T) {
ctx := context.Background()
st, err := OpenSQLite(ctx, filepath.Join(t.TempDir(), "notification-dead-letter.db"))
if err != nil {
t.Fatal(err)
}
defer st.Close()
user, err := st.EnsureBootstrapUser(ctx, "admin", "test-password-hash")
if err != nil {
t.Fatal(err)
}
created, inserted, err := st.CreateNotificationDelivery(ctx, model.NotificationDelivery{
UserID: user.ID, Event: "active", Channel: "telegram", Title: "test", Body: "body",
Destination: "12345", DestinationMasked: "***2345", MaxAttempts: 1,
}, "lease-dead-letter")
if err != nil || !inserted {
t.Fatalf("create delivery: inserted=%v err=%v", inserted, err)
}
now := time.Now().UTC().Add(time.Second)
claimed, found, err := st.ClaimNotificationDelivery(ctx, created.ID, now, time.Minute)
if err != nil || !found || claimed.AttemptCount != 1 {
t.Fatalf("initial claim: delivery=%#v found=%v err=%v", claimed, found, err)
}
if deliveries, err := st.ClaimNotificationDeliveries(ctx, now.Add(2*time.Minute), time.Minute, 10); err != nil || len(deliveries) != 0 {
t.Fatalf("exhausted lease was reclaimed: deliveries=%#v err=%v", deliveries, err)
}
history, err := st.ListNotificationDeliveries(ctx, NotificationListOptions{UserID: user.ID})
if err != nil || len(history) != 1 || history[0].Status != "dead_letter" {
t.Fatalf("delivery was not moved to dead letter: %#v err=%v", history, err)
}
queued, inserted, err := st.CreateNotificationDelivery(ctx, model.NotificationDelivery{
UserID: user.ID, Event: "test", Channel: "email", Title: "queued", Body: "body",
Destination: "owner@example.test", DestinationMasked: "o***@example.test",
}, "retention-keeps-queued")
if err != nil || !inserted || queued.Status != "queued" {
t.Fatalf("create queued delivery: %#v inserted=%v err=%v", queued, inserted, err)
}
deleted, err := st.PurgeNotificationDeliveriesBefore(ctx, time.Now().UTC().Add(24*time.Hour))
if err != nil || deleted != 1 {
t.Fatalf("purge terminal history: deleted=%d err=%v", deleted, err)
}
history, err = st.ListNotificationDeliveries(ctx, NotificationListOptions{UserID: user.ID})
if err != nil || len(history) != 1 || history[0].ID != queued.ID {
t.Fatalf("purge removed non-terminal delivery: %#v err=%v", history, err)
}
}
+24
View File
@@ -198,7 +198,12 @@ CREATE TABLE IF NOT EXISTS notification_deliveries (
destination TEXT NOT NULL,
destination_masked TEXT NOT NULL,
error TEXT NOT NULL DEFAULT '',
attempt_count INTEGER NOT NULL DEFAULT 0,
max_attempts INTEGER NOT NULL DEFAULT 5,
created_at TEXT NOT NULL,
last_attempt_at TEXT,
next_attempt_at TEXT NOT NULL DEFAULT '',
lease_expires_at TEXT,
sent_at TEXT,
updated_at TEXT NOT NULL,
FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE,
@@ -358,6 +363,11 @@ CREATE TABLE IF NOT EXISTS device_access_sessions (
`ALTER TABLE remote_sessions ADD COLUMN updated_at TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE users ADD COLUMN display_name TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE users ADD COLUMN email TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE notification_deliveries ADD COLUMN attempt_count INTEGER NOT NULL DEFAULT 0`,
`ALTER TABLE notification_deliveries ADD COLUMN max_attempts INTEGER NOT NULL DEFAULT 5`,
`ALTER TABLE notification_deliveries ADD COLUMN last_attempt_at TEXT`,
`ALTER TABLE notification_deliveries ADD COLUMN next_attempt_at TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE notification_deliveries ADD COLUMN lease_expires_at TEXT`,
} {
if _, err := s.db.ExecContext(ctx, stmt); err != nil && !isDuplicateColumnError(err) {
return err
@@ -376,11 +386,25 @@ CREATE TABLE IF NOT EXISTS device_access_sessions (
`CREATE INDEX IF NOT EXISTS idx_remote_sessions_status_expires_at ON remote_sessions(status, expires_at)`,
`CREATE INDEX IF NOT EXISTS idx_notification_deliveries_user_created ON notification_deliveries(user_id, created_at DESC)`,
`CREATE INDEX IF NOT EXISTS idx_notification_deliveries_status_created ON notification_deliveries(status, created_at)`,
`CREATE INDEX IF NOT EXISTS idx_notification_deliveries_ready ON notification_deliveries(status, next_attempt_at, lease_expires_at)`,
} {
if _, err := s.db.ExecContext(ctx, stmt); err != nil {
return err
}
}
if _, err := s.db.ExecContext(ctx, `
UPDATE notification_deliveries
SET status = 'dead_letter',
attempt_count = CASE WHEN attempt_count = 0 THEN 1 ELSE attempt_count END,
next_attempt_at = ''
WHERE status = 'failed';
UPDATE notification_deliveries
SET next_attempt_at = created_at
WHERE status IN ('queued', 'retry') AND next_attempt_at = ''
`); err != nil {
return err
}
return s.migrateDeviceTokens(ctx)
}
+13 -2
View File
@@ -1638,9 +1638,13 @@ function renderNotificationHistory() {
for (const delivery of state.notifications) {
const row = document.createElement("article");
row.className = `notification-history-row is-${delivery.status || "queued"}`;
const attempts = delivery.attempt_count > 0 ? ` · попытка ${delivery.attempt_count}/${delivery.max_attempts || "?"}` : "";
const retryAt = delivery.status === "retry" && delivery.next_attempt_at
? ` · повтор ${formatDate(delivery.next_attempt_at)}`
: "";
row.innerHTML = `
<span class="notification-channel">${delivery.channel === "telegram" ? "Telegram" : "E-mail"}</span>
<div><strong>${escapeHtml(delivery.title || "Уведомление")}</strong><small>${escapeHtml(delivery.destination || "")}${delivery.error ? ` · ${escapeHtml(delivery.error)}` : ""}</small></div>
<div><strong>${escapeHtml(delivery.title || "Уведомление")}</strong><small>${escapeHtml(delivery.destination || "")}${delivery.error ? ` · ${escapeHtml(delivery.error)}` : ""}${escapeHtml(attempts)}${escapeHtml(retryAt)}</small></div>
<span class="notification-delivery-status">${escapeHtml(notificationStatusLabel(delivery.status))}</span>
<time>${escapeHtml(formatDate(delivery.sent_at || delivery.created_at))}</time>
`;
@@ -1649,7 +1653,14 @@ function renderNotificationHistory() {
}
function notificationStatusLabel(status) {
return { queued: "В очереди", sent: "Отправлено", failed: "Ошибка" }[status] || status || "Неизвестно";
return {
queued: "В очереди",
sending: "Отправляется",
retry: "Повтор ожидается",
sent: "Отправлено",
dead_letter: "Не доставлено",
failed: "Ошибка",
}[status] || status || "Неизвестно";
}
function notificationErrorMessage(error) {
+3 -1
View File
@@ -351,7 +351,9 @@
.notification-channel { color: var(--muted); font-size: 11px; font-weight: 750; text-transform: uppercase; }
.notification-delivery-status { color: var(--warn); font-size: 12px; font-weight: 750; }
.notification-history-row.is-sent .notification-delivery-status { color: var(--good); }
.notification-history-row.is-failed .notification-delivery-status { color: var(--bad); }
.notification-history-row.is-failed .notification-delivery-status,
.notification-history-row.is-dead_letter .notification-delivery-status { color: var(--bad); }
.notification-history-row.is-sending .notification-delivery-status { color: var(--accent); }
@media (max-width: 900px) {
.landing-header nav a:not(.landing-login-link) { display: none; }