libredesk/internal/conversation/message.go

package conversation

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"html"
	"net/mail"
	"path/filepath"
	"regexp"
	"slices"
	"strings"
	"text/template"
	"time"

	"github.com/abhinavxd/libredesk/internal/attachment"
	amodels "github.com/abhinavxd/libredesk/internal/automation/models"
	"github.com/abhinavxd/libredesk/internal/conversation/models"
	"github.com/abhinavxd/libredesk/internal/envelope"
	"github.com/abhinavxd/libredesk/internal/image"
	"github.com/abhinavxd/libredesk/internal/inbox"
	"github.com/abhinavxd/libredesk/internal/inbox/channel/livechat"
	whatsappChannel "github.com/abhinavxd/libredesk/internal/inbox/channel/whatsapp"
	imodels "github.com/abhinavxd/libredesk/internal/inbox/models"
	mmodels "github.com/abhinavxd/libredesk/internal/media/models"
	"github.com/abhinavxd/libredesk/internal/sla"
	"github.com/abhinavxd/libredesk/internal/stringutil"
	tmpl "github.com/abhinavxd/libredesk/internal/template"
	umodels "github.com/abhinavxd/libredesk/internal/user/models"
	wmodels "github.com/abhinavxd/libredesk/internal/webhook/models"
	"github.com/jmoiron/sqlx"
	"github.com/lib/pq"
	"github.com/volatiletech/null/v9"
)

const (
	maxMessagesPerPage = 500
	// Only allow visitor-to-contact upgrade within this window after the last continuity email.
	upgradeWindowTTL    = 7 * 24 * time.Hour
	maxPreviousMessages = 10
)

// Matches <img ... src="URL"> and captures the URL for downstream parsing.
var imgSrcPattern = regexp.MustCompile(`(?i)<img\b[^>]*?\bsrc=["']([^"']*)["']`)

// fromNameVars is the template context for an inbox's from-name template.
type fromNameVars struct {
	Agent fromNameAgent
	Inbox fromNameInbox
}

type fromNameAgent struct{ FirstName, LastName, FullName string }

type fromNameInbox struct{ Name string }

// previousMessage is one earlier message exposed to the outgoing email template. String fields are HTML-escaped.
type previousMessage struct {
	SenderName  string
	SenderEmail string
	CreatedAt   time.Time
	IsIncoming  bool
	Content     string
}

type previousMessageRow struct {
	CreatedAt   time.Time `db:"created_at"`
	IsIncoming  bool      `db:"is_incoming"`
	Content     string    `db:"content"`
	TextContent string    `db:"text_content"`
	ContentType string    `db:"content_type"`
	FirstName   string    `db:"first_name"`
	LastName    string    `db:"last_name"`
	Email       string    `db:"email"`
}

// Run starts a pool of worker goroutines to handle message dispatching via inbox's channel and processes incoming messages. It scans for
// pending outgoing messages at the specified read interval and pushes them to the outgoing queue to be sent.
func (m *Manager) Run(ctx context.Context, incomingQWorkers, outgoingQWorkers, scanInterval time.Duration) {
	dbScanner := time.NewTicker(scanInterval)
	defer dbScanner.Stop()

	for range outgoingQWorkers {
		m.wg.Add(1)
		go func() {
			defer m.wg.Done()
			m.MessageSenderWorker(ctx)
		}()
	}
	for range incomingQWorkers {
		m.wg.Add(1)
		go func() {
			defer m.wg.Done()
			m.IncomingMessageWorker(ctx)
		}()
	}

	// Scan pending outgoing messages and send them.
	for {
		select {
		case <-ctx.Done():
			return
		case <-dbScanner.C:
			pendingMessages := []models.Message{}
			processingConversationIDs := m.getOutgoingProcessingConversationIDs()

			if err := m.q.GetOutgoingPendingMessages.Select(&pendingMessages, pq.Array(processingConversationIDs)); err != nil {
				m.lo.Error("error fetching pending messages from db", "error", err)
				continue
			}

			for _, message := range pendingMessages {
				m.outgoingProcessingMessages.Store(message.ID, message.ConversationID)
				m.outgoingMessageQueue <- message
			}
		}
	}
}

// Close signals the Manager to stop processing messages, closes channels,
// and waits for all worker goroutines to finish processing.
func (m *Manager) Close() {
	m.closedMu.Lock()
	defer m.closedMu.Unlock()
	m.closed = true
	close(m.outgoingMessageQueue)
	close(m.incomingMessageQueue)
	m.wg.Wait()
}

// IncomingMessageWorker processes incoming messages from the incoming message queue.
func (m *Manager) IncomingMessageWorker(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		case msg, ok := <-m.incomingMessageQueue:
			if !ok {
				return
			}
			if _, err := m.ProcessIncomingMessage(msg); err != nil {
				m.lo.Error("error processing incoming msg", "error", err)
			}
		}
	}
}

// MessageSenderWorker sends outgoing pending messages.
func (m *Manager) MessageSenderWorker(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		case message, ok := <-m.outgoingMessageQueue:
			if !ok {
				return
			}
			m.sendOutgoingMessage(message)
		}
	}
}

// sendOutgoingMessage sends an outgoing message.
func (m *Manager) sendOutgoingMessage(message models.Message) {
	defer m.outgoingProcessingMessages.Delete(message.ID)

	// Helper function to handle errors
	handleError := func(err error, errorMsg string) bool {
		if err != nil {
			m.lo.Error(errorMsg, "error", err, "message_id", message.ID)
			m.UpdateMessageStatus(message.UUID, models.MessageStatusFailed)
			return true
		}
		return false
	}

	// Get inbox
	inb, err := m.inboxStore.Get(message.InboxID)
	if handleError(err, "error fetching inbox") {
		return
	}

	// Render content in template
	if err := m.RenderMessageInTemplate(inb.Channel(), &message); err != nil {
		handleError(err, "error rendering content in template")
		return
	}

	// Attach attachments to the message
	if err := m.attachAttachmentsToMessage(&message); err != nil {
		handleError(err, "error attaching attachments to message")
		return
	}

	// Convert to OutboundMessage for transport
	outbound := message.ToOutbound()

	if inb.Channel() == inbox.ChannelEmail {
		inboxRecord, err := m.inboxStore.GetDBRecord(message.InboxID)
		if handleError(err, "error fetching inbox sender addresses") {
			return
		}
		selected := message.SendFrom()
		if selected == "" {
			selected = inboxRecord.From
		}
		selected, err = m.resolveSendFrom(message.ConversationUUID, inboxRecord, selected)
		if handleError(err, "invalid email sender") {
			return
		}
		outbound.From = m.emailFromAddress(inb, message, selected)
		if len(messageArticleIDs(message.Meta)) > 0 {
			outbound.AltContent = stringutil.HTML2TextMarkdownLinks(message.Content)
		}

		// Emails with no subject (e.g. arrived via IMAP with an empty Subject header) are
		// rejected by some SMTP servers. Fall back to the reference number so replies go through.
		if outbound.Subject == "" {
			outbound.Subject = m.formatRefMarker(message.ReferenceNumber)
		}

		// Set "In-Reply-To" and "References" headers for email threading.
		outbound.References, outbound.InReplyTo = m.BuildEmailThreadingHeaders(message.ConversationID, outbound.SourceID)
	}

	// Send message
	err = inb.Send(outbound)
	if err != nil && err != livechat.ErrClientNotConnected {
		if inb.Channel() == inbox.ChannelWhatsApp {
			m.RecordWhatsAppSendFailure(message.UUID, err.Error())
		}
		handleError(err, "error sending message")
		return
	}

	// Update status as sent.
	m.UpdateMessageStatus(message.UUID, models.MessageStatusSent)

	// Skip system user replies since we only update timestamps and SLA for human replies.
	systemUser, err := m.userStore.GetSystemUser()
	if err != nil {
		m.lo.Error("error fetching system user", "error", err)
		return
	}
	if message.SenderID != systemUser.ID {
		conversation, err := m.GetConversation(message.ConversationID, "", "")
		if err != nil {
			m.lo.Error("error fetching conversation", "conversation_id", message.ConversationID, "error", err)
			return
		}

		now := time.Now()
		nowStr := now.Format(time.RFC3339)
		wsData := map[string]any{"last_reply_at": nowStr, "waiting_since": nil}

		var isFirstReply bool
		if err := m.q.UpdateConversationReplyTimestamps.QueryRow(message.ConversationID, now).Scan(&isFirstReply); err != nil {
			m.lo.Error("error updating conversation reply timestamps", "error", err)
		} else if isFirstReply {
			wsData["first_reply_at"] = nowStr
			// Stamp the first-response SLA immediately.
			if err := m.slaStore.EvaluateConversationSLA(message.ConversationID); err != nil {
				m.lo.Error("error evaluating SLA after first reply", "conversation_id", message.ConversationID, "error", err)
			}
		}

		// Mark latest SLA event for next response metric as met.
		metAt, err := m.slaStore.SetLatestSLAEventMetAt(conversation.AppliedSLAID.Int, sla.MetricNextResponse)
		if err != nil && !errors.Is(err, sla.ErrLatestSLAEventNotFound) {
			m.lo.Error("error setting next response SLA event `met_at`", "conversation_id", conversation.ID, "metric", sla.MetricNextResponse, "applied_sla_id", conversation.AppliedSLAID.Int, "error", err)
		} else if !metAt.IsZero() {
			wsData["next_response_met_at"] = metAt.Format(time.RFC3339)
		}

		m.BroadcastConversationUpdate(message.ConversationUUID, wsData)

		if message.ShouldEvaluateAutomation(systemUser.ID) {
			m.automation.EvaluateConversationUpdateRulesByID(message.ConversationID, "", amodels.EventConversationMessageOutgoing, umodels.User{ID: message.SenderID})
		}
	}
}

