index.ts

import { AdminForthPlugin, Filters, Sorts } from "adminforth";
import type { IAdminForth, IHttpServer, AdminForthResource, AdminUser, AdminForthComponentDeclarationFull } from "adminforth";
import type { PluginOptions } from './types.js';
import { afLogger } from "adminforth";
import pLimit from 'p-limit';
import { Level } from 'level';
import fs from 'fs/promises';
import { Mutex } from 'async-mutex';
import { z } from "zod";

const jobIdBodySchema = z.object({
  jobId: z.union([z.string(), z.number()]),
}).strict();

const getTasksBodySchema = z.object({
  jobId: z.union([z.string(), z.number()]),
  limit: z.number().int().positive().max(100),
  offset: z.number().int().nonnegative(),
  fieldsToReturn: z.array(z.string()).optional(),
}).strict();

type TaskStatus = 'SCHEDULED' | 'IN_PROGRESS' | 'DONE' | 'FAILED' | 'DELETED';
type JobStatus = 'QUEUED' | 'IN_PROGRESS' | 'DONE' | 'DONE_WITH_ERRORS' | 'CANCELLED';
const TERMINAL_JOB_STATUSES: JobStatus[] = ['DONE', 'DONE_WITH_ERRORS', 'CANCELLED'];
const DEFAULT_PARALLEL_LIMIT = 3;
const DEFAULT_CONCURRENCY_LIMIT = 1;
const LEVEL_DB_IDLE_CLOSE_MS = 1000 * 60 * 60 * 4; // 4 hours
type setStateFieldParams = {
  (fieldName: string, value: any): Promise<void>;
  (state: Record<string, any>): Promise<void>;
};
type getStateFieldParams = {
  (fieldName: string): Promise<any>;
  (): Promise<Record<string, any>>;
};
type getStateParams = () => Promise<Record<string, any>>;
type taskHandlerType = ( { jobId, setTaskStateField, getTaskStateField, getState }: { jobId: string; setTaskStateField: setStateFieldParams; getTaskStateField: getStateFieldParams; getState: getStateParams } ) => Promise<void>;
type allTasksDoneStatusType = {
  jobId: string;
  failedTasks: number;
  succeededTasks: number;
};
type beforeJobFinishStatusType = allTasksDoneStatusType & {
  finishAttemptNumber: number;
};
type onAllTasksDoneType = (status: allTasksDoneStatusType) => Promise<void> | void;
type beforeJobFinishType = (status: beforeJobFinishStatusType) => Promise<void> | void;
type taskType = {
  skip?: boolean;
  state: Record<string, any>;
}
type indexedTaskType = taskType & {
  taskIndex: number;
}
type startNewJobOptions = {
  /**
   * When true the job is always created in QUEUED status, even when its queue has a free concurrency slot.
   * Without it the job is created in QUEUED status only when the queue is busy.
   */
  queued?: boolean;
  /**
   * When false a job which landed in QUEUED status stays there until something else drains the queue
   * (startNextQueuedJob, a finishing job of the same queue, or an application restart). Default: true.
   */
  autoStart?: boolean;
}
/**
 * Everything needed to run the tasks of a single job. Kept as one object so it can be passed through the
 * recursive task processing without a long positional argument list.
 */
type jobRunContext = {
  jobHandlerName: string;
  handleTask: taskHandlerType;
  parrallelLimit: number;
  onAllTasksDone?: onAllTasksDoneType;
  beforeJobFinish?: beforeJobFinishType;
}

function encodeStateFieldName(fieldName: string): string {
  return encodeURIComponent(fieldName);
}
 
export default class BackgroundJobsPlugin extends AdminForthPlugin {
  options: PluginOptions;
  private taskHandlers: Record<string, taskHandlerType> = {};
  private onAllTasksDoneHandlers: Partial<Record<string, onAllTasksDoneType>> = {};
  private beforeJobFinishHandlers: Partial<Record<string, beforeJobFinishType>> = {};
  private jobCustomComponents: Record<string, AdminForthComponentDeclarationFull> = {};
  private jobParallelLimits: Record<string, number> = {};
  private jobConcurrencyLimits: Record<string, number> = {};
  private levelDbInstances: Record<string, Level> = {};
  private levelDbJobIds = new WeakMap<Level, string>();
  private levelDbLastAccessAt: Record<string, number> = {};
  private levelDbCloseTimers: Record<string, ReturnType<typeof setTimeout>> = {};
  private jobStateMutexes: Record<string, Mutex> = {};
  private jobQueueMutexes: Record<string, Mutex> = {};
  private jobTasksMutexes: Record<string, Mutex> = {};
  private addTasksToRunningJob: Record<string, (firstTaskIndex: number, tasks: taskType[]) => void> = {};
  private deprecatedWarningsShown = new Set<string>();

  constructor(options: PluginOptions) {
    super(options, import.meta.url);
    this.options = options;
    this.shouldHaveSingleInstancePerWholeApp = () => true;
  }

  private getResourcePk(): string {
    const resourcePk = this.resourceConfig.columns.find(c => c.primaryKey)?.name;
    return resourcePk;
  }

  private getResourceId(): string {
    return this.resourceConfig.resourceId;
  }

  async modifyResourceConfig(adminforth: IAdminForth, resourceConfig: AdminForthResource) {
    super.modifyResourceConfig(adminforth, resourceConfig);
    if (!adminforth.config.customization?.globalInjections?.header) {
      adminforth.config.customization.globalInjections.header = [];
    }
    (adminforth.config.customization.globalInjections.header).push({
      file: this.componentPath('NavbarJobs.vue'),
      meta: {
        pluginInstanceId: this.pluginInstanceId,
      }
    });

    // Global API injection: exposes OpenJobInfoPopup(jobId) to open job details from anywhere
    (adminforth.config.customization.globalInjections.header).push({
      file: this.componentPath('GlobalJobApi.vue'),
      meta: {
        pluginInstanceId: this.pluginInstanceId,
      }
    });

    if (!this.adminforth.config.componentsToExplicitRegister) {
      this.adminforth.config.componentsToExplicitRegister = [];
    }
    this.adminforth.config.componentsToExplicitRegister.push(
      {
        file: this.componentPath('StateToIcon.vue')
      }
    );

    if (!this.resourceConfig.hooks) {
      this.resourceConfig.hooks = {};
    }
    if (!this.resourceConfig.hooks.delete) {
      this.resourceConfig.hooks.delete = {};
    }
    if (!this.resourceConfig.hooks.delete.beforeSave) {
      this.resourceConfig.hooks.delete.beforeSave = [];
    }
    this.resourceConfig.hooks.delete.beforeSave.push(async ({record, recordId}: {record: any, recordId: any}) => {

      const levelDbPath = `${this.options.levelDbPath || './background-jobs-dbs/'}job_${recordId}`;
      const jobLevelDb = this.levelDbInstances[recordId];

      //close level db instance if it's open and delete the level db folder for the job
      if (jobLevelDb) {
        await jobLevelDb.close();
        this.forgetLevelDb(recordId);
      }

      // cleanup per-job mutexes as well
      delete this.jobTasksMutexes[recordId];

      //delete level db folder for the job
      await fs.rm(levelDbPath, {
        recursive: true,
        force: true,
      });

      return {ok: true};
    })
  }

