gochat/internal/service/conversation_maintenance_worker.go

package service

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/gochat/gochat/internal/campaign"
	"github.com/gochat/gochat/internal/channel"
	"github.com/gochat/gochat/internal/model"
	"github.com/gochat/gochat/internal/worker"
	"gorm.io/gorm"
)

const (
	TaskTypeScheduledTriggerItems            = "scheduled:trigger_items"
	TaskTypeCampaignTriggerOneoff            = "campaign:trigger_oneoff"
	TaskTypeConversationReopenSnoozed        = "conversation:reopen_snoozed"
	TaskTypeConversationResolutionScheduler  = "account:conversations_resolution_scheduler"
	TaskTypeConversationResolutionForAccount = "conversation:resolution"
	TaskTypeConversationUpdateMessageStatus  = "conversation:update_message_status"
	TaskTypeConversationBulkAction           = "conversation:bulk_action"
	TaskTypeContactBulkAction                = "contact:bulk_action"
)

const (
	scheduledItemsInterval      = time.Hour
	scheduledItemsLookback      = 3 * 24 * time.Hour
	conversationResolutionLimit = 100
)

type campaignTriggerOneoffJob struct {
	CampaignID uint `json:"campaign_id"`
}

type conversationResolutionJob struct {
	AccountID uint `json:"account_id"`
}

type conversationUpdateMessageStatusJob struct {
	ConversationID uint   `json:"conversation_id"`
	Timestamp      int64  `json:"timestamp"`
	Status         string `json:"status"`
}

type ConversationBulkActionParams struct {
	Type         string                       `json:"type"`
	ActionName   string                       `json:"action_name,omitempty"`
	IDs          []uint                       `json:"ids"`
	Fields       ConversationBulkActionFields `json:"fields,omitempty"`
	Labels       ConversationBulkActionLabels `json:"labels,omitempty"`
	SnoozedUntil string                       `json:"snoozed_until,omitempty"`
}

type ConversationBulkActionFields struct {
	Status     *string `json:"status,omitempty"`
	AssigneeID *uint   `json:"assignee_id,omitempty"`
	TeamID     *uint   `json:"team_id,omitempty"`
}

type ConversationBulkActionLabels struct {
	Add    []string `json:"add,omitempty"`
	Remove []string `json:"remove,omitempty"`
}

type ContactBulkActionParams struct {
	Type       string                       `json:"type"`
	ActionName string                       `json:"action_name,omitempty"`
	IDs        []uint                       `json:"ids"`
	Labels     ConversationBulkActionLabels `json:"labels,omitempty"`
}

type conversationBulkActionJob struct {
	AccountID uint                         `json:"account_id"`
	UserID    uint                         `json:"user_id,omitempty"`
	Params    ConversationBulkActionParams `json:"params"`
}

type contactBulkActionJob struct {
	AccountID uint                    `json:"account_id"`
	UserID    uint                    `json:"user_id,omitempty"`
	Params    ContactBulkActionParams `json:"params"`
}

var conversationMaintenanceRegistrations sync.Map

// RegisterConversationMaintenanceJobs wires Chatwoot scheduled maintenance jobs
// into the durable worker: TriggerScheduledItemsJob fans out to campaign,
// snooze-reopen, and auto-resolution jobs.
func RegisterConversationMaintenanceJobs(wp *worker.WorkerPool, db *gorm.DB, dispatchers ...*channel.Dispatcher) {
	registerConversationMaintenanceJobsWithNow(wp, db, time.Now, dispatchers...)
}

func RegisterContactBulkActionSearchIndexer(wp *worker.WorkerPool, db *gorm.DB, indexer SearchIndexer) {
	RegisterConversationMaintenanceSearchIndexer(wp, db, indexer)
}

func RegisterBulkActionSearchIndexer(wp *worker.WorkerPool, db *gorm.DB, indexer SearchIndexer) {
	RegisterConversationMaintenanceSearchIndexer(wp, db, indexer)
}