// BuildTemplateData builds the common template data map for rendering message content variables.
func (m *Manager) BuildTemplateData(conversationUUID string, senderID int) (map[string]any, error) {
	conversation, err := m.GetConversation(0, conversationUUID, "")
	if err != nil {
		return nil, fmt.Errorf("fetching conversation: %w", err)
	}

	sender, err := m.userStore.GetAgentCachedOrLoad(senderID)
	if err != nil {
		return nil, fmt.Errorf("fetching message sender user: %w", err)
	}

	recipient := umodels.User{
		FirstName: conversation.Contact.FirstName,
		LastName:  conversation.Contact.LastName,
		Email:     conversation.Contact.Email,
	}
	data := tmpl.BuildConversationData(conversation, recipient, sender)
	data["Recipient"] = data["Contact"]
	return data, nil
}

func (m *Manager) RenderMessageInTemplate(channel string, message *models.Message) error {
	switch channel {
	case inbox.ChannelEmail:
		data, err := m.BuildTemplateData(message.ConversationUUID, message.SenderID)
		if err != nil {
			return err
		}

		content := m.emailTemplateContent(message, data)
		isContinuity, _ := data["IsContinuityEmail"].(bool)
		if !isContinuity && m.template.DefaultOutgoingEmailTemplateContains("PreviousMessages") {
			data["PreviousMessages"] = m.previousMessagesData(message.ConversationID, message.ID)
		}
		message.Content, err = m.template.RenderEmailWithTemplate(data, content)
		if err != nil {
			m.lo.Error("could not render email content using template", "id", message.ID, "error", err)
			return fmt.Errorf("could not render email content using template: %w", err)
		}
	case inbox.ChannelLiveChat, inbox.ChannelWhatsApp:
		m.RenderArticleReferences(message)
		return nil
	default:
		m.lo.Warn("unknown message channel", "channel", channel)
		return fmt.Errorf("unknown message channel: %s", channel)
	}
	return nil
}

func (m *Manager) emailTemplateContent(message *models.Message, data map[string]any) string {
	content := articleCitationRegexp.ReplaceAllString(message.Content, "")
	var (
		isContinuity bool
		meta         map[string]any
	)
	if len(message.Meta) > 0 && json.Unmarshal(message.Meta, &meta) == nil {
		isContinuity, _ = meta["continuity_email"].(bool)
	}
	data["IsContinuityEmail"] = isContinuity
	if isContinuity {
		data["ContinuityContent"] = content
		return "{{ .ContinuityContent }}"
	}
	if references := m.emailArticleReferences(message); references != "" {
		data["ArticleReferences"] = references
		return content + "{{ .ArticleReferences }}"
	}
	return content
}

// previousMessagesData returns the conversation's earlier messages exchanged with the contact, newest first, with incoming reply chains removed.
func (m *Manager) previousMessagesData(conversationID, messageID int) []previousMessage {
	out := []previousMessage{}
	var rows []previousMessageRow
	if err := m.q.GetPreviousMessages.Select(&rows, conversationID, messageID, maxPreviousMessages); err != nil {
		m.lo.Error("error fetching previous messages", "conversation_id", conversationID, "error", err)
		return out
	}
	for _, r := range rows {
		text := strings.TrimSpace(r.TextContent)
		if r.IsIncoming {
			text = textWithoutQuotedReplies(models.Message{Content: r.Content, TextContent: r.TextContent, ContentType: r.ContentType})
		}
		if text == "" {
			continue
		}
		sender := umodels.User{FirstName: r.FirstName, LastName: r.LastName}
		out = append(out, previousMessage{
			SenderName:  html.EscapeString(strings.TrimSpace(sender.FullName())),
			SenderEmail: html.EscapeString(r.Email),
			CreatedAt:   r.CreatedAt,
			IsIncoming:  r.IsIncoming,
			Content:     strings.ReplaceAll(html.EscapeString(text), "\n", "<br>"),
		})
	}
	return out
}

// GetConversationMessages retrieves messages for a specific conversation.
func (m *Manager) GetConversationMessages(conversationUUID string, page, pageSize int, private *bool, msgTypes []string) ([]models.Message, int, error) {
	var (
		messages = make([]models.Message, 0)
		qArgs    []any
	)

	// Convert msgTypes slice to pq.StringArray for PostgreSQL
	var typesArg any
	if len(msgTypes) > 0 {
		typesArg = pq.StringArray(msgTypes)
	}

	qArgs = append(qArgs, conversationUUID, private, typesArg)
	query, pageSize, qArgs, err := m.generateMessagesQuery(m.q.GetMessages, qArgs, page, pageSize)
	if err != nil {
		m.lo.Error("error generating messages query", "error", err)
		return messages, pageSize, envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.somethingWentWrong"), nil)
	}

	tx, err := m.db.BeginTxx(context.Background(), &sql.TxOptions{
		ReadOnly: true,
	})
	defer tx.Rollback()
	if err != nil {
		m.lo.Error("error preparing get messages query", "error", err)
		return messages, pageSize, envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.somethingWentWrong"), nil)
	}

	if err := tx.Select(&messages, query, qArgs...); err != nil {
		m.lo.Error("error fetching conversations", "error", err)
		return messages, pageSize, envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.somethingWentWrong"), nil)
	}

	return messages, pageSize, nil
}

// GetAllConversationMessages returns the newest messages in a conversation in chronological order, capped at limit; a non-positive limit returns them all.
func (m *Manager) GetAllConversationMessages(conversationUUID string, private *bool, msgTypes []string, limit int) ([]models.Message, error) {
	var all []models.Message
	pageSize := maxMessagesPerPage
	if limit > 0 && limit < pageSize {
		pageSize = limit
	}
	for page := 1; ; page++ {
		messages, _, err := m.GetConversationMessages(conversationUUID, page, pageSize, private, msgTypes)
		if err != nil {
			return nil, err
		}
		all = append(all, messages...)
		if len(messages) == 0 || len(all) >= messages[0].Total || (limit > 0 && len(all) >= limit) {
			break
		}
	}
	if limit > 0 && len(all) > limit {
		all = all[:limit]
	}
	slices.Reverse(all)
	return all, nil
}

// GetMessage retrieves a message by UUID.
func (m *Manager) GetMessage(uuid string) (models.Message, error) {
	var message models.Message
	if err := m.q.GetMessage.Get(&message, uuid); err != nil {
		m.lo.Error("error fetching message", "uuid", uuid, "error", err)
		return message, envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.somethingWentWrong"), nil)
	}

	// Generate signed URLs for attachments.
	m.SignAttachmentURLs(message.Attachments)

	return message, nil
}

// SignAttachmentURLs adds access URLs for the original image and its thumbnail.
func (m *Manager) SignAttachmentURLs(attachments attachment.Attachments) {
	for i := range attachments {
		attachments[i].URL = m.mediaStore.GetURL(attachments[i].UUID, attachments[i].ContentType, attachments[i].Name)
		if strings.HasPrefix(attachments[i].ContentType, "image/") {
			attachments[i].ThumbnailURL = m.mediaStore.GetThumbnailURL(attachments[i].UUID)
		}
	}
}

// UpdateMessageStatus updates the status of a message.
func (m *Manager) UpdateMessageStatus(messageUUID string, status string) error {
	res, err := m.q.UpdateMessageStatus.Exec(status, messageUUID)
	if err != nil {
		m.lo.Error("error updating message status", "message_uuid", messageUUID, "error", err)
		return err
	}
	// The sent-onto-failed guard can make this a no-op. A status that wasn't applied must not be broadcast.
	if n, _ := res.RowsAffected(); n == 0 {
		return nil
	}

	// Broadcast message status update to all conversation subscribers.
	conversationUUID, _ := m.getConversationUUIDFromMessageUUID(messageUUID)
	m.BroadcastMessageUpdate(conversationUUID, messageUUID, map[string]any{"status": status})

	// Trigger webhook for message update.
	if message, err := m.GetMessage(messageUUID); err != nil {
		m.lo.Error("error fetching message for webhook event", "uuid", messageUUID, "error", err)
	} else {
		m.webhookStore.TriggerEvent(wmodels.EventMessageUpdated, message)
	}

	return nil
}

func (m *Manager) UpdateMessageSourceID(messageUUID, sourceID string) error {
	if messageUUID == "" || sourceID == "" {
		return nil
	}
	if _, err := m.q.UpdateMessageSourceIDByUUID.Exec(messageUUID, sourceID); err != nil {
		m.lo.Error("error updating message source id", "message_uuid", messageUUID, "error", err)
		return err
	}
	return nil
}