  private cleanupJobMutexIfTerminalStatus(jobId: string, status: JobStatus) {
    // Keep mutex while job is active to preserve atomicity between concurrent tasks.
    if (TERMINAL_JOB_STATUSES.includes(status)) {
      // safe to drop even while held: the terminal status is already stored, so a caller who creates a new
      // mutex for this job reads that status and rejects the new tasks
      delete this.jobTasksMutexes[jobId];
    }
  }

  private checkIfFieldInResource(resourceConfig: AdminForthResource, fieldName: string, fieldString?: string) {
    if (!fieldName) {
      throw new Error(`Field name for ${fieldString} is not provided. Please check your plugin options.`);
    }
    const fieldInConfig = resourceConfig.columns.find(f => f.name === fieldName);
    if (!fieldInConfig) {
      throw new Error(`Field ${fieldName} not found in resource config. Please check your plugin options.`);
    }
  }

  private async createLevelDbTaskRecord(levelDb: Level, taskId: string, initialState: Record<string, any>) {
    //create record in level db with task id as key and initial state as value and status SCHEDULED
    this.touchLevelDb(levelDb);
    await levelDb.put(taskId, JSON.stringify({ state: initialState, status: 'SCHEDULED' }));
  }

  private async setLevelDbTaskStateField(levelDb: Level, taskId: string, state: Record<string, any>) {
    //update record in level db with task id as key and new state as value
    this.touchLevelDb(levelDb);
    const status = await this.getLevelDbTaskStatusField(levelDb, taskId);
    await levelDb.del(taskId);
    await levelDb.put(taskId, JSON.stringify({ state, status }));
  }

  private async setLevelDbTaskStatusField(levelDb: Level, taskId: string, status: TaskStatus) {
    this.touchLevelDb(levelDb);
    const state = await this.getLevelDbTaskStateField(levelDb, taskId);
    await levelDb.del(taskId);
    await levelDb.put(taskId, JSON.stringify({ state, status }));
  }

  private async getLevelDbTaskStateField(levelDb: Level, taskId: string): Promise<Record<string, any>> {
    //get record from level db with task id as key and return the value of the key in the state
    this.touchLevelDb(levelDb);
    const state = await levelDb.get(taskId);
    if (state) {
      const parsedState = JSON.parse(state);
      return parsedState.state;
    }
    return Promise.resolve(null);
  }

  private async getLevelDbTaskStatusField(levelDb: Level, taskId: string): Promise<TaskStatus> {
    this.touchLevelDb(levelDb);
    const state = await levelDb.get(taskId);
    if (state) {
      const parsedState = JSON.parse(state);
      return parsedState.status;
    }
    return Promise.resolve(null);
  }

  private async getTotalTasksInLevelDb(levelDb: Level): Promise<number> {
    this.touchLevelDb(levelDb);
    const count = await levelDb.get('_meta:count');
    return count ? parseInt(count, 10) : 0;
  }

  private async getAllTasksDoneStatus(levelDb: Level): Promise<Omit<allTasksDoneStatusType, 'jobId'>> {
    const totalTasks = await this.getTotalTasksInLevelDb(levelDb);
    let failedTasks = 0;
    let succeededTasks = 0;

    for (let taskIndex = 0; taskIndex < totalTasks; taskIndex++) {
      const status = await this.getLevelDbTaskStatusField(levelDb, taskIndex.toString());
      if (status === 'FAILED') {
        failedTasks++;
      } else if (status === 'DONE') {
        succeededTasks++;
      }
    }

    return { failedTasks, succeededTasks };
  }

  private async getLevelDbForTheJob(jobId: string): Promise<Level> {
    const levelDbPath = `${this.options.levelDbPath || './background-jobs-dbs/'}job_${jobId}`;
    let jobLevelDb: Level;
    if (this.levelDbInstances[jobId]) {
      jobLevelDb = this.levelDbInstances[jobId];
    } else {
      try {
        jobLevelDb = new Level(levelDbPath, { valueEncoding: 'json' });
        this.levelDbInstances[jobId] = jobLevelDb;
        this.levelDbJobIds.set(jobLevelDb, jobId);
      } catch (error) {
        throw new Error(`Failed to access task storage for job with id ${jobId}.`);
      }
    }
    this.touchLevelDb(jobLevelDb);
    this.scheduleLevelDbClose(jobId);
    return jobLevelDb;
  }

  /**
   * Marks the LevelDB of a job as used right now, so the idle close timer does not close a db which is
   * still being read/written (for example by a job running longer than the idle window).
   */
  private touchLevelDb(levelDb: Level) {
    const jobId = this.levelDbJobIds.get(levelDb);
    if (jobId) {
      this.levelDbLastAccessAt[jobId] = Date.now();
    }
  }

  private scheduleLevelDbClose(jobId: string, delay: number = LEVEL_DB_IDLE_CLOSE_MS) {
    if (this.levelDbCloseTimers[jobId]) {
      clearTimeout(this.levelDbCloseTimers[jobId]);
    }
    this.levelDbCloseTimers[jobId] = setTimeout(() => { this.closeLevelDbIfIdle(jobId); }, delay);
  }

  private async closeLevelDbIfIdle(jobId: string) {
    const jobLevelDb = this.levelDbInstances[jobId];
    if (!jobLevelDb) {
      this.forgetLevelDb(jobId);
      return;
    }

    const idleFor = Date.now() - (this.levelDbLastAccessAt[jobId] ?? 0);
    if (idleFor < LEVEL_DB_IDLE_CLOSE_MS) {
      // db was used after this timer was scheduled, so wait for the rest of the idle window
      this.scheduleLevelDbClose(jobId, LEVEL_DB_IDLE_CLOSE_MS - idleFor);
      return;
    }

    try {
      await jobLevelDb.close();
    } catch (error) {
      afLogger.error(`Failed to close LevelDB for job ${jobId}: ${error instanceof Error ? error.message : String(error)}`);
      // keep the instance and try again after another idle window
      this.scheduleLevelDbClose(jobId);
      return;
    }
    this.forgetLevelDb(jobId);
  }

  private forgetLevelDb(jobId: string) {
    if (this.levelDbCloseTimers[jobId]) {
      clearTimeout(this.levelDbCloseTimers[jobId]);
    }
    delete this.levelDbCloseTimers[jobId];
    delete this.levelDbLastAccessAt[jobId];
    delete this.levelDbInstances[jobId];
  }

  private publishJobStateField(jobId: string, fieldName: string, value: any) {
    this.adminforth.websocket.publish(`/background-jobs-state-update/${jobId}/${encodeStateFieldName(fieldName)}`, {
      jobId,
      fieldName,
      value,
    });
  }

