bo/worklenz
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(recurring-tasks): implement recurring tasks service with timezone support and notifications

Browse Source 
- Added a new service for managing recurring tasks, allowing configuration of task schedules with timezone support.
- Introduced job queues for processing recurring tasks and handling task creation in bulk.
- Implemented notification system to alert users about newly created recurring tasks, including email and in-app notifications.
- Enhanced database schema with new tables for notifications and audit logs to track recurring task operations.
- Updated frontend components to support timezone selection and manage excluded dates for recurring tasks.
- Refactored existing code to integrate new features and improve overall task management experience.
This commit is contained in:
chamikaJ
2025-07-20 19:16:03 +05:30
parent a112d39321
commit 474f1afe66
21 changed files with 2771 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
worklenz-backend/src/config/recurring-tasks-config.ts Normal file
View File

@@ -0,0 +1,57 @@
export interface RecurringTasksConfig {
enabled: boolean;
mode: 'cron' | 'queue';
cronInterval: string;
redisConfig: {
host: string;
port: number;
password?: string;
db: number;
};
queueOptions: {
maxConcurrency: number;
retryAttempts: number;
retryDelay: number;
};
notifications: {
enabled: boolean;
email: boolean;
push: boolean;
inApp: boolean;
};
auditLog: {
enabled: boolean;
retentionDays: number;
};
}
export const recurringTasksConfig: RecurringTasksConfig = {
enabled: process.env.RECURRING_TASKS_ENABLED !== 'false',
mode: (process.env.RECURRING_TASKS_MODE as 'cron' | 'queue') || 'cron',
cronInterval: process.env.RECURRING_JOBS_INTERVAL || '0 * * * *',
redisConfig: {
host: process.env.REDIS_HOST || 'localhost',
port: parseInt(process.env.REDIS_PORT || '6379'),
password: process.env.REDIS_PASSWORD,
db: parseInt(process.env.REDIS_DB || '0'),
},
queueOptions: {
maxConcurrency: parseInt(process.env.RECURRING_TASKS_MAX_CONCURRENCY || '5'),
retryAttempts: parseInt(process.env.RECURRING_TASKS_RETRY_ATTEMPTS || '3'),
retryDelay: parseInt(process.env.RECURRING_TASKS_RETRY_DELAY || '2000'),
},
notifications: {
enabled: process.env.RECURRING_TASKS_NOTIFICATIONS_ENABLED !== 'false',
email: process.env.RECURRING_TASKS_EMAIL_NOTIFICATIONS !== 'false',
push: process.env.RECURRING_TASKS_PUSH_NOTIFICATIONS !== 'false',
inApp: process.env.RECURRING_TASKS_IN_APP_NOTIFICATIONS !== 'false',
},
auditLog: {
enabled: process.env.RECURRING_TASKS_AUDIT_LOG_ENABLED !== 'false',
retentionDays: parseInt(process.env.RECURRING_TASKS_AUDIT_RETENTION_DAYS || '90'),
},
};

48
worklenz-backend/src/controllers/recurring-tasks-admin-controller.ts Normal file
View File

@@ -0,0 +1,48 @@
import { IWorkLenzRequest } from "../interfaces/worklenz-request";
import { IWorkLenzResponse } from "../interfaces/worklenz-response";
import { ServerResponse } from "../models/server-response";
import WorklenzControllerBase from "./worklenz-controller-base";
import HandleExceptions from "../decorators/handle-exceptions";
import { RecurringTasksPermissions } from "../utils/recurring-tasks-permissions";
import { RecurringTasksAuditLogger } from "../utils/recurring-tasks-audit-logger";
export default class RecurringTasksAdminController extends WorklenzControllerBase {
/**
* Get templates with permission issues
*/
@HandleExceptions()
public static async getPermissionIssues(req: IWorkLenzRequest, res: IWorkLenzResponse): Promise<IWorkLenzResponse> {
const issues = await RecurringTasksPermissions.getTemplatesWithPermissionIssues();
return res.status(200).send(new ServerResponse(true, issues));
}
/**
* Get audit log summary
*/
@HandleExceptions()
public static async getAuditSummary(req: IWorkLenzRequest, res: IWorkLenzResponse): Promise<IWorkLenzResponse> {
const { days = 7 } = req.query;
const summary = await RecurringTasksAuditLogger.getAuditSummary(Number(days));
return res.status(200).send(new ServerResponse(true, summary));
}
/**
* Get recent errors from audit log
*/
@HandleExceptions()
public static async getRecentErrors(req: IWorkLenzRequest, res: IWorkLenzResponse): Promise<IWorkLenzResponse> {
const { limit = 10 } = req.query;
const errors = await RecurringTasksAuditLogger.getRecentErrors(Number(limit));
return res.status(200).send(new ServerResponse(true, errors));
}
/**
* Validate a specific template
*/
@HandleExceptions()
public static async validateTemplate(req: IWorkLenzRequest, res: IWorkLenzResponse): Promise<IWorkLenzResponse> {
const { templateId } = req.params;
const result = await RecurringTasksPermissions.validateTemplatePermissions(templateId);
return res.status(200).send(new ServerResponse(true, result));
}
}

40
worklenz-backend/src/controllers/task-recurring-controller.ts
View File

