cyberhybridhub/server/test/integration/market_data_retention_test.dart

@Tags(['integration', 'postgres'])
library;

import 'package:cyberhybridhub_server/trading/market_data_db.dart';
import 'package:cyberhybridhub_server/trading/market_data_retention.dart';
import 'package:cyberhybridhub_server/trading/market_history_session_slot.dart';
import 'package:postgres/postgres.dart';
import 'package:test/test.dart';

import '../helpers/test_db.dart';

void main() {
  TestDb? testDb;

  setUpAll(() async {
    ensureMarketHistoryTimezonesInitialized();
    testDb = await TestDb.open();
  });

  /// After US close so [windowFirstSlotStart] is stable in tests.
  final DateTime retentionNow = DateTime.utc(2026, 5, 26, 21, 0);

  tearDown(() async {
    if (testDb != null) {
      await testDb!.connection.execute('TRUNCATE TABLE market_data_archive');
      await testDb!.truncateTradingTables();
    }
  });

  tearDownAll(() async {
    await testDb?.close();
  });

  MarketDataRetention retention({void Function(String sql)? onExecute}) {
    return MarketDataRetention(
      connection: testDb!.connection,
      onExecute: onExecute,
    );
  }

  group('runCleanup (hard delete)', () {
    test('deletes rows older than window and keeps rows within window',
        () async {
      if (testDb == null) {
        markTestSkipped(
          'Set DATABASE_URL or TEST_DATABASE_URL for integration tests',
        );
        return;
      }

      final MarketDataDb db = testDb!.marketDataDb;
      final DateTime now = retentionNow;
      final DateTime cutoff =
          MarketHistorySessionSlot.windowFirstSlotStart(now, 5);

      await db.upsertSnapshot(
        symbol: 'SPY',
        metric: 'bar',
        timeframe: 'sessionHalf',
        asOf: cutoff.subtract(const Duration(days: 2)),
        price: 480,
      );
      await db.upsertSnapshot(
        symbol: 'SPY',
        metric: 'bar',
        timeframe: 'sessionHalf',
        asOf: cutoff,
        price: 490,
      );
      await db.upsertSnapshot(
        symbol: 'SPY',
        metric: 'bar',
        timeframe: 'sessionHalf',
        asOf: cutoff.add(const Duration(hours: 3, minutes: 15)),
        price: 500,
      );

      final MarketDataRetentionResult result = await retention().runCleanup(
        now: now,
      );

      expect(result.error, isNull);
      expect(result.rowsRemoved, 1);

      final Result remaining = await testDb!.connection.execute(
        'SELECT COUNT(*)::int FROM market_data_snapshots',
      );
      expect((remaining.first[0]! as num).toInt(), 2);
    });

    test('deletes prospective questions when older slot is before cutoff',
        () async {
      if (testDb == null) {
        markTestSkipped(
          'Set DATABASE_URL or TEST_DATABASE_URL for integration tests',
        );
        return;
      }

      final DateTime now = retentionNow;
      final DateTime cutoff =
          MarketHistorySessionSlot.windowFirstSlotStart(now, 5);
      final DateTime olderSlot = cutoff.subtract(const Duration(days: 2));
      final DateTime newerSlot = cutoff.add(const Duration(hours: 3, minutes: 15));

      await testDb!.connection.execute(
        Sql.named(
          '''
          INSERT INTO market_history_prospective_questions (
            compare_until, newer_slot_start, older_slot_start,
            symbol, question_text, correct_answer,
            price_delta_pct, volume_delta_pct, avg_volume_usd,
            older_slot, newer_slot
          ) VALUES (
            @compare_until, @newer_slot_start, @older_slot_start,
            'SPY', 'test', 10, 10, 0, 1000,
            '{}'::jsonb, '{}'::jsonb
          )
          ''',
        ),
        parameters: <String, dynamic>{
          'compare_until': newerSlot.add(const Duration(hours: 3, minutes: 15)),
          'newer_slot_start': newerSlot,
          'older_slot_start': olderSlot,
        },
      );

      final MarketDataRetentionResult result = await MarketDataRetention(
        connection: testDb!.connection,
        windowDays: 5,
      ).runCleanup(now: now);

      expect(result.error, isNull);
      expect(result.rowsRemoved, greaterThanOrEqualTo(1));

      final Result remaining = await testDb!.connection.execute(
        'SELECT COUNT(*)::int FROM market_history_prospective_questions',
      );
      expect((remaining.first[0]! as num).toInt(), 0);
    });

    test('empty table returns rowsRemoved 0 without throwing', () async {
      if (testDb == null) {
        markTestSkipped(
          'Set DATABASE_URL or TEST_DATABASE_URL for integration tests',
        );
        return;
      }

      final MarketDataRetentionResult result = await retention().runCleanup(
        now: retentionNow,
      );

      expect(result.rowsRemoved, 0);
      expect(result.error, isNull);
    });

    test('batchSize issues multiple DELETE statements for large backlog',
        () async {
      if (testDb == null) {
        markTestSkipped(
          'Set DATABASE_URL or TEST_DATABASE_URL for integration tests',
        );
        return;
      }

      const int rowCount = 5000;
      const int batchSize = 1000;
      final DateTime now = retentionNow;
      final DateTime oldAsOf = MarketHistorySessionSlot.windowFirstSlotStart(
        now,
        5,
      ).subtract(const Duration(days: 2));

      await testDb!.connection.execute(
        Sql.named(
          '''
          INSERT INTO market_data_snapshots (symbol, metric, timeframe, as_of, price)
          SELECT 'SYM' || g::text, 'bar', '1Day', @as_of, 100
          FROM generate_series(1, @count) AS g
          ''',
        ),
        parameters: <String, dynamic>{
          'as_of': oldAsOf,
          'count': rowCount,
        },
      );

      int deleteStatements = 0;
      final MarketDataRetentionResult result = await MarketDataRetention(
        connection: testDb!.connection,
        batchSize: batchSize,
        onExecute: (String sql) {
          if (sql.toUpperCase().contains('DELETE FROM MARKET_DATA_SNAPSHOTS')) {
            deleteStatements++;
          }
        },
      ).runCleanup(now: now);

      expect(result.rowsRemoved, rowCount);
      expect(deleteStatements, greaterThanOrEqualTo(5));
    });

    test('writes market_data_sync_runs row with kind cleanup and rows_removed',
        () async {
      if (testDb == null) {
        markTestSkipped(
          'Set DATABASE_URL or TEST_DATABASE_URL for integration tests',
        );
        return;
      }

      final DateTime now = retentionNow;
      final DateTime cutoff =
          MarketHistorySessionSlot.windowFirstSlotStart(now, 5);
      await testDb!.marketDataDb.upsertSnapshot(
        symbol: 'SPY',
        metric: 'bar',
        timeframe: 'sessionHalf',
        asOf: cutoff.subtract(const Duration(days: 2)),
        price: 480,
      );

      await retention().runCleanup(now: now);

      final Result runs = await testDb!.connection.execute(
        '''
        SELECT kind, rows_removed, finished_at, error
        FROM market_data_sync_runs
        ORDER BY id DESC
        LIMIT 1
        ''',
      );
      expect(runs, hasLength(1));
      expect(runs.first[0], 'cleanup');
      expect((runs.first[1]! as num).toInt(), 1);
      expect(runs.first[2], isNotNull);
      expect(runs.first[3], isNull);
    });

    test('rows within window are never deleted (by id)', () async {
      if (testDb == null) {
        markTestSkipped(
          'Set DATABASE_URL or TEST_DATABASE_URL for integration tests',
        );
        return;
      }

      final MarketDataDb db = testDb!.marketDataDb;
      final DateTime now = retentionNow;
      final DateTime cutoff =
          MarketHistorySessionSlot.windowFirstSlotStart(now, 5);

      final MarketDataSnapshot kept = await db.upsertSnapshot(
        symbol: 'SPY',
        metric: 'bar',
        timeframe: 'sessionHalf',
        asOf: cutoff.add(const Duration(hours: 3, minutes: 15)),
        price: 500,
      );
      await db.upsertSnapshot(
        symbol: 'SPY',
        metric: 'bar',
        timeframe: 'sessionHalf',
        asOf: cutoff.subtract(const Duration(days: 2)),
        price: 480,
      );

      await retention().runCleanup(now: now);

      final Result row = await testDb!.connection.execute(
        Sql.named('SELECT id FROM market_data_snapshots WHERE id = @id'),
        parameters: <String, dynamic>{'id': kept.id},
      );
      expect(row, hasLength(1));
    });
  });

  group('runArchiveAndCleanup', () {
    test('archiveEnabled copies expired rows then deletes them', () async {
      if (testDb == null) {
        markTestSkipped(
          'Set DATABASE_URL or TEST_DATABASE_URL for integration tests',
        );
        return;
      }

      final DateTime now = retentionNow;
      final DateTime cutoff =
          MarketHistorySessionSlot.windowFirstSlotStart(now, 5);
      await testDb!.marketDataDb.upsertSnapshot(
        symbol: 'SPY',
        metric: 'bar',
        timeframe: 'sessionHalf',
        asOf: cutoff.subtract(const Duration(days: 2)),
        price: 480,
        volume: 1000,
        raw: <String, dynamic>{'c': 480},
      );
      await testDb!.marketDataDb.upsertSnapshot(
        symbol: 'SPY',
        metric: 'bar',
        timeframe: 'sessionHalf',
        asOf: cutoff.add(const Duration(hours: 3, minutes: 15)),
        price: 500,
      );

      final MarketDataRetentionResult result =
          await retention().runArchiveAndCleanup(now: now);

      expect(result.error, isNull);
      expect(result.rowsRemoved, 1);

      final Result archiveCount = await testDb!.connection.execute(
        'SELECT COUNT(*)::int FROM market_data_archive',
      );
      expect((archiveCount.first[0]! as num).toInt(), 1);

      final Result liveCount = await testDb!.connection.execute(
        'SELECT COUNT(*)::int FROM market_data_snapshots',
      );
      expect((liveCount.first[0]! as num).toInt(), 1);
    });

    test('archive failure rolls back delete and records error', () async {
      if (testDb == null) {
        markTestSkipped(
          'Set DATABASE_URL or TEST_DATABASE_URL for integration tests',
        );
        return;
      }

      final DateTime now = retentionNow;
      final DateTime cutoff =
          MarketHistorySessionSlot.windowFirstSlotStart(now, 5);
      await testDb!.marketDataDb.upsertSnapshot(
        symbol: 'SPY',
        metric: 'bar',
        timeframe: 'sessionHalf',
        asOf: cutoff.subtract(const Duration(days: 2)),
        price: 480,
      );

      await testDb!.connection.execute(
        'ALTER TABLE market_data_archive RENAME TO market_data_archive_hidden',
      );

      final MarketDataRetentionResult result =
          await retention().runArchiveAndCleanup(now: now);

      expect(result.error, isNotNull);
      expect(result.rowsRemoved, 0);

      final Result liveCount = await testDb!.connection.execute(
        'SELECT COUNT(*)::int FROM market_data_snapshots',
      );
      expect((liveCount.first[0]! as num).toInt(), 1);

      await testDb!.connection.execute(
        'ALTER TABLE market_data_archive_hidden RENAME TO market_data_archive',
      );
    });

    test('runCleanup does not write to archive table', () async {
      if (testDb == null) {
        markTestSkipped(
          'Set DATABASE_URL or TEST_DATABASE_URL for integration tests',
        );
        return;
      }

      final DateTime now = retentionNow;
      final DateTime cutoff =
          MarketHistorySessionSlot.windowFirstSlotStart(now, 5);
      await testDb!.marketDataDb.upsertSnapshot(
        symbol: 'SPY',
        metric: 'bar',
        timeframe: 'sessionHalf',
        asOf: cutoff.subtract(const Duration(days: 2)),
        price: 480,
      );

      await retention().runCleanup(now: now);

      final Result archiveCount = await testDb!.connection.execute(
        'SELECT COUNT(*)::int FROM market_data_archive',
      );
      expect((archiveCount.first[0]! as num).toInt(), 0);
    });
  });
}