// UpdateConversationLastInboundAt advances the clock gating business-initiated messages (WhatsApp 24h window).
func (m *Manager) UpdateConversationLastInboundAt(conversationID int, at time.Time) error {
	if at.IsZero() {
		at = time.Now()
	}
	var row struct {
		ContactID int `db:"contact_id"`
		InboxID   int `db:"inbox_id"`
	}
	if err := m.q.UpdateConversationLastInboundAt.QueryRow(conversationID, at).Scan(&row.ContactID, &row.InboxID); err != nil {
		m.lo.Error("error updating conversation last_inbound_at", "conversation_id", conversationID, "error", err)
		return err
	}
	var windowAt sql.NullTime
	if err := m.q.GetContactWindowInboundAt.Get(&windowAt, row.ContactID, row.InboxID); err != nil {
		m.lo.Error("error fetching contact window for broadcast", "contact_id", row.ContactID, "inbox_id", row.InboxID, "error", err)
		return nil
	}
	if !windowAt.Valid {
		return nil
	}
	var uuids []string
	if err := m.q.GetConversationUUIDsByContactInbox.Select(&uuids, row.ContactID, row.InboxID); err != nil {
		m.lo.Error("error fetching contact's conversations for broadcast", "contact_id", row.ContactID, "inbox_id", row.InboxID, "error", err)
		return nil
	}
	for _, uuid := range uuids {
		m.BroadcastConversationUpdate(uuid, map[string]any{"contact_last_inbound_at": windowAt.Time.Format(time.RFC3339)})
	}
	return nil
}

// GetLatestOpenConversationForContact returns the most recent non-resolved conversation for a (contact, inbox) pair, or sql.ErrNoRows.
func (m *Manager) GetLatestOpenConversationForContact(contactID, inboxID int) (int, string, error) {
	var row struct {
		ID   int    `db:"id"`
		UUID string `db:"uuid"`
	}
	if err := m.q.GetLatestOpenConversationByContact.Get(&row, contactID, inboxID); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return 0, "", err
		}
		m.lo.Error("error fetching open conversation for contact", "contact_id", contactID, "inbox_id", inboxID, "error", err)
		return 0, "", envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.somethingWentWrong"), nil)
	}
	return row.ID, row.UUID, nil
}

// GetReopenableConversationForContact returns the most recent resolved conversation for a (contact, inbox) pair last resolved within windowHours, or sql.ErrNoRows.
func (m *Manager) GetReopenableConversationForContact(contactID, inboxID, windowHours int) (int, string, error) {
	var row struct {
		ID   int    `db:"id"`
		UUID string `db:"uuid"`
	}
	if err := m.q.GetReopenableConversationByContact.Get(&row, contactID, inboxID, windowHours); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return 0, "", err
		}
		m.lo.Error("error fetching reopenable conversation for contact", "contact_id", contactID, "inbox_id", inboxID, "error", err)
		return 0, "", envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.somethingWentWrong"), nil)
	}
	return row.ID, row.UUID, nil
}

func (m *Manager) MarkMessageAsPending(uuid string) error {
	if _, err := m.q.MarkMessagePendingForRetry.Exec(uuid); err != nil {
		m.lo.Error("error marking message as pending", "uuid", uuid, "error", err)
		return envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.errorSendingMessage"), nil)
	}
	conversationUUID, _ := m.getConversationUUIDFromMessageUUID(uuid)
	m.BroadcastMessageUpdate(conversationUUID, uuid, map[string]any{"status": models.MessageStatusPending})
	return nil
}

// ClearHandoffFormPending clears the handoff form flag on a message and reports whether it was set.
func (m *Manager) ClearHandoffFormPending(messageUUID string) (bool, error) {
	res, err := m.q.ClearMessageHandoffFormPending.Exec(messageUUID)
	if err != nil {
		m.lo.Error("error clearing handoff form flag", "message_uuid", messageUUID, "error", err)
		return false, err
	}
	n, err := res.RowsAffected()
	return n > 0, err
}

// SendPrivateNote inserts a private message in a conversation.
func (m *Manager) SendPrivateNote(media []mmodels.Media, senderID int, conversationUUID, content string, mentions []models.MentionInput) (models.Message, error) {
	// Best-effort render template variables before saving.
	if data, err := m.BuildTemplateData(conversationUUID, senderID); err == nil {
		content = m.template.RenderString(data, content)
	}

	message := models.Message{
		ConversationUUID: conversationUUID,
		SenderID:         senderID,
		UploadUserID:     senderID,
		Type:             models.MessageOutgoing,
		SenderType:       models.SenderTypeAgent,
		Status:           models.MessageStatusSent,
		Content:          content,
		ContentType:      models.ContentTypeHTML,
		Private:          true,
		Media:            media,
	}
	if err := m.InsertMessage(&message); err != nil {
		return models.Message{}, err
	}

	// Insert mentions if any.
	if len(mentions) > 0 {
		if err := m.InsertMentions(message.ConversationID, message.ID, senderID, mentions); err != nil {
			m.lo.Error("error inserting mentions", "error", err)
		}
		go m.NotifyMention(conversationUUID, message, mentions, senderID)
	}

	return message, nil
}

// CreateContactMessage creates a contact message in a conversation.
// sourceID is the bare RFC 5322 Message-ID of the inbound message; it is normalized and stored on the message so replies thread on it, mirroring the IMAP ingestion path. Empty leaves the column NULL.
func (m *Manager) CreateContactMessage(media []mmodels.Media, contactID int, conversationUUID, content, contentType string, isNewConversation bool, sourceID string, actorUserID int) (models.Message, error) {
	sourceID = stringutil.NormalizeMessageID(sourceID)
	message := models.Message{
		ConversationUUID: conversationUUID,
		SenderID:         contactID,
		UploadUserID:     actorUserID,
		Type:             models.MessageIncoming,
		SenderType:       models.SenderTypeContact,
		Status:           models.MessageStatusReceived,
		Content:          content,
		ContentType:      contentType,
		Private:          false,
		Media:            media,
		SourceID:         null.NewString(sourceID, sourceID != ""),
	}
	if err := m.InsertMessage(&message); err != nil {
		return models.Message{}, err
	}

	// Process post-message hooks (reopen, waiting since, automation, SLA).
	if err := m.ProcessIncomingMessageHooks(message, isNewConversation); err != nil {
		m.lo.Error("error processing incoming message hooks", "conversation_uuid", conversationUUID, "error", err)
	}

	return message, nil
}

// QueueReply queues a reply message in a conversation.
func (m *Manager) QueueReply(media []mmodels.Media, inboxID, senderID, contactID int, conversationUUID, content string, to, cc, bcc []string, sendFrom string, metaMap map[string]interface{}) (models.Message, error) {
	var (
		message = models.Message{}
	)

	inboxRecord, err := m.inboxStore.GetDBRecord(inboxID)
	if err != nil {
		m.lo.Error("error fetching inbox record", "inbox_id", inboxID, "error", err)
		return models.Message{}, err
	}

	if !inboxRecord.Enabled {
		return models.Message{}, envelope.NewError(envelope.InputError, m.i18n.T("status.disabledInbox"), nil)
	}

	isWhatsAppTemplate := inboxRecord.Channel == inbox.ChannelWhatsApp && extractInt(metaMap, "whatsapp_template_id") > 0
	if !isWhatsAppTemplate {
		if data, err := m.BuildTemplateData(conversationUUID, senderID); err == nil {
			content = m.template.RenderString(data, content)
		}
	}

	var (
		sourceID    string
		contentType = models.ContentTypeHTML
	)
	switch inboxRecord.Channel {
	case inbox.ChannelEmail:
		sendFrom, err = m.resolveSendFrom(conversationUUID, inboxRecord, sendFrom)
		if err != nil {
			return models.Message{}, err
		}
		metaMap["send_from"] = sendFrom
		// Add `to`, `cc`, and `bcc` recipients to meta map.
		to = stringutil.RemoveEmpty(to)
		cc = stringutil.RemoveEmpty(cc)
		bcc = stringutil.RemoveEmpty(bcc)
		if len(to) > 0 {
			metaMap["to"] = to
		}
		if len(cc) > 0 {
			metaMap["cc"] = cc
		}
		if len(bcc) > 0 {
			metaMap["bcc"] = bcc
		}
		if len(to) == 0 {
			return message, envelope.NewError(envelope.GeneralError, m.i18n.Ts("globals.messages.empty", "name", "`to`"), nil)
		}
		sourceID, err = stringutil.GenerateEmailMessageID(conversationUUID, inboxRecord.From)
		if err != nil {
			m.lo.Error("error generating source message id", "error", err)
			return models.Message{}, envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.somethingWentWrong"), nil)
		}
	case inbox.ChannelWhatsApp:
		// Meta accepts one media per message, so a multi-attachment reply must be sent as separate messages.
		if len(media) > 1 {
			return models.Message{}, envelope.NewError(envelope.InputError, m.i18n.T("conversation.whatsapp.error.oneAttachment"), nil)
		}
		// Reject unsendable media here. Meta's upload endpoint enforces the same caps and would only fail after the message is queued.
		for _, md := range media {
			if reason := whatsappChannel.RejectMediaReason(md.Filename, md.ContentType, md.Size); reason != "" {
				return models.Message{}, envelope.NewError(envelope.InputError, reason, nil)
			}
		}
		rendered, err := m.prepareWhatsAppOutbound(inboxRecord, conversationUUID, content, len(media) > 0, metaMap)
		if err != nil {
			return models.Message{}, err
		}
		content = rendered
		// A rendered template body is plain text. Storing it as HTML drops its line breaks in the timeline.
		if isWhatsAppTemplate {
			contentType = models.ContentTypeText
		}
	}

	// Marshal meta.
	metaJSON, err := json.Marshal(metaMap)
	if err != nil {
		m.lo.Error("error marshalling message meta map to JSON", "error", err)
		return models.Message{}, envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.somethingWentWrong"), nil)
	}

	// Insert the message into the database
	message = models.Message{
		ConversationUUID:  conversationUUID,
		SenderID:          senderID,
		UploadUserID:      senderID,
		Type:              models.MessageOutgoing,
		SenderType:        models.SenderTypeAgent,
		Status:            models.MessageStatusPending,
		Content:           content,
		ContentType:       contentType,
		Private:           false,
		Media:             media,
		SourceID:          null.NewString(sourceID, sourceID != ""),
		MessageReceiverID: contactID,
		Meta:              metaJSON,
	}
	if err := m.InsertMessage(&message); err != nil {
		return models.Message{}, err
	}
	return message, nil
}

