package repository

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

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/warmbly/warmbly/internal/models"
)

type AdvancedOutreachRepository interface {
	GetOutreachSettings(ctx context.Context, organizationID uuid.UUID) (*models.AdvancedOutreachSettings, error)
	UpsertOutreachSettings(ctx context.Context, organizationID, updatedBy uuid.UUID, settings *models.AdvancedOutreachSettings) error

	GetCampaignAdvancedSettings(ctx context.Context, campaignID uuid.UUID) (*models.CampaignAdvancedSettings, error)
	UpsertCampaignAdvancedSettings(ctx context.Context, campaignID uuid.UUID, settings *models.AdvancedOutreachSettings) error

	ListABVariants(ctx context.Context, campaignID uuid.UUID) ([]models.CampaignABVariant, error)
	CreateABVariant(ctx context.Context, campaignID uuid.UUID, req *models.CreateCampaignABVariantRequest) (*models.CampaignABVariant, error)
	UpdateABVariant(ctx context.Context, campaignID, variantID uuid.UUID, req *models.UpdateCampaignABVariantRequest) (*models.CampaignABVariant, error)
	DeleteABVariant(ctx context.Context, campaignID, variantID uuid.UUID) error
	GetAssignedVariant(ctx context.Context, campaignID, contactID uuid.UUID) (*models.CampaignABVariant, error)
	AssignVariant(ctx context.Context, campaignID, contactID, variantID uuid.UUID) error
	MarkVariantEvent(ctx context.Context, campaignID, contactID uuid.UUID, eventType string) error

	IsRecipientSuppressed(ctx context.Context, organizationID uuid.UUID, email string) (*models.SuppressedRecipient, error)
	UpsertSuppressedRecipient(ctx context.Context, entry *models.SuppressedRecipient) error
	// UpsertSuppressedRecipients writes a batch in one transaction, so a
	// pasted list lands whole or not at all.
	UpsertSuppressedRecipients(ctx context.Context, entries []models.SuppressedRecipient) error
	// ListSuppressedRecipients pages the active list newest first. q filters by
	// address or domain substring; before is the keyset (created_at, id).
	ListSuppressedRecipients(ctx context.Context, organizationID uuid.UUID, q string, beforeAt *time.Time, beforeID *uuid.UUID, limit int) ([]models.SuppressedRecipient, error)
	GetSuppressedRecipient(ctx context.Context, organizationID, id uuid.UUID) (*models.SuppressedRecipient, error)
	DeleteSuppressedRecipient(ctx context.Context, organizationID, id uuid.UUID) (bool, error)
	// DeleteSuppressionByEmail removes the address entry with the given source
	// (the recipient's own resubscribe only undoes a recipient-made entry).
	DeleteSuppressionByEmail(ctx context.Context, organizationID uuid.UUID, email string, source models.DeliverabilityEventType) (bool, error)

	CreateDeliverabilityEvent(ctx context.Context, event *models.DeliverabilityEvent) error
	GetDeliverabilityDashboard(ctx context.Context, organizationID uuid.UUID, from, to time.Time) (*models.DeliverabilityDashboard, error)

	StartTaskExecution(ctx context.Context, taskID uuid.UUID, executionKey string, metadata map[string]interface{}) (bool, error)
	CompleteTaskExecution(ctx context.Context, taskID uuid.UUID, executionKey, status string, metadata map[string]interface{}) error

	CreateTaskDeadLetter(ctx context.Context, item *models.TaskDeadLetter) error
	ListTaskDeadLetters(ctx context.Context, organizationID uuid.UUID, status string, limit int) ([]models.TaskDeadLetter, error)
	GetTaskDeadLetter(ctx context.Context, id uuid.UUID, organizationID uuid.UUID) (*models.TaskDeadLetter, error)
	MarkTaskDeadLetterReplayed(ctx context.Context, id uuid.UUID) error

	CreateReplyIntent(ctx context.Context, record *models.ReplyIntentRecord) error
	CreatePreflightReport(ctx context.Context, report *models.PreflightReport) error

	GetABVariantStats(ctx context.Context, campaignID uuid.UUID) ([]models.ABVariantStats, error)

	// DLQ auto-retry
	ListRetryableDeadLetters(ctx context.Context, limit int) ([]models.TaskDeadLetter, error)
	IncrementDeadLetterAttempt(ctx context.Context, id uuid.UUID, nextRetryAt *time.Time) error
}

type advancedOutreachRepository struct {
	db *pgxpool.Pool
}

func NewAdvancedOutreachRepository(db *pgxpool.Pool) AdvancedOutreachRepository {
	return &advancedOutreachRepository{db: db}
}

func marshalJSON(v interface{}) ([]byte, error) {
	if v == nil {
		return []byte(`{}`), nil
	}
	return json.Marshal(v)
}

func unmarshalJSON[T any](b []byte, out *T) error {
	if len(b) == 0 {
		return nil
	}
	return json.Unmarshal(b, out)
}

func (r *advancedOutreachRepository) GetOutreachSettings(ctx context.Context, organizationID uuid.UUID) (*models.AdvancedOutreachSettings, error) {
	query := `SELECT settings FROM outreach_settings WHERE organization_id = $1`
	var raw []byte
	if err := r.db.QueryRow(ctx, query, organizationID).Scan(&raw); err != nil {
		if err == pgx.ErrNoRows {
			def := models.DefaultAdvancedOutreachSettings()
			return &def, nil
		}
		return nil, err
	}

	settings := models.DefaultAdvancedOutreachSettings()
	if err := unmarshalJSON(raw, &settings); err != nil {
		return nil, err
	}
	return &settings, nil
}

func (r *advancedOutreachRepository) UpsertOutreachSettings(ctx context.Context, organizationID, updatedBy uuid.UUID, settings *models.AdvancedOutreachSettings) error {
	raw, err := marshalJSON(settings)
	if err != nil {
		return err
	}
	query := `
		INSERT INTO outreach_settings (organization_id, settings, updated_by, updated_at)
		VALUES ($1, $2, $3, NOW())
		ON CONFLICT (organization_id)
		DO UPDATE SET settings = EXCLUDED.settings, updated_by = EXCLUDED.updated_by, updated_at = NOW()
	`
	_, err = r.db.Exec(ctx, query, organizationID, raw, updatedBy)
	return err
}

