/**
 * db/agent.js — Agent tunnel + registration token persistence.
 *
 * Owns: agent_tunnels (one per connection), agent_reg_tokens (short-lived install tokens).
 * Does NOT own: oracle_connections CRUD (server.js), API key generation/encryption (setup-fresh.js).
 */

'use strict';

const pool = require('./index');

// ── Connection name uniqueness ─────────────────────────────────────────────

class DuplicateConnectionNameError extends Error {
  constructor(name) {
    super(`DUPLICATE_CONNECTION_NAME:${name}`);
    this.name = 'DuplicateConnectionNameError';
  }
}

async function connectionNameExists(userId, name) {
  if (!userId || !name) return false;
  const result = await pool.query(
    `SELECT id FROM oracle_connections WHERE user_id = :p1 AND name = :p2 FETCH FIRST 1 ROWS ONLY`,
    { p1: userId, p2: name.trim() }
  );
  return result.rows.length > 0;
}

// ── Registration tokens ─────────────────────────────────────────────────────

async function createRegToken({ token, connectionId, userId }) {
  // Multi-column RETURNING -> id through an OUT bind, then read the row back.
  const ins = await pool.query(
    `INSERT INTO agent_reg_tokens (token, connection_id, user_id)
     VALUES (:p1, :p2, :p3)
     RETURNING id INTO :out_id`,
    { p1: token, p2: connectionId, p3: userId, out_id: pool.outNumber() }
  );
  const result = await pool.query(
    `SELECT id, token, connection_id, user_id, expires_at
     FROM agent_reg_tokens WHERE id = :p1`,
    { p1: ins.outBinds.out_id[0] }
  );
  return result.rows[0];
}

// Non-consuming look-up of a registration token (does NOT mark it used).
// Used by /api/agent/provision to validate server-type BEFORE redeeming, so a
// mismatch rejection never burns the token.
async function peekRegToken(token) {
  const result = await pool.query(
    `SELECT connection_id, used, expires_at FROM agent_reg_tokens WHERE token = :p1`,
    { p1: token }
  );
  return result.rows[0] || null;
}

async function redeemRegToken(token) {
  // Atomic: find valid unused token, mark used, return connection_id.
  // The single UPDATE stays the atomic step — the read-back afterwards is only
  // fetching columns for a row this statement has already claimed, so two callers
  // racing on the same token still produce exactly one winner.
  const upd = await pool.query(
    `UPDATE agent_reg_tokens
     SET used = TRUE
     WHERE token = :p1
       AND used = FALSE
       AND expires_at > CURRENT_TIMESTAMP
     RETURNING id INTO :out_id`,
    { p1: token, out_id: pool.outNumber() }
  );
  if (!upd.rowCount) return null;
  const result = await pool.query(
    `SELECT id, connection_id, user_id FROM agent_reg_tokens WHERE id = :p1`,
    { p1: upd.outBinds.out_id[0] }
  );
  return result.rows[0] || null;
}

async function cleanExpiredTokens() {
  await pool.query(
    `DELETE FROM agent_reg_tokens WHERE expires_at < CURRENT_TIMESTAMP - INTERVAL '1' HOUR`
  );
}

// ── Agent tunnels ───────────────────────────────────────────────────────────

async function upsertTunnel({ connectionId, tunnelUuid, tunnelName, dnsHostname, status }) {
  // ON CONFLICT DO UPDATE ... RETURNING * -> MERGE + read-back (MERGE has no
  // RETURNING). EXCLUDED.col becomes s.col; the self-reference
  // COALESCE(agent_tunnels.provisioned_at, NOW()) becomes COALESCE(t.provisioned_at, ...)
  // so an existing provisioned_at is still preserved. oracle_sids was a text[]
  // reset to '{}' and is a JSON column here, so the empty value is '[]'.
  await pool.query(
    `MERGE INTO agent_tunnels t
     USING (SELECT :p1 AS connection_id, :p2 AS tunnel_uuid, :p3 AS tunnel_name,
                   :p4 AS dns_hostname, :p5 AS status FROM DUAL) s
        ON (t.connection_id = s.connection_id)
     WHEN MATCHED THEN UPDATE SET
       t.tunnel_uuid    = s.tunnel_uuid,
       t.tunnel_name    = s.tunnel_name,
       t.dns_hostname   = s.dns_hostname,
       t.status         = s.status,
       t.oracle_sids    = '[]',
       t.provisioned_at = COALESCE(t.provisioned_at, CURRENT_TIMESTAMP),
       t.updated_at     = CURRENT_TIMESTAMP
     WHEN NOT MATCHED THEN INSERT
       (connection_id, tunnel_uuid, tunnel_name, dns_hostname, status, provisioned_at, updated_at)
       VALUES (s.connection_id, s.tunnel_uuid, s.tunnel_name, s.dns_hostname, s.status,
               CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)`,
    { p1: connectionId, p2: tunnelUuid, p3: tunnelName, p4: dnsHostname, p5: status }
  );
  const result = await pool.query(
    `SELECT * FROM agent_tunnels WHERE connection_id = :p1`,
    { p1: connectionId }
  );
  return result.rows[0];
}