// InsertMessage inserts a message and attaches the media to the message.
func (m *Manager) InsertMessage(message *models.Message) error {
	tx, err := m.db.Beginx()
	if err != nil {
		m.lo.Error("error beginning message insert transaction", "error", err)
		return envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.somethingWentWrong"), nil)
	}
	defer tx.Rollback()

	inlineUUIDs, err := m.InsertMessageTx(tx, message)
	if err != nil {
		return err
	}

	if err := tx.Commit(); err != nil {
		m.lo.Error("error committing message insert transaction", "error", err)
		return envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.somethingWentWrong"), nil)
	}

	m.AfterMessageInsert(message, inlineUUIDs)
	return nil
}

// InsertMessageTx inserts a message inside the caller's transaction; the caller must run AfterMessageInsert once it commits.
func (m *Manager) InsertMessageTx(tx *sqlx.Tx, message *models.Message) ([]string, error) {
	if message.Private {
		message.Status = models.MessageStatusSent
	}
	if len(message.Meta) == 0 || string(message.Meta) == "null" {
		message.Meta = json.RawMessage(`{}`)
	}

	// Handle empty content type enum, default to text.
	if message.ContentType == "" {
		message.ContentType = models.ContentTypeText
	}

	inlineUUIDs := extractInlineImageUUIDs(message.Content)

	// Rewrite inline image URLs to cid:ldsk-<uuid>. The read API resolves them back to signed URLs.
	message.Content = rewriteInlineImagesToCID(message.Content)

	// Convert content to plain text for search.
	if message.ContentType == models.ContentTypeText {
		message.TextContent = message.Content
	} else {
		message.TextContent = stringutil.HTML2Text(message.Content)
	}

	if err := tx.Stmtx(m.q.InsertMessage).Get(message, message.Type, message.Status, message.ConversationID, message.ConversationUUID, message.Content, message.TextContent, message.SenderID, message.SenderType,
		message.Private, message.ContentType, message.SourceID, message.Meta); err != nil {
		m.lo.Error("error inserting message in db", "error", err)
		return nil, envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.somethingWentWrong"), nil)
	}

	if err := m.mediaStore.LinkMessageMediaTx(tx, message.ID, message.Media, inlineUUIDs, message.UploadUserID); err != nil {
		return nil, envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.somethingWentWrong"), nil)
	}
	return inlineUUIDs, nil
}

// AfterMessageInsert runs the post-commit side effects of a message insert: participant, last message and broadcast.
func (m *Manager) AfterMessageInsert(message *models.Message, inlineUUIDs []string) {
	m.addConversationParticipant(message.SenderID, message.ConversationUUID)

	// Skip updating last_message and broadcasting for continuity emails.
	if !message.IsContinuityMessage() {
		// Hide CSAT message content as it contains a public link to the survey.
		lastMessage := message.TextContent
		if message.HasCSAT() {
			lastMessage = "Please rate your experience with us"
		}

		// HTML2Text drops <img> tags, so image-only messages have empty text. Fall back to a media-type preview.
		if strings.TrimSpace(lastMessage) == "" {
			switch {
			case len(message.Media) > 0:
				lastMessage = m.getMediaPreview(message.Media[0])
			case len(inlineUUIDs) > 0:
				lastMessage = m.i18n.T("globals.terms.image")
			}
		}

		// Update conversation last message details (also conditionally updates last_interaction if not activity/private).
		m.UpdateConversationLastMessage(message.ConversationID, message.ConversationUUID, lastMessage, message.SenderType, message.Type, message.Private, message.CreatedAt, message.SenderID)

		var convItem *models.ConversationListItem
		if item, err := m.GetConversationListItem(message.ConversationUUID); err == nil {
			convItem = &item
		} else {
			m.lo.Error("error fetching conversation list item for broadcast", "uuid", message.ConversationUUID, "error", err)
		}
		m.BroadcastNewMessage(message, convItem, lastMessage)
	}

	// Refetch the message to get all fields populated (e.g., author, media URLs).
	refetchedMessage, err := m.GetMessage(message.UUID)
	if err != nil {
		m.lo.Error("error fetching message after insert", "error", err)
	} else {
		*message = refetchedMessage
	}

	// Trigger webhook for new message created.
	m.webhookStore.TriggerEvent(wmodels.EventMessageCreated, message)
}

// RecordAssigneeUserChange records an activity for a user assignee change.
func (m *Manager) RecordAssigneeUserChange(conversationUUID string, assigneeID int, actor umodels.User) error {
	// Self assignment.
	if assigneeID == actor.ID {
		return m.InsertConversationActivity(models.ActivitySelfAssign, conversationUUID, actor.FullName(), actor)
	}

	// Assignment to another user.
	assignee, err := m.userStore.GetAgentCachedOrLoad(assigneeID)
	if err != nil {
		return err
	}
	return m.InsertConversationActivity(models.ActivityAssignedUserChange, conversationUUID, assignee.FullName(), actor)
}

// RecordAssigneeUserRemoval records an activity for the removal of a user assignee.
func (m *Manager) RecordAssigneeUserRemoval(conversationUUID string, assigneeID int, actor umodels.User) error {
	if assigneeID == actor.ID {
		return m.InsertConversationActivity(models.ActivitySelfUnassign, conversationUUID, actor.FullName(), actor)
	}

	assignee, err := m.userStore.GetAgentCachedOrLoad(assigneeID)
	if err != nil {
		return err
	}
	return m.InsertConversationActivity(models.ActivityAssigneeUserRemoved, conversationUUID, assignee.FullName(), actor)
}

// RecordAssigneeTeamChange records an activity for a team assignee change.
func (m *Manager) RecordAssigneeTeamChange(conversationUUID string, teamID int, actor umodels.User) error {
	team, err := m.teamStore.Get(teamID)
	if err != nil {
		return err
	}
	return m.InsertConversationActivity(models.ActivityAssignedTeamChange, conversationUUID, team.Name, actor)
}

// RecordPriorityChange records an activity for a priority change.
func (m *Manager) RecordPriorityChange(priority, conversationUUID string, actor umodels.User) error {
	return m.InsertConversationActivity(models.ActivityPriorityChange, conversationUUID, priority, actor)
}

// RecordStatusChange records an activity for a status change.
func (m *Manager) RecordStatusChange(status, conversationUUID string, actor umodels.User) error {
	return m.InsertConversationActivity(models.ActivityStatusChange, conversationUUID, status, actor)
}

// RecordSLASet records an activity for an SLA set.
func (m *Manager) RecordSLASet(conversationUUID string, slaName string, actor umodels.User) error {
	return m.InsertConversationActivity(models.ActivitySLASet, conversationUUID, slaName, actor)
}

// RecordTagAddition records an activity for a tag addition.
func (m *Manager) RecordTagAddition(conversationUUID string, tag string, actor umodels.User) error {
	return m.InsertConversationActivity(models.ActivityTagAdded, conversationUUID, tag, actor)
}

// RecordTagRemoval records an activity for a tag removal.
func (m *Manager) RecordTagRemoval(conversationUUID string, tag string, actor umodels.User) error {
	return m.InsertConversationActivity(models.ActivityTagRemoved, conversationUUID, tag, actor)
}

// InsertConversationActivity inserts an activity message.
func (m *Manager) InsertConversationActivity(activityType, conversationUUID, newValue string, actor umodels.User) error {
	content, err := m.getMessageActivityContent(activityType, newValue, actor.FullName())
	if err != nil {
		m.lo.Error("error could not generate activity content", "error", err)
		return envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.somethingWentWrong"), nil)
	}

	// Store the activity type structurally so callers can filter activities without parsing i18n content.
	meta, _ := json.Marshal(map[string]string{"activity_type": activityType})

	message := models.Message{
		Type:             models.MessageActivity,
		Status:           models.MessageStatusSent,
		Content:          content,
		ContentType:      models.ContentTypeText,
		ConversationUUID: conversationUUID,
		Private:          true,
		SenderID:         actor.ID,
		SenderType:       models.SenderTypeAgent,
		Meta:             meta,
	}

	if err := m.InsertMessage(&message); err != nil {
		m.lo.Error("error inserting activity message", "error", err)
		return envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.somethingWentWrong"), nil)
	}
	return nil
}