  private publishJobUpdate(data: Record<string, any>) {
    this.adminforth.websocket.publish('/background-jobs-job-update', data);
  }

  private async getJobById(jobId: string | number): Promise<Record<string, any> | null> {
    return this.adminforth.resource(this.getResourceId()).get(Filters.EQ(this.getResourcePk(), jobId));
  }

  private publishTaskStateFields(jobId: string, taskIndex: number, state: Record<string, any>) {
    for (const [fieldName, value] of Object.entries(state)) {
      this.adminforth.websocket.publish(`/background-jobs-task-state-update/${jobId}/${encodeStateFieldName(fieldName)}`, {
        jobId,
        taskIndex,
        fieldName,
        value,
      });
    }
  }

  private warnDeprecatedOnce(key: string, message: string) {
    if (this.deprecatedWarningsShown.has(key)) {
      return;
    }

    this.deprecatedWarningsShown.add(key);
    afLogger.warn(message);
  }

  private async triggerOnAllTasksDone(onAllTasksDone: onAllTasksDoneType | undefined, levelDb: Level, jobId: string) {
    if (!onAllTasksDone) {
      return;
    }

    try {
      const status = await this.getAllTasksDoneStatus(levelDb);
      await onAllTasksDone({ jobId, ...status });
    } catch (error) {
      const errorMessage = error instanceof Error ? error.message : String(error);
      afLogger.error(`Error in onAllTasksDone callback for job ${jobId}: ${errorMessage}`);
    }
  }

  private async triggerBeforeJobFinish(beforeJobFinish: beforeJobFinishType | undefined, levelDb: Level, jobId: string, finishAttemptNumber: number) {
    if (!beforeJobFinish) {
      return;
    }

    try {
      const status = await this.getAllTasksDoneStatus(levelDb);
      await beforeJobFinish({ jobId, ...status, finishAttemptNumber });
    } catch (error) {
      const errorMessage = error instanceof Error ? error.message : String(error);
      afLogger.error(`Error in beforeJobFinish callback for job ${jobId}: ${errorMessage}`);
      throw error;
    }
  }
  
  public registerTaskHandler({ jobHandlerName, handler, parallelLimit = DEFAULT_PARALLEL_LIMIT, concurrencyLimit = DEFAULT_CONCURRENCY_LIMIT, onAllTasksDone, beforeJobFinish,
  }:{jobHandlerName: string, handler: taskHandlerType, parallelLimit?: number, concurrencyLimit?: number, onAllTasksDone?: onAllTasksDoneType, beforeJobFinish?: beforeJobFinishType}) {
    if (!Number.isInteger(concurrencyLimit) || concurrencyLimit < 1) {
      throw new Error(`concurrencyLimit for jobHandler ${jobHandlerName} must be an integer greater than 0, got ${concurrencyLimit}.`);
    }
    //register the handler in a map with jobHandlerName as key and handler as value
    this.taskHandlers[jobHandlerName] = handler;
    this.jobParallelLimits[jobHandlerName] = parallelLimit;
    this.jobConcurrencyLimits[jobHandlerName] = concurrencyLimit;
    if (onAllTasksDone) {
      this.onAllTasksDoneHandlers[jobHandlerName] = onAllTasksDone;
    } else {
      delete this.onAllTasksDoneHandlers[jobHandlerName];
    }
    if (beforeJobFinish) {
      this.beforeJobFinishHandlers[jobHandlerName] = beforeJobFinish;
    } else {
      delete this.beforeJobFinishHandlers[jobHandlerName];
    }
  }

  public registerTaskDetailsComponent({
    jobHandlerName,
    component,
  }:{jobHandlerName: string, component: AdminForthComponentDeclarationFull}) {
    this.jobCustomComponents[jobHandlerName] = component;
  }


  private buildJobRunContext(jobHandlerName: string): jobRunContext | null {
    const handleTask: taskHandlerType = this.taskHandlers[jobHandlerName];
    if (!handleTask) {
      return null;
    }
    return {
      jobHandlerName,
      handleTask,
      parrallelLimit: this.jobParallelLimits[jobHandlerName] || DEFAULT_PARALLEL_LIMIT,
      onAllTasksDone: this.onAllTasksDoneHandlers[jobHandlerName],
      beforeJobFinish: this.beforeJobFinishHandlers[jobHandlerName],
    };
  }

  private async canStartJobImmediately(jobHandlerName: string): Promise<boolean> {
    const concurrencyLimit = this.jobConcurrencyLimits[jobHandlerName] || DEFAULT_CONCURRENCY_LIMIT;
    const runningJobsCount = await this.getRunningJobsCount(jobHandlerName);
    if (runningJobsCount >= concurrencyLimit) {
      return false;
    }
    const [queuedJob] = await this.listJobsByStatus(jobHandlerName, 'QUEUED', 1);
    return !queuedJob;
  }

  public async startNewJob(
    jobName: string,
    adminUser: AdminUser | null,
    tasks: taskType[],
    jobHandlerName: string,
    initialState: Record<string, any> = {},
    options: startNewJobOptions = {},
  ): Promise<string> {
    const jobRunContext = this.buildJobRunContext(jobHandlerName);
    if (!jobRunContext) {
      throw new Error(`No handler registered for jobHandler ${jobHandlerName}. Please register a handler using the registerTaskHandler method before starting a job with this jobHandler.`);
    }
    const customComponent = this.jobCustomComponents[jobHandlerName];
    const { parrallelLimit } = jobRunContext;
    const startedBy = adminUser?.pk ?? null;

    const { createdRecord, initialStatus, jobLevelDb } = await this.getQueueMutex(jobHandlerName).runExclusive(async () => {
      const initialStatus: JobStatus = !options.queued && await this.canStartJobImmediately(jobHandlerName)
        ? 'IN_PROGRESS'
        : 'QUEUED';
      //create a record for the job in the database with status in progress (or queued, when the queue is busy)
      const objectToSave = {
        [this.options.nameField]: jobName,
        [this.options.startedByField]: startedBy,
        [this.options.progressField]: 0,
        [this.options.statusField]: initialStatus,
        [this.options.jobHandlerField]: jobHandlerName,
        [this.options.stateField]: initialState
      }

      const creationResult = await this.adminforth.resource(this.getResourceId()).create(objectToSave);
      if (creationResult.ok !== true) {
        throw new Error(`Failed to create a record for the job. Error: ${creationResult.error}`);
      }
      const createdRecord = creationResult.createdRecord as Record<string, any>;

      // tasks are stored before the queue mutex is released, otherwise startNextQueuedJob could take the QUEUED job
      // while it has no tasks yet and finish it right away, leaving the tasks written after that unprocessed
      //create a level db instance for the job with name as jobId
      const jobLevelDb = await this.getLevelDbForTheJob(createdRecord[this.getResourcePk()]);
      await jobLevelDb.put('_meta:count', `${tasks.length}`);
      const limit2 = pLimit(parrallelLimit);
      const createTaskRecordsPromises = tasks.map((task, index) => {
        return limit2(() => this.createLevelDbTaskRecord(jobLevelDb, index.toString(), task.state));
      });

      await Promise.all(createTaskRecordsPromises);
      return { createdRecord, initialStatus, jobLevelDb };
    });
    const jobId = createdRecord[this.getResourcePk()];

    this.publishJobUpdate({
      jobId,
      status: initialStatus,
      name: jobName,
      progress: 0,
      createdAt: createdRecord[this.options.createdAtField],
      customComponent,
    });

    if (initialStatus === 'IN_PROGRESS') {
      this.runProcessingTasks(tasks, jobLevelDb, jobId, jobRunContext).catch((error) => {
        void this.handleJobProcessingFailure(jobId, jobHandlerName, error);
      });
      return jobId;
    }

    if (options.autoStart !== false) {
      await this.startQueuedJobs(jobHandlerName);
    }
    return jobId;
  }

