transceiver-db/packages/scraper/src/utils/db.ts

import { Pool } from "pg";
import { config } from "dotenv";
import { join } from "path";

config({ path: join(__dirname, "..", "..", "..", "..", ".env") });

export const pool = new Pool({
  host: process.env.POSTGRES_HOST || "localhost",
  port: parseInt(process.env.POSTGRES_PORT || "5433"),
  database: process.env.POSTGRES_DB || "transceiver_db",
  user: process.env.POSTGRES_USER || "tip",
  password: process.env.POSTGRES_PASSWORD || "tip_dev_2026",
  max: 5,
  idleTimeoutMillis: 10000,
  connectionTimeoutMillis: 5000,
});

// Alias — some scrapers import { db } instead of { pool }
export const db = pool;

// Per-form-factor price bounds [min, max] in USD equivalent
const PRICE_BOUNDS: Record<string, [number, number]> = {
  "SFP":       [2,    3000],
  "SFP+":      [4,    5000],
  "SFP28":     [10,   8000],
  "SFP56":     [20,  10000],
  "SFP-DD":    [30,  12000],
  "QSFP+":     [15,   6000],
  "QSFP28":    [20,  10000],
  "QSFP56":    [50,  15000],
  "QSFP-DD":   [60,  20000],
  "QSFP112":   [80,  25000],
  "OSFP":      [100, 35000],
  "OSFP112":   [150, 40000],
  "OSFP224":   [200, 60000],
  "CFP":       [100, 30000],
  "CFP2":      [100, 30000],
  "XFP":       [10,   5000],
  "GBIC":      [2,    2000],
};

async function isPriceAnomalous(transceiverId: string, priceUsd: number): Promise<boolean> {
  const row = await pool.query(
    `SELECT form_factor FROM transceivers WHERE id = $1`,
    [transceiverId]
  );
  const formFactor = row.rows[0]?.form_factor as string | undefined;
  if (!formFactor) return false;
  const bounds = PRICE_BOUNDS[formFactor];
  if (!bounds) return false;
  return priceUsd < bounds[0] || priceUsd > bounds[1];
}

export async function upsertPriceObservation(params: {
  transceiverId: string;
  sourceVendorId: string;
  price: number;
  currency: string;
  stockLevel: string;
  quantityAvailable?: number;
  leadTimeDays?: number;
  url?: string;
  contentHash: string;
}): Promise<boolean> {
  // Normalize price to USD for sanity check (rough conversion)
  const priceUsd = params.currency === "EUR" ? params.price * 1.09
    : params.currency === "GBP" ? params.price * 1.27
    : params.price;

  const anomalous = await isPriceAnomalous(params.transceiverId, priceUsd);
  if (anomalous) {
    return false; // Reject price outside form-factor bounds
  }

  // Check if price changed via content hash
  const existing = await pool.query(
    `SELECT content_hash FROM price_observations
     WHERE transceiver_id = $1 AND source_vendor_id = $2
     ORDER BY time DESC LIMIT 1`,
    [params.transceiverId, params.sourceVendorId]
  );

  // Check if vendor is a competitor (non-Flexoptix) for competitor_verified flag
  const vendorRow = await pool.query(
    `SELECT is_competitor FROM vendors WHERE id = $1`,
    [params.sourceVendorId]
  );
  const isCompetitor = vendorRow.rows[0]?.is_competitor === true;

  if (existing.rows.length > 0 && existing.rows[0].content_hash === params.contentHash) {
    // Price unchanged — still ensure verified flags are current
    await pool.query(
      `UPDATE transceivers SET
         price_verified = true
         ${isCompetitor ? ", competitor_verified = true, competitor_verified_at = COALESCE(competitor_verified_at, NOW())" : ""}
       WHERE id = $1 AND (price_verified IS NULL OR price_verified = false OR ${isCompetitor ? "competitor_verified IS NULL OR competitor_verified = false" : "false"})`,
      [params.transceiverId]
    );
    return false; // No change
  }

  await pool.query(
    `INSERT INTO price_observations (time, transceiver_id, source_vendor_id, price, currency, stock_level, quantity_available, lead_time_days, url, content_hash)
     VALUES (NOW(), $1, $2, $3, $4, $5, $6, $7, $8, $9)`,
    [
      params.transceiverId,
      params.sourceVendorId,
      params.price,
      params.currency,
      params.stockLevel,
      params.quantityAvailable || null,
      params.leadTimeDays || null,
      params.url || null,
      params.contentHash,
    ]
  );

  // Mark price_verified always; competitor_verified only for non-Flexoptix vendors
  if (isCompetitor) {
    await pool.query(
      `UPDATE transceivers SET
         price_verified = true,
         competitor_verified = true,
         competitor_verified_at = COALESCE(competitor_verified_at, NOW())
       WHERE id = $1`,
      [params.transceiverId]
    );
  } else {
    await pool.query(
      `UPDATE transceivers SET price_verified = true WHERE id = $1 AND (price_verified IS NULL OR price_verified = false)`,
      [params.transceiverId]
    );
  }
  return true; // New observation written
}

