Implementa o MVP do Adaptador de Planilhas no Elo Pro.
O módulo usa a identidade Supabase, isola cada workspace e registra storage, TTL, runner e biblioteca XLSX como decisões ainda abertas. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
camilafagundesandCursor committed
commit
6e23f1533b
144 files changed
+19060
No files matched your search
@@ -0,0 +1,316 @@
|
||||
import { AdapterError, isAdapterError } from "@/domain/spreadsheet-adapter/errors";
|
||||
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 { buildPdf } from "@/infra/export/pdf";
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
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 assertTempKey(key: string): void {
|
||||
if (!key.includes("/tmp/") || key.includes("/models/") || key.includes("..")) {
|
||||
throw new AdapterError("PATH_TRAVERSAL", "Chave temporária rejeitada.");
|
||||
}
|
||||
}
|
||||
|
||||
export async function runProcess(deps: AppDeps, ctx: RequestContext, jobId: string, sourceId: string): Promise<void> {
|
||||
const userLock = lockKey(ctx.userId);
|
||||
const written: string[] = [];
|
||||
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;
|
||||
for (const _row of transformed) {
|
||||
done += 1;
|
||||
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`;
|
||||
assertTempKey(csvKey);
|
||||
assertTempKey(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 failed: JobProgress = {
|
||||
phase: "failed",
|
||||
label: "Falhou",
|
||||
percent: null,
|
||||
sourceIndex: 1,
|
||||
sourceCount: 1,
|
||||
sourceId,
|
||||
};
|
||||
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'`, [
|
||||
sourceId,
|
||||
code,
|
||||
]);
|
||||
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)],
|
||||
);
|
||||
await audit(query, ctx, "job.failed", jobId, "error", { error_code: code, source_id: sourceId });
|
||||
})
|
||||
.catch(() => undefined);
|
||||
await deps.coordinator.setProgress(jobId, failed).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 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);
|
||||
assertTempKey(file.output_csv_key);
|
||||
const csv = (await deps.storage.get(file.output_csv_key)).toString("utf8");
|
||||
const grid = parseCsvText(csv);
|
||||
const headers = grid[0] ?? [];
|
||||
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);
|
||||
assertTempKey(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'`,
|
||||
[(deps.now ?? (() => new Date()))().toISOString()],
|
||||
),
|
||||
);
|
||||
for (const row of rows) {
|
||||
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);
|
||||
}
|
||||
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]),
|
||||
);
|
||||
}
|
||||
return rows.length;
|
||||
}
|
||||
|
||||
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.spreadsheet_jobs where status = 'processing'`,
|
||||
),
|
||||
);
|
||||
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(
|
||||
`update public.spreadsheet_jobs set status = 'failed', error_code = 'INTERRUPTED', updated_at = now() where id = $1 and status = 'processing'`,
|
||||
[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.spreadsheet_jobs where id = $1 and status = 'queued'`,
|
||||
[jobId],
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
export async function listQueuedJobIds(deps: AppDeps): Promise<string[]> {
|
||||
const rows = await deps.db.withService((query) =>
|
||||
query.query<{ id: string }>(`select id from public.spreadsheet_jobs where status = 'queued' order by created_at asc limit 5`),
|
||||
);
|
||||
return rows.map((row) => row.id);
|
||||
}
|
||||
|
||||
export { mapDbError, readJob };
|
||||
Reference in new issue
Block a user