/**
 * Listing change outbox: claims `queued` rows of CUST_MarketplaceListingChange,
 * coalesces them per listing (last write wins), hands them to the marketplace
 * adapter and records the outcome. Runs from the Nitro plugin timer
 * (server/plugins/listing-outbox-flush.ts, service token) and from
 * POST /api/sales/sms/flush ("Send now", caller token).
 *
 * State machine (change): queued → sending → sent → confirmed | error
 *                                   ↘ queued (transient, ≤ 3 attempts)   cancelled / superseded are terminal
 * Listing.SyncStatus mirrors the latest change: queued → sending → sent → synced | error.
 *
 * Confirmation: feed deliveries via the feed result document; patch deliveries by
 * a bounded spot-check (getListingsItem on `sent` rows older than SPOT_CHECK_AFTER_MS)
 * — a full report reconcile is phase 3.
 */
import { acquire, LISTINGS_READ_BUCKET } from './rateLimiter'
import { amazonConnectionFromOrderSource, getListingsItem } from '../amazonSpApi'
import { getListingAdapter, marketplaceKeyOf, supportsListings } from './adapters'
import type { AdapterCtx, PushItem, PushResult } from './adapters/types'
import {
  type IdempiereFetch, type ChangeRow, type ListingRow, serviceFetch, listChanges, updateChange, updateListing,
  getListingsByIds, listingTablesAvailable, restDate
} from './listingStore'

export interface FlushOptions {
  orderSourceId?: number
  trigger: 'timer' | 'manual'
  fetch?: IdempiereFetch
  maxRunMs?: number
  ignoreDebounce?: boolean
}

export interface FlushSourceSummary { orderSourceId: number; name: string; marketplace: string; claimed: number; sent: number; errors: number; requeued: number; mode: 'patch' | 'feed' | 'none'; feedId?: string | null; skipped?: string }
export interface FlushSummary {
  runId: string; trigger: string; startedAt: string; durationMs: number
  sources: FlushSourceSummary[]; feedsPolled: number; confirmed: number; spotChecked: number; skipped?: string; error?: string
}

const DEBOUNCE_MS = Math.max(0, Number(process.env.LISTING_FLUSH_DEBOUNCE_MS ?? 20_000))
const STALE_CLAIM_MS = 30 * 60 * 1000
const MAX_ATTEMPTS = 3
const SPOT_CHECK_AFTER_MS = 5 * 60 * 1000
const SPOT_CHECK_MAX = 10
const PRICE_TOL = 0.005

let running = false
let lastRun: FlushSummary | null = null
export const getLastFlushRun = () => lastRun

const log = (msg: string, extra?: any) => extra ? console.log(`[listing-outbox] ${msg}`, extra) : console.log(`[listing-outbox] ${msg}`)
const logErr = (msg: string, extra?: any) => extra ? console.error(`[listing-outbox] ${msg}`, extra) : console.error(`[listing-outbox] ${msg}`)
const ageMs = (iso: string | null) => iso ? Date.now() - new Date(iso).getTime() : 0
const now = () => restDate(new Date())
const num = (v: any): number | null => (v == null || v === '') ? null : (Number.isFinite(Number(v)) ? Number(v) : null)

const loadOrderSource = async (f: IdempiereFetch, id: number): Promise<any | null> => {
  try { return await f(`models/c_ordersource/${Number(id)}`) } catch (e: any) { if (Number(e?.status || e?.statusCode) === 404) return null; throw e }
}

const makeCtx = (f: IdempiereFetch, orderSource: any): AdapterCtx => ({
  orderSource, orderSourceId: Number(orderSource.id), orgId: Number(orderSource?.AD_Org_ID?.id) || 0,
  marketplace: marketplaceKeyOf(orderSource), fetch: f,
  log: (msg, extra) => log(`[${orderSource?.Name || orderSource.id}] ${msg}`, extra)
})

/** `sending` rows nobody finished (crash mid-run) go back to `queued`. */
export const recoverStaleClaims = async (f: IdempiereFetch, olderThanMs = STALE_CLAIM_MS): Promise<number> => {
  const stuck = (await listChanges(f, { status: ['sending'] }, 500)).filter(c => ageMs(c.updated) > olderThanMs)
  for (const c of stuck) {
    await updateChange(f, c.id, { changeStatus: 'queued', errorMsg: 'recovered from stale sending state' })
    if (c.listingId) await updateListing(f, c.listingId, { syncStatus: 'queued' })
  }
  if (stuck.length) log(`recovered ${stuck.length} stale claim(s)`)
  return stuck.length
}

