new-site/api/src/routes/cdr.ts

/**
 * CDR Portal API — profiles, upload tokens, bucket mappings, webhook,
 * paywalled traffic-study response.
 *
 * Paywall model:
 *   The classified traffic study for a reporting year is locked until
 *   the customer has a `cdr_study_access_grants` row for that (profile,
 *   year). Grants are issued by the payment webhook in checkout.ts on
 *   fcc-499a / fcc-499a-499q / fcc-full-compliance / cdr-analysis.
 *
 *   Response shape when locked: counts + ingestion health only, no %s,
 *   no pre-signed PDF URL. Admin bypass ignores the grant check.
 */

import { Router } from "express";
import type { Request, Response } from "express";
import { randomBytes, createHmac } from "crypto";
import { pool } from "../db.js";

const router = Router();

const WORKER_URL = process.env.WORKER_URL || "http://workers:8090";

/** Ask the worker for a presigned MinIO URL (GET or PUT). */
async function presign(key: string, method: "GET" | "PUT", expires: number): Promise<string | null> {
  if (!key) return null;
  try {
    const r = await fetch(`${WORKER_URL}/jobs/presign`, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ key, expires, method }),
    });
    if (!r.ok) return null;
    const data = (await r.json()) as { url?: string };
    return data.url || null;
  } catch {
    return null;
  }
}

// ── Profile lookup helpers ────────────────────────────────────────────

async function loadProfile(profileId: number) {
  const r = await pool.query(
    `SELECT * FROM cdr_ingestion_profiles WHERE id = $1`,
    [profileId],
  );
  return r.rows[0] || null;
}

async function hasGrant(profileId: number, year: number): Promise<string | null> {
  const r = await pool.query(
    `SELECT granted_by_order FROM cdr_study_access_grants
      WHERE profile_id = $1 AND reporting_year = $2
      LIMIT 1`,
    [profileId, year],
  );
  return r.rows[0]?.granted_by_order ?? null;
}

function isAdminRequest(req: Request): boolean {
  // Admin bypass — any request bearing the admin token header sees full data.
  const token = (req.headers["x-admin-token"] || "").toString().trim();
  const expected = process.env.ADMIN_API_TOKEN || "";
  return Boolean(expected) && token === expected;
}

// ── GET profile id by telecom_entity_id (wizard pre-fill convenience) ─

router.get(
  "/api/v1/cdr/profile/by-entity/:entity_id",
  async (req: Request, res: Response) => {
    const entityId = Number(req.params.entity_id);
    if (!Number.isFinite(entityId) || entityId <= 0) {
      res.status(400).json({ error: "bad entity_id" }); return;
    }
    const r = await pool.query(
      `SELECT id FROM cdr_ingestion_profiles WHERE telecom_entity_id = $1`,
      [entityId],
    );
    if (r.rows.length === 0) {
      res.status(404).json({ error: "no cdr profile for this entity" }); return;
    }
    res.json({ profile_id: r.rows[0].id });
  },
);

// ── GET traffic study (paywalled) ────────────────────────────────────

