import 'server-only'

import { createAdminClient } from '@/lib/supabase/admin'
import { sendPlatformEmail } from '@/lib/email/send'
import { createJoinToken } from '@/lib/email/tokens'
import { SITE_URL } from '@/lib/app-url'
import { DEFAULT_TEMPLATES } from '@/lib/email-templates'
import { resolveTemplateMeta } from '@/lib/domain/email-policy'
import { logLeadEvent, type LeadEventType } from '@/lib/data/broker-lead-hooks'
import {
  campaignReadiness, campaignTransition, dailyAllowance, frequencyCapOk, frequencyCapReleaseAt, leadKindConfig, leadMatchesFilters, leadSendBlock, leadStage, leadTemplateData,
  nextWindowStart, normalizeRamp, normalizeSteps, normalizeWindow, recipientStopReason, sendingDayIndex, stepDueAt, timezoneForLead, windowOpen,
  type CampaignStatus, type CampaignStep, type LeadKind, type LeadStage, type Ramp, type SegmentFilters, type SendWindow, LEAD_STAGES,
} from '@/lib/domain/broker-leads'
import type { BrokerAgentLead } from '@/types/db'

/**
 * Freight Broker Leads — the I/O (Nash, 2026-09-23). Rules live in
 * src/lib/domain/broker-leads.ts; the small hooks other modules call live in
 * broker-lead-hooks.ts. Needs migration 0030; `brokerLeadsAvailable` says so.
 *
 * Scope reminder: a lead is a freight broker and the brokerage they are
 * pre-verified for. Nothing here touches carriers.
 */

type Admin = ReturnType<typeof createAdminClient>
const missing = (e: { message?: string } | null | undefined) => !!e && /does not exist|schema cache|column/i.test(e.message ?? '')
/** One engine, two kinds (2026-09-25): every read and write is scoped by `kind`; 'broker' keeps every existing call site unchanged. */
const campaignTypeFor = (kind: LeadKind) => leadKindConfig(kind).campaignType
const kindOfCampaignType = (t: string | null | undefined): LeadKind => (t === 'carrier_recruitment' ? 'carrier' : 'broker')
const PAGE_SIZE = 50

/* ------------------------------------------------------------------ types */

export interface LeadRow extends BrokerAgentLead {
  state: string | null
  mailbox_type: 'person' | 'role'
  email_domain_check: 'matches_brokerage' | 'differs_from_brokerage' | 'brokerage_email_unknown' | null
  source_system: string | null
  source_last_seen_at: string | null
  tags: string[]
  notes: string | null
  suppressed_at: string | null
  suppressed_reason: string | null
  suppressed_by: string | null
  emails_sent: number
  last_emailed_at: string | null
  last_template_key: string | null
  last_delivered_at: string | null
  last_opened_at: string | null
  last_clicked_at: string | null
  bounced_at: string | null
  unsubscribed_at: string | null
  reminded_at: string | null
  signed_up_at: string | null
  signed_up_user_id: string | null
  join_token_issued_at: string | null
  /** Joined in: the brokerage's display name and state (from companies). For a carrier lead, the free-text company name. */
  brokerage_name: string | null
  brokerage_state: string | null
  stage: LeadStage
  lead_kind: LeadKind
  lead_type: 'dispatcher' | 'driver' | 'both' | null
  company_name: string | null
  source_record_id: string | null
  source_status_active: boolean | null
  source_created_at: string | null
}

export interface LeadEvent { id: string; lead_id: string; event_type: LeadEventType; campaign_id: string | null; message_id: string | null; template_key: string | null; step: number | null; actor_label: string | null; detail: Record<string, unknown> | null; created_at: string }
export interface LeadMessage { id: string; template_key: string | null; subject: string | null; status: string; campaign_id: string | null; sent_at: string | null; delivered_at: string | null; opened_at: string | null; clicked_at: string | null; bounce_reason: string | null; suppressed_reason: string | null; error: string | null; created_at: string }
export interface LeadSegment { id: string; name: string; description: string | null; filters: SegmentFilters; created_by: string | null; created_at: string; updated_at: string }
export interface BrokerageRollup { id: string; name: string; mc_number: string | null; state: string | null; verification_status: string | null; total: number; stages: Partial<Record<LeadStage, number>> }

export interface LeadCampaign {
  id: string
  kind: LeadKind
  name: string
  description: string | null
  status: CampaignStatus
  steps: CampaignStep[]
  audience_kind: 'segment' | 'leads'
  audience: SegmentFilters | { lead_ids: string[] }
  audience_count: number | null
  send_window: SendWindow
  batch_size: number
  ramp: Ramp | null
  frequency_cap_days: number
  daily_send_limit: number
  start_date: string | null
  end_date: string | null
  sent_count: number; delivered_count: number; bounce_count: number; hard_bounce_count: number; soft_bounce_count: number; unsubscribe_count: number; complaint_count: number; open_count: number; click_count: number
  conversion_count: number; signed_up_count: number; enrolled_count: number; done_count: number; stopped_count: number
  first_run_date: string | null
  last_run_at: string | null
  last_run_summary: Record<string, unknown> | null
  pause_reason: string | null
  paused_at: string | null
  approved_by: string | null
  approved_at: string | null
  created_by: string | null
  created_at: string
  updated_at: string
  test_sent_at: string | null
  /** Live counts from the recipients table. */
  recipients: { pending: number; scheduled: number; sent: number; skipped: number; failed: number; stopped: number; done: number; converted: number }
}

export interface LeadsSettings {
  cron_enabled: boolean
  default_daily_limit: number
  default_batch_size: number
  default_send_window: SendWindow
  default_ramp: Ramp | null
  default_frequency_cap_days: number
  updated_by: string | null
  updated_at: string | null
}

export interface LeadTemplate { key: string; name: string; subject: string; active: boolean; version: number; stream: string; category: string; is_custom: boolean }

export interface Actor { id: string | null; label: string }

/* ------------------------------------------------------------ available */

export async function brokerLeadsAvailable(admin: Admin = createAdminClient(), kind: LeadKind = 'broker'): Promise<boolean> {
  const { error } = await admin.from('broker_agent_leads').select(kind === 'carrier' ? 'signed_up_at, lead_kind, lead_type' : 'signed_up_at').limit(1)
  if (error) return false
  const s = await admin.from('broker_leads_settings').select('id').limit(1)
  return !s.error
}

/* -------------------------------------------------------------- settings */

const DEFAULT_SETTINGS: LeadsSettings = { cron_enabled: false, default_daily_limit: 100, default_batch_size: 25, default_send_window: normalizeWindow(null), default_ramp: { start: 50, step: 50, max: 300 }, default_frequency_cap_days: 7, updated_by: null, updated_at: null }

export async function loadLeadsSettings(admin: Admin = createAdminClient(), kind: LeadKind = 'broker'): Promise<LeadsSettings> {
  const { data, error } = await admin.from('broker_leads_settings').select('*').eq('id', leadKindConfig(kind).settingsId).maybeSingle()
  if (error || !data) return DEFAULT_SETTINGS
  const d = data as Record<string, unknown>
  return {
    cron_enabled: !!d.cron_enabled, default_daily_limit: Number(d.default_daily_limit ?? 100), default_batch_size: Number(d.default_batch_size ?? 25),
    default_send_window: normalizeWindow(d.default_send_window as Partial<SendWindow>), default_ramp: normalizeRamp(d.default_ramp as Partial<Ramp>),
    default_frequency_cap_days: Number(d.default_frequency_cap_days ?? 7), updated_by: (d.updated_by as string | null) ?? null, updated_at: (d.updated_at as string | null) ?? null,
  }
}

export async function saveLeadsSettings(patch: Partial<LeadsSettings>, actor: Actor, admin: Admin = createAdminClient(), kind: LeadKind = 'broker') {
  const row: Record<string, unknown> = { id: leadKindConfig(kind).settingsId, ...(kind === 'carrier' ? { lead_kind: 'carrier' } : {}), updated_by: actor.label, updated_at: new Date().toISOString() }
  if (patch.cron_enabled !== undefined) row.cron_enabled = !!patch.cron_enabled
  if (patch.default_daily_limit !== undefined) row.default_daily_limit = Math.max(1, Number(patch.default_daily_limit))
  if (patch.default_batch_size !== undefined) row.default_batch_size = Math.max(1, Number(patch.default_batch_size))
  if (patch.default_send_window !== undefined) row.default_send_window = normalizeWindow(patch.default_send_window)
  if (patch.default_ramp !== undefined) row.default_ramp = normalizeRamp(patch.default_ramp)
  if (patch.default_frequency_cap_days !== undefined) row.default_frequency_cap_days = Math.max(0, Number(patch.default_frequency_cap_days))
  const { error } = await admin.from('broker_leads_settings').upsert(row, { onConflict: 'id' })
  return error ? { ok: false as const, error: error.message } : { ok: true as const }
}

