import * as XLSX from '@e965/xlsx';
import {
  AssetStatus,
  BookStatus,
  ImportItemStatus,
  ImportRunStatus,
  Prisma,
} from '@prisma/client';
import crypto from 'crypto';
import fs from 'fs/promises';
import path from 'path';
import sanitizeHtml from 'sanitize-html';
import prisma from '../config/database';
import {
  asText,
  cleanText,
  imageUrlsFromCell,
  lookupKey,
  normalizeFormat,
  operationalSku,
  operationalSlug,
  parseAffirmative,
  parseEnabled,
  parseMoney,
  parseNonNegativeInteger,
  parseOptionalInteger,
  parsePercent,
  slugify,
  sourceKey,
} from '../utils/catalog.utils';

const EXPECTED_SHEETS = [
  'Products',
  'Inventory',
  'Categories',
  'Sub Categories',
  'Authors',
  'Publishers',
  'Product Text',
  'Custom Attributes',
  'Homepage Categories',
  'Header Categories',
  'Currencies',
  'Display Limits',
  'Catalogues',
  'Reconciliation',
] as const;

const EXPECTED_COUNTS = {
  Products: 3083,
  Categories: 47,
  Authors: 1072,
  Publishers: 300,
  'Custom Attributes': 23166,
  Catalogues: 43,
} as const;

type RowRecord = Record<string, string> & { __rowNumber: string };

export interface CatalogImportOptions {
  workbookPath: string;
  dryRun?: boolean;
  resume?: boolean;
  assetMode?: 'download';
  concurrency?: number;
  batchSize?: number;
  logsDir?: string;
  storageDir?: string;
  assetPublicBaseUrl?: string;
  catalogueMapPath?: string;
}

interface PreflightData {
  sheetRows: Record<string, number>;
  skuCounts: Map<string, number>;
  isbnCounts: Map<string, number>;
  baseSlugCounts: Map<string, number>;
  productIds: Set<string>;
  categoryTokens: Set<string>;
  categoryNames: Set<string>;
  authorValues: Set<string>;
  productAuthorValues: Set<string>;
  publisherValues: Set<string>;
  productPublisherValues: Set<string>;
  attributeProductIds: Set<string>;
  productTextProductIds: Set<string>;
  missingFields: Record<string, number>;
  invalidPriceRows: number[];
  invalidStockRows: number[];
  discountAbovePriceRows: number[];
  imageUrlCount: number;
  productsWithImages: number;
  productTextRows: number;
  cataloguePdfUrls: Set<string>;
  nonContiguousAttributeGroups: number;
  nonContiguousTextGroups: number;
}

interface ImportSummary {
  sourceWorkbook: string;
  sourceFingerprint: string;
  dryRun: boolean;
  startedAt: string;
  completedAt?: string;
  executionTimeMs?: number;
  runId?: string;
  sheetRows: Record<string, number>;
  imported: Record<string, number>;
  updated: Record<string, number>;
  failed: Record<string, number>;
  skipped: Record<string, number>;
  warnings: Record<string, number>;
  duplicates: Record<string, number>;
  missing: Record<string, number>;
  verification?: Record<string, unknown>;
  logPath: string;
  errorLogPath: string;
}

const increment = (map: Map<string, number>, key: string): void => {
  map.set(key, (map.get(key) ?? 0) + 1);
};

const duplicateGroupCount = (map: Map<string, number>): number =>
  [...map.values()].filter((count) => count > 1).length;

const rowNumber = (row: RowRecord): number => Number(row.__rowNumber);

const rowValue = (row: RowRecord, column: string): string => row[column] ?? '';

const countRecord = (record: Record<string, number>, key: string, amount = 1): void => {
  record[key] = (record[key] ?? 0) + amount;
};

class ImportLogger {
  readonly logPath: string;
  readonly errorLogPath: string;
  readonly summaryPath: string;
  private writeQueue: Promise<void> = Promise.resolve();

  constructor(logsDir: string, sourceFingerprint: string) {
    const timestamp = new Date().toISOString().replace(/[:.]/g, '-');
    const prefix = `catalog-import-${timestamp}-${sourceFingerprint.slice(0, 10)}`;
    this.logPath = path.join(logsDir, `${prefix}.jsonl`);
    this.errorLogPath = path.join(logsDir, `${prefix}.errors.jsonl`);
    this.summaryPath = path.join(logsDir, `${prefix}.summary.json`);
  }

  async initialize(): Promise<void> {
    await fs.mkdir(path.dirname(this.logPath), { recursive: true });
    // A successful run must still deliver an explicit (possibly empty) error log.
    await Promise.all([
      fs.writeFile(this.logPath, '', 'utf8'),
      fs.writeFile(this.errorLogPath, '', 'utf8'),
    ]);
  }

  private enqueue(filePath: string, payload: Record<string, unknown>): Promise<void> {
    const line = `${JSON.stringify({ timestamp: new Date().toISOString(), ...payload })}\n`;
    this.writeQueue = this.writeQueue.then(() => fs.appendFile(filePath, line, 'utf8'));
    return this.writeQueue;
  }

  event(event: string, payload: Record<string, unknown> = {}): Promise<void> {
    return this.enqueue(this.logPath, { event, ...payload });
  }

  error(event: string, payload: Record<string, unknown> = {}): Promise<void> {
    void this.enqueue(this.logPath, { event, level: 'error', ...payload });
    return this.enqueue(this.errorLogPath, { event, level: 'error', ...payload });
  }

  async writeSummary(summary: ImportSummary): Promise<void> {
    await this.writeQueue;
    await fs.writeFile(this.summaryPath, `${JSON.stringify(summary, null, 2)}\n`, 'utf8');
  }

  async flush(): Promise<void> {
    await this.writeQueue;
  }
}

const workbookFingerprint = async (workbookPath: string): Promise<string> => {
  const data = await fs.readFile(workbookPath);
  return crypto.createHash('sha256').update(data).digest('hex');
};

const recordFromArray = (headers: string[], values: unknown[], sourceRow: number): RowRecord => {
  const record: RowRecord = { __rowNumber: String(sourceRow) };
  headers.forEach((header, index) => {
    record[header] = asText(values[index]);
  });
  return record;
};

const streamWorkbook = async (
  workbookPath: string,
  onRow: (sheetName: string, row: RowRecord) => Promise<void> | void,
  onSheet?: (sheetName: string) => Promise<void> | void
): Promise<void> => {
  // SheetJS is used for compatibility with the verified workbook's OPC entry
  // ordering. The file is only 4.8 MB; rows are released sheet-by-sheet and all
  // database writes are still processed in bounded chunks.
  const workbook = XLSX.readFile(workbookPath, {
    cellDates: false,
    cellFormula: true,
    cellText: true,
    dense: true,
  });

  for (const sheetName of workbook.SheetNames) {
    await onSheet?.(sheetName);
    const worksheet = workbook.Sheets[sheetName];
    const rows = XLSX.utils.sheet_to_json<unknown[]>(worksheet, {
      header: 1,
      raw: false,
      defval: '',
      blankrows: true,
    });
    const headerRowIndex = rows.findIndex((row) => row.some((value) => cleanText(value)));
    if (headerRowIndex < 0) continue;
    const headers = rows[headerRowIndex].map((value) => cleanText(value));
    for (let rowIndex = headerRowIndex + 1; rowIndex < rows.length; rowIndex += 1) {
      const values = rows[rowIndex];
      if (!values.some((value) => cleanText(value))) continue;
      await onRow(sheetName, recordFromArray(headers, values, rowIndex + 1));
    }
  }
};

