server.js

const express = require("express");
const { Pool } = require("pg");

const PORT = process.env.PORT || 3000;
const DATABASE_URL = process.env.DATABASE_URL;

if (!DATABASE_URL) {
  console.error("Missing DATABASE_URL environment variable.");
  process.exit(1);
}

const app = express();
app.use(express.json({ limit: "1mb" }));

const pool = new Pool({
  connectionString: DATABASE_URL,
  ssl: process.env.PGSSLMODE === "disable" ? false : { rejectUnauthorized: false }
});

async function initDb() {
  await pool.query(`
    CREATE TABLE IF NOT EXISTS events (
      id SERIAL PRIMARY KEY,
      type TEXT,
      time BIGINT,
      resource TEXT,
      user_id TEXT,
      environment_id TEXT,
      project_id TEXT,
      start_time BIGINT,
      end_time BIGINT,
      end_reason TEXT,
      platform TEXT,
      call_sid TEXT,
      call_type TEXT,
      user_number TEXT,
      agent_number TEXT,
      raw JSONB
    )
  `);

  await pool.query(`
    CREATE TABLE IF NOT EXISTS calls (
      id SERIAL PRIMARY KEY,
      call_sid TEXT NOT NULL,
      start_time BIGINT NOT NULL,
      end_time BIGINT,
      end_reason TEXT,
      user_id TEXT,
      environment_id TEXT,
      project_id TEXT,
      platform TEXT,
      call_type TEXT,
      user_number TEXT,
      agent_number TEXT,
      resource TEXT,
      first_event_time BIGINT,
      last_event_time BIGINT,
      raw_start JSONB,
      raw_end JSONB,
      UNIQUE (call_sid, start_time)
    )
  `);

  await pool.query(`
    CREATE TABLE IF NOT EXISTS voiceflow_postcall (
      id BIGSERIAL PRIMARY KEY,
      created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
      user_id TEXT,
      session_id TEXT,
      ts_unix BIGINT,
      vf_now TEXT,
      address TEXT,
      purchase_price NUMERIC,
      ownmoneytoinvest NUMERIC,
      rehab_budget NUMERIC,
      after_repair_value NUMERIC,
      name TEXT,
      phone TEXT,
      email TEXT,
      raw_payload JSONB NOT NULL
    )
  `);

  await pool.query(`
    CREATE INDEX IF NOT EXISTS voiceflow_postcall_user_id_idx
      ON voiceflow_postcall (user_id)
  `);
  await pool.query(`
    CREATE INDEX IF NOT EXISTS voiceflow_postcall_session_id_idx
      ON voiceflow_postcall (session_id)
  `);
  await pool.query(`
    CREATE INDEX IF NOT EXISTS voiceflow_postcall_raw_payload_gin
      ON voiceflow_postcall USING gin (raw_payload)
  `);
}

function extract(payload) {
  const data = payload.data || {};
  const metadata = data.metadata || {};
  return {
    type: payload.type || null,
    time: payload.time || null,
    resource: payload.resource || null,
    user_id: data.userID || null,
    environment_id: data.environmentID || null,
    project_id: data.projectID || null,
    start_time: data.startTime || null,
    end_time: data.endTime || null,
    end_reason: data.endReason || null,
    platform: data.platform || null,
    call_sid: metadata.callSid || null,
    call_type: metadata.callType || null,
    user_number: metadata.userNumber || null,
    agent_number: metadata.agentNumber || null
  };
}

function toNumber(value) {
  if (value === null || value === undefined || value === "") return null;
  const n = Number(value);
  return Number.isFinite(n) ? n : null;
}