async function confirmTunnel({ connectionId, osInfo, oracleHome, oracleSids, pdbServices }) {
  // Always overwrite oracle_sids + pdb_services — installer is ground truth.
  // On re-registration the old values must be replaced, never merged/appended.
  // Out-of-order placeholders ($2..$5 before $1) — named binds keep the mapping.
  // oracle_sids / pdb_services were text[] bound as JS arrays; they are JSON
  // columns here, so they must be serialised.
  const upd = await pool.query(
    `UPDATE agent_tunnels
     SET status        = 'confirmed',
         os_info       = :p2,
         oracle_home   = :p3,
         oracle_sids   = :p4,
         pdb_services  = :p5,
         confirmed_at  = CURRENT_TIMESTAMP,
         updated_at    = CURRENT_TIMESTAMP
     WHERE connection_id = :p1`,
    {
      p1: connectionId, p2: osInfo, p3: oracleHome,
      p4: JSON.stringify(oracleSids || []), p5: JSON.stringify(pdbServices || []),
    }
  );
  const result = upd.rowCount
    ? await pool.query(`SELECT * FROM agent_tunnels WHERE connection_id = :p1`, { p1: connectionId })
    : { rows: [] };
  if (!result.rows[0]) {
    console.warn(`[agent-db] confirmTunnel: no tunnel row for connection_id=${connectionId} — confirm had no effect`);
  }
  return result.rows[0] || null;
}

/**
 * Update oracle_sids + pdb_services after a re-detect-sids probe.
 * Called from the re-detect-sids route — always overwrites both arrays.
 */
async function updateTunnelSids({ connectionId, oracleSids, pdbServices }) {
  await pool.query(
    `UPDATE agent_tunnels
     SET oracle_sids  = :p2,
         pdb_services = :p3,
         updated_at   = CURRENT_TIMESTAMP
     WHERE connection_id = :p1`,
    { p1: connectionId,
      p2: JSON.stringify(oracleSids || []), p3: JSON.stringify(pdbServices || []) }
  );
}

