package advanced

import (
	"context"
	"encoding/json"
	"fmt"
	"hash/fnv"
	"math/rand"
	"net/mail"
	"regexp"
	"strings"
	"time"

	"github.com/rs/zerolog/log"
	"github.com/warmbly/warmbly/internal/pkg/emailverify"
	"github.com/warmbly/warmbly/internal/utils/validate"

	"github.com/google/uuid"
	"github.com/warmbly/warmbly/internal/app/listgate"
	"github.com/warmbly/warmbly/internal/app/replyclassify"
	warmupapp "github.com/warmbly/warmbly/internal/app/warmup"
	"github.com/warmbly/warmbly/internal/errx"
	"github.com/warmbly/warmbly/internal/models"
	"github.com/warmbly/warmbly/internal/pkg/warmlint"
	"github.com/warmbly/warmbly/internal/repository"
	"github.com/warmbly/warmbly/internal/tasks/proto"
	"github.com/warmbly/warmbly/internal/tasksched"
)

type Service interface {
	GetOrganizationSettings(ctx context.Context, organizationID uuid.UUID) (*models.AdvancedOutreachSettings, *errx.Error)
	UpdateOrganizationSettings(ctx context.Context, organizationID, updatedBy uuid.UUID, settings *models.AdvancedOutreachSettings) *errx.Error

	GetCampaignSettings(ctx context.Context, campaignID uuid.UUID) (*models.CampaignAdvancedSettings, *errx.Error)
	// EffectiveSettings is the org settings with the campaign's overrides
	// merged in — what every per-campaign decision must read.
	EffectiveSettings(ctx context.Context, organizationID, campaignID uuid.UUID) (*models.AdvancedOutreachSettings, *errx.Error)
	UpdateCampaignSettings(ctx context.Context, campaignID uuid.UUID, settings *models.AdvancedOutreachSettings) *errx.Error

	ListABVariants(ctx context.Context, campaignID uuid.UUID) ([]models.CampaignABVariant, *errx.Error)
	CreateABVariant(ctx context.Context, campaignID uuid.UUID, req *models.CreateCampaignABVariantRequest) (*models.CampaignABVariant, *errx.Error)
	UpdateABVariant(ctx context.Context, campaignID, variantID uuid.UUID, req *models.UpdateCampaignABVariantRequest) (*models.CampaignABVariant, *errx.Error)
	DeleteABVariant(ctx context.Context, campaignID, variantID uuid.UUID) *errx.Error

	RunPreflight(ctx context.Context, organizationID, campaignID uuid.UUID) (*models.PreflightReport, *errx.Error)
	GetDeliverabilityDashboard(ctx context.Context, organizationID uuid.UUID, from, to time.Time) (*models.DeliverabilityDashboard, *errx.Error)

	IngestDeliverabilityEvent(ctx context.Context, organizationID uuid.UUID, req *models.IngestDeliverabilityEventRequest) *errx.Error

	// RecordInboundBounce resolves a permanent NDR (parsed worker-side) back to
	// the original campaign send via its Message-ID and records a bounce
	// deliverability event. Best-effort: unresolvable bounces are a no-op.
	RecordInboundBounce(ctx context.Context, emailAccountID uuid.UUID, originalMessageID, failedRecipient, reason string) *errx.Error
	// RecordInboundComplaint resolves an abuse feedback report (RFC 5965) back
	// to the campaign send it names and records it as a complaint.
	RecordInboundComplaint(ctx context.Context, emailAccountID uuid.UUID, originalMessageID, complainedRecipient, provider string) *errx.Error

	ShouldSuppressRecipient(ctx context.Context, organizationID uuid.UUID, recipient string) (bool, string, *errx.Error)
	// Unsubscribe suppresses a contact in response to a List-Unsubscribe action
	// (one-click POST or the manual link). Always suppresses — it's an explicit
	// recipient request, independent of the auto-suppress settings.
	Unsubscribe(ctx context.Context, campaignID, contactID uuid.UUID) *errx.Error
	// UnsubscribeFromLink is Unsubscribe for a verified link token: the
	// organization in the token must own the campaign, and via names the
	// mechanism ("one_click" for the RFC 8058 POST, "link" for a click).
	UnsubscribeFromLink(ctx context.Context, organizationID, campaignID, contactID uuid.UUID, via string) *errx.Error
	// Resubscribe undoes a recipient's own unsubscribe from the hosted page.
	// Only an entry the recipient made (source "unsubscribe") is removed; a
	// bounce, complaint or manual entry stays.
	Resubscribe(ctx context.Context, organizationID, contactID uuid.UUID) *errx.Error

	// The workspace suppression list.
	ListSuppressions(ctx context.Context, organizationID uuid.UUID, q string, beforeAt *time.Time, beforeID *uuid.UUID, limit int) ([]models.SuppressedRecipient, *errx.Error)
	AddSuppressions(ctx context.Context, organizationID, actorID uuid.UUID, req *models.AddSuppressionsRequest) (*models.AddSuppressionsResult, *errx.Error)
	RemoveSuppression(ctx context.Context, organizationID, id uuid.UUID) (*models.SuppressedRecipient, *errx.Error)
	SelectVariant(ctx context.Context, organizationID, campaignID, contactID, sequenceID uuid.UUID, subject, bodyHTML, bodyPlain string) (*models.VariantSelection, *errx.Error)
	OptimizeSendTime(ctx context.Context, organizationID uuid.UUID, contact *models.Contact, base time.Time) (time.Time, *errx.Error)

	StartTaskExecution(ctx context.Context, taskID uuid.UUID, executionKey string, metadata map[string]interface{}) (bool, *errx.Error)
	CompleteTaskExecution(ctx context.Context, taskID uuid.UUID, executionKey, status string, metadata map[string]interface{}) *errx.Error
	CaptureTaskDeadLetter(ctx context.Context, taskID uuid.UUID, taskType string, payload map[string]interface{}, lastError string, attempts int) *errx.Error
	ListDeadLetters(ctx context.Context, organizationID uuid.UUID, status string, limit int) ([]models.TaskDeadLetter, *errx.Error)
	ReplayDeadLetter(ctx context.Context, organizationID, deadLetterID uuid.UUID) *errx.Error

	ProcessIncomingReply(ctx context.Context, emailAccountID uuid.UUID, msg *models.EmailMessageStoreData) *errx.Error
	GetABWinnerAnalysis(ctx context.Context, organizationID, campaignID uuid.UUID) (*models.ABWinnerAnalysis, *errx.Error)

	// CreateContactTask creates a CRM task for a contact, used by the campaign
	// "create task" action node. createdBy is the campaign owner; the task's
	// AssignedTo (set in data) is the teammate chosen on the step. Records a
	// task_created activity on the contact.
	CreateContactTask(ctx context.Context, orgID, createdBy uuid.UUID, data *models.CreateCRMTask) (*models.CRMTask, *errx.Error)

	// CreateContactDeal opens a CRM deal for a contact, used by the campaign
	// "create_deal" action node (typically off a positive reply branch). data
	// carries pipeline/stage/name/value/currency; CampaignID is stamped for deal
	// attribution. Records a deal-created activity on the contact.
	CreateContactDeal(ctx context.Context, orgID uuid.UUID, createdBy uuid.UUID, data *models.CreateDeal) (*models.Deal, *errx.Error)

	// MoveContactDealStage moves the contact's most-recent OPEN deal in
	// pipelineID to stageID, used by the campaign "move_deal_stage" action node.
	// When the contact has no open deal in that pipeline it is a no-op (returns
	// nil, nil) rather than an error, so a chained reply automation doesn't fail
	// just because a deal hasn't been created yet.
	MoveContactDealStage(ctx context.Context, orgID, contactID, pipelineID, stageID uuid.UUID) (*models.Deal, *errx.Error)

	// LabelThread additively applies unibox conversation labels (categories owned
	// by userID) to a thread, for the "label_email" automation action. No-op on
	// empty input; categories not owned by userID are silently ignored.
	LabelThread(ctx context.Context, userID uuid.UUID, threadID string, categoryIDs []uuid.UUID) error
	// LabelLatestThreadForContact finds the contact's most recent conversation in
	// userID's unibox and labels it, for the "label_email" campaign step action
	// (which knows the contact but not the thread id). Returns the labeled thread
	// id, or "" when the contact has no conversation yet.
	LabelLatestThreadForContact(ctx context.Context, userID uuid.UUID, contactEmail string, categoryIDs []uuid.UUID) (string, error)
	// LatestInboundFromContact returns the subject + snippet of the newest email
	// received from the contact ("" when none). Backs the campaign AI step's
	// incoming-email context.
	LatestInboundFromContact(ctx context.Context, userID uuid.UUID, contactEmail string) (string, string, error)

	// ListCategories returns the user's contact categories, which double as
	// unibox conversation labels (same registry). An AI agent step offers these
	// by name and resolves the model's pick to an id. CreateCategory mints a new
	// one for the agent's create-on-the-fly path (opt-in per step).
	ListCategories(ctx context.Context, userID uuid.UUID) ([]models.MiniCategory, error)
	CreateCategory(ctx context.Context, userID uuid.UUID, title, color string) (models.MiniCategory, error)
	// ListPipelines returns the org's CRM pipelines with stages hydrated (both
	// ordered by position), so an AI agent step can pick a valid pipeline+stage
	// (defaulting to the first pipeline and its first stage).
	ListPipelines(ctx context.Context, orgID uuid.UUID) ([]models.Pipeline, error)

	// WireDispatcher attaches the event dispatcher that fans classified
	// replies + deliverability events out to customer webhooks and third-party
	// integration actions (Slack ping, CRM upsert).
	WireDispatcher(d EventDispatcher)
	// WireNotifier attaches the in-app notification gate (reply/bounce/complaint).
	WireNotifier(n Notifier)
	// WireRealtime attaches the org-scoped EMAIL_REPLIED realtime pulse.
	WireRealtime(p ReplyRealtimePublisher)
	// WireAutomationRunner attaches the automation runner so instant
	// "run_automation" action nodes (reply/open/click branches) can launch a flow.
	WireAutomationRunner(r AutomationRunner)
	// WireInboxAgent attaches the inbox agent so an inbound human reply drafts a
	// suggested reply for review (M10). Best-effort; nil = feature off.
	WireInboxAgent(a InboxAgent)

	// EmitCampaignEvent dispatches a campaign event (e.g. from a sequence
	// "notify" action node) to customer webhooks and wired integrations.
	EmitCampaignEvent(ctx context.Context, orgID uuid.UUID, eventType models.WebhookEventType, data map[string]any)

	// FireCampaignEvent publishes a developer-defined "fire event" to the realtime
	// gateway from a campaign step (subscribers receive it over the API websocket).
	FireCampaignEvent(ctx context.Context, orgID uuid.UUID, sourceID, name string, fields []models.ActionKV, contact *models.Contact)

	// FireInstantActions runs the matched INSTANT branch's action chain for a
	// contact the moment an engagement signal lands for them, instead of waiting
	// for the next scheduled step boundary. eventKind is "reply", "open", or
	// "click" and selects which branch fields can fire (reply -> reply_* intent
	// fields; open -> "opened"; click -> "clicked"). The signal must already be
	// recorded on the contact's progress row before this is called. Best-effort
	// and non-blocking: it never returns an error and must never block the caller's
	// hot path. The tracking consumer calls this after RecordEmailOpened /
	// RecordEmailClicked; ProcessIncomingReply calls the unexported path with
	// "reply".
	FireInstantActions(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID, eventKind string)

	// DLQ auto-retry
	ProcessRetryableDeadLetters(ctx context.Context) (int, *errx.Error)
}