  public async queueNewJob(
    jobName: string,
    adminUser: AdminUser | null,
    tasks: taskType[],
    jobHandlerName: string,
    initialState: Record<string, any> = {},
    options: Omit<startNewJobOptions, 'queued'> = {},
  ): Promise<string> {
    return this.startNewJob(jobName, adminUser, tasks, jobHandlerName, initialState, { ...options, queued: true });
  }


  private getQueueMutex(jobHandlerName: string): Mutex {
    let mutex = this.jobQueueMutexes[jobHandlerName];
    if (!mutex) {
      mutex = new Mutex();
      this.jobQueueMutexes[jobHandlerName] = mutex;
    }
    return mutex;
  }

  /**
   * Taken by addNewTasksToExistingJob and by the last unfinished-task scan with the terminal status write of a job,
   * so a task is either added before that scan and processed, or rejected because the job is already finished.
   */
  private getJobTasksMutex(jobId: string): Mutex {
    let mutex = this.jobTasksMutexes[jobId];
    if (!mutex) {
      mutex = new Mutex();
      this.jobTasksMutexes[jobId] = mutex;
    }
    return mutex;
  }

  private async listJobsByStatus(
    jobHandlerName: string,
    status: JobStatus,
    limit: number | null = null,
  ): Promise<Record<string, any>[]> {
    return this.adminforth.resource(this.getResourceId()).list(
      Filters.AND(
        Filters.EQ(this.options.jobHandlerField, jobHandlerName),
        Filters.EQ(this.options.statusField, status),
      ),
      limit,
      0,
      Sorts.ASC(this.options.createdAtField),
    );
  }


  private async getRunningJobsCount(jobHandlerName: string): Promise<number> {
    const runningJobs = await this.listJobsByStatus(jobHandlerName, 'IN_PROGRESS');
    return runningJobs.length;
  }


  public async startNextQueuedJob(jobHandlerName: string): Promise<string | null> {
    return this.getQueueMutex(jobHandlerName).runExclusive(async () => {
      const jobRunContext = this.buildJobRunContext(jobHandlerName);
      if (!jobRunContext) {
        afLogger.error(`No handler registered for jobHandler ${jobHandlerName}. Cannot start queued jobs for this jobHandler.`);
        return null;
      }

      const concurrencyLimit = this.jobConcurrencyLimits[jobHandlerName] || DEFAULT_CONCURRENCY_LIMIT;
      const runningJobsCount = await this.getRunningJobsCount(jobHandlerName);
      if (runningJobsCount >= concurrencyLimit) {
        return null;
      }

      const [oldestQueuedJob] = await this.listJobsByStatus(jobHandlerName, 'QUEUED', 1);
      if (!oldestQueuedJob) {
        return null;
      }

      const jobId = oldestQueuedJob[this.getResourcePk()];
      const updateResult = await this.adminforth.resource(this.getResourceId()).update(jobId, {
        [this.options.statusField]: 'IN_PROGRESS',
      });
      if (updateResult?.ok === false) {
        afLogger.error(`Failed to start queued job ${jobId}: ${updateResult.error}`);
        return null;
      }

      this.publishJobUpdate({
        jobId,
        status: 'IN_PROGRESS',
        name: oldestQueuedJob[this.options.nameField],
        progress: oldestQueuedJob[this.options.progressField],
        createdAt: oldestQueuedJob[this.options.createdAtField],
        customComponent: this.jobCustomComponents[jobHandlerName],
      });

      // not awaited on purpose: processing runs until the whole job is finished, while the queue mutex
      // only has to cover taking the slot
      this.runProcessingJobTasksFromStorage(oldestQueuedJob, jobRunContext).catch((error) => {
        void this.handleJobProcessingFailure(
          jobId,
          jobHandlerName,
          error,
        );
      });

      return jobId;
    });
  }


  public async startQueuedJobs(jobHandlerName: string): Promise<string[]> {
    const startedJobIds: string[] = [];
    while (true) {
      const startedJobId = await this.startNextQueuedJob(jobHandlerName);
      if (!startedJobId) {
        return startedJobIds;
      }
      startedJobIds.push(startedJobId);
    }
  }

  private async startQueuedJobsSafely(jobHandlerName: string) {
    if (!jobHandlerName) {
      return;
    }
    try {
      await this.startQueuedJobs(jobHandlerName);
    } catch (error) {
      const errorMessage = error instanceof Error ? error.message : String(error);
      afLogger.error(`Failed to start queued jobs for jobHandler ${jobHandlerName}: ${errorMessage}`);
    }
  }

  public async addNewTasksToExistingJob(
    jobId: string,
    tasks: taskType[],
  ) {
    await this.getJobTasksMutex(jobId).runExclusive(async () => {
      const jobRecord = await this.adminforth.resource(this.getResourceId()).get(Filters.EQ(this.getResourcePk(), jobId));
      if (!jobRecord) {
        throw new Error(`Job with id ${jobId} not found.`);
      }
      const jobStatus = jobRecord[this.options.statusField];
      if (jobStatus !== 'IN_PROGRESS' && jobStatus !== 'QUEUED') {
        throw new Error(`Cannot add tasks to a job with status ${jobStatus}. Only jobs with status IN_PROGRESS or QUEUED can be added new tasks.`);
      }
      const jobLevelDb = await this.getLevelDbForTheJob(jobId);
      const currentTotalTasks = await this.getTotalTasksInLevelDb(jobLevelDb);
      const newTotalTasks = currentTotalTasks + tasks.length;
      await jobLevelDb.put('_meta:count', `${newTotalTasks}`);
      const createTaskRecordsPromises = tasks.map((task, index) => {
        return this.createLevelDbTaskRecord(jobLevelDb, (currentTotalTasks + index).toString(), task.state);
      });

      await Promise.all(createTaskRecordsPromises);
      // a running job takes the new tasks into free slots of its parallel limit right away, otherwise they are
      // picked up from storage when the job starts or rescans its unfinished tasks
      this.addTasksToRunningJob[jobId]?.(currentTotalTasks, tasks);
    });
  }

