package dm import ( "context" "database/sql" "encoding/json" "errors" "log/slog" "net/http" "sort" "strconv" "git.dustin.coffee/hobokenchicken/dumpsterChat/internal/gateway" "git.dustin.coffee/hobokenchicken/dumpsterChat/internal/middleware" "github.com/go-chi/chi/v5" ) // Handler handles direct-message conversations. type Handler struct { db *sql.DB hub *gateway.Hub logger *slog.Logger } // NewHandler creates a new DM handler. func NewHandler(db *sql.DB, hub *gateway.Hub, logger *slog.Logger) *Handler { return &Handler{db: db, hub: hub, 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) }) } 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 } if len(req.UserIDs) == 0 { http.Error(w, `{"error":"at least one user_id is required"}`, http.StatusBadRequest) return } // Build unique member set including creator. memberSet := map[string]struct{}{userID: {}} for _, uid := range req.UserIDs { if uid != "" { memberSet[uid] = struct{}{} } } if len(memberSet) == 1 { http.Error(w, `{"error":"cannot create conversation with yourself"}`, http.StatusBadRequest) return } 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) == 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, 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, 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"` } // 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 ASC 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 ASC 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) } 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 } if h.hub != nil { h.hub.BroadcastToConversation(convID, gateway.Event{ Type: gateway.EventMessageCreate, Data: msg, }) } 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 }