/**
 * db/agent-command-queue.js — durable agent command queue + channel state.
 *
 * Owns: CRUD for agent_command_queue and agent_channel_state tables.
 * Does NOT own: HTTP long-poll mechanics, response resolution (services/agent-channel.js).
 *
 * Design invariants:
 *  - enqueueCommand()      INSERT a row, returns the row id + generated request_id payload field.
 *  - claimNextCommand()    SELECT FOR UPDATE SKIP LOCKED — at-most-one delivery per poll.
 *  - completeCommand()     UPDATE status → completed|failed, writes result JSON.
 *  - expireStaleClaims()   — reverts claimed rows with claim_expires_at < NOW() back to pending.
 *    Call this at the top of every poll handler so crashed workers never orphan commands.
 *  - upsertChannelState()  — records last_poll_at for observability.
 *  - getQueueDepth()       — scalar metric for Datadog observability.
 *  - getOldestPendingAgeSec() — scalar metric for Datadog 30s alert threshold.
 *
 * Oracle notes (migrated from Postgres/Neon):
 *  - SKIP LOCKED works natively, but Oracle rejects `FETCH FIRST n ROWS ONLY`
 *    together with `FOR UPDATE` (ORA-03049). The dequeue therefore runs the
 *    `FOR UPDATE SKIP LOCKED` SELECT with an execute-time `maxRows: 1` on a
 *    dedicated connection (a transaction), then UPDATEs the claimed id and
 *    commits. This is the standard Oracle skip-locked-queue shape.
 *  - payload / result are JSON columns: bind writes as JSON.stringify(obj) (a raw
 *    JS object is NJS-044); reads come back already parsed. request_id lookups use
 *    JSON_VALUE(payload, '$.request_id').
 *  - The former pg LISTEN/NOTIFY signalling is gone — services/agent-channel.js
 *    now polls this queue on an interval (Oracle has no LISTEN/NOTIFY equivalent
 *    we rely on here).
 */

'use strict';

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

const CLAIM_TTL_SEC = 60; // 60s — a crashed worker's claim reverts to pending after this

// How many times a single command may be handed to an agent before we stop.
// Each redelivery makes the agent RE-RUN the work, so this bounds real cost on the
// customer's server, not just queue churn. 3 tolerates genuine transient failures
// (deploy restart, brief network drop) while ending a deterministic one quickly.
const MAX_DELIVERY_ATTEMPTS = 3;
// Rows abandoned during one claim call are skipped, so a call may look past them.
const MAX_CLAIM_SCAN = 10;

// ── Enqueue ──────────────────────────────────────────────────────────────────

/**
 * Enqueue a proxy-request command for an agent.
 * Returns { id, requestId } where requestId is embedded in payload for
 * round-trip correlation when the agent calls /api/agent/respond.
 *
 * @param {number} agentId
 * @param {object} requestPayload — { method, path, body, headers }
 * @returns {Promise<{ id: number, requestId: string }>}
 */
async function enqueueCommand(agentId, requestPayload) {
  const requestId = crypto.randomBytes(8).toString('hex');
  const payload = { ...requestPayload, request_id: requestId };

  const res = await pool.query(
    `INSERT INTO agent_command_queue (agent_id, command_type, payload)
     VALUES (:p1, 'proxy_request', :p2)
     RETURNING id INTO :out_id`,
    { p1: agentId, p2: JSON.stringify(payload), out_id: pool.outNumber() }
  );

  return { id: res.outBinds.out_id[0], requestId };
}

// ── Claim ─────────────────────────────────────────────────────────────────────

/**
 * Claim the next pending command for an agent.
 * Uses SELECT FOR UPDATE SKIP LOCKED so concurrent polls on multi-worker
 * deployments never double-deliver.
 *
 * Returns the row ({ id, payload, enqueued_at }) or null if the queue is empty.
 * `payload` is the parsed work-item object.
 *
 * @param {number} agentId
 * @param {string} claimHolder — unique label for this worker (e.g. process.pid)
 * @returns {Promise<object|null>}
 */