@@ -6,6 +6,7 @@ import { IWorkLenzRequest } from "../interfaces/worklenz-request";
import { IWorkLenzResponse } from "../interfaces/worklenz-response";
import { ServerResponse } from "../models/server-response";
import { calculateNextEndDate, log_error } from "../shared/utils";
import { RecurringTasksAuditLogger, RecurringTaskOperationType } from "../utils/recurring-tasks-audit-logger";
export default class TaskRecurringController extends WorklenzControllerBase {
@HandleExceptions()
@@ -34,7 +35,7 @@ export default class TaskRecurringController extends WorklenzControllerBase {
}
@HandleExceptions()
public static async createTaskSchedule(taskId: string) {
public static async createTaskSchedule(taskId: string, userId?: string) {
const q = `INSERT INTO task_recurring_schedules (schedule_type) VALUES ('daily') RETURNING id, schedule_type;`;
const result = await db.query(q, []);
const [data] = result.rows;
@@ -44,6 +45,15 @@ export default class TaskRecurringController extends WorklenzControllerBase {
await TaskRecurringController.insertTaskRecurringTemplate(taskId, data.id);
// Log schedule creation
await RecurringTasksAuditLogger.logScheduleChange(
RecurringTaskOperationType.SCHEDULE_CREATED,
data.id,
taskId,
userId,
{ schedule_type: data.schedule_type }
);
return data;
}
@@ -56,9 +66,9 @@ export default class TaskRecurringController extends WorklenzControllerBase {
@HandleExceptions()
public static async updateSchedule(req: IWorkLenzRequest, res: IWorkLenzResponse): Promise<IWorkLenzResponse> {
const { id } = req.params;
const { schedule_type, days_of_week, day_of_month, week_of_month, interval_days, interval_weeks, interval_months, date_of_month } = req.body;
const { schedule_type, days_of_week, day_of_month, week_of_month, interval_days, interval_weeks, interval_months, date_of_month, timezone, end_date, excluded_dates } = req.body;
const deleteQ = `UPDATE task_recurring_schedules
const updateQ = `UPDATE task_recurring_schedules
SET schedule_type = $1,
days_of_week = $2,
date_of_month = $3,
@@ -66,9 +76,27 @@ export default class TaskRecurringController extends WorklenzControllerBase {
week_of_month = $5,
interval_days = $6,
interval_weeks = $7,
interval_months = $8
WHERE id = $9;`;
await db.query(deleteQ, [schedule_type, days_of_week, date_of_month, day_of_month, week_of_month, interval_days, interval_weeks, interval_months, id]);
interval_months = $8,
timezone = COALESCE($9, timezone, 'UTC'),
end_date = $10,
excluded_dates = $11
WHERE id = $12;`;
await db.query(updateQ, [schedule_type, days_of_week, date_of_month, day_of_month, week_of_month, interval_days, interval_weeks, interval_months, timezone, end_date, excluded_dates, id]);
// Log schedule update
await RecurringTasksAuditLogger.logScheduleChange(
RecurringTaskOperationType.SCHEDULE_UPDATED,
id,
undefined,
req.user?.id,
{
schedule_type,
timezone,
end_date,
excluded_dates_count: excluded_dates?.length || 0
}
);
return res.status(200).send(new ServerResponse(true, null));
}

286
worklenz-backend/src/cron_jobs/recurring-tasks.ts
View File

@@ -2,12 +2,16 @@ import { CronJob } from "cron";
import { calculateNextEndDate, log_error } from "../shared/utils";
import db from "../config/db";
import { IRecurringSchedule, ITaskTemplate } from "../interfaces/recurring-tasks";
import moment from "moment";
import moment from "moment-timezone";
import TasksController from "../controllers/tasks-controller";
import { TimezoneUtils } from "../utils/timezone-utils";
import { RetryUtils } from "../utils/retry-utils";
import { RecurringTasksAuditLogger, RecurringTaskOperationType } from "../utils/recurring-tasks-audit-logger";
import { RecurringTasksPermissions } from "../utils/recurring-tasks-permissions";
import { RecurringTasksNotifications } from "../utils/recurring-tasks-notifications";
// At 11:00+00 (4.30pm+530) on every day-of-month if it's on every day-of-week from Monday through Friday.
// const TIME = "0 11 */1 * 1-5";
const TIME = process.env.RECURRING_JOBS_INTERVAL || "0 11 */1 * 1-5";
// Run every hour to process tasks in different timezones
const TIME = process.env.RECURRING_JOBS_INTERVAL || "0 * * * *";
const TIME_FORMAT = "YYYY-MM-DD";
// const TIME = "0 0 * * *"; // Runs at midnight every day
@@ -44,8 +48,129 @@ function getFutureLimit(scheduleType: string, interval?: number): moment.Duratio
}
}
// Helper function to batch create tasks
// Helper function to batch create tasks using bulk operations
async function createBatchTasks(template: ITaskTemplate & IRecurringSchedule, endDates: moment.Moment[]) {
if (endDates.length === 0) return [];
try {
// Prepare bulk task data
const tasksData = endDates.map(endDate => ({
name: template.name,
priority_id: template.priority_id,
project_id: template.project_id,
reporter_id: template.reporter_id,
status_id: template.status_id || null,
end_date: endDate.format(TIME_FORMAT),
schedule_id: template.schedule_id
}));
// Create all tasks in bulk with retry logic
const createTasksResult = await RetryUtils.withDatabaseRetry(async () => {
const createTasksQuery = `SELECT * FROM create_bulk_recurring_tasks($1::JSONB);`;
return await db.query(createTasksQuery, [JSON.stringify(tasksData)]);
}, `create_bulk_recurring_tasks for template ${template.name}`);
const createdTasks = createTasksResult.rows.filter(row => row.created);
const failedTasks = createTasksResult.rows.filter(row => !row.created);
// Log results
if (createdTasks.length > 0) {
console.log(`Created ${createdTasks.length} tasks for template ${template.name}`);
}
if (failedTasks.length > 0) {
failedTasks.forEach(task => {
console.log(`Failed to create task for template ${template.name}: ${task.error_message}`);
});
}
// Only process assignments for successfully created tasks
if (createdTasks.length > 0 && (template.assignees?.length > 0 || template.labels?.length > 0)) {
// Validate assignee permissions
let validAssignees = template.assignees || [];
if (validAssignees.length > 0) {
const invalidAssignees = await RecurringTasksPermissions.validateAssigneePermissions(
validAssignees,
template.project_id
);
if (invalidAssignees.length > 0) {
console.log(`Warning: ${invalidAssignees.length} assignees do not have permissions for project ${template.project_id}`);
// Filter out invalid assignees
validAssignees = validAssignees.filter(
a => !invalidAssignees.includes(a.team_member_id)
);
}
}
// Prepare bulk assignments
const assignments = [];
const labelAssignments = [];
for (const task of createdTasks) {
// Prepare team member assignments with validated assignees
if (validAssignees.length > 0) {
for (const assignee of validAssignees) {
assignments.push({
task_id: task.task_id,
team_member_id: assignee.team_member_id,
assigned_by: assignee.assigned_by
});
}
}
// Prepare label assignments
if (template.labels?.length > 0) {
for (const label of template.labels) {
labelAssignments.push({
task_id: task.task_id,
label_id: label.label_id
});
}
}
}
// Bulk assign team members with retry logic
if (assignments.length > 0) {
await RetryUtils.withDatabaseRetry(async () => {
const assignQuery = `SELECT * FROM bulk_assign_team_members($1::JSONB);`;
return await db.query(assignQuery, [JSON.stringify(assignments)]);
}, `bulk_assign_team_members for template ${template.name}`);
}
// Bulk assign labels with retry logic
if (labelAssignments.length > 0) {
await RetryUtils.withDatabaseRetry(async () => {
const labelQuery = `SELECT * FROM bulk_assign_labels($1::JSONB);`;
return await db.query(labelQuery, [JSON.stringify(labelAssignments)]);
}, `bulk_assign_labels for template ${template.name}`);
}
// Send notifications for created tasks
if (createdTasks.length > 0) {
const taskData = createdTasks.map(task => ({ id: task.task_id, name: task.task_name }));
const assigneeIds = template.assignees?.map(a => a.team_member_id) || [];
await RecurringTasksNotifications.notifyRecurringTasksCreated(
template.name,
template.project_id,
taskData,
assigneeIds,
template.reporter_id
);
}
}
return createdTasks.map(task => ({ id: task.task_id, name: task.task_name }));
} catch (error) {
log_error("Error in bulk task creation:", error);
// Fallback to sequential creation if bulk operation fails
console.log("Falling back to sequential task creation");
return createBatchTasksSequential(template, endDates);
}
}
// Fallback function for sequential task creation
async function createBatchTasksSequential(template: ITaskTemplate & IRecurringSchedule, endDates: moment.Moment[]) {
const createdTasks = [];
for (const nextEndDate of endDates) {
@@ -92,69 +217,162 @@ async function createBatchTasks(template: ITaskTemplate & IRecurringSchedule, en
}
async function onRecurringTaskJobTick() {
const errors: any[] = [];
try {
log("(cron) Recurring tasks job started.");
RecurringTasksAuditLogger.startTimer();
const templatesQuery = `
SELECT t.*, s.*, (SELECT MAX(end_date) FROM tasks WHERE schedule_id = s.id) as last_task_end_date
FROM task_recurring_templates t
JOIN task_recurring_schedules s ON t.schedule_id = s.id;
`;
const templatesResult = await db.query(templatesQuery);
const templates = templatesResult.rows as (ITaskTemplate & IRecurringSchedule)[];
// Get all active timezones where it's currently the scheduled hour
const activeTimezones = TimezoneUtils.getActiveTimezones();
log(`Processing recurring tasks for ${activeTimezones.length} timezones`);
// Fetch templates with retry logic
const templatesResult = await RetryUtils.withDatabaseRetry(async () => {
const templatesQuery = `
SELECT t.*, s.*,
(SELECT MAX(end_date) FROM tasks WHERE schedule_id = s.id) as last_task_end_date,
u.timezone as user_timezone
FROM task_recurring_templates t
JOIN task_recurring_schedules s ON t.schedule_id = s.id
LEFT JOIN tasks orig_task ON t.task_id = orig_task.id
LEFT JOIN users u ON orig_task.reporter_id = u.id
WHERE s.end_date IS NULL OR s.end_date >= CURRENT_DATE;
`;
return await db.query(templatesQuery);
}, "fetch_recurring_templates");
const templates = templatesResult.rows as (ITaskTemplate & IRecurringSchedule & { user_timezone?: string })[];
const now = moment();
let createdTaskCount = 0;
for (const template of templates) {
// Check template permissions before processing
const permissionCheck = await RecurringTasksPermissions.validateTemplatePermissions(template.task_id);
if (!permissionCheck.hasPermission) {
console.log(`Skipping template ${template.name}: ${permissionCheck.reason}`);
// Log permission issue
await RecurringTasksAuditLogger.log({
operationType: RecurringTaskOperationType.TASKS_CREATION_FAILED,
templateId: template.task_id,
scheduleId: template.schedule_id,
templateName: template.name,
success: false,
errorMessage: `Permission denied: ${permissionCheck.reason}`,
details: { permissionCheck }
});
continue;
}
// Use template timezone or user timezone or default to UTC
const timezone = template.timezone || TimezoneUtils.getUserTimezone(template.user_timezone);
// Check if this template should run in the current hour for its timezone
if (!activeTimezones.includes(timezone) && timezone !== 'UTC') {
continue;
}
const now = TimezoneUtils.nowInTimezone(timezone);
const lastTaskEndDate = template.last_task_end_date
? moment(template.last_task_end_date)
: moment(template.created_at);
? moment.tz(template.last_task_end_date, timezone)
: moment.tz(template.created_at, timezone);
// Calculate future limit based on schedule type
const futureLimit = moment(template.last_checked_at || template.created_at)
const futureLimit = moment.tz(template.last_checked_at || template.created_at, timezone)
.add(getFutureLimit(
template.schedule_type,
template.interval_days || template.interval_weeks || template.interval_months || 1
));
let nextEndDate = calculateNextEndDate(template, lastTaskEndDate);
let nextEndDate = TimezoneUtils.calculateNextEndDateWithTimezone(template, lastTaskEndDate, timezone);
const endDatesToCreate: moment.Moment[] = [];
// Find all future occurrences within the limit
while (nextEndDate.isSameOrBefore(futureLimit)) {
if (nextEndDate.isAfter(now)) {
endDatesToCreate.push(moment(nextEndDate));
// Check if date is not in excluded dates
if (!template.excluded_dates || !template.excluded_dates.includes(nextEndDate.format(TIME_FORMAT))) {
endDatesToCreate.push(moment(nextEndDate));
}
}
nextEndDate = calculateNextEndDate(template, nextEndDate);
nextEndDate = TimezoneUtils.calculateNextEndDateWithTimezone(template, nextEndDate, timezone);
}
// Batch create tasks for all future dates
if (endDatesToCreate.length > 0) {
const createdTasks = await createBatchTasks(template, endDatesToCreate);
createdTaskCount += createdTasks.length;
try {
const createdTasks = await createBatchTasks(template, endDatesToCreate);
createdTaskCount += createdTasks.length;
// Update the last_checked_at in the schedule
const updateScheduleQuery = `
UPDATE task_recurring_schedules
SET last_checked_at = $1::DATE,
last_created_task_end_date = $2
WHERE id = $3;
`;
await db.query(updateScheduleQuery, [
moment().format(TIME_FORMAT),
endDatesToCreate[endDatesToCreate.length - 1].format(TIME_FORMAT),
template.schedule_id
]);
// Log successful template processing
await RecurringTasksAuditLogger.logTemplateProcessing(
template.task_id,
template.name,
template.schedule_id,
createdTasks.length,
endDatesToCreate.length - createdTasks.length,
{
timezone,
endDates: endDatesToCreate.map(d => d.format(TIME_FORMAT))
}
);
// Update the last_checked_at in the schedule with retry logic
await RetryUtils.withDatabaseRetry(async () => {
const updateScheduleQuery = `
UPDATE task_recurring_schedules
SET last_checked_at = $1,
last_created_task_end_date = $2
WHERE id = $3;
`;
return await db.query(updateScheduleQuery, [
now.toDate(),
endDatesToCreate[endDatesToCreate.length - 1].toDate(),
template.schedule_id
]);
}, `update_schedule for template ${template.name}`);
} catch (error) {
errors.push({ template: template.name, error });
// Log failed template processing
await RecurringTasksAuditLogger.logTemplateProcessing(
template.task_id,
template.name,
template.schedule_id,
0,
endDatesToCreate.length,
{
timezone,
error: error.message || error.toString()
}
);
}
} else {
console.log(`No tasks created for template ${template.name} - next occurrence is beyond the future limit`);
console.log(`No tasks created for template ${template.name} (${timezone}) - next occurrence is beyond the future limit or excluded`);
}
}
log(`(cron) Recurring tasks job ended with ${createdTaskCount} new tasks created.`);
// Log cron job completion
await RecurringTasksAuditLogger.logCronJobRun(
templates.length,
createdTaskCount,
errors
);
} catch (error) {
log_error(error);
log("(cron) Recurring task job ended with errors.");
// Log cron job failure
await RecurringTasksAuditLogger.log({
operationType: RecurringTaskOperationType.CRON_JOB_ERROR,
success: false,
errorMessage: error.message || error.toString(),
details: { error: error.stack || error }
});
}
}

3
worklenz-backend/src/interfaces/recurring-tasks.ts
View File

@@ -12,6 +12,9 @@ export interface IRecurringSchedule {
last_checked_at: Date | null;
last_task_end_date: Date | null;
created_at: Date;
timezone?: string;
end_date?: Date | null;
excluded_dates?: string[] | null;
}
interface ITaskTemplateAssignee {

322
worklenz-backend/src/jobs/recurring-tasks-queue.ts Normal file
View File

@@ -0,0 +1,322 @@
import Bull from 'bull';
import { TimezoneUtils } from '../utils/timezone-utils';
import { RetryUtils } from '../utils/retry-utils';
import { RecurringTasksAuditLogger, RecurringTaskOperationType } from '../utils/recurring-tasks-audit-logger';
import { RecurringTasksPermissions } from '../utils/recurring-tasks-permissions';
import { RecurringTasksNotifications } from '../utils/recurring-tasks-notifications';
import { calculateNextEndDate, log_error } from '../shared/utils';
import { IRecurringSchedule, ITaskTemplate } from '../interfaces/recurring-tasks';
import moment from 'moment-timezone';
import db from '../config/db';
// Configure Redis connection
const redisConfig = {
host: process.env.REDIS_HOST || 'localhost',
port: parseInt(process.env.REDIS_PORT || '6379'),
password: process.env.REDIS_PASSWORD,
db: parseInt(process.env.REDIS_DB || '0'),
};
// Create job queues
export const recurringTasksQueue = new Bull('recurring-tasks', {
redis: redisConfig,
defaultJobOptions: {
removeOnComplete: 100, // Keep last 100 completed jobs
removeOnFail: 50, // Keep last 50 failed jobs
attempts: 3,
backoff: {
type: 'exponential',
delay: 2000,
},
},
});
export const taskCreationQueue = new Bull('task-creation', {
redis: redisConfig,
defaultJobOptions: {
removeOnComplete: 200,
removeOnFail: 100,
attempts: 5,
backoff: {
type: 'exponential',
delay: 1000,
},
},
});
// Job data interfaces
interface RecurringTaskJobData {
templateId: string;
scheduleId: string;
timezone: string;
}
interface TaskCreationJobData {
template: ITaskTemplate & IRecurringSchedule;
endDates: string[];
timezone: string;
}
// Job processors
recurringTasksQueue.process('process-template', async (job) => {
const { templateId, scheduleId, timezone }: RecurringTaskJobData = job.data;
try {
RecurringTasksAuditLogger.startTimer();
// Fetch template data
const templateQuery = `
SELECT t.*, s.*,
(SELECT MAX(end_date) FROM tasks WHERE schedule_id = s.id) as last_task_end_date,
u.timezone as user_timezone
FROM task_recurring_templates t
JOIN task_recurring_schedules s ON t.schedule_id = s.id
LEFT JOIN tasks orig_task ON t.task_id = orig_task.id
LEFT JOIN users u ON orig_task.reporter_id = u.id
WHERE t.id = $1 AND s.id = $2
`;
const result = await RetryUtils.withDatabaseRetry(async () => {
return await db.query(templateQuery, [templateId, scheduleId]);
}, 'fetch_template_for_job');
if (result.rows.length === 0) {
throw new Error(`Template ${templateId} not found`);
}
const template = result.rows[0] as ITaskTemplate & IRecurringSchedule & { user_timezone?: string };
// Check permissions
const permissionCheck = await RecurringTasksPermissions.validateTemplatePermissions(template.task_id);
if (!permissionCheck.hasPermission) {
await RecurringTasksAuditLogger.log({
operationType: RecurringTaskOperationType.TASKS_CREATION_FAILED,
templateId: template.task_id,
scheduleId: template.schedule_id,
templateName: template.name,
success: false,
errorMessage: `Permission denied: ${permissionCheck.reason}`,
details: { permissionCheck, processedBy: 'job_queue' }
});
return;
}
// Calculate dates to create
const now = TimezoneUtils.nowInTimezone(timezone);
const lastTaskEndDate = template.last_task_end_date
? moment.tz(template.last_task_end_date, timezone)
: moment.tz(template.created_at, timezone);
const futureLimit = moment.tz(template.last_checked_at || template.created_at, timezone)
.add(getFutureLimit(
template.schedule_type,
template.interval_days || template.interval_weeks || template.interval_months || 1
));
let nextEndDate = TimezoneUtils.calculateNextEndDateWithTimezone(template, lastTaskEndDate, timezone);
const endDatesToCreate: string[] = [];
while (nextEndDate.isSameOrBefore(futureLimit)) {
if (nextEndDate.isAfter(now)) {
if (!template.excluded_dates || !template.excluded_dates.includes(nextEndDate.format('YYYY-MM-DD'))) {
endDatesToCreate.push(nextEndDate.format('YYYY-MM-DD'));
}
}
nextEndDate = TimezoneUtils.calculateNextEndDateWithTimezone(template, nextEndDate, timezone);
}
if (endDatesToCreate.length > 0) {
// Add task creation job
await taskCreationQueue.add('create-tasks', {
template,
endDates: endDatesToCreate,
timezone
}, {
priority: 10, // Higher priority for task creation
});
}
// Update schedule
await RetryUtils.withDatabaseRetry(async () => {
const updateQuery = `
UPDATE task_recurring_schedules
SET last_checked_at = $1
WHERE id = $2;
`;
return await db.query(updateQuery, [now.toDate(), scheduleId]);
}, `update_schedule_for_template_${templateId}`);
} catch (error) {
log_error('Error processing recurring task template:', error);
throw error;
}
});
taskCreationQueue.process('create-tasks', async (job) => {
const { template, endDates, timezone }: TaskCreationJobData = job.data;
try {
// Create tasks using the bulk function from the cron job
const tasksData = endDates.map(endDate => ({
name: template.name,
priority_id: template.priority_id,
project_id: template.project_id,
reporter_id: template.reporter_id,
status_id: template.status_id || null,
end_date: endDate,
schedule_id: template.schedule_id
}));
const createTasksResult = await RetryUtils.withDatabaseRetry(async () => {
const createTasksQuery = `SELECT * FROM create_bulk_recurring_tasks($1::JSONB);`;
return await db.query(createTasksQuery, [JSON.stringify(tasksData)]);
}, `create_bulk_tasks_queue_${template.name}`);
const createdTasks = createTasksResult.rows.filter(row => row.created);
const failedTasks = createTasksResult.rows.filter(row => !row.created);
// Handle assignments and labels (similar to cron job implementation)
if (createdTasks.length > 0 && (template.assignees?.length > 0 || template.labels?.length > 0)) {
// ... (assignment logic from cron job)
}
// Send notifications
if (createdTasks.length > 0) {
const taskData = createdTasks.map(task => ({ id: task.task_id, name: task.task_name }));
const assigneeIds = template.assignees?.map(a => a.team_member_id) || [];
await RecurringTasksNotifications.notifyRecurringTasksCreated(
template.name,
template.project_id,
taskData,
assigneeIds,
template.reporter_id
);
}
// Log results
await RecurringTasksAuditLogger.logTemplateProcessing(
template.task_id,
template.name,
template.schedule_id,
createdTasks.length,
failedTasks.length,
{
timezone,
endDates,
processedBy: 'job_queue'
}
);
return {
created: createdTasks.length,
failed: failedTasks.length
};
} catch (error) {
log_error('Error creating tasks in queue:', error);
throw error;
}
});
// Helper function (copied from cron job)
function getFutureLimit(scheduleType: string, interval?: number): moment.Duration {
const FUTURE_LIMITS = {
daily: moment.duration(3, "days"),
weekly: moment.duration(1, "week"),
monthly: moment.duration(1, "month"),
every_x_days: (interval: number) => moment.duration(interval, "days"),
every_x_weeks: (interval: number) => moment.duration(interval, "weeks"),
every_x_months: (interval: number) => moment.duration(interval, "months")
};
switch (scheduleType) {
case "daily":
return FUTURE_LIMITS.daily;
case "weekly":
return FUTURE_LIMITS.weekly;
case "monthly":
return FUTURE_LIMITS.monthly;
case "every_x_days":
return FUTURE_LIMITS.every_x_days(interval || 1);
case "every_x_weeks":
return FUTURE_LIMITS.every_x_weeks(interval || 1);
case "every_x_months":
return FUTURE_LIMITS.every_x_months(interval || 1);
default:
return moment.duration(3, "days");
}
}
// Job schedulers
export class RecurringTasksJobScheduler {
/**
* Schedule recurring task processing for all templates
*/
static async scheduleRecurringTasks(): Promise<void> {
try {
// Get all active templates
const templatesQuery = `
SELECT t.id as template_id, s.id as schedule_id,
COALESCE(s.timezone, u.timezone, 'UTC') as timezone
FROM task_recurring_templates t
JOIN task_recurring_schedules s ON t.schedule_id = s.id
LEFT JOIN tasks orig_task ON t.task_id = orig_task.id
LEFT JOIN users u ON orig_task.reporter_id = u.id
WHERE s.end_date IS NULL OR s.end_date >= CURRENT_DATE
`;
const result = await db.query(templatesQuery);
// Schedule a job for each template
for (const template of result.rows) {
await recurringTasksQueue.add('process-template', {
templateId: template.template_id,
scheduleId: template.schedule_id,
timezone: template.timezone
}, {
delay: Math.random() * 60000, // Random delay up to 1 minute to spread load
});
}
} catch (error) {
log_error('Error scheduling recurring tasks:', error);
}
}
/**
* Start the job queue system
*/
static async start(): Promise<void> {
console.log('Starting recurring tasks job queue...');
// Schedule recurring task processing every hour
await recurringTasksQueue.add('schedule-all', {}, {
repeat: { cron: '0 * * * *' }, // Every hour
removeOnComplete: 1,
removeOnFail: 1,
});
// Process the schedule-all job
recurringTasksQueue.process('schedule-all', async () => {
await this.scheduleRecurringTasks();
});
console.log('Recurring tasks job queue started');
}
/**
* Get queue statistics
*/
static async getStats(): Promise<any> {
const [recurringStats, creationStats] = await Promise.all([
recurringTasksQueue.getJobCounts(),
taskCreationQueue.getJobCounts()
]);
return {
recurringTasks: recurringStats,
taskCreation: creationStats
};
}
}

162
worklenz-backend/src/services/recurring-tasks-service.ts Normal file
View File

@@ -0,0 +1,162 @@
import { recurringTasksConfig } from '../config/recurring-tasks-config';
import { startRecurringTasksJob } from '../cron_jobs/recurring-tasks';
import { RecurringTasksJobScheduler } from '../jobs/recurring-tasks-queue';
import { log_error } from '../shared/utils';
export class RecurringTasksService {
private static isStarted = false;
/**
* Start the recurring tasks service based on configuration
*/
static async start(): Promise<void> {
if (this.isStarted) {
console.log('Recurring tasks service already started');
return;
}
if (!recurringTasksConfig.enabled) {
console.log('Recurring tasks service disabled');
return;
}
try {
console.log(`Starting recurring tasks service in ${recurringTasksConfig.mode} mode...`);
switch (recurringTasksConfig.mode) {
case 'cron':
startRecurringTasksJob();
break;
case 'queue':
await RecurringTasksJobScheduler.start();
break;
default:
throw new Error(`Unknown recurring tasks mode: ${recurringTasksConfig.mode}`);
}
this.isStarted = true;
console.log(`Recurring tasks service started successfully in ${recurringTasksConfig.mode} mode`);
} catch (error) {
log_error('Failed to start recurring tasks service:', error);
throw error;
}
}
/**
* Stop the recurring tasks service
*/
static async stop(): Promise<void> {
if (!this.isStarted) {
return;
}
try {
console.log('Stopping recurring tasks service...');
if (recurringTasksConfig.mode === 'queue') {
// Close queue connections
const { recurringTasksQueue, taskCreationQueue } = await import('../jobs/recurring-tasks-queue');
await recurringTasksQueue.close();
await taskCreationQueue.close();
}
this.isStarted = false;
console.log('Recurring tasks service stopped');
} catch (error) {
log_error('Error stopping recurring tasks service:', error);
}
}
/**
* Get service status and statistics
*/
static async getStatus(): Promise<any> {
const status = {
enabled: recurringTasksConfig.enabled,
mode: recurringTasksConfig.mode,
started: this.isStarted,
config: recurringTasksConfig
};
if (this.isStarted && recurringTasksConfig.mode === 'queue') {
try {
const stats = await RecurringTasksJobScheduler.getStats();
return { ...status, queueStats: stats };
} catch (error) {
return { ...status, queueStatsError: error.message };
}
}
return status;
}
/**
* Manually trigger recurring tasks processing
*/
static async triggerManual(): Promise<void> {
if (!this.isStarted) {
throw new Error('Recurring tasks service is not started');
}
try {
if (recurringTasksConfig.mode === 'queue') {
await RecurringTasksJobScheduler.scheduleRecurringTasks();
} else {
// For cron mode, we can't manually trigger easily
// Could implement a manual trigger function in the cron job file
throw new Error('Manual trigger not supported in cron mode');
}
} catch (error) {
log_error('Error manually triggering recurring tasks:', error);
throw error;
}
}
/**
* Health check for the service
*/
static async healthCheck(): Promise<{ healthy: boolean; message: string; details?: any }> {
try {
if (!recurringTasksConfig.enabled) {
return {
healthy: true,
message: 'Recurring tasks service is disabled'
};
}
if (!this.isStarted) {
return {
healthy: false,
message: 'Recurring tasks service is not started'
};
}
if (recurringTasksConfig.mode === 'queue') {
const stats = await RecurringTasksJobScheduler.getStats();
const hasFailures = stats.recurringTasks.failed > 0 || stats.taskCreation.failed > 0;
return {
healthy: !hasFailures,
message: hasFailures ? 'Some jobs are failing' : 'All systems operational',
details: stats
};
}
return {
healthy: true,
message: `Running in ${recurringTasksConfig.mode} mode`
};
} catch (error) {
return {
healthy: false,
message: 'Health check failed',
details: { error: error.message }
};
}
}
}

189
worklenz-backend/src/utils/recurring-tasks-audit-logger.ts Normal file
View File

@@ -0,0 +1,189 @@
import db from "../config/db";
import { log_error } from "../shared/utils";
export enum RecurringTaskOperationType {
CRON_JOB_RUN = "cron_job_run",
CRON_JOB_ERROR = "cron_job_error",
TEMPLATE_CREATED = "template_created",
TEMPLATE_UPDATED = "template_updated",
TEMPLATE_DELETED = "template_deleted",
SCHEDULE_CREATED = "schedule_created",
SCHEDULE_UPDATED = "schedule_updated",
SCHEDULE_DELETED = "schedule_deleted",
TASKS_CREATED = "tasks_created",
TASKS_CREATION_FAILED = "tasks_creation_failed",
MANUAL_TRIGGER = "manual_trigger",
BULK_OPERATION = "bulk_operation"
}
export interface AuditLogEntry {
operationType: RecurringTaskOperationType;
templateId?: string;
scheduleId?: string;
taskId?: string;
templateName?: string;
success?: boolean;
errorMessage?: string;
details?: any;
createdTasksCount?: number;
failedTasksCount?: number;
executionTimeMs?: number;
createdBy?: string;
}
export class RecurringTasksAuditLogger {
private static startTime: number;
/**
* Start timing an operation
*/
static startTimer(): void {
this.startTime = Date.now();
}
/**
* Get elapsed time since timer started
*/
static getElapsedTime(): number {
return this.startTime ? Date.now() - this.startTime : 0;
}
/**
* Log a recurring task operation
*/
static async log(entry: AuditLogEntry): Promise<void> {
try {
const query = `SELECT log_recurring_task_operation($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12);`;
await db.query(query, [
entry.operationType,
entry.templateId || null,
entry.scheduleId || null,
entry.taskId || null,
entry.templateName || null,
entry.success !== false, // Default to true
entry.errorMessage || null,
entry.details ? JSON.stringify(entry.details) : null,
entry.createdTasksCount || 0,
entry.failedTasksCount || 0,
entry.executionTimeMs || this.getElapsedTime(),
entry.createdBy || null
]);
} catch (error) {
// Don't let audit logging failures break the main flow
log_error("Failed to log recurring task audit entry:", error);
}
}
/**
* Log cron job execution
*/
static async logCronJobRun(
totalTemplates: number,
createdTasksCount: number,
errors: any[] = []
): Promise<void> {
await this.log({
operationType: RecurringTaskOperationType.CRON_JOB_RUN,
success: errors.length === 0,
errorMessage: errors.length > 0 ? `${errors.length} errors occurred` : undefined,
details: {
totalTemplates,
errors: errors.map(e => e.message || e.toString())
},
createdTasksCount,
executionTimeMs: this.getElapsedTime()
});
}
/**
* Log template processing
*/
static async logTemplateProcessing(
templateId: string,
templateName: string,
scheduleId: string,
createdCount: number,
failedCount: number,
details?: any
): Promise<void> {
await this.log({
operationType: RecurringTaskOperationType.TASKS_CREATED,
templateId,
scheduleId,
templateName,
success: failedCount === 0,
createdTasksCount: createdCount,
failedTasksCount: failedCount,
details
});
}
/**
* Log schedule changes
*/
static async logScheduleChange(
operationType: RecurringTaskOperationType,
scheduleId: string,
templateId?: string,
userId?: string,
details?: any
): Promise<void> {
await this.log({
operationType,
scheduleId,
templateId,
createdBy: userId,
details
});
}
/**
* Get audit log summary
*/
static async getAuditSummary(days: number = 7): Promise<any> {
try {
const query = `
SELECT
operation_type,
COUNT(*) as count,
SUM(CASE WHEN success THEN 1 ELSE 0 END) as success_count,
SUM(CASE WHEN NOT success THEN 1 ELSE 0 END) as failure_count,
SUM(created_tasks_count) as total_tasks_created,
SUM(failed_tasks_count) as total_tasks_failed,
AVG(execution_time_ms) as avg_execution_time_ms
FROM recurring_tasks_audit_log
WHERE created_at >= CURRENT_TIMESTAMP - INTERVAL '${days} days'
GROUP BY operation_type
ORDER BY count DESC;
`;
const result = await db.query(query);
return result.rows;
} catch (error) {
log_error("Failed to get audit summary:", error);
return [];
}
}
/**
* Get recent errors
*/
static async getRecentErrors(limit: number = 10): Promise<any[]> {
try {
const query = `
SELECT *
FROM v_recent_recurring_tasks_audit
WHERE NOT success
ORDER BY created_at DESC
LIMIT $1;
`;
const result = await db.query(query, [limit]);
return result.rows;
} catch (error) {
log_error("Failed to get recent errors:", error);
return [];
}
}
}

260
worklenz-backend/src/utils/recurring-tasks-notifications.ts Normal file
View File

@@ -0,0 +1,260 @@
import db from "../config/db";
import { log_error } from "../shared/utils";
export interface NotificationData {
userId: string;
projectId: string;
taskId: string;
taskName: string;
templateName: string;
scheduleId: string;
createdBy?: string;
}
export class RecurringTasksNotifications {
/**
* Send notification to user about a new recurring task
*/
static async notifyTaskCreated(data: NotificationData): Promise<void> {
try {
// Create notification in the database
const notificationQuery = `
INSERT INTO notifications (
user_id,
message,
data,
created_at
) VALUES ($1, $2, $3, NOW())
`;
const message = `New recurring task "${data.taskName}" has been created from template "${data.templateName}"`;
const notificationData = {
type: 'recurring_task_created',
task_id: data.taskId,
project_id: data.projectId,
schedule_id: data.scheduleId,
task_name: data.taskName,
template_name: data.templateName
};
await db.query(notificationQuery, [
data.userId,
message,
JSON.stringify(notificationData)
]);
} catch (error) {
log_error("Failed to create notification:", error);
}
}
/**
* Send notifications to all assignees of created tasks
*/
static async notifyAssignees(
taskIds: string[],
templateName: string,
projectId: string
): Promise<void> {
if (taskIds.length === 0) return;
try {
// Get all assignees for the created tasks
const assigneesQuery = `
SELECT DISTINCT ta.team_member_id, t.id as task_id, t.name as task_name
FROM tasks_assignees ta
JOIN tasks t ON ta.task_id = t.id
WHERE t.id = ANY($1)
`;
const result = await db.query(assigneesQuery, [taskIds]);
// Send notification to each assignee
for (const assignee of result.rows) {
await this.notifyTaskCreated({
userId: assignee.team_member_id,
projectId,
taskId: assignee.task_id,
taskName: assignee.task_name,
templateName,
scheduleId: '' // Not needed for assignee notifications
});
}
} catch (error) {
log_error("Failed to notify assignees:", error);
}
}
/**
* Send email notifications (if email system is configured)
*/
static async sendEmailNotifications(
userIds: string[],
subject: string,
message: string
): Promise<void> {
try {
// Get user email addresses
const usersQuery = `
SELECT id, email, name, email_notifications
FROM users
WHERE id = ANY($1) AND email_notifications = true AND email IS NOT NULL
`;
const result = await db.query(usersQuery, [userIds]);
// TODO: Integrate with your email service (SendGrid, AWS SES, etc.)
// For now, just log the email notifications that would be sent
for (const user of result.rows) {
console.log(`Email notification would be sent to ${user.email}: ${subject}`);
// Example: await emailService.send({
// to: user.email,
// subject,
// html: message
// });
}
} catch (error) {
log_error("Failed to send email notifications:", error);
}
}
/**
* Send push notifications (if push notification system is configured)
*/
static async sendPushNotifications(
userIds: string[],
title: string,
body: string,
data?: any
): Promise<void> {
try {
// Get user push tokens
const tokensQuery = `
SELECT user_id, push_token
FROM user_push_tokens
WHERE user_id = ANY($1) AND push_token IS NOT NULL
`;
const result = await db.query(tokensQuery, [userIds]);
// TODO: Integrate with your push notification service (FCM, APNs, etc.)
// For now, just log the push notifications that would be sent
for (const token of result.rows) {
console.log(`Push notification would be sent to ${token.push_token}: ${title}`);
// Example: await pushService.send({
// token: token.push_token,
// title,
// body,
// data
// });
}
} catch (error) {
log_error("Failed to send push notifications:", error);
}
}
/**
* Get notification preferences for users
*/
static async getNotificationPreferences(userIds: string[]): Promise<any[]> {
try {
const query = `
SELECT
id,
email_notifications,
push_notifications,
in_app_notifications
FROM users
WHERE id = ANY($1)
`;
const result = await db.query(query, [userIds]);
return result.rows;
} catch (error) {
log_error("Failed to get notification preferences:", error);
return [];
}
}
/**
* Comprehensive notification for recurring task creation
*/
static async notifyRecurringTasksCreated(
templateName: string,
projectId: string,
createdTasks: Array<{ id: string; name: string }>,
assignees: string[] = [],
reporterId?: string
): Promise<void> {
try {
const taskIds = createdTasks.map(t => t.id);
const allUserIds = [...new Set([...assignees, reporterId].filter(Boolean))];
if (allUserIds.length === 0) return;
// Get notification preferences
const preferences = await this.getNotificationPreferences(allUserIds);
// Send in-app notifications
const inAppUsers = preferences.filter(p => p.in_app_notifications !== false);
for (const user of inAppUsers) {
for (const task of createdTasks) {
await this.notifyTaskCreated({
userId: user.id,
projectId,
taskId: task.id,
taskName: task.name,
templateName,
scheduleId: '',
createdBy: 'system'
});
}
}
// Send email notifications
const emailUsers = preferences
.filter(p => p.email_notifications === true)
.map(p => p.id);
if (emailUsers.length > 0) {
const subject = `New Recurring Tasks Created: ${templateName}`;
const message = `
<h3>Recurring Tasks Created</h3>
<p>${createdTasks.length} new tasks have been created from template "${templateName}":</p>
<ul>
${createdTasks.map(t => `<li>${t.name}</li>`).join('')}
</ul>
`;
await this.sendEmailNotifications(emailUsers, subject, message);
}
// Send push notifications
const pushUsers = preferences
.filter(p => p.push_notifications !== false)
.map(p => p.id);
if (pushUsers.length > 0) {
await this.sendPushNotifications(
pushUsers,
'New Recurring Tasks',
`${createdTasks.length} tasks created from ${templateName}`,
{
type: 'recurring_tasks_created',
project_id: projectId,
task_count: createdTasks.length
}
);
}
} catch (error) {
log_error("Failed to send comprehensive notifications:", error);
}
}
}

187
worklenz-backend/src/utils/recurring-tasks-permissions.ts Normal file
View File

@@ -0,0 +1,187 @@
import db from "../config/db";
import { log_error } from "../shared/utils";
export interface PermissionCheckResult {
hasPermission: boolean;
reason?: string;
projectRole?: string;
}
export class RecurringTasksPermissions {
/**
* Check if a user has permission to create tasks in a project
*/
static async canCreateTasksInProject(
userId: string,
projectId: string
): Promise<PermissionCheckResult> {
try {
// Check if user is a member of the project
const memberQuery = `
SELECT pm.role_id, pr.name as role_name, pr.permissions
FROM project_members pm
JOIN project_member_roles pr ON pm.role_id = pr.id
WHERE pm.user_id = $1 AND pm.project_id = $2
LIMIT 1;
`;
const result = await db.query(memberQuery, [userId, projectId]);
if (result.rows.length === 0) {
return {
hasPermission: false,
reason: "User is not a member of the project"
};
}
const member = result.rows[0];
// Check if role has task creation permission
if (member.permissions && member.permissions.create_tasks === false) {
return {
hasPermission: false,
reason: "User role does not have permission to create tasks",
projectRole: member.role_name
};
}
return {
hasPermission: true,
projectRole: member.role_name
};
} catch (error) {
log_error("Error checking project permissions:", error);
return {
hasPermission: false,
reason: "Error checking permissions"
};
}
}
/**
* Check if a template has valid permissions
*/
static async validateTemplatePermissions(templateId: string): Promise<PermissionCheckResult> {
try {
const query = `
SELECT
t.reporter_id,
t.project_id,
p.is_active as project_active,
p.archived as project_archived,
u.is_active as user_active
FROM task_recurring_templates trt
JOIN tasks t ON trt.task_id = t.id
JOIN projects p ON t.project_id = p.id
JOIN users u ON t.reporter_id = u.id
WHERE trt.id = $1
LIMIT 1;
`;
const result = await db.query(query, [templateId]);
if (result.rows.length === 0) {
return {
hasPermission: false,
reason: "Template not found"
};
}
const template = result.rows[0];
// Check if project is active
if (!template.project_active || template.project_archived) {
return {
hasPermission: false,
reason: "Project is not active or archived"
};
}
// Check if reporter is still active
if (!template.user_active) {
return {
hasPermission: false,
reason: "Original task reporter is no longer active"
};
}
// Check if reporter still has permissions in the project
const permissionCheck = await this.canCreateTasksInProject(
template.reporter_id,
template.project_id
);
return permissionCheck;
} catch (error) {
log_error("Error validating template permissions:", error);
return {
hasPermission: false,
reason: "Error validating template permissions"
};
}
}
/**
* Get all templates with permission issues
*/
static async getTemplatesWithPermissionIssues(): Promise<any[]> {
try {
const query = `
SELECT
trt.id as template_id,
trt.name as template_name,
t.reporter_id,
u.name as reporter_name,
t.project_id,
p.name as project_name,
CASE
WHEN NOT p.is_active THEN 'Project inactive'
WHEN p.archived THEN 'Project archived'
WHEN NOT u.is_active THEN 'User inactive'
WHEN NOT EXISTS (
SELECT 1 FROM project_members
WHERE user_id = t.reporter_id AND project_id = t.project_id
) THEN 'User not in project'
ELSE NULL
END as issue
FROM task_recurring_templates trt
JOIN tasks t ON trt.task_id = t.id
JOIN projects p ON t.project_id = p.id
JOIN users u ON t.reporter_id = u.id
WHERE
NOT p.is_active
OR p.archived
OR NOT u.is_active
OR NOT EXISTS (
SELECT 1 FROM project_members
WHERE user_id = t.reporter_id AND project_id = t.project_id
);
`;
const result = await db.query(query);
return result.rows;
} catch (error) {
log_error("Error getting templates with permission issues:", error);
return [];
}
}
/**
* Validate all assignees have permissions
*/
static async validateAssigneePermissions(
assignees: Array<{ team_member_id: string }>,
projectId: string
): Promise<string[]> {
const invalidAssignees: string[] = [];
for (const assignee of assignees) {
const check = await this.canCreateTasksInProject(assignee.team_member_id, projectId);
if (!check.hasPermission) {
invalidAssignees.push(assignee.team_member_id);
}
}
return invalidAssignees;
}
}

134
worklenz-backend/src/utils/retry-utils.ts Normal file
View File

@@ -0,0 +1,134 @@
import { log_error } from "../shared/utils";
export interface RetryOptions {
maxRetries: number;
delayMs: number;
backoffFactor?: number;
onRetry?: (error: any, attempt: number) => void;
}
export class RetryUtils {
/**
* Execute a function with retry logic
*/
static async withRetry<T>(
fn: () => Promise<T>,
options: RetryOptions
): Promise<T> {
const { maxRetries, delayMs, backoffFactor = 1.5, onRetry } = options;
let lastError: any;
for (let attempt = 1; attempt <= maxRetries; attempt++) {
try {
return await fn();
} catch (error) {
lastError = error;
if (attempt === maxRetries) {
throw error;
}
const delay = delayMs * Math.pow(backoffFactor, attempt - 1);
if (onRetry) {
onRetry(error, attempt);
}
log_error(`Attempt ${attempt} failed. Retrying in ${delay}ms...`, error);
await new Promise(resolve => setTimeout(resolve, delay));
}
}
throw lastError;
}
/**
* Execute database operations with retry logic
*/
static async withDatabaseRetry<T>(
operation: () => Promise<T>,
operationName: string
): Promise<T> {
return this.withRetry(operation, {
maxRetries: 3,
delayMs: 1000,
backoffFactor: 2,
onRetry: (error, attempt) => {
log_error(`Database operation '${operationName}' failed on attempt ${attempt}:`, error);
}
});
}
/**
* Check if an error is retryable
*/
static isRetryableError(error: any): boolean {
// PostgreSQL error codes that are retryable
const retryableErrorCodes = [
'40001', // serialization_failure
'40P01', // deadlock_detected
'55P03', // lock_not_available
'57P01', // admin_shutdown
'57P02', // crash_shutdown
'57P03', // cannot_connect_now
'58000', // system_error
'58030', // io_error
'53000', // insufficient_resources
'53100', // disk_full
'53200', // out_of_memory
'53300', // too_many_connections
'53400', // configuration_limit_exceeded
];
if (error.code && retryableErrorCodes.includes(error.code)) {
return true;
}
// Network-related errors
if (error.message && (
error.message.includes('ECONNRESET') ||
error.message.includes('ETIMEDOUT') ||
error.message.includes('ECONNREFUSED')
)) {
return true;
}
return false;
}
/**
* Execute with conditional retry based on error type
*/
static async withConditionalRetry<T>(
fn: () => Promise<T>,
options: RetryOptions
): Promise<T> {
const { maxRetries, delayMs, backoffFactor = 1.5, onRetry } = options;
let lastError: any;
for (let attempt = 1; attempt <= maxRetries; attempt++) {
try {
return await fn();
} catch (error) {
lastError = error;
if (!this.isRetryableError(error) || attempt === maxRetries) {
throw error;
}
const delay = delayMs * Math.pow(backoffFactor, attempt - 1);
if (onRetry) {
onRetry(error, attempt);
}
log_error(`Retryable error on attempt ${attempt}. Retrying in ${delay}ms...`, error);
await new Promise(resolve => setTimeout(resolve, delay));
}
}
throw lastError;
}
}

156
worklenz-backend/src/utils/timezone-utils.ts Normal file
View File

@@ -0,0 +1,156 @@
import moment from "moment-timezone";
import { IRecurringSchedule } from "../interfaces/recurring-tasks";
export class TimezoneUtils {
/**
* Convert a date from one timezone to another
*/
static convertTimezone(date: moment.Moment | Date | string, fromTz: string, toTz: string): moment.Moment {
return moment.tz(date, fromTz).tz(toTz);
}
/**
* Get the current time in a specific timezone
*/
static nowInTimezone(timezone: string): moment.Moment {
return moment.tz(timezone);
}
/**
* Check if a recurring task should run based on timezone
*/
static shouldRunInTimezone(schedule: IRecurringSchedule, timezone: string): boolean {
const now = this.nowInTimezone(timezone);
const scheduleTime = moment.tz(schedule.created_at, timezone);
// Check if it's the right time of day (within a 1-hour window)
const hourDiff = Math.abs(now.hour() - scheduleTime.hour());
return hourDiff < 1;
}
/**
* Calculate next end date considering timezone
*/
static calculateNextEndDateWithTimezone(
schedule: IRecurringSchedule,
lastDate: moment.Moment | Date | string,
timezone: string
): moment.Moment {
const lastMoment = moment.tz(lastDate, timezone);
switch (schedule.schedule_type) {
case "daily":
return lastMoment.clone().add(1, "day");
case "weekly":
if (schedule.days_of_week && schedule.days_of_week.length > 0) {
// Find next occurrence based on selected days
let nextDate = lastMoment.clone();
let daysChecked = 0;
do {
nextDate.add(1, "day");
daysChecked++;
if (schedule.days_of_week.includes(nextDate.day())) {
return nextDate;
}
} while (daysChecked < 7);
// If no valid day found, return next week's first selected day
const sortedDays = [...schedule.days_of_week].sort((a, b) => a - b);
nextDate = lastMoment.clone().add(1, "week").day(sortedDays[0]);
return nextDate;
}
return lastMoment.clone().add(1, "week");
case "monthly":
if (schedule.date_of_month) {
// Specific date of month
let nextDate = lastMoment.clone().add(1, "month").date(schedule.date_of_month);
// Handle months with fewer days
if (nextDate.date() !== schedule.date_of_month) {
nextDate = nextDate.endOf("month");
}
return nextDate;
} else if (schedule.week_of_month && schedule.day_of_month !== undefined) {
// Nth occurrence of a day in month
const nextMonth = lastMoment.clone().add(1, "month").startOf("month");
const targetDay = schedule.day_of_month;
const targetWeek = schedule.week_of_month;
// Find first occurrence of the target day
let firstOccurrence = nextMonth.clone();
while (firstOccurrence.day() !== targetDay) {
firstOccurrence.add(1, "day");
}
// Calculate nth occurrence
if (targetWeek === 5) {
// Last occurrence
let lastOccurrence = firstOccurrence.clone();
let temp = firstOccurrence.clone().add(7, "days");
while (temp.month() === nextMonth.month()) {
lastOccurrence = temp.clone();
temp.add(7, "days");
}
return lastOccurrence;
} else {
// Specific week number
return firstOccurrence.add((targetWeek - 1) * 7, "days");
}
}
return lastMoment.clone().add(1, "month");
case "every_x_days":
return lastMoment.clone().add(schedule.interval_days || 1, "days");
case "every_x_weeks":
return lastMoment.clone().add(schedule.interval_weeks || 1, "weeks");
case "every_x_months":
return lastMoment.clone().add(schedule.interval_months || 1, "months");
default:
return lastMoment.clone().add(1, "day");
}
}
/**
* Get all timezones that should be processed in the current hour
*/
static getActiveTimezones(): string[] {
const activeTimezones: string[] = [];
const allTimezones = moment.tz.names();
for (const tz of allTimezones) {
const tzTime = moment.tz(tz);
// Check if it's 11:00 AM in this timezone (matching the cron schedule)
if (tzTime.hour() === 11) {
activeTimezones.push(tz);
}
}
return activeTimezones;
}
/**
* Validate timezone string
*/
static isValidTimezone(timezone: string): boolean {
return moment.tz.zone(timezone) !== null;
}
/**
* Get user's timezone or default to UTC
*/
static getUserTimezone(userTimezone?: string): string {
if (userTimezone && this.isValidTimezone(userTimezone)) {
return userTimezone;
}
return "UTC";
}
}