unrip/test/quote-lifecycle-retention.test.mjs

import test from 'node:test';
import assert from 'node:assert/strict';

import { extendMakerTiming } from '../src/core/maker-timing.mjs';
import {
  DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY,
  QUOTE_LIFECYCLE_DATA_CLASSES,
  buildQuoteLifecycleRollups,
  classifyQuoteLifecycleDataClass,
  decideQuoteLifecycleRetentionMode,
  normalizeQuoteLifecycleRetentionPolicy,
} from '../src/core/quote-lifecycle-retention.mjs';
import {
  countQuoteLifecyclePruneCandidates,
  runQuoteLifecycleRetentionMaintenance,
} from '../src/lib/postgres.mjs';

test('quote lifecycle classifier separates success, in-flight, non-success, raw firehose, and rollups', () => {
  assert.equal(
    classifyQuoteLifecycleDataClass({
      table: 'quote_outcome_attributions',
      row: { outcome_status: 'completed', attribution_status: 'heuristic_match' },
    }),
    QUOTE_LIFECYCLE_DATA_CLASSES.successfulTradeEvidence,
  );
  assert.equal(
    classifyQuoteLifecycleDataClass({
      table: 'quote_outcome_attributions',
      row: { outcome_status: 'submitted', attribution_status: 'linked_settlement' },
    }),
    QUOTE_LIFECYCLE_DATA_CLASSES.successfulTradeEvidence,
  );
  assert.equal(
    classifyQuoteLifecycleDataClass({
      table: 'raw_near_intents_quotes',
      row: { ingested_at: '2026-06-06T10:00:00.000Z', quote_id: null },
      now: '2026-06-06T11:00:00.000Z',
    }),
    QUOTE_LIFECYCLE_DATA_CLASSES.rawDebugFirehose,
  );
  assert.equal(
    classifyQuoteLifecycleDataClass({
      table: 'trade_decisions',
      row: { ingested_at: '2026-06-06T10:55:00.000Z', quote_id: 'quote-recent' },
      now: '2026-06-06T11:00:00.000Z',
      detailRetentionMs: 10 * 60 * 1000,
    }),
    QUOTE_LIFECYCLE_DATA_CLASSES.inFlightDetail,
  );
  assert.equal(
    classifyQuoteLifecycleDataClass({
      table: 'trade_execution_results',
      row: { ingested_at: '2026-06-06T09:00:00.000Z', quote_id: 'quote-old' },
      now: '2026-06-06T11:00:00.000Z',
      detailRetentionMs: 10 * 60 * 1000,
    }),
    QUOTE_LIFECYCLE_DATA_CLASSES.nonSuccessDetail,
  );
  assert.equal(
    classifyQuoteLifecycleDataClass({
      table: 'quote_lifecycle_rollups',
      row: { data_class: 'aggregate_rollup' },
    }),
    QUOTE_LIFECYCLE_DATA_CLASSES.aggregateRollup,
  );
});

test('retention policy validation fails closed on missing or unsafe policy fields', () => {
  assert.equal(normalizeQuoteLifecycleRetentionPolicy(null).ok, false);
  assert.match(normalizeQuoteLifecycleRetentionPolicy(null).block_reason, /missing/);

  const unsafe = normalizeQuoteLifecycleRetentionPolicy({
    ...DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY,
    preserve_successful_evidence: false,
  });
  assert.equal(unsafe.ok, false);
  assert.match(unsafe.block_reason, /preserve_successful_evidence/);
});

test('retention mode enters pressure only from DB-backed policy thresholds', () => {
  const normal = decideQuoteLifecycleRetentionMode({
    policy: DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY,
    storageMetrics: {
      database: { total_bytes: 10 },
      tables: [{ table: 'trade_decisions', total_bytes: 100 }],
    },
  });
  assert.equal(normal.mode, 'normal');
  assert.equal(normal.detail_retention_ms, DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY.normal_detail_retention_ms);

  const pressure = decideQuoteLifecycleRetentionMode({
    policy: DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY,
    storageMetrics: {
      database: { total_bytes: 10 },
      tables: [{
        table: 'trade_decisions',
        total_bytes: DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY.pressure_table_bytes,
      }],
    },
  });
  assert.equal(pressure.mode, 'pressure');
  assert.equal(pressure.detail_retention_ms, DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY.pressure_detail_retention_ms);
});