type service struct {
	repo                 repository.AdvancedOutreachRepository
	campaignRepo         repository.CampaignRepository
	emailRepo            repository.EmailRepository
	taskRepo             repository.TaskRepository
	contactRepo          repository.ContactRepository
	segmentRepo          repository.SegmentRepository
	campaignProgressRepo repository.CampaignProgressRepository
	crmRepo              repository.CRMRepository
	categoryRepo         repository.GroupRepository
	uniboxRepo           repository.UniboxRepository
	tasksClient          tasksched.Scheduler
	warmupService        warmupapp.Service
	dispatcher           EventDispatcher
	// audienceRepo measures a campaign's list for the preflight report.
	// Optional/nil-safe: without it the list check is simply absent.
	audienceRepo repository.CampaignAudienceRepository
	// attachmentRepo lets preflight weigh attachments as the send path does.
	// Optional/nil-safe: without it the content check scores none.
	attachmentRepo repository.AttachmentRepository
	notifier       Notifier
	realtime       ReplyRealtimePublisher
	// evidence teaches verification what replies and bounces showed.
	evidence         EvidenceRecorder
	automationRunner AutomationRunner
	inboxAgent       InboxAgent
}

func NewService(
	repo repository.AdvancedOutreachRepository,
	campaignRepo repository.CampaignRepository,
	emailRepo repository.EmailRepository,
	taskRepo repository.TaskRepository,
	contactRepo repository.ContactRepository,
	campaignProgressRepo repository.CampaignProgressRepository,
	crmRepo repository.CRMRepository,
	categoryRepo repository.GroupRepository,
	uniboxRepo repository.UniboxRepository,
	tasksClient tasksched.Scheduler,
	warmupService warmupapp.Service,
) Service {
	return &service{
		repo:                 repo,
		campaignRepo:         campaignRepo,
		emailRepo:            emailRepo,
		taskRepo:             taskRepo,
		contactRepo:          contactRepo,
		campaignProgressRepo: campaignProgressRepo,
		crmRepo:              crmRepo,
		categoryRepo:         categoryRepo,
		uniboxRepo:           uniboxRepo,
		tasksClient:          tasksClient,
		warmupService:        warmupService,
	}
}

func toErrx(err error) *errx.Error {
	if err == nil {
		return nil
	}
	if xerr, ok := err.(*errx.Error); ok {
		return xerr
	}
	return errx.InternalError()
}

func (s *service) GetOrganizationSettings(ctx context.Context, organizationID uuid.UUID) (*models.AdvancedOutreachSettings, *errx.Error) {
	settings, err := s.repo.GetOutreachSettings(ctx, organizationID)
	if err != nil {
		return nil, toErrx(err)
	}
	return settings, nil
}

func (s *service) UpdateOrganizationSettings(ctx context.Context, organizationID, updatedBy uuid.UUID, settings *models.AdvancedOutreachSettings) *errx.Error {
	if settings == nil {
		return errx.New(errx.BadRequest, "settings are required")
	}
	settings.Normalize()
	if err := s.repo.UpsertOutreachSettings(ctx, organizationID, updatedBy, settings); err != nil {
		return toErrx(err)
	}
	return nil
}

func (s *service) GetCampaignSettings(ctx context.Context, campaignID uuid.UUID) (*models.CampaignAdvancedSettings, *errx.Error) {
	cfg, err := s.repo.GetCampaignAdvancedSettings(ctx, campaignID)
	if err != nil {
		return nil, toErrx(err)
	}
	if cfg == nil {
		return &models.CampaignAdvancedSettings{
			CampaignID: campaignID,
			Overrides:  models.DefaultAdvancedOutreachSettings(),
			UpdatedAt:  time.Now().UTC(),
		}, nil
	}
	return cfg, nil
}

func (s *service) UpdateCampaignSettings(ctx context.Context, campaignID uuid.UUID, settings *models.AdvancedOutreachSettings) *errx.Error {
	if settings == nil {
		return errx.New(errx.BadRequest, "settings are required")
	}
	settings.Normalize()
	if err := s.repo.UpsertCampaignAdvancedSettings(ctx, campaignID, settings); err != nil {
		return toErrx(err)
	}
	return nil
}

func (s *service) ListABVariants(ctx context.Context, campaignID uuid.UUID) ([]models.CampaignABVariant, *errx.Error) {
	out, err := s.repo.ListABVariants(ctx, campaignID)
	if err != nil {
		return nil, toErrx(err)
	}
	return out, nil
}

func (s *service) CreateABVariant(ctx context.Context, campaignID uuid.UUID, req *models.CreateCampaignABVariantRequest) (*models.CampaignABVariant, *errx.Error) {
	if req == nil || strings.TrimSpace(req.Name) == "" {
		return nil, errx.New(errx.BadRequest, "variant name is required")
	}
	out, err := s.repo.CreateABVariant(ctx, campaignID, req)
	if err != nil {
		return nil, toErrx(err)
	}
	return out, nil
}

func (s *service) UpdateABVariant(ctx context.Context, campaignID, variantID uuid.UUID, req *models.UpdateCampaignABVariantRequest) (*models.CampaignABVariant, *errx.Error) {
	out, err := s.repo.UpdateABVariant(ctx, campaignID, variantID, req)
	if err != nil {
		return nil, toErrx(err)
	}
	return out, nil
}

func (s *service) DeleteABVariant(ctx context.Context, campaignID, variantID uuid.UUID) *errx.Error {
	if err := s.repo.DeleteABVariant(ctx, campaignID, variantID); err != nil {
		return toErrx(err)
	}
	return nil
}

func (s *service) EffectiveSettings(ctx context.Context, organizationID, campaignID uuid.UUID) (*models.AdvancedOutreachSettings, *errx.Error) {
	return s.effectiveSettings(ctx, organizationID, campaignID)
}

func (s *service) effectiveSettings(ctx context.Context, organizationID, campaignID uuid.UUID) (*models.AdvancedOutreachSettings, *errx.Error) {
	orgSettings, err := s.repo.GetOutreachSettings(ctx, organizationID)
	if err != nil {
		return nil, toErrx(err)
	}
	campaignSettings, err := s.repo.GetCampaignAdvancedSettings(ctx, campaignID)
	if err != nil {
		return nil, toErrx(err)
	}
	if campaignSettings == nil {
		return orgSettings, nil
	}
	merged, err := mergeSettings(orgSettings, &campaignSettings.Overrides)
	if err != nil {
		return nil, toErrx(err)
	}
	return merged, nil
}

func mergeSettings(base, override *models.AdvancedOutreachSettings) (*models.AdvancedOutreachSettings, error) {
	if base == nil {
		def := models.DefaultAdvancedOutreachSettings()
		base = &def
	}
	if override == nil {
		copy := *base
		return &copy, nil
	}

	baseRaw, err := json.Marshal(base)
	if err != nil {
		return nil, err
	}
	overrideRaw, err := json.Marshal(override)
	if err != nil {
		return nil, err
	}

	var baseMap map[string]interface{}
	var overrideMap map[string]interface{}
	if err := json.Unmarshal(baseRaw, &baseMap); err != nil {
		return nil, err
	}
	if err := json.Unmarshal(overrideRaw, &overrideMap); err != nil {
		return nil, err
	}
	mergedMap := mergeMap(baseMap, overrideMap)
	outRaw, err := json.Marshal(mergedMap)
	if err != nil {
		return nil, err
	}
	var out models.AdvancedOutreachSettings
	if err := json.Unmarshal(outRaw, &out); err != nil {
		return nil, err
	}
	return &out, nil
}

func mergeMap(base, override map[string]interface{}) map[string]interface{} {
	out := make(map[string]interface{}, len(base))
	for k, v := range base {
		out[k] = v
	}
	for k, ov := range override {
		if ovm, ok := ov.(map[string]interface{}); ok {
			if bv, exists := out[k]; exists {
				if bvm, ok := bv.(map[string]interface{}); ok {
					out[k] = mergeMap(bvm, ovm)
					continue
				}
			}
		}
		out[k] = ov
	}
	return out
}

func (s *service) ShouldSuppressRecipient(ctx context.Context, organizationID uuid.UUID, recipient string) (bool, string, *errx.Error) {
	entry, err := s.repo.IsRecipientSuppressed(ctx, organizationID, recipient)
	if err != nil {
		return false, "", toErrx(err)
	}
	if entry == nil {
		return false, "", nil
	}
	return true, entry.Reason, nil
}

// Unsubscribe resolves the campaign + contact behind a List-Unsubscribe link and
// suppresses the recipient org-wide. Always suppresses (an explicit recipient
// request), then fans out the campaign.unsubscribed event for Slack/CRM.
func (s *service) CreateContactTask(ctx context.Context, orgID, createdBy uuid.UUID, data *models.CreateCRMTask) (*models.CRMTask, *errx.Error) {
	if s.crmRepo == nil {
		return nil, errx.InternalError()
	}
	task, err := s.crmRepo.CreateCRMTask(ctx, orgID, createdBy, data)
	if err != nil {
		return nil, errx.InternalError()
	}
	if task.ContactID != nil {
		_ = s.crmRepo.RecordActivity(ctx, orgID, *task.ContactID, &createdBy, models.ActivityTaskCreated, map[string]interface{}{
			"task_id":    task.ID.String(),
			"task_title": task.Title,
			"source":     "campaign",
		})
	}
	return task, nil
}

// CreateContactDeal opens a deal for the contact (campaign "create_deal" node)
// and records a deal_created activity. Mirrors CreateContactTask.
func (s *service) CreateContactDeal(ctx context.Context, orgID uuid.UUID, createdBy uuid.UUID, data *models.CreateDeal) (*models.Deal, *errx.Error) {
	if s.crmRepo == nil {
		return nil, errx.InternalError()
	}
	deal, err := s.crmRepo.CreateDeal(ctx, orgID, data)
	if err != nil {
		return nil, toErrx(err)
	}
	if deal.ContactID != nil {
		_ = s.crmRepo.RecordActivity(ctx, orgID, *deal.ContactID, &createdBy, models.ActivityDealCreated, map[string]interface{}{
			"deal_id":   deal.ID.String(),
			"deal_name": deal.Name,
			"source":    "campaign",
		})
	}
	return deal, nil
}