async function claimNextCommand(agentId, claimHolder) {
  // A row that has exhausted its attempts is abandoned rather than served, and we
  // then want the NEXT pending row instead — so one call may examine several rows.
  // Bounded so a queue full of exhausted rows cannot spin here.
  for (let scan = 0; scan < MAX_CLAIM_SCAN; scan++) {
    const outcome = await claimOneCommand(agentId, claimHolder);
    if (outcome.kind === 'claimed') return outcome.row;
    if (outcome.kind === 'empty') return null;
    // 'abandoned' — that row is now terminal; look at the next one.
  }
  return null;
}

/**
 * Claim (or abandon) exactly one pending row, in its own transaction.
 *
 * @returns {Promise<{kind:'claimed',row:object}|{kind:'abandoned'}|{kind:'empty'}>}
 */
async function claimOneCommand(agentId, claimHolder) {
  // Own connection = own transaction. autoCommit stays off until we commit the
  // claim, so the row lock taken by FOR UPDATE holds across the UPDATE. maxRows:1
  // makes the driver fetch (and therefore lock) exactly the one row we claim.
  const conn = await pool.getConnection();
  try {
    const sel = await conn.execute(
      `SELECT id AS "id", payload AS "payload", enqueued_at AS "enqueued_at",
              NVL(delivery_attempts, 0) AS "delivery_attempts"
         FROM agent_command_queue
        WHERE agent_id = :a AND status = 'pending'
        ORDER BY enqueued_at
          FOR UPDATE SKIP LOCKED`,
      { a: agentId },
      { maxRows: 1, autoCommit: false }
    );

    if (!sel.rows.length) {
      await conn.rollback();
      return { kind: 'empty' };
    }
    const row = sel.rows[0]; // { id, payload, enqueued_at, delivery_attempts }
    const attempts = Number(row.delivery_attempts) || 0;

    // ── Poison-pill backstop ────────────────────────────────────────────────
    // Stop re-serving a command that keeps coming back undelivered. Re-serving is
    // not free: the agent RE-RUNS the work every time, so a command that can never
    // return its result burns the agent in a ~68s cycle for ever and starves every
    // command queued behind it — which is what a dead agent looks like from the UI.
    // The agent has its own guard for failures it can recognise; this one holds
    // regardless of WHY delivery failed, and covers agents too old to have it.
    if (attempts >= MAX_DELIVERY_ATTEMPTS) {
      // The result envelope has to be a real one. getCompletedResult() feeds
      // sendToAgent() directly, and a bare { error } would surface as a 200 with an
      // empty body — an undeliverable command reading as success.
      const abandoned = {
        status_code: 502,
        body: {
          success: false,
          error: `Command abandoned after ${attempts} delivery attempts — the agent ran it but could not return the result. If this was a query, narrow it or add a row limit.`,
          error_code: 'DELIVERY_ATTEMPTS_EXCEEDED',
          delivery_attempts: attempts,
        },
      };
      await conn.execute(
        `UPDATE agent_command_queue
            SET status = 'failed',
                completed_at = CURRENT_TIMESTAMP,
                result = :r,
                claim_expires_at = NULL
          WHERE id = :id`,
        { r: JSON.stringify(abandoned), id: row.id },
        { autoCommit: false }
      );
      await conn.commit();
      console.warn(
        '[agent-command-queue] abandoned command id=%s agent=%s after %d delivery attempts — not re-serving',
        row.id, agentId, attempts
      );
      return { kind: 'abandoned' };
    }

    await conn.execute(
      `UPDATE agent_command_queue
          SET status = 'claimed',
              claimed_at = CURRENT_TIMESTAMP,
              claim_holder = :h,
              delivery_attempts = NVL(delivery_attempts, 0) + 1,
              claim_expires_at = CURRENT_TIMESTAMP + NUMTODSINTERVAL(${CLAIM_TTL_SEC}, 'SECOND')
        WHERE id = :id`,
      { h: claimHolder, id: row.id },
      { autoCommit: false }
    );
    await conn.commit();
    return { kind: 'claimed', row };
  } catch (err) {
    await conn.rollback().catch(() => {});
    throw err;
  } finally {
    await conn.close().catch(() => {});
  }
}