export async function findOrCreateScrapedTransceiver(params: {
  partNumber: string;
  vendorId: string;
  formFactor?: string;
  speedGbps?: number;
  speed?: string;
  reachMeters?: number;
  reachLabel?: string;
  fiberType?: string;
  wavelengths?: string;
  category?: string;
  imageUrl?: string;
}): Promise<string> {
  // Try to match existing transceiver by part number + vendor
  const existing = await pool.query(
    `SELECT id, image_url FROM transceivers WHERE part_number = $1 AND vendor_id = $2`,
    [params.partNumber, params.vendorId]
  );

  if (existing.rows.length > 0) {
    // Update image_url and image_verified if we have a new image for a record without one
    if (params.imageUrl && !existing.rows[0].image_url) {
      await pool.query(
        `UPDATE transceivers SET image_url = $1, image_verified = true, updated_at = NOW() WHERE id = $2`,
        [params.imageUrl, existing.rows[0].id]
      );
    }
    return existing.rows[0].id;
  }

  // Create new transceiver entry
  const slug = `scraped-${params.partNumber.toLowerCase().replace(/[^a-z0-9]+/g, "-")}`;
  const result = await pool.query(
    `INSERT INTO transceivers (slug, part_number, vendor_id, form_factor, speed_gbps, speed, reach_meters, reach_label, fiber_type, wavelengths, category, market_status, image_url, image_verified)
     VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, 'Mainstream', $12, $13)
     ON CONFLICT (slug) DO UPDATE SET image_url = COALESCE(transceivers.image_url, EXCLUDED.image_url), image_verified = COALESCE(transceivers.image_verified, EXCLUDED.image_verified), updated_at = NOW()
     RETURNING id`,
    [
      slug,
      params.partNumber,
      params.vendorId,
      params.formFactor || "SFP",
      params.speedGbps || 0,
      params.speed || "Unknown",
      params.reachMeters || 0,
      params.reachLabel || "",
      params.fiberType || "",
      params.wavelengths || "",
      params.category || "DataCenter",
      params.imageUrl || null,
      params.imageUrl ? true : false,
    ]
  );
  return result.rows[0].id;
}

export interface SwitchParams {
  model: string;
  vendorId: string;
  series?: string;
  category?: string;
  layer?: string;
  portsConfig?: Record<string, number>;
  totalPorts?: number;
  uplinkSpeedGbps?: number;
  maxSpeedGbps?: number;
  switchingCapacityTbps?: number;
  forwardingRateMpps?: number;
  asicVendor?: string;
  asicModel?: string;
  asicSeries?: string;
  asicGeneration?: string;
  rackUnits?: number;
  maxPowerW?: number;
  typicalPowerW?: number;
  poeSupport?: string;
  stackingSupport?: boolean;
  vxlanSupport?: boolean;
  evpnSupport?: boolean;
  bgpSupport?: boolean;
  mplsSupport?: boolean;
  openconfigSupport?: boolean;
  sonicCompatible?: boolean;
  macsecSupport?: boolean;
  lifecycleStatus?: string;
  releaseDate?: string;
  eolDate?: string;
  msrpUsd?: number;
  tags?: string[];
  // Whitebox-specific fields
  isWhitebox?: boolean;
  isOcpAccepted?: boolean;
  ocpStatus?: string;
  supportedNos?: string[];
  onlCompatible?: boolean;
  dentCompatible?: boolean;
  cumulusCompatible?: boolean;
  fbossCompatible?: boolean;
  cpu?: string;
  cpuCores?: number;
  ramGb?: number;
  storageGb?: number;
  storageType?: string;
  transceiverFormFactors?: string[];
  catalogUrl?: string;
  sonicHwsku?: string;
  onieSupport?: boolean;
  scrapeSource?: string;
}