// getConversationUUIDFromMessageUUID returns conversation UUID from message UUID.
func (m *Manager) getConversationUUIDFromMessageUUID(uuid string) (string, error) {
	var conversationUUID string
	if err := m.q.GetConversationUUIDFromMessageUUID.Get(&conversationUUID, uuid); err != nil {
		m.lo.Error("error fetching conversation uuid from message uuid", "uuid", uuid, "error", err)
		return conversationUUID, err
	}
	return conversationUUID, nil
}

// getMessageActivityContent generates activity content based on the activity type.
func (m *Manager) getMessageActivityContent(activityType, newValue, actorName string) (string, error) {
	var content = ""
	switch activityType {
	case models.ActivityAssignedUserChange:
		content = fmt.Sprintf("Assigned to %s by %s", newValue, actorName)
	case models.ActivityAssignedTeamChange:
		content = fmt.Sprintf("Assigned to %s team by %s", newValue, actorName)
	case models.ActivityAssigneeUserRemoved:
		content = fmt.Sprintf("%s removed %s as assignee", actorName, newValue)
	case models.ActivitySelfAssign:
		content = fmt.Sprintf("%s self-assigned this conversation", actorName)
	case models.ActivitySelfUnassign:
		content = fmt.Sprintf("%s unassigned themselves", actorName)
	case models.ActivityPriorityChange:
		content = fmt.Sprintf("%s set priority to %s", actorName, newValue)
	case models.ActivityStatusChange:
		content = fmt.Sprintf("%s marked the conversation as %s", actorName, newValue)
	case models.ActivityTagAdded:
		content = fmt.Sprintf("%s added tag %s", actorName, newValue)
	case models.ActivityTagRemoved:
		content = fmt.Sprintf("%s removed tag %s", actorName, newValue)
	case models.ActivitySLASet:
		content = fmt.Sprintf("%s set %s SLA policy", actorName, newValue)
	case models.ActivityParticipantAdded:
		content = fmt.Sprintf("%s joined the conversation", newValue)
	case models.ActivityCSATNotSent:
		content = m.i18n.T("conversation.whatsapp.csatNotSent")
	default:
		return "", fmt.Errorf("invalid activity type %s", activityType)
	}
	return content, nil
}

// ProcessIncomingMessage handles the insertion of an incoming message and
// associated contact. It finds or creates the contact, checks for existing
// conversations, and creates a new conversation if necessary. It also
// inserts the message, uploads any attachments, and queues the conversation evaluation of automation rules.
func (m *Manager) ProcessIncomingMessage(in models.IncomingMessage) (models.Message, error) {
	// Return early if this message already exists (same source ID).
	dupConvID, err := m.messageExistsBySourceID([]string{in.SourceID.String})
	if err != nil && err != errConversationNotFound {
		return models.Message{}, err
	}
	if dupConvID > 0 {
		return models.Message{}, nil
	}

	// Resolve sender and conversation from plus addressing.
	senderID, conversationID, conversationUUID, err := m.resolveSender(&in)
	if err != nil {
		return models.Message{}, err
	}

	// Find or create contact.
	if senderID == 0 {
		user := umodels.User{
			FirstName: in.Contact.FirstName,
			LastName:  in.Contact.LastName,
			Email:     in.Contact.Email,
			Type:      umodels.UserTypeContact,
		}
		if err := m.userStore.ResolveContact(&user, umodels.ContactSync); err != nil {
			m.lo.Error("error creating contact for incoming message", "message_source_id", in.SourceID.String, "error", err)
			return models.Message{}, fmt.Errorf("creating contact: %w", err)
		}
		senderID = user.ID
		in.Contact.ID = senderID
	}

	// Match conversation if not already matched by plus-addressing.
	var isNewConversation bool
	if conversationID == 0 {
		conversationID, conversationUUID, isNewConversation, err = m.findOrCreateConversation(in)
		if err != nil {
			m.lo.Error("error finding or creating conversation for incoming message", "message_source_id", in.SourceID.String, "error", err)
			return models.Message{}, err
		}
	}

	// For existing conversations, override sender with the conversation's contact when emails match.
	if !isNewConversation && conversationID > 0 {
		conversation, convErr := m.GetConversation(conversationID, "", "")
		if convErr == nil && strings.EqualFold(conversation.Contact.Email.String, in.Contact.Email.String) {
			senderID = conversation.ContactID
			in.Contact.ID = senderID
		}
	}

	// Convert to Message for attachment upload and insertion.
	msg := in.ToMessage(senderID, conversationID, conversationUUID)

	// Upload message attachments. On failure, delete the conversation if it was just created for this message.
	if upErr := m.uploadMessageAttachments(&msg); upErr != nil {
		m.lo.Error("error uploading message attachments", "message_source_id", in.SourceID.String, "error", upErr)
		if isNewConversation && conversationUUID != "" {
			m.lo.Info("deleting conversation as message attachment upload failed", "conversation_uuid", conversationUUID, "message_source_id", in.SourceID.String)
			if err := m.DeleteConversation(conversationUUID); err != nil {
				return models.Message{}, fmt.Errorf("deleting conversation after message attachment upload failure: %w", err)
			}
		}
		return models.Message{}, fmt.Errorf("uploading message attachments: %w", upErr)
	}

	// Insert message. On failure, delete the conversation if it was just created for this message.
	if err = m.InsertMessage(&msg); err != nil {
		m.lo.Error("error inserting incoming message", "message_source_id", in.SourceID.String, "conversation_uuid", conversationUUID, "is_new", isNewConversation, "error", err)
		if isNewConversation && conversationUUID != "" {
			if delErr := m.DeleteConversation(conversationUUID); delErr != nil {
				return models.Message{}, fmt.Errorf("deleting conversation after message insert failure: %w", delErr)
			}
		}
		return models.Message{}, fmt.Errorf("inserting message: %w", err)
	}

	// When a customer replies to a continuity emailsync the message to their live chat widget via WebSocket.
	// No-op if the conversation's inbox isn't livechat.
	m.broadcastMessageToWidgetClients(&msg)

	// Process post-message hooks (automation rules, webhooks, SLA, etc.).
	if err := m.ProcessIncomingMessageHooks(msg, isNewConversation); err != nil {
		m.lo.Error("error processing incoming message hooks", "conversation_uuid", msg.ConversationUUID, "error", err)
		return models.Message{}, fmt.Errorf("processing incoming message hooks: %w", err)
	}
	return msg, nil
}

// resolveSender resolves the sender for an incoming message via plus-addressing.
// Returns senderID, and optionally conversationID/UUID if matched.
// If sender is not resolved here, ProcessIncomingMessage handles it with conversation context.
func (m *Manager) resolveSender(in *models.IncomingMessage) (senderID, conversationID int, conversationUUID string, err error) {
	if in.ConversationUUIDFromReplyTo != "" {
		senderID, conversationID, conversationUUID, err = m.resolveByPlusAddress(in)
		if err != nil {
			return 0, 0, "", err
		}
		if senderID > 0 {
			in.Contact.ID = senderID
		}
	}
	return senderID, conversationID, conversationUUID, nil
}

// resolveByPlusAddress attempts to match a conversation via plus-addressed Reply-To
// (e.g., inbox+conv-{uuid}@domain). If the conversation contact is a visitor, it upgrades
// them to a contact (proving email ownership). Returns senderID > 0 if resolved.
func (m *Manager) resolveByPlusAddress(in *models.IncomingMessage) (senderID, conversationID int, conversationUUID string, err error) {
	conversation, err := m.GetConversation(0, in.ConversationUUIDFromReplyTo, "")
	if err != nil {
		// Not found return with no error.
		if envErr, ok := err.(envelope.Error); ok && envErr.ErrorType == envelope.NotFoundError {
			return 0, 0, "", nil
		}
		// Other errors.
		return 0, 0, "", fmt.Errorf("fetching conversation: %w", err)
	}

	m.lo.Debug("matched conversation by plus-addressed Reply-To", "conversation_uuid", conversation.UUID, "contact_email", in.Contact.Email.String)

	conversationID = conversation.ID
	conversationUUID = conversation.UUID
	senderID = conversation.Contact.ID

	// Already a contact - if same email, return as sender. If different email, let contact resolution find the actual sender.
	if conversation.Contact.Type == umodels.UserTypeContact {
		if !strings.EqualFold(conversation.Contact.Email.String, in.Contact.Email.String) {
			return 0, conversationID, conversationUUID, nil
		}
		return senderID, conversationID, conversationUUID, nil
	}

	// Visitor upgrade requires email match - proves email ownership.
	// If a different email replied, thread the message but don't upgrade.
	if !strings.EqualFold(conversation.Contact.Email.String, in.Contact.Email.String) {
		return 0, conversationID, conversationUUID, nil
	}

	user, contactErr := m.userStore.Get(0, in.Contact.Email.String, []string{umodels.UserTypeContact})
	if contactErr == nil {
		m.lo.Debug("a contact already exists with the same email as visitor; not upgrading visitor", "conversation_uuid", conversation.UUID, "contact_email", in.Contact.Email.String, "contact_user_id", user.ID)
		// A contact with this email already exists; don't upgrade visitor.
		// Let contact resolution find the correct sender ID.
		return 0, conversationID, conversationUUID, nil
	}

	if envErr, ok := contactErr.(envelope.Error); !ok || envErr.ErrorType != envelope.NotFoundError {
		return 0, 0, "", fmt.Errorf("fetching contact by email: %w", contactErr)
	}

	// Block upgrade if continuity email TTL expired.
	if !m.isVisitorUpgradeSafe(conversation) {
		return 0, conversationID, conversationUUID, nil
	}

	// Upgrade visitor as no contact exist with this email.
	if err := m.userStore.UpgradeVisitorToContact(conversation.Contact.ID); err != nil {
		return 0, 0, "", fmt.Errorf("upgrading visitor to contact: %w", err)
	}

	m.lo.Debug("upgraded visitor to contact", "conversation_uuid", conversation.UUID, "contact_id", conversation.Contact.ID)

	// Notify conversation subscribers that the contact type has changed.
	m.BroadcastContactUpdate(conversation.ContactID, map[string]any{"type": umodels.UserTypeContact})

	return senderID, conversationID, conversationUUID, nil
}