func (r *advancedOutreachRepository) GetCampaignAdvancedSettings(ctx context.Context, campaignID uuid.UUID) (*models.CampaignAdvancedSettings, error) {
	query := `SELECT settings, updated_at FROM campaign_advanced_settings WHERE campaign_id = $1`
	var raw []byte
	var updatedAt time.Time
	if err := r.db.QueryRow(ctx, query, campaignID).Scan(&raw, &updatedAt); err != nil {
		if err == pgx.ErrNoRows {
			return nil, nil
		}
		return nil, err
	}
	settings := models.DefaultAdvancedOutreachSettings()
	if err := unmarshalJSON(raw, &settings); err != nil {
		return nil, err
	}
	return &models.CampaignAdvancedSettings{
		CampaignID: campaignID,
		Overrides:  settings,
		UpdatedAt:  updatedAt,
	}, nil
}

func (r *advancedOutreachRepository) UpsertCampaignAdvancedSettings(ctx context.Context, campaignID uuid.UUID, settings *models.AdvancedOutreachSettings) error {
	raw, err := marshalJSON(settings)
	if err != nil {
		return err
	}
	query := `
		INSERT INTO campaign_advanced_settings (campaign_id, settings, updated_at)
		VALUES ($1, $2, NOW())
		ON CONFLICT (campaign_id)
		DO UPDATE SET settings = EXCLUDED.settings, updated_at = NOW()
	`
	_, err = r.db.Exec(ctx, query, campaignID, raw)
	return err
}

func scanABVariant(rows pgx.Row, v *models.CampaignABVariant) error {
	var metadata []byte
	if err := rows.Scan(
		&v.ID,
		&v.CampaignID,
		&v.SequenceID,
		&v.Name,
		&v.Weight,
		&v.Subject,
		&v.BodyHTML,
		&v.BodyPlain,
		&v.IsControl,
		&v.IsActive,
		&metadata,
		&v.CreatedAt,
		&v.UpdatedAt,
	); err != nil {
		return err
	}
	if len(metadata) > 0 {
		if err := json.Unmarshal(metadata, &v.Metadata); err != nil {
			return err
		}
	}
	if v.Metadata == nil {
		v.Metadata = map[string]interface{}{}
	}
	return nil
}

