mirror of
https://github.com/Dvorinka/Containr.git
synced 2026-10-28 07:41:27 +00:00
feat(notifications): ntfy/Gotify push channels with live delivery (#69)
Per-user notification_channels (ntfy + Gotify): every
insertUserNotification now fans out best-effort to enabled channels —
deployment/backup/security/agent alerts reach the user's phone via the
ntfy app with zero extra infrastructure.
- POST/GET/DELETE /notifications/channels + POST .../:id/test which
pushes a real probe so misconfiguration surfaces before alerts depend
on it (502 DELIVERY_FAILED on endpoint errors).
- ntfy: POST to endpoint (topic in path), Title/Tags headers, optional
Bearer token. Gotify: POST {endpoint}/message?token=...
- Delivery is best-effort by design — a dead relay never breaks the
producing flow; one shared 5s-timeout client.
- CLI `notifications channels list|add|test|remove`, MCP tools,
OpenAPI paths + NotificationChannel schema.
- Web settings "Push notifications" section and a Push channels card on
the mobile notifications tab (add/test/remove).
Verified live: channel CRUD, test-ping hit a recording endpoint with
correct headers/body, and a real deployment fanned out "Deployment
succeeded" through the channel end-to-end.
Generated with [Devin](https://devin.ai)
Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
1 parent
60145f7b21
commit
9243d020cc
19 files changed
+1261
-6
No files matched your search
@@ -0,0 +1,225 @@
|
||||
package api
|
||||
|
||||
// Push notification channels — self-hosted ntfy and Gotify endpoints users
|
||||
// attach to their account. insertUserNotification fans out best-effort.
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"containr/internal/database"
|
||||
"containr/internal/database/sqlcdb"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
// jarvis: ceiling — one shared client; no retry/backoff, delivery is
|
||||
// best-effort and failures are silent by design (notifications must never
|
||||
// break the producing flow).
|
||||
var pushHTTPClient = &http.Client{Timeout: 5 * time.Second}
|
||||
|
||||
type notificationChannelDTO struct {
|
||||
ID string `json:"id"`
|
||||
Kind string `json:"kind"`
|
||||
Endpoint string `json:"endpoint"`
|
||||
Enabled bool `json:"enabled"`
|
||||
CreatedAt string `json:"created_at"`
|
||||
}
|
||||
|
||||
func mapChannel(row sqlcdb.NotificationChannel) notificationChannelDTO {
|
||||
return notificationChannelDTO{
|
||||
ID: row.ID.String(),
|
||||
Kind: row.Kind,
|
||||
Endpoint: row.Endpoint,
|
||||
Enabled: row.Enabled,
|
||||
CreatedAt: row.CreatedAt.Time.Format(time.RFC3339),
|
||||
}
|
||||
}
|
||||
|
||||
func handleListNotificationChannels(c *gin.Context) {
|
||||
userID, ok := requireAuthenticatedUserID(c)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
db := c.MustGet("db").(*database.DB)
|
||||
uid, _ := uuid.Parse(userID)
|
||||
rows, err := sqlcdb.New(db.DB).ListAllNotificationChannelsByUser(c.Request.Context(), uid)
|
||||
if err != nil {
|
||||
respondError(c, http.StatusInternalServerError, "INTERNAL_ERROR", "Failed to list channels")
|
||||
return
|
||||
}
|
||||
out := make([]notificationChannelDTO, 0, len(rows))
|
||||
for _, r := range rows {
|
||||
out = append(out, mapChannel(r))
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"channels": out})
|
||||
}
|
||||
|
||||
func handleCreateNotificationChannel(c *gin.Context) {
|
||||
userID, ok := requireAuthenticatedUserID(c)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
var req struct {
|
||||
Kind string `json:"kind" binding:"required,oneof=ntfy gotify"`
|
||||
Endpoint string `json:"endpoint" binding:"required"`
|
||||
Token string `json:"token"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
respondError(c, http.StatusBadRequest, "VALIDATION", "kind (ntfy|gotify) and endpoint are required")
|
||||
return
|
||||
}
|
||||
endpoint := strings.TrimRight(strings.TrimSpace(req.Endpoint), "/")
|
||||
u, err := url.Parse(endpoint)
|
||||
if err != nil || u.Scheme == "" || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https") {
|
||||
respondError(c, http.StatusBadRequest, "VALIDATION", "endpoint must be a valid http(s) URL")
|
||||
return
|
||||
}
|
||||
db := c.MustGet("db").(*database.DB)
|
||||
uid, _ := uuid.Parse(userID)
|
||||
row, err := sqlcdb.New(db.DB).InsertNotificationChannel(c.Request.Context(), sqlcdb.InsertNotificationChannelParams{
|
||||
UserID: uid,
|
||||
Kind: req.Kind,
|
||||
Endpoint: endpoint,
|
||||
Token: sql.NullString{String: req.Token, Valid: req.Token != ""},
|
||||
})
|
||||
if err != nil {
|
||||
respondError(c, http.StatusInternalServerError, "INTERNAL_ERROR", "Failed to create channel")
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusCreated, gin.H{"channel": mapChannel(row)})
|
||||
}
|
||||
|
||||
func handleDeleteNotificationChannel(c *gin.Context) {
|
||||
userID, ok := requireAuthenticatedUserID(c)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
id, err := uuid.Parse(c.Param("id"))
|
||||
if err != nil {
|
||||
respondError(c, http.StatusBadRequest, "VALIDATION", "Invalid channel id")
|
||||
return
|
||||
}
|
||||
db := c.MustGet("db").(*database.DB)
|
||||
uid, _ := uuid.Parse(userID)
|
||||
n, err := sqlcdb.New(db.DB).DeleteNotificationChannel(c.Request.Context(), sqlcdb.DeleteNotificationChannelParams{
|
||||
ID: id, UserID: uid,
|
||||
})
|
||||
if err != nil {
|
||||
respondError(c, http.StatusInternalServerError, "INTERNAL_ERROR", "Failed to delete channel")
|
||||
return
|
||||
}
|
||||
if n == 0 {
|
||||
respondError(c, http.StatusNotFound, "NOT_FOUND", "Channel not found")
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"message": "Channel deleted"})
|
||||
}
|
||||
|
||||
// handleTestNotificationChannel pushes a probe notification through the
|
||||
// channel — surfaces misconfiguration before real alerts depend on it.
|
||||
func handleTestNotificationChannel(c *gin.Context) {
|
||||
userID, ok := requireAuthenticatedUserID(c)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
id, err := uuid.Parse(c.Param("id"))
|
||||
if err != nil {
|
||||
respondError(c, http.StatusBadRequest, "VALIDATION", "Invalid channel id")
|
||||
return
|
||||
}
|
||||
db := c.MustGet("db").(*database.DB)
|
||||
uid, _ := uuid.Parse(userID)
|
||||
rows, err := sqlcdb.New(db.DB).ListAllNotificationChannelsByUser(c.Request.Context(), uid)
|
||||
if err != nil {
|
||||
respondError(c, http.StatusInternalServerError, "INTERNAL_ERROR", "Failed to load channel")
|
||||
return
|
||||
}
|
||||
var target *sqlcdb.NotificationChannel
|
||||
for i := range rows {
|
||||
if rows[i].ID == id {
|
||||
target = &rows[i]
|
||||
break
|
||||
}
|
||||
}
|
||||
if target == nil {
|
||||
respondError(c, http.StatusNotFound, "NOT_FOUND", "Channel not found")
|
||||
return
|
||||
}
|
||||
if err := deliverPush(c.Request.Context(), *target, "test", "Containr test notification", "Your push channel is configured correctly."); err != nil {
|
||||
respondError(c, http.StatusBadGateway, "DELIVERY_FAILED", fmt.Sprintf("Push failed: %v", err))
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"message": "Test notification sent"})
|
||||
}
|
||||
|
||||
// deliverPush sends one notification through a channel. ntfy expects the
|
||||
// body at POST {endpoint} with Title/Priority headers; Gotify takes JSON at
|
||||
// POST {endpoint}/message?token=...
|
||||
func deliverPush(ctx context.Context, ch sqlcdb.NotificationChannel, kind, title, body string) error {
|
||||
var req *http.Request
|
||||
var err error
|
||||
switch ch.Kind {
|
||||
case "ntfy":
|
||||
req, err = http.NewRequestWithContext(ctx, http.MethodPost, ch.Endpoint, strings.NewReader(body))
|
||||
if err == nil {
|
||||
req.Header.Set("Title", title)
|
||||
req.Header.Set("Tags", "containr,"+kind)
|
||||
if ch.Token.Valid && ch.Token.String != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+ch.Token.String)
|
||||
}
|
||||
}
|
||||
case "gotify":
|
||||
payload, _ := json.Marshal(map[string]interface{}{
|
||||
"title": title,
|
||||
"message": body,
|
||||
"priority": 5,
|
||||
})
|
||||
endpoint := ch.Endpoint + "/message"
|
||||
if ch.Token.Valid && ch.Token.String != "" {
|
||||
endpoint += "?token=" + url.QueryEscape(ch.Token.String)
|
||||
}
|
||||
req, err = http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(payload))
|
||||
if err == nil {
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
}
|
||||
default:
|
||||
return fmt.Errorf("unsupported channel kind %q", ch.Kind)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
res, err := pushHTTPClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer res.Body.Close()
|
||||
if res.StatusCode >= 400 {
|
||||
return fmt.Errorf("push endpoint returned %d", res.StatusCode)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// fanoutUserPush delivers a notification to every enabled channel the user
|
||||
// configured. Errors are swallowed — the in-app row already landed.
|
||||
func fanoutUserPush(db *database.DB, userID, kind, title, body string) {
|
||||
uid, err := uuid.Parse(userID)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
rows, err := sqlcdb.New(db.DB).ListNotificationChannelsByUser(context.Background(), uid)
|
||||
if err != nil || len(rows) == 0 {
|
||||
return
|
||||
}
|
||||
for _, ch := range rows {
|
||||
_ = deliverPush(context.Background(), ch, kind, title, body)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"containr/internal/database/sqlcdb"
|
||||
)
|
||||
|
||||
func TestDeliverPushNtfy(t *testing.T) {
|
||||
var gotAuth, gotTitle, gotTags, gotBody string
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
gotAuth = r.Header.Get("Authorization")
|
||||
gotTitle = r.Header.Get("Title")
|
||||
gotTags = r.Header.Get("Tags")
|
||||
b, _ := io.ReadAll(r.Body)
|
||||
gotBody = string(b)
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
ch := sqlcdb.NotificationChannel{
|
||||
Kind: "ntfy",
|
||||
Endpoint: srv.URL + "/alerts",
|
||||
Token: sql.NullString{String: "tk_abc", Valid: true},
|
||||
}
|
||||
if err := deliverPush(context.Background(), ch, "deployment", "Deploy done", "svc is live"); err != nil {
|
||||
t.Fatalf("deliverPush: %v", err)
|
||||
}
|
||||
if gotAuth != "Bearer tk_abc" {
|
||||
t.Fatalf("auth header = %q", gotAuth)
|
||||
}
|
||||
if gotTitle != "Deploy done" || gotTags != "containr,deployment" || gotBody != "svc is live" {
|
||||
t.Fatalf("title=%q tags=%q body=%q", gotTitle, gotTags, gotBody)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeliverPushGotify(t *testing.T) {
|
||||
var gotPath, gotToken string
|
||||
var payload map[string]interface{}
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
gotPath = r.URL.Path
|
||||
gotToken = r.URL.Query().Get("token")
|
||||
_ = json.NewDecoder(r.Body).Decode(&payload)
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
ch := sqlcdb.NotificationChannel{
|
||||
Kind: "gotify",
|
||||
Endpoint: srv.URL,
|
||||
Token: sql.NullString{String: "apptoken", Valid: true},
|
||||
}
|
||||
if err := deliverPush(context.Background(), ch, "backup", "Backup done", "db backup ok"); err != nil {
|
||||
t.Fatalf("deliverPush: %v", err)
|
||||
}
|
||||
if gotPath != "/message" || gotToken != "apptoken" {
|
||||
t.Fatalf("path=%q token=%q", gotPath, gotToken)
|
||||
}
|
||||
if payload["title"] != "Backup done" || payload["message"] != "db backup ok" {
|
||||
t.Fatalf("payload = %v", payload)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeliverPushFailure(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
ch := sqlcdb.NotificationChannel{Kind: "ntfy", Endpoint: srv.URL}
|
||||
if err := deliverPush(context.Background(), ch, "k", "t", "b"); err == nil {
|
||||
t.Fatal("expected error on 500")
|
||||
}
|
||||
}
|
||||
@@ -41,6 +41,7 @@ func insertUserNotification(db *database.DB, userID, kind, title, body, resource
|
||||
ResourceID: sql.NullString{String: resourceID, Valid: resourceID != ""},
|
||||
CreatedAt: sql.NullTime{Time: time.Now(), Valid: true},
|
||||
})
|
||||
go fanoutUserPush(db, userID, kind, title, body)
|
||||
}
|
||||
|
||||
func mapNotification(row sqlcdb.Notification) NotificationItem {
|
||||
|
||||
@@ -364,6 +364,10 @@ func SetupRoutes(router *gin.Engine, db *database.DB, redis *database.Redis, cfg
|
||||
authed.GET("/notifications", handleListNotifications)
|
||||
authed.POST("/notifications/:id/read", handleMarkNotificationRead)
|
||||
authed.POST("/notifications/read-all", handleMarkAllNotificationsRead)
|
||||
authed.GET("/notifications/channels", handleListNotificationChannels)
|
||||
authed.POST("/notifications/channels", handleCreateNotificationChannel)
|
||||
authed.POST("/notifications/channels/:id/test", handleTestNotificationChannel)
|
||||
authed.DELETE("/notifications/channels/:id", handleDeleteNotificationChannel)
|
||||
|
||||
// WebSocket endpoint
|
||||
authed.GET("/ws", handleWebSocket)
|
||||
|
||||
Reference in new issue
Block a user