Feat: Implemented bulk payments - csv mass disbursement for merchants

backend/package.json

@@ -54,6 +54,7 @@
     "bullmq": "^5.71.1",
     "class-transformer": "^0.5.1",
     "class-validator": "^0.15.1",
+    "csv-parser": "^3.2.0",
     "dotenv": "^17.3.1",
     "imurmurhash": "^0.1.4",
     "ioredis": "^5.10.1",
@@ -85,6 +86,7 @@
     "@types/express": "^5.0.0",
     "@types/imurmurhash": "^0.1.4",
     "@types/jest": "^30.0.0",
+    "@types/multer": "^2.1.0",
     "@types/node": "^22.10.7",
     "@types/passport-jwt": "^4.0.1",
     "@types/qrcode": "^1.5.5",

backend/src/admin/admin.controller.ts

@@ -25,6 +25,8 @@ import { Request } from 'express';
 import { AuditInterceptor, Audit } from '../audit/audit.interceptor';
 import { ReferralAnalyticsService } from '../referrals/referral-analytics.service';
 import { FunnelStatsDto, TopReferrersDto, CohortComparisonDto, RewardSpendDto, UserReferralStatsDto } from '../referrals/dto/referral-analytics.dto';
+import { OffRampService } from '../offramp/offramp.service';
+import { AdminOffRampQueryDto, OffRampResponseDto } from '../offramp/dto/offramp.dto';
 
 @ApiTags('admin')
 @ApiBearerAuth()
@@ -36,6 +38,7 @@ export class AdminController {
     private readonly adminService: AdminService,
     private readonly receiptService: ReceiptService,
     private readonly referralAnalyticsService: ReferralAnalyticsService,
+    private readonly offRampService: OffRampService,
   ) {}
 
   @Get('users')
@@ -120,4 +123,28 @@ export class AdminController {
   async getTransactionReceipt(@Param('id') id: string) {
     return this.receiptService.generateTransactionReceiptAdmin(id);
   }
+
+  // ── Off-Ramp Admin ──────────────────────────────────────────────────────────
+
+  @Get('offramps')
+  @Roles(AdminRole.ADMIN, AdminRole.SUPERADMIN)
+  @ApiOperation({ summary: 'List all off-ramp orders with optional filters' })
+  async listOffRamps(@Query() query: AdminOffRampQueryDto) {
+    return this.offRampService.adminList(query);
+  }
+
+  @Get('offramps/:id')
+  @Roles(AdminRole.ADMIN, AdminRole.SUPERADMIN)
+  @ApiOperation({ summary: 'Get a single off-ramp order by ID' })
+  async getOffRamp(@Param('id') id: string): Promise<OffRampResponseDto> {
+    return this.offRampService.adminGetById(id);
+  }
+
+  @Post('offramps/:id/refund')
+  @Roles(AdminRole.SUPERADMIN)
+  @Audit({ action: 'offramp.refund', resourceType: 'off_ramp', resourceIdParam: 'id' })
+  @ApiOperation({ summary: 'Manually trigger USDC refund for a failed off-ramp order' })
+  async refundOffRamp(@Param('id') id: string): Promise<OffRampResponseDto> {
+    return this.offRampService.adminRefund(id);
+  }
 }

backend/src/admin/admin.module.ts

@@ -16,9 +16,8 @@ import { CronModule } from '../cron/cron.module';
 import { CronAdminController } from './cron-admin.controller';
 import { AuditModule } from '../audit/audit.module';
 import { ReferralsModule } from '../referrals/referrals.module';
-
-import { AnalyticsModule } from './analytics/analytics.module';
 import { ReceiptModule } from '../receipt/receipt.module';