async function recordHeartbeat(connectionId, agentVersion, meta) {
  // Transition any non-terminal state to 'active' on heartbeat.
  // agentVersion is optional — sent by oracle-proxy.py since v3.5.5+.
  // meta: { proxy_version, installed_at, last_upgrade_at, python_version, cx_oracle_version,
  //         os_id, kernel, agent_status, last_oracle_error, oracle_retry_count, uptime_seconds,
  //         oracle_mode, instant_client_path, verifier_workaround_active }
  const m = (meta && typeof meta === 'object') ? meta : {};

  // agent_status from oracle_worker: 'starting' | 'healthy' | 'oracle_unreachable'
  // Null-safe — older agents that don't send this field leave the column unchanged.
  const agentStatus = m.agent_status || null;
  const lastOracleError = m.last_oracle_error !== undefined ? m.last_oracle_error : undefined;

  // Guard against NaN strings from Python agents (e.g., str(NaN) → "NaN").
  // parseInt("NaN", 10) returns NaN, not null — PostgreSQL rejects NaN for integer columns.
  const safeInt = (v) => { const n = parseInt(v, 10); return isNaN(n) ? null : n; };
  const oracleRetryCount = m.oracle_retry_count !== undefined ? safeInt(m.oracle_retry_count) : null;
  const uptimeSeconds = m.uptime_seconds !== undefined ? safeInt(m.uptime_seconds) : null;

  // Thick-mode fallback fields (v6.2+ agent, Task #1728054).
  // Null-safe — older agents that omit these leave the columns unchanged via COALESCE.
  const oracleMode = m.oracle_mode || null;                 // 'thin' | 'thick'
  const instantClientPath = m.instant_client_path || null;
  const verifierWorkaroundActive = m.verifier_workaround_active != null
    ? Boolean(m.verifier_workaround_active) : null;

  // last_oracle_error is a JSON column. The original wrapped it in
  // CASE WHEN $4::jsonb IS NOT NULL THEN $4::jsonb ELSE last_oracle_error END,
  // which Oracle rejects with ORA-00932: a CASE cannot mix a character bind with
  // a JSON column. A plain SET converts fine, so the assignment is emitted only
  // when a value was supplied — same semantics (absent => column left unchanged)
  // without the type conflict.
  const errBind = lastOracleError !== undefined ? JSON.stringify(lastOracleError) : null;
  const setError = errBind !== null ? 'last_oracle_error = :p4,' : '';

  const upd = await pool.query(
    `UPDATE agent_tunnels
     SET last_heartbeat              = CURRENT_TIMESTAMP,
         status                      = CASE WHEN status != 'uninstalled' THEN 'active' ELSE status END,
         agent_version               = COALESCE(:p2, agent_version),
         agent_status                = COALESCE(:p3, agent_status),
         ${setError}
         oracle_retry_count          = COALESCE(:p5, oracle_retry_count),
         uptime_seconds              = COALESCE(:p6, uptime_seconds),
         oracle_status_at            = CASE WHEN :p3 IS NOT NULL THEN CURRENT_TIMESTAMP ELSE oracle_status_at END,
         oracle_mode                 = COALESCE(:p7, oracle_mode),
         instant_client_path         = COALESCE(:p8, instant_client_path),
         verifier_workaround_active  = COALESCE(:p9, verifier_workaround_active),
         updated_at                  = CURRENT_TIMESTAMP
     WHERE connection_id = :p1`,
    Object.assign({
      p1: connectionId,
      p2: agentVersion || null,
      p3: agentStatus,
      // Typed explicitly — each of these can be null and meets a non-VARCHAR
      // column inside COALESCE, where an untyped null binds as CHAR (ORA-00932).
      p5: pool.num(oracleRetryCount),
      p6: pool.num(uptimeSeconds),
      p7: oracleMode,
      p8: instantClientPath,
      p9: pool.bool(verifierWorkaroundActive),
    }, errBind !== null ? { p4: errBind } : {})
  );
  const result = upd.rowCount
    ? await pool.query(
        `SELECT id, connection_id, status, last_heartbeat, agent_version, agent_status,
                oracle_mode, instant_client_path, verifier_workaround_active
         FROM agent_tunnels WHERE connection_id = :p1`,
        { p1: connectionId })
    : { rows: [] };

  // Persist extended proxy metadata to oracle_connections (backward-compatible — all nullable)
  if (m.proxy_version || m.installed_at || m.python_version || m.cx_oracle_version || m.os_id || m.kernel) {
    await pool.query(
      `UPDATE oracle_connections SET
         proxy_version      = COALESCE(:p2, proxy_version),
         installed_at       = COALESCE(:p3, installed_at),
         last_upgrade_at    = COALESCE(:p4, last_upgrade_at),
         python_version     = COALESCE(:p5, python_version),
         cx_oracle_version  = COALESCE(:p6, cx_oracle_version),
         os_id              = COALESCE(:p7, os_id),
         kernel_version     = COALESCE(:p8, kernel_version),
         updated_at         = CURRENT_TIMESTAMP
       WHERE id = :p1`,
      {
        p1: connectionId,
        p2: m.proxy_version    || null,
        // Typed explicitly: these meet TIMESTAMP columns inside COALESCE, and an
        // untyped null binds as CHAR, which Oracle refuses to mix (ORA-00932).
        p3: pool.tstz(m.installed_at),
        p4: pool.tstz(m.last_upgrade_at),
        p5: m.python_version   || null,
        p6: m.cx_oracle_version|| null,
        p7: m.os_id            || null,
        p8: m.kernel           || null,
      }
    );
  }

  return result.rows[0] || null;
}

async function getTunnel(connectionId) {
  const result = await pool.query(
    `SELECT * FROM agent_tunnels WHERE connection_id = :p1`,
    { p1: connectionId }
  );
  return result.rows[0] || null;
}

