Update: 2026-07-06 17:00:43

backend/pricing-engine/src/db/connection.ts

@@ -0,0 +1,142 @@
+import mysql, { RowDataPacket, ResultSetHeader } from 'mysql2/promise';
+import dotenv from 'dotenv';
+import path from 'path';
+
+dotenv.config({ path: path.resolve(__dirname, '../../.env') });
+
+let mysqlPool: mysql.Pool | null = null;
+
+export async function getMySQL(): Promise<mysql.Pool> {
+  if (!mysqlPool) {
+    mysqlPool = mysql.createPool({
+      host: process.env.DB_HOST || '127.0.0.1',
+      port: parseInt(process.env.DB_PORT || '3306'),
+      database: process.env.DB_NAME || 'siro',
+      user: process.env.DB_USER || 'root',
+      password: process.env.DB_PASS || '',
+      waitForConnections: true,
+      connectionLimit: 5,
+      queueLimit: 0,
+    });
+  }
+  return mysqlPool;
+}
+
+export async function fetchSamples(
+  pool: mysql.Pool,
+  competitorName?: string,
+  countryCode?: string,
+  hoursBack?: number
+): Promise<RowDataPacket[]> {
+  const conditions: string[] = ['distance_km > 0', 'duration_min > 0', 'price_amount > 0'];
+  const params: (string | number)[] = [];
+
+  if (competitorName) {
+    conditions.push('competitor_name = ?');
+    params.push(competitorName);
+  }
+  if (countryCode) {
+    conditions.push('country_code = ?');
+    params.push(countryCode);
+  }
+  if (hoursBack) {
+    conditions.push('scraped_at >= DATE_SUB(NOW(), INTERVAL ? HOUR)');
+    params.push(hoursBack);
+  }
+
+  const sql = `SELECT * FROM scraped_competitor_prices WHERE ${conditions.join(' AND ')} ORDER BY id DESC LIMIT 10000`;
+  const [rows] = await pool.query<RowDataPacket[]>(sql, params);
+  return rows;
+}
+
+export async function saveFormulas(
+  pool: mysql.Pool,
+  formulas: Array<{
+    competitorName: string;
+    countryCode: string;
+    tier: string;
+    baseFare: number;
+    kmRate: number;
+    minRate: number;
+    minFare: number;
+    rmse: number;
+    rSquared: number;
+    sampleCount: number;
+    surgeMultiplier: number;
+    peakHours: string;
+  }>
+): Promise<void> {
+  if (formulas.length === 0) return;
+
+  // Batch INSERT with ON DUPLICATE KEY UPDATE
+  const values = formulas.map(f => `(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())`).join(',');
+  const flatParams: (string | number)[] = [];
+
+  for (const f of formulas) {
+    flatParams.push(
+      f.competitorName, f.countryCode, f.tier,
+      f.baseFare, f.kmRate, f.minRate, f.minFare,
+      f.rmse, f.rSquared, f.surgeMultiplier,
+      f.sampleCount, f.peakHours
+    );
+  }
+
+  const sql = `INSERT INTO competitor_secret_formulas 
+    (competitor_name, country_code, tier, base_fare, price_per_km, price_per_min, min_fare, rmse, r_squared, surge_multiplier, sample_size, peak_hours, last_updated)
+    VALUES ${values}
+    ON DUPLICATE KEY UPDATE
+    base_fare = VALUES(base_fare),
+    price_per_km = VALUES(price_per_km),
+    price_per_min = VALUES(price_per_min),
+    min_fare = VALUES(min_fare),
+    rmse = VALUES(rmse),
+    r_squared = VALUES(r_squared),
+    surge_multiplier = VALUES(surge_multiplier),
+    sample_size = VALUES(sample_size),
+    peak_hours = VALUES(peak_hours),
+    last_updated = NOW()`;
+
+  await pool.execute(sql, flatParams);
+}
+
+export async function saveSurgeInsights(
+  pool: mysql.Pool,
+  insights: Array<{
+    competitorName: string;
+    countryCode: string;
+    surgeMultiplier: number;
+    peakStartHour: number;
+    peakEndHour: number;
+    sampleCount: number;
+  }>
+): Promise<void> {
+  if (insights.length === 0) return;
+
+  const values = insights.map(() => `(?, ?, ?, ?, ?, ?, NOW())`).join(',');
+  const flatParams: (string | number)[] = [];
+
+  for (const ins of insights) {
+    flatParams.push(
+      ins.competitorName, ins.countryCode,
+      ins.surgeMultiplier, ins.peakStartHour,
+      ins.peakEndHour, ins.sampleCount
+    );
+  }
+
+  const sql = `INSERT INTO competitor_surge_insights 
+    (competitor_name, country_code, surge_multiplier, peak_start_hour, peak_end_hour, sample_count, detected_at)
+    VALUES ${values}
+    ON DUPLICATE KEY UPDATE
+    surge_multiplier = VALUES(surge_multiplier),
+    sample_count = VALUES(sample_count),
+    detected_at = NOW()`;
+
+  await pool.execute(sql, flatParams);
+}
+
+export async function closeConnections(): Promise<void> {
+  if (mysqlPool) {
+    await mysqlPool.end();
+    mysqlPool = null;
+  }
+}