// isVisitorUpgradeSafe checks whether a visitor-to-contact upgrade should proceed.
// Blocks upgrade if the continuity email TTL has expired.
func (m *Manager) isVisitorUpgradeSafe(conversation models.Conversation) bool {
	if conversation.LastContinuityEmailSentAt.Valid {
		if time.Since(conversation.LastContinuityEmailSentAt.Time) > upgradeWindowTTL {
			m.lo.Info("visitor upgrade blocked: continuity email TTL expired",
				"conversation_uuid", conversation.UUID,
				"last_sent", conversation.LastContinuityEmailSentAt.Time,
				"age", time.Since(conversation.LastContinuityEmailSentAt.Time).String(),
				"max_ttl", upgradeWindowTTL.String())
			return false
		}
	}
	return true
}

// ProcessIncomingLiveChatMessage handles incoming live chat messages.
func (m *Manager) ProcessIncomingLiveChatMessage(msg models.Message) (models.Message, error) {
	// Upload message attachments.
	if err := m.uploadMessageAttachments(&msg); err != nil {
		return models.Message{}, fmt.Errorf("uploading message attachments: %w", err)
	}

	// Insert message.
	if err := m.InsertMessage(&msg); err != nil {
		return models.Message{}, err
	}

	// Advance contact_last_seen_at.
	if err := m.UpdateConversationContactLastSeen(msg.ConversationUUID); err != nil {
		m.lo.Error("error updating contact last seen after livechat message", "conversation_uuid", msg.ConversationUUID, "error", err)
	}

	// Process post-message hooks (automation rules, webhooks, SLA, etc.).
	// isNewConversation = false since conversation always exists for live chat.
	if err := m.ProcessIncomingMessageHooks(msg, false); err != nil {
		m.lo.Error("error processing incoming message hooks", "conversation_uuid", msg.ConversationUUID, "error", err)
	}

	return msg, nil
}

// ProcessIncomingWhatsAppMessage inserts an inbound message and advances the 24h window clock.
func (m *Manager) ProcessIncomingWhatsAppMessage(msg models.Message, isNewConversation bool, inboundAt time.Time) (models.Message, error) {
	if err := m.uploadMessageAttachments(&msg); err != nil {
		return models.Message{}, fmt.Errorf("uploading whatsapp attachments: %w", err)
	}

	if err := m.InsertMessage(&msg); err != nil {
		return models.Message{}, err
	}

	// Hooks run even if the window update fails: the message is stored, so the queue retry only repairs the window.
	windowErr := m.UpdateConversationLastInboundAt(msg.ConversationID, inboundAt)

	if err := m.ProcessIncomingMessageHooks(msg, isNewConversation); err != nil {
		m.lo.Error("error processing incoming message hooks", "conversation_uuid", msg.ConversationUUID, "error", err)
	}

	if windowErr != nil {
		return models.Message{}, windowErr
	}

	return msg, nil
}

// MessageExists checks if a message with the given messageID exists.
func (m *Manager) MessageExists(messageID string) (bool, error) {
	_, err := m.messageExistsBySourceID([]string{messageID})
	if err != nil {
		if errors.Is(err, errConversationNotFound) {
			return false, nil
		}
		m.lo.Error("error fetching message from db", "error", err)
		return false, err
	}
	return true, nil
}

func (m *Manager) AdvanceWhatsAppWindowForMessage(messageID string, inboundAt time.Time) (bool, error) {
	conversationID, err := m.messageExistsBySourceID([]string{messageID})
	if errors.Is(err, errConversationNotFound) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	if err := m.UpdateConversationLastInboundAt(conversationID, inboundAt); err != nil {
		return true, err
	}
	return true, nil
}

// EnqueueIncoming enqueues an incoming message for inserting in db.
func (m *Manager) EnqueueIncoming(message models.IncomingMessage) error {
	m.closedMu.Lock()
	defer m.closedMu.Unlock()
	if m.closed {
		return errors.New("incoming message queue is closed")
	}

	select {
	case m.incomingMessageQueue <- message:
		return nil
	default:
		m.lo.Warn("WARNING: incoming message queue is full")
		return errors.New("incoming message queue is full")
	}
}

// GetConversationByMessageID returns conversation by message id.
func (m *Manager) GetConversationByMessageID(id int) (models.Conversation, error) {
	var conversation = models.Conversation{}
	if err := m.q.GetConversationByMessageID.Get(&conversation, id); err != nil {
		if err == sql.ErrNoRows {
			return conversation, envelope.NewError(envelope.NotFoundError, m.i18n.T("validation.notFoundConversation"), nil)
		}
		m.lo.Error("error fetching message from DB", "error", err)
		return conversation, envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.somethingWentWrong"), nil)
	}
	return conversation, nil
}

// generateMessagesQuery generates the SQL query for fetching messages in a conversation.
func (c *Manager) generateMessagesQuery(baseQuery string, qArgs []interface{}, page, pageSize int) (string, int, []interface{}, error) {
	if pageSize > maxMessagesPerPage {
		pageSize = maxMessagesPerPage
	}

	// Calculate the offset
	offset := (page - 1) * pageSize

	// Append LIMIT and OFFSET to query arguments
	qArgs = append(qArgs, pageSize, offset)

	// Include LIMIT and OFFSET in the SQL query
	sqlQuery := fmt.Sprintf(baseQuery, fmt.Sprintf("LIMIT $%d OFFSET $%d", len(qArgs)-1, len(qArgs)))
	return sqlQuery, pageSize, qArgs, nil
}

// extractInlineImageUUIDs returns unique media UUIDs from <img src="..."> URLs in order of first appearance, skipping the cid: form.
func extractInlineImageUUIDs(content string) []string {
	matches := imgSrcPattern.FindAllStringSubmatch(content, -1)
	seen := make(map[string]bool, len(matches))
	out := make([]string, 0, len(matches))
	for _, m := range matches {
		url := m[1]
		if strings.HasPrefix(url, "cid:") {
			continue
		}
		u := stringutil.ExtractUUID(url)
		if u == "" || seen[u] {
			continue
		}
		seen[u] = true
		out = append(out, u)
	}
	return out
}

// extractInlineContentIDs returns unique content_ids referenced via <img src="cid:..."> in the body.
func extractInlineContentIDs(content string) []string {
	matches := imgSrcPattern.FindAllStringSubmatch(content, -1)
	seen := make(map[string]bool, len(matches))
	out := make([]string, 0, len(matches))
	for _, m := range matches {
		url := m[1]
		if !strings.HasPrefix(url, "cid:") {
			continue
		}
		cid := strings.TrimPrefix(url, "cid:")
		if cid == "" || seen[cid] {
			continue
		}
		seen[cid] = true
		out = append(out, cid)
	}
	return out
}

// rewriteInlineImagesToCID rewrites every <img src="...<uuid>..."> to <img src="cid:ldsk-<uuid>">. Already-cid form is left alone.
func rewriteInlineImagesToCID(content string) string {
	return imgSrcPattern.ReplaceAllStringFunc(content, func(match string) string {
		sub := imgSrcPattern.FindStringSubmatch(match)
		url := sub[1]
		if strings.HasPrefix(url, "cid:") {
			return match
		}
		u := stringutil.ExtractUUID(url)
		if u == "" {
			return match
		}
		return strings.Replace(match, url, "cid:"+inlineContentID(u), 1)
	})
}

