import { and, eq, getTableColumns, inArray, isNull, lt, notInArray, sql, type SQL } from 'drizzle-orm'
import type { PgTable } from 'drizzle-orm/pg-core'
import type { Db } from './db.js'
import {
  attributeSetAttributes, attributeSets, attributes, auditLog, categories, categoryStoreData, inventorySalable, inventorySourceItems,
  productCategories, productImages, productLinks, productRelations, products, productStoreData, stores, syncRuns, syncState,
} from './schema.js'

/** `SET col = excluded.col` for every column except the conflict keys. */
function excludedSet<T extends PgTable>(table: T, keys: string[]): Record<string, SQL> {
  const set: Record<string, SQL> = {}
  for (const [prop, col] of Object.entries(getTableColumns(table))) {
    if (keys.includes(prop)) continue
    set[prop] = sql.raw(`excluded."${col.name}"`)
  }
  return set
}

function chunk<T>(rows: T[], size = 500): T[][] {
  const out: T[][] = []
  for (let i = 0; i < rows.length; i += size) out.push(rows.slice(i, i + size))
  return out
}

// ── Record shapes produced by apps/sync mappers ────────────────────────────

export type StoreRow = typeof stores.$inferInsert
export type AttributeRow = typeof attributes.$inferInsert
export type CategoryRow = typeof categories.$inferInsert
export type CategoryStoreRow = typeof categoryStoreData.$inferInsert
export type ProductRow = typeof products.$inferInsert
export type ProductStoreRow = typeof productStoreData.$inferInsert
export type SourceItemRow = typeof inventorySourceItems.$inferInsert

/** A product as read from Magento's default (`all`) scope: global data + store-0 content + children. */
export interface ProductRecord {
  product: ProductRow
  storeData: ProductStoreRow
  relations: Array<{ childSku: string; kind: 'configurable' | 'bundle' | 'grouped' }>
  links: Array<{ linkType: string; linkedSku: string; position: number }>
  categories: Array<{ categoryId: number; position: number }>
  images: Array<{ file: string; label: string | null; position: number; roles: string[]; disabled: boolean; mediaType: string }>
}

// ── Reference data ─────────────────────────────────────────────────────────

export async function upsertStores(db: Db, rows: StoreRow[]) {
  if (!rows.length) return
  await db.insert(stores).values(rows).onConflictDoUpdate({ target: stores.storeId, set: excludedSet(stores, ['storeId']) })
}

export async function upsertAttributes(db: Db, rows: AttributeRow[]) {
  for (const batch of chunk(rows)) {
    await db.insert(attributes).values(batch).onConflictDoUpdate({ target: attributes.attributeCode, set: excludedSet(attributes, ['attributeCode']) })
  }
}

export async function replaceAttributeSet(db: Db, set: { attributeSetId: number; name: string }, members: Array<{ attributeCode: string; isRequired: boolean }>) {
  await db.transaction(async tx => {
    await tx.insert(attributeSets).values(set).onConflictDoUpdate({ target: attributeSets.attributeSetId, set: { name: set.name, syncedAt: sql`now()` } })
    await tx.delete(attributeSetAttributes).where(eq(attributeSetAttributes.attributeSetId, set.attributeSetId))
    if (members.length) await tx.insert(attributeSetAttributes).values(members.map(m => ({ ...m, attributeSetId: set.attributeSetId })))
  })
}

export async function upsertCategories(db: Db, rows: CategoryRow[], storeRows: CategoryStoreRow[]) {
  await db.transaction(async tx => {
    for (const batch of chunk(rows)) {
      await tx.insert(categories).values(batch).onConflictDoUpdate({ target: categories.categoryId, set: excludedSet(categories, ['categoryId']) })
    }
    const known = new Set(rows.map(r => r.categoryId))
    for (const batch of chunk(storeRows.filter(r => known.has(r.categoryId)))) {
      await tx
        .insert(categoryStoreData)
        .values(batch)
        .onConflictDoUpdate({ target: [categoryStoreData.categoryId, categoryStoreData.storeId], set: excludedSet(categoryStoreData, ['categoryId', 'storeId']) })
    }
  })
}