func RegisterConversationMaintenanceSearchIndexer(wp *worker.WorkerPool, db *gorm.DB, indexer SearchIndexer) {
	if wp == nil || db == nil {
		return
	}
	now := time.Now
	if existing, ok := conversationMaintenanceRegistrations.Load(wp); ok {
		if runner, ok := existing.(*conversationMaintenanceRunner); ok && runner.now != nil {
			now = runner.now
		}
	}
	runner := &conversationMaintenanceRunner{wp: wp, db: db, now: now, searchIndexer: indexer}
	wp.Register(TaskTypeConversationReopenSnoozed, runner.performReopenSnoozed)
	wp.Register(TaskTypeConversationResolutionForAccount, runner.performResolutionForAccount)
	wp.Register(TaskTypeConversationUpdateMessageStatus, runner.performUpdateMessageStatus)
	wp.Register(TaskTypeConversationBulkAction, runner.performConversationBulkAction)
	wp.Register(TaskTypeContactBulkAction, runner.performContactBulkAction)
}

func registerConversationMaintenanceJobsWithNow(wp *worker.WorkerPool, db *gorm.DB, now func() time.Time, dispatchers ...*channel.Dispatcher) {
	if wp == nil || db == nil {
		return
	}
	var dispatcher *channel.Dispatcher
	if len(dispatchers) > 0 {
		dispatcher = dispatchers[0]
	}
	runner := &conversationMaintenanceRunner{wp: wp, db: db, now: now, dispatcher: dispatcher}
	if _, loaded := conversationMaintenanceRegistrations.LoadOrStore(wp, runner); loaded {
		return
	}
	wp.Register(TaskTypeScheduledTriggerItems, runner.performScheduledTriggerItems)
	wp.Register(TaskTypeCampaignTriggerOneoff, runner.performCampaignTriggerOneoff)
	wp.Register(TaskTypeConversationReopenSnoozed, runner.performReopenSnoozed)
	wp.Register(TaskTypeConversationResolutionScheduler, runner.performResolutionScheduler)
	wp.Register(TaskTypeConversationResolutionForAccount, runner.performResolutionForAccount)
	wp.Register(TaskTypeConversationUpdateMessageStatus, runner.performUpdateMessageStatus)
	wp.Register(TaskTypeConversationBulkAction, runner.performConversationBulkAction)
	wp.Register(TaskTypeContactBulkAction, runner.performContactBulkAction)
}

func EnqueueScheduledItemsTrigger(ctx context.Context, wp *worker.WorkerPool, scheduledAt time.Time) (*model.BackgroundJob, error) {
	if wp == nil {
		return nil, nil
	}
	return wp.Enqueue(ctx, TaskTypeScheduledTriggerItems, nil,
		worker.WithQueue("scheduled_jobs"),
		worker.WithScheduledAt(scheduledAt),
		worker.WithMaxAttempts(3),
		worker.WithIdempotencyKey(scheduledItemsIdempotencyKey(scheduledAt)),
	)
}

func EnqueueConversationMessageStatusUpdate(ctx context.Context, wp *worker.WorkerPool, conversationID uint, timestamp time.Time, status string) (*model.BackgroundJob, error) {
	if wp == nil {
		return nil, nil
	}
	if status == "" {
		status = string(model.MessageStatusRead)
	}
	payload := conversationUpdateMessageStatusJob{ConversationID: conversationID, Timestamp: timestamp.UTC().Unix(), Status: status}
	return wp.Enqueue(ctx, TaskTypeConversationUpdateMessageStatus, payload,
		worker.WithQueue("deferred"),
		worker.WithMaxAttempts(3),
		worker.WithIdempotencyKey(fmt.Sprintf("conversation:update_message_status:%d:%s:%d", conversationID, status, payload.Timestamp)),
	)
}

func EnqueueConversationBulkAction(ctx context.Context, wp *worker.WorkerPool, accountID, userID uint, params ConversationBulkActionParams) (*model.BackgroundJob, error) {
	if wp == nil {
		return nil, nil
	}
	if accountID == 0 {
		return nil, fmt.Errorf("invalid conversation bulk action payload: account_id=%d ids=%v", accountID, params.IDs)
	}
	return wp.Enqueue(ctx, TaskTypeConversationBulkAction, conversationBulkActionJob{AccountID: accountID, UserID: userID, Params: params},
		worker.WithQueue("medium"),
		worker.WithMaxAttempts(3),
	)
}

func EnqueueContactBulkAction(ctx context.Context, wp *worker.WorkerPool, accountID, userID uint, params ContactBulkActionParams) (*model.BackgroundJob, error) {
	if wp == nil {
		return nil, nil
	}
	if accountID == 0 {
		return nil, fmt.Errorf("invalid contact bulk action payload: account_id=%d ids=%v", accountID, params.IDs)
	}
	return wp.Enqueue(ctx, TaskTypeContactBulkAction, contactBulkActionJob{AccountID: accountID, UserID: userID, Params: params},
		worker.WithQueue("medium"),
		worker.WithMaxAttempts(3),
	)
}