const OP_RANK: Record<string, number> = { delete: 5, create: 4, delist: 3, relist: 2, update: 1 }

/** Coalesce the queued changes of ONE listing into a push item; returns superseded ids too. */
const coalesce = (listing: ListingRow, changes: ChangeRow[]): { item: PushItem; superseded: number[] } => {
  const sorted = [...changes].sort((a, b) => (a.created || '').localeCompare(b.created || '') || a.id - b.id)
  const latest = new Map<string, ChangeRow>()
  const superseded: number[] = []
  for (const c of sorted) {
    const key = c.changeType === 'update' ? `update:${c.fieldName || ''}` : c.changeType
    const prev = latest.get(key)
    if (prev) superseded.push(prev.id)
    latest.set(key, c)
  }
  const kept = [...latest.values()]
  const op = kept.map(c => c.changeType).sort((a, b) => (OP_RANK[b] || 0) - (OP_RANK[a] || 0))[0] || 'update'
  const field = (name: string) => kept.find(c => c.changeType === 'update' && c.fieldName === name)
  const item: PushItem = { listing, changes: kept, op }
  const p = field('price'); if (p) item.price = num(p.newValue)
  const q = field('qty'); if (q) item.qty = num(q.newValue)
  const mn = field('minprice'); if (mn) item.minPrice = num(mn.newValue)
  const mx = field('maxprice'); if (mx) item.maxPrice = num(mx.newValue)
  const s = field('sale')
  if (s) { try { const j = JSON.parse(s.newValue || 'null'); item.sale = j ? { price: num(j.price), from: j.from || null, to: j.to || null } : { price: null, from: null, to: null } } catch { item.sale = null } }
  if (op === 'create') { item.price = item.price ?? listing.price; item.qty = item.qty ?? listing.qty ?? 0; item.minPrice = item.minPrice ?? listing.minPrice; item.maxPrice = item.maxPrice ?? listing.maxPrice }
  return { item, superseded }
}

const applyConfirmed = async (f: IdempiereFetch, listing: ListingRow, items: { change: ChangeRow; ok: boolean; error?: string }[]) => {
  const okOnes = items.filter(i => i.ok)
  const patch: Partial<ListingRow> = {}
  for (const { change } of okOnes) {
    if (change.changeType === 'update' && change.fieldName === 'price') patch.remotePrice = num(change.newValue)
    if (change.changeType === 'update' && change.fieldName === 'qty') patch.remoteQty = num(change.newValue)
    if (change.changeType === 'create' || change.changeType === 'relist') { patch.listingStatus = 'active'; patch.dateDelisted = null; if (change.changeType === 'create') { patch.dateListed = listing.dateListed || now(); patch.remotePrice = listing.price; patch.remoteQty = listing.qty } }
    if (change.changeType === 'delist') { patch.listingStatus = 'delisted'; patch.dateDelisted = now(); patch.remoteQty = 0 }
    if (change.changeType === 'delete') { patch.listingStatus = 'delisted'; patch.dateDelisted = now(); patch.isActive = false }
  }
  const failed = items.filter(i => !i.ok)
  const stillOpen = (await listChanges(f, { listingId: listing.id, status: ['queued', 'sending', 'sent'] }, 50)).length
  if (failed.length) { patch.syncStatus = 'error'; patch.syncError = failed.map(i => i.error).filter(Boolean).join('; ').slice(0, 2000) }
  else if (!stillOpen) { patch.syncStatus = 'synced'; patch.syncError = ''; patch.lastSyncedAt = now() }
  await updateListing(f, listing.id, patch)
}

