|
|
@@ -1,18 +1,32 @@
|
|
|
/**
|
|
|
* Job queue system with persistence
|
|
|
- * Ensures only one job runs at a time and persists queue to disk
|
|
|
+ * Supports both single-queue mode and per-repository parallel queues
|
|
|
*/
|
|
|
import { readFileSync, writeFileSync, existsSync } from 'fs';
|
|
|
import { resolve } from 'path';
|
|
|
import logger from './logger.js';
|
|
|
import commandExecutor from './commandExecutor.js';
|
|
|
+import config from './config.js';
|
|
|
|
|
|
export class JobQueue {
|
|
|
constructor(options = {}) {
|
|
|
this.queueFile = options.queueFile || resolve(process.cwd(), 'queue.json');
|
|
|
- this.queue = [];
|
|
|
- this.isProcessing = false;
|
|
|
- this.currentJob = null;
|
|
|
+
|
|
|
+ // Check if per-repository parallel queues are enabled
|
|
|
+ this.parallelPerRepository = config.getBoolean('QUEUE_PARALLEL_PER_REPOSITORY', false);
|
|
|
+
|
|
|
+ if (this.parallelPerRepository) {
|
|
|
+ // Multi-queue mode: separate queue for each repository
|
|
|
+ this.queues = new Map(); // key: full_repo_name, value: { queue: [], isProcessing: false, currentJob: null }
|
|
|
+ logger.info('Job queue initialized in parallel per-repository mode');
|
|
|
+ } else {
|
|
|
+ // Single-queue mode: original behavior
|
|
|
+ this.queue = [];
|
|
|
+ this.isProcessing = false;
|
|
|
+ this.currentJob = null;
|
|
|
+ logger.info('Job queue initialized in single-queue mode');
|
|
|
+ }
|
|
|
+
|
|
|
this.jobIdCounter = 0;
|
|
|
|
|
|
// Track recently processed issue events to prevent duplicates
|
|
|
@@ -31,27 +45,66 @@ export class JobQueue {
|
|
|
const data = readFileSync(this.queueFile, 'utf-8');
|
|
|
const savedState = JSON.parse(data);
|
|
|
|
|
|
- this.queue = savedState.queue || [];
|
|
|
this.jobIdCounter = savedState.jobIdCounter || 0;
|
|
|
|
|
|
- // Reset any jobs that were running when server stopped
|
|
|
- this.queue.forEach(job => {
|
|
|
- if (job.status === 'running') {
|
|
|
- job.status = 'pending';
|
|
|
- logger.info(`Reset job ${job.id} from running to pending after restart`);
|
|
|
+ if (this.parallelPerRepository) {
|
|
|
+ // Load multi-queue state
|
|
|
+ const queuesData = savedState.queues || {};
|
|
|
+ let totalJobs = 0;
|
|
|
+
|
|
|
+ for (const [repoName, queueData] of Object.entries(queuesData)) {
|
|
|
+ const queue = queueData.queue || [];
|
|
|
+
|
|
|
+ // Reset any jobs that were running when server stopped
|
|
|
+ queue.forEach(job => {
|
|
|
+ if (job.status === 'running') {
|
|
|
+ job.status = 'pending';
|
|
|
+ logger.info(`Reset job ${job.id} in ${repoName} from running to pending after restart`);
|
|
|
+ }
|
|
|
+ });
|
|
|
+
|
|
|
+ this.queues.set(repoName, {
|
|
|
+ queue: queue,
|
|
|
+ isProcessing: false,
|
|
|
+ currentJob: null
|
|
|
+ });
|
|
|
+
|
|
|
+ totalJobs += queue.length;
|
|
|
+
|
|
|
+ // Start processing if there are jobs for this repository
|
|
|
+ if (queue.length > 0) {
|
|
|
+ this.processNext(repoName);
|
|
|
+ }
|
|
|
}
|
|
|
- });
|
|
|
|
|
|
- logger.info(`Loaded ${this.queue.length} jobs from queue file`);
|
|
|
+ logger.info(`Loaded ${totalJobs} jobs from queue file across ${this.queues.size} repositories`);
|
|
|
+ } else {
|
|
|
+ // Load single-queue state
|
|
|
+ this.queue = savedState.queue || [];
|
|
|
+
|
|
|
+ // Reset any jobs that were running when server stopped
|
|
|
+ this.queue.forEach(job => {
|
|
|
+ if (job.status === 'running') {
|
|
|
+ job.status = 'pending';
|
|
|
+ logger.info(`Reset job ${job.id} from running to pending after restart`);
|
|
|
+ }
|
|
|
+ });
|
|
|
+
|
|
|
+ logger.info(`Loaded ${this.queue.length} jobs from queue file`);
|
|
|
|
|
|
- // Start processing if there are jobs
|
|
|
- if (this.queue.length > 0) {
|
|
|
- this.processNext();
|
|
|
+ // Start processing if there are jobs
|
|
|
+ if (this.queue.length > 0) {
|
|
|
+ this.processNext();
|
|
|
+ }
|
|
|
}
|
|
|
}
|
|
|
} catch (error) {
|
|
|
logger.error('Failed to load queue from file', error);
|
|
|
- this.queue = [];
|
|
|
+ if (this.parallelPerRepository) {
|
|
|
+ this.queues.clear();
|
|
|
+ } else {
|
|
|
+ this.queue = [];
|
|
|
+ }
|
|
|
this.jobIdCounter = 0;
|
|
|
}
|
|
|
}
|
|
|
@@ -62,17 +115,54 @@ export class JobQueue {
|
|
|
saveQueue() {
|
|
|
try {
|
|
|
const state = {
|
|
|
- queue: this.queue,
|
|
|
jobIdCounter: this.jobIdCounter,
|
|
|
lastSaved: new Date().toISOString()
|
|
|
};
|
|
|
|
|
|
+ if (this.parallelPerRepository) {
|
|
|
+ // Save multi-queue state
|
|
|
+ state.queues = {};
|
|
|
+ for (const [repoName, queueData] of this.queues.entries()) {
|
|
|
+ state.queues[repoName] = {
|
|
|
+ queue: queueData.queue
|
|
|
+ };
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ // Save single-queue state
|
|
|
+ state.queue = this.queue;
|
|
|
+ }
|
|
|
+
|
|
|
writeFileSync(this.queueFile, JSON.stringify(state, null, 2), 'utf-8');
|
|
|
} catch (error) {
|
|
|
logger.error('Failed to save queue to file', error);
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ /**
|
|
|
+ * Get repository name from payload
|
|
|
+ * @param {object} payload - The webhook payload
|
|
|
+ * @returns {string} Repository full name (owner/repo)
|
|
|
+ */
|
|
|
+ getRepositoryName(payload) {
|
|
|
+ return payload.repository?.full_name || payload.repository?.name || 'unknown';
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Get or create queue for a repository (multi-queue mode only)
|
|
|
+ * @param {string} repoName - Repository full name
|
|
|
+ * @returns {object} Queue data { queue: [], isProcessing: false, currentJob: null }
|
|
|
+ */
|
|
|
+ getOrCreateQueue(repoName) {
|
|
|
+ if (!this.queues.has(repoName)) {
|
|
|
+ this.queues.set(repoName, {
|
|
|
+ queue: [],
|
|
|
+ isProcessing: false,
|
|
|
+ currentJob: null
|
|
|
+ });
|
|
|
+ }
|
|
|
+ return this.queues.get(repoName);
|
|
|
+ }
|
|
|
+
|
|
|
/**
|
|
|
* Add a job to the queue
|
|
|
* @param {string} eventType - The webhook event type
|
|
|
@@ -125,24 +215,48 @@ export class JobQueue {
|
|
|
result: null
|
|
|
};
|
|
|
|
|
|
- this.queue.push(job);
|
|
|
- this.saveQueue();
|
|
|
-
|
|
|
// Extract issue number for logging if this is an issue-related event
|
|
|
const issueNumber = payload.issue?.number || payload.number || null;
|
|
|
const commandName = config.name || config.command || 'unknown';
|
|
|
|
|
|
- // Use structured logging
|
|
|
- logger.jobQueued(job.id, eventType, issueNumber, commandName);
|
|
|
+ if (this.parallelPerRepository) {
|
|
|
+ // Multi-queue mode: add to repository-specific queue
|
|
|
+ const repoName = this.getRepositoryName(payload);
|
|
|
+ const queueData = this.getOrCreateQueue(repoName);
|
|
|
+
|
|
|
+ queueData.queue.push(job);
|
|
|
+ this.saveQueue();
|
|
|
+
|
|
|
+ // Use structured logging
|
|
|
+ logger.jobQueued(job.id, eventType, issueNumber, commandName);
|
|
|
+
|
|
|
+ logger.info(`Job ${job.id} added to queue for ${repoName} (${eventType})`, {
|
|
|
+ queueLength: queueData.queue.length,
|
|
|
+ jobId: job.id,
|
|
|
+ repository: repoName
|
|
|
+ });
|
|
|
|
|
|
- logger.info(`Job ${job.id} added to queue (${eventType})`, {
|
|
|
- queueLength: this.queue.length,
|
|
|
- jobId: job.id
|
|
|
- });
|
|
|
+ // Start processing if not already processing for this repository
|
|
|
+ if (!queueData.isProcessing) {
|
|
|
+ this.processNext(repoName);
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ // Single-queue mode: original behavior
|
|
|
+ this.queue.push(job);
|
|
|
+ this.saveQueue();
|
|
|
|
|
|
- // Start processing if not already processing
|
|
|
- if (!this.isProcessing) {
|
|
|
- this.processNext();
|
|
|
+ // Use structured logging
|
|
|
+ logger.jobQueued(job.id, eventType, issueNumber, commandName);
|
|
|
+
|
|
|
+ logger.info(`Job ${job.id} added to queue (${eventType})`, {
|
|
|
+ queueLength: this.queue.length,
|
|
|
+ jobId: job.id
|
|
|
+ });
|
|
|
+
|
|
|
+ // Start processing if not already processing
|
|
|
+ if (!this.isProcessing) {
|
|
|
+ this.processNext();
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
return job;
|
|
|
@@ -150,85 +264,185 @@ export class JobQueue {
|
|
|
|
|
|
/**
|
|
|
* Process the next job in the queue
|
|
|
+ * @param {string} repoName - Repository name (required in multi-queue mode)
|
|
|
*/
|
|
|
- async processNext() {
|
|
|
- // If already processing or queue is empty, do nothing
|
|
|
- if (this.isProcessing || this.queue.length === 0) {
|
|
|
- return;
|
|
|
- }
|
|
|
+ async processNext(repoName = null) {
|
|
|
+ if (this.parallelPerRepository) {
|
|
|
+ // Multi-queue mode: process jobs for specific repository
|
|
|
+ if (!repoName) {
|
|
|
+ logger.error('Repository name required in parallel per-repository mode');
|
|
|
+ return;
|
|
|
+ }
|
|
|
|
|
|
- this.isProcessing = true;
|
|
|
- const job = this.queue[0]; // Get first job (FIFO)
|
|
|
- this.currentJob = job;
|
|
|
+ const queueData = this.queues.get(repoName);
|
|
|
+ if (!queueData) {
|
|
|
+ logger.error(`Queue not found for repository: ${repoName}`);
|
|
|
+ return;
|
|
|
+ }
|
|
|
|
|
|
- // Update job status
|
|
|
- job.status = 'running';
|
|
|
- job.startedAt = new Date().toISOString();
|
|
|
- this.saveQueue();
|
|
|
+ // If already processing or queue is empty, do nothing
|
|
|
+ if (queueData.isProcessing || queueData.queue.length === 0) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
|
|
|
- // Extract issue number and command name for logging
|
|
|
- const issueNumber = job.payload.issue?.number || job.payload.number || null;
|
|
|
- const commandName = job.config.name || job.config.command || 'unknown';
|
|
|
+ queueData.isProcessing = true;
|
|
|
+ const job = queueData.queue[0]; // Get first job (FIFO)
|
|
|
+ queueData.currentJob = job;
|
|
|
|
|
|
- // Use structured logging
|
|
|
- logger.jobStarted(job.id, job.eventType, issueNumber, commandName);
|
|
|
+ // Update job status
|
|
|
+ job.status = 'running';
|
|
|
+ job.startedAt = new Date().toISOString();
|
|
|
+ this.saveQueue();
|
|
|
|
|
|
- logger.info(`Starting job ${job.id} (${job.eventType})`, {
|
|
|
- queueLength: this.queue.length,
|
|
|
- jobId: job.id
|
|
|
- });
|
|
|
+ // Extract issue number and command name for logging
|
|
|
+ const issueNumber = job.payload.issue?.number || job.payload.number || null;
|
|
|
+ const commandName = job.config.name || job.config.command || 'unknown';
|
|
|
|
|
|
- try {
|
|
|
- // Execute the webhook command
|
|
|
- const result = await commandExecutor.executeWebhookCommand(
|
|
|
- job.eventType,
|
|
|
- job.payload,
|
|
|
- job.headers,
|
|
|
- job.config
|
|
|
- );
|
|
|
+ // Use structured logging
|
|
|
+ logger.jobStarted(job.id, job.eventType, issueNumber, commandName);
|
|
|
+
|
|
|
+ logger.info(`Starting job ${job.id} for ${repoName} (${job.eventType})`, {
|
|
|
+ queueLength: queueData.queue.length,
|
|
|
+ jobId: job.id,
|
|
|
+ repository: repoName
|
|
|
+ });
|
|
|
+
|
|
|
+ try {
|
|
|
+ // Execute the webhook command
|
|
|
+ const result = await commandExecutor.executeWebhookCommand(
|
|
|
+ job.eventType,
|
|
|
+ job.payload,
|
|
|
+ job.headers,
|
|
|
+ job.config
|
|
|
+ );
|
|
|
+
|
|
|
+ // Update job with result
|
|
|
+ job.status = 'completed';
|
|
|
+ job.completedAt = new Date().toISOString();
|
|
|
+ job.result = result;
|
|
|
+
|
|
|
+ const durationMs = new Date(job.completedAt) - new Date(job.startedAt);
|
|
|
+
|
|
|
+ // Use structured logging
|
|
|
+ logger.jobCompleted(job.id, job.eventType, issueNumber, durationMs);
|
|
|
+
|
|
|
+ logger.info(`Job ${job.id} for ${repoName} completed successfully`, {
|
|
|
+ jobId: job.id,
|
|
|
+ duration: durationMs,
|
|
|
+ repository: repoName
|
|
|
+ });
|
|
|
+ } catch (error) {
|
|
|
+ // Update job with error
|
|
|
+ job.status = 'failed';
|
|
|
+ job.completedAt = new Date().toISOString();
|
|
|
+ job.result = {
|
|
|
+ success: false,
|
|
|
+ error: error.message
|
|
|
+ };
|
|
|
+
|
|
|
+ // Use structured logging
|
|
|
+ logger.jobFailed(job.id, job.eventType, issueNumber, error.message);
|
|
|
+
|
|
|
+ logger.error(`Job ${job.id} for ${repoName} failed`, error);
|
|
|
+ }
|
|
|
+
|
|
|
+ // Remove completed job from queue
|
|
|
+ queueData.queue.shift();
|
|
|
+ queueData.currentJob = null;
|
|
|
+ queueData.isProcessing = false;
|
|
|
+ this.saveQueue();
|
|
|
+
|
|
|
+ // Process next job if queue is not empty
|
|
|
+ if (queueData.queue.length > 0) {
|
|
|
+ logger.info(`${queueData.queue.length} jobs remaining in queue for ${repoName}`);
|
|
|
+ // Use setImmediate to avoid deep recursion
|
|
|
+ setImmediate(() => this.processNext(repoName));
|
|
|
+ } else {
|
|
|
+ logger.info(`Queue is empty for ${repoName}`);
|
|
|
+ // Remove empty queue to keep memory clean
|
|
|
+ this.queues.delete(repoName);
|
|
|
+ this.saveQueue();
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ // Single-queue mode: original behavior
|
|
|
+ // If already processing or queue is empty, do nothing
|
|
|
+ if (this.isProcessing || this.queue.length === 0) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ this.isProcessing = true;
|
|
|
+ const job = this.queue[0]; // Get first job (FIFO)
|
|
|
+ this.currentJob = job;
|
|
|
|
|
|
- // Update job with result
|
|
|
- job.status = 'completed';
|
|
|
- job.completedAt = new Date().toISOString();
|
|
|
- job.result = result;
|
|
|
+ // Update job status
|
|
|
+ job.status = 'running';
|
|
|
+ job.startedAt = new Date().toISOString();
|
|
|
+ this.saveQueue();
|
|
|
|
|
|
- const durationMs = new Date(job.completedAt) - new Date(job.startedAt);
|
|
|
+ // Extract issue number and command name for logging
|
|
|
+ const issueNumber = job.payload.issue?.number || job.payload.number || null;
|
|
|
+ const commandName = job.config.name || job.config.command || 'unknown';
|
|
|
|
|
|
// Use structured logging
|
|
|
- logger.jobCompleted(job.id, job.eventType, issueNumber, durationMs);
|
|
|
+ logger.jobStarted(job.id, job.eventType, issueNumber, commandName);
|
|
|
|
|
|
- logger.info(`Job ${job.id} completed successfully`, {
|
|
|
- jobId: job.id,
|
|
|
- duration: durationMs
|
|
|
+ logger.info(`Starting job ${job.id} (${job.eventType})`, {
|
|
|
+ queueLength: this.queue.length,
|
|
|
+ jobId: job.id
|
|
|
});
|
|
|
- } catch (error) {
|
|
|
- // Update job with error
|
|
|
- job.status = 'failed';
|
|
|
- job.completedAt = new Date().toISOString();
|
|
|
- job.result = {
|
|
|
- success: false,
|
|
|
- error: error.message
|
|
|
- };
|
|
|
|
|
|
- // Use structured logging
|
|
|
- logger.jobFailed(job.id, job.eventType, issueNumber, error.message);
|
|
|
+ try {
|
|
|
+ // Execute the webhook command
|
|
|
+ const result = await commandExecutor.executeWebhookCommand(
|
|
|
+ job.eventType,
|
|
|
+ job.payload,
|
|
|
+ job.headers,
|
|
|
+ job.config
|
|
|
+ );
|
|
|
|
|
|
- logger.error(`Job ${job.id} failed`, error);
|
|
|
- }
|
|
|
+ // Update job with result
|
|
|
+ job.status = 'completed';
|
|
|
+ job.completedAt = new Date().toISOString();
|
|
|
+ job.result = result;
|
|
|
|
|
|
- // Remove completed job from queue
|
|
|
- this.queue.shift();
|
|
|
- this.currentJob = null;
|
|
|
- this.isProcessing = false;
|
|
|
- this.saveQueue();
|
|
|
+ const durationMs = new Date(job.completedAt) - new Date(job.startedAt);
|
|
|
|
|
|
- // Process next job if queue is not empty
|
|
|
- if (this.queue.length > 0) {
|
|
|
- logger.info(`${this.queue.length} jobs remaining in queue`);
|
|
|
- // Use setImmediate to avoid deep recursion
|
|
|
- setImmediate(() => this.processNext());
|
|
|
- } else {
|
|
|
- logger.info('Queue is empty');
|
|
|
+ // Use structured logging
|
|
|
+ logger.jobCompleted(job.id, job.eventType, issueNumber, durationMs);
|
|
|
+
|
|
|
+ logger.info(`Job ${job.id} completed successfully`, {
|
|
|
+ jobId: job.id,
|
|
|
+ duration: durationMs
|
|
|
+ });
|
|
|
+ } catch (error) {
|
|
|
+ // Update job with error
|
|
|
+ job.status = 'failed';
|
|
|
+ job.completedAt = new Date().toISOString();
|
|
|
+ job.result = {
|
|
|
+ success: false,
|
|
|
+ error: error.message
|
|
|
+ };
|
|
|
+
|
|
|
+ // Use structured logging
|
|
|
+ logger.jobFailed(job.id, job.eventType, issueNumber, error.message);
|
|
|
+
|
|
|
+ logger.error(`Job ${job.id} failed`, error);
|
|
|
+ }
|
|
|
+
|
|
|
+ // Remove completed job from queue
|
|
|
+ this.queue.shift();
|
|
|
+ this.currentJob = null;
|
|
|
+ this.isProcessing = false;
|
|
|
+ this.saveQueue();
|
|
|
+
|
|
|
+ // Process next job if queue is not empty
|
|
|
+ if (this.queue.length > 0) {
|
|
|
+ logger.info(`${this.queue.length} jobs remaining in queue`);
|
|
|
+ // Use setImmediate to avoid deep recursion
|
|
|
+ setImmediate(() => this.processNext());
|
|
|
+ } else {
|
|
|
+ logger.info('Queue is empty');
|
|
|
+ }
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -236,46 +450,133 @@ export class JobQueue {
|
|
|
* Get queue statistics
|
|
|
*/
|
|
|
getStats() {
|
|
|
- return {
|
|
|
- queueLength: this.queue.length,
|
|
|
- isProcessing: this.isProcessing,
|
|
|
- currentJob: this.currentJob ? {
|
|
|
- id: this.currentJob.id,
|
|
|
- eventType: this.currentJob.eventType,
|
|
|
- status: this.currentJob.status,
|
|
|
- startedAt: this.currentJob.startedAt
|
|
|
- } : null,
|
|
|
- pendingJobs: this.queue.filter(j => j.status === 'pending').length,
|
|
|
- runningJobs: this.queue.filter(j => j.status === 'running').length
|
|
|
- };
|
|
|
+ if (this.parallelPerRepository) {
|
|
|
+ // Multi-queue mode: aggregate stats across all repositories
|
|
|
+ const repositoryStats = {};
|
|
|
+ let totalJobs = 0;
|
|
|
+ let totalPending = 0;
|
|
|
+ let totalRunning = 0;
|
|
|
+
|
|
|
+ for (const [repoName, queueData] of this.queues.entries()) {
|
|
|
+ const pending = queueData.queue.filter(j => j.status === 'pending').length;
|
|
|
+ const running = queueData.queue.filter(j => j.status === 'running').length;
|
|
|
+
|
|
|
+ repositoryStats[repoName] = {
|
|
|
+ queueLength: queueData.queue.length,
|
|
|
+ isProcessing: queueData.isProcessing,
|
|
|
+ currentJob: queueData.currentJob ? {
|
|
|
+ id: queueData.currentJob.id,
|
|
|
+ eventType: queueData.currentJob.eventType,
|
|
|
+ status: queueData.currentJob.status,
|
|
|
+ startedAt: queueData.currentJob.startedAt
|
|
|
+ } : null,
|
|
|
+ pendingJobs: pending,
|
|
|
+ runningJobs: running
|
|
|
+ };
|
|
|
+
|
|
|
+ totalJobs += queueData.queue.length;
|
|
|
+ totalPending += pending;
|
|
|
+ totalRunning += running;
|
|
|
+ }
|
|
|
+
|
|
|
+ return {
|
|
|
+ mode: 'parallel-per-repository',
|
|
|
+ repositories: repositoryStats,
|
|
|
+ totalRepositories: this.queues.size,
|
|
|
+ totalJobs: totalJobs,
|
|
|
+ totalPending: totalPending,
|
|
|
+ totalRunning: totalRunning
|
|
|
+ };
|
|
|
+ } else {
|
|
|
+ // Single-queue mode: original stats
|
|
|
+ return {
|
|
|
+ mode: 'single-queue',
|
|
|
+ queueLength: this.queue.length,
|
|
|
+ isProcessing: this.isProcessing,
|
|
|
+ currentJob: this.currentJob ? {
|
|
|
+ id: this.currentJob.id,
|
|
|
+ eventType: this.currentJob.eventType,
|
|
|
+ status: this.currentJob.status,
|
|
|
+ startedAt: this.currentJob.startedAt
|
|
|
+ } : null,
|
|
|
+ pendingJobs: this.queue.filter(j => j.status === 'pending').length,
|
|
|
+ runningJobs: this.queue.filter(j => j.status === 'running').length
|
|
|
+ };
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* Get all jobs in the queue
|
|
|
*/
|
|
|
getQueue() {
|
|
|
- return [...this.queue];
|
|
|
+ if (this.parallelPerRepository) {
|
|
|
+ // Multi-queue mode: return all jobs from all repositories
|
|
|
+ const allJobs = [];
|
|
|
+ for (const [repoName, queueData] of this.queues.entries()) {
|
|
|
+ for (const job of queueData.queue) {
|
|
|
+ allJobs.push({ ...job, repository: repoName });
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return allJobs;
|
|
|
+ } else {
|
|
|
+ // Single-queue mode: return all jobs
|
|
|
+ return [...this.queue];
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* Clear completed jobs from history (if we decide to keep history)
|
|
|
*/
|
|
|
clearCompleted() {
|
|
|
- const beforeLength = this.queue.length;
|
|
|
- this.queue = this.queue.filter(j => j.status !== 'completed');
|
|
|
- this.saveQueue();
|
|
|
-
|
|
|
- logger.info(`Cleared ${beforeLength - this.queue.length} completed jobs`);
|
|
|
+ if (this.parallelPerRepository) {
|
|
|
+ // Multi-queue mode: clear completed jobs from all repositories
|
|
|
+ let totalCleared = 0;
|
|
|
+ for (const [repoName, queueData] of this.queues.entries()) {
|
|
|
+ const beforeLength = queueData.queue.length;
|
|
|
+ queueData.queue = queueData.queue.filter(j => j.status !== 'completed');
|
|
|
+ totalCleared += beforeLength - queueData.queue.length;
|
|
|
+ }
|
|
|
+ this.saveQueue();
|
|
|
+ logger.info(`Cleared ${totalCleared} completed jobs across all repositories`);
|
|
|
+ } else {
|
|
|
+ // Single-queue mode: original behavior
|
|
|
+ const beforeLength = this.queue.length;
|
|
|
+ this.queue = this.queue.filter(j => j.status !== 'completed');
|
|
|
+ this.saveQueue();
|
|
|
+ logger.info(`Cleared ${beforeLength - this.queue.length} completed jobs`);
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* Graceful shutdown - save queue state
|
|
|
*/
|
|
|
async shutdown() {
|
|
|
- logger.info('Shutting down job queue', {
|
|
|
- remainingJobs: this.queue.length,
|
|
|
- currentJob: this.currentJob?.id
|
|
|
- });
|
|
|
+ if (this.parallelPerRepository) {
|
|
|
+ // Multi-queue mode: save all queues
|
|
|
+ let totalJobs = 0;
|
|
|
+ const runningJobs = [];
|
|
|
+
|
|
|
+ for (const [repoName, queueData] of this.queues.entries()) {
|
|
|
+ totalJobs += queueData.queue.length;
|
|
|
+ if (queueData.currentJob) {
|
|
|
+ runningJobs.push({ repo: repoName, jobId: queueData.currentJob.id });
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ logger.info('Shutting down job queue', {
|
|
|
+ mode: 'parallel-per-repository',
|
|
|
+ totalRepositories: this.queues.size,
|
|
|
+ remainingJobs: totalJobs,
|
|
|
+ runningJobs: runningJobs
|
|
|
+ });
|
|
|
+ } else {
|
|
|
+ // Single-queue mode: original behavior
|
|
|
+ logger.info('Shutting down job queue', {
|
|
|
+ mode: 'single-queue',
|
|
|
+ remainingJobs: this.queue.length,
|
|
|
+ currentJob: this.currentJob?.id
|
|
|
+ });
|
|
|
+ }
|
|
|
|
|
|
this.saveQueue();
|
|
|
|