// MoveContactDealStage moves the contact's most-recent OPEN deal in pipelineID
// to stageID. "Most-recent open deal" is resolved by scanning the contact's
// deals (GetDealsByContact returns them created_at DESC) and taking the first
// one whose pipeline matches and whose status is still "open". No open deal in
// that pipeline is a deliberate no-op (returns nil, nil) so a chained reply
// automation doesn't error just because nothing has been created yet.
func (s *service) MoveContactDealStage(ctx context.Context, orgID, contactID, pipelineID, stageID uuid.UUID) (*models.Deal, *errx.Error) {
	if s.crmRepo == nil {
		return nil, errx.InternalError()
	}
	deals, err := s.crmRepo.GetDealsByContact(ctx, orgID, contactID)
	if err != nil {
		return nil, toErrx(err)
	}
	var target *models.Deal
	for i := range deals {
		d := deals[i]
		if d.PipelineID == pipelineID && d.Status == models.DealStatusOpen {
			target = &d
			break // GetDealsByContact is created_at DESC => first match is newest
		}
	}
	if target == nil {
		// No open deal in this pipeline: documented no-op (not an error).
		return nil, nil
	}
	if target.StageID == stageID {
		return target, nil // already there
	}
	updated, uerr := s.crmRepo.UpdateDeal(ctx, orgID, target.ID, &models.UpdateDeal{StageID: &stageID})
	if uerr != nil {
		return nil, toErrx(uerr)
	}
	_ = s.crmRepo.RecordActivity(ctx, orgID, contactID, nil, models.ActivityDealStageChange, map[string]interface{}{
		"deal_id": updated.ID.String(),
		"from":    target.StageID.String(),
		"to":      stageID.String(),
		"source":  "campaign",
	})
	return updated, nil
}

// ListCategories returns the user's categories (contact tags == unibox labels).
func (s *service) ListCategories(ctx context.Context, userID uuid.UUID) ([]models.MiniCategory, error) {
	if s.categoryRepo == nil {
		return nil, nil
	}
	groups, err := s.categoryRepo.List(ctx, userID)
	if err != nil {
		return nil, err
	}
	out := make([]models.MiniCategory, 0, len(groups))
	for _, g := range groups {
		out = append(out, models.MiniCategory{ID: g.ID, Title: g.Title, Color: g.Color})
	}
	return out, nil
}

// CreateCategory mints a new category (tag/label) for the agent's opt-in
// create-on-the-fly path. GroupRepository.Create validates the title (1-50) and
// enforces the per-user cap; color defaults to slate when blank.
func (s *service) CreateCategory(ctx context.Context, userID uuid.UUID, title, color string) (models.MiniCategory, error) {
	if s.categoryRepo == nil {
		return models.MiniCategory{}, errx.New(errx.BadRequest, "categories are not available")
	}
	if strings.TrimSpace(color) == "" {
		color = "#64748b"
	}
	g, err := s.categoryRepo.Create(ctx, userID, &models.GroupCreate{Title: strings.TrimSpace(title), Color: color})
	if err != nil {
		return models.MiniCategory{}, err
	}
	return models.MiniCategory{ID: g.ID, Title: g.Title, Color: g.Color}, nil
}

// ListPipelines passes through the org's CRM pipelines (stages hydrated).
func (s *service) ListPipelines(ctx context.Context, orgID uuid.UUID) ([]models.Pipeline, error) {
	if s.crmRepo == nil {
		return nil, nil
	}
	return s.crmRepo.ListPipelines(ctx, orgID)
}

func (s *service) Unsubscribe(ctx context.Context, campaignID, contactID uuid.UUID) *errx.Error {
	return s.unsubscribe(ctx, nil, campaignID, contactID, "action")
}

func (s *service) UnsubscribeFromLink(ctx context.Context, organizationID, campaignID, contactID uuid.UUID, via string) *errx.Error {
	if via != "one_click" {
		via = "link"
	}
	return s.unsubscribe(ctx, &organizationID, campaignID, contactID, via)
}

// unsubscribe records an explicit opt-out: the address goes on the workspace
// suppression list and the contact's own subscription flag is cleared, so the
// CRM and the send gate tell the same story.
func (s *service) unsubscribe(ctx context.Context, expectOrg *uuid.UUID, campaignID, contactID uuid.UUID, via string) *errx.Error {
	campaign, err := s.campaignRepo.GetByID(ctx, campaignID)
	if err != nil || campaign == nil || campaign.OrganizationID == nil {
		return errx.New(errx.BadRequest, "invalid unsubscribe link")
	}
	if expectOrg != nil && *expectOrg != *campaign.OrganizationID {
		return errx.New(errx.BadRequest, "invalid unsubscribe link")
	}
	contact, cerr := s.contactRepo.GetByID(ctx, contactID)
	if cerr != nil || contact == nil || contact.Email == "" {
		return errx.New(errx.BadRequest, "invalid unsubscribe link")
	}

	reason := map[string]string{
		"one_click": "one-click unsubscribe (mail client)",
		"link":      "clicked the unsubscribe link",
		"action":    "unsubscribed by a sequence action",
	}[via]
	if err := s.repo.UpsertSuppressedRecipient(ctx, &models.SuppressedRecipient{
		OrganizationID: *campaign.OrganizationID,
		Email:          contact.Email,
		Kind:           models.SuppressionKindEmail,
		Reason:         reason,
		Source:         models.DeliverabilityEventUnsubscribe,
		CampaignID:     &campaignID,
		Metadata:       map[string]interface{}{"via": via},
	}); err != nil {
		return toErrx(err)
	}
	if err := s.contactRepo.SetSubscribedByEmail(ctx, *campaign.OrganizationID, contact.Email, false); err != nil {
		log.Warn().Err(err).Str("contact_id", contactID.String()).Msg("unsubscribe: could not clear the contact's subscription flag")
	}

	s.emit(ctx, *campaign.OrganizationID, models.WebhookEventCampaignUnsubscribed, map[string]any{
		"campaign_id":   campaignID.String(),
		"contact_id":    contactID.String(),
		"contact_email": contact.Email,
		"source":        via,
	})
	return nil
}

func (s *service) Resubscribe(ctx context.Context, organizationID, contactID uuid.UUID) *errx.Error {
	contact, cerr := s.contactRepo.GetByID(ctx, contactID)
	if cerr != nil || contact == nil || contact.Email == "" {
		return errx.New(errx.BadRequest, "invalid unsubscribe link")
	}
	// The contact must belong to the organization in the token.
	if owned, oerr := s.contactRepo.GetByEmailAndOrganization(ctx, organizationID, contact.Email); oerr != nil || owned == nil {
		return errx.New(errx.BadRequest, "invalid unsubscribe link")
	}
	if _, err := s.repo.DeleteSuppressionByEmail(ctx, organizationID, contact.Email, models.DeliverabilityEventUnsubscribe); err != nil {
		return toErrx(err)
	}
	if err := s.contactRepo.SetSubscribedByEmail(ctx, organizationID, contact.Email, true); err != nil {
		return toErrx(err)
	}
	return nil
}

func (s *service) ListSuppressions(ctx context.Context, organizationID uuid.UUID, q string, beforeAt *time.Time, beforeID *uuid.UUID, limit int) ([]models.SuppressedRecipient, *errx.Error) {
	out, err := s.repo.ListSuppressedRecipients(ctx, organizationID, q, beforeAt, beforeID, limit)
	if err != nil {
		return nil, toErrx(err)
	}
	return out, nil
}

// suppressionDomain matches a bare host ("acme.com", "mail.acme.co.uk").
var suppressionDomain = regexp.MustCompile(`^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?(?:\.[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)+$`)

// AddSuppressions adds each value as an address, or as a domain when it has
// no local part ("acme.com" or "@acme.com"). Unparseable values are reported
// back rather than failing the whole batch, because the batch is usually a
// pasted list with a stray header or blank line in it.
func (s *service) AddSuppressions(ctx context.Context, organizationID, actorID uuid.UUID, req *models.AddSuppressionsRequest) (*models.AddSuppressionsResult, *errx.Error) {
	if req == nil || len(req.Entries) == 0 {
		return nil, errx.New(errx.BadRequest, "entries are required")
	}
	if len(req.Entries) > 5000 {
		return nil, errx.New(errx.BadRequest, "at most 5000 entries per request")
	}
	source := models.SuppressionSourceManual
	if len(req.Entries) > 1 {
		source = models.SuppressionSourceImport
	}
	res := &models.AddSuppressionsResult{Skipped: []string{}}
	seen := map[string]bool{}
	entries := make([]models.SuppressedRecipient, 0, len(req.Entries))
	for _, e := range req.Entries {
		raw := strings.TrimSpace(e.Value)
		value := strings.ToLower(strings.TrimPrefix(raw, "@"))
		if value == "" || seen[value] {
			continue
		}
		kind := models.SuppressionKindEmail
		if strings.Contains(value, "@") {
			if !validate.Email(value) {
				res.Skipped = append(res.Skipped, raw)
				continue
			}
		} else if suppressionDomain.MatchString(value) {
			kind = models.SuppressionKindDomain
		} else {
			res.Skipped = append(res.Skipped, raw)
			continue
		}
		seen[value] = true
		reason := strings.TrimSpace(e.Reason)
		if reason == "" {
			reason = strings.TrimSpace(req.Reason)
		}
		if reason == "" {
			reason = "added to the suppression list"
		}
		if r := []rune(reason); len(r) > models.UnsubscribeCopyMaxLen {
			reason = string(r[:models.UnsubscribeCopyMaxLen])
		}
		entries = append(entries, models.SuppressedRecipient{
			OrganizationID: organizationID,
			Email:          value,
			Kind:           kind,
			Reason:         reason,
			Source:         source,
			Metadata:       map[string]interface{}{"added_by": actorID.String()},
		})
	}
	// One transaction: a pasted list lands whole or not at all, so a failure
	// part-way never leaves the caller guessing which half got in.
	if err := s.repo.UpsertSuppressedRecipients(ctx, entries); err != nil {
		return nil, toErrx(err)
	}
	for _, e := range entries {
		if e.Kind != models.SuppressionKindEmail {
			continue
		}
		if err := s.contactRepo.SetSubscribedByEmail(ctx, organizationID, e.Email, false); err != nil {
			log.Warn().Err(err).Msg("suppression: could not clear the contact's subscription flag")
		}
	}
	res.Added = len(entries)
	return res, nil
}

func (s *service) RemoveSuppression(ctx context.Context, organizationID, id uuid.UUID) (*models.SuppressedRecipient, *errx.Error) {
	entry, err := s.repo.GetSuppressedRecipient(ctx, organizationID, id)
	if err != nil {
		return nil, toErrx(err)
	}
	if entry == nil {
		return nil, errx.New(errx.NotFound, "suppression entry not found")
	}
	if _, err := s.repo.DeleteSuppressedRecipient(ctx, organizationID, id); err != nil {
		return nil, toErrx(err)
	}
	// Lifting an address's suppression restores the contact too; otherwise
	// the send gate still refuses it on the subscription flag and the list
	// says one thing while the contact says another.
	if entry.Kind == models.SuppressionKindEmail {
		if err := s.contactRepo.SetSubscribedByEmail(ctx, organizationID, entry.Email, true); err != nil {
			log.Warn().Err(err).Msg("suppression: could not restore the contact's subscription flag")
		}
	}
	return entry, nil
}