func scheduledItemsIdempotencyKey(scheduledAt time.Time) string {
	bucket := scheduledAt.UTC().Truncate(scheduledItemsInterval).Unix()
	return fmt.Sprintf("scheduled:trigger_items:%d", bucket)
}

type conversationMaintenanceRunner struct {
	wp            *worker.WorkerPool
	db            *gorm.DB
	now           func() time.Time
	searchIndexer SearchIndexer
	dispatcher    *channel.Dispatcher
}

func (r *conversationMaintenanceRunner) performScheduledTriggerItems(ctx context.Context, job *model.BackgroundJob) error {
	now := r.now()
	var campaignIDs []uint
	if err := r.db.WithContext(ctx).Model(&campaign.Campaign{}).
		Where("campaign_type = ? AND campaign_status = ? AND enabled = ?", campaign.CampaignTypeOneOff, campaign.CampaignStatusActive, true).
		Where("scheduled_at BETWEEN ? AND ?", now.Add(-scheduledItemsLookback), now).
		Pluck("id", &campaignIDs).Error; err != nil {
		return fmt.Errorf("find due one-off campaigns: %w", err)
	}
	for _, campaignID := range campaignIDs {
		_, err := r.wp.Enqueue(ctx, TaskTypeCampaignTriggerOneoff, campaignTriggerOneoffJob{CampaignID: campaignID},
			worker.WithQueue("low"),
			worker.WithMaxAttempts(3),
			worker.WithIdempotencyKey(fmt.Sprintf("campaign:trigger_oneoff:%d", campaignID)),
		)
		if err != nil {
			return fmt.Errorf("enqueue campaign %d: %w", campaignID, err)
		}
	}
	if _, err := r.wp.Enqueue(ctx, TaskTypeConversationReopenSnoozed, nil, worker.WithQueue("low"), worker.WithMaxAttempts(3)); err != nil {
		return fmt.Errorf("enqueue reopen snoozed conversations: %w", err)
	}
	if _, err := r.wp.Enqueue(ctx, TaskTypeConversationResolutionScheduler, nil, worker.WithQueue("scheduled_jobs"), worker.WithMaxAttempts(3)); err != nil {
		return fmt.Errorf("enqueue conversation resolution scheduler: %w", err)
	}
	_, err := EnqueueScheduledItemsTrigger(ctx, r.wp, now.Add(scheduledItemsInterval))
	return err
}

func (r *conversationMaintenanceRunner) performCampaignTriggerOneoff(ctx context.Context, job *model.BackgroundJob) error {
	var payload campaignTriggerOneoffJob
	if err := json.Unmarshal(job.Payload, &payload); err != nil {
		return fmt.Errorf("unmarshal one-off campaign job: %w", err)
	}
	if payload.CampaignID == 0 {
		return fmt.Errorf("invalid one-off campaign job payload: %#v", payload)
	}

	claimed := false
	if err := r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		result := tx.Model(&campaign.Campaign{}).
			Where("id = ? AND campaign_type = ? AND campaign_status = ? AND enabled = ?", payload.CampaignID, campaign.CampaignTypeOneOff, campaign.CampaignStatusActive, true).
			Update("campaign_status", campaign.CampaignStatusProcessing)
		if result.Error != nil {
			return result.Error
		}
		claimed = result.RowsAffected == 1
		return nil
	}); err != nil {
		return fmt.Errorf("claim one-off campaign %d: %w", payload.CampaignID, err)
	}
	if !claimed {
		return nil
	}
	if err := campaign.NewCampaignService(r.db, r.dispatcher).TriggerCampaign(ctx, payload.CampaignID); err != nil {
		_ = r.db.WithContext(ctx).Model(&campaign.Campaign{}).Where("id = ?", payload.CampaignID).Update("campaign_status", campaign.CampaignStatusActive).Error
		return err
	}
	return r.db.WithContext(ctx).Model(&campaign.Campaign{}).Where("id = ?", payload.CampaignID).Update("campaign_status", campaign.CampaignStatusCompleted).Error
}