app.post("/webhook", async (req, res) => {
  const payload = req.body;

  if (!payload || !payload.type || !payload.data) {
    return res.status(400).json({ ok: false, error: "Invalid payload" });
  }

  const fields = extract(payload);
  const raw = payload;

  try {
    await pool.query(
      `
      INSERT INTO events (
        type, time, resource, user_id, environment_id, project_id,
        start_time, end_time, end_reason, platform,
        call_sid, call_type, user_number, agent_number, raw
      ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)
      `,
      [
        fields.type,
        fields.time,
        fields.resource,
        fields.user_id,
        fields.environment_id,
        fields.project_id,
        fields.start_time,
        fields.end_time,
        fields.end_reason,
        fields.platform,
        fields.call_sid,
        fields.call_type,
        fields.user_number,
        fields.agent_number,
        raw
      ]
    );

    const isStart = fields.type === "runtime.call.start";
    const isEnd = fields.type === "runtime.call.end";

    if (isStart || isEnd) {
      const rawStart = isStart ? raw : null;
      const rawEnd = isEnd ? raw : null;
      const firstEvent = fields.time;
      const lastEvent = fields.time;

      await pool.query(
        `
        INSERT INTO calls (
          call_sid, start_time, end_time, end_reason, user_id, environment_id, project_id,
          platform, call_type, user_number, agent_number, resource,
          first_event_time, last_event_time, raw_start, raw_end
        ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16)
        ON CONFLICT (call_sid, start_time) DO UPDATE SET
          end_time = COALESCE(EXCLUDED.end_time, calls.end_time),
          end_reason = COALESCE(EXCLUDED.end_reason, calls.end_reason),
          user_id = COALESCE(EXCLUDED.user_id, calls.user_id),
          environment_id = COALESCE(EXCLUDED.environment_id, calls.environment_id),
          project_id = COALESCE(EXCLUDED.project_id, calls.project_id),
          platform = COALESCE(EXCLUDED.platform, calls.platform),
          call_type = COALESCE(EXCLUDED.call_type, calls.call_type),
          user_number = COALESCE(EXCLUDED.user_number, calls.user_number),
          agent_number = COALESCE(EXCLUDED.agent_number, calls.agent_number),
          resource = COALESCE(EXCLUDED.resource, calls.resource),
          first_event_time = COALESCE(calls.first_event_time, EXCLUDED.first_event_time),
          last_event_time = GREATEST(COALESCE(calls.last_event_time, EXCLUDED.last_event_time), EXCLUDED.last_event_time),
          raw_start = COALESCE(calls.raw_start, EXCLUDED.raw_start),
          raw_end = COALESCE(EXCLUDED.raw_end, calls.raw_end)
        `,
        [
          fields.call_sid,
          fields.start_time,
          fields.end_time,
          fields.end_reason,
          fields.user_id,
          fields.environment_id,
          fields.project_id,
          fields.platform,
          fields.call_type,
          fields.user_number,
          fields.agent_number,
          fields.resource,
          firstEvent,
          lastEvent,
          rawStart,
          rawEnd
        ]
      );
    }

    res.json({ ok: true });
  } catch (err) {
    res.status(500).json({ ok: false, error: String(err) });
  }
});

app.post("/postcall", async (req, res) => {
  const payload = req.body || {};

  try {
    await pool.query(
      `
      INSERT INTO voiceflow_postcall (
        user_id, session_id, ts_unix, vf_now,
        address, purchase_price, ownmoneytoinvest, rehab_budget, after_repair_value,
        name, phone, email, raw_payload
      ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)
      `,
      [
        payload.user_id || null,
        payload.session_id || null,
        toNumber(payload.timestamp),
        payload.vf_now || null,
        payload.address || null,
        toNumber(payload.purchase_price),
        toNumber(payload.ownmoneytoinvest),
        toNumber(payload.rehab_budget),
        toNumber(payload.After_Repair_Value || payload.after_repair_value),
        payload.name || null,
        payload.phone || null,
        payload.email || null,
        payload
      ]
    );

    res.json({ ok: true });
  } catch (err) {
    res.status(500).json({ ok: false, error: String(err) });
  }
});

app.get("/health", (req, res) => {
  res.json({ ok: true });
});

function buildWhere(allowedMap, query) {
  const clauses = [];
  const params = [];
  for (const [paramKey, column] of Object.entries(allowedMap)) {
    if (query[paramKey] !== undefined) {
      clauses.push(`${column} = $${params.length + 1}`);
      params.push(query[paramKey]);
    }
  }
  return { where: clauses.length ? `WHERE ${clauses.join(" AND ")}` : "", params };
}

app.get("/events", async (req, res) => {
  const limit = Number(req.query.limit || 100);
  const offset = Number(req.query.offset || 0);
  const allowed = {
    type: "type",
    callSid: "call_sid",
    userID: "user_id",
    projectID: "project_id",
    environmentID: "environment_id",
    resource: "resource"
  };
  const { where, params } = buildWhere(allowed, req.query);
  const sql = `
    SELECT *
    FROM events
    ${where}
    ORDER BY time DESC
    LIMIT $${params.length + 1} OFFSET $${params.length + 2}
  `;
  try {
    const result = await pool.query(sql, [...params, limit, offset]);
    res.json({ ok: true, rows: result.rows });
  } catch (err) {
    res.status(500).json({ ok: false, error: String(err) });
  }
});

app.get("/calls", async (req, res) => {
  const limit = Number(req.query.limit || 100);
  const offset = Number(req.query.offset || 0);
  const allowed = {
    callSid: "call_sid",
    userID: "user_id",
    projectID: "project_id",
    environmentID: "environment_id",
    resource: "resource"
  };
  const { where, params } = buildWhere(allowed, req.query);
  const sql = `
    SELECT *
    FROM calls
    ${where}
    ORDER BY COALESCE(end_time, start_time) DESC
    LIMIT $${params.length + 1} OFFSET $${params.length + 2}
  `;
  try {
    const result = await pool.query(sql, [...params, limit, offset]);
    res.json({ ok: true, rows: result.rows });
  } catch (err) {
    res.status(500).json({ ok: false, error: String(err) });
  }
});

initDb()
  .then(() => {
    app.listen(PORT, () => {
      console.log(`Listening on http://localhost:${PORT}`);
    });
  })
  .catch((err) => {
    console.error("Failed to initialize database:", err);
    process.exit(1);
  });