// pickVariantWeightedRandom does a weighted random draw over active variants.
func pickVariantWeightedRandom(variants []models.CampaignABVariant) *models.CampaignABVariant {
	total := 0
	for i := range variants {
		if variants[i].Weight <= 0 {
			variants[i].Weight = 100
		}
		total += variants[i].Weight
	}
	if total <= 0 {
		return nil
	}
	pick := rand.Intn(total)
	running := 0
	for i := range variants {
		running += variants[i].Weight
		if pick < running {
			return &variants[i]
		}
	}
	return &variants[len(variants)-1]
}

// abControlWeight is the implicit weight of a step's original content (the
// control arm) in a step-scoped A/B split, matching the default variant weight
// so one variant yields an even 50/50 split with the original.
const abControlWeight = 100

// pickVariantDeterministic does a weighted draw seeded by a stable string, so
// the same seed always picks the same variant (used for per-step assignment).
func pickVariantDeterministic(variants []models.CampaignABVariant, seed string) *models.CampaignABVariant {
	total := 0
	for i := range variants {
		if variants[i].Weight <= 0 {
			variants[i].Weight = 100
		}
		total += variants[i].Weight
	}
	if total <= 0 {
		return nil
	}
	h := fnv.New32a()
	_, _ = h.Write([]byte(seed))
	pick := int(h.Sum32() % uint32(total))
	running := 0
	for i := range variants {
		running += variants[i].Weight
		if pick < running {
			return &variants[i]
		}
	}
	return &variants[len(variants)-1]
}

func (s *service) SelectVariant(ctx context.Context, organizationID, campaignID, contactID, sequenceID uuid.UUID, subject, bodyHTML, bodyPlain string) (*models.VariantSelection, *errx.Error) {
	settings, xerr := s.effectiveSettings(ctx, organizationID, campaignID)
	if xerr != nil {
		return nil, xerr
	}
	if !settings.ABTesting.Enabled {
		return &models.VariantSelection{Subject: subject, BodyHTML: bodyHTML, BodyPlain: bodyPlain}, nil
	}

	variants, err := s.repo.ListABVariants(ctx, campaignID)
	if err != nil {
		return nil, toErrx(err)
	}
	// Partition active variants into step-scoped (this step) and campaign-level.
	var stepVariants, campaignVariants []models.CampaignABVariant
	for _, v := range variants {
		if !v.IsActive {
			continue
		}
		if v.SequenceID != nil {
			if *v.SequenceID == sequenceID {
				stepVariants = append(stepVariants, v)
			}
		} else {
			campaignVariants = append(campaignVariants, v)
		}
	}

	var selected *models.CampaignABVariant
	if len(stepVariants) > 0 {
		// Step-scoped: the step's own content is the control arm. Contacts split
		// across the original PLUS the active variants by weight, deterministically
		// per (contact, step). The original's share is the weight of its is_control
		// row when the user has set one, otherwise the default control weight. The
		// control arm carries the zero id (or an is_control row, whose content is
		// ignored); if it wins we send the step's own content.
		controlWeight := abControlWeight
		arms := make([]models.CampaignABVariant, 0, len(stepVariants))
		for _, v := range stepVariants {
			if v.IsControl {
				controlWeight = v.Weight
				continue
			}
			arms = append(arms, v)
		}
		pool := append([]models.CampaignABVariant{{Weight: controlWeight}}, arms...)
		selected = pickVariantDeterministic(pool, contactID.String()+":"+sequenceID.String())
		if selected != nil && (selected.ID == uuid.Nil || selected.IsControl) {
			return &models.VariantSelection{Subject: subject, BodyHTML: bodyHTML, BodyPlain: bodyPlain}, nil
		}
	} else if len(campaignVariants) > 0 {
		// Campaign-level (legacy): keep the assignment-based selection so a
		// contact stays on one variant across the whole campaign.
		assigned, aerr := s.repo.GetAssignedVariant(ctx, campaignID, contactID)
		if aerr != nil {
			return nil, toErrx(aerr)
		}
		if assigned != nil && assigned.IsActive && assigned.SequenceID == nil {
			selected = assigned
		} else {
			selected = pickVariantWeightedRandom(campaignVariants)
			if selected != nil {
				_ = s.repo.AssignVariant(ctx, campaignID, contactID, selected.ID)
			}
		}
	}

	if selected == nil {
		return &models.VariantSelection{Subject: subject, BodyHTML: bodyHTML, BodyPlain: bodyPlain}, nil
	}

	finalSubject := subject
	if selected.Subject != "" {
		finalSubject = selected.Subject
	}
	finalHTML := bodyHTML
	if selected.BodyHTML != "" {
		finalHTML = selected.BodyHTML
	}
	finalPlain := bodyPlain
	if selected.BodyPlain != "" {
		finalPlain = selected.BodyPlain
	}
	return &models.VariantSelection{
		VariantID: &selected.ID,
		Subject:   finalSubject,
		BodyHTML:  finalHTML,
		BodyPlain: finalPlain,
	}, nil
}

func parseSenderEmail(addrs []string) string {
	if len(addrs) == 0 {
		return ""
	}
	primary := strings.TrimSpace(addrs[0])
	if primary == "" {
		return ""
	}
	if parsed, err := mail.ParseAddress(primary); err == nil {
		return strings.ToLower(strings.TrimSpace(parsed.Address))
	}
	return strings.ToLower(strings.Trim(primary, "<>"))
}

func cleanMessageID(mid string) string {
	return strings.TrimSpace(strings.Trim(mid, "<>"))
}

// buildReplyHeaders synthesizes the header map the reply classifier's Layer 1
// (header) scan reads. EmailMessageStoreData does not carry the full raw header
// block, but it does carry the structured fields the deterministic markers care
// about (From, Reply-To) plus any custom headers the worker folded into Flags
// using the "Header-Name:value" convention (the same encoding extractHeaderValue
// relies on for the warmup token). That lets RFC 3834 / Precedence / X-Autoreply
// markers reach the classifier when the worker forwarded them, while the From
// (mailer-daemon / no-reply) and Subject ("Automatic reply") signals work from
// the always-present structured fields.
func buildReplyHeaders(msg *models.EmailMessageStoreData) map[string][]string {
	if msg == nil {
		return nil
	}
	h := map[string][]string{}
	if len(msg.FromAddr) > 0 {
		h["From"] = msg.FromAddr
	}
	if len(msg.ReplyTo) > 0 {
		h["Reply-To"] = msg.ReplyTo
	}
	if msg.Subject != "" {
		h["Subject"] = []string{msg.Subject}
	}
	// Custom headers the worker stored as "Header-Name:value" flags (auto-reply
	// markers, Precedence, etc.). Split on the FIRST colon so header values that
	// contain ':' survive intact.
	for _, flag := range msg.Flags {
		if i := strings.Index(flag, ":"); i > 0 {
			name := strings.TrimSpace(flag[:i])
			val := strings.TrimSpace(flag[i+1:])
			if name != "" && !strings.HasPrefix(name, "\\") {
				h[name] = append(h[name], val)
			}
		}
	}
	return h
}

func firstNonEmpty(vals ...string) string {
	for _, v := range vals {
		if strings.TrimSpace(v) != "" {
			return v
		}
	}
	return ""
}

func uuidString(id *uuid.UUID) string {
	if id == nil {
		return ""
	}
	return id.String()
}

func containsAnyKeyword(text string, keywords []string) bool {
	if text == "" {
		return false
	}
	lower := strings.ToLower(text)
	for _, k := range keywords {
		if k == "" {
			continue
		}
		if strings.Contains(lower, strings.ToLower(k)) {
			return true
		}
	}
	return false
}

func classifyReply(text string, cfg models.ReplyIntentSettings) (models.ReplyIntentType, float64) {
	lower := strings.ToLower(text)
	if strings.TrimSpace(lower) == "" {
		return models.ReplyIntentNeutral, 0.2
	}

	score := map[models.ReplyIntentType]float64{
		models.ReplyIntentPositive:    0,
		models.ReplyIntentNegative:    0,
		models.ReplyIntentOutOfOffice: 0,
		models.ReplyIntentQuestion:    0,
	}

	for _, kw := range cfg.PositiveKeywords {
		if kw != "" && strings.Contains(lower, strings.ToLower(kw)) {
			score[models.ReplyIntentPositive] += 1.2
		}
	}
	for _, kw := range cfg.NegativeKeywords {
		if kw != "" && strings.Contains(lower, strings.ToLower(kw)) {
			score[models.ReplyIntentNegative] += 1.5
		}
	}
	for _, kw := range cfg.OutOfOfficeKeywords {
		if kw != "" && strings.Contains(lower, strings.ToLower(kw)) {
			score[models.ReplyIntentOutOfOffice] += 2
		}
	}
	for _, kw := range cfg.QuestionKeywords {
		if kw != "" && strings.Contains(lower, strings.ToLower(kw)) {
			score[models.ReplyIntentQuestion] += 1
		}
	}
	if strings.Contains(lower, "?") {
		score[models.ReplyIntentQuestion] += 0.8
	}

	best := models.ReplyIntentNeutral
	bestScore := 0.0
	for intent, sc := range score {
		if sc > bestScore {
			best = intent
			bestScore = sc
		}
	}
	if bestScore == 0 {
		return models.ReplyIntentNeutral, 0.35
	}
	conf := bestScore / (bestScore + 1.5)
	if conf > 0.99 {
		conf = 0.99
	}
	return best, conf
}

