Job cancellation

[noxify]

Extracted feature request from #41

---

• a new status cancelled
• a new field cancelledAt
• a new method cancelJob on queue level, which can be called to trigger the cancellation ( and sets the status + cancelledAt )
• a new event job:cancelled
• a new event job:cancel:{jobId}

To process the cancellation, we're extending the processJob and processBatchJobs and listen to the job:cancel:${jobId} event and stop the processing ( inside processJob we just have to extend the Promise.race.

For processBatchJobs, we have to modify the current implementation ( maybe, we just have to call processJob from processBatchJobs 🤔 ).

it's similar to the current timeout implementation.

---

Note: We should also check if an AbortController ( as @maelp mention here: #41 (comment) ) could work.

[noxify]

based on some prompts, possible implementation could be something like this:

private async processJob(job: BaseJob): Promise<void> {
  const handler = this.handlers.get(job.name)
  if (!handler) {
    await this.failJob(job, serializeError(new Error(`No handler registered for job: ${job.name}`)))
    return
  }

  await this.adapter.updateJobStatus(job.id, "processing")
  this.emit("job:processing", { ...job, status: "processing" })

  const controller = new AbortController()
  const { signal } = controller

  // Timeout handling
  const timeout = job.timeout ?? this.config.defaultJobOptions.timeout ?? 30000
  const timeoutId = setTimeout(() => controller.abort("timeout"), timeout)

  // Cancel event handling
  const cancelListener = (cancelledJob: BaseJob) => {
    if (cancelledJob.id === job.id) controller.abort("cancel")
  }
  this.on("job:cancelled", cancelListener)

  try {
    const result = await handler(createJobWrapper(job, this), signal)
    if (signal.aborted) {
      if (signal.reason === "cancel") {
        await this.adapter.updateJobStatus(job.id, "cancelled")
        this.emit("job:cancelled", { ...job, status: "cancelled", cancelledAt: new Date() })
        return
      }
      if (signal.reason === "timeout") {
        // Retry nur bei Timeout
        throw new Error("Job timeout")
      }
    }
    await this.adapter.updateJobStatus(job.id, "completed", undefined, result)
    this.emit("job:completed", { ...job, status: "completed", completedAt: new Date(), result })
    await this.cleanupCompletedJob()
    this.handleRecurringJob(job).catch(() => {})
  } catch (error) {
    if (signal.aborted && signal.reason === "timeout") {
      await this.handleJobError(job, error)
    } else if (signal.aborted && signal.reason === "cancel") {
      // Kein Retry, Job als cancelled markieren
      await this.adapter.updateJobStatus(job.id, "cancelled")
      this.emit("job:cancelled", { ...job, status: "cancelled", cancelledAt: new Date() })
    } else {
      await this.handleJobError(job, error)
    }
  } finally {
    clearTimeout(timeoutId)
    this.listeners.get("job:cancelled")?.splice(
      this.listeners.get("job:cancelled")?.indexOf(cancelListener) ?? -1, 1
    )
  }
}