async function markUninstalled(connectionId) {
  // Sets uninstalled_at for the 30-day restore window; nulls heartbeat so fleet
  // status reads as offline immediately.
  await pool.query(
    `UPDATE agent_tunnels
     SET status         = 'uninstalled',
         uninstalled_at = CURRENT_TIMESTAMP,
         last_heartbeat = NULL,
         updated_at     = CURRENT_TIMESTAMP
     WHERE connection_id = :p1`,
    { p1: connectionId }
  );
}

async function restoreConnection(connectionId) {
  // Flip status back to 'pending' (agent needs re-install to become active again).
  // Clears uninstalled_at so the row no longer shows in the Removed section.
  await pool.query(
    `UPDATE agent_tunnels
     SET status         = 'pending',
         uninstalled_at = NULL,
         updated_at     = CURRENT_TIMESTAMP
     WHERE connection_id = :p1`,
    { p1: connectionId }
  );
}

// ── Connection helpers (agent-scoped reads/writes on oracle_connections) ────

async function createAgentConnection({ name, host, port, serviceName, username, encryptedPassword, encryptedKey, userId, privilegeModel }) {
  const displayName = (name || '').trim();
  if (!displayName) throw new Error('Connection name is required');
  if (await connectionNameExists(userId, displayName)) {
    throw new DuplicateConnectionNameError(displayName);
  }
  const model = (privilegeModel === 'sysdba') ? 'sysdba' : 'reader';
  const ins = await pool.query(
    `INSERT INTO oracle_connections
       (name, host, port, service_name, username, encrypted_password,
        connection_type, proxy_url, proxy_api_key_enc, user_id, proxy_key_created_at, privilege_model)
     VALUES (:p1, :p2, :p3, :p4, :p5, :p6, 'proxy', :p7, :p8, :p9, CURRENT_TIMESTAMP, :p10)
     RETURNING id INTO :out_id`,
    {
      p1: name, p2: host, p3: port, p4: serviceName, p5: username, p6: encryptedPassword,
      p7: 'https://pending.tunevault.agent', p8: encryptedKey, p9: userId, p10: model,
      out_id: pool.outNumber(),
    }
  );
  const result = await pool.query(
    `SELECT id, name FROM oracle_connections WHERE id = :p1`,
    { p1: ins.outBinds.out_id[0] }
  );
  return result.rows[0];
}

/**
 * Agent-only connection — no DB credentials needed upfront.
 * The agent auto-detects Oracle SIDs after install and updates the record.
 * hostIp/sshUser are collected from the UI; proxy_url is built from hostIp
 * so health checks can reach the proxy before a tunnel is provisioned.
 */
async function createAgentOnlyConnection({ name, encryptedKey, userId, hostIp, sshUser, privilegeModel }) {
  const displayName = (name || '').trim();
  if (!displayName) throw new Error('Connection name is required');
  if (await connectionNameExists(userId, displayName)) {
    throw new DuplicateConnectionNameError(displayName);
  }
  // Build proxy_url from hostIp if provided; placeholder only as last resort
  const proxyUrl = hostIp ? `http://${hostIp}:3100` : 'https://pending.tunevault.agent';
  const model = (privilegeModel === 'sysdba') ? 'sysdba' : 'reader';
  const ins = await pool.query(
    `INSERT INTO oracle_connections
       (name, host, connection_type, proxy_url, proxy_api_key_enc, user_id, proxy_key_created_at, privilege_model)
     VALUES (:p1, :p2, 'proxy', :p3, :p4, :p5, CURRENT_TIMESTAMP, :p6)
     RETURNING id INTO :out_id`,
    { p1: name, p2: hostIp || null, p3: proxyUrl, p4: encryptedKey, p5: userId, p6: model,
      out_id: pool.outNumber() }
  );
  const result = await pool.query(
    `SELECT id, name FROM oracle_connections WHERE id = :p1`,
    { p1: ins.outBinds.out_id[0] }
  );
  return result.rows[0];
}

async function getConnectionById(connectionId) {
  const result = await pool.query(
    `SELECT id, name, user_id, proxy_api_key_enc, proxy_key_last_used_at, last_test_success, server_type
     FROM oracle_connections WHERE id = :p1`,
    { p1: connectionId }
  );
  return result.rows[0] || null;
}