test('quote lifecycle rollups group by pair, direction, request, edge, result, failure, age, notional, and outcome', () => {
  const nbtc = 'nep141:nbtc.bridge.near';
  const eure = 'nep141:eure.omft.near';
  const usdc = 'nep141:usdc.omft.near';
  const rows = [
    {
      quote_id: 'quote-completed',
      activity_at: '2026-06-06T10:00:00.010Z',
      quote: {
        quote_id: 'quote-completed',
        pair: `${nbtc}->${eure}`,
        request_kind: 'exact_in',
        maker_timing: { quote_received_at: '2026-06-06T10:00:00.000Z' },
      },
      decision: {
        decision_id: 'decision-completed',
        quote_id: 'quote-completed',
        pair: `${nbtc}->${eure}`,
        direction: 'base_to_quote',
        request_kind: 'exact_in',
        edge_bps: '49',
        notional: '5',
        notional_symbol: 'EURe',
      },
      command: {
        command_id: 'command-completed',
        quote_id: 'quote-completed',
      },
      execution: {
        quote_id: 'quote-completed',
        status: 'submitted',
        result_code: 'quote_response_ok',
        maker_timing: extendMakerTiming({ quote_received_at: '2026-06-06T10:00:00.000Z' }, {
          relay_result_at: '2026-06-06T10:00:00.090Z',
        }),
      },
      outcome: {
        quote_id: 'quote-completed',
        outcome_status: 'completed',
        attribution_status: 'heuristic_match',
      },
    },
    {
      quote_id: 'quote-failed',
      activity_at: '2026-06-06T10:00:00.020Z',
      decision: {
        decision_id: 'decision-failed',
        quote_id: 'quote-failed',
        pair: `${nbtc}->${usdc}`,
        direction: 'base_to_quote',
        request_kind: 'exact_in',
        edge_bps: '20',
        notional: '8',
        notional_symbol: 'USDC',
      },
      execution: {
        quote_id: 'quote-failed',
        status: 'failed',
        result_code: 'submission_failed',
        failure_category: 'quote_not_found_or_finished',
        maker_timing: extendMakerTiming({ quote_received_at: '2026-06-06T10:00:00.000Z' }, {
          relay_result_at: '2026-06-06T10:00:00.400Z',
        }),
      },
    },
  ];

  const first = buildQuoteLifecycleRollups({
    rows,
    now: '2026-06-06T11:00:00.000Z',
    computedAt: '2026-06-06T11:00:00.000Z',
  });
  const second = buildQuoteLifecycleRollups({
    rows,
    now: '2026-06-06T11:00:00.000Z',
    computedAt: '2026-06-06T11:00:00.000Z',
  });

  assert.deepEqual(first, second);
  assert.ok(first.some((rollup) => (
    rollup.pair === `${nbtc}->${eure}`
    && rollup.request_kind === 'exact_in'
    && rollup.edge_bps === '49'
    && rollup.notional_bucket === '5-25 EURe'
    && rollup.outcome_status === 'completed'
    && rollup.completed_count === 1
    && rollup.relay_accepted_count === 1
  )));
  assert.ok(first.some((rollup) => (
    rollup.pair === `${nbtc}->${usdc}`
    && rollup.result_code === 'submission_failed'
    && rollup.failure_category === 'quote_not_found_or_finished'
    && rollup.quote_age_bucket === '250-500ms'
    && rollup.notional_bucket === '5-25 USDC'
    && rollup.relay_failed_count === 1
  )));
});

test('policy-driven prune candidate counting fails closed when policy is invalid', async () => {
  const pool = {
    async query() {
      throw new Error('query should not run for invalid policy');
    },
  };
  const result = await countQuoteLifecyclePruneCandidates(pool, {
    policy: {
      ...DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY,
      enabled: false,
    },
  });

  assert.equal(result.ok, false);
  assert.equal(result.total, 0);
  assert.match(result.block_reason, /enabled/);
});

test('retention maintenance writes rollups before deleting detail and preserves successful predicates', async () => {
  const queries = [];
  let watermarkReads = 0;
  const policyRow = {
    ...DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY,
    updated_at: '2026-06-06T09:00:00.000Z',
  };
  const pool = {
    async query(sql, params = []) {
      queries.push({ sql, params });
      if (sql.includes('FROM retention_policies')) return { rows: [policyRow], rowCount: 1 };
      if (sql.includes('pg_database_size')) return { rows: [{ total_bytes: 10 }], rowCount: 1 };
      if (sql.includes('pg_stat_user_tables')) return { rows: [], rowCount: 0 };
      if (sql.includes('COUNT(*)::INT AS count') && sql.includes('quote_outcome_attributions')) {
        return { rows: [{ count: 1 }], rowCount: 1 };
      }
      if (sql.includes('FROM quote_lifecycle_rollup_watermarks')) {
        watermarkReads += 1;
        return watermarkReads === 1
          ? { rows: [], rowCount: 0 }
          : {
            rows: [{
              rollup_key: 'quote_lifecycle',
              covered_until: '2026-06-06T10:30:00.000Z',
              computed_at: '2026-06-06T11:00:00.000Z',
              status: 'success',
              row_count: 1,
              rollup_count: 1,
            }],
            rowCount: 1,
          };
      }
      if (sql.includes('MIN(activity_at)')) {
        return { rows: [{ earliest_at: '2026-06-06T10:00:00.000Z' }], rowCount: 1 };
      }
      if (sql.includes('WITH subject_events')) {
        return { rows: [], rowCount: 0 };
      }
      if (sql.includes('quote_id IS NULL')) {
        return {
          rows: [{
            event_id: 'raw-old',
            quote_id: null,
            activity_at: '2026-06-06T10:00:00.000Z',
            pair: 'a->b',
            quote_payload: {
              pair: 'a->b',
              request_kind: 'exact_in',
              maker_timing: { quote_received_at: '2026-06-06T10:00:00.000Z' },
            },
            quote_observed_at: '2026-06-06T10:00:00.000Z',
            quote_ingested_at: '2026-06-06T10:00:00.001Z',
          }],
          rowCount: 1,
        };
      }
      if (sql.includes('INSERT INTO quote_lifecycle_rollups')) return { rows: [], rowCount: 1 };
      if (sql.includes('INSERT INTO quote_lifecycle_rollup_watermarks')) return { rows: [], rowCount: 1 };
      if (sql.includes('DELETE FROM')) return { rows: [], rowCount: 2 };
      if (sql.includes('INSERT INTO quote_lifecycle_retention_runs')) return { rows: [], rowCount: 1 };
      return { rows: [], rowCount: 0 };
    },
  };

  const result = await runQuoteLifecycleRetentionMaintenance(pool, {
    now: '2026-06-06T11:00:00.000Z',
  });
  const rollupIndex = queries.findIndex((query) => query.sql.includes('INSERT INTO quote_lifecycle_rollups'));
  const firstDeleteIndex = queries.findIndex((query) => query.sql.includes('DELETE FROM'));

  assert.equal(result.status, 'success');
  assert.ok(rollupIndex >= 0);
  assert.ok(firstDeleteIndex > rollupIndex);
  assert.ok(queries.some((query) => (
    query.sql.includes('DELETE FROM quote_outcome_attributions')
    && query.sql.includes("target.outcome_status = 'completed'")
    && query.sql.includes('linked_settlement')
  )));
});

