storeBlockHeadersBulk method

@override Future<void> storeBlockHeadersBulk(List<(BlockHeader, int)> headers)

Bulk store block headers: a CDN import, or a run of headers a peer sent (BlockHeaderChain.acceptHeaders).

Same upsert semantics as storeBlockHeader.

Parameters:

• headers: List of (BlockHeader, height) pairs to store

Implementations should use batch/transaction writes for performance. Default implementation falls back to sequential storeBlockHeader calls.

Implementation

@override
Future<void> storeBlockHeadersBulk(List<(BlockHeader, int)> headers) async {
  _ensureInitialized();
  if (headers.isEmpty) return;

  final now = DateTime.now();

  // One upsert row per hash; the last occurrence wins. A multi-row
  // ON CONFLICT DO UPDATE may not touch the same row twice.
  final byHash = <String, (BlockHeader, int)>{};
  for (final entry in headers) {
    byHash[entry.$1.blockHash().toString()] = entry;
  }
  final rows = byHash.entries.toList();
  final chunkSize = headerInsertChunkSize < 1 ? 1 : headerInsertChunkSize;

  // One transaction per batch (a failure leaves no partial chain) and one
  // multi-row INSERT per chunk: a CDN sync was one round-trip per header
  // (audit S-08). Upsert semantics as storeBlockHeader (S-12).
  await _pool!.runTx((session) async {
    for (var start = 0; start < rows.length; start += chunkSize) {
      final chunk = rows.sublist(start, (start + chunkSize).clamp(0, rows.length));
      final values = <String>[];
      final params = <String, dynamic>{'storedAt': now};
      for (var i = 0; i < chunk.length; i++) {
        final hash = chunk[i].key;
        final (header, height) = chunk[i].value;
        values.add('(@h$i, @hash$i, @prev$i, @merkle$i, @ts$i, '
            '@version$i, @bits$i, @nonce$i, false, @storedAt)');
        params['h$i'] = height;
        params['hash$i'] = hash;
        params['prev$i'] = header.prevBlock.toString();
        params['merkle$i'] = header.merkleRoot.toString();
        params['ts$i'] = header.timestamp.millisecondsSinceEpoch ~/ 1000;
        params['version$i'] = header.version;
        params['bits$i'] = header.bits;
        params['nonce$i'] = header.nonce;
      }
      await session.execute(
        Sql.named('''
          INSERT INTO block_headers (
            height, hash, prev_block_hash, merkle_root, timestamp,
            version, bits, nonce, is_orphaned, stored_at
          ) VALUES ${values.join(', ')}
          ON CONFLICT (hash) DO UPDATE SET
            is_orphaned = false,
            height = EXCLUDED.height
        '''),
        parameters: params,
      );
      onHeaderInsertStatement?.call(chunk.length);
    }
  });
}