func (r *conversationMaintenanceRunner) performReopenSnoozed(ctx context.Context, job *model.BackgroundJob) error {
	now := r.now()
	nowUnix := now.Unix()
	lookbackUnix := now.Add(-scheduledItemsLookback).Unix()
	var conversationIDs []uint
	if err := r.db.WithContext(ctx).Model(&model.Conversation{}).
		Where("status = ?", string(model.ConversationStatusSnoozed)).
		Where("snoozed_until BETWEEN ? AND ?", lookbackUnix, nowUnix).
		Pluck("id", &conversationIDs).Error; err != nil {
		return err
	}
	if len(conversationIDs) == 0 {
		return nil
	}
	updates := map[string]any{
		"status":        string(model.ConversationStatusOpen),
		"snoozed_until": nil,
		"resumed_at":    now,
	}
	if err := r.db.WithContext(ctx).Model(&model.Conversation{}).
		Where("id IN ?", conversationIDs).
		Updates(updates).Error; err != nil {
		return err
	}
	return r.indexConversationSearchDocumentsByIDs(ctx, conversationIDs)
}

func (r *conversationMaintenanceRunner) performResolutionScheduler(ctx context.Context, job *model.BackgroundJob) error {
	var accountIDs []uint
	if err := r.db.WithContext(ctx).Model(&model.Account{}).
		Where("auto_resolve_duration > 0").
		Pluck("id", &accountIDs).Error; err != nil {
		return fmt.Errorf("find auto-resolve accounts: %w", err)
	}
	for _, accountID := range accountIDs {
		if _, err := r.wp.Enqueue(ctx, TaskTypeConversationResolutionForAccount, conversationResolutionJob{AccountID: accountID}, worker.WithQueue("low"), worker.WithMaxAttempts(3)); err != nil {
			return fmt.Errorf("enqueue account conversation resolution %d: %w", accountID, err)
		}
	}
	return nil
}

func (r *conversationMaintenanceRunner) performResolutionForAccount(ctx context.Context, job *model.BackgroundJob) error {
	var payload conversationResolutionJob
	if err := json.Unmarshal(job.Payload, &payload); err != nil {
		return fmt.Errorf("unmarshal conversation resolution job: %w", err)
	}
	if payload.AccountID == 0 {
		return fmt.Errorf("invalid conversation resolution job payload: %#v", payload)
	}
	var account model.Account
	if err := r.db.WithContext(ctx).First(&account, payload.AccountID).Error; err != nil {
		return fmt.Errorf("load auto-resolve account %d: %w", payload.AccountID, err)
	}
	if account.AutoResolveDuration <= 0 {
		return nil
	}
	cutoff := r.now().Add(-time.Duration(account.AutoResolveDuration) * time.Minute).Unix()
	now := r.now()
	var conversationIDs []uint
	if err := r.db.WithContext(ctx).Model(&model.Conversation{}).
		Where("account_id = ? AND status = ? AND contact_id <> 0", account.ID, string(model.ConversationStatusOpen)).
		Where("last_activity_at IS NOT NULL AND last_activity_at < ?", cutoff).
		Limit(conversationResolutionLimit).
		Pluck("id", &conversationIDs).Error; err != nil {
		return err
	}
	if len(conversationIDs) == 0 {
		return nil
	}
	updates := map[string]any{
		"status":      string(model.ConversationStatusResolved),
		"resolved_at": now,
	}
	if err := r.db.WithContext(ctx).Model(&model.Conversation{}).
		Where("id IN ?", conversationIDs).
		Updates(updates).Error; err != nil {
		return err
	}
	return r.indexConversationSearchDocumentsByIDs(ctx, conversationIDs)
}

func (r *conversationMaintenanceRunner) performUpdateMessageStatus(ctx context.Context, job *model.BackgroundJob) error {
	var payload conversationUpdateMessageStatusJob
	if err := json.Unmarshal(job.Payload, &payload); err != nil {
		return fmt.Errorf("unmarshal conversation message status job: %w", err)
	}
	if payload.ConversationID == 0 || payload.Timestamp == 0 {
		return fmt.Errorf("invalid conversation message status job payload: %#v", payload)
	}
	if !validConversationMessageStatus(payload.Status) {
		return nil
	}

	var conversation model.Conversation
	if err := r.db.WithContext(ctx).First(&conversation, payload.ConversationID).Error; err != nil {
		if err == gorm.ErrRecordNotFound {
			return nil
		}
		return fmt.Errorf("load conversation %d for message status update: %w", payload.ConversationID, err)
	}

	var messageIDs []uint
	if err := r.db.WithContext(ctx).Model(&model.Message{}).
		Where("conversation_id = ?", conversation.ID).
		Where("status IN ?", []string{string(model.MessageStatusSent), string(model.MessageStatusDelivered)}).
		Where("message_type <> ?", "incoming").
		Where("created_at <= ?", time.Unix(payload.Timestamp, 0).UTC()).
		Pluck("id", &messageIDs).Error; err != nil {
		return err
	}
	if len(messageIDs) == 0 {
		return nil
	}
	if err := r.db.WithContext(ctx).Model(&model.Message{}).
		Where("id IN ?", messageIDs).
		Update("status", payload.Status).Error; err != nil {
		return err
	}
	if err := r.indexMessageSearchDocuments(ctx, conversation.AccountID, messageIDs); err != nil {
		return err
	}
	return r.indexConversationSearchDocumentsByIDs(ctx, []uint{conversation.ID})
}

