Merge pull request #210 from Stephan-Thomas/feat/data-export-gdpr-165

feat(users): add data export flow, entity, service, processor wiring …

Author: RUKAYAT-CODER

BackEnd/src/config/export.config.ts

@@ -0,0 +1,7 @@
+export default () => ({
+  export: {
+    storageBaseUrl: process.env.EXPORT_STORAGE_BASE_URL || 'https://storage.example.com/exports',
+    allowedFormats: ['json', 'csv'],
+    downloadExpirySeconds: parseInt(process.env.EXPORT_DOWNLOAD_EXPIRY || '86400', 10),
+  },
+});

BackEnd/src/modules/jobs/job.types.ts

@@ -116,6 +116,7 @@ export interface DataExportPayload extends JobPayload {
   exportType: 'users' | 'payouts' | 'quests' | 'analytics';
   format: 'csv' | 'json' | 'xlsx';
   userId: string; // Who requested the export
+  exportId?: string; // optional DB record id to correlate
 }
 
 export interface ReportGeneratePayload extends JobPayload {

BackEnd/src/modules/jobs/jobs.module.ts

@@ -12,10 +12,11 @@ import { WebhookProcessor } from './processors/webhook.processor';
 import { AnalyticsProcessor } from './processors/analytics.processor';
 import { QuestProcessor } from './processors/quest.processor';
 import { JobLog, JobLogRetry, JobDependency, JobSchedule } from './entities/job-log.entity';
+import { DataExport } from '../users/entities/data-export.entity';
 
 @Module({
   imports: [
-    TypeOrmModule.forFeature([JobLog, JobLogRetry, JobDependency, JobSchedule]),
+    TypeOrmModule.forFeature([JobLog, JobLogRetry, JobDependency, JobSchedule, DataExport]),
   ],
   providers: [
     JobsService,

BackEnd/src/modules/jobs/jobs.service.ts

@@ -1,6 +1,7 @@
 import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
 import { Queue, Worker, Job } from 'bullmq';
 import { QUEUES, DEFAULT_JOB_OPTIONS } from './jobs.constants';
+import { DataExportProcessor } from './processors/export.processor';
 
 const redisConnection = () => {
   const url = process.env.REDIS_URL || 'redis://127.0.0.1:6379';
@@ -14,6 +15,8 @@ export class JobsService implements OnModuleInit, OnModuleDestroy {
   private workers: Worker[] = [];
   private emailProcessor: ((messageId: string, dto: any) => Promise<void>) | null = null;
 
+  constructor(private readonly dataExportProcessor?: DataExportProcessor) {}
+
   registerEmailProcessor(processor: (messageId: string, dto: any) => Promise<void>) {
     this.emailProcessor = processor;
   }
@@ -25,12 +28,17 @@ export class JobsService implements OnModuleInit, OnModuleDestroy {
     this.queues[QUEUES.SCHEDULED] = new Queue(QUEUES.SCHEDULED, redisConnection() as any);
     this.queues[QUEUES.DEAD_LETTER] = new Queue(QUEUES.DEAD_LETTER, redisConnection() as any);
     this.queues[QUEUES.EMAIL] = new Queue(QUEUES.EMAIL, redisConnection() as any);
+    this.queues[QUEUES.EXPORTS] = new Queue(QUEUES.EXPORTS, redisConnection() as any);
 
     this.createWorker(QUEUES.NOTIFICATIONS, this.handleNotification.bind(this));
     this.createWorker(QUEUES.ANALYTICS, this.handleAnalytics.bind(this));
     this.createWorker(QUEUES.CLEANUP, this.handleCleanup.bind(this));
     this.createWorker(QUEUES.SCHEDULED, this.handleScheduled.bind(this));
     this.createWorker(QUEUES.EMAIL, this.handleEmail.bind(this));
+    // register exports worker if processor available
+    if (this.dataExportProcessor && typeof this.dataExportProcessor.processExport === 'function') {
+      this.createWorker(QUEUES.EXPORTS, this.dataExportProcessor.processExport.bind(this.dataExportProcessor));
+    }
   }
 
   async onModuleDestroy() {

BackEnd/src/modules/jobs/processors/export.processor.ts

@@ -1,7 +1,10 @@
 import { Injectable, Logger } from '@nestjs/common';
 import { Job } from 'bullmq';
+import { InjectRepository } from '@nestjs/typeorm';
+import { Repository } from 'typeorm';
 import { DataExportPayload, ReportGeneratePayload, JobResult } from '../job.types';
 import { JobLogService } from './job-log.service';
+import { DataExport, DataExportStatus } from '../../users/entities/data-export.entity';
 
 /**
  * Data Export Processor
@@ -11,7 +14,11 @@ import { JobLogService } from './job-log.service';
 export class DataExportProcessor {
   private readonly logger = new Logger(DataExportProcessor.name);
 
-  constructor(private readonly jobLogService: JobLogService) {}
+  constructor(
+    private readonly jobLogService: JobLogService,
+    @InjectRepository(DataExport)
+    private readonly dataExportRepo: Repository<DataExport>,
+  ) {}
 
   /**
    * Process data export job
@@ -52,7 +59,7 @@ export class DataExportProcessor {
 
       // Simulate file generation
       const fileName = `export-${exportType}-${Date.now()}.${this.getFileExtension(format)}`;
-      const downloadUrl = `https://storage.example.com/exports/${fileName}`;
+      const downloadUrl = `${process.env.EXPORT_STORAGE_BASE_URL || 'https://storage.example.com/exports'}/${fileName}`;
 
       await job.updateProgress(75);
 
@@ -75,6 +82,21 @@ export class DataExportProcessor {
         duration: Date.now() - job.timestamp,
       };
 
+      // Update DB record if present
+      if (job.data && (job.data as any).exportId) {
+        try {
+          await this.dataExportRepo.update((job.data as any).exportId, {
+            status: DataExportStatus.COMPLETED,
+            fileName,
+            downloadUrl,
+            recordCount: result.data.recordCount,
+            exportedAt: result.data.exportedAt,
+          } as any);
+        } catch (err) {
+          this.logger.error('Failed to update data export record', err?.stack || err);
+        }
+      }
+
       this.logger.log(`Data export completed: ${fileName}`);
       return result;
     } catch (error) {
@@ -83,6 +105,17 @@ export class DataExportProcessor {
         error.stack,
       );
 
+      // mark DB record as failed if exportId provided
+      if ((job.data as any)?.exportId) {
+        try {
+          await this.dataExportRepo.update((job.data as any).exportId, {
+            status: DataExportStatus.FAILED,
+          } as any);
+        } catch (err) {
+          this.logger.error('Failed to mark export record failed', err?.stack || err);
+        }
+      }
+
       throw error;
     }
   }

BackEnd/src/modules/users/data-export.service.ts

@@ -0,0 +1,57 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+import { Repository } from 'typeorm';
+import { DataExport, DataExportStatus } from './entities/data-export.entity';
+import { JobsService } from '../jobs/jobs.service';
+import { QUEUES } from '../jobs/jobs.constants';
+
+@Injectable()
+export class DataExportService {
+  private readonly logger = new Logger(DataExportService.name);
+
+  constructor(
+    @InjectRepository(DataExport)
+    private readonly dataExportRepo: Repository<DataExport>,
+    private readonly jobsService: JobsService,
+  ) {}
+
+  async requestExport(userId: string, exportType: string, format: string) {
+    const exportRecord = this.dataExportRepo.create({
+      userId,
+      exportType,
+      format,
+      status: DataExportStatus.PENDING,
+    });
+
+    const saved = await this.dataExportRepo.save(exportRecord);
+
+    try {
+      await this.jobsService.addJob(QUEUES.EXPORTS, {
+        organizationId: null,
+        exportType,
+        format,
+        userId,
+        exportId: saved.id,
+      });
+      this.logger.log(`Queued export job for user ${userId} id=${saved.id}`);
+    } catch (err) {
+      this.logger.error('Failed to enqueue export job', err?.stack || err);
+      saved.status = DataExportStatus.FAILED;
+      await this.dataExportRepo.save(saved);
+    }
+
+    return saved;
+  }
+
+  async markProcessing(id: string) {
+    await this.dataExportRepo.update(id, { status: DataExportStatus.PROCESSING });
+  }
+
+  async markCompleted(id: string, payload: Partial<DataExport>) {
+    await this.dataExportRepo.update(id, { status: DataExportStatus.COMPLETED, ...payload });
+  }
+
+  async markFailed(id: string, error?: string) {
+    await this.dataExportRepo.update(id, { status: DataExportStatus.FAILED });
+  }
+}

BackEnd/src/modules/users/entities/data-export.entity.ts

@@ -0,0 +1,44 @@
+import { Entity, PrimaryGeneratedColumn, Column, CreateDateColumn, UpdateDateColumn } from 'typeorm';
+
+export enum DataExportStatus {
+  PENDING = 'pending',
+  PROCESSING = 'processing',
+  COMPLETED = 'completed',
+  FAILED = 'failed',
+}
+
+@Entity('data_exports')
+export class DataExport {
+  @PrimaryGeneratedColumn('uuid')
+  id: string;
+
+  @Column({ nullable: false })
+  userId: string;
+
+  @Column({ default: 'users' })
+  exportType: string;
+
+  @Column({ default: 'json' })
+  format: string;
+
+  @Column({ type: 'enum', enum: DataExportStatus, default: DataExportStatus.PENDING })
+  status: DataExportStatus;
+
+  @Column({ nullable: true })
+  fileName?: string;
+
+  @Column({ nullable: true })
+  downloadUrl?: string;
+
+  @Column({ type: 'int', nullable: true })
+  recordCount?: number;
+
+  @Column({ type: 'timestamptz', nullable: true })
+  exportedAt?: Date;
+
+  @CreateDateColumn()
+  createdAt: Date;
+
+  @UpdateDateColumn()
+  updatedAt: Date;
+}

BackEnd/src/modules/users/user.controller.ts

@@ -1,6 +1,7 @@
 import {
   Controller,
   Get,
+  Post,
   Patch,
   Delete,
   Param,
@@ -30,11 +31,15 @@ import { UsersService } from './user.service';
 import { User } from './entities/user.entity';
 import { Role } from '../../common/enums/role.enum';
 import { Roles } from '../auth/decorators/roles.decorator';
+import { DataExportService } from './data-export.service';
 
 @ApiTags('users')
 @Controller('users')
 export class UsersController {
-  constructor(private readonly usersService: UsersService) {}
+  constructor(
+    private readonly usersService: UsersService,
+    private readonly dataExportService: DataExportService,
+  ) {}
 
   @Get('search')
   @ApiOperation({ summary: 'Search users by username or address' })
@@ -150,4 +155,27 @@ export class UsersController {
   ) {
     await this.usersService.deleteUser(address, requestingUser);
   }
+
+  @Post('export')
+  @UseGuards(JwtAuthGuard)
+  @ApiBearerAuth()
+  @ApiOperation({ summary: 'Request data export (async)' })
+  @ApiResponse({ status: 202, description: 'Export queued' })
+  async requestExport(
+    @CurrentUser() user: User,
+    @Body('format') format: string = 'json',
+    @Body('exportType') exportType: string = 'users',
+  ) {
+    if (!user || !user.id) {
+      throw new BadRequestException('Invalid user');
+    }
+
+    const record = await this.dataExportService.requestExport(user.id, exportType, format);
+
+    return {
+      id: record.id,
+      status: record.status,
+      message: 'Export request queued. You will be notified when ready.',
+    };
+  }
 }

BackEnd/src/modules/users/users.module.ts

@@ -8,14 +8,18 @@ import { Submission } from '../submissions/entities/submission.entity';
 import { Payout } from '../payouts/entities/payout.entity';
 import { CacheModule } from '../cache/cache.module';
 import { UserExperienceListener } from './events/user-experience.listener';
+import { DataExport } from './entities/data-export.entity';
+import { DataExportService } from './data-export.service';
+import { JobsModule } from '../jobs/jobs.module';
 
 @Module({
   imports: [
-    TypeOrmModule.forFeature([User, Quest, Submission, Payout]),
+    TypeOrmModule.forFeature([User, Quest, Submission, Payout, DataExport]),
     CacheModule,
+    JobsModule,
   ],
   controllers: [UsersController],
-  providers: [UsersService, UserExperienceListener],
-  exports: [UsersService],
+  providers: [UsersService, UserExperienceListener, DataExportService],
+  exports: [UsersService, DataExportService],
 })
 export class UsersModule {}