/** Remove categories Magento no longer returns (only call with the complete id list). */
export async function deleteCategoriesNotIn(db: Db, categoryIds: number[]) {
  if (!categoryIds.length) return 0
  const res = await db.delete(categories).where(notInArray(categories.categoryId, categoryIds)).returning({ id: categories.categoryId })
  return res.length
}

// ── Products ───────────────────────────────────────────────────────────────

/** Upsert products from the default scope, replacing their child collections. */
export async function upsertProducts(db: Db, records: ProductRecord[]) {
  if (!records.length) return
  await db.transaction(async tx => {
    const ids = records.map(r => r.product.productId)

    // A SKU can be re-created in Magento with a new entity_id; drop the stale row first.
    await tx.delete(products).where(and(inArray(products.sku, records.map(r => r.product.sku)), notInArray(products.productId, ids)))

    await tx
      .insert(products)
      .values(records.map(r => ({ ...r.product, deletedAt: null })))
      .onConflictDoUpdate({ target: products.productId, set: excludedSet(products, ['productId']) })

    await tx
      .insert(productStoreData)
      .values(records.map(r => r.storeData))
      .onConflictDoUpdate({ target: [productStoreData.productId, productStoreData.storeId], set: excludedSet(productStoreData, ['productId', 'storeId']) })

    await tx.delete(productRelations).where(inArray(productRelations.parentId, ids))
    await tx.delete(productLinks).where(inArray(productLinks.productId, ids))
    await tx.delete(productCategories).where(inArray(productCategories.productId, ids))
    await tx.delete(productImages).where(inArray(productImages.productId, ids))

    const rel = records.flatMap(r => r.relations.map(x => ({ ...x, parentId: r.product.productId })))
    const lnk = records.flatMap(r => r.links.map(x => ({ ...x, productId: r.product.productId })))
    const cat = records.flatMap(r => r.categories.map(x => ({ ...x, productId: r.product.productId })))
    const img = records.flatMap(r => r.images.map(x => ({ ...x, productId: r.product.productId })))
    for (const b of chunk(rel)) await tx.insert(productRelations).values(b).onConflictDoNothing()
    for (const b of chunk(lnk)) await tx.insert(productLinks).values(b).onConflictDoNothing()
    for (const b of chunk(cat)) await tx.insert(productCategories).values(b).onConflictDoNothing()
    for (const b of chunk(img)) await tx.insert(productImages).values(b).onConflictDoNothing()
  })
}

/** Upsert store-view content for products that already exist (default scope is synced first). */
export async function upsertProductStoreData(db: Db, rows: ProductStoreRow[]) {
  if (!rows.length) return 0
  const existing = await db
    .select({ id: products.productId })
    .from(products)
    .where(inArray(products.productId, rows.map(r => r.productId)))
  const known = new Set(existing.map(e => e.id))
  const keep = rows.filter(r => known.has(r.productId))
  for (const batch of chunk(keep)) {
    await db
      .insert(productStoreData)
      .values(batch)
      .onConflictDoUpdate({ target: [productStoreData.productId, productStoreData.storeId], set: excludedSet(productStoreData, ['productId', 'storeId']) })
  }
  return keep.length
}

/** Soft-delete live products whose id is not in `seenIds` (the complete set from Magento). */
export async function softDeleteProductsNotIn(db: Db, seenIds: number[]) {
  if (!seenIds.length) return 0 // never wipe the catalog because Magento returned nothing
  const res = await db
    .update(products)
    .set({ deletedAt: sql`now()` })
    .where(and(isNull(products.deletedAt), sql`${products.productId} <> ALL(${sql.param(seenIds)}::int[])`))
    .returning({ id: products.productId })
  return res.length
}

