package dm import ( "context" "database/sql" "encoding/json" "errors" "fmt" "log/slog" "net/http" "sort" "strconv" "strings" "git.dustin.coffee/hobokenchicken/dumpsterChat/internal/gateway" "git.dustin.coffee/hobokenchicken/dumpsterChat/internal/middleware" "git.dustin.coffee/hobokenchicken/dumpsterChat/internal/push" "github.com/go-chi/chi/v5" ) // Handler handles direct-message conversations. type Handler struct { db *sql.DB hub *gateway.Hub pushHandler *push.Handler logger *slog.Logger } // NewHandler creates a new DM handler. func NewHandler(db *sql.DB, hub *gateway.Hub, pushHandler *push.Handler, logger *slog.Logger) *Handler { return &Handler{db: db, hub: hub, pushHandler: pushHandler, logger: logger} } // RegisterRoutes registers conversation routes. func (h *Handler) RegisterRoutes(r chi.Router) { r.Post("/", h.Create) r.Get("/", h.List) r.Route("/{conversationID}", func(r chi.Router) { r.Get("/", h.Get) r.Get("/messages", h.ListMessages) r.Post("/messages", h.SendMessage) r.Route("/messages/{messageID}/reactions", func(r chi.Router) { r.Post("/", h.AddReaction) r.Delete("/{emoji}", h.RemoveReaction) }) }) } type createConversationRequest struct { UserIDs []string `json:"user_ids"` } type conversationResponse struct { ID string `json:"id"` Type string `json:"type"` Name string `json:"name"` Members []member `json:"members"` CreatedAt string `json:"created_at"` } type member struct { ID string `json:"id"` Username string `json:"username"` DisplayName string `json:"display_name"` Avatar string `json:"avatar"` } // Create creates a new DM or group DM. func (h *Handler) Create(w http.ResponseWriter, r *http.Request) { userID, ok := middleware.UserIDFromContext(r.Context()) if !ok { http.Error(w, `{"error":"unauthorized"}`, http.StatusUnauthorized) return } var req createConversationRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, `{"error":"invalid request"}`, http.StatusBadRequest) return } // ponytail: user_ids can be empty for self-DM (notes to self) // Build unique member set including creator. memberSet := map[string]struct{}{userID: {}} for _, uid := range req.UserIDs { if uid != "" { memberSet[uid] = struct{}{} } } // ponytail: allow self-DM (notes to self) — len(memberSet)==1 is fine memberIDs := make([]string, 0, len(memberSet)) for uid := range memberSet { memberIDs = append(memberIDs, uid) } sort.Strings(memberIDs) // For a 1:1 DM, check if it already exists. if len(memberIDs) == 1 { var existingID string err := h.db.QueryRowContext(r.Context(), ` SELECT c.id FROM conversations c WHERE c.type = 'dm' AND (SELECT COUNT(*) FROM conversation_members cm WHERE cm.conversation_id = c.id) = 1 AND EXISTS (SELECT 1 FROM conversation_members cm2 WHERE cm2.conversation_id = c.id AND cm2.user_id = $1) LIMIT 1 `, memberIDs[0]).Scan(&existingID) if err == nil { h.getByID(w, r, existingID) return } if !errors.Is(err, sql.ErrNoRows) { http.Error(w, `{"error":"server error"}`, http.StatusInternalServerError) return } } else if len(memberIDs) == 2 { var existingID string err := h.db.QueryRowContext(r.Context(), ` SELECT c.id FROM conversations c WHERE c.type = 'dm' AND (SELECT COUNT(*) FROM conversation_members cm WHERE cm.conversation_id = c.id) = 2 AND NOT EXISTS ( SELECT 1 FROM conversation_members cm2 WHERE cm2.conversation_id = c.id AND cm2.user_id NOT IN ($1, $2) ) LIMIT 1 `, memberIDs[0], memberIDs[1]).Scan(&existingID) if err == nil { h.getByID(w, r, existingID) return } if !errors.Is(err, sql.ErrNoRows) { http.Error(w, `{"error":"server error"}`, http.StatusInternalServerError) return } } convType := "dm" if len(memberIDs) > 2 { convType = "group_dm" } tx, err := h.db.BeginTx(r.Context(), nil) if err != nil { http.Error(w, `{"error":"server error"}`, http.StatusInternalServerError) return } defer tx.Rollback() var convID string var createdAt string err = tx.QueryRowContext(r.Context(), ` INSERT INTO conversations (type, created_by) VALUES ($1, $2) RETURNING id, created_at::text `, convType, userID).Scan(&convID, &createdAt) if err != nil { http.Error(w, `{"error":"failed to create conversation"}`, http.StatusInternalServerError) return } for _, uid := range memberIDs { _, err = tx.ExecContext(r.Context(), ` INSERT INTO conversation_members (conversation_id, user_id) VALUES ($1, $2) `, convID, uid) if err != nil { http.Error(w, `{"error":"failed to add member"}`, http.StatusInternalServerError) return } } if err := tx.Commit(); err != nil { http.Error(w, `{"error":"server error"}`, http.StatusInternalServerError) return } resp, err := h.buildResponse(r.Context(), convID) if err != nil { http.Error(w, `{"error":"server error"}`, http.StatusInternalServerError) return } w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusCreated) json.NewEncoder(w).Encode(resp) } // Get returns a single conversation. func (h *Handler) Get(w http.ResponseWriter, r *http.Request) { convID := chi.URLParam(r, "conversationID") resp, err := h.buildResponse(r.Context(), convID) if err != nil { http.Error(w, `{"error":"conversation not found"}`, http.StatusNotFound) return } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(resp) } func (h *Handler) getByID(w http.ResponseWriter, r *http.Request, convID string) { resp, err := h.buildResponse(r.Context(), convID) if err != nil { http.Error(w, `{"error":"conversation not found"}`, http.StatusNotFound) return } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(resp) } // List returns the current user's conversations. func (h *Handler) List(w http.ResponseWriter, r *http.Request) { userID, ok := middleware.UserIDFromContext(r.Context()) if !ok { http.Error(w, `{"error":"unauthorized"}`, http.StatusUnauthorized) return } rows, err := h.db.QueryContext(r.Context(), ` SELECT c.id FROM conversations c JOIN conversation_members cm ON cm.conversation_id = c.id WHERE cm.user_id = $1 ORDER BY c.created_at DESC `, userID) if err != nil { http.Error(w, `{"error":"server error"}`, http.StatusInternalServerError) return } defer rows.Close() var conversations []conversationResponse for rows.Next() { var id string if err := rows.Scan(&id); err != nil { continue } resp, err := h.buildResponse(r.Context(), id) if err != nil { continue } conversations = append(conversations, resp) } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(conversations) } func (h *Handler) buildResponse(ctx context.Context, convID string) (conversationResponse, error) { var resp conversationResponse err := h.db.QueryRowContext(ctx, ` SELECT id, type, COALESCE(name, ''), created_at::text FROM conversations WHERE id = $1 `, convID).Scan(&resp.ID, &resp.Type, &resp.Name, &resp.CreatedAt) if err != nil { return resp, err } memberRows, err := h.db.QueryContext(ctx, ` SELECT u.id, u.username, COALESCE(u.display_name, ''), COALESCE(u.avatar, '') FROM conversation_members cm JOIN users u ON u.id = cm.user_id WHERE cm.conversation_id = $1 `, convID) if err != nil { return resp, err } defer memberRows.Close() for memberRows.Next() { var m member if err := memberRows.Scan(&m.ID, &m.Username, &m.DisplayName, &m.Avatar); err != nil { continue } resp.Members = append(resp.Members, m) } return resp, nil } type messageResponse struct { ID string `json:"id"` ConversationID string `json:"conversation_id"` AuthorID string `json:"author_id"` AuthorName string `json:"author_username"` DisplayName *string `json:"author_display_name"` Content string `json:"content"` EditedAt *string `json:"edited_at"` CreatedAt string `json:"created_at"` Reactions []emojiGroup `json:"reactions"` } // ListMessages lists messages in a conversation. func (h *Handler) ListMessages(w http.ResponseWriter, r *http.Request) { userID, ok := middleware.UserIDFromContext(r.Context()) if !ok { http.Error(w, `{"error":"unauthorized"}`, http.StatusUnauthorized) return } convID := chi.URLParam(r, "conversationID") if !h.isMember(r.Context(), convID, userID) { http.Error(w, `{"error":"not a member"}`, http.StatusForbidden) return } limitStr := r.URL.Query().Get("limit") limit := 50 if limitStr != "" { if l, err := strconv.Atoi(limitStr); err == nil && l > 0 && l <= 100 { limit = l } } before := r.URL.Query().Get("before") var rows *sql.Rows var err error if before != "" { rows, err = h.db.QueryContext(r.Context(), ` SELECT m.id, m.conversation_id, m.author_id, u.username, u.display_name, m.content, m.edited_at::text, m.created_at::text FROM conversation_messages m JOIN users u ON u.id = m.author_id WHERE m.conversation_id = $1 AND m.created_at < (SELECT created_at FROM conversation_messages WHERE id = $2) ORDER BY m.created_at DESC LIMIT $3 `, convID, before, limit) } else { rows, err = h.db.QueryContext(r.Context(), ` SELECT m.id, m.conversation_id, m.author_id, u.username, u.display_name, m.content, m.edited_at::text, m.created_at::text FROM conversation_messages m JOIN users u ON u.id = m.author_id WHERE m.conversation_id = $1 ORDER BY m.created_at DESC LIMIT $2 `, convID, limit) } if err != nil { http.Error(w, `{"error":"server error"}`, http.StatusInternalServerError) return } defer rows.Close() messages := make([]messageResponse, 0) for rows.Next() { var msg messageResponse var editedAt sql.NullString var createdAt sql.NullString var displayName sql.NullString if err := rows.Scan(&msg.ID, &msg.ConversationID, &msg.AuthorID, &msg.AuthorName, &displayName, &msg.Content, &editedAt, &createdAt); err != nil { continue } if editedAt.Valid { msg.EditedAt = &editedAt.String } msg.CreatedAt = createdAt.String if displayName.Valid { msg.DisplayName = &displayName.String } messages = append(messages, msg) } messages = h.attachReactions(r.Context(), messages) // Reverse: query returns DESC (newest first), client expects ASC (oldest first). for i, j := 0, len(messages)-1; i < j; i, j = i+1, j-1 { messages[i], messages[j] = messages[j], messages[i] } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(messages) } // SendMessage sends a message to a conversation. func (h *Handler) SendMessage(w http.ResponseWriter, r *http.Request) { userID, ok := middleware.UserIDFromContext(r.Context()) if !ok { http.Error(w, `{"error":"unauthorized"}`, http.StatusUnauthorized) return } convID := chi.URLParam(r, "conversationID") if !h.isMember(r.Context(), convID, userID) { http.Error(w, `{"error":"not a member"}`, http.StatusForbidden) return } var req struct { Content string `json:"content"` } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, `{"error":"invalid request"}`, http.StatusBadRequest) return } if req.Content == "" || len(req.Content) > 4000 { http.Error(w, `{"error":"invalid content"}`, http.StatusBadRequest) return } var msg messageResponse var editedAt sql.NullString var createdAt sql.NullString var displayName sql.NullString err := h.db.QueryRowContext(r.Context(), ` INSERT INTO conversation_messages (conversation_id, author_id, content) VALUES ($1, $2, $3) RETURNING id, conversation_id, author_id, content, edited_at::text, created_at::text `, convID, userID, req.Content).Scan(&msg.ID, &msg.ConversationID, &msg.AuthorID, &msg.Content, &editedAt, &createdAt) if err != nil { http.Error(w, `{"error":"failed to send message"}`, http.StatusInternalServerError) return } err = h.db.QueryRowContext(r.Context(), `SELECT username, display_name FROM users WHERE id = $1`, userID).Scan(&msg.AuthorName, &displayName) if err != nil { http.Error(w, `{"error":"server error"}`, http.StatusInternalServerError) return } if editedAt.Valid { msg.EditedAt = &editedAt.String } msg.CreatedAt = createdAt.String if displayName.Valid { msg.DisplayName = &displayName.String } msg = h.attachReactions(r.Context(), []messageResponse{msg})[0] if h.hub != nil { h.hub.BroadcastToConversation(convID, gateway.Event{ Type: gateway.EventMessageCreate, Data: msg, }) } // Push notifications for DM participants if h.pushHandler != nil { go func() { members, err := h.MemberIDs(context.Background(), convID) if err != nil { return } payload := map[string]interface{}{ "title": msg.AuthorName, "body": req.Content, "url": "/conversations/" + convID, } for _, memberID := range members { if memberID != userID { h.pushHandler.SendPush(context.Background(), memberID, payload) } } }() } w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusCreated) json.NewEncoder(w).Encode(msg) } func (h *Handler) isMember(ctx context.Context, convID, userID string) bool { var exists bool err := h.db.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM conversation_members WHERE conversation_id = $1 AND user_id = $2)`, convID, userID).Scan(&exists) return err == nil && exists } // MemberIDs returns all user IDs in a conversation. func (h *Handler) MemberIDs(ctx context.Context, convID string) ([]string, error) { rows, err := h.db.QueryContext(ctx, `SELECT user_id FROM conversation_members WHERE conversation_id = $1`, convID) if err != nil { return nil, err } defer rows.Close() var ids []string for rows.Next() { var id string if err := rows.Scan(&id); err == nil { ids = append(ids, id) } } return ids, nil } type emojiGroup struct { Emoji string `json:"emoji"` Count int `json:"count"` Users []string `json:"users"` Details []reactionResponse `json:"details"` } func (h *Handler) attachReactions(ctx context.Context, messages []messageResponse) []messageResponse { if len(messages) == 0 { return messages } msgIDs := make([]string, len(messages)) msgMap := make(map[string]int) for i, msg := range messages { msgIDs[i] = msg.ID msgMap[msg.ID] = i messages[i].Reactions = make([]emojiGroup, 0) } placeholders := make([]string, len(msgIDs)) args := make([]interface{}, len(msgIDs)) for i, id := range msgIDs { placeholders[i] = fmt.Sprintf("$%d", i+1) args[i] = id } query := fmt.Sprintf(` SELECT id, message_id, user_id, emoji, created_at::text FROM conversation_reactions WHERE message_id IN (%s) ORDER BY created_at ASC `, strings.Join(placeholders, ", ")) rows, err := h.db.QueryContext(ctx, query, args...) if err != nil { return messages } defer rows.Close() reactionsMap := make(map[string]map[string]*emojiGroup) emojiOrderMap := make(map[string][]string) for rows.Next() { var reaction reactionResponse var createdAt sql.NullString if err := rows.Scan(&reaction.ID, &reaction.MessageID, &reaction.UserID, &reaction.Emoji, &createdAt); err != nil { continue } reaction.CreatedAt = createdAt.String msgID := reaction.MessageID if _, ok := reactionsMap[msgID]; !ok { reactionsMap[msgID] = make(map[string]*emojiGroup) emojiOrderMap[msgID] = make([]string, 0) } group, ok := reactionsMap[msgID][reaction.Emoji] if !ok { group = &emojiGroup{ Emoji: reaction.Emoji, Users: []string{}, Details: []reactionResponse{}, } reactionsMap[msgID][reaction.Emoji] = group emojiOrderMap[msgID] = append(emojiOrderMap[msgID], reaction.Emoji) } group.Count++ group.Users = append(group.Users, reaction.UserID) group.Details = append(group.Details, reaction) } for msgID, emojisMap := range reactionsMap { idx, ok := msgMap[msgID] if !ok { continue } order := emojiOrderMap[msgID] for _, emoji := range order { messages[idx].Reactions = append(messages[idx].Reactions, *emojisMap[emoji]) } } return messages } type reactionResponse struct { ID string `json:"id"` MessageID string `json:"message_id"` ConversationID string `json:"conversation_id"` UserID string `json:"user_id"` Emoji string `json:"emoji"` CreatedAt string `json:"created_at"` } // @Summary Add a reaction to a DM // @Router /conversations/{conversationID}/messages/{messageID}/reactions [post] func (h *Handler) AddReaction(w http.ResponseWriter, r *http.Request) { convID := chi.URLParam(r, "conversationID") messageID := chi.URLParam(r, "messageID") userID, ok := middleware.UserIDFromContext(r.Context()) if !ok || !h.isMember(r.Context(), convID, userID) { http.Error(w, `{"error":"unauthorized"}`, http.StatusUnauthorized) return } var req struct { Emoji string `json:"emoji"` } if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.Emoji == "" { http.Error(w, `{"error":"invalid request"}`, http.StatusBadRequest) return } var reaction reactionResponse reaction.ConversationID = convID var createdAt sql.NullString err := h.db.QueryRowContext(r.Context(), ` INSERT INTO conversation_reactions (message_id, user_id, emoji) VALUES ($1, $2, $3) ON CONFLICT (message_id, user_id, emoji) DO UPDATE SET emoji = EXCLUDED.emoji RETURNING id, message_id, user_id, emoji, created_at::text `, messageID, userID, req.Emoji).Scan( &reaction.ID, &reaction.MessageID, &reaction.UserID, &reaction.Emoji, &createdAt, ) if err != nil { h.logger.Error("failed to add dm reaction", "error", err) http.Error(w, `{"error":"server error"}`, http.StatusInternalServerError) return } reaction.CreatedAt = createdAt.String h.hub.BroadcastToConversation(convID, gateway.Event{ Type: gateway.EventReactionAdd, Data: map[string]interface{}{ "reaction": reaction, "conversation_id": convID, "message_id": messageID, }, }) w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusCreated) json.NewEncoder(w).Encode(reaction) } // @Summary Remove a reaction from a DM // @Router /conversations/{conversationID}/messages/{messageID}/reactions/{emoji} [delete] func (h *Handler) RemoveReaction(w http.ResponseWriter, r *http.Request) { convID := chi.URLParam(r, "conversationID") messageID := chi.URLParam(r, "messageID") emoji := chi.URLParam(r, "emoji") userID, ok := middleware.UserIDFromContext(r.Context()) if !ok || !h.isMember(r.Context(), convID, userID) { http.Error(w, `{"error":"unauthorized"}`, http.StatusUnauthorized) return } result, err := h.db.ExecContext(r.Context(), ` DELETE FROM conversation_reactions WHERE message_id = $1 AND user_id = $2 AND emoji = $3 `, messageID, userID, emoji) if err != nil { h.logger.Error("failed to remove dm reaction", "error", err) http.Error(w, `{"error":"server error"}`, http.StatusInternalServerError) return } rowsAffected, _ := result.RowsAffected() if rowsAffected > 0 { h.hub.BroadcastToConversation(convID, gateway.Event{ Type: gateway.EventReactionRemove, Data: map[string]interface{}{ "conversation_id": convID, "message_id": messageID, "user_id": userID, "emoji": emoji, }, }) } w.WriteHeader(http.StatusNoContent) }