const preflightWorkbook = async (
  workbookPath: string,
  logger: ImportLogger
): Promise<PreflightData> => {
  const data: PreflightData = {
    sheetRows: {},
    skuCounts: new Map(),
    isbnCounts: new Map(),
    baseSlugCounts: new Map(),
    productIds: new Set(),
    categoryTokens: new Set(),
    categoryNames: new Set(),
    authorValues: new Set(),
    productAuthorValues: new Set(),
    publisherValues: new Set(),
    productPublisherValues: new Set(),
    attributeProductIds: new Set(),
    productTextProductIds: new Set(),
    missingFields: {},
    invalidPriceRows: [],
    invalidStockRows: [],
    discountAbovePriceRows: [],
    imageUrlCount: 0,
    productsWithImages: 0,
    productTextRows: 0,
    cataloguePdfUrls: new Set(),
    nonContiguousAttributeGroups: 0,
    nonContiguousTextGroups: 0,
  };

  let lastAttributeProductId = '';
  let lastTextProductId = '';
  const completedAttributeProducts = new Set<string>();
  const completedTextProducts = new Set<string>();

  await streamWorkbook(workbookPath, (sheetName, row) => {
    data.sheetRows[sheetName] = (data.sheetRows[sheetName] ?? 0) + 1;

    if (sheetName === 'Products') {
      const productId = cleanText(rowValue(row, 'Product ID'));
      const sku = asText(rowValue(row, 'SKU / ISBN Code')).trim();
      const isbn = asText(rowValue(row, 'ISBN')).trim();
      const sourceSlug = rowValue(row, 'SEO Slug');
      const title = rowValue(row, 'Book Name');
      const baseSlug = slugify(sourceSlug) || slugify(title) || 'book';
      const price = parseMoney(rowValue(row, 'Price'));
      const stock = parseNonNegativeInteger(rowValue(row, 'Stock Quantity'));
      const discount = parseMoney(rowValue(row, 'Discount Price'));
      const imageUrls = imageUrlsFromCell(rowValue(row, 'Product Image URLs'));

      data.productIds.add(productId);
      increment(data.skuCounts, sku);
      increment(data.isbnCounts, isbn);
      increment(data.baseSlugCounts, baseSlug);
      if (price.warning) data.invalidPriceRows.push(rowNumber(row));
      if (stock.warning) data.invalidStockRows.push(rowNumber(row));
      if (!discount.warning && !price.warning && discount.value > price.value) {
        data.discountAbovePriceRows.push(rowNumber(row));
      }

      data.imageUrlCount += imageUrls.length;
      if (imageUrls.length > 0) data.productsWithImages += 1;

      for (const column of [
        'Product ID',
        'Book Name',
        'ISBN',
        'SKU / ISBN Code',
        'Categories (Normalized)',
        'SEO Slug',
        'Price',
        'Discount Price',
        'Stock Quantity',
        'Status',
        'Product Image URLs',
        'Author',
        'Publisher',
        'Description (Plain Text Preview)',
      ]) {
        if (!cleanText(rowValue(row, column))) countRecord(data.missingFields, column);
      }

      for (const categoryName of rowValue(row, 'Categories (Normalized)').split('|')) {
        if (cleanText(categoryName)) data.categoryTokens.add(lookupKey(categoryName));
      }
      if (asText(rowValue(row, 'Author'))) data.productAuthorValues.add(asText(rowValue(row, 'Author')));
      if (asText(rowValue(row, 'Publisher'))) {
        data.productPublisherValues.add(asText(rowValue(row, 'Publisher')));
      }
    } else if (sheetName === 'Categories') {
      data.categoryNames.add(lookupKey(rowValue(row, 'Category Name')));
    } else if (sheetName === 'Authors') {
      data.authorValues.add(asText(rowValue(row, 'Author Value (as stored)')));
    } else if (sheetName === 'Publishers') {
      data.publisherValues.add(asText(rowValue(row, 'Publisher Value (as stored)')));
    } else if (sheetName === 'Product Text') {
      const productId = cleanText(rowValue(row, 'Product ID'));
      data.productTextRows += 1;
      data.productTextProductIds.add(productId);
      if (lastTextProductId && lastTextProductId !== productId) completedTextProducts.add(lastTextProductId);
      if (completedTextProducts.has(productId)) data.nonContiguousTextGroups += 1;
      lastTextProductId = productId;
    } else if (sheetName === 'Custom Attributes') {
      const productId = cleanText(rowValue(row, 'Product ID'));
      data.attributeProductIds.add(productId);
      if (lastAttributeProductId && lastAttributeProductId !== productId) {
        completedAttributeProducts.add(lastAttributeProductId);
      }
      if (completedAttributeProducts.has(productId)) data.nonContiguousAttributeGroups += 1;
      lastAttributeProductId = productId;
    } else if (sheetName === 'Catalogues') {
      const pdfUrl = cleanText(rowValue(row, 'PDF URL'));
      if (pdfUrl) data.cataloguePdfUrls.add(pdfUrl);
    }
  }, (sheetName) => {
    data.sheetRows[sheetName] = data.sheetRows[sheetName] ?? 0;
  });

  const missingSheets = EXPECTED_SHEETS.filter((sheetName) => !(sheetName in data.sheetRows));
  if (missingSheets.length > 0) {
    throw new Error(`Workbook is missing required sheets: ${missingSheets.join(', ')}`);
  }

  for (const [sheetName, expected] of Object.entries(EXPECTED_COUNTS)) {
    const actual = data.sheetRows[sheetName] ?? 0;
    if (actual !== expected) {
      throw new Error(`Sheet ${sheetName} has ${actual} rows; expected ${expected}.`);
    }
  }

  const missingCategoryNames = [...data.categoryTokens].filter((name) => !data.categoryNames.has(name));
  const missingAuthorValues = [...data.productAuthorValues].filter((name) => !data.authorValues.has(name));
  const missingPublisherValues = [...data.productPublisherValues].filter(
    (name) => !data.publisherValues.has(name)
  );

  if (missingCategoryNames.length || missingAuthorValues.length || missingPublisherValues.length) {
    throw new Error(
      `Workbook relationship validation failed: ${missingCategoryNames.length} categories, ` +
        `${missingAuthorValues.length} authors, and ${missingPublisherValues.length} publishers are absent from their masters.`
    );
  }

  if (data.nonContiguousAttributeGroups || data.nonContiguousTextGroups) {
    throw new Error(
      'Product Text or Custom Attributes rows are not grouped by Product ID; safe streaming import cannot continue.'
    );
  }

  await logger.event('preflight.completed', {
    sheetRows: data.sheetRows,
    duplicateIsbnGroups: duplicateGroupCount(data.isbnCounts),
    duplicateSkuGroups: duplicateGroupCount(data.skuCounts),
    duplicateSlugGroups: duplicateGroupCount(data.baseSlugCounts),
    missingFields: data.missingFields,
    invalidPriceRows: data.invalidPriceRows,
    invalidStockRows: data.invalidStockRows,
    discountAbovePriceRows: data.discountAbovePriceRows,
    imageUrlCount: data.imageUrlCount,
    productsWithImages: data.productsWithImages,
  });

  return data;
};

interface MasterRows {
  categories: RowRecord[];
  authors: RowRecord[];
  publishers: RowRecord[];
  homepageCategories: RowRecord[];
  currencies: RowRecord[];
  displayLimits: RowRecord[];
  catalogues: RowRecord[];
}

const collectMasterRows = async (workbookPath: string): Promise<MasterRows> => {
  const masters: MasterRows = {
    categories: [],
    authors: [],
    publishers: [],
    homepageCategories: [],
    currencies: [],
    displayLimits: [],
    catalogues: [],
  };

  await streamWorkbook(workbookPath, (sheetName, row) => {
    if (sheetName === 'Categories') masters.categories.push(row);
    if (sheetName === 'Authors') masters.authors.push(row);
    if (sheetName === 'Publishers') masters.publishers.push(row);
    if (sheetName === 'Homepage Categories') masters.homepageCategories.push(row);
    if (sheetName === 'Currencies') masters.currencies.push(row);
    if (sheetName === 'Display Limits') masters.displayLimits.push(row);
    if (sheetName === 'Catalogues') masters.catalogues.push(row);
  });

  return masters;
};

const asyncPool = async <T>(
  items: T[],
  concurrency: number,
  worker: (item: T) => Promise<void>
): Promise<void> => {
  let cursor = 0;
  const workers = Array.from({ length: Math.max(1, concurrency) }, async () => {
    while (cursor < items.length) {
      const index = cursor;
      cursor += 1;
      await worker(items[index]);
    }
  });
  await Promise.all(workers);
};

const checkpointRows = async (
  runId: string,
  sheetName: string,
  rows: RowRecord[],
  status: ImportItemStatus,
  message?: string
): Promise<void> => {
  if (rows.length === 0) return;
  await prisma.catalogImportItem.createMany({
    data: rows.map((row) => ({
      runId,
      sheetName,
      sourceRow: rowNumber(row),
      externalKey:
        cleanText(rowValue(row, 'Product ID')) ||
        cleanText(rowValue(row, 'Category ID')) ||
        cleanText(rowValue(row, 'Catalogue ID')) ||
        null,
      status,
      message,
    })),
    skipDuplicates: true,
  });
};

