Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | import { Injectable, Logger } from '@nestjs/common'; import { InjectModel } from '@nestjs/mongoose'; import { Model } from 'mongoose'; import { ImportJob } from './schemas/import.schema'; import { ExportJob } from './schemas/export.schema'; import { StorageHubService } from '../storage/storage-hub.service'; export interface ImportEntityHandler { validateRow(row: any): Promise<{ isValid: boolean; error?: string }>; processRow(tenantId: string, row: any): Promise<void>; } @Injectable() export class ImportExportService { private readonly logger = new Logger(ImportExportService.name); private readonly handlers = new Map<string, ImportEntityHandler>(); constructor( @InjectModel(ImportJob.name) private readonly importJobModel: Model<ImportJob>, @InjectModel(ExportJob.name) private readonly exportJobModel: Model<ExportJob>, private readonly storageHub: StorageHubService, ) {} registerHandler(entityType: string, handler: ImportEntityHandler): void { this.handlers.set(entityType, handler); } async startImport( tenantId: string, entityType: string, fileBuffer: Buffer, filename: string, performedBy?: string, ): Promise<ImportJob> { const handler = this.handlers.get(entityType); if (!handler) { throw new Error(`No import handler registered for: ${entityType}`); } // Upload key registry const key = `tenants/${tenantId}/imports`; const storageObj = await this.storageHub.uploadFile(tenantId, { key, filename, mimeType: 'text/csv', buffer: fileBuffer, }); const job = await this.importJobModel.create({ tenantId, entityType, status: 'pending', filePath: storageObj.key, performedBy, }); // Execute parsing asynchronously this.executeImport(job, fileBuffer, handler); return job; } private async executeImport( job: any, buffer: Buffer, handler: ImportEntityHandler, ): Promise<void> { job.status = 'processing'; await job.save(); try { const rows = this.parseCsv(buffer.toString('utf8')); job.totalRows = rows.length; for (let i = 0; i < rows.length; i++) { const row = rows[i]; const validation = await handler.validateRow(row); if (validation.isValid) { await handler.processRow(job.tenantId, row); job.processedRows += 1; } else { job.failedRows += 1; job.validationErrors.push({ rowNumber: i + 1, fieldName: 'row', errorMessage: validation.error || 'Row validation failed', }); } await job.save(); } job.status = 'completed'; } catch (err: any) { job.status = 'failed'; this.logger.error(`Import execution failed: ${err.message}`); } finally { await job.save(); } } private parseCsv(content: string): any[] { const lines = content.split('\n').filter((l) => l.trim().length > 0); if (lines.length <= 1) return []; const headers = lines[0].split(',').map((h) => h.trim()); const result: any[] = []; for (let i = 1; i < lines.length; i++) { const currentline = lines[i].split(','); const obj: any = {}; for (let j = 0; j < headers.length; j++) { obj[headers[j]] = currentline[j] ? currentline[j].trim() : ''; } result.push(obj); } return result; } } |