package store import ( "context" "errors" "sort" "time" "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 ( msgID *int64 msgConvID *int64 msgSender *int64 msgBody *string msgCreated *time.Time msgEdited *time.Time msgDeleted *time.Time ) if err := rows.Scan( &cs.ID, &cs.Type, &cs.CreatedAt, &msgID, &msgConvID, &msgSender, &msgBody, &msgCreated, &msgEdited, &msgDeleted, &cs.UnreadCount, ); err != nil { return nil, err } if msgID != nil { cs.LastMessage = &Message{ ID: *msgID, ConversationID: *msgConvID, SenderID: *msgSender, Body: *msgBody, CreatedAt: *msgCreated, EditedAt: msgEdited, DeletedAt: msgDeleted, } } 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 }