unrip/test/postgres-quote-lifecycle-statistics.test.mjs

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

import {
  buildQuoteLifecycleStatistics,
} from '../src/core/quote-lifecycle-statistics.mjs';
import {
  ensureSchemaIndex,
  loadQuoteLifecycleStatistics,
  refreshQuoteLifecycleStatistics,
} from '../src/lib/postgres.mjs';

test('postgres quote lifecycle statistics refresh idempotently upserts state buckets by original quote window', async () => {
  const upserts = [];
  const subjects = new Map();
  const issuedSql = [];
  let includeFailure = false;
  const pool = {
    async query(sql, params = []) {
      issuedSql.push(sql);
      if (sql.includes('MAX(updated_at) AS latest_subject_updated_at')) {
        return {
          rows: [{
            has_statistic_subjects: subjects.size > 0,
            latest_subject_updated_at: subjects.size ? '2026-06-12T17:32:00.000Z' : null,
          }],
        };
      }
      if (sql.includes('INSERT INTO quote_lifecycle_stat_subjects')) {
        subjects.set('quote-stat', {
          quote_id: 'quote-stat',
          quote_activity_at: '2026-06-12T17:31:01.000Z',
          latest_stage_at: includeFailure
            ? '2026-06-12T17:36:10.000Z'
            : '2026-06-12T17:31:03.000Z',
          statistics_bucket: includeFailure ? 'submission_failed' : 'awaiting_executor',
          lifecycle_state: includeFailure ? 'failed' : 'command_emitted',
          reason_code: includeFailure ? 'quote_not_found_or_finished' : 'awaiting_executor',
        });
        if (subjects.size > 0 && includeFailure) {
          assert.equal(params[1], '2026-06-12T17:27:00.000Z');
        }
        return { rows: [], rowCount: 1 };
      }
      if (sql.includes('SELECT COUNT(*)::INT AS subject_count')) {
        return { rows: [{ subject_count: subjects.size }] };
      }
      if (sql.includes('WITH subjects AS')) {
        return {
          rows: buildQuoteLifecycleStatistics({
            lifecycleRows: [...subjects.values()],
            grains: params[0],
            computedAt: params[1],
          }),
        };
      }
      if (sql.includes('INSERT INTO quote_lifecycle_statistics')) {
        upserts.push({ sql, params });
        return { rows: [], rowCount: 1 };
      }
      throw new Error(`unexpected query: ${sql}`);
    },
  };

  await refreshQuoteLifecycleStatistics(pool, {
    now: '2026-06-12T17:32:00.000Z',
  });
  includeFailure = true;
  await refreshQuoteLifecycleStatistics(pool, {
    now: '2026-06-12T17:37:00.000Z',
  });

  const fiveMinuteUpserts = upserts.filter((query) => (
    query.params[1] === 'five_minute'
    && query.params[2] === '2026-06-12T17:30:00.000Z'
  ));
  assert.equal(fiveMinuteUpserts.length, 2);
  assert.equal(fiveMinuteUpserts.at(-1).params[4], 1);
  const latestBuckets = JSON.parse(fiveMinuteUpserts.at(-1).params[7]);
  assert.equal(latestBuckets.submission_failed, 1);
  assert.equal(latestBuckets.awaiting_executor, 0);
  assert.equal(
    upserts.some((query) => (
      query.params[1] === 'five_minute'
      && query.params[2] === '2026-06-12T17:35:00.000Z'
    )),
    false,
  );
  assert.equal(
    issuedSql.some((sql) => sql.includes('ORDER BY COALESCE') && sql.includes('FROM swap_demand_events')),
    false,
  );
  assert.equal(
    issuedSql.some((sql) => sql.includes('COALESCE(ingested_at, observed_at) >= $2::timestamptz')),
    true,
  );
});

