diff --git a/README.md b/README.md index 7917ff3..4029466 100644 --- a/README.md +++ b/README.md @@ -753,6 +753,16 @@ The API workspace includes a resumable HAF sync skeleton. It reads HAF block-sea npm run indexer:run --workspace=@hivelore/api ``` +The indexer stores the last processed block number, operation index, and block hash. If a later HAF page shows a different hash for the checkpoint block, or the next block points at a different parent hash, the indexer deletes rebuildable `HiveEvent` rows from the fork point onward and replays the canonical rows. Projection writes are keyed by transaction ID and operation index, so replaying the same block range updates existing rows instead of double-counting events. + +To rebuild a range explicitly, pass the first block that should be replayed: + +```bash +npm run indexer:replay --workspace=@hivelore/api -- 123456 +``` + +Rebuild time is bounded by `INDEXER_BATCH_SIZE`, `INDEXER_MAX_BLOCKS_PER_RUN`, HAF latency, and database write throughput. With the default `INDEXER_MAX_BLOCKS_PER_RUN=1000`, one invocation rebuilds at most 1,000 blocks; larger backfills should be run repeatedly or under an always-on worker and timed in staging before relying on a public freshness SLO. + Optional API environment variables: ```env diff --git a/apps/api/package.json b/apps/api/package.json index 755d79a..9be2eae 100644 --- a/apps/api/package.json +++ b/apps/api/package.json @@ -7,6 +7,7 @@ "predev": "prisma generate", "dev": "tsx watch src/server.ts", "indexer:run": "tsx src/indexer/runner.ts", + "indexer:replay": "tsx src/indexer/runner.ts --from-block", "search:reindex": "tsx src/scripts/reindex-search.ts", "smoke:hive-mainnet": "tsx src/scripts/hive-mainnet-smoke.ts", "prisma:generate": "prisma generate", diff --git a/apps/api/prisma/migrations/20260907100000_add_indexer_block_hashes/migration.sql b/apps/api/prisma/migrations/20260907100000_add_indexer_block_hashes/migration.sql new file mode 100644 index 0000000..262300d --- /dev/null +++ b/apps/api/prisma/migrations/20260907100000_add_indexer_block_hashes/migration.sql @@ -0,0 +1,10 @@ +-- Add block linkage metadata so the HAF indexer can detect forked history and replay +-- rebuildable projections from the divergent block. +ALTER TABLE "HiveEvent" +ADD COLUMN "blockHash" TEXT, +ADD COLUMN "previousBlockHash" TEXT; + +ALTER TABLE "IndexerWatermark" +ADD COLUMN "lastProcessedBlockHash" TEXT; + +CREATE INDEX "HiveEvent_blockHash_idx" ON "HiveEvent"("blockHash"); diff --git a/apps/api/prisma/schema.prisma b/apps/api/prisma/schema.prisma index 7beab31..c0f5440 100644 --- a/apps/api/prisma/schema.prisma +++ b/apps/api/prisma/schema.prisma @@ -736,6 +736,8 @@ model HiveReference { model HiveEvent { id String @id @default(cuid()) blockNumber BigInt + blockHash String? + previousBlockHash String? transactionId String operationIndex Int eventType HiveEventType @@ -749,6 +751,7 @@ model HiveEvent { @@unique([transactionId, operationIndex]) @@index([blockNumber, operationIndex]) + @@index([blockHash]) @@index([eventType]) @@index([blockchainTimestamp]) } @@ -757,6 +760,7 @@ model IndexerWatermark { name String @id lastProcessedBlock BigInt @default(0) lastProcessedOperationIndex Int @default(-1) + lastProcessedBlockHash String? lastRunStartedAt DateTime? lastRunFinishedAt DateTime? createdAt DateTime @default(now()) diff --git a/apps/api/src/indexer/haf-sync.test.ts b/apps/api/src/indexer/haf-sync.test.ts index 381fb0f..4f017ca 100644 --- a/apps/api/src/indexer/haf-sync.test.ts +++ b/apps/api/src/indexer/haf-sync.test.ts @@ -2,74 +2,131 @@ import assert from 'node:assert/strict'; import { describe, test } from 'node:test'; import { buildHiveLoreCommentOperation } from '../lib/hive/operations.js'; +import type { HafOperationRow } from '../lib/hive/types.js'; import { HafSyncService, type HafSyncDatabase } from './haf-sync.js'; +const operation = buildHiveLoreCommentOperation({ + author: 'emberquill.dev', + permlink: 'canon-lore', + title: 'Canon Lore', + body: 'A verified entry.', + kind: 'canon_lore', + entityType: 'LORE_ENTRY', + entityId: 'lore-1', +}); + describe('HAF sync service', () => { - test('projects new rows and advances the durable watermark', async () => { - const projectedEvents: unknown[] = []; - const watermarkUpdates: unknown[] = []; - const operation = buildHiveLoreCommentOperation({ - author: 'emberquill.dev', - permlink: 'canon-lore', - title: 'Canon Lore', - body: 'A verified entry.', - kind: 'canon_lore', - entityType: 'LORE_ENTRY', - entityId: 'lore-1', + test('projects new rows and advances the durable hash checkpoint', async () => { + const store = createIndexerStore({ + lastProcessedBlock: 100n, + lastProcessedBlockHash: 'block-100-a', + lastProcessedOperationIndex: 1, }); - const database: HafSyncDatabase = { - hiveEvent: { - async upsert(args) { - projectedEvents.push(args); - }, - }, - indexerWatermark: { - async findUnique() { - return { - lastProcessedBlock: 100n, - lastProcessedOperationIndex: 1, - }; - }, - async upsert(args) { - watermarkUpdates.push(args); - }, - }, - }; const service = new HafSyncService( + createHafClient([ + row({ block: 100, blockHash: 'block-100-a', operationIndex: 1, transactionId: 'skip-me' }), + row({ + block: 100, + blockHash: 'block-100-a', + operationIndex: 2, + transactionId: 'project-me', + }), + ]), + store.database, { - async getHeadBlock() { - return 105; - }, - async searchBlocks() { - return { - operations: [ - { - block: 100, - trx_id: 'skip-me', - op_pos: 1, - timestamp: '2026-07-30T08:00:00.000Z', - op: { - type: 'comment_operation', - value: operation.comment_operation, - }, - }, - { - block: 100, - trx_id: 'project-me', - op_pos: 2, - timestamp: '2026-07-30T08:01:00.000Z', - op: { - type: 'comment_operation', - value: operation.comment_operation, - }, - }, - ], - page: 1, - totalPages: 1, - }; + startBlock: 100, + maxBlocksPerRun: 10, + }, + ); + + const result = await service.runOnce(new Date('2026-07-30T09:00:00.000Z')); + + assert.equal(result.projectedOperations, 1); + assert.equal(result.rolledBackEvents, 0); + assert.equal(store.events.size, 1); + assert.deepEqual(store.watermark, { + lastProcessedBlock: 100n, + lastProcessedBlockHash: 'block-100-a', + lastProcessedOperationIndex: 2, + }); + }); + + test('replays safely when a crash happens after projection but before checkpoint', async () => { + const store = createIndexerStore({ + lastProcessedBlock: 100n, + lastProcessedBlockHash: 'block-100-a', + lastProcessedOperationIndex: 1, + }); + const duplicate = row({ + block: 100, + blockHash: 'block-100-a', + operationIndex: 2, + transactionId: 'project-me', + }); + await store.database.hiveEvent.upsert({ + where: { + transactionId_operationIndex: { + operationIndex: 2, + transactionId: 'project-me', }, }, - database, + create: { + blockHash: 'block-100-a', + blockNumber: 100n, + blockchainTimestamp: new Date('2026-07-30T08:02:00.000Z'), + eventType: 'COMMENT', + operationIndex: 2, + payload: operation, + previousBlockHash: 'block-99-a', + transactionId: 'project-me', + }, + update: { + blockHash: 'block-100-a', + blockNumber: 100n, + blockchainTimestamp: new Date('2026-07-30T08:02:00.000Z'), + eventType: 'COMMENT', + payload: operation, + previousBlockHash: 'block-99-a', + }, + }); + const service = new HafSyncService(createHafClient([duplicate]), store.database, { + startBlock: 100, + maxBlocksPerRun: 10, + }); + + const result = await service.runOnce(new Date('2026-07-30T09:00:00.000Z')); + + assert.equal(result.projectedOperations, 1); + assert.equal(store.events.size, 1); + assert.equal(store.upsertsByKey.get('project-me:2'), 2); + assert.deepEqual(store.watermark, { + lastProcessedBlock: 100n, + lastProcessedBlockHash: 'block-100-a', + lastProcessedOperationIndex: 2, + }); + }); + + test('rewinds and converges when the stored head block hash changes', async () => { + const store = createIndexerStore({ + lastProcessedBlock: 100n, + lastProcessedBlockHash: 'block-100-old', + lastProcessedOperationIndex: 0, + }); + store.events.set('old-tx:0', { + blockNumber: 100n, + operationIndex: 0, + transactionId: 'old-tx', + }); + const service = new HafSyncService( + createHafClient([ + row({ + block: 100, + blockHash: 'block-100-new', + operationIndex: 0, + transactionId: 'new-tx', + }), + ]), + store.database, { startBlock: 100, maxBlocksPerRun: 10, @@ -78,20 +135,170 @@ describe('HAF sync service', () => { const result = await service.runOnce(new Date('2026-07-30T09:00:00.000Z')); + assert.equal(result.rolledBackEvents, 1); assert.equal(result.projectedOperations, 1); - assert.equal(projectedEvents.length, 1); - assert.equal( - ( - watermarkUpdates.find( - (update) => - typeof update === 'object' && - update !== null && - 'update' in update && - (update.update as { lastProcessedOperationIndex?: number }) - .lastProcessedOperationIndex === 2, - ) as { update: { lastProcessedBlock: bigint; lastProcessedOperationIndex: number } } - ).update.lastProcessedBlock, - 100n, + assert.equal(store.events.has('old-tx:0'), false); + assert.equal(store.events.has('new-tx:0'), true); + assert.deepEqual(store.watermark, { + lastProcessedBlock: 100n, + lastProcessedBlockHash: 'block-100-new', + lastProcessedOperationIndex: 0, + }); + }); + + test('rewinds from an explicit replay block before rebuilding projections', async () => { + const store = createIndexerStore({ + lastProcessedBlock: 105n, + lastProcessedBlockHash: 'block-105-a', + lastProcessedOperationIndex: 0, + }); + store.events.set('kept:0', { + blockNumber: 99n, + operationIndex: 0, + transactionId: 'kept', + }); + store.events.set('deleted:0', { + blockNumber: 100n, + operationIndex: 0, + transactionId: 'deleted', + }); + const service = new HafSyncService( + createHafClient([ + row({ + block: 100, + blockHash: 'block-100-a', + operationIndex: 0, + transactionId: 'rebuilt', + }), + ]), + store.database, + { + startBlock: 99, + maxBlocksPerRun: 10, + }, ); + + const result = await service.replayFromBlock(100, new Date('2026-07-30T09:00:00.000Z')); + + assert.equal(result.replayedFromBlock, 100); + assert.equal(result.rolledBackEvents, 1); + assert.equal(result.projectedOperations, 1); + assert.equal(store.events.has('kept:0'), true); + assert.equal(store.events.has('deleted:0'), false); + assert.equal(store.events.has('rebuilt:0'), true); }); }); + +function createHafClient(rows: HafOperationRow[]) { + return { + async getHeadBlock() { + return 105; + }, + async searchBlocks() { + return { + operations: rows, + page: 1, + totalPages: 1, + }; + }, + }; +} + +function createIndexerStore(watermark: { + lastProcessedBlock: bigint; + lastProcessedOperationIndex: number; + lastProcessedBlockHash: string | null; +}) { + const events = new Map< + string, + { + blockNumber: bigint; + transactionId: string; + operationIndex: number; + } + >(); + const upsertsByKey = new Map(); + const store = { + events, + upsertsByKey, + watermark: { ...watermark }, + }; + const database: HafSyncDatabase = { + hiveEvent: { + async deleteMany(args) { + let count = 0; + + for (const [key, event] of events) { + if (event.blockNumber >= args.where.blockNumber.gte) { + events.delete(key); + count += 1; + } + } + + return { count }; + }, + async upsert(args) { + const key = `${args.where.transactionId_operationIndex.transactionId}:${args.where.transactionId_operationIndex.operationIndex}`; + + upsertsByKey.set(key, (upsertsByKey.get(key) ?? 0) + 1); + events.set(key, { + blockNumber: args.create.blockNumber, + operationIndex: args.create.operationIndex, + transactionId: args.create.transactionId, + }); + }, + }, + indexerWatermark: { + async findUnique() { + return store.watermark; + }, + async upsert(args) { + store.watermark = { + lastProcessedBlock: + args.update.lastProcessedBlock ?? + store.watermark.lastProcessedBlock ?? + args.create.lastProcessedBlock, + lastProcessedBlockHash: + args.update.lastProcessedBlockHash === undefined + ? (store.watermark.lastProcessedBlockHash ?? + args.create.lastProcessedBlockHash ?? + null) + : args.update.lastProcessedBlockHash, + lastProcessedOperationIndex: + args.update.lastProcessedOperationIndex ?? + store.watermark.lastProcessedOperationIndex ?? + args.create.lastProcessedOperationIndex, + }; + }, + }, + }; + + return { + database, + events, + upsertsByKey, + get watermark() { + return store.watermark; + }, + }; +} + +function row(input: { + block: number; + blockHash: string; + operationIndex: number; + transactionId: string; +}): HafOperationRow { + return { + block: input.block, + blockHash: input.blockHash, + op: { + type: 'comment_operation', + value: operation.comment_operation, + }, + op_pos: input.operationIndex, + previousBlockHash: `block-${input.block - 1}-a`, + timestamp: '2026-07-30T08:00:00.000Z', + trx_id: input.transactionId, + }; +} diff --git a/apps/api/src/indexer/haf-sync.ts b/apps/api/src/indexer/haf-sync.ts index 263b277..4cd7e45 100644 --- a/apps/api/src/indexer/haf-sync.ts +++ b/apps/api/src/indexer/haf-sync.ts @@ -3,7 +3,15 @@ import { normalizeHafOperation, projectHiveOperation } from '../lib/hive/project import type { HafOperationRow } from '../lib/hive/types.js'; export interface HafSyncDatabase { - hiveEvent: Parameters[0]['hiveEvent']; + hiveEvent: Parameters[0]['hiveEvent'] & { + deleteMany(args: { + where: { + blockNumber: { + gte: bigint; + }; + }; + }): Promise<{ count: number } | unknown>; + }; indexerWatermark: { upsert(args: { where: { @@ -13,12 +21,14 @@ export interface HafSyncDatabase { name: string; lastProcessedBlock: bigint; lastProcessedOperationIndex: number; + lastProcessedBlockHash?: string | null; lastRunStartedAt?: Date; lastRunFinishedAt?: Date; }; update: Partial<{ lastProcessedBlock: bigint; lastProcessedOperationIndex: number; + lastProcessedBlockHash: string | null; lastRunStartedAt: Date; lastRunFinishedAt: Date; }>; @@ -30,11 +40,9 @@ export interface HafSyncDatabase { select: { lastProcessedBlock: true; lastProcessedOperationIndex: true; + lastProcessedBlockHash: true; }; - }): Promise<{ - lastProcessedBlock: bigint; - lastProcessedOperationIndex: number; - } | null>; + }): Promise; }; } @@ -51,6 +59,14 @@ export interface HafSyncResult { toBlock: number; headBlock: number; projectedOperations: number; + rolledBackEvents: number; + replayedFromBlock?: number | undefined; +} + +interface IndexerWatermark { + lastProcessedBlock: bigint; + lastProcessedOperationIndex: number; + lastProcessedBlockHash: string | null; } const DEFAULT_INDEXER_NAME = 'hivelore-haf'; @@ -84,21 +100,59 @@ export class HafSyncService { this.hafClient.getHeadBlock(), this.getWatermark(), ]); + + return this.syncRange({ headBlock, watermark }); + } + + async replayFromBlock(fromBlock: number, now = new Date()): Promise { + if (!Number.isInteger(fromBlock) || fromBlock < this.startBlock) { + throw new Error(`Replay block must be an integer >= ${this.startBlock}.`); + } + + await this.markStarted(now); + + const [headBlock, rewind] = await Promise.all([ + this.hafClient.getHeadBlock(), + this.rewindToBlock(fromBlock), + ]); + const result = await this.syncRange({ + headBlock, + watermark: { + lastProcessedBlock: BigInt(fromBlock), + lastProcessedBlockHash: null, + lastProcessedOperationIndex: -1, + }, + }); + + return { + ...result, + replayedFromBlock: fromBlock, + rolledBackEvents: result.rolledBackEvents + rewind.deletedEvents, + }; + } + + private async syncRange(input: { + headBlock: number; + watermark: IndexerWatermark; + }): Promise { + let watermark = input.watermark; const fromBlock = Math.max(Number(watermark.lastProcessedBlock), this.startBlock); - const toBlock = Math.min(headBlock, fromBlock + this.maxBlocksPerRun - 1); + const toBlock = Math.min(input.headBlock, fromBlock + this.maxBlocksPerRun - 1); if (toBlock < fromBlock) { await this.markFinished(new Date()); return { fromBlock, - toBlock, - headBlock, + headBlock: input.headBlock, projectedOperations: 0, + rolledBackEvents: 0, + toBlock, }; } let projectedOperations = 0; + let rolledBackEvents = 0; let page = 1; while (true) { @@ -109,13 +163,30 @@ export class HafSyncService { page, pageSize: this.batchSize, }); + const forkBlock = findForkBlock(response.operations, watermark); + + if (forkBlock !== undefined) { + const rewind = await this.rewindToBlock(forkBlock); + + rolledBackEvents += rewind.deletedEvents; + watermark = { + lastProcessedBlock: BigInt(forkBlock), + lastProcessedBlockHash: null, + lastProcessedOperationIndex: -1, + }; + } + const rows = response.operations.filter((row) => shouldProcessRow(row, watermark)); for (const row of rows) { const operation = normalizeHafOperation(row); await projectHiveOperation(this.database, operation); - await this.saveWatermark(operation.blockNumber, operation.operationIndex); + await this.saveWatermark( + operation.blockNumber, + operation.operationIndex, + operation.blockHash ?? null, + ); projectedOperations += 1; } @@ -130,16 +201,14 @@ export class HafSyncService { return { fromBlock, - toBlock, - headBlock, + headBlock: input.headBlock, projectedOperations, + rolledBackEvents, + toBlock, }; } - private async getWatermark(): Promise<{ - lastProcessedBlock: bigint; - lastProcessedOperationIndex: number; - }> { + private async getWatermark(): Promise { return ( (await this.database.indexerWatermark.findUnique({ where: { @@ -147,10 +216,12 @@ export class HafSyncService { }, select: { lastProcessedBlock: true, + lastProcessedBlockHash: true, lastProcessedOperationIndex: true, }, })) ?? { lastProcessedBlock: BigInt(this.startBlock), + lastProcessedBlockHash: null, lastProcessedOperationIndex: -1, } ); @@ -164,6 +235,7 @@ export class HafSyncService { create: { name: this.name, lastProcessedBlock: BigInt(this.startBlock), + lastProcessedBlockHash: null, lastProcessedOperationIndex: -1, lastRunStartedAt: startedAt, }, @@ -181,6 +253,7 @@ export class HafSyncService { create: { name: this.name, lastProcessedBlock: BigInt(this.startBlock), + lastProcessedBlockHash: null, lastProcessedOperationIndex: -1, lastRunFinishedAt: finishedAt, }, @@ -190,7 +263,11 @@ export class HafSyncService { }); } - private async saveWatermark(blockNumber: bigint, operationIndex: number): Promise { + private async saveWatermark( + blockNumber: bigint, + operationIndex: number, + blockHash: string | null, + ): Promise { await this.database.indexerWatermark.upsert({ where: { name: this.name, @@ -198,22 +275,83 @@ export class HafSyncService { create: { name: this.name, lastProcessedBlock: blockNumber, + lastProcessedBlockHash: blockHash, lastProcessedOperationIndex: operationIndex, }, update: { lastProcessedBlock: blockNumber, + lastProcessedBlockHash: blockHash, lastProcessedOperationIndex: operationIndex, }, }); } + + private async rewindToBlock(blockNumber: number): Promise<{ deletedEvents: number }> { + const deleted = await this.database.hiveEvent.deleteMany({ + where: { + blockNumber: { + gte: BigInt(blockNumber), + }, + }, + }); + + await this.saveWatermark(BigInt(blockNumber), -1, null); + + return { + deletedEvents: + typeof deleted === 'object' && + deleted !== null && + 'count' in deleted && + typeof deleted.count === 'number' + ? deleted.count + : 0, + }; + } } -function shouldProcessRow( - row: HafOperationRow, - watermark: { lastProcessedBlock: bigint; lastProcessedOperationIndex: number }, -): boolean { +function findForkBlock(rows: HafOperationRow[], watermark: IndexerWatermark): number | undefined { + if (!watermark.lastProcessedBlockHash) { + return undefined; + } + + for (const row of rows) { + const blockNumber = getNumeric(row.block_num ?? row.blockNumber ?? row.block); + + if (blockNumber === undefined || BigInt(blockNumber) < watermark.lastProcessedBlock) { + continue; + } + + const blockHash = getHash(row.block_hash ?? row.blockHash ?? row.block_id ?? row.blockId); + + if (BigInt(blockNumber) === watermark.lastProcessedBlock) { + if (blockHash && blockHash !== watermark.lastProcessedBlockHash) { + return blockNumber; + } + + continue; + } + + const previousBlockHash = getHash( + row.previous_block_hash ?? + row.previousBlockHash ?? + row.previous ?? + row.prev_block ?? + row.prevBlock, + ); + + if (previousBlockHash && previousBlockHash !== watermark.lastProcessedBlockHash) { + return blockNumber - 1; + } + } + + return undefined; +} + +function shouldProcessRow(row: HafOperationRow, watermark: IndexerWatermark): boolean { const blockNumber = getNumeric(row.block_num ?? row.blockNumber ?? row.block); - const operationIndex = getNumeric(row.operation_id ?? row.operationIndex ?? row.op_pos); + const operationIndex = getNumeric( + row.operation_id ?? row.operationIndex ?? row.op_in_trx ?? row.op_pos, + ); if (blockNumber === undefined || operationIndex === undefined) { return true; @@ -244,3 +382,7 @@ function getNumeric(value: unknown): number | undefined { ? numericValue : undefined; } + +function getHash(value: unknown): string | undefined { + return typeof value === 'string' && value.length > 0 ? value : undefined; +} diff --git a/apps/api/src/indexer/runner.ts b/apps/api/src/indexer/runner.ts index 8bb0075..9b76405 100644 --- a/apps/api/src/indexer/runner.ts +++ b/apps/api/src/indexer/runner.ts @@ -23,6 +23,9 @@ const database: HafSyncDatabase = { }, }); }, + async deleteMany(args) { + return prisma.hiveEvent.deleteMany(args); + }, }, indexerWatermark: { async findUnique(args) { @@ -42,9 +45,30 @@ const syncService = new HafSyncService(createHafClient(), database, { }); try { - const result = await syncService.runOnce(); + const replayFromBlock = getReplayFromBlock(process.argv.slice(2)); + const result = + replayFromBlock === undefined + ? await syncService.runOnce() + : await syncService.replayFromBlock(replayFromBlock); console.info('HAF indexer run completed', result); } finally { await prisma.$disconnect(); } + +function getReplayFromBlock(args: string[]): number | undefined { + const replayFlagIndex = args.findIndex((arg) => arg === '--from-block'); + + if (replayFlagIndex === -1) { + return undefined; + } + + const rawBlock = args[replayFlagIndex + 1]; + const fromBlock = rawBlock === undefined ? Number.NaN : Number(rawBlock); + + if (!Number.isInteger(fromBlock) || fromBlock <= 0) { + throw new Error('Usage: npm run indexer:replay --workspace=@hivelore/api -- N'); + } + + return fromBlock; +} diff --git a/apps/api/src/lib/hive/projection.test.ts b/apps/api/src/lib/hive/projection.test.ts index 9c45a54..f06e318 100644 --- a/apps/api/src/lib/hive/projection.test.ts +++ b/apps/api/src/lib/hive/projection.test.ts @@ -98,16 +98,20 @@ describe('HAF projection helpers', () => { }, create: { blockNumber: 456n, + blockHash: null, transactionId: 'def456', operationIndex: 0, + previousBlockHash: null, eventType: 'COMMENT', blockchainTimestamp: new Date('2026-07-25T18:30:00.000Z'), payload: operation, }, update: { blockNumber: 456n, + blockHash: null, eventType: 'COMMENT', blockchainTimestamp: new Date('2026-07-25T18:30:00.000Z'), + previousBlockHash: null, payload: operation, }, }); diff --git a/apps/api/src/lib/hive/projection.ts b/apps/api/src/lib/hive/projection.ts index f8cdc40..5b15074 100644 --- a/apps/api/src/lib/hive/projection.ts +++ b/apps/api/src/lib/hive/projection.ts @@ -8,6 +8,8 @@ export interface HiveProjectionDatabase { where: { transactionId_operationIndex: { transactionId: string; operationIndex: number } }; create: { blockNumber: bigint; + blockHash: string | null; + previousBlockHash: string | null; transactionId: string; operationIndex: number; eventType: HiveProjectionEventType; @@ -16,6 +18,8 @@ export interface HiveProjectionDatabase { }; update: { blockNumber: bigint; + blockHash: string | null; + previousBlockHash: string | null; eventType: HiveProjectionEventType; blockchainTimestamp: Date; payload: unknown; @@ -36,6 +40,7 @@ export function normalizeHafOperation(row: HafOperationRow): NormalizedHiveOpera blockNumber: BigInt( requiredNumeric(row.block_num ?? row.blockNumber ?? row.block, 'block number'), ), + ...optionalBlockHashes(row), transactionId: requiredString( row.transaction_id ?? row.transactionId ?? row.trx_id, 'transaction id', @@ -63,21 +68,48 @@ export async function projectHiveOperation( }, create: { blockNumber: operation.blockNumber, + blockHash: operation.blockHash ?? null, transactionId: operation.transactionId, operationIndex: operation.operationIndex, + previousBlockHash: operation.previousBlockHash ?? null, eventType: toHiveEventType(operation.operationType), blockchainTimestamp: operation.blockchainTimestamp, payload: operation.operation, }, update: { blockNumber: operation.blockNumber, + blockHash: operation.blockHash ?? null, eventType: toHiveEventType(operation.operationType), blockchainTimestamp: operation.blockchainTimestamp, + previousBlockHash: operation.previousBlockHash ?? null, payload: operation.operation, }, }); } +function optionalBlockHashes(row: HafOperationRow): { + blockHash?: string | undefined; + previousBlockHash?: string | undefined; +} { + const blockHash = optionalHash(row.block_hash ?? row.blockHash ?? row.block_id ?? row.blockId); + const previousBlockHash = optionalHash( + row.previous_block_hash ?? + row.previousBlockHash ?? + row.previous ?? + row.prev_block ?? + row.prevBlock, + ); + + return { + ...(blockHash === undefined ? {} : { blockHash }), + ...(previousBlockHash === undefined ? {} : { previousBlockHash }), + }; +} + +function optionalHash(value: unknown): string | undefined { + return typeof value === 'string' && value.length > 0 ? value : undefined; +} + function unwrapHafOperation(row: HafOperationRow): HiveLoreOperation { const source = row.operation ?? row.op ?? row.body; diff --git a/apps/api/src/lib/hive/types.ts b/apps/api/src/lib/hive/types.ts index d5087c0..93de3e8 100644 --- a/apps/api/src/lib/hive/types.ts +++ b/apps/api/src/lib/hive/types.ts @@ -73,6 +73,15 @@ export interface HafOperationRow { block_num?: number | string; blockNumber?: number | string; block?: number | string; + block_hash?: string; + blockHash?: string; + block_id?: string; + blockId?: string; + previous_block_hash?: string; + previousBlockHash?: string; + previous?: string; + prev_block?: string; + prevBlock?: string; transaction_num?: number | string; transaction_id?: string; transactionId?: string; @@ -99,6 +108,8 @@ export interface HafBlockSearchPage { export interface NormalizedHiveOperation { blockNumber: bigint; + blockHash?: string | undefined; + previousBlockHash?: string | undefined; transactionId: string; operationIndex: number; blockchainTimestamp: Date;