func (s *service) ProcessIncomingReply(ctx context.Context, emailAccountID uuid.UUID, msg *models.EmailMessageStoreData) *errx.Error {
	account, xerr := s.emailRepo.GetByID(ctx, emailAccountID)
	if xerr != nil {
		return xerr
	}
	if account == nil || account.OrganizationID == nil {
		return nil
	}

	settings, err := s.repo.GetOutreachSettings(ctx, *account.OrganizationID)
	if err != nil {
		return toErrx(err)
	}
	if !settings.ReplyIntent.Enabled {
		return nil
	}

	sender := parseSenderEmail(msg.FromAddr)
	if sender == "" {
		return nil
	}

	text := strings.TrimSpace(msg.Snippet)
	text = strings.TrimSpace(text + "\n" + msg.Subject)
	intent, confidence := classifyReply(text, settings.ReplyIntent)

	// Layered reply classification (header -> lexicon -> optional model) is run
	// further down, once the campaign context is known to store it on. Classifying
	// only inside that block means a reply with no campaign match never spends a
	// model call.

	var campaignID *uuid.UUID
	var sequenceID *uuid.UUID
	var contactID *uuid.UUID
	var taskID *uuid.UUID

	// First, try exact message threading via In-Reply-To.
	for _, mid := range msg.InReplyTo {
		candidate := cleanMessageID(mid)
		if candidate == "" {
			continue
		}
		task, err := s.taskRepo.GetTaskByMessageID(ctx, candidate)
		if err != nil || task == nil || task.TaskType != "campaign" {
			continue
		}
		taskID = &task.ID
		ct, err := s.taskRepo.GetCampaignTask(ctx, task.ID)
		if err == nil && ct != nil {
			campaignID = ct.CampaignID
			contactID = ct.ContactID
			sequenceID = ct.SequenceID
		}
		break
	}

	if contactID == nil {
		contact, xerr := s.contactRepo.GetByEmailAndOrganization(ctx, *account.OrganizationID, sender)
		if xerr != nil {
			return xerr
		}
		if contact != nil {
			contactID = &contact.ID
		}
	}

	if campaignID == nil && contactID != nil {
		latest, err := s.campaignProgressRepo.GetLatestCampaignSequenceForContact(ctx, *contactID)
		if err == nil && latest != nil {
			campaignID = &latest.CampaignID
			sequenceID = &latest.SequenceID
		}
	}

	if campaignID != nil && contactID != nil && sequenceID != nil {
		cID, ctID, sID := *campaignID, *contactID, *sequenceID

		// Cost guard for the optional AI layer (Layer 3). Layers 1-2 (headers +
		// lexicon) are free and always run; the paid model is only spent on a
		// genuinely new, non-trivial, human-looking reply from a contact we have
		// not already classified. So one contact spamming replies costs at most a
		// single model call, and trivial/boilerplate bodies never reach the model.
		gate := func() bool {
			if !replyclassify.WorthModeling(replyclassify.Input{BodyText: msg.Snippet}) {
				return false
			}
			if prior, perr := s.campaignProgressRepo.GetLatestReplyClass(ctx, ctID, cID); perr == nil &&
				prior != "" && prior != replyclassify.ClassUnknown {
				return false
			}
			return true
		}

		replyResult := replyclassify.ClassifyGated(ctx, replyclassify.Input{
			Headers:  buildReplyHeaders(msg),
			Subject:  msg.Subject,
			BodyText: msg.Snippet,
		}, gate)

		// Always persist the classifier verdict so reply_* branches can route on
		// it (including reply_automated for OOO / autoresponders). Layers 1-2 run
		// for every reply, so OOO/unsubscribe stay correct even when the gate
		// skipped the model.
		_ = s.campaignProgressRepo.RecordReplyClassification(ctx, cID, ctID, sID, replyResult.Class, replyResult.Source, replyResult.Confidence)

		// OOO trap fix: only a HUMAN reply stamps replied_at. An auto_reply /
		// out_of_office must NOT count as a reply, or it would (a) trip
		// stop_on_reply and silently halt the sequence, and (b) match the plain
		// "replied" branch. Both stop_on_reply and the "replied" condition key off
		// replied_at IS NOT NULL, so gating the stamp here fixes both at once.
		// Any reply, human or automatic, proves the mailbox is live; only a
		// human one counts as engagement.
		if s.evidence != nil {
			kind := "replied"
			if replyclassify.IsAutomated(replyResult.Class) {
				kind = "auto_replied"
			}
			s.evidence.RecordEvidence(ctx, ctID, kind, msg.ID.String(), "")
		}
		if !replyclassify.IsAutomated(replyResult.Class) {
			_ = s.campaignProgressRepo.RecordEmailReplied(ctx, cID, ctID, sID)
			_ = s.repo.MarkVariantEvent(ctx, cID, ctID, string(models.DeliverabilityEventReply))

			// Live org-wide pulse: the team sees the reply land on the
			// campaign without a refresh.
			if s.realtime != nil && account.OrganizationID != nil {
				s.realtime.PublishEmailReplied(ctx, account.OrganizationID.String(), account.UserID, cID.String(), ctID.String(), sender, sID.String())
			}

			// Inbox agent (M10): best-effort, paid + opt-in checked inside. Drafts a
			// suggested reply for this human reply, awaiting human approval in the
			// unibox. Self-detaches, so it never blocks reply ingest.
			if s.inboxAgent != nil && account.OrganizationID != nil {
				ownerID, _ := uuid.Parse(account.UserID)
				s.inboxAgent.DraftForReply(ctx, models.InboxAgentReply{
					OrganizationID:  *account.OrganizationID,
					EmailAccountID:  emailAccountID,
					OwnerUserID:     ownerID,
					SourceMessageID: msg.ID,
					ThreadID:        msg.ThreadID,
					Counterpart:     sender,
					Subject:         msg.Subject,
					Snippet:         msg.Snippet,
					BodyText:        msg.BodyText,
					InReplyTo:       msg.MessageID,
					ContactID:       ctID,
					CampaignID:      cID,
					IntentClass:     replyResult.Class,
					Confidence:      replyResult.Confidence,
				})
			}
		}

		// INSTANT reply trigger: if the contact's CURRENT step has a reply_* intent
		// branch matching this just-classified reply, run that branch's
		// action chain for THIS contact right now (instead of waiting for the next
		// scheduled step boundary). Best-effort and non-blocking like the rest of
		// reply handling — a failure must never block inbox ingest. Fires for both
		// human and automated replies (reply_automated drives the auto-reply case)
		// and exactly once per reply event via the instant_fired["reply"] gate. The
		// just-classified reply_class and (human-only) replied_at have already been
		// persisted above, so the matcher reads them off the loaded progress row.
		s.fireInstantActions(ctx, cID, ctID, sID, "reply")
	}

	actionTaken := ""
	if settings.ReplyIntent.AutoPauseOnNegative && intent == models.ReplyIntentNegative && campaignID != nil {
		_ = s.campaignRepo.UpdateStatus(ctx, *campaignID, "paused")
		actionTaken = "paused_campaign"
	}

	// Reply-based opt-out is what makes the plain "just reply and I'll stop"
	// line a real mechanism. The check ignores the quoted history (which
	// carries our own opt-out wording) and matches whole phrases only.
	if settings.ReplyIntent.AutoSuppressOnUnsubWord &&
		replyclassify.IsOptOut(msg.Subject, firstNonEmpty(msg.BodyText, msg.Snippet)) {
		_ = s.repo.UpsertSuppressedRecipient(ctx, &models.SuppressedRecipient{
			OrganizationID: *account.OrganizationID,
			Email:          sender,
			Kind:           models.SuppressionKindEmail,
			Reason:         "asked to stop in a reply",
			Source:         models.DeliverabilityEventUnsubscribe,
			CampaignID:     campaignID,
			Metadata: map[string]interface{}{
				"via": "reply",
			},
		})
		if err := s.contactRepo.SetSubscribedByEmail(ctx, *account.OrganizationID, sender, false); err != nil {
			log.Warn().Err(err).Msg("reply opt-out: could not clear the contact's subscription flag")
		}
		s.emit(ctx, *account.OrganizationID, models.WebhookEventCampaignUnsubscribed, map[string]any{
			"campaign_id":   uuidString(campaignID),
			"contact_id":    uuidString(contactID),
			"contact_email": sender,
			"source":        "reply",
		})
		if actionTaken == "" {
			actionTaken = "suppressed_recipient"
		} else {
			actionTaken += ",suppressed_recipient"
		}
	}

	if settings.ReplyIntent.AutoCreateCRMTask && s.crmRepo != nil && contactID != nil {
		owner, parseErr := uuid.Parse(account.UserID)
		if parseErr == nil {
			title := fmt.Sprintf("Follow up reply intent: %s (%s)", intent, sender)
			_, _ = s.crmRepo.CreateCRMTask(ctx, *account.OrganizationID, owner, &models.CreateCRMTask{
				ContactID:  contactID,
				Title:      title,
				Priority:   "high",
				DueDate:    ptrTime(time.Now().UTC().Add(24 * time.Hour)),
				AssignedTo: &owner,
			})
			if actionTaken == "" {
				actionTaken = "created_crm_task"
			} else {
				actionTaken += ",created_crm_task"
			}
		}
	}

	_ = s.repo.CreateReplyIntent(ctx, &models.ReplyIntentRecord{
		OrganizationID: *account.OrganizationID,
		ContactEmail:   sender,
		CampaignID:     campaignID,
		TaskID:         taskID,
		Intent:         intent,
		Confidence:     confidence,
		ActionTaken:    actionTaken,
		Metadata: map[string]interface{}{
			"subject": msg.Subject,
		},
	})

	// Fan the classified reply out to customer webhooks + integration actions
	// (Slack ping, CRM upsert). The intent/confidence fields let an integration
	// automation filter for e.g. only "positive" replies. This is the trigger
	// behind "notify me when a prospect replies".
	payload := map[string]any{
		"contact_email": sender,
		"intent":        string(intent),
		"confidence":    confidence,
		"subject":       msg.Subject,
		"snippet":       msg.Snippet,
		"action_taken":  actionTaken,
		"trigger":       "campaign_reply",
		// thread_id lets a "label email" automation action tag the conversation
		// this reply belongs to. _user_id is the mailbox owner (categories are per
		// user); the leading underscore keeps it out of outbound customer webhook
		// bodies (publicEventData strips _-prefixed keys) while staying available
		// to native actions, which read the raw event data.
		"thread_id": msg.ThreadID,
		"_user_id":  account.UserID,
	}
	if campaignID != nil {
		payload["campaign_id"] = campaignID.String()
	}
	if contactID != nil {
		payload["contact_id"] = contactID.String()
	}
	s.emit(ctx, *account.OrganizationID, models.WebhookEventCampaignReplyReceived, payload)
	if intent == models.ReplyIntentNegative {
		// Negative/unsubscribe-leaning replies also fire the unsubscribe event
		// only when we actually suppressed; otherwise the reply event is enough.
		if strings.Contains(actionTaken, "suppressed_recipient") {
			s.emit(ctx, *account.OrganizationID, models.WebhookEventCampaignUnsubscribed, payload)
		}
	}

	// Raise an in-app notification to the mailbox owner (gated by their prefs).
	if uid, perr := uuid.Parse(account.UserID); perr == nil {
		cat := models.NotifInboundReply
		title := "New reply from " + sender
		if intent == models.ReplyIntentOutOfOffice {
			cat = models.NotifInboundOOO
			title = "Out-of-office from " + sender
		}
		s.notify(uid, account.OrganizationID, cat, title, msg.Subject, "/app/unibox", map[string]any{"intent": string(intent)})
	}

	return nil
}

func ptrTime(t time.Time) *time.Time {
	return &t
}

