Messenger/server/internal/store/conversations.go

package store

import (
	"context"
	"errors"
	"sort"

	"github.com/jackc/pgx/v5"
)

// CreateDM returns an existing direct-message conversation between two users or creates it.
func (s *Store) CreateDM(ctx context.Context, userA, userB int64) (*Conversation, error) {
	// Check existing DM
	var conv Conversation
	err := s.pool.QueryRow(ctx, `
       SELECT c.id, c.type, c.created_at
       FROM conversations c
       JOIN conversation_members m1 ON m1.conversation_id=c.id AND m1.user_id=$1
       JOIN conversation_members m2 ON m2.conversation_id=c.id AND m2.user_id=$2
       WHERE c.type='dm'
       LIMIT 1
   `, userA, userB).Scan(&conv.ID, &conv.Type, &conv.CreatedAt)
	if err == nil {
		return &conv, nil
	}
	if !errors.Is(err, pgx.ErrNoRows) {
		return nil, err
	}

	conn, err := s.pool.Acquire(ctx)
	if err != nil {
		return nil, err
	}
	defer conn.Release()

	tx, err := conn.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)

	err = tx.QueryRow(ctx, `
       INSERT INTO conversations(type) VALUES('dm')
       RETURNING id, type, created_at
   `).Scan(&conv.ID, &conv.Type, &conv.CreatedAt)
	if err != nil {
		return nil, err
	}
	// Add members
	members := []int64{userA, userB}
	for _, uid := range members {
		if _, err := tx.Exec(ctx, `
           INSERT INTO conversation_members(conversation_id, user_id)
           VALUES($1, $2)
       `, conv.ID, uid); err != nil {
			return nil, err
		}
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, err
	}
	return &conv, nil
}

// GetConversationByID retrieves a conversation by ID or returns ErrNotFound.
func (s *Store) GetConversationByID(ctx context.Context, id int64) (*Conversation, error) {
	var conv Conversation
	err := s.pool.QueryRow(ctx, `
       SELECT id, type, created_at FROM conversations WHERE id=$1
   `, id).Scan(&conv.ID, &conv.Type, &conv.CreatedAt)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	return &conv, nil
}

// IsMember reports whether a user is a member of a conversation.
func (s *Store) IsMember(ctx context.Context, conversationID, userID int64) (bool, error) {
	var exists bool
	err := s.pool.QueryRow(ctx, `
       SELECT EXISTS(
           SELECT 1 FROM conversation_members WHERE conversation_id=$1 AND user_id=$2
       )
   `, conversationID, userID).Scan(&exists)
	if err != nil {
		return false, err
	}
	return exists, nil
}

// ConversationIDsForUser returns all conversation IDs the user is a member of.
func (s *Store) ConversationIDsForUser(ctx context.Context, userID int64) ([]int64, error) {
	rows, err := s.pool.Query(ctx, `
       SELECT conversation_id FROM conversation_members WHERE user_id=$1
   `, userID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var ids []int64
	for rows.Next() {
		var id int64
		if err := rows.Scan(&id); err != nil {
			return nil, err
		}
		ids = append(ids, id)
	}
	sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
	return ids, nil
}

// ListConversationsForUser returns summaries of conversations for a user.
func (s *Store) ListConversationsForUser(ctx context.Context, userID int64) ([]ConversationSummary, error) {
	rows, err := s.pool.Query(ctx, `
       SELECT c.id, c.type, c.created_at,
           m.id, m.conversation_id, m.sender_id, m.body, m.created_at, m.edited_at, m.deleted_at,
           uc.unread_count
       FROM conversation_members cm
       JOIN conversations c ON c.id = cm.conversation_id
       LEFT JOIN LATERAL (
           SELECT id, conversation_id, sender_id, body, created_at, edited_at, deleted_at
           FROM messages WHERE conversation_id = cm.conversation_id
           ORDER BY id DESC LIMIT 1
       ) m ON true
       LEFT JOIN LATERAL (
           SELECT COUNT(*) AS unread_count
           FROM messages msg
           WHERE msg.conversation_id = cm.conversation_id
             AND msg.sender_id <> $1
             AND (cm.last_read_at IS NULL OR msg.created_at > cm.last_read_at)
       ) uc ON true
       WHERE cm.user_id = $1
       ORDER BY m.id DESC NULLS LAST
   `, userID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var list []ConversationSummary
	for rows.Next() {
		var cs ConversationSummary
		var msg Message
		if err := rows.Scan(
			&cs.ID, &cs.Type, &cs.CreatedAt,
			&msg.ID, &msg.ConversationID, &msg.SenderID, &msg.Body, &msg.CreatedAt, &msg.EditedAt, &msg.DeletedAt,
			&cs.UnreadCount,
		); err != nil {
			return nil, err
		}
		if msg.ID != 0 {
			cs.LastMessage = &msg
		}
		list = append(list, cs)
	}
	return list, nil
}

// UpdateLastRead sets the last_read_at to now for a user's conversation.
func (s *Store) UpdateLastRead(ctx context.Context, conversationID, userID int64) error {
	_, err := s.pool.Exec(ctx, `
       UPDATE conversation_members SET last_read_at = now()
       WHERE conversation_id=$1 AND user_id=$2
   `, conversationID, userID)
	return err
}