func validConversationMessageStatus(status string) bool {
	switch status {
	case string(model.MessageStatusRead), string(model.MessageStatusDelivered):
		return true
	default:
		return false
	}
}

func (r *conversationMaintenanceRunner) performConversationBulkAction(ctx context.Context, job *model.BackgroundJob) error {
	var payload conversationBulkActionJob
	if err := json.Unmarshal(job.Payload, &payload); err != nil {
		return fmt.Errorf("unmarshal conversation bulk action job: %w", err)
	}
	if payload.AccountID == 0 {
		return fmt.Errorf("invalid conversation bulk action job payload: %#v", payload)
	}
	if len(payload.Params.IDs) == 0 {
		return nil
	}
	if payload.Params.Type != "" && payload.Params.Type != "Conversation" {
		return nil
	}

	var conversations []model.Conversation
	if err := r.db.WithContext(ctx).
		Where("account_id = ? AND display_id IN ?", payload.AccountID, payload.Params.IDs).
		Find(&conversations).Error; err != nil {
		return fmt.Errorf("load bulk action conversations: %w", err)
	}
	updatedConversationIDs := make([]uint, 0, len(conversations))
	for i := range conversations {
		conversation := conversations[i]
		updates := map[string]any{}
		if payload.Params.Fields.Status != nil && *payload.Params.Fields.Status != "" {
			updates["status"] = *payload.Params.Fields.Status
			now := r.now()
			switch *payload.Params.Fields.Status {
			case string(model.ConversationStatusResolved):
				updates["resolved_at"] = now
			case string(model.ConversationStatusOpen):
				updates["resumed_at"] = now
				updates["snoozed_until"] = nil
			}
		}
		if payload.Params.Fields.AssigneeID != nil {
			updates["assignee_id"] = payload.Params.Fields.AssigneeID
		}
		if payload.Params.Fields.TeamID != nil {
			updates["team_id"] = payload.Params.Fields.TeamID
		}
		if payload.Params.SnoozedUntil != "" {
			if snoozedUntil, ok := parseBulkActionTime(payload.Params.SnoozedUntil); ok {
				updates["snoozed_until"] = snoozedUntil.Unix()
			}
		}
		if len(payload.Params.Labels.Add) > 0 || len(payload.Params.Labels.Remove) > 0 {
			updates["labels"] = mergeConversationLabels(conversation.Labels, payload.Params.Labels.Add, payload.Params.Labels.Remove)
		}
		if len(updates) == 0 {
			continue
		}
		if err := r.db.WithContext(ctx).Model(&model.Conversation{}).Where("id = ?", conversation.ID).Updates(updates).Error; err != nil {
			return fmt.Errorf("bulk update conversation %d: %w", conversation.ID, err)
		}
		updatedConversationIDs = append(updatedConversationIDs, conversation.ID)
	}
	return r.indexConversationSearchDocuments(ctx, payload.AccountID, updatedConversationIDs)
}

func (r *conversationMaintenanceRunner) indexConversationSearchDocuments(ctx context.Context, accountID uint, conversationIDs []uint) error {
	if r.searchIndexer == nil || len(conversationIDs) == 0 {
		return nil
	}
	var conversations []model.Conversation
	if err := r.db.WithContext(ctx).
		Preload("Contact").
		Preload("Inbox").
		Preload("Assignee").
		Preload("Messages", func(db *gorm.DB) *gorm.DB { return db.Order("created_at ASC, id ASC") }).
		Where("account_id = ? AND id IN ?", accountID, conversationIDs).
		Find(&conversations).Error; err != nil {
		return err
	}
	for i := range conversations {
		if err := r.searchIndexer.IndexConversation(ctx, &conversations[i]); err != nil {
			return err
		}
	}
	return nil
}