func (s *service) IngestDeliverabilityEvent(ctx context.Context, organizationID uuid.UUID, req *models.IngestDeliverabilityEventRequest) *errx.Error {
	if req == nil {
		return errx.New(errx.BadRequest, "event payload is required")
	}
	if req.RecipientEmail == "" {
		return errx.New(errx.BadRequest, "recipient_email is required")
	}

	eventType := req.EventType
	switch eventType {
	case models.DeliverabilityEventBounce,
		models.DeliverabilityEventComplaint,
		models.DeliverabilityEventUnsubscribe,
		models.DeliverabilityEventOpen,
		models.DeliverabilityEventClick,
		models.DeliverabilityEventReply:
	default:
		return errx.New(errx.BadRequest, "invalid event_type")
	}

	idempotencyKey := strings.TrimSpace(req.IdempotencyKey)
	if idempotencyKey == "" {
		idempotencyKey = uuid.NewString()
	}

	provider := strings.TrimSpace(req.Provider)
	if provider == "" {
		provider = "manual"
	}

	if err := s.repo.CreateDeliverabilityEvent(ctx, &models.DeliverabilityEvent{
		OrganizationID: organizationID,
		CampaignID:     req.CampaignID,
		TaskID:         req.TaskID,
		ContactID:      req.ContactID,
		EventType:      eventType,
		Provider:       provider,
		RecipientEmail: req.RecipientEmail,
		Reason:         req.Reason,
		IdempotencyKey: idempotencyKey,
		Metadata:       req.Metadata,
	}); err != nil {
		return toErrx(err)
	}

	settings, err := s.repo.GetOutreachSettings(ctx, organizationID)
	if err != nil {
		return toErrx(err)
	}

	shouldSuppress := (eventType == models.DeliverabilityEventBounce && settings.BouncePipeline.AutoSuppressOnBounce) ||
		(eventType == models.DeliverabilityEventComplaint && settings.BouncePipeline.AutoSuppressOnComplaint) ||
		(eventType == models.DeliverabilityEventUnsubscribe && settings.BouncePipeline.AutoSuppressOnUnsubscribe)

	if shouldSuppress {
		_ = s.repo.UpsertSuppressedRecipient(ctx, &models.SuppressedRecipient{
			OrganizationID: organizationID,
			Email:          req.RecipientEmail,
			Reason:         fmt.Sprintf("%s: %s", eventType, req.Reason),
			Source:         eventType,
			CampaignID:     req.CampaignID,
			Metadata:       req.Metadata,
		})
	}

	if req.CampaignID != nil && req.ContactID != nil {
		_ = s.repo.MarkVariantEvent(ctx, *req.CampaignID, *req.ContactID, string(eventType))
	}

	// Record bounces + complaints in campaign progress so analytics and the
	// breaker work correctly. Complaints were previously never recorded, which
	// is why complaint-rate auto-pause could never fire.
	if req.CampaignID != nil && req.ContactID != nil && req.TaskID != nil &&
		(eventType == models.DeliverabilityEventBounce || eventType == models.DeliverabilityEventComplaint) {
		campaignTask, cErr := s.taskRepo.GetCampaignTask(ctx, *req.TaskID)
		if cErr == nil && campaignTask != nil && campaignTask.SequenceID != nil {
			switch eventType {
			case models.DeliverabilityEventBounce:
				_ = s.campaignProgressRepo.RecordEmailBounced(ctx, *req.CampaignID, *req.ContactID, *campaignTask.SequenceID)
				// Only a bounce that names the recipient is evidence against
				// the address; a full mailbox or a policy block is not.
				if s.evidence != nil {
					kind := "bounced_other"
					if emailverify.NamesRecipient(req.Reason) {
						kind = "bounced_recipient"
					}
					s.evidence.RecordEvidence(ctx, *req.ContactID, kind, req.IdempotencyKey, req.Reason)
				}
			case models.DeliverabilityEventComplaint:
				_ = s.campaignProgressRepo.RecordEmailComplained(ctx, *req.CampaignID, *req.ContactID, *campaignTask.SequenceID)
			}
		}
	}

	if req.CampaignID != nil &&
		(eventType == models.DeliverabilityEventBounce || eventType == models.DeliverabilityEventComplaint) {
		s.evaluateCampaignBreaker(ctx, organizationID, *req.CampaignID, settings)
	}

	// Trigger warmup health re-evaluation on bounce or complaint events.
	// This connects deliverability signals to the warmup pool health system.
	if s.warmupService != nil && req.TaskID != nil &&
		(eventType == models.DeliverabilityEventBounce || eventType == models.DeliverabilityEventComplaint) {
		task, tErr := s.taskRepo.GetTask(ctx, *req.TaskID)
		if tErr == nil && task != nil {
			_, _ = s.warmupService.ApplySpamReport(ctx, uuid.Nil, task.EmailAccountID, req.IdempotencyKey, string(eventType))
		}
	}

	// Fan bounce / complaint / unsubscribe out to customer webhooks +
	// integration actions so a Slack channel or CRM can react in real time.
	payload := map[string]any{
		"contact_email": req.RecipientEmail,
		"recipient":     req.RecipientEmail,
		"event_type":    string(eventType),
		"provider":      provider,
		"reason":        req.Reason,
	}
	if req.CampaignID != nil {
		payload["campaign_id"] = req.CampaignID.String()
	}
	if req.ContactID != nil {
		payload["contact_id"] = req.ContactID.String()
	}
	switch eventType {
	case models.DeliverabilityEventBounce:
		s.emit(ctx, organizationID, models.WebhookEventCampaignEmailBounced, payload)
		s.emit(ctx, organizationID, models.WebhookEventDeliverabilityBounce, payload)
	case models.DeliverabilityEventComplaint:
		s.emit(ctx, organizationID, models.WebhookEventDeliverabilityComplaint, payload)
	case models.DeliverabilityEventUnsubscribe:
		s.emit(ctx, organizationID, models.WebhookEventCampaignUnsubscribed, payload)
	}

	// In-app notification to the campaign owner for bounce/complaint (gated by
	// their prefs). Resolvable only when the event is tied to a campaign.
	if (eventType == models.DeliverabilityEventBounce || eventType == models.DeliverabilityEventComplaint) && req.CampaignID != nil {
		if camp, cerr := s.campaignRepo.GetByID(ctx, *req.CampaignID); cerr == nil && camp != nil {
			if uid, perr := uuid.Parse(camp.UserID); perr == nil {
				cat := models.NotifHealthBounce
				title := "Bounce: " + req.RecipientEmail
				if eventType == models.DeliverabilityEventComplaint {
					cat = models.NotifHealthComplaint
					title = "Spam complaint: " + req.RecipientEmail
				}
				org := organizationID
				s.notify(uid, &org, cat, title, req.Reason, "/app/deliverability", map[string]any{"provider": provider})
			}
		}
	}

	return nil
}

// Campaign deliverability circuit breaker tuning. Mirrors Instantly's safeguards:
// a minimum sample before acting (so a single early bounce can't pause a
// campaign) and a rolling window so the breaker reacts to recent behaviour
// rather than a campaign's lifetime average.
const (
	campaignBreakerWindow    = 7 * 24 * time.Hour
	campaignBreakerMinSample = 50
	// Early-warning band: emit a warning webhook at half the pause threshold.
	campaignBreakerWarnRatio = 0.5
)

// evaluateCampaignBreaker auto-pauses a campaign when its rolling bounce or
// complaint rate breaches the configured threshold, and emits an early-warning
// webhook in the band below. Rolling-first with a cumulative fallback when the
// recent window is too small a sample.
func (s *service) evaluateCampaignBreaker(ctx context.Context, orgID, campaignID uuid.UUID, settings *models.AdvancedOutreachSettings) {
	if settings == nil || !settings.BouncePipeline.AutoPauseCampaignOnSpike {
		return
	}
	bounceThresh := settings.BouncePipeline.PauseBounceRateThreshold
	complaintThresh := settings.BouncePipeline.PauseComplaintRateThreshold
	if bounceThresh <= 0 && complaintThresh <= 0 {
		return
	}

	sent, bounced, complained := 0, 0, 0
	if rolling, err := s.campaignProgressRepo.GetCampaignRollingRates(ctx, campaignID, time.Now().Add(-campaignBreakerWindow)); err == nil && rolling != nil && rolling.Sent >= campaignBreakerMinSample {
		sent, bounced, complained = rolling.Sent, rolling.Bounced, rolling.Complained
	} else if progress, pErr := s.campaignProgressRepo.GetCampaignProgress(ctx, campaignID); pErr == nil && progress != nil {
		sent, bounced, complained = progress.EmailsSent, progress.EmailsBounced, progress.EmailsComplained
	}

	// Not enough delivered volume yet to judge — never pause on a tiny sample.
	if sent < campaignBreakerMinSample {
		return
	}

	bounceRate := float64(bounced) / float64(sent) * 100
	complaintRate := float64(complained) / float64(sent) * 100

	pauseBounce := bounceThresh > 0 && bounceRate >= bounceThresh
	pauseComplaint := complaintThresh > 0 && complaintRate >= complaintThresh
	if pauseBounce || pauseComplaint {
		if err := s.campaignRepo.UpdateStatus(ctx, campaignID, "paused"); err == nil {
			s.emit(ctx, orgID, models.WebhookEventCampaignPaused, map[string]any{
				"campaign_id":    campaignID.String(),
				"reason":         "deliverability_auto_pause",
				"bounce_rate":    bounceRate,
				"complaint_rate": complaintRate,
				"sample_size":    sent,
				"breached":       breachLabel(pauseBounce, pauseComplaint),
			})
		}
		return
	}

	warnBounce := bounceThresh > 0 && bounceRate >= bounceThresh*campaignBreakerWarnRatio
	warnComplaint := complaintThresh > 0 && complaintRate >= complaintThresh*campaignBreakerWarnRatio
	if warnBounce || warnComplaint {
		s.emit(ctx, orgID, models.WebhookEventCampaignDeliverabilityWarning, map[string]any{
			"campaign_id":    campaignID.String(),
			"bounce_rate":    bounceRate,
			"complaint_rate": complaintRate,
			"sample_size":    sent,
		})
	}
}

func breachLabel(bounce, complaint bool) string {
	switch {
	case bounce && complaint:
		return "bounce_and_complaint"
	case bounce:
		return "bounce"
	default:
		return "complaint"
	}
}

func (s *service) OptimizeSendTime(ctx context.Context, organizationID uuid.UUID, contact *models.Contact, base time.Time) (time.Time, *errx.Error) {
	settings, err := s.repo.GetOutreachSettings(ctx, organizationID)
	if err != nil {
		return base, toErrx(err)
	}
	if !settings.SendTimeOptimization.Enabled {
		return base, nil
	}

	targetTZ := settings.SendTimeOptimization.DefaultContactTimezone
	if targetTZ == "" {
		targetTZ = "UTC"
	}
	if settings.SendTimeOptimization.UseContactTimezone && contact != nil {
		if tz, ok := contact.CustomFields["timezone"]; ok && strings.TrimSpace(tz) != "" {
			targetTZ = strings.TrimSpace(tz)
		}
	}
	loc, lerr := time.LoadLocation(targetTZ)
	if lerr != nil {
		loc = time.UTC
	}

	candidate := base.In(loc)
	preferred := settings.SendTimeOptimization.PreferredHours
	if len(preferred) == 0 {
		preferred = []int{9, 10, 11, 14, 15, 16}
	}

	hour := candidate.Hour()
	chosenHour := -1
	for _, h := range preferred {
		if h >= hour {
			chosenHour = h
			break
		}
	}
	if chosenHour == -1 {
		candidate = candidate.Add(24 * time.Hour)
		chosenHour = preferred[0]
	}
	candidate = time.Date(candidate.Year(), candidate.Month(), candidate.Day(), chosenHour, 0, 0, 0, loc)

	if candidate.Weekday() == time.Saturday || candidate.Weekday() == time.Sunday {
		if settings.SendTimeOptimization.WeekendWeightMultiplier < 1 {
			for candidate.Weekday() == time.Saturday || candidate.Weekday() == time.Sunday {
				candidate = candidate.Add(24 * time.Hour)
			}
			candidate = time.Date(candidate.Year(), candidate.Month(), candidate.Day(), preferred[0], 0, 0, 0, loc)
		}
	}

	return candidate.UTC(), nil
}

