cferraz/adaptador-planilhas
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Inicio do Projeto Piloto
ci / verify (push) Has been cancelled
Details

Browse Source
This commit is contained in:
camilafagundes committed 2026-10-02 23:59:36 -03:00
1 parent 6e23f1533b
commit 419325db12
52 files changed
+1329 -110
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/server/spreadsheet-adapter/process-job.ts
+72 -30
View File
@@ -1,4 +1,5 @@
import { AdapterError, isAdapterError } from "@/domain/spreadsheet-adapter/errors";
import { shouldRequeue } from "@/domain/spreadsheet-adapter/failure-policy";
import { neutralizeRow, neutralizeSpreadsheetFormula } from "@/domain/spreadsheet-adapter/csv-injection";
import { buildOutputFilename } from "@/domain/spreadsheet-adapter/filenames";
import { validateMappings } from "@/domain/spreadsheet-adapter/mapping";
@@ -22,6 +23,12 @@ function assertUse(ctx: RequestContext): void {
}
}
function assertOwnedTempKey(workspaceId: string, key: string): void {
if (!key.startsWith(`workspaces/${workspaceId}/tmp/`) || !key.includes("/tmp/") || key.includes("/models/") || key.includes("..")) {
throw new AdapterError("PATH_TRAVERSAL", "Chave temporária rejeitada.");
}
}
async function audit(
query: Query,
ctx: RequestContext,
@@ -37,15 +44,18 @@ async function audit(
);
}
function assertTempKey(key: string): void {
if (!key.includes("/tmp/") || key.includes("/models/") || key.includes("..")) {
throw new AdapterError("PATH_TRAVERSAL", "Chave temporária rejeitada.");
}
function processTimeoutMs(): number {
const raw = process.env.SPREADSHEET_JOB_TIMEOUT_MS;
if (!raw) return 120_000;
const value = Number(raw);
return Number.isFinite(value) && value >= 0 ? value : 120_000;
}
export async function runProcess(deps: AppDeps, ctx: RequestContext, jobId: string, sourceId: string): Promise<void> {
const userLock = lockKey(ctx.userId);
const written: string[] = [];
const startedAt = Date.now();
const timeoutMs = processTimeoutMs();
try {
const owns = await deps.coordinator.ownsLock(userLock, jobId);
if (!owns) {
@@ -120,8 +130,14 @@ export async function runProcess(deps: AppDeps, ctx: RequestContext, jobId: stri
).map(neutralizeRow);
const headers = columns.map((column) => neutralizeSpreadsheetFormula(column.name));
let done = 0;
if (Date.now() - startedAt >= timeoutMs) {
throw new AdapterError("TIMEOUT", "O processamento excedeu o tempo limite.", 504);
}
for (const _row of transformed) {
done += 1;
if (Date.now() - startedAt >= timeoutMs) {
throw new AdapterError("TIMEOUT", "O processamento excedeu o tempo limite.", 504);
}
if (done % 5000 === 0) {
await progress({
phase: "processing",
@@ -136,8 +152,8 @@ export async function runProcess(deps: AppDeps, ctx: RequestContext, jobId: stri
await progress({ phase: "generating", label: "Gerando arquivo", percent: null, sourceIndex, sourceCount: total, sourceId });
const csvKey = `workspaces/${ctx.workspaceId}/tmp/${jobId}/${sourceId}/out.csv`;
const xlsxKey = `workspaces/${ctx.workspaceId}/tmp/${jobId}/${sourceId}/out.xlsx`;
assertTempKey(csvKey);
assertTempKey(xlsxKey);
assertOwnedTempKey(ctx.workspaceId, csvKey);
assertOwnedTempKey(ctx.workspaceId, xlsxKey);
const csv = buildCsv(headers, transformed);
const xlsx = await buildXlsx(headers, transformed);
await deps.storage.put(csvKey, csv);
@@ -180,28 +196,38 @@ export async function runProcess(deps: AppDeps, ctx: RequestContext, jobId: stri
} catch (error) {
for (const key of written) await deps.storage.delete(key).catch(() => undefined);
const code = isAdapterError(error) ? error.code : "PROCESS_FAILED";
const failed: JobProgress = {
phase: "failed",
label: "Falhou",
percent: null,
sourceIndex: 1,
sourceCount: 1,
sourceId,
};
const httpStatus = isAdapterError(error) ? error.status : 500;
let requeue = false;
await deps.db
.withUser(ctx.userId, async (query) => {
await query.query(`update public.spreadsheet_job_sources set status = 'failed', error_code = $2 where id = $1 and status <> 'completed'`, [
const current = await query.query<{ progress: unknown }>(`select progress from public.spreadsheet_jobs where id = $1`, [jobId]);
const prev = asJson<JobProgress | null>(current[0]?.progress, null);
const attempt = (prev?.attempt ?? 0) + 1;
requeue = shouldRequeue(code, httpStatus, attempt);
const next: JobProgress = {
phase: requeue ? "validating" : "failed",
label: requeue ? "Nova tentativa" : "Falhou",
percent: null,
sourceIndex: 1,
sourceCount: 1,
sourceId,
code,
]);
attempt,
};
await query.query(
`update public.spreadsheet_jobs set status = 'failed', error_code = $2, progress = $3::jsonb, finished_at = now(), updated_at = now() where id = $1`,
[jobId, code, JSON.stringify(failed)],
`update public.spreadsheet_job_sources set status = $2, error_code = $3 where id = $1 and status <> 'completed'`,
[sourceId, requeue ? "mapped" : "failed", code],
);
await audit(query, ctx, "job.failed", jobId, "error", { error_code: code, source_id: sourceId });
await query.query(
`update public.spreadsheet_jobs
set status = $2, error_code = $3, progress = $4::jsonb,
finished_at = case when $2 = 'failed' then now() else finished_at end, updated_at = now()
where id = $1`,
[jobId, requeue ? "queued" : "failed", code, JSON.stringify(next)],
);
await audit(query, ctx, "job.failed", jobId, "error", { error_code: code, source_id: sourceId, attempt });
})
.catch(() => undefined);
await deps.coordinator.setProgress(jobId, failed).catch(() => undefined);
if (requeue) await deps.coordinator.enqueue(jobId).catch(() => undefined);
if (!isAdapterError(error) || error.status >= 500) {
deps.capture?.(error, { module: "spreadsheet-adapter", job_id: jobId, workspace_id: ctx.workspaceId, error_code: code });
}
@@ -220,6 +246,16 @@ const CONTENT_TYPE: Record<OutputFormat, string> = {
export async function downloadOutput(deps: AppDeps, ctx: RequestContext, jobId: string, sourceId: string, format: OutputFormat) {
assertUse(ctx);
const flags = await deps.db.withUser(ctx.userId, (query) =>
query.query<{ enabled: boolean }>(
`select enabled from public.workspace_module_flags where workspace_id = $1 and module_key = 'spreadsheet_adapter'`,
[ctx.workspaceId],
),
);
const flag = flags[0]?.enabled as unknown;
if (flag !== true && flag !== "t" && flag !== "true") {
throw new AdapterError("FEATURE_DISABLED", "O Adaptador não está habilitado neste workspace.", 403);
}
const file = await deps.db.withUser(ctx.userId, async (query) => {
const rows = await query.query<{
model_name: string;
@@ -242,7 +278,7 @@ export async function downloadOutput(deps: AppDeps, ctx: RequestContext, jobId:
const filename = buildOutputFilename(file.model_name, format, deps.now?.() ?? new Date());
if (format === "pdf") {
if (!file.output_csv_key) throw new AdapterError("NOT_FOUND", "Saída indisponível.", 404);
assertTempKey(file.output_csv_key);
assertOwnedTempKey(ctx.workspaceId, file.output_csv_key);
const csv = (await deps.storage.get(file.output_csv_key)).toString("utf8");
const grid = parseCsvText(csv);
const headers = grid[0] ?? [];
@@ -251,29 +287,35 @@ export async function downloadOutput(deps: AppDeps, ctx: RequestContext, jobId:
}
const key = format === "csv" ? file.output_csv_key : file.output_xlsx_key;
if (!key) throw new AdapterError("NOT_FOUND", "Saída indisponível.", 404);
assertTempKey(key);
assertOwnedTempKey(ctx.workspaceId, key);
return { filename, contentType: CONTENT_TYPE[format], body: await deps.storage.get(key) };
}
export async function cleanupExpired(deps: AppDeps): Promise<number> {
const rows = await deps.db.withService((query) =>
query.query<{ id: string; temp_storage_key: string; output_csv_key: string | null; output_xlsx_key: string | null }>(
`select id, temp_storage_key, output_csv_key, output_xlsx_key
from public.spreadsheet_job_sources
where expires_at < $1::timestamptz and status <> 'expired'`,
query.query<{ id: string; workspace_id: string; temp_storage_key: string; output_csv_key: string | null; output_xlsx_key: string | null }>(
`select s.id, j.workspace_id, s.temp_storage_key, s.output_csv_key, s.output_xlsx_key
from public.spreadsheet_job_sources s
join public.spreadsheet_jobs j on j.id = s.job_id
where s.expires_at < $1::timestamptz and s.status <> 'expired'`,
[(deps.now ?? (() => new Date()))().toISOString()],
),
);
let removed = 0;
for (const row of rows) {
const prefix = `workspaces/${row.workspace_id}/tmp/`;
for (const key of [row.temp_storage_key, row.output_csv_key, row.output_xlsx_key]) {
if (!key || !key.includes("/tmp/") || key.includes("/models/")) continue;
await deps.storage.delete(key).catch(() => undefined);
if (!key || !key.startsWith(prefix) || key.includes("..") || key.includes("/models/")) continue;
await deps.storage.delete(key).catch((error) => {
deps.capture?.(error, { module: "spreadsheet-adapter", workspace_id: row.workspace_id, error_code: "CLEANUP_FAILED" });
});
}
await deps.db.withService((query) =>
query.query(`update public.spreadsheet_job_sources set status = 'expired', output_csv_key = null, output_xlsx_key = null where id = $1`, [row.id]),
);
removed += 1;
}
return rows.length;
return removed;
}
export async function recoverStaleJobs(deps: AppDeps): Promise<number> {