feat(scenario): add scheduler, cancelled step status, and runs list endpoint

- install @nestjs/schedule and register ScheduleModule in AppModule
- add ScenarioSchedulerService with two 1s interval jobs:
  - activatePendingRuns: flips pending runs to in_progress
  - processPendingStepRuns: executes next pending step run (login or exec),
    runs optional validateCode, advances or fails the run accordingly
- on step pass: next waiting step becomes pending; last step pass flips run to pass
- on step fail: remaining steps set to cancelled, run flipped to fail
- add cancelled to RunStepStatus union
- add GET /scenarios/:id/runs with pagination and optional ?status= filter
This commit is contained in:
2026-04-07 14:51:37 +03:00
parent caf120741d
commit cff36ffeff
9 changed files with 355 additions and 4 deletions
+46
View File
@@ -15,6 +15,7 @@
"@nestjs/config": "^4.0.3",
"@nestjs/core": "^11.1.18",
"@nestjs/platform-express": "^11.1.18",
"@nestjs/schedule": "^6.1.1",
"@nestjs/swagger": "^11.2.6",
"@nestjs/typeorm": "^11.0.1",
"acorn": "^8.16.0",
@@ -1174,6 +1175,19 @@
"@nestjs/core": "^11.0.0"
}
},
"node_modules/@nestjs/schedule": {
"version": "6.1.1",
"resolved": "https://registry.npmjs.org/@nestjs/schedule/-/schedule-6.1.1.tgz",
"integrity": "sha512-kQl1RRgi02GJ0uaUGCrXHCcwISsCsJDciCKe38ykJZgnAeeoeVWs8luWtBo4AqAAXm4nS5K8RlV0smHUJ4+2FA==",
"license": "MIT",
"dependencies": {
"cron": "4.4.0"
},
"peerDependencies": {
"@nestjs/common": "^10.0.0 || ^11.0.0",
"@nestjs/core": "^10.0.0 || ^11.0.0"
}
},
"node_modules/@nestjs/schematics": {
"version": "11.0.10",
"resolved": "https://registry.npmjs.org/@nestjs/schematics/-/schematics-11.0.10.tgz",
@@ -1468,6 +1482,12 @@
"dev": true,
"license": "MIT"
},
"node_modules/@types/luxon": {
"version": "3.7.1",
"resolved": "https://registry.npmjs.org/@types/luxon/-/luxon-3.7.1.tgz",
"integrity": "sha512-H3iskjFIAn5SlJU7OuxUmTEpebK6TKB8rxZShDslBMZJ5u9S//KM1sbdAisiSrqwLQncVjnpi2OK2J51h+4lsg==",
"license": "MIT"
},
"node_modules/@types/mozilla__readability": {
"version": "0.4.2",
"resolved": "https://registry.npmjs.org/@types/mozilla__readability/-/mozilla__readability-0.4.2.tgz",
@@ -2445,6 +2465,23 @@
}
}
},
"node_modules/cron": {
"version": "4.4.0",
"resolved": "https://registry.npmjs.org/cron/-/cron-4.4.0.tgz",
"integrity": "sha512-fkdfq+b+AHI4cKdhZlppHveI/mgz2qpiYxcm+t5E5TsxX7QrLS1VE0+7GENEk9z0EeGPcpSciGv6ez24duWhwQ==",
"license": "MIT",
"dependencies": {
"@types/luxon": "~3.7.0",
"luxon": "~3.7.0"
},
"engines": {
"node": ">=18.x"
},
"funding": {
"type": "ko-fi",
"url": "https://ko-fi.com/intcreator"
}
},
"node_modules/cross-spawn": {
"version": "7.0.6",
"resolved": "https://registry.npmjs.org/cross-spawn/-/cross-spawn-7.0.6.tgz",
@@ -3836,6 +3873,15 @@
"node": "20 || >=22"
}
},
"node_modules/luxon": {
"version": "3.7.2",
"resolved": "https://registry.npmjs.org/luxon/-/luxon-3.7.2.tgz",
"integrity": "sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew==",
"license": "MIT",
"engines": {
"node": ">=12"
}
},
"node_modules/magic-string": {
"version": "0.30.17",
"resolved": "https://registry.npmjs.org/magic-string/-/magic-string-0.30.17.tgz",
+1
View File
@@ -19,6 +19,7 @@
"@nestjs/config": "^4.0.3",
"@nestjs/core": "^11.1.18",
"@nestjs/platform-express": "^11.1.18",
"@nestjs/schedule": "^6.1.1",
"@nestjs/swagger": "^11.2.6",
"@nestjs/typeorm": "^11.0.1",
"acorn": "^8.16.0",
+2
View File
@@ -1,6 +1,7 @@
import { Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { TypeOrmModule } from '@nestjs/typeorm';
import { ScheduleModule } from '@nestjs/schedule';
import { HealthController } from './health/health.controller';
import { AuthModule } from './auth/auth.module';
import { BrowserModule } from './browser/browser.module';
@@ -20,6 +21,7 @@ import { ScenarioModule } from './scenario/scenario.module';
isGlobal: true,
envFilePath: '.env',
}),
ScheduleModule.forRoot(),
TypeOrmModule.forRootAsync({
inject: [ConfigService],
useFactory: (config: ConfigService) => ({
+28
View File
@@ -0,0 +1,28 @@
import { ApiPropertyOptional } from '@nestjs/swagger';
import { Type } from 'class-transformer';
import { IsIn, IsInt, IsOptional, Min } from 'class-validator';
import { RunStatus } from '../scenario-run.entity';
export class RunsQueryDto {
@ApiPropertyOptional({ example: 1, default: 1 })
@IsOptional()
@Type(() => Number)
@IsInt()
@Min(1)
page?: number = 1;
@ApiPropertyOptional({ example: 20, default: 20 })
@IsOptional()
@Type(() => Number)
@IsInt()
@Min(1)
limit?: number = 20;
@ApiPropertyOptional({
enum: ['pending', 'in_progress', 'pass', 'fail'],
description: 'Filter by run status',
})
@IsOptional()
@IsIn(['pending', 'in_progress', 'pass', 'fail'])
status?: RunStatus;
}
+1 -1
View File
@@ -10,7 +10,7 @@ import {
import { ScenarioRunEntity } from './scenario-run.entity';
import { ScenarioStepEntity } from './scenario-step.entity';
export type RunStepStatus = 'waiting' | 'pending' | 'in_progress' | 'pass' | 'fail';
export type RunStepStatus = 'waiting' | 'pending' | 'in_progress' | 'pass' | 'fail' | 'cancelled';
@Entity('scenario_run_steps')
export class ScenarioRunStepEntity {
+238
View File
@@ -0,0 +1,238 @@
import { Injectable, Logger } from '@nestjs/common';
import { Interval } from '@nestjs/schedule';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { chromium } from 'playwright';
import { ScenarioRunEntity } from './scenario-run.entity';
import { ScenarioRunStepEntity } from './scenario-run-step.entity';
import { ScenarioStepEntity } from './scenario-step.entity';
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;
}
@Injectable()
export class ScenarioSchedulerService {
private readonly logger = new Logger(ScenarioSchedulerService.name);
private isProcessingStep = false;
constructor(
@InjectRepository(ScenarioRunEntity)
private readonly runRepo: Repository<ScenarioRunEntity>,
@InjectRepository(ScenarioRunStepEntity)
private readonly runStepRepo: Repository<ScenarioRunStepEntity>,
private readonly authService: AuthService,
private readonly codeExecutor: CodeExecutorService,
private readonly sessionService: SessionService,
) {}
// ── Job 1: flip pending runs → in_progress ─────────────────────────────────
@Interval(1000)
async activatePendingRuns(): Promise<void> {
const pending = await this.runRepo.find({ where: { status: 'pending' } });
if (pending.length === 0) return;
for (const run of pending) {
run.status = 'in_progress';
await this.runRepo.save(run);
this.logger.log(`Run #${run.id} → in_progress`);
}
}
// ── Job 2: process one pending step run per tick ───────────────────────────
@Interval(1000)
async processPendingStepRuns(): Promise<void> {
if (this.isProcessingStep) return;
this.isProcessingStep = true;
try {
await this.processNextPendingStep();
} finally {
this.isProcessingStep = false;
}
}
private async processNextPendingStep(): Promise<void> {
const stepRun = await this.runStepRepo.findOne({
where: { status: 'pending' },
relations: ['scenarioStep'],
order: { order: 'ASC' },
});
if (!stepRun) return;
const step = stepRun.scenarioStep as ScenarioStepEntity;
// Mark in_progress
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 {
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);
}
}
// ── 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}`);
// Optional validation
if (step.validateCode) {
const validateResult = await this.runValidateCode(step.sessionName, step.validateCode);
if (!validateResult.success) {
throw new Error(validateResult.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);
await this.runExecCode(step.sessionName, step.execCode);
this.logger.log(`StepRun #${stepRun.id}: exec OK`);
// Optional validation
if (step.validateCode) {
const validateResult = await this.runValidateCode(step.sessionName, step.validateCode);
if (!validateResult.success) {
throw new Error(validateResult.description ?? 'Validation failed');
}
}
await this.passStepRun(stepRun, null);
}
// ── Browser helpers ────────────────────────────────────────────────────────
private async runExecCode(sessionName: string, code: string): Promise<unknown> {
const session = await this.sessionService.findBySessionName(sessionName);
if (!session) throw new Error(`Session not found: ${sessionName}`);
const cookies: unknown[] = JSON.parse(session.cookies);
const localStorageData: Record<string, string> = JSON.parse(session.localStorage);
const browser = await chromium.launch({ headless: true });
try {
const context = await browser.newContext();
await context.addCookies(cookies as Parameters<typeof context.addCookies>[0]);
const page = await context.newPage();
await context.addInitScript((entries: Record<string, string>) => {
for (const [k, v] of Object.entries(entries)) window.localStorage.setItem(k, v);
}, localStorageData);
const { result } = await this.codeExecutor.execute(page, context, code);
return result;
} finally {
await browser.close();
}
}
private async runValidateCode(sessionName: string, validateCode: string): Promise<ValidateResult> {
let raw: unknown;
try {
raw = await this.runExecCode(sessionName, validateCode);
} catch (err) {
return { success: false, description: (err as Error).message };
}
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,
};
}
// Anything truthy = pass
return { success: Boolean(raw) };
}
// ── Pass / fail helpers ────────────────────────────────────────────────────
private async passStepRun(stepRun: ScenarioRunStepEntity, description: string | null): Promise<void> {
stepRun.status = 'pass';
stepRun.description = description;
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);
this.logger.log(`StepRun #${nextStep.id} → pending (next in sequence)`);
} 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.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.runRepo.update(stepRun.runId, { status: 'fail' });
this.logger.log(`Run #${stepRun.runId} → fail`);
}
}
+12
View File
@@ -17,6 +17,7 @@ import { UpdateScenarioDto } from './dto/update-scenario.dto';
import { CreateScenarioStepDto } from './dto/create-scenario-step.dto';
import { UpdateScenarioStepDto } from './dto/update-scenario-step.dto';
import { PaginationQueryDto } from './dto/pagination-query.dto';
import { RunsQueryDto } from './dto/runs-query.dto';
@ApiTags('scenarios')
@Controller('scenarios')
@@ -114,6 +115,17 @@ export class ScenarioController {
// ── Runs ──────────────────────────────────────────────────────────────────
@Get(':id/runs')
@ApiOperation({ summary: 'List runs for a scenario (paginated, filterable by status)' })
@ApiResponse({ status: 200 })
@ApiResponse({ status: 404, description: 'Scenario not found' })
findRuns(
@Param('id', ParseIntPipe) id: number,
@Query() query: RunsQueryDto,
) {
return this.scenarioService.findRuns(id, query);
}
@Post(':id/run')
@ApiOperation({ summary: 'Create a new run for a scenario' })
@ApiResponse({ status: 201, description: 'Run created with step runs' })
+11 -2
View File
@@ -6,11 +6,20 @@ import { ScenarioRunEntity } from './scenario-run.entity';
import { ScenarioRunStepEntity } from './scenario-run-step.entity';
import { ScenarioService } from './scenario.service';
import { ScenarioController } from './scenario.controller';
import { ScenarioSchedulerService } from './scenario-scheduler.service';
import { AuthModule } from '../auth/auth.module';
import { CodeExecutorModule } from '../code-executor/code-executor.module';
import { SessionModule } from '../session/session.module';
@Module({
imports: [TypeOrmModule.forFeature([ScenarioEntity, ScenarioStepEntity, ScenarioRunEntity, ScenarioRunStepEntity])],
imports: [
TypeOrmModule.forFeature([ScenarioEntity, ScenarioStepEntity, ScenarioRunEntity, ScenarioRunStepEntity]),
AuthModule,
CodeExecutorModule,
SessionModule,
],
controllers: [ScenarioController],
providers: [ScenarioService],
providers: [ScenarioService, ScenarioSchedulerService],
exports: [ScenarioService],
})
export class ScenarioModule {}
+16 -1
View File
@@ -10,6 +10,7 @@ import { UpdateScenarioDto } from './dto/update-scenario.dto';
import { CreateScenarioStepDto } from './dto/create-scenario-step.dto';
import { UpdateScenarioStepDto } from './dto/update-scenario-step.dto';
import { PaginationQueryDto } from './dto/pagination-query.dto';
import { RunsQueryDto } from './dto/runs-query.dto';
export interface PaginatedResult<T> {
data: T[];
@@ -100,7 +101,21 @@ export class ScenarioService {
await this.stepRepo.delete(stepId);
}
// ── Runs ──────────────────────────────────────────────────────────────────
async findRuns(scenarioId: number, query: RunsQueryDto): Promise<PaginatedResult<ScenarioRunEntity>> {
await this.findOne(scenarioId); // 404 guard
const page = query.page ?? 1;
const limit = query.limit ?? 20;
const where: Record<string, unknown> = { scenarioId };
if (query.status) where['status'] = query.status;
const [data, total] = await this.runRepo.findAndCount({
where,
relations: ['stepRuns'],
order: { id: 'DESC', stepRuns: { order: 'ASC' } },
skip: (page - 1) * limit,
take: limit,
});
return { data, total, page, limit };
}
async createRun(scenarioId: number): Promise<ScenarioRunEntity> {
const scenario = await this.findOne(scenarioId);