func (s *service) StartTaskExecution(ctx context.Context, taskID uuid.UUID, executionKey string, metadata map[string]interface{}) (bool, *errx.Error) {
	duplicate, err := s.repo.StartTaskExecution(ctx, taskID, executionKey, metadata)
	if err != nil {
		return false, toErrx(err)
	}
	return duplicate, nil
}

func (s *service) CompleteTaskExecution(ctx context.Context, taskID uuid.UUID, executionKey, status string, metadata map[string]interface{}) *errx.Error {
	if err := s.repo.CompleteTaskExecution(ctx, taskID, executionKey, status, metadata); err != nil {
		return toErrx(err)
	}
	return nil
}

func (s *service) CaptureTaskDeadLetter(ctx context.Context, taskID uuid.UUID, taskType string, payload map[string]interface{}, lastError string, attempts int) *errx.Error {
	maxAttempts := 5

	// Compute next retry time using exponential backoff: 30s * 2^attempts
	var nextRetryAt *time.Time
	if attempts < maxAttempts {
		backoff := time.Duration(30*(1<<uint(attempts))) * time.Second
		t := time.Now().UTC().Add(backoff)
		nextRetryAt = &t
	}

	item := &models.TaskDeadLetter{
		TaskID:      taskID,
		TaskType:    taskType,
		Payload:     payload,
		LastError:   lastError,
		Attempts:    attempts,
		MaxAttempts: maxAttempts,
		Status:      "pending",
		NextRetryAt: nextRetryAt,
	}
	if err := s.repo.CreateTaskDeadLetter(ctx, item); err != nil {
		return toErrx(err)
	}
	return nil
}

func (s *service) ListDeadLetters(ctx context.Context, organizationID uuid.UUID, status string, limit int) ([]models.TaskDeadLetter, *errx.Error) {
	out, err := s.repo.ListTaskDeadLetters(ctx, organizationID, status, limit)
	if err != nil {
		return nil, toErrx(err)
	}
	return out, nil
}

func (s *service) ReplayDeadLetter(ctx context.Context, organizationID, deadLetterID uuid.UUID) *errx.Error {
	if s.tasksClient == nil {
		return errx.New(errx.BadRequest, "cloud tasks client not configured")
	}

	dlq, err := s.repo.GetTaskDeadLetter(ctx, deadLetterID, organizationID)
	if err != nil {
		return toErrx(err)
	}
	if dlq == nil {
		return errx.ErrNotFound
	}

	task, err := s.taskRepo.GetTask(ctx, dlq.TaskID)
	if err != nil {
		return toErrx(err)
	}
	if task == nil {
		return errx.ErrNotFound
	}

	scheduleAt := time.Now().UTC().Add(10 * time.Second)
	cloudTaskName, err := s.tasksClient.CreateTask(ctx, &proto.ProcessTask{TaskId: task.ID.String()}, scheduleAt)
	if err != nil {
		return toErrx(err)
	}
	if err := s.taskRepo.UpdateTaskScheduledAt(ctx, task.ID, scheduleAt, cloudTaskName); err != nil {
		return toErrx(err)
	}
	if err := s.taskRepo.UpdateTaskStatus(ctx, task.ID, "pending"); err != nil {
		return toErrx(err)
	}
	if err := s.repo.MarkTaskDeadLetterReplayed(ctx, deadLetterID); err != nil {
		return toErrx(err)
	}
	return nil
}

func (s *service) RunPreflight(ctx context.Context, organizationID, campaignID uuid.UUID) (*models.PreflightReport, *errx.Error) {
	campaign, err := s.campaignRepo.GetByID(ctx, campaignID)
	if err != nil || campaign == nil {
		return nil, errx.ErrNotFound
	}
	if campaign.OrganizationID == nil || *campaign.OrganizationID != organizationID {
		return nil, errx.ErrForbidden
	}

	settings, xerr := s.effectiveSettings(ctx, organizationID, campaignID)
	if xerr != nil {
		return nil, xerr
	}

	checks := make([]models.PreflightCheckResult, 0, 6)
	recommendations := make([]string, 0, 6)

	readyErr := s.campaignRepo.ValidateCampaignReady(ctx, campaignID)
	if readyErr != nil {
		checks = append(checks, models.PreflightCheckResult{
			Key:         "campaign_ready",
			Passed:      false,
			Severity:    "error",
			Message:     "Campaign has missing prerequisites (contacts, sequences, or sender accounts).",
			Remediation: "Add contacts, sequences, and at least one sender account tag match.",
		})
		recommendations = append(recommendations, "Complete core campaign setup before start.")
	} else {
		checks = append(checks, models.PreflightCheckResult{
			Key:      "campaign_ready",
			Passed:   true,
			Severity: "info",
			Message:  "Campaign has required entities configured.",
		})
	}

	if settings.Preflight.CheckScheduleWindow {
		// Per-day windows (when set) define the schedule; otherwise fall back to
		// the legacy start_time/end_time check.
		pass := !campaign.ScheduleWindows.IsEmpty() || campaign.StartTime < campaign.EndTime
		check := models.PreflightCheckResult{
			Key:      "schedule_window",
			Passed:   pass,
			Severity: "error",
			Message:  "Campaign schedule window is valid.",
		}
		if !pass {
			check.Message = "Campaign start_time must be before end_time."
			check.Remediation = "Update campaign schedule window."
			recommendations = append(recommendations, "Fix send window: start_time must be earlier than end_time.")
		}
		checks = append(checks, check)
	}

	if settings.Preflight.CheckDailyLimit {
		pass := campaign.DailyLimit > 0
		check := models.PreflightCheckResult{
			Key:      "daily_limit",
			Passed:   pass,
			Severity: "error",
			Message:  "Daily limit is configured.",
		}
		if !pass {
			check.Message = "Daily limit must be greater than zero."
			check.Remediation = "Set daily_limit > 0."
			recommendations = append(recommendations, "Set a valid daily limit to avoid burst sends.")
		}
		checks = append(checks, check)
	}

	if settings.Preflight.CheckUnsubscribeHeader {
		optOut := settings.Unsubscribe.Effective(campaign.UnsubscribeMode)
		bodyOptOut := optOut.Mode != models.UnsubscribeModeOff
		pass := campaign.UnsubscribeHeader || bodyOptOut
		check := models.PreflightCheckResult{
			Key:      "unsubscribe_header",
			Passed:   pass,
			Severity: "warning",
			Message:  "Recipients have a way to opt out.",
		}
		switch {
		case !pass:
			check.Message = "Recipients have no way to opt out: the unsubscribe header and the opt-out line are both off."
			check.Remediation = "Turn the opt-out line back on in Settings > Sending or on the campaign, or enable the unsubscribe header."
			recommendations = append(recommendations, "Give recipients a way to opt out.")
		case !campaign.UnsubscribeHeader:
			check.Message = "Opt-out line is on; the List-Unsubscribe header is off."
		case !bodyOptOut:
			check.Message = "List-Unsubscribe header is on; no opt-out line in the body."
		}
		checks = append(checks, check)
	}

	if settings.Preflight.CheckTrackingDomain && (campaign.OpenTracking || campaign.LinkTracking) {
		scope := repository.NewAccountScope(campaign.OrganizationID)
		accounts, err := s.emailRepo.GetByTags(ctx, scope, campaign.EmailTags)
		if err != nil || len(accounts) == 0 {
			checks = append(checks, models.PreflightCheckResult{
				Key:         "tracking_domain",
				Passed:      false,
				Severity:    "error",
				Message:     "No sender accounts available for tracking validation.",
				Remediation: "Attach sender accounts to campaign tags.",
			})
			recommendations = append(recommendations, "Attach at least one sender account with tracking domain.")
		} else {
			// Unverified counts as missing: an unverified domain is not used
			// at send time, so those mailboxes track on the shared host just
			// like the ones with nothing set.
			missing, unverified := 0, 0
			for _, account := range accounts {
				switch {
				case strings.TrimSpace(account.TrackingDomain) == "":
					missing++
				case !account.TrackingDomainVerified:
					unverified++
				}
			}
			pass := missing == 0 && unverified == 0
			check := models.PreflightCheckResult{
				Key:      "tracking_domain",
				Passed:   pass,
				Severity: "warning",
				Message:  "Tracking domain configured for all senders.",
			}
			if !pass {
				switch {
				case missing > 0 && unverified > 0:
					check.Message = fmt.Sprintf("%d sender account(s) missing tracking domain, %d not verified.", missing, unverified)
				case unverified > 0:
					check.Message = fmt.Sprintf("%d sender account(s) have an unverified tracking domain.", unverified)
				default:
					check.Message = fmt.Sprintf("%d sender account(s) missing tracking domain.", missing)
				}
				check.Remediation = "Set a tracking domain on every sender account used by this campaign and verify its CNAME."
				recommendations = append(recommendations, "Configure and verify tracking domains on all sender accounts.")
			}
			checks = append(checks, check)
		}
	}

	if settings.Preflight.CheckABVariantConfigured && settings.ABTesting.Enabled {
		variants, err := s.repo.ListABVariants(ctx, campaignID)
		if err != nil {
			return nil, toErrx(err)
		}
		active := 0
		for _, v := range variants {
			if v.IsActive {
				active++
			}
		}
		pass := active >= 2
		check := models.PreflightCheckResult{
			Key:      "ab_variants",
			Passed:   pass,
			Severity: "warning",
			Message:  "A/B variants are configured.",
		}
		if !pass {
			check.Message = "At least two active A/B variants are required."
			check.Remediation = "Create at least two active variants or disable AB testing."
			recommendations = append(recommendations, "Create more A/B variants.")
		}
		checks = append(checks, check)
	}

	if s.audienceRepo != nil {
		checks = append(checks, s.listQualityCheck(ctx, organizationID, campaignID, &recommendations))
	}

	if settings.Preflight.CheckContentScore {
		checks = append(checks, s.contentScoreCheck(ctx, campaignID, settings.Preflight.MinContentScore, &recommendations))
	}

	passedCount := 0
	for _, c := range checks {
		if c.Passed {
			passedCount++
		}
	}
	score := 100
	if len(checks) > 0 {
		score = int(float64(passedCount) / float64(len(checks)) * 100.0)
	}
	passed := passedCount == len(checks)

	report := &models.PreflightReport{
		ID:              uuid.New(),
		OrganizationID:  organizationID,
		CampaignID:      campaignID,
		Passed:          passed,
		Score:           score,
		Checks:          checks,
		Recommendations: recommendations,
		CreatedAt:       time.Now().UTC(),
	}
	if err := s.repo.CreatePreflightReport(ctx, report); err != nil {
		return nil, toErrx(err)
	}
	return report, nil
}