// Persist wizard-selected server type + (encrypted) EBS app-tier passwords onto a
// draft connection before install. Partial update — only provided keys are written.
async function setConnectionServerConfig(connectionId, { serverType, appsPwdEnc, weblogicPwdEnc, clientName, clientSlug }) {
  const sets = [];
  const vals = [];
  let i = 1;
  if (serverType !== undefined)     { sets.push(`server_type = $${i++}`);      vals.push(serverType); }
  if (appsPwdEnc !== undefined)     { sets.push(`apps_pwd_enc = $${i++}`);     vals.push(appsPwdEnc); }
  if (weblogicPwdEnc !== undefined) { sets.push(`weblogic_pwd_enc = $${i++}`); vals.push(weblogicPwdEnc); }
  if (clientName !== undefined)     { sets.push(`client_name = $${i++}`);      vals.push(clientName); }
  if (clientSlug !== undefined)     { sets.push(`client_slug = $${i++}`);      vals.push(clientSlug); }
  if (!sets.length) return;
  vals.push(connectionId);
  await pool.query(`UPDATE oracle_connections SET ${sets.join(', ')} WHERE id = $${i}`, vals);
}

async function updateConnectionProxyUrl(connectionId, proxyUrl) {
  await pool.query(
    `UPDATE oracle_connections SET proxy_url = :p1, updated_at = CURRENT_TIMESTAMP WHERE id = :p2`,
    { p1: proxyUrl, p2: connectionId }
  );
}

async function touchConnectionKeyUsage(connectionId) {
  await pool.query(
    `UPDATE oracle_connections SET proxy_key_last_used_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = :p1`,
    { p1: connectionId }
  );
}

async function updateConnectionSid(connectionId, serviceName) {
  await pool.query(
    `UPDATE oracle_connections SET service_name = :p1, updated_at = CURRENT_TIMESTAMP WHERE id = :p2`,
    { p1: serviceName, p2: connectionId }
  );
}

async function updateConnectionInstallerInfo(connectionId, { serverType, ebsService, ebsInstanceName, ebsContextFile }) {
  const isEbs = serverType === 'apps' || serverType === 'both' ? true : null;
  // Explicit casts on nullable params prevent "could not determine data type of parameter $N"
  // when pg driver sends JS null without a type hint.
  await pool.query(
    // is_ebs is BOOLEAN and nullable here, so it is bound with an explicit type:
    // an untyped null binds as CHAR and Oracle refuses CHAR vs BOOLEAN (ORA-00932).
    // The COALESCE'd text columns need no cast — a null CHAR bind is compatible
    // with VARCHAR2, which is why only this one changes shape.
    `UPDATE oracle_connections SET
      server_type       = COALESCE(:p1, server_type),
      ebs_service       = COALESCE(:p2, ebs_service),
      is_ebs            = COALESCE(:p3, is_ebs),
      ebs_instance_name = COALESCE(:p4, ebs_instance_name),
      ebs_context_file  = COALESCE(:p5, ebs_context_file),
      updated_at        = CURRENT_TIMESTAMP
     WHERE id = :p6`,
    { p1: serverType || null, p2: ebsService || null, p3: pool.bool(isEbs),
      p4: ebsInstanceName || null, p5: ebsContextFile || null, p6: connectionId }
  );
}

async function getConnectionContextFile(connectionId) {
  const r = await pool.query(
    `SELECT ebs_context_file FROM oracle_connections WHERE id = :p1`,
    { p1: connectionId }
  );
  return r.rows[0]?.ebs_context_file || null;
}
async function clearConnectionProxy(connectionId) {
  await pool.query(
    `UPDATE oracle_connections SET proxy_url = NULL, updated_at = CURRENT_TIMESTAMP WHERE id = :p1`,
    { p1: connectionId }
  );
}

// Returns a status summary for a single agent connection.
// Used by GET /api/connections/:id/agent-status.
// Pulls from agent_tunnels + oracle_connections — no new schema needed.
async function getAgentStatus(connectionId) {
  const result = await pool.query(
    `SELECT
       oc.id,
       oc.connection_type,
       oc.proxy_url,
       oc.proxy_key_last_used_at,
       at.tunnel_uuid,
       at.dns_hostname,
       at.status        AS tunnel_status,
       at.last_heartbeat,
       at.oracle_sids   AS detected_sids,
       at.os_info,
       at.agent_version
     FROM oracle_connections oc
     LEFT JOIN agent_tunnels at ON at.connection_id = oc.id
     WHERE oc.id = :p1`,
    { p1: connectionId }
  );
  return result.rows[0] || null;
}