// uploadMessageAttachments uploads all attachments for a message.
func (m *Manager) uploadMessageAttachments(message *models.Message) error {
	if len(message.Attachments) == 0 {
		return nil
	}

	for _, attachment := range message.Attachments {
		contentID := attachment.ContentID
		if contentID != "" {
			storedCID, exists, mediaUUID := m.findExistingMedia(contentID, message.ConversationUUID)

			// Make body's cid match the stored content_id so the read path can find it.
			if storedCID != contentID {
				message.Content = strings.ReplaceAll(message.Content, fmt.Sprintf("cid:%s", contentID), fmt.Sprintf("cid:%s", storedCID))
			}

			if exists {
				m.lo.Debug("inline attachment exists, reusing", "content_id", storedCID, "media_uuid", mediaUUID)
				continue
			}
			contentID = storedCID
		}

		attachment.Name = stringutil.SanitizeFilename(attachment.Name)

		if len(attachment.Content) == 0 {
			m.lo.Warn("skipping empty attachment", "name", attachment.Name, "content_id", contentID, "content_type", attachment.ContentType, "disposition", attachment.Disposition, "message_source_id", message.SourceID.String, "conversation_uuid", message.ConversationUUID)
			continue
		}

		m.lo.Debug("uploading message attachment", "name", attachment.Name, "content_id", contentID, "size", attachment.Size, "content_type", attachment.ContentType, "disposition", attachment.Disposition)

		// Upload and insert entry in media table.
		attachReader := bytes.NewReader(attachment.Content)
		media, err := m.mediaStore.UploadAndInsert(
			attachment.Name,
			attachment.ContentType,
			contentID,
			/** Linking media to message happens later **/
			null.String{}, /** modelType */
			null.Int{},    /** modelID **/
			attachReader,
			attachment.Size,
			null.StringFrom(attachment.Disposition),
			[]byte("{}"), /** meta **/
			true,         /** private **/
		)
		if err != nil {
			m.lo.Error("failed to upload attachment", "name", attachment.Name, "content_type", attachment.ContentType, "size", attachment.Size, "content_id", contentID, "disposition", attachment.Disposition, "conversation_uuid", message.ConversationUUID, "message_source_id", message.SourceID.String, "error", err)
			return fmt.Errorf("failed to upload media %s: %w", attachment.Name, err)
		}

		// If the attachment is an image, generate and upload a thumbnail. Log any errors and continue.
		attachmentExt := strings.TrimPrefix(strings.ToLower(filepath.Ext(attachment.Name)), ".")
		if slices.Contains(image.Exts, attachmentExt) && image.IsImageByContent(bytes.NewReader(attachment.Content)) {
			if err := m.uploadThumbnailForMedia(media, attachment.Content); err != nil {
				m.lo.Warn("skipping thumbnail, unsupported image format", "error", err)
			}
		}

		message.Media = append(message.Media, media)
	}
	return nil
}

// findOrCreateConversation finds or creates a conversation for the given incoming message.
func (m *Manager) findOrCreateConversation(in models.IncomingMessage) (int, string, bool, error) {
	var (
		conversationID   int
		conversationUUID string
		err              error
	)

	// Search for existing conversation using the in-reply-to and references.
	m.lo.Debug("searching conversation using in-reply-to and references", "in_reply_to", in.InReplyTo, "references", in.References)

	sourceIDs := append([]string{in.InReplyTo}, in.References...)
	conversationID, err = m.messageExistsBySourceID(sourceIDs)
	if err != nil && err != errConversationNotFound {
		return 0, "", false, err
	}

	// Conversation not found, create one.
	if conversationID == 0 {
		m.lo.Debug("no conversation found with in-reply-to and references, creating new conversation", "in_reply_to", in.InReplyTo, "references", in.References)
		lastMessage := stringutil.HTML2Text(in.Content)
		lastMessageAt := time.Now()
		conversationID, conversationUUID, err = m.CreateConversation(in.Contact.ID,
			in.InboxID,
			lastMessage,
			lastMessageAt,
			in.Subject,
			false, /**append reference number to subject**/
			nil,   /** meta **/
			nil,   /** customer attributes **/
			0,     /** max conversation **/
			0,     /** rate limit window **/
		)
		if err != nil || conversationID == 0 {
			return 0, "", false, err
		}
		return conversationID, conversationUUID, true, nil
	}

	// Get UUID for the found conversation ID.
	conversationUUID, err = m.GetConversationUUID(conversationID)
	if err != nil {
		return 0, "", false, err
	}
	return conversationID, conversationUUID, false, nil
}

// messageExistsBySourceID returns conversation ID if a message with any of the given source IDs exists.
func (m *Manager) messageExistsBySourceID(messageSourceIDs []string) (int, error) {
	messageSourceIDs = stringutil.RemoveEmpty(messageSourceIDs)
	if len(messageSourceIDs) == 0 {
		return 0, errConversationNotFound
	}
	var conversationID int
	if err := m.q.MessageExistsBySourceID.QueryRow(pq.Array(messageSourceIDs)).Scan(&conversationID); err != nil {
		if err == sql.ErrNoRows {
			return conversationID, errConversationNotFound
		}
		m.lo.Error("error fetching msg from DB", "error", err)
		return conversationID, err
	}
	return conversationID, nil
}

// GetInlineMediaRefs returns media referenced via cid: in the body but linked to other messages (quoted history).
func (m *Manager) GetInlineMediaRefs(message *models.Message) ([]mmodels.Media, error) {
	cids := extractInlineContentIDs(message.Content)
	if len(cids) == 0 {
		return nil, nil
	}
	existing := make(map[string]bool, len(message.Attachments))
	for _, a := range message.Attachments {
		if a.ContentID != "" {
			existing[a.ContentID] = true
		}
	}
	missing := make([]string, 0, len(cids))
	for _, cid := range cids {
		if !existing[cid] {
			missing = append(missing, cid)
		}
	}
	if len(missing) == 0 {
		return nil, nil
	}
	return m.mediaStore.GetByContentIDs(missing, message.ConversationUUID)
}

// fetchMessageAttachments fetches attachments (also inline images) for a single message ID.
func (m *Manager) fetchMessageAttachments(messageID int) (attachment.Attachments, error) {
	var attachments attachment.Attachments

	// Get all media for this message.
	medias, err := m.mediaStore.GetByModel(messageID, mmodels.ModelMessages)
	if err != nil {
		return attachments, fmt.Errorf("error fetching message attachments: %w", err)
	}

	// Fetch blobs for each media item.
	for _, media := range medias {
		blob, err := m.mediaStore.GetBlob(media.UUID)
		if err != nil {
			return attachments, fmt.Errorf("error fetching media blob: %w", err)
		}

		contentID := media.ContentID
		if contentID == "" {
			contentID = media.UUID
		}

		attachment := attachment.Attachment{
			Name:        media.Filename,
			UUID:        media.UUID,
			ContentType: media.ContentType,
			ContentID:   contentID,
			Content:     blob,
			Size:        media.Size,
			Header:      attachment.MakeHeader(media.ContentType, contentID, media.Filename, "base64", media.Disposition.String),
			URL:         m.mediaStore.GetURL(media.UUID, media.ContentType, media.Filename),
		}
		if strings.HasPrefix(media.ContentType, "image/") {
			attachment.ThumbnailURL = m.mediaStore.GetThumbnailURL(media.UUID)
		}
		attachments = append(attachments, attachment)
	}

	return attachments, nil
}

// attachAttachmentsToMessage attaches attachment blobs to message.
func (m *Manager) attachAttachmentsToMessage(message *models.Message) error {
	attachments, err := m.fetchMessageAttachments(message.ID)
	if err != nil {
		m.lo.Error("error fetching message attachments", "error", err)
		return err
	}

	// Attach attachments.
	message.Attachments = attachments

	return nil
}

func (m *Manager) getOutgoingProcessingConversationIDs() []int {
	out := make([]int, 0)
	m.outgoingProcessingMessages.Range(func(_, value any) bool {
		if conversationID, ok := value.(int); ok {
			out = append(out, conversationID)
		}
		return true
	})
	return out
}

// uploadThumbnailForMedia prepares and uploads a thumbnail for an image attachment.
func (m *Manager) uploadThumbnailForMedia(media mmodels.Media, content []byte) error {
	// Create a reader from the content
	file := bytes.NewReader(content)

	// Seek to the beginning of the file
	file.Seek(0, 0)

	// Create the thumbnail
	thumbFile, err := image.CreateThumb(image.DefThumbSize, file)
	if err != nil {
		return fmt.Errorf("error creating thumbnail: %w", err)
	}

	// Generate thumbnail name
	thumbName := fmt.Sprintf("thumb_%s", media.UUID)

	// Upload the thumbnail
	if _, _, err := m.mediaStore.Upload(thumbName, media.ContentType, thumbFile); err != nil {
		m.lo.Error("error uploading thumbnail", "error", err)
		return fmt.Errorf("error uploading thumbnail: %w", err)
	}
	return nil
}