const importMasters = async (
  workbookPath: string,
  runId: string,
  summary: ImportSummary,
  logger: ImportLogger,
  concurrency: number
): Promise<void> => {
  const masters = await collectMasterRows(workbookPath);

  await asyncPool(masters.categories, concurrency, async (row) => {
    const sourceCategoryId = cleanText(rowValue(row, 'Category ID'));
    const name = asText(rowValue(row, 'Category Name'));
    try {
      await prisma.category.upsert({
        where: { sourceCategoryId },
        create: {
          sourceCategoryId,
          name,
          slug: slugify(name) || `category-${sourceCategoryId}`,
          isActive: parseEnabled(rowValue(row, 'Status')),
          displayOrder: rowNumber(row) - 1,
        },
        update: {
          name,
          slug: slugify(name) || `category-${sourceCategoryId}`,
          isActive: parseEnabled(rowValue(row, 'Status')),
          displayOrder: rowNumber(row) - 1,
        },
      });
      countRecord(summary.imported, 'Categories');
      await checkpointRows(runId, 'Categories', [row], ImportItemStatus.IMPORTED);
    } catch (error) {
      countRecord(summary.failed, 'Categories');
      await logger.error('category.failed', {
        sourceRow: rowNumber(row),
        sourceCategoryId,
        error: error instanceof Error ? error.message : String(error),
      });
      await checkpointRows(
        runId,
        'Categories',
        [row],
        ImportItemStatus.FAILED,
        error instanceof Error ? error.message : String(error)
      );
    }
  });

  const authorData = masters.authors.map((row) => {
    const name = asText(rowValue(row, 'Author Value (as stored)'));
    const key = sourceKey('author', name);
    return {
      name,
      sourceKey: key,
      slug: `${slugify(name) || 'author'}-${key.slice(0, 10)}`,
    };
  });
  try {
    const result = await prisma.author.createMany({ data: authorData, skipDuplicates: true });
    countRecord(summary.imported, 'Authors', result.count);
    countRecord(summary.updated, 'Authors', masters.authors.length - result.count);
    await checkpointRows(runId, 'Authors', masters.authors, ImportItemStatus.IMPORTED);
  } catch (error) {
    await logger.error('authors.batch.failed', {
      error: error instanceof Error ? error.message : String(error),
    });
    await asyncPool(masters.authors, concurrency, async (row) => {
      const name = asText(rowValue(row, 'Author Value (as stored)'));
      const key = sourceKey('author', name);
      try {
        await prisma.author.upsert({
          where: { sourceKey: key },
          create: { name, sourceKey: key, slug: `${slugify(name) || 'author'}-${key.slice(0, 10)}` },
          update: { name },
        });
        countRecord(summary.imported, 'Authors');
        await checkpointRows(runId, 'Authors', [row], ImportItemStatus.IMPORTED);
      } catch (rowError) {
        countRecord(summary.failed, 'Authors');
        await logger.error('author.failed', {
          sourceRow: rowNumber(row),
          error: rowError instanceof Error ? rowError.message : String(rowError),
        });
      }
    });
  }

  const publisherData = masters.publishers.map((row) => {
    const name = asText(rowValue(row, 'Publisher Value (as stored)'));
    const key = sourceKey('publisher', name);
    return {
      name,
      sourceKey: key,
      slug: `${slugify(name) || 'publisher'}-${key.slice(0, 10)}`,
    };
  });
  try {
    const result = await prisma.publisher.createMany({ data: publisherData, skipDuplicates: true });
    countRecord(summary.imported, 'Publishers', result.count);
    countRecord(summary.updated, 'Publishers', masters.publishers.length - result.count);
    await checkpointRows(runId, 'Publishers', masters.publishers, ImportItemStatus.IMPORTED);
  } catch (error) {
    await logger.error('publishers.batch.failed', {
      error: error instanceof Error ? error.message : String(error),
    });
    await asyncPool(masters.publishers, concurrency, async (row) => {
      const name = asText(rowValue(row, 'Publisher Value (as stored)'));
      const key = sourceKey('publisher', name);
      try {
        await prisma.publisher.upsert({
          where: { sourceKey: key },
          create: {
            name,
            sourceKey: key,
            slug: `${slugify(name) || 'publisher'}-${key.slice(0, 10)}`,
          },
          update: { name },
        });
        countRecord(summary.imported, 'Publishers');
        await checkpointRows(runId, 'Publishers', [row], ImportItemStatus.IMPORTED);
      } catch (rowError) {
        countRecord(summary.failed, 'Publishers');
        await logger.error('publisher.failed', {
          sourceRow: rowNumber(row),
          error: rowError instanceof Error ? rowError.message : String(rowError),
        });
      }
    });
  }

  for (const row of masters.homepageCategories) {
    const sourceHomepageCategoryId = cleanText(rowValue(row, 'Category ID'));
    const name = asText(rowValue(row, 'Category Name'));
    await prisma.homepageCategory.upsert({
      where: { sourceHomepageCategoryId },
      create: {
        sourceHomepageCategoryId,
        name,
        slug: `${slugify(name) || 'homepage'}-${sourceHomepageCategoryId}`,
        isActive: parseEnabled(rowValue(row, 'Status')),
        adminSourceUrl: cleanText(rowValue(row, 'Admin Source URL')) || null,
      },
      update: {
        name,
        isActive: parseEnabled(rowValue(row, 'Status')),
        adminSourceUrl: cleanText(rowValue(row, 'Admin Source URL')) || null,
      },
    });
  }
  await checkpointRows(
    runId,
    'Homepage Categories',
    masters.homepageCategories,
    ImportItemStatus.IMPORTED
  );
  countRecord(summary.imported, 'Homepage Categories', masters.homepageCategories.length);

  for (const row of masters.currencies) {
    const sourceCurrencyId = cleanText(rowValue(row, 'Currency ID'));
    const exchangeRate = parseMoney(rowValue(row, 'Currency Value'), 1);
    await prisma.storeCurrency.upsert({
      where: { sourceCurrencyId },
      create: {
        sourceCurrencyId,
        name: asText(rowValue(row, 'Name')),
        currency: asText(rowValue(row, 'Currency')),
        exchangeRate: exchangeRate.value,
        isActive: parseEnabled(rowValue(row, 'Status')),
        adminSourceUrl: cleanText(rowValue(row, 'Admin Source URL')) || null,
      },
      update: {
        name: asText(rowValue(row, 'Name')),
        currency: asText(rowValue(row, 'Currency')),
        exchangeRate: exchangeRate.value,
        isActive: parseEnabled(rowValue(row, 'Status')),
        adminSourceUrl: cleanText(rowValue(row, 'Admin Source URL')) || null,
      },
    });
  }
  await checkpointRows(runId, 'Currencies', masters.currencies, ImportItemStatus.IMPORTED);
  countRecord(summary.imported, 'Currencies', masters.currencies.length);

  for (const row of masters.displayLimits) {
    const sourceDisplayLimitId = cleanText(rowValue(row, 'Limit ID'));
    const productsPerPage = parseNonNegativeInteger(rowValue(row, 'Products Per Page'), 20).value;
    await prisma.displayLimit.upsert({
      where: { sourceDisplayLimitId },
      create: {
        sourceDisplayLimitId,
        productsPerPage,
        isActive: parseEnabled(rowValue(row, 'Status')),
        adminSourceUrl: cleanText(rowValue(row, 'Admin Source URL')) || null,
      },
      update: {
        productsPerPage,
        isActive: parseEnabled(rowValue(row, 'Status')),
        adminSourceUrl: cleanText(rowValue(row, 'Admin Source URL')) || null,
      },
    });
  }
  await checkpointRows(runId, 'Display Limits', masters.displayLimits, ImportItemStatus.IMPORTED);
  countRecord(summary.imported, 'Display Limits', masters.displayLimits.length);

  await asyncPool(masters.catalogues, concurrency, async (row) => {
    const sourceCatalogueId = cleanText(rowValue(row, 'Catalogue ID'));
    const title = asText(rowValue(row, 'Title'));
    const sourcePdfUrl = cleanText(rowValue(row, 'PDF URL'));
    try {
      const existing = await prisma.catalogue.findUnique({ where: { sourceCatalogueId } });
      await prisma.catalogue.upsert({
        where: { sourceCatalogueId },
        create: {
          sourceCatalogueId,
          title,
          slug: `${slugify(title) || 'catalogue'}-${sourceCatalogueId}`,
          pdfUrl: sourcePdfUrl,
          sourcePdfUrl,
          adminSourceUrl: cleanText(rowValue(row, 'Admin Source URL')) || null,
          isActive: parseEnabled(rowValue(row, 'Status')),
          assetStatus: AssetStatus.LINKED,
        },
        update: {
          title,
          sourcePdfUrl,
          pdfUrl:
            existing?.assetStatus === AssetStatus.DOWNLOADED && existing.sourcePdfUrl === sourcePdfUrl
              ? existing.pdfUrl
              : sourcePdfUrl,
          adminSourceUrl: cleanText(rowValue(row, 'Admin Source URL')) || null,
          isActive: parseEnabled(rowValue(row, 'Status')),
          assetStatus:
            existing?.assetStatus === AssetStatus.DOWNLOADED && existing.sourcePdfUrl === sourcePdfUrl
              ? AssetStatus.DOWNLOADED
              : AssetStatus.LINKED,
        },
      });
      countRecord(summary.imported, 'Catalogues');
      await checkpointRows(runId, 'Catalogues', [row], ImportItemStatus.IMPORTED);
    } catch (error) {
      countRecord(summary.failed, 'Catalogues');
      await logger.error('catalogue.failed', {
        sourceRow: rowNumber(row),
        sourceCatalogueId,
        error: error instanceof Error ? error.message : String(error),
      });
    }
  });

  await logger.event('masters.completed', {
    categories: masters.categories.length,
    authors: masters.authors.length,
    publishers: masters.publishers.length,
    catalogues: masters.catalogues.length,
  });
};

