77 lines
3.3 KiB
TypeScript
77 lines
3.3 KiB
TypeScript
import path from "node:path";
|
|
import { createFileDb } from "../src/infra/db/pglite";
|
|
import { createPostgresDb } from "../src/infra/db/postgres";
|
|
import { createCoordinator } from "../src/infra/redis";
|
|
import { createStorage } from "../src/infra/storage";
|
|
import { migrate } from "../src/infra/db/migrate";
|
|
import { cleanupExpired, listQueuedJobIds, readJobForWorker, recoverStaleJobs, runProcess } from "../src/server/spreadsheet-adapter/process-job";
|
|
import { asJson } from "../src/server/spreadsheet-adapter/db-utils";
|
|
import type { JobProgress, RequestContext, Role } from "../src/domain/spreadsheet-adapter/types";
|
|
|
|
import { assertProductionEnv } from "../src/config/env";
|
|
|
|
assertProductionEnv(process.env);
|
|
const databaseUrl = process.env.DATABASE_URL;
|
|
if (!databaseUrl && process.env.NODE_ENV === "production") {
|
|
throw new Error("DATABASE_URL é obrigatório para o worker.");
|
|
}
|
|
const db = databaseUrl ? createPostgresDb(databaseUrl) : await createFileDb(path.join(process.cwd(), ".data", "pglite"));
|
|
await migrate(db);
|
|
const deps = { db, storage: createStorage(), coordinator: createCoordinator() };
|
|
|
|
async function tick() {
|
|
await recoverStaleJobs(deps).catch((error) => {
|
|
console.error("worker", error instanceof Error ? error.name : "error");
|
|
});
|
|
await cleanupExpired(deps).catch((error) => {
|
|
console.error("worker", "cleanup", error instanceof Error ? error.name : "error");
|
|
});
|
|
const queued = new Set(await listQueuedJobIds(deps));
|
|
for (let index = 0; index < 20; index += 1) {
|
|
const jobId = await deps.coordinator.dequeue().catch(() => null);
|
|
if (!jobId) break;
|
|
queued.add(jobId);
|
|
}
|
|
for (const jobId of queued) {
|
|
const rows = await readJobForWorker(deps, jobId);
|
|
const row = rows[0];
|
|
if (!row) continue;
|
|
const progress = asJson<JobProgress | null>(row.progress, null);
|
|
if (!progress?.sourceId) continue;
|
|
const profile = await db.withService((query) =>
|
|
query.query<{ display_name: string; is_superadmin: boolean; role: Role; workspace_name: string; module_enabled: boolean; allow_upload: boolean }>(
|
|
`select p.display_name, p.is_superadmin, w.name as workspace_name, m.role,
|
|
coalesce(f.enabled, false) as module_enabled,
|
|
coalesce(s.allow_non_admin_model_upload, false) as allow_upload
|
|
from public.profiles p
|
|
join public.workspace_members m on m.user_id = p.id and m.workspace_id = $2
|
|
join public.workspaces w on w.id = $2
|
|
left join public.workspace_module_flags f on f.workspace_id = w.id and f.module_key = 'spreadsheet_adapter'
|
|
left join public.spreadsheet_adapter_settings s on s.workspace_id = w.id
|
|
where p.id = $1`,
|
|
[row.user_id, row.workspace_id],
|
|
),
|
|
);
|
|
const found = profile[0];
|
|
if (!found) continue;
|
|
const ctx: RequestContext = {
|
|
userId: row.user_id,
|
|
displayName: found.display_name,
|
|
isSuperadmin: found.is_superadmin,
|
|
workspaceId: row.workspace_id,
|
|
workspaceName: found.workspace_name,
|
|
role: found.role,
|
|
moduleEnabled: found.module_enabled,
|
|
allowNonAdminModelUpload: found.allow_upload,
|
|
};
|
|
await runProcess(deps, ctx, jobId, progress.sourceId).catch((error) => {
|
|
console.error("worker", error instanceof Error ? error.name : "error");
|
|
});
|
|
}
|
|
}
|
|
|
|
await tick();
|
|
setInterval(() => {
|
|
void tick();
|
|
}, 2000);
|