import 'server-only'

import { createHash } from 'node:crypto'
import { createAdminClient } from '@/lib/supabase/admin'
import { emailPattern } from '@/lib/like'
import { normalizeEmail } from '@/lib/email/send'
import { campaignPauseDecision, normalizeEventType, softBounceDecision, type EmailStream, type NormalizedEventType } from '@/lib/domain/email-policy'
import { recordLeadEmailEvent } from '@/lib/data/broker-lead-hooks'
import { parseZeptoMailWebhook } from '@/lib/domain/zeptomail-webhook'
import { applyDeliveryEvents, pilotNetworkAvailable } from '@/lib/data/pilot-network'
import { logSecurityEvent } from '@/lib/security/events'

/**
 * Central email-event processor (2026-09-22, §24):
 *   webhook → verify → normalize → store (idempotent) → update message +
 *   delivery state → suppression rules → campaign counters / auto-pause.
 * Both Mail Agents feed this one path.
 */

export interface IncomingEvent {
  agent: 'transactional' | 'updates'
  rawEvent: string
  type: NormalizedEventType
  recipient: string
  requestId: string | null
  clientReference: string | null
  reason: string | null
  occurredAt: string | null
  payload: unknown
}

/** Walk the payload for the identifiers ZeptoMail includes (request_id, client_reference, processed_time). */
export function extractIdentifiers(payload: unknown): { requestId: string | null; clientReference: string | null; occurredAt: string | null } {
  const out = { requestId: null as string | null, clientReference: null as string | null, occurredAt: null as string | null }
  const walk = (v: unknown, depth: number) => {
    if (!v || typeof v !== 'object' || depth > 6) return
    for (const [k, val] of Object.entries(v as Record<string, unknown>)) {
      if (typeof val === 'string') {
        if (!out.requestId && /^request_id$/i.test(k)) out.requestId = val
        if (!out.clientReference && /^client_reference$/i.test(k)) out.clientReference = val
        if (!out.occurredAt && /^(processed_time|time|timestamp|event_time)$/i.test(k) && !Number.isNaN(Date.parse(val))) out.occurredAt = new Date(val).toISOString()
      } else if (typeof val === 'number' && !out.occurredAt && /^(processed_time|time|timestamp)$/i.test(k)) {
        out.occurredAt = new Date(val > 1e12 ? val : val * 1000).toISOString()
      } else walk(val, depth + 1)
    }
  }
  walk(payload, 0)
  return out
}

/** One ZeptoMail POST → zero or more normalized recipient events. */
export function eventsFromPayload(agent: IncomingEvent['agent'], payload: unknown): IncomingEvent[] {
  const ids = extractIdentifiers(payload)
  return parseZeptoMailWebhook(payload).flatMap((ev) =>
    ev.emails.map((email) => ({ agent, rawEvent: ev.raw_event, type: normalizeEventType(ev.raw_event), recipient: normalizeEmail(email), requestId: ids.requestId, clientReference: ids.clientReference, reason: ev.reason, occurredAt: ids.occurredAt, payload })),
  )
}

function dedupeKey(e: IncomingEvent): string {
  return createHash('sha256').update([e.agent, e.type, e.recipient, e.requestId ?? '', e.clientReference ?? '', e.occurredAt ?? JSON.stringify(e.payload).slice(0, 2000)].join('|')).digest('hex')
}

async function settings() {
  const { data } = await createAdminClient().from('email_settings').select('*').eq('id', 1).maybeSingle()
  return {
    soft: { threshold: data?.soft_bounce_threshold ?? 3, windowDays: data?.soft_bounce_window_days ?? 14, suppressDays: data?.soft_bounce_suppress_days ?? 7 },
    campaign: { minSent: data?.campaign_pause_min_sent ?? 20, pausePct: Number(data?.campaign_pause_bounce_pct ?? 5) },
  }
}

