package store import ( "context" "errors" "fmt" "time" "go.mongodb.org/mongo-driver/v2/bson" "go.mongodb.org/mongo-driver/v2/mongo" "go.mongodb.org/mongo-driver/v2/mongo/options" "bountyboard/internal/ulid" ) // Conversation per §4.6. type Conversation struct { ID string `bson:"_id" json:"id"` Kind string `bson:"kind" json:"kind"` // dm | group | project CustomerID string `bson:"customerId,omitempty" json:"customerId,omitempty"` Title string `bson:"title,omitempty" json:"title,omitempty"` CreatorID string `bson:"creatorId,omitempty" json:"creatorId,omitempty"` ParticipantIDs []string `bson:"participantIds" json:"participantIds"` LastMessageAt time.Time `bson:"lastMessageAt" json:"lastMessageAt"` CreatedAt time.Time `bson:"createdAt" json:"createdAt"` } // MessageAttachment per §4.6. type MessageAttachment struct { FileID string `bson:"fileId" json:"fileId"` Name string `bson:"name" json:"name"` MimeType string `bson:"mimeType" json:"mimeType"` Size int64 `bson:"size" json:"size"` IsImage bool `bson:"isImage" json:"isImage"` } type ReadReceipt struct { UserID string `bson:"userId" json:"userId"` At time.Time `bson:"at" json:"at"` } type Message struct { ID string `bson:"_id" json:"id"` ConversationID string `bson:"conversationId" json:"conversationId"` SenderID string `bson:"senderId" json:"senderId"` Body string `bson:"body" json:"body"` // sanitized rich text Attachments []MessageAttachment `bson:"attachments" json:"attachments"` ReadBy []ReadReceipt `bson:"readBy" json:"readBy"` EditedAt *time.Time `bson:"editedAt" json:"editedAt"` DeletedAt *time.Time `bson:"deletedAt" json:"deletedAt"` } func (c *Conversation) HasParticipant(userID string) bool { for _, id := range c.ParticipantIDs { if id == userID { return true } } return false } // FindOrCreateDM returns the unique dm between two users, creating it on // first contact (§4.6: participantIds is the dm unique key). func (s *Store) FindOrCreateDM(ctx context.Context, a, b string) (*Conversation, error) { var conv Conversation err := s.DB.Collection("conversations").FindOne(ctx, bson.M{ "kind": "dm", "participantIds": bson.M{"$all": []string{a, b}, "$size": 2}, }).Decode(&conv) if err == nil { return &conv, nil } if !errors.Is(err, mongo.ErrNoDocuments) { return nil, fmt.Errorf("find dm: %w", err) } conv = Conversation{ ID: ulid.New(), Kind: "dm", ParticipantIDs: []string{a, b}, LastMessageAt: time.Now().UTC(), CreatedAt: time.Now().UTC(), } if _, err := s.DB.Collection("conversations").InsertOne(ctx, conv); err != nil { return nil, fmt.Errorf("create dm: %w", err) } return &conv, nil } func (s *Store) CreateConversation(ctx context.Context, c *Conversation) error { c.ID = ulid.New() c.CreatedAt = time.Now().UTC() c.LastMessageAt = c.CreatedAt if _, err := s.DB.Collection("conversations").InsertOne(ctx, c); err != nil { return fmt.Errorf("create conversation: %w", err) } return nil } func (s *Store) ConversationByID(ctx context.Context, id string) (*Conversation, error) { var c Conversation err := s.DB.Collection("conversations").FindOne(ctx, bson.M{"_id": id}).Decode(&c) if errors.Is(err, mongo.ErrNoDocuments) { return nil, ErrNotFound } if err != nil { return nil, fmt.Errorf("find conversation: %w", err) } return &c, nil } // AddParticipant adds a user to a conversation (idempotent via $addToSet). func (s *Store) AddParticipant(ctx context.Context, convID, userID string) error { _, err := s.DB.Collection("conversations").UpdateOne(ctx, bson.M{"_id": convID}, bson.M{"$addToSet": bson.M{"participantIds": userID}}) if err != nil { return fmt.Errorf("add participant: %w", err) } return nil } // RemoveParticipant removes a user from a conversation. func (s *Store) RemoveParticipant(ctx context.Context, convID, userID string) error { _, err := s.DB.Collection("conversations").UpdateOne(ctx, bson.M{"_id": convID}, bson.M{"$pull": bson.M{"participantIds": userID}}) if err != nil { return fmt.Errorf("remove participant: %w", err) } return nil } func (s *Store) ListConversations(ctx context.Context, userID string) ([]Conversation, error) { cur, err := s.DB.Collection("conversations").Find(ctx, bson.M{"participantIds": userID}, options.Find().SetSort(bson.D{{Key: "lastMessageAt", Value: -1}}).SetLimit(100)) if err != nil { return nil, fmt.Errorf("list conversations: %w", err) } out := []Conversation{} if err := cur.All(ctx, &out); err != nil { return nil, fmt.Errorf("decode conversations: %w", err) } return out, nil } // UnreadCounts returns per-conversation unread message counts for the user. func (s *Store) UnreadCounts(ctx context.Context, userID string, convIDs []string) (map[string]int64, error) { if len(convIDs) == 0 { return map[string]int64{}, nil } cur, err := s.DB.Collection("messages").Aggregate(ctx, mongo.Pipeline{ {{Key: "$match", Value: bson.M{ "conversationId": bson.M{"$in": convIDs}, "senderId": bson.M{"$ne": userID}, "deletedAt": nil, "readBy.userId": bson.M{"$ne": userID}, }}}, {{Key: "$group", Value: bson.M{"_id": "$conversationId", "n": bson.M{"$sum": 1}}}}, }) if err != nil { return nil, fmt.Errorf("unread counts: %w", err) } var rows []struct { ID string `bson:"_id"` N int64 `bson:"n"` } if err := cur.All(ctx, &rows); err != nil { return nil, fmt.Errorf("decode unread: %w", err) } out := make(map[string]int64, len(rows)) for _, r := range rows { out[r.ID] = r.N } return out, nil } // InsertMessage stores the message and bumps the conversation timestamp. func (s *Store) InsertMessage(ctx context.Context, m *Message) error { m.ID = ulid.New() if m.Attachments == nil { m.Attachments = []MessageAttachment{} } m.ReadBy = []ReadReceipt{{UserID: m.SenderID, At: time.Now().UTC()}} if _, err := s.DB.Collection("messages").InsertOne(ctx, m); err != nil { return fmt.Errorf("insert message: %w", err) } _, err := s.DB.Collection("conversations").UpdateOne(ctx, bson.M{"_id": m.ConversationID}, bson.M{"$set": bson.M{"lastMessageAt": time.Now().UTC()}}) if err != nil { return fmt.Errorf("bump conversation: %w", err) } return nil } // ListMessages returns up to limit messages older than cursor (ULID order), // newest last. func (s *Store) ListMessages(ctx context.Context, convID, cursor string, limit int) ([]Message, error) { if limit <= 0 || limit > 100 { limit = 50 } q := bson.M{"conversationId": convID} if cursor != "" { q["_id"] = bson.M{"$lt": cursor} } cur, err := s.DB.Collection("messages").Find(ctx, q, options.Find().SetSort(bson.D{{Key: "_id", Value: -1}}).SetLimit(int64(limit))) if err != nil { return nil, fmt.Errorf("list messages: %w", err) } out := []Message{} if err := cur.All(ctx, &out); err != nil { return nil, fmt.Errorf("decode messages: %w", err) } // reverse to chronological order for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 { out[i], out[j] = out[j], out[i] } return out, nil } // MarkConversationRead adds a read receipt to every unread message. func (s *Store) MarkConversationRead(ctx context.Context, convID, userID string) (int64, error) { res, err := s.DB.Collection("messages").UpdateMany(ctx, bson.M{ "conversationId": convID, "readBy.userId": bson.M{"$ne": userID}, }, bson.M{"$push": bson.M{"readBy": ReadReceipt{UserID: userID, At: time.Now().UTC()}}}) if err != nil { return 0, fmt.Errorf("mark read: %w", err) } return res.ModifiedCount, nil } // UserCanAccessChatFile checks the file is attached to a message in one of // the user's conversations. func (s *Store) UserCanAccessChatFile(ctx context.Context, userID, fileID string) (bool, error) { var msg Message err := s.DB.Collection("messages").FindOne(ctx, bson.M{"attachments.fileId": fileID}).Decode(&msg) if errors.Is(err, mongo.ErrNoDocuments) { return false, nil } if err != nil { return false, fmt.Errorf("find file message: %w", err) } conv, err := s.ConversationByID(ctx, msg.ConversationID) if err != nil { return false, err } return conv.HasParticipant(userID), nil }