Parse large Excel imports asynchronously to avoid gateway 502 timeouts.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
2026-10-03 18:14:17 +03:30
co-authored by Cursor
parent d3fb92a5e2
commit ac57b8ac33
5 changed files with 270 additions and 60 deletions
+166 -15
View File
@@ -1,4 +1,7 @@
import { randomBytes } from 'node:crypto';
import * as fs from 'node:fs';
import * as os from 'node:os';
import * as path from 'node:path';
import { BadRequestException } from '@nestjs/common';
export const EXCEL_IMPORT_CHUNK_SIZE = 150;
@@ -6,6 +9,7 @@ export const EXCEL_IMPORT_CHUNK_SIZE = 150;
export const EXCEL_IMPORT_LARGE_THRESHOLD = 80;
const SESSION_TTL_MS = 45 * 60 * 1000;
const MAX_SESSIONS = 40;
const MAX_ASYNC_ROWS = 100_000;
export type ExcelImportLookupCaches = {
user: Map<string, string | null>;
@@ -14,19 +18,46 @@ export type ExcelImportLookupCaches = {
deal: Map<string, string | null>;
};
export type ExcelImportSessionStatus = {
sessionId: string;
status: 'parsing' | 'ready' | 'error';
total: number;
parsed: number;
error?: string;
hasMore?: boolean;
nextOffset?: number;
};
type ExcelImportSession = {
organizationId: string;
rows: Array<Record<string, string>>;
expiresAt: number;
caches: ExcelImportLookupCaches;
status: 'parsing' | 'ready' | 'error';
error?: string;
total: number;
parsed: number;
};
const sessions = new Map<string, ExcelImportSession>();
const TMP_DIR = path.join(os.tmpdir(), 'mento-excel-import');
function emptyCaches(): ExcelImportLookupCaches {
return {
user: new Map(),
person: new Map(),
company: new Map(),
deal: new Map(),
};
}
function purgeExpired() {
const now = Date.now();
for (const [id, session] of sessions) {
if (session.expiresAt <= now) sessions.delete(id);
if (session.expiresAt <= now) {
sessions.delete(id);
cleanupSessionFiles(id);
}
}
if (sessions.size <= MAX_SESSIONS) return;
const ordered = [...sessions.entries()].sort(
@@ -35,9 +66,24 @@ function purgeExpired() {
while (sessions.size > MAX_SESSIONS && ordered.length) {
const [id] = ordered.shift()!;
sessions.delete(id);
cleanupSessionFiles(id);
}
}
function cleanupSessionFiles(sessionId: string) {
for (const ext of ['.xlsx', '.jsonl']) {
try {
fs.unlinkSync(path.join(TMP_DIR, `${sessionId}${ext}`));
} catch {
/* ignore */
}
}
}
function touch(session: ExcelImportSession) {
session.expiresAt = Date.now() + SESSION_TTL_MS;
}
export function createExcelImportSession(
organizationId: string,
rows: Array<Record<string, string>>,
@@ -48,16 +94,115 @@ export function createExcelImportSession(
organizationId,
rows,
expiresAt: Date.now() + SESSION_TTL_MS,
caches: {
user: new Map(),
person: new Map(),
company: new Map(),
deal: new Map(),
},
caches: emptyCaches(),
status: 'ready',
total: rows.length,
parsed: rows.length,
});
return id;
}
/**
* آپلود را فوری قبول می‌کند و پارس را در پس‌زمینه انجام می‌دهد
* تا پروکسی nginx با ۵۰۲ قطع نکند.
*/
export function beginExcelImportPrepare(
organizationId: string,
buffer: Buffer,
loadRows: (file: Express.Multer.File) => Promise<Array<Record<string, string>>>,
): string {
purgeExpired();
fs.mkdirSync(TMP_DIR, { recursive: true });
const id = randomBytes(16).toString('hex');
const xlsxPath = path.join(TMP_DIR, `${id}.xlsx`);
fs.writeFileSync(xlsxPath, buffer);
sessions.set(id, {
organizationId,
rows: [],
expiresAt: Date.now() + SESSION_TTL_MS,
caches: emptyCaches(),
status: 'parsing',
total: 0,
parsed: 0,
});
const fakeFile = {
fieldname: 'file',
originalname: 'import.xlsx',
encoding: '7bit',
mimetype:
'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
size: buffer.length,
buffer,
destination: '',
filename: '',
path: xlsxPath,
} as Express.Multer.File;
void (async () => {
const session = sessions.get(id);
if (!session) return;
try {
const rows = await loadRows(fakeFile);
const current = sessions.get(id);
if (!current) return;
if (rows.length > MAX_ASYNC_ROWS) {
current.status = 'error';
current.error = `فایل بیش از ${MAX_ASYNC_ROWS} ردیف دارد. فایل را تقسیم کنید.`;
current.total = rows.length;
current.parsed = rows.length;
touch(current);
cleanupSessionFiles(id);
return;
}
current.rows = rows;
current.total = rows.length;
current.parsed = rows.length;
current.status = 'ready';
touch(current);
try {
fs.unlinkSync(xlsxPath);
} catch {
/* ignore */
}
} catch (e) {
const current = sessions.get(id);
if (!current) return;
current.status = 'error';
current.error =
e instanceof Error ? e.message : 'خواندن فایل اکسل ناموفق بود';
touch(current);
cleanupSessionFiles(id);
}
})();
return id;
}
export function getExcelImportPrepareStatus(
sessionId: string,
organizationId: string,
): ExcelImportSessionStatus {
purgeExpired();
const session = sessions.get(sessionId);
if (!session || session.organizationId !== organizationId) {
throw new BadRequestException(
'جلسه ورود اکسل منقضی یا نامعتبر است. فایل را دوباره انتخاب کنید.',
);
}
touch(session);
return {
sessionId,
status: session.status,
total: session.total,
parsed: session.parsed,
error: session.error,
hasMore: session.status === 'ready' ? session.total > 0 : true,
nextOffset: 0,
};
}
export function getExcelImportSession(
sessionId: string,
organizationId: string,
@@ -69,13 +214,24 @@ export function getExcelImportSession(
'جلسه ورود اکسل منقضی یا نامعتبر است. فایل را دوباره انتخاب کنید.',
);
}
session.expiresAt = Date.now() + SESSION_TTL_MS;
if (session.status === 'parsing') {
throw new BadRequestException(
'فایل هنوز در حال آماده‌سازی است. چند لحظه صبر کنید.',
);
}
if (session.status === 'error') {
throw new BadRequestException(
session.error || 'آماده‌سازی فایل اکسل ناموفق بود.',
);
}
touch(session);
return session.rows;
}
export function destroyExcelImportSession(sessionId: string | undefined) {
if (!sessionId) return;
sessions.delete(sessionId);
cleanupSessionFiles(sessionId);
}
export function getExcelImportSessionCaches(
@@ -85,14 +241,9 @@ export function getExcelImportSessionCaches(
purgeExpired();
const session = sessions.get(sessionId);
if (!session || session.organizationId !== organizationId) {
return {
user: new Map(),
person: new Map(),
company: new Map(),
deal: new Map(),
};
return emptyCaches();
}
session.expiresAt = Date.now() + SESSION_TTL_MS;
touch(session);
return session.caches;
}