  public async deleteTasksFromExistingJob(
    jobId: string,
    taskIndex: number,
  ): Promise<void> {
    if (taskIndex < 0) {
      throw new Error(`Invalid task index ${taskIndex}.`);
    }
    const jobRecord = await this.adminforth.resource(this.getResourceId()).get(Filters.EQ(this.getResourcePk(), jobId));
    if (!jobRecord) {
      throw new Error(`Job with id ${jobId} not found.`);
    }
    const jobStatus = jobRecord[this.options.statusField];
    if (jobStatus !== 'IN_PROGRESS' && jobStatus !== 'QUEUED') {
      throw new Error(`Cannot delete tasks from a job with status ${jobStatus}. Only jobs with status IN_PROGRESS or QUEUED can have tasks deleted.`);
    }
    const jobLevelDb = await this.getLevelDbForTheJob(jobId);
    const currentTotalTasks = await this.getTotalTasksInLevelDb(jobLevelDb);
    if (taskIndex >= currentTotalTasks) {
      throw new Error(`Invalid task index ${taskIndex}.`);
    }
    const taskStatus = await this.getLevelDbTaskStatusField(jobLevelDb, taskIndex.toString());
    if (taskStatus === 'IN_PROGRESS') {
      throw new Error(`Cannot delete task ${taskIndex} because it is already in progress.`);
    }
    // running workers address their task by its index, so the task is only marked as deleted and keeps its
    // index, removing it would shift the following tasks under the workers which are processing them
    await this.setLevelDbTaskStatusField(jobLevelDb, taskIndex.toString(), 'DELETED');
    this.adminforth.websocket.publish(`/background-jobs-task-update/${jobId}`, { taskIndex, status: 'DELETED' });
  }

  private async getUnfinishedTasksFromLevelDb(levelDb: Level): Promise<indexedTaskType[]> {
    const totalTasks = await this.getTotalTasksInLevelDb(levelDb);
    const unfinishedTasks: indexedTaskType[] = [];
    for (let taskIndex = 0; taskIndex < totalTasks; taskIndex++) {
      const status = await this.getLevelDbTaskStatusField(levelDb, taskIndex.toString());
      if (status === 'IN_PROGRESS' || status === 'SCHEDULED') {
        const state = await this.getLevelDbTaskStateField(levelDb, taskIndex.toString());
        unfinishedTasks.push({ state, taskIndex });
      }
    }
    return unfinishedTasks;
  }

  private async buildTasksToReprocess(levelDb: Level, unfinishedTasks: indexedTaskType[]): Promise<taskType[]> {
    const totalTasks = await this.getTotalTasksInLevelDb(levelDb);
    const tasksToReprocess: taskType[] = Array.from(
      { length: totalTasks },
      () => ({ skip: true, state: {} }),
    );
    for (const { taskIndex, state } of unfinishedTasks) {
      tasksToReprocess[taskIndex] = { state };
    }
    return tasksToReprocess;
  }