// ── Complete / Fail ───────────────────────────────────────────────────────────

/**
 * Mark a command as completed and store the result.
 *
 * @param {number} commandId
 * @param {object} result — { status_code, body }
 * @returns {Promise<void>}
 */
async function completeCommand(commandId, result) {
  await pool.query(
    `UPDATE agent_command_queue
        SET status = 'completed',
            completed_at = CURRENT_TIMESTAMP,
            result = :p2,
            claim_expires_at = NULL
      WHERE id = :p1`,
    { p1: commandId, p2: JSON.stringify(result) }
  );
}

/**
 * Mark a command as failed.
 *
 * @param {number} commandId
 * @param {string} reason
 * @returns {Promise<void>}
 */
async function failCommand(commandId, reason) {
  await pool.query(
    `UPDATE agent_command_queue
        SET status = 'failed',
            completed_at = CURRENT_TIMESTAMP,
            result = :p2,
            claim_expires_at = NULL
      WHERE id = :p1`,
    { p1: commandId, p2: JSON.stringify({ error: reason }) }
  );
}

// ── Claim expiry ──────────────────────────────────────────────────────────────

/**
 * Revert stale claimed rows (claim_expires_at < NOW()) back to 'pending'.
 * Must be called at the top of every poll handler.
 * Never throws — failures are logged and swallowed so they don't break polls.
 *
 * @param {number} agentId  — scoped per-agent to minimise lock surface
 * @returns {Promise<number>} count of reverted rows
 */
async function expireStaleClaims(agentId) {
  try {
    const { rowCount } = await pool.query(
      `UPDATE agent_command_queue
          SET status = 'pending',
              claimed_at = NULL,
              claim_holder = NULL,
              claim_expires_at = NULL
        WHERE agent_id = :p1
          AND status = 'claimed'
          AND claim_expires_at < CURRENT_TIMESTAMP`,
      { p1: agentId }
    );
    return rowCount || 0;
  } catch (err) {
    console.error('[agent-command-queue] expireStaleClaims error:', err.message);
    return 0;
  }
}

// ── Channel state (observability) ─────────────────────────────────────────────

/**
 * Upsert the channel state row for an agent — records last_poll_at.
 * Fire-and-forget; never throws.
 *
 * @param {number} agentId
 */
async function upsertChannelState(agentId) {
  try {
    await pool.query(
      `MERGE INTO agent_channel_state t
       USING (SELECT :p1 AS agent_id FROM DUAL) s
       ON (t.agent_id = s.agent_id)
       WHEN MATCHED THEN
         UPDATE SET last_poll_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP
       WHEN NOT MATCHED THEN
         INSERT (agent_id, last_poll_at, updated_at)
         VALUES (s.agent_id, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)`,
      { p1: agentId }
    );
  } catch (err) {
    console.error('[agent-command-queue] upsertChannelState error:', err.message);
  }
}

// ── Observability metrics ─────────────────────────────────────────────────────

/**
 * Total pending commands across all agents.
 * Used by Datadog metric pg_agent_queue_depth.
 *
 * @returns {Promise<number>}
 */
async function getQueueDepth() {
  const { rows } = await pool.query(
    `SELECT COUNT(*) AS n FROM agent_command_queue WHERE status = 'pending'`
  );
  return parseInt(rows[0].n, 10);
}

/**
 * Age in seconds of the oldest pending command, or 0 if the queue is empty.
 * Used by Datadog metric pg_agent_queue_oldest_pending_seconds (alert >30s).
 *
 * Oracle has no EXTRACT(EPOCH FROM interval); sum the interval's day/hour/
 * minute/second components instead. MIN() over an empty set is NULL → NULL age → 0.
 *
 * @returns {Promise<number>}
 */
async function getOldestPendingAgeSec() {
  const { rows } = await pool.query(
    `SELECT EXTRACT(DAY    FROM (SYSTIMESTAMP - MIN(enqueued_at))) * 86400
          + EXTRACT(HOUR   FROM (SYSTIMESTAMP - MIN(enqueued_at))) * 3600
          + EXTRACT(MINUTE FROM (SYSTIMESTAMP - MIN(enqueued_at))) * 60
          + EXTRACT(SECOND FROM (SYSTIMESTAMP - MIN(enqueued_at)))  AS age_sec
       FROM agent_command_queue
      WHERE status = 'pending'`
  );
  const age = parseFloat(rows[0].age_sec);
  return isNaN(age) ? 0 : Math.round(age);
}