// Returns enriched diagnostics for a single agent connection.
// Extends getAgentStatus with the most recent health check status/error
// and the connection owner user_id for access control.
// Used by GET /api/connections/:id/diagnostics.
async function getDiagnostics(connectionId) {
  const result = await pool.query(
    `SELECT
       oc.id,
       oc.user_id,
       oc.name          AS connection_name,
       oc.connection_type,
       oc.proxy_url,
       oc.proxy_key_last_used_at,
       at.tunnel_uuid,
       at.dns_hostname,
       at.status        AS tunnel_status,
       at.last_heartbeat,
       at.oracle_sids   AS detected_sids,
       at.os_info,
       at.agent_version,
       hc.id            AS last_hc_id,
       hc.status        AS last_hc_status,
       hc.created_at    AS last_hc_at,
       hc.summary_text  AS last_hc_error
     FROM oracle_connections oc
     LEFT JOIN agent_tunnels at ON at.connection_id = oc.id
     LEFT JOIN LATERAL (
       SELECT id, status, created_at, summary_text
       FROM health_checks
       WHERE connection_id = oc.id
       ORDER BY created_at DESC
       FETCH FIRST 1 ROWS ONLY
     ) hc ON 1 = 1
     WHERE oc.id = :p1`,
    { p1: connectionId }
  );
  return result.rows[0] || null;
}

// Returns all agent connections for a user that are stale or unregistered.
// Used by dashboard banner: "N agent connections need attention".
async function getStaleAgentConnections(userId) {
  const fiveMinAgo = new Date(Date.now() - 5 * 60 * 1000);
  const result = await pool.query(
    `SELECT oc.id, oc.name, at.last_heartbeat, at.status AS tunnel_status
     FROM oracle_connections oc
     LEFT JOIN agent_tunnels at ON at.connection_id = oc.id
     WHERE (oc.user_id = :p1 OR oc.user_id IS NULL)
       AND oc.connection_type = 'proxy'
       AND (
         at.id IS NULL
         OR at.status = 'uninstalled'
         OR at.last_heartbeat IS NULL
         OR at.last_heartbeat < :p2
       )`,
    { p1: userId, p2: pool.tstz(fiveMinAgo) }
  );
  return result.rows;
}

/**
 * Return the most recent unused, non-expired registration token for a connection.
 * Used by the SSH-install route to retrieve the token issued at form-submit time.
 */
async function getLatestRegToken(connectionId) {
  const result = await pool.query(
    `SELECT token FROM agent_reg_tokens
     WHERE connection_id = :p1
       AND used = FALSE
       AND expires_at > CURRENT_TIMESTAMP
     ORDER BY id DESC
     FETCH FIRST 1 ROWS ONLY`,
    [connectionId]
  );
  return result.rows[0] || null;
}

// Returns minimal connection row needed for the ping endpoint.
// Includes encrypted_password so the caller can decrypt and forward to the agent.
async function getConnectionForPing(connectionId) {
  const result = await pool.query(
    `SELECT id, user_id, connection_type, host, port, service_name, username, encrypted_password
     FROM oracle_connections WHERE id = :p1`,
    { p1: connectionId }
  );
  return result.rows[0] || null;
}

// Returns connection row with credentials for the run-diagnostics endpoint.
// Also fetches detected_sids from agent_tunnels for SID auto-selection.
async function getConnectionForDiagnostics(connectionId) {
  const result = await pool.query(
    `SELECT oc.id, oc.user_id, oc.connection_type, oc.host, oc.port,
            oc.service_name, oc.username, oc.encrypted_password,
            at.oracle_sids AS detected_sids
     FROM oracle_connections oc
     LEFT JOIN agent_tunnels at ON at.connection_id = oc.id
     WHERE oc.id = :p1`,
    { p1: connectionId }
  );
  return result.rows[0] || null;
}

// Stamp last_diagnostics_at = NOW() after a successful diagnostics run.
async function touchDiagnosticsAt(connectionId) {
  await pool.query(
    `UPDATE oracle_connections SET last_diagnostics_at = CURRENT_TIMESTAMP WHERE id = :p1`,
    { p1: connectionId }
  );
}

// ── Key rotation helpers ──────────────────────────────────────────────────────

