Merge pull request #245 from prismn/feat/setup-bullmq

Feat/setup bullmq

Author: mftee

backend/.env.example

@@ -47,4 +47,14 @@ THROTTLE_LIMIT=10
 # File Upload
 UPLOAD_DIR=./uploads
 
+# Stellar Configuration
+STELLAR_SECRET_KEY=your-stellar-secret-key
+STELLAR_HORIZON_URL=https://horizon-testnet.stellar.org
+STELLAR_NETWORK=Test SDF Network ; September 2015
+
+# Redis Configuration
+REDIS_HOST=localhost
+REDIS_PORT=6379
+REDIS_PASSWORD=
+
 

backend/package-lock.json

@@ -40,6 +40,7 @@
     "class-transformer": "^0.5.1",
     "class-validator": "^0.14.2",
     "exceljs": "^4.4.0",
+    "ioredis": "^5.10.1",
     "nodemailer": "^7.0.12",
     "passport": "^0.7.0",
     "passport-github2": "^0.1.12",

backend/package.json

@@ -5,7 +5,10 @@ import { AppController } from './app.controller';
 import { AppService } from './app.service';
 import { AuthModule } from './auth/auth.module';
 import { DocumentsModule } from './documents/documents.module';
+import { MailModule } from './mail/mail.module';
+import { QueueModule } from './queue/queue.module';
 import { RiskAssessmentModule } from './risk-assessment/risk-assessment.module';
+import { StellarModule } from './stellar/stellar.module';
 import { UsersModule } from './users/users.module';
 import { VerificationModule } from './verification/verification.module';
 
@@ -33,7 +36,10 @@ import { VerificationModule } from './verification/verification.module';
     AuthModule,
     DocumentsModule,
     RiskAssessmentModule,
+    StellarModule,
     VerificationModule,
+    MailModule,
+    QueueModule,
   ],
   controllers: [AppController],
   providers: [AppService],

backend/src/app.module.ts