/* ----------------------------------------------------------------- leads */

const LEAD_COLUMNS = 'id, first_name, last_name, email, email_normalized, phone, phone_digits, broker_company_id, broker_company_name, mc_number, usdot_number, historical_relationship_verified, historical_verification_source, historical_verified_at, claim_status, relationship_confirmation_status, invitation_status, invited_at, claimed_user_id, claimed_at, confirmed_at, manual_review_reason, import_batch, created_at, updated_at, state, mailbox_type, email_domain_check, source_system, source_last_seen_at, tags, notes, suppressed_at, suppressed_reason, suppressed_by, emails_sent, last_emailed_at, last_template_key, last_delivered_at, last_opened_at, last_clicked_at, bounced_at, unsubscribed_at, reminded_at, signed_up_at, signed_up_user_id, join_token_issued_at, lead_kind, lead_type, company_name, source_record_id, source_status_active, source_created_at'
/** Before migration 0039 the kind columns do not exist; broker reads fall back to this. */
const LEAD_COLUMNS_0030 = LEAD_COLUMNS.replace(', lead_kind, lead_type, company_name, source_record_id, source_status_active, source_created_at', '')

async function fetchAll<T>(build: (from: number, to: number) => PromiseLike<{ data: unknown; error: { message?: string } | null }>): Promise<T[]> {
  const out: T[] = []
  const CHUNK = 1000
  for (let from = 0; ; from += CHUNK) {
    const { data, error } = await build(from, from + CHUNK - 1)
    if (error) throw new Error(error.message)
    const rows = (data ?? []) as T[]
    out.push(...rows)
    if (rows.length < CHUNK) break
  }
  return out
}

let companyCache: { at: number; map: Map<string, { name: string; mc: string | null; state: string | null; verification_status: string | null }> } | null = null
async function brokerCompanies(admin: Admin, fresh = false) {
  if (!fresh && companyCache && Date.now() - companyCache.at < 60_000) return companyCache.map
  const rows = await fetchAll<{ id: string; display_name: string; mc_number: string | null; physical_address: string | null; verification_status: string | null }>((from, to) =>
    admin.from('companies').select('id, display_name, mc_number, physical_address, verification_status').eq('company_type', 'broker').is('merged_into', null).order('display_name').range(from, to))
  const map = new Map(rows.map((c) => [c.id, { name: c.display_name, mc: c.mc_number, state: stateFromAddress(c.physical_address), verification_status: c.verification_status }]))
  companyCache = { at: Date.now(), map }
  return map
}

/** "600 E WILLIAM ST SUITE 206, CARSON CITY, NV 89701" → "NV". */
export function stateFromAddress(addr: string | null | undefined): string | null {
  const m = (addr ?? '').match(/,\s*([A-Z]{2})\s+\d{5}(?:-\d{4})?\s*$/i)
  return m ? m[1].toUpperCase() : null
}

function decorate(l: Record<string, unknown>, companies: Map<string, { name: string; mc: string | null; state: string | null }>): LeadRow {
  const c = l.broker_company_id ? companies.get(l.broker_company_id as string) : null
  const kind: LeadKind = l.lead_kind === 'carrier' ? 'carrier' : 'broker'
  const row = {
    ...l, tags: Array.isArray(l.tags) ? (l.tags as string[]) : [], emails_sent: Number(l.emails_sent ?? 0), mailbox_type: (l.mailbox_type as 'person' | 'role') ?? 'person',
    lead_kind: kind, lead_type: (l.lead_type as LeadRow['lead_type']) ?? null, company_name: (l.company_name as string | null) ?? null,
    source_record_id: (l.source_record_id as string | null) ?? null, source_status_active: (l.source_status_active as boolean | null) ?? null, source_created_at: (l.source_created_at as string | null) ?? null,
    brokerage_name: kind === 'carrier' ? ((l.company_name as string | null) ?? null) : (c?.name ?? (l.broker_company_name as string | null) ?? null), brokerage_state: c?.state ?? null,
    state: (l.state as string | null) ?? c?.state ?? null,
  } as unknown as LeadRow
  row.stage = leadStage(row)
  return row
}

/** Every lead of one kind (a few thousand rows) with the brokerage joined in. The page and the roll-ups work on this. */
export async function loadAllLeads(admin: Admin = createAdminClient(), kind: LeadKind = 'broker'): Promise<LeadRow[]> {
  const companies = await brokerCompanies(admin)
  let rows: Record<string, unknown>[]
  try {
    rows = await fetchAll<Record<string, unknown>>((from, to) => admin.from('broker_agent_leads').select(LEAD_COLUMNS).eq('lead_kind', kind).order('created_at', { ascending: true }).range(from, to))
  } catch (e) {
    if (!/does not exist|schema cache|column/i.test(String(e))) throw e
    if (kind === 'carrier') return [] // 0039 not applied: there are no carrier leads yet
    try {
      rows = await fetchAll<Record<string, unknown>>((from, to) => admin.from('broker_agent_leads').select(LEAD_COLUMNS_0030).order('created_at', { ascending: true }).range(from, to))
    } catch {
      rows = await fetchAll<Record<string, unknown>>((from, to) => admin.from('broker_agent_leads').select('*').order('created_at', { ascending: true }).range(from, to))
    }
  }
  return rows.map((r) => decorate(r, companies))
}

export function filterLeads(all: LeadRow[], f: SegmentFilters): LeadRow[] {
  return all.filter((l) => leadMatchesFilters(l, f))
}

export interface LeadsPage { rows: LeadRow[]; total: number; page: number; pageSize: number; pages: number }

export function pageLeads(matching: LeadRow[], page: number, sort: 'activity' | 'name' | 'brokerage' = 'brokerage'): LeadsPage {
  const sorted = [...matching].sort((a, b) => {
    if (sort === 'activity') return (b.last_emailed_at ?? b.updated_at ?? '').localeCompare(a.last_emailed_at ?? a.updated_at ?? '')
    if (sort === 'name') return (a.last_name ?? '').localeCompare(b.last_name ?? '') || (a.first_name ?? '').localeCompare(b.first_name ?? '')
    return (a.brokerage_name ?? '').localeCompare(b.brokerage_name ?? '') || (a.last_name ?? '').localeCompare(b.last_name ?? '') || (a.first_name ?? '').localeCompare(b.first_name ?? '')
  })
  const pages = Math.max(1, Math.ceil(sorted.length / PAGE_SIZE))
  const p = Math.min(Math.max(1, page), pages)
  return { rows: sorted.slice((p - 1) * PAGE_SIZE, p * PAGE_SIZE), total: sorted.length, page: p, pageSize: PAGE_SIZE, pages }
}

export interface LeadsSummary { total: number; stages: Record<LeadStage, number>; brokerages: number; states: Array<[string, number]>; tags: Array<[string, number]>; sources: Array<[string, number]>; emails_sent: number; opened: number; clicked: number }

export function summarizeLeads(all: LeadRow[]): LeadsSummary {
  const stages = Object.fromEntries(LEAD_STAGES.map((s) => [s, 0])) as Record<LeadStage, number>
  const states = new Map<string, number>(), tags = new Map<string, number>(), sources = new Map<string, number>(), brokerages = new Set<string>()
  let emails = 0, opened = 0, clicked = 0
  for (const l of all) {
    stages[l.stage]++
    if (l.state) states.set(l.state, (states.get(l.state) ?? 0) + 1)
    for (const t of l.tags) tags.set(t, (tags.get(t) ?? 0) + 1)
    const src = l.source_system ?? l.import_batch ?? 'unknown'
    sources.set(src, (sources.get(src) ?? 0) + 1)
    if (l.broker_company_id) brokerages.add(l.broker_company_id)
    else if (l.company_name) brokerages.add(l.company_name)
    emails += l.emails_sent
    if (l.last_opened_at) opened++
    if (l.last_clicked_at) clicked++
  }
  const top = (m: Map<string, number>) => [...m.entries()].sort((a, b) => b[1] - a[1])
  return { total: all.length, stages, brokerages: brokerages.size, states: top(states), tags: top(tags), sources: top(sources), emails_sent: emails, opened, clicked }
}

