353 lines
16 KiB
TypeScript
353 lines
16 KiB
TypeScript
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";
|
|
import { canUseSpreadsheetAdapter } from "@/domain/spreadsheet-adapter/permissions";
|
|
import { transformRows } from "@/domain/spreadsheet-adapter/transform";
|
|
import type { JobProgress, OutputFormat, RequestContext } from "@/domain/spreadsheet-adapter/types";
|
|
import { buildCsv, buildXlsx } from "@/infra/export/tabular";
|
|
import type { Query } from "@/infra/db/types";
|
|
import { lockKey } from "@/infra/redis/redis";
|
|
import { sanitizeAuditMetadata } from "@/infra/observability/scrub";
|
|
import { parseCsvText } from "@/domain/spreadsheet-adapter/csv";
|
|
import { readSpreadsheet } from "@/infra/spreadsheet/parse";
|
|
import { asJson, mapDbError, sha256 } from "./db-utils";
|
|
import type { AppDeps } from "./deps";
|
|
import { readJob } from "./read-model";
|
|
|
|
function assertUse(ctx: RequestContext): void {
|
|
if (!ctx.moduleEnabled || !canUseSpreadsheetAdapter(ctx)) {
|
|
throw new AdapterError("FEATURE_DISABLED", "O Adaptador não está habilitado neste workspace.", 403);
|
|
}
|
|
}
|
|
|
|
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,
|
|
action: string,
|
|
resourceId: string,
|
|
result: "success" | "error",
|
|
metadata?: Record<string, unknown>,
|
|
): Promise<void> {
|
|
await query.query(
|
|
`insert into public.audit_log (workspace_id, actor_id, action, resource_type, resource_id, result, metadata)
|
|
values ($1,$2,$3,'job',$4,$5,$6::jsonb)`,
|
|
[ctx.workspaceId, ctx.userId, action, resourceId, result, JSON.stringify(sanitizeAuditMetadata(metadata))],
|
|
);
|
|
}
|
|
|
|
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) {
|
|
const acquired = await deps.coordinator.acquireLock(userLock, jobId, 120);
|
|
if (!acquired) throw new AdapterError("JOB_LOCKED", "Já existe uma transferência em andamento para este usuário.", 409);
|
|
}
|
|
const loaded = await deps.db.withUser(ctx.userId, async (query) => {
|
|
const rows = await query.query<{
|
|
storage_key: string;
|
|
content_sha256: string;
|
|
model_name: string;
|
|
columns_schema: unknown;
|
|
temp_storage_key: string;
|
|
source_name: string;
|
|
selected_sheet: string | null;
|
|
header_row: number;
|
|
source_count: number;
|
|
}>(
|
|
`select m.storage_key, m.content_sha256, m.name as model_name, m.columns_schema,
|
|
s.temp_storage_key, s.source_name, s.selected_sheet, s.header_row, j.source_count
|
|
from public.spreadsheet_jobs j
|
|
join public.spreadsheet_models m on m.id = j.model_id
|
|
join public.spreadsheet_job_sources s on s.job_id = j.id
|
|
where j.id = $1 and j.workspace_id = $2 and j.user_id = $3 and s.id = $4`,
|
|
[jobId, ctx.workspaceId, ctx.userId, sourceId],
|
|
);
|
|
const mappings = await query.query<{ source_column: string; model_column: string }>(
|
|
`select source_column, model_column from public.spreadsheet_job_mappings where job_source_id = $1 order by position`,
|
|
[sourceId],
|
|
);
|
|
return { row: rows[0], mappings };
|
|
});
|
|
if (!loaded.row) throw new AdapterError("NOT_FOUND", "Fonte da operação não encontrada.", 404);
|
|
const columns = asJson<{ name: string; required: boolean }[]>(loaded.row.columns_schema, []);
|
|
const validation = validateMappings(
|
|
columns,
|
|
[],
|
|
loaded.mappings.map((mapping) => ({ sourceColumn: mapping.source_column, modelColumn: mapping.model_column })),
|
|
);
|
|
const modelBytes = await deps.storage.get(loaded.row.storage_key);
|
|
if (sha256(modelBytes) !== loaded.row.content_sha256) {
|
|
throw new AdapterError("MODEL_TAMPERED", "O arquivo do modelo não confere com o cadastro.", 409);
|
|
}
|
|
const sourceBytes = await deps.storage.get(loaded.row.temp_storage_key);
|
|
const parsed = await readSpreadsheet(loaded.row.source_name, sourceBytes, {
|
|
sheet: loaded.row.selected_sheet,
|
|
headerRow: loaded.row.header_row,
|
|
});
|
|
const checked = validateMappings(
|
|
columns,
|
|
parsed.grid.headers,
|
|
loaded.mappings.map((mapping) => ({ sourceColumn: mapping.source_column, modelColumn: mapping.model_column })),
|
|
);
|
|
if (!checked.ok || validation.missingRequired.length) {
|
|
throw new AdapterError(checked.code, checked.missingRequired.length ? "Mapeamento obrigatório incompleto." : "Mapeamento inválido.", 422);
|
|
}
|
|
if (parsed.grid.columns.some((column) => column.inferredType === "mixed")) {
|
|
throw new AdapterError("TYPE_INCONSISTENT", "A fonte mudou e agora tem tipos inconsistentes.", 422);
|
|
}
|
|
const sourceIndex = 1;
|
|
const total = Number(loaded.row.source_count) || 1;
|
|
const progress = async (value: JobProgress) => {
|
|
await deps.coordinator.setProgress(jobId, value).catch(() => undefined);
|
|
await deps.coordinator.renewLock(userLock, jobId, 120).catch(() => undefined);
|
|
};
|
|
await progress({ phase: "processing", label: "Processando", percent: 0, sourceIndex, sourceCount: total, sourceId });
|
|
const transformed = transformRows(
|
|
columns.map((column) => column.name),
|
|
parsed.grid.headers,
|
|
parsed.grid.rows,
|
|
loaded.mappings.map((mapping) => ({ sourceColumn: mapping.source_column, modelColumn: mapping.model_column })),
|
|
).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",
|
|
label: "Processando",
|
|
percent: Math.round((done / transformed.length) * 100),
|
|
sourceIndex,
|
|
sourceCount: total,
|
|
sourceId,
|
|
});
|
|
}
|
|
}
|
|
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`;
|
|
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);
|
|
written.push(csvKey);
|
|
await deps.storage.put(xlsxKey, xlsx);
|
|
written.push(xlsxKey);
|
|
const after = await deps.storage.get(loaded.row.storage_key);
|
|
if (!after.equals(modelBytes)) {
|
|
throw new AdapterError("MODEL_TAMPERED", "O modelo original foi modificado durante a transferência.", 500);
|
|
}
|
|
const finished: JobProgress = {
|
|
phase: "completed",
|
|
label: "Concluído",
|
|
percent: 100,
|
|
sourceIndex,
|
|
sourceCount: total,
|
|
sourceId,
|
|
};
|
|
await deps.db.withUser(ctx.userId, async (query) => {
|
|
await query.query(
|
|
`update public.spreadsheet_job_sources
|
|
set status = 'completed', output_csv_key = $2, output_xlsx_key = $3, processed_at = now(), error_code = null
|
|
where id = $1`,
|
|
[sourceId, csvKey, xlsxKey],
|
|
);
|
|
const pending = await query.query<{ total: number }>(
|
|
`select count(*)::int as total from public.spreadsheet_job_sources where job_id = $1 and status <> 'completed'`,
|
|
[jobId],
|
|
);
|
|
const status = Number(pending[0]?.total ?? 0) === 0 ? "completed" : "ready";
|
|
await query.query(
|
|
`update public.spreadsheet_jobs
|
|
set status = $2, progress = $3::jsonb, error_code = null, finished_at = case when $2 = 'completed' then now() else finished_at end, updated_at = now()
|
|
where id = $1`,
|
|
[jobId, status, JSON.stringify(finished)],
|
|
);
|
|
await audit(query, ctx, "job.completed", jobId, "success", { source_id: sourceId, output_rows: transformed.length });
|
|
});
|
|
await deps.coordinator.setProgress(jobId, finished).catch(() => undefined);
|
|
} catch (error) {
|
|
for (const key of written) await deps.storage.delete(key).catch(() => undefined);
|
|
const code = isAdapterError(error) ? error.code : "PROCESS_FAILED";
|
|
const httpStatus = isAdapterError(error) ? error.status : 500;
|
|
let requeue = false;
|
|
await deps.db
|
|
.withUser(ctx.userId, async (query) => {
|
|
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,
|
|
attempt,
|
|
};
|
|
await query.query(
|
|
`update public.spreadsheet_job_sources set status = $2, error_code = $3 where id = $1 and status <> 'completed'`,
|
|
[sourceId, requeue ? "mapped" : "failed", code],
|
|
);
|
|
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);
|
|
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 });
|
|
}
|
|
if (isAdapterError(error)) throw error;
|
|
throw new AdapterError("PROCESS_FAILED", "O processamento falhou. Você pode tentar novamente.", 500);
|
|
} finally {
|
|
await deps.coordinator.releaseLock(userLock, jobId).catch(() => undefined);
|
|
}
|
|
}
|
|
|
|
const CONTENT_TYPE: Record<OutputFormat, string> = {
|
|
xlsx: "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
csv: "text/csv; charset=utf-8",
|
|
pdf: "application/pdf",
|
|
};
|
|
|
|
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;
|
|
output_csv_key: string | null;
|
|
output_xlsx_key: string | null;
|
|
expires_at: string | Date;
|
|
status: string;
|
|
}>(
|
|
`select m.name as model_name, s.output_csv_key, s.output_xlsx_key, s.expires_at, s.status
|
|
from public.spreadsheet_job_sources s
|
|
join public.spreadsheet_jobs j on j.id = s.job_id
|
|
join public.spreadsheet_models m on m.id = j.model_id
|
|
where s.id = $1 and j.id = $2 and j.workspace_id = $3 and j.user_id = $4 and s.expires_at > $5::timestamptz`,
|
|
[sourceId, jobId, ctx.workspaceId, ctx.userId, (deps.now ?? (() => new Date()))().toISOString()],
|
|
);
|
|
if (!rows[0] || rows[0].status !== "completed") throw new AdapterError("NOT_FOUND", "Saída indisponível ou expirada.", 404);
|
|
await audit(query, ctx, "job.download", jobId, "success", { format, source_id: sourceId });
|
|
return rows[0];
|
|
});
|
|
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);
|
|
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] ?? [];
|
|
// Import tardio: o worker não gera PDF (o PDF sai sob demanda no download),
|
|
// e carregar @react-pdf/renderer fora do bundler do Next falha ao resolver
|
|
// o subpath de hifenização. Sem isto o processo do worker nem subia.
|
|
const { buildPdf } = await import("@/infra/export/pdf");
|
|
const body = await buildPdf(file.model_name, headers, grid.slice(1));
|
|
return { filename, contentType: CONTENT_TYPE.pdf, body: body.body };
|
|
}
|
|
const key = format === "csv" ? file.output_csv_key : file.output_xlsx_key;
|
|
if (!key) throw new AdapterError("NOT_FOUND", "Saída indisponível.", 404);
|
|
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; workspace_id: string; temp_storage_key: string; output_csv_key: string | null; output_xlsx_key: string | null }>(
|
|
`select id, workspace_id, temp_storage_key, output_csv_key, output_xlsx_key
|
|
from public.adapter_list_expired_sources($1::timestamptz)`,
|
|
[(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.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(`select public.adapter_mark_source_expired($1::uuid)`, [row.id]));
|
|
removed += 1;
|
|
}
|
|
return removed;
|
|
}
|
|
|
|
export async function recoverStaleJobs(deps: AppDeps): Promise<number> {
|
|
const rows = await deps.db.withService((query) =>
|
|
query.query<{ id: string; user_id: string; workspace_id: string }>(
|
|
`select id, user_id, workspace_id from public.adapter_list_processing_jobs()`,
|
|
),
|
|
);
|
|
let recovered = 0;
|
|
for (const row of rows) {
|
|
const owns = await deps.coordinator.ownsLock(lockKey(row.user_id), row.id).catch(() => false);
|
|
if (owns) continue;
|
|
await deps.db.withService((query) => query.query(`select public.adapter_mark_job_interrupted($1::uuid)`, [row.id]));
|
|
recovered += 1;
|
|
}
|
|
return recovered;
|
|
}
|
|
|
|
export async function readJobForWorker(deps: AppDeps, jobId: string) {
|
|
return deps.db.withService((query) =>
|
|
query.query<{ user_id: string; workspace_id: string; progress: unknown }>(
|
|
`select user_id, workspace_id, progress from public.adapter_read_queued_job($1::uuid)`,
|
|
[jobId],
|
|
),
|
|
);
|
|
}
|
|
|
|
export async function listQueuedJobIds(deps: AppDeps): Promise<string[]> {
|
|
const rows = await deps.db.withService((query) =>
|
|
query.query<{ id: string }>(`select id from public.adapter_list_queued_job_ids()`),
|
|
);
|
|
return rows.map((row) => row.id);
|
|
}
|
|
|
|
export { mapDbError, readJob };
|