feat: add cron task management features including enabling/disabling tasks and immediate execution, along with log rotation configuration and UI enhancements for better user interaction

Author: tacxou

Dockerfile

@@ -22,6 +22,8 @@ ENV ALLOW_RUNTIME_BUILD=true
 ENV DO_NOT_TRACK=1
 ENV PYTHONWARNINGS=ignore::UserWarning
 ENV TZ=Europe/Paris
+ENV SESAME_CRON_LOG_ROTATE_MAX_SIZE_BYTES=10485760
+ENV SESAME_CRON_LOG_ROTATE_MAX_FILES=5
 
 WORKDIR /data
 
@@ -41,7 +43,8 @@ RUN apk add --no-cache \
   tzdata \
   bash \
   nano && \
-  mkdir -p /var/log/supervisor
+  mkdir -p /var/log/supervisor /data/apps/api/logs/handlers && \
+  chmod -R 0777 /data/apps/api/logs
 
 RUN ARCH=$(uname -m) && \
   if [ "$ARCH" = "x86_64" ]; then \

apps/api/src/_common/functions/handler-logger.ts

@@ -21,16 +21,54 @@ function stripAnsiCodes(str: string): string {
   return str.replace(/\x1b\[[0-9;?]*[A-Za-z]/g, '')
 }
 
+function rotateFile(logFile: string, maxFiles: number) {
+  for (let index = maxFiles - 1; index >= 1; index--) {
+    const source = `${logFile}.${index}`
+    const destination = `${logFile}.${index + 1}`
+
+    if (fs.existsSync(source)) {
+      fs.renameSync(source, destination)
+    }
+  }
+
+  fs.renameSync(logFile, `${logFile}.1`)
+}
+
+function rotateHandlerLogIfNeeded(logFile: string, maxSizeBytes: number, maxFiles: number) {
+  if (!fs.existsSync(logFile)) {
+    return
+  }
+
+  const stats = fs.statSync(logFile)
+  if (stats.size < maxSizeBytes) {
+    return
+  }
+
+  try {
+    const oldestArchive = `${logFile}.${maxFiles}`
+    if (fs.existsSync(oldestArchive)) {
+      fs.unlinkSync(oldestArchive)
+    }
+
+    rotateFile(logFile, maxFiles)
+  } catch (err) {
+    Logger.error(`Could not rotate handler log file ${logFile}: ${(err as Error).message}`, rotateHandlerLogIfNeeded.name)
+  }
+}
+
 export function createHandlerLogger(config: ConfigService, handler: string) {
   const safeHandler = toSafeHandlerName(handler)
   const logDir = config.get('cron.logDirectory') || path.join(process.cwd(), 'logs', 'handlers')
   const logFile = path.join(logDir, `${safeHandler}.log`)
+  const logRotateMaxSizeBytes = config.get<number>('cron.logRotateMaxSizeBytes') || 10 * 1024 * 1024
+  const logRotateMaxFiles = config.get<number>('cron.logRotateMaxFiles') || 5
 
   let stream: fs.WriteStream | null = null
   let streamError: Error | null = null
 
   try {
     ensureDir(logDir)
+    rotateHandlerLogIfNeeded(logFile, logRotateMaxSizeBytes, logRotateMaxFiles)
     stream = fs.createWriteStream(logFile, { flags: 'a' })
 
     stream.on('error', (err) => {

apps/api/src/config.ts

@@ -158,6 +158,18 @@ export const validationSchema = Joi.object({
     .string()
     .default(path.join(process.cwd(), 'logs', 'handlers')),
 
+  SESAME_CRON_LOG_ROTATE_MAX_SIZE_BYTES: Joi
+    .number()
+    .integer()
+    .min(1)
+    .default(10 * 1024 * 1024),
+
+  SESAME_CRON_LOG_ROTATE_MAX_FILES: Joi
+    .number()
+    .integer()
+    .min(1)
+    .default(5),
+
   SESAME_IDENTITY_DOUBLON_SEARCH_ATTRIBUTES: Joi
     .string()
     .default(''),
@@ -214,6 +226,8 @@ export interface ConfigInstance {
   cron: {
     handlerExpression: string
     logDirectory: string
+    logRotateMaxSizeBytes: number
+    logRotateMaxFiles: number
   }
   factorydrive: {
     options:
@@ -331,6 +345,8 @@ export default (): ConfigInstance => ({
   cron: {
     handlerExpression: process.env['SESAME_CRON_HANDLER_EXPRESSION'] || CronExpression.EVERY_HOUR,
     logDirectory: process.env['SESAME_CRON_LOG_DIRECTORY'] || path.join(process.cwd(), 'logs', 'handlers'),
+    logRotateMaxSizeBytes: parseInt(process.env['SESAME_CRON_LOG_ROTATE_MAX_SIZE_BYTES'], 10) || 10 * 1024 * 1024,
+    logRotateMaxFiles: parseInt(process.env['SESAME_CRON_LOG_ROTATE_MAX_FILES'], 10) || 5,
   },
   factorydrive: {
     options: {

apps/api/src/core/cron/cron-hooks.service.ts

@@ -98,6 +98,34 @@ export class CronHooksService {
     this.logger.log('CronHooksService (bootstrap) initialized')
   }
 
+  public async syncCronJobs(): Promise<void> {
+    await this.handleCron()
+  }
+
+  public async runTaskNow(name: string): Promise<boolean> {
+    let currentTask = this.cronTasks.find((task) => task.name === name)
+    if (!currentTask) {
+      await this.refreshCronTasksFileCache()
+      currentTask = this.cronTasks.find((task) => task.name === name)
+    }
+
+    if (!currentTask) {
+      return false
+    }
+
+    this.logger.warn(`Running cron task manually: ${currentTask.name}`)
+    // Fire-and-forget: the HTTP caller should get an immediate response.
+    void this.executeHandlerCommand(currentTask.name, currentTask.handler, currentTask.options)
+      .then(() => {
+        this.logger.log(`Manual cron task <${currentTask.name}> finished`)
+      })
+      .catch((error) => {
+        this.logger.error(`Manual cron task <${currentTask.name}> failed`, error?.message || error)
+      })
+
+    return true
+  }
+
   private async handleCron(): Promise<void> {
     this.logger.verbose('Syncing cron tasks jobs from configs/cron/*.yml')
 

apps/api/src/core/cron/cron.controller.ts

@@ -1,5 +1,5 @@
 
-import { Controller, Get, Res, HttpStatus, Query, Param, DefaultValuePipe, ParseIntPipe, NotFoundException } from '@nestjs/common'
+import { Body, Controller, DefaultValuePipe, Get, HttpStatus, NotFoundException, Param, ParseIntPipe, Patch, Post, Query, Res } from '@nestjs/common'
 import { ApiTags } from '@nestjs/swagger'
 import { CronService } from './cron.service'
 import { Response } from 'express'
@@ -10,6 +10,12 @@ import { PickProjectionHelper } from '~/_common/helpers/pick-projection.helper'
 import { CronDto } from './_dto/cron.dto'
 import { PartialProjectionType } from '~/_common/types/partial-projection.type'
 import { ApiReadResponseDecorator } from '~/_common/decorators/api-read-response.decorator'
+import { IsBoolean } from 'class-validator'
+
+class UpdateCronEnabledBody {
+  @IsBoolean()
+  enabled: boolean
+}
 
 /**
  * Contrôleur Cron - Endpoints pour les tâches planifiées.
@@ -101,4 +107,51 @@ export class CronController {
       data,
     })
   }
+
+  @Patch(':name/enabled')
+  @UseRoles({
+    resource: '/core/cron',
+    action: AC_ACTIONS.UPDATE,
+    possession: AC_DEFAULT_POSSESSION,
+  })
+  public async updateEnabled(
+    @Param('name') name: string,
+    @Body() body: UpdateCronEnabledBody,
+    @Res() res: Response,
+  ): Promise<Response> {
+    const data = await this.cronService.setEnabled(name, body.enabled)
+    if (!data) {
+      throw new NotFoundException(`Cron task <${name}> not found`)
+    }
+
+    return res.json({
+      statusCode: HttpStatus.OK,
+      data,
+    })
+  }
+
+  @Post(':name/run-immediately')
+  @UseRoles({
+    resource: '/core/cron',
+    action: AC_ACTIONS.UPDATE,
+    possession: AC_DEFAULT_POSSESSION,
+  })
+  public async runImmediately(
+    @Param('name') name: string,
+    @Res() res: Response,
+  ): Promise<Response> {
+    const launched = await this.cronService.runImmediately(name)
+    if (!launched) {
+      throw new NotFoundException(`Cron task <${name}> not found`)
+    }
+
+    return res.json({
+      statusCode: HttpStatus.OK,
+      data: {
+        name,
+        launched: true,
+        status: 'in_progress',
+      },
+    })
+  }
 }

apps/api/src/core/cron/cron.service.ts

@@ -3,12 +3,13 @@ import { Injectable, Logger } from '@nestjs/common'
 import { SchedulerRegistry } from '@nestjs/schedule'
 import { CronHooksService } from './cron-hooks.service'
 import { pick } from 'radash'
-import { CronTaskDTO } from './_dto/config-task.dto'
+import { ConfigTaskDTO, CronTaskDTO } from './_dto/config-task.dto'
 import { CronJob } from 'cron'
 import path from 'node:path'
-import { existsSync, readFileSync, statSync } from 'node:fs'
+import { closeSync, existsSync, openSync, readFileSync, readSync, readdirSync, statSync, writeFileSync } from 'node:fs'
 import { ConfigService } from '@nestjs/config'
 import { toSafeHandlerName } from '~/_common/functions/handler-logger'
+import { parse, stringify } from 'yaml'
 
 @Injectable()
 export class CronService {
@@ -118,6 +119,29 @@ export class CronService {
     }
   }
 
+  public async setEnabled(name: string, enabled: boolean): Promise<CronTaskDTO & { _job: Partial<CronJob> } | null> {
+    const updated = this.updateTaskInConfig(name, (task) => {
+      task.enabled = enabled
+    })
+
+    if (!updated) {
+      return null
+    }
+
+    await this.cronHooksService.syncCronJobs()
+    return this.read(name)
+  }
+
+  public async runImmediately(name: string): Promise<boolean> {
+    const task = await this.read(name)
+    if (!task) {
+      return false
+    }
+
+    await this.cronHooksService.runTaskNow(name)
+    return true
+  }
+
   public async readLogs(name: string, tail = 500): Promise<{
     name: string
     exists: boolean
@@ -140,10 +164,46 @@ export class CronService {
     }
 
     const stats = statSync(logFile)
-    const fullContent = readFileSync(logFile, 'utf-8')
-    const boundedTail = Math.min(Math.max(tail || 200, 1), 2_000)
+    const boundedTail = Math.max(tail || 200, 1)
     const maxLineChars = 4_000
-    const maxContentChars = 200_000
+    const maxContentChars = this.configService.get<number>('cron.logRotateMaxSizeBytes') || 10 * 1024 * 1024
+    const maxReadableBytes = Math.max(maxContentChars, 1)
+    const chunkSize = 64 * 1024
+
+    // Read the file backwards in chunks to avoid loading everything in memory.
+    const fileDescriptor = openSync(logFile, 'r')
+    let filePosition = stats.size
+    let bytesReadTotal = 0
+    let lineBreakCount = 0
+    const chunks: Buffer[] = []
+
+    try {
+      while (filePosition > 0 && bytesReadTotal < maxReadableBytes && lineBreakCount <= boundedTail) {
+        const remainingBudget = maxReadableBytes - bytesReadTotal
+        const readSize = Math.min(chunkSize, filePosition, remainingBudget)
+        filePosition -= readSize
+
+        const chunkBuffer = Buffer.allocUnsafe(readSize)
+        const readBytes = readSync(fileDescriptor, chunkBuffer, 0, readSize, filePosition)
+        if (readBytes <= 0) {
+          break
+        }
+
+        const chunk = readBytes === readSize ? chunkBuffer : chunkBuffer.subarray(0, readBytes)
+        chunks.unshift(chunk)
+        bytesReadTotal += readBytes
+
+        for (let index = 0; index < chunk.length; index++) {
+          if (chunk[index] === 0x0a) {
+            lineBreakCount++
+          }
+        }
+      }
+    } finally {
+      closeSync(fileDescriptor)
+    }
+
+    const fullContent = Buffer.concat(chunks).toString('utf-8')
 
     const tailedLines = fullContent
       .split('\n')
@@ -168,4 +228,35 @@ export class CronService {
       content,
     }
   }
+
+  private updateTaskInConfig(name: string, updater: (task: CronTaskDTO) => void): boolean {
+    const configDir = path.join(process.cwd(), 'configs', 'cron')
+    if (!existsSync(configDir)) {
+      return false
+    }
+
+    const files = readdirSync(configDir).filter((file) => file.endsWith('.yml') || file.endsWith('.yaml'))
+
+    for (const file of files) {
+      const filePath = path.join(configDir, file)
+      const raw = readFileSync(filePath, 'utf-8')
+      const parsed = parse(raw) as ConfigTaskDTO
+      const tasks = parsed?.tasks
+
+      if (!tasks || !Array.isArray(tasks)) {
+        continue
+      }
+
+      const targetTask = tasks.find((task) => task.name === name)
+      if (!targetTask) {
+        continue
+      }
+
+      updater(targetTask)
+      writeFileSync(filePath, stringify(parsed))
+      return true
+    }
+
+    return false
+  }
 }