func (s *service) GetDeliverabilityDashboard(ctx context.Context, organizationID uuid.UUID, from, to time.Time) (*models.DeliverabilityDashboard, *errx.Error) {
	out, err := s.repo.GetDeliverabilityDashboard(ctx, organizationID, from, to)
	if err != nil {
		return nil, toErrx(err)
	}
	return out, nil
}

func (s *service) GetABWinnerAnalysis(ctx context.Context, organizationID, campaignID uuid.UUID) (*models.ABWinnerAnalysis, *errx.Error) {
	campaign, err := s.campaignRepo.GetByID(ctx, campaignID)
	if err != nil || campaign == nil {
		return nil, errx.ErrNotFound
	}
	if campaign.OrganizationID == nil || *campaign.OrganizationID != organizationID {
		return nil, errx.ErrForbidden
	}

	settings, xerr := s.effectiveSettings(ctx, organizationID, campaignID)
	if xerr != nil {
		return nil, xerr
	}

	stats, err := s.repo.GetABVariantStats(ctx, campaignID)
	if err != nil {
		return nil, toErrx(err)
	}

	analysis := &models.ABWinnerAnalysis{
		CampaignID:  campaignID,
		Variants:    stats,
		WinningRule: settings.ABTesting.DefaultWinningRule,
	}

	if len(stats) == 0 {
		analysis.Confidence = "none"
		return analysis, nil
	}

	// Determine winner based on the winning rule
	var bestIdx int
	var bestScore float64
	for i, v := range stats {
		var score float64
		switch settings.ABTesting.DefaultWinningRule {
		case "reply_rate":
			score = v.ReplyRate
		case "click_rate":
			score = v.ClickRate
		case "open_rate":
			score = v.OpenRate
		default:
			score = v.ReplyRate
		}
		if score > bestScore {
			bestScore = score
			bestIdx = i
		}
	}

	minSample := settings.ABTesting.MinSampleSize
	if minSample <= 0 {
		minSample = 30
	}

	winner := stats[bestIdx]
	if winner.TotalSent >= minSample {
		analysis.WinnerID = &winner.VariantID
		analysis.WinnerName = winner.VariantName
		if winner.TotalSent >= minSample*3 {
			analysis.Confidence = "high"
		} else if winner.TotalSent >= minSample {
			analysis.Confidence = "medium"
		}
	} else {
		analysis.Confidence = "low"
	}

	return analysis, nil
}

func (s *service) ProcessRetryableDeadLetters(ctx context.Context) (int, *errx.Error) {
	if s.tasksClient == nil {
		return 0, nil
	}

	items, err := s.repo.ListRetryableDeadLetters(ctx, 10)
	if err != nil {
		return 0, toErrx(err)
	}

	retried := 0
	for _, dlq := range items {
		task, err := s.taskRepo.GetTask(ctx, dlq.TaskID)
		if err != nil || task == nil {
			// Mark as exhausted if the task no longer exists
			_ = s.repo.MarkTaskDeadLetterReplayed(ctx, dlq.ID)
			continue
		}

		// Check if attempts exceeded
		if dlq.Attempts >= dlq.MaxAttempts {
			_ = s.repo.IncrementDeadLetterAttempt(ctx, dlq.ID, nil)
			continue
		}

		// Schedule retry via Cloud Tasks
		scheduleAt := time.Now().UTC().Add(10 * time.Second)
		cloudTaskName, err := s.tasksClient.CreateTask(ctx, &proto.ProcessTask{TaskId: task.ID.String()}, scheduleAt)
		if err != nil {
			// Compute next retry with exponential backoff
			backoff := time.Duration(30*(1<<uint(dlq.Attempts+1))) * time.Second
			nextRetry := time.Now().UTC().Add(backoff)
			_ = s.repo.IncrementDeadLetterAttempt(ctx, dlq.ID, &nextRetry)
			continue
		}

		if err := s.taskRepo.UpdateTaskScheduledAt(ctx, task.ID, scheduleAt, cloudTaskName); err != nil {
			continue
		}
		_ = s.taskRepo.UpdateTaskStatus(ctx, task.ID, "pending")
		_ = s.repo.MarkTaskDeadLetterReplayed(ctx, dlq.ID)
		retried++
	}

	return retried, nil
}

// worstStepContentScore returns the lowest-scoring email step's score, number,
// leading issue, and how many steps were scored. Only email steps carry copy: a
// wait or action node would otherwise score as the campaign's worst content.
// attachmentsFor gives the file count of one step's send, which differs per
// step now that a file can be scoped to one.
func worstStepContentScore(seqs []models.Sequence, attachmentsFor func(models.Sequence) int) (worst, worstStep int, issue string, scored int) {
	worst = 101
	for _, seq := range seqs {
		if seq.Kind != "" && seq.Kind != "email" {
			continue
		}
		scored++
		attachments := 0
		if attachmentsFor != nil {
			attachments = attachmentsFor(seq)
		}
		r := warmlint.ScoreWithAttachments(seq.Subject, seq.BodyHTML, seq.BodyPlain, attachments)
		if r.Score >= worst {
			continue
		}
		worst, worstStep, issue = r.Score, seq.Position+1, ""
		for _, is := range r.Issues {
			if is.Severity == "high" {
				issue = is.Message
				break
			}
		}
		if issue == "" && len(r.Issues) > 0 {
			issue = r.Issues[0].Message
		}
	}
	return worst, worstStep, issue, scored
}

// contentScoreCheck scores every step's copy and reports the worst. A step list
// it could not read reports as FAILED, not passed: a check that did not run
// must never look like one that succeeded.
func (s *service) contentScoreCheck(ctx context.Context, campaignID uuid.UUID, floor int, recommendations *[]string) models.PreflightCheckResult {
	// Out of range means a row written before the floor was clamped.
	if floor <= 0 || floor > 100 {
		floor = 60
	}
	seqs, err := s.campaignRepo.GetSequencesByCampaignID(ctx, campaignID)
	if err != nil {
		*recommendations = append(*recommendations, "Re-run preflight; the campaign's steps could not be read.")
		return models.PreflightCheckResult{
			Key:         "content_score",
			Passed:      false,
			Severity:    "warning",
			Message:     "Could not read the campaign's steps to score their copy.",
			Remediation: "Re-run preflight.",
		}
	}
	if len(seqs) == 0 {
		return models.PreflightCheckResult{
			Key:      "content_score",
			Passed:   true,
			Severity: "info",
			Message:  "No steps to score yet.",
		}
	}

	// The send path scores the files each step actually carries, so preflight
	// counts them the same way (campaign-wide plus that step's own) rather than
	// reporting a score the feed later contradicts.
	campaignWide := 0
	perStep := map[uuid.UUID]int{}
	if s.attachmentRepo != nil {
		atts, aerr := s.attachmentRepo.ListByCampaign(ctx, campaignID)
		if aerr != nil {
			// Scoring as none would pass copy the send path then warns about.
			*recommendations = append(*recommendations, "Re-run preflight; the campaign's attachments could not be read.")
			return models.PreflightCheckResult{
				Key:         "content_score",
				Passed:      false,
				Severity:    "warning",
				Message:     "Could not read the campaign's attachments to score its copy.",
				Remediation: "Re-run preflight.",
			}
		}
		for _, a := range atts {
			if a.SequenceID == nil {
				campaignWide++
				continue
			}
			perStep[*a.SequenceID]++
		}
	}

	worst, worstStep, issue, scored := worstStepContentScore(seqs, func(seq models.Sequence) int {
		return campaignWide + perStep[seq.ID]
	})
	if scored == 0 {
		return models.PreflightCheckResult{
			Key:      "content_score",
			Passed:   true,
			Severity: "info",
			Message:  "No email steps to score yet.",
		}
	}

	if worst >= floor {
		return models.PreflightCheckResult{
			Key:      "content_score",
			Passed:   true,
			Severity: "info",
			Message:  fmt.Sprintf("Copy scores %d/100 for spam signals.", worst),
		}
	}
	*recommendations = append(*recommendations, "Rewrite the lowest-scoring step's copy before sending at volume.")
	return models.PreflightCheckResult{
		Key:         "content_score",
		Passed:      false,
		Severity:    "warning",
		Message:     fmt.Sprintf("Step %d scores %d/100 for spam signals (floor %d). %s", worstStep, worst, floor, issue),
		Remediation: "Trim spam-trigger wording, links, images and stacked punctuation in that step.",
	}
}

// listQualityCheck projects the campaign's bounce rate for the preflight
// report, using the same rule StartCampaign refuses on, so the launch dialog
// cannot say one thing and the launch another.
func (s *service) listQualityCheck(ctx context.Context, orgID, campaignID uuid.UUID, recommendations *[]string) models.PreflightCheckResult {
	audience, err := s.audienceRepo.GetCampaignAudience(ctx, orgID, campaignID)
	if err != nil {
		return models.PreflightCheckResult{
			Key:      "list_quality",
			Passed:   false,
			Severity: "warning",
			Message:  "Could not read the campaign's audience to check it.",
		}
	}
	v := listgate.Project(audience)
	check := models.PreflightCheckResult{
		Key:         "list_quality",
		Passed:      !v.Block && !v.Warn,
		Severity:    "warning",
		Message:     v.Summary,
		Remediation: v.Remediation,
	}
	if v.Block {
		check.Severity = "error"
	}
	if !check.Passed {
		*recommendations = append(*recommendations, "Clean or verify the recipient list before sending at volume.")
	}
	return check
}

// WireAudience attaches the launch-time list measurement.
// EvidenceRecorder mirrors emailverify.EvidenceRecorder without importing it.
type EvidenceRecorder interface {
	RecordEvidence(ctx context.Context, contactID uuid.UUID, kind, ref, detail string)
}

// EvidenceAware lets main hand the service the verification evidence ledger.
type EvidenceAware interface {
	WireEvidence(e EvidenceRecorder)
}

func (s *service) WireEvidence(e EvidenceRecorder) { s.evidence = e }

func (s *service) WireAudience(r repository.CampaignAudienceRepository) {
	s.audienceRepo = r
}

// AudienceAware is the optional capability the caller uses to attach it.
type AudienceAware interface {
	WireAudience(r repository.CampaignAudienceRepository)
}

// WireAttachments attaches the campaign attachment counter the content check
// scores with, so preflight and the send path weigh attachments the same way.
func (s *service) WireAttachments(r repository.AttachmentRepository) {
	s.attachmentRepo = r
}

// AttachmentAware is the optional capability the caller uses to attach it.
type AttachmentAware interface {
	WireAttachments(r repository.AttachmentRepository)
}

// main attaches this by type assertion, which fails silently, so pin it here.
var _ AttachmentAware = (*service)(nil)
