feat: Stellar Transaction History Importer module [#642]

backend/package.json

@@ -40,6 +40,10 @@
     "rxjs": "^7.8.1",
     "sharp": "^0.34.4",
     "socket.io": "^4.8.1",
+    "@nestjs/bullmq": "^11.0.2",
+    "@nestjs/event-emitter": "^3.0.1",
+    "bullmq": "^5.56.0",
+    "ioredis": "^5.6.1",
     "stellar-sdk": "^13.3.0",
     "typeorm": "^0.3.26"
   },

backend/src/app.module.ts

@@ -1,6 +1,7 @@
 import { Module } from '@nestjs/common';
-import { ConfigModule } from '@nestjs/config';
+import { ConfigModule, ConfigService } from '@nestjs/config';
 import { EventEmitterModule } from '@nestjs/event-emitter';
+import { BullModule } from '@nestjs/bullmq';
 import { AppController } from './app.controller';
 import { AppService } from './app.service';
 import * as path from 'path';
@@ -52,6 +53,17 @@ function loadModules(): (new () => any)[] {
     DatabaseModule,
     HealthModule,
     EventEmitterModule.forRoot(),
+    BullModule.forRootAsync({
+      imports: [ConfigModule],
+      inject: [ConfigService],
+      useFactory: (config: ConfigService) => ({
+        connection: {
+          host: config.get<string>('REDIS_HOST', 'localhost'),
+          port: config.get<number>('REDIS_PORT', 6379),
+          password: config.get<string>('REDIS_PASSWORD'),
+        },
+      }),
+    }),
     CacheModule,
     LeaderboardsModule,
     ...loadModules(),

backend/src/stellar-history-importer/dto/import-status.dto.ts

@@ -0,0 +1,17 @@
+import { ImportJobStatus } from '../entities/history-import-job.entity';
+
+export class ImportStatusDto {
+  jobId!: string;
+  walletId!: string;
+  status!: ImportJobStatus;
+  totalImported!: number;
+  cursor?: string;
+  errorMessage?: string;
+  startedAt!: Date;
+  completedAt?: Date;
+}
+
+export class TriggerImportResponseDto {
+  jobId!: string;
+  message!: string;
+}

backend/src/stellar-history-importer/entities/history-import-job.entity.ts

@@ -0,0 +1,48 @@
+import {
+  Entity,
+  PrimaryGeneratedColumn,
+  Column,
+  CreateDateColumn,
+  Index,
+} from 'typeorm';
+
+export enum ImportJobStatus {
+  PENDING = 'PENDING',
+  RUNNING = 'RUNNING',
+  COMPLETED = 'COMPLETED',
+  FAILED = 'FAILED',
+}
+
+@Entity('history_import_jobs')
+@Index(['walletId'])
+@Index(['status'])
+export class HistoryImportJob {
+  @PrimaryGeneratedColumn('uuid')
+  id!: string;
+
+  @Column('uuid')
+  walletId!: string;
+
+  @Column({
+    type: 'enum',
+    enum: ImportJobStatus,
+    default: ImportJobStatus.PENDING,
+  })
+  status!: ImportJobStatus;
+
+  @Column({ type: 'int', default: 0 })
+  totalImported!: number;
+
+  /** Horizon paging_token of the last successfully imported transaction */
+  @Column({ length: 100, nullable: true })
+  cursor?: string;
+
+  @Column({ type: 'text', nullable: true })
+  errorMessage?: string;
+
+  @CreateDateColumn()
+  startedAt!: Date;
+
+  @Column({ type: 'timestamp', nullable: true })
+  completedAt?: Date;
+}

backend/src/stellar-history-importer/listeners/wallet-verified.listener.ts

@@ -0,0 +1,27 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { OnEvent } from '@nestjs/event-emitter';
+import { StellarHistoryImporterService } from '../stellar-history-importer.service';
+
+export interface WalletVerifiedEvent {
+  walletId: string;
+  walletAddress: string;
+}
+
+/**
+ * Listens for wallet.verified events (emitted by WalletIntegrationService after
+ * a successful wallet connection) and automatically queues a history import.
+ */
+@Injectable()
+export class WalletVerifiedListener {
+  private readonly logger = new Logger(WalletVerifiedListener.name);
+
+  constructor(private readonly importerService: StellarHistoryImporterService) {}
+
+  @OnEvent('wallet.verified')
+  async handleWalletVerified(event: WalletVerifiedEvent): Promise<void> {
+    this.logger.log(
+      `wallet.verified event received — queuing history import for wallet ${event.walletId}`,
+    );
+    await this.importerService.triggerImport(event.walletId, event.walletAddress);
+  }
+}

backend/src/stellar-history-importer/processors/history-import.processor.ts

@@ -0,0 +1,33 @@
+import { Processor, WorkerHost } from '@nestjs/bullmq';
+import { Logger } from '@nestjs/common';
+import { Job } from 'bullmq';
+import { StellarHistoryImporterService } from '../stellar-history-importer.service';
+
+export const HISTORY_IMPORT_QUEUE = 'stellar-history-import';
+
+export interface HistoryImportJobData {
+  jobId: string;
+  walletId: string;
+  walletAddress: string;
+}
+
+@Processor(HISTORY_IMPORT_QUEUE)
+export class HistoryImportProcessor extends WorkerHost {
+  private readonly logger = new Logger(HistoryImportProcessor.name);
+
+  constructor(private readonly importerService: StellarHistoryImporterService) {
+    super();
+  }
+
+  async process(job: Job<HistoryImportJobData>): Promise<void> {
+    const { jobId, walletId, walletAddress } = job.data;
+    this.logger.log(`Processing history import job ${jobId} for wallet ${walletId}`);
+
+    try {
+      await this.importerService.importHistory(jobId, walletId, walletAddress);
+    } catch (error) {
+      this.logger.error(`History import job ${jobId} failed: ${error.message}`);
+      throw error;
+    }
+  }
+}

backend/src/stellar-history-importer/stellar-history-importer.controller.ts

@@ -0,0 +1,46 @@
+import {
+  Controller,
+  Post,
+  Get,
+  Param,
+  Body,
+  HttpCode,
+  HttpStatus,
+} from '@nestjs/common';
+import { StellarHistoryImporterService } from './stellar-history-importer.service';
+import { ImportStatusDto, TriggerImportResponseDto } from './dto/import-status.dto';
+
+class TriggerImportBodyDto {
+  walletAddress!: string;
+}
+
+@Controller('wallets')
+export class StellarHistoryImporterController {
+  constructor(private readonly importerService: StellarHistoryImporterService) {}
+
+  /**
+   * POST /wallets/:id/import-history
+   * Trigger a full history import for the given wallet.
+   */
+  @Post(':id/import-history')
+  @HttpCode(HttpStatus.ACCEPTED)
+  async triggerImport(
+    @Param('id') walletId: string,
+    @Body() body: TriggerImportBodyDto,
+  ): Promise<TriggerImportResponseDto> {
+    const job = await this.importerService.triggerImport(walletId, body.walletAddress);
+    return {
+      jobId: job.id,
+      message: 'History import queued successfully',
+    };
+  }
+
+  /**
+   * GET /wallets/:id/import-status
+   * Return the latest import job status for the given wallet.
+   */
+  @Get(':id/import-status')
+  async getImportStatus(@Param('id') walletId: string): Promise<ImportStatusDto> {
+    return this.importerService.getImportStatus(walletId);
+  }
+}

backend/src/stellar-history-importer/stellar-history-importer.module.ts

@@ -0,0 +1,26 @@
+import { Module } from '@nestjs/common';
+import { TypeOrmModule } from '@nestjs/typeorm';
+import { BullModule } from '@nestjs/bullmq';
+import { ConfigModule } from '@nestjs/config';
+import { HistoryImportJob } from './entities/history-import-job.entity';
+import { StellarHistoryImporterService } from './stellar-history-importer.service';
+import { StellarHistoryImporterController } from './stellar-history-importer.controller';
+import {
+  HistoryImportProcessor,
+  HISTORY_IMPORT_QUEUE,
+} from './processors/history-import.processor';
+import { WalletVerifiedListener } from './listeners/wallet-verified.listener';
+
+@Module({
+  imports: [
+    ConfigModule,
+    TypeOrmModule.forFeature([HistoryImportJob]),
+    BullModule.registerQueue({
+      name: HISTORY_IMPORT_QUEUE,
+    }),
+  ],
+  controllers: [StellarHistoryImporterController],
+  providers: [StellarHistoryImporterService, HistoryImportProcessor, WalletVerifiedListener],
+  exports: [StellarHistoryImporterService],
+})
+export class StellarHistoryImporterModule {}