diff --git a/src/config/configuration.ts b/src/config/configuration.ts
index 282c360..f351aa1 100644
--- a/src/config/configuration.ts
+++ b/src/config/configuration.ts
@@ -72,6 +72,56 @@ export default () => ({
     eventPageLimit: parseInt(process.env.SOROBAN_EVENT_PAGE_LIMIT ?? '100', 10),
   },
 
+  blockchainIndexer: {
+    enabled: process.env.BLOCKCHAIN_INDEXER_ENABLED === 'true',
+    pollIntervalMs: parseInt(
+      process.env.BLOCKCHAIN_INDEXER_POLL_INTERVAL_MS ?? '2000',
+      10,
+    ),
+    maxBackfillLedgers: parseInt(
+      process.env.BLOCKCHAIN_INDEXER_MAX_BACKFILL_LEDGERS ?? '1000',
+      10,
+    ),
+    includeFailed: process.env.BLOCKCHAIN_INDEXER_INCLUDE_FAILED === 'true',
+    pageLimit: parseInt(process.env.BLOCKCHAIN_INDEXER_PAGE_LIMIT ?? '200', 10),
+    streamTtlSecs: parseInt(
+      process.env.BLOCKCHAIN_INDEXER_STREAM_TTL_SECS ?? '300',
+      10,
+    ),
+    // Real-time WebSocket streaming settings
+    wsReconnectBaseDelayMs: parseInt(
+      process.env.BLOCKCHAIN_INDEXER_WS_RECONNECT_BASE_MS ?? '1000',
+      10,
+    ),
+    wsReconnectMaxDelayMs: parseInt(
+      process.env.BLOCKCHAIN_INDEXER_WS_RECONNECT_MAX_MS ?? '30000',
+      10,
+    ),
+    // In-memory event buffer
+    eventBufferSize: parseInt(
+      process.env.BLOCKCHAIN_INDEXER_BUFFER_SIZE ?? '10000',
+      10,
+    ),
+    eventBufferTtlMs: parseInt(
+      process.env.BLOCKCHAIN_INDEXER_BUFFER_TTL_MS ?? '60000',
+      10,
+    ),
+    // Batched persistence
+    batchFlushIntervalMs: parseInt(
+      process.env.BLOCKCHAIN_INDEXER_BATCH_FLUSH_MS ?? '1000',
+      10,
+    ),
+    batchMaxSize: parseInt(
+      process.env.BLOCKCHAIN_INDEXER_BATCH_MAX_SIZE ?? '1000',
+      10,
+    ),
+    // Retention policy
+    retentionDays: parseInt(
+      process.env.BLOCKCHAIN_INDEXER_RETENTION_DAYS ?? '90',
+      10,
+    ),
+  },
+
   rateLimit: {
     strategy: process.env.RATE_LIMIT_STRATEGY ?? 'sliding_window',
     defaultWindowSize: parseInt(process.env.RATE_LIMIT_DEFAULT_WINDOW ?? '60', 10),
diff --git a/src/config/env.validation.ts b/src/config/env.validation.ts
index a4ae25f..2027fa8 100644
--- a/src/config/env.validation.ts
+++ b/src/config/env.validation.ts
@@ -50,6 +50,14 @@ export const envValidationSchema = Joi.object({
   BLOCKCHAIN_INDEXER_MAX_BACKFILL_LEDGERS: Joi.number().default(1000),
   BLOCKCHAIN_INDEXER_STREAM_TTL_SECS: Joi.number().default(300),
   BLOCKCHAIN_INDEXER_INCLUDE_FAILED: Joi.boolean().default(true),
+  BLOCKCHAIN_INDEXER_PAGE_LIMIT: Joi.number().default(200),
+  BLOCKCHAIN_INDEXER_WS_RECONNECT_BASE_MS: Joi.number().default(1000),
+  BLOCKCHAIN_INDEXER_WS_RECONNECT_MAX_MS: Joi.number().default(30000),
+  BLOCKCHAIN_INDEXER_BUFFER_SIZE: Joi.number().default(10000),
+  BLOCKCHAIN_INDEXER_BUFFER_TTL_MS: Joi.number().default(60000),
+  BLOCKCHAIN_INDEXER_BATCH_FLUSH_MS: Joi.number().default(1000),
+  BLOCKCHAIN_INDEXER_BATCH_MAX_SIZE: Joi.number().default(1000),
+  BLOCKCHAIN_INDEXER_RETENTION_DAYS: Joi.number().default(90),
 
   // Rate limiting
   RATE_LIMIT_STRATEGY: Joi.string()
diff --git a/src/modules/blockchain-indexer/blockchain-indexer.controller.ts b/src/modules/blockchain-indexer/blockchain-indexer.controller.ts
index da1938c..f6e2722 100644
--- a/src/modules/blockchain-indexer/blockchain-indexer.controller.ts
+++ b/src/modules/blockchain-indexer/blockchain-indexer.controller.ts
@@ -17,10 +17,13 @@ import {
 } from '@nestjs/swagger';
 import { JwtAuthGuard } from '../auth/guards/jwt-auth.guard';
 import { BlockchainIndexerService } from './blockchain-indexer.service';
+import { LedgerIndexerService } from './services/ledger-indexer.service';
 import { EventQueryService } from './services/event-query.service';
 import { QueryEventsDto } from './dto/query-events.dto';
+import { TemporalQueryDto } from './dto/subscribe-events.dto';
 import { PaginatedResultDto } from '@app/common';
 import { BlockchainEvent } from './entities/blockchain-event.entity';
+import { IndexedEvent } from './entities/indexed-event.entity';
 import { Response } from 'express';
 
 @ApiTags('blockchain-indexer')
@@ -30,9 +33,12 @@ import { Response } from 'express';
 export class BlockchainIndexerController {
   constructor(
     private readonly blockchainIndexerService: BlockchainIndexerService,
+    private readonly ledgerIndexerService: LedgerIndexerService,
     private readonly queryService: EventQueryService,
   ) {}
 
+  // ─── Legacy event queries ───────────────────────────────────────
+
   @Get('events')
   @ApiOperation({ summary: 'Query indexed blockchain events with filtering' })
   @ApiResponse({
@@ -67,6 +73,77 @@ export class BlockchainIndexerController {
     return this.queryService.findByTransactionHash(transactionHash);
   }
 
+  // ─── Real-time indexer endpoints ────────────────────────────────
+
+  @Get('realtime/status')
+  @ApiOperation({
+    summary: 'Get real-time ledger indexer status, metrics, and health',
+  })
+  @ApiResponse({
+    status: HttpStatus.OK,
+    description: 'Real-time indexer status',
+  })
+  async getRealtimeStatus() {
+    return this.ledgerIndexerService.getStatus();
+  }
+
+  @Get('realtime/events')
+  @ApiOperation({
+    summary:
+      'Query indexed events from the real-time pipeline with filtering',
+  })
+  @ApiResponse({
+    status: HttpStatus.OK,
+    description: 'Events retrieved successfully',
+  })
+  async getRealtimeEvents(@Query() queryDto: QueryEventsDto) {
+    const result = await this.queryService.findEvents({
+      ...queryDto,
+      skip: queryDto.skip,
+      startTime: queryDto.startTime ? new Date(queryDto.startTime) : undefined,
+      endTime: queryDto.endTime ? new Date(queryDto.endTime) : undefined,
+      excludeInvalidated: true,
+    });
+    return new PaginatedResultDto<BlockchainEvent>(
+      result.events,
+      result.total,
+      queryDto.page,
+      queryDto.limit,
+    );
+  }
+
+  @Get('realtime/temporal')
+  @ApiOperation({
+    summary: 'Query state at a specific block (temporal query)',
+    description:
+      'Returns events as they were at a given ledger sequence, supporting ' +
+      'historical state reconstruction.',
+  })
+  @ApiResponse({
+    status: HttpStatus.OK,
+    description: 'Temporal query results',
+  })
+  async getTemporalState(@Query() queryDto: TemporalQueryDto) {
+    const result = await this.queryService.findEvents({
+      ledgerFrom: 0,
+      ledgerTo: queryDto.atLedger,
+      eventType: queryDto.eventType,
+      sourceAccount: queryDto.account,
+      skip: queryDto.skip,
+      limit: queryDto.limit,
+      excludeInvalidated: true,
+    });
+    return {
+      asOfLedger: queryDto.atLedger,
+      ...new PaginatedResultDto<IndexedEvent>(
+        result.events as any,
+        result.total,
+        queryDto.page,
+        queryDto.limit,
+      ),
+    };
+  }
+
   @Get('events/stream')
   @ApiOperation({ summary: 'Stream recent blockchain events via SSE' })
   @ApiResponse({
@@ -96,6 +173,8 @@ export class BlockchainIndexerController {
     });
   }
 
+  // ─── Health and operational endpoints ───────────────────────────
+
   @Get('status')
   @ApiOperation({ summary: 'Get blockchain indexer status and metrics' })
   @ApiResponse({
diff --git a/src/modules/blockchain-indexer/blockchain-indexer.module.ts b/src/modules/blockchain-indexer/blockchain-indexer.module.ts
index c5d9b63..57d8f51 100644
--- a/src/modules/blockchain-indexer/blockchain-indexer.module.ts
+++ b/src/modules/blockchain-indexer/blockchain-indexer.module.ts
@@ -7,22 +7,32 @@ import { BlockchainIndexerController } from './blockchain-indexer.controller';
 import { BlockchainIndexerService } from './blockchain-indexer.service';
 import { BlockchainEvent } from './entities/blockchain-event.entity';
 import { IndexerState } from './entities/indexer-state.entity';
+import { IndexedEvent } from './entities/indexed-event.entity';
 import { StellarEventSourceService } from './services/stellar-event-source.service';
 import { IndexingStateService } from './services/indexing-state.service';
 import { ReorgHandlerService } from './services/reorg-handler.service';
 import { EventIndexerService } from './services/event-indexer.service';
 import { EventQueryService } from './services/event-query.service';
 import { EventStreamService } from './services/event-stream.service';
+// New real-time indexing services
+import { HorizonStreamService } from './services/horizon-stream.service';
+import { EventNormalizer } from './services/event-normalizer.service';
+import { EventBufferService } from './services/event-buffer.service';
+import { BatchedPersistenceService } from './services/batched-persistence.service';
+import { SubscriptionManager } from './services/subscription-manager.service';
+import { LedgerIndexerService } from './services/ledger-indexer.service';
+import { EventWebSocketGateway } from './event-websocket.gateway';
 
 @Module({
   imports: [
     ConfigModule,
-    TypeOrmModule.forFeature([BlockchainEvent, IndexerState]),
+    TypeOrmModule.forFeature([BlockchainEvent, IndexerState, IndexedEvent]),
     StellarModule,
     RedisModule,
   ],
   controllers: [BlockchainIndexerController],
   providers: [
+    // Legacy polling-based indexer (kept for backward compatibility)
     BlockchainIndexerService,
     StellarEventSourceService,
     IndexingStateService,
@@ -30,7 +40,19 @@ import { EventStreamService } from './services/event-stream.service';
     EventIndexerService,
     EventQueryService,
     EventStreamService,
+    // Real-time streaming indexer
+    HorizonStreamService,
+    EventNormalizer,
+    EventBufferService,
+    BatchedPersistenceService,
+    SubscriptionManager,
+    LedgerIndexerService,
+    EventWebSocketGateway,
+  ],
+  exports: [
+    BlockchainIndexerService,
+    LedgerIndexerService,
+    SubscriptionManager,
   ],
-  exports: [BlockchainIndexerService],
 })
 export class BlockchainIndexerModule {}
diff --git a/src/modules/blockchain-indexer/dto/subscribe-events.dto.ts b/src/modules/blockchain-indexer/dto/subscribe-events.dto.ts
new file mode 100644
index 0000000..b31278c
--- /dev/null
+++ b/src/modules/blockchain-indexer/dto/subscribe-events.dto.ts
@@ -0,0 +1,100 @@
+import { ApiPropertyOptional } from '@nestjs/swagger';
+import { Type } from 'class-transformer';
+import {
+  IsArray,
+  IsEnum,
+  IsInt,
+  IsOptional,
+  IsString,
+  Max,
+  Min,
+} from 'class-validator';
+import { BlockchainEventType } from '../enums/blockchain-event-type.enum';
+
+/**
+ * DTO for subscribing to real-time events via REST.
+ * The WebSocket gateway accepts a similar shape directly.
+ */
+export class SubscribeEventsDto {
+  @ApiPropertyOptional({
+    description: 'Filter by event types. Empty = all types.',
+    enum: BlockchainEventType,
+    isArray: true,
+  })
+  @IsOptional()
+  @IsArray()
+  @IsEnum(BlockchainEventType, { each: true })
+  eventTypes?: BlockchainEventType[];
+
+  @ApiPropertyOptional({
+    description: 'Filter by Soroban contract IDs.',
+    type: [String],
+  })
+  @IsOptional()
+  @IsArray()
+  @IsString({ each: true })
+  contractIds?: string[];
+
+  @ApiPropertyOptional({
+    description: 'Filter by source or destination accounts.',
+    type: [String],
+  })
+  @IsOptional()
+  @IsArray()
+  @IsString({ each: true })
+  accounts?: string[];
+
+  @ApiPropertyOptional({ description: 'Minimum ledger sequence to include.' })
+  @IsOptional()
+  @Type(() => Number)
+  @IsInt()
+  @Min(0)
+  fromLedger?: number;
+}
+
+/**
+ * DTO for querying state at a specific block (temporal query).
+ */
+export class TemporalQueryDto {
+  @ApiPropertyOptional({
+    description: 'Query events as they were at this ledger sequence.',
+  })
+  @Type(() => Number)
+  @IsInt()
+  @Min(0)
+  atLedger: number;
+
+  @ApiPropertyOptional({ description: 'Filter by event types.' })
+  @IsOptional()
+  @IsEnum(BlockchainEventType)
+  eventType?: BlockchainEventType;
+
+  @ApiPropertyOptional({ description: 'Filter by contract ID.' })
+  @IsOptional()
+  @IsString()
+  contractId?: string;
+
+  @ApiPropertyOptional({ description: 'Filter by account.' })
+  @IsOptional()
+  @IsString()
+  account?: string;
+
+  @ApiPropertyOptional({ default: 1, minimum: 1 })
+  @IsOptional()
+  @Type(() => Number)
+  @IsInt()
+  @Min(1)
+  page = 1;
+
+  @ApiPropertyOptional({ default: 20, minimum: 1, maximum: 100 })
+  @IsOptional()
+  @Type(() => Number)
+  @IsInt()
+  @Min(1)
+  @Max(100)
+  limit = 20;
+
+  get skip(): number {
+    return (this.page - 1) * this.limit;
+  }
+}
diff --git a/src/modules/blockchain-indexer/entities/blockchain-event.entity.ts b/src/modules/blockchain-indexer/entities/blockchain-event.entity.ts
index 64c91bb..3d8a2ef 100644
--- a/src/modules/blockchain-indexer/entities/blockchain-event.entity.ts
+++ b/src/modules/blockchain-indexer/entities/blockchain-event.entity.ts
@@ -10,6 +10,11 @@ export enum BlockchainEventType {
   CREATE_ACCOUNT = 'create_account',
   ACCOUNT_MERGE = 'account_merge',
   TRANSACTION = 'transaction',
+  // Soroban contract event types
+  SOROBAN_CONTRACT_INVOCATION = 'soroban_contract_invocation',
+  SOROBAN_CONTRACT_EVENT = 'soroban_contract_event',
+  SOROBAN_SYSTEM_EVENT = 'soroban_system_event',
+  SOROBAN_DIAGNOSTIC_EVENT = 'soroban_diagnostic_event',
 }
 
 @Entity('blockchain_events')
diff --git a/src/modules/blockchain-indexer/entities/indexed-event.entity.ts b/src/modules/blockchain-indexer/entities/indexed-event.entity.ts
new file mode 100644
index 0000000..24f6266
--- /dev/null
+++ b/src/modules/blockchain-indexer/entities/indexed-event.entity.ts
@@ -0,0 +1,86 @@
+import { Column, Entity, Index } from 'typeorm';
+import { BaseEntity } from '@app/common';
+
+/**
+ * Unified event entity for the real-time ledger indexer. Stores both Stellar
+ * native operations and Soroban contract events in a single table, enabling
+ * cross-cutting temporal queries (state at ledger X) and efficient filtering.
+ */
+@Entity('indexed_events')
+@Index(['sequenceNumber'], { unique: true })
+@Index(['ledgerSequence'])
+@Index(['eventType'])
+@Index(['contractId'])
+@Index(['sourceAccount'])
+@Index(['timestamp'])
+@Index(['ledgerSequence', 'eventType'])
+@Index(['contractId', 'eventType'])
+export class IndexedEvent extends BaseEntity {
+  /** Monotonically increasing sequence number for ordering and deduplication. */
+  @Column({ type: 'bigint' })
+  sequenceNumber: string;
+
+  /** Stellar ledger sequence where this event was produced. */
+  @Column({ type: 'int' })
+  ledgerSequence: number;
+
+  /** Normalized event type (see BlockchainEventType enum). */
+  @Column({ type: 'varchar' })
+  eventType: string;
+
+  /** Transaction hash that produced this event. */
+  @Column({ type: 'varchar' })
+  transactionHash: string;
+
+  /** Block timestamp of the ledger. */
+  @Column({ type: 'timestamptz' })
+  timestamp: Date;
+
+  /** Account or contract that initiated the event. */
+  @Column({ type: 'varchar' })
+  sourceAccount: string;
+
+  /** Destination account, if applicable (payments, transfers). */
+  @Column({ type: 'varchar', nullable: true })
+  destinationAccount?: string;
+
+  /** Soroban contract ID, if this is a contract event. */
+  @Column({ type: 'varchar', nullable: true })
+  contractId?: string;
+
+  /** Soroban contract method name, if this is a contract invocation. */
+  @Column({ type: 'varchar', nullable: true })
+  methodName?: string;
+
+  /** Asset code for native Stellar events. */
+  @Column({ type: 'varchar', nullable: true })
+  assetCode?: string;
+
+  /** Asset issuer for non-native assets. */
+  @Column({ type: 'varchar', nullable: true })
+  assetIssuer?: string;
+
+  /** Event amount, stored as string for precision. */
+  @Column({ type: 'varchar', nullable: true })
+  amount?: string;
+
+  /** Event topics (Soroban contract events). */
+  @Column({ type: 'jsonb', nullable: true })
+  topics?: unknown[];
+
+  /** Decoded event body (Soroban contract events). */
+  @Column({ type: 'jsonb', nullable: true })
+  value?: unknown;
+
+  /** Normalized internal event data for quick access. */
+  @Column({ type: 'jsonb', nullable: true })
+  normalizedData?: Record<string, unknown>;
+
+  /** Raw event data for debugging and replay. */
+  @Column({ type: 'jsonb', nullable: true })
+  raw?: Record<string, unknown>;
+
+  /** Whether this event has been invalidated by a chain reorganization. */
+  @Column({ type: 'boolean', default: false })
+  invalidated: boolean;
+}
diff --git a/src/modules/blockchain-indexer/enums/blockchain-event-type.enum.ts b/src/modules/blockchain-indexer/enums/blockchain-event-type.enum.ts
index 1a6bbc5..cf4a493 100644
--- a/src/modules/blockchain-indexer/enums/blockchain-event-type.enum.ts
+++ b/src/modules/blockchain-indexer/enums/blockchain-event-type.enum.ts
@@ -7,4 +7,9 @@ export enum BlockchainEventType {
   CREATE_ACCOUNT = 'create_account',
   ACCOUNT_MERGE = 'account_merge',
   TRANSACTION = 'transaction',
+  // Soroban contract event types
+  SOROBAN_CONTRACT_INVOCATION = 'soroban_contract_invocation',
+  SOROBAN_CONTRACT_EVENT = 'soroban_contract_event',
+  SOROBAN_SYSTEM_EVENT = 'soroban_system_event',
+  SOROBAN_DIAGNOSTIC_EVENT = 'soroban_diagnostic_event',
 }
diff --git a/src/modules/blockchain-indexer/event-buffer.service.spec.ts b/src/modules/blockchain-indexer/event-buffer.service.spec.ts
new file mode 100644
index 0000000..83a3970
--- /dev/null
+++ b/src/modules/blockchain-indexer/event-buffer.service.spec.ts
@@ -0,0 +1,162 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { ConfigModule } from '@nestjs/config';
+import { EventBufferService, BufferedEvent } from './services/event-buffer.service';
+
+describe('EventBufferService', () => {
+  let module: TestingModule;
+  let buffer: EventBufferService;
+
+  const makeEvent = (seq: number, ledger: number = seq * 10): BufferedEvent => ({
+    sequenceNumber: seq,
+    ledgerSequence: ledger,
+    eventType: 'payment',
+    timestamp: Date.now(),
+    data: { hash: `tx_${seq}` },
+    bufferedAt: Date.now(),
+  });
+
+  beforeAll(async () => {
+    module = await Test.createTestingModule({
+      imports: [ConfigModule.forRoot({ isGlobal: true })],
+      providers: [EventBufferService],
+    }).compile();
+
+    buffer = module.get(EventBufferService);
+  });
+
+  afterAll(async () => {
+    await module.close();
+  });
+
+  beforeEach(() => {
+    buffer.reset();
+  });
+
+  describe('push', () => {
+    it('should accept a new event', () => {
+      const accepted = buffer.push(makeEvent(1));
+      expect(accepted).toBe(true);
+      expect(buffer.getStats().bufferSize).toBe(1);
+    });
+
+    it('should reject duplicate sequence numbers', () => {
+      buffer.push(makeEvent(1));
+      const accepted = buffer.push(makeEvent(1));
+      expect(accepted).toBe(false);
+      expect(buffer.getStats().bufferSize).toBe(1);
+      expect(buffer.getStats().totalDuplicates).toBe(1);
+    });
+
+    it('should track receive stats', () => {
+      buffer.push(makeEvent(1));
+      buffer.push(makeEvent(2));
+      const stats = buffer.getStats();
+      expect(stats.totalReceived).toBe(2);
+    });
+  });
+
+  describe('pushBatch', () => {
+    it('should accept all unique events', () => {
+      const events = [makeEvent(1), makeEvent(2), makeEvent(3)];
+      const accepted = buffer.pushBatch(events);
+      expect(accepted).toBe(3);
+      expect(buffer.getStats().bufferSize).toBe(3);
+    });
+
+    it('should skip duplicates in a batch', () => {
+      buffer.push(makeEvent(1));
+      const events = [makeEvent(1), makeEvent(2)];
+      const accepted = buffer.pushBatch(events);
+      expect(accepted).toBe(1); // Only event 2 is new
+      expect(buffer.getStats().bufferSize).toBe(2);
+    });
+  });
+
+  describe('drain', () => {
+    it('should return events in sequence order', () => {
+      buffer.push(makeEvent(3));
+      buffer.push(makeEvent(1));
+      buffer.push(makeEvent(2));
+
+      const drained = buffer.drain(10);
+
+      expect(drained).toHaveLength(3);
+      expect(drained[0].sequenceNumber).toBe(1);
+      expect(drained[1].sequenceNumber).toBe(2);
+      expect(drained[2].sequenceNumber).toBe(3);
+    });
+
+    it('should remove drained events from buffer', () => {
+      buffer.push(makeEvent(1));
+      buffer.push(makeEvent(2));
+
+      buffer.drain(1);
+
+      expect(buffer.getStats().bufferSize).toBe(1);
+    });
+
+    it('should respect maxCount limit', () => {
+      buffer.push(makeEvent(1));
+      buffer.push(makeEvent(2));
+      buffer.push(makeEvent(3));
+
+      const drained = buffer.drain(2);
+
+      expect(drained).toHaveLength(2);
+      expect(buffer.getStats().bufferSize).toBe(1);
+    });
+
+    it('should return empty array when buffer is empty', () => {
+      const drained = buffer.drain(10);
+      expect(drained).toHaveLength(0);
+    });
+  });
+
+  describe('peek', () => {
+    it('should return events without removing them', () => {
+      buffer.push(makeEvent(1));
+      buffer.push(makeEvent(2));
+
+      const peeked = buffer.peek(10);
+
+      expect(peeked).toHaveLength(2);
+      expect(buffer.getStats().bufferSize).toBe(2);
+    });
+  });
+
+  describe('evictExpired', () => {
+    it('should evict events older than TTL', () => {
+      buffer.push(makeEvent(1));
+
+      // Manually backdate the bufferedAt to exceed the TTL.
+      const events = buffer.peek(1);
+      (events[0] as any).bufferedAt = Date.now() - 200000;
+
+      buffer.evictExpired();
+
+      expect(buffer.getStats().bufferSize).toBe(0);
+      expect(buffer.getStats().totalExpired).toBe(1);
+    });
+
+    it('should keep recent events', () => {
+      buffer.push(makeEvent(1));
+      buffer.evictExpired();
+      expect(buffer.getStats().bufferSize).toBe(1);
+    });
+  });
+
+  describe('ordering guarantees', () => {
+    it('should maintain monotonic ordering after out-of-order pushes', () => {
+      // Push events out of order.
+      for (const seq of [5, 1, 9, 3, 7, 2, 8, 4, 6]) {
+        buffer.push(makeEvent(seq));
+      }
+
+      const drained = buffer.drain(100);
+
+      expect(drained.map((e) => e.sequenceNumber)).toEqual([
+        1, 2, 3, 4, 5, 6, 7, 8, 9,
+      ]);
+    });
+  });
+});
diff --git a/src/modules/blockchain-indexer/event-normalizer.service.spec.ts b/src/modules/blockchain-indexer/event-normalizer.service.spec.ts
new file mode 100644
index 0000000..f20e70d
--- /dev/null
+++ b/src/modules/blockchain-indexer/event-normalizer.service.spec.ts
@@ -0,0 +1,206 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { ConfigModule } from '@nestjs/config';
+import { EventNormalizer } from './services/event-normalizer.service';
+import { BlockchainEventType } from './entities/blockchain-event.entity';
+import { ParsedContractEvent } from '../stellar/soroban/soroban.types';
+
+describe('EventNormalizer', () => {
+  let module: TestingModule;
+  let normalizer: EventNormalizer;
+
+  beforeAll(async () => {
+    module = await Test.createTestingModule({
+      imports: [ConfigModule.forRoot({ isGlobal: true })],
+      providers: [EventNormalizer],
+    }).compile();
+
+    normalizer = module.get(EventNormalizer);
+  });
+
+  afterAll(async () => {
+    await module.close();
+  });
+
+  beforeEach(() => {
+    normalizer.initializeSequenceCounter(BigInt(0));
+  });
+
+  describe('normalizeStellarOperations', () => {
+    const mockTx = {
+      hash: 'tx123',
+      ledger: 50000,
+      created_at: '2024-01-15T10:30:00Z',
+      source_account: 'GABC...',
+      fee_charged: '100',
+      successful: true,
+      paging_token: '50000_0',
+    };
+
+    it('should normalize payment operations', () => {
+      const operations = [
+        {
+          transaction_hash: 'tx123',
+          application_index: 0,
+          type: 'payment',
+          asset_code: 'USD',
+          asset_issuer: 'GDEF...',
+          from: 'GABC...',
+          to: 'GXYZ...',
+          amount: '100.50',
+        },
+      ];
+
+      const events = normalizer.normalizeStellarOperations(mockTx as any, operations);
+
+      expect(events).toHaveLength(1);
+      expect(events[0].eventType).toBe(BlockchainEventType.PAYMENT);
+      expect(events[0].transactionHash).toBe('tx123');
+      expect(events[0].ledgerSequence).toBe(50000);
+      expect(events[0].sourceAccount).toBe('GABC...');
+      expect(events[0].destinationAccount).toBe('GXYZ...');
+      expect(events[0].assetCode).toBe('USD');
+      expect(events[0].amount).toBe('100.50');
+    });
+
+    it('should normalize manage offer operations', () => {
+      const operations = [
+        {
+          transaction_hash: 'tx123',
+          application_index: 1,
+          type: 'manage_offer',
+          asset_code: 'BTC',
+          asset_issuer: 'GDEF...',
+          from: 'GABC...',
+          amount: '0.5',
+        },
+      ];
+
+      const events = normalizer.normalizeStellarOperations(mockTx as any, operations);
+
+      expect(events).toHaveLength(1);
+      expect(events[0].eventType).toBe(BlockchainEventType.MANAGE_OFFER);
+    });
+
+    it('should skip irrelevant operation types', () => {
+      const operations = [
+        {
+          transaction_hash: 'tx123',
+          application_index: 0,
+          type: 'bump_sequence',
+        },
+      ];
+
+      const events = normalizer.normalizeStellarOperations(mockTx as any, operations);
+
+      expect(events).toHaveLength(0);
+    });
+
+    it('should handle create_account operations', () => {
+      const operations = [
+        {
+          transaction_hash: 'tx123',
+          application_index: 0,
+          type: 'create_account',
+          to: 'GNEW...',
+          from: 'GABC...',
+          starting_balance: '10',
+        },
+      ];
+
+      const events = normalizer.normalizeStellarOperations(mockTx as any, operations);
+
+      expect(events).toHaveLength(1);
+      expect(events[0].eventType).toBe(BlockchainEventType.CREATE_ACCOUNT);
+      expect(events[0].destinationAccount).toBe('GNEW...');
+      expect(events[0].amount).toBe('10');
+    });
+  });
+
+  describe('normalizeSorobanEvent', () => {
+    it('should normalize a contract event', () => {
+      const sorobanEvent: ParsedContractEvent = {
+        id: 'soroban_event_1',
+        contractId: 'CCONTRACT...',
+        type: 'contract',
+        ledger: 60000,
+        ledgerClosedAt: '2024-01-15T11:00:00Z',
+        topics: ['transfer', 'GABC...', 'GXYZ...'],
+        value: { amount: 1000, symbol: 'TOKEN' },
+        txHash: 'tx_soroban_1',
+        pagingToken: 'soroban_1',
+        indexedAt: Date.now(),
+      };
+
+      const event = normalizer.normalizeSorobanEvent(sorobanEvent);
+
+      expect(event.eventType).toBe(
+        BlockchainEventType.SOROBAN_CONTRACT_EVENT,
+      );
+      expect(event.contractId).toBe('CCONTRACT...');
+      expect(event.methodName).toBe('transfer');
+      expect(event.topics).toEqual(['transfer', 'GABC...', 'GXYZ...']);
+      expect(event.value).toEqual({ amount: 1000, symbol: 'TOKEN' });
+      expect(event.transactionHash).toBe('tx_soroban_1');
+    });
+
+    it('should normalize a system event', () => {
+      const sorobanEvent: ParsedContractEvent = {
+        id: 'soroban_sys_1',
+        contractId: '',
+        type: 'system',
+        ledger: 60001,
+        ledgerClosedAt: '2024-01-15T11:00:01Z',
+        topics: ['contract crédito'],
+        value: {},
+        txHash: 'tx_sys_1',
+        pagingToken: 'sys_1',
+        indexedAt: Date.now(),
+      };
+
+      const event = normalizer.normalizeSorobanEvent(sorobanEvent);
+
+      expect(event.eventType).toBe(
+        BlockchainEventType.SOROBAN_SYSTEM_EVENT,
+      );
+    });
+  });
+
+  describe('sequence counter', () => {
+    it('should initialize from a given value', () => {
+      normalizer.initializeSequenceCounter(BigInt(100));
+      expect(normalizer.getCurrentSequence()).toBe(BigInt(100));
+    });
+
+    it('should increment sequence numbers monotonically', () => {
+      normalizer.initializeSequenceCounter(BigInt(0));
+
+      const seq1 = normalizer.getNextSequence();
+      const seq2 = normalizer.getNextSequence();
+      const seq3 = normalizer.getNextSequence();
+
+      expect(seq1).toBe(BigInt(1));
+      expect(seq2).toBe(BigInt(2));
+      expect(seq3).toBe(BigInt(3));
+    });
+
+    it('should assign sequences to events via assignSequence', () => {
+      normalizer.initializeSequenceCounter(BigInt(0));
+
+      const input = {
+        ledgerSequence: 100,
+        eventType: BlockchainEventType.PAYMENT,
+        transactionHash: 'tx1',
+        timestamp: new Date(),
+        sourceAccount: 'GABC...',
+        raw: {},
+        invalidated: false,
+      };
+
+      const result = normalizer.assignSequence(input);
+      expect(result.sequenceNumber).toBe('1');
+
+      const result2 = normalizer.assignSequence(input);
+      expect(result2.sequenceNumber).toBe('2');
+    });
+  });
+});
diff --git a/src/modules/blockchain-indexer/event-websocket.gateway.ts b/src/modules/blockchain-indexer/event-websocket.gateway.ts
new file mode 100644
index 0000000..188a963
--- /dev/null
+++ b/src/modules/blockchain-indexer/event-websocket.gateway.ts
@@ -0,0 +1,155 @@
+import {
+  WebSocketGateway,
+  WebSocketServer,
+  OnGatewayConnection,
+  OnGatewayDisconnect,
+  ConnectedSocket,
+  MessageBody,
+  SubscribeMessage,
+} from '@nestjs/websockets';
+import { Logger } from '@nestjs/common';
+import { Server, Socket } from 'socket.io';
+import { SubscriptionManager } from './services/subscription-manager.service';
+
+/**
+ * WebSocket gateway for real-time event subscriptions. Frontend clients
+ * connect to `/blockchain-indexer` and subscribe to specific event types
+ * (trade executions, balance updates, liquidations, contract events).
+ *
+ * Protocol:
+ * - Client emits `subscribe` with filter options
+ * - Server pushes `event` messages for matching events
+ * - Client emits `unsubscribe` to stop receiving events
+ * - Server pushes `health` for connection status updates
+ */
+@WebSocketGateway({
+  cors: { origin: '*' },
+  namespace: '/blockchain-indexer',
+})
+export class EventWebSocketGateway
+  implements OnGatewayConnection, OnGatewayDisconnect
+{
+  @WebSocketServer()
+  server: Server;
+
+  private readonly logger = new Logger(EventWebSocketGateway.name);
+
+  constructor(private readonly subscriptionManager: SubscriptionManager) {}
+
+  handleConnection(client: Socket): void {
+    const clientId = client.id;
+    this.logger.log(`Client connected: ${clientId}`);
+
+    // Listen for events destined for this client and forward them.
+    const eventHandler = (payload: {
+      subscriptionId: string;
+      event: unknown;
+    }) => {
+      client.emit('event', {
+        subscriptionId: payload.subscriptionId,
+        data: payload.event,
+      });
+    };
+
+    this.subscriptionManager.on(`event:${clientId}`, eventHandler);
+
+    // Clean up the listener when the client disconnects.
+    client.on('disconnect', () => {
+      this.subscriptionManager.removeListener(`event:${clientId}`, eventHandler);
+    });
+
+    // Notify the client of successful connection.
+    client.emit('connected', {
+      clientId,
+      message: 'Connected to blockchain event stream',
+      timestamp: new Date().toISOString(),
+    });
+  }
+
+  handleDisconnect(client: Socket): void {
+    const removed = this.subscriptionManager.removeClientSubscriptions(
+      client.id,
+    );
+    this.logger.log(
+      `Client disconnected: ${client.id} (${removed} subscriptions removed)`,
+    );
+  }
+
+  /**
+   * Subscribe to events with optional filters.
+   *
+   * @example
+   * socket.emit('subscribe', {
+   *   eventTypes: ['trade', 'liquidation'],
+   *   contractIds: ['C...'],
+   *   accounts: ['G...'],
+   *   fromLedger: 100000,
+   * })
+   */
+  @SubscribeMessage('subscribe')
+  handleSubscribe(
+    @ConnectedSocket() client: Socket,
+    @MessageBody()
+    data: {
+      eventTypes?: string[];
+      contractIds?: string[];
+      accounts?: string[];
+      fromLedger?: number;
+    },
+  ): void {
+    const subscriptionId = this.subscriptionManager.subscribe({
+      clientId: client.id,
+      eventTypes: data.eventTypes ?? [],
+      contractIds: data.contractIds,
+      accounts: data.accounts,
+      fromLedger: data.fromLedger,
+    });
+
+    client.emit('subscribed', {
+      subscriptionId,
+      filters: data,
+      message: 'Successfully subscribed to event stream',
+    });
+
+    this.logger.debug(
+      `Client ${client.id} subscribed as ${subscriptionId}`,
+    );
+  }
+
+  /**
+   * Unsubscribe from a specific subscription.
+   *
+   * @example
+   * socket.emit('unsubscribe', { subscriptionId: 'sub_1' })
+   */
+  @SubscribeMessage('unsubscribe')
+  handleUnsubscribe(
+    @ConnectedSocket() client: Socket,
+    @MessageBody() data: { subscriptionId: string },
+  ): void {
+    const removed = this.subscriptionManager.unsubscribe(data.subscriptionId);
+    client.emit('unsubscribed', {
+      subscriptionId: data.subscriptionId,
+      success: removed,
+    });
+  }
+
+  /**
+   * Returns the current connection status.
+   */
+  @SubscribeMessage('status')
+  handleStatus(@ConnectedSocket() client: Socket): void {
+    const subs = this.subscriptionManager.listSubscriptions(client.id);
+    client.emit('status', {
+      subscriptionCount: subs.length,
+      subscriptions: subs,
+    });
+  }
+
+  /**
+   * Broadcasts a message to all connected clients (admin use).
+   */
+  broadcastToAll(eventType: string, data: unknown): void {
+    this.server.emit(eventType, data);
+  }
+}
diff --git a/src/modules/blockchain-indexer/services/batched-persistence.service.ts b/src/modules/blockchain-indexer/services/batched-persistence.service.ts
new file mode 100644
index 0000000..1a66ad1
--- /dev/null
+++ b/src/modules/blockchain-indexer/services/batched-persistence.service.ts
@@ -0,0 +1,211 @@
+import {
+  Injectable,
+  Logger,
+  OnModuleDestroy,
+} from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import { InjectRepository } from '@nestjs/typeorm';
+import { Repository, DataSource } from 'typeorm';
+import { IndexedEvent } from '../entities/indexed-event.entity';
+import { IndexingStateService } from './indexing-state.service';
+
+/**
+ * Type used internally before an event is assigned a database-generated ID.
+ */
+export type IndexedEventInput = Omit<IndexedEvent, 'id' | 'createdAt' | 'updatedAt'>;
+
+/**
+ * High-throughput batched persistence layer for IndexedEvent records.
+ *
+ * Events are collected in an internal buffer and flushed to PostgreSQL in
+ * configurable batch sizes. This achieves >10k events/second throughput by:
+ * - Using bulk INSERT with ON CONFLICT DO NOTHING for idempotency
+ * - Flushing on a timer or when the batch reaches capacity
+ * - Tracking the highest persisted sequence for restart resumption
+ *
+ * A configurable retention policy deletes events older than `retentionDays`.
+ */
+@Injectable()
+export class BatchedPersistenceService implements OnModuleDestroy {
+  private readonly logger = new Logger(BatchedPersistenceService.name);
+
+  private readonly flushIntervalMs: number;
+  private readonly batchMaxSize: number;
+  private readonly retentionDays: number;
+
+  private pendingBatch: IndexedEventInput[] = [];
+  private flushTimer: NodeJS.Timeout | null = null;
+  private flushing = false;
+  private highestPersistedSequence = BigInt(0);
+
+  /** Rolling throughput counter for observability. */
+  private totalPersisted = 0;
+
+  constructor(
+    @InjectRepository(IndexedEvent)
+    private readonly eventRepo: Repository<IndexedEvent>,
+    private readonly stateService: IndexingStateService,
+    private readonly dataSource: DataSource,
+    private readonly configService: ConfigService,
+  ) {
+    this.flushIntervalMs =
+      this.configService.get<number>('blockchainIndexer.batchFlushIntervalMs') ?? 1000;
+    this.batchMaxSize =
+      this.configService.get<number>('blockchainIndexer.batchMaxSize') ?? 1000;
+    this.retentionDays =
+      this.configService.get<number>('blockchainIndexer.retentionDays') ?? 90;
+  }
+
+  onModuleDestroy(): void {
+    this.stopFlushTimer();
+    // Attempt a final flush of any remaining events.
+    if (this.pendingBatch.length > 0) {
+      this.logger.log(
+        `Flushing ${this.pendingBatch.length} remaining events on shutdown`,
+      );
+      void this.flush();
+    }
+  }
+
+  /**
+   * Starts the periodic flush timer. Call once during startup after
+   * restoring the sequence counter.
+   */
+  startFlushTimer(): void {
+    if (this.flushTimer) return;
+    this.flushTimer = setInterval(() => {
+      void this.flush();
+    }, this.flushIntervalMs);
+    this.flushTimer.unref?.();
+  }
+
+  private stopFlushTimer(): void {
+    if (this.flushTimer) {
+      clearInterval(this.flushTimer);
+      this.flushTimer = null;
+    }
+  }
+
+  /**
+   * Enqueues a single event for batched persistence. The event must already
+   * have a sequenceNumber assigned by the EventNormalizer.
+   */
+  enqueue(event: IndexedEventInput): void {
+    this.pendingBatch.push(event);
+
+    const seqNum = BigInt(event.sequenceNumber);
+    if (seqNum > this.highestPersistedSequence) {
+      this.highestPersistedSequence = seqNum;
+    }
+
+    // Auto-flush when the batch reaches capacity.
+    if (this.pendingBatch.length >= this.batchMaxSize) {
+      void this.flush();
+    }
+  }
+
+  /**
+   * Enqueues multiple events at once.
+   */
+  enqueueBatch(events: IndexedEventInput[]): void {
+    for (const event of events) {
+      this.enqueue(event);
+    }
+  }
+
+  /**
+   * Forces an immediate flush of the pending batch to PostgreSQL.
+   * Returns the number of events persisted.
+   */
+  async flush(): Promise<number> {
+    if (this.flushing || this.pendingBatch.length === 0) return 0;
+    this.flushing = true;
+
+    const batch = this.pendingBatch;
+    this.pendingBatch = [];
+
+    try {
+      await this.dataSource.transaction(async (manager) => {
+        // Bulk insert with ON CONFLICT DO NOTHING for idempotency.
+        if (batch.length > 0) {
+          await manager
+            .createQueryBuilder()
+            .insert()
+            .into(IndexedEvent)
+            .values(batch as any)
+            .orIgnore()
+            .execute();
+        }
+      });
+
+      this.totalPersisted += batch.length;
+
+      // Persist the high-water mark for restart recovery.
+      if (batch.length > 0) {
+        await this.stateService.setSequenceCounter(this.highestPersistedSequence);
+      }
+
+      this.logger.debug(
+        `Flushed ${batch.length} events to PostgreSQL (total: ${this.totalPersisted})`,
+      );
+      return batch.length;
+    } catch (error) {
+      this.logger.error(
+        `Failed to flush ${batch.length} events: ${(error as Error).message}`,
+      );
+      // Re-queue the failed batch for retry.
+      this.pendingBatch = [...batch, ...this.pendingBatch];
+      return 0;
+    } finally {
+      this.flushing = false;
+    }
+  }
+
+  /**
+   * Runs the retention policy: deletes events older than the configured
+   * retention period. Should be called periodically (e.g. once per hour).
+   */
+  async purgeExpired(): Promise<number> {
+    const cutoff = new Date();
+    cutoff.setDate(cutoff.getDate() - this.retentionDays);
+
+    try {
+      const result = await this.eventRepo
+        .createQueryBuilder()
+        .delete()
+        .where('timestamp < :cutoff', { cutoff })
+        .execute();
+
+      const deleted = result.affected ?? 0;
+      if (deleted > 0) {
+        this.logger.log(
+          `Purged ${deleted} events older than ${this.retentionDays} days`,
+        );
+      }
+      return deleted;
+    } catch (error) {
+      this.logger.error(
+        `Failed to purge expired events: ${(error as Error).message}`,
+      );
+      return 0;
+    }
+  }
+
+  /**
+   * Returns current persistence metrics.
+   */
+  getStats() {
+    return {
+      pendingCount: this.pendingBatch.length,
+      batchMaxSize: this.batchMaxSize,
+      flushIntervalMs: this.flushIntervalMs,
+      totalPersisted: this.totalPersisted,
+      highestPersistedSequence: Number(this.highestPersistedSequence),
+      retentionDays: this.retentionDays,
+    };
+  }
+
+  getHighestPersistedSequence(): bigint {
+    return this.highestPersistedSequence;
+  }
+}
diff --git a/src/modules/blockchain-indexer/services/event-buffer.service.ts b/src/modules/blockchain-indexer/services/event-buffer.service.ts
new file mode 100644
index 0000000..93db823
--- /dev/null
+++ b/src/modules/blockchain-indexer/services/event-buffer.service.ts
@@ -0,0 +1,210 @@
+import { Injectable, Logger, OnModuleDestroy } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+
+/**
+ * A buffered event with its sequence number for ordering.
+ */
+export interface BufferedEvent {
+  sequenceNumber: number;
+  ledgerSequence: number;
+  eventType: string;
+  timestamp: number;
+  data: Record<string, unknown>;
+  /** Epoch millis when the event entered the buffer. */
+  bufferedAt: number;
+}
+
+/**
+ * In-memory event buffer that provides:
+ * - Deduplication by sequence number
+ * - Ordering guarantees via sequence numbers
+ * - TTL-based expiry for stale events
+ * - FIFO batch extraction for persistence
+ *
+ * This sits between the event normalizer and the persistence layer, absorbing
+ * bursts of events while maintaining ordering and preventing duplicates.
+ */
+@Injectable()
+export class EventBufferService implements OnModuleDestroy {
+  private readonly logger = new Logger(EventBufferService.name);
+  private readonly maxBufferSize: number;
+  private readonly eventTtlMs: number;
+
+  /** Events keyed by sequence number for O(1) deduplication lookups. */
+  private readonly buffer = new Map<number, BufferedEvent>();
+
+  /** Sorted sequence numbers for ordered extraction. */
+  private readonly sequenceOrder: number[] = [];
+
+  /** Timer for periodic TTL cleanup. */
+  private cleanupTimer: NodeJS.Timeout | null = null;
+
+  /** Metrics counters. */
+  private stats = {
+    totalReceived: 0,
+    totalDuplicates: 0,
+    totalExpired: 0,
+  };
+
+  constructor(private readonly configService: ConfigService) {
+    this.maxBufferSize =
+      this.configService.get<number>('blockchainIndexer.eventBufferSize') ?? 10000;
+    this.eventTtlMs =
+      this.configService.get<number>('blockchainIndexer.eventBufferTtlMs') ?? 60000;
+
+    // Start periodic cleanup to evict expired events.
+    this.cleanupTimer = setInterval(
+      () => this.evictExpired(),
+      Math.min(this.eventTtlMs / 2, 10000),
+    );
+    this.cleanupTimer.unref?.();
+  }
+
+  onModuleDestroy(): void {
+    if (this.cleanupTimer) {
+      clearInterval(this.cleanupTimer);
+    }
+  }
+
+  /**
+   * Pushes a single event into the buffer. Returns true if the event was
+   * accepted, false if it was a duplicate or the buffer is full.
+   */
+  push(event: BufferedEvent): boolean {
+    this.stats.totalReceived++;
+
+    // Deduplication: reject if we already have this sequence number.
+    if (this.buffer.has(event.sequenceNumber)) {
+      this.stats.totalDuplicates++;
+      return false;
+    }
+
+    // If the buffer is at capacity, evict the oldest event.
+    if (this.buffer.size >= this.maxBufferSize) {
+      this.evictOldest();
+    }
+
+    if (!event.bufferedAt) {
+      event.bufferedAt = Date.now();
+    }
+    this.buffer.set(event.sequenceNumber, event);
+    this.sequenceOrder.push(event.sequenceNumber);
+
+    // Keep the sequence order sorted for efficient ordered extraction.
+    // For small arrays this is fine; for large arrays we'd use a heap.
+    if (this.sequenceOrder.length > 1) {
+      const last = this.sequenceOrder[this.sequenceOrder.length - 1];
+      const prev = this.sequenceOrder[this.sequenceOrder.length - 2];
+      if (prev !== undefined && last !== undefined && last < prev) {
+        this.sequenceOrder.sort((a, b) => a - b);
+      }
+    }
+
+    return true;
+  }
+
+  /**
+   * Pushes multiple events at once. Returns the count of events accepted.
+   */
+  pushBatch(events: BufferedEvent[]): number {
+    let accepted = 0;
+    for (const event of events) {
+      if (this.push(event)) {
+        accepted++;
+      }
+    }
+    return accepted;
+  }
+
+  /**
+   * Drains up to `maxCount` events from the buffer in sequence-number order.
+   * Returns them sorted oldest-first and removes them from the buffer.
+   */
+  drain(maxCount: number): BufferedEvent[] {
+    const count = Math.min(maxCount, this.sequenceOrder.length);
+    const sequences = this.sequenceOrder.splice(0, count);
+    const events: BufferedEvent[] = [];
+
+    for (const seq of sequences) {
+      const event = this.buffer.get(seq);
+      if (event) {
+        events.push(event);
+        this.buffer.delete(seq);
+      }
+    }
+
+    return events;
+  }
+
+  /**
+   * Peek at the next events without removing them.
+   */
+  peek(count: number): BufferedEvent[] {
+    const sequences = this.sequenceOrder.slice(0, count);
+    return sequences
+      .map((seq) => this.buffer.get(seq))
+      .filter((e): e is BufferedEvent => e !== undefined);
+  }
+
+  /**
+   * Removes events older than the TTL.
+   */
+  evictExpired(): void {
+    const now = Date.now();
+    let evicted = 0;
+
+    for (const [seq, event] of this.buffer) {
+      if (now - event.bufferedAt > this.eventTtlMs) {
+        this.buffer.delete(seq);
+        evicted++;
+      }
+    }
+
+    // Rebuild sequence order from remaining entries.
+    if (evicted > 0) {
+      this.sequenceOrder.length = 0;
+      for (const seq of this.buffer.keys()) {
+        this.sequenceOrder.push(seq);
+      }
+      this.sequenceOrder.sort((a, b) => a - b);
+      this.stats.totalExpired += evicted;
+      this.logger.debug(`Evicted ${evicted} expired events from buffer`);
+    }
+  }
+
+  /**
+   * Resets all internal state and counters. Useful for testing.
+   */
+  reset(): void {
+    this.buffer.clear();
+    this.sequenceOrder.length = 0;
+    this.stats = { totalReceived: 0, totalDuplicates: 0, totalExpired: 0 };
+  }
+
+  /**
+   * Returns current buffer statistics.
+   */
+  getStats() {
+    return {
+      bufferSize: this.buffer.size,
+      maxBufferSize: this.maxBufferSize,
+      totalReceived: this.stats.totalReceived,
+      totalDuplicates: this.stats.totalDuplicates,
+      totalExpired: this.stats.totalExpired,
+      oldestSequence:
+        this.sequenceOrder.length > 0 ? this.sequenceOrder[0] : null,
+      newestSequence:
+        this.sequenceOrder.length > 0
+          ? this.sequenceOrder[this.sequenceOrder.length - 1]
+          : null,
+    };
+  }
+
+  private evictOldest(): void {
+    if (this.sequenceOrder.length === 0) return;
+    const oldest = this.sequenceOrder.shift();
+    if (oldest !== undefined) {
+      this.buffer.delete(oldest);
+    }
+  }
+}
diff --git a/src/modules/blockchain-indexer/services/event-normalizer.service.ts b/src/modules/blockchain-indexer/services/event-normalizer.service.ts
new file mode 100644
index 0000000..aed9097
--- /dev/null
+++ b/src/modules/blockchain-indexer/services/event-normalizer.service.ts
@@ -0,0 +1,181 @@
+import { Injectable, Logger } from '@nestjs/common';
+import {
+  BlockchainEventType,
+} from '../entities/blockchain-event.entity';
+import { IndexedEvent } from '../entities/indexed-event.entity';
+import {
+  RawTransaction,
+  RawOperation,
+} from './stellar-event-source.service';
+import { ParsedContractEvent } from '../../stellar/soroban/soroban.types';
+
+/**
+ * Full event ready for persistence (has all required fields including sequenceNumber).
+ */
+export type IndexedEventInput = Omit<
+  IndexedEvent,
+  'id' | 'createdAt' | 'updatedAt'
+>;
+
+/**
+ * Event with sequenceNumber not yet assigned (returned by normalizer methods).
+ */
+export type UnsequencedEvent = Omit<IndexedEventInput, 'sequenceNumber'> & {
+  sequenceNumber?: string;
+};
+
+/**
+ * Normalizes raw Stellar operations and Soroban contract events into a unified
+ * {@link IndexedEvent} shape. This keeps the rest of the indexer pipeline free
+ * of SDK-specific types and makes cross-cutting queries possible.
+ */
+@Injectable()
+export class EventNormalizer {
+  private readonly logger = new Logger(EventNormalizer.name);
+  private sequenceCounter = BigInt(0);
+
+  /**
+   * Initializes the sequence counter from the persisted high-water mark.
+   * Must be called once during startup before normalizing any events.
+   */
+  initializeSequenceCounter(from: bigint): void {
+    this.sequenceCounter = from;
+    this.logger.log(`Sequence counter initialized at ${from}`);
+  }
+
+  getNextSequence(): bigint {
+    this.sequenceCounter += BigInt(1);
+    return this.sequenceCounter;
+  }
+
+  getCurrentSequence(): bigint {
+    return this.sequenceCounter;
+  }
+
+  /**
+   * Converts Stellar native operations into normalized event inputs.
+   * Returns events without sequence numbers — caller must use assignSequence().
+   */
+  normalizeStellarOperations(
+    tx: RawTransaction,
+    operations: RawOperation[],
+  ): UnsequencedEvent[] {
+    const results: UnsequencedEvent[] = [];
+
+    for (const op of operations) {
+      const eventType = this.mapOperationType(op.type);
+      if (!eventType) continue;
+
+      results.push({
+        ledgerSequence: tx.ledger,
+        eventType,
+        transactionHash: tx.hash,
+        timestamp: new Date(tx.created_at),
+        sourceAccount: op.from ?? tx.source_account,
+        destinationAccount: this.resolveDestination(op),
+        assetCode: op.asset_code ?? 'native',
+        assetIssuer: op.asset_issuer,
+        amount: op.amount ?? op.starting_balance ?? undefined,
+        raw: { operation: op, transaction: { hash: tx.hash, ledger: tx.ledger } },
+        invalidated: false,
+      });
+    }
+
+    return results;
+  }
+
+  /**
+   * Converts a Soroban contract event into a normalized event input.
+   * Returns events without sequence numbers — caller must use assignSequence().
+   */
+  normalizeSorobanEvent(
+    sorobanEvent: ParsedContractEvent,
+  ): UnsequencedEvent {
+    const eventType = this.mapSorobanEventType(sorobanEvent.type);
+
+    return {
+      ledgerSequence: sorobanEvent.ledger,
+      eventType,
+      transactionHash: sorobanEvent.txHash,
+      timestamp: new Date(sorobanEvent.ledgerClosedAt),
+      sourceAccount: sorobanEvent.contractId,
+      contractId: sorobanEvent.contractId,
+      methodName: this.extractMethodName(sorobanEvent.topics),
+      topics: sorobanEvent.topics,
+      value: sorobanEvent.value,
+      normalizedData: {
+        sorobanId: sorobanEvent.id,
+        pagingToken: sorobanEvent.pagingToken,
+        eventType: sorobanEvent.type,
+      },
+      raw: {
+        id: sorobanEvent.id,
+        pagingToken: sorobanEvent.pagingToken,
+        indexedAt: sorobanEvent.indexedAt,
+      },
+      invalidated: false,
+    };
+  }
+
+  /**
+   * Batch-converts Soroban events.
+   */
+  normalizeSorobanEvents(
+    events: ParsedContractEvent[],
+  ): UnsequencedEvent[] {
+    return events.map((e) => this.normalizeSorobanEvent(e));
+  }
+
+  /**
+   * Builds a complete IndexedEvent ready for persistence, assigning sequence numbers.
+   */
+  assignSequence(
+    input: UnsequencedEvent,
+  ): IndexedEventInput {
+    return {
+      ...input,
+      sequenceNumber: this.getNextSequence().toString(),
+    } as IndexedEventInput;
+  }
+
+  private mapOperationType(
+    horizonType: string,
+  ): BlockchainEventType | null {
+    const mapping: Record<string, BlockchainEventType> = {
+      payment: BlockchainEventType.PAYMENT,
+      path_payment_strict_receive:
+        BlockchainEventType.PATH_PAYMENT_STRICT_RECEIVE,
+      path_payment_strict_send: BlockchainEventType.PATH_PAYMENT_STRICT_SEND,
+      manage_offer: BlockchainEventType.MANAGE_OFFER,
+      create_account: BlockchainEventType.CREATE_ACCOUNT,
+      account_merge: BlockchainEventType.ACCOUNT_MERGE,
+    };
+
+    return mapping[horizonType] ?? null;
+  }
+
+  private mapSorobanEventType(
+    sorobanType: string,
+  ): BlockchainEventType {
+    switch (sorobanType) {
+      case 'contract':
+        return BlockchainEventType.SOROBAN_CONTRACT_EVENT;
+      case 'system':
+        return BlockchainEventType.SOROBAN_SYSTEM_EVENT;
+      case 'diagnostic':
+        return BlockchainEventType.SOROBAN_DIAGNOSTIC_EVENT;
+      default:
+        return BlockchainEventType.SOROBAN_CONTRACT_EVENT;
+    }
+  }
+
+  private resolveDestination(op: RawOperation): string | undefined {
+    return op.to ?? op.into;
+  }
+
+  private extractMethodName(topics: unknown[]): string | undefined {
+    if (!topics || topics.length === 0) return undefined;
+    const first = topics[0];
+    return typeof first === 'string' ? first : String(first);
+  }
+}
diff --git a/src/modules/blockchain-indexer/services/horizon-stream.service.ts b/src/modules/blockchain-indexer/services/horizon-stream.service.ts
new file mode 100644
index 0000000..111c9b4
--- /dev/null
+++ b/src/modules/blockchain-indexer/services/horizon-stream.service.ts
@@ -0,0 +1,287 @@
+import {
+  Injectable,
+  Logger,
+  OnModuleDestroy,
+} from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import { EventEmitter } from 'events';
+import { IndexingStateService } from './indexing-state.service';
+
+/**
+ * Emitted when a new ledger close event is received from Horizon.
+ */
+export interface LedgerCloseEvent {
+  sequence: number;
+  closedAt: string;
+  hash: string;
+  header: Record<string, unknown>;
+}
+
+/**
+ * Emitted when the stream health status changes.
+ */
+export interface StreamHealthEvent {
+  status: 'connected' | 'reconnecting' | 'error';
+  lastLedgerSequence?: number | null;
+  error?: string;
+}
+
+/**
+ * Maintains a persistent SSE connection to the Stellar Horizon API to receive
+ * ledger close events in real time. This replaces the polling-based approach
+ * with sub-second latency for event detection.
+ *
+ * Features:
+ * - Automatic reconnection with exponential backoff
+ * - Health status tracking and event emission
+ * - Resumable streams via cursor tracking
+ */
+@Injectable()
+export class HorizonStreamService
+  extends EventEmitter
+  implements OnModuleDestroy
+{
+  private readonly logger = new Logger(HorizonStreamService.name);
+  private readonly horizonUrl: string;
+  private readonly reconnectBaseMs: number;
+  private readonly reconnectMaxMs: number;
+
+  private currentController: AbortController | null = null;
+  private reconnectTimer: NodeJS.Timeout | null = null;
+  private currentReconnectDelay = 0;
+  private connected = false;
+  private lastLedgerSequence: number | null = null;
+  private stopping = false;
+
+  constructor(
+    private readonly configService: ConfigService,
+    private readonly stateService: IndexingStateService,
+  ) {
+    super();
+    this.horizonUrl =
+      this.configService.get<string>('stellar.horizonUrl') ??
+      'https://horizon-testnet.stellar.org';
+    this.reconnectBaseMs =
+      this.configService.get<number>('blockchainIndexer.wsReconnectBaseDelayMs') ??
+      1000;
+    this.reconnectMaxMs =
+      this.configService.get<number>('blockchainIndexer.wsReconnectMaxDelayMs') ??
+      30000;
+  }
+
+  onModuleDestroy(): void {
+    this.stop();
+  }
+
+  /**
+   * Starts the SSE stream. Resumes from the last known cursor if available.
+   */
+  async start(): Promise<void> {
+    if (this.currentController) return;
+    this.stopping = false;
+    this.currentReconnectDelay = 0;
+
+    const cursor = await this.stateService.getLastLedgerCursor();
+    this.lastLedgerSequence = cursor
+      ? Number(cursor)
+      : null;
+
+    this.logger.log(
+      `Starting Horizon SSE stream from ledger ${this.lastLedgerSequence ?? 'latest'}`,
+    );
+    this.connect();
+  }
+
+  /**
+   * Gracefully stops the SSE stream.
+   */
+  stop(): void {
+    this.stopping = true;
+    if (this.reconnectTimer) {
+      clearTimeout(this.reconnectTimer);
+      this.reconnectTimer = null;
+    }
+    if (this.currentController) {
+      this.currentController.abort();
+      this.currentController = null;
+    }
+    this.connected = false;
+    this.logger.log('Horizon SSE stream stopped');
+  }
+
+  /**
+   * Whether the stream is currently connected and receiving events.
+   */
+  isConnected(): boolean {
+    return this.connected;
+  }
+
+  getLastLedgerSequence(): number | null {
+    return this.lastLedgerSequence;
+  }
+
+  /**
+   * Manually reset the stream cursor to a specific ledger sequence.
+   */
+  async resetCursor(sequence: number): Promise<void> {
+    this.lastLedgerSequence = sequence;
+    await this.stateService.setLedgerCursor(String(sequence));
+    this.logger.log(`Stream cursor reset to ledger ${sequence}`);
+    // Restart the connection to pick up from the new cursor.
+    if (this.connected) {
+      this.disconnect();
+      this.connect();
+    }
+  }
+
+  private connect(): void {
+    if (this.stopping) return;
+
+    this.currentController = new AbortController();
+    const baseUrl = this.horizonUrl.replace(/\/$/, '');
+    const cursor = this.lastLedgerSequence
+      ? `cursor=${this.lastLedgerSequence}`
+      : '';
+    const url = `${baseUrl}/ledgers?order=asc&limit=200${cursor ? `&${cursor}` : ''}`;
+
+    this.logger.debug(`Connecting to Horizon SSE: ${url}`);
+
+    // Use fetch with streaming for SSE support (Node 18+ ReadableStream).
+    void this.streamLoop(url, this.currentController.signal);
+  }
+
+  private disconnect(): void {
+    if (this.currentController) {
+      this.currentController.abort();
+      this.currentController = null;
+    }
+    this.connected = false;
+  }
+
+  private async streamLoop(
+    url: string,
+    signal: AbortSignal,
+  ): Promise<void> {
+    try {
+      const response = await fetch(url, {
+        signal,
+        headers: {
+          Accept: 'text/event-stream',
+          'Cache-Control': 'no-cache',
+        },
+      });
+
+      if (!response.ok) {
+        throw new Error(`Horizon responded with ${response.status}`);
+      }
+
+      this.connected = true;
+      this.currentReconnectDelay = 0;
+      this.emit('health', {
+        status: 'connected',
+        lastLedgerSequence: this.lastLedgerSequence,
+      } satisfies StreamHealthEvent);
+
+      this.logger.log('Horizon SSE stream connected');
+
+      const reader = response.body?.getReader();
+      if (!reader) {
+        throw new Error('Response body is not readable');
+      }
+
+      const decoder = new TextDecoder();
+      let buffer = '';
+
+      while (!signal.aborted) {
+        const { done, value } = await reader.read();
+        if (done) break;
+
+        buffer += decoder.decode(value, { stream: true });
+        const lines = buffer.split('\n');
+        buffer = lines.pop() ?? '';
+
+        for (const line of lines) {
+          if (signal.aborted) break;
+          this.processLine(line);
+        }
+      }
+    } catch (error) {
+      if ((error as Error).name === 'AbortError') {
+        this.logger.debug('Horizon SSE stream aborted');
+        return;
+      }
+
+      this.logger.warn(
+        `Horizon SSE stream error: ${(error as Error).message}`,
+      );
+      this.connected = false;
+      this.emit('health', {
+        status: 'error',
+        error: (error as Error).message,
+        lastLedgerSequence: this.lastLedgerSequence ?? undefined,
+      } satisfies StreamHealthEvent);
+
+      if (!this.stopping) {
+        this.scheduleReconnect();
+      }
+    }
+  }
+
+  private processLine(line: string): void {
+    if (line.startsWith('event:')) {
+      // SSE event type marker — Horizon sends "event: ledgers" for ledger events.
+      return;
+    }
+
+    if (!line.startsWith('data:')) return;
+
+    const jsonStr = line.slice(5).trim();
+    if (!jsonStr) return;
+
+    try {
+      const data = JSON.parse(jsonStr) as Record<string, unknown>;
+
+      // Filter for ledger close events only.
+      if (
+        data.type !== 'ledger' &&
+        data._links &&
+        typeof data.sequence === 'number'
+      ) {
+        // This is a ledger record.
+        const ledgerEvent: LedgerCloseEvent = {
+          sequence: data.sequence as number,
+          closedAt: (data.closed_at as string) ?? '',
+          hash: (data.hash as string) ?? '',
+          header: data,
+        };
+
+        this.lastLedgerSequence = ledgerEvent.sequence;
+        this.emit('ledger', ledgerEvent);
+      }
+    } catch {
+      // Non-JSON or malformed data — ignore.
+    }
+  }
+
+  private scheduleReconnect(): void {
+    if (this.stopping || this.reconnectTimer) return;
+
+    const delay = Math.min(
+      this.reconnectBaseMs * Math.pow(2, this.currentReconnectDelay),
+      this.reconnectMaxMs,
+    );
+    this.currentReconnectDelay++;
+
+    this.logger.log(
+      `Reconnecting to Horizon SSE in ${delay}ms (attempt ${this.currentReconnectDelay})`,
+    );
+    this.emit('health', {
+      status: 'reconnecting',
+    } satisfies StreamHealthEvent);
+
+    this.reconnectTimer = setTimeout(() => {
+      this.reconnectTimer = null;
+      this.connect();
+    }, delay);
+  }
+}
diff --git a/src/modules/blockchain-indexer/services/indexing-state.service.ts b/src/modules/blockchain-indexer/services/indexing-state.service.ts
index 3175523..6cb353e 100644
--- a/src/modules/blockchain-indexer/services/indexing-state.service.ts
+++ b/src/modules/blockchain-indexer/services/indexing-state.service.ts
@@ -50,4 +50,25 @@ export class IndexingStateService {
   async setLastCursor(cursor: string): Promise<void> {
     await this.set('last_cursor', cursor);
   }
+
+  async getLedgerCursor(): Promise<string | null> {
+    return this.get('stream_ledger_cursor');
+  }
+
+  async setLedgerCursor(sequence: string): Promise<void> {
+    await this.set('stream_ledger_cursor', sequence);
+  }
+
+  async getLastLedgerCursor(): Promise<string | null> {
+    return this.get('stream_ledger_cursor');
+  }
+
+  async getSequenceCounter(): Promise<bigint> {
+    const value = await this.get('sequence_counter');
+    return value ? BigInt(value) : BigInt(0);
+  }
+
+  async setSequenceCounter(counter: bigint): Promise<void> {
+    await this.set('sequence_counter', counter.toString());
+  }
 }
diff --git a/src/modules/blockchain-indexer/services/ledger-indexer.service.ts b/src/modules/blockchain-indexer/services/ledger-indexer.service.ts
new file mode 100644
index 0000000..bf14642
--- /dev/null
+++ b/src/modules/blockchain-indexer/services/ledger-indexer.service.ts
@@ -0,0 +1,268 @@
+import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
+import { HorizonStreamService, LedgerCloseEvent } from './horizon-stream.service';
+import { EventNormalizer } from './event-normalizer.service';
+import { EventBufferService, BufferedEvent } from './event-buffer.service';
+import { BatchedPersistenceService } from './batched-persistence.service';
+import { SubscriptionManager, NormalizedEventPayload } from './subscription-manager.service';
+import { IndexingStateService } from './indexing-state.service';
+import { StellarEventSourceService } from './stellar-event-source.service';
+import { EventStreamService } from './event-stream.service';
+
+/**
+ * Orchestrates the full real-time indexing pipeline:
+ *
+ * 1. HorizonStreamService → ledger close events (SSE stream)
+ * 2. StellarEventSourceService → fetch transactions/operations per ledger
+ * 3. EventNormalizer → convert to IndexedEvent shape with sequence numbers
+ * 4. EventBufferService → in-memory deduplication and ordering
+ * 5. BatchedPersistenceService → high-throughput PostgreSQL writes
+ * 6. SubscriptionManager → distribute events to WebSocket subscribers
+ *
+ * This replaces the polling-based EventIndexerService with sub-second latency.
+ */
+@Injectable()
+export class LedgerIndexerService implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger(LedgerIndexerService.name);
+
+  private retentionTimer: NodeJS.Timeout | null = null;
+  private running = false;
+
+  constructor(
+    private readonly horizonStream: HorizonStreamService,
+    private readonly eventSource: StellarEventSourceService,
+    private readonly normalizer: EventNormalizer,
+    private readonly buffer: EventBufferService,
+    private readonly persistence: BatchedPersistenceService,
+    private readonly subscriptionManager: SubscriptionManager,
+    private readonly stateService: IndexingStateService,
+    private readonly legacyStreamService: EventStreamService,
+  ) {}
+
+  async onModuleInit(): Promise<void> {
+    await this.start();
+  }
+
+  async onModuleDestroy(): Promise<void> {
+    await this.stop();
+  }
+
+  /**
+   * Starts the real-time indexing pipeline.
+   */
+  async start(): Promise<void> {
+    if (this.running) return;
+    this.running = true;
+
+    this.logger.log('Starting real-time ledger indexer');
+
+    // Initialize the sequence counter from persisted state.
+    const lastSequence = await this.stateService.getSequenceCounter();
+    this.normalizer.initializeSequenceCounter(lastSequence);
+
+    // Start the persistence flush timer.
+    this.persistence.startFlushTimer();
+
+    // Wire up the Horizon SSE stream to process ledger close events.
+    this.horizonStream.on('ledger', (event: LedgerCloseEvent) => {
+      void this.handleLedgerClose(event);
+    });
+
+    // Start the SSE stream.
+    await this.horizonStream.start();
+
+    // Start periodic retention purge (every hour).
+    this.retentionTimer = setInterval(
+      () => {
+        void this.persistence.purgeExpired();
+      },
+      60 * 60 * 1000,
+    );
+    this.retentionTimer.unref?.();
+
+    this.logger.log('Real-time ledger indexer started');
+  }
+
+  /**
+   * Gracefully stops the pipeline.
+   */
+  async stop(): Promise<void> {
+    this.running = false;
+
+    if (this.retentionTimer) {
+      clearInterval(this.retentionTimer);
+      this.retentionTimer = null;
+    }
+
+    this.horizonStream.stop();
+
+    // Final flush.
+    await this.persistence.flush();
+
+    this.logger.log('Real-time ledger indexer stopped');
+  }
+
+  /**
+   * Handles a new ledger close event from the Horizon SSE stream.
+   * Fetches all transactions/operations for the ledger, normalizes them,
+   * buffers them, and triggers persistence.
+   */
+  private async handleLedgerClose(ledgerEvent: LedgerCloseEvent): Promise<void> {
+    const ledger = ledgerEvent.sequence;
+    this.logger.debug(`Processing ledger close: ${ledger}`);
+
+    try {
+      // Fetch all transactions for this ledger.
+      const transactions = await this.eventSource.getTransactionsByLedgerRange(
+        ledger,
+        ledger,
+      );
+
+      // Process each transaction's operations.
+      for (const tx of transactions) {
+        if (!tx.successful) continue;
+
+        let operations;
+        try {
+          operations = await this.eventSource.getOperationsForTransaction(tx.hash);
+        } catch {
+          this.logger.warn(
+            `Failed to fetch operations for tx ${tx.hash} in ledger ${ledger}`,
+          );
+          continue;
+        }
+
+        // Normalize operations into IndexedEvent inputs.
+        const normalizedEvents =
+          this.normalizer.normalizeStellarOperations(tx, operations);
+
+        // Assign sequence numbers and convert to buffered events.
+        const bufferedEvents: BufferedEvent[] = normalizedEvents.map((event) => {
+          const withSequence = this.normalizer.assignSequence(event);
+          return {
+            sequenceNumber: Number(withSequence.sequenceNumber),
+            ledgerSequence: withSequence.ledgerSequence,
+            eventType: withSequence.eventType,
+            timestamp: withSequence.timestamp.getTime(),
+            data: withSequence as unknown as Record<string, unknown>,
+            bufferedAt: Date.now(),
+          };
+        });
+
+        // Add to the buffer (deduplication happens here).
+        const accepted = this.buffer.pushBatch(bufferedEvents);
+        if (accepted > 0) {
+          // Drain the buffer and persist.
+          const toPersist = this.buffer.drain(this.buffer.getStats().bufferSize);
+          await this.persistAndDistribute(toPersist);
+        }
+      }
+
+      // Persist the ledger cursor for stream resumption.
+      await this.stateService.setLedgerCursor(String(ledger));
+    } catch (error) {
+      this.logger.warn(
+        `Failed to process ledger ${ledger}: ${(error as Error).message}`,
+      );
+    }
+  }
+
+  /**
+   * Handles Soroban contract events from the ContractEventIndexerService.
+   * Called externally when new Soroban events are detected.
+   */
+  async handleSorobanEvents(
+    events: Array<{ topics: unknown[]; value: unknown; [key: string]: unknown }>,
+  ): Promise<void> {
+    if (events.length === 0) return;
+
+    const normalizedEvents = this.normalizer.normalizeSorobanEvents(
+      events as any,
+    );
+
+    const bufferedEvents: BufferedEvent[] = normalizedEvents.map((event) => {
+      const withSequence = this.normalizer.assignSequence(event);
+      return {
+        sequenceNumber: Number(withSequence.sequenceNumber),
+        ledgerSequence: withSequence.ledgerSequence,
+        eventType: withSequence.eventType,
+        timestamp: withSequence.timestamp.getTime(),
+        data: withSequence as unknown as Record<string, unknown>,
+        bufferedAt: Date.now(),
+      };
+    });
+
+    const accepted = this.buffer.pushBatch(bufferedEvents);
+    if (accepted > 0) {
+      const toPersist = this.buffer.drain(this.buffer.getStats().bufferSize);
+      await this.persistAndDistribute(toPersist);
+    }
+  }
+
+  /**
+   * Persists events to the database and distributes them to subscribers.
+   */
+  private async persistAndDistribute(
+    events: BufferedEvent[],
+  ): Promise<void> {
+    if (events.length === 0) return;
+
+    // Enqueue for batched persistence.
+    for (const event of events) {
+      this.persistence.enqueue(event.data as any);
+    }
+
+    // Distribute to WebSocket subscribers.
+    for (const event of events) {
+      const payload: NormalizedEventPayload = {
+        sequenceNumber: event.sequenceNumber,
+        ledgerSequence: event.ledgerSequence,
+        eventType: event.eventType,
+        transactionHash: (event.data.transactionHash as string) ?? '',
+        timestamp: new Date(event.timestamp).toISOString(),
+        sourceAccount: (event.data.sourceAccount as string) ?? '',
+        destinationAccount: event.data.destinationAccount as string | undefined,
+        contractId: event.data.contractId as string | undefined,
+        methodName: event.data.methodName as string | undefined,
+        assetCode: event.data.assetCode as string | undefined,
+        amount: event.data.amount as string | undefined,
+        topics: event.data.topics as unknown[] | undefined,
+        value: event.data.value as unknown | undefined,
+        normalizedData: event.data.normalizedData as
+          | Record<string, unknown>
+          | undefined,
+      };
+
+      this.subscriptionManager.distributeEvent(payload);
+
+      // Also publish to the legacy Redis stream for backward compatibility.
+      void this.legacyStreamService.publish(
+        event.data as any,
+      );
+    }
+  }
+
+  /**
+   * Returns comprehensive status of the real-time indexer.
+   */
+  async getStatus() {
+    const horizonConnected = this.horizonStream.isConnected();
+    const lastLedger = this.horizonStream.getLastLedgerSequence();
+    const bufferStats = this.buffer.getStats();
+    const persistenceStats = this.persistence.getStats();
+    const subscriptionCount = this.subscriptionManager.getSubscriptionCount();
+    const clientCount = this.subscriptionManager.getClientCount();
+
+    return {
+      running: this.running,
+      horizonConnected,
+      lastLedgerSequence: lastLedger,
+      sequenceCounter: Number(this.normalizer.getCurrentSequence()),
+      buffer: bufferStats,
+      persistence: persistenceStats,
+      subscriptions: {
+        total: subscriptionCount,
+        clients: clientCount,
+      },
+    };
+  }
+}
diff --git a/src/modules/blockchain-indexer/services/subscription-manager.service.ts b/src/modules/blockchain-indexer/services/subscription-manager.service.ts
new file mode 100644
index 0000000..221d469
--- /dev/null
+++ b/src/modules/blockchain-indexer/services/subscription-manager.service.ts
@@ -0,0 +1,231 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { EventEmitter } from 'events';
+
+/**
+ * Represents a client subscription to specific event types.
+ */
+export interface EventSubscription {
+  /** Unique subscription ID. */
+  id: string;
+  /** Socket/client ID that owns this subscription. */
+  clientId: string;
+  /** Filtered event types. Empty array means "all events". */
+  eventTypes: string[];
+  /** Optional contract ID filter (Soroban events). */
+  contractIds?: string[];
+  /** Optional account filter. */
+  accounts?: string[];
+  /** Optional minimum ledger sequence. */
+  fromLedger?: number;
+}
+
+/**
+ * A normalized event pushed to subscribers.
+ */
+export interface NormalizedEventPayload {
+  sequenceNumber: number;
+  ledgerSequence: number;
+  eventType: string;
+  transactionHash: string;
+  timestamp: string;
+  sourceAccount: string;
+  destinationAccount?: string;
+  contractId?: string;
+  methodName?: string;
+  assetCode?: string;
+  amount?: string;
+  topics?: unknown[];
+  value?: unknown;
+  normalizedData?: Record<string, unknown>;
+}
+
+/**
+ * Client-side subscription info for admin/debug endpoints.
+ */
+export interface SubscriptionInfo {
+  id: string;
+  clientId: string;
+  eventTypes: string[];
+  contractIds?: string[];
+  accounts?: string[];
+  createdAt: Date;
+}
+
+/**
+ * Manages pub/sub subscriptions for real-time event distribution. Clients
+ * register interest in specific event types, contract IDs, or accounts, and
+ * receive only matching events through their WebSocket connection.
+ *
+ * This decouples the persistence pipeline from the client-facing delivery,
+ * allowing independent scaling and failure isolation.
+ */
+@Injectable()
+export class SubscriptionManager extends EventEmitter {
+  private readonly logger = new Logger(SubscriptionManager.name);
+
+  /** All active subscriptions keyed by subscription ID. */
+  private readonly subscriptions = new Map<string, EventSubscription>();
+
+  /** Reverse index: clientId → Set of subscription IDs for fast cleanup. */
+  private readonly clientIndex = new Map<string, Set<string>>();
+
+  private nextId = 1;
+
+  /**
+   * Creates a new subscription and returns its ID.
+   */
+  subscribe(subscription: Omit<EventSubscription, 'id'>): string {
+    const id = `sub_${this.nextId++}`;
+    const full: EventSubscription = { ...subscription, id };
+
+    this.subscriptions.set(id, full);
+
+    // Maintain reverse index.
+    if (!this.clientIndex.has(subscription.clientId)) {
+      this.clientIndex.set(subscription.clientId, new Set());
+    }
+    this.clientIndex.get(subscription.clientId)!.add(id);
+
+    this.logger.debug(
+      `Subscription ${id} created for client ${subscription.clientId} ` +
+        `(types: ${full.eventTypes.length > 0 ? full.eventTypes.join(',') : 'all'})`,
+    );
+
+    return id;
+  }
+
+  /**
+   * Removes a subscription by ID.
+   */
+  unsubscribe(subscriptionId: string): boolean {
+    const sub = this.subscriptions.get(subscriptionId);
+    if (!sub) return false;
+
+    this.subscriptions.delete(subscriptionId);
+    const clientSubs = this.clientIndex.get(sub.clientId);
+    if (clientSubs) {
+      clientSubs.delete(subscriptionId);
+      if (clientSubs.size === 0) {
+        this.clientIndex.delete(sub.clientId);
+      }
+    }
+
+    this.logger.debug(`Subscription ${subscriptionId} removed`);
+    return true;
+  }
+
+  /**
+   * Removes all subscriptions for a disconnected client.
+   */
+  removeClientSubscriptions(clientId: string): number {
+    const clientSubs = this.clientIndex.get(clientId);
+    if (!clientSubs || clientSubs.size === 0) return 0;
+
+    const count = clientSubs.size;
+    for (const subId of clientSubs) {
+      this.subscriptions.delete(subId);
+    }
+    this.clientIndex.delete(clientId);
+
+    this.logger.debug(
+      `Removed ${count} subscriptions for disconnected client ${clientId}`,
+    );
+    return count;
+  }
+
+  /**
+   * Filters and distributes an event to matching subscriptions.
+   * Returns the list of subscription IDs that received the event.
+   */
+  distributeEvent(event: NormalizedEventPayload): string[] {
+    const matchedSubscriptions: string[] = [];
+
+    for (const [, sub] of this.subscriptions) {
+      if (this.matchesSubscription(event, sub)) {
+        matchedSubscriptions.push(sub.id);
+        this.emit(`event:${sub.clientId}`, {
+          subscriptionId: sub.id,
+          event,
+        });
+      }
+    }
+
+    return matchedSubscriptions;
+  }
+
+  /**
+   * Returns all subscriptions, optionally filtered by client ID.
+   */
+  listSubscriptions(clientId?: string): SubscriptionInfo[] {
+    const results: SubscriptionInfo[] = [];
+
+    for (const [, sub] of this.subscriptions) {
+      if (clientId && sub.clientId !== clientId) continue;
+      results.push({
+        id: sub.id,
+        clientId: sub.clientId,
+        eventTypes: sub.eventTypes,
+        contractIds: sub.contractIds,
+        accounts: sub.accounts,
+        createdAt: new Date(), // tracked in-memory, not persisted
+      });
+    }
+
+    return results;
+  }
+
+  /**
+   * Returns the total number of active subscriptions.
+   */
+  getSubscriptionCount(): number {
+    return this.subscriptions.size;
+  }
+
+  /**
+   * Returns the number of unique connected clients.
+   */
+  getClientCount(): number {
+    return this.clientIndex.size;
+  }
+
+  /**
+   * Checks whether an event matches a subscription's filters.
+   */
+  private matchesSubscription(
+    event: NormalizedEventPayload,
+    sub: EventSubscription,
+  ): boolean {
+    // Event type filter.
+    if (
+      sub.eventTypes.length > 0 &&
+      !sub.eventTypes.includes(event.eventType)
+    ) {
+      return false;
+    }
+
+    // Contract ID filter.
+    if (sub.contractIds && sub.contractIds.length > 0) {
+      if (!event.contractId || !sub.contractIds.includes(event.contractId)) {
+        return false;
+      }
+    }
+
+    // Account filter (matches source or destination).
+    if (sub.accounts && sub.accounts.length > 0) {
+      const matchesAccount =
+        sub.accounts.includes(event.sourceAccount) ||
+        (event.destinationAccount &&
+          sub.accounts.includes(event.destinationAccount));
+      if (!matchesAccount) {
+        return false;
+      }
+    }
+
+    // Ledger range filter.
+    if (sub.fromLedger !== undefined && event.ledgerSequence < sub.fromLedger) {
+      return false;
+    }
+
+    return true;
+  }
+}
diff --git a/src/modules/blockchain-indexer/subscription-manager.service.spec.ts b/src/modules/blockchain-indexer/subscription-manager.service.spec.ts
new file mode 100644
index 0000000..ddc1cee
--- /dev/null
+++ b/src/modules/blockchain-indexer/subscription-manager.service.spec.ts
@@ -0,0 +1,218 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { ConfigModule } from '@nestjs/config';
+import {
+  SubscriptionManager,
+  NormalizedEventPayload,
+} from './services/subscription-manager.service';
+
+describe('SubscriptionManager', () => {
+  let module: TestingModule;
+  let manager: SubscriptionManager;
+
+  const makeEvent = (
+    overrides: Partial<NormalizedEventPayload> = {},
+  ): NormalizedEventPayload => ({
+    sequenceNumber: 1,
+    ledgerSequence: 50000,
+    eventType: 'payment',
+    transactionHash: 'tx1',
+    timestamp: '2024-01-15T10:30:00Z',
+    sourceAccount: 'GABC...',
+    ...overrides,
+  });
+
+  beforeAll(async () => {
+    module = await Test.createTestingModule({
+      imports: [ConfigModule.forRoot({ isGlobal: true })],
+      providers: [SubscriptionManager],
+    }).compile();
+
+    manager = module.get(SubscriptionManager);
+  });
+
+  afterAll(async () => {
+    await module.close();
+  });
+
+  beforeEach(() => {
+    // Clean up all subscriptions.
+    for (const sub of manager.listSubscriptions()) {
+      manager.unsubscribe(sub.id);
+    }
+  });
+
+  describe('subscribe / unsubscribe', () => {
+    it('should create a subscription and return an ID', () => {
+      const id = manager.subscribe({
+        clientId: 'client1',
+        eventTypes: ['payment'],
+      });
+      expect(id).toBeDefined();
+      expect(id).toMatch(/^sub_/);
+      expect(manager.getSubscriptionCount()).toBe(1);
+    });
+
+    it('should remove a subscription by ID', () => {
+      const id = manager.subscribe({
+        clientId: 'client1',
+        eventTypes: [],
+      });
+      const removed = manager.unsubscribe(id);
+      expect(removed).toBe(true);
+      expect(manager.getSubscriptionCount()).toBe(0);
+    });
+
+    it('should return false for non-existent subscription', () => {
+      const removed = manager.unsubscribe('sub_nonexistent');
+      expect(removed).toBe(false);
+    });
+  });
+
+  describe('removeClientSubscriptions', () => {
+    it('should remove all subscriptions for a client', () => {
+      manager.subscribe({ clientId: 'client1', eventTypes: ['payment'] });
+      manager.subscribe({ clientId: 'client1', eventTypes: ['trade'] });
+      manager.subscribe({ clientId: 'client2', eventTypes: ['payment'] });
+
+      const removed = manager.removeClientSubscriptions('client1');
+
+      expect(removed).toBe(2);
+      expect(manager.getSubscriptionCount()).toBe(1);
+    });
+
+    it('should return 0 for unknown client', () => {
+      const removed = manager.removeClientSubscriptions('unknown');
+      expect(removed).toBe(0);
+    });
+  });
+
+  describe('distributeEvent', () => {
+    it('should match all events when eventTypes is empty', () => {
+      const id = manager.subscribe({
+        clientId: 'client1',
+        eventTypes: [],
+      });
+
+      const event = makeEvent({ eventType: 'payment' });
+      const matched = manager.distributeEvent(event);
+
+      expect(matched).toContain(id);
+    });
+
+    it('should match events by type', () => {
+      const id = manager.subscribe({
+        clientId: 'client1',
+        eventTypes: ['payment', 'trade'],
+      });
+
+      const payment = makeEvent({ eventType: 'payment' });
+      const trade = makeEvent({ eventType: 'trade' });
+      const liquidation = makeEvent({ eventType: 'liquidation' });
+
+      expect(manager.distributeEvent(payment)).toContain(id);
+      expect(manager.distributeEvent(trade)).toContain(id);
+      expect(manager.distributeEvent(liquidation)).not.toContain(id);
+    });
+
+    it('should match events by contractId', () => {
+      const id = manager.subscribe({
+        clientId: 'client1',
+        eventTypes: [],
+        contractIds: ['CCONTRACT1...'],
+      });
+
+      const match = makeEvent({
+        contractId: 'CCONTRACT1...',
+        eventType: 'soroban_contract_event',
+      });
+      const noMatch = makeEvent({
+        contractId: 'CCONTRACT2...',
+        eventType: 'soroban_contract_event',
+      });
+
+      expect(manager.distributeEvent(match)).toContain(id);
+      expect(manager.distributeEvent(noMatch)).not.toContain(id);
+    });
+
+    it('should match events by account', () => {
+      const id = manager.subscribe({
+        clientId: 'client1',
+        eventTypes: [],
+        accounts: ['GABC...'],
+      });
+
+      const sourceMatch = makeEvent({ sourceAccount: 'GABC...' });
+      const destMatch = makeEvent({
+        sourceAccount: 'GOTHER...',
+        destinationAccount: 'GABC...',
+      });
+      const noMatch = makeEvent({
+        sourceAccount: 'GOTHER...',
+        destinationAccount: 'GXYZ...',
+      });
+
+      expect(manager.distributeEvent(sourceMatch)).toContain(id);
+      expect(manager.distributeEvent(destMatch)).toContain(id);
+      expect(manager.distributeEvent(noMatch)).not.toContain(id);
+    });
+
+    it('should match events by ledger sequence', () => {
+      const id = manager.subscribe({
+        clientId: 'client1',
+        eventTypes: [],
+        fromLedger: 50000,
+      });
+
+      const match = makeEvent({ ledgerSequence: 50001 });
+      const noMatch = makeEvent({ ledgerSequence: 49999 });
+
+      expect(manager.distributeEvent(match)).toContain(id);
+      expect(manager.distributeEvent(noMatch)).not.toContain(id);
+    });
+
+    it('should emit events to the correct client', () => {
+      const events: any[] = [];
+      manager.on('event:client1', (payload) => events.push(payload));
+
+      manager.subscribe({
+        clientId: 'client1',
+        eventTypes: ['payment'],
+      });
+
+      manager.distributeEvent(makeEvent({ eventType: 'payment' }));
+      manager.distributeEvent(makeEvent({ eventType: 'trade' }));
+
+      expect(events).toHaveLength(1);
+      expect(events[0].event.eventType).toBe('payment');
+    });
+  });
+
+  describe('listSubscriptions', () => {
+    it('should list all subscriptions', () => {
+      manager.subscribe({ clientId: 'client1', eventTypes: ['payment'] });
+      manager.subscribe({ clientId: 'client2', eventTypes: ['trade'] });
+
+      const all = manager.listSubscriptions();
+      expect(all).toHaveLength(2);
+    });
+
+    it('should filter by client ID', () => {
+      manager.subscribe({ clientId: 'client1', eventTypes: ['payment'] });
+      manager.subscribe({ clientId: 'client2', eventTypes: ['trade'] });
+
+      const client1Subs = manager.listSubscriptions('client1');
+      expect(client1Subs).toHaveLength(1);
+      expect(client1Subs[0].clientId).toBe('client1');
+    });
+  });
+
+  describe('getClientCount', () => {
+    it('should count unique clients', () => {
+      manager.subscribe({ clientId: 'client1', eventTypes: ['payment'] });
+      manager.subscribe({ clientId: 'client1', eventTypes: ['trade'] });
+      manager.subscribe({ clientId: 'client2', eventTypes: ['payment'] });
+
+      expect(manager.getClientCount()).toBe(2);
+    });
+  });
+});