func (r *conversationMaintenanceRunner) indexConversationSearchDocumentsByIDs(ctx context.Context, conversationIDs []uint) error {
	if r.searchIndexer == nil || len(conversationIDs) == 0 {
		return nil
	}
	var conversations []model.Conversation
	if err := r.db.WithContext(ctx).
		Preload("Contact").
		Preload("Inbox").
		Preload("Assignee").
		Preload("Messages", func(db *gorm.DB) *gorm.DB { return db.Order("created_at ASC, id ASC") }).
		Where("id IN ?", conversationIDs).
		Find(&conversations).Error; err != nil {
		return err
	}
	for i := range conversations {
		if err := r.searchIndexer.IndexConversation(ctx, &conversations[i]); err != nil {
			return err
		}
	}
	return nil
}

func (r *conversationMaintenanceRunner) indexMessageSearchDocuments(ctx context.Context, accountID uint, messageIDs []uint) error {
	if r.searchIndexer == nil || len(messageIDs) == 0 {
		return nil
	}
	var messages []model.Message
	if err := r.db.WithContext(ctx).
		Preload("Conversation").
		Preload("Attachments", func(db *gorm.DB) *gorm.DB { return db.Order("id ASC") }).
		Where("account_id = ? AND id IN ?", accountID, messageIDs).
		Find(&messages).Error; err != nil {
		return err
	}
	for i := range messages {
		if err := r.searchIndexer.IndexMessage(ctx, &messages[i]); err != nil {
			return err
		}
	}
	return nil
}

func (r *conversationMaintenanceRunner) performContactBulkAction(ctx context.Context, job *model.BackgroundJob) error {
	var payload contactBulkActionJob
	if err := json.Unmarshal(job.Payload, &payload); err != nil {
		return fmt.Errorf("unmarshal contact bulk action job: %w", err)
	}
	if payload.AccountID == 0 {
		return fmt.Errorf("invalid contact bulk action job payload: %#v", payload)
	}
	if payload.Params.Type != "" && payload.Params.Type != "Contact" {
		return nil
	}
	if len(payload.Params.IDs) == 0 {
		return nil
	}

	switch {
	case payload.Params.ActionName == "delete":
		contactIDs, err := scopedContactIDs(ctx, r.db, payload.AccountID, payload.Params.IDs)
		if err != nil {
			return err
		}
		if len(contactIDs) == 0 {
			return nil
		}
		if err := r.db.WithContext(ctx).
			Where("account_id = ? AND id IN ?", payload.AccountID, contactIDs).
			Delete(&model.Contact{}).Error; err != nil {
			return err
		}
		r.deleteContactSearchIndexes(ctx, payload.AccountID, contactIDs)
		return nil
	case len(payload.Params.Labels.Add) > 0:
		contactIDs, err := bulkAddContactLabels(ctx, r.db, payload.AccountID, payload.Params.IDs, payload.Params.Labels.Add)
		if err != nil {
			return err
		}
		return r.indexContactSearchDocuments(ctx, payload.AccountID, contactIDs)
	case len(payload.Params.Labels.Remove) > 0:
		contactIDs, err := bulkRemoveContactLabels(ctx, r.db, payload.AccountID, payload.Params.IDs, payload.Params.Labels.Remove)
		if err != nil {
			return err
		}
		return r.indexContactSearchDocuments(ctx, payload.AccountID, contactIDs)
	default:
		return nil
	}
}

func bulkAddContactLabels(ctx context.Context, db *gorm.DB, accountID uint, contactIDs []uint, labels []string) ([]uint, error) {
	labels = normalizeContactServiceLabels(labels)
	if len(contactIDs) == 0 || len(labels) == 0 {
		return nil, nil
	}
	scopedIDs, err := scopedContactIDs(ctx, db, accountID, contactIDs)
	if err != nil {
		return nil, err
	}
	if len(scopedIDs) == 0 {
		return nil, nil
	}
	err = db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		for _, label := range labels {
			tag := model.Tag{AccountID: accountID, Name: label}
			if err := tx.Where("account_id = ? AND name = ?", accountID, label).FirstOrCreate(&tag).Error; err != nil {
				return err
			}
			for _, contactID := range scopedIDs {
				contactLabel := model.ContactLabel{AccountID: accountID, ContactID: contactID, TagID: tag.ID}
				if err := tx.Where("account_id = ? AND contact_id = ? AND tag_id = ?", accountID, contactID, tag.ID).FirstOrCreate(&contactLabel).Error; err != nil {
					return err
				}
			}
		}
		return nil
	})
	return scopedIDs, err
}