export async function processIncomingEvents(events: IncomingEvent[]): Promise<{ stored: number; duplicates: number; suppressed: number; unmatched: number }> {
  const admin = createAdminClient()
  const out = { stored: 0, duplicates: 0, suppressed: 0, unmatched: 0 }
  const cfg = await settings()
  for (const e of events) {
    const stream: EmailStream = e.agent
    const key = dedupeKey(e)
    const ins = await admin.from('email_webhook_events').insert({ agent: e.agent, email_stream: stream, dedupe_key: key, raw_event: e.rawEvent, normalized_type: e.type, request_id: e.requestId, client_reference: e.clientReference, recipient: e.recipient, reason: e.reason, payload: e.payload }).select('id').single()
    if (ins.error) {
      if (/duplicate|unique/i.test(ins.error.message)) { out.duplicates++; continue }
      if (/does not exist|schema cache/i.test(ins.error.message)) { out.unmatched++; continue } // 0028 not applied — nothing to update
      console.error('email_webhook_events insert failed', ins.error.message); continue
    }
    out.stored++
    const now = e.occurredAt ?? new Date().toISOString()

    // 1. the message this belongs to (client_reference first, then request id, then newest to this recipient)
    let msg = e.clientReference ? (await admin.from('email_messages').select('id, campaign_id, campaign_type').eq('client_reference', e.clientReference).maybeSingle()).data : null
    if (!msg && e.requestId) msg = (await admin.from('email_messages').select('id, campaign_id, campaign_type').eq('zepto_request_id', e.requestId).maybeSingle()).data
    if (!msg) msg = (await admin.from('email_messages').select('id, campaign_id, campaign_type').eq('email_normalized', e.recipient).eq('email_stream', stream).order('created_at', { ascending: false }).limit(1).maybeSingle()).data
    if (msg) {
      const patch: Record<string, unknown> = { updated_at: now }
      if (e.type === 'delivered') Object.assign(patch, { status: 'delivered', delivered_at: now })
      if (e.type === 'opened') Object.assign(patch, { opened_at: now })
      if (e.type === 'clicked') Object.assign(patch, { clicked_at: now })
      if (e.type === 'soft_bounce') Object.assign(patch, { status: 'soft_bounced', bounce_type: 'soft', bounce_reason: e.reason })
      if (e.type === 'hard_bounce') Object.assign(patch, { status: 'hard_bounced', bounce_type: 'hard', bounce_reason: e.reason })
      if (e.type === 'spam_complaint') Object.assign(patch, { status: 'complained', complaint_at: now })
      await admin.from('email_messages').update(patch).eq('id', msg.id)
    } else out.unmatched++
    // Freight Broker Leads (2026-09-23): the lead's timeline and last-activity columns follow the same event.
    if (e.type !== 'other') await recordLeadEmailEvent({ email: e.recipient, type: e.type, messageId: msg?.id ?? null, campaignId: msg?.campaign_id ?? null, at: now, reason: e.reason })

    // 2. delivery state + suppression
    const { data: st } = await admin.from('email_delivery_state').select('*').eq('email_normalized', e.recipient).maybeSingle()
    const state = st ?? { email_normalized: e.recipient, status: 'ok', soft_bounce_count: 0, first_soft_bounce_at: null }
    const upsertState = (patch: Record<string, unknown>) => admin.from('email_delivery_state').upsert({ email_normalized: e.recipient, ...patch, last_event: e.rawEvent, updated_at: now }, { onConflict: 'email_normalized' })
    const suppress = (scope: 'global' | 'marketing', reason: string, expiresAt: string | null) =>
      admin.from('email_suppressions').upsert({ email_normalized: e.recipient, scope, campaign_type: null, reason, source: 'zeptomail webhook', agent: e.agent, event_id: key, detail: { raw_event: e.rawEvent, reason: e.reason }, expires_at: expiresAt }, { onConflict: 'email_normalized,scope,campaign_type', ignoreDuplicates: true })

    if (e.type === 'delivered') {
      await upsertState({ status: state.status === 'hard_bounced' ? 'hard_bounced' : 'ok', soft_bounce_count: 0, first_soft_bounce_at: null, last_delivered_at: now })
    } else if (e.type === 'hard_bounce') {
      // §11: delivery-wide, immediately.
      await upsertState({ status: 'hard_bounced', hard_bounced_at: now })
      await suppress('global', 'hard_bounce', null)
      out.suppressed++
    } else if (e.type === 'soft_bounce') {
      // §12: count inside the window; suppress temporarily at the threshold.
      const d = softBounceDecision({ soft_bounce_count: state.soft_bounce_count ?? 0, first_soft_bounce_at: state.first_soft_bounce_at ?? null }, cfg.soft, Date.parse(now))
      await upsertState({ status: state.status === 'hard_bounced' ? 'hard_bounced' : 'soft_bouncing', soft_bounce_count: d.count, first_soft_bounce_at: d.firstAt, last_soft_bounce_at: now })
      if (d.suppress) { await suppress('global', 'repeated_soft_bounce', new Date(Date.parse(now) + cfg.soft.suppressDays * 86_400_000).toISOString()); out.suppressed++ }
    } else if (e.type === 'spam_complaint') {
      // §13: marketing stops at once; transactional stays possible (never blindly cut account mail).
      await upsertState({ last_complaint_at: now })
      await suppress('marketing', 'spam_complaint', null)
      out.suppressed++
      if (e.agent === 'transactional') await logSecurityEvent({ event: 'api_rate_limited', detail: { note: 'spam complaint on a transactional email — review', recipient: e.recipient, raw_event: e.rawEvent } })
    }

    // 3. campaign counters + auto-pause (§30)
    if (msg?.campaign_id) {
      const { data: c } = await admin.from('email_campaigns').select('*').eq('id', msg.campaign_id).maybeSingle()
      if (c) {
        const inc: Record<string, number> = {}
        if (e.type === 'delivered') inc.delivered_count = c.delivered_count + 1
        if (e.type === 'hard_bounce') { inc.hard_bounce_count = c.hard_bounce_count + 1; inc.bounce_count = c.bounce_count + 1 }
        if (e.type === 'soft_bounce') { inc.soft_bounce_count = c.soft_bounce_count + 1; inc.bounce_count = c.bounce_count + 1 }
        if (e.type === 'spam_complaint') inc.complaint_count = c.complaint_count + 1
        if (e.type === 'opened') inc.open_count = c.open_count + 1
        if (e.type === 'clicked') inc.click_count = c.click_count + 1
        const next = { ...c, ...inc }
        const pause = c.status === 'running' && campaignPauseDecision(next, cfg.campaign)
        await admin.from('email_campaigns').update({ ...inc, updated_at: now, ...(pause ? { status: 'paused', pause_reason: `hard bounce rate ${((next.hard_bounce_count / next.sent_count) * 100).toFixed(1)}% ≥ ${cfg.campaign.pausePct}%`, paused_at: now } : {}) }).eq('id', c.id)
      } else if (e.type === 'hard_bounce' || e.type === 'spam_complaint') {
        // Pilot pipeline campaigns keep their own table; a bad list pauses there too.
        const { data: pc } = await admin.from('pilot_campaigns').select('id, status').eq('id', msg.campaign_id).maybeSingle()
        if (pc?.status === 'running') {
          const { count: sent } = await admin.from('email_messages').select('id', { count: 'exact', head: true }).eq('campaign_id', pc.id).in('status', ['sent', 'delivered', 'hard_bounced', 'soft_bounced', 'complained', 'delivered'])
          const { count: hard } = await admin.from('email_messages').select('id', { count: 'exact', head: true }).eq('campaign_id', pc.id).eq('status', 'hard_bounced')
          if (campaignPauseDecision({ sent_count: sent ?? 0, hard_bounce_count: hard ?? 0 }, cfg.campaign)) await admin.from('pilot_campaigns').update({ status: 'paused', updated_at: now }).eq('id', pc.id)
        }
      }
    }

    // 4. keep the pilot pipeline's own history and suppression list in step (unchanged behaviour)
    if ((e.type === 'hard_bounce' || e.type === 'soft_bounce' || e.type === 'spam_complaint') && (await pilotNetworkAvailable(admin))) {
      const { data: profile } = await admin.from('pilot_profiles').select('id').ilike('email', emailPattern(e.recipient)).maybeSingle()
      if (profile) await applyDeliveryEvents([{ kind: e.type === 'spam_complaint' ? 'complaint' : e.type === 'hard_bounce' ? 'hardbounce' : 'softbounce', raw_event: e.rawEvent, emails: [e.recipient], reason: e.reason }], e.payload, admin)
    }
    await admin.from('email_webhook_events').update({ processed_at: new Date().toISOString(), note: msg ? 'matched' : 'no message matched' }).eq('id', ins.data.id)
  }
  return out
}