  private async runProcessingTasks(
    tasks: taskType[],
    jobLevelDb: Level,
    jobId: string,
    jobRunContext: jobRunContext,
    finishAttemptNumber = 0,
  ) {
    const { jobHandlerName, handleTask, parrallelLimit, onAllTasksDone, beforeJobFinish } = jobRunContext;
    let totalTasks = tasks.length;
    let completedTasks = 0;
    let lastJobStatus = 'IN_PROGRESS';

    const progressMutex = new Mutex();
    const finishTask = (wasTaskSkipped: boolean = false) => progressMutex.runExclusive(async () => {
      completedTasks = await this.handleFinishTask(completedTasks, totalTasks, jobId, wasTaskSkipped);
    });

    const taskHandler = async ( taskIndex: number, task: taskType ) => {
      totalTasks = await this.getTotalTasksInLevelDb(jobLevelDb);
      if (task.skip) {
        await finishTask(true);
        return;
      }
      if (lastJobStatus === 'CANCELLED') {
        afLogger.debug(`Job ${jobId} was cancelled. Skipping task ${taskIndex}.`);
        return;
      }
      const currentJobStatus = await this.getLastJobStatus(jobId);

      if (currentJobStatus === 'CANCELLED') {
        lastJobStatus = currentJobStatus;
        afLogger.debug(`Job ${jobId} was cancelled. Skipping task ${taskIndex}.`);
        return;
      }
      // the task can be deleted while it waits for a free slot
      const taskStatus = await this.getLevelDbTaskStatusField(jobLevelDb, taskIndex.toString());
      if (taskStatus === 'DELETED') {
        afLogger.debug(`Task ${taskIndex} of job ${jobId} was deleted. Skipping processing.`);
        // a deleted task stays in the total task count, so it has to be counted as finished for the progress
        await finishTask();
        return;
      }
      const getState = async () => {
        return await this.getLevelDbTaskStateField(jobLevelDb, taskIndex.toString());
      }
      const setTaskStateField: setStateFieldParams = async (fieldNameOrState: string | Record<string, any>, value?: any) => {
        if (typeof fieldNameOrState === 'string') {
          const state = await getState();
          const updatedState = {
            ...state,
            [fieldNameOrState]: value,
          };
          await this.setLevelDbTaskStateField(jobLevelDb, taskIndex.toString(), updatedState);
          this.publishTaskStateFields(jobId, taskIndex, { [fieldNameOrState]: value });
          return;
        }

        this.warnDeprecatedOnce(
          'setTaskStateField-object',
          'BackgroundJobsPlugin: setTaskStateField(stateObject) is deprecated and will be removed soon. Use setTaskStateField(fieldName: string, value: any) instead. Use getState() when you need the full task state.',
        );
        await this.setLevelDbTaskStateField(jobLevelDb, taskIndex.toString(), fieldNameOrState);
        this.publishTaskStateFields(jobId, taskIndex, fieldNameOrState);
      }
      const getTaskStateField: getStateFieldParams = async (fieldName?: string) => {
        const state = await getState();
        if (typeof fieldName === 'string') {
          return state[fieldName];
        }

        this.warnDeprecatedOnce(
          'getTaskStateField-no-args',
          'BackgroundJobsPlugin: getTaskStateField() without a field name is deprecated and will be removed soon. Use getTaskStateField(fieldName: string) for one field, or getState() for the full task state.',
        );
        return state;
      }

      await this.setLevelDbTaskStatusField(jobLevelDb, taskIndex.toString(), 'IN_PROGRESS');
      this.adminforth.websocket.publish(`/background-jobs-task-update/${jobId}`, { taskIndex, status: "IN_PROGRESS" });

      //handling the task 
      try {
        await handleTask({ jobId, setTaskStateField, getTaskStateField, getState });

        //Set task status to completed in level db
        await this.setLevelDbTaskStatusField(jobLevelDb, taskIndex.toString(), 'DONE');
        this.adminforth.websocket.publish(`/background-jobs-task-update/${jobId}`, { taskIndex, status: "DONE" });
      } catch (error) {
        const errorMessage = error?.message || 'Unknown error';
        afLogger.error(`Error in handling task ${taskIndex} of job ${jobId}: ${errorMessage}`, );
        await this.setJobStateField(jobId, 'error', errorMessage);
        // persist the error inside the task state so it is returned when tasks are fetched later
        const taskState = await this.getLevelDbTaskStateField(jobLevelDb, taskIndex.toString());
        await this.setLevelDbTaskStateField(jobLevelDb, taskIndex.toString(), { ...taskState, error: errorMessage });
        await this.setLevelDbTaskStatusField(jobLevelDb, taskIndex.toString(), 'FAILED');
        this.adminforth.websocket.publish(`/background-jobs-task-update/${jobId}`, { taskIndex, status: "FAILED", error: errorMessage });
        return;
      } finally {
        //Update progress
        const currentJobStatus = await this.getLastJobStatus(jobId);
        if (currentJobStatus === 'CANCELLED') {
          lastJobStatus = currentJobStatus;
          afLogger.debug(`Job ${jobId} was cancelled during processing of task ${taskIndex}. Progress will not be updated.`);
          return;
        }

        await finishTask();
      }
    }

    const limit = pLimit(parrallelLimit);
    let pendingTasks: Promise<void>[] = [];
    const enqueueTask = (taskIndex: number, task: taskType) => {
      pendingTasks.push(limit(() => taskHandler(taskIndex, task)));
    };
    tasks.forEach((task, taskIndex) => enqueueTask(taskIndex, task));

    this.addTasksToRunningJob[jobId] = (firstTaskIndex, newTasks) => {
      newTasks.forEach((task, offset) => {
        const taskIndex = firstTaskIndex + offset;
        // a run built from storage after the task was saved already has it
        if (taskIndex >= tasks.length) {
          enqueueTask(taskIndex, task);
        }
      });
    };
    try {
      // tasks enqueued while a batch runs form the next batch, so the loop ends only when nothing is left
      while (pendingTasks.length > 0) {
        const tasksBatch = pendingTasks;
        pendingTasks = [];
        await Promise.all(tasksBatch);
      }
    } catch (error) {
      // tasks enqueued after the failed batch are not awaited anymore, so their rejections must not go unhandled
      pendingTasks.forEach((pendingTask) => pendingTask.catch(() => {}));
      throw error;
    } finally {
      delete this.addTasksToRunningJob[jobId];
    }

    if (lastJobStatus === 'CANCELLED') {
      this.cleanupJobMutexIfTerminalStatus(jobId, 'CANCELLED');
      await this.startQueuedJobsSafely(jobHandlerName);
      return;
    }
    const jobStatusAfterExecution = await this.getLastJobStatus(jobId);
    if (jobStatusAfterExecution === 'CANCELLED') {
      this.cleanupJobMutexIfTerminalStatus(jobId, 'CANCELLED');
      await this.startQueuedJobsSafely(jobHandlerName);
      return;
    }

    const unfinishedTasks = await this.getUnfinishedTasksFromLevelDb(jobLevelDb);
    if (unfinishedTasks.length > 0) {
      const tasksToReprocess = await this.buildTasksToReprocess(jobLevelDb, unfinishedTasks);
      await this.runProcessingTasks(tasksToReprocess, jobLevelDb, jobId, jobRunContext, finishAttemptNumber);
    } else {
      const nextFinishAttemptNumber = finishAttemptNumber + 1;
      await this.triggerBeforeJobFinish(beforeJobFinish, jobLevelDb, jobId, nextFinishAttemptNumber);

      const currentJobStatusAfterFinishCallback = await this.getLastJobStatus(jobId);
      if (currentJobStatusAfterFinishCallback === 'CANCELLED') {
        this.cleanupJobMutexIfTerminalStatus(jobId, 'CANCELLED');
        await this.startQueuedJobsSafely(jobHandlerName);
        return;
      }

      // the last scan and the terminal status write hold the job tasks mutex, so addNewTasksToExistingJob can not
      // store a task between them which nobody would process
      const { unfinishedTasksAfterFinishCallback, finalStatus } = await this.getJobTasksMutex(jobId).runExclusive(async () => {
        const unfinishedTasksAfterFinishCallback = await this.getUnfinishedTasksFromLevelDb(jobLevelDb);
        if (unfinishedTasksAfterFinishCallback.length > 0) {
          return { unfinishedTasksAfterFinishCallback, finalStatus: null };
        }
        const allTasksDoneStatus = await this.getAllTasksDoneStatus(jobLevelDb);
        const finalStatus: JobStatus = allTasksDoneStatus.failedTasks === 0 ? 'DONE' : 'DONE_WITH_ERRORS';
        await this.adminforth.resource(this.getResourceId()).update(jobId, {
          [this.options.statusField]: finalStatus,
          [this.options.finishedAtField]: (new Date()).toISOString(),
        })
        return { unfinishedTasksAfterFinishCallback, finalStatus };
      });
      if (unfinishedTasksAfterFinishCallback.length > 0) {
        const tasksToReprocess = await this.buildTasksToReprocess(jobLevelDb, unfinishedTasksAfterFinishCallback);
        await this.runProcessingTasks(tasksToReprocess, jobLevelDb, jobId, jobRunContext, nextFinishAttemptNumber);
        return;
      }

      if (finalStatus === 'DONE') {
        this.publishJobUpdate({ jobId, status: 'DONE', finishedAt: (new Date()).toISOString() });
        this.cleanupJobMutexIfTerminalStatus(jobId, 'DONE');
        await this.triggerOnAllTasksDone(onAllTasksDone, jobLevelDb, jobId);
      } else {
        const jobError = await this.getJobStateField(jobId, 'error');
        this.publishJobUpdate({ jobId, status: 'DONE_WITH_ERRORS', finishedAt: (new Date()).toISOString(), error: jobError });
        this.cleanupJobMutexIfTerminalStatus(jobId, 'DONE_WITH_ERRORS');
        await this.triggerOnAllTasksDone(onAllTasksDone, jobLevelDb, jobId);
      }
      await this.startQueuedJobsSafely(jobHandlerName);
    }
  }

  private async getLastJobStatus(jobId: string): Promise<string> {
    const currentJobRecord = await this.adminforth.resource(this.getResourceId()).get(Filters.EQ(this.getResourcePk(), jobId));
    const currentJobStatus = currentJobRecord[this.options.statusField];
    return currentJobStatus;
  }

  private async handleFinishTask(completedTasks: number, totalTasks: number, jobId: string, wasTaskSkipped: boolean) {
    completedTasks++;
    if (wasTaskSkipped) {
      return completedTasks;
    }
    const progress = Math.round((completedTasks / totalTasks) * 100);
    await this.adminforth.resource(this.getResourceId()).update(jobId, {
      [this.options.progressField]: progress,
    })
    this.publishJobUpdate({ jobId, progress });
    return completedTasks;
  }

