Files
spsays-bff/internal/service/chat_service.go
T
2026-09-02 23:00:55 +05:30

2127 lines
70 KiB
Go

package service
import (
"bytes"
"context"
"crypto/rand"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"html/template"
"net/url"
"path"
"strings"
"time"
"github.com/prabhupadasays/spsays-bff/internal/model"
"github.com/prabhupadasays/spsays-bff/internal/repository"
"github.com/prabhupadasays/spsays-bff/pkg/mailer"
"github.com/prabhupadasays/spsays-bff/pkg/realtime"
"github.com/prabhupadasays/spsays-bff/pkg/unfurl"
"go.uber.org/zap"
)
var (
ErrChatBlocked = errors.New("cannot message this user")
ErrChatEmptyBody = errors.New("message is empty")
ErrChatBodyTooLong = errors.New("message is too long")
ErrChatReadOnly = errors.New("this conversation cannot be posted to")
ErrChatBadMuteDuration = errors.New("duration must be one of 8h, 1w, always, off")
ErrChatNoPreview = errors.New("no preview available for that link")
ErrChatLinkNotInBody = errors.New("that link does not appear in the message")
ErrChatMediaOff = errors.New("media uploads are not configured")
ErrChatBadMedia = errors.New("unsupported attachment type")
ErrChatNotSender = errors.New("only the sender can delete a message for everyone")
ErrMessageDeleted = errors.New("that message was deleted")
ErrChatBadReport = errors.New("a report category is required")
ErrChatNotAdmin = errors.New("only group admins can do that")
ErrChatNotGroup = errors.New("that conversation is not a group")
ErrChatBadRole = errors.New("role must be admin or member")
ErrChatBadPolicy = errors.New("send policy must be everyone or admins")
ErrChatOwnerImmutable = errors.New("the group owner's role cannot be changed")
ErrChatAdminsOnly = errors.New("only admins can post in this group")
ErrChatBadAvatarKey = errors.New("that image was not uploaded for this group")
ErrAvatarUnset = errors.New("this group has no photo")
ErrChatNotAccepting = errors.New("this person isn't accepting messages from you")
ErrChatBadWhoCanMessage = errors.New("unknown messaging restriction")
ErrChatBadQuote = errors.New("that quote is not a valid passage reference")
ErrChatBadEmoji = errors.New("that is not a valid reaction")
ErrChatBadEntity = errors.New("that post cannot be shared")
ErrChatBadAccountAge = errors.New("account age must be between 1 and 120 months")
ErrChatGenderUnset = errors.New("add your gender in profile settings to use this option")
)
// maxBodyRunes caps a single message. Generous for conversation, small enough
// that one message can't be used to stuff the database.
const maxBodyRunes = 8000
// defaultPageSize / maxPageSize bound history paging.
const (
defaultPageSize = 30
maxPageSize = 100
)
// ChatService owns conversation and message rules: who may read a thread, who
// may post to it, and who hears about it in realtime.
type ChatService struct {
chat *repository.ChatRepo
blocks *repository.BlockRepo
// posts resolves shared post cards. Optional: nil leaves entity
// attachments unresolved rather than failing a whole thread load.
posts *repository.PostRepo
// push, prefs and users deliver out-of-band message alerts. All optional:
// nil means no push, never a failed send.
push *PushSender
prefs *repository.NotificationPrefsRepo
users *repository.UserRepo
// mail nudges someone who has never used this thread and isn't connected.
// Optional: nil disables first-contact email and nothing else.
mail mailer.Mailer
frontendURL string
rt *realtime.SoketiClient
r2 *repository.R2
links *unfurl.Fetcher
cache *repository.Redis
logger *zap.Logger
}
func NewChatService(
chat *repository.ChatRepo,
blocks *repository.BlockRepo,
posts *repository.PostRepo,
push *PushSender,
prefs *repository.NotificationPrefsRepo,
users *repository.UserRepo,
mail mailer.Mailer,
frontendURL string,
rt *realtime.SoketiClient,
r2 *repository.R2,
cache *repository.Redis,
logger *zap.Logger,
) *ChatService {
return &ChatService{
chat: chat, blocks: blocks, posts: posts,
push: push, prefs: prefs, users: users, mail: mail,
frontendURL: frontendURL, rt: rt, r2: r2,
links: unfurl.New(), cache: cache, logger: logger,
}
}
// ConversationChannel is the private realtime channel for one thread. Exported
// so the channel-authorization rule and the publisher agree on the name.
func ConversationChannel(conversationID string) string {
return "private-conversation." + conversationID
}
// KindsForTab maps an inbox tab to the conversation kinds it shows. The two
// tabs are a filter over one pipeline, not two implementations.
func KindsForTab(tab string) []string {
if tab == "kautilya" {
return []string{model.ConversationAI}
}
return []string{model.ConversationDM, model.ConversationGroup}
}
// IsParticipant backs the realtime channel-authorization rule. It returns
// (false, nil) for "not a member" and an error only when the check itself
// failed, so a database blip is never mistaken for a denial.
func (s *ChatService) IsParticipant(ctx context.Context, userID, conversationID string) (bool, error) {
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
if errors.Is(err, repository.ErrNotParticipant) {
return false, nil
}
return false, err
}
return true, nil
}
// Settings returns the caller's chat preferences.
func (s *ChatService) Settings(ctx context.Context, userID string) (model.ChatSettings, error) {
return s.chat.Settings(ctx, userID)
}
// SaveSettings validates and stores the caller's chat preferences.
func (s *ChatService) SaveSettings(ctx context.Context, userID string, in model.ChatSettings) (model.ChatSettings, error) {
switch in.WhoCanMessage {
case model.WhoAnyone, model.WhoFriends, model.WhoAccountAge, model.WhoSameGender:
default:
return model.ChatSettings{}, ErrChatBadWhoCanMessage
}
if in.MinAccountAgeMonths < 1 || in.MinAccountAgeMonths > 120 {
return model.ChatSettings{}, ErrChatBadAccountAge
}
current, err := s.chat.Settings(ctx, userID)
if err != nil {
return model.ChatSettings{}, err
}
// Choosing same-gender without a gender on the profile would silently do
// nothing, so refuse and let the UI say why.
if in.WhoCanMessage == model.WhoSameGender && !current.GenderSet {
return model.ChatSettings{}, ErrChatGenderUnset
}
if err := s.chat.SaveSettings(ctx, userID, in); err != nil {
return model.ChatSettings{}, err
}
return s.chat.Settings(ctx, userID)
}
// Inbox lists the caller's threads for one tab.
func (s *ChatService) Inbox(ctx context.Context, userID, tab string, limit int) ([]model.InboxItem, error) {
if limit <= 0 || limit > maxPageSize {
limit = maxPageSize
}
return s.chat.Inbox(ctx, userID, KindsForTab(tab), limit)
}
// UnreadCount reports how many threads are waiting on the caller.
func (s *ChatService) UnreadCount(ctx context.Context, userID string) (int, error) {
return s.chat.UnreadConversations(ctx, userID)
}
// OpenDM returns (creating if needed) the DM between the caller and peer.
//
// Blocking is enforced in BOTH directions: a blocked user must not be able to
// open a thread, and neither should someone who blocked the peer — silently
// leaving a one-way channel open would be worse than refusing.
func (s *ChatService) OpenDM(ctx context.Context, userID, peerID string) (*model.Conversation, error) {
if userID == peerID {
return nil, errors.New("cannot open a conversation with yourself")
}
blocked, blockedBy, err := s.blocks.BlockState(ctx, userID, peerID)
if err != nil {
return nil, fmt.Errorf("block state: %w", err)
}
if blocked || blockedBy {
return nil, ErrChatBlocked
}
// The recipient's "who can message you" rule gates opening the thread. An
// existing conversation is unaffected, so tightening it never strands
// people mid-exchange.
allowed, err := s.chat.CanStartConversation(ctx, userID, peerID)
if err != nil {
return nil, err
}
if !allowed {
return nil, ErrChatNotAccepting
}
return s.chat.EnsureDM(ctx, userID, peerID)
}
// OpenDMByHandle resolves a handle and opens the DM.
func (s *ChatService) OpenDMByHandle(ctx context.Context, userID, handle string) (*model.Conversation, error) {
ids, err := s.chat.UserIDsByHandles(ctx, []string{handle})
if err != nil {
return nil, err
}
if len(ids) == 0 {
return nil, repository.ErrUserGone
}
return s.OpenDM(ctx, userID, ids[0])
}
// CreateGroup opens a group thread. Members who have blocked the creator (or
// are blocked by them) are dropped rather than failing the whole call.
func (s *ChatService) CreateGroup(ctx context.Context, userID, title string, memberHandles []string) (*model.Conversation, error) {
title = strings.TrimSpace(title)
if title == "" {
return nil, errors.New("group title is required")
}
memberIDs, err := s.chat.UserIDsByHandles(ctx, memberHandles)
if err != nil {
return nil, err
}
allowed := make([]string, 0, len(memberIDs))
for _, m := range memberIDs {
if m == userID {
continue
}
blocked, blockedBy, err := s.blocks.BlockState(ctx, userID, m)
if err != nil {
return nil, fmt.Errorf("block state: %w", err)
}
if blocked || blockedBy {
continue
}
allowed = append(allowed, m)
}
return s.chat.CreateGroup(ctx, userID, title, allowed)
}
// OpenAIThread returns the caller's Kautilya thread.
func (s *ChatService) OpenAIThread(ctx context.Context, userID string) (*model.Conversation, error) {
return s.chat.EnsureAIThread(ctx, userID)
}
// Detail returns one thread with its participants and the caller's own state.
func (s *ChatService) Detail(ctx context.Context, userID, conversationID string) (*model.ConversationDetail, error) {
me, err := s.chat.Participant(ctx, conversationID, userID)
if err != nil {
return nil, err
}
conv, err := s.chat.Conversation(ctx, conversationID)
if err != nil {
return nil, err
}
people, err := s.chat.Participants(ctx, conversationID)
if err != nil {
return nil, err
}
detail := &model.ConversationDetail{
Conversation: *conv,
Participants: people,
Unread: max64(conv.LastSeq-me.LastReadSeq, 0),
MutedUntil: me.MutedUntil,
MyRole: me.Role,
}
if conv.Kind == model.ConversationDM {
for i := range people {
if people[i].ID != userID {
detail.Peer = &people[i]
break
}
}
if detail.Peer != nil {
rel, err := s.relationship(ctx, userID, detail.Peer.ID)
if err != nil {
return nil, err
}
detail.Relationship = rel
}
}
return detail, nil
}
// relationship assembles the follow and block state the chat profile shows.
func (s *ChatService) relationship(ctx context.Context, userID, peerID string) (*model.ChatRelationship, error) {
following, followedBy, since, err := s.chat.FollowEdges(ctx, userID, peerID)
if err != nil {
return nil, err
}
blocked, blockedBy, err := s.blocks.BlockState(ctx, userID, peerID)
if err != nil {
return nil, fmt.Errorf("block state: %w", err)
}
return &model.ChatRelationship{
Following: following, FollowedBy: followedBy,
Blocked: blocked, BlockedBy: blockedBy, FriendsSince: since,
}, nil
}
// Gallery lists a thread's shared media or links for the chat profile.
func (s *ChatService) Gallery(ctx context.Context, userID, conversationID, tab string, before *time.Time, limit int) ([]model.Attachment, error) {
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return nil, err
}
if limit <= 0 || limit > maxPageSize {
limit = maxPageSize
}
kinds := repository.GalleryKindsMedia
if tab == "links" {
kinds = repository.GalleryKindsLink
}
return s.chat.Gallery(ctx, conversationID, userID, kinds, before, limit)
}
// SetStar stars or unstars a message for the caller.
func (s *ChatService) SetStar(ctx context.Context, userID, conversationID, messageID string, starred bool) error {
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return err
}
return s.chat.SetStar(ctx, conversationID, messageID, userID, starred)
}
// Starred lists the caller's starred messages in a thread.
func (s *ChatService) Starred(ctx context.Context, userID, conversationID string, limit int) ([]model.Message, error) {
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return nil, err
}
if limit <= 0 || limit > maxPageSize {
limit = maxPageSize
}
msgs, err := s.chat.Starred(ctx, conversationID, userID, limit)
if err != nil {
return nil, err
}
return s.hydrateAttachments(ctx, userID, msgs)
}
// hydrateAttachments fills in each message's attachments in one query, then
// resolves any shared entities for this particular viewer.
func (s *ChatService) hydrateAttachments(ctx context.Context, viewerID string, msgs []model.Message) ([]model.Message, error) {
ids := make([]string, 0, len(msgs))
for i := range msgs {
ids = append(ids, msgs[i].ID)
}
byMessage, err := s.chat.AttachmentsFor(ctx, ids)
if err != nil {
return nil, err
}
// One grouped query for the whole page. Counts depend on the viewer
// (`mine`), which is why this is hydrated per request rather than cached.
reactions, err := s.chat.ReactionsFor(ctx, ids, viewerID)
if err != nil {
return nil, err
}
for i := range msgs {
if a := byMessage[msgs[i].ID]; a != nil {
msgs[i].Attachments = a
} else {
// Always a slice, never nil, so clients need no guard.
msgs[i].Attachments = []model.Attachment{}
}
if r := reactions[msgs[i].ID]; r != nil {
msgs[i].Reactions = r
} else {
msgs[i].Reactions = []model.Reaction{}
}
}
pages := make([][]model.Attachment, 0, len(msgs))
for i := range msgs {
pages = append(pages, msgs[i].Attachments)
}
s.resolveEntities(ctx, viewerID, pages)
return msgs, nil
}
func max64(a, b int64) int64 {
if a > b {
return a
}
return b
}
// Messages returns a page of history after checking membership.
func (s *ChatService) Messages(ctx context.Context, userID, conversationID string, beforeSeq int64, limit int) ([]model.Message, error) {
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return nil, err
}
if limit <= 0 {
limit = defaultPageSize
}
if limit > maxPageSize {
limit = maxPageSize
}
msgs, err := s.chat.Messages(ctx, conversationID, userID, beforeSeq, limit)
if err != nil {
return nil, err
}
return s.hydrateAttachments(ctx, userID, msgs)
}
// SendRequest is one outgoing message from a client.
type SendRequest struct {
// Body is raw composer text; formatting markers are parsed out here, so
// clients never send markup or entity spans of their own.
Body string
// ClientMsgID makes a retry idempotent.
ClientMsgID *string
ReplyToID *string
// PreviewURL asks for a link card. Only the URL is accepted — the BFF
// fetches the title, description and image itself, because a client that
// could supply them could make any link claim to be anything.
PreviewURL *string
// Entity is something else in the app — a post — shared as a card.
Entity *EntityAttachment
// StickerKey sends a piece of reward artwork. Ownership is checked
// server-side; the key alone proves nothing.
StickerKey *string
// Quote is a passage shared from the library, rendered as a quote the
// reader can tap to return to the exact phrase.
Quote *QuoteAttachment
// Attachments are files already uploaded to the private bucket via
// PresignAttachment; their keys are checked against the caller's prefix.
Attachments []MediaAttachment
}
// MediaAttachment is one uploaded file being attached to a message.
type MediaAttachment struct {
Key string
Mime string
SizeBytes *int64
Width *int
Height *int
DurationMs *int
Blurhash *string
FileName *string
}
// Send validates, persists and publishes a message.
//
// Entities are parsed server-side rather than trusted from the client: they
// drive rendering, so accepting client-supplied spans would let one user
// control how their text is displayed in someone else's thread.
func (s *ChatService) Send(ctx context.Context, userID, conversationID string, req SendRequest) (*model.Message, error) {
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return nil, err
}
if err := s.assertCanPost(ctx, userID, conversationID); err != nil {
return nil, err
}
if req.Quote != nil && !req.Quote.valid() {
return nil, ErrChatBadQuote
}
body := strings.TrimSpace(req.Body)
// A photo or a shared passage with no caption is an ordinary message; only
// a message with nothing at all in it is empty.
if body == "" && len(req.Attachments) == 0 && req.Quote == nil &&
req.Entity == nil && req.StickerKey == nil {
return nil, ErrChatEmptyBody
}
if len([]rune(body)) > maxBodyRunes {
return nil, ErrChatBodyTooLong
}
clean, entities := ParseEntities(body)
encoded, err := json.Marshal(entities)
if err != nil {
return nil, fmt.Errorf("encode entities: %w", err)
}
if entities == nil {
encoded = []byte(`[]`)
}
msg, created, err := s.chat.Append(ctx, repository.AppendParams{
ConversationID: conversationID,
SenderID: &userID,
Kind: model.MessageText,
Body: clean,
Entities: encoded,
ClientMsgID: req.ClientMsgID,
ReplyToID: req.ReplyToID,
})
if err != nil {
return nil, err
}
// A retry returns the stored message as-is: its attachments were persisted
// the first time, and re-publishing would duplicate the bubble for
// everyone else.
if !created {
return s.hydrateSent(ctx, userID, conversationID, msg), nil
}
if req.StickerKey != nil {
if err := s.attachSticker(ctx, conversationID, userID, msg, *req.StickerKey); err != nil {
s.logger.Warn("chat: sticker refused",
zap.String("user_id", userID), zap.Error(err))
}
}
if err := s.attachEntity(ctx, conversationID, userID, msg, req.Entity); err != nil {
s.logger.Warn("chat: entity share refused", zap.String("user_id", userID), zap.Error(err))
}
s.attachQuote(ctx, conversationID, msg, req.Quote)
s.attachMedia(ctx, conversationID, userID, msg, req.Attachments)
s.attachLinkPreview(ctx, conversationID, msg, entities, req.PreviewURL)
full := s.hydrateSent(ctx, userID, conversationID, msg)
s.publish(ctx, conversationID, userID, full)
return full, nil
}
// hydrateSent re-reads a just-written message through the full read path.
//
// Append RETURNINGs the row it inserted, which carries reply_to_id but not the
// quoted preview — that is built by a LATERAL join only the read query has. So
// a reply sent over the wire, and the realtime event announcing it, would both
// arrive with nothing to quote and stay that way until the thread was
// reloaded. Reading it back costs one query and makes the sent message
// identical to the one everyone else receives.
func (s *ChatService) hydrateSent(ctx context.Context, viewerID, conversationID string, msg *model.Message) *model.Message {
full, err := s.chat.Message(ctx, conversationID, msg.ID, viewerID)
if err != nil {
// Fall back to the unhydrated row: a missing quote is better than a
// message that appears not to have sent at all.
full = msg
}
hydrated, err := s.withAttachments(ctx, viewerID, full)
if err != nil {
return full
}
return hydrated
}
// withAttachments hydrates a message's attachments for the response.
func (s *ChatService) withAttachments(ctx context.Context, viewerID string, msg *model.Message) (*model.Message, error) {
byMessage, err := s.chat.AttachmentsFor(ctx, []string{msg.ID})
if err != nil {
return msg, err
}
msg.Attachments = byMessage[msg.ID]
if msg.Attachments == nil {
msg.Attachments = []model.Attachment{}
}
if reactions, rErr := s.chat.ReactionsFor(ctx, []string{msg.ID}, viewerID); rErr == nil {
msg.Reactions = reactions[msg.ID]
}
if msg.Reactions == nil {
msg.Reactions = []model.Reaction{}
}
s.resolveEntities(ctx, viewerID, [][]model.Attachment{msg.Attachments})
return msg, nil
}
// attachMedia persists already-uploaded files.
//
// Each key must sit under this conversation and this uploader's prefix. Without
// that check a caller could attach any key it could guess — including one
// uploaded to a thread it is not in.
func (s *ChatService) attachMedia(ctx context.Context, conversationID, userID string, msg *model.Message, media []MediaAttachment) {
prefix := path.Join("chat", conversationID, userID) + "/"
for i, m := range media {
if !strings.HasPrefix(m.Key, prefix) {
s.logger.Warn("chat: refusing attachment key outside the caller's prefix",
zap.String("user_id", userID), zap.String("conversation_id", conversationID))
continue
}
kind, ok := attachmentKindFor(m.Mime)
if !ok {
continue
}
key := m.Key
mime := m.Mime
if _, err := s.chat.AddAttachment(ctx, repository.AddAttachmentParams{
MessageID: msg.ID, ConversationID: conversationID, Ord: i, Kind: kind,
R2Key: &key, Mime: &mime, SizeBytes: m.SizeBytes, Width: m.Width,
Height: m.Height, DurationMs: m.DurationMs, Blurhash: m.Blurhash,
FileName: m.FileName,
}); err != nil {
s.logger.Warn("chat: attach media failed", zap.Error(err))
}
}
}
// EntityAttachment is a reference to something else in the app, shared into a
// chat. Only posts for now; the column allows comments, profiles and
// collections when their cards are built.
type EntityAttachment struct {
Type string
ID string
}
// postSnapshot is the copy stored alongside the reference.
//
// It is a FALLBACK, never the source of truth: resolveEntities re-reads the
// live post for each viewer, because a post can be deleted, hidden, or become
// invisible to one participant after it was shared. Rendering the snapshot
// unconditionally would leak a post to someone who has since lost access.
type postSnapshot struct {
ID string `json:"id"`
AuthorHandle string `json:"author_handle"`
AuthorName string `json:"author_name"`
AuthorAvatarURL *string `json:"author_avatar_url,omitempty"`
Caption string `json:"caption"`
ImageURL *string `json:"image_url,omitempty"`
SourceTitle string `json:"source_title,omitempty"`
}
func snapshotOf(p *model.Post) postSnapshot {
snap := postSnapshot{
ID: p.ID, AuthorHandle: p.AuthorHandle, AuthorName: p.AuthorName,
AuthorAvatarURL: p.AuthorAvatarURL, Caption: p.Caption,
SourceTitle: p.SourceTitle,
}
if len(p.Images) > 0 {
url := p.Images[0].URL
snap.ImageURL = &url
}
return snap
}
// postVisibleTo decides whether a viewer may see a post inside a chat bubble.
//
// Same rules the feed applies: a hidden post is the owner's alone, and a block
// in either direction hides the author's content.
func (s *ChatService) postVisibleTo(ctx context.Context, viewerID string, p *model.Post) bool {
if p.IsHidden && p.UserID != viewerID {
return false
}
if p.UserID == viewerID {
return true
}
blocked, blockedBy, err := s.blocks.BlockState(ctx, viewerID, p.UserID)
if err != nil {
// Fail closed: an unreadable block table must not open a post up.
return false
}
return !blocked && !blockedBy
}
// attachEntity stores a shared post after checking the SENDER can see it —
// you cannot forward what you were never allowed to read.
func (s *ChatService) attachEntity(ctx context.Context, conversationID, userID string, msg *model.Message, e *EntityAttachment) error {
if e == nil {
return nil
}
if e.Type != model.EntityPost {
return ErrChatBadEntity
}
post, err := s.posts.FindByID(ctx, e.ID)
if err != nil || post == nil {
return ErrChatBadEntity
}
if !s.postVisibleTo(ctx, userID, post) {
return ErrChatBadEntity
}
snap, err := json.Marshal(snapshotOf(post))
if err != nil {
return ErrChatBadEntity
}
entityType, entityID := model.EntityPost, post.ID
if _, err := s.chat.AddAttachment(ctx, repository.AddAttachmentParams{
MessageID: msg.ID, ConversationID: conversationID, Ord: 0,
Kind: model.AttachmentEntity, EntityType: &entityType,
EntityID: &entityID, Snapshot: snap,
}); err != nil {
s.logger.Warn("chat: attach entity failed", zap.Error(err))
}
return nil
}
// resolveEntities re-reads every shared entity on a page for THIS viewer.
//
// This is the permission re-check that makes sharing safe: access is decided
// at read time, per person, not frozen at send time. A post that was visible
// when shared and is hidden now renders as unavailable.
//
// It takes the whole page at once — three queries (posts, their images,
// blocks) regardless of how many shares are on it. Resolving per message
// instead would cost two queries per share and could not dedupe a post quoted
// in several messages.
func (s *ChatService) resolveEntities(ctx context.Context, viewerID string, pages [][]model.Attachment) {
if s.posts == nil {
return
}
// Gather the distinct posts referenced anywhere on the page.
shared := map[string]bool{}
for _, atts := range pages {
for i := range atts {
if atts[i].Kind == model.AttachmentEntity && atts[i].EntityID != nil {
shared[*atts[i].EntityID] = true
}
}
}
if len(shared) == 0 {
return
}
ids := make([]string, 0, len(shared))
for id := range shared {
ids = append(ids, id)
}
posts, err := s.posts.FindByIDs(ctx, ids)
if err != nil {
// Fail closed: every card becomes a tombstone rather than falling back
// to snapshots whose visibility we could not confirm.
s.logger.Warn("chat: resolving shared posts failed", zap.Error(err))
posts = nil
}
authors := make([]string, 0, len(posts))
for _, post := range posts {
if post.UserID != viewerID {
authors = append(authors, post.UserID)
}
}
blocked, err := s.blocks.BlockedWith(ctx, viewerID, authors)
if err != nil {
s.logger.Warn("chat: block lookup for shared posts failed", zap.Error(err))
// Fail closed again: an unreadable block table must not open posts up.
blocked = nil
for _, id := range authors {
if blocked == nil {
blocked = map[string]bool{}
}
blocked[id] = true
}
}
// Resolve each referenced post once, then fan the result out.
resolved := make(map[string]json.RawMessage, len(shared))
for id := range shared {
post := posts[id]
if post == nil {
continue
}
if post.IsHidden && post.UserID != viewerID {
continue
}
if blocked[post.UserID] {
continue
}
if b, mErr := json.Marshal(snapshotOf(post)); mErr == nil {
resolved[id] = b
}
}
for _, atts := range pages {
for i := range atts {
a := &atts[i]
if a.Kind != model.AttachmentEntity || a.EntityID == nil {
continue
}
// The stored snapshot is never sent to a client — only a copy
// resolved for this viewer — so a stale one cannot leak.
a.Entity = resolved[*a.EntityID]
a.EntityUnavailable = a.Entity == nil
}
}
}
// QuoteAttachment is a passage shared out of the library.
type QuoteAttachment struct {
// Path is an APP-RELATIVE deep link, e.g. "/books/bg/2/13?hl=…".
//
// Deliberately not a full URL. A quote renders as trusted in-app content,
// so allowing an absolute URL would let a sender dress an off-site link up
// as a scripture citation. A path cannot leave the app, which removes the
// question of which origins to trust.
Path string
// Title is the source reference shown above the passage.
Title string
// Text is the quoted passage itself.
Text string
}
const (
maxQuoteTitleRunes = 200
maxQuoteTextRunes = 2000
maxQuotePathBytes = 2048
)
func (q *QuoteAttachment) valid() bool {
if strings.TrimSpace(q.Text) == "" {
return false
}
if len([]rune(q.Title)) > maxQuoteTitleRunes || len([]rune(q.Text)) > maxQuoteTextRunes {
return false
}
return safeAppPath(q.Path)
}
// safeAppPath accepts only a path within this app.
//
// "//evil.com" and "/\evil.com" are protocol-relative and would navigate
// off-site, so a leading slash alone is not enough to check.
func safeAppPath(p string) bool {
if p == "" || len(p) > maxQuotePathBytes {
return false
}
if p[0] != '/' {
return false
}
if strings.HasPrefix(p, "//") || strings.HasPrefix(p, "/\\") {
return false
}
if strings.ContainsAny(p, "\\") {
return false
}
for _, r := range p {
if r < 0x20 || r == 0x7f {
return false
}
}
// A parsed path must not carry a scheme or host of its own.
u, err := url.Parse(p)
if err != nil || u.Scheme != "" || u.Host != "" {
return false
}
return true
}
// attachQuote stores a shared passage. Unlike a link card, the contents come
// from the client — but they are only ever rendered as a quotation, and the
// path has already been constrained to this app.
func (s *ChatService) attachQuote(ctx context.Context, conversationID string, msg *model.Message, q *QuoteAttachment) {
if q == nil {
return
}
path, title, text := q.Path, strings.TrimSpace(q.Title), strings.TrimSpace(q.Text)
if _, err := s.chat.AddAttachment(ctx, repository.AddAttachmentParams{
MessageID: msg.ID, ConversationID: conversationID, Ord: 0,
Kind: model.AttachmentQuote, URL: &path, Title: &title, Description: &text,
}); err != nil {
s.logger.Warn("chat: attach quote failed", zap.Error(err))
}
}
// attachLinkPreview unfurls the chosen link and stores the card.
//
// Two rules make this safe. The URL must be one the parser actually found in
// the message body — otherwise a message could carry a card for a link it never
// mentions. And the metadata comes from our own fetch, never from the client.
//
// Unfurling happens after the message is stored, so a slow remote page delays
// the card, never the message.
func (s *ChatService) attachLinkPreview(ctx context.Context, conversationID string, msg *model.Message, entities []Entity, previewURL *string) {
if previewURL == nil || *previewURL == "" {
return // the sender dismissed the preview, or there was none
}
if !urlInEntities(*previewURL, entities) {
s.logger.Warn("chat: preview url not present in message body")
return
}
preview, err := s.LinkPreview(ctx, *previewURL)
if err != nil {
return // no card rather than a broken one
}
params := repository.AddAttachmentParams{
MessageID: msg.ID, ConversationID: conversationID,
Ord: len(msg.Attachments), Kind: model.AttachmentLink,
URL: &preview.URL,
}
if preview.Title != "" {
params.Title = &preview.Title
}
if preview.Description != "" {
params.Description = &preview.Description
}
if preview.SiteName != "" {
params.SiteName = &preview.SiteName
}
if preview.ImageURL != "" {
params.ImageURL = &preview.ImageURL
}
if _, err := s.chat.AddAttachment(ctx, params); err != nil {
s.logger.Warn("chat: attach link preview failed", zap.Error(err))
}
}
// urlInEntities reports whether the parser found this URL in the body.
func urlInEntities(target string, entities []Entity) bool {
for _, e := range entities {
if e.Type == EntityURL && e.URL == target {
return true
}
}
return false
}
// linkCacheTTL keeps a fetched preview around long enough that the composer's
// lookup and the send that follows it don't fetch the same page twice.
const linkCacheTTL = 30 * time.Minute
// LinkPreview fetches metadata for a URL the user is about to send.
//
// Cached by URL because the composer asks as soon as a link is typed and the
// send asks again moments later; without it every message with a link would
// fetch the remote page twice.
func (s *ChatService) LinkPreview(ctx context.Context, rawURL string) (*unfurl.Preview, error) {
key := "chat:unfurl:" + rawURL
if s.cache != nil {
if cached, err := s.cache.Client.Get(ctx, key).Bytes(); err == nil {
var p unfurl.Preview
if json.Unmarshal(cached, &p) == nil {
return &p, nil
}
}
}
p, err := s.links.Fetch(ctx, rawURL)
if err != nil {
// A blocked or unreadable link isn't an error worth showing in detail:
// the composer simply offers no card.
return nil, ErrChatNoPreview
}
if s.cache != nil {
if encoded, err := json.Marshal(p); err == nil {
_ = s.cache.Client.Set(ctx, key, encoded, linkCacheTTL).Err()
}
}
return p, nil
}
// PresignAttachment issues a one-shot upload URL into the PRIVATE bucket.
//
// Keys are namespaced by conversation and uploader, which is what lets the
// send path verify that a key being attached was uploaded by this caller for
// this thread rather than guessed.
func (s *ChatService) PresignAttachment(ctx context.Context, userID, conversationID, contentType string) (uploadURL, key string, err error) {
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return "", "", err
}
if _, ok := attachmentKindFor(contentType); !ok {
return "", "", ErrChatBadMedia
}
if s.r2 == nil || !s.r2.PrivateConfigured() {
return "", "", ErrChatMediaOff
}
suffix := make([]byte, 12)
if _, err := rand.Read(suffix); err != nil {
return "", "", err
}
key = path.Join("chat", conversationID, userID, hex.EncodeToString(suffix))
uploadURL, err = s.r2.PresignPutPrivate(ctx, key, contentType, 5*time.Minute)
if err != nil {
return "", "", err
}
return uploadURL, key, nil
}
// AttachmentURL mints a short-lived read URL, but only for a participant.
//
// The client holds a stable /chat/attachments/{id} URL and this re-checks
// membership on every fetch — so access revoked after the link was shared
// stops working, which a long-lived signed URL in the payload could not do.
func (s *ChatService) AttachmentURL(ctx context.Context, userID, attachmentID string) (string, error) {
conversationID, key, err := s.chat.AttachmentLocation(ctx, attachmentID)
if err != nil {
return "", err
}
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return "", err
}
if s.r2 == nil || !s.r2.PrivateConfigured() {
return "", ErrChatMediaOff
}
return s.r2.PresignGet(ctx, key, 5*time.Minute)
}
// attachmentKindFor maps a MIME type onto an attachment kind, and reports
// whether we accept it at all.
func attachmentKindFor(mime string) (string, bool) {
switch {
case strings.HasPrefix(mime, "image/"):
return model.AttachmentImage, true
case strings.HasPrefix(mime, "video/"):
return model.AttachmentVideo, true
case strings.HasPrefix(mime, "audio/"):
return model.AttachmentAudio, true
case mime == "application/pdf":
return model.AttachmentFile, true
default:
return "", false
}
}
// DeleteScope selects which kind of delete the caller asked for.
const (
DeleteForMe = "me"
DeleteForEveryone = "everyone"
reportContextSpan = 3 // messages either side captured with a report
maxReportReasonLen = 2000
)
// DeleteMessage removes a message for the caller, or for everyone.
//
// "For everyone" is the sender's alone: letting a recipient erase someone
// else's words for all participants would be a moderation power, not a chat
// feature. Anyone in the thread may hide a message for themselves.
func (s *ChatService) DeleteMessage(ctx context.Context, userID, messageID, scope string) error {
conversationID, senderID, err := s.chat.MessageConversation(ctx, messageID)
if err != nil {
return err
}
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return err
}
if scope == DeleteForEveryone {
if senderID == nil || *senderID != userID {
return ErrChatNotSender
}
if err := s.chat.DeleteMessage(ctx, messageID, userID); err != nil {
return err
}
// Tell the open thread so the bubble is replaced in place rather than
// lingering until a reload.
if s.rt != nil {
if err := s.rt.Trigger(ctx, ConversationChannel(conversationID), "message_deleted",
map[string]any{"message_id": messageID}); err != nil {
s.logger.Debug("chat: delete publish failed", zap.Error(err))
}
}
return nil
}
return s.chat.HideMessage(ctx, messageID, userID)
}
// ForwardMessage copies a message into another thread.
//
// Two memberships are checked, not one: the caller must be able to SEE the
// source and to POST to the destination. Checking only the destination would
// turn a message id into a way to copy text out of a thread you were never in.
func (s *ChatService) ForwardMessage(ctx context.Context, userID, messageID, toConversationID string) (*model.Message, error) {
sourceID, _, err := s.chat.MessageConversation(ctx, messageID)
if err != nil {
return nil, err
}
if _, err := s.chat.Participant(ctx, sourceID, userID); err != nil {
return nil, err
}
if _, err := s.chat.Participant(ctx, toConversationID, userID); err != nil {
return nil, err
}
if err := s.assertCanPost(ctx, userID, toConversationID); err != nil {
return nil, err
}
original, err := s.chat.MessageByID(ctx, messageID)
if err != nil {
return nil, err
}
if original.DeletedAt != nil {
return nil, ErrMessageDeleted
}
msg, created, err := s.chat.Append(ctx, repository.AppendParams{
ConversationID: toConversationID,
SenderID: &userID,
Kind: model.MessageText,
Body: original.Body,
Entities: original.Entities,
ForwardedFromID: &messageID,
})
if err != nil {
return nil, err
}
if created {
// Carry the attachments across. Without this a quote, sticker, photo
// or shared post forwards as an empty bubble: its content lives in an
// attachment row, not in the body.
if err := s.chat.CopyAttachments(ctx, messageID, msg.ID, toConversationID); err != nil {
s.logger.Warn("chat: forwarding attachments failed",
zap.String("message_id", messageID), zap.Error(err))
}
msg = s.hydrateSent(ctx, userID, toConversationID, msg)
s.publish(ctx, toConversationID, userID, msg)
}
return msg, nil
}
// ReportMessage files a report, copying the message and a little context into
// the report itself.
//
// The snapshot is the point: moderators read a report record and never need
// read access to live conversations, so a report cannot become a way to browse
// someone's private threads.
func (s *ChatService) ReportMessage(ctx context.Context, userID, messageID, category, reason string) error {
conversationID, senderID, err := s.chat.MessageConversation(ctx, messageID)
if err != nil {
return err
}
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return err
}
category = strings.TrimSpace(category)
if category == "" {
return ErrChatBadReport
}
if len(reason) > maxReportReasonLen {
reason = reason[:maxReportReasonLen]
}
context, err := s.chat.MessageWithContext(ctx, messageID, reportContextSpan)
if err != nil {
return err
}
snapshot, err := json.Marshal(map[string]any{
"reported_message_id": messageID,
"conversation_id": conversationID,
"captured_at": time.Now().UTC(),
"messages": context,
})
if err != nil {
return fmt.Errorf("encode report snapshot: %w", err)
}
return s.chat.ReportMessage(ctx, userID, messageID, conversationID, senderID, category, reason, snapshot)
}
// isAdmin reports whether a role may administer a group.
func isAdmin(role string) bool {
return role == model.RoleOwner || role == model.RoleAdmin
}
// requireGroupAdmin loads the caller's membership and insists they can
// administer this group.
func (s *ChatService) requireGroupAdmin(ctx context.Context, userID, conversationID string) (*model.Conversation, *model.Participant, error) {
me, err := s.chat.Participant(ctx, conversationID, userID)
if err != nil {
return nil, nil, err
}
conv, err := s.chat.Conversation(ctx, conversationID)
if err != nil {
return nil, nil, err
}
if conv.Kind != model.ConversationGroup {
return nil, nil, ErrChatNotGroup
}
if !isAdmin(me.Role) {
return nil, nil, ErrChatNotAdmin
}
return conv, me, nil
}
// systemMessage records a membership change in the thread itself, so the
// history explains how the group came to look the way it does.
func (s *ChatService) systemMessage(ctx context.Context, conversationID, body string) {
msg, created, err := s.chat.Append(ctx, repository.AppendParams{
ConversationID: conversationID,
Kind: model.MessageSystem,
Body: body,
})
if err != nil {
s.logger.Warn("chat: system message failed", zap.Error(err))
return
}
if created {
// Publish the hydrated message. Append leaves Attachments nil, which
// serialises to null and breaks clients that trust the "always an
// array" contract this field's own doc comment promises.
if full, hErr := s.withAttachments(ctx, "", msg); hErr == nil {
msg = full
} else if msg.Attachments == nil {
msg.Attachments = []model.Attachment{}
}
s.publish(ctx, conversationID, "", msg)
}
}
// AddMembers adds people to a group.
//
// Blocked pairs are skipped rather than failing the whole call — one awkward
// relationship shouldn't stop the other five invitations.
func (s *ChatService) AddMembers(ctx context.Context, userID, conversationID string, memberHandles []string) error {
if _, _, err := s.requireGroupAdmin(ctx, userID, conversationID); err != nil {
return err
}
memberIDs, err := s.chat.UserIDsByHandles(ctx, memberHandles)
if err != nil {
return err
}
allowed := make([]string, 0, len(memberIDs))
for _, id := range memberIDs {
if id == userID {
continue
}
blocked, blockedBy, err := s.blocks.BlockState(ctx, userID, id)
if err != nil {
return fmt.Errorf("block state: %w", err)
}
if blocked || blockedBy {
continue
}
allowed = append(allowed, id)
}
if len(allowed) == 0 {
return nil
}
if err := s.chat.AddParticipants(ctx, conversationID, allowed); err != nil {
return err
}
names, err := s.chat.Participants(ctx, conversationID)
if err == nil {
s.systemMessage(ctx, conversationID, addedNames(names, allowed))
}
return nil
}
func addedNames(people []model.ChatPeer, added []string) string {
byID := map[string]string{}
for _, p := range people {
byID[p.ID] = p.Name
}
labels := make([]string, 0, len(added))
for _, id := range added {
if n, ok := byID[id]; ok {
labels = append(labels, n)
}
}
if len(labels) == 0 {
return "Members were added"
}
return strings.Join(labels, ", ") + " joined the group"
}
// RemoveMember removes someone from a group, or lets them leave.
//
// Anyone may remove themselves. Removing someone else needs admin rights, and
// an admin cannot remove another admin or the owner — otherwise a single
// promoted member could empty the group's leadership.
func (s *ChatService) RemoveMember(ctx context.Context, userID, conversationID, targetHandle string) error {
targetIDs, err := s.chat.UserIDsByHandles(ctx, []string{targetHandle})
if err != nil {
return err
}
if len(targetIDs) == 0 {
return repository.ErrNotParticipant
}
targetID := targetIDs[0]
leaving := userID == targetID
me, err := s.chat.Participant(ctx, conversationID, userID)
if err != nil {
return err
}
conv, err := s.chat.Conversation(ctx, conversationID)
if err != nil {
return err
}
if conv.Kind != model.ConversationGroup {
return ErrChatNotGroup
}
if !leaving {
if !isAdmin(me.Role) {
return ErrChatNotAdmin
}
target, err := s.chat.Participant(ctx, conversationID, targetID)
if err != nil {
return err
}
if target.Role == model.RoleOwner {
return ErrChatOwnerImmutable
}
if target.Role == model.RoleAdmin && me.Role != model.RoleOwner {
return ErrChatNotAdmin
}
}
// A departing owner hands the group on, so it is never left unadministered.
if leaving && me.Role == model.RoleOwner {
successor, err := s.chat.SuccessorOwner(ctx, conversationID, userID)
if err != nil {
return err
}
if successor != "" {
if err := s.chat.SetRole(ctx, conversationID, successor, model.RoleOwner); err != nil {
return err
}
}
}
if err := s.chat.RemoveParticipant(ctx, conversationID, targetID); err != nil {
return err
}
name := targetID
if people, err := s.chat.Participants(ctx, conversationID); err == nil {
for _, p := range people {
if p.ID == targetID {
name = p.Name
}
}
}
if leaving {
s.systemMessage(ctx, conversationID, "A member left the group")
} else {
s.systemMessage(ctx, conversationID, name+" was removed from the group")
}
return nil
}
// SetMemberRole promotes a member to admin or demotes an admin.
//
// Only the owner may change roles: an admin promoting other admins would let
// one compromised account escalate the whole group.
func (s *ChatService) SetMemberRole(ctx context.Context, userID, conversationID, targetHandle, role string) error {
if role != model.RoleAdmin && role != model.RoleMember {
return ErrChatBadRole
}
targetIDs, err := s.chat.UserIDsByHandles(ctx, []string{targetHandle})
if err != nil {
return err
}
if len(targetIDs) == 0 {
return repository.ErrNotParticipant
}
targetID := targetIDs[0]
_, me, adminErr := s.requireGroupAdmin(ctx, userID, conversationID)
if adminErr != nil {
return adminErr
}
if me.Role != model.RoleOwner {
return ErrChatNotAdmin
}
if targetID == userID {
return ErrChatOwnerImmutable
}
target, err := s.chat.Participant(ctx, conversationID, targetID)
if err != nil {
return err
}
if target.Role == model.RoleOwner {
return ErrChatOwnerImmutable
}
if err := s.chat.SetRole(ctx, conversationID, targetID, role); err != nil {
return err
}
name := targetID
if people, err := s.chat.Participants(ctx, conversationID); err == nil {
for _, p := range people {
if p.ID == targetID {
name = p.Name
}
}
}
if role == model.RoleAdmin {
s.systemMessage(ctx, conversationID, name+" is now an admin")
} else {
s.systemMessage(ctx, conversationID, name+" is no longer an admin")
}
return nil
}
// groupAvatarPrefix namespaces a group's avatar objects.
func groupAvatarPrefix(conversationID string) string {
return path.Join("chat", conversationID, "avatar") + "/"
}
// PresignGroupAvatar issues an upload URL for a new group picture.
//
// The key is namespaced under this conversation's avatar prefix, which is what
// lets UpdateGroup verify the key being set was uploaded for THIS group.
func (s *ChatService) PresignGroupAvatar(ctx context.Context, userID, conversationID, contentType string) (uploadURL, key string, err error) {
if _, _, err := s.requireGroupAdmin(ctx, userID, conversationID); err != nil {
return "", "", err
}
if kind, ok := attachmentKindFor(contentType); !ok || kind != model.AttachmentImage {
return "", "", ErrChatBadMedia
}
if s.r2 == nil || !s.r2.PrivateConfigured() {
return "", "", ErrChatMediaOff
}
suffix := make([]byte, 12)
if _, err := rand.Read(suffix); err != nil {
return "", "", err
}
key = groupAvatarPrefix(conversationID) + hex.EncodeToString(suffix)
uploadURL, err = s.r2.PresignPutPrivate(ctx, key, contentType, 5*time.Minute)
if err != nil {
return "", "", err
}
return uploadURL, key, nil
}
// GroupAvatarURL mints a short-lived read URL for a group's picture.
//
// Members only: a group's photo is group content, so it lives in the private
// bucket behind the same membership check as everything else in the thread.
func (s *ChatService) GroupAvatarURL(ctx context.Context, userID, conversationID string) (string, error) {
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return "", err
}
conv, err := s.chat.Conversation(ctx, conversationID)
if err != nil {
return "", err
}
if conv.AvatarKey == nil || *conv.AvatarKey == "" {
return "", ErrAvatarUnset
}
if s.r2 == nil || !s.r2.PrivateConfigured() {
return "", ErrChatMediaOff
}
return s.r2.PresignGet(ctx, *conv.AvatarKey, 5*time.Minute)
}
// UpdateGroup renames a group, changes who may post in it, or sets its photo.
func (s *ChatService) UpdateGroup(ctx context.Context, userID, conversationID string, title, sendPolicy, avatarKey *string) error {
if _, _, err := s.requireGroupAdmin(ctx, userID, conversationID); err != nil {
return err
}
if sendPolicy != nil && *sendPolicy != model.SendPolicyEveryone && *sendPolicy != model.SendPolicyAdmins {
return ErrChatBadPolicy
}
if title != nil {
trimmed := strings.TrimSpace(*title)
if trimmed == "" {
return errors.New("group title is required")
}
title = &trimmed
}
if avatarKey != nil {
// The key must sit under THIS group's avatar prefix. Without the check
// an admin could point avatar_key at another thread's media, and every
// member here would be served a file from a conversation they are not
// in.
if !strings.HasPrefix(*avatarKey, groupAvatarPrefix(conversationID)) {
return ErrChatBadAvatarKey
}
}
if err := s.chat.UpdateGroup(ctx, conversationID, title, sendPolicy, avatarKey); err != nil {
return err
}
if avatarKey != nil {
s.systemMessage(ctx, conversationID, "The group photo was updated")
}
if sendPolicy != nil {
if *sendPolicy == model.SendPolicyAdmins {
s.systemMessage(ctx, conversationID, "Only admins can send messages now")
} else {
s.systemMessage(ctx, conversationID, "Everyone can send messages now")
}
}
return nil
}
// assertCanPost applies the per-kind posting rules.
func (s *ChatService) assertCanPost(ctx context.Context, userID, conversationID string) error {
conv, err := s.chat.Conversation(ctx, conversationID)
if err != nil {
return err
}
switch conv.Kind {
case model.ConversationAI:
// Kautilya is still coming soon: the thread and its tab exist so the
// plumbing is in place, but nothing consumes messages yet, so refuse
// rather than silently swallow them.
return ErrChatReadOnly
case model.ConversationDM:
// Either direction of a block closes a DM. A one-way channel — where
// the blocker can still be written to — would be worse than refusing.
ids, err := s.chat.ParticipantIDs(ctx, conversationID)
if err != nil {
return err
}
for _, id := range ids {
if id == userID {
continue
}
blocked, blockedBy, err := s.blocks.BlockState(ctx, userID, id)
if err != nil {
return fmt.Errorf("block state: %w", err)
}
if blocked || blockedBy {
return ErrChatBlocked
}
}
return nil
default:
// Groups: blocks are enforced when members are added, not on every
// send — one blocked pair shouldn't silence the whole room. What does
// gate a send is the group's own policy.
if conv.SendPolicy == model.SendPolicyAdmins {
me, err := s.chat.Participant(ctx, conversationID, userID)
if err != nil {
return err
}
if !isAdmin(me.Role) {
return ErrChatAdminsOnly
}
}
return nil
}
}
// MarkRead advances the caller's read cursor and tells the other participants,
// so their ticks turn blue without polling.
func (s *ChatService) MarkRead(ctx context.Context, userID, conversationID string, seq int64) error {
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return err
}
// The cursor still advances — it is what clears the caller's own unread
// count. What changes is whether anyone else is told.
if err := s.chat.MarkRead(ctx, conversationID, userID, seq); err != nil {
return err
}
mine, err := s.chat.Settings(ctx, userID)
if err != nil {
return err
}
if mine.ReadReceipts {
s.publishReceipt(ctx, conversationID, userID, "read", seq)
}
return nil
}
// MarkDelivered advances the delivery cursor only.
func (s *ChatService) MarkDelivered(ctx context.Context, userID, conversationID string, seq int64) error {
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return err
}
if err := s.chat.MarkDelivered(ctx, conversationID, userID, seq); err != nil {
return err
}
s.publishReceipt(ctx, conversationID, userID, "delivered", seq)
return nil
}
// MuteDuration values accepted by SetMute.
const (
MuteEightHours = "8h"
MuteOneWeek = "1w"
MuteAlways = "always"
MuteOff = "off"
)
// muteForever is the timestamp stored for an indefinite mute. A sentinel date
// rather than a nullable-plus-flag pair keeps every read a single comparison
// (muted_until > now), and Postgres `infinity` doesn't scan cleanly into
// time.Time.
var muteForever = time.Date(9999, 12, 31, 0, 0, 0, 0, time.UTC)
// SetMute silences a thread for the caller alone.
//
// Muting suppresses alerts, not delivery: unread counts keep accruing and the
// thread keeps updating live, which is what users expect from a muted chat.
func (s *ChatService) SetMute(ctx context.Context, userID, conversationID, duration string) (*time.Time, error) {
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return nil, err
}
var until *time.Time
switch duration {
case MuteOff:
until = nil
case MuteEightHours:
t := time.Now().Add(8 * time.Hour)
until = &t
case MuteOneWeek:
t := time.Now().Add(7 * 24 * time.Hour)
until = &t
case MuteAlways:
t := muteForever
until = &t
default:
return nil, ErrChatBadMuteDuration
}
if err := s.chat.SetMute(ctx, conversationID, userID, until); err != nil {
return nil, err
}
return until, nil
}
// IsMuted reports whether the caller currently has the thread muted, expiring
// a lapsed mute without needing a sweep job.
func (s *ChatService) IsMuted(ctx context.Context, userID, conversationID string) (bool, error) {
p, err := s.chat.Participant(ctx, conversationID, userID)
if err != nil {
return false, err
}
return p.MutedUntil != nil && p.MutedUntil.After(time.Now()), nil
}
// HeartEmoji is what a double-tap sends, and the one reaction with its own
// notification copy.
const HeartEmoji = "❤️"
// maxEmojiRunes bounds a reaction. Generous enough for a flag or a skin-tone
// sequence, small enough that the column can't be used as free text storage.
const maxEmojiRunes = 8
// MessageReactions lists who reacted to a message and with what.
//
// Authorized against the message's OWN conversation, so a message id alone
// never reveals who is in a thread the caller isn't part of.
func (s *ChatService) MessageReactions(ctx context.Context, userID, messageID string) ([]model.ReactionDetail, error) {
conversationID, _, err := s.chat.MessageConversation(ctx, messageID)
if err != nil {
return nil, err
}
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return nil, err
}
return s.chat.ReactionDetails(ctx, messageID)
}
// Message returns one hydrated message for this viewer.
//
// Exists because reaction counts carry a per-viewer `mine` flag, so a client
// told "message X changed" cannot compute the new chips itself — it has to ask.
func (s *ChatService) Message(ctx context.Context, userID, conversationID, messageID string) (*model.Message, error) {
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return nil, err
}
msg, err := s.chat.Message(ctx, conversationID, messageID, userID)
if err != nil {
return nil, err
}
return s.withAttachments(ctx, userID, msg)
}
// Stickers lists what the caller can send: sādhana rewards they earned, plus
// stickers they saved from a chat.
func (s *ChatService) Stickers(ctx context.Context, userID string) ([]model.Sticker, error) {
return s.chat.Stickers(ctx, userID)
}
// SaveSticker keeps one someone sent, after checking it really is on that
// message and the caller is in the thread.
func (s *ChatService) SaveSticker(ctx context.Context, userID, messageID, key string) error {
return s.chat.SaveSticker(ctx, userID, key, messageID)
}
// attachSticker stores the artwork on a message.
//
// Ownership is checked here, not trusted from the client: the key comes over
// the wire, and without this anyone could send any sticker in the catalog by
// guessing its key. The image and name are denormalized because the catalog
// is static art — there is nothing viewer-dependent to re-resolve on read.
func (s *ChatService) attachSticker(ctx context.Context, conversationID, userID string, msg *model.Message, key string) error {
sticker, err := s.chat.Sticker(ctx, userID, key)
if err != nil {
return err
}
entityType := model.EntitySticker
if _, err := s.chat.AddAttachment(ctx, repository.AddAttachmentParams{
MessageID: msg.ID, ConversationID: conversationID, Ord: 0,
Kind: model.AttachmentSticker, EntityType: &entityType,
EntityID: &sticker.Key, ImageURL: &sticker.ImageURL, Title: &sticker.Name,
}); err != nil {
s.logger.Warn("chat: attach sticker failed", zap.Error(err))
}
return nil
}
// Typing broadcasts that someone is composing.
//
// Never persisted and never pushed: a typing indicator is only meaningful to
// someone already looking at the thread, and it is stale within seconds.
// Publishing failures are ignored for the same reason.
func (s *ChatService) Typing(ctx context.Context, userID, conversationID string) error {
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return err
}
if s.rt == nil {
return nil
}
if err := s.rt.Trigger(ctx, ConversationChannel(conversationID), "typing", map[string]any{
"user_id": userID,
}); err != nil {
s.logger.Debug("chat: typing publish failed", zap.Error(err))
}
return nil
}
// React sets the caller's emoji on a message, replacing any previous one.
//
// Authorization is by membership of the message's OWN conversation, read back
// from the insert — a message id alone must not let someone react into a
// thread they were never in.
func (s *ChatService) React(ctx context.Context, userID, messageID, emoji string) error {
emoji = strings.TrimSpace(emoji)
if emoji == "" || len([]rune(emoji)) > maxEmojiRunes {
return ErrChatBadEmoji
}
conversationID, author, err := s.chat.MessageConversation(ctx, messageID)
if err != nil {
return err
}
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return err
}
if _, err := s.chat.SetReaction(ctx, messageID, userID, emoji); err != nil {
return err
}
s.publishReaction(ctx, conversationID, messageID)
s.notifyReaction(ctx, conversationID, author, userID, emoji)
return nil
}
// Unreact removes the caller's reaction.
func (s *ChatService) Unreact(ctx context.Context, userID, messageID string) error {
conversationID, _, err := s.chat.MessageConversation(ctx, messageID)
if err != nil {
return err
}
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return err
}
if _, err := s.chat.ClearReaction(ctx, messageID, userID); err != nil {
// Nothing to remove is the desired end state, not a failure.
if errors.Is(err, repository.ErrMessageNotHere) {
return nil
}
return err
}
s.publishReaction(ctx, conversationID, messageID)
return nil
}
// publishReaction tells the open thread to refresh one message's chips. The
// event carries only the id: counts depend on the viewer, so each client
// re-reads its own.
func (s *ChatService) publishReaction(ctx context.Context, conversationID, messageID string) {
if s.rt == nil {
return
}
if err := s.rt.Trigger(ctx, ConversationChannel(conversationID), "reaction", map[string]any{
"message_id": messageID,
}); err != nil {
s.logger.Debug("chat: reaction publish failed", zap.Error(err))
}
}
// notifyReaction pushes to the message's author.
//
// Never for reacting to your own message, and it reuses the chat push rules —
// mute, the Messages preference, presence and the debounce — so a reaction
// cannot become a way around them.
func (s *ChatService) notifyReaction(ctx context.Context, conversationID string, author *string, actorID, emoji string) {
if s.push == nil || !s.push.Enabled() {
return
}
// No author on system and assistant messages, and reacting to your own
// message must never notify you.
if author == nil || *author == actorID {
return
}
conv, err := s.chat.Conversation(ctx, conversationID)
if err != nil {
return
}
actor := "Someone"
if s.users != nil {
if u, uErr := s.users.FindByID(ctx, actorID); uErr == nil && u != nil && u.Name != "" {
actor = u.Name
}
}
body := actor + " reacted " + emoji + " to your message"
if emoji == HeartEmoji {
body = actor + " liked your message"
}
s.pushTo(ctx, conv, *author, actor, body)
}
// Receipts reports how far EVERY other participant has got, which is what
// draws the caller's tick states.
//
// Read receipts are reciprocal, as they are everywhere else this feature
// exists: someone who switches them off stops broadcasting their own reads AND
// stops seeing other people's. A one-way version would let a user watch others
// while staying invisible. Delivery ticks are unaffected — they say the device
// received the message, not that a person read it.
func (s *ChatService) Receipts(ctx context.Context, userID, conversationID string) (model.Receipts, error) {
if _, err := s.chat.Participant(ctx, conversationID, userID); err != nil {
return model.Receipts{}, err
}
receipts, err := s.chat.Receipts(ctx, conversationID, userID)
if err != nil {
return model.Receipts{}, err
}
mine, err := s.chat.Settings(ctx, userID)
if err != nil {
return model.Receipts{}, err
}
if !mine.ReadReceipts {
receipts.ReadSeq = 0
return receipts, nil
}
// A read cursor only counts if that participant shares theirs. If any of
// them doesn't, the message can never be "read by everyone", which is what
// the blue tick means here.
others, err := s.chat.ParticipantIDs(ctx, conversationID)
if err != nil {
return model.Receipts{}, err
}
enabled, err := s.chat.ReadReceiptsEnabled(ctx, others)
if err != nil {
return model.Receipts{}, err
}
for _, id := range others {
if id != userID && !enabled[id] {
receipts.ReadSeq = 0
break
}
}
return receipts, nil
}
// publish fans a new message out to the open thread, and separately nudges each
// other participant's inbox.
func (s *ChatService) publish(ctx context.Context, conversationID, senderID string, msg *model.Message) {
// Out-of-band alerts for anyone not currently connected. Best-effort and
// after the realtime fan-out, so a slow push never delays delivery to
// people who are online.
if conv, err := s.chat.Conversation(ctx, conversationID); err == nil {
s.notifyNewMessage(ctx, conv, senderID, msg)
}
if s.rt == nil {
return
}
if err := s.rt.Trigger(ctx, ConversationChannel(conversationID), "message", msg); err != nil {
s.logger.Warn("chat: publish message failed",
zap.String("conversation_id", conversationID), zap.Error(err))
}
ids, err := s.chat.ParticipantIDs(ctx, conversationID)
if err != nil {
s.logger.Warn("chat: participant fan-out failed", zap.Error(err))
return
}
for _, id := range ids {
if id == senderID {
continue
}
// Deliberately NOT TriggerUser: that also publishes to the legacy
// PUBLIC notifications channel, which would expose who is messaging
// whom to anyone subscribed to it. Chat only ever uses the private one.
//
// Deliberately NOT the notification service either: chat activity
// highlights the thread in the inbox and must not create a
// notifications row or appear in the tray.
if err := s.rt.Trigger(ctx, realtime.UserChannel(id), "chat_activity", map[string]any{
"conversation_id": conversationID,
"seq": msg.Seq,
"sender_id": senderID,
}); err != nil {
s.logger.Debug("chat: activity nudge failed", zap.String("user_id", id), zap.Error(err))
}
}
}
// pushConfig tunes out-of-band delivery.
const (
// One push per thread per window, however many messages arrive. A burst
// should buzz a pocket once.
pushDebounceWindow = 90 * time.Second
// How long a live WebSocket connection suppresses push for.
presenceTTL = 2 * time.Minute
// Lock-screen previews are deliberately senderless of content — see
// notifyNewMessage.
maxPushPreviewRunes = 120
)
// presenceKey marks "this user has a live realtime connection".
func presenceKey(userID string) string { return "chat:online:" + userID }
// MarkPresence records that a user's realtime channel became occupied or
// vacated. Driven by Soketi's webhooks.
func (s *ChatService) MarkPresence(ctx context.Context, userID string, online bool) {
if s.cache == nil {
return
}
if online {
s.cache.Client.Set(ctx, presenceKey(userID), "1", presenceTTL)
return
}
s.cache.Client.Del(ctx, presenceKey(userID))
}
// notifyNewMessage delivers a Web Push alert to participants who are not
// currently connected.
//
// Deliberately NOT routed through NotificationService: chat must not create a
// notifications row or land in the tray. It borrows only the preference matrix
// so "Messages → Push" can be turned off like anything else.
//
// Four things suppress a push, in the order that costs least to check:
// the thread is muted, the user opted out, they have a live connection, or one
// already went out for this thread recently.
func (s *ChatService) notifyNewMessage(ctx context.Context, conv *model.Conversation, senderID string, msg *model.Message) {
if s.push == nil || !s.push.Enabled() {
return
}
candidates, err := s.chat.PushCandidates(ctx, conv.ID, senderID)
if err != nil {
s.logger.Warn("chat: push candidates failed", zap.Error(err))
return
}
title, body := s.pushCopy(ctx, conv, senderID, msg)
now := time.Now()
for _, c := range candidates {
if c.MutedUntil != nil && c.MutedUntil.After(now) {
continue
}
// First contact: push only reaches a device that has already opted
// in, which someone who has never used chat almost certainly hasn't.
// Email is the one channel that reaches them — and only the FIRST
// message qualifies, since after that the thread exists and push
// carries it.
if msg.Seq == 1 && !s.online(ctx, c.UserID) {
s.emailFirstContact(ctx, c.UserID, senderID, conv)
}
s.pushTo(ctx, conv, c.UserID, title, body)
}
}
// pushTo delivers one chat notification, applying the suppression rules every
// chat push obeys: the Messages preference, a live connection, and the
// per-thread debounce. Reactions go through here too, so they cannot become a
// way around any of them.
//
// Mute is checked by callers, since they already hold the participant row.
func (s *ChatService) pushTo(ctx context.Context, conv *model.Conversation, userID, title, body string) {
if s.push == nil || !s.push.Enabled() {
return
}
if s.prefs != nil {
if off, err := s.prefs.DisabledSet(ctx, userID); err == nil &&
prefOff(off, NotifChatMessage, repository.ChannelPush) {
return
}
}
if s.online(ctx, userID) {
return
}
if !s.claimPushSlot(ctx, userID, conv.ID) {
return
}
s.push.SendToUser(ctx, userID, PushPayload{
Title: title,
Body: body,
URL: "/chats/" + conv.ID,
// Same tag per thread: a later notification replaces the earlier one
// rather than stacking up.
Tag: "chat:" + conv.ID,
})
}
// emailFirstContact tells someone a conversation has been opened with them,
// when nothing else would reach them.
//
// Deliberately says nothing about the message itself: an email subject line is
// the least private surface in the product, and "you have a message" is enough
// to bring someone back.
func (s *ChatService) emailFirstContact(ctx context.Context, recipientID, senderID string, conv *model.Conversation) {
if s.mail == nil || s.users == nil {
return
}
// The same preference as chat push, on its email channel — so turning
// Messages off silences both rather than leaving one leaking through.
if s.prefs != nil {
if off, err := s.prefs.DisabledSet(ctx, recipientID); err == nil &&
prefOff(off, NotifChatMessage, repository.ChannelEmail) {
return
}
}
recipient, err := s.users.FindByID(ctx, recipientID)
if err != nil || recipient == nil ||
recipient.Email == nil || strings.TrimSpace(*recipient.Email) == "" {
// OAuth-only accounts may have no address on file.
return
}
sender, err := s.users.FindByID(ctx, senderID)
if err != nil || sender == nil {
return
}
var buf bytes.Buffer
if err := firstContactEmailTpl.Execute(&buf, map[string]any{
"RecipientName": recipient.Name,
"SenderName": actorDisplay(sender),
"SenderHandle": sender.Handle,
"ChatURL": s.frontendURL + "/chats/" + conv.ID,
}); err != nil {
s.logger.Warn("chat: first-contact email render failed", zap.Error(err))
return
}
subject := actorDisplay(sender) + " sent you a message — Prabhupāda Says"
if err := s.mail.Send(strings.TrimSpace(*recipient.Email), subject, buf.String()); err != nil {
s.logger.Warn("chat: first-contact email failed",
zap.String("recipient", recipientID), zap.Error(err))
}
}
// Styled to match the follow-request email so the two read as one product.
var firstContactEmailTpl = template.Must(template.New("firstContact").Parse(`<div style="font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,Helvetica,Arial,sans-serif;color:#2a2118;max-width:560px;margin:0 auto;font-size:17px;line-height:1.7">
<p style="font-size:18px;color:#9a6b2f;letter-spacing:0.04em;margin-bottom:1em">Hare Kṛṣṇa{{if .RecipientName}}, {{.RecipientName}}{{end}} 🙏</p>
<p><strong>{{.SenderName}}</strong> (@{{.SenderHandle}}) has sent you a message.
Reply to them on Prabhupāda Says.</p>
<p style="margin:1.6em 0">
<a href="{{.ChatURL}}" style="background:#9a6b2f;color:#fff;text-decoration:none;padding:12px 22px;border-radius:8px;font-weight:600;display:inline-block">Open the conversation</a>
</p>
<p style="font-size:14px;color:#7a6a55">You are receiving this because someone started a conversation with you.
You can turn these off under Settings → Notifications.</p>
</div>`))
// pushCopy builds the lock-screen text.
func (s *ChatService) pushCopy(ctx context.Context, conv *model.Conversation, senderID string, msg *model.Message) (title, body string) {
sender := "Someone"
if s.users != nil {
if u, err := s.users.FindByID(ctx, senderID); err == nil && u != nil {
if u.Name != "" {
sender = u.Name
} else if u.Handle != "" {
sender = "@" + u.Handle
}
}
}
title = sender
if conv.Kind == model.ConversationGroup && conv.Title != nil && *conv.Title != "" {
title = sender + " · " + *conv.Title
}
switch {
case strings.TrimSpace(msg.Body) != "":
body = truncateRunes(strings.TrimSpace(msg.Body), maxPushPreviewRunes)
case len(msg.Attachments) > 0:
body = "Sent an attachment"
default:
body = "Sent you a message"
}
return title, body
}
// online reports whether the user holds a live realtime connection.
//
// Best-effort and fails OPEN: if the cache is unavailable — or Soketi is not
// configured to send channel_occupied — we treat them as offline and push. A
// redundant notification while they are reading is a smaller failure than a
// message that never arrives.
func (s *ChatService) online(ctx context.Context, userID string) bool {
if s.cache == nil {
return false
}
n, err := s.cache.Client.Exists(ctx, presenceKey(userID)).Result()
return err == nil && n > 0
}
// claimPushSlot returns true for the first push to a thread within the window.
// Fails OPEN for the same reason as online().
func (s *ChatService) claimPushSlot(ctx context.Context, userID, conversationID string) bool {
if s.cache == nil {
return true
}
ok, err := s.cache.Client.SetNX(ctx,
"chat:pushed:"+userID+":"+conversationID, "1", pushDebounceWindow).Result()
if err != nil {
return true
}
return ok
}
func truncateRunes(s string, n int) string {
r := []rune(s)
if len(r) <= n {
return s
}
return strings.TrimSpace(string(r[:n])) + "…"
}
func (s *ChatService) publishReceipt(ctx context.Context, conversationID, userID, kind string, seq int64) {
if s.rt == nil {
return
}
if err := s.rt.Trigger(ctx, ConversationChannel(conversationID), "receipt", map[string]any{
"user_id": userID, "kind": kind, "seq": seq,
}); err != nil {
s.logger.Debug("chat: receipt publish failed", zap.Error(err))
}
}