diff --git a/Backend/src/app.module.ts b/Backend/src/app.module.ts index 8c6006a2..2a6f342c 100644 --- a/Backend/src/app.module.ts +++ b/Backend/src/app.module.ts @@ -18,6 +18,7 @@ import { CorrelationIdMiddleware } from './common/middleware/correlation-id.midd import { InFlightRequestMiddleware } from './common/shutdown/in-flight.middleware'; import { buildWinstonOptions } from './common/logger/winston.config'; import { IndexerModule } from './indexer/indexer.module'; +import { StatsModule } from './stats/stats.module'; @Module({ imports: [ @@ -64,6 +65,7 @@ import { IndexerModule } from './indexer/indexer.module'; // Enables the Soroban indexer when the application starts IndexerModule, + StatsModule, ], providers: [ diff --git a/Backend/src/database/migrations/1787200000000-CreateIndexedEventsTable.ts b/Backend/src/database/migrations/1787200000000-CreateIndexedEventsTable.ts new file mode 100644 index 00000000..8a43773e --- /dev/null +++ b/Backend/src/database/migrations/1787200000000-CreateIndexedEventsTable.ts @@ -0,0 +1,38 @@ +import { MigrationInterface, QueryRunner } from 'typeorm'; + +export class CreateIndexedEventsTable1787200000000 implements MigrationInterface { + name = 'CreateIndexedEventsTable1787200000000'; + + public async up(queryRunner: QueryRunner): Promise { + await queryRunner.query(` + CREATE TABLE "indexed_events" ( + "id" uuid NOT NULL DEFAULT gen_random_uuid(), + "event_id" character varying NOT NULL, + "ledger" bigint NOT NULL, + "event_type" character varying NOT NULL, + "stellar_gist_id" character varying, + "observed_at" TIMESTAMP NOT NULL DEFAULT now(), + "payload" jsonb, + CONSTRAINT "UQ_indexed_events_event_id" UNIQUE ("event_id"), + CONSTRAINT "PK_indexed_events_id" PRIMARY KEY ("id") + ) + `); + + await queryRunner.query( + `CREATE INDEX "IDX_indexed_events_event_type" ON "indexed_events" ("event_type")`, + ); + await queryRunner.query( + `CREATE INDEX "IDX_indexed_events_stellar_gist_id" ON "indexed_events" ("stellar_gist_id")`, + ); + await queryRunner.query( + `CREATE INDEX "IDX_indexed_events_observed_at" ON "indexed_events" ("observed_at")`, + ); + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query(`DROP INDEX "IDX_indexed_events_observed_at"`); + await queryRunner.query(`DROP INDEX "IDX_indexed_events_stellar_gist_id"`); + await queryRunner.query(`DROP INDEX "IDX_indexed_events_event_type"`); + await queryRunner.query(`DROP TABLE "indexed_events"`); + } +} diff --git a/Backend/src/indexer/contract-indexer.integration-spec.ts b/Backend/src/indexer/contract-indexer.integration-spec.ts index c43f6397..4649336f 100644 --- a/Backend/src/indexer/contract-indexer.integration-spec.ts +++ b/Backend/src/indexer/contract-indexer.integration-spec.ts @@ -95,6 +95,7 @@ describe('Contract-Indexer Integration Flow (integration)', () => { sorobanService, gistRepository, geoService, + { record: jest.fn().mockResolvedValue(undefined) } as any, indexerStateRepo as any, ); }); @@ -107,6 +108,7 @@ describe('Contract-Indexer Integration Flow (integration)', () => { { type: 'gist_posted', ledger: 100, + id: 'evt-int-1', gist: { gistId: stellarGistId, locationCell: 's1t7d8c', @@ -135,6 +137,7 @@ describe('Contract-Indexer Integration Flow (integration)', () => { { type: 'gist_posted', ledger: 100, + id: 'evt-int-1', gist: { gistId: stellarGistId, locationCell: 's1t7d8c', diff --git a/Backend/src/indexer/event-log.repository.spec.ts b/Backend/src/indexer/event-log.repository.spec.ts new file mode 100644 index 00000000..89455a8e --- /dev/null +++ b/Backend/src/indexer/event-log.repository.spec.ts @@ -0,0 +1,54 @@ +import { EventLogRepository } from './event-log.repository'; +import { PG_UNIQUE_VIOLATION } from '../gists/gist.repository'; +import type { GistPostedEvent } from '../soroban/soroban.service'; + +describe('EventLogRepository', () => { + let repository: { create: jest.Mock; insert: jest.Mock }; + let service: EventLogRepository; + + const event: GistPostedEvent = { + type: 'gist_posted', + ledger: 100, + id: 'evt-1', + gist: { + gistId: 'gist-1', + locationCell: 'u4pruyd', + contentHash: 'Qm123', + author: null, + createdAt: 1700000000, + expiresAt: 1700086400, + hidden: false, + }, + }; + + beforeEach(() => { + repository = { + create: jest.fn((data) => data), + insert: jest.fn(), + }; + service = new EventLogRepository(repository as any); + }); + + it('inserts a row keyed by the event id', async () => { + repository.insert.mockResolvedValue(undefined); + + await service.record(event); + + expect(repository.create).toHaveBeenCalledWith( + expect.objectContaining({ eventId: 'evt-1', ledger: 100, eventType: 'gist_posted', stellarGistId: 'gist-1' }), + ); + expect(repository.insert).toHaveBeenCalledTimes(1); + }); + + it('is idempotent: a duplicate event id does not throw', async () => { + repository.insert.mockRejectedValue({ code: PG_UNIQUE_VIOLATION }); + + await expect(service.record(event)).resolves.toBeUndefined(); + }); + + it('re-throws non-duplicate errors', async () => { + repository.insert.mockRejectedValue(new Error('connection lost')); + + await expect(service.record(event)).rejects.toThrow('connection lost'); + }); +}); diff --git a/Backend/src/indexer/event-log.repository.ts b/Backend/src/indexer/event-log.repository.ts new file mode 100644 index 00000000..f8fa83ae --- /dev/null +++ b/Backend/src/indexer/event-log.repository.ts @@ -0,0 +1,56 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; + +import { IndexedEvent } from './indexed-event.entity'; +import { PG_UNIQUE_VIOLATION } from '../gists/gist.repository'; +import type { GistRegistryEvent } from '../soroban/soroban.service'; + +/** + * Extracts the gist id a decoded event refers to, when it has one. + * `gist_posted`/`gist_edited` carry it nested under `gist.gistId`; the + * rest carry a bare `gistId`. + */ +function extractStellarGistId(event: GistRegistryEvent): string | null { + if (event.type === 'gist_posted' || event.type === 'gist_edited') { + return event.gist.gistId ?? null; + } + return 'gistId' in event ? event.gistId : null; +} + +@Injectable() +export class EventLogRepository { + private readonly logger = new Logger(EventLogRepository.name); + + constructor( + @InjectRepository(IndexedEvent) + private readonly repository: Repository, + ) {} + + /** + * Append one decoded event to the log. Idempotent: replaying an event + * with an `id` already on file is a no-op, not an error — matching the + * indexer's "one malformed/duplicate event must never wedge the poll + * loop" design. + */ + async record(event: GistRegistryEvent): Promise { + const entity = this.repository.create({ + eventId: event.id, + ledger: event.ledger, + eventType: event.type, + stellarGistId: extractStellarGistId(event), + payload: event as unknown as Record, + }); + + try { + await this.repository.insert(entity); + } catch (err) { + const code = (err as { code?: string })?.code; + if (code === PG_UNIQUE_VIOLATION) { + this.logger.debug(`Event ${event.id} already logged — duplicate handled`); + return; + } + throw err; + } + } +} diff --git a/Backend/src/indexer/indexed-event.entity.ts b/Backend/src/indexer/indexed-event.entity.ts new file mode 100644 index 00000000..f09ee868 --- /dev/null +++ b/Backend/src/indexer/indexed-event.entity.ts @@ -0,0 +1,44 @@ +import { + Column, + CreateDateColumn, + Entity, + Index, + PrimaryGeneratedColumn, +} from 'typeorm'; + +/** + * Append-only log of every decoded on-chain event, including event types + * that don't otherwise mutate the `gists` table. Lets stats endpoints + * derive counts of edits/hides/reports over time without those actions + * being separately tracked elsewhere. + * + * `event_id` is the Soroban RPC's own unique event id — replaying the + * same event (e.g. after an indexer restart re-fetches a ledger range) + * is idempotent via the unique constraint on that column. + */ +@Entity('indexed_events') +export class IndexedEvent { + @PrimaryGeneratedColumn('uuid') + id: string; + + @Column({ name: 'event_id', unique: true }) + eventId: string; + + @Column({ type: 'bigint' }) + ledger: number; + + @Index() + @Column({ name: 'event_type' }) + eventType: string; + + @Index() + @Column({ name: 'stellar_gist_id', nullable: true }) + stellarGistId: string | null; + + @Index() + @CreateDateColumn({ name: 'observed_at' }) + observedAt: Date; + + @Column({ type: 'jsonb', nullable: true }) + payload: Record | null; +} diff --git a/Backend/src/indexer/indexer.module.ts b/Backend/src/indexer/indexer.module.ts index fe2b38f4..8fca5225 100644 --- a/Backend/src/indexer/indexer.module.ts +++ b/Backend/src/indexer/indexer.module.ts @@ -3,6 +3,8 @@ import { TypeOrmModule } from '@nestjs/typeorm'; import { IndexerService } from './indexer.service'; import { IndexerState } from './indexer-state.entity'; +import { IndexedEvent } from './indexed-event.entity'; +import { EventLogRepository } from './event-log.repository'; import { SorobanModule } from '../soroban/soroban.module'; import { GistsModule } from '../gists/gists.module'; @@ -10,11 +12,12 @@ import { GeoModule } from '../geo/geo.module'; @Module({ imports: [ - TypeOrmModule.forFeature([IndexerState]), + TypeOrmModule.forFeature([IndexerState, IndexedEvent]), SorobanModule, GistsModule, GeoModule, ], - providers: [IndexerService], + providers: [IndexerService, EventLogRepository], + exports: [EventLogRepository], }) export class IndexerModule {} \ No newline at end of file diff --git a/Backend/src/indexer/indexer.service.spec.ts b/Backend/src/indexer/indexer.service.spec.ts index 98aa01cd..972696bd 100644 --- a/Backend/src/indexer/indexer.service.spec.ts +++ b/Backend/src/indexer/indexer.service.spec.ts @@ -24,6 +24,7 @@ function makeEvent( return { type: 'gist_posted', ledger: 100, + id: `evt-${Math.random().toString(36).slice(2)}`, gist: { gistId: 'gist-1', locationCell: 'u4pruyd', @@ -65,6 +66,7 @@ describe('IndexerService', () => { create: jest.Mock; save: jest.Mock; }; + let eventLogRepo: { record: jest.Mock }; let logSpy: ReturnType; let warnSpy: ReturnType; @@ -115,10 +117,15 @@ describe('IndexerService', () => { async (state: Partial) => state, ); + eventLogRepo = { + record: jest.fn().mockResolvedValue(undefined), + }; + service = new IndexerService( soroban, gistRepo, geoService, + eventLogRepo as any, indexerStateRepo as any, ); @@ -576,6 +583,7 @@ describe('IndexerService', () => { { type: 'gist_edited', ledger: 105, + id: 'evt-edited-1', gist: { gistId: 'gist-1', locationCell: 'u4pruyd', @@ -595,8 +603,8 @@ describe('IndexerService', () => { it('handles gist_deleted and gist_removed events by setting is_active = false', async () => { soroban.getEventsSince.mockResolvedValue([ - { type: 'gist_deleted', ledger: 106, gistId: 'gist-1' }, - { type: 'gist_removed', ledger: 107, gistId: 'gist-2' }, + { type: 'gist_deleted', ledger: 106, gistId: 'gist-1', id: 'evt-deleted-1' }, + { type: 'gist_removed', ledger: 107, gistId: 'gist-2', id: 'evt-removed-1' }, ]); await service.poll(); @@ -607,8 +615,8 @@ describe('IndexerService', () => { it('handles gist_hidden and gist_unhidden events by updating hidden column', async () => { soroban.getEventsSince.mockResolvedValue([ - { type: 'gist_hidden', ledger: 108, gistId: 'gist-1' }, - { type: 'gist_unhidden', ledger: 109, gistId: 'gist-1' }, + { type: 'gist_hidden', ledger: 108, gistId: 'gist-1', id: 'evt-hidden-1' }, + { type: 'gist_unhidden', ledger: 109, gistId: 'gist-1', id: 'evt-unhidden-1' }, ]); await service.poll(); @@ -619,7 +627,7 @@ describe('IndexerService', () => { it('handles gist_reported event by updating report count', async () => { soroban.getEventsSince.mockResolvedValue([ - { type: 'gist_reported', ledger: 110, gistId: 'gist-1', count: 3 }, + { type: 'gist_reported', ledger: 110, gistId: 'gist-1', count: 3, id: 'evt-reported-1' }, ]); await service.poll(); diff --git a/Backend/src/indexer/indexer.service.ts b/Backend/src/indexer/indexer.service.ts index 552371cd..a06647c9 100644 --- a/Backend/src/indexer/indexer.service.ts +++ b/Backend/src/indexer/indexer.service.ts @@ -7,6 +7,7 @@ import { SorobanService } from '../soroban/soroban.service'; import { GistRepository, PG_UNIQUE_VIOLATION } from '../gists/gist.repository'; import { GeoService } from '../geo/geo.service'; import { IndexerState } from './indexer-state.entity'; +import { EventLogRepository } from './event-log.repository'; @Injectable() export class IndexerService { @@ -23,6 +24,7 @@ export class IndexerService { private readonly soroban: SorobanService, private readonly gistRepository: GistRepository, private readonly geoService: GeoService, + private readonly eventLogRepository: EventLogRepository, @InjectRepository(IndexerState) private readonly indexerStateRepository: Repository, @@ -75,6 +77,21 @@ export class IndexerService { event.ledger, ); + // Log every decoded event — including types handleEvent doesn't + // otherwise persist to the gists table — so stats can be derived + // from the log alone. Isolated from handleEvent's own + // success/failure: a log-write failure must not skip the + // gists-table update, and vice versa. + try { + await this.eventLogRepository.record(event); + } catch (err) { + this.logger.warn( + `Failed to log event at ledger ${event.ledger}: ${ + err instanceof Error ? err.message : String(err) + }`, + ); + } + try { await this.handleEvent(event); } catch (err) { diff --git a/Backend/src/soroban/soroban.service.ts b/Backend/src/soroban/soroban.service.ts index 51708a7d..0fdaa9d1 100644 --- a/Backend/src/soroban/soroban.service.ts +++ b/Backend/src/soroban/soroban.service.ts @@ -126,36 +126,43 @@ export interface GistPostedEvent { type: 'gist_posted'; gist: GistRecord; ledger: number; + /** Unique Soroban RPC event id (e.g. `-`), used to dedupe the event log. */ + id: string; } export interface GistEditedEvent { type: 'gist_edited'; gist: GistRecord; ledger: number; + id: string; } export interface GistDeletedEvent { type: 'gist_deleted'; gistId: string; ledger: number; + id: string; } export interface GistHiddenEvent { type: 'gist_hidden'; gistId: string; ledger: number; + id: string; } export interface GistUnhiddenEvent { type: 'gist_unhidden'; gistId: string; ledger: number; + id: string; } export interface GistRemovedEvent { type: 'gist_removed'; gistId: string; ledger: number; + id: string; } export interface GistReportedEvent { @@ -163,6 +170,7 @@ export interface GistReportedEvent { gistId: string; count: number; ledger: number; + id: string; } export type GistRegistryEvent = @@ -218,6 +226,7 @@ export function decodeGistRegistryEvent( eventName: string, value: xdr.ScVal | undefined, ledger: number, + id: string, ): GistRegistryEvent | null { if (!value) return null; @@ -229,7 +238,7 @@ export function decodeGistRegistryEvent( case GIST_EVENT_TOPICS.EDITED: { if (!native || typeof native !== 'object') return null; const gist = decodeGistRecord(native as Record); - return { type: eventName, gist, ledger }; + return { type: eventName, gist, ledger, id }; } case GIST_EVENT_TOPICS.DELETED: @@ -242,7 +251,7 @@ export function decodeGistRegistryEvent( if (raw == null) return null; const gistId = decoderReadString(raw); if (gistId === '[object Object]') return null; - return { type: eventName, gistId, ledger }; + return { type: eventName, gistId, ledger, id }; } case GIST_EVENT_TOPICS.REPORTED: { @@ -253,6 +262,7 @@ export function decodeGistRegistryEvent( gistId: decoderReadString(record.gist_id ?? record.gistId), count: decoderReadNumber(record.count), ledger, + id, }; } @@ -887,7 +897,7 @@ export class SorobanService { } const eventName = typeof topic[0] === 'string' ? topic[0] : ''; - return decodeGistRegistryEvent(eventName, event.value, event.ledger); + return decodeGistRegistryEvent(eventName, event.value, event.ledger, event.id); } private readString(value: unknown): string { diff --git a/Backend/src/stats/dto/query-stats-events.dto.ts b/Backend/src/stats/dto/query-stats-events.dto.ts new file mode 100644 index 00000000..213dd710 --- /dev/null +++ b/Backend/src/stats/dto/query-stats-events.dto.ts @@ -0,0 +1,31 @@ +import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; +import { Type } from 'class-transformer'; +import { IsIn, IsISO8601, IsOptional, Max, Min } from 'class-validator'; + +export const STATS_BUCKETS = ['hour', 'day'] as const; +export type StatsBucket = (typeof STATS_BUCKETS)[number]; + +export class QueryStatsEventsDto { + @ApiProperty({ description: 'Range start (ISO 8601)' }) + @IsISO8601() + from: string; + + @ApiProperty({ description: 'Range end (ISO 8601)' }) + @IsISO8601() + to: string; + + @ApiProperty({ description: 'Bucket granularity', enum: STATS_BUCKETS }) + @IsIn(STATS_BUCKETS) + bucket: StatsBucket; + + @ApiPropertyOptional({ description: 'Filter to a single event type' }) + @IsOptional() + type?: string; + + @ApiPropertyOptional({ description: 'Max recent events to return (1-100)', default: 20 }) + @IsOptional() + @Min(1) + @Max(100) + @Type(() => Number) + recentLimit?: number = 20; +} diff --git a/Backend/src/stats/stats.controller.ts b/Backend/src/stats/stats.controller.ts new file mode 100644 index 00000000..6cd2cdf8 --- /dev/null +++ b/Backend/src/stats/stats.controller.ts @@ -0,0 +1,19 @@ +import { Controller, Get, Query } from '@nestjs/common'; +import { ApiOperation, ApiTags } from '@nestjs/swagger'; +import { SkipThrottle } from '@nestjs/throttler'; + +import { StatsService } from './stats.service'; +import { QueryStatsEventsDto } from './dto/query-stats-events.dto'; + +@ApiTags('stats') +@Controller({ path: 'stats', version: '1' }) +export class StatsController { + constructor(private readonly statsService: StatsService) {} + + @Get('events') + @SkipThrottle() + @ApiOperation({ summary: 'Event counts per type per time bucket, plus recent events' }) + getEvents(@Query() query: QueryStatsEventsDto) { + return this.statsService.getEventStats(query); + } +} diff --git a/Backend/src/stats/stats.module.ts b/Backend/src/stats/stats.module.ts new file mode 100644 index 00000000..e6302982 --- /dev/null +++ b/Backend/src/stats/stats.module.ts @@ -0,0 +1,10 @@ +import { Module } from '@nestjs/common'; + +import { StatsService } from './stats.service'; +import { StatsController } from './stats.controller'; + +@Module({ + controllers: [StatsController], + providers: [StatsService], +}) +export class StatsModule {} diff --git a/Backend/src/stats/stats.service.spec.ts b/Backend/src/stats/stats.service.spec.ts new file mode 100644 index 00000000..c9bc346b --- /dev/null +++ b/Backend/src/stats/stats.service.spec.ts @@ -0,0 +1,54 @@ +import { BadRequestException } from '@nestjs/common'; +import { StatsService } from './stats.service'; + +describe('StatsService', () => { + let dataSource: { query: jest.Mock }; + let service: StatsService; + + beforeEach(() => { + dataSource = { query: jest.fn() }; + service = new StatsService(dataSource as any); + }); + + it('rejects a range producing more buckets than the max', async () => { + await expect( + service.getEventStats({ + from: '2020-01-01T00:00:00.000Z', + to: '2021-01-01T00:00:00.000Z', + bucket: 'hour', + } as any), + ).rejects.toThrow(BadRequestException); + }); + + it('zero-fills buckets with no matching rows', async () => { + dataSource.query.mockResolvedValueOnce([]).mockResolvedValueOnce([]); + + const result = await service.getEventStats({ + from: '2026-01-01T00:00:00.000Z', + to: '2026-01-01T02:00:00.000Z', + bucket: 'hour', + type: 'gist_posted', + } as any); + + expect(result.buckets).toHaveLength(2); + expect(result.buckets[0].counts.gist_posted).toBe(0); + }); + + it('returns the recent array from the second query', async () => { + dataSource.query + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([ + { event_type: 'gist_posted', ledger: '10', stellar_gist_id: 'g1', observed_at: new Date() }, + ]); + + const result = await service.getEventStats({ + from: '2026-01-01T00:00:00.000Z', + to: '2026-01-01T01:00:00.000Z', + bucket: 'hour', + } as any); + + expect(result.recent).toEqual([ + expect.objectContaining({ type: 'gist_posted', ledger: 10, stellarGistId: 'g1' }), + ]); + }); +}); diff --git a/Backend/src/stats/stats.service.ts b/Backend/src/stats/stats.service.ts new file mode 100644 index 00000000..b09bc7e2 --- /dev/null +++ b/Backend/src/stats/stats.service.ts @@ -0,0 +1,109 @@ +import { BadRequestException, Injectable } from '@nestjs/common'; +import { InjectDataSource } from '@nestjs/typeorm'; +import { DataSource } from 'typeorm'; + +import { GIST_EVENT_TOPICS } from '../soroban/soroban.service'; +import { QueryStatsEventsDto } from './dto/query-stats-events.dto'; + +const BUCKET_MS: Record = { + hour: 60 * 60 * 1000, + day: 24 * 60 * 60 * 1000, +}; + +const MAX_BUCKETS = 500; +const ALL_EVENT_TYPES = Object.values(GIST_EVENT_TOPICS); + +export interface RecentEvent { + type: string; + ledger: number; + stellarGistId: string | null; + observedAt: Date; +} + +@Injectable() +export class StatsService { + constructor(@InjectDataSource() private readonly dataSource: DataSource) {} + + async getEventStats(query: QueryStatsEventsDto) { + const from = new Date(query.from); + const to = new Date(query.to); + if (Number.isNaN(from.getTime()) || Number.isNaN(to.getTime()) || from >= to) { + throw new BadRequestException('`from` must be a valid ISO date before `to`'); + } + + const bucketMs = BUCKET_MS[query.bucket]; + const bucketCount = Math.ceil((to.getTime() - from.getTime()) / bucketMs); + if (bucketCount > MAX_BUCKETS) { + throw new BadRequestException(`Range too large: ${bucketCount} buckets exceeds max of ${MAX_BUCKETS}`); + } + + const params: unknown[] = [from, to]; + let typeClause = ''; + if (query.type) { + params.push(query.type); + typeClause = `AND event_type = $${params.length}`; + } + + const rows = await this.dataSource.query>( + ` + SELECT date_trunc('${query.bucket}', observed_at) AS bucket, event_type, COUNT(*) AS count + FROM indexed_events + WHERE observed_at >= $1 AND observed_at < $2 ${typeClause} + GROUP BY bucket, event_type + ORDER BY bucket + `, + params, + ); + + const types = query.type ? [query.type] : ALL_EVENT_TYPES; + const countsByBucket = new Map>(); + for (const row of rows) { + const key = row.bucket.toISOString(); + const entry = countsByBucket.get(key) ?? {}; + entry[row.event_type] = Number(row.count); + countsByBucket.set(key, entry); + } + + const buckets: Array<{ bucket: string; counts: Record }> = []; + for (let t = from.getTime(); t < to.getTime(); t += bucketMs) { + const bucketDate = new Date(t); + const key = bucketDate.toISOString(); + const existing = countsByBucket.get(key) ?? {}; + const counts: Record = {}; + for (const type of types) { + counts[type] = existing[type] ?? 0; + } + buckets.push({ bucket: key, counts }); + } + + const recentParams: unknown[] = [from, to]; + let recentTypeClause = ''; + if (query.type) { + recentParams.push(query.type); + recentTypeClause = `AND event_type = $${recentParams.length}`; + } + recentParams.push(query.recentLimit ?? 20); + + const recentRows = await this.dataSource.query< + Array<{ event_type: string; ledger: string; stellar_gist_id: string | null; observed_at: Date }> + >( + ` + SELECT event_type, ledger, stellar_gist_id, observed_at + FROM indexed_events + WHERE observed_at >= $1 AND observed_at < $2 ${recentTypeClause} + ORDER BY observed_at DESC + LIMIT $${recentParams.length} + `, + recentParams, + ); + + const recent: RecentEvent[] = recentRows.map((r) => ({ + type: r.event_type, + ledger: Number(r.ledger), + stellarGistId: r.stellar_gist_id, + observedAt: r.observed_at, + })); + + return { buckets, recent }; + } +}