interface MasterMaps {
  categoryByName: Map<string, string>;
  authorBySourceKey: Map<string, string>;
  publisherBySourceKey: Map<string, string>;
}

const loadMasterMaps = async (): Promise<MasterMaps> => {
  const [categories, authors, publishers] = await Promise.all([
    prisma.category.findMany({
      where: { sourceCategoryId: { not: null } },
      select: { id: true, name: true },
    }),
    prisma.author.findMany({
      where: { sourceKey: { not: null } },
      select: { id: true, sourceKey: true },
    }),
    prisma.publisher.findMany({
      where: { sourceKey: { not: null } },
      select: { id: true, sourceKey: true },
    }),
  ]);

  return {
    categoryByName: new Map(categories.map((category) => [lookupKey(category.name), category.id])),
    authorBySourceKey: new Map(
      authors.filter((author) => author.sourceKey).map((author) => [author.sourceKey as string, author.id])
    ),
    publisherBySourceKey: new Map(
      publishers
        .filter((publisher) => publisher.sourceKey)
        .map((publisher) => [publisher.sourceKey as string, publisher.id])
    ),
  };
};

interface ProductWorkItem {
  row: RowRecord;
  slug: string;
  sku: string;
}

const processProduct = async (
  item: ProductWorkItem,
  runId: string,
  masters: MasterMaps,
  summary: ImportSummary,
  logger: ImportLogger
): Promise<void> => {
  const { row, slug, sku } = item;
  const sourceRow = rowNumber(row);
  const sourceProductId = cleanText(rowValue(row, 'Product ID'));
  const title = asText(rowValue(row, 'Book Name'));

  if (!sourceProductId || !title) {
    const message = 'Product ID and Book Name are required.';
    countRecord(summary.failed, 'Products');
    await logger.error('product.failed', { sourceRow, sourceProductId, message });
    await prisma.catalogImportItem.upsert({
      where: { runId_sheetName_sourceRow: { runId, sheetName: 'Products', sourceRow } },
      create: {
        runId,
        sheetName: 'Products',
        sourceRow,
        externalKey: sourceProductId || null,
        status: ImportItemStatus.FAILED,
        message,
      },
      update: { status: ImportItemStatus.FAILED, message, processedAt: new Date() },
    });
    return;
  }

  const priceResult = parseMoney(rowValue(row, 'Price'));
  const discountRaw = cleanText(rowValue(row, 'Discount Price'));
  const discountResult = discountRaw ? parseMoney(discountRaw) : null;
  const stockResult = parseNonNegativeInteger(rowValue(row, 'Stock Quantity'));
  const gstResult = parsePercent(rowValue(row, 'GST'));
  const warnings = [
    priceResult.warning,
    discountResult?.warning,
    stockResult.warning,
    gstResult.warning,
  ].filter(Boolean) as string[];

  const authorRaw = asText(rowValue(row, 'Author'));
  const publisherRaw = asText(rowValue(row, 'Publisher'));
  const authorId = authorRaw ? masters.authorBySourceKey.get(sourceKey('author', authorRaw)) : undefined;
  const publisherId = publisherRaw
    ? masters.publisherBySourceKey.get(sourceKey('publisher', publisherRaw))
    : undefined;
  const categoryNames = rowValue(row, 'Categories (Normalized)')
    .split('|')
    .map((value) => cleanText(value))
    .filter(Boolean);
  const categoryIds = [
    ...new Set(categoryNames.map((name) => masters.categoryByName.get(lookupKey(name))).filter(Boolean)),
  ] as string[];
  const imageUrls = imageUrlsFromCell(rowValue(row, 'Product Image URLs'));
  const description = asText(rowValue(row, 'Description (Plain Text Preview)'));
  const sourceLanguage = asText(rowValue(row, 'Language'));
  const status = parseEnabled(rowValue(row, 'Status')) ? BookStatus.ENABLED : BookStatus.DISABLED;

  if (authorRaw && !authorId) warnings.push(`Author master record not found for "${authorRaw}".`);
  if (publisherRaw && !publisherId) warnings.push(`Publisher master record not found for "${publisherRaw}".`);
  if (categoryIds.length !== categoryNames.length) {
    warnings.push('One or more category relationships could not be resolved.');
  }

  try {
    const result = await prisma.$transaction(
      async (tx) => {
        const existing = await tx.book.findUnique({
          where: { sourceProductId },
          include: { images: { orderBy: { sortOrder: 'asc' } } },
        });
        const existingImageSources = existing?.images.map((image) => image.sourceUrl ?? image.url) ?? [];
        const imagesUnchanged =
          existingImageSources.length === imageUrls.length &&
          existingImageSources.every((sourceUrl, index) => sourceUrl === imageUrls[index]);

        const bookData: Prisma.BookUncheckedCreateInput = {
          sourceProductId,
          adminSerial: parseOptionalInteger(rowValue(row, 'Admin Serial')),
          title,
          slug,
          sourceSlug: asText(rowValue(row, 'SEO Slug')) || null,
          isbn: asText(rowValue(row, 'ISBN')),
          sku,
          sourceSku: asText(rowValue(row, 'SKU / ISBN Code')),
          barcode: asText(rowValue(row, 'Barcode')) || null,
          description,
          additionalAttributesText:
            asText(rowValue(row, 'Additional Attributes (Plain Text)')) || null,
          price: priceResult.value,
          discountPrice: discountResult?.warning ? null : discountResult?.value ?? null,
          gstRate: gstResult.value,
          stock: stockResult.value,
          format: normalizeFormat(rowValue(row, 'Binding')),
          binding: asText(rowValue(row, 'Binding')) || null,
          language: cleanText(sourceLanguage) || 'UNKNOWN',
          sourceLanguage: sourceLanguage || null,
          edition: asText(rowValue(row, 'Edition')) || null,
          classGrade: asText(rowValue(row, 'Class / Grade')) || null,
          publicationYear: cleanText(rowValue(row, 'Year')).slice(0, 64) || null,
          pageCount: parseOptionalInteger(rowValue(row, 'Pages')),
          pagesText: asText(rowValue(row, 'Pages')) || null,
          publisherId: publisherId ?? null,
          coverImage: imageUrls[0] ?? null,
          tagsRaw: asText(rowValue(row, 'Tags (Raw)')) || null,
          sourceCategoryIds: asText(rowValue(row, 'Category IDs')) || null,
          homepageCategoriesRaw: asText(rowValue(row, 'Homepage Categories')) || null,
          homepageCategoryIdsRaw: asText(rowValue(row, 'Homepage Category IDs')) || null,
          sourceInStock: asText(rowValue(row, 'In Stock')) || null,
          sourceNewRelease: asText(rowValue(row, 'New Release')) || null,
          publicProductUrl: asText(rowValue(row, 'Public Product Page')) || null,
          adminEditUrl: asText(rowValue(row, 'Admin Edit URL')) || null,
          adminImagesUrl: asText(rowValue(row, 'Admin Images URL')) || null,
          detailSource: asText(rowValue(row, 'Detail Source')) || null,
          status,
          isActive: status === BookStatus.ENABLED,
          displayOrder:
            parseOptionalInteger(rowValue(row, 'Admin Serial')) ?? Math.max(0, sourceRow - 2),
          isNewRelease: parseAffirmative(rowValue(row, 'New Release')),
          isFeatured: Boolean(cleanText(rowValue(row, 'Homepage Categories'))),
        };

        let bookId: string;
        if (existing) {
          const { coverImage: _coverImage, ...updateDataWithoutCover } = bookData;
          const updated = await tx.book.update({
            where: { id: existing.id },
            data: imagesUnchanged ? updateDataWithoutCover : bookData,
          });
          bookId = updated.id;
        } else {
          const created = await tx.book.create({ data: bookData });
          bookId = created.id;
        }

        await tx.bookInventory.upsert({
          where: { bookId },
          create: { bookId, stock: stockResult.value, reserved: 0 },
          update: { stock: stockResult.value },
        });

        await tx.bookAuthor.deleteMany({ where: { bookId } });
        if (authorId) await tx.bookAuthor.create({ data: { bookId, authorId } });

        await tx.bookCategory.deleteMany({ where: { bookId } });
        if (categoryIds.length > 0) {
          await tx.bookCategory.createMany({
            data: categoryIds.map((categoryId) => ({ bookId, categoryId })),
            skipDuplicates: true,
          });
        }

        if (!imagesUnchanged) {
          await tx.bookImage.deleteMany({ where: { bookId } });
          if (imageUrls.length > 0) {
            await tx.bookImage.createMany({
              data: imageUrls.map((url, index) => ({
                bookId,
                url,
                sourceUrl: url,
                altText: title,
                sortOrder: index,
                isPrimary: index === 0,
                assetStatus: AssetStatus.LINKED,
              })),
            });
          }
        }

        const itemStatus = existing ? ImportItemStatus.UPDATED : ImportItemStatus.IMPORTED;
        await tx.catalogImportItem.upsert({
          where: { runId_sheetName_sourceRow: { runId, sheetName: 'Products', sourceRow } },
          create: {
            runId,
            sheetName: 'Products',
            sourceRow,
            externalKey: sourceProductId,
            status: itemStatus,
            message: warnings.length ? warnings.join(' ') : null,
          },
          update: {
            externalKey: sourceProductId,
            status: itemStatus,
            message: warnings.length ? warnings.join(' ') : null,
            processedAt: new Date(),
          },
        });

        return { existed: Boolean(existing) };
      },
      { timeout: 30000 }
    );

    countRecord(result.existed ? summary.updated : summary.imported, 'Products');
    if (warnings.length) {
      countRecord(summary.warnings, 'Product validation', warnings.length);
      await logger.event('product.warning', { sourceRow, sourceProductId, warnings });
    }
  } catch (error) {
    const message = error instanceof Error ? error.message : String(error);
    countRecord(summary.failed, 'Products');
    await logger.error('product.failed', { sourceRow, sourceProductId, message });
    await prisma.catalogImportItem.upsert({
      where: { runId_sheetName_sourceRow: { runId, sheetName: 'Products', sourceRow } },
      create: {
        runId,
        sheetName: 'Products',
        sourceRow,
        externalKey: sourceProductId,
        status: ImportItemStatus.FAILED,
        message,
      },
      update: { status: ImportItemStatus.FAILED, message, processedAt: new Date() },
    });
  }
};

