import { gunzipSync } from 'node:zlib';
import { z } from 'zod';
import { adapters, BaseConnector, html as H, type ConnectorMeta, type CrawlContext, type RawRecordInput, type RawRecordLike } from '@rareindex/connectors';
import type { NormalizedRecord } from '@rareindex/shared';
import { amount, cardHouseCategory, certFromTitle, isBundleTitle, isCurrency, isoDate, jsonAfterKey, lotAttributes, makeSale, nextFlightText, safeYear, saleGrade } from '../_g7-auctions-na-lib/index.js';

const SITE = 'https://www.fanaticscollect.com';
const HOUSE = 'Fanatics Collect';
const PARSER_VERSION = '1.0.0';

const MoneySchema = z.object({ amountInCents: z.number().nullable().optional(), currency: z.string().nullable().optional() }).nullable().optional();
export const ListingSchema = z.object({
  id: z.string(),
  title: z.string(),
  listingType: z.string().nullable().optional(),
  status: z.string().nullable().optional(),
  bidCount: z.number().nullable().optional(),
  currentBid: MoneySchema,
  startingPrice: MoneySchema,
  collectSales: z.array(z.object({ soldDate: z.string().nullable().optional(), soldFor: MoneySchema })).default([]),
  auction: z.object({ id: z.string().nullable().optional(), name: z.string().nullable().optional(), shortName: z.string().nullable().optional(), startsAt: z.string().nullable().optional(), endsAt: z.string().nullable().optional(), status: z.string().nullable().optional() }).nullable().optional(),
  lotString: z.string().nullable().optional(),
  slug: z.string().nullable().optional(),
  description: z.string().nullable().optional(),
  integerId: z.number().nullable().optional(),
  insertedAt: z.string().nullable().optional(),
  updatedAt: z.string().nullable().optional(),
  vaultItem: z.object({ id: z.string().nullable().optional(), integerId: z.number().nullable().optional() }).nullable().optional(),
  images: z.array(z.string()).default([]),
});
export type Listing = z.infer<typeof ListingSchema>;
export const PayloadSchema = z.object({ kind: z.literal('fc_item'), url: z.string(), lastmod: z.string().nullable(), listing: ListingSchema });
export type Payload = z.infer<typeof PayloadSchema>;

type RawListing = Record<string, unknown> & { imageSets?: Array<{ large?: string | null; medium?: string | null; small?: string | null }> | null; description?: string | null; vaultItem?: Record<string, unknown> | null; auction?: Record<string, unknown> | null };

/** Item page HTML → trimmed CollectListing (RSC flight payload, JSON-LD Product as fallback for title/images). */
export function parseItemPage(html: string, url: string, lastmod: string | null = null): Payload | null {
  const text = nextFlightText(html);
  const raw = jsonAfterKey<RawListing>(text, '"prefetchedItemData":');
  const ld = H.jsonLd(html, 'Product')[0];
  if (!raw && !ld) return null;
  const images: string[] = [];
  for (const s of raw?.imageSets ?? []) {
    const u = s?.large ?? s?.medium ?? s?.small;
    if (u && !images.includes(u)) images.push(u);
  }
  if (!images.length && ld?.image) for (const u of Array.isArray(ld.image) ? ld.image : [ld.image]) if (typeof u === 'string') images.push(u);
  const pick = (o: Record<string, unknown> | null | undefined, keys: string[]) => (o ? Object.fromEntries(keys.filter((k) => k in o).map((k) => [k, o[k]])) : o ?? null);
  const listing = {
    ...pick(raw ?? {}, ['id', 'title', 'listingType', 'status', 'bidCount', 'currentBid', 'startingPrice', 'collectSales', 'lotString', 'slug', 'integerId', 'insertedAt', 'updatedAt']),
    id: (raw?.id as string | undefined) ?? (typeof ld?.sku === 'string' ? ld.sku : url.match(/\/(?:weekly|fixed|premier)\/([0-9a-f-]{36})/i)?.[1]),
    title: (raw?.title as string | undefined) ?? (typeof ld?.name === 'string' ? ld.name : undefined),
    auction: pick(raw?.auction ?? null, ['id', 'name', 'shortName', 'startsAt', 'endsAt', 'status']),
    vaultItem: pick(raw?.vaultItem ?? null, ['id', 'integerId']),
    description: typeof raw?.description === 'string' ? raw.description.slice(0, 2000) : null,
    images: images.slice(0, 6),
  };
  const parsed = ListingSchema.safeParse(listing);
  if (!parsed.success) return null;
  return { kind: 'fc_item', url, lastmod, listing: parsed.data };
}