  private async handleJobProcessingFailure(
    jobId: string,
    jobHandlerName: string,
    error: unknown,
  ) {
    const errorMessage = error instanceof Error ? error.message : String(error);
    afLogger.error(`Error while processing tasks of job ${jobId}: ${errorMessage}`);

    try {
      const currentStatus = await this.getLastJobStatus(jobId);
      if (TERMINAL_JOB_STATUSES.includes(currentStatus as JobStatus)) {
        return;
      }

      await this.setJobStateField(jobId, 'error', errorMessage);
      const finishedAt = new Date().toISOString();
      await this.adminforth.resource(this.getResourceId()).update(jobId, {
        [this.options.statusField]: 'DONE_WITH_ERRORS',
        [this.options.finishedAtField]: finishedAt,
      });
      this.publishJobUpdate({ jobId, status: 'DONE_WITH_ERRORS', finishedAt, error: errorMessage });
      this.cleanupJobMutexIfTerminalStatus(jobId, 'DONE_WITH_ERRORS');
      await this.startQueuedJobsSafely(jobHandlerName);
    } catch (failureHandlingError) {
      const failureMessage = failureHandlingError instanceof Error ? failureHandlingError.message : String(failureHandlingError);
      afLogger.error(`Failed to mark job ${jobId} as DONE_WITH_ERRORS: ${failureMessage}`);
    }
  }


  private async runProcessingJobTasksFromStorage(
    job: Record<string, any>,
    knownJobRunContext?: jobRunContext,
  ) {
    const jobId = job[this.getResourcePk()];
    const jobLevelDb = await this.getLevelDbForTheJob(jobId);
    const jobHandlerName = job[this.options.jobHandlerField];
    const jobRunContext = knownJobRunContext || this.buildJobRunContext(jobHandlerName);
    if (!jobRunContext) {
      afLogger.error(`No handler registered for jobHandler ${jobHandlerName}. Cannot process unfinished tasks for job ${jobId}.`);
      return;
    }
    const unfinishedTasks: taskType[] = [];
    let taskIndex = 0;
    while (true) {
      const taskData = await jobLevelDb.get(taskIndex.toString());
      if (!taskData) {   
        break;
      }
      let parsedTaskData: { state: Record<string, any>, status: TaskStatus };
      try {
        parsedTaskData = JSON.parse(taskData);
      } catch (error) {
        afLogger.error(`Error parsing task data for task ${taskIndex} of job ${jobId}: ${error}`);
        taskIndex++;
        continue;
      }
      if (parsedTaskData.status === 'IN_PROGRESS' || parsedTaskData.status === 'SCHEDULED') {
        unfinishedTasks.push({ state: parsedTaskData.state });
      } else {
        unfinishedTasks.push({ state: parsedTaskData.state, skip: true });
      }
      taskIndex++;
    }
    await this.runProcessingTasks(
      unfinishedTasks,
      jobLevelDb,
      jobId,
      jobRunContext,
    );

  }

  public async setJobStateField(jobId: string, key: string, value: any) {
    const jobRecord = await this.adminforth.resource(this.getResourceId()).get(Filters.EQ(this.getResourcePk(), jobId));
    const state = jobRecord[this.options.stateField];
    state[key] = value;
    await this.adminforth.resource(this.getResourceId()).update(jobId, {
      [this.options.stateField]: state,
    });
    this.publishJobStateField(jobId, key, value);
  }

  public async getJobStateField(jobId: string, key: string) {
    const jobRecord = await this.adminforth.resource(this.getResourceId()).get(Filters.EQ(this.getResourcePk(), jobId));
    const state = jobRecord[this.options.stateField];
    return state[key];
  }

  public async getJobState(jobId: string) {
    const jobRecord = await this.adminforth.resource(this.getResourceId()).get(Filters.EQ(this.getResourcePk(), jobId));
    return jobRecord[this.options.stateField];
  }

  public async setJobField(jobId: string, key: string, value: any) {
    this.warnDeprecatedOnce(
      'setJobField',
      'BackgroundJobsPlugin: setJobField(jobId, key, value) is deprecated and will be removed soon. Use setJobStateField(jobId, fieldName: string, value: any) instead.',
    );
    return this.setJobStateField(jobId, key, value);
  }

  public async getJobField(jobId: string, key: string) {
    this.warnDeprecatedOnce(
      'getJobField',
      'BackgroundJobsPlugin: getJobField(jobId, key) is deprecated and will be removed soon. Use getJobStateField(jobId, fieldName: string) instead.',
    );
    return this.getJobStateField(jobId, key);
  }

  public async updateJobFieldsAtomically(jobId: string, updateFunction: () => Promise<void>) {
    if (!jobId) {
      throw new Error('updateJobFieldsAtomically: jobId is required');
    }
    if (typeof updateFunction !== 'function') {
      throw new Error('updateJobFieldsAtomically: updateFunction must be a function');
    }

    // Ensure updates are atomic per jobId.
    // Different jobs are not blocked by each other.
    let mutex = this.jobStateMutexes[jobId];
    if (!mutex) {
      mutex = new Mutex();
      this.jobStateMutexes[jobId] = mutex;
    }

    try {
      return await mutex.runExclusive(async () => {
        await updateFunction();
      });
    } finally {
      // drop the mutex only when nobody holds or waits for it, otherwise the next caller would create a second
      // mutex for the same job and run its update concurrently with the current owner
      if (!mutex.isLocked() && this.jobStateMutexes[jobId] === mutex) {
        delete this.jobStateMutexes[jobId];
      }
    }
  }

  private async processAllUnfinishedJobs() {
    const resourceId = this.getResourceId();
    const unprocessedJobs = await this.adminforth.resource(resourceId).list(Filters.EQ(this.options.statusField, 'IN_PROGRESS'));
    for (const job of unprocessedJobs) {
      const jobName = job[this.options.nameField];
      afLogger.info(`Processing unfinished job with name ${jobName} on startup.`);
      this.runProcessingJobTasksFromStorage(job).catch((error) => {
        void this.handleJobProcessingFailure(
          job[this.getResourcePk()],
          job[this.options.jobHandlerField],
          error,
        );
      });
    }

    // resumed jobs above are already IN_PROGRESS in the database, so they are counted as occupying a
    // concurrency slot and queued jobs are started only for queues which still have a free one
    const queuedJobs = await this.adminforth.resource(resourceId).list(Filters.EQ(this.options.statusField, 'QUEUED'));
    const queuedJobHandlerNames = new Set<string>(queuedJobs.map((job) => job[this.options.jobHandlerField]));
    for (const jobHandlerName of queuedJobHandlerNames) {
      afLogger.info(`Starting queued jobs for jobHandler ${jobHandlerName} on startup.`);
      // without a job name every queue of the handler is drained
      await this.startQueuedJobsSafely(jobHandlerName);
    }
  }

  
  async validateConfigAfterDiscover(adminforth: IAdminForth, resourceConfig: AdminForthResource) {
    // optional method where you can safely check field types after database discovery was performed
    this.checkIfFieldInResource(resourceConfig, this.options.createdAtField, 'createdAtField');
    this.checkIfFieldInResource(resourceConfig, this.options.finishedAtField, 'finishedAtField');
    this.checkIfFieldInResource(resourceConfig, this.options.startedByField, 'startedByField');
    this.checkIfFieldInResource(resourceConfig, this.options.stateField, 'stateField');
    this.checkIfFieldInResource(resourceConfig, this.options.progressField, 'progressField');
    this.checkIfFieldInResource(resourceConfig, this.options.statusField, 'statusField');
    this.checkIfFieldInResource(resourceConfig, this.options.nameField, 'nameField');
    this.checkIfFieldInResource(resourceConfig, this.options.jobHandlerField, 'jobHandlerField');


    //Add temp delay to make sure, that all resources active. Probably should be fixed
    await new Promise(resolve => setTimeout(resolve, 1000));
    this.processAllUnfinishedJobs();
  }