test('postgres quote lifecycle statistics retain per-quote buckets after detail pruning', async () => {
  const subjects = new Map();
  const allTimeUpserts = [];
  let sourceAvailable = true;
  const pool = {
    async query(sql, params = []) {
      if (sql.includes('MAX(updated_at) AS latest_subject_updated_at')) {
        return {
          rows: [{
            has_statistic_subjects: subjects.size > 0,
            latest_subject_updated_at: subjects.size ? '2026-06-12T14:12:00.000Z' : null,
          }],
        };
      }
      if (sql.includes('INSERT INTO quote_lifecycle_stat_subjects')) {
        if (sourceAvailable) {
          subjects.set('quote-completed', {
            quote_id: 'quote-completed',
            quote_activity_at: '2026-06-12T14:11:01.000Z',
            latest_stage_at: '2026-06-12T14:11:10.000Z',
            statistics_bucket: 'success',
            lifecycle_state: 'completed',
            reason_code: 'completed',
          });
        }
        return { rows: [], rowCount: sourceAvailable ? 1 : 0 };
      }
      if (sql.includes('SELECT COUNT(*)::INT AS subject_count')) {
        return { rows: [{ subject_count: subjects.size }] };
      }
      if (sql.includes('WITH subjects AS')) {
        return {
          rows: buildQuoteLifecycleStatistics({
            lifecycleRows: [...subjects.values()],
            grains: params[0],
            computedAt: params[1],
          }),
        };
      }
      if (sql.includes('INSERT INTO quote_lifecycle_statistics')) {
        if (params[1] === 'all_time') allTimeUpserts.push({ params });
        return { rows: [], rowCount: 1 };
      }
      throw new Error(`unexpected query: ${sql}`);
    },
  };

  await refreshQuoteLifecycleStatistics(pool, {
    now: '2026-06-12T14:12:00.000Z',
  });
  sourceAvailable = false;
  await refreshQuoteLifecycleStatistics(pool, {
    now: '2026-06-12T15:12:00.000Z',
  });

  assert.equal(allTimeUpserts.length, 2);
  assert.equal(allTimeUpserts[0].params[4], 1);
  assert.equal(JSON.parse(allTimeUpserts[0].params[7]).success, 1);
  assert.equal(allTimeUpserts[1].params[4], 1);
  assert.equal(JSON.parse(allTimeUpserts[1].params[7]).success, 1);
});

test('postgres quote lifecycle statistics loader returns all-time and clamped grain rows', async () => {
  let issuedSql = '';
  const pool = {
    async query(sql, params = []) {
      issuedSql = sql;
      assert.match(sql, /FROM quote_lifecycle_statistics/);
      assert.deepEqual(params, [['all_time', 'five_minute'], 2]);
      return {
        rows: [{
          stat_id: 'quote-lifecycle-stat:all_time:all',
          grain: 'all_time',
          window_start: null,
          window_end: null,
          quote_count: 3,
          missing_identifier_count: 1,
          missing_timestamp_count: 0,
          bucket_counts: { success: 1, submission_failed: 2 },
          computed_at: '2026-06-12T17:40:00.000Z',
        }],
      };
    },
  };

  const rows = await loadQuoteLifecycleStatistics(pool, {
    grains: ['all_time', 'five_minute'],
    limit: 2,
  });

  assert.equal(rows.length, 1);
  assert.equal(rows[0].quote_count, 3);
  assert.equal(rows[0].bucket_counts.success, 1);
  assert.equal(rows[0].bucket_counts.submitted_no_reply, 0);
  assert.match(issuedSql, /PARTITION BY grain, window_start, window_end/);
  assert.match(issuedSql, /duplicate_rank = 1/);
});

test('postgres schema index creation tolerates concurrent already-created index race', async () => {
  let queryCount = 0;
  const pool = {
    async query() {
      queryCount += 1;
      const error = new Error('duplicate key value violates unique constraint "pg_class_relname_nsp_index"');
      error.code = '23505';
      error.constraint = 'pg_class_relname_nsp_index';
      throw error;
    },
  };

  await ensureSchemaIndex(pool, 'CREATE INDEX IF NOT EXISTS quote_lifecycle_stat_subjects_updated_idx ON quote_lifecycle_stat_subjects (updated_at DESC)');
  assert.equal(queryCount, 1);
});

test('postgres schema index creation still rejects unrelated duplicate errors', async () => {
  const pool = {
    async query() {
      const error = new Error('duplicate key value violates unrelated constraint');
      error.code = '23505';
      error.constraint = 'other_unique_constraint';
      throw error;
    },
  };

  await assert.rejects(
    () => ensureSchemaIndex(pool, 'CREATE INDEX IF NOT EXISTS unrelated_idx ON some_table (id)'),
    /unrelated constraint/,
  );
});