export function brokerageRollups(all: LeadRow[], companies: Map<string, { name: string; mc: string | null; state: string | null; verification_status: string | null }>, kind: LeadKind = 'broker'): BrokerageRollup[] {
  const by = new Map<string, BrokerageRollup>()
  if (kind === 'broker') for (const [id, c] of companies) by.set(id, { id, name: c.name, mc_number: c.mc, state: c.state, verification_status: c.verification_status, total: 0, stages: {} })
  for (const l of all) {
    // Carrier leads have no company row: the free-text company name is the key.
    const key = l.lead_kind === 'carrier' ? (l.company_name ?? '') : (l.broker_company_id ?? '')
    if (!key) continue
    const r = by.get(key) ?? { id: key, name: l.brokerage_name ?? key, mc_number: l.mc_number, state: l.state, verification_status: null, total: 0, stages: {} }
    r.total++
    r.stages[l.stage] = (r.stages[l.stage] ?? 0) + 1
    by.set(key, r)
  }
  return [...by.values()].sort((a, b) => b.total - a.total || a.name.localeCompare(b.name))
}

export async function loadBrokerCompanyMap(admin: Admin = createAdminClient()) {
  return brokerCompanies(admin, true)
}

/* ------------------------------------------------------------ one lead */

export interface LeadDetail { lead: LeadRow; events: LeadEvent[]; messages: LeadMessage[]; campaigns: Array<{ campaign_id: string; name: string; status: string; step: number; recipient_status: string; scheduled_for: string | null; last_sent_at: string | null; sends: number; stop_reason: string | null; skip_reason: string | null }> }

export async function loadLead(id: string, admin: Admin = createAdminClient()): Promise<LeadDetail | null> {
  const companies = await brokerCompanies(admin)
  const q1 = await admin.from('broker_agent_leads').select(LEAD_COLUMNS).eq('id', id).maybeSingle()
  const q2 = q1.error && missing(q1.error) ? await admin.from('broker_agent_leads').select(LEAD_COLUMNS_0030).eq('id', id).maybeSingle() : null
  const l = ((q2 ?? q1).data ?? null) as Record<string, unknown> | null
  if (!l) return null
  const lead = decorate(l as Record<string, unknown>, companies)
  const [ev, msgs, recs] = await Promise.all([
    admin.from('broker_lead_events').select('*').eq('lead_id', id).order('created_at', { ascending: false }).limit(200),
    admin.from('email_messages').select('id, template_key, subject, status, campaign_id, sent_at, delivered_at, opened_at, clicked_at, bounce_reason, suppressed_reason, error, created_at').eq('email_normalized', lead.email_normalized).order('created_at', { ascending: false }).limit(50),
    admin.from('email_campaign_recipients').select('campaign_id, step, status, scheduled_for, last_sent_at, sends, stop_reason, skip_reason').eq('lead_id', id),
  ])
  const recRows = (recs.data ?? []) as Array<{ campaign_id: string; step: number; status: string; scheduled_for: string | null; last_sent_at: string | null; sends: number; stop_reason: string | null; skip_reason: string | null }>
  const campaignIds = [...new Set(recRows.map((r) => r.campaign_id))]
  const { data: cs } = campaignIds.length ? await admin.from('email_campaigns').select('id, name, status').in('id', campaignIds) : { data: [] }
  const cmap = new Map(((cs ?? []) as Array<{ id: string; name: string; status: string }>).map((c) => [c.id, c]))
  return {
    lead,
    events: (ev.data ?? []) as LeadEvent[],
    messages: (msgs.data ?? []) as LeadMessage[],
    campaigns: recRows.map((r) => ({ campaign_id: r.campaign_id, name: cmap.get(r.campaign_id)?.name ?? '?', status: cmap.get(r.campaign_id)?.status ?? '?', step: r.step, recipient_status: r.status, scheduled_for: r.scheduled_for, last_sent_at: r.last_sent_at, sends: r.sends, stop_reason: r.stop_reason, skip_reason: r.skip_reason })),
  }
}

export async function updateLead(id: string, patch: { first_name?: string; last_name?: string; phone?: string; state?: string; notes?: string; tags?: string[] }, actor: Actor, admin: Admin = createAdminClient()) {
  const { data: before } = await admin.from('broker_agent_leads').select('first_name, last_name, phone, state, notes, tags').eq('id', id).maybeSingle()
  if (!before) return { ok: false as const, error: 'Lead not found.' }
  const row: Record<string, unknown> = { updated_at: new Date().toISOString() }
  if (patch.first_name !== undefined) row.first_name = patch.first_name.trim() || null
  if (patch.last_name !== undefined) row.last_name = patch.last_name.trim() || null
  if (patch.phone !== undefined) { row.phone = patch.phone.trim() || null; const d = patch.phone.replace(/\D/g, ''); row.phone_digits = d.length >= 10 ? (d.length === 11 && d.startsWith('1') ? d.slice(1) : d) : null }
  if (patch.state !== undefined) row.state = patch.state.trim().toUpperCase() || null
  if (patch.notes !== undefined) row.notes = patch.notes.trim() || null
  if (patch.tags !== undefined) row.tags = [...new Set(patch.tags.map((t) => t.trim().toLowerCase()).filter(Boolean))]
  const { error } = await admin.from('broker_agent_leads').update(row).eq('id', id)
  if (error) return { ok: false as const, error: error.message }
  const b = before as Record<string, unknown>
  if (patch.notes !== undefined && patch.notes.trim() !== ((b.notes as string | null) ?? '')) await logLeadEvent({ leadId: id, type: 'note', actorLabel: actor.label, detail: { notes: patch.notes.trim() } }, admin)
  if (patch.tags !== undefined) {
    const was = new Set((b.tags as string[]) ?? []), now = new Set(row.tags as string[])
    const added = [...now].filter((t) => !was.has(t)), removed = [...was].filter((t) => !now.has(t))
    if (added.length) await logLeadEvent({ leadId: id, type: 'tagged', actorLabel: actor.label, detail: { tags: added } }, admin)
    if (removed.length) await logLeadEvent({ leadId: id, type: 'untagged', actorLabel: actor.label, detail: { tags: removed } }, admin)
  }
  const changed = ['first_name', 'last_name', 'phone', 'state'].filter((k) => k in row && row[k] !== (b[k] ?? null))
  if (changed.length) await logLeadEvent({ leadId: id, type: 'edited', actorLabel: actor.label, detail: { fields: changed, before: Object.fromEntries(changed.map((k) => [k, b[k] ?? null])), after: Object.fromEntries(changed.map((k) => [k, row[k]])) } }, admin)
  return { ok: true as const }
}

/** Bulk tag: add or remove one tag on many leads. */
export async function tagLeads(ids: string[], tag: string, action: 'add' | 'remove', actor: Actor, admin: Admin = createAdminClient()) {
  const t = tag.trim().toLowerCase()
  if (!t) return { ok: false as const, error: 'Enter a tag.' }
  const { data } = await admin.from('broker_agent_leads').select('id, tags').in('id', ids)
  let n = 0
  for (const l of (data ?? []) as Array<{ id: string; tags: string[] | null }>) {
    const tags = new Set(l.tags ?? [])
    if (action === 'add' ? tags.has(t) : !tags.has(t)) continue
    if (action === 'add') tags.add(t); else tags.delete(t)
    await admin.from('broker_agent_leads').update({ tags: [...tags], updated_at: new Date().toISOString() }).eq('id', l.id)
    await logLeadEvent({ leadId: l.id, type: action === 'add' ? 'tagged' : 'untagged', actorLabel: actor.label, detail: { tags: [t] } }, admin)
    n++
  }
  return { ok: true as const, changed: n }
}