/** Sales-history children from the sitemap index, newest first (higher N = newer; fixed-price after weekly of the same N). */
export function salesHistorySitemaps(indexXml: string): string[] {
  const locs = adapters.parseSitemapIndex(indexXml).map((e) => e.loc).filter((l) => /sales-history/.test(l));
  const n = (l: string) => Number(l.match(/-(\d+)\.xml/)?.[1] ?? 0);
  const kind = (l: string) => (/fixed-price/.test(l) ? 1 : 0);
  return locs.sort((a, b) => n(b) - n(a) || kind(b) - kind(a));
}

function gunzipMaybe(buf: Uint8Array | null | undefined, text: string | null | undefined): string | null {
  if (buf && buf.byteLength) {
    const b = Buffer.from(buf);
    return (b.length >= 2 && b[0] === 0x1f && b[1] === 0x8b ? gunzipSync(b) : b).toString('utf8');
  }
  return text ?? null;
}

/**
 * Fanatics Collect (formerly PWCC) — sold results of the weekly/premier auctions and fixed-price sales.
 * Discovery through the public sitemap index (sales-history-*.xml.gz children with lastmod); each public
 * item page embeds the listing (incl. collectSales) in its Next.js RSC payload. See meta.json accessNotes.
 */
export class FanaticsCollectConnector extends BaseConnector {
  readonly version = '1.0.0';
  readonly parserVersion = PARSER_VERSION;
  override readonly urlPatterns = [/^https?:\/\/(www\.)?fanaticscollect\.com\/(weekly|fixed|premier)\/[0-9a-f-]{36}/i];
  protected override minIntervalMs = 1500;

  private async fetchText(ctx: CrawlContext, url: string, binary = false): Promise<{ text: string | null; status: number | null; fetchedAt: Date }> {
    await this.throttle(url);
    const res = await ctx.fetch(url, { engines: ['api'], responseType: binary ? 'binary' : 'text', minQuality: 0, force: binary, timeoutMs: 45_000 });
    if (!res.success) {
      ctx.anomaly('page_fetch_failed', `${url}: ${res.error ?? res.httpStatus}`);
      return { text: null, status: res.httpStatus, fetchedAt: res.fetchedAt };
    }
    return { text: gunzipMaybe(res.buffer, res.html), status: res.httpStatus, fetchedAt: res.fetchedAt };
  }

  private async childEntries(ctx: CrawlContext, url: string): Promise<Array<{ loc: string; lastmod: string | null }>> {
    const r = await this.fetchText(ctx, url, true);
    if (!r.text) return [];
    const entries = adapters.parseUrlset(r.text).map((e) => ({ loc: e.loc, lastmod: e.lastmod }));
    if (!entries.length) ctx.anomaly('pagination_failure', `${url}: empty sitemap`);
    return entries;
  }

  private async itemRecord(ctx: CrawlContext, url: string, lastmod: string | null): Promise<RawRecordInput | null> {
    const r = await this.fetchText(ctx, url);
    if (!r.text) return null;
    const payload = parseItemPage(r.text, url, lastmod);
    if (!payload) {
      ctx.anomaly('parse_failure_page', `${url}: no prefetchedItemData / Product JSON-LD`);
      return null;
    }
    return { url, externalId: payload.listing.id, kind: 'sale', engine: 'api', httpStatus: r.status, payload, fetchedAt: r.fetchedAt };
  }

