ars9
1f3a604940
- add global runs page and per-scenario runs/run-detail pages - run detail page polls every 1s until pass/fail, cleans up on unmount - runs list pages poll every 10s with AutoRefreshIndicator (pulse on data load) - fix scheduler: set run to pass after empty step loop to prevent stuck in_progress - fix table header colors and link cell color for readability - fix play button to navigate to the new run after creation - fix package.json import paths in server for Docker build context
420 lines
14 KiB
TypeScript
420 lines
14 KiB
TypeScript
import { Injectable } from "@nestjs/common";
|
|
import { TraceLogger } from "../common/trace-logger";
|
|
import { traceStorage } from "../common/trace-context";
|
|
import { Interval } from "@nestjs/schedule";
|
|
import { InjectRepository } from "@nestjs/typeorm";
|
|
import { Repository } from "typeorm";
|
|
import * as crypto from "crypto";
|
|
import { chromium } from "playwright";
|
|
import type { Browser, BrowserContext, Page } from "playwright";
|
|
import { ScenarioRunEntity } from "./scenario-run.entity";
|
|
import { ScenarioRunStepEntity } from "./scenario-run-step.entity";
|
|
import { ScenarioRunLogEntity } from "./scenario-run-log.entity";
|
|
import { ScenarioStepEntity } from "./scenario-step.entity";
|
|
import type { ScriptLogger } from "../code-executor/code-executor.service";
|
|
import { AuthService } from "../auth/auth.service";
|
|
import { CodeExecutorService } from "../code-executor/code-executor.service";
|
|
import { SessionService } from "../session/session.service";
|
|
|
|
interface ValidateResult {
|
|
success: boolean;
|
|
description?: string;
|
|
}
|
|
|
|
interface BrowserHandle {
|
|
browser: Browser;
|
|
context: BrowserContext;
|
|
page: Page;
|
|
}
|
|
|
|
@Injectable()
|
|
export class ScenarioSchedulerService {
|
|
private readonly logger = new TraceLogger(ScenarioSchedulerService.name);
|
|
private readonly activeRuns = new Set<number>();
|
|
private readonly runBrowsers = new Map<number, BrowserHandle>();
|
|
|
|
constructor(
|
|
@InjectRepository(ScenarioRunEntity)
|
|
private readonly runRepo: Repository<ScenarioRunEntity>,
|
|
@InjectRepository(ScenarioRunStepEntity)
|
|
private readonly runStepRepo: Repository<ScenarioRunStepEntity>,
|
|
@InjectRepository(ScenarioRunLogEntity)
|
|
private readonly runLogRepo: Repository<ScenarioRunLogEntity>,
|
|
private readonly authService: AuthService,
|
|
private readonly codeExecutor: CodeExecutorService,
|
|
private readonly sessionService: SessionService,
|
|
) {}
|
|
|
|
private persistLog(
|
|
runId: number,
|
|
stepRunId: number | null,
|
|
level: "log" | "warn" | "error",
|
|
message: string,
|
|
): void {
|
|
void this.runLogRepo.save(
|
|
this.runLogRepo.create({ runId, stepRunId, level, message }),
|
|
);
|
|
}
|
|
|
|
private stepLogger(stepRunId: number, runId: number): ScriptLogger {
|
|
return (level, msg) => {
|
|
this.logger[level](`StepRun #${stepRunId} script: ${msg}`);
|
|
this.persistLog(runId, stepRunId, level, msg);
|
|
};
|
|
}
|
|
|
|
private makeGetStepOutput(
|
|
stepRun: ScenarioRunStepEntity,
|
|
): (order: number) => Promise<unknown> {
|
|
return async (order: number) => {
|
|
const targetOrder = order < 0 ? stepRun.order + order : order;
|
|
if (targetOrder < 0) return null;
|
|
const sr = await this.runStepRepo.findOne({
|
|
where: { runId: stepRun.runId, order: targetOrder },
|
|
});
|
|
if (!sr?.output) return null;
|
|
try {
|
|
return JSON.parse(sr.output) as unknown;
|
|
} catch {
|
|
return sr.output;
|
|
}
|
|
};
|
|
}
|
|
|
|
// ── Job: pick up pending runs and process each to completion ─────────────
|
|
|
|
@Interval(1000)
|
|
async pickUpPendingRuns(): Promise<void> {
|
|
const pending = await this.runRepo.find({ where: { status: "pending" } });
|
|
for (const run of pending) {
|
|
if (this.activeRuns.has(run.id)) continue;
|
|
this.activeRuns.add(run.id);
|
|
run.status = "in_progress";
|
|
await this.runRepo.save(run);
|
|
this.logger.log(`Run #${run.id} → in_progress`);
|
|
const traceId = crypto.randomUUID();
|
|
void traceStorage.run({ traceId }, () =>
|
|
this.processRunToCompletion(run.id),
|
|
);
|
|
}
|
|
}
|
|
|
|
private async processRunToCompletion(runId: number): Promise<void> {
|
|
try {
|
|
let stepRun = await this.runStepRepo.findOne({
|
|
where: { runId, status: "pending" },
|
|
relations: ["scenarioStep"],
|
|
order: { order: "ASC" },
|
|
});
|
|
while (stepRun) {
|
|
await this.executeStepRun(stepRun);
|
|
stepRun = await this.runStepRepo.findOne({
|
|
where: { runId, status: "pending" },
|
|
relations: ["scenarioStep"],
|
|
order: { order: "ASC" },
|
|
});
|
|
}
|
|
// Guard: if there were no steps (or all steps already resolved via passStepRun),
|
|
// ensure the run is not left in in_progress.
|
|
await this.runRepo
|
|
.createQueryBuilder()
|
|
.update()
|
|
.set({ status: "pass" })
|
|
.where("id = :id AND status = 'in_progress'", { id: runId })
|
|
.execute();
|
|
} catch (err) {
|
|
this.logger.error(
|
|
`Run #${runId}: unexpected error: ${(err as Error).message}`,
|
|
);
|
|
await this.runRepo.update(runId, { status: "fail" });
|
|
} finally {
|
|
this.activeRuns.delete(runId);
|
|
}
|
|
}
|
|
|
|
private async executeStepRun(stepRun: ScenarioRunStepEntity): Promise<void> {
|
|
const step = stepRun.scenarioStep as ScenarioStepEntity;
|
|
|
|
stepRun.status = "in_progress";
|
|
await this.runStepRepo.save(stepRun);
|
|
this.logger.log(
|
|
`StepRun #${stepRun.id} (run #${stepRun.runId}, step #${step.id} order=${step.order}) → in_progress`,
|
|
);
|
|
|
|
try {
|
|
if (step.type === "login") {
|
|
await this.executeLoginStep(stepRun, step);
|
|
} else if (step.type === "sign") {
|
|
await this.executeSignStep(stepRun, step);
|
|
} else {
|
|
await this.executeExecStep(stepRun, step);
|
|
}
|
|
} catch (err) {
|
|
const msg = (err as Error).message ?? String(err);
|
|
this.logger.error(`StepRun #${stepRun.id} threw: ${msg}`);
|
|
await this.failStepRun(stepRun, msg);
|
|
}
|
|
}
|
|
|
|
// ── Shared browser per run ─────────────────────────────────────────────────
|
|
|
|
private async getOrCreateBrowserHandle(
|
|
runId: number,
|
|
sessionName: string,
|
|
): Promise<BrowserHandle> {
|
|
const existing = this.runBrowsers.get(runId);
|
|
if (existing) return existing;
|
|
|
|
const session = await this.sessionService.findBySessionName(sessionName);
|
|
if (!session) throw new Error(`Session not found: ${sessionName}`);
|
|
|
|
const cookies = JSON.parse(session.cookies) as Parameters<
|
|
BrowserContext["addCookies"]
|
|
>[0];
|
|
const localStorageData: Record<string, string> = JSON.parse(
|
|
session.localStorage,
|
|
);
|
|
|
|
const browser = await chromium.launch({
|
|
headless: true,
|
|
// TODO: env var move to config
|
|
executablePath: process.env.PLAYWRIGHT_CHROMIUM_EXECUTABLE_PATH,
|
|
});
|
|
const context = await browser.newContext();
|
|
await context.addCookies(cookies);
|
|
await context.addInitScript((entries: Record<string, string>) => {
|
|
for (const [k, v] of Object.entries(entries))
|
|
window.localStorage.setItem(k, v);
|
|
}, localStorageData);
|
|
const page = await context.newPage();
|
|
|
|
const handle: BrowserHandle = { browser, context, page };
|
|
this.runBrowsers.set(runId, handle);
|
|
this.logger.log(`Run #${runId}: browser created (session: ${sessionName})`);
|
|
return handle;
|
|
}
|
|
|
|
private async closeBrowserHandle(runId: number): Promise<void> {
|
|
const handle = this.runBrowsers.get(runId);
|
|
if (!handle) return;
|
|
this.runBrowsers.delete(runId);
|
|
try {
|
|
await handle.browser.close();
|
|
this.logger.log(`Run #${runId}: browser closed`);
|
|
} catch (err) {
|
|
this.logger.warn(
|
|
`Run #${runId}: error closing browser: ${(err as Error).message}`,
|
|
);
|
|
}
|
|
}
|
|
|
|
// ── Login step ─────────────────────────────────────────────────────────────
|
|
|
|
private async executeLoginStep(
|
|
stepRun: ScenarioRunStepEntity,
|
|
step: ScenarioStepEntity,
|
|
): Promise<void> {
|
|
// execCode must be a JSON object: { "keyId": "...", "environmentName": "..." }
|
|
let params: { keyId: string; environmentName: string };
|
|
try {
|
|
params = JSON.parse(step.execCode ?? "{}");
|
|
} catch {
|
|
throw new Error(
|
|
"login step execCode must be valid JSON with keyId and environmentName",
|
|
);
|
|
}
|
|
if (!params.keyId || !params.environmentName) {
|
|
throw new Error(
|
|
"login step execCode must include keyId and environmentName",
|
|
);
|
|
}
|
|
|
|
const loginResult = await this.authService.login(
|
|
params.keyId,
|
|
params.environmentName,
|
|
step.sessionName,
|
|
);
|
|
this.logger.log(
|
|
`StepRun #${stepRun.id}: login OK, session=${loginResult.sessionName}`,
|
|
);
|
|
|
|
if (step.validateCode) {
|
|
const { page, context } = await this.getOrCreateBrowserHandle(
|
|
stepRun.runId,
|
|
step.sessionName,
|
|
);
|
|
this.codeExecutor.validate(step.validateCode);
|
|
const { result } = await this.codeExecutor.execute(
|
|
page,
|
|
context,
|
|
step.validateCode,
|
|
this.stepLogger(stepRun.id, stepRun.runId),
|
|
this.makeGetStepOutput(stepRun),
|
|
);
|
|
const vr = this.parseValidateResult(result);
|
|
if (!vr.success) throw new Error(vr.description ?? "Validation failed");
|
|
}
|
|
|
|
await this.passStepRun(stepRun, null);
|
|
}
|
|
|
|
// ── Exec step ──────────────────────────────────────────────────────────────
|
|
|
|
private async executeExecStep(
|
|
stepRun: ScenarioRunStepEntity,
|
|
step: ScenarioStepEntity,
|
|
): Promise<void> {
|
|
if (!step.execCode) throw new Error("exec step has no execCode");
|
|
|
|
this.codeExecutor.validate(step.execCode);
|
|
const { page, context } = await this.getOrCreateBrowserHandle(
|
|
stepRun.runId,
|
|
step.sessionName,
|
|
);
|
|
const getStepOutput = this.makeGetStepOutput(stepRun);
|
|
const { result: execOutput } = await this.codeExecutor.execute(
|
|
page,
|
|
context,
|
|
step.execCode,
|
|
this.stepLogger(stepRun.id, stepRun.runId),
|
|
getStepOutput,
|
|
);
|
|
this.logger.log(`StepRun #${stepRun.id}: exec OK`);
|
|
|
|
if (step.validateCode) {
|
|
this.codeExecutor.validate(step.validateCode);
|
|
const { result } = await this.codeExecutor.execute(
|
|
page,
|
|
context,
|
|
step.validateCode,
|
|
this.stepLogger(stepRun.id, stepRun.runId),
|
|
getStepOutput,
|
|
);
|
|
const vr = this.parseValidateResult(result);
|
|
if (!vr.success) throw new Error(vr.description ?? "Validation failed");
|
|
}
|
|
|
|
await this.passStepRun(stepRun, null, execOutput);
|
|
}
|
|
|
|
// ── Sign step ──────────────────────────────────────────────────────────────
|
|
|
|
private async executeSignStep(
|
|
stepRun: ScenarioRunStepEntity,
|
|
step: ScenarioStepEntity,
|
|
): Promise<void> {
|
|
// execCode must be a JSON object: { "keyId": "..." }
|
|
let params: { keyId: string };
|
|
try {
|
|
params = JSON.parse(step.execCode ?? "{}");
|
|
} catch {
|
|
throw new Error("sign step execCode must be valid JSON with keyId");
|
|
}
|
|
if (!params.keyId) throw new Error("sign step execCode must include keyId");
|
|
|
|
const { page, context } = await this.getOrCreateBrowserHandle(
|
|
stepRun.runId,
|
|
step.sessionName,
|
|
);
|
|
await this.authService.signWithKey(params.keyId, page);
|
|
this.logger.log(`StepRun #${stepRun.id}: sign OK`);
|
|
|
|
if (step.validateCode) {
|
|
this.codeExecutor.validate(step.validateCode);
|
|
const { result } = await this.codeExecutor.execute(
|
|
page,
|
|
context,
|
|
step.validateCode,
|
|
this.stepLogger(stepRun.id, stepRun.runId),
|
|
this.makeGetStepOutput(stepRun),
|
|
);
|
|
const vr = this.parseValidateResult(result);
|
|
if (!vr.success) throw new Error(vr.description ?? "Validation failed");
|
|
}
|
|
|
|
await this.passStepRun(stepRun, null);
|
|
}
|
|
|
|
// ── Validation helper ──────────────────────────────────────────────────────
|
|
|
|
private parseValidateResult(raw: unknown): ValidateResult {
|
|
if (typeof raw === "boolean") return { success: raw };
|
|
if (raw && typeof raw === "object") {
|
|
const r = raw as Record<string, unknown>;
|
|
return {
|
|
success: Boolean(r["success"]),
|
|
description:
|
|
r["description"] != null ? String(r["description"]) : undefined,
|
|
};
|
|
}
|
|
return { success: Boolean(raw) };
|
|
}
|
|
|
|
// ── Pass / fail helpers ────────────────────────────────────────────────────
|
|
|
|
private async passStepRun(
|
|
stepRun: ScenarioRunStepEntity,
|
|
description: string | null,
|
|
output: unknown = null,
|
|
): Promise<void> {
|
|
stepRun.status = "pass";
|
|
stepRun.description = description;
|
|
stepRun.output =
|
|
output !== null && output !== undefined ? JSON.stringify(output) : null;
|
|
await this.runStepRepo.save(stepRun);
|
|
this.logger.log(`StepRun #${stepRun.id} → pass`);
|
|
|
|
// Find the next waiting step in this run (next by order)
|
|
const nextStep = await this.runStepRepo.findOne({
|
|
where: { runId: stepRun.runId, status: "waiting" },
|
|
order: { order: "ASC" },
|
|
});
|
|
|
|
if (nextStep) {
|
|
nextStep.status = "pending";
|
|
await this.runStepRepo.save(nextStep);
|
|
} else {
|
|
// No more waiting steps — check if any are still in_progress/pending (shouldn't be, but guard anyway)
|
|
const remaining = await this.runStepRepo.count({
|
|
where: [
|
|
{ runId: stepRun.runId, status: "pending" },
|
|
{ runId: stepRun.runId, status: "in_progress" },
|
|
{ runId: stepRun.runId, status: "waiting" },
|
|
],
|
|
});
|
|
if (remaining === 0) {
|
|
await this.closeBrowserHandle(stepRun.runId);
|
|
await this.runRepo.update(stepRun.runId, { status: "pass" });
|
|
this.logger.log(`Run #${stepRun.runId} → pass (all steps passed)`);
|
|
}
|
|
}
|
|
}
|
|
|
|
private async failStepRun(
|
|
stepRun: ScenarioRunStepEntity,
|
|
description: string,
|
|
): Promise<void> {
|
|
stepRun.status = "fail";
|
|
stepRun.description = description;
|
|
await this.runStepRepo.save(stepRun);
|
|
this.logger.log(`StepRun #${stepRun.id} → fail: ${description}`);
|
|
|
|
// Cancel all remaining waiting/pending step runs in this run
|
|
await this.runStepRepo
|
|
.createQueryBuilder()
|
|
.update()
|
|
.set({ status: "cancelled" })
|
|
.where("runId = :runId AND status IN (:...statuses)", {
|
|
runId: stepRun.runId,
|
|
statuses: ["waiting", "pending"],
|
|
})
|
|
.execute();
|
|
|
|
this.logger.log(`Run #${stepRun.runId}: remaining steps cancelled`);
|
|
|
|
await this.closeBrowserHandle(stepRun.runId);
|
|
await this.runRepo.update(stepRun.runId, { status: "fail" });
|
|
this.logger.log(`Run #${stepRun.runId} → fail`);
|
|
}
|
|
}
|