func bulkRemoveContactLabels(ctx context.Context, db *gorm.DB, accountID uint, contactIDs []uint, labels []string) ([]uint, error) {
	labels = normalizeContactServiceLabels(labels)
	if len(contactIDs) == 0 || len(labels) == 0 {
		return nil, nil
	}
	scopedIDs, err := scopedContactIDs(ctx, db, accountID, contactIDs)
	if err != nil {
		return nil, err
	}
	if len(scopedIDs) == 0 {
		return nil, nil
	}
	var tagIDs []uint
	if err := db.WithContext(ctx).Model(&model.Tag{}).
		Where("account_id = ? AND name IN ?", accountID, labels).
		Pluck("id", &tagIDs).Error; err != nil {
		return nil, err
	}
	if len(tagIDs) == 0 {
		return scopedIDs, nil
	}
	err = db.WithContext(ctx).
		Where("account_id = ? AND contact_id IN ? AND tag_id IN ?", accountID, scopedIDs, tagIDs).
		Delete(&model.ContactLabel{}).Error
	return scopedIDs, err
}

func scopedContactIDs(ctx context.Context, db *gorm.DB, accountID uint, contactIDs []uint) ([]uint, error) {
	if len(contactIDs) == 0 {
		return nil, nil
	}
	var ids []uint
	err := db.WithContext(ctx).Model(&model.Contact{}).
		Where("account_id = ? AND id IN ?", accountID, contactIDs).
		Pluck("id", &ids).Error
	return ids, err
}

func (r *conversationMaintenanceRunner) indexContactSearchDocuments(ctx context.Context, accountID uint, contactIDs []uint) error {
	if r.searchIndexer == nil || len(contactIDs) == 0 {
		return nil
	}
	labelsByContactID, err := contactLabelsForSearch(ctx, r.db, accountID, contactIDs)
	if err != nil {
		return err
	}
	var contacts []model.Contact
	if err := r.db.WithContext(ctx).Where("account_id = ? AND id IN ?", accountID, contactIDs).Find(&contacts).Error; err != nil {
		return err
	}
	for i := range contacts {
		contacts[i].Labels = labelsByContactID[contacts[i].ID]
		if err := r.searchIndexer.IndexContact(ctx, &contacts[i]); err != nil {
			return err
		}
	}
	return nil
}

func (r *conversationMaintenanceRunner) deleteContactSearchIndexes(ctx context.Context, accountID uint, contactIDs []uint) {
	if r.searchIndexer == nil {
		return
	}
	for _, contactID := range contactIDs {
		logSearchIndexError("contact", contactID, r.searchIndexer.DeleteContact(ctx, accountID, contactID))
	}
}

func parseBulkActionTime(value string) (time.Time, bool) {
	if value == "" {
		return time.Time{}, false
	}
	if ts, err := time.Parse(time.RFC3339, value); err == nil {
		return ts.UTC(), true
	}
	if ts, err := time.Parse("2006-01-02T15:04:05.000Z", value); err == nil {
		return ts.UTC(), true
	}
	return time.Time{}, false
}

func mergeConversationLabels(existing string, add, remove []string) string {
	labels := map[string]bool{}
	order := []string{}
	for _, label := range splitConversationLabels(existing) {
		if !labels[label] {
			labels[label] = true
			order = append(order, label)
		}
	}
	for _, label := range add {
		label = strings.TrimSpace(label)
		if label == "" || labels[label] {
			continue
		}
		labels[label] = true
		order = append(order, label)
	}
	for _, label := range remove {
		delete(labels, strings.TrimSpace(label))
	}
	merged := make([]string, 0, len(order))
	for _, label := range order {
		if labels[label] {
			merged = append(merged, label)
		}
	}
	return strings.Join(merged, ",")
}

func splitConversationLabels(existing string) []string {
	parts := strings.Split(existing, ",")
	labels := make([]string, 0, len(parts))
	for _, part := range parts {
		label := strings.TrimSpace(part)
		if label != "" {
			labels = append(labels, label)
		}
	}
	return labels
}