const importProducts = async (
  workbookPath: string,
  preflight: PreflightData,
  runId: string,
  summary: ImportSummary,
  logger: ImportLogger,
  concurrency: number,
  batchSize: number
): Promise<void> => {
  const masters = await loadMasterMaps();
  const existingBooks = await prisma.book.findMany({
    select: { sourceProductId: true, slug: true, sku: true },
  });
  const slugOwners = new Map(
    existingBooks.filter((book) => book.slug).map((book) => [book.slug, book.sourceProductId ?? book.slug])
  );
  const skuOwners = new Map(
    existingBooks
      .filter((book) => book.sku)
      .map((book) => [book.sku as string, book.sourceProductId ?? (book.sku as string)])
  );
  const completedRows = new Set(
    (
      await prisma.catalogImportItem.findMany({
        where: {
          runId,
          sheetName: 'Products',
          status: { in: [ImportItemStatus.IMPORTED, ImportItemStatus.UPDATED, ImportItemStatus.SKIPPED] },
        },
        select: { sourceRow: true },
      })
    ).map((item) => item.sourceRow)
  );

  let batch: ProductWorkItem[] = [];
  const flushBatch = async (): Promise<void> => {
    const current = batch;
    batch = [];
    await asyncPool(current, concurrency, (item) =>
      processProduct(item, runId, masters, summary, logger)
    );
  };

  await streamWorkbook(workbookPath, async (sheetName, row) => {
    if (sheetName !== 'Products') return;
    const sourceRow = rowNumber(row);
    if (completedRows.has(sourceRow)) {
      countRecord(summary.skipped, 'Products');
      return;
    }

    const sourceProductId = cleanText(rowValue(row, 'Product ID'));
    const rawSku = asText(rowValue(row, 'SKU / ISBN Code')).trim();
    const baseSlug =
      slugify(rowValue(row, 'SEO Slug')) || slugify(rowValue(row, 'Book Name')) || 'book';
    let slug = operationalSlug(
      rowValue(row, 'SEO Slug'),
      rowValue(row, 'Book Name'),
      sourceProductId,
      preflight.baseSlugCounts.get(baseSlug) ?? 1
    );
    if (slugOwners.has(slug) && slugOwners.get(slug) !== sourceProductId) {
      slug = operationalSlug(slug, rowValue(row, 'Book Name'), sourceProductId, 2);
    }
    while (slugOwners.has(slug) && slugOwners.get(slug) !== sourceProductId) {
      slug = `${slug.slice(0, 177).replace(/-+$/g, '')}-${sourceKey('slug', `${slug}:${sourceProductId}`).slice(0, 12)}`;
    }
    slugOwners.set(slug, sourceProductId);

    let sku = operationalSku(rawSku, sourceProductId, preflight.skuCounts.get(rawSku) ?? 1);
    if (skuOwners.has(sku) && skuOwners.get(sku) !== sourceProductId) {
      sku = operationalSku(rawSku, sourceProductId, 2);
    }
    while (skuOwners.has(sku) && skuOwners.get(sku) !== sourceProductId) {
      const suffix = `::${sourceKey('sku', `${sku}:${sourceProductId}`).slice(0, 12)}`;
      sku = `${sku.slice(0, Math.max(1, 191 - suffix.length))}${suffix}`;
    }
    skuOwners.set(sku, sourceProductId);

    batch.push({ row, slug, sku });
    if (batch.length >= batchSize) await flushBatch();
  });
  await flushBatch();

  await logger.event('products.completed', {
    imported: summary.imported.Products ?? 0,
    updated: summary.updated.Products ?? 0,
    skipped: summary.skipped.Products ?? 0,
    failed: summary.failed.Products ?? 0,
  });
};

const streamGroupedSheet = async (
  workbookPath: string,
  targetSheet: string,
  onGroup: (productId: string, rows: RowRecord[]) => Promise<void>
): Promise<void> => {
  let currentProductId = '';
  let currentRows: RowRecord[] = [];

  const flush = async (): Promise<void> => {
    if (!currentProductId || currentRows.length === 0) return;
    const rows = currentRows;
    const productId = currentProductId;
    currentRows = [];
    await onGroup(productId, rows);
  };

  await streamWorkbook(workbookPath, async (sheetName, row) => {
    if (sheetName !== targetSheet) return;
    const productId = cleanText(rowValue(row, 'Product ID'));
    if (currentProductId && productId !== currentProductId) await flush();
    currentProductId = productId;
    currentRows.push(row);
  });
  await flush();
};

const safeCatalogHtml = (html: string): string =>
  sanitizeHtml(html, {
    allowedTags: [
      'p',
      'br',
      'strong',
      'b',
      'em',
      'i',
      'u',
      'ul',
      'ol',
      'li',
      'table',
      'thead',
      'tbody',
      'tfoot',
      'tr',
      'th',
      'td',
      'blockquote',
      'h1',
      'h2',
      'h3',
      'h4',
      'h5',
      'h6',
      'span',
      'div',
      'a',
    ],
    allowedAttributes: {
      a: ['href', 'title', 'target', 'rel'],
      table: ['border', 'cellpadding', 'cellspacing'],
      td: ['colspan', 'rowspan'],
      th: ['colspan', 'rowspan'],
      '*': ['style'],
    },
    allowedStyles: {
      '*': {
        'text-align': [/^left$/, /^right$/, /^center$/, /^justify$/],
        'font-weight': [/^bold$/, /^\d{3}$/],
        'font-style': [/^italic$/],
        'text-decoration': [/^underline$/],
      },
    },
    allowedSchemes: ['http', 'https', 'mailto'],
    transformTags: {
      a: sanitizeHtml.simpleTransform('a', { rel: 'noopener noreferrer' }, true),
    },
  });

const plainTextFromHtml = (html: string): string =>
  cleanText(sanitizeHtml(html, { allowedTags: [], allowedAttributes: {} }));

