// backend\chat_websocket.go package main import ( "context" "encoding/json" "errors" "net/http" "net/url" "strings" "sync" "time" "github.com/coder/websocket" "github.com/coder/websocket/wsjson" ) type chatSocketCommand struct { Type string `json:"type"` ClientID string `json:"clientId"` ConversationID string `json:"conversationId"` Body string `json:"body"` AttachmentIDs []string `json:"attachmentIds"` ReplyToMessageID string `json:"replyToMessageId"` ForwardedFromMessageID string `json:"forwardedFromMessageId"` Location *ChatLocation `json:"location"` } type chatSocketEvent struct { Type string `json:"type"` ClientID string `json:"clientId,omitempty"` ConversationID string `json:"conversationId,omitempty"` UserID string `json:"userId,omitempty"` UserDisplayName string `json:"userDisplayName,omitempty"` Typing bool `json:"typing,omitempty"` Message *ChatMessage `json:"message,omitempty"` Conversation *ChatConversation `json:"conversation,omitempty"` Error string `json:"error,omitempty"` } type chatSocketClient struct { userID string ch chan chatSocketEvent } type chatSocketBroker struct { mu sync.RWMutex clients map[*chatSocketClient]struct{} } var chatSockets = &chatSocketBroker{ clients: map[*chatSocketClient]struct{}{}, } func (b *chatSocketBroker) subscribe(userID string) *chatSocketClient { client := &chatSocketClient{ userID: userID, ch: make(chan chatSocketEvent, 32), } b.mu.Lock() b.clients[client] = struct{}{} b.mu.Unlock() return client } func (b *chatSocketBroker) unsubscribe(client *chatSocketClient) { b.mu.Lock() delete(b.clients, client) b.mu.Unlock() } func (b *chatSocketBroker) publish(userID string, event chatSocketEvent) { b.mu.RLock() defer b.mu.RUnlock() for client := range b.clients { if client.userID != userID { continue } select { case client.ch <- event: default: } } } func (c *chatSocketClient) send(event chatSocketEvent) { select { case c.ch <- event: default: } } func (s *Server) handleChatWebSocket(w http.ResponseWriter, r *http.Request) { user, ok := userFromContext(r.Context()) if !ok { writeError(w, http.StatusUnauthorized, "Unauthorized") return } // coder/websocket matcht OriginPatterns gegen den Host der Origin (z. B. // "localhost:5173"). Daher nur den Host übergeben – schema-unabhängig, damit // http/https sowie ein Dev-Proxy mit geändertem Host (changeOrigin) den // Upgrade nicht blockieren. originPatterns := []string{} if frontendOrigin := strings.TrimRight(s.frontendOrigin, "/"); frontendOrigin != "" { host := frontendOrigin if parsed, err := url.Parse(frontendOrigin); err == nil && parsed.Host != "" { host = parsed.Host } originPatterns = append(originPatterns, host) } conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{ OriginPatterns: originPatterns, }) if err != nil { return } defer conn.CloseNow() conn.SetReadLimit(8 * 1024) client := chatSockets.subscribe(user.ID) defer chatSockets.unsubscribe(client) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go s.readChatSocket(ctx, cancel, conn, client, user) client.send(chatSocketEvent{Type: "connected"}) pingTicker := time.NewTicker(30 * time.Second) defer pingTicker.Stop() for { select { case <-ctx.Done(): return case event, open := <-client.ch: if !open { return } writeCtx, writeCancel := context.WithTimeout(ctx, 10*time.Second) err := wsjson.Write(writeCtx, conn, event) writeCancel() if err != nil { return } case <-pingTicker.C: pingCtx, pingCancel := context.WithTimeout(ctx, 10*time.Second) err := conn.Ping(pingCtx) pingCancel() if err != nil { return } } } } func (s *Server) readChatSocket( ctx context.Context, cancel context.CancelFunc, conn *websocket.Conn, client *chatSocketClient, user User, ) { defer cancel() for { var command chatSocketCommand if err := wsjson.Read(ctx, conn, &command); err != nil { return } switch command.Type { case "typing.start", "typing.stop": if !s.canPostToConversation(ctx, command.ConversationID, user.ID) { client.send(chatSocketEvent{ Type: "error", Error: "Dieser Chat ist schreibgeschützt", }) continue } s.publishChatTyping( ctx, user, command.ConversationID, command.Type == "typing.start", ) continue case "message.send": message, err := s.createChatMessage( ctx, user, command.ConversationID, createChatMessageInput{ Body: command.Body, AttachmentIDs: command.AttachmentIDs, ReplyToMessageID: command.ReplyToMessageID, ForwardedFromMessageID: command.ForwardedFromMessageID, Location: command.Location, }, ) if err != nil { client.send(chatSocketEvent{ Type: "error", ClientID: command.ClientID, Error: chatSocketErrorMessage(err), }) continue } s.publishChatMessage(ctx, message, command.ClientID) continue default: client.send(chatSocketEvent{ Type: "error", ClientID: command.ClientID, Error: "Unbekannte Chat-Aktion", }) } } } func chatSocketErrorMessage(err error) string { switch { case errors.Is(err, errChatAccessDenied): return "Kein Zugriff auf diesen Chat" case errors.Is(err, errChatMessageEmpty): return "Nachricht darf nicht leer sein" case errors.Is(err, errChatMessageTooLong): return "Nachricht darf höchstens 4000 Zeichen enthalten" case errors.Is(err, errChatLocationInvalid): return "Ungültiger Standort" case errors.Is(err, errChatForwardSame): return "Nachrichten können nicht in denselben Chat weitergeleitet werden" case errors.Is(err, errChatReadOnly): return "Channels sind schreibgeschützt" default: return "Nachricht konnte nicht gesendet werden" } } func (s *Server) publishChatTyping( ctx context.Context, user User, conversationID string, typing bool, ) { memberIDs, err := s.chatConversationMemberIDs(ctx, conversationID) if err != nil { return } event := chatSocketEvent{ Type: "typing.updated", ConversationID: conversationID, UserID: user.ID, UserDisplayName: user.DisplayName, Typing: typing, } if event.UserDisplayName == "" { event.UserDisplayName = user.Username } if event.UserDisplayName == "" { event.UserDisplayName = user.Email } for _, memberID := range memberIDs { if memberID != user.ID { chatSockets.publish(memberID, event) } } } func (s *Server) publishChatMessage( ctx context.Context, message ChatMessage, clientID string, ) { rows, err := s.db.Query( ctx, ` SELECT cm.user_id::TEXT, cm.muted FROM conversation_members cm WHERE cm.conversation_id = $1 `, message.ConversationID, ) if err != nil { return } defer rows.Close() event := chatSocketEvent{ Type: "message.created", ClientID: clientID, Message: &message, } recipientIDs := []string{} for rows.Next() { var recipientID string var muted bool if err := rows.Scan(&recipientID, &muted); err != nil { continue } chatSockets.publish(recipientID, event) if recipientID != message.Sender.ID && !muted { recipientIDs = append(recipientIDs, recipientID) } } rows.Close() // System-Nachrichten (Gruppen-Ereignisse, Selbstlöschen) erscheinen im // Verlauf, lösen aber keine Benachrichtigung / Web-Push aus. if message.Type == "system" { return } title, conversationType, conversationImage, err := s.chatNotificationTitle(ctx, message) if err != nil { return } preview := strings.TrimSpace(message.Body) if preview == "" && len(message.Attachments) > 0 { preview = "Datei angehängt" } previewRunes := []rune(preview) if len(previewRunes) > 160 { preview = string(previewRunes[:160]) + "..." } data, err := json.Marshal(map[string]any{ "conversationId": message.ConversationID, "messageId": message.ID, "senderId": message.Sender.ID, "senderAvatar": message.Sender.Avatar, "senderName": chatUserDisplayName(message.Sender), "conversationType": conversationType, "conversationImage": conversationImage, }) if err != nil { return } for _, recipientID := range recipientIDs { notificationType := "chat.message" if conversationType == "channel" { notificationType = "channel.message" } _ = s.createAndPublishNotification( ctx, recipientID, notificationType, title, preview, "chat", message.ConversationID, data, false, ) } s.scheduleLinkPreview(message) } func (s *Server) chatConversationMemberIDs(ctx context.Context, conversationID string) ([]string, error) { rows, err := s.db.Query( ctx, ` SELECT cm.user_id::TEXT FROM conversation_members cm WHERE cm.conversation_id = $1 `, conversationID, ) if err != nil { return nil, err } defer rows.Close() memberIDs := []string{} for rows.Next() { var memberID string if err := rows.Scan(&memberID); err != nil { return nil, err } memberIDs = append(memberIDs, memberID) } return memberIDs, rows.Err() } func (s *Server) publishChatMessageUpdate(ctx context.Context, message ChatMessage) { memberIDs, err := s.chatConversationMemberIDs(ctx, message.ConversationID) if err != nil { return } event := chatSocketEvent{ Type: "message.updated", Message: &message, } for _, memberID := range memberIDs { chatSockets.publish(memberID, event) } } // publishConversationUpdate sendet jedem Mitglied die aktualisierte // Konversation (Name/Bild/Selbstlöschen/Mitglieder), damit Seitenleiste und // geöffneter Chat live aktualisiert werden. func (s *Server) publishConversationUpdate(ctx context.Context, conversationID string) { rows, err := s.db.Query( ctx, ` SELECT cm.user_id::TEXT, u.show_last_seen FROM conversation_members cm JOIN users u ON u.id = cm.user_id WHERE cm.conversation_id = $1 `, conversationID, ) if err != nil { return } defer rows.Close() type member struct { id string showLastSeen bool } members := []member{} for rows.Next() { var m member if err := rows.Scan(&m.id, &m.showLastSeen); err != nil { return } members = append(members, m) } rows.Close() for _, m := range members { conversation, err := s.getChatConversation(ctx, conversationID, m.id, m.showLastSeen) if err != nil { continue } chatSockets.publish(m.id, chatSocketEvent{ Type: "conversation.updated", Conversation: &conversation, }) } } func (s *Server) publishChatMessagesReload(ctx context.Context, conversationID string) { memberIDs, err := s.chatConversationMemberIDs(ctx, conversationID) if err != nil { return } for _, memberID := range memberIDs { chatSockets.publish(memberID, chatSocketEvent{ Type: "messages.reload", ConversationID: conversationID, }) } } func chatUserDisplayName(user ChatUser) string { if user.DisplayName != "" { return user.DisplayName } if user.Username != "" { return user.Username } return user.Email } func (s *Server) chatNotificationTitle( ctx context.Context, message ChatMessage, ) (string, string, string, error) { var conversationType string var conversationName string var conversationImage string err := s.db.QueryRow( ctx, ` SELECT type, name, channel_image FROM conversations WHERE id = $1 `, message.ConversationID, ).Scan(&conversationType, &conversationName, &conversationImage) if err != nil { return "", "", "", err } senderName := chatUserDisplayName(message.Sender) if conversationType == "group" && conversationName != "" { return senderName + " in " + conversationName, conversationType, conversationImage, nil } if conversationType == "channel" && conversationName != "" { return "Neue Meldung in " + conversationName, conversationType, conversationImage, nil } return "Neue Nachricht von " + senderName, conversationType, conversationImage, nil }