export async function findOrCreateSwitch(params: SwitchParams): Promise<string> {
  const existing = await pool.query(
    `SELECT id FROM switches WHERE model = $1 AND vendor_id = $2`,
    [params.model, params.vendorId]
  );

  if (existing.rows.length > 0) {
    await pool.query(
      `UPDATE switches SET
        series = COALESCE($2, series),
        category = COALESCE($3, category),
        ports_config = COALESCE($4, ports_config),
        total_ports = COALESCE($5, total_ports),
        max_speed_gbps = COALESCE($6, max_speed_gbps),
        switching_capacity_tbps = COALESCE($7, switching_capacity_tbps),
        is_whitebox = COALESCE($8, is_whitebox),
        supported_nos = COALESCE($9, supported_nos),
        sonic_compatible = COALESCE($10, sonic_compatible),
        sonic_hwsku = COALESCE($11, sonic_hwsku),
        cpu = COALESCE($12, cpu),
        ram_gb = COALESCE($13, ram_gb),
        storage_gb = COALESCE($14, storage_gb),
        transceiver_form_factors = COALESCE($15, transceiver_form_factors),
        catalog_url = COALESCE($16, catalog_url),
        is_ocp_accepted = COALESCE($17, is_ocp_accepted),
        ocp_status = COALESCE($18, ocp_status),
        onie_support = COALESCE($19, onie_support),
        asic_series = COALESCE($20, asic_series),
        last_scraped = CASE WHEN $21::text IS NOT NULL THEN NOW() ELSE last_scraped END,
        scrape_source = COALESCE($21, scrape_source),
        updated_at = NOW()
       WHERE id = $1`,
      [
        existing.rows[0].id,
        params.series || null,
        params.category || null,
        params.portsConfig ? JSON.stringify(params.portsConfig) : null,
        params.totalPorts || null,
        params.maxSpeedGbps || null,
        params.switchingCapacityTbps || null,
        params.isWhitebox ?? null,
        params.supportedNos?.length ? params.supportedNos : null,
        params.sonicCompatible ?? null,
        params.sonicHwsku || null,
        params.cpu || null,
        params.ramGb || null,
        params.storageGb || null,
        params.transceiverFormFactors?.length ? params.transceiverFormFactors : null,
        params.catalogUrl || null,
        params.isOcpAccepted ?? null,
        params.ocpStatus || null,
        params.onieSupport ?? null,
        params.asicSeries || null,
        params.scrapeSource || null,
      ]
    );
    return existing.rows[0].id;
  }

  const result = await pool.query(
    `INSERT INTO switches (
      model, vendor_id, series, category, layer,
      ports_config, total_ports, uplink_speed_gbps, max_speed_gbps,
      switching_capacity_tbps, forwarding_rate_mpps,
      asic_vendor, asic_model, asic_series, asic_generation,
      rack_units, max_power_w, typical_power_w,
      poe_support, stacking_support, vxlan_support, evpn_support,
      bgp_support, mpls_support, openconfig_support, sonic_compatible, macsec_support,
      lifecycle_status, release_date, eol_date, msrp_usd, tags,
      is_whitebox, is_ocp_accepted, ocp_status, supported_nos,
      onl_compatible, dent_compatible, cumulus_compatible, fboss_compatible,
      cpu, cpu_cores, ram_gb, storage_gb, storage_type,
      transceiver_form_factors, catalog_url, sonic_hwsku, onie_support,
      last_scraped, scrape_source
    ) VALUES (
      $1, $2, $3, $4, $5,
      $6, $7, $8, $9,
      $10, $11,
      $12, $13, $14, $15,
      $16, $17, $18,
      $19, $20, $21, $22,
      $23, $24, $25, $26, $27,
      $28, $29, $30, $31, $32,
      $33, $34, $35, $36,
      $37, $38, $39, $40,
      $41, $42, $43, $44, $45,
      $46, $47, $48, $49,
      $50, $51
    )
    ON CONFLICT (vendor_id, model) DO UPDATE SET updated_at = NOW()
    RETURNING id`,
    [
      params.model,
      params.vendorId,
      params.series || null,
      params.category || 'DataCenter',
      params.layer || 'L3',
      JSON.stringify(params.portsConfig || {}),
      params.totalPorts || null,
      params.uplinkSpeedGbps || null,
      params.maxSpeedGbps || null,
      params.switchingCapacityTbps || null,
      params.forwardingRateMpps || null,
      params.asicVendor || null,
      params.asicModel || null,
      params.asicSeries || null,
      params.asicGeneration || null,
      params.rackUnits || null,
      params.maxPowerW || null,
      params.typicalPowerW || null,
      params.poeSupport || 'None',
      params.stackingSupport || false,
      params.vxlanSupport || false,
      params.evpnSupport || false,
      params.bgpSupport || false,
      params.mplsSupport || false,
      params.openconfigSupport || false,
      params.sonicCompatible || false,
      params.macsecSupport || false,
      params.lifecycleStatus || 'Active',
      params.releaseDate || null,
      params.eolDate || null,
      params.msrpUsd || null,
      params.tags || [],
      params.isWhitebox || false,
      params.isOcpAccepted || false,
      params.ocpStatus || 'None',
      params.supportedNos || [],
      params.onlCompatible || false,
      params.dentCompatible || false,
      params.cumulusCompatible || false,
      params.fbossCompatible || false,
      params.cpu || null,
      params.cpuCores || null,
      params.ramGb || null,
      params.storageGb || null,
      params.storageType || null,
      params.transceiverFormFactors || [],
      params.catalogUrl || null,
      params.sonicHwsku || null,
      params.onieSupport || false,
      params.scrapeSource ? new Date() : null,
      params.scrapeSource || null,
    ]
  );
  return result.rows[0].id;
}