const importProductText = async (
  workbookPath: string,
  preflight: PreflightData,
  runId: string,
  summary: ImportSummary,
  logger: ImportLogger
): Promise<void> => {
  const books = await prisma.book.findMany({
    where: { sourceProductId: { not: null } },
    select: { id: true, sourceProductId: true },
  });
  const bookIds = new Map(
    books.filter((book) => book.sourceProductId).map((book) => [book.sourceProductId as string, book.id])
  );

  await streamGroupedSheet(workbookPath, 'Product Text', async (productId, rows) => {
    const bookId = bookIds.get(productId);
    if (!bookId) {
      countRecord(summary.failed, 'Product Text', rows.length);
      await logger.error('product_text.orphan', {
        productId,
        sourceRows: rows.map(rowNumber),
      });
      return;
    }

    const byField = new Map<string, RowRecord[]>();
    for (const row of rows) {
      const field = cleanText(rowValue(row, 'Field'));
      if (!byField.has(field)) byField.set(field, []);
      byField.get(field)?.push(row);
    }
    const descriptionHtmlOriginal = (byField.get('Description HTML') ?? [])
      .sort((a, b) => Number(rowValue(a, 'Part Number')) - Number(rowValue(b, 'Part Number')))
      .map((row) => asText(rowValue(row, 'Original Content')))
      .join('');
    const attributesHtmlOriginal = (byField.get('Additional Attributes HTML') ?? [])
      .sort((a, b) => Number(rowValue(a, 'Part Number')) - Number(rowValue(b, 'Part Number')))
      .map((row) => asText(rowValue(row, 'Original Content')))
      .join('');

    try {
      await prisma.$transaction(
        async (tx) => {
          await tx.bookText.deleteMany({ where: { bookId } });
          await tx.bookText.createMany({
            data: rows.map((row) => ({
              bookId,
              field: cleanText(rowValue(row, 'Field')),
              partNumber: parseNonNegativeInteger(rowValue(row, 'Part Number'), 1).value,
              originalContent: asText(rowValue(row, 'Original Content')),
              sourceRow: rowNumber(row),
            })),
          });

          const book = await tx.book.findUnique({ where: { id: bookId }, select: { description: true } });
          const derivedDescription = descriptionHtmlOriginal
            ? plainTextFromHtml(descriptionHtmlOriginal)
            : '';
          await tx.book.update({
            where: { id: bookId },
            data: {
              descriptionHtml: descriptionHtmlOriginal ? safeCatalogHtml(descriptionHtmlOriginal) : null,
              additionalAttributesHtml: attributesHtmlOriginal
                ? safeCatalogHtml(attributesHtmlOriginal)
                : null,
              description: book?.description || derivedDescription,
            },
          });

          await tx.catalogImportItem.deleteMany({
            where: {
              runId,
              sheetName: 'Product Text',
              sourceRow: { in: rows.map(rowNumber) },
            },
          });
          await tx.catalogImportItem.createMany({
            data: rows.map((row) => ({
              runId,
              sheetName: 'Product Text',
              sourceRow: rowNumber(row),
              externalKey: productId,
              status: ImportItemStatus.IMPORTED,
            })),
          });
        },
        { timeout: 30000 }
      );
      countRecord(summary.imported, 'Product Text', rows.length);
    } catch (error) {
      countRecord(summary.failed, 'Product Text', rows.length);
      await logger.error('product_text.failed', {
        productId,
        sourceRows: rows.map(rowNumber),
        error: error instanceof Error ? error.message : String(error),
      });
    }
  });

  const booksWithoutText = books.filter(
    (book) => book.sourceProductId && !preflight.productTextProductIds.has(book.sourceProductId)
  );
  await asyncPool(booksWithoutText, 4, async (book) => {
    await prisma.$transaction([
      prisma.bookText.deleteMany({ where: { bookId: book.id } }),
      prisma.book.update({
        where: { id: book.id },
        data: { descriptionHtml: null, additionalAttributesHtml: null },
      }),
    ]);
  });

  await logger.event('product_text.completed', {
    imported: summary.imported['Product Text'] ?? 0,
    failed: summary.failed['Product Text'] ?? 0,
  });
};

const importCustomAttributes = async (
  workbookPath: string,
  preflight: PreflightData,
  runId: string,
  summary: ImportSummary,
  logger: ImportLogger
): Promise<void> => {
  const books = await prisma.book.findMany({
    where: { sourceProductId: { not: null } },
    select: { id: true, sourceProductId: true },
  });
  const bookIds = new Map(
    books.filter((book) => book.sourceProductId).map((book) => [book.sourceProductId as string, book.id])
  );

  await streamGroupedSheet(workbookPath, 'Custom Attributes', async (productId, rows) => {
    const bookId = bookIds.get(productId);
    if (!bookId) {
      countRecord(summary.failed, 'Custom Attributes', rows.length);
      await logger.error('custom_attributes.orphan', {
        productId,
        sourceRows: rows.map(rowNumber),
      });
      return;
    }

    try {
      await prisma.$transaction(
        async (tx) => {
          await tx.bookAttribute.deleteMany({ where: { bookId } });
          await tx.bookAttribute.createMany({
            data: rows.map((row, index) => ({
              bookId,
              name: asText(rowValue(row, 'Attribute Name')),
              value: asText(rowValue(row, 'Attribute Value')),
              sourceRow: rowNumber(row),
              sortOrder: index,
            })),
          });
          await tx.catalogImportItem.deleteMany({
            where: {
              runId,
              sheetName: 'Custom Attributes',
              sourceRow: { in: rows.map(rowNumber) },
            },
          });
          await tx.catalogImportItem.createMany({
            data: rows.map((row) => ({
              runId,
              sheetName: 'Custom Attributes',
              sourceRow: rowNumber(row),
              externalKey: productId,
              status: ImportItemStatus.IMPORTED,
            })),
          });
        },
        { timeout: 30000 }
      );
      countRecord(summary.imported, 'Custom Attributes', rows.length);
    } catch (error) {
      countRecord(summary.failed, 'Custom Attributes', rows.length);
      await logger.error('custom_attributes.failed', {
        productId,
        sourceRows: rows.map(rowNumber),
        error: error instanceof Error ? error.message : String(error),
      });
    }
  });

  const booksWithoutAttributes = books.filter(
    (book) => book.sourceProductId && !preflight.attributeProductIds.has(book.sourceProductId)
  );
  await asyncPool(booksWithoutAttributes, 4, async (book) => {
    await prisma.bookAttribute.deleteMany({ where: { bookId: book.id } });
  });

  await logger.event('custom_attributes.completed', {
    imported: summary.imported['Custom Attributes'] ?? 0,
    failed: summary.failed['Custom Attributes'] ?? 0,
  });
};

const extensionForImage = (contentType: string, url: string, bytes: Uint8Array): string => {
  const normalizedType = contentType.toLocaleLowerCase('en');
  if (normalizedType.includes('png') || (bytes[0] === 0x89 && bytes[1] === 0x50)) return '.png';
  if (normalizedType.includes('webp') || String.fromCharCode(...bytes.slice(0, 4)) === 'RIFF') {
    return '.webp';
  }
  if (normalizedType.includes('gif') || String.fromCharCode(...bytes.slice(0, 3)) === 'GIF') return '.gif';
  if (normalizedType.includes('avif')) return '.avif';
  if (normalizedType.includes('jpeg') || normalizedType.includes('jpg') || (bytes[0] === 0xff && bytes[1] === 0xd8)) {
    return '.jpg';
  }
  const pathnameExtension = path.extname(new URL(url).pathname).toLocaleLowerCase('en');
  if (['.jpg', '.jpeg', '.png', '.gif', '.webp', '.avif'].includes(pathnameExtension)) {
    return pathnameExtension === '.jpeg' ? '.jpg' : pathnameExtension;
  }
  return '.img';
};

const downloadAsset = async (
  url: string,
  targetWithoutExtension: string,
  kind: 'image' | 'pdf'
): Promise<{ absolutePath: string; contentType: string }> => {
  const response = await fetch(url, {
    redirect: 'follow',
    signal: AbortSignal.timeout(45000),
    headers: { 'user-agent': 'IndianBooksWorldwide-CatalogImporter/1.0' },
  });
  if (!response.ok) throw new Error(`HTTP ${response.status} ${response.statusText}`);
  const contentType = response.headers.get('content-type') ?? '';
  const bytes = new Uint8Array(await response.arrayBuffer());
  if (bytes.length === 0) throw new Error('Downloaded asset is empty.');

  let extension: string;
  if (kind === 'pdf') {
    const signature = String.fromCharCode(...bytes.slice(0, 5));
    if (signature !== '%PDF-') throw new Error(`Downloaded file is not a valid PDF (${contentType}).`);
    extension = '.pdf';
  } else {
    extension = extensionForImage(contentType, url, bytes);
    if (extension === '.img' && !contentType.toLocaleLowerCase('en').startsWith('image/')) {
      throw new Error(`Downloaded file is not a recognized image (${contentType}).`);
    }
  }

  const absolutePath = `${targetWithoutExtension}${extension}`;
  await fs.mkdir(path.dirname(absolutePath), { recursive: true });
  await fs.writeFile(absolutePath, bytes);
  return { absolutePath, contentType };
};

const publicAssetUrl = (baseUrl: string, storageKey: string): string =>
  `${baseUrl.replace(/\/$/, '')}/${storageKey
    .split('/')
    .map((segment) => encodeURIComponent(segment))
    .join('/')}`;