+import { OffRampModule } from '../offramp/offramp.module';
 
 @Module({
   imports: [
@@ -38,6 +37,7 @@ import { ReceiptModule } from '../receipt/receipt.module';
     CronModule,
     ReceiptModule,
     ReferralsModule,
+    OffRampModule,
   ],
   providers: [AdminService],
   controllers: [AdminController, CronAdminController],

backend/src/database/migrations/1700000000000-BulkDisbursement.ts

@@ -0,0 +1,57 @@
+import { MigrationInterface, QueryRunner } from "typeorm";
+
+export class BulkDisbursement1700000000000 implements MigrationInterface {
+    name = 'BulkDisbursement1700000000000'
+
+    public async up(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(`
+            CREATE TYPE "bulk_disbursement_status_enum" AS ENUM('pending', 'processing', 'completed', 'failed');
+        `);
+        await queryRunner.query(`
+            CREATE TABLE "bulk_disbursements" (
+                "id" uuid NOT NULL DEFAULT uuid_generate_v4(),
+                "created_at" TIMESTAMP NOT NULL DEFAULT now(),
+                "updated_at" TIMESTAMP NOT NULL DEFAULT now(),
+                "user_id" uuid NOT NULL,
+                "file_name" character varying(255) NOT NULL,
+                "reference" character varying(100) NOT NULL,
+                "total_items" integer NOT NULL DEFAULT 0,
+                "processed_items" integer NOT NULL DEFAULT 0,
+                "failed_items" integer NOT NULL DEFAULT 0,
+                "total_amount_usdc" numeric(24,8) NOT NULL DEFAULT 0,
+                "status" "bulk_disbursement_status_enum" NOT NULL DEFAULT 'pending',
+                "failure_reason" text,
+                CONSTRAINT "UQ_bulk_disbursements_reference" UNIQUE ("reference"),
+                CONSTRAINT "PK_bulk_disbursements_id" PRIMARY KEY ("id")
+            )
+        `);
+        await queryRunner.query(`
+            CREATE INDEX "IDX_bulk_disbursements_user_id" ON "bulk_disbursements" ("user_id")
+        `);
+        await queryRunner.query(`
+            CREATE INDEX "IDX_bulk_disbursements_user_id_created_at" ON "bulk_disbursements" ("user_id", "created_at")
+        `);
+
+        // Modify off_ramps table
+        await queryRunner.query(`
+            ALTER TABLE "off_ramps" ALTER COLUMN "bank_account_id" DROP NOT NULL;
+        `);
+        await queryRunner.query(`
+            ALTER TABLE "off_ramps" ADD "bulk_disbursement_id" uuid;
+        `);
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(`
+            ALTER TABLE "off_ramps" DROP COLUMN "bulk_disbursement_id";
+        `);
+        await queryRunner.query(`
+            ALTER TABLE "off_ramps" ALTER COLUMN "bank_account_id" SET NOT NULL;
+        `);
+
+        await queryRunner.query(`DROP INDEX "IDX_bulk_disbursements_user_id_created_at"`);
+        await queryRunner.query(`DROP INDEX "IDX_bulk_disbursements_user_id"`);
+        await queryRunner.query(`DROP TABLE "bulk_disbursements"`);
+        await queryRunner.query(`DROP TYPE "bulk_disbursement_status_enum"`);
+    }
+}

backend/src/offramp/bulk-disbursement.processor.ts

@@ -0,0 +1,111 @@
+import { Process, Processor } from '@nestjs/bull';
+import { Job } from 'bullmq';
+import { Logger } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+import { Repository } from 'typeorm';
+import * as fs from 'fs';
+import csv = require('csv-parser');
+import { BulkDisbursement, BulkDisbursementStatus } from './entities/bulk-disbursement.entity';
+import { OffRampService } from './offramp.service';
+
+@Processor('offramp-jobs')
+export class BulkDisbursementProcessor {
+  private readonly logger = new Logger(BulkDisbursementProcessor.name);
+
+  constructor(
+    @InjectRepository(BulkDisbursement)
+    private readonly bulkDisbursementRepo: Repository<BulkDisbursement>,
+    private readonly offRampService: OffRampService,
+  ) {}
+
+  @Process('bulk-disbursement')
+  async handleBulkDisbursement(job: Job<{ bulkDisbursementId: string; filePath: string; userId: string }>) {
+    const { bulkDisbursementId, filePath, userId } = job.data;
+
+    this.logger.log(`Starting processing for bulk disbursement: ${bulkDisbursementId}`);
+
+    const bulkRecord = await this.bulkDisbursementRepo.findOne({ where: { id: bulkDisbursementId } });
+    if (!bulkRecord) {
+      this.logger.error(`BulkDisbursement ${bulkDisbursementId} not found`);
+      return;
+    }
+
+    await this.bulkDisbursementRepo.update(bulkDisbursementId, { status: BulkDisbursementStatus.PROCESSING });
+
+    const rows: any[] = [];
+
+    try {
+      await new Promise<void>((resolve, reject) => {
+        fs.createReadStream(filePath)
+          .pipe(csv())
+          .on('data', (data) => rows.push(data))
+          .on('end', () => resolve())
+          .on('error', (err) => reject(err));
+      });
+    } catch (err: any) {
+      await this.bulkDisbursementRepo.update(bulkDisbursementId, {
+        status: BulkDisbursementStatus.FAILED,
+        failureReason: `Failed to parse CSV: ${err.message}`,
+      });
+      return;
+    }
+
+    await this.bulkDisbursementRepo.update(bulkDisbursementId, { totalItems: rows.length });
+
+    let processed = 0;
+    let failed = 0;
+    let totalUsdc = 0;
+
+    for (const row of rows) {
+      try {
+        const amountUsdc = parseFloat(row.amountUsdc || row.amount);
+        const bankCode = row.bankCode;
+        const accountNumber = row.accountNumber;
+        const accountName = row.accountName || 'Unknown Configured';
+
+        if (isNaN(amountUsdc) || !bankCode || !accountNumber) {
+          throw new Error('Invalid row format (missing amount, bankCode or accountNumber)');
+        }
+
+        await this.offRampService.executeBulkItem(userId, {
+          amountUsdc,
+          bankCode,
+          accountNumber,
+          accountName,
+          bulkDisbursementId,
+        });
+
+        processed++;
+        totalUsdc += amountUsdc;
+      } catch (err: any) {
+        this.logger.error(`Row offramp failed: ${err.message}`);
+        failed++;
+      }
+
+      // Periodically update progression
+      if ((processed + failed) % 10 === 0) {
+        await this.bulkDisbursementRepo.update(bulkDisbursementId, {
+          processedItems: processed,
+          failedItems: failed,
+          totalAmountUsdc: totalUsdc.toFixed(8)
+        });
+      }
+    }
+
+    await this.bulkDisbursementRepo.update(bulkDisbursementId, {
+      processedItems: processed,
+      failedItems: failed,
+      totalAmountUsdc: totalUsdc.toFixed(8),
+      status: failed === rows.length ? BulkDisbursementStatus.FAILED : BulkDisbursementStatus.COMPLETED,
+    });
+
+    // Cleanup CSV
+    try {
+      await fs.promises.unlink(filePath);
+    } catch (e) {
+      this.logger.warn(`Failed to unlink temp file ${filePath}`);
+    }
+
+    this.logger.log(`Finished bulk disbursement ${bulkDisbursementId}: ${processed} success, ${failed} failed`);
+  }
+}

backend/src/offramp/dto/bulk-disbursement.dto.ts

@@ -0,0 +1,32 @@
+import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
+import { BulkDisbursement, BulkDisbursementStatus } from '../entities/bulk-disbursement.entity';
+
+export class BulkDisbursementResponseDto {
+  @ApiProperty() id!: string;
+  @ApiProperty() reference!: string;
+  @ApiProperty() fileName!: string;
+  @ApiProperty() totalItems!: number;
+  @ApiProperty() processedItems!: number;
+  @ApiProperty() failedItems!: number;
+  @ApiProperty() totalAmountUsdc!: string;
+  @ApiProperty({ enum: BulkDisbursementStatus }) status!: BulkDisbursementStatus;
+  @ApiPropertyOptional() failureReason!: string | null;
+  @ApiProperty() createdAt!: Date;
+  @ApiProperty() updatedAt!: Date;
+
+  static from(o: BulkDisbursement): BulkDisbursementResponseDto {
+    const dto = new BulkDisbursementResponseDto();
+    dto.id = o.id;
+    dto.reference = o.reference;
+    dto.fileName = o.fileName;
+    dto.totalItems = o.totalItems;
+    dto.processedItems = o.processedItems;
+    dto.failedItems = o.failedItems;
+    dto.totalAmountUsdc = o.totalAmountUsdc;
+    dto.status = o.status;
+    dto.failureReason = o.failureReason;
+    dto.createdAt = o.createdAt;
+    dto.updatedAt = o.updatedAt;
+    return dto;
+  }
+}