// ── Inventory ──────────────────────────────────────────────────────────────

export async function upsertSourceItems(db: Db, rows: SourceItemRow[]) {
  for (const batch of chunk(rows, 1000)) {
    await db
      .insert(inventorySourceItems)
      .values(batch)
      .onConflictDoUpdate({ target: [inventorySourceItems.sku, inventorySourceItems.sourceCode], set: excludedSet(inventorySourceItems, ['sku', 'sourceCode']) })
  }
}

/** After a full inventory sweep, drop source rows not touched since `sweepStartedAt`. */
export async function deleteStaleSourceItems(db: Db, sweepStartedAt: Date) {
  const res = await db.delete(inventorySourceItems).where(lt(inventorySourceItems.syncedAt, sweepStartedAt)).returning({ sku: inventorySourceItems.sku })
  return res.length
}

export async function upsertSalable(db: Db, row: typeof inventorySalable.$inferInsert) {
  await db
    .insert(inventorySalable)
    .values({ ...row, checkedAt: new Date() })
    .onConflictDoUpdate({ target: [inventorySalable.sku, inventorySalable.stockId], set: excludedSet(inventorySalable, ['sku', 'stockId']) })
}

// ── Sync bookkeeping ───────────────────────────────────────────────────────

export async function getCursor(db: Db, stream: string): Promise<string | null> {
  const [row] = await db.select({ cursor: syncState.cursor }).from(syncState).where(eq(syncState.stream, stream))
  return row?.cursor ?? null
}

export async function setCursor(db: Db, stream: string, cursor: string | null) {
  await db
    .insert(syncState)
    .values({ stream, cursor, lastSuccessAt: new Date(), updatedAt: new Date() })
    .onConflictDoUpdate({ target: syncState.stream, set: { cursor, lastSuccessAt: sql`now()`, updatedAt: sql`now()` } })
}

/** Record a job run around `fn`; the stats object it returns is stored on the run. */
export async function recordRun<T extends Record<string, number>>(db: Db, job: string, fn: () => Promise<T>): Promise<T> {
  const [run] = await db.insert(syncRuns).values({ job }).returning({ id: syncRuns.id })
  try {
    const stats = await fn()
    await db.update(syncRuns).set({ status: 'succeeded', finishedAt: sql`now()`, stats }).where(eq(syncRuns.id, run!.id))
    return stats
  } catch (err) {
    await db
      .update(syncRuns)
      .set({ status: 'failed', finishedAt: sql`now()`, error: (err as Error).message.slice(0, 2000) })
      .where(eq(syncRuns.id, run!.id))
    throw err
  }
}

export async function softDeleteSku(db: Db, sku: string) {
  const res = await db.update(products).set({ deletedAt: sql`now()` }).where(and(eq(products.sku, sku), isNull(products.deletedAt))).returning({ id: products.productId })
  return res.length
}

export async function skusByProductIds(db: Db, ids: number[]): Promise<Map<number, string>> {
  if (!ids.length) return new Map()
  const rows = await db.select({ id: products.productId, sku: products.sku }).from(products).where(inArray(products.productId, ids))
  return new Map(rows.map(r => [r.id, r.sku]))
}

export async function writeAudit(db: Db, row: typeof auditLog.$inferInsert) {
  await db.insert(auditLog).values(row)
}

/** True when a run of `job` started in the last `withinMinutes` and has not finished (crashed runs age out). */
export async function isRunning(db: Db, job: string, withinMinutes = 90) {
  const [row] = await db
    .select({ id: syncRuns.id })
    .from(syncRuns)
    .where(and(eq(syncRuns.job, job), eq(syncRuns.status, 'running'), sql`${syncRuns.startedAt} > now() - make_interval(mins => ${withinMinutes})`))
    .limit(1)
  return Boolean(row)
}