package repository import ( "context" "encoding/json" "errors" "fmt" "time" "github.com/jackc/pgx/v5" "github.com/prabhupadasays/spsays-bff/internal/model" ) var ( // ErrNotParticipant is the authorization failure for every chat read and // write: if you are not in the thread, it does not exist for you. ErrNotParticipant = errors.New("not a participant in this conversation") ErrConversationGone = errors.New("conversation not found") // ErrReplyTargetNotHere guards against replying to a message from another // thread. The foreign key only proves the target exists somewhere, so // without this check a client could quote a conversation it was never in. ErrReplyTargetNotHere = errors.New("cannot reply to a message from another conversation") ErrAttachmentGone = errors.New("attachment not found") ErrMessageNotHere = errors.New("message is not in this conversation") ErrUserGone = errors.New("user not found") ) type ChatRepo struct{ pg *Postgres } func NewChatRepo(pg *Postgres) *ChatRepo { return &ChatRepo{pg: pg} } // dmKey builds the sorted pair key that makes "one DM thread per pair" a // database guarantee rather than a service-layer race. func dmKey(a, b string) string { if a < b { return a + ":" + b } return b + ":" + a } // EnsureDM returns the DM between two users, creating it if absent. // // Idempotent under concurrency: two people tapping "message" simultaneously // race on the partial unique index, and the loser falls through to the SELECT // rather than erroring or creating a second thread. func (r *ChatRepo) EnsureDM(ctx context.Context, userA, userB string) (*model.Conversation, error) { tx, err := r.pg.Pool.Begin(ctx) if err != nil { return nil, fmt.Errorf("begin dm tx: %w", err) } defer tx.Rollback(ctx) //nolint:errcheck // no-op once committed key := dmKey(userA, userB) var c model.Conversation err = tx.QueryRow(ctx, ` INSERT INTO conversations (kind, dm_key, created_by) VALUES ('dm', $1, $2) ON CONFLICT (dm_key) WHERE kind = 'dm' DO NOTHING RETURNING id, kind, title, avatar_key, send_policy, last_seq, last_message_at, created_at`, key, userA, ).Scan(&c.ID, &c.Kind, &c.Title, &c.AvatarKey, &c.SendPolicy, &c.LastSeq, &c.LastMessageAt, &c.CreatedAt) if errors.Is(err, pgx.ErrNoRows) { // Already existed — read it back. err = tx.QueryRow(ctx, ` SELECT id, kind, title, avatar_key, send_policy, last_seq, last_message_at, created_at FROM conversations WHERE dm_key = $1 AND kind = 'dm'`, key, ).Scan(&c.ID, &c.Kind, &c.Title, &c.AvatarKey, &c.SendPolicy, &c.LastSeq, &c.LastMessageAt, &c.CreatedAt) } if err != nil { return nil, fmt.Errorf("ensure dm: %w", err) } // Participants are inserted unconditionally: a thread created moments ago // by the other party may not list us yet. if _, err := tx.Exec(ctx, ` INSERT INTO conversation_participants (conversation_id, user_id) VALUES ($1, $2), ($1, $3) ON CONFLICT (conversation_id, user_id) DO NOTHING`, c.ID, userA, userB, ); err != nil { return nil, fmt.Errorf("ensure dm participants: %w", err) } if err := tx.Commit(ctx); err != nil { return nil, fmt.Errorf("commit dm: %w", err) } return &c, nil } // CreateGroup opens a group thread with creator as owner and members joined. func (r *ChatRepo) CreateGroup(ctx context.Context, creatorID, title string, memberIDs []string) (*model.Conversation, error) { tx, err := r.pg.Pool.Begin(ctx) if err != nil { return nil, fmt.Errorf("begin group tx: %w", err) } defer tx.Rollback(ctx) //nolint:errcheck var c model.Conversation if err := tx.QueryRow(ctx, ` INSERT INTO conversations (kind, title, created_by) VALUES ('group', $1, $2) RETURNING id, kind, title, avatar_key, send_policy, last_seq, last_message_at, created_at`, title, creatorID, ).Scan(&c.ID, &c.Kind, &c.Title, &c.AvatarKey, &c.SendPolicy, &c.LastSeq, &c.LastMessageAt, &c.CreatedAt); err != nil { return nil, fmt.Errorf("create group: %w", err) } if _, err := tx.Exec(ctx, ` INSERT INTO conversation_participants (conversation_id, user_id, role) VALUES ($1, $2, 'owner') ON CONFLICT DO NOTHING`, c.ID, creatorID, ); err != nil { return nil, fmt.Errorf("add group owner: %w", err) } for _, m := range memberIDs { if m == creatorID { continue } if _, err := tx.Exec(ctx, ` INSERT INTO conversation_participants (conversation_id, user_id) VALUES ($1, $2) ON CONFLICT DO NOTHING`, c.ID, m, ); err != nil { return nil, fmt.Errorf("add group member: %w", err) } } if err := tx.Commit(ctx); err != nil { return nil, fmt.Errorf("commit group: %w", err) } return &c, nil } // EnsureAIThread returns the caller's Kautilya thread, creating it on first // use. One participant; the assistant has no user row. func (r *ChatRepo) EnsureAIThread(ctx context.Context, userID string) (*model.Conversation, error) { var c model.Conversation err := r.pg.Pool.QueryRow(ctx, ` SELECT c.id, c.kind, c.title, c.avatar_key, c.send_policy, c.last_seq, c.last_message_at, c.created_at FROM conversations c JOIN conversation_participants p ON p.conversation_id = c.id AND p.user_id = $1 WHERE c.kind = 'ai' ORDER BY c.created_at LIMIT 1`, userID, ).Scan(&c.ID, &c.Kind, &c.Title, &c.AvatarKey, &c.SendPolicy, &c.LastSeq, &c.LastMessageAt, &c.CreatedAt) if err == nil { return &c, nil } if !errors.Is(err, pgx.ErrNoRows) { return nil, fmt.Errorf("find ai thread: %w", err) } tx, err := r.pg.Pool.Begin(ctx) if err != nil { return nil, fmt.Errorf("begin ai tx: %w", err) } defer tx.Rollback(ctx) //nolint:errcheck if err := tx.QueryRow(ctx, ` INSERT INTO conversations (kind, created_by) VALUES ('ai', $1) RETURNING id, kind, title, avatar_key, send_policy, last_seq, last_message_at, created_at`, userID, ).Scan(&c.ID, &c.Kind, &c.Title, &c.AvatarKey, &c.SendPolicy, &c.LastSeq, &c.LastMessageAt, &c.CreatedAt); err != nil { return nil, fmt.Errorf("create ai thread: %w", err) } if _, err := tx.Exec(ctx, ` INSERT INTO conversation_participants (conversation_id, user_id) VALUES ($1, $2)`, c.ID, userID, ); err != nil { return nil, fmt.Errorf("add ai participant: %w", err) } if err := tx.Commit(ctx); err != nil { return nil, fmt.Errorf("commit ai thread: %w", err) } return &c, nil } // Conversation loads a thread by id, without any membership check — callers // pair it with Participant when the caller's access matters. func (r *ChatRepo) Conversation(ctx context.Context, id string) (*model.Conversation, error) { var c model.Conversation err := r.pg.Pool.QueryRow(ctx, ` SELECT id, kind, title, avatar_key, send_policy, last_seq, last_message_at, created_at FROM conversations WHERE id = $1`, id, ).Scan(&c.ID, &c.Kind, &c.Title, &c.AvatarKey, &c.SendPolicy, &c.LastSeq, &c.LastMessageAt, &c.CreatedAt) if errors.Is(err, pgx.ErrNoRows) { return nil, ErrConversationGone } if err != nil { return nil, fmt.Errorf("load conversation: %w", err) } return &c, nil } // Participant loads the caller's membership row — the authorization primitive // behind every other chat call and the realtime channel rule. func (r *ChatRepo) Participant(ctx context.Context, conversationID, userID string) (*model.Participant, error) { var p model.Participant err := r.pg.Pool.QueryRow(ctx, ` SELECT conversation_id, user_id, role, last_read_seq, last_delivered_seq, muted_until, joined_at, left_at FROM conversation_participants WHERE conversation_id = $1 AND user_id = $2 AND left_at IS NULL`, conversationID, userID, ).Scan(&p.ConversationID, &p.UserID, &p.Role, &p.LastReadSeq, &p.LastDeliveredSeq, &p.MutedUntil, &p.JoinedAt, &p.LeftAt) if errors.Is(err, pgx.ErrNoRows) { return nil, ErrNotParticipant } if err != nil { return nil, fmt.Errorf("load participant: %w", err) } return &p, nil } // Participants lists everyone still in the thread with the details needed to // render a name and avatar beside their messages. func (r *ChatRepo) Participants(ctx context.Context, conversationID string) ([]model.ChatPeer, error) { rows, err := r.pg.Pool.Query(ctx, ` SELECT u.id, u.handle, u.name, u.avatar_url, p.role, p.joined_at FROM conversation_participants p JOIN users u ON u.id = p.user_id WHERE p.conversation_id = $1 AND p.left_at IS NULL -- Owner first, then admins, then members by seniority: the order the -- member list is read in. ORDER BY CASE p.role WHEN 'owner' THEN 0 WHEN 'admin' THEN 1 ELSE 2 END, p.joined_at`, conversationID) if err != nil { return nil, fmt.Errorf("query participants: %w", err) } defer rows.Close() people := []model.ChatPeer{} for rows.Next() { var c model.ChatPeer if err := rows.Scan(&c.ID, &c.Handle, &c.Name, &c.AvatarURL, &c.Role, &c.JoinedAt); err != nil { return nil, fmt.Errorf("scan participant: %w", err) } people = append(people, c) } return people, rows.Err() } // ParticipantIDs lists everyone still in the thread — the fan-out list for // realtime publishes and push. func (r *ChatRepo) ParticipantIDs(ctx context.Context, conversationID string) ([]string, error) { rows, err := r.pg.Pool.Query(ctx, ` SELECT user_id FROM conversation_participants WHERE conversation_id = $1 AND left_at IS NULL`, conversationID) if err != nil { return nil, fmt.Errorf("participant ids: %w", err) } defer rows.Close() var ids []string for rows.Next() { var id string if err := rows.Scan(&id); err != nil { return nil, fmt.Errorf("scan participant id: %w", err) } ids = append(ids, id) } return ids, rows.Err() } // Inbox lists the caller's threads of the given kinds, most recent first. // // The last-message line comes from a LATERAL on (conversation_id, seq DESC) // rather than a denormalized column, so an edit or delete is reflected without // a second write path that could drift. Messages the caller deleted for // themselves are skipped, so their inbox preview matches their thread. func (r *ChatRepo) Inbox(ctx context.Context, userID string, kinds []string, limit int) ([]model.InboxItem, error) { rows, err := r.pg.Pool.Query(ctx, ` SELECT c.id, c.kind, c.title, c.avatar_key, c.send_policy, c.last_seq, c.last_message_at, c.created_at, GREATEST(c.last_seq - p.last_read_seq, 0) AS unread, p.muted_until, m.seq, m.sender_id, m.kind, m.body, m.deleted_at, COALESCE(su.name, ''), peer.id, peer.handle, peer.name, peer.avatar_url FROM conversation_participants p JOIN conversations c ON c.id = p.conversation_id LEFT JOIN LATERAL ( SELECT msg.seq, msg.sender_id, msg.kind, msg.body, msg.deleted_at FROM messages msg WHERE msg.conversation_id = c.id AND NOT EXISTS (SELECT 1 FROM message_hides h WHERE h.message_id = msg.id AND h.user_id = $1) ORDER BY msg.seq DESC LIMIT 1 ) m ON TRUE LEFT JOIN users su ON su.id = m.sender_id LEFT JOIN LATERAL ( SELECT u2.id, u2.handle, u2.name, u2.avatar_url FROM conversation_participants p2 JOIN users u2 ON u2.id = p2.user_id WHERE p2.conversation_id = c.id AND p2.user_id <> $1 LIMIT 1 ) peer ON c.kind = 'dm' WHERE p.user_id = $1 AND p.left_at IS NULL AND c.kind = ANY($2) ORDER BY c.last_message_at DESC NULLS LAST, c.created_at DESC LIMIT $3`, userID, kinds, limit) if err != nil { return nil, fmt.Errorf("query inbox: %w", err) } defer rows.Close() items := []model.InboxItem{} for rows.Next() { var it model.InboxItem var seq *int64 var senderID, msgKind, body *string var deletedAt *time.Time var senderName string var peerID, peerHandle, peerName, peerAvatar *string if err := rows.Scan( &it.ID, &it.Kind, &it.Title, &it.AvatarKey, &it.SendPolicy, &it.LastSeq, &it.LastMessageAt, &it.CreatedAt, &it.Unread, &it.MutedUntil, &seq, &senderID, &msgKind, &body, &deletedAt, &senderName, &peerID, &peerHandle, &peerName, &peerAvatar, ); err != nil { return nil, fmt.Errorf("scan inbox row: %w", err) } if seq != nil { it.LastMessage = &model.MessagePreview{ Seq: *seq, SenderID: senderID, SenderName: senderName, Kind: derefOr(msgKind, model.MessageText), Body: derefOr(body, ""), DeletedAt: deletedAt, } } if peerID != nil { it.Peer = &model.ChatPeer{ ID: *peerID, Handle: derefOr(peerHandle, ""), Name: derefOr(peerName, ""), AvatarURL: peerAvatar, } } items = append(items, it) } return items, rows.Err() } // Messages returns one page of history, newest first. beforeSeq == 0 starts at // the newest message; otherwise it is an exclusive upper bound, which makes // paging stable even as new messages arrive at the top. func (r *ChatRepo) Messages(ctx context.Context, conversationID, userID string, beforeSeq int64, limit int) ([]model.Message, error) { rows, err := r.pg.Pool.Query(ctx, ` SELECT m.id, m.conversation_id, m.seq, m.sender_id, m.kind, m.body, m.entities, m.client_msg_id, m.reply_to_id, m.forwarded_from_id, m.edited_at, m.deleted_at, m.created_at, rep.id, rep.seq, rep.sender_id, rep.body, rep.deleted_at, COALESCE(rep.sender_name, '') FROM messages m LEFT JOIN LATERAL ( SELECT r.id, r.seq, r.sender_id, r.body, r.deleted_at, ru.name AS sender_name FROM messages r LEFT JOIN users ru ON ru.id = r.sender_id -- The conversation guard is repeated here on purpose: even a row -- written before the write-side check existed cannot leak another -- thread's text through a quote. WHERE r.id = m.reply_to_id AND r.conversation_id = m.conversation_id ) rep ON TRUE WHERE m.conversation_id = $1 AND ($2 = 0 OR m.seq < $2) AND NOT EXISTS (SELECT 1 FROM message_hides h WHERE h.message_id = m.id AND h.user_id = $3) ORDER BY m.seq DESC LIMIT $4`, conversationID, beforeSeq, userID, limit) if err != nil { return nil, fmt.Errorf("query messages: %w", err) } defer rows.Close() msgs := []model.Message{} for rows.Next() { var m model.Message var entities []byte var repID *string var repSeq *int64 var repSender *string var repBody *string var repDeleted *time.Time var repSenderName string if err := rows.Scan(&m.ID, &m.ConversationID, &m.Seq, &m.SenderID, &m.Kind, &m.Body, &entities, &m.ClientMsgID, &m.ReplyToID, &m.ForwardedFromID, &m.EditedAt, &m.DeletedAt, &m.CreatedAt, &repID, &repSeq, &repSender, &repBody, &repDeleted, &repSenderName); err != nil { return nil, fmt.Errorf("scan message: %w", err) } m.Entities = json.RawMessage(entities) if repID != nil { m.ReplyTo = &model.ReplyPreview{ ID: *repID, Seq: derefInt64(repSeq), SenderID: repSender, SenderName: repSenderName, Body: derefOr(repBody, ""), Deleted: repDeleted != nil, } } msgs = append(msgs, m) } return msgs, rows.Err() } // AppendParams carries everything needed to persist one outgoing message. type AppendParams struct { ConversationID string SenderID *string Kind string Body string Entities json.RawMessage ClientMsgID *string ReplyToID *string ForwardedFromID *string } // Append persists a message and assigns its sequence number. // // The conversation row is locked FOR UPDATE first. That serialises sends within // a thread — which is what makes `seq` gapless — and it also means the // duplicate check below runs under the lock, so a retried send returns the // original message instead of consuming a sequence number and leaving a hole. func (r *ChatRepo) Append(ctx context.Context, p AppendParams) (*model.Message, bool, error) { tx, err := r.pg.Pool.Begin(ctx) if err != nil { return nil, false, fmt.Errorf("begin append tx: %w", err) } defer tx.Rollback(ctx) //nolint:errcheck var lastSeq int64 if err := tx.QueryRow(ctx, `SELECT last_seq FROM conversations WHERE id = $1 FOR UPDATE`, p.ConversationID, ).Scan(&lastSeq); err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, false, ErrConversationGone } return nil, false, fmt.Errorf("lock conversation: %w", err) } if p.ClientMsgID != nil { existing, err := scanMessage(tx.QueryRow(ctx, ` SELECT id, conversation_id, seq, sender_id, kind, body, entities, client_msg_id, reply_to_id, forwarded_from_id, edited_at, deleted_at, created_at FROM messages WHERE conversation_id = $1 AND client_msg_id = $2`, p.ConversationID, *p.ClientMsgID)) if err == nil { return existing, false, tx.Commit(ctx) } if !errors.Is(err, pgx.ErrNoRows) { return nil, false, fmt.Errorf("dedup lookup: %w", err) } } // Checked under the conversation lock we already hold. if p.ReplyToID != nil { var sameThread bool if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM messages WHERE id = $1 AND conversation_id = $2)`, *p.ReplyToID, p.ConversationID, ).Scan(&sameThread); err != nil { return nil, false, fmt.Errorf("validate reply target: %w", err) } if !sameThread { return nil, false, ErrReplyTargetNotHere } } entities := p.Entities if len(entities) == 0 { entities = json.RawMessage(`[]`) } seq := lastSeq + 1 m, err := scanMessage(tx.QueryRow(ctx, ` INSERT INTO messages (conversation_id, seq, sender_id, kind, body, entities, client_msg_id, reply_to_id, forwarded_from_id) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9) RETURNING id, conversation_id, seq, sender_id, kind, body, entities, client_msg_id, reply_to_id, forwarded_from_id, edited_at, deleted_at, created_at`, p.ConversationID, seq, p.SenderID, p.Kind, p.Body, entities, p.ClientMsgID, p.ReplyToID, p.ForwardedFromID)) if err != nil { return nil, false, fmt.Errorf("insert message: %w", err) } if _, err := tx.Exec(ctx, ` UPDATE conversations SET last_seq = $2, last_message_at = now(), updated_at = now() WHERE id = $1`, p.ConversationID, seq); err != nil { return nil, false, fmt.Errorf("bump conversation: %w", err) } // The sender has trivially seen their own message; advancing their cursors // here keeps their unread count at zero without a follow-up write. if p.SenderID != nil { if _, err := tx.Exec(ctx, ` UPDATE conversation_participants SET last_read_seq = GREATEST(last_read_seq, $3), last_delivered_seq = GREATEST(last_delivered_seq, $3) WHERE conversation_id = $1 AND user_id = $2`, p.ConversationID, *p.SenderID, seq); err != nil { return nil, false, fmt.Errorf("advance sender cursor: %w", err) } } if err := tx.Commit(ctx); err != nil { return nil, false, fmt.Errorf("commit append: %w", err) } return m, true, nil } // MarkRead advances the caller's read cursor. GREATEST means an out-of-order // or replayed request can never move a cursor backwards and resurrect unreads. // Reading implies delivery, so both cursors move together. func (r *ChatRepo) MarkRead(ctx context.Context, conversationID, userID string, seq int64) error { _, err := r.pg.Pool.Exec(ctx, ` UPDATE conversation_participants SET last_read_seq = GREATEST(last_read_seq, $3), last_delivered_seq = GREATEST(last_delivered_seq, $3) WHERE conversation_id = $1 AND user_id = $2`, conversationID, userID, seq) if err != nil { return fmt.Errorf("mark read: %w", err) } return nil } // MarkDelivered advances only the delivery cursor — the client received the // message but the user has not opened the thread. func (r *ChatRepo) MarkDelivered(ctx context.Context, conversationID, userID string, seq int64) error { _, err := r.pg.Pool.Exec(ctx, ` UPDATE conversation_participants SET last_delivered_seq = GREATEST(last_delivered_seq, $3) WHERE conversation_id = $1 AND user_id = $2`, conversationID, userID, seq) if err != nil { return fmt.Errorf("mark delivered: %w", err) } return nil } // AddAttachmentParams describes one attachment to persist. type AddAttachmentParams struct { MessageID string ConversationID string Ord int Kind string R2Key *string Mime *string SizeBytes *int64 Width *int Height *int DurationMs *int Blurhash *string FileName *string URL *string Title *string Description *string SiteName *string ImageURL *string // Shared entity: the durable reference plus a snapshot used only as a // fallback. Live data is re-resolved per viewer at read time. EntityType *string EntityID *string Snapshot []byte } // AddAttachment persists one attachment. conversation_id is denormalized here // so the chat profile's media and links tabs are a single index scan. func (r *ChatRepo) AddAttachment(ctx context.Context, p AddAttachmentParams) (*model.Attachment, error) { var a model.Attachment err := r.pg.Pool.QueryRow(ctx, ` INSERT INTO message_attachments (message_id, conversation_id, ord, kind, r2_key, mime, size_bytes, width, height, duration_ms, blurhash, file_name, url, title, description, site_name, image_url, entity_type, entity_id, snapshot) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20) RETURNING id, message_id, ord, kind, mime, size_bytes, width, height, duration_ms, blurhash, file_name, url, title, description, site_name, image_url, entity_type, entity_id, snapshot, created_at`, p.MessageID, p.ConversationID, p.Ord, p.Kind, p.R2Key, p.Mime, p.SizeBytes, p.Width, p.Height, p.DurationMs, p.Blurhash, p.FileName, p.URL, p.Title, p.Description, p.SiteName, p.ImageURL, p.EntityType, p.EntityID, p.Snapshot, ).Scan(&a.ID, &a.MessageID, &a.Ord, &a.Kind, &a.Mime, &a.SizeBytes, &a.Width, &a.Height, &a.DurationMs, &a.Blurhash, &a.FileName, &a.URL, &a.Title, &a.Description, &a.SiteName, &a.ImageURL, &a.EntityType, &a.EntityID, &a.Entity, &a.CreatedAt) if err != nil { return nil, fmt.Errorf("add attachment: %w", err) } return &a, nil } // AttachmentsFor loads the attachments of the given messages, keyed by message. func (r *ChatRepo) AttachmentsFor(ctx context.Context, messageIDs []string) (map[string][]model.Attachment, error) { out := map[string][]model.Attachment{} if len(messageIDs) == 0 { return out, nil } rows, err := r.pg.Pool.Query(ctx, ` SELECT id, message_id, ord, kind, mime, size_bytes, width, height, duration_ms, blurhash, file_name, url, title, description, site_name, image_url, entity_type, entity_id, snapshot, created_at FROM message_attachments WHERE message_id = ANY($1) ORDER BY message_id, ord`, messageIDs) if err != nil { return nil, fmt.Errorf("query attachments: %w", err) } defer rows.Close() for rows.Next() { var a model.Attachment if err := rows.Scan(&a.ID, &a.MessageID, &a.Ord, &a.Kind, &a.Mime, &a.SizeBytes, &a.Width, &a.Height, &a.DurationMs, &a.Blurhash, &a.FileName, &a.URL, &a.Title, &a.Description, &a.SiteName, &a.ImageURL, &a.EntityType, &a.EntityID, &a.Entity, &a.CreatedAt); err != nil { return nil, fmt.Errorf("scan attachment: %w", err) } out[a.MessageID] = append(out[a.MessageID], a) } return out, rows.Err() } // AttachmentLocation returns an attachment's storage key and the conversation // it belongs to, so a download can be authorized before a URL is minted. func (r *ChatRepo) AttachmentLocation(ctx context.Context, attachmentID string) (conversationID, r2Key string, err error) { var key *string err = r.pg.Pool.QueryRow(ctx, ` SELECT conversation_id, r2_key FROM message_attachments WHERE id = $1`, attachmentID).Scan(&conversationID, &key) if errors.Is(err, pgx.ErrNoRows) { return "", "", ErrAttachmentGone } if err != nil { return "", "", fmt.Errorf("attachment location: %w", err) } if key == nil || *key == "" { return "", "", ErrAttachmentGone } return conversationID, *key, nil } // GalleryKindsMedia / GalleryKindsLink name the two chat-profile tabs. var ( GalleryKindsMedia = []string{model.AttachmentImage, model.AttachmentVideo, model.AttachmentAudio, model.AttachmentFile} GalleryKindsLink = []string{model.AttachmentLink} ) // Gallery lists a thread's attachments of the given kinds, newest first. // // Served by message_attachments_gallery_idx, which is why conversation_id is // denormalized onto attachments: this is one index range scan rather than a // join back through messages. // // Attachments of messages the caller deleted for themselves are excluded, so // the gallery matches the thread they actually see. func (r *ChatRepo) Gallery(ctx context.Context, conversationID, userID string, kinds []string, before *time.Time, limit int) ([]model.Attachment, error) { rows, err := r.pg.Pool.Query(ctx, ` SELECT a.id, a.message_id, a.ord, a.kind, a.mime, a.size_bytes, a.width, a.height, a.duration_ms, a.blurhash, a.file_name, a.url, a.title, a.description, a.site_name, a.image_url, a.entity_type, a.entity_id, a.snapshot, a.created_at FROM message_attachments a JOIN messages m ON m.id = a.message_id WHERE a.conversation_id = $1 AND a.kind = ANY($2) AND m.deleted_at IS NULL AND ($3::timestamptz IS NULL OR a.created_at < $3) AND NOT EXISTS (SELECT 1 FROM message_hides h WHERE h.message_id = a.message_id AND h.user_id = $4) ORDER BY a.created_at DESC LIMIT $5`, conversationID, kinds, before, userID, limit) if err != nil { return nil, fmt.Errorf("query gallery: %w", err) } defer rows.Close() out := []model.Attachment{} for rows.Next() { var a model.Attachment if err := rows.Scan(&a.ID, &a.MessageID, &a.Ord, &a.Kind, &a.Mime, &a.SizeBytes, &a.Width, &a.Height, &a.DurationMs, &a.Blurhash, &a.FileName, &a.URL, &a.Title, &a.Description, &a.SiteName, &a.ImageURL, &a.EntityType, &a.EntityID, &a.Entity, &a.CreatedAt); err != nil { return nil, fmt.Errorf("scan gallery row: %w", err) } out = append(out, a) } return out, rows.Err() } // SetStar stars or unstars a message for one user. // // The message must belong to the conversation: the caller's membership is // checked against the thread, so accepting a message id from elsewhere would // let them star something they cannot see. func (r *ChatRepo) SetStar(ctx context.Context, conversationID, messageID, userID string, starred bool) error { if !starred { _, err := r.pg.Pool.Exec(ctx, `DELETE FROM message_stars WHERE message_id = $1 AND user_id = $2`, messageID, userID) if err != nil { return fmt.Errorf("unstar: %w", err) } return nil } tag, err := r.pg.Pool.Exec(ctx, ` INSERT INTO message_stars (message_id, user_id, conversation_id) SELECT $1, $2, $3 WHERE EXISTS (SELECT 1 FROM messages WHERE id = $1 AND conversation_id = $3) ON CONFLICT (message_id, user_id) DO NOTHING`, messageID, userID, conversationID) if err != nil { return fmt.Errorf("star: %w", err) } // No row inserted and no conflict means the message isn't in this thread. if tag.RowsAffected() == 0 { var exists bool if err := r.pg.Pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM message_stars WHERE message_id=$1 AND user_id=$2)`, messageID, userID).Scan(&exists); err != nil { return fmt.Errorf("star check: %w", err) } if !exists { return ErrMessageNotHere } } return nil } // Starred lists the caller's starred messages in a thread, newest first. func (r *ChatRepo) Starred(ctx context.Context, conversationID, userID string, limit int) ([]model.Message, error) { rows, err := r.pg.Pool.Query(ctx, ` SELECT m.id, m.conversation_id, m.seq, m.sender_id, m.kind, m.body, m.entities, m.client_msg_id, m.reply_to_id, m.forwarded_from_id, m.edited_at, m.deleted_at, m.created_at FROM message_stars st JOIN messages m ON m.id = st.message_id WHERE st.user_id = $1 AND st.conversation_id = $2 AND m.deleted_at IS NULL ORDER BY st.created_at DESC LIMIT $3`, userID, conversationID, limit) if err != nil { return nil, fmt.Errorf("query starred: %w", err) } defer rows.Close() msgs := []model.Message{} for rows.Next() { var m model.Message var entities []byte if err := rows.Scan(&m.ID, &m.ConversationID, &m.Seq, &m.SenderID, &m.Kind, &m.Body, &entities, &m.ClientMsgID, &m.ReplyToID, &m.ForwardedFromID, &m.EditedAt, &m.DeletedAt, &m.CreatedAt); err != nil { return nil, fmt.Errorf("scan starred: %w", err) } m.Entities = json.RawMessage(entities) m.Attachments = []model.Attachment{} msgs = append(msgs, m) } return msgs, rows.Err() } // StarredIDs reports which of the given messages the caller has starred, so a // thread can show the star on each bubble. func (r *ChatRepo) StarredIDs(ctx context.Context, userID string, messageIDs []string) (map[string]bool, error) { out := map[string]bool{} if len(messageIDs) == 0 { return out, nil } rows, err := r.pg.Pool.Query(ctx, `SELECT message_id FROM message_stars WHERE user_id = $1 AND message_id = ANY($2)`, userID, messageIDs) if err != nil { return nil, fmt.Errorf("query stars: %w", err) } defer rows.Close() for rows.Next() { var id string if err := rows.Scan(&id); err != nil { return nil, fmt.Errorf("scan star: %w", err) } out[id] = true } return out, rows.Err() } // FollowEdges returns the accepted-follow state between two users in both // directions, plus when the first of them was accepted. func (r *ChatRepo) FollowEdges(ctx context.Context, a, b string) (aFollowsB, bFollowsA bool, since *time.Time, err error) { rows, qErr := r.pg.Pool.Query(ctx, ` SELECT follower_id, status, COALESCE(responded_at, created_at) FROM follows WHERE (follower_id = $1 AND following_id = $2) OR (follower_id = $2 AND following_id = $1)`, a, b) if qErr != nil { return false, false, nil, fmt.Errorf("follow edges: %w", qErr) } defer rows.Close() for rows.Next() { var follower, status string var at time.Time if err := rows.Scan(&follower, &status, &at); err != nil { return false, false, nil, fmt.Errorf("scan follow edge: %w", err) } if status != "accepted" { continue // a pending request is not a friendship } if follower == a { aFollowsB = true } else { bFollowsA = true } // "Friends since" is the earliest accepted edge either way. if since == nil || at.Before(*since) { at := at since = &at } } return aFollowsB, bFollowsA, since, rows.Err() } // Settings returns a user's chat preferences, falling back to the defaults // when they have never changed anything. func (r *ChatRepo) Settings(ctx context.Context, userID string) (model.ChatSettings, error) { out := model.DefaultChatSettings() var gender *string err := r.pg.Pool.QueryRow(ctx, ` SELECT COALESCE(s.read_receipts, TRUE), COALESCE(s.who_can_message, 'anyone'), COALESCE(s.min_account_age_months, 3), COALESCE(s.train_kautilya, TRUE), u.gender FROM users u LEFT JOIN chat_settings s ON s.user_id = u.id WHERE u.id = $1`, userID, ).Scan(&out.ReadReceipts, &out.WhoCanMessage, &out.MinAccountAgeMonths, &out.TrainKautilya, &gender) if errors.Is(err, pgx.ErrNoRows) { return out, ErrUserGone } if err != nil { return out, fmt.Errorf("chat settings: %w", err) } out.GenderSet = gender != nil && *gender != "" return out, nil } // SaveSettings upserts a user's chat preferences. func (r *ChatRepo) SaveSettings(ctx context.Context, userID string, s model.ChatSettings) error { _, err := r.pg.Pool.Exec(ctx, ` INSERT INTO chat_settings (user_id, read_receipts, who_can_message, min_account_age_months, train_kautilya) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (user_id) DO UPDATE SET read_receipts = EXCLUDED.read_receipts, who_can_message = EXCLUDED.who_can_message, min_account_age_months = EXCLUDED.min_account_age_months, train_kautilya = EXCLUDED.train_kautilya, updated_at = now()`, userID, s.ReadReceipts, s.WhoCanMessage, s.MinAccountAgeMonths, s.TrainKautilya) if err != nil { return fmt.Errorf("save chat settings: %w", err) } return nil } // CanStartConversation applies the recipient's "who can message you" rule. // // Only gates STARTING a thread. An existing conversation keeps working, so // tightening the setting never strands people mid-exchange. func (r *ChatRepo) CanStartConversation(ctx context.Context, senderID, recipientID string) (bool, error) { settings, err := r.Settings(ctx, recipientID) if err != nil { return false, err } switch settings.WhoCanMessage { case model.WhoAnyone: return true, nil case model.WhoFriends: // An accepted follow in EITHER direction counts as a connection. following, followedBy, _, err := r.FollowEdges(ctx, senderID, recipientID) if err != nil { return false, err } return following || followedBy, nil case model.WhoAccountAge: var oldEnough bool if err := r.pg.Pool.QueryRow(ctx, ` SELECT created_at <= now() - make_interval(months => $2) FROM users WHERE id = $1`, senderID, settings.MinAccountAgeMonths, ).Scan(&oldEnough); err != nil { return false, fmt.Errorf("account age: %w", err) } return oldEnough, nil case model.WhoSameGender: // Unusable without the recipient's own gender; treated as "anyone" // rather than locking them out of their own inbox. if !settings.GenderSet { return true, nil } var same bool if err := r.pg.Pool.QueryRow(ctx, ` SELECT s.gender IS NOT NULL AND s.gender = r.gender FROM users s, users r WHERE s.id = $1 AND r.id = $2`, senderID, recipientID).Scan(&same); err != nil { return false, fmt.Errorf("gender match: %w", err) } return same, nil } return true, nil } // ReadReceiptsEnabled reports which of the given users allow read receipts. func (r *ChatRepo) ReadReceiptsEnabled(ctx context.Context, userIDs []string) (map[string]bool, error) { out := map[string]bool{} if len(userIDs) == 0 { return out, nil } rows, err := r.pg.Pool.Query(ctx, ` SELECT u.id, COALESCE(s.read_receipts, TRUE) FROM users u LEFT JOIN chat_settings s ON s.user_id = u.id WHERE u.id = ANY($1)`, userIDs) if err != nil { return nil, fmt.Errorf("read receipt settings: %w", err) } defer rows.Close() for rows.Next() { var id string var on bool if err := rows.Scan(&id, &on); err != nil { return nil, fmt.Errorf("scan receipt setting: %w", err) } out[id] = on } return out, rows.Err() } // UserIDsByHandles resolves handles to ids, skipping any that don't exist. // // Member operations take handles because that is what the UI has: search // results and follow lists identify people by handle, and handles are what the // rest of the API keys on (/u/{handle}/block, /u/{handle}/followers). func (r *ChatRepo) UserIDsByHandles(ctx context.Context, handles []string) ([]string, error) { if len(handles) == 0 { return nil, nil } rows, err := r.pg.Pool.Query(ctx, // Banned users are skipped: they cannot be added to a group. `SELECT id FROM users WHERE handle = ANY($1) AND banned_at IS NULL`, handles) if err != nil { return nil, fmt.Errorf("resolve handles: %w", err) } defer rows.Close() ids := []string{} for rows.Next() { var id string if err := rows.Scan(&id); err != nil { return nil, fmt.Errorf("scan handle id: %w", err) } ids = append(ids, id) } return ids, rows.Err() } // AddParticipants adds members to a group, reviving anyone who had left. func (r *ChatRepo) AddParticipants(ctx context.Context, conversationID string, userIDs []string) error { for _, id := range userIDs { if _, err := r.pg.Pool.Exec(ctx, ` INSERT INTO conversation_participants (conversation_id, user_id, role, joined_at) VALUES ($1, $2, 'member', now()) ON CONFLICT (conversation_id, user_id) -- Re-adding someone who left clears left_at and restarts their -- tenure; it must not resurrect an old role. DO UPDATE SET left_at = NULL, role = 'member', joined_at = now() WHERE conversation_participants.left_at IS NOT NULL`, conversationID, id); err != nil { return fmt.Errorf("add participant: %w", err) } } return nil } // RemoveParticipant marks someone as having left. The row stays so their past // messages still resolve to a member when history is rendered. func (r *ChatRepo) RemoveParticipant(ctx context.Context, conversationID, userID string) error { _, err := r.pg.Pool.Exec(ctx, ` UPDATE conversation_participants SET left_at = now() WHERE conversation_id = $1 AND user_id = $2 AND left_at IS NULL`, conversationID, userID) if err != nil { return fmt.Errorf("remove participant: %w", err) } return nil } // SetRole changes a participant's role. func (r *ChatRepo) SetRole(ctx context.Context, conversationID, userID, role string) error { _, err := r.pg.Pool.Exec(ctx, ` UPDATE conversation_participants SET role = $3 WHERE conversation_id = $1 AND user_id = $2 AND left_at IS NULL`, conversationID, userID, role) if err != nil { return fmt.Errorf("set role: %w", err) } return nil } // UpdateGroup changes a group's title and/or send policy. Nil leaves a field // untouched. func (r *ChatRepo) UpdateGroup(ctx context.Context, conversationID string, title, sendPolicy, avatarKey *string) error { _, err := r.pg.Pool.Exec(ctx, ` UPDATE conversations SET title = COALESCE($2, title), send_policy = COALESCE($3, send_policy), avatar_key = COALESCE($4, avatar_key), updated_at = now() WHERE id = $1`, conversationID, title, sendPolicy, avatarKey) if err != nil { return fmt.Errorf("update group: %w", err) } return nil } // SuccessorOwner picks who inherits a group when its owner leaves: the // longest-serving remaining admin, else the longest-serving member. Empty when // nobody is left. // // Without this a group whose owner walks away would have no one able to // administer it. func (r *ChatRepo) SuccessorOwner(ctx context.Context, conversationID, leavingUserID string) (string, error) { var id string err := r.pg.Pool.QueryRow(ctx, ` SELECT user_id FROM conversation_participants WHERE conversation_id = $1 AND user_id <> $2 AND left_at IS NULL ORDER BY CASE role WHEN 'admin' THEN 0 ELSE 1 END, joined_at LIMIT 1`, conversationID, leavingUserID).Scan(&id) if errors.Is(err, pgx.ErrNoRows) { return "", nil } if err != nil { return "", fmt.Errorf("successor owner: %w", err) } return id, nil } // MessageConversation returns the thread a message belongs to, plus its // sender — enough to authorize a delete, a forward or a report. func (r *ChatRepo) MessageConversation(ctx context.Context, messageID string) (conversationID string, senderID *string, err error) { err = r.pg.Pool.QueryRow(ctx, `SELECT conversation_id, sender_id FROM messages WHERE id = $1`, messageID, ).Scan(&conversationID, &senderID) if errors.Is(err, pgx.ErrNoRows) { return "", nil, ErrMessageNotHere } if err != nil { return "", nil, fmt.Errorf("message conversation: %w", err) } return conversationID, senderID, nil } // MessageByID loads a single message. Callers authorize separately — this // only fetches. func (r *ChatRepo) MessageByID(ctx context.Context, messageID string) (*model.Message, error) { m, err := scanMessage(r.pg.Pool.QueryRow(ctx, ` SELECT id, conversation_id, seq, sender_id, kind, body, entities, client_msg_id, reply_to_id, forwarded_from_id, edited_at, deleted_at, created_at FROM messages WHERE id = $1`, messageID)) if errors.Is(err, pgx.ErrNoRows) { return nil, ErrMessageNotHere } if err != nil { return nil, fmt.Errorf("message by id: %w", err) } return m, nil } // HideMessage removes a message from one user's view only. Others keep seeing // it — this is "delete for me". func (r *ChatRepo) HideMessage(ctx context.Context, messageID, userID string) error { _, err := r.pg.Pool.Exec(ctx, ` INSERT INTO message_hides (message_id, user_id) VALUES ($1, $2) ON CONFLICT DO NOTHING`, messageID, userID) if err != nil { return fmt.Errorf("hide message: %w", err) } return nil } // DeleteMessage tombstones a message for everyone. // // The row survives on purpose: `seq` must stay gapless for the client's // catch-up cursor, and other participants need something to replace the bubble // with rather than a hole. Body and entities are cleared so the text is // actually gone, and attachments go with it. func (r *ChatRepo) DeleteMessage(ctx context.Context, messageID, byUserID string) error { tx, err := r.pg.Pool.Begin(ctx) if err != nil { return fmt.Errorf("begin delete tx: %w", err) } defer tx.Rollback(ctx) //nolint:errcheck if _, err := tx.Exec(ctx, ` UPDATE messages SET deleted_at = now(), deleted_by = $2, body = '', entities = '[]'::jsonb WHERE id = $1`, messageID, byUserID); err != nil { return fmt.Errorf("delete message: %w", err) } if _, err := tx.Exec(ctx, `DELETE FROM message_attachments WHERE message_id = $1`, messageID); err != nil { return fmt.Errorf("delete message attachments: %w", err) } return tx.Commit(ctx) } // MessageWithContext returns a message and the few around it — the snapshot a // report captures so moderators never need to read the live thread. func (r *ChatRepo) MessageWithContext(ctx context.Context, messageID string, around int) ([]model.Message, error) { rows, err := r.pg.Pool.Query(ctx, ` WITH target AS ( SELECT conversation_id, seq FROM messages WHERE id = $1 ) SELECT m.id, m.conversation_id, m.seq, m.sender_id, m.kind, m.body, m.entities, m.client_msg_id, m.reply_to_id, m.forwarded_from_id, m.edited_at, m.deleted_at, m.created_at FROM messages m, target t WHERE m.conversation_id = t.conversation_id AND m.seq BETWEEN t.seq - $2 AND t.seq + $2 ORDER BY m.seq`, messageID, around) if err != nil { return nil, fmt.Errorf("message context: %w", err) } defer rows.Close() msgs := []model.Message{} for rows.Next() { var m model.Message var entities []byte if err := rows.Scan(&m.ID, &m.ConversationID, &m.Seq, &m.SenderID, &m.Kind, &m.Body, &entities, &m.ClientMsgID, &m.ReplyToID, &m.ForwardedFromID, &m.EditedAt, &m.DeletedAt, &m.CreatedAt); err != nil { return nil, fmt.Errorf("scan context message: %w", err) } m.Entities = json.RawMessage(entities) msgs = append(msgs, m) } return msgs, rows.Err() } // ReportMessage records a report with its evidence copied in. func (r *ChatRepo) ReportMessage(ctx context.Context, reporterID, messageID, conversationID string, reportedUserID *string, category, reason string, snapshot []byte) error { _, err := r.pg.Pool.Exec(ctx, ` INSERT INTO message_reports (reporter_id, message_id, conversation_id, reported_user_id, category, reason, snapshot) VALUES ($1,$2,$3,$4,$5,$6,$7)`, reporterID, messageID, conversationID, reportedUserID, category, reason, snapshot) if err != nil { return fmt.Errorf("report message: %w", err) } return nil } // SetMute sets or clears the caller's mute on a thread. A nil until unmutes. // // Mute is per participant, not per conversation: muting a group silences it for // you alone. func (r *ChatRepo) SetMute(ctx context.Context, conversationID, userID string, until *time.Time) error { _, err := r.pg.Pool.Exec(ctx, ` UPDATE conversation_participants SET muted_until = $3 WHERE conversation_id = $1 AND user_id = $2`, conversationID, userID, until) if err != nil { return fmt.Errorf("set mute: %w", err) } return nil } // Receipts returns the delivery/read cursors across everyone EXCEPT the // caller, so the caller can draw ticks on their own messages. MIN is the // deliberate choice: in a group, blue ticks mean everyone has read it. func (r *ChatRepo) Receipts(ctx context.Context, conversationID, excludeUserID string) (model.Receipts, error) { var rc model.Receipts err := r.pg.Pool.QueryRow(ctx, ` SELECT COALESCE(MIN(last_delivered_seq), 0), COALESCE(MIN(last_read_seq), 0) FROM conversation_participants WHERE conversation_id = $1 AND user_id <> $2 AND left_at IS NULL`, conversationID, excludeUserID).Scan(&rc.DeliveredSeq, &rc.ReadSeq) if err != nil { return rc, fmt.Errorf("receipts: %w", err) } return rc, nil } func scanMessage(row pgx.Row) (*model.Message, error) { var m model.Message var entities []byte if err := row.Scan(&m.ID, &m.ConversationID, &m.Seq, &m.SenderID, &m.Kind, &m.Body, &entities, &m.ClientMsgID, &m.ReplyToID, &m.ForwardedFromID, &m.EditedAt, &m.DeletedAt, &m.CreatedAt); err != nil { return nil, err } m.Entities = json.RawMessage(entities) return &m, nil } func derefInt64(v *int64) int64 { if v == nil { return 0 } return *v } func derefOr(s *string, fallback string) string { if s == nil { return fallback } return *s }