// ProcessIncomingMessageHooks handles automation rules, webhooks, SLA events, and other post-processing
// for incoming messages. This allows other channels to insert messages first and then call this
// function to trigger the necessary hooks.
func (m *Manager) ProcessIncomingMessageHooks(message models.Message, isNewConversation bool) error {
	conversationUUID := message.ConversationUUID

	// Start waiting since clock, cleared when agent replies.
	m.StartConversationWaitingSince(conversationUUID, time.Now())

	// Handle new conversation events.
	if isNewConversation {
		var recipients struct {
			To []string `json:"to"`
		}
		if len(message.Meta) > 0 {
			if err := json.Unmarshal(message.Meta, &recipients); err != nil {
				m.lo.Warn("error reading incoming message recipients", "conversation_uuid", conversationUUID, "error", err)
			}
		}
		for i, addr := range recipients.To {
			recipients.To[i] = stringutil.RemoveConvPlusAddress(addr)
		}

		conversation, err := m.GetConversation(0, conversationUUID, "")
		if err == nil {
			conversation.IncomingTo = recipients.To
			m.webhookStore.TriggerEvent(wmodels.EventConversationCreated, conversation)
			m.automation.EvaluateNewConversationRules(conversation)
		}
		return nil
	}

	// Snapshot before reopening so previous_* filters see the pre-reopen state.
	var previousValues map[string]string
	if preReopen, err := m.GetConversation(0, conversationUUID, ""); err == nil {
		previousValues = amodels.PreviousValues(preReopen)
	}

	// Reopen conversation if it's not Open.
	var reopened bool
	systemUser, err := m.userStore.GetSystemUser()
	if err != nil {
		m.lo.Error("error fetching system user", "error", err)
	} else {
		var err error
		if reopened, err = m.ReOpenConversation(conversationUUID, systemUser); err != nil {
			m.lo.Error("error reopening conversation", "error", err)
		}
	}

	// Create SLA event for next response if a SLA is applied and has next response time set, subsequent agent replies will mark this event as met.
	// This cycle continues for next response time SLA metric.
	conversation, err := m.GetConversation(0, conversationUUID, "")
	if err != nil {
		m.lo.Error("error fetching conversation for incoming message hooks", "conversation_uuid", conversationUUID, "error", err)
	} else {
		// Trigger automations on incoming message event.
		m.automation.EvaluateConversationUpdateRules(conversation, amodels.EventConversationMessageIncoming, previousValues, umodels.User{ID: conversation.ContactID})

		go m.NotifyNewReply(conversation, message, reopened)

		// If assigned to an AI assistant, let it respond to this inbound customer message.
		if m.aiAgent != nil && conversation.AssignedUserID.Valid {
			m.aiAgent.HandleConversationEvent(conversation.ID, conversation.AssignedUserID.Int)
		}

		if conversation.SLAPolicyID.Int == 0 {
			m.lo.Info("no SLA policy applied to conversation, skipping next response SLA event creation")
			return nil
		}
		if deadline, err := m.slaStore.CreateNextResponseSLAEvent(conversation.ID, conversation.AppliedSLAID.Int, conversation.SLAPolicyID.Int, conversation.AssignedTeamID.Int); err != nil && !errors.Is(err, sla.ErrUnmetSLAEventAlreadyExists) {
			m.lo.Error("error creating next response SLA event", "conversation_id", conversation.ID, "error", err)
		} else if !deadline.IsZero() {
			m.lo.Info("next response SLA event created for conversation", "conversation_id", conversation.ID, "deadline", deadline, "sla_policy_id", conversation.SLAPolicyID.Int)
			m.BroadcastConversationUpdate(conversationUUID, map[string]any{
				"next_response_deadline_at": deadline.Format(time.RFC3339),
				"next_response_met_at":      nil,
			})
		}
	}
	return nil
}

// broadcastMessageToWidgetClients sends a message to widget clients if the conversation belongs to a livechat inbox.
func (m *Manager) broadcastMessageToWidgetClients(message *models.Message) {
	conversation, err := m.GetConversation(0, message.ConversationUUID, "")
	if err != nil {
		return
	}

	inboxInstance, err := m.inboxStore.Get(conversation.InboxID)
	if err != nil {
		return
	}

	liveChatInbox, ok := inboxInstance.(*livechat.LiveChat)
	if !ok {
		return
	}

	m.SignAttachmentURLs(message.Attachments)
	m.SignAvatarURL(&message.Author.AvatarURL)
	liveChatInbox.BroadcastMessageToClients(message.ConversationUUID, conversation.ContactID, models.ChatMessage{
		ID:               message.ID,
		UUID:             message.UUID,
		Status:           message.Status,
		ConversationUUID: message.ConversationUUID,
		CreatedAt:        message.CreatedAt,
		Content:          message.Content,
		TextContent:      message.TextContent,
		ContentType:      message.ContentType,
		Author:           message.Author,
		Attachments:      message.Attachments,
		Meta:             message.Meta,
	})
}

// getMediaPreview returns a localized preview string based on attachment type.
func (m *Manager) getMediaPreview(media mmodels.Media) string {
	contentType := media.ContentType
	switch {
	case strings.HasPrefix(contentType, "image/"):
		return m.i18n.T("globals.terms.image")
	case strings.HasPrefix(contentType, "video/"):
		return m.i18n.T("globals.terms.video")
	case strings.HasPrefix(contentType, "audio/"):
		return m.i18n.T("globals.terms.audio")
	default:
		return m.i18n.T("globals.terms.file")
	}
}

// inlineContentID lowercases the uuid to match the content_id the DB stamps from uuid::TEXT.
func inlineContentID(uuid string) string {
	return "ldsk-" + strings.ToLower(uuid)
}

// findExistingMedia resolves an inbound cid to its stored form: ldsk-* is left as-is, others are namespaced by conversation to avoid cross-conversation collisions.
func (m *Manager) findExistingMedia(rawContentID, conversationUUID string) (string, bool, string) {
	storedCID := rawContentID
	if !strings.HasPrefix(rawContentID, "ldsk-") {
		storedCID = conversationUUID + "_" + rawContentID
	}
	exists, mediaUUID, err := m.mediaStore.ContentIDExists(storedCID, conversationUUID)
	if err != nil {
		m.lo.Error("error checking media existence by content ID", "content_id", storedCID, "error", err)
	}
	return storedCID, exists, mediaUUID
}

// emailFromAddress returns the From header for the sender address, named after the inbox From or the from-name template for agents.
func (m *Manager) emailFromAddress(inb inbox.Inbox, message models.Message, sender string) string {
	from := inb.FromAddress()
	if emailInbox, ok := inb.(inbox.EmailInbox); ok && sender != emailInbox.PrimaryAddress() {
		from = sender
		if addr, err := mail.ParseAddress(inb.FromAddress()); err == nil && addr.Name != "" {
			from = (&mail.Address{Name: addr.Name, Address: sender}).String()
		}
	}

	tpl := inb.FromNameTemplate()
	if tpl == "" || message.SenderType != models.SenderTypeAgent {
		return from
	}

	agent, err := m.userStore.GetAgentCachedOrLoad(message.SenderID)
	if err != nil {
		m.lo.Error("error fetching agent for from name template", "error", err, "sender_id", message.SenderID)
		return from
	}
	if agent.IsSystemUser() {
		return from
	}

	addr, err := mail.ParseAddress(from)
	if err != nil {
		m.lo.Error("error parsing inbox from address for name template", "error", err, "from", from)
		return from
	}

	firstName := strings.TrimSpace(agent.FirstName)
	lastName := strings.TrimSpace(agent.LastName)
	data := fromNameVars{
		Agent: fromNameAgent{
			FirstName: firstName,
			LastName:  lastName,
			FullName:  strings.TrimSpace(firstName + " " + lastName),
		},
		Inbox: fromNameInbox{Name: inb.Name()},
	}

	t, err := template.New("from").Parse(tpl)
	if err != nil {
		m.lo.Error("error parsing from name template", "error", err, "template", tpl)
		return from
	}

	var buf bytes.Buffer
	if err := t.Execute(&buf, data); err != nil {
		m.lo.Error("error executing from name template", "error", err, "template", tpl)
		return from
	}

	name := strings.TrimSpace(buf.String())
	if name == "" {
		return from
	}
	addr.Name = name
	return addr.String()
}

func (m *Manager) resolveSendFrom(conversationUUID string, inboxRecord imodels.Inbox, requested string) (string, error) {
	sendable, err := inbox.SendableEmailAddresses(inboxRecord.From, inboxRecord.Aliases)
	if err != nil {
		m.lo.Error("error resolving inbox sender addresses", "inbox_id", inboxRecord.ID, "error", err)
		return "", envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.somethingWentWrong"), nil)
	}

	if requested != "" {
		normalized, err := inbox.NormalizeEmailAddress(requested)
		if err != nil || !slices.Contains(sendable, normalized) {
			return "", envelope.NewError(envelope.InputError, m.i18n.T("validation.invalidEmail"), nil)
		}
		return normalized, nil
	}

	var latest models.Message
	if err := m.q.GetEmailSenderMessage.Get(&latest, conversationUUID); err != nil && err != sql.ErrNoRows {
		m.lo.Error("error fetching latest message for sender address", "conversation_uuid", conversationUUID, "error", err)
		return "", envelope.NewError(envelope.GeneralError, m.i18n.T("globals.messages.somethingWentWrong"), nil)
	}
	previous := latest.SendFrom()
	if latest.Type == models.MessageIncoming {
		previous = latest.InboxAddress()
	}
	if normalized, err := inbox.NormalizeEmailAddress(previous); err == nil && slices.Contains(sendable, normalized) {
		return normalized, nil
	}
	return sendable[0], nil
}