const importAssets = async (
  options: Required<Pick<CatalogImportOptions, 'assetMode' | 'storageDir' | 'assetPublicBaseUrl' | 'concurrency'>>,
  summary: ImportSummary,
  logger: ImportLogger
): Promise<void> => {
  const [images, catalogues] = await Promise.all([
    prisma.bookImage.findMany({
      where: { book: { sourceProductId: { not: null } } },
      include: { book: { select: { id: true, sourceProductId: true } } },
      orderBy: [{ bookId: 'asc' }, { sortOrder: 'asc' }],
    }),
    prisma.catalogue.findMany({ orderBy: { sourceCatalogueId: 'asc' } }),
  ]);

  const placeholderStorageKey = 'books/placeholder.jpg';
  const placeholderPath = path.join(options.storageDir, ...placeholderStorageKey.split('/'));
  await fs.access(placeholderPath);
  const placeholderUrl = publicAssetUrl(options.assetPublicBaseUrl, placeholderStorageKey);

  await asyncPool(images, options.concurrency, async (image) => {
    const sourceUrl = image.sourceUrl ?? image.url;
    const sourceProductId = image.book.sourceProductId as string;
    const existingStoragePath = image.storageKey
      ? path.join(options.storageDir, ...image.storageKey.split('/'))
      : null;
    if (image.assetStatus === AssetStatus.DOWNLOADED && existingStoragePath) {
      try {
        await fs.access(existingStoragePath);
        countRecord(summary.skipped, 'Images Downloaded');
        return;
      } catch {
        // Re-download a missing local file.
      }
    }

    try {
      const basePath = path.join(
        options.storageDir,
        'books',
        sourceProductId,
        String(image.sortOrder + 1)
      );
      const downloaded = await downloadAsset(sourceUrl, basePath, 'image');
      const storageKey = path.relative(options.storageDir, downloaded.absolutePath).split(path.sep).join('/');
      const url = publicAssetUrl(options.assetPublicBaseUrl, storageKey);
      await prisma.$transaction([
        prisma.bookImage.update({
          where: { id: image.id },
          data: { url, storageKey, assetStatus: AssetStatus.DOWNLOADED },
        }),
        ...(image.isPrimary
          ? [prisma.book.update({ where: { id: image.book.id }, data: { coverImage: url } })]
          : []),
      ]);
      countRecord(summary.imported, 'Images Downloaded');
    } catch (error) {
      countRecord(summary.failed, 'Images Downloaded');
      await prisma.$transaction([
        prisma.bookImage.update({
          where: { id: image.id },
          data: {
            url: placeholderUrl,
            storageKey: placeholderStorageKey,
            assetStatus: AssetStatus.MISSING,
          },
        }),
        ...(image.isPrimary
          ? [prisma.book.update({ where: { id: image.book.id }, data: { coverImage: placeholderUrl } })]
          : []),
      ]);
      await logger.error('image.download_failed', {
        sourceProductId,
        sourceUrl,
        error: error instanceof Error ? error.message : String(error),
      });
    }
  });

  await asyncPool(catalogues, options.concurrency, async (catalogue) => {
    const existingStoragePath = catalogue.storageKey
      ? path.join(options.storageDir, ...catalogue.storageKey.split('/'))
      : null;
    if (catalogue.assetStatus === AssetStatus.DOWNLOADED && existingStoragePath) {
      try {
        await fs.access(existingStoragePath);
        countRecord(summary.skipped, 'PDFs Downloaded');
        return;
      } catch {
        // Re-download a missing local file.
      }
    }

    try {
      const basePath = path.join(
        options.storageDir,
        'catalogues',
        `${catalogue.sourceCatalogueId}-${slugify(catalogue.title) || 'catalogue'}`
      );
      const downloaded = await downloadAsset(catalogue.sourcePdfUrl, basePath, 'pdf');
      const storageKey = path.relative(options.storageDir, downloaded.absolutePath).split(path.sep).join('/');
      const pdfUrl = publicAssetUrl(options.assetPublicBaseUrl, storageKey);
      await prisma.catalogue.update({
        where: { id: catalogue.id },
        data: { pdfUrl, storageKey, assetStatus: AssetStatus.DOWNLOADED },
      });
      countRecord(summary.imported, 'PDFs Downloaded');
    } catch (error) {
      countRecord(summary.failed, 'PDFs Downloaded');
      await prisma.catalogue.update({
        where: { id: catalogue.id },
        data: { assetStatus: AssetStatus.FAILED },
      });
      await logger.error('catalogue_pdf.download_failed', {
        sourceCatalogueId: catalogue.sourceCatalogueId,
        sourceUrl: catalogue.sourcePdfUrl,
        error: error instanceof Error ? error.message : String(error),
      });
    }
  });

  await logger.event('assets.download.completed', {
    imagesDownloaded: summary.imported['Images Downloaded'] ?? 0,
    imagesFailed: summary.failed['Images Downloaded'] ?? 0,
    pdfsDownloaded: summary.imported['PDFs Downloaded'] ?? 0,
    pdfsFailed: summary.failed['PDFs Downloaded'] ?? 0,
  });
};

const applyCatalogueMap = async (
  catalogueMapPath: string | undefined,
  summary: ImportSummary,
  logger: ImportLogger
): Promise<void> => {
  if (!catalogueMapPath) {
    countRecord(summary.missing, 'Explicit product-to-catalogue mapping', 1);
    await logger.event('catalogue_map.skipped', {
      reason:
        'The workbook has no product-to-catalogue or category-to-catalogue mapping. No semantic links were inferred.',
    });
    return;
  }

  const raw = await fs.readFile(path.resolve(catalogueMapPath), 'utf8');
  const mapping = JSON.parse(raw) as Record<string, string[]>;
  const [catalogues, categories] = await Promise.all([
    prisma.catalogue.findMany({ select: { id: true, sourceCatalogueId: true } }),
    prisma.category.findMany({ select: { id: true, name: true } }),
  ]);
  const catalogueBySourceId = new Map(
    catalogues.map((catalogue) => [catalogue.sourceCatalogueId, catalogue.id])
  );
  const categoryByName = new Map(categories.map((category) => [lookupKey(category.name), category.id]));

  for (const [sourceCatalogueId, categoryNames] of Object.entries(mapping)) {
    const catalogueId = catalogueBySourceId.get(sourceCatalogueId);
    if (!catalogueId) throw new Error(`Catalogue map references unknown Catalogue ID ${sourceCatalogueId}.`);
    const categoryIds = categoryNames.map((name) => {
      const categoryId = categoryByName.get(lookupKey(name));
      if (!categoryId) throw new Error(`Catalogue map references unknown category "${name}".`);
      return categoryId;
    });

    await prisma.$transaction(async (tx) => {
      await tx.categoryCatalogue.deleteMany({ where: { catalogueId } });
      await tx.bookCatalogue.deleteMany({ where: { catalogueId } });
      if (categoryIds.length === 0) return;
      await tx.categoryCatalogue.createMany({
        data: categoryIds.map((categoryId) => ({ categoryId, catalogueId })),
        skipDuplicates: true,
      });
      const bookCategories = await tx.bookCategory.findMany({
        where: { categoryId: { in: categoryIds } },
        select: { bookId: true },
        distinct: ['bookId'],
      });
      await tx.bookCatalogue.createMany({
        data: bookCategories.map(({ bookId }) => ({ bookId, catalogueId })),
        skipDuplicates: true,
      });
    });
  }

  const productLinks = await prisma.bookCatalogue.count();
  countRecord(summary.imported, 'Product Catalogue Links', productLinks);
  await logger.event('catalogue_map.completed', {
    mappedCatalogues: Object.keys(mapping).length,
    productLinks,
  });
};