/**
 * Atomically rotate the API key for a connection.
 *
 * - Moves current proxy_api_key_enc → proxy_api_key_enc_previous (5-min grace)
 * - Sets proxy_api_key_enc = newEncryptedKey
 * - Sets key_rotation_status = 'pending' (agent hasn't ACKed yet)
 * - Records rotated_at + actor email
 *
 * The grace window lets in-flight poll/respond requests that carry the OLD key
 * still succeed for 5 minutes while the agent picks up the new key.
 */
async function rotateConnectionKey(connectionId, newEncryptedKey, actorEmail) {
  await pool.query(
    `UPDATE oracle_connections
     SET proxy_api_key_enc_previous = proxy_api_key_enc,
         proxy_api_key_enc          = :p1,
         key_rotated_at             = CURRENT_TIMESTAMP,
         key_rotation_status        = 'pending',
         key_rotation_actor         = :p2,
         updated_at                 = CURRENT_TIMESTAMP
     WHERE id = :p3`,
    { p1: newEncryptedKey, p2: actorEmail || null, p3: connectionId }
  );
}

/**
 * Mark a rotation as acknowledged (agent confirmed it received the new key).
 */
async function ackKeyRotation(connectionId) {
  await pool.query(
    `UPDATE oracle_connections
     SET key_rotation_status = 'acknowledged',
         updated_at           = CURRENT_TIMESTAMP
     WHERE id = :p1`,
    { p1: connectionId }
  );
}

/**
 * Return current rotation state for a connection.
 * Also returns proxy_api_key_enc_previous so the grace-window check in
 * verifyApiKey can decrypt and compare without a second query.
 */
async function getConnectionKeyState(connectionId) {
  const result = await pool.query(
    `SELECT id, user_id, proxy_api_key_enc, proxy_api_key_enc_previous,
            key_rotated_at, key_rotation_status, key_rotation_actor, server_type, status
     FROM oracle_connections WHERE id = :p1`,
    { p1: connectionId }
  );
  return result.rows[0] || null;
}

async function activateOnHeartbeat(connectionId) {
  // Flip pending_registration → active on first successful poll/heartbeat.
  // No-op for connections already active or in any other terminal state.
  await pool.query(
    `UPDATE oracle_connections
     SET status = 'active', last_tested_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP
     WHERE id = :p1 AND status = 'pending_registration'`,
    { p1: connectionId }
  );
}

/**
 * Return last 5 key rotation events for the history list in the UI.
 * We read from activity_log (action_type = 'settings_change', detail.event = 'connection_key_rotated').
 */
async function getKeyRotationHistory(connectionId) {
  const result = await pool.query(
    // detail->>'event' -> JSON_VALUE(detail, '$.event') on the JSON column.
    `SELECT user_email, detail, created_at
     FROM activity_log
     WHERE connection_id = :p1
       AND action_type   = 'settings_change'
       AND JSON_VALUE(detail, '$.event') = 'connection_key_rotated'
     ORDER BY created_at DESC
     FETCH FIRST 5 ROWS ONLY`,
    { p1: connectionId }
  );
  return result.rows;
}

// ── Connection status + install token hash (v6 wizard) ──────────────────────

async function setConnectionStatus(connectionId, status) {
  await pool.query(
    `UPDATE oracle_connections SET status = :p1 WHERE id = :p2`,
    { p1: status, p2: connectionId }
  );
}

async function setInstallTokenHash(connectionId, tokenHash) {
  await pool.query(
    `UPDATE oracle_connections SET install_token_hash = :p1 WHERE id = :p2`,
    { p1: tokenHash, p2: connectionId }
  );
}

async function clearInstallTokenHash(connectionId) {
  await pool.query(
    `UPDATE oracle_connections SET install_token_hash = NULL WHERE id = :p1`,
    { p1: connectionId }
  );
}

// ── Install timing stamps ────────────────────────────────────────────────────

/**
 * Stamp install_token_issued_at = NOW() the first time a token is minted.
 * Uses COALESCE so re-issues don't overwrite the original timestamp —
 * the original issuance time is the start of the stall clock.
 */
async function stampInstallTokenIssuedAt(connectionId) {
  await pool.query(
    `UPDATE oracle_connections
     SET install_token_issued_at = COALESCE(install_token_issued_at, CURRENT_TIMESTAMP),
         updated_at               = CURRENT_TIMESTAMP
     WHERE id = :p1`,
    { p1: connectionId }
  );
}

/**
 * Stamp first_heartbeat_at = NOW() on the first successful heartbeat.
 * No-op if already set (COALESCE guard).
 * Clears the stalled state permanently — once an agent phones home the row
 * is no longer stalled regardless of subsequent silences.
 */
