From 609c95652497204bf7351e3dc64f0ef3e295ff95 Mon Sep 17 00:00:00 2001 From: Fury03 Date: Thu, 27 Aug 2026 20:18:11 +0100 Subject: [PATCH] feat(events): persist last processed ledger for crash recovery Persist the Stellar event poller's last processed ledger to the database after every poll so the service resumes from that checkpoint on restart instead of replaying from the chain tip and dropping events. - add EventPollerCheckpoint model + migration (event_poller_checkpoints) - EventsService loads the checkpoint on boot and resumes from it, with a guard that ignores a checkpoint pointing past the current chain tip - checkpoint is written in a finally block after each poll; write failures are logged and counted but never block polling - expose checkpoint state via GET /health/checkpoint and as an informational indicator in the aggregate GET /health check - specs cover resume, cold start, stale-checkpoint guard, post-poll writes, write-failure tolerance and the health routes --- .../migration.sql | 19 ++ prisma/schema.prisma | 17 ++ src/modules/events/events.module.ts | 1 + src/modules/events/events.service.spec.ts | 186 ++++++++++++++++++ src/modules/events/events.service.ts | 168 +++++++++++++++- src/modules/health/health.controller.spec.ts | 99 ++++++++++ src/modules/health/health.controller.ts | 41 +++- src/modules/health/health.module.ts | 3 +- 8 files changed, 523 insertions(+), 11 deletions(-) create mode 100644 prisma/migrations/20260827000000_add_event_poller_checkpoint/migration.sql create mode 100644 src/modules/events/events.service.spec.ts create mode 100644 src/modules/health/health.controller.spec.ts diff --git a/prisma/migrations/20260827000000_add_event_poller_checkpoint/migration.sql b/prisma/migrations/20260827000000_add_event_poller_checkpoint/migration.sql new file mode 100644 index 0000000..fd5b1e8 --- /dev/null +++ b/prisma/migrations/20260827000000_add_event_poller_checkpoint/migration.sql @@ -0,0 +1,19 @@ +-- Migration: add_event_poller_checkpoint +-- Created for: Issue #80 — Persist last processed ledger for crash recovery + +-- CreateTable: event_poller_checkpoints +CREATE TABLE "event_poller_checkpoints" ( + "id" TEXT NOT NULL, + "key" TEXT NOT NULL, + "lastProcessedLedger" INTEGER NOT NULL DEFAULT 0, + "lastPolledAt" TIMESTAMP(3), + "lastWriteError" TEXT, + "consecutiveWriteFailures" INTEGER NOT NULL DEFAULT 0, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "event_poller_checkpoints_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE UNIQUE INDEX "event_poller_checkpoints_key_key" ON "event_poller_checkpoints"("key"); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index d895353..40ceea1 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -551,6 +551,23 @@ model ChainEvent { @@map("chain_events") } +/// Durable checkpoint for the Stellar event poller (EventsService). +/// A single row (keyed by `key`) records the last Stellar ledger the poller +/// finished processing so the service can resume from it after a restart or +/// crash instead of replaying from the chain tip and missing events. +model EventPollerCheckpoint { + id String @id @default(uuid()) + key String @unique + lastProcessedLedger Int @default(0) + lastPolledAt DateTime? + lastWriteError String? + consecutiveWriteFailures Int @default(0) + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@map("event_poller_checkpoints") +} + model Notification { id String @id @default(uuid()) userId String diff --git a/src/modules/events/events.module.ts b/src/modules/events/events.module.ts index ddebabb..31f603b 100644 --- a/src/modules/events/events.module.ts +++ b/src/modules/events/events.module.ts @@ -10,5 +10,6 @@ import { CoursesModule } from '../courses/courses.module'; imports: [NotificationsModule, EnrollmentsModule, CoursesModule], providers: [EventsService], controllers: [EventsController], + exports: [EventsService], }) export class EventsModule {} diff --git a/src/modules/events/events.service.spec.ts b/src/modules/events/events.service.spec.ts new file mode 100644 index 0000000..99194a3 --- /dev/null +++ b/src/modules/events/events.service.spec.ts @@ -0,0 +1,186 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { PrismaService } from '../../common/prisma/prisma.service'; +import { StellarService } from '../../common/stellar/stellar.service'; +import { NotificationsService } from '../notifications/notifications.service'; +import { CoursesService } from '../courses/courses.service'; +import { EventsService, EVENT_POLLER_CHECKPOINT_KEY } from './events.service'; + +describe('EventsService — ledger checkpoint (Issue #80)', () => { + let service: EventsService; + + const mockPrisma = { + eventPollerCheckpoint: { + findUnique: jest.fn(), + upsert: jest.fn(), + updateMany: jest.fn(), + }, + chainEvent: { + create: jest.fn(), + }, + }; + + const mockStellar = { + getLatestLedger: jest.fn(), + fetchContractEvents: jest.fn(), + }; + + const mockNotifications = { notifyUser: jest.fn() }; + const mockCourses = {}; + + beforeEach(async () => { + const module: TestingModule = await Test.createTestingModule({ + providers: [ + EventsService, + { provide: PrismaService, useValue: mockPrisma }, + { provide: StellarService, useValue: mockStellar }, + { provide: NotificationsService, useValue: mockNotifications }, + { provide: CoursesService, useValue: mockCourses }, + ], + }).compile(); + + service = module.get(EventsService); + jest.clearAllMocks(); + mockPrisma.eventPollerCheckpoint.upsert.mockResolvedValue({}); + mockPrisma.eventPollerCheckpoint.updateMany.mockResolvedValue({ count: 1 }); + mockPrisma.chainEvent.create.mockResolvedValue({}); + }); + + describe('onModuleInit — resume from checkpoint', () => { + it('resumes from the persisted ledger instead of the chain tip', async () => { + mockPrisma.eventPollerCheckpoint.findUnique.mockResolvedValue({ + key: EVENT_POLLER_CHECKPOINT_KEY, + lastProcessedLedger: 4242, + }); + mockStellar.getLatestLedger.mockResolvedValue(9999); + + await service.onModuleInit(); + + const status = await service.getCheckpointStatus(); + expect(status.lastProcessedLedger).toBe(4242); + expect(status.resumedFromCheckpoint).toBe(true); + }); + + it('cold-starts near the chain tip and writes an initial checkpoint when none exists', async () => { + mockPrisma.eventPollerCheckpoint.findUnique.mockResolvedValue(null); + mockStellar.getLatestLedger.mockResolvedValue(1000); + + await service.onModuleInit(); + + const status = await service.getCheckpointStatus(); + expect(status.lastProcessedLedger).toBe(990); // 1000 - COLD_START_LOOKBACK + expect(status.resumedFromCheckpoint).toBe(false); + expect(mockPrisma.eventPollerCheckpoint.upsert).toHaveBeenCalledWith( + expect.objectContaining({ + where: { key: EVENT_POLLER_CHECKPOINT_KEY }, + }), + ); + }); + + it('ignores a checkpoint that points past the chain tip (stale / wrong network)', async () => { + mockPrisma.eventPollerCheckpoint.findUnique.mockResolvedValue({ + key: EVENT_POLLER_CHECKPOINT_KEY, + lastProcessedLedger: 50_000, + }); + mockStellar.getLatestLedger.mockResolvedValue(1000); + + await service.onModuleInit(); + + const status = await service.getCheckpointStatus(); + expect(status.lastProcessedLedger).toBe(990); + expect(status.resumedFromCheckpoint).toBe(false); + }); + }); + + describe('pollEvents — checkpoint writes after each poll', () => { + beforeEach(async () => { + mockPrisma.eventPollerCheckpoint.findUnique.mockResolvedValue(null); + mockStellar.getLatestLedger.mockResolvedValue(100); + await service.onModuleInit(); + jest.clearAllMocks(); + mockPrisma.eventPollerCheckpoint.upsert.mockResolvedValue({}); + mockPrisma.chainEvent.create.mockResolvedValue({}); + }); + + it('advances and persists the ledger after processing events', async () => { + mockStellar.fetchContractEvents.mockResolvedValue([ + { ledger: 105, topic: ['course_registered'], value: 'COURSE-1', txHash: 'a' }, + { ledger: 107, topic: ['course_registered'], value: 'COURSE-2', txHash: 'b' }, + ]); + + await service.pollEvents(); + + expect(mockPrisma.eventPollerCheckpoint.upsert).toHaveBeenCalledWith( + expect.objectContaining({ + update: expect.objectContaining({ lastProcessedLedger: 108 }), + }), + ); + const status = await service.getCheckpointStatus(); + expect(status.lastProcessedLedger).toBe(108); + }); + + it('still writes a checkpoint when a poll finds no events', async () => { + mockStellar.fetchContractEvents.mockResolvedValue([]); + + await service.pollEvents(); + + expect(mockPrisma.eventPollerCheckpoint.upsert).toHaveBeenCalledTimes(1); + }); + }); + + describe('pollEvents — checkpoint write failures do not block polling', () => { + beforeEach(async () => { + mockPrisma.eventPollerCheckpoint.findUnique.mockResolvedValue(null); + mockStellar.getLatestLedger.mockResolvedValue(100); + await service.onModuleInit(); + jest.clearAllMocks(); + mockPrisma.chainEvent.create.mockResolvedValue({}); + }); + + it('swallows the write error, records the failure and keeps advancing in memory', async () => { + mockStellar.fetchContractEvents.mockResolvedValue([ + { ledger: 110, topic: ['course_registered'], value: 'COURSE-1', txHash: 'a' }, + ]); + mockPrisma.eventPollerCheckpoint.upsert.mockRejectedValue(new Error('db down')); + mockPrisma.eventPollerCheckpoint.updateMany.mockResolvedValue({ count: 1 }); + + await expect(service.pollEvents()).resolves.not.toThrow(); + + const status = await service.getCheckpointStatus(); + expect(status.consecutiveWriteFailures).toBe(1); + expect(status.lastWriteError).toContain('db down'); + expect(status.healthy).toBe(false); + // in-memory progress is not lost + expect(status.lastProcessedLedger).toBe(111); + }); + + it('recovers (failure counter resets) once a later write succeeds', async () => { + mockStellar.fetchContractEvents.mockResolvedValue([]); + mockPrisma.eventPollerCheckpoint.upsert + .mockRejectedValueOnce(new Error('transient')) + .mockResolvedValueOnce({}); + mockPrisma.eventPollerCheckpoint.updateMany.mockResolvedValue({ count: 1 }); + + await service.pollEvents(); + expect((await service.getCheckpointStatus()).consecutiveWriteFailures).toBe(1); + + await service.pollEvents(); + const status = await service.getCheckpointStatus(); + expect(status.consecutiveWriteFailures).toBe(0); + expect(status.healthy).toBe(true); + }); + }); + + describe('getCheckpointStatus', () => { + it('reports the persisted ledger from the database', async () => { + mockPrisma.eventPollerCheckpoint.findUnique.mockResolvedValue({ + key: EVENT_POLLER_CHECKPOINT_KEY, + lastProcessedLedger: 777, + lastPolledAt: new Date('2026-08-27T00:00:00Z'), + }); + + const status = await service.getCheckpointStatus(); + expect(status.persistedLedger).toBe(777); + expect(status.key).toBe(EVENT_POLLER_CHECKPOINT_KEY); + }); + }); +}); diff --git a/src/modules/events/events.service.ts b/src/modules/events/events.service.ts index 5f098be..42b1030 100644 --- a/src/modules/events/events.service.ts +++ b/src/modules/events/events.service.ts @@ -6,6 +6,27 @@ import { NotificationsService } from '../notifications/notifications.service'; import { CoursesService } from '../courses/courses.service'; import { NotificationType } from '@prisma/client'; +/** Fixed primary key for the singleton poller checkpoint row. */ +export const EVENT_POLLER_CHECKPOINT_KEY = 'stellar-event-poller'; + +/** Number of ledgers to rewind behind the chain tip on a cold start. */ +const COLD_START_LOOKBACK = 10; + +export interface CheckpointStatus { + key: string; + /** Ledger the running process will poll from next. */ + lastProcessedLedger: number; + /** Ledger currently persisted in the database (null if never written). */ + persistedLedger: number | null; + lastPolledAt: Date | null; + /** How the poller obtained its starting ledger on the last boot. */ + resumedFromCheckpoint: boolean; + consecutiveWriteFailures: number; + lastWriteError: string | null; + /** true when the last checkpoint write succeeded. */ + healthy: boolean; +} + /** * EventsService * @@ -19,12 +40,22 @@ import { NotificationType } from '@prisma/client'; * - certificate_revoked → update DB flag * - course_paused → update course status * - course_archived → update course status + * + * The last processed Stellar ledger is persisted to the `event_poller_checkpoints` + * table after every poll. On boot the service resumes from that checkpoint so a + * restart or crash does not skip events that landed while the process was down. */ @Injectable() export class EventsService implements OnModuleInit { private readonly logger = new Logger(EventsService.name); private lastProcessedLedger = 0; + // Checkpoint bookkeeping (in-memory mirror of the DB row). + private resumedFromCheckpoint = false; + private consecutiveWriteFailures = 0; + private lastWriteError: string | null = null; + private lastPolledAt: Date | null = null; + constructor( private readonly prisma: PrismaService, private readonly stellar: StellarService, @@ -33,33 +64,152 @@ export class EventsService implements OnModuleInit { ) {} async onModuleInit() { + const checkpoint = await this.loadCheckpoint(); + let latestLedger: number | null = null; try { - const latest = await this.stellar.getLatestLedger(); - this.lastProcessedLedger = Math.max(1, latest - 10); - this.logger.log(`Event poller initialised at ledger ${this.lastProcessedLedger}`); + latestLedger = await this.stellar.getLatestLedger(); } catch { - this.lastProcessedLedger = 1; this.logger.warn('Could not fetch latest ledger on init'); } + + if (checkpoint && checkpoint.lastProcessedLedger > 0) { + // Guard against a corrupt / stale checkpoint that points past the chain + // tip (e.g. restored from another network) — fall back to a cold start. + if (latestLedger !== null && checkpoint.lastProcessedLedger > latestLedger + 1) { + this.lastProcessedLedger = Math.max(1, latestLedger - COLD_START_LOOKBACK); + this.resumedFromCheckpoint = false; + this.logger.warn( + `Persisted checkpoint ledger ${checkpoint.lastProcessedLedger} is ahead of ` + + `chain tip ${latestLedger}; ignoring it and restarting near the tip at ` + + `${this.lastProcessedLedger}`, + ); + } else { + this.lastProcessedLedger = checkpoint.lastProcessedLedger; + this.resumedFromCheckpoint = true; + this.logger.log( + `Resumed event poller from persisted ledger ${this.lastProcessedLedger}`, + ); + } + return; + } + + // Cold start — no usable checkpoint yet. + this.lastProcessedLedger = + latestLedger !== null ? Math.max(1, latestLedger - COLD_START_LOOKBACK) : 1; + this.resumedFromCheckpoint = false; + this.logger.log(`Event poller initialised at ledger ${this.lastProcessedLedger}`); + await this.persistCheckpoint(); } @Cron(CronExpression.EVERY_5_SECONDS) async pollEvents() { try { const events = await this.stellar.fetchContractEvents(this.lastProcessedLedger); - if (!events.length) return; - this.logger.log(`Processing ${events.length} chain event(s)`); + if (events.length) { + this.logger.log(`Processing ${events.length} chain event(s)`); - for (const event of events) { - await this.processEvent(event); - this.lastProcessedLedger = Math.max(this.lastProcessedLedger, event.ledger + 1); + for (const event of events) { + await this.processEvent(event); + this.lastProcessedLedger = Math.max(this.lastProcessedLedger, event.ledger + 1); + } } } catch (error) { this.logger.error('Event polling failed', error.message); + } finally { + // Persist progress even when a poll finds nothing — a checkpoint write + // failure is logged but never allowed to stop the poll loop. + this.lastPolledAt = new Date(); + await this.persistCheckpoint(); } } + // ---------------------------------------------------------- + // CHECKPOINT PERSISTENCE (Issue #80 — crash recovery) + // ---------------------------------------------------------- + + /** Read the persisted checkpoint row, tolerating any DB error. */ + private async loadCheckpoint() { + try { + return await this.prisma.eventPollerCheckpoint.findUnique({ + where: { key: EVENT_POLLER_CHECKPOINT_KEY }, + }); + } catch (error) { + this.logger.error('Could not read event poller checkpoint', error.message); + return null; + } + } + + /** + * Write the current ledger to the checkpoint row. Best-effort: a failure + * increments the failure counter and is surfaced via the health endpoint, + * but is swallowed so polling keeps running. + */ + private async persistCheckpoint(): Promise { + try { + await this.prisma.eventPollerCheckpoint.upsert({ + where: { key: EVENT_POLLER_CHECKPOINT_KEY }, + create: { + key: EVENT_POLLER_CHECKPOINT_KEY, + lastProcessedLedger: this.lastProcessedLedger, + lastPolledAt: this.lastPolledAt, + }, + update: { + lastProcessedLedger: this.lastProcessedLedger, + lastPolledAt: this.lastPolledAt, + lastWriteError: null, + consecutiveWriteFailures: 0, + }, + }); + this.consecutiveWriteFailures = 0; + this.lastWriteError = null; + } catch (error) { + this.consecutiveWriteFailures += 1; + this.lastWriteError = error?.message ?? String(error); + this.logger.error( + `Failed to persist ledger checkpoint (consecutive failure ` + + `#${this.consecutiveWriteFailures}); polling continues`, + this.lastWriteError, + ); + // Try to record the failure on the row itself, but do not care if this + // also fails — the poll loop must not be blocked by checkpoint I/O. + try { + await this.prisma.eventPollerCheckpoint.updateMany({ + where: { key: EVENT_POLLER_CHECKPOINT_KEY }, + data: { + lastWriteError: this.lastWriteError.slice(0, 500), + consecutiveWriteFailures: this.consecutiveWriteFailures, + }, + }); + } catch { + /* already logged above */ + } + } + } + + /** Current checkpoint state, exposed through GET /health/checkpoint. */ + async getCheckpointStatus(): Promise { + let persisted: Awaited> = null; + try { + persisted = await this.prisma.eventPollerCheckpoint.findUnique({ + where: { key: EVENT_POLLER_CHECKPOINT_KEY }, + }); + } catch (error) { + this.logger.error('Could not read checkpoint for health report', error.message); + } + + return { + key: EVENT_POLLER_CHECKPOINT_KEY, + lastProcessedLedger: this.lastProcessedLedger, + persistedLedger: persisted?.lastProcessedLedger ?? null, + lastPolledAt: persisted?.lastPolledAt ?? this.lastPolledAt, + resumedFromCheckpoint: this.resumedFromCheckpoint, + consecutiveWriteFailures: this.consecutiveWriteFailures, + lastWriteError: this.lastWriteError, + healthy: this.consecutiveWriteFailures === 0, + }; + } + private async processEvent(event: any) { const eventName = this.extractEventName(event); const payload = this.extractPayload(event); diff --git a/src/modules/health/health.controller.spec.ts b/src/modules/health/health.controller.spec.ts new file mode 100644 index 0000000..964eb8e --- /dev/null +++ b/src/modules/health/health.controller.spec.ts @@ -0,0 +1,99 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { TerminusModule } from '@nestjs/terminus'; +import { HealthController } from './health.controller'; +import { PrismaService } from '../../common/prisma/prisma.service'; +import { EventsService } from '../events/events.service'; +import type { CheckpointStatus } from '../events/events.service'; + +/** + * Exercises the real health routes end-to-end: the controller is resolved from + * a Nest module with the real TerminusModule wired in, so `check()` runs the + * actual HealthCheckService aggregation and `checkpoint()` runs the real + * handler that GET /health/checkpoint is bound to. + */ +describe('HealthController (Issue #80 — checkpoint via health endpoint)', () => { + let controller: HealthController; + + const healthyStatus: CheckpointStatus = { + key: 'stellar-event-poller', + lastProcessedLedger: 5123, + persistedLedger: 5123, + lastPolledAt: new Date('2026-08-27T12:00:00Z'), + resumedFromCheckpoint: true, + consecutiveWriteFailures: 0, + lastWriteError: null, + healthy: true, + }; + + const mockEvents = { getCheckpointStatus: jest.fn() }; + // PrismaHealthIndicator.pingDb tries $runCommandRaw first, then falls back to + // $queryRawUnsafe('SELECT 1') for SQL providers. + const mockPrisma = { + $runCommandRaw: jest + .fn() + .mockRejectedValue(new Error('Use the mongodb provider')), + $queryRawUnsafe: jest.fn().mockResolvedValue([{ result: 1 }]), + }; + + beforeEach(async () => { + const module: TestingModule = await Test.createTestingModule({ + imports: [TerminusModule], + controllers: [HealthController], + providers: [ + { provide: PrismaService, useValue: mockPrisma }, + { provide: EventsService, useValue: mockEvents }, + ], + }).compile(); + + controller = module.get(HealthController); + jest.clearAllMocks(); + mockPrisma.$runCommandRaw.mockRejectedValue(new Error('Use the mongodb provider')); + mockPrisma.$queryRawUnsafe.mockResolvedValue([{ result: 1 }]); + }); + + it('GET /health/checkpoint returns the current ledger checkpoint payload', async () => { + mockEvents.getCheckpointStatus.mockResolvedValue(healthyStatus); + + const result = await controller.checkpoint(); + + expect(result).toEqual(healthyStatus); + expect(mockEvents.getCheckpointStatus).toHaveBeenCalledTimes(1); + }); + + it('GET /health includes the event poller checkpoint indicator and stays 200 (ok)', async () => { + mockEvents.getCheckpointStatus.mockResolvedValue(healthyStatus); + + const result = await controller.check(); + + expect(result.status).toBe('ok'); + expect(result.info?.event_poller_checkpoint).toEqual( + expect.objectContaining({ + status: 'up', + checkpointHealthy: true, + lastProcessedLedger: 5123, + persistedLedger: 5123, + }), + ); + }); + + it('GET /health surfaces a degraded checkpoint without failing the overall check', async () => { + mockEvents.getCheckpointStatus.mockResolvedValue({ + ...healthyStatus, + consecutiveWriteFailures: 3, + lastWriteError: 'connection reset', + healthy: false, + }); + + const result = await controller.check(); + + expect(result.status).toBe('ok'); + expect(result.info?.event_poller_checkpoint).toEqual( + expect.objectContaining({ + status: 'up', + checkpointHealthy: false, + consecutiveWriteFailures: 3, + lastWriteError: 'connection reset', + }), + ); + }); +}); diff --git a/src/modules/health/health.controller.ts b/src/modules/health/health.controller.ts index 63a1985..f96496d 100644 --- a/src/modules/health/health.controller.ts +++ b/src/modules/health/health.controller.ts @@ -1,7 +1,13 @@ import { Controller, Get } from '@nestjs/common'; -import { HealthCheck, HealthCheckService, PrismaHealthIndicator } from '@nestjs/terminus'; +import { + HealthCheck, + HealthCheckService, + HealthIndicatorResult, + PrismaHealthIndicator, +} from '@nestjs/terminus'; import { ApiTags, ApiOperation } from '@nestjs/swagger'; import { PrismaService } from '../../common/prisma/prisma.service'; +import { EventsService } from '../events/events.service'; @ApiTags('health') @Controller('health') @@ -10,6 +16,7 @@ export class HealthController { private readonly health: HealthCheckService, private readonly prismaHealth: PrismaHealthIndicator, private readonly prisma: PrismaService, + private readonly events: EventsService, ) {} @Get() @@ -18,6 +25,38 @@ export class HealthController { check() { return this.health.check([ () => this.prismaHealth.pingCheck('database', this.prisma), + () => this.eventPollerCheckpointIndicator(), ]); } + + @Get('checkpoint') + @ApiOperation({ + summary: 'Current Stellar event-poller ledger checkpoint (crash recovery)', + }) + checkpoint() { + return this.events.getCheckpointStatus(); + } + + /** + * Reports the event-poller checkpoint inside the aggregate health check. + * Informational only: it always reports `up` (with a `checkpointHealthy` + * flag and failure details) so a transient checkpoint write failure on the + * poller never sends a 503 that pulls the whole service out of rotation. + * Use GET /health/checkpoint for the authoritative checkpoint state. + */ + private async eventPollerCheckpointIndicator(): Promise { + const status = await this.events.getCheckpointStatus(); + return { + event_poller_checkpoint: { + status: 'up', + checkpointHealthy: status.healthy, + lastProcessedLedger: status.lastProcessedLedger, + persistedLedger: status.persistedLedger, + lastPolledAt: status.lastPolledAt, + resumedFromCheckpoint: status.resumedFromCheckpoint, + consecutiveWriteFailures: status.consecutiveWriteFailures, + lastWriteError: status.lastWriteError, + }, + }; + } } diff --git a/src/modules/health/health.module.ts b/src/modules/health/health.module.ts index 79578ed..33d61af 100644 --- a/src/modules/health/health.module.ts +++ b/src/modules/health/health.module.ts @@ -2,9 +2,10 @@ import { Module } from '@nestjs/common'; import { TerminusModule } from '@nestjs/terminus'; import { HealthController } from './health.controller'; +import { EventsModule } from '../events/events.module'; @Module({ - imports: [TerminusModule], + imports: [TerminusModule, EventsModule], controllers: [HealthController], }) export class HealthModule {}