const verifyImport = async (preflight: PreflightData): Promise<Record<string, unknown>> => {
  const [
    productCount,
    categoryCount,
    authorCount,
    publisherCount,
    attributeCount,
    productTextCount,
    catalogueCount,
    imageCount,
    productsWithoutCategories,
    productCatalogueLinks,
    downloadedImages,
    downloadedPdfs,
    books,
  ] = await Promise.all([
    prisma.book.count({ where: { sourceProductId: { not: null } } }),
    prisma.category.count({ where: { sourceCategoryId: { not: null } } }),
    prisma.author.count({ where: { sourceKey: { not: null } } }),
    prisma.publisher.count({ where: { sourceKey: { not: null } } }),
    prisma.bookAttribute.count({ where: { book: { sourceProductId: { not: null } } } }),
    prisma.bookText.count({ where: { book: { sourceProductId: { not: null } } } }),
    prisma.catalogue.count(),
    prisma.bookImage.count({ where: { book: { sourceProductId: { not: null } } } }),
    prisma.book.count({
      where: { sourceProductId: { not: null }, bookCategories: { none: {} } },
    }),
    prisma.bookCatalogue.count(),
    prisma.bookImage.count({ where: { assetStatus: AssetStatus.DOWNLOADED } }),
    prisma.catalogue.count({ where: { assetStatus: AssetStatus.DOWNLOADED } }),
    prisma.book.findMany({
      where: { sourceProductId: { not: null } },
      select: { sourceProductId: true, sku: true },
    }),
  ]);

  const skuCounts = new Map<string, number>();
  for (const book of books) increment(skuCounts, book.sku ?? '');
  const duplicateOperationalSkuGroups = duplicateGroupCount(skuCounts);

  return {
    products: { actual: productCount, expected: EXPECTED_COUNTS.Products, passed: productCount === 3083 },
    categories: { actual: categoryCount, expected: EXPECTED_COUNTS.Categories, passed: categoryCount === 47 },
    authors: { actual: authorCount, expected: EXPECTED_COUNTS.Authors, passed: authorCount === 1072 },
    publishers: { actual: publisherCount, expected: EXPECTED_COUNTS.Publishers, passed: publisherCount === 300 },
    customAttributes: {
      actual: attributeCount,
      expected: EXPECTED_COUNTS['Custom Attributes'],
      passed: attributeCount === 23166,
    },
    productText: {
      actual: productTextCount,
      expected: preflight.productTextRows,
      passed: productTextCount === preflight.productTextRows,
    },
    catalogues: { actual: catalogueCount, expected: EXPECTED_COUNTS.Catalogues, passed: catalogueCount === 43 },
    images: { actual: imageCount, expected: preflight.imageUrlCount, passed: imageCount === preflight.imageUrlCount },
    productsWithoutCategories,
    duplicateOperationalSkuGroups,
    productCatalogueLinks,
    downloadedImages,
    downloadedPdfs,
    sourceBlankAuthors: preflight.missingFields.Author ?? 0,
    sourceBlankPublishers: preflight.missingFields.Publisher ?? 0,
    sourceMissingImages: preflight.sheetRows.Products - preflight.productsWithImages,
  };
};

const totalCount = (record: Record<string, number>): number =>
  Object.values(record).reduce((sum, value) => sum + value, 0);

export const runCatalogImport = async (
  inputOptions: CatalogImportOptions
): Promise<ImportSummary> => {
  const startedAtMs = Date.now();
  const workbookPath = path.resolve(inputOptions.workbookPath);
  await fs.access(workbookPath);
  const sourceFingerprint = await workbookFingerprint(workbookPath);
  const logsDir = path.resolve(inputOptions.logsDir ?? path.join(process.cwd(), 'logs', 'catalog-import'));
  const storageDir = path.resolve(inputOptions.storageDir ?? path.join(process.cwd(), 'storage'));
  const logger = new ImportLogger(logsDir, sourceFingerprint);
  await logger.initialize();

  const options = {
    dryRun: inputOptions.dryRun ?? false,
    resume: inputOptions.resume ?? true,
    assetMode: inputOptions.assetMode ?? 'download',
    concurrency: Math.max(1, Math.min(12, inputOptions.concurrency ?? 4)),
    batchSize: Math.max(1, Math.min(500, inputOptions.batchSize ?? 50)),
    storageDir,
    assetPublicBaseUrl:
      inputOptions.assetPublicBaseUrl ?? process.env.ASSET_PUBLIC_BASE_URL ?? '/assets',
  } as const;

  const summary: ImportSummary = {
    sourceWorkbook: workbookPath,
    sourceFingerprint,
    dryRun: options.dryRun,
    startedAt: new Date(startedAtMs).toISOString(),
    sheetRows: {},
    imported: {},
    updated: {},
    failed: {},
    skipped: {},
    warnings: {},
    duplicates: {},
    missing: {},
    logPath: logger.logPath,
    errorLogPath: logger.errorLogPath,
  };

  let runId: string | undefined;
  try {
    await logger.event('import.started', { workbookPath, sourceFingerprint, options });
    const preflight = await preflightWorkbook(workbookPath, logger);
    summary.sheetRows = preflight.sheetRows;
    summary.duplicates = {
      'ISBN groups': duplicateGroupCount(preflight.isbnCounts),
      'Source SKU groups': duplicateGroupCount(preflight.skuCounts),
      'Source slug groups': duplicateGroupCount(preflight.baseSlugCounts),
    };
    summary.missing = { ...preflight.missingFields };
    summary.warnings = {
      'Invalid price rows': preflight.invalidPriceRows.length,
      'Invalid stock rows': preflight.invalidStockRows.length,
      'Discount above price rows': preflight.discountAbovePriceRows.length,
    };

    if (options.dryRun) {
      summary.imported = {
        'Products validated': preflight.sheetRows.Products,
        'Categories validated': preflight.sheetRows.Categories,
        'Authors validated': preflight.sheetRows.Authors,
        'Publishers validated': preflight.sheetRows.Publishers,
        'Custom Attributes validated': preflight.sheetRows['Custom Attributes'],
        'Catalogues validated': preflight.sheetRows.Catalogues,
        'Image URLs validated': preflight.imageUrlCount,
      };
      summary.completedAt = new Date().toISOString();
      summary.executionTimeMs = Date.now() - startedAtMs;
      await logger.event('dry_run.completed', { summary });
      await logger.writeSummary(summary);
      return summary;
    }

    await prisma.$connect();
    const resumableRun = options.resume
      ? await prisma.catalogImportRun.findFirst({
          where: { sourceFingerprint, status: ImportRunStatus.RUNNING },
          orderBy: { startedAt: 'desc' },
        })
      : null;
    const run =
      resumableRun ??
      (await prisma.catalogImportRun.create({
        data: {
          sourceFingerprint,
          workbookPath,
          status: ImportRunStatus.RUNNING,
          phase: 'PREFLIGHT_COMPLETE',
          options: options as unknown as Prisma.InputJsonValue,
          logPath: logger.logPath,
          errorLogPath: logger.errorLogPath,
        },
      }));
    runId = run.id;
    summary.runId = runId;

    await prisma.catalogImportRun.update({ where: { id: runId }, data: { phase: 'MASTERS' } });
    await importMasters(workbookPath, runId, summary, logger, options.concurrency);

    await prisma.catalogImportRun.update({ where: { id: runId }, data: { phase: 'PRODUCTS' } });
    await importProducts(
      workbookPath,
      preflight,
      runId,
      summary,
      logger,
      options.concurrency,
      options.batchSize
    );

    await prisma.catalogImportRun.update({ where: { id: runId }, data: { phase: 'PRODUCT_TEXT' } });
    await importProductText(workbookPath, preflight, runId, summary, logger);

    await prisma.catalogImportRun.update({
      where: { id: runId },
      data: { phase: 'CUSTOM_ATTRIBUTES' },
    });
    await importCustomAttributes(workbookPath, preflight, runId, summary, logger);

    await prisma.catalogImportRun.update({ where: { id: runId }, data: { phase: 'CATALOGUE_MAP' } });
    await applyCatalogueMap(inputOptions.catalogueMapPath, summary, logger);

    await prisma.catalogImportRun.update({ where: { id: runId }, data: { phase: 'ASSETS' } });
    await importAssets(
      {
        assetMode: options.assetMode,
        storageDir: options.storageDir,
        assetPublicBaseUrl: options.assetPublicBaseUrl,
        concurrency: options.concurrency,
      },
      summary,
      logger
    );

    await prisma.catalogImportRun.update({ where: { id: runId }, data: { phase: 'VERIFY' } });
    summary.verification = await verifyImport(preflight);
    summary.completedAt = new Date().toISOString();
    summary.executionTimeMs = Date.now() - startedAtMs;
    const status =
      totalCount(summary.failed) > 0
        ? ImportRunStatus.COMPLETED_WITH_ERRORS
        : ImportRunStatus.COMPLETED;
    await prisma.catalogImportRun.update({
      where: { id: runId },
      data: {
        status,
        phase: 'COMPLETE',
        counters: summary as unknown as Prisma.InputJsonValue,
        completedAt: new Date(),
        executionTimeMs: summary.executionTimeMs,
      },
    });
    await logger.event('import.completed', { status, summary });
    await logger.writeSummary(summary);
    return summary;
  } catch (error) {
    const message = error instanceof Error ? error.message : String(error);
    countRecord(summary.failed, 'Fatal import errors');
    summary.completedAt = new Date().toISOString();
    summary.executionTimeMs = Date.now() - startedAtMs;
    await logger.error('import.failed', { message });
    if (runId) {
      await prisma.catalogImportRun
        .update({
          where: { id: runId },
          data: {
            status: ImportRunStatus.FAILED,
            counters: summary as unknown as Prisma.InputJsonValue,
            completedAt: new Date(),
            executionTimeMs: summary.executionTimeMs,
          },
        })
        .catch(() => undefined);
    }
    await logger.writeSummary(summary);
    throw error;
  } finally {
    await logger.flush();
    if (!options.dryRun) await prisma.$disconnect();
  }
};