783 lines
30 KiB
TypeScript
783 lines
30 KiB
TypeScript
import cron from 'node-cron';
|
|
import { spawn, ChildProcess } from 'child_process';
|
|
import fs from 'fs';
|
|
import path from 'path';
|
|
import dotenv from 'dotenv';
|
|
dotenv.config({ path: path.resolve(process.cwd(), '.env') });
|
|
|
|
import { Pool } from 'pg';
|
|
import { PrismaPg } from '@prisma/adapter-pg';
|
|
import { PrismaClient } from '@prisma/client';
|
|
|
|
// Fresh DB connection for scheduler
|
|
const pool = new Pool({ connectionString: process.env.DATABASE_URL });
|
|
const adapter = new PrismaPg(pool);
|
|
const prisma = new PrismaClient({ adapter });
|
|
|
|
// ─── Configuration ───────────────────────────────────────────────────────────
|
|
|
|
const LOGS_DIR = '/app/logs';
|
|
const LOG_RETENTION_DAYS = 30;
|
|
const JOB_POLL_INTERVAL_MS = 30_000; // Check pipeline state every 30 seconds
|
|
const CYCLE_COOLDOWN_MS = 6 * 60 * 60 * 1000; // 6 hours between cycles
|
|
const JOB_TIMEOUT_MS = 24 * 60 * 60 * 1000; // 24 hours max per job
|
|
const STALE_JOB_TIMEOUT_MS = 24 * 60 * 60 * 1000; // 24 hours — mark non-pipeline running jobs as stale
|
|
|
|
// ─── Pipeline Phases ────────────────────────────────────────────────────────
|
|
|
|
interface PipelinePhase {
|
|
name: string;
|
|
type: string;
|
|
language?: string;
|
|
config: Record<string, unknown>;
|
|
}
|
|
|
|
interface PipelineGroup {
|
|
name: string;
|
|
phases: PipelinePhase[];
|
|
mode: 'sequential' | 'parallel';
|
|
}
|
|
|
|
const PIPELINE_GROUPS: PipelineGroup[] = [
|
|
{
|
|
name: 'imports',
|
|
mode: 'sequential',
|
|
phases: [
|
|
{ name: 'osm-import-p1', type: 'osm-import', config: { priority: 1 } },
|
|
{ name: 'gcatholic-import', type: 'gcatholic-import', config: { delay: 2000 } },
|
|
{ name: 'orarimesse-import', type: 'orarimesse-import', config: {} },
|
|
{ name: 'mass-schedules-ph-import', type: 'mass-schedules-ph-import', config: {} },
|
|
{ name: 'philmass-import', type: 'philmass-import', config: {} },
|
|
{ name: 'horariosmisas-import', type: 'horariosmisas-import', config: {} },
|
|
{ name: 'msze-info-import', type: 'msze-info-import', config: {} },
|
|
{ name: 'weekdaymasses-import', type: 'weekdaymasses-import', config: {} },
|
|
{ name: 'messesinfo-import', type: 'messesinfo-import', config: {} },
|
|
{ name: 'bohosluzby-import', type: 'bohosluzby-import', config: {} },
|
|
{ name: 'miserend-import', type: 'miserend-import', config: {} },
|
|
{ name: 'kerknet-import', type: 'kerknet-import', config: {} },
|
|
{ name: 'gottesdienstzeiten-import', type: 'gottesdienstzeiten-import', config: {} },
|
|
{ name: 'masstimes-api-import', type: 'masstimes-api-import', config: {} },
|
|
],
|
|
},
|
|
{
|
|
name: 'scrapers-batch-1',
|
|
mode: 'parallel',
|
|
phases: [
|
|
{ name: 'scraper-english', type: 'scraper', language: 'english', config: { allMode: true, maxFailures: 10, language: 'english' } },
|
|
{ name: 'scraper-french', type: 'scraper', language: 'french', config: { allMode: true, maxFailures: 10, language: 'french' } },
|
|
{ name: 'scraper-german', type: 'scraper', language: 'german', config: { allMode: true, maxFailures: 10, language: 'german' } },
|
|
],
|
|
},
|
|
{
|
|
name: 'scrapers-batch-2',
|
|
mode: 'parallel',
|
|
phases: [
|
|
{ name: 'scraper-polish', type: 'scraper', language: 'polish', config: { allMode: true, maxFailures: 10, language: 'polish' } },
|
|
{ name: 'scraper-spanish', type: 'scraper', language: 'spanish', config: { allMode: true, maxFailures: 10, language: 'spanish' } },
|
|
{ name: 'scraper-italian', type: 'scraper', language: 'italian', config: { allMode: true, maxFailures: 10, language: 'italian' } },
|
|
{ name: 'scraper-portuguese', type: 'scraper', language: 'portuguese', config: { allMode: true, maxFailures: 10, language: 'portuguese' } },
|
|
{ name: 'scraper-czech', type: 'scraper', language: 'czech', config: { allMode: true, maxFailures: 10, language: 'czech' } },
|
|
],
|
|
},
|
|
{
|
|
name: 'scrapers-batch-3',
|
|
mode: 'parallel',
|
|
phases: [
|
|
{ name: 'scraper-dutch', type: 'scraper', language: 'dutch', config: { allMode: true, maxFailures: 10, language: 'dutch' } },
|
|
{ name: 'scraper-hungarian', type: 'scraper', language: 'hungarian', config: { allMode: true, maxFailures: 10, language: 'hungarian' } },
|
|
{ name: 'scraper-generic', type: 'scraper', language: 'generic', config: { allMode: true, maxFailures: 10, language: 'generic' } },
|
|
],
|
|
},
|
|
];
|
|
|
|
// ─── Cycle State ────────────────────────────────────────────────────────────
|
|
|
|
interface CycleState {
|
|
currentGroupIndex: number;
|
|
currentSequentialPhaseIndex: number;
|
|
cycleNumber: number;
|
|
cycleStartedAt: Date | null;
|
|
lastCycleCompletedAt: Date | null;
|
|
waitingForCooldown: boolean;
|
|
activeGroupJobs: number;
|
|
}
|
|
|
|
const cycleState: CycleState = {
|
|
currentGroupIndex: 0,
|
|
currentSequentialPhaseIndex: 0,
|
|
cycleNumber: 0,
|
|
cycleStartedAt: null,
|
|
lastCycleCompletedAt: null,
|
|
waitingForCooldown: false,
|
|
activeGroupJobs: 0,
|
|
};
|
|
|
|
// Map job type+language to script command
|
|
function getJobCommand(type: string, language?: string | null, config?: Record<string, unknown> | null): { command: string; args: string[] } {
|
|
const limit = (config?.limit as number) || undefined;
|
|
|
|
switch (type) {
|
|
case 'scraper': {
|
|
const args = ['tsx', 'scripts/scrape-churches.ts', '--all'];
|
|
if (language) args.push('--language', language);
|
|
if (config?.maxFailures) args.push('--max-failures', String(config.maxFailures));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'freesearch-enrichment': {
|
|
const args = ['tsx', 'scripts/enrich-with-freesearch.ts'];
|
|
if (config?.continuous) args.push('--continuous');
|
|
if (config?.reSearch) args.push('--re-search');
|
|
if (limit) args.push('--limit', String(limit));
|
|
if (config?.country) args.push('--country', String(config.country));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'reverse-geocode-enrichment': {
|
|
const args = ['tsx', 'scripts/enrich-with-reverse-geocode.ts'];
|
|
if (config?.continuous) args.push('--continuous');
|
|
if (limit) args.push('--limit', String(limit));
|
|
if (config?.country) args.push('--country', String(config.country));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'match-search-results': {
|
|
const args = ['tsx', 'scripts/match-search-results.ts'];
|
|
if (limit) args.push('--limit', String(limit));
|
|
if (config?.country) args.push('--country', String(config.country));
|
|
if (config?.threshold) args.push('--threshold', String(config.threshold));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'diocese-directory': {
|
|
const args = ['tsx', 'scripts/scrape-diocese-directory.ts', '--all'];
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'wikidata-enrichment': {
|
|
return { command: 'npx', args: ['tsx', 'scripts/enrich-with-wikidata.ts'] };
|
|
}
|
|
case 'osm-import': {
|
|
const args = ['tsx', 'scripts/import-osm-churches.ts'];
|
|
if (config?.priority) args.push('--priority', String(config.priority));
|
|
else args.push('--all');
|
|
if (config?.resumeFrom) args.push('--resume-from', String(config.resumeFrom));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'baidu-import': {
|
|
const args = ['tsx', 'scripts/import-baidu-churches.ts'];
|
|
if (config?.resumeFromCell) args.push('--resume-from-cell', String(config.resumeFromCell));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'gcatholic-import': {
|
|
const args = ['tsx', 'scripts/import-gcatholic.ts', '--all'];
|
|
if (config?.country) args.push('--country', String(config.country));
|
|
if (config?.resumeFrom) args.push('--resume-from', String(config.resumeFrom));
|
|
if (config?.delay) args.push('--delay', String(config.delay));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'orarimesse-import': {
|
|
const args = ['tsx', 'scripts/import-orarimesse.ts', '--all'];
|
|
if (config?.diocese) args.push('--diocese', String(config.diocese));
|
|
if (config?.resumeFrom) args.push('--resume-from', String(config.resumeFrom));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'mass-schedules-ph-import': {
|
|
const args = ['tsx', 'scripts/import-mass-schedules-ph.ts', '--all'];
|
|
if (config?.churchId) args.push('--church-id', String(config.churchId));
|
|
if (config?.resumeFrom) args.push('--resume-from', String(config.resumeFrom));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'philmass-import': {
|
|
const args = ['tsx', 'scripts/import-philmass.ts', '--all'];
|
|
if (config?.province) args.push('--province', String(config.province));
|
|
if (config?.resumeFrom) args.push('--resume-from', String(config.resumeFrom));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'horariosmisas-import': {
|
|
const args = ['tsx', 'scripts/import-horariosmisas.ts', '--all', '--geocode'];
|
|
if (config?.province) args.push('--province', String(config.province));
|
|
if (config?.resumeFrom) args.push('--resume-from', String(config.resumeFrom));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'msze-info-import': {
|
|
const args = ['tsx', 'scripts/import-msze-info.ts', '--all'];
|
|
if (config?.resumeFrom) args.push('--resume-from', String(config.resumeFrom));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'weekdaymasses-import': {
|
|
const args = ['tsx', 'scripts/import-weekdaymasses.ts', '--all'];
|
|
if (config?.resumeFrom) args.push('--resume-from', String(config.resumeFrom));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'messesinfo-import': {
|
|
const args = ['tsx', 'scripts/import-messesinfo.ts', '--all'];
|
|
if (config?.resumeFrom) args.push('--resume-from', String(config.resumeFrom));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'bohosluzby-import': {
|
|
const args = ['tsx', 'scripts/import-bohosluzby.ts', '--all'];
|
|
if (config?.resumeFrom) args.push('--resume-from', String(config.resumeFrom));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'miserend-import': {
|
|
const args = ['tsx', 'scripts/import-miserend.ts', '--all'];
|
|
if (config?.resumeFrom) args.push('--resume-from', String(config.resumeFrom));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'kerknet-import': {
|
|
const args = ['tsx', 'scripts/import-kerknet.ts', '--all'];
|
|
if (config?.resumeFrom) args.push('--resume-from', String(config.resumeFrom));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'gottesdienstzeiten-import': {
|
|
const args = ['tsx', 'scripts/import-gottesdienstzeiten.ts', '--all'];
|
|
if (config?.resumeFrom) args.push('--resume-from', String(config.resumeFrom));
|
|
return { command: 'npx', args };
|
|
}
|
|
case 'masstimes-api-import': {
|
|
const args = ['tsx', 'scripts/import-masstimes-api.ts', '--all', '--skip-us'];
|
|
if (config?.resumeFrom) args.push('--resume-from', String(config.resumeFrom));
|
|
if (config?.region) args.splice(2, 1, '--region', String(config.region)); // replace --all with --region
|
|
return { command: 'npx', args };
|
|
}
|
|
default:
|
|
throw new Error(`Unknown job type: ${type}`);
|
|
}
|
|
}
|
|
|
|
// ─── State ───────────────────────────────────────────────────────────────────
|
|
|
|
const runningJobs = new Map<string, { process: ChildProcess; jobId: string }>();
|
|
|
|
/** Kill entire process group (child + all descendants like Chromium).
|
|
* Falls back to child.kill() if process group kill fails. */
|
|
function killProcessGroup(child: ChildProcess, signal: NodeJS.Signals = 'SIGTERM'): void {
|
|
try {
|
|
if (child.pid) {
|
|
process.kill(-child.pid, signal); // negative PID = kill process group
|
|
}
|
|
} catch (err: unknown) {
|
|
const code = (err as NodeJS.ErrnoException).code;
|
|
if (code !== 'ESRCH') {
|
|
logError(`killProcessGroup(-${child.pid}, ${signal}) failed: ${err}`);
|
|
}
|
|
// Process group already dead, try direct kill as fallback
|
|
try { child.kill(signal); } catch { /* already dead */ }
|
|
}
|
|
}
|
|
|
|
// ─── Logging ─────────────────────────────────────────────────────────────────
|
|
|
|
function log(message: string): void {
|
|
const timestamp = new Date().toISOString();
|
|
console.log(`[${timestamp}] ${message}`);
|
|
}
|
|
|
|
function logError(message: string): void {
|
|
const timestamp = new Date().toISOString();
|
|
console.error(`[${timestamp}] ERROR: ${message}`);
|
|
}
|
|
|
|
// ─── Environment Validation ──────────────────────────────────────────────────
|
|
|
|
function validateEnvironment(): void {
|
|
if (!process.env.DATABASE_URL) {
|
|
logError('Missing required environment variable: DATABASE_URL');
|
|
process.exit(1);
|
|
}
|
|
|
|
log('Environment variables validated');
|
|
log(` DATABASE_URL: ${process.env.DATABASE_URL?.substring(0, 30)}...`);
|
|
}
|
|
|
|
// ─── Log File Management ─────────────────────────────────────────────────────
|
|
|
|
function ensureLogsDir(): void {
|
|
if (!fs.existsSync(LOGS_DIR)) {
|
|
fs.mkdirSync(LOGS_DIR, { recursive: true });
|
|
log(`Created logs directory: ${LOGS_DIR}`);
|
|
}
|
|
}
|
|
|
|
function getLogFilePath(jobType: string): string {
|
|
const timestamp = new Date().toISOString().replace(/[:.]/g, '-');
|
|
return path.join(LOGS_DIR, `${jobType}-${timestamp}.log`);
|
|
}
|
|
|
|
function cleanOldLogs(): void {
|
|
log('Cleaning old log files...');
|
|
const cutoff = Date.now() - LOG_RETENTION_DAYS * 24 * 60 * 60 * 1000;
|
|
let cleaned = 0;
|
|
|
|
try {
|
|
const files = fs.readdirSync(LOGS_DIR);
|
|
for (const file of files) {
|
|
if (!file.endsWith('.log')) continue;
|
|
const filePath = path.join(LOGS_DIR, file);
|
|
const stat = fs.statSync(filePath);
|
|
if (stat.mtimeMs < cutoff) {
|
|
fs.unlinkSync(filePath);
|
|
cleaned++;
|
|
}
|
|
}
|
|
log(`Cleaned ${cleaned} log files older than ${LOG_RETENTION_DAYS} days`);
|
|
} catch (err) {
|
|
logError(`Failed to clean logs: ${err}`);
|
|
}
|
|
}
|
|
|
|
// ─── Crash Recovery ─────────────────────────────────────────────────────────
|
|
|
|
async function recoverFromCrash(): Promise<void> {
|
|
log('Running crash recovery...');
|
|
|
|
// Re-queue orphaned 'running' pipeline jobs as 'pending' so they get retried
|
|
const requeued = await prisma.backgroundJob.updateMany({
|
|
where: {
|
|
status: 'running',
|
|
config: { path: ['pipelineManaged'], equals: true },
|
|
},
|
|
data: { status: 'pending', error: null, completedAt: null },
|
|
});
|
|
if (requeued.count > 0) {
|
|
log(`Re-queued ${requeued.count} pipeline jobs interrupted by restart`);
|
|
}
|
|
|
|
// Mark non-pipeline running jobs as failed (manual jobs can't be safely resumed)
|
|
const orphanedRunning = await prisma.backgroundJob.updateMany({
|
|
where: {
|
|
status: 'running',
|
|
NOT: { config: { path: ['pipelineManaged'], equals: true } },
|
|
},
|
|
data: { status: 'failed', error: 'Scheduler restarted', completedAt: new Date() },
|
|
});
|
|
if (orphanedRunning.count > 0) {
|
|
log(`Marked ${orphanedRunning.count} orphaned manual jobs as failed`);
|
|
}
|
|
|
|
// Keep stale 'pending' pipeline jobs — they'll be picked up by the pipeline
|
|
// Only cancel non-pipeline stale pending jobs
|
|
const stalePending = await prisma.backgroundJob.updateMany({
|
|
where: {
|
|
status: 'pending',
|
|
NOT: { config: { path: ['pipelineManaged'], equals: true } },
|
|
},
|
|
data: { status: 'failed', error: 'Scheduler restarted', completedAt: new Date() },
|
|
});
|
|
if (stalePending.count > 0) {
|
|
log(`Cancelled ${stalePending.count} stale pending manual jobs`);
|
|
}
|
|
|
|
// Resume pipeline from the last incomplete phase
|
|
const lastRunningPipelineJob = await prisma.backgroundJob.findFirst({
|
|
where: {
|
|
status: 'pending',
|
|
config: { path: ['pipelineManaged'], equals: true },
|
|
},
|
|
orderBy: { createdAt: 'desc' },
|
|
});
|
|
|
|
if (lastRunningPipelineJob) {
|
|
for (let gi = 0; gi < PIPELINE_GROUPS.length; gi++) {
|
|
const group = PIPELINE_GROUPS[gi];
|
|
const phaseIdx = group.phases.findIndex(
|
|
p => p.type === lastRunningPipelineJob.type &&
|
|
(p.language || null) === (lastRunningPipelineJob.language || null)
|
|
);
|
|
if (phaseIdx >= 0) {
|
|
cycleState.currentGroupIndex = gi;
|
|
cycleState.currentSequentialPhaseIndex = group.mode === 'sequential' ? phaseIdx : 0;
|
|
log(`Resuming pipeline from group ${gi + 1}: ${group.name}`);
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
log('Crash recovery complete');
|
|
}
|
|
|
|
async function cleanStaleJobs(): Promise<void> {
|
|
const cutoff = new Date(Date.now() - STALE_JOB_TIMEOUT_MS);
|
|
|
|
const stale = await prisma.backgroundJob.updateMany({
|
|
where: {
|
|
status: 'running',
|
|
startedAt: { lt: cutoff },
|
|
NOT: { config: { path: ['pipelineManaged'], equals: true } },
|
|
},
|
|
data: {
|
|
status: 'failed',
|
|
error: 'Stale: exceeded 24h runtime without completion',
|
|
completedAt: new Date(),
|
|
},
|
|
});
|
|
|
|
if (stale.count > 0) {
|
|
log(`Cleaned ${stale.count} stale non-pipeline job(s) running >24h`);
|
|
}
|
|
}
|
|
|
|
// ─── Database-Driven Job Queue ───────────────────────────────────────────────
|
|
|
|
async function createPendingJob(type: string, language?: string, config?: Record<string, unknown>): Promise<string> {
|
|
// Don't create if a job of this type is already pending or running
|
|
const existing = await prisma.backgroundJob.findFirst({
|
|
where: {
|
|
type,
|
|
language: language || null,
|
|
status: { in: ['pending', 'running'] },
|
|
},
|
|
});
|
|
|
|
if (existing) {
|
|
log(`Skipping ${type}${language ? `:${language}` : ''} — already ${existing.status} (${existing.id})`);
|
|
return existing.id;
|
|
}
|
|
|
|
const job = await prisma.backgroundJob.create({
|
|
data: {
|
|
type,
|
|
language: language || null,
|
|
status: 'pending',
|
|
config: config || null,
|
|
},
|
|
});
|
|
|
|
log(`Created pending job: ${type}${language ? `:${language}` : ''} (${job.id})`);
|
|
return job.id;
|
|
}
|
|
|
|
// ─── Continuous Pipeline Logic ──────────────────────────────────────────────
|
|
|
|
async function pollAndAdvancePipeline(): Promise<void> {
|
|
try {
|
|
// 1. Check for manual pending jobs from admin API (priority over pipeline)
|
|
if (runningJobs.size === 0) {
|
|
const manualJob = await prisma.backgroundJob.findFirst({
|
|
where: {
|
|
status: 'pending',
|
|
NOT: { config: { path: ['pipelineManaged'], equals: true } },
|
|
},
|
|
orderBy: { createdAt: 'asc' },
|
|
});
|
|
|
|
if (manualJob) {
|
|
log(`Found manual job: ${manualJob.type}${manualJob.language ? `:${manualJob.language}` : ''} (${manualJob.id})`);
|
|
await startJobProcess(
|
|
manualJob.id,
|
|
manualJob.type,
|
|
manualJob.language,
|
|
manualJob.config as Record<string, unknown> | null
|
|
);
|
|
return;
|
|
}
|
|
}
|
|
|
|
// 2. If jobs are still running for the current group, wait
|
|
if (cycleState.activeGroupJobs > 0) {
|
|
return;
|
|
}
|
|
|
|
// 3. If in cooldown, check if expired
|
|
if (cycleState.waitingForCooldown) {
|
|
if (cycleState.lastCycleCompletedAt) {
|
|
const elapsed = Date.now() - cycleState.lastCycleCompletedAt.getTime();
|
|
if (elapsed < CYCLE_COOLDOWN_MS) {
|
|
const remaining = Math.round((CYCLE_COOLDOWN_MS - elapsed) / 60_000);
|
|
if (remaining % 30 === 0 || remaining <= 5) {
|
|
log(`Cooldown: ${remaining} minutes remaining before next cycle`);
|
|
}
|
|
return;
|
|
}
|
|
}
|
|
cycleState.waitingForCooldown = false;
|
|
cycleState.currentGroupIndex = 0;
|
|
cycleState.currentSequentialPhaseIndex = 0;
|
|
log('Cooldown expired, starting new cycle');
|
|
}
|
|
|
|
// 4. If past the last group, complete the cycle
|
|
if (cycleState.currentGroupIndex >= PIPELINE_GROUPS.length) {
|
|
cycleState.cycleNumber++;
|
|
cycleState.lastCycleCompletedAt = new Date();
|
|
cycleState.waitingForCooldown = true;
|
|
const cooldownHours = CYCLE_COOLDOWN_MS / (60 * 60 * 1000);
|
|
log(`=== Cycle ${cycleState.cycleNumber} complete! Entering ${cooldownHours}h cooldown ===`);
|
|
return;
|
|
}
|
|
|
|
// 5. Start the current group
|
|
const group = PIPELINE_GROUPS[cycleState.currentGroupIndex];
|
|
|
|
if (cycleState.currentGroupIndex === 0 && cycleState.currentSequentialPhaseIndex === 0 && !cycleState.cycleStartedAt) {
|
|
cycleState.cycleStartedAt = new Date();
|
|
log(`=== Starting cycle ${cycleState.cycleNumber + 1} ===`);
|
|
}
|
|
|
|
if (group.mode === 'parallel') {
|
|
// Launch all phases in the group concurrently
|
|
log(`Pipeline group ${cycleState.currentGroupIndex + 1}/${PIPELINE_GROUPS.length}: ${group.name} (parallel, ${group.phases.length} jobs)`);
|
|
cycleState.activeGroupJobs = group.phases.length;
|
|
|
|
for (const phase of group.phases) {
|
|
const jobId = await createPendingJob(
|
|
phase.type,
|
|
phase.language,
|
|
{ ...phase.config, pipelineManaged: true }
|
|
);
|
|
await startJobProcess(jobId, phase.type, phase.language || null, phase.config, true);
|
|
}
|
|
} else {
|
|
// Sequential: run one phase at a time within the group
|
|
const phaseIndex = cycleState.currentSequentialPhaseIndex;
|
|
if (phaseIndex >= group.phases.length) {
|
|
// All phases in this sequential group are done
|
|
cycleState.currentGroupIndex++;
|
|
cycleState.currentSequentialPhaseIndex = 0;
|
|
return; // Will pick up next group on next poll
|
|
}
|
|
|
|
const phase = group.phases[phaseIndex];
|
|
log(`Pipeline group ${cycleState.currentGroupIndex + 1}/${PIPELINE_GROUPS.length}: ${group.name} (sequential ${phaseIndex + 1}/${group.phases.length}: ${phase.name})`);
|
|
cycleState.activeGroupJobs = 1;
|
|
|
|
const jobId = await createPendingJob(
|
|
phase.type,
|
|
phase.language,
|
|
{ ...phase.config, pipelineManaged: true }
|
|
);
|
|
await startJobProcess(jobId, phase.type, phase.language || null, phase.config, true);
|
|
}
|
|
} catch (err) {
|
|
logError(`Error in pipeline: ${err}`);
|
|
}
|
|
}
|
|
|
|
function onJobCompleted(): void {
|
|
cycleState.activeGroupJobs--;
|
|
|
|
if (cycleState.activeGroupJobs <= 0) {
|
|
cycleState.activeGroupJobs = 0;
|
|
const group = PIPELINE_GROUPS[cycleState.currentGroupIndex];
|
|
|
|
if (group?.mode === 'sequential') {
|
|
cycleState.currentSequentialPhaseIndex++;
|
|
if (cycleState.currentSequentialPhaseIndex < group.phases.length) {
|
|
return; // Don't advance group yet
|
|
}
|
|
}
|
|
|
|
cycleState.currentGroupIndex++;
|
|
cycleState.currentSequentialPhaseIndex = 0;
|
|
log(`Group "${group?.name}" complete, advancing to group ${cycleState.currentGroupIndex + 1}`);
|
|
}
|
|
}
|
|
|
|
async function startJobProcess(
|
|
jobId: string,
|
|
type: string,
|
|
language: string | null,
|
|
config: Record<string, unknown> | null,
|
|
isPipelineJob = false
|
|
): Promise<void> {
|
|
const { command, args } = getJobCommand(type, language, config);
|
|
|
|
// Add --job-id to args so the script tracks this job
|
|
args.push('--job-id', jobId);
|
|
|
|
const logFile = getLogFilePath(type);
|
|
const logStream = fs.createWriteStream(logFile, { flags: 'a' });
|
|
const typeKey = `${type}:${language || 'all'}`;
|
|
const stderrBuffer: string[] = []; // Keep last 20 lines of stderr for error reporting
|
|
|
|
log(`Starting job: ${typeKey} (${jobId})`);
|
|
log(` Command: ${command} ${args.join(' ')}`);
|
|
log(` Log file: ${logFile}`);
|
|
|
|
const header = `[${new Date().toISOString()}] Starting ${typeKey}: ${command} ${args.join(' ')}\n`;
|
|
logStream.write(header);
|
|
|
|
const child = spawn(command, args, {
|
|
cwd: '/app',
|
|
env: process.env as NodeJS.ProcessEnv,
|
|
stdio: ['ignore', 'pipe', 'pipe'],
|
|
detached: true, // own process group so we can kill Chromium grandchildren
|
|
});
|
|
|
|
runningJobs.set(typeKey, { process: child, jobId });
|
|
|
|
child.stdout?.on('data', (data: Buffer) => {
|
|
logStream.write(data);
|
|
});
|
|
|
|
child.stderr?.on('data', (data: Buffer) => {
|
|
logStream.write(data);
|
|
const lines = data.toString().split('\n').filter(l => l.trim());
|
|
stderrBuffer.push(...lines);
|
|
if (stderrBuffer.length > 20) {
|
|
stderrBuffer.splice(0, stderrBuffer.length - 20);
|
|
}
|
|
});
|
|
|
|
// Timeout — kill entire process group (including Chromium grandchildren)
|
|
const timeout = setTimeout(() => {
|
|
logError(`Job ${typeKey} (${jobId}) timed out after ${JOB_TIMEOUT_MS / 3600000}h, killing process group...`);
|
|
logStream.write(`\n[${new Date().toISOString()}] TIMEOUT: killed after ${JOB_TIMEOUT_MS / 3600000}h\n`);
|
|
killProcessGroup(child, 'SIGTERM');
|
|
setTimeout(() => {
|
|
if (runningJobs.has(typeKey)) {
|
|
killProcessGroup(child, 'SIGKILL');
|
|
}
|
|
}, 10_000);
|
|
}, JOB_TIMEOUT_MS);
|
|
|
|
child.on('close', async (code: number | null) => {
|
|
clearTimeout(timeout);
|
|
runningJobs.delete(typeKey);
|
|
const footer = `\n[${new Date().toISOString()}] Finished ${typeKey}: exit code ${code}\n`;
|
|
logStream.write(footer);
|
|
logStream.end();
|
|
|
|
if (code === 0) {
|
|
log(`Job ${typeKey} (${jobId}) completed successfully`);
|
|
} else {
|
|
logError(`Job ${typeKey} (${jobId}) failed with exit code ${code}`);
|
|
// Mark as failed if the script didn't do it itself
|
|
try {
|
|
const job = await prisma.backgroundJob.findUnique({ where: { id: jobId } });
|
|
if (job && job.status === 'running') {
|
|
const stderrTail = stderrBuffer.join('\n').slice(-500);
|
|
const errorMsg = stderrTail
|
|
? `Exit code: ${code}\n${stderrTail}`
|
|
: `Exit code: ${code}`;
|
|
await prisma.backgroundJob.update({
|
|
where: { id: jobId },
|
|
data: { status: 'failed', error: errorMsg, completedAt: new Date() },
|
|
});
|
|
}
|
|
} catch (err) {
|
|
logError(`Failed to update job status: ${err}`);
|
|
}
|
|
}
|
|
|
|
// Advance pipeline regardless of success/failure
|
|
// Failed languages are logged but pipeline continues
|
|
if (isPipelineJob) onJobCompleted();
|
|
});
|
|
|
|
child.on('error', async (err: Error) => {
|
|
clearTimeout(timeout);
|
|
runningJobs.delete(typeKey);
|
|
logStream.write(`\n[${new Date().toISOString()}] ERROR: ${err.message}\n`);
|
|
logStream.end();
|
|
logError(`Job ${typeKey} (${jobId}) failed to start: ${err.message}`);
|
|
|
|
try {
|
|
await prisma.backgroundJob.update({
|
|
where: { id: jobId },
|
|
data: { status: 'failed', error: err.message, completedAt: new Date() },
|
|
});
|
|
} catch (updateErr) {
|
|
logError(`Failed to update job status: ${updateErr}`);
|
|
}
|
|
|
|
// Advance pipeline even on error
|
|
if (isPipelineJob) onJobCompleted();
|
|
});
|
|
}
|
|
|
|
// ─── Graceful Shutdown ───────────────────────────────────────────────────────
|
|
|
|
async function shutdown(signal: string): Promise<void> {
|
|
log(`Received ${signal}, shutting down gracefully...`);
|
|
|
|
// Mark running jobs as pending so they can be resumed on restart
|
|
for (const [name, { process: child, jobId }] of runningJobs) {
|
|
log(`Stopping running job: ${name} (${jobId})`);
|
|
killProcessGroup(child, 'SIGTERM');
|
|
|
|
// Re-queue the job for the next startup
|
|
try {
|
|
await prisma.backgroundJob.update({
|
|
where: { id: jobId },
|
|
data: { status: 'pending', error: null },
|
|
});
|
|
log(`Re-queued job ${jobId} for restart`);
|
|
} catch (err) {
|
|
logError(`Failed to re-queue job ${jobId}: ${err}`);
|
|
}
|
|
}
|
|
|
|
// Give jobs 10 seconds to finish, then force exit
|
|
setTimeout(() => {
|
|
if (runningJobs.size > 0) {
|
|
logError(`Force killing ${runningJobs.size} remaining jobs`);
|
|
for (const [, { process: child }] of runningJobs) {
|
|
killProcessGroup(child, 'SIGKILL');
|
|
}
|
|
}
|
|
prisma.$disconnect().then(() => pool.end()).then(() => process.exit(0));
|
|
}, 10_000);
|
|
|
|
// If no jobs running, exit immediately
|
|
if (runningJobs.size === 0) {
|
|
prisma.$disconnect().then(() => pool.end()).then(() => process.exit(0));
|
|
}
|
|
}
|
|
|
|
process.on('SIGTERM', () => shutdown('SIGTERM'));
|
|
process.on('SIGINT', () => shutdown('SIGINT'));
|
|
|
|
// ─── Main ────────────────────────────────────────────────────────────────────
|
|
|
|
async function main(): Promise<void> {
|
|
log('=== NearestMass Scheduler Starting (Continuous Pipeline) ===');
|
|
|
|
validateEnvironment();
|
|
ensureLogsDir();
|
|
|
|
// Crash recovery: mark orphaned jobs as failed
|
|
await recoverFromCrash();
|
|
|
|
// Log cleanup at 03:00 UTC
|
|
cron.schedule('0 3 * * *', () => cleanOldLogs(), { timezone: 'UTC' });
|
|
log('Registered cron job: log-cleanup (0 3 * * * UTC)');
|
|
|
|
// Stale job cleanup every 6 hours
|
|
cron.schedule('0 */6 * * *', () => cleanStaleJobs(), { timezone: 'UTC' });
|
|
log('Registered cron job: stale-job-cleanup (every 6h)');
|
|
|
|
// Heartbeat every hour — logs cycle state and writes heartbeat file for Docker healthcheck
|
|
cron.schedule('0 * * * *', () => {
|
|
const currentGroup = cycleState.currentGroupIndex < PIPELINE_GROUPS.length
|
|
? PIPELINE_GROUPS[cycleState.currentGroupIndex].name
|
|
: 'none';
|
|
const jobs = runningJobs.size > 0
|
|
? `Running: ${[...runningJobs.keys()].join(', ')}`
|
|
: 'No jobs running';
|
|
const state = cycleState.waitingForCooldown
|
|
? 'cooldown'
|
|
: `group ${cycleState.currentGroupIndex + 1}/${PIPELINE_GROUPS.length} (${currentGroup})`;
|
|
log(`Heartbeat: Cycle ${cycleState.cycleNumber + 1}, ${state}. ${jobs}`);
|
|
fs.writeFileSync(path.join(LOGS_DIR, 'scheduler.heartbeat'), new Date().toISOString());
|
|
}, { timezone: 'UTC' });
|
|
log('Registered cron job: heartbeat (hourly)');
|
|
|
|
// Poll and advance pipeline every 30 seconds
|
|
setInterval(() => {
|
|
pollAndAdvancePipeline();
|
|
}, JOB_POLL_INTERVAL_MS);
|
|
log(`Pipeline polling every ${JOB_POLL_INTERVAL_MS / 1000}s`);
|
|
|
|
// Start pipeline immediately
|
|
await pollAndAdvancePipeline();
|
|
|
|
log('=== Scheduler running (parallel grouped pipeline) ===');
|
|
log(`Pipeline groups (${PIPELINE_GROUPS.length}):`);
|
|
for (let i = 0; i < PIPELINE_GROUPS.length; i++) {
|
|
const g = PIPELINE_GROUPS[i];
|
|
const phaseNames = g.phases.map(p => p.name).join(', ');
|
|
log(` ${i + 1}. ${g.name} [${g.mode}]: ${phaseNames}`);
|
|
}
|
|
log(`Cooldown between cycles: ${CYCLE_COOLDOWN_MS / (60 * 60 * 1000)}h`);
|
|
log(`Job timeout: ${JOB_TIMEOUT_MS / (60 * 60 * 1000)}h`);
|
|
}
|
|
|
|
main();
|