export async function suppressLead(id: string, reason: string, actor: Actor, admin: Admin = createAdminClient()) {
  const { data: l } = await admin.from('broker_agent_leads').select('id, email_normalized').eq('id', id).maybeSingle()
  if (!l) return { ok: false as const, error: 'Lead not found.' }
  const now = new Date().toISOString()
  const note = reason.trim() || 'admin_block'
  const { error } = await admin.from('broker_agent_leads').update({ suppressed_at: now, suppressed_reason: note, suppressed_by: actor.label, updated_at: now }).eq('id', id)
  if (error) return { ok: false as const, error: error.message }
  await admin.from('email_suppressions').upsert({ email_normalized: (l as { email_normalized: string }).email_normalized, scope: 'marketing', campaign_type: null, reason: 'admin_block', source: `broker leads manager · ${actor.label}`, detail: { note } }, { onConflict: 'email_normalized,scope,campaign_type', ignoreDuplicates: true })
  await admin.from('email_campaign_recipients').update({ status: 'stopped', stop_reason: 'suppressed', updated_at: now }).eq('lead_id', id).in('status', ['pending', 'scheduled'])
  await logLeadEvent({ leadId: id, type: 'suppressed', actorLabel: actor.label, detail: { reason: note } }, admin)
  return { ok: true as const }
}

export async function unsuppressLead(id: string, actor: Actor, admin: Admin = createAdminClient()) {
  const { data: l } = await admin.from('broker_agent_leads').select('id, email_normalized').eq('id', id).maybeSingle()
  if (!l) return { ok: false as const, error: 'Lead not found.' }
  const now = new Date().toISOString()
  const { error } = await admin.from('broker_agent_leads').update({ suppressed_at: null, suppressed_reason: null, suppressed_by: null, updated_at: now }).eq('id', id)
  if (error) return { ok: false as const, error: error.message }
  await admin.from('email_suppressions').update({ overridden_at: now, manually_overridden_by: actor.label }).eq('email_normalized', (l as { email_normalized: string }).email_normalized).eq('scope', 'marketing').eq('reason', 'admin_block').is('overridden_at', null)
  await logLeadEvent({ leadId: id, type: 'unsuppressed', actorLabel: actor.label }, admin)
  return { ok: true as const }
}

/** Manual review queue: accept (back to unclaimed) or drop (suppress). */
export async function resolveReview(id: string, action: 'accept' | 'drop', actor: Actor, admin: Admin = createAdminClient()) {
  const now = new Date().toISOString()
  if (action === 'drop') {
    const r = await suppressLead(id, 'manual review: dropped', actor, admin)
    if (!r.ok) return r
  }
  const { error } = await admin.from('broker_agent_leads').update({ claim_status: 'unclaimed', manual_review_reason: null, updated_at: now }).eq('id', id).eq('claim_status', 'manual_review')
  if (error) return { ok: false as const, error: error.message }
  await logLeadEvent({ leadId: id, type: 'review_resolved', actorLabel: actor.label, detail: { action } }, admin)
  return { ok: true as const }
}

/* ------------------------------------------------------------- segments */

export async function loadSegments(admin: Admin = createAdminClient(), kind: LeadKind = 'broker'): Promise<LeadSegment[]> {
  let { data, error } = await admin.from('broker_lead_segments').select('*').eq('lead_kind', kind).order('name')
  if (error && missing(error) && kind === 'broker') ({ data, error } = await admin.from('broker_lead_segments').select('*').order('name'))
  if (error) return []
  return ((data ?? []) as Array<Record<string, unknown>>).map((s) => ({ id: s.id as string, name: s.name as string, description: (s.description as string | null) ?? null, filters: (s.filters as SegmentFilters) ?? {}, created_by: (s.created_by as string | null) ?? null, created_at: s.created_at as string, updated_at: s.updated_at as string }))
}

export async function saveSegment(input: { id?: string | null; name: string; description?: string | null; filters: SegmentFilters }, actor: Actor, admin: Admin = createAdminClient(), kind: LeadKind = 'broker') {
  const name = input.name.trim()
  if (!name) return { ok: false as const, error: 'Give the segment a name.' }
  const row = { name, description: input.description?.trim() || null, filters: input.filters, updated_at: new Date().toISOString(), ...(kind === 'carrier' ? { lead_kind: 'carrier' } : {}) }
  if (input.id) {
    const { error } = await admin.from('broker_lead_segments').update(row).eq('id', input.id)
    return error ? { ok: false as const, error: error.message } : { ok: true as const, id: input.id }
  }
  const { data, error } = await admin.from('broker_lead_segments').insert({ ...row, created_by: actor.label }).select('id').single()
  return error ? { ok: false as const, error: error.message } : { ok: true as const, id: (data as { id: string }).id }
}

export async function deleteSegment(id: string, admin: Admin = createAdminClient()) {
  const { error } = await admin.from('broker_lead_segments').delete().eq('id', id)
  return error ? { ok: false as const, error: error.message } : { ok: true as const }
}

/* ------------------------------------------------------------ templates */

/** Templates a lead campaign may use: the updates stream, the kind's invitations category. */
export async function loadLeadTemplates(admin: Admin = createAdminClient(), kind: LeadKind = 'broker'): Promise<LeadTemplate[]> {
  const category = leadKindConfig(kind).category
  const { data } = await admin.from('email_templates').select('key, name, subject, active, version, email_stream, category_key, category')
  const saved = new Map(((data ?? []) as Array<Record<string, unknown>>).map((t) => [t.key as string, t]))
  const out: LeadTemplate[] = []
  const seen = new Set<string>()
  for (const d of DEFAULT_TEMPLATES) {
    const o = saved.get(d.key)
    const meta = resolveTemplateMeta(d.key, o as { email_stream?: string | null; category_key?: string | null } | undefined)
    if (meta.category !== category) continue
    seen.add(d.key)
    out.push({ key: d.key, name: (o?.name as string) ?? d.name, subject: (o?.subject as string) ?? d.subject, active: o ? !!o.active : true, version: Number(o?.version ?? 0), stream: meta.stream, category: meta.category, is_custom: false })
  }
  for (const [key, o] of saved) {
    if (seen.has(key) || o.category === 'pilot_invitation') continue
    const meta = resolveTemplateMeta(key, o as { email_stream?: string | null; category_key?: string | null })
    if (meta.category !== category) continue
    out.push({ key, name: o.name as string, subject: o.subject as string, active: !!o.active, version: Number(o.version ?? 1), stream: meta.stream, category: meta.category, is_custom: true })
  }
  return out
}

/* ------------------------------------------------------------ campaigns */

function rowToCampaign(r: Record<string, unknown>, recipients: LeadCampaign['recipients']): LeadCampaign {
  const n = (k: string) => Number(r[k] ?? 0)
  return {
    id: r.id as string, kind: kindOfCampaignType(r.campaign_type as string), name: r.name as string, description: (r.description as string | null) ?? null, status: r.status as CampaignStatus,
    steps: normalizeSteps(r.steps), audience_kind: ((r.audience_kind as string) === 'leads' ? 'leads' : 'segment'), audience: (r.audience as LeadCampaign['audience']) ?? {},
    audience_count: r.audience_count == null ? null : Number(r.audience_count), send_window: normalizeWindow(r.send_window as Partial<SendWindow>), batch_size: Math.max(1, n('batch_size') || 25),
    ramp: normalizeRamp(r.ramp as Partial<Ramp>), frequency_cap_days: n('frequency_cap_days'), daily_send_limit: n('daily_send_limit'), start_date: (r.start_date as string | null) ?? null, end_date: (r.end_date as string | null) ?? null,
    sent_count: n('sent_count'), delivered_count: n('delivered_count'), bounce_count: n('bounce_count'), hard_bounce_count: n('hard_bounce_count'), soft_bounce_count: n('soft_bounce_count'), unsubscribe_count: n('unsubscribe_count'), complaint_count: n('complaint_count'), open_count: n('open_count'), click_count: n('click_count'),
    conversion_count: n('conversion_count'), signed_up_count: n('signed_up_count'), enrolled_count: n('enrolled_count'), done_count: n('done_count'), stopped_count: n('stopped_count'),
    first_run_date: (r.first_run_date as string | null) ?? null, last_run_at: (r.last_run_at as string | null) ?? null, last_run_summary: (r.last_run_summary as Record<string, unknown> | null) ?? null,
    pause_reason: (r.pause_reason as string | null) ?? null, paused_at: (r.paused_at as string | null) ?? null, approved_by: (r.approved_by as string | null) ?? null, approved_at: (r.approved_at as string | null) ?? null,
    created_by: (r.created_by as string | null) ?? null, created_at: r.created_at as string, updated_at: r.updated_at as string, test_sent_at: (r.test_sent_at as string | null) ?? null, recipients,
  }
}