// ── Lookup by request_id ──────────────────────────────────────────────────────

/**
 * Find a completed/failed command by its embedded request_id (for deliverResult).
 * The request_id is stored inside the payload JSON field.
 *
 * @param {string} requestId
 * @param {number} agentId
 * @returns {Promise<object|null>}
 */
async function findByRequestId(requestId, agentId) {
  const { rows } = await pool.query(
    `SELECT id, status, result FROM agent_command_queue
      WHERE agent_id = :p1
        AND JSON_VALUE(payload, '$.request_id') = :p2
      ORDER BY enqueued_at DESC
      FETCH FIRST 1 ROWS ONLY`,
    { p1: agentId, p2: requestId }
  );
  return rows[0] || null;
}

/**
 * Find a command by request_id without filtering by agent_id.
 * Used in /respond to correlate job_id stored in payload.
 *
 * @param {string} requestId
 * @returns {Promise<object|null>}
 */
async function findByRequestIdOnly(requestId) {
  const { rows } = await pool.query(
    `SELECT id, status, result, payload FROM agent_command_queue
      WHERE JSON_VALUE(payload, '$.request_id') = :p1
      ORDER BY enqueued_at DESC
      FETCH FIRST 1 ROWS ONLY`,
    { p1: requestId }
  );
  return rows[0] || null;
}

/**
 * Fetch the stored result for a completed command by request_id.
 * Returns the parsed result object, or null if not completed yet.
 *
 * @param {string} requestId
 * @returns {Promise<object|null>}
 */
async function getCompletedResult(requestId) {
  const { rows } = await pool.query(
    `SELECT result FROM agent_command_queue
      WHERE JSON_VALUE(payload, '$.request_id') = :p1
        AND status IN ('completed', 'failed')
      ORDER BY completed_at DESC
      FETCH FIRST 1 ROWS ONLY`,
    { p1: requestId }
  );
  return rows[0]?.result || null;
}

/**
 * Check if an agent polled recently (last_poll_at within last 30s).
 *
 * @param {number} agentId
 * @returns {Promise<boolean>}
 */
async function isRecentlyPolled(agentId) {
  // An agent that is EXECUTING work is not polling — and on modest hardware an op
  // routinely runs longer than the 30s window (measured 8-36s for EBS status ops on
  // a 2-VM lab). Judging liveness on last_poll_at alone therefore reports "agent
  // offline" precisely when the agent is busiest: five ops once 503'd in 10-15ms
  // straight after a 36s op, and all five passed when re-run alone. An unexpired
  // claimed row is positive proof the agent took the work, so accept it as alive.
  // deliverResult() also refreshes last_poll_at, so a long op keeps the agent live
  // as each result lands rather than only between polls.
  const { rows } = await pool.query(
    `SELECT 1 AS ok FROM agent_channel_state
      WHERE agent_id = :p1
        AND last_poll_at > SYSTIMESTAMP - INTERVAL '30' SECOND
      FETCH FIRST 1 ROWS ONLY`,
    { p1: agentId }
  );
  if (rows.length > 0) return true;

  const { rows: busy } = await pool.query(
    `SELECT 1 AS ok FROM agent_command_queue
      WHERE agent_id = :p1
        AND status = 'claimed'
        AND claim_expires_at > SYSTIMESTAMP
      FETCH FIRST 1 ROWS ONLY`,
    { p1: agentId }
  );
  return busy.length > 0;
}

module.exports = {
  enqueueCommand,
  claimNextCommand,
  completeCommand,
  failCommand,
  expireStaleClaims,
  upsertChannelState,
  getQueueDepth,
  getOldestPendingAgeSec,
  findByRequestId,
  findByRequestIdOnly,
  getCompletedResult,
  isRecentlyPolled,
};