export async function ensureWhiteboxVendor(
  name: string,
  website?: string,
  options?: { isOdm?: boolean; ocpMember?: boolean; sonicContributor?: boolean }
): Promise<string> {
  const existing = await pool.query(`SELECT id FROM vendors WHERE name ILIKE $1`, [name]);
  if (existing.rows.length > 0) {
    if (options) {
      await pool.query(
        `UPDATE vendors SET
          is_whitebox_vendor = TRUE,
          is_odm = COALESCE($2, is_odm),
          ocp_member = COALESCE($3, ocp_member),
          sonic_contributor = COALESCE($4, sonic_contributor),
          updated_at = NOW()
         WHERE id = $1`,
        [existing.rows[0].id, options.isOdm ?? null, options.ocpMember ?? null, options.sonicContributor ?? null]
      );
    }
    return existing.rows[0].id;
  }

  const slug = name.toLowerCase().replace(/[^a-z0-9]+/g, "-");
  const result = await pool.query(
    `INSERT INTO vendors (name, slug, type, website, is_whitebox_vendor, is_odm, ocp_member, sonic_contributor)
     VALUES ($1, $2, 'manufacturer', $3, TRUE, $4, $5, $6)
     RETURNING id`,
    [
      name, slug, website || null,
      options?.isOdm ?? true,
      options?.ocpMember ?? false,
      options?.sonicContributor ?? false,
    ]
  );
  return result.rows[0].id;
}

export async function getVendorId(name: string): Promise<string | null> {
  const result = await pool.query(`SELECT id FROM vendors WHERE name = $1`, [name]);
  return result.rows[0]?.id || null;
}

export async function ensureVendor(
  name: string,
  type: string,
  website?: string,
  shopUrl?: string
): Promise<string> {
  // Try to find existing vendor first
  const existing = await pool.query(`SELECT id FROM vendors WHERE name ILIKE $1`, [name]);
  if (existing.rows.length > 0) return existing.rows[0].id;

  const slug = name.toLowerCase().replace(/[^a-z0-9]+/g, "-");
  try {
    const result = await pool.query(
      `INSERT INTO vendors (name, slug, type, website, shop_url, is_competitor)
       VALUES ($1, $2, $3, $4, $5, true)
       RETURNING id`,
      [name, slug, type, website || null, shopUrl || null]
    );
    return result.rows[0].id;
  } catch (err: unknown) {
    // Handle race condition — re-query if insert fails on unique constraint
    const existing2 = await pool.query(`SELECT id FROM vendors WHERE name ILIKE $1 OR slug = $2`, [name, slug]);
    if (existing2.rows.length > 0) return existing2.rows[0].id;
    throw err;
  }
}