async function recipientCounts(ids: string[], admin: Admin): Promise<Map<string, LeadCampaign['recipients']>> {
  const out = new Map<string, LeadCampaign['recipients']>()
  const blank = (): LeadCampaign['recipients'] => ({ pending: 0, scheduled: 0, sent: 0, skipped: 0, failed: 0, stopped: 0, done: 0, converted: 0 })
  for (const id of ids) out.set(id, blank())
  if (!ids.length) return out
  const rows = await fetchAll<{ campaign_id: string; status: string }>((from, to) => admin.from('email_campaign_recipients').select('campaign_id, status').in('campaign_id', ids).range(from, to))
  for (const r of rows) {
    const c = out.get(r.campaign_id) ?? blank()
    if (r.status in c) (c as unknown as Record<string, number>)[r.status]++
    out.set(r.campaign_id, c)
  }
  return out
}

export async function loadCampaigns(admin: Admin = createAdminClient(), kind: LeadKind = 'broker'): Promise<LeadCampaign[]> {
  const { data, error } = await admin.from('email_campaigns').select('*').eq('campaign_type', campaignTypeFor(kind)).order('created_at', { ascending: false })
  if (error) return []
  const rows = (data ?? []) as Array<Record<string, unknown>>
  const counts = await recipientCounts(rows.map((r) => r.id as string), admin)
  return rows.map((r) => rowToCampaign(r, counts.get(r.id as string)!))
}

export async function loadCampaign(id: string, admin: Admin = createAdminClient()): Promise<LeadCampaign | null> {
  const { data } = await admin.from('email_campaigns').select('*').eq('id', id).in('campaign_type', ['broker_recruitment', 'carrier_recruitment']).maybeSingle()
  if (!data) return null
  const counts = await recipientCounts([id], admin)
  return rowToCampaign(data as Record<string, unknown>, counts.get(id)!)
}

export interface CampaignInput {
  name: string
  description?: string | null
  steps: Array<{ template_key: string; delay_days: number }>
  audience_kind: 'segment' | 'leads'
  audience: SegmentFilters | { lead_ids: string[] }
  send_window: Partial<SendWindow>
  batch_size: number
  ramp: Partial<Ramp> | null
  frequency_cap_days: number
  daily_send_limit: number
  start_date: string | null
  end_date: string | null
}

function campaignRow(input: Partial<CampaignInput>): Record<string, unknown> {
  const row: Record<string, unknown> = {}
  if (input.name !== undefined) row.name = input.name.trim()
  if (input.description !== undefined) row.description = input.description?.trim() || null
  if (input.steps !== undefined) { const steps = normalizeSteps(input.steps); row.steps = steps; row.template_key = steps[0]?.template_key ?? null }
  if (input.audience_kind !== undefined) row.audience_kind = input.audience_kind
  if (input.audience !== undefined) row.audience = input.audience
  if (input.send_window !== undefined) row.send_window = normalizeWindow(input.send_window)
  if (input.batch_size !== undefined) row.batch_size = Math.max(1, Math.min(500, Number(input.batch_size) || 25))
  if (input.ramp !== undefined) row.ramp = normalizeRamp(input.ramp)
  if (input.frequency_cap_days !== undefined) row.frequency_cap_days = Math.max(0, Number(input.frequency_cap_days) || 0)
  if (input.daily_send_limit !== undefined) row.daily_send_limit = Math.max(1, Math.min(5000, Number(input.daily_send_limit) || 100))
  if (input.start_date !== undefined) row.start_date = input.start_date || null
  if (input.end_date !== undefined) row.end_date = input.end_date || null
  return row
}

export async function createCampaign(input: CampaignInput, actor: Actor, admin: Admin = createAdminClient(), kind: LeadKind = 'broker') {
  if (!input.name?.trim()) return { ok: false as const, error: 'Give the campaign a name.' }
  const row = { ...campaignRow(input), campaign_type: campaignTypeFor(kind), sender_stream: 'updates', status: 'draft', created_by: actor.label, created_by_user_id: actor.id, updated_at: new Date().toISOString() }
  const { data, error } = await admin.from('email_campaigns').insert(row).select('id').single()
  if (error) return { ok: false as const, error: missing(error) ? 'Campaigns need database migration 0030.' : error.message }
  return { ok: true as const, id: (data as { id: string }).id }
}

export async function updateCampaign(id: string, input: Partial<CampaignInput>, actor: Actor, admin: Admin = createAdminClient()) {
  const c = await loadCampaign(id, admin)
  if (!c) return { ok: false as const, error: 'Campaign not found.' }
  if (c.status === 'stopped' || c.status === 'completed') return { ok: false as const, error: 'A finished campaign cannot be edited. Duplicate it instead.' }
  const row = campaignRow(input)
  if (c.status !== 'draft') {
    // running / paused: schedule and throttle only — the sequence and audience are fixed once people are enrolled
    for (const k of ['steps', 'template_key', 'audience', 'audience_kind']) delete row[k]
  }
  const { error } = await admin.from('email_campaigns').update({ ...row, updated_at: new Date().toISOString() }).eq('id', id)
  if (error) return { ok: false as const, error: error.message }
  void actor
  return { ok: true as const }
}

export async function duplicateCampaign(id: string, actor: Actor, admin: Admin = createAdminClient()) {
  const c = await loadCampaign(id, admin)
  if (!c) return { ok: false as const, error: 'Campaign not found.' }
  return createCampaign({ name: `${c.name} (copy)`, description: c.description, steps: c.steps, audience_kind: c.audience_kind, audience: c.audience, send_window: c.send_window, batch_size: c.batch_size, ramp: c.ramp, frequency_cap_days: c.frequency_cap_days, daily_send_limit: c.daily_send_limit, start_date: null, end_date: null }, actor, admin, c.kind)
}

export async function deleteCampaign(id: string, admin: Admin = createAdminClient()) {
  const c = await loadCampaign(id, admin)
  if (!c) return { ok: false as const, error: 'Campaign not found.' }
  if (c.status !== 'draft') return { ok: false as const, error: 'Only a draft can be deleted. Stop a running campaign instead.' }
  const { error } = await admin.from('email_campaigns').delete().eq('id', id)
  return error ? { ok: false as const, error: error.message } : { ok: true as const }
}

/** Who the campaign would reach right now, and who is excluded and why. */
export async function previewAudience(c: Pick<LeadCampaign, 'audience_kind' | 'audience' | 'id'> & { kind?: LeadKind }, all?: LeadRow[], admin: Admin = createAdminClient()) {
  const leads = all ?? (await loadAllLeads(admin, c.kind ?? 'broker'))
  const matching = c.audience_kind === 'leads' ? leads.filter((l) => ((c.audience as { lead_ids: string[] }).lead_ids ?? []).includes(l.id)) : filterLeads(leads, c.audience as SegmentFilters)
  const { data: enrolled } = c.id ? await admin.from('email_campaign_recipients').select('lead_id').eq('campaign_id', c.id) : { data: [] }
  const already = new Set(((enrolled ?? []) as Array<{ lead_id: string }>).map((r) => r.lead_id))
  const blocked: Record<string, number> = {}
  const eligible: LeadRow[] = []
  for (const l of matching) {
    const b = leadSendBlock(l)
    if (b) { blocked[b] = (blocked[b] ?? 0) + 1; continue }
    eligible.push(l)
  }
  return { matching: matching.length, eligible: eligible.filter((l) => !already.has(l.id)).length, already_enrolled: matching.filter((l) => already.has(l.id)).length, blocked, sample: eligible.slice(0, 8).map((l) => ({ id: l.id, name: [l.first_name, l.last_name].filter(Boolean).join(' ') || l.email, email: l.email, brokerage: l.brokerage_name, state: l.state })), eligibleRows: eligible.filter((l) => !already.has(l.id)) }
}