/** Feed deliveries: ask the adapter whether the feed finished and settle every message. */
export const pollOpenFeeds = async (f: IdempiereFetch, orderSourceId?: number): Promise<{ polled: number; confirmed: number }> => {
  const open = await listChanges(f, { status: ['sent'], deliveryMode: 'feed', ...(orderSourceId ? { orderSourceId } : {}) }, 2000)
  const byFeed = new Map<string, ChangeRow[]>()
  for (const c of open) { if (!c.feedId) continue; const arr = byFeed.get(c.feedId) || []; arr.push(c); byFeed.set(c.feedId, arr) }
  let polled = 0, confirmed = 0
  for (const [feedId, changes] of byFeed) {
    const osId = changes[0]!.orderSourceId!
    try {
      const os = await loadOrderSource(f, osId)
      const adapter = os && getListingAdapter(os)
      if (!os || !adapter?.pollFeed) continue
      polled++
      const res = await adapter.pollFeed(makeCtx(f, os), feedId, [...new Set(changes.map(c => c.messageId!).filter(Boolean))])
      if (!res.done) continue
      const listings = await getListingsByIds(f, [...new Set(changes.map(c => c.listingId!).filter(Boolean))])
      const perListing = new Map<number, { change: ChangeRow; ok: boolean; error?: string }[]>()
      for (const c of changes) {
        let ok = false, error: string | undefined
        if (res.fatal) error = res.fatal
        else {
          const r = (res.results || []).find((x: { messageId: number }) => x.messageId === c.messageId)
          ok = !!r?.ok; error = r ? (r.errors.join('; ') || undefined) : 'no result for this message in the feed report'
        }
        await updateChange(f, c.id, ok ? { changeStatus: 'confirmed', confirmedAt: now(), errorMsg: '' } : { changeStatus: 'error', errorMsg: error || 'feed processing failed' })
        if (ok) confirmed++
        const arr = perListing.get(c.listingId!) || []; arr.push({ change: c, ok, error }); perListing.set(c.listingId!, arr)
      }
      for (const [lid, items] of perListing) { const l = listings.find((x: ListingRow) => x.id === lid); if (l) await applyConfirmed(f, l, items) }
      log(`feed ${feedId}: ${confirmed} confirmed, ${changes.length - confirmed} failed`)
    } catch (e: any) { logErr(`feed ${feedId} poll failed: ${e?.message || e}`) }
  }
  return { polled, confirmed }
}

/** Patch deliveries: compare a few `sent` rows against the live listing (Amazon reflects within 1–5 min). */
export const spotCheckSent = async (f: IdempiereFetch, orderSourceId?: number, max = SPOT_CHECK_MAX): Promise<number> => {
  const sent = (await listChanges(f, { status: ['sent'], deliveryMode: 'patch', ...(orderSourceId ? { orderSourceId } : {}) }, 500))
    .filter(c => ageMs(c.sentAt || c.updated) > SPOT_CHECK_AFTER_MS)
  if (!sent.length) return 0
  const byListing = new Map<number, ChangeRow[]>()
  for (const c of sent) { const arr = byListing.get(c.listingId!) || []; arr.push(c); byListing.set(c.listingId!, arr) }
  const listingIds = [...byListing.keys()].slice(0, max)
  const listings = await getListingsByIds(f, listingIds)
  const connCache = new Map<number, any>()
  let confirmed = 0
  for (const l of listings) {
    const changes = byListing.get(l.id) || []
    try {
      let conn = connCache.get(l.orderSourceId!)
      if (!conn) {
        const os = await loadOrderSource(f, l.orderSourceId!)
        if (!os || marketplaceKeyOf(os) !== 'amazon') continue
        conn = await amazonConnectionFromOrderSource(os); connCache.set(l.orderSourceId!, conn)
      }
      await acquire(`amazon:read:${conn.sellerId}`, LISTINGS_READ_BUCKET)
      const live = await getListingsItem(conn.accessToken, conn.sellerId, l.sellerSku, conn.marketplaceId).catch((e: any) => { if (Number(e?.status) === 404) return null; throw e })
      const results: { change: ChangeRow; ok: boolean; error?: string }[] = []
      for (const c of changes) {
        let ok: boolean | null = null
        const want = num(c.newValue)
        if (c.changeType === 'update' && c.fieldName === 'price') ok = live?.price != null && want != null ? Math.abs(live.price - want) < PRICE_TOL : null
        else if (c.changeType === 'update' && c.fieldName === 'qty') ok = live?.quantity != null && want != null ? Math.trunc(live.quantity) === Math.trunc(want) : null
        else if (c.changeType === 'delist') ok = live?.quantity != null ? live.quantity === 0 : null
        else if (c.changeType === 'relist') ok = live?.quantity != null ? live.quantity > 0 : null
        else if (c.changeType === 'create') ok = live ? true : null
        else if (c.changeType === 'delete') ok = live ? null : true
        else ok = true   // min/max/sale cannot be read back → trust ACCEPTED
        if (ok === true) { await updateChange(f, c.id, { changeStatus: 'confirmed', confirmedAt: now(), errorMsg: '' }); results.push({ change: c, ok: true }); confirmed++ }
        else if (ageMs(c.sentAt || c.updated) > 24 * 60 * 60 * 1000) { await updateChange(f, c.id, { changeStatus: 'error', errorMsg: 'not applied by Amazon within 24 h' }); results.push({ change: c, ok: false, error: 'not applied by Amazon within 24 h' }) }
      }
      if (results.length) {
        await applyConfirmed(f, l, results)
        if (live) await updateListing(f, l.id, { remotePrice: live.price, remoteQty: live.quantity, remoteStatus: live.status || null, lastSyncedAt: now(), ...(live.productType && !l.productType ? { productType: live.productType } : {}) })
      }
    } catch (e: any) { logErr(`spot-check listing ${l.id} failed: ${e?.message || e}`) }
  }
  return confirmed
}