test('retention maintenance bounds historical rollup catch-up and fails pruning closed until covered', async () => {
  const queries = [];
  let watermarkReads = 0;
  const policyRow = {
    ...DEFAULT_QUOTE_LIFECYCLE_RETENTION_POLICY,
    updated_at: '2026-06-06T09:00:00.000Z',
  };
  const pool = {
    async query(sql, params = []) {
      queries.push({ sql, params });
      if (sql.includes('FROM retention_policies')) return { rows: [policyRow], rowCount: 1 };
      if (sql.includes('pg_database_size')) return { rows: [{ total_bytes: 10 }], rowCount: 1 };
      if (sql.includes('pg_stat_user_tables')) return { rows: [], rowCount: 0 };
      if (sql.includes('COUNT(*)::INT AS count') && sql.includes('quote_outcome_attributions')) {
        return { rows: [{ count: 0 }], rowCount: 1 };
      }
      if (sql.includes('FROM quote_lifecycle_rollup_watermarks')) {
        watermarkReads += 1;
        return watermarkReads === 1
          ? { rows: [], rowCount: 0 }
          : {
            rows: [{
              rollup_key: 'quote_lifecycle',
              covered_until: '2026-06-05T04:00:00.000Z',
              computed_at: '2026-06-06T11:00:00.000Z',
              status: 'success',
              row_count: 0,
              rollup_count: 0,
            }],
            rowCount: 1,
          };
      }
      if (sql.includes('MIN(activity_at)')) {
        return { rows: [{ earliest_at: '2026-06-05T00:00:00.000Z' }], rowCount: 1 };
      }
      if (sql.includes('WITH subject_events')) return { rows: [], rowCount: 0 };
      if (sql.includes('quote_id IS NULL')) return { rows: [], rowCount: 0 };
      if (sql.includes('INSERT INTO quote_lifecycle_rollup_watermarks')) return { rows: [], rowCount: 1 };
      if (sql.includes('INSERT INTO quote_lifecycle_retention_runs')) return { rows: [], rowCount: 1 };
      // Unconditional pruning (preserving successful evidence) runs even when the watermark is stale.
      if (sql.includes('DELETE FROM')) return { rows: [], rowCount: 0 };
      return { rows: [], rowCount: 0 };
    },
  };

  const result = await runQuoteLifecycleRetentionMaintenance(pool, {
    now: '2026-06-06T11:00:00.000Z',
  });
  const subjectQuery = queries.find((query) => query.sql.includes('WITH subject_events'));

  assert.equal(result.status, 'skipped');
  assert.equal(result.mode, 'blocked_rollup_stale');
  assert.equal(result.rollup_counts.partial, true);
  assert.equal(result.rollup_counts.covered_until, '2026-06-05T04:00:00.000Z');
  assert.equal(result.rollup_counts.target_covered_until, '2026-06-06T10:30:00.000Z');
  assert.deepEqual(subjectQuery?.params, [
    '2026-06-05T00:00:00.000Z',
    '2026-06-05T04:00:00.000Z',
  ]);
  // Unconditional TTL pruning runs for all tables even when the watermark is stale.
  // Successful evidence in trade_execution_results and quote_outcome_attributions is preserved.
  const deleteQueries = queries.filter((query) => query.sql.includes('DELETE FROM'));
  assert.ok(deleteQueries.length >= 1, 'expected at least one DELETE query');
});