/** Put the audience into the campaign (pending, step 1). Re-runnable: only new leads are added. */
export async function enrollCampaign(id: string, actor: Actor, admin: Admin = createAdminClient()) {
  const c = await loadCampaign(id, admin)
  if (!c) return { ok: false as const, error: 'Campaign not found.' }
  if (c.status === 'stopped' || c.status === 'completed') return { ok: false as const, error: 'This campaign is finished.' }
  const preview = await previewAudience(c, undefined, admin)
  const now = new Date()
  const rows = preview.eligibleRows.map((l) => {
    const tz = timezoneForLead(l.state, c.send_window)
    const start = c.start_date && Date.parse(`${c.start_date}T00:00:00Z`) > now.getTime() ? new Date(`${c.start_date}T12:00:00Z`) : now
    return { campaign_id: id, lead_id: l.id, email_normalized: l.email_normalized, step: 1, status: 'pending', scheduled_for: nextWindowStart(c.send_window, start, tz).toISOString() }
  })
  let added = 0
  for (let i = 0; i < rows.length; i += 500) {
    const chunk = rows.slice(i, i + 500)
    const { error } = await admin.from('email_campaign_recipients').upsert(chunk, { onConflict: 'campaign_id,lead_id', ignoreDuplicates: true })
    if (error) return { ok: false as const, error: error.message }
    added += chunk.length
    await admin.from('broker_lead_events').insert(chunk.map((r) => ({ lead_id: r.lead_id, event_type: 'enrolled', campaign_id: id, actor_label: actor.label, detail: { campaign: c.name } })))
  }
  const counts = await recipientCounts([id], admin)
  const r = counts.get(id)!
  const enrolledTotal = r.pending + r.scheduled + r.sent + r.skipped + r.failed + r.stopped + r.done + r.converted
  await admin.from('email_campaigns').update({ enrolled_count: enrolledTotal, audience_count: preview.matching, updated_at: new Date().toISOString() }).eq('id', id)
  return { ok: true as const, added, enrolled: enrolledTotal, matching: preview.matching, blocked: preview.blocked }
}

export async function campaignProblems(c: LeadCampaign, admin: Admin = createAdminClient()): Promise<string[]> {
  const templates = await loadLeadTemplates(admin, c.kind)
  const active = Object.fromEntries(templates.map((t) => [t.key, t.active]))
  const enrolled = c.recipients.pending + c.recipients.scheduled + c.recipients.sent
  return campaignReadiness({ name: c.name, steps: c.steps, start_date: c.start_date, daily_send_limit: c.daily_send_limit, enrolled, templatesActive: active })
}

export async function setCampaignStatus(id: string, action: 'approve' | 'pause' | 'resume' | 'stop', actor: Actor, admin: Admin = createAdminClient()) {
  const c = await loadCampaign(id, admin)
  if (!c) return { ok: false as const, error: 'Campaign not found.' }
  const t = campaignTransition(c.status, action)
  if (!t.ok) return { ok: false as const, error: t.error }
  const now = new Date().toISOString()
  const patch: Record<string, unknown> = { status: t.to, updated_at: now }
  if (action === 'approve') {
    const problems = await campaignProblems(c, admin)
    if (problems.length) return { ok: false as const, error: problems.join(' ') }
    Object.assign(patch, { approved_by: actor.label, approved_at: now, pause_reason: null, paused_at: null })
  }
  if (action === 'resume') Object.assign(patch, { pause_reason: null, paused_at: null })
  if (action === 'pause') Object.assign(patch, { pause_reason: `paused by ${actor.label}`, paused_at: now })
  if (action === 'stop') {
    await admin.from('email_campaign_recipients').update({ status: 'stopped', stop_reason: 'campaign_stopped', updated_at: now }).eq('campaign_id', id).in('status', ['pending', 'scheduled'])
  }
  const { error } = await admin.from('email_campaigns').update(patch).eq('id', id)
  return error ? { ok: false as const, error: error.message } : { ok: true as const, status: t.to }
}

/* -------------------------------------------------------------- sending */

export interface LeadSendOutcome { status: 'sent' | 'recorded_not_delivered' | 'suppressed' | 'skipped' | 'failed' | 'blocked'; reason?: string; messageId?: string | null; subject?: string | null }

/** One email to one lead through the central sender, with the lead's counters, status and timeline updated. */
export async function sendLeadEmail(params: { lead: LeadRow; templateKey: string; campaignId?: string | null; campaignName?: string | null; step?: number | null; actor: Actor; origin?: string }, admin: Admin = createAdminClient()): Promise<LeadSendOutcome> {
  const { lead } = params
  const block = leadSendBlock(lead)
  if (block) { return { status: 'blocked', reason: block } }
  const origin = params.origin ?? SITE_URL
  const token = await createJoinToken({ leadId: lead.id, email: lead.email })
  const joinLink = `${origin}/join/${token}`
  const data = leadTemplateData({ first_name: lead.first_name, last_name: lead.last_name, broker_company_name: lead.brokerage_name ?? lead.broker_company_name, mc_number: lead.mc_number, email: lead.email, company_name: lead.company_name, lead_type: lead.lead_type }, joinLink)
  const r = await sendPlatformEmail({ templateKey: params.templateKey, to: lead.email, data, campaignId: params.campaignId ?? null, campaignType: campaignTypeFor(lead.lead_kind), actorUserId: params.actor.id, origin, tags: { lead: lead.id, step: String(params.step ?? 1) } })
  const { data: m } = await admin.from('email_messages').select('id').eq('client_reference', r.clientReference).maybeSingle()
  const messageId = (m as { id: string } | null)?.id ?? null
  const now = new Date().toISOString()
  if (r.status === 'sent' || r.status === 'recorded_not_delivered') {
    const first = !lead.invited_at && !lead.last_emailed_at
    const patch: Record<string, unknown> = {
      emails_sent: lead.emails_sent + 1, last_emailed_at: now, last_template_key: params.templateKey, join_token_issued_at: now, updated_at: now,
      invitation_status: first ? 'invited' : 'reminded', invited_at: lead.invited_at ?? now, ...(first ? {} : { reminded_at: now }),
    }
    let up = await admin.from('broker_agent_leads').update(patch).eq('id', lead.id)
    if (up.error && missing(up.error)) up = await admin.from('broker_agent_leads').update({ invitation_status: patch.invitation_status, invited_at: patch.invited_at, updated_at: now }).eq('id', lead.id)
    await logLeadEvent({ leadId: lead.id, type: 'email_sent', campaignId: params.campaignId, messageId, templateKey: params.templateKey, step: params.step, actorLabel: params.actor.label, detail: { status: r.status, subject: r.subject, campaign: params.campaignName ?? null, demo: r.status === 'recorded_not_delivered' } }, admin)
    return { status: r.status, messageId, subject: r.subject }
  }
  if (r.status === 'failed') {
    await logLeadEvent({ leadId: lead.id, type: 'email_failed', campaignId: params.campaignId, messageId, templateKey: params.templateKey, step: params.step, actorLabel: params.actor.label, detail: { error: r.reason ?? null } }, admin)
    return { status: 'failed', reason: r.reason, messageId }
  }
  // suppressed / skipped by the central policy — reflect the strongest reasons on the lead
  if (r.reason === 'hard_bounced') await admin.from('broker_agent_leads').update({ bounced_at: now, updated_at: now }).eq('id', lead.id)
  if (r.reason === 'marketing_suppression' || r.reason === 'preference_off') await admin.from('broker_agent_leads').update({ unsubscribed_at: now, invitation_status: 'opted_out', updated_at: now }).eq('id', lead.id)
  await logLeadEvent({ leadId: lead.id, type: 'email_skipped', campaignId: params.campaignId, messageId, templateKey: params.templateKey, step: params.step, actorLabel: params.actor.label, detail: { reason: r.reason ?? r.status } }, admin)
  return { status: r.status, reason: r.reason, messageId }
}