export const flushOutbox = async (opts: FlushOptions): Promise<FlushSummary> => {
  const startedAt = Date.now()
  const runId = `${new Date(startedAt).toISOString().replace(/[-:T]/g, '').slice(0, 12)}-${opts.trigger}`
  const summary: FlushSummary = { runId, trigger: opts.trigger, startedAt: new Date(startedAt).toISOString(), durationMs: 0, sources: [], feedsPolled: 0, confirmed: 0, spotChecked: 0 }
  if (running) { summary.skipped = 'a flush is already running'; return summary }
  running = true
  const deadline = startedAt + (opts.maxRunMs || 10 * 60 * 1000)
  const timeLeft = () => Date.now() < deadline
  try {
    const f = opts.fetch || serviceFetch()
    if (!(await listingTablesAvailable(f))) { summary.skipped = 'listing tables not available'; return summary }
    await recoverStaleClaims(f)

    const queued = (await listChanges(f, { status: ['queued'], ...(opts.orderSourceId ? { orderSourceId: opts.orderSourceId } : {}) }, 2000))
      .filter(c => opts.ignoreDebounce || opts.trigger === 'manual' || ageMs(c.created) >= DEBOUNCE_MS)
    const bySource = new Map<number, ChangeRow[]>()
    for (const c of queued) { if (!c.orderSourceId || !c.listingId) continue; const arr = bySource.get(c.orderSourceId) || []; arr.push(c); bySource.set(c.orderSourceId, arr) }

    for (const [osId, changes] of bySource) {
      if (!timeLeft()) break
      const os = await loadOrderSource(f, osId)
      const src: FlushSourceSummary = { orderSourceId: osId, name: os?.Name || String(osId), marketplace: os ? marketplaceKeyOf(os) : 'unknown', claimed: 0, sent: 0, errors: 0, requeued: 0, mode: 'none' }
      summary.sources.push(src)
      if (!os) { src.skipped = 'order source not found'; continue }
      const adapter = getListingAdapter(os)
      if (!adapter || !supportsListings(os)) { src.skipped = `no listings adapter for ${src.marketplace}`; continue }
      const ctx = makeCtx(f, os)
      try {
        const byListing = new Map<number, ChangeRow[]>()
        for (const c of changes) { const arr = byListing.get(c.listingId!) || []; arr.push(c); byListing.set(c.listingId!, arr) }
        const listings = await getListingsByIds(f, [...byListing.keys()])
        const items: PushItem[] = []
        for (const [lid, cs] of byListing) {
          const l = listings.find(x => x.id === lid)
          if (!l) { for (const c of cs) await updateChange(f, c.id, { changeStatus: 'error', errorMsg: 'listing no longer exists' }); src.errors += cs.length; continue }
          const { item, superseded } = coalesce(l, cs)
          for (const id of superseded) await updateChange(f, id, { changeStatus: 'superseded' })
          for (const c of item.changes) await updateChange(f, c.id, { changeStatus: 'sending', batchId: runId })
          await updateListing(f, l.id, { syncStatus: 'sending' })
          src.claimed += item.changes.length
          items.push(item)
        }
        if (!items.length) continue
        // one feed still processing for this source → no second feed, per-item calls instead
        const openFeed = (await listChanges(f, { orderSourceId: osId, status: ['sent'], deliveryMode: 'feed' }, 1)).length > 0
        const results: PushResult[] = await adapter.push(ctx, items, { allowFeed: !openFeed })
        src.mode = results.some(r => r.deliveryMode === 'feed') ? 'feed' : 'patch'
        src.feedId = results.find(r => r.feedId)?.feedId || null
        for (const r of results) {
          const item = items.find(i => i.listing.id === r.listingId)!
          const lPatch: Partial<ListingRow> = {}
          if (r.productType && !item.listing.productType) lPatch.productType = r.productType
          if (r.remote) {
            if (r.remote.listingKey) lPatch.listingKey = r.remote.listingKey
            if (r.remote.externalProductId !== undefined) lPatch.externalProductId = r.remote.externalProductId
            if (r.remote.externalInventoryId !== undefined) lPatch.externalInventoryId = r.remote.externalInventoryId
            if (r.remote.meta !== undefined) lPatch.meta = r.remote.meta
          }
          if (r.status === 'sent' && r.confirmed) {
            // synchronous marketplace (Shopify): applied already → confirmed, listing settled now
            const warn = r.warnings?.length ? `warnings: ${r.warnings.join('; ')}`.slice(0, 2000) : ''
            for (const c of item.changes) await updateChange(f, c.id, { changeStatus: 'confirmed', deliveryMode: r.deliveryMode, submissionId: r.submissionId || null, sentAt: now(), confirmedAt: now(), attemptCount: c.attemptCount + 1, errorMsg: warn })
            if (Object.keys(lPatch).length) await updateListing(f, item.listing.id, lPatch)
            await applyConfirmed(f, { ...item.listing, ...lPatch }, item.changes.map(c => ({ change: c, ok: true })))
            src.sent += item.changes.length
            summary.confirmed += item.changes.length
          } else if (r.status === 'sent') {
            for (const c of item.changes) await updateChange(f, c.id, { changeStatus: 'sent', deliveryMode: r.deliveryMode, submissionId: r.submissionId || null, feedId: r.feedId || null, messageId: r.messageId ?? null, sentAt: now(), attemptCount: c.attemptCount + 1, errorMsg: r.warnings?.length ? `warnings: ${r.warnings.join('; ')}`.slice(0, 2000) : '' })
            await updateListing(f, item.listing.id, { ...lPatch, syncStatus: 'sent', syncError: '' })
            src.sent += item.changes.length
          } else {
            const retry = !!r.transient && item.changes.every(c => c.attemptCount + 1 < MAX_ATTEMPTS)
            for (const c of item.changes) await updateChange(f, c.id, { changeStatus: retry ? 'queued' : 'error', attemptCount: c.attemptCount + 1, errorMsg: r.error || 'unknown error' })
            await updateListing(f, item.listing.id, { ...lPatch, syncStatus: retry ? 'queued' : 'error', syncError: r.error || 'unknown error' })
            if (retry) src.requeued += item.changes.length; else src.errors += item.changes.length
          }
        }
        ctx.log(`claimed ${src.claimed}, sent ${src.sent}, errors ${src.errors}, requeued ${src.requeued}, mode ${src.mode}`)
      } catch (e: any) {
        // whole-source failure (credentials, LWA, network): release the claims
        logErr(`[${src.name}] flush failed: ${e?.message || e}`)
        src.skipped = String(e?.message || e).slice(0, 300)
        for (const c of changes) { try { await updateChange(f, c.id, { changeStatus: 'queued', errorMsg: String(e?.message || e).slice(0, 2000) }) } catch {} }
        for (const lid of new Set(changes.map(c => c.listingId!))) { try { await updateListing(f, lid, { syncStatus: 'queued', syncError: String(e?.message || e).slice(0, 2000) }) } catch {} }
      }
    }

    if (timeLeft()) { const fp = await pollOpenFeeds(f, opts.orderSourceId); summary.feedsPolled = fp.polled; summary.confirmed += fp.confirmed }
    if (timeLeft()) { const sc = await spotCheckSent(f, opts.orderSourceId); summary.spotChecked = sc; summary.confirmed += sc }
  } catch (e: any) {
    summary.error = String(e?.message || e)
    logErr(`run ${runId} failed: ${summary.error}`)
  } finally {
    running = false
    summary.durationMs = Date.now() - startedAt
    lastRun = summary
  }
  return summary
}