@@ -1,6 +1,10 @@
-import {
+import {
   BadRequestException,
+  ConflictException,
   Controller,
+  Get,
+  NotFoundException,
+  Param,
   Post,
   Req,
   Res,
@@ -20,6 +24,8 @@ import { DocumentsService } from './documents.service';
 import { DocumentStatus } from './entities/document.entity';
 import { JwtAuthGuard } from '../auth/guards/jwt-auth.guard';
 import { User } from '../users/entities/user.entity';
+import { QueueService } from '../queue/queue.service';
+import { VerificationService } from '../verification/verification.service';
 
 const ALLOWED_MIME_TYPES = ['application/pdf', 'image/png', 'image/jpeg'];
 const MAX_FILE_SIZE_BYTES = 20 * 1024 * 1024;
@@ -39,6 +45,8 @@ export class DocumentsController {
   constructor(
     private readonly documentsService: DocumentsService,
     private readonly configService: ConfigService,
+    private readonly queueService: QueueService,
+    private readonly verificationService: VerificationService,
   ) {}
 
   @Post('upload')
@@ -74,7 +82,7 @@ export class DocumentsController {
     await fs.mkdir(uploadDir, { recursive: true });
 
     const extension = extname(file.originalname) || '';
-    const filename = ${fileHash};
+    const filename = `${fileHash}${extension}`;
     const targetPath = join(uploadDir, filename);
     await fs.writeFile(targetPath, file.buffer);
 
@@ -88,6 +96,43 @@ export class DocumentsController {
       status: DocumentStatus.PENDING,
     });
 
-    return res.status(201).send(document);
+    await this.queueService.enqueueAnalyze(document.id);
+    return res.status(202).send(document);
+  }
+
+  @Post(':id/verify')
+  @UseGuards(JwtAuthGuard)
+  async verifyDocument(@Param('id') id: string, @Res() res: Response) {
+    const document = await this.documentsService.findById(id);
+    if (!document) {
+      throw new NotFoundException('Document not found');
+    }
+
+    if (document.status === DocumentStatus.VERIFIED) {
+      throw new ConflictException('Document has already been verified');
+    }
+
+    await this.queueService.enqueueAnchor(document.id);
+
+    return res.status(202).json({
+      message: 'Verification queued',
+      documentId: document.id,
+    });
+  }
+
+  @Get(':id/verification')
+  @UseGuards(JwtAuthGuard)
+  async getVerification(@Param('id') id: string) {
+    const document = await this.documentsService.findById(id);
+    if (!document) {
+      throw new NotFoundException('Document not found');
+    }
+
+    const record = await this.verificationService.findLatestByDocument(id);
+    if (!record) {
+      throw new NotFoundException('No verification record found for this document');
+    }
+
+    return record;
   }
 }

backend/src/documents/documents.controller.ts

@@ -1,12 +1,21 @@
-import { Module } from '@nestjs/common';
+import { forwardRef, Module } from '@nestjs/common';
 import { ConfigModule } from '@nestjs/config';
 import { TypeOrmModule } from '@nestjs/typeorm';
 import { DocumentsController } from './documents.controller';
 import { DocumentsService } from './documents.service';
 import { Document } from './entities/document.entity';
+import { StellarModule } from '../stellar/stellar.module';
+import { VerificationModule } from '../verification/verification.module';
+import { QueueModule } from '../queue/queue.module';
 
 @Module({
-  imports: [ConfigModule, TypeOrmModule.forFeature([Document])],
+  imports: [
+    ConfigModule,
+    TypeOrmModule.forFeature([Document]),
+    StellarModule,
+    VerificationModule,
+    forwardRef(() => QueueModule),
+  ],
   controllers: [DocumentsController],
   providers: [DocumentsService],
   exports: [DocumentsService],

backend/src/documents/documents.module.ts

@@ -0,0 +1,10 @@
+import { Module } from '@nestjs/common';
+import { ConfigModule } from '@nestjs/config';
+import { MailService } from './mail.service';
+
+@Module({
+  imports: [ConfigModule],
+  providers: [MailService],
+  exports: [MailService],
+})
+export class MailModule {}

backend/src/mail/mail.module.ts

@@ -0,0 +1,76 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import nodemailer, { SendMailOptions, Transporter } from 'nodemailer';
+
+@Injectable()
+export class MailService {
+  private readonly logger = new Logger(MailService.name);
+  private readonly transporter: Transporter | null;
+  private readonly from?: string;
+
+  constructor(private readonly configService: ConfigService) {
+    const host = this.configService.get<string>('MAIL_HOST');
+    const port = Number(this.configService.get<string>('MAIL_PORT'));
+    const user = this.configService.get<string>('MAIL_USER');
+    const pass = this.configService.get<string>('MAIL_PASSWORD');
+    this.from = this.configService.get<string>('MAIL_FROM');
+
+    if (!host || !port || !user || !pass || !this.from) {
+      this.logger.warn('SMTP configuration is incomplete; email delivery will be disabled');
+      this.transporter = null;
+      return;
+    }
+
+    this.transporter = nodemailer.createTransport({
+      host,
+      port,
+      secure: port === 465,
+      auth: { user, pass },
+    });
+  }
+
+  private async sendMail(options: SendMailOptions) {
+    if (!this.transporter) {
+      this.logger.warn(`Skipping email to ${options.to} because SMTP is not configured`);
+      return;
+    }
+
+    await this.transporter.sendMail({
+      from: this.from,
+      ...options,
+    });
+  }
+
+  async sendWelcome(to: string, name: string): Promise<void> {
+    await this.sendMail({
+      to,
+      subject: 'Welcome to Smalda',
+      html: `<p>Hi ${name},</p><p>Thank you for joining Smalda. We are excited to help you secure your land documents.</p>`,
+    });
+  }
+
+  async sendVerificationComplete(to: string, documentTitle: string, txHash: string): Promise<void> {
+    await this.sendMail({
+      to,
+      subject: 'Document Verification Complete',
+      html: `
+        <p>Your document <strong>${documentTitle}</strong> has been anchored on the Stellar network.</p>
+        <p>Transaction hash: <code>${txHash}</code></p>
+        <p>You can view the transaction via the Stellar Horizon explorer.</p>
+      `,
+    });
+  }
+
+  async sendRiskAlert(to: string, documentTitle: string, flags: string[]): Promise<void> {
+    const flagList = flags.map((flag) => `<li>${flag}</li>`).join('');
+    await this.sendMail({
+      to,
+      subject: 'Risk Alert: Document Needs Attention',
+      html: `
+        <p>The document <strong>${documentTitle}</strong> triggered the following risk flags:</p>
+        <ul>${flagList}</ul>
+        <p>Please review the document and supply any missing information.</p>
+      `,
+    });
+  }
+}

backend/src/mail/mail.service.ts

@@ -0,0 +1,70 @@
+import { Injectable, Logger, OnModuleDestroy } from '@nestjs/common';
+import { QueueScheduler, Worker } from 'bullmq';
+
+import { DocumentsService } from '../documents/documents.service';
+import { DocumentStatus } from '../documents/entities/document.entity';
+import { VerificationService } from '../verification/verification.service';
+import { VerificationStatus } from '../verification/entities/verification-record.entity';
+import { RiskAssessmentService } from '../risk-assessment/risk-assessment.service';
+import { StellarService } from '../stellar/stellar.service';
+import { QueueService } from './queue.service';
+
+@Injectable()
+export class DocumentProcessor implements OnModuleDestroy {
+  private readonly logger = new Logger(DocumentProcessor.name);
+  private readonly worker: Worker;
+  private readonly scheduler: QueueScheduler;
+
+  constructor(
+    private readonly queueService: QueueService,
+    private readonly riskService: RiskAssessmentService,
+    private readonly documentsService: DocumentsService,
+    private readonly stellarService: StellarService,
+    private readonly verificationService: VerificationService,
+  ) {
+    const connection = this.queueService.getConnectionOptions();
+    this.scheduler = new QueueScheduler(this.queueService.queueName, { connection });
+    this.worker = new Worker(
+      this.queueService.queueName,
+      async (job) => {
+        if (job.name === 'analyze') {
+          await this.riskService.assessDocument(job.data.documentId);
+          return;
+        }
+        if (job.name === 'anchor') {
+          await this.handleAnchor(job.data.documentId);
+        }
+      },
+      { connection },
+    );
+
+    this.worker.on('failed', (job, err) => {
+      this.logger.error(`Job ${job.id} (${job.name}) failed`, err?.message, err?.stack);
+    });
+  }
+
+  private async handleAnchor(documentId: string) {
+    const document = await this.documentsService.findById(documentId);
+    if (!document) {
+      this.logger.warn(`Document ${documentId} not found for anchor job`);
+      return;
+    }
+
+    const { txHash, ledger } = await this.stellarService.anchorHash(document.fileHash);
+    await this.verificationService.create({
+      documentId,
+      stellarTxHash: txHash,
+      stellarLedger: ledger,
+      anchoredAt: new Date(),
+      status: VerificationStatus.CONFIRMED,
+    });
+
+    await this.documentsService.updateStatus(documentId, DocumentStatus.VERIFIED);
+    this.logger.log(`Document ${documentId} verified on ledger ${ledger}`);
+  }
+
+  async onModuleDestroy(): Promise<void> {
+    await this.worker?.close();
+    await this.scheduler?.close();
+  }
+}