/** "Invite now" on selected leads: immediate, capped, outside any campaign. */
export async function inviteLeadsNow(params: { leadIds: string[]; templateKey: string; limit?: number; actor: Actor; origin?: string }, admin: Admin = createAdminClient(), kind: LeadKind = 'broker') {
  const settings = await loadLeadsSettings(admin, kind)
  const limit = Math.max(1, Math.min(500, params.limit ?? settings.default_daily_limit))
  const companies = await brokerCompanies(admin)
  const ids = params.leadIds.slice(0, 2000)
  const q1 = await admin.from('broker_agent_leads').select(LEAD_COLUMNS).in('id', ids)
  const q2 = q1.error && missing(q1.error) && kind === 'broker' ? await admin.from('broker_agent_leads').select(LEAD_COLUMNS_0030).in('id', ids) : null
  const data = ((q2 ?? q1).data ?? null) as Record<string, unknown>[] | null
  const error = (q2 ?? q1).error
  if (error) return { ok: false as const, error: missing(error) ? 'Lead invitations need database migration 0030.' : error.message }
  const leads = ((data ?? []) as Array<Record<string, unknown>>).map((r) => decorate(r, companies))
  const now = new Date()
  const out = { sent: 0, recorded: 0, blocked: 0, suppressed: 0, failed: 0, capped: 0, results: [] as Array<{ id: string; email: string; status: string; reason?: string }> }
  for (const lead of leads) {
    if (out.sent + out.recorded >= limit) { out.capped++; continue }
    if (!frequencyCapOk(lead.last_emailed_at, settings.default_frequency_cap_days, now)) { out.blocked++; out.results.push({ id: lead.id, email: lead.email, status: 'blocked', reason: 'frequency_cap' }); continue }
    const r = await sendLeadEmail({ lead, templateKey: params.templateKey, step: 1, actor: params.actor, origin: params.origin }, admin)
    if (r.status === 'sent') out.sent++
    else if (r.status === 'recorded_not_delivered') out.recorded++
    else if (r.status === 'blocked') out.blocked++
    else if (r.status === 'failed') out.failed++
    else out.suppressed++
    out.results.push({ id: lead.id, email: lead.email, status: r.status, reason: r.reason })
  }
  return { ok: true as const, ...out }
}

/** A test email to a staff address, rendered with sample data. */
export async function sendTestEmail(params: { templateKey: string; to: string; campaignId?: string | null; actor: Actor; origin?: string }, admin: Admin = createAdminClient(), kind: LeadKind = 'broker') {
  const origin = params.origin ?? SITE_URL
  const data = kind === 'carrier'
    ? leadTemplateData({ first_name: 'Sam', last_name: 'Rivera', broker_company_name: null, mc_number: null, email: params.to, company_name: 'ABC TRANSPORT LLC', lead_type: 'both' }, `${origin}/join/test-link-not-valid`)
    : leadTemplateData({ first_name: 'Sam', last_name: 'Rivera', broker_company_name: 'ABC FREIGHT BROKERAGE LLC', mc_number: '654321', email: params.to }, `${origin}/join/test-link-not-valid`)
  const r = await sendPlatformEmail({ templateKey: params.templateKey, to: params.to, data, campaignType: campaignTypeFor(kind), actorUserId: params.actor.id, origin, tags: { test: 'true' } })
  if (params.campaignId && (r.status === 'sent' || r.status === 'recorded_not_delivered')) await admin.from('email_campaigns').update({ test_sent_at: new Date().toISOString() }).eq('id', params.campaignId)
  return r.status === 'sent' || r.status === 'recorded_not_delivered' ? { ok: true as const, status: r.status, subject: r.subject } : { ok: false as const, error: `${r.status}${r.reason ? `: ${r.reason}` : ''}` }
}

/* --------------------------------------------------------------- runner */

export interface RunSummary { ran: boolean; reason?: string; at: string; campaigns: Array<{ id: string; name: string; allowance: number; sent_before: number; sent: number; rescheduled: number; stopped: number; skipped: number; failed: number; completed: boolean }> }

/**
 * The scheduler's pass (hourly cron or "Run now"). For every running campaign:
 * today's allowance (daily limit, warm-up ramp, what already went today) →
 * due recipients → stop rules → the lead's send window → the frequency cap →
 * send → schedule the next step or finish. Repeated calls are safe.
 */
export async function runBrokerCampaigns(params: { now?: Date; origin: string; actor: Actor; manual: boolean }, admin: Admin = createAdminClient(), kind: LeadKind = 'broker'): Promise<RunSummary> {
  const now = params.now ?? new Date()
  const at = now.toISOString()
  const settings = await loadLeadsSettings(admin, kind)
  if (!params.manual && !settings.cron_enabled) return { ran: false, reason: 'Automatic sending is switched off.', at, campaigns: [] }
  const today = at.slice(0, 10)
  const campaigns = (await loadCampaigns(admin, kind)).filter((c) => c.status === 'running' && (!c.start_date || c.start_date <= today) && (!c.end_date || c.end_date >= today))
  const summary: RunSummary = { ran: true, at, campaigns: [] }
  const companies = await brokerCompanies(admin)

  for (const c of campaigns) {
    if (!c.first_run_date) { await admin.from('email_campaigns').update({ first_run_date: today }).eq('id', c.id); c.first_run_date = today }
    const allowance = dailyAllowance({ daily_send_limit: c.daily_send_limit, ramp: c.ramp }, sendingDayIndex(c.first_run_date, today))
    const { count: sentToday } = await admin.from('email_campaign_recipients').select('id', { count: 'exact', head: true }).eq('campaign_id', c.id).gte('last_sent_at', `${today}T00:00:00Z`)
    const line = { id: c.id, name: c.name, allowance, sent_before: sentToday ?? 0, sent: 0, rescheduled: 0, stopped: 0, skipped: 0, failed: 0, completed: false }
    let remaining = Math.min(c.batch_size, Math.max(0, allowance - (sentToday ?? 0)))
    if (remaining > 0) {
      const { data: due } = await admin.from('email_campaign_recipients').select('*').eq('campaign_id', c.id).in('status', ['pending', 'scheduled']).or(`scheduled_for.is.null,scheduled_for.lte.${at}`).order('scheduled_for', { ascending: true, nullsFirst: true }).limit(remaining * 4)
      for (const r of (due ?? []) as Array<{ id: string; lead_id: string; step: number; sends: number }>) {
        if (remaining <= 0) break
        const first = await admin.from('broker_agent_leads').select(LEAD_COLUMNS).eq('id', r.lead_id).maybeSingle()
        const lrow = first.error && missing(first.error) ? (await admin.from('broker_agent_leads').select(LEAD_COLUMNS_0030).eq('id', r.lead_id).maybeSingle()).data : first.data
        if (!lrow) { await admin.from('email_campaign_recipients').update({ status: 'stopped', stop_reason: 'lead_deleted', updated_at: at }).eq('id', r.id); line.stopped++; continue }
        const lead = decorate(lrow as Record<string, unknown>, companies)
        const stop = recipientStopReason(lead)
        if (stop) {
          await admin.from('email_campaign_recipients').update({ status: stop === 'already_converted' ? 'converted' : 'stopped', stop_reason: stop, updated_at: at }).eq('id', r.id)
          await logLeadEvent({ leadId: lead.id, type: 'stopped', campaignId: c.id, step: r.step, actorLabel: params.actor.label, detail: { reason: stop, campaign: c.name } }, admin)
          line.stopped++
          continue
        }
        const tz = timezoneForLead(lead.state, c.send_window)
        if (!windowOpen(c.send_window, now, tz)) {
          await admin.from('email_campaign_recipients').update({ status: 'scheduled', scheduled_for: nextWindowStart(c.send_window, now, tz).toISOString(), updated_at: at }).eq('id', r.id)
          line.rescheduled++
          continue
        }
        if (!frequencyCapOk(lead.last_emailed_at, c.frequency_cap_days, now)) {
          const release = nextWindowStart(c.send_window, frequencyCapReleaseAt(lead.last_emailed_at!, c.frequency_cap_days), tz)
          await admin.from('email_campaign_recipients').update({ status: 'scheduled', scheduled_for: release.toISOString(), updated_at: at }).eq('id', r.id)
          line.rescheduled++
          continue
        }
        const step = c.steps.find((s) => s.step === r.step)
        if (!step) { await admin.from('email_campaign_recipients').update({ status: 'done', updated_at: at }).eq('id', r.id); continue }
        const out = await sendLeadEmail({ lead, templateKey: step.template_key, campaignId: c.id, campaignName: c.name, step: r.step, actor: params.actor, origin: params.origin }, admin)
        if (out.status === 'sent' || out.status === 'recorded_not_delivered') {
          const next = c.steps.find((s) => s.step === r.step + 1)
          const patch: Record<string, unknown> = { sends: r.sends + 1, last_sent_at: at, last_message_id: out.messageId ?? null, updated_at: at }
          if (next) Object.assign(patch, { step: next.step, status: 'scheduled', scheduled_for: stepDueAt(now, next.delay_days, c.send_window, tz).toISOString() })
          else Object.assign(patch, { status: 'done' })
          await admin.from('email_campaign_recipients').update(patch).eq('id', r.id)
          line.sent++
          remaining--
        } else if (out.status === 'failed') {
          line.failed++
          remaining-- // counts against the batch so a broken provider cannot spin
        } else {
          await admin.from('email_campaign_recipients').update({ status: 'skipped', skip_reason: out.reason ?? out.status, updated_at: at }).eq('id', r.id)
          line.skipped++
        }
      }
    }
    // counters + completion
    const counts = (await recipientCounts([c.id], admin)).get(c.id)!
    const open = counts.pending + counts.scheduled
    const completed = open === 0 && (counts.sent + counts.done + counts.converted + counts.stopped + counts.skipped + counts.failed) > 0
    line.completed = completed
    await admin.from('email_campaigns').update({
      sent_count: c.sent_count + line.sent, done_count: counts.done + counts.converted, stopped_count: counts.stopped, last_run_at: at, last_run_summary: line, updated_at: at,
      ...(completed ? { status: 'completed' } : {}),
    }).eq('id', c.id)
    summary.campaigns.push(line)
  }
  return summary
}