  async *crawl(ctx: CrawlContext): AsyncIterable<RawRecordInput> {
    const mode = ctx.options.mode;
    const cfg = this.meta.config;
    const pagesPerRun = mode === 'probe' ? Number(cfg.probePages ?? 3) : Number(cfg.pagesPerRun ?? 300);
    let fetched = 0;
    let noSale = 0;
    let count = 0;
    const track = (rec: RawRecordInput | null) => {
      fetched++;
      if (rec && (rec.payload as Payload).listing.collectSales.length === 0) noSale++;
    };
    if (ctx.options.seeds?.length) {
      for (const url of ctx.options.seeds) {
        if (ctx.signal?.aborted || this.reached(ctx, count)) break;
        const rec = await this.itemRecord(ctx, url, null);
        track(rec);
        if (rec) {
          count++;
          yield rec;
        }
      }
      return;
    }
    const idx = await this.fetchText(ctx, `${SITE}/sitemap.xml`);
    if (!idx.text) return;
    const children = salesHistorySitemaps(idx.text);
    if (!children.length) {
      ctx.anomaly('pagination_failure', 'sitemap index has no sales-history children');
      return;
    }
    const cursor = ctx.options.cursor ?? {};
    if (mode === 'backfill') {
      // Oldest → newest, resumable by child url + offset.
      const order = [...children].reverse();
      let startIdx = typeof cursor.sitemap === 'string' ? Math.max(0, order.indexOf(cursor.sitemap)) : 0;
      let offset = startIdx === order.indexOf(cursor.sitemap as string) && typeof cursor.offset === 'number' ? cursor.offset : 0;
      let urlsDone = typeof cursor.urlsDone === 'number' ? cursor.urlsDone : 0;
      let items = typeof cursor.itemsProcessed === 'number' ? cursor.itemsProcessed : 0;
      for (let ci = startIdx; ci < order.length; ci++) {
        if (ctx.signal?.aborted || fetched >= pagesPerRun) break;
        const child = order[ci]!;
        const entries = await this.childEntries(ctx, child);
        for (let i = offset; i < entries.length; i++) {
          if (ctx.signal?.aborted || fetched >= pagesPerRun || this.reached(ctx, count)) {
            await ctx.setCursor({ sitemap: child, offset: i, urlsDone, itemsProcessed: items, updatedAt: new Date().toISOString() });
            await ctx.progress({ page: urlsDone, totalPages: null, itemsProcessed: items, cursor: { sitemap: child, offset: i, urlsDone, itemsProcessed: items } });
            this.reportNoSale(ctx, noSale, fetched);
            return;
          }
          const e = entries[i]!;
          const rec = await this.itemRecord(ctx, e.loc, e.lastmod);
          track(rec);
          urlsDone++;
          if (rec) {
            items++;
            count++;
            yield rec;
          }
          if (urlsDone % 25 === 0) {
            await ctx.setCursor({ sitemap: child, offset: i + 1, urlsDone, itemsProcessed: items, updatedAt: new Date().toISOString() });
            await ctx.progress({ page: urlsDone, totalPages: null, itemsProcessed: items });
          }
        }
        offset = 0;
        startIdx = ci + 1;
        await ctx.setCursor({ sitemap: order[ci + 1] ?? child, offset: 0, urlsDone, itemsProcessed: items, updatedAt: new Date().toISOString() });
      }
      if (startIdx >= order.length) await ctx.setCursor({ done: true, urlsDone, itemsProcessed: items, updatedAt: new Date().toISOString() });
      this.reportNoSale(ctx, noSale, fetched);
      return;
    }
    // Incremental / probe: newest children first, only entries newer than the last fully processed lastmod.
    const since = typeof cursor.since === 'string' ? cursor.since : '';
    let newest = since;
    const maxChildren = mode === 'probe' ? 1 : children.length;
    for (const child of children.slice(0, maxChildren)) {
      if (ctx.signal?.aborted || fetched >= pagesPerRun) break;
      const entries = await this.childEntries(ctx, child);
      const fresh = entries.filter((e) => !since || (e.lastmod ?? '') > since).sort((a, b) => (b.lastmod ?? '').localeCompare(a.lastmod ?? ''));
      if (!fresh.length) break; // whole child older than the cursor → stop descending
      for (const e of fresh) {
        if (ctx.signal?.aborted || fetched >= pagesPerRun || this.reached(ctx, count)) break;
        const rec = await this.itemRecord(ctx, e.loc, e.lastmod);
        track(rec);
        if (e.lastmod && e.lastmod > newest) newest = e.lastmod;
        if (rec) {
          count++;
          yield rec;
        }
      }
    }
    if (mode !== 'probe' && newest && newest !== since) await ctx.setCursor({ since: newest, updatedAt: new Date().toISOString() });
    this.reportNoSale(ctx, noSale, fetched);
  }