  instanceUniqueRepresentation(pluginOptions: any) : string {
    return `BackgroundJobsPlugin`;
  }

  setupEndpoints(server: IHttpServer) {
    server.endpoint({
      method: 'POST',
      path: `/plugin/${this.pluginInstanceId}/get-list-of-jobs`,
      handler: async () => {
        const resourcePk = this.getResourcePk();
        const listOfJobs = await this.adminforth.resource(this.resourceConfig.resourceId).list(
          Filters.IS_NOT_EMPTY(resourcePk),
          100,
          0,
          Sorts.DESC(this.options.createdAtField),
        );
        
        const jobsToReturn = listOfJobs.map(job => {
          return {
            id: job[resourcePk],
            name: job[this.options.nameField],
            createdAt: job[this.options.createdAtField],
            finishedAt: job[this.options.finishedAtField] || null,
            status: job[this.options.statusField],
            progress: job[this.options.progressField],
            customComponent: this.jobCustomComponents[job[this.options.jobHandlerField]],
          }
        });
        return { jobs: jobsToReturn };
      }
    });

    server.endpoint({
      method: 'POST',
      path: `/plugin/get-background-job-info`,
      request_schema: jobIdBodySchema,
      handler: async ({ body }) => {
        const data = body as z.infer<typeof jobIdBodySchema>;
        const jobId = data.jobId;

        const job = await this.getJobById(jobId);
        if (!job) {
          return { ok: false, message: `Job with id ${jobId} not found.` };
        }
        const jobToReturn = {
          id: job[this.getResourcePk()],
          name: job[this.options.nameField],
          createdAt: job[this.options.createdAtField],
          finishedAt: job[this.options.finishedAtField] || null,
          status: job[this.options.statusField],
          state: job[this.options.stateField],
          progress: job[this.options.progressField],
          customComponent: this.jobCustomComponents[job[this.options.jobHandlerField]],
        };
        return { ok: true, job: jobToReturn };
      }
    });


    server.endpoint({
      method: 'POST',
      path: `/plugin/${this.pluginInstanceId}/cancel-job`,
      agent: { requiresHumanApproval: true },
      request_schema: jobIdBodySchema,
      handler: async ({ body }) => {
        const data = body as z.infer<typeof jobIdBodySchema>;
        const jobId = data.jobId;
        const currentJob = await this.getJobById(jobId);
        if (!currentJob) {
          return { ok: false, message: `Job with id ${jobId} not found.` };
        }
        const jobHandlerName = currentJob[this.options.jobHandlerField];
        try {
          // startNextQueuedJob reads a QUEUED job and promotes it under the same queue mutex, so the status is
          // checked and CANCELLED is written under it too, otherwise a promotion could overwrite CANCELLED with IN_PROGRESS
          const cancelResult = await this.getQueueMutex(jobHandlerName).runExclusive(async () => {
            const oldStatus = (await this.getJobById(jobId))[this.options.statusField];
            if (TERMINAL_JOB_STATUSES.includes(oldStatus)) {
              return { ok: false, message: `Cannot cancel a job with status ${oldStatus}.` };
            }
            await this.adminforth.resource(this.getResourceId()).update(jobId, {
              [this.options.statusField]: 'CANCELLED',
              [this.options.finishedAtField]: (new Date()).toISOString(),
            });
            this.publishJobUpdate({
              jobId,
              status: 'CANCELLED',
            });
            return { ok: true };
          });
          if (!cancelResult.ok) {
            return cancelResult;
          }
          // a cancelled job frees its concurrency slot, so the next queued job of the same handler can start;
          // runs after the queue mutex is released because startNextQueuedJob takes it again and it is not reentrant
          await this.startQueuedJobsSafely(jobHandlerName);
          return { ok: true };
        } catch (error) {
          return { ok: false, message: `Failed to cancel job with id ${jobId}.` };
        }
      }
    });

    server.endpoint({
      method: 'POST',
      path: `/plugin/${this.pluginInstanceId}/get-tasks`,
      request_schema: getTasksBodySchema,
      handler: async ({ body }) => {
        const data = body as z.infer<typeof getTasksBodySchema>;
        const { jobId, limit, offset, fieldsToReturn } = data;
        const job = await this.getJobById(jobId);
        if (!job) {
          return { ok: false, message: 'Job not found' };
        }

        const jobLevelDb: Level = await this.getLevelDbForTheJob(jobId as string);
        if (!jobLevelDb) {
          return { ok: false, message: `Job with id ${jobId} not found.` };
        }
        // `_meta:count` is written for every job when it is created, even for a job without tasks, so a
        // missing key means the LevelDB of the job was deleted (container storage wiped, old job dbs
        // cleaned up) and task details can not be restored anymore
        const tasksCountMeta = await jobLevelDb.get('_meta:count');
        if (!tasksCountMeta) {
          afLogger.warn(`Task storage for job with id ${jobId} is missing, task details can not be returned.`);
          return {
            ok: true,
            data: {
              tasks: [],
              total: 0,
              storageLost: true,
              message: `Task storage of job with id ${jobId} was deleted, task details can not be restored.`,
            },
          };
        }
        const tasks = [];
        let taskIndex = 0 + offset;
        while (true) {
          if (limit && tasks.length >= limit) {
            break;
          }
          const taskData = await jobLevelDb.get(taskIndex.toString());
          if (!taskData) {   
            break;
          }
          let parsedTaskData: { state: Record<string, any>, status: TaskStatus };
          try {
            parsedTaskData = JSON.parse(taskData);
          } catch (error) {
            afLogger.error(`Error parsing task data for task ${taskIndex} of job ${jobId}: ${error}`);
            taskIndex++;
            continue;
          }
          if (fieldsToReturn && fieldsToReturn.length > 0) {
            const filteredState: Record<string, any> = {};
            for (const field of fieldsToReturn) {
              filteredState[field] = parsedTaskData.state[field];
            }
            parsedTaskData.state = filteredState;
          }
          tasks.push(parsedTaskData);
          taskIndex++;
        }
          
        const total = await this.getTotalTasksInLevelDb(jobLevelDb);
        return { ok: true, data: { tasks, total } };
      }
    });
  }

}