/* ------------------------------------------------------------ analytics */

export interface LeadsAnalytics {
  funnel: { leads: number; emailed: number; delivered: number; opened: number; clicked: number; signed_up: number; converted: number; bounced: number; unsubscribed: number }
  byCampaign: Array<{ id: string; name: string; status: string; sent: number; delivered: number; opened: number; clicked: number; signed_up: number; converted: number; bounced: number; unsubscribed: number }>
  byTemplate: Array<{ template_key: string; version: number | null; sent: number; delivered: number; opened: number; clicked: number }>
  byState: Array<{ state: string; leads: number; emailed: number; signed_up: number; converted: number }>
  byBrokerage: Array<{ id: string; name: string; leads: number; emailed: number; signed_up: number; converted: number }>
  last30: Array<{ day: string; sent: number; delivered: number; opened: number; clicked: number; signed_up: number }>
}

export async function loadLeadsAnalytics(all: LeadRow[], campaigns: LeadCampaign[], admin: Admin = createAdminClient(), kind: LeadKind = 'broker'): Promise<LeadsAnalytics> {
  const funnel = { leads: all.length, emailed: 0, delivered: 0, opened: 0, clicked: 0, signed_up: 0, converted: 0, bounced: 0, unsubscribed: 0 }
  const byState = new Map<string, { state: string; leads: number; emailed: number; signed_up: number; converted: number }>()
  const byBrokerage = new Map<string, { id: string; name: string; leads: number; emailed: number; signed_up: number; converted: number }>()
  for (const l of all) {
    if (l.last_emailed_at) funnel.emailed++
    if (l.last_delivered_at) funnel.delivered++
    if (l.last_opened_at) funnel.opened++
    if (l.last_clicked_at) funnel.clicked++
    if (l.signed_up_at) funnel.signed_up++
    if (l.stage === 'converted') funnel.converted++
    if (l.bounced_at) funnel.bounced++
    if (l.unsubscribed_at || l.invitation_status === 'opted_out') funnel.unsubscribed++
    const st = l.state ?? '—'
    const s = byState.get(st) ?? { state: st, leads: 0, emailed: 0, signed_up: 0, converted: 0 }
    s.leads++; if (l.last_emailed_at) s.emailed++; if (l.signed_up_at) s.signed_up++; if (l.stage === 'converted') s.converted++
    byState.set(st, s)
    const companyKey = l.lead_kind === 'carrier' ? l.company_name : l.broker_company_id
    if (companyKey) {
      const b = byBrokerage.get(companyKey) ?? { id: companyKey, name: l.brokerage_name ?? companyKey, leads: 0, emailed: 0, signed_up: 0, converted: 0 }
      b.leads++; if (l.last_emailed_at) b.emailed++; if (l.signed_up_at) b.signed_up++; if (l.stage === 'converted') b.converted++
      byBrokerage.set(companyKey, b)
    }
  }
  // messages of this campaign type in the last 30 days → per template + per day
  const since = new Date(Date.now() - 30 * 86_400_000).toISOString()
  const { data: msgs } = await admin.from('email_messages').select('template_key, template_version, status, sent_at, delivered_at, opened_at, clicked_at, created_at').eq('campaign_type', campaignTypeFor(kind)).gte('created_at', since).limit(5000)
  const byTemplate = new Map<string, { template_key: string; version: number | null; sent: number; delivered: number; opened: number; clicked: number }>()
  const byDay = new Map<string, { day: string; sent: number; delivered: number; opened: number; clicked: number; signed_up: number }>()
  for (const m of (msgs ?? []) as Array<{ template_key: string | null; template_version: number | null; status: string; sent_at: string | null; delivered_at: string | null; opened_at: string | null; clicked_at: string | null; created_at: string }>) {
    const k = `${m.template_key ?? '?'}@${m.template_version ?? 0}`
    const t = byTemplate.get(k) ?? { template_key: m.template_key ?? '?', version: m.template_version, sent: 0, delivered: 0, opened: 0, clicked: 0 }
    const wasSent = !!m.sent_at || ['sent', 'delivered', 'soft_bounced', 'hard_bounced', 'complained'].includes(m.status)
    if (wasSent) t.sent++; if (m.delivered_at) t.delivered++; if (m.opened_at) t.opened++; if (m.clicked_at) t.clicked++
    byTemplate.set(k, t)
    const day = (m.sent_at ?? m.created_at).slice(0, 10)
    const d = byDay.get(day) ?? { day, sent: 0, delivered: 0, opened: 0, clicked: 0, signed_up: 0 }
    if (wasSent) d.sent++; if (m.delivered_at) d.delivered++; if (m.opened_at) d.opened++; if (m.clicked_at) d.clicked++
    byDay.set(day, d)
  }
  for (const l of all) if (l.signed_up_at && l.signed_up_at >= since) { const day = l.signed_up_at.slice(0, 10); const d = byDay.get(day) ?? { day, sent: 0, delivered: 0, opened: 0, clicked: 0, signed_up: 0 }; d.signed_up++; byDay.set(day, d) }
  return {
    funnel,
    byCampaign: campaigns.map((c) => ({ id: c.id, name: c.name, status: c.status, sent: c.sent_count, delivered: c.delivered_count, opened: c.open_count, clicked: c.click_count, signed_up: c.signed_up_count, converted: c.conversion_count, bounced: c.bounce_count, unsubscribed: c.unsubscribe_count })),
    byTemplate: [...byTemplate.values()].sort((a, b) => b.sent - a.sent),
    byState: [...byState.values()].sort((a, b) => b.leads - a.leads),
    byBrokerage: [...byBrokerage.values()].sort((a, b) => b.leads - a.leads).slice(0, 50),
    last30: [...byDay.values()].sort((a, b) => a.day.localeCompare(b.day)),
  }
}

/* ---------------------------------------------------------------- export */

export function leadsToCsv(rows: LeadRow[], kind: LeadKind = 'broker'): string {
  const cols = kind === 'carrier'
    ? (['first_name', 'last_name', 'email', 'phone', 'company_name', 'lead_type', 'source_status_active', 'source_created_at', 'stage', 'invitation_status', 'emails_sent', 'last_emailed_at', 'last_opened_at', 'last_clicked_at', 'signed_up_at', 'tags', 'source_system', 'source_record_id', 'mailbox_type'] as const)
    : (['first_name', 'last_name', 'email', 'phone', 'brokerage_name', 'mc_number', 'state', 'stage', 'invitation_status', 'emails_sent', 'last_emailed_at', 'last_opened_at', 'last_clicked_at', 'signed_up_at', 'confirmed_at', 'tags', 'source_system', 'source_last_seen_at', 'email_domain_check', 'mailbox_type'] as const)
  const cell = (v: unknown) => { const s = Array.isArray(v) ? v.join(';') : v == null ? '' : String(v); return /[",\n]/.test(s) ? `"${s.replace(/"/g, '""')}"` : s }
  return [cols.join(','), ...rows.map((r) => cols.map((c) => cell((r as unknown as Record<string, unknown>)[c])).join(','))].join('\n') + '\n'
}