spsays-bff/internal/service/chat_service.go

package service

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"path"
	"strings"
	"time"

	"github.com/prabhupadasays/spsays-bff/internal/model"
	"github.com/prabhupadasays/spsays-bff/internal/repository"
	"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")
	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
	rt     *realtime.SoketiClient
	r2     *repository.R2
	links  *unfurl.Fetcher
	cache  *repository.Redis
	logger *zap.Logger
}

func NewChatService(
	chat *repository.ChatRepo,
	blocks *repository.BlockRepo,
	rt *realtime.SoketiClient,
	r2 *repository.R2,
	cache *repository.Redis,
	logger *zap.Logger,
) *ChatService {
	return &ChatService{
		chat: chat, blocks: blocks, 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)
}

// 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, msgs)
}

// hydrateAttachments fills in each message's attachments in one query.
func (s *ChatService) hydrateAttachments(ctx context.Context, 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
	}
	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{}
		}
	}
	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, 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
	// 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
	}

	body := strings.TrimSpace(req.Body)
	// A photo with no caption is an ordinary message; only a message with
	// neither text nor attachments is empty.
	if body == "" && len(req.Attachments) == 0 {
		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.withAttachments(ctx, msg)
	}

	s.attachMedia(ctx, conversationID, userID, msg, req.Attachments)
	s.attachLinkPreview(ctx, conversationID, msg, entities, req.PreviewURL)

	full, err := s.withAttachments(ctx, msg)
	if err != nil {
		full = msg
	}
	s.publish(ctx, conversationID, userID, full)
	return full, nil
}

// withAttachments hydrates a message's attachments for the response.
func (s *ChatService) withAttachments(ctx context.Context, 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{}
	}
	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))
		}
	}
}

// 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 {
		full, hErr := s.withAttachments(ctx, msg)
		if hErr == nil {
			msg = full
		}
		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 {
		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
}

// 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) {
	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))
		}
	}
}

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))
	}
}