func (r *advancedOutreachRepository) ListABVariants(ctx context.Context, campaignID uuid.UUID) ([]models.CampaignABVariant, error) {
	query := `
		SELECT id, campaign_id, sequence_id, name, weight, subject, body_html, body_plain, is_control, is_active, metadata, created_at, updated_at
		FROM campaign_ab_variants
		WHERE campaign_id = $1
		ORDER BY is_control DESC, created_at ASC
	`
	rows, err := r.db.Query(ctx, query, campaignID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var out []models.CampaignABVariant
	for rows.Next() {
		var v models.CampaignABVariant
		if err := scanABVariant(rows, &v); err != nil {
			return nil, err
		}
		out = append(out, v)
	}
	return out, rows.Err()
}

func (r *advancedOutreachRepository) CreateABVariant(ctx context.Context, campaignID uuid.UUID, req *models.CreateCampaignABVariantRequest) (*models.CampaignABVariant, error) {
	metadata, err := marshalJSON(req.Metadata)
	if err != nil {
		return nil, err
	}
	isActive := true
	if req.IsActive != nil {
		isActive = *req.IsActive
	}
	weight := req.Weight
	if weight <= 0 {
		weight = 100
	}
	query := `
		INSERT INTO campaign_ab_variants (
			campaign_id, sequence_id, name, weight, subject, body_html, body_plain, is_control, is_active, metadata, created_at, updated_at
		)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, NOW(), NOW())
		RETURNING id, campaign_id, sequence_id, name, weight, subject, body_html, body_plain, is_control, is_active, metadata, created_at, updated_at
	`
	var out models.CampaignABVariant
	if err := scanABVariant(r.db.QueryRow(ctx, query,
		campaignID,
		req.SequenceID,
		req.Name,
		weight,
		req.Subject,
		req.BodyHTML,
		req.BodyPlain,
		req.IsControl,
		isActive,
		metadata,
	), &out); err != nil {
		return nil, err
	}
	return &out, nil
}

func (r *advancedOutreachRepository) UpdateABVariant(ctx context.Context, campaignID, variantID uuid.UUID, req *models.UpdateCampaignABVariantRequest) (*models.CampaignABVariant, error) {
	sets := make([]string, 0, 8)
	args := []interface{}{campaignID, variantID}
	argPos := 3

	if req.Name != nil {
		sets = append(sets, fmt.Sprintf("name = $%d", argPos))
		args = append(args, *req.Name)
		argPos++
	}
	if req.Weight != nil {
		sets = append(sets, fmt.Sprintf("weight = $%d", argPos))
		args = append(args, *req.Weight)
		argPos++
	}
	if req.Subject != nil {
		sets = append(sets, fmt.Sprintf("subject = $%d", argPos))
		args = append(args, *req.Subject)
		argPos++
	}
	if req.BodyHTML != nil {
		sets = append(sets, fmt.Sprintf("body_html = $%d", argPos))
		args = append(args, *req.BodyHTML)
		argPos++
	}
	if req.BodyPlain != nil {
		sets = append(sets, fmt.Sprintf("body_plain = $%d", argPos))
		args = append(args, *req.BodyPlain)
		argPos++
	}
	if req.IsControl != nil {
		sets = append(sets, fmt.Sprintf("is_control = $%d", argPos))
		args = append(args, *req.IsControl)
		argPos++
	}
	if req.IsActive != nil {
		sets = append(sets, fmt.Sprintf("is_active = $%d", argPos))
		args = append(args, *req.IsActive)
		argPos++
	}
	if req.Metadata != nil {
		metadata, err := marshalJSON(req.Metadata)
		if err != nil {
			return nil, err
		}
		sets = append(sets, fmt.Sprintf("metadata = $%d", argPos))
		args = append(args, metadata)
		argPos++
	}

	if len(sets) == 0 {
		return nil, fmt.Errorf("no fields to update")
	}
	sets = append(sets, "updated_at = NOW()")

	query := fmt.Sprintf(`
		UPDATE campaign_ab_variants
		SET %s
		WHERE campaign_id = $1 AND id = $2
		RETURNING id, campaign_id, sequence_id, name, weight, subject, body_html, body_plain, is_control, is_active, metadata, created_at, updated_at
	`, strings.Join(sets, ", "))

	var out models.CampaignABVariant
	if err := scanABVariant(r.db.QueryRow(ctx, query, args...), &out); err != nil {
		return nil, err
	}
	return &out, nil
}

func (r *advancedOutreachRepository) DeleteABVariant(ctx context.Context, campaignID, variantID uuid.UUID) error {
	query := `DELETE FROM campaign_ab_variants WHERE campaign_id = $1 AND id = $2`
	cmd, err := r.db.Exec(ctx, query, campaignID, variantID)
	if err != nil {
		return err
	}
	if cmd.RowsAffected() == 0 {
		return pgx.ErrNoRows
	}
	return nil
}

func (r *advancedOutreachRepository) GetAssignedVariant(ctx context.Context, campaignID, contactID uuid.UUID) (*models.CampaignABVariant, error) {
	query := `
		SELECT v.id, v.campaign_id, v.sequence_id, v.name, v.weight, v.subject, v.body_html, v.body_plain, v.is_control, v.is_active, v.metadata, v.created_at, v.updated_at
		FROM campaign_ab_assignments a
		JOIN campaign_ab_variants v ON v.id = a.variant_id
		WHERE a.campaign_id = $1 AND a.contact_id = $2
	`
	var out models.CampaignABVariant
	if err := scanABVariant(r.db.QueryRow(ctx, query, campaignID, contactID), &out); err != nil {
		if err == pgx.ErrNoRows {
			return nil, nil
		}
		return nil, err
	}
	return &out, nil
}

func (r *advancedOutreachRepository) AssignVariant(ctx context.Context, campaignID, contactID, variantID uuid.UUID) error {
	query := `
		INSERT INTO campaign_ab_assignments (campaign_id, contact_id, variant_id, assigned_at)
		VALUES ($1, $2, $3, NOW())
		ON CONFLICT (campaign_id, contact_id)
		DO UPDATE SET variant_id = EXCLUDED.variant_id, assigned_at = NOW()
	`
	_, err := r.db.Exec(ctx, query, campaignID, contactID, variantID)
	return err
}

func (r *advancedOutreachRepository) MarkVariantEvent(ctx context.Context, campaignID, contactID uuid.UUID, eventType string) error {
	var setClause string
	switch eventType {
	case string(models.DeliverabilityEventOpen):
		setClause = "opened_at = COALESCE(opened_at, NOW())"
	case string(models.DeliverabilityEventClick):
		setClause = "clicked_at = COALESCE(clicked_at, NOW())"
	case string(models.DeliverabilityEventReply):
		setClause = "replied_at = COALESCE(replied_at, NOW())"
	case string(models.DeliverabilityEventBounce):
		setClause = "bounced_at = COALESCE(bounced_at, NOW())"
	default:
		return nil
	}
	query := fmt.Sprintf(`UPDATE campaign_ab_assignments SET %s WHERE campaign_id = $1 AND contact_id = $2`, setClause)
	_, err := r.db.Exec(ctx, query, campaignID, contactID)
	return err
}

const suppressedRecipientColumns = `id, organization_id, email, kind, reason, source, campaign_id, expires_at, metadata, created_at, updated_at`

func scanSuppressedRecipient(row pgx.Row) (*models.SuppressedRecipient, error) {
	var out models.SuppressedRecipient
	var metadata []byte
	if err := row.Scan(
		&out.ID,
		&out.OrganizationID,
		&out.Email,
		&out.Kind,
		&out.Reason,
		&out.Source,
		&out.CampaignID,
		&out.ExpiresAt,
		&metadata,
		&out.CreatedAt,
		&out.UpdatedAt,
	); err != nil {
		return nil, err
	}
	if len(metadata) > 0 {
		_ = json.Unmarshal(metadata, &out.Metadata)
	}
	return &out, nil
}

// IsRecipientSuppressed returns the entry that blocks email: its own address
// row first, else a row for its domain. Same predicate as the SQL function
// recipient_suppressed() the send gates use, spelled out here because the
// caller wants the row, not a boolean. Stored values are lowercase (every
// write folds them; migration 000124 folded the rest), so the comparison is
// an equality the unique index serves.
func (r *advancedOutreachRepository) IsRecipientSuppressed(ctx context.Context, organizationID uuid.UUID, email string) (*models.SuppressedRecipient, error) {
	query := `
		SELECT ` + suppressedRecipientColumns + `
		FROM suppressed_recipients
		WHERE organization_id = $1
		  AND (expires_at IS NULL OR expires_at > NOW())
		  AND (
		        (kind = 'email' AND email = LOWER($2))
		     OR (kind = 'domain' AND email = split_part(LOWER($2), '@', 2))
		  )
		ORDER BY (kind = 'email') DESC
		LIMIT 1
	`
	out, err := scanSuppressedRecipient(r.db.QueryRow(ctx, query, organizationID, email))
	if err != nil {
		if err == pgx.ErrNoRows {
			return nil, nil
		}
		return nil, err
	}
	return out, nil
}

func (r *advancedOutreachRepository) UpsertSuppressedRecipient(ctx context.Context, entry *models.SuppressedRecipient) error {
	metadata, err := marshalJSON(entry.Metadata)
	if err != nil {
		return err
	}
	kind := entry.Kind
	if kind == "" {
		kind = models.SuppressionKindEmail
	}
	query := `
		INSERT INTO suppressed_recipients (organization_id, email, kind, reason, source, campaign_id, expires_at, metadata, created_at, updated_at)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW(), NOW())
		ON CONFLICT (organization_id, email)
		DO UPDATE SET
			kind = EXCLUDED.kind,
			reason = EXCLUDED.reason,
			source = EXCLUDED.source,
			campaign_id = EXCLUDED.campaign_id,
			expires_at = EXCLUDED.expires_at,
			metadata = EXCLUDED.metadata,
			updated_at = NOW()
	`
	_, err = r.db.Exec(ctx, query, entry.OrganizationID, strings.ToLower(strings.TrimSpace(entry.Email)), kind, entry.Reason, entry.Source, entry.CampaignID, entry.ExpiresAt, metadata)
	return err
}

const upsertSuppressedRecipientSQL = `
		INSERT INTO suppressed_recipients (organization_id, email, kind, reason, source, campaign_id, expires_at, metadata, created_at, updated_at)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW(), NOW())
		ON CONFLICT (organization_id, email)
		DO UPDATE SET
			kind = EXCLUDED.kind,
			reason = EXCLUDED.reason,
			source = EXCLUDED.source,
			campaign_id = EXCLUDED.campaign_id,
			expires_at = EXCLUDED.expires_at,
			metadata = EXCLUDED.metadata,
			updated_at = NOW()
	`

func (r *advancedOutreachRepository) UpsertSuppressedRecipients(ctx context.Context, entries []models.SuppressedRecipient) error {
	if len(entries) == 0 {
		return nil
	}
	tx, err := r.db.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	batch := &pgx.Batch{}
	for i := range entries {
		e := &entries[i]
		metadata, err := marshalJSON(e.Metadata)
		if err != nil {
			return err
		}
		kind := e.Kind
		if kind == "" {
			kind = models.SuppressionKindEmail
		}
		batch.Queue(upsertSuppressedRecipientSQL, e.OrganizationID, strings.ToLower(strings.TrimSpace(e.Email)), kind, e.Reason, e.Source, e.CampaignID, e.ExpiresAt, metadata)
	}
	res := tx.SendBatch(ctx, batch)
	for range entries {
		if _, err := res.Exec(); err != nil {
			_ = res.Close()
			return err
		}
	}
	if err := res.Close(); err != nil {
		return err
	}
	return tx.Commit(ctx)
}

func (r *advancedOutreachRepository) ListSuppressedRecipients(ctx context.Context, organizationID uuid.UUID, q string, beforeAt *time.Time, beforeID *uuid.UUID, limit int) ([]models.SuppressedRecipient, error) {
	args := []any{organizationID, limit}
	where := `organization_id = $1 AND (expires_at IS NULL OR expires_at > NOW())`
	if q = strings.ToLower(strings.TrimSpace(q)); q != "" {
		args = append(args, "%"+q+"%")
		where += fmt.Sprintf(` AND email ILIKE $%d`, len(args))
	}
	if beforeAt != nil && beforeID != nil {
		args = append(args, *beforeAt, *beforeID)
		where += fmt.Sprintf(` AND (created_at, id) < ($%d, $%d)`, len(args)-1, len(args))
	}
	rows, err := r.db.Query(ctx, `SELECT `+suppressedRecipientColumns+` FROM suppressed_recipients WHERE `+where+` ORDER BY created_at DESC, id DESC LIMIT $2`, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := make([]models.SuppressedRecipient, 0, limit)
	for rows.Next() {
		entry, err := scanSuppressedRecipient(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, *entry)
	}
	return out, rows.Err()
}

func (r *advancedOutreachRepository) GetSuppressedRecipient(ctx context.Context, organizationID, id uuid.UUID) (*models.SuppressedRecipient, error) {
	out, err := scanSuppressedRecipient(r.db.QueryRow(ctx, `SELECT `+suppressedRecipientColumns+` FROM suppressed_recipients WHERE organization_id = $1 AND id = $2`, organizationID, id))
	if err != nil {
		if err == pgx.ErrNoRows {
			return nil, nil
		}
		return nil, err
	}
	return out, nil
}

func (r *advancedOutreachRepository) DeleteSuppressedRecipient(ctx context.Context, organizationID, id uuid.UUID) (bool, error) {
	tag, err := r.db.Exec(ctx, `DELETE FROM suppressed_recipients WHERE organization_id = $1 AND id = $2`, organizationID, id)
	if err != nil {
		return false, err
	}
	return tag.RowsAffected() > 0, nil
}

func (r *advancedOutreachRepository) DeleteSuppressionByEmail(ctx context.Context, organizationID uuid.UUID, email string, source models.DeliverabilityEventType) (bool, error) {
	tag, err := r.db.Exec(ctx, `DELETE FROM suppressed_recipients WHERE organization_id = $1 AND kind = 'email' AND email = LOWER($2) AND source = $3`, organizationID, strings.TrimSpace(email), source)
	if err != nil {
		return false, err
	}
	return tag.RowsAffected() > 0, nil
}

func (r *advancedOutreachRepository) CreateDeliverabilityEvent(ctx context.Context, event *models.DeliverabilityEvent) error {
	metadata, err := marshalJSON(event.Metadata)
	if err != nil {
		return err
	}
	query := `
		INSERT INTO deliverability_events (
			organization_id, campaign_id, task_id, contact_id, event_type, provider,
			recipient_email, reason, idempotency_key, metadata, created_at
		)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, NOW())
		ON CONFLICT (idempotency_key) DO NOTHING
	`
	_, err = r.db.Exec(ctx, query,
		event.OrganizationID,
		event.CampaignID,
		event.TaskID,
		event.ContactID,
		event.EventType,
		event.Provider,
		strings.ToLower(strings.TrimSpace(event.RecipientEmail)),
		event.Reason,
		event.IdempotencyKey,
		metadata,
	)
	return err
}

func (r *advancedOutreachRepository) GetDeliverabilityDashboard(ctx context.Context, organizationID uuid.UUID, from, to time.Time) (*models.DeliverabilityDashboard, error) {
	out := &models.DeliverabilityDashboard{
		From: from,
		To:   to,
	}

	queryEvents := `
		SELECT
			COUNT(*) AS total,
			COUNT(*) FILTER (WHERE event_type = 'bounce') AS bounces,
			COUNT(*) FILTER (WHERE event_type = 'complaint') AS complaints,
			COUNT(*) FILTER (WHERE event_type = 'unsubscribe') AS unsubscribes,
			COUNT(*) FILTER (WHERE event_type = 'reply') AS replies,
			COUNT(*) FILTER (WHERE event_type = 'open') AS opens,
			COUNT(*) FILTER (WHERE event_type = 'click') AS clicks
		FROM deliverability_events
		WHERE organization_id = $1
		  AND created_at >= $2
		  AND created_at <= $3
	`
	if err := r.db.QueryRow(ctx, queryEvents, organizationID, from, to).Scan(
		&out.EventsTotal,
		&out.BounceCount,
		&out.ComplaintCount,
		&out.UnsubscribeCount,
		&out.ReplyCount,
		&out.OpenCount,
		&out.ClickCount,
	); err != nil {
		return nil, err
	}

	querySuppressed := `SELECT COUNT(*) FROM suppressed_recipients WHERE organization_id = $1 AND (expires_at IS NULL OR expires_at > NOW())`
	if err := r.db.QueryRow(ctx, querySuppressed, organizationID).Scan(&out.SuppressedRecipients); err != nil {
		return nil, err
	}

	queryDLQ := `
		SELECT COUNT(*)
		FROM task_dead_letters d
		JOIN tasks t ON t.id = d.task_id
		JOIN email_accounts ea ON ea.id = t.email_account_id
		WHERE ea.organization_id = $1
		  AND d.status = 'pending'
	`
	if err := r.db.QueryRow(ctx, queryDLQ, organizationID).Scan(&out.DLQPending); err != nil {
		return nil, err
	}

	queryIntents := `
		SELECT
			COUNT(*) FILTER (WHERE intent = 'positive') AS positive,
			COUNT(*) FILTER (WHERE intent = 'negative') AS negative,
			COUNT(*) FILTER (WHERE intent = 'out_of_office') AS ooo,
			COUNT(*) FILTER (WHERE intent = 'question') AS question,
			COUNT(*) FILTER (WHERE intent = 'neutral') AS neutral
		FROM reply_intents
		WHERE organization_id = $1
		  AND created_at >= $2
		  AND created_at <= $3
	`
	if err := r.db.QueryRow(ctx, queryIntents, organizationID, from, to).Scan(
		&out.IntentPositive,
		&out.IntentNegative,
		&out.IntentOOO,
		&out.IntentQuestion,
		&out.IntentNeutral,
	); err != nil {
		return nil, err
	}

	// Rate denominator: completed campaign sends in the window. Best-effort — a
	// rate/breakdown query failing must not break the (already-built) core rollup.
	sentQuery := `
		SELECT COUNT(*) FROM tasks t
		JOIN email_accounts ea ON ea.id = t.email_account_id
		WHERE ea.organization_id = $1 AND t.task_type = 'campaign' AND t.status = 'completed'
		  AND t.completed_at >= $2 AND t.completed_at <= $3
		  AND ` + taskDispatchedEmail
	_ = r.db.QueryRow(ctx, sentQuery, organizationID, from, to).Scan(&out.EmailsSent)
	out.BounceRate = models.Rate(out.BounceCount, out.EmailsSent)
	out.ComplaintRate = models.Rate(out.ComplaintCount, out.EmailsSent)
	out.OpenRate = models.Rate(out.OpenCount, out.EmailsSent)
	out.ClickRate = models.Rate(out.ClickCount, out.EmailsSent)
	out.ReplyRate = models.Rate(out.ReplyCount, out.EmailsSent)

	out.Timeseries = r.deliverabilityTimeseries(ctx, organizationID, from, to)
	out.ByCampaign = r.deliverabilityByCampaign(ctx, organizationID, from, to)
	out.ByMailbox = r.deliverabilityByMailbox(ctx, organizationID, from, to)

	// Seed inbox-placement per provider (optional — nil rates when there are no
	// seed samples). The flat org totals are derived from the same rows.
	var spam, inbox, placementTotal int
	byProvider := map[string]*models.ProviderPlacement{}
	placementQuery := `
		SELECT pr.provider, pr.folder, COUNT(*)
		FROM placement_results pr JOIN placement_tests pt ON pt.id = pr.test_id
		WHERE pt.organization_id = $1 AND pr.detected_at >= $2 AND pr.detected_at <= $3 AND pr.folder <> 'pending'
		GROUP BY pr.provider, pr.folder`
	if rows, perr := r.db.Query(ctx, placementQuery, organizationID, from, to); perr == nil {
		for rows.Next() {
			var provider, folder string
			var n int
			if rows.Scan(&provider, &folder, &n) == nil {
				placementTotal += n
				p := byProvider[provider]
				if p == nil {
					p = &models.ProviderPlacement{Provider: provider}
					byProvider[provider] = p
				}
				p.Samples += n
				switch folder {
				case "spam":
					spam += n
					p.Spam += n
				case "inbox":
					inbox += n
					p.Inbox += n
				case "promotions":
					p.Promotions += n
				default:
					p.Other += n
				}
			}
		}
		rows.Close()
	}
	out.ByProvider = make([]models.ProviderPlacement, 0, len(byProvider))
	for _, p := range byProvider {
		p.InboxRate = models.Rate(p.Inbox, p.Samples)
		p.SpamRate = models.Rate(p.Spam, p.Samples)
		out.ByProvider = append(out.ByProvider, *p)
	}
	sort.Slice(out.ByProvider, func(i, j int) bool {
		if out.ByProvider[i].Samples != out.ByProvider[j].Samples {
			return out.ByProvider[i].Samples > out.ByProvider[j].Samples
		}
		return out.ByProvider[i].Provider < out.ByProvider[j].Provider
	})

	out.PlacementSamples = placementTotal
	spamRate := 0.0
	if placementTotal > 0 {
		sr := models.Rate(spam, placementTotal)
		ir := models.Rate(inbox, placementTotal)
		out.SpamPlacementRate = &sr
		out.InboxPlacementRate = &ir
		spamRate = sr
	}

	out.WarmupPlacement = r.warmupPlacementByDomain(ctx, organizationID, from, to)

	out.Band = models.DeliverabilityBand(out.BounceRate, out.ComplaintRate, spamRate)
	out.Score = models.DeliverabilityScore(out.BounceRate, out.ComplaintRate, spamRate)
	return out, nil
}

// warmupPlacementByDomain rolls the continuous warmup placement signal up per
// recipient domain: delivered = verified warmup arrivals at partner mailboxes
// (warmup_received), spam = the subset the recipient's provider filed into
// junk (warmup_spam_reports, report_type=spam_placement). Org scope is the
// sending account. Best-effort: an error returns an empty list.
func (r *advancedOutreachRepository) warmupPlacementByDomain(ctx context.Context, orgID uuid.UUID, from, to time.Time) []models.WarmupDomainPlacement {
	out := []models.WarmupDomainPlacement{}
	byDomain := map[string]*models.WarmupDomainPlacement{}

	deliveredQ := `
		SELECT rcpt.provider, split_part(lower(rcpt.email), '@', 2), COUNT(*)
		FROM warmup_received wr
		JOIN email_accounts snd ON snd.id = wr.sender_account_id
		JOIN email_accounts rcpt ON rcpt.id = wr.email_account_id
		WHERE snd.organization_id = $1 AND wr.created_at >= $2 AND wr.created_at <= $3
		GROUP BY 1, 2`
	if rows, err := r.db.Query(ctx, deliveredQ, orgID, from, to); err == nil {
		for rows.Next() {
			var provider, domain string
			var n int
			if rows.Scan(&provider, &domain, &n) == nil && domain != "" {
				byDomain[domain] = &models.WarmupDomainPlacement{Provider: provider, Domain: domain, Delivered: n}
			}
		}
		rows.Close()
	}

	spamQ := `
		SELECT sr.recipient_provider, sr.recipient_domain, COUNT(*)
		FROM warmup_spam_reports sr
		JOIN email_accounts snd ON snd.id = sr.reported_account_id
		WHERE snd.organization_id = $1 AND sr.report_type = 'spam_placement'
		  AND sr.created_at >= $2 AND sr.created_at <= $3 AND sr.recipient_domain <> ''
		GROUP BY 1, 2`
	if rows, err := r.db.Query(ctx, spamQ, orgID, from, to); err == nil {
		for rows.Next() {
			var provider, domain string
			var n int
			if rows.Scan(&provider, &domain, &n) == nil {
				p := byDomain[domain]
				if p == nil {
					p = &models.WarmupDomainPlacement{Provider: provider, Domain: domain}
					byDomain[domain] = p
				}
				p.Spam += n
				// A spam-flagged arrival can be reported without (or before) its
				// warmup_received row; keep delivered >= spam so rates stay sane.
				if p.Delivered < p.Spam {
					p.Delivered = p.Spam
				}
			}
		}
		rows.Close()
	}

	for _, p := range byDomain {
		p.InboxRate = models.Rate(p.Delivered-p.Spam, p.Delivered)
		p.SpamRate = models.Rate(p.Spam, p.Delivered)
		out = append(out, *p)
	}
	sort.Slice(out, func(i, j int) bool {
		if out[i].Delivered != out[j].Delivered {
			return out[i].Delivered > out[j].Delivered
		}
		return out[i].Domain < out[j].Domain
	})
	if len(out) > 25 {
		out = out[:25]
	}
	return out
}

// deliverabilityTimeseries returns gap-filled UTC daily points for the window.
func (r *advancedOutreachRepository) deliverabilityTimeseries(ctx context.Context, orgID uuid.UUID, from, to time.Time) []models.DeliverabilityDailyPoint {
	byDay := map[string]*models.DeliverabilityDailyPoint{}
	// Bucket by UTC calendar day explicitly (created_at::date alone uses the DB
	// session timezone) so the SQL days line up with the Go gap-fill loop below.
	evQ := `
		SELECT (created_at AT TIME ZONE 'UTC')::date,
			COUNT(*) FILTER (WHERE event_type='bounce'),
			COUNT(*) FILTER (WHERE event_type='complaint'),
			COUNT(*) FILTER (WHERE event_type='open'),
			COUNT(*) FILTER (WHERE event_type='click'),
			COUNT(*) FILTER (WHERE event_type='reply'),
			COUNT(*) FILTER (WHERE event_type='unsubscribe')
		FROM deliverability_events
		WHERE organization_id=$1 AND created_at >= $2 AND created_at <= $3
		GROUP BY 1`
	if rows, err := r.db.Query(ctx, evQ, orgID, from, to); err == nil {
		for rows.Next() {
			var d time.Time
			var b, c, o, cl, rep, u int
			if rows.Scan(&d, &b, &c, &o, &cl, &rep, &u) == nil {
				key := d.Format("2006-01-02")
				byDay[key] = &models.DeliverabilityDailyPoint{Date: key, Bounces: b, Complaints: c, Opens: o, Clicks: cl, Replies: rep, Unsubscribes: u}
			}
		}
		rows.Close()
	}
	sentQ := `
		SELECT (t.completed_at AT TIME ZONE 'UTC')::date, COUNT(*)
		FROM tasks t JOIN email_accounts ea ON ea.id = t.email_account_id
		WHERE ea.organization_id=$1 AND t.task_type='campaign' AND t.status='completed'
		  AND t.completed_at >= $2 AND t.completed_at <= $3
		GROUP BY 1`
	if rows, err := r.db.Query(ctx, sentQ, orgID, from, to); err == nil {
		for rows.Next() {
			var d time.Time
			var s int
			if rows.Scan(&d, &s) == nil {
				key := d.Format("2006-01-02")
				if p := byDay[key]; p != nil {
					p.Sent = s
				} else {
					byDay[key] = &models.DeliverabilityDailyPoint{Date: key, Sent: s}
				}
			}
		}
		rows.Close()
	}
	out := []models.DeliverabilityDailyPoint{}
	for d := from.UTC().Truncate(24 * time.Hour); !d.After(to.UTC()); d = d.Add(24 * time.Hour) {
		key := d.Format("2006-01-02")
		if p := byDay[key]; p != nil {
			out = append(out, *p)
		} else {
			out = append(out, models.DeliverabilityDailyPoint{Date: key})
		}
	}
	return out
}

// deliverabilityByCampaign ranks campaigns by bounce+complaint in the window,
// with the per-campaign sent denominator from campaign_contact_progress.
func (r *advancedOutreachRepository) deliverabilityByCampaign(ctx context.Context, orgID uuid.UUID, from, to time.Time) []models.CampaignDeliverability {
	out := []models.CampaignDeliverability{}
	q := `
		SELECT de.campaign_id, c.name,
			COUNT(*) FILTER (WHERE de.event_type='bounce'),
			COUNT(*) FILTER (WHERE de.event_type='complaint')
		FROM deliverability_events de JOIN campaigns c ON c.id = de.campaign_id
		WHERE de.organization_id=$1 AND de.created_at >= $2 AND de.created_at <= $3 AND de.campaign_id IS NOT NULL
		GROUP BY de.campaign_id, c.name
		ORDER BY (COUNT(*) FILTER (WHERE de.event_type='bounce') + COUNT(*) FILTER (WHERE de.event_type='complaint')) DESC
		LIMIT 20`
	rows, err := r.db.Query(ctx, q, orgID, from, to)
	if err != nil {
		return out
	}
	type rec struct {
		id   uuid.UUID
		name string
		b, c int
	}
	items := []rec{}
	for rows.Next() {
		var x rec
		if rows.Scan(&x.id, &x.name, &x.b, &x.c) == nil {
			items = append(items, x)
		}
	}
	rows.Close()
	if len(items) == 0 {
		return out
	}
	sent := map[uuid.UUID]int{}
	sq := `
		SELECT ccp.campaign_id, COUNT(*)
		FROM campaign_contact_progress ccp JOIN campaigns c ON c.id = ccp.campaign_id
		WHERE c.organization_id=$1 AND ccp.sent_at IS NOT NULL AND ccp.sent_at >= $2 AND ccp.sent_at <= $3
		GROUP BY ccp.campaign_id`
	if srows, serr := r.db.Query(ctx, sq, orgID, from, to); serr == nil {
		for srows.Next() {
			var id uuid.UUID
			var n int
			if srows.Scan(&id, &n) == nil {
				sent[id] = n
			}
		}
		srows.Close()
	}
	for _, x := range items {
		s := sent[x.id]
		br := models.Rate(x.b, s)
		cr := models.Rate(x.c, s)
		out = append(out, models.CampaignDeliverability{CampaignID: x.id, Name: x.name, Sent: s, Bounces: x.b, Complaints: x.c, BounceRate: br, ComplaintRate: cr, Band: models.DeliverabilityBand(br, cr, 0)})
	}
	return out
}

// deliverabilityByMailbox ranks mailboxes by bounce+complaint in the window,
// with the per-mailbox sent denominator from completed campaign tasks.
func (r *advancedOutreachRepository) deliverabilityByMailbox(ctx context.Context, orgID uuid.UUID, from, to time.Time) []models.MailboxDeliverability {
	out := []models.MailboxDeliverability{}
	q := `
		SELECT ea.id, ea.email,
			COUNT(*) FILTER (WHERE de.event_type='bounce'),
			COUNT(*) FILTER (WHERE de.event_type='complaint')
		FROM deliverability_events de
		JOIN tasks t ON t.id = de.task_id
		JOIN email_accounts ea ON ea.id = t.email_account_id
		WHERE de.organization_id=$1 AND de.created_at >= $2 AND de.created_at <= $3 AND de.task_id IS NOT NULL
		GROUP BY ea.id, ea.email
		ORDER BY (COUNT(*) FILTER (WHERE de.event_type='bounce') + COUNT(*) FILTER (WHERE de.event_type='complaint')) DESC
		LIMIT 50`
	rows, err := r.db.Query(ctx, q, orgID, from, to)
	if err != nil {
		return out
	}
	type rec struct {
		id    uuid.UUID
		email string
		b, c  int
	}
	items := []rec{}
	for rows.Next() {
		var x rec
		if rows.Scan(&x.id, &x.email, &x.b, &x.c) == nil {
			items = append(items, x)
		}
	}
	rows.Close()
	if len(items) == 0 {
		return out
	}
	sent := map[uuid.UUID]int{}
	sq := `
		SELECT t.email_account_id, COUNT(*)
		FROM tasks t JOIN email_accounts ea ON ea.id = t.email_account_id
		WHERE ea.organization_id=$1 AND t.task_type='campaign' AND t.status='completed'
		  AND t.completed_at >= $2 AND t.completed_at <= $3
		GROUP BY t.email_account_id`
	if srows, serr := r.db.Query(ctx, sq, orgID, from, to); serr == nil {
		for srows.Next() {
			var id uuid.UUID
			var n int
			if srows.Scan(&id, &n) == nil {
				sent[id] = n
			}
		}
		srows.Close()
	}
	for _, x := range items {
		s := sent[x.id]
		br := models.Rate(x.b, s)
		cr := models.Rate(x.c, s)
		out = append(out, models.MailboxDeliverability{EmailAccountID: x.id, Email: x.email, Sent: s, Bounces: x.b, Complaints: x.c, BounceRate: br, ComplaintRate: cr, Band: models.DeliverabilityBand(br, cr, 0)})
	}
	return out
}

// executionLockTTL is the maximum time an in_progress execution lock is considered valid.
// If a lock has been in_progress longer than this, it is treated as expired (worker crash).
const executionLockTTL = 5 * time.Minute

func (r *advancedOutreachRepository) StartTaskExecution(ctx context.Context, taskID uuid.UUID, executionKey string, metadata map[string]interface{}) (bool, error) {
	var existingStatus string
	var lastSeenAt time.Time
	err := r.db.QueryRow(ctx,
		`SELECT status, last_seen_at FROM task_execution_keys WHERE task_id = $1 AND execution_key = $2`,
		taskID, executionKey,
	).Scan(&existingStatus, &lastSeenAt)
	if err == nil {
		switch existingStatus {
		case "completed":
			_, _ = r.db.Exec(ctx, `UPDATE task_execution_keys SET attempts = attempts + 1, last_seen_at = NOW() WHERE task_id = $1 AND execution_key = $2`, taskID, executionKey)
			return true, nil
		case "in_progress":
			// Check if the lock has expired (worker likely crashed)
			if time.Since(lastSeenAt) > executionLockTTL {
				// Expired lock - reclaim it
				meta, _ := marshalJSON(metadata)
				_, err := r.db.Exec(ctx, `
					UPDATE task_execution_keys
					SET attempts = attempts + 1, last_seen_at = NOW(), status = 'in_progress', metadata = $3
					WHERE task_id = $1 AND execution_key = $2
				`, taskID, executionKey, meta)
				return false, err
			}
			_, _ = r.db.Exec(ctx, `UPDATE task_execution_keys SET attempts = attempts + 1, last_seen_at = NOW() WHERE task_id = $1 AND execution_key = $2`, taskID, executionKey)
			return true, nil
		default:
			meta, _ := marshalJSON(metadata)
			_, err := r.db.Exec(ctx, `
				UPDATE task_execution_keys
				SET attempts = attempts + 1, last_seen_at = NOW(), status = 'in_progress', metadata = $3
				WHERE task_id = $1 AND execution_key = $2
			`, taskID, executionKey, meta)
			return false, err
		}
	}
	if err != pgx.ErrNoRows {
		return false, err
	}

	meta, _ := marshalJSON(metadata)
	_, err = r.db.Exec(ctx, `
		INSERT INTO task_execution_keys (task_id, execution_key, status, metadata, first_seen_at, last_seen_at, attempts)
		VALUES ($1, $2, 'in_progress', $3, NOW(), NOW(), 1)
	`, taskID, executionKey, meta)
	return false, err
}

func (r *advancedOutreachRepository) CompleteTaskExecution(ctx context.Context, taskID uuid.UUID, executionKey, status string, metadata map[string]interface{}) error {
	meta, _ := marshalJSON(metadata)
	_, err := r.db.Exec(ctx, `
		UPDATE task_execution_keys
		SET status = $3, metadata = $4, last_seen_at = NOW()
		WHERE task_id = $1 AND execution_key = $2
	`, taskID, executionKey, status, meta)
	return err
}

func (r *advancedOutreachRepository) CreateTaskDeadLetter(ctx context.Context, item *models.TaskDeadLetter) error {
	payload, err := marshalJSON(item.Payload)
	if err != nil {
		return err
	}
	query := `
		INSERT INTO task_dead_letters (task_id, task_type, payload, last_error, attempts, max_attempts, status, next_retry_at, created_at, updated_at)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW(), NOW())
	`
	if item.MaxAttempts <= 0 {
		item.MaxAttempts = 5
	}
	if item.Status == "" {
		item.Status = "pending"
	}
	_, err = r.db.Exec(ctx, query, item.TaskID, item.TaskType, payload, item.LastError, item.Attempts, item.MaxAttempts, item.Status, item.NextRetryAt)
	return err
}

func (r *advancedOutreachRepository) ListTaskDeadLetters(ctx context.Context, organizationID uuid.UUID, status string, limit int) ([]models.TaskDeadLetter, error) {
	if limit <= 0 || limit > 200 {
		limit = 100
	}
	query := `
		SELECT d.id, d.task_id, d.task_type, d.payload, d.last_error, d.attempts, d.max_attempts, d.status, d.next_retry_at, d.replayed_at, d.created_at, d.updated_at
		FROM task_dead_letters d
		JOIN tasks t ON t.id = d.task_id
		JOIN email_accounts ea ON ea.id = t.email_account_id
		WHERE ea.organization_id = $1
		  AND ($2 = '' OR d.status = $2)
		ORDER BY d.updated_at DESC
		LIMIT $3
	`
	rows, err := r.db.Query(ctx, query, organizationID, status, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	out := make([]models.TaskDeadLetter, 0, limit)
	for rows.Next() {
		var d models.TaskDeadLetter
		var payload []byte
		if err := rows.Scan(
			&d.ID,
			&d.TaskID,
			&d.TaskType,
			&payload,
			&d.LastError,
			&d.Attempts,
			&d.MaxAttempts,
			&d.Status,
			&d.NextRetryAt,
			&d.ReplayedAt,
			&d.CreatedAt,
			&d.UpdatedAt,
		); err != nil {
			return nil, err
		}
		if len(payload) > 0 {
			_ = json.Unmarshal(payload, &d.Payload)
		}
		if d.Payload == nil {
			d.Payload = map[string]interface{}{}
		}
		out = append(out, d)
	}
	return out, rows.Err()
}

func (r *advancedOutreachRepository) GetTaskDeadLetter(ctx context.Context, id uuid.UUID, organizationID uuid.UUID) (*models.TaskDeadLetter, error) {
	query := `
		SELECT d.id, d.task_id, d.task_type, d.payload, d.last_error, d.attempts, d.max_attempts, d.status, d.next_retry_at, d.replayed_at, d.created_at, d.updated_at
		FROM task_dead_letters d
		JOIN tasks t ON t.id = d.task_id
		JOIN email_accounts ea ON ea.id = t.email_account_id
		WHERE d.id = $1 AND ea.organization_id = $2
	`
	var d models.TaskDeadLetter
	var payload []byte
	if err := r.db.QueryRow(ctx, query, id, organizationID).Scan(
		&d.ID,
		&d.TaskID,
		&d.TaskType,
		&payload,
		&d.LastError,
		&d.Attempts,
		&d.MaxAttempts,
		&d.Status,
		&d.NextRetryAt,
		&d.ReplayedAt,
		&d.CreatedAt,
		&d.UpdatedAt,
	); err != nil {
		if err == pgx.ErrNoRows {
			return nil, nil
		}
		return nil, err
	}
	if len(payload) > 0 {
		_ = json.Unmarshal(payload, &d.Payload)
	}
	if d.Payload == nil {
		d.Payload = map[string]interface{}{}
	}
	return &d, nil
}

func (r *advancedOutreachRepository) MarkTaskDeadLetterReplayed(ctx context.Context, id uuid.UUID) error {
	_, err := r.db.Exec(ctx, `
		UPDATE task_dead_letters
		SET status = 'replayed', replayed_at = NOW(), updated_at = NOW()
		WHERE id = $1
	`, id)
	return err
}

func (r *advancedOutreachRepository) CreateReplyIntent(ctx context.Context, record *models.ReplyIntentRecord) error {
	metadata, err := marshalJSON(record.Metadata)
	if err != nil {
		return err
	}
	query := `
		INSERT INTO reply_intents (
			organization_id, contact_email, campaign_id, task_id, intent, confidence, action_taken, metadata, created_at
		)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW())
	`
	_, err = r.db.Exec(ctx, query,
		record.OrganizationID,
		strings.ToLower(strings.TrimSpace(record.ContactEmail)),
		record.CampaignID,
		record.TaskID,
		record.Intent,
		record.Confidence,
		record.ActionTaken,
		metadata,
	)
	return err
}

func (r *advancedOutreachRepository) CreatePreflightReport(ctx context.Context, report *models.PreflightReport) error {
	checks, err := marshalJSON(report.Checks)
	if err != nil {
		return err
	}
	recommendations, err := marshalJSON(report.Recommendations)
	if err != nil {
		return err
	}
	query := `
		INSERT INTO preflight_reports (organization_id, campaign_id, passed, score, checks, recommendations, created_at)
		VALUES ($1, $2, $3, $4, $5, $6, NOW())
	`
	_, err = r.db.Exec(ctx, query, report.OrganizationID, report.CampaignID, report.Passed, report.Score, checks, recommendations)
	return err
}

func (r *advancedOutreachRepository) GetABVariantStats(ctx context.Context, campaignID uuid.UUID) ([]models.ABVariantStats, error) {
	query := `
		SELECT
			v.id, v.name,
			COUNT(a.contact_id) AS total_sent,
			COUNT(a.opened_at) AS opened,
			COUNT(a.clicked_at) AS clicked,
			COUNT(a.replied_at) AS replied,
			COUNT(a.bounced_at) AS bounced
		FROM campaign_ab_variants v
		LEFT JOIN campaign_ab_assignments a ON a.variant_id = v.id AND a.campaign_id = v.campaign_id
		WHERE v.campaign_id = $1 AND v.is_control = false
		GROUP BY v.id, v.name
		ORDER BY v.created_at
	`
	rows, err := r.db.Query(ctx, query, campaignID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var stats []models.ABVariantStats
	for rows.Next() {
		var s models.ABVariantStats
		if err := rows.Scan(&s.VariantID, &s.VariantName, &s.TotalSent, &s.Opened, &s.Clicked, &s.Replied, &s.Bounced); err != nil {
			return nil, err
		}
		if s.TotalSent > 0 {
			s.OpenRate = float64(s.Opened) / float64(s.TotalSent) * 100
			s.ClickRate = float64(s.Clicked) / float64(s.TotalSent) * 100
			s.ReplyRate = float64(s.Replied) / float64(s.TotalSent) * 100
			s.BounceRate = float64(s.Bounced) / float64(s.TotalSent) * 100
		}
		stats = append(stats, s)
	}
	return stats, rows.Err()
}

func (r *advancedOutreachRepository) ListRetryableDeadLetters(ctx context.Context, limit int) ([]models.TaskDeadLetter, error) {
	if limit <= 0 || limit > 50 {
		limit = 10
	}
	query := `
		SELECT id, task_id, task_type, payload, last_error, attempts, max_attempts, status, next_retry_at, replayed_at, created_at, updated_at
		FROM task_dead_letters
		WHERE status = 'pending'
		  AND next_retry_at IS NOT NULL
		  AND next_retry_at <= NOW()
		  AND attempts < max_attempts
		ORDER BY next_retry_at ASC
		LIMIT $1
	`
	rows, err := r.db.Query(ctx, query, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var items []models.TaskDeadLetter
	for rows.Next() {
		var d models.TaskDeadLetter
		var payload []byte
		if err := rows.Scan(&d.ID, &d.TaskID, &d.TaskType, &payload, &d.LastError, &d.Attempts, &d.MaxAttempts, &d.Status, &d.NextRetryAt, &d.ReplayedAt, &d.CreatedAt, &d.UpdatedAt); err != nil {
			return nil, err
		}
		if len(payload) > 0 {
			_ = json.Unmarshal(payload, &d.Payload)
		}
		items = append(items, d)
	}
	return items, rows.Err()
}

func (r *advancedOutreachRepository) IncrementDeadLetterAttempt(ctx context.Context, id uuid.UUID, nextRetryAt *time.Time) error {
	_, err := r.db.Exec(ctx, `
		UPDATE task_dead_letters
		SET attempts = attempts + 1, next_retry_at = $2, updated_at = NOW()
		WHERE id = $1
	`, id, nextRetryAt)
	return err
}