async function stampFirstHeartbeatAt(connectionId) {
  await pool.query(
    `UPDATE oracle_connections
     SET first_heartbeat_at = COALESCE(first_heartbeat_at, CURRENT_TIMESTAMP),
         updated_at          = CURRENT_TIMESTAMP
     WHERE id = :p1`,
    { p1: connectionId }
  );
}

/**
 * Return agent tunnels for a user's connections that match a given SSH host.
 * Used by install-jobs to detect an already-online agent (double-install guard).
 */
async function getTunnelsByUserAndHost(userId, host) {
  const result = await pool.query(
    `SELECT at.connection_id, at.status, at.last_heartbeat, at.agent_version
     FROM agent_tunnels at
     JOIN oracle_connections oc ON oc.id = at.connection_id
     WHERE oc.user_id = :p1
       AND (oc.host = :p2 OR oc.ssh_db_host = :p2)`,
    { p1: userId, p2: host }
  );
  return result.rows;
}

// ── Expire pending_registration drafts older than 24h ───────────────────────
// Called by the cleanup cron in routes/agent.js cleanup interval.

async function expirePendingDrafts() {
  const result = await pool.query(
    `DELETE FROM oracle_connections
     WHERE status = 'pending_registration'
       AND created_at < CURRENT_TIMESTAMP - INTERVAL '24' HOUR
       AND NOT EXISTS (
         SELECT 1 FROM health_checks WHERE connection_id = oracle_connections.id
       )
     RETURNING id INTO :out_ids`,
    { out_ids: pool.outNumber() }
  );
  // Multi-row DML RETURNING ... INTO collects every affected value into the OUT
  // bind, so the caller still gets the full list of deleted ids (not just a count).
  return (result.outBinds && result.outBinds.out_ids) || [];
}

/**
 * Persist the agent's self-reported agent.env config (non-secret keys only).
 * oracle-proxy.py applies a positive allowlist before sending; this side
 * re-filters so a downgraded or tampered agent cannot widen it.
 * JSON column: stringify on write — the Oracle driver takes no JS object.
 */
const AGENT_CONFIG_KEYS = new Set([
  'ORACLE_HOME', 'ORACLE_SID', 'ORACLE_SIDS', 'SERVER_TYPE',
  'APPS_ENV_FILE', 'APPS_USER', 'EBS_DB_HOST', 'INSTANCE_NAME',
  'CONTEXT_FILE', 'APPS_BASE',
]);

async function updateAgentConfig(connectionId, cfg) {
  if (!cfg || typeof cfg !== 'object') return;
  const clean = {};
  for (const [k, v] of Object.entries(cfg)) {
    if (AGENT_CONFIG_KEYS.has(k) && typeof v === 'string') {
      clean[k] = v.slice(0, 512);
    }
  }
  if (!Object.keys(clean).length) return;
  await pool.query(
    `UPDATE oracle_connections
        SET agent_config = :p1, agent_config_at = CURRENT_TIMESTAMP
      WHERE id = :p2`,
    { p1: JSON.stringify(clean), p2: connectionId }
  );
}

module.exports = {
  createRegToken,
  peekRegToken,
  redeemRegToken,
  cleanExpiredTokens,
  getLatestRegToken,
  upsertTunnel,
  confirmTunnel,
  recordHeartbeat,
  getTunnel,
  markUninstalled,
  restoreConnection,
  createAgentConnection,
  createAgentOnlyConnection,
  getConnectionById,
  setConnectionServerConfig,
  updateConnectionProxyUrl,
  updateConnectionSid,
  updateConnectionInstallerInfo,
  updateAgentConfig,
  getConnectionContextFile,
  touchConnectionKeyUsage,
  clearConnectionProxy,
  getAgentStatus,
  getStaleAgentConnections,
  getDiagnostics,
  getConnectionForPing,
  getConnectionForDiagnostics,
  touchDiagnosticsAt,
  rotateConnectionKey,
  ackKeyRotation,
  getConnectionKeyState,
  activateOnHeartbeat,
  getKeyRotationHistory,
  setConnectionStatus,
  setInstallTokenHash,
  clearInstallTokenHash,
  expirePendingDrafts,
  updateTunnelSids,
  stampInstallTokenIssuedAt,
  stampFirstHeartbeatAt,
  getTunnelsByUserAndHost,
};