router.get(
  "/api/v1/cdr/profile/:profile_id/study",
  async (req: Request, res: Response) => {
    const profileId = Number(req.params.profile_id);
    const year = Number(req.query.year) || new Date().getUTCFullYear();
    if (!Number.isFinite(profileId) || profileId <= 0) {
      res.status(400).json({ error: "bad profile_id" }); return;
    }

    const profile = await loadProfile(profileId);
    if (!profile) { res.status(404).json({ error: "profile not found" }); return; }

    // Ingestion health — always visible regardless of paywall
    const meter = await pool.query(
      `SELECT bytes_stored, rows_ingested, last_measured_at
         FROM cdr_usage_meters
        WHERE profile_id = $1 AND reporting_year = $2`,
      [profileId, year],
    );
    const uploads = await pool.query(
      `SELECT COUNT(*)::int AS total_uploads,
              MAX(created_at) AS last_upload_at,
              COALESCE(SUM(rows_accepted), 0)::int   AS rows_accepted,
              COALESCE(SUM(rows_quarantined), 0)::int AS rows_quarantined
         FROM cdr_ingestion_uploads
        WHERE profile_id = $1`,
      [profileId],
    );
    const ingestion = {
      profile_configured: true,
      total_uploads: uploads.rows[0].total_uploads,
      last_upload_at: uploads.rows[0].last_upload_at,
      rows_accepted: uploads.rows[0].rows_accepted,
      rows_quarantined: uploads.rows[0].rows_quarantined,
      bytes_stored: meter.rows[0]?.bytes_stored ?? 0,
      rows_this_year: meter.rows[0]?.rows_ingested ?? 0,
      last_measured_at: meter.rows[0]?.last_measured_at ?? null,
    };

    const grantOrder = await hasGrant(profileId, year);
    const admin = isAdminRequest(req);

    if (!grantOrder && !admin) {
      // LOCKED — show counts only.
      const siteBase =
        process.env.SITE_URL ||
        (process.env.DOMAIN ? `https://${process.env.DOMAIN}` : "https://performancewest.net");
      const unlockUrl =
        `${siteBase}/order/fcc-499a?entity=${profile.telecom_entity_id}` +
        `&year=${year}`;
      res.json({
        status: "locked",
        reporting_year: year,
        unlock_reason:
          "Pay for your " + year + " Form 499-A filing (or the standalone " +
          "CDR traffic study) to unlock the classified report.",
        unlock_url: unlockUrl,
        ingestion,
        classified_report: null,
      });
      return;
    }

    // UNLOCKED — load the study row if one exists.
    const study = await pool.query(
      `SELECT * FROM cdr_traffic_studies
        WHERE profile_id = $1 AND reporting_year = $2
        ORDER BY CASE reporting_period WHEN 'ANNUAL' THEN 0
               WHEN 'Q4' THEN 1 WHEN 'Q3' THEN 2
               WHEN 'Q2' THEN 3 WHEN 'Q1' THEN 4 ELSE 5 END
        LIMIT 1`,
      [profileId, year],
    );
    if (study.rows.length === 0) {
      res.json({
        status: "unlocked_pending_study",
        reporting_year: year,
        granted_by_order: grantOrder,
        message: "No traffic study has been generated yet for this period.",
        ingestion,
        classified_report: null,
      });
      return;
    }

    const row = study.rows[0];
    // Pre-signed URLs for the PDF + XLSX — only produced when unlocked.
    const [pdfUrl, xlsxUrl] = await Promise.all([
      row.pdf_minio_path  ? presign(row.pdf_minio_path,  "GET", 3600) : null,
      row.xlsx_minio_path ? presign(row.xlsx_minio_path, "GET", 3600) : null,
    ]);
    res.json({
      status: "unlocked",
      reporting_year: year,
      granted_by_order: grantOrder,
      ingestion,
      classified_report: {
        reporting_period: row.reporting_period,
        total_calls: Number(row.total_calls || 0),
        total_minutes: Number(row.total_minutes || 0),
        total_revenue_cents: Number(row.total_revenue_cents || 0),
        interstate_pct: row.interstate_pct,
        intrastate_pct: row.intrastate_pct,
        international_pct: row.international_pct,
        indeterminate_pct: row.indeterminate_pct,
        interstate_pct_minutes: row.interstate_pct_minutes,
        intrastate_pct_minutes: row.intrastate_pct_minutes,
        international_pct_minutes: row.international_pct_minutes,
        indeterminate_pct_minutes: row.indeterminate_pct_minutes,
        wholesale_minutes: Number(row.wholesale_minutes || 0),
        retail_minutes: Number(row.retail_minutes || 0),
        orig_state_regions: row.orig_state_regions_json,
        billing_state_regions: row.billing_state_regions_json,
        methodology: row.methodology,
        pdf_minio_path:  row.pdf_minio_path,
        xlsx_minio_path: row.xlsx_minio_path,
        pdf_download_url:  pdfUrl,
        xlsx_download_url: xlsxUrl,
        download_expires_in_seconds: 3600,
      },
    });
  },
);

