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 {}