  private reportNoSale(ctx: CrawlContext, noSale: number, fetched: number): void {
    if (fetched > 0 && noSale / fetched > 0.3) ctx.anomaly('no_sale_on_page', `${noSale} of ${fetched} item pages carried no collectSales`);
  }

  async lookup(url: string, ctx: CrawlContext): Promise<RawRecordInput[]> {
    const rec = await this.itemRecord(ctx, url, null);
    return rec ? [rec] : [];
  }

  async normalize(raw: RawRecordLike): Promise<NormalizedRecord[]> {
    const p = PayloadSchema.parse(raw.payload);
    const l = p.listing;
    const sale = l.collectSales.find((s) => s.soldFor?.amountInCents && s.soldDate) ?? l.collectSales[0];
    if (!sale) return [];
    const price = amount((sale.soldFor?.amountInCents ?? 0) / 100);
    const saleDate = isoDate(sale.soldDate);
    if (!price || !saleDate || saleDate.getTime() > Date.now() + 86_400_000) return [];
    const currency = isCurrency(sale.soldFor?.currency) ? sale.soldFor!.currency! : 'USD';
    const type = (l.listingType ?? '').toUpperCase();
    const isAuction = type === 'WEEKLY' || type === 'PREMIER' || type === 'FLASH' || type === 'AUCTION';
    const hammer = isAuction ? amount((l.currentBid?.amountInCents ?? 0) / 100) : null;
    const g = saleGrade(l.title);
    const cert = certFromTitle(l.title) ?? (l.description ? certFromTitle(l.description) : null);
    const identifiers: Record<string, string> = { fanatics_listing_id: l.id };
    if (l.vaultItem?.id) identifiers.fanatics_vault_item_id = l.vaultItem.id;
    const attributes = lotAttributes({
      categorySlug: cardHouseCategory(l.title),
      name: l.title,
      year: safeYear(l.title),
      identifiers,
      metadata: { auction_name: l.auction?.name ?? null, auction_short_name: l.auction?.shortName ?? null, auction_ends_at: l.auction?.endsAt ?? null, listing_type: l.listingType ?? null, bid_count: l.bidCount ?? null, fanatics_status: l.status ?? null, hammer_price: hammer, buyer_premium_pct: isAuction && hammer && price > hammer ? Math.round(((price / hammer) - 1) * 1000) / 10 : null, sitemap_lastmod: p.lastmod },
    });
    const record = makeSale({
      meta: this.meta,
      sourceUrl: p.url,
      externalId: l.id,
      rawTitle: l.title,
      description: l.description ?? null,
      attributes,
      price,
      currency,
      saleDate,
      buyerPremiumIncluded: isAuction ? true : null,
      auctionHouse: HOUSE,
      lotNumber: l.lotString?.match(/Lot:?\s*([A-Za-z0-9-]+)/i)?.[1] ?? null,
      imageUrls: l.images,
      location: 'US',
      observedAt: raw.fetchedAt,
      parserVersion: PARSER_VERSION,
      grader: g.grader,
      grade: g.grade,
      isBundle: isBundleTitle(l.title),
      saleType: isAuction ? 'auction' : 'fixed_price',
      confidence: g.grader ? 0.9 : 0.85,
    });
    record.grade.qualifier = g.qualifier;
    record.grade.certificationNumber = cert;
    return [record];
  }
}

export default (meta: ConnectorMeta) => new FanaticsCollectConnector(meta);