package main import ( "context" "encoding/json" "errors" "net/http" "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"` 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 } originPatterns := []string{} if frontendOrigin := strings.TrimRight(s.frontendOrigin, "/"); frontendOrigin != "" { originPatterns = append(originPatterns, frontendOrigin) } 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.canAccessConversation(ctx, command.ConversationID, user.ID) || !s.isConversationWritable(ctx, command.ConversationID) { 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 JOIN conversations c ON c.id = cm.conversation_id WHERE cm.conversation_id = $1 AND ( c.type <> 'team' OR EXISTS ( SELECT 1 FROM user_teams ut WHERE ut.team_id = c.team_id AND ut.user_id = cm.user_id ) ) `, 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() title, conversationType, 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, }) if err != nil { return } for _, recipientID := range recipientIDs { _ = s.createAndPublishNotification( ctx, recipientID, "chat.message", 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 JOIN conversations c ON c.id = cm.conversation_id WHERE cm.conversation_id = $1 AND ( c.type <> 'team' OR EXISTS ( SELECT 1 FROM user_teams ut WHERE ut.team_id = c.team_id AND ut.user_id = cm.user_id ) ) `, 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) } } 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, error) { var conversationType string var conversationName string err := s.db.QueryRow( ctx, ` SELECT type, name FROM conversations WHERE id = $1 `, message.ConversationID, ).Scan(&conversationType, &conversationName) if err != nil { return "", "", err } senderName := chatUserDisplayName(message.Sender) if conversationType == "team" && conversationName != "" { return senderName + " in " + conversationName, conversationType, nil } if conversationType == "channel" && conversationName != "" { return "Neue Meldung in " + conversationName, conversationType, nil } return "Neue Nachricht von " + senderName, conversationType, nil }