teg/backend/chat_websocket.go

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"`
	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
	}

	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
		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
}