package main import ( "context" "crypto/rand" "crypto/sha256" "crypto/subtle" "encoding/base64" "encoding/hex" "errors" "net/http" "regexp" "strings" "time" "github.com/jackc/pgx/v5" ) const zabbixWebhookScript = `var params = JSON.parse(value), request = new HttpRequest(), payload = { subject: params.Subject, message: params.Message, host: params.Host, severity: params.Severity, status: params.Status, eventId: params.EventId, eventUrl: params.EventUrl }; request.addHeader('Content-Type: application/json'); request.addHeader('Authorization: Bearer ' + params.Token); var response = request.post(params.URL, JSON.stringify(payload)), status = request.getStatus(); if (status < 200 || status >= 300) { throw 'Webhook failed with HTTP status ' + status + ': ' + response; } return response;` var channelSlugPattern = regexp.MustCompile(`^[a-z0-9]+(?:-[a-z0-9]+)*$`) type AdminChannel struct { ID string `json:"id"` Name string `json:"name"` Slug string `json:"slug"` SourceName string `json:"sourceName"` Image string `json:"image"` AllUsers bool `json:"allUsers"` UserIDs []string `json:"userIds"` WebhookEnabled bool `json:"webhookEnabled"` TokenConfigured bool `json:"tokenConfigured"` WebhookTokenUpdatedAt *time.Time `json:"webhookTokenUpdatedAt"` WebhookPath string `json:"webhookPath"` } type AdminChannelUser struct { ID string `json:"id"` Username string `json:"username"` DisplayName string `json:"displayName"` Email string `json:"email"` Unit string `json:"unit"` Avatar string `json:"avatar"` } type saveAdminChannelRequest struct { Name string `json:"name"` Slug string `json:"slug"` SourceName string `json:"sourceName"` Image string `json:"image"` AllUsers bool `json:"allUsers"` UserIDs []string `json:"userIds"` WebhookEnabled bool `json:"webhookEnabled"` } type channelWebhookPayload struct { Subject string `json:"subject"` Message string `json:"message"` Host string `json:"host"` Severity string `json:"severity"` Status string `json:"status"` EventID string `json:"eventId"` EventURL string `json:"eventUrl"` } func (s *Server) ensureAllUserChannels(ctx context.Context, userID string) error { _, err := s.db.Exec( ctx, ` INSERT INTO conversation_members (conversation_id, user_id) SELECT c.id, $1 FROM conversations c WHERE c.type = 'channel' AND c.channel_all_users ON CONFLICT (conversation_id, user_id) DO NOTHING `, userID, ) return err } func normalizeChannelSlug(value string) string { return strings.ToLower(strings.TrimSpace(value)) } func validateAdminChannelInput(input saveAdminChannelRequest) (saveAdminChannelRequest, error) { input.Name = strings.TrimSpace(input.Name) input.Slug = normalizeChannelSlug(input.Slug) input.SourceName = strings.TrimSpace(input.SourceName) input.Image = strings.TrimSpace(input.Image) userIDs := input.UserIDs uniqueUserIDs := make(map[string]struct{}, len(userIDs)) input.UserIDs = make([]string, 0, len(userIDs)) for _, userID := range userIDs { userID = strings.TrimSpace(userID) if userID == "" { continue } if _, exists := uniqueUserIDs[userID]; exists { continue } uniqueUserIDs[userID] = struct{}{} input.UserIDs = append(input.UserIDs, userID) } if input.Name == "" { return input, errors.New("Channelname ist erforderlich") } if !channelSlugPattern.MatchString(input.Slug) { return input, errors.New("Der Webhook-Slug darf nur Kleinbuchstaben, Zahlen und Bindestriche enthalten") } if input.SourceName == "" { input.SourceName = input.Name } if len(input.Image) > 3*1024*1024 { return input, errors.New("Das Channel-Bild ist zu groß") } return input, nil } func (s *Server) handleAdminListChannels(w http.ResponseWriter, r *http.Request) { channels, err := s.listAdminChannels(r.Context()) if err != nil { writeError(w, http.StatusInternalServerError, "Channels konnten nicht geladen werden") return } users, err := s.listAdminChannelUsers(r.Context()) if err != nil { writeError(w, http.StatusInternalServerError, "Benutzer konnten nicht geladen werden") return } writeJSON(w, http.StatusOK, map[string]any{ "channels": channels, "users": users, "zabbixScript": zabbixWebhookScript, }) } func (s *Server) listAdminChannels(ctx context.Context) ([]AdminChannel, error) { rows, err := s.db.Query( ctx, ` SELECT c.id::TEXT, c.name, c.slug, c.channel_source_name, c.channel_image, c.channel_all_users, c.webhook_enabled, c.webhook_token_hash <> '', c.webhook_token_updated_at, COALESCE(array_agg(cm.user_id::TEXT) FILTER (WHERE cm.user_id IS NOT NULL), '{}') FROM conversations c LEFT JOIN conversation_members cm ON cm.conversation_id = c.id WHERE c.type = 'channel' GROUP BY c.id ORDER BY lower(c.name) `, ) if err != nil { return nil, err } defer rows.Close() channels := []AdminChannel{} for rows.Next() { var channel AdminChannel if err := rows.Scan( &channel.ID, &channel.Name, &channel.Slug, &channel.SourceName, &channel.Image, &channel.AllUsers, &channel.WebhookEnabled, &channel.TokenConfigured, &channel.WebhookTokenUpdatedAt, &channel.UserIDs, ); err != nil { return nil, err } channel.WebhookPath = "/webhooks/channels/" + channel.Slug channels = append(channels, channel) } return channels, rows.Err() } func (s *Server) listAdminChannelUsers(ctx context.Context) ([]AdminChannelUser, error) { rows, err := s.db.Query( ctx, ` SELECT id::TEXT, COALESCE(username, ''), COALESCE(display_name, ''), email, COALESCE(unit, ''), COALESCE(avatar, '') FROM users ORDER BY lower(COALESCE(NULLIF(display_name, ''), username, email)) `, ) if err != nil { return nil, err } defer rows.Close() users := []AdminChannelUser{} for rows.Next() { var user AdminChannelUser if err := rows.Scan( &user.ID, &user.Username, &user.DisplayName, &user.Email, &user.Unit, &user.Avatar, ); err != nil { return nil, err } users = append(users, user) } return users, rows.Err() } func (s *Server) handleAdminCreateChannel(w http.ResponseWriter, r *http.Request) { var input saveAdminChannelRequest if err := readJSON(r, &input); err != nil { writeError(w, http.StatusBadRequest, "Ungültige Anfrage") return } input, err := validateAdminChannelInput(input) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } tx, err := s.db.Begin(r.Context()) if err != nil { writeError(w, http.StatusInternalServerError, "Channel konnte nicht erstellt werden") return } defer tx.Rollback(r.Context()) var channelID string err = tx.QueryRow( r.Context(), ` INSERT INTO conversations ( type, name, slug, channel_source_name, channel_image, channel_all_users, webhook_enabled ) VALUES ('channel', $1, $2, $3, $4, $5, $6) RETURNING id::TEXT `, input.Name, input.Slug, input.SourceName, input.Image, input.AllUsers, input.WebhookEnabled, ).Scan(&channelID) if err != nil { writeError(w, http.StatusConflict, "Channelname oder Webhook-Slug wird bereits verwendet") return } if err := syncChannelMembers(r.Context(), tx, channelID, input.AllUsers, input.UserIDs); err != nil { writeError(w, http.StatusInternalServerError, "Channelmitglieder konnten nicht gespeichert werden") return } if err := tx.Commit(r.Context()); err != nil { writeError(w, http.StatusInternalServerError, "Channel konnte nicht gespeichert werden") return } writeJSON(w, http.StatusCreated, map[string]any{"id": channelID}) } func (s *Server) handleAdminUpdateChannel(w http.ResponseWriter, r *http.Request) { channelID := strings.TrimSpace(r.PathValue("id")) var input saveAdminChannelRequest if err := readJSON(r, &input); err != nil { writeError(w, http.StatusBadRequest, "Ungültige Anfrage") return } input, err := validateAdminChannelInput(input) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return } tx, err := s.db.Begin(r.Context()) if err != nil { writeError(w, http.StatusInternalServerError, "Channel konnte nicht aktualisiert werden") return } defer tx.Rollback(r.Context()) commandTag, err := tx.Exec( r.Context(), ` UPDATE conversations SET name = $2, slug = $3, channel_source_name = $4, channel_image = $5, channel_all_users = $6, webhook_enabled = $7, updated_at = now() WHERE id = $1 AND type = 'channel' `, channelID, input.Name, input.Slug, input.SourceName, input.Image, input.AllUsers, input.WebhookEnabled, ) if err != nil { writeError(w, http.StatusConflict, "Channelname oder Webhook-Slug wird bereits verwendet") return } if commandTag.RowsAffected() == 0 { writeError(w, http.StatusNotFound, "Channel wurde nicht gefunden") return } if err := syncChannelMembers(r.Context(), tx, channelID, input.AllUsers, input.UserIDs); err != nil { writeError(w, http.StatusInternalServerError, "Channelmitglieder konnten nicht gespeichert werden") return } if err := tx.Commit(r.Context()); err != nil { writeError(w, http.StatusInternalServerError, "Channel konnte nicht gespeichert werden") return } writeJSON(w, http.StatusOK, map[string]any{"id": channelID}) } func syncChannelMembers( ctx context.Context, tx pgx.Tx, channelID string, allUsers bool, userIDs []string, ) error { if allUsers { _, err := tx.Exec( ctx, ` INSERT INTO conversation_members (conversation_id, user_id) SELECT $1, id FROM users ON CONFLICT (conversation_id, user_id) DO NOTHING `, channelID, ) return err } if _, err := tx.Exec( ctx, ` DELETE FROM conversation_members WHERE conversation_id = $1 AND NOT ( user_id::TEXT = ANY( COALESCE($2::TEXT[], ARRAY[]::TEXT[]) ) ) `, channelID, userIDs, ); err != nil { return err } for _, userID := range userIDs { if _, err := tx.Exec( ctx, ` INSERT INTO conversation_members (conversation_id, user_id) SELECT $1, id FROM users WHERE id = $2 ON CONFLICT (conversation_id, user_id) DO NOTHING `, channelID, strings.TrimSpace(userID), ); err != nil { return err } } return nil } func generateChannelWebhookToken() (string, string, error) { raw := make([]byte, 32) if _, err := rand.Read(raw); err != nil { return "", "", err } token := base64.RawURLEncoding.EncodeToString(raw) hash := sha256.Sum256([]byte(token)) return token, hex.EncodeToString(hash[:]), nil } func (s *Server) handleAdminRotateChannelToken(w http.ResponseWriter, r *http.Request) { channelID := strings.TrimSpace(r.PathValue("id")) token, tokenHash, err := generateChannelWebhookToken() if err != nil { writeError(w, http.StatusInternalServerError, "Webhook-Token konnte nicht erzeugt werden") return } commandTag, err := s.db.Exec( r.Context(), ` UPDATE conversations SET webhook_token_hash = $2, webhook_token_updated_at = now(), updated_at = now() WHERE id = $1 AND type = 'channel' `, channelID, tokenHash, ) if err != nil { writeError(w, http.StatusInternalServerError, "Webhook-Token konnte nicht gespeichert werden") return } if commandTag.RowsAffected() == 0 { writeError(w, http.StatusNotFound, "Channel wurde nicht gefunden") return } writeJSON(w, http.StatusOK, map[string]any{"token": token}) } func (s *Server) handleChannelWebhook(w http.ResponseWriter, r *http.Request) { r.Body = http.MaxBytesReader(w, r.Body, 64*1024) slug := normalizeChannelSlug(r.PathValue("slug")) var channelID string var tokenHash string var enabled bool err := s.db.QueryRow( r.Context(), ` SELECT id::TEXT, webhook_token_hash, webhook_enabled FROM conversations WHERE type = 'channel' AND slug = $1 `, slug, ).Scan(&channelID, &tokenHash, &enabled) if err != nil || !enabled || tokenHash == "" { writeError(w, http.StatusNotFound, "Webhook wurde nicht gefunden") return } token := strings.TrimSpace(strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")) if token == "" { token = strings.TrimSpace(r.Header.Get("X-Webhook-Token")) } hash := sha256.Sum256([]byte(token)) providedHash := hex.EncodeToString(hash[:]) if subtle.ConstantTimeCompare([]byte(providedHash), []byte(tokenHash)) != 1 { writeError(w, http.StatusUnauthorized, "Ungültiges Webhook-Token") return } var input channelWebhookPayload if err := readJSON(r, &input); err != nil { writeError(w, http.StatusBadRequest, "Ungültige Webhook-Nachricht") return } body := formatChannelWebhookMessage(input) if body == "" { writeError(w, http.StatusBadRequest, "Nachricht darf nicht leer sein") return } message, err := s.createChannelMessage(r.Context(), channelID, body) if err != nil { writeError(w, http.StatusInternalServerError, "Channel-Nachricht konnte nicht gespeichert werden") return } s.publishChatMessage(r.Context(), message, "") writeJSON(w, http.StatusCreated, map[string]any{ "messageId": message.ID, "status": "accepted", }) } func formatChannelWebhookMessage(input channelWebhookPayload) string { parts := []string{} if subject := strings.TrimSpace(input.Subject); subject != "" { parts = append(parts, subject) } if message := strings.TrimSpace(input.Message); message != "" { parts = append(parts, message) } details := []string{} for _, detail := range []struct { label string value string }{ {"Host", input.Host}, {"Status", input.Status}, {"Schweregrad", input.Severity}, {"Event-ID", input.EventID}, {"Link", input.EventURL}, } { if value := strings.TrimSpace(detail.value); value != "" { details = append(details, detail.label+": "+value) } } if len(details) > 0 { parts = append(parts, strings.Join(details, "\n")) } body := strings.Join(parts, "\n\n") runes := []rune(body) if len(runes) > 4000 { body = string(runes[:4000]) } return body } func (s *Server) createChannelMessage( ctx context.Context, conversationID string, body string, ) (ChatMessage, error) { tx, err := s.db.Begin(ctx) if err != nil { return ChatMessage{}, err } defer tx.Rollback(ctx) var messageID string err = tx.QueryRow( ctx, ` INSERT INTO messages (conversation_id, body, expires_at) SELECT id, $2, CASE WHEN disappearing_seconds > 0 THEN now() + make_interval(secs => disappearing_seconds) ELSE NULL END FROM conversations WHERE id = $1 AND type = 'channel' RETURNING id::TEXT `, conversationID, body, ).Scan(&messageID) if err != nil { return ChatMessage{}, err } if _, err := tx.Exec( ctx, ` UPDATE conversations SET last_message_at = now(), updated_at = now() WHERE id = $1 `, conversationID, ); err != nil { return ChatMessage{}, err } if err := tx.Commit(ctx); err != nil { return ChatMessage{}, err } return s.getChatMessage(ctx, messageID) }