From 7c4f236cb07d263f81676876a1ba0f51fcaeaa47 Mon Sep 17 00:00:00 2001 From: rodribp Date: Mon, 10 Aug 2026 01:43:46 -0600 Subject: [PATCH] add: flag_values table and repository --- backend/src/api/database-migration.ts | 19 ++- .../src/repositories/FlagValueRepository.ts | 133 ++++++++++++++++++ 2 files changed, 151 insertions(+), 1 deletion(-) create mode 100644 backend/src/repositories/FlagValueRepository.ts diff --git a/backend/src/api/database-migration.ts b/backend/src/api/database-migration.ts index 3932179ed..39bd7787f 100644 --- a/backend/src/api/database-migration.ts +++ b/backend/src/api/database-migration.ts @@ -7,7 +7,7 @@ import cpfpRepository from '../repositories/CpfpRepository'; import { RowDataPacket } from 'mysql2'; class DatabaseMigration { - private static currentVersion = 111; + private static currentVersion = 112; private queryTimeout = 3600_000; private statisticsAddedIndexed = false; private uniqueLogs: string[] = []; @@ -1255,6 +1255,11 @@ class DatabaseMigration { await this.$executeQuery('ALTER TABLE `compact_cpfp_clusters` ADD template_algo TINYINT UNSIGNED NOT NULL DEFAULT 0'); await this.updateToSchemaVersion(111); } + + if (databaseSchemaVersion < 112) { + await this.$executeQuery(this.getCreateFlagsValuesTableQuery(), await this.$checkIfTableExists('flag_values')); + await this.updateToSchemaVersion(112); + } } /** @@ -1841,6 +1846,18 @@ class DatabaseMigration { ) ENGINE=InnoDB DEFAULT CHARSET=utf8;`; } + private getCreateFlagsValuesTableQuery(): string { + return `CREATE TABLE IF NOT EXISTS flag_values ( + bucket_size enum('1', '1008', '4032') NOT NULL, + start_height int unsigned NOT NULL, + avg_timestamp timestamp NOT NULL, + flag_value bigint unsigned NOT NULL, + tx_count int unsigned NOT NULL, + vsize_total int unsigned NOT NULL, + PRIMARY KEY (bucket_size, start_height, flag_value) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8`; + } + /** @asyncUnsafe */ public async $blocksReindexingTruncate(): Promise { logger.warn(`Truncating pools, blocks, hashrates and difficulty_adjustments tables for re-indexing (using '--reindex-blocks'). You can cancel this command within 5 seconds`); diff --git a/backend/src/repositories/FlagValueRepository.ts b/backend/src/repositories/FlagValueRepository.ts new file mode 100644 index 000000000..66f8495c1 --- /dev/null +++ b/backend/src/repositories/FlagValueRepository.ts @@ -0,0 +1,133 @@ +import DB from '../database'; +import logger from '../logger'; + +export const INDEXING_PRESETS = [ + {name: 'per block', bucketSize: 1, retentionSpan: 144}, // block span of ~1 day + {name: 'per week', bucketSize: 1008, retentionSpan: -1}, // all + {name: 'per month', bucketSize: 4032, retentionSpan: -1}, // all +]; + +export const INTERVAL_PRESETS = { + '24h': {retentionSpan: 144, bucketSizes: [1]}, + '6m': {retentionSpan: 24192, bucketSizes: [1008, 4032]}, + '1y': {retentionSpan: 48384, bucketSizes: [1008, 4032]}, + '2y': {retentionSpan: 96768, bucketSizes: [1008, 4032]}, + '3y': {retentionSpan: 145152, bucketSizes: [1008, 4032]}, + 'all': {retentionSpan: -1, bucketSizes: [1008, 4032]}, +}; + +class FlagValuesRepository { + /** + * Get the latest indexed day from the database + * + * @asyncSafe */ + public async $getTipAndTailIndexedByBucketSize(bucketSize: number): Promise<{tip: number, tail: number} | null> { + try { + const [rows]: any[] = await DB.query(`SELECT (MAX(start_height) + ?) as tip, MIN(start_height) as tail FROM flag_values WHERE bucket_size = ?`, [bucketSize, bucketSize.toString()]); + if (rows !== null && rows.length > 0 && rows[0].tip !== null && rows[0].tail !== null) { + return rows[0]; + } + } catch (e) { + logger.err(`Cannot get tip and tail indexed from flag_values. Reason: ` + (e instanceof Error ? e.message : e)); + } + return null; + } + + /** + * Get the set of bucket that area already indexed between heights by bucketSize + * + * @asyncSafe */ + public async $getIndexedStartHeights(bucketSize: number, startHeight: number, latestHeight: number): Promise { + try { + const [rows]: any[] = await DB.query( + `SELECT DISTINCT start_height FROM flag_values WHERE bucket_size = ? AND start_height <= ? AND start_height >= ?`, + [bucketSize.toString(), startHeight, latestHeight] + ); + return rows.map(row => row.start_height); + } catch (e) { + logger.err(`Cannot get indexed start heights from flag_values. Reason: ` + (e instanceof Error ? e.message : e)); + } + return []; + } + + public async $saveBatchFlagValues(bucketSize: number, startHeight: number, dataPerFlag: Record>, avgTimestamp: number): Promise { + const params: any[] = []; + const distinctFlags = Object.keys(dataPerFlag); + const avgDate = new Date(Math.round(avgTimestamp) * 1000); + for (const flag of distinctFlags) { + params.push([bucketSize.toString(), startHeight, avgDate, BigInt(flag), dataPerFlag[flag].txCount, dataPerFlag[flag].vSizeTotal]); + } + try { + await DB.query(` + INSERT INTO flag_values (bucket_size, start_height, avg_timestamp, flag_value, tx_count, vsize_total) VALUES ? + ON DUPLICATE KEY UPDATE + avg_timestamp = VALUES(avg_timestamp), tx_count = VALUES(tx_count), vsize_total = VALUES(vsize_total) + `, [params]); + } catch (e) { + logger.debug(`Cannot save flag batched values. Reason: ${e instanceof Error ? e.message : e}`); + throw e; + } + } + + public async $queryTxCountBasedOnMask(mask: bigint, bucketSize: number, op: 'and' | 'or' | 'nor' | undefined, startHeight: number): Promise<{bucketSize: string, startHeight: number, avgTimestamp: number, txCount: number, vSizeTotal: number}[]> { + let flagPredicate = ''; + let params: any[]= []; + switch (op) { + case 'and': { + flagPredicate = 'AND (flag_value & ?) = ?'; + params = [bucketSize.toString(), startHeight, mask, mask]; + } break; + case 'or': { + flagPredicate = 'AND (flag_value & ?) > 0'; + params = [bucketSize.toString(), startHeight, mask]; + } break; + case 'nor': { + flagPredicate = 'AND (flag_value & ?) = 0'; + params = [bucketSize.toString(), startHeight, mask]; + } break; + case undefined: { // op not passed, no boolean operations + params = [bucketSize.toString(), startHeight]; + break; + } + default: throw new Error(`Invalid op '${op}', expected 'and' | 'or' | 'nor' | undefined`); + } + try { + const [rows]: any[] = await DB.query(` + SELECT bucket_size as bucketSize, start_height as startHeight, UNIX_TIMESTAMP(avg_timestamp) as avgTimestamp, + SUM(tx_count) as txCount, SUM(vsize_total) as vSizeTotal + FROM flag_values + WHERE bucket_size = ? AND start_height >= ? ${flagPredicate} + GROUP BY start_height ORDER BY start_height DESC + `, params); + if (rows !== null && rows.length > 0) { + return rows; + } + } catch (e) { + logger.debug(`Cannot get tx counts. Reason: ${e instanceof Error ? e.message : e}`); + } + return []; + } + + /** @asyncSafe */ + public async $deleteFlagValuesBelowHeight(height: number, bucketSize: number): Promise { + try { + await DB.query(`DELETE FROM flag_values WHERE start_height < ? AND bucket_size = ?`, [height, bucketSize.toString()]); + } catch(e) { + logger.err(`Cannot delete flag values below block #${height}. Reason: ` + (e instanceof Error ? e.message : e)); + } + } + + /** @asyncSafe */ + public async $deleteFlagValuesFromHeight(height: number): Promise { + try { + for (const preset of INDEXING_PRESETS) { + const startHeight = Math.floor(height / preset.bucketSize) * preset.bucketSize; + await DB.query(`DELETE FROM flag_values WHERE start_height >= ? AND bucket_size = ?`, [startHeight, preset.bucketSize.toString()]); + } + } catch (e) { + logger.err(`Cannot delete flag values above ${height}. Reason: ` + (e instanceof Error ? e.message : e)); + } + } +} + +export default new FlagValuesRepository();