// ── Presigned upload token (browser drag-drop path) ─────────────────

router.post("/api/v1/cdr/upload-token", async (req: Request, res: Response) => {
  const { profile_id, file_name } = req.body ?? {};
  if (!profile_id || !file_name) {
    res.status(400).json({ error: "profile_id and file_name required" });
    return;
  }
  const profile = await loadProfile(Number(profile_id));
  if (!profile) { res.status(404).json({ error: "profile not found" }); return; }

  // Short-lived token; browser PUTs to MinIO directly.
  const token = randomBytes(16).toString("hex");
  const safeName = String(file_name).replace(/[^A-Za-z0-9._-]/g, "_");
  const minioKey =
    `cdr-uploads/${profile.customer_id}/raw/browser/` +
    `${new Date().toISOString().replace(/[:.]/g, "")}_${token}_${safeName}`;

  await pool.query(
    `INSERT INTO cdr_ingestion_uploads
       (profile_id, source, raw_minio_path, raw_sha256, status, summary_json)
     VALUES ($1, 'browser', $2, $3, 'pending', $4::jsonb)`,
    [profile.id, minioKey, `pending_${token}`, JSON.stringify({ token, file_name: safeName })],
  );

  // Generate the pre-signed MinIO PUT URL — browser PUTs directly, no
  // API bandwidth.
  const minioPutUrl = await presign(minioKey, "PUT", 24 * 3600);

  res.status(201).json({
    token,
    minio_key: minioKey,
    minio_put_url: minioPutUrl,
    expires_in_seconds: 24 * 3600,
  });
});

// ── Webhook (per-call stream from a switch) ─────────────────────────

router.post(
  "/api/v1/cdr/webhook/:customer_token",
  async (req: Request, res: Response) => {
    const token = req.params.customer_token;
    // Look up the profile by webhook token (stored on preset_config).
    const r = await pool.query(
      `SELECT id FROM cdr_ingestion_profiles
        WHERE preset_config->>'webhook_token' = $1`,
      [token],
    );
    if (r.rows.length === 0) {
      res.status(404).json({ error: "unknown webhook token" }); return;
    }
    // Buffer body to MinIO under webhook/ prefix; the ingester will
    // pick it up on next cycle. Implementation left to the MinIO helper.
    res.status(202).json({ received: true });
  },
);

// ── Bucket-mapping CRUD ─────────────────────────────────────────────

router.get(
  "/api/v1/cdr/profile/:profile_id/bucket-mappings",
  async (req: Request, res: Response) => {
    const pid = Number(req.params.profile_id);
    const rows = await pool.query(
      `SELECT id, match_type, match_value, bucket, override_priority
         FROM cdr_bucket_mappings WHERE profile_id = $1
         ORDER BY match_type, match_value`,
      [pid],
    );
    res.json({ mappings: rows.rows });
  },
);

router.put(
  "/api/v1/cdr/profile/:profile_id/bucket-mappings",
  async (req: Request, res: Response) => {
    const pid = Number(req.params.profile_id);
    const incoming = (req.body?.mappings ?? []) as Array<{
      match_type: string; match_value: string; bucket: string;
    }>;
    await pool.query(
      "DELETE FROM cdr_bucket_mappings WHERE profile_id = $1",
      [pid],
    );
    for (const m of incoming) {
      if (!["trunk_group", "account_id"].includes(m.match_type)) continue;
      if (!["wholesale", "retail"].includes(m.bucket)) continue;
      await pool.query(
        `INSERT INTO cdr_bucket_mappings
           (profile_id, match_type, match_value, bucket)
         VALUES ($1, $2, $3, $4)
         ON CONFLICT (profile_id, match_type, match_value) DO UPDATE
           SET bucket = EXCLUDED.bucket`,
        [pid, m.match_type, m.match_value, m.bucket],
      );
    }
    res.json({ saved: incoming.length });
  },
);

export default router;