task scheduler rework

task manager implementation to start/stop tasks WIP
implemented socket client for internal communication between threads - needs more testing as connection string is hardcoded
added loopback in ws server to pass through toast ws messages from threads - needs more testing

fixed potential bug during config fetch function when syncing

backend/classes/emby-api.js

@@ -11,15 +11,22 @@ class EmbyAPI {
   //Helper classes
   #checkReadyStatus() {
     let checkConfigError = setInterval(async () => {
-      const _config = await new configClass().getConfig();
-      if (!_config.error && _config.state === 2) {
+      const success = await this.#fetchConfig();
+      if (success) {
         clearInterval(checkConfigError);
-        this.config = _config;
-        this.configReady = true;
       }
     }, 5000); // Check every 5 seconds
   }
 
+  async #fetchConfig() {
+    const _config = await new configClass().getConfig();
+    if (!_config.error && _config.state === 2) {
+      this.config = _config;
+      this.configReady = true;
+      return true;
+    }
+    return false;
+  }
   #errorHandler(error, url) {
     if (error.response) {
       console.log("[EMBY-API]: " + this.#httpErrorMessageHandler(error));
@@ -292,7 +299,10 @@ class EmbyAPI {
 
   async getLibraries() {
     if (!this.configReady) {
-      return [];
+      const success = await this.#fetchConfig();
+      if (!success) {
+        return [];
+      }
     }
     try {
       let url = `${this.config.JF_HOST}/Library/MediaFolders`;

backend/classes/jellyfin-api.js

@@ -11,15 +11,23 @@ class JellyfinAPI {
   //Helper classes
   #checkReadyStatus() {
     let checkConfigError = setInterval(async () => {
-      const _config = await new configClass().getConfig();
-      if (!_config.error && _config.state === 2) {
+      const success = await this.#fetchConfig();
+      if (success) {
         clearInterval(checkConfigError);
-        this.config = _config;
-        this.configReady = true;
       }
     }, 5000); // Check every 5 seconds
   }
 
+  async #fetchConfig() {
+    const _config = await new configClass().getConfig();
+    if (!_config.error && _config.state === 2) {
+      this.config = _config;
+      this.configReady = true;
+      return true;
+    }
+    return false;
+  }
+
   #errorHandler(error, url) {
     if (error.response) {
       console.log("[JELLYFIN-API]: " + this.#httpErrorMessageHandler(error));
@@ -289,7 +297,10 @@ class JellyfinAPI {
 
   async getLibraries() {
     if (!this.configReady) {
-      return [];
+      const success = await this.#fetchConfig();
+      if (!success) {
+        return [];
+      }
     }
     try {
       let url = `${this.config.JF_HOST}/Library/MediaFolders`;

backend/classes/task-manager-singleton.js

@@ -0,0 +1,16 @@
+const TaskManager = require("./task-manager");
+
+class TaskManagerSingleton {
+  constructor() {
+    if (!TaskManagerSingleton.instance) {
+      TaskManagerSingleton.instance = new TaskManager();
+      console.log("Task Manager Singleton created");
+    }
+  }
+
+  getInstance() {
+    return TaskManagerSingleton.instance;
+  }
+}
+
+module.exports = TaskManagerSingleton;

backend/classes/task-manager.js

@@ -0,0 +1,86 @@
+const { Worker } = require("worker_threads");
+const TaskList = require("../global/task-list");
+const { sendUpdate } = require("../ws");
+
+class TaskManager {
+  constructor() {
+    this.tasks = {};
+    this.taskList = TaskList;
+    this.emitTaskList();
+  }
+
+  addTask({ task, onComplete, onError, onExit }) {
+    if (this.tasks[task.name]) {
+      console.log(`Task ${task.name} already exists.`);
+      return false;
+    }
+
+    const worker = new Worker(task.path);
+
+    worker.on("message", (message) => {
+      if (message.status === "complete" && onComplete) {
+        onComplete();
+      }
+      if (message.status === "error" && onError) {
+        onError(new Error(message.message));
+      }
+      delete this.tasks[task.name];
+    });
+
+    worker.on("error", (error) => {
+      if (onError) {
+        onError(error);
+      }
+      console.error(`Error from ${task.name}:`, error);
+      delete this.tasks[task.name];
+    });
+
+    worker.on("exit", (code) => {
+      if (code !== 0) {
+        console.error(`Worker ${task.name} stopped with exit code ${code}`);
+      }
+      if (onExit) {
+        onExit();
+      }
+      delete this.tasks[task.name];
+    });
+
+    this.tasks[task.name] = { worker };
+    return true;
+  }
+
+  startTask(task, triggerType) {
+    const taskExists = this.tasks[task.name];
+    if (!taskExists) {
+      console.log(`Task ${task.name} does not exist.`);
+      return;
+    }
+    taskExists.worker.postMessage({ command: "start", triggertype: triggerType });
+  }
+
+  stopTask(task) {
+    const taskExists = this.tasks[task.name];
+    if (!taskExists) {
+      console.log(`Task ${task.name} does not exist.`);
+      return;
+    }
+
+    taskExists.worker.terminate();
+    delete this.tasks[task.name];
+  }
+
+  isTaskRunning(taskName) {
+    return !!this.tasks[taskName];
+  }
+
+  emitTaskList() {
+    let emitTasks = setInterval(async () => {
+      const taskList = Object.keys(this.taskList).map((key) => {
+        return { task: key, name: this.taskList[key].name, running: this.isTaskRunning(this.taskList[key].name) };
+      });
+      sendUpdate("task-list", taskList);
+    }, 1000);
+  }
+}
+
+module.exports = TaskManager;

backend/classes/task-scheduler-singleton.js

@@ -0,0 +1,16 @@
+const TaskScheduler = require("./task-scheduler.js");
+
+class TaskSchedulerSingleton {
+  constructor() {
+    if (!TaskSchedulerSingleton.instance) {
+      TaskSchedulerSingleton.instance = new TaskScheduler();
+      console.log("Task Scheduler Singleton created");
+    }
+  }
+
+  getInstance() {
+    return TaskSchedulerSingleton.instance;
+  }
+}
+
+module.exports = TaskSchedulerSingleton;

backend/classes/task-scheduler.js

@@ -0,0 +1,236 @@
+const TaskManager = require("./task-manager-singleton");
+const db = require("../db");
+const TaskList = require("../global/task-list");
+const { sendUpdate } = require("../ws");
+const triggertype = require("../logging/triggertype");
+const taskstate = require("../logging/taskstate");
+
+class TaskScheduler {
+  constructor() {
+    this.taskManager = new TaskManager().getInstance();
+    this.scheduledTasks = {};
+    this.taskHistory = [];
+
+    // Predefined tasks and default intervals (in minutes)
+    this.defaultIntervals = {
+      PartialJellyfinSync: {
+        Interval: 60,
+        ...TaskList.PartialJellyfinSync,
+      },
+      JellyfinSync: {
+        Interval: 1440,
+        ...TaskList.JellyfinSync,
+      },
+      Backup: {
+        Interval: 1440,
+        ...TaskList.Backup,
+      },
+      // Add more tasks as needed
+    };
+
+    // Initialize tasks with default intervals
+    this.initializeTasks();
+  }
+
+  delay(ms) {
+    return new Promise((resolve) => setTimeout(resolve, ms));
+  }
+
+  async initializeTasks() {
+    await this.updateIntervalsFromDB();
+    await this.getTaskHistory();
+    await this.clearRunningTasks();
+    this.mainSchedulerUpdateLoop();
+  }
+
+  async clearRunningTasks() {
+    try {
+      await db.query(`UPDATE jf_logging SET "Result"='${taskstate.FAILED}' WHERE "Result"='${taskstate.RUNNING}'`);
+    } catch (error) {
+      console.log("Clear Running Tasks Error: " + error);
+    }
+  }
+
+  async getTaskHistory() {
+    try {
+      const historyjson = await db
+        .query(
+          `
+        with latest_tasks as
+          (SELECT DISTINCT ON ("Name")
+            "Id",
+            "Name",
+            "Type",
+            "ExecutionType",
+            "Duration",
+            "TimeRun",
+            "Log",
+            "Result"
+          FROM public.jf_logging
+          ORDER BY  "Name", "TimeRun" DESC
+          )
+
+        select * from latest_tasks
+        ORDER BY  "TimeRun" DESC;`
+        )
+        .then((res) =>
+          res.rows.map((row) => {
+            return {
+              Name: row.Name,
+              Type: row.Type,
+              ExecutionType: row.ExecutionType,
+              Duration: row.Duration,
+              TimeRun: row.TimeRun,
+              Result: row.Result,
+            };
+          })
+        );
+
+      this.taskHistory = historyjson;
+      this.getTimeTillNextRun();
+    } catch (error) {
+      console.log("Get Task History Error: " + error);
+    }
+  }
+
+  async updateIntervalsFromDB() {
+    try {
+      const settingsjson = await db.query('SELECT settings FROM app_config where "ID"=1').then((res) => res.rows);
+
+      if (settingsjson.length > 0) {
+        const settings = settingsjson[0].settings || {};
+
+        for (const taskEnumKey in this.defaultIntervals) {
+          const taskSettings = settings.Tasks?.[taskEnumKey] || {};
+          if (taskSettings.Interval) {
+            this.defaultIntervals[taskEnumKey].Interval = taskSettings.Interval;
+          } else {
+            taskSettings.Interval = this.defaultIntervals[taskEnumKey];
+          }
+
+          if (!settings.Tasks) {
+            settings.Tasks = {};
+          }
+          settings.Tasks[taskEnumKey] = taskSettings;
+        }
+
+        let query = 'UPDATE app_config SET settings=$1 where "ID"=1';
+        await db.query(query, [settings]);
+      }
+    } catch (error) {
+      console.log("Sync Task Settings Error: " + error);
+    }
+  }
+
+  getTimeTillNextRun() {
+    try {
+      for (const taskEnumKey in this.defaultIntervals) {
+        const task = this.defaultIntervals[taskEnumKey];
+        const interval = task.Interval;
+        const lastRun = this.taskHistory.find((history) => history.Name === task.name);
+        const currentTime = new Date().getTime();
+        if (!lastRun) {
+          const nextRunTime = currentTime + interval * 60000;
+          this.defaultIntervals[taskEnumKey].NextRunTime = taskEnumKey == "JellyfinSync" ? 0 : nextRunTime;
+        } else {
+          const lastRunTime = new Date(lastRun.TimeRun).getTime();
+          const nextRunTime = lastRunTime + interval * 60000;
+          this.defaultIntervals[taskEnumKey].NextRunTime = nextRunTime;
+        }
+      }
+    } catch (error) {
+      console.log(error);
+    }
+  }
+
+  mainSchedulerUpdateLoop() {
+    setInterval(() => {
+      const currentTime = new Date().getTime();
+
+      for (const taskEnumKey in this.defaultIntervals) {
+        const task = this.defaultIntervals[taskEnumKey];
+        const nextRunTime = task.NextRunTime;
+        if (currentTime >= nextRunTime && this.taskManager.isTaskRunning(task.name) === false) {
+          console.log(`Running task ${task.name}...`);
+          this.beginTask(taskEnumKey);
+        }
+      }
+    }, 10000);
+  }
+  beginTask(taskEnumKey) {
+    switch (taskEnumKey) {
+      case "PartialJellyfinSync":
+        this.addPartialSyncTask();
+        break;
+      case "JellyfinSync":
+        this.addFullSyncTask();
+        break;
+      case "Backup":
+        this.addBackupTask();
+        break;
+      default:
+        console.log(`Unknown task: ${taskEnumKey}`);
+    }
+  }
+
+  // Add tasks here
+
+  addPartialSyncTask() {
+    const success = this.taskManager.addTask({
+      task: this.taskManager.taskList.PartialJellyfinSync,
+      onComplete: async () => {
+        await this.getTaskHistory();
+      },
+      onError: async (error) => {
+        await this.getTaskHistory();
+        console.error(error);
+      },
+    });
+    if (success) {
+      this.taskManager.startTask(this.taskManager.taskList.PartialJellyfinSync, triggertype.Automatic);
+      return;
+    }
+  }
+
+  addFullSyncTask() {
+    const success = this.taskManager.addTask({
+      task: this.taskManager.taskList.JellyfinSync,
+      onComplete: async () => {
+        await this.getTaskHistory();
+      },
+      onError: async (error) => {
+        await this.getTaskHistory();
+        console.error(error);
+      },
+    });
+    if (success) {
+      this.taskManager.startTask(this.taskManager.taskList.JellyfinSync, triggertype.Automatic);
+      return;
+    }
+  }
+
+  addBackupTask() {
+    const success = this.taskManager.addTask({
+      task: this.taskManager.taskList.Backup,
+      onComplete: async () => {
+        await this.getTaskHistory();
+        sendUpdate("BackupTask", { type: "Success", message: triggertype.Automatic + " Backup Completed" });
+      },
+      onError: async (error) => {
+        console.error(error);
+        await this.getTaskHistory();
+        sendUpdate("BackupTask", { type: "Error", message: "Error: Backup failed" });
+      },
+      onExit: async () => {
+        await this.getTaskHistory();
+        sendUpdate("BackupTask", { type: "Error", message: "Backup Task Stopped" });
+      },
+    });
+    if (success) {
+      this.taskManager.startTask(this.taskManager.taskList.Backup, triggertype.Automatic);
+      return;
+    }
+  }
+}
+
+module.exports = TaskScheduler;

backend/global/task-list.js

@@ -0,0 +1,12 @@
+const TaskName = require("../logging/taskName");
+
+const Tasks = {
+  Backup: { path: "./tasks/BackupTask.js", name: TaskName.backup },
+  Restore: { path: "./tasks/BackupTask.js", name: TaskName.restore },
+  JellyfinSync: { path: "./tasks/FullSyncTask.js", name: TaskName.fullsync },
+  PartialJellyfinSync: { path: "./tasks/RecentlyAddedItemsSyncTask.js", name: TaskName.partialsync },
+  JellyfinPlaybackReportingPluginSync: { path: "./tasks/PlaybackReportingPluginSyncTask.js", name: TaskName.import },
+  // Add more tasks as needed
+};
+
+module.exports = Tasks;

backend/routes/api.js

@@ -7,13 +7,14 @@ const dbHelper = require("../classes/db-helper");
 const pgp = require("pg-promise")();
 const { randomUUID } = require("crypto");
 
-const { axios } = require("../classes/axios");
 const configClass = require("../classes/config");
 const { checkForUpdates } = require("../version-control");
 const API = require("../classes/api-loader");
 const { sendUpdate } = require("../ws");
 const moment = require("moment");
 const { tables } = require("../global/backup_tables");
+const TaskScheduler = require("../classes/task-scheduler-singleton");
+const TaskManager = require("../classes/task-manager-singleton.js");
 
 const router = express.Router();
 
@@ -873,6 +874,9 @@ router.post("/setTaskSettings", async (req, res) => {
       let query = 'UPDATE app_config SET settings=$1 where "ID"=1';
 
       await db.query(query, [settings]);
+      const taskScheduler = new TaskScheduler().getInstance();
+      await taskScheduler.updateIntervalsFromDB();
+      await taskScheduler.getTaskHistory();
       res.status(200);
       res.send(tasksettings);
     } else {
@@ -1863,6 +1867,36 @@ router.post("/getActivityTimeLine", async (req, res) => {
   }
 });
 
+//Tasks
+
+router.get("/stopTask", async (req, res) => {
+  const { task } = req.query;
+
+  if (task === undefined) {
+    res.status(400);
+    res.send("No Task provided");
+    return;
+  }
+  const taskManager = new TaskManager().getInstance();
+  if (taskManager.taskList[task] === undefined) {
+    res.status(404);
+    res.send("Task not found");
+    return;
+  }
+
+  const _task = taskManager.taskList[task];
+
+  if (taskManager.isTaskRunning(_task.name)) {
+    taskManager.stopTask(_task);
+    res.send("Task Stopped");
+    return;
+  } else {
+    res.status(400);
+    res.send("Task is not running");
+    return;
+  }
+});
+
 // Handle other routes
 router.use((req, res) => {
   res.status(404).send({ error: "Not Found" });

backend/routes/backup.js

@@ -6,16 +6,16 @@ const { randomUUID } = require("crypto");
 const multer = require("multer");
 
 const Logging = require("../classes/logging");
-const backup = require("../classes/backup");
 const triggertype = require("../logging/triggertype");
 const taskstate = require("../logging/taskstate");
 const taskName = require("../logging/taskName");
 const sanitizeFilename = require("../utils/sanitizer");
 
 const { sendUpdate } = require("../ws");
-const db = require("../db");
 
 const router = express.Router();
+const TaskManager = require("../classes/task-manager-singleton");
+const TaskScheduler = require("../classes/task-scheduler-singleton");
 
 // Database connection parameters
 const postgresUser = process.env.POSTGRES_USER;
@@ -114,31 +114,28 @@ async function restore(file, refLog) {
 // Route handler for backup endpoint
 router.get("/beginBackup", async (req, res) => {
   try {
-    const last_execution = await db
-      .query(
-        `SELECT "Result"
-    FROM public.jf_logging
-    WHERE "Name"='${taskName.backup}'
-    ORDER BY "TimeRun" DESC
-    LIMIT 1`
-      )
-      .then((res) => res.rows);
-
-    if (last_execution.length !== 0) {
-      if (last_execution[0].Result === taskstate.RUNNING) {
-        sendUpdate("TaskError", "Error: Backup is already running");
-        res.send();
-        return;
-      }
+    const taskManager = new TaskManager().getInstance();
+    const taskScheduler = new TaskScheduler().getInstance();
+    const success = taskManager.addTask({
+      task: taskManager.taskList.Backup,
+      onComplete: async () => {
+        console.log("Backup completed successfully");
+        await taskScheduler.getTaskHistory();
+        res.send("Backup completed successfully");
+      },
+      onError: (error) => {
+        console.error(error);
+        res.status(500).send("Backup failed");
+        sendUpdate("BackupTask", { type: "Error", message: "Error: Backup failed" });
+      },
+    });
+    if (!success) {
+      res.status(500).send("Backup already running");
+      sendUpdate("BackupTask", { type: "Error", message: "Backup is already running" });
+      return;
     }
 
-    const uuid = randomUUID();
-    let refLog = { logData: [], uuid: uuid };
-    await Logging.insertLog(uuid, triggertype.Manual, taskName.backup);
-    await backup(refLog);
-    Logging.updateLog(uuid, refLog.logData, taskstate.SUCCESS);
-    res.send("Backup completed successfully");
-    sendUpdate("TaskComplete", { message: triggertype + " Backup Completed" });
+    taskManager.startTask(taskManager.taskList.Backup, triggertype.Manual);
   } catch (error) {
     console.error(error);
     res.status(500).send("Backup failed");

backend/routes/sync.js

@@ -1,5 +1,4 @@
 const express = require("express");
-const pgp = require("pg-promise")();
 const db = require("../db");
 
 const moment = require("moment");
@@ -13,6 +12,8 @@ const triggertype = require("../logging/triggertype");
 
 const configClass = require("../classes/config");
 const API = require("../classes/api-loader");
+const TaskManager = require("../classes/task-manager-singleton");
+const TaskScheduler = require("../classes/task-scheduler-singleton");
 
 const router = express.Router();
 
@@ -441,97 +442,113 @@ async function migrateArchivedActivty() {
 }
 
 async function syncPlaybackPluginData() {
-  PlaybacksyncTask.loggedData.push({ color: "lawngreen", Message: "Syncing..." });
+  try {
+    const uuid = randomUUID();
+    PlaybacksyncTask = { loggedData: [], uuid: uuid };
 
-  //Playback Reporting Plugin Check
-  const installed_plugins = await API.getInstalledPlugins();
+    await logging.insertLog(uuid, triggertype.Manual, taskName.import);
+    sendUpdate("PlaybackSyncTask", { type: "Start", message: "Playback Plugin Sync Started" });
 
-  const hasPlaybackReportingPlugin = installed_plugins.filter(
-    (plugins) => ["playback_reporting.xml", "Jellyfin.Plugin.PlaybackReporting.xml"].includes(plugins?.ConfigurationFileName) //TO-DO Change this to the correct plugin name
-  );
+    PlaybacksyncTask.loggedData.push({ color: "lawngreen", Message: "Syncing..." });
+
+    //Playback Reporting Plugin Check
+    const installed_plugins = await API.getInstalledPlugins();
+
+    const hasPlaybackReportingPlugin = installed_plugins.filter(
+      (plugins) => ["playback_reporting.xml", "Jellyfin.Plugin.PlaybackReporting.xml"].includes(plugins?.ConfigurationFileName) //TO-DO Change this to the correct plugin name
+    );
 
-  if (!hasPlaybackReportingPlugin || hasPlaybackReportingPlugin.length === 0) {
     if (!hasPlaybackReportingPlugin || hasPlaybackReportingPlugin.length === 0) {
-      PlaybacksyncTask.loggedData.push({ color: "dodgerblue", Message: `No new data to insert.` });
-    } else {
-      PlaybacksyncTask.loggedData.push({ color: "lawngreen", Message: "Playback Reporting Plugin not detected. Skipping step." });
-    }
-  } else {
-    //
-
-    PlaybacksyncTask.loggedData.push({ color: "dodgerblue", Message: "Determining query constraints." });
-    const OldestPlaybackActivity = await db
-      .query('SELECT  MIN("ActivityDateInserted") "OldestPlaybackActivity" FROM public.jf_playback_activity')
-      .then((res) => res.rows[0]?.OldestPlaybackActivity);
-
-    const NewestPlaybackActivity = await db
-      .query('SELECT  MAX("ActivityDateInserted") "OldestPlaybackActivity" FROM public.jf_playback_activity')
-      .then((res) => res.rows[0]?.OldestPlaybackActivity);
-
-    const MaxPlaybackReportingPluginID = await db
-      .query('SELECT MAX(rowid) "MaxRowId" FROM jf_playback_reporting_plugin_data')
-      .then((res) => res.rows[0]?.MaxRowId);
-
-    //Query Builder
-    let query = `SELECT rowid, * FROM PlaybackActivity`;
-
-    if (OldestPlaybackActivity && NewestPlaybackActivity) {
-      const formattedDateTimeOld = moment(OldestPlaybackActivity).format("YYYY-MM-DD HH:mm:ss");
-      const formattedDateTimeNew = moment(NewestPlaybackActivity).format("YYYY-MM-DD HH:mm:ss");
-      query = query + ` WHERE (DateCreated < '${formattedDateTimeOld}' or DateCreated > '${formattedDateTimeNew}')`;
-    }
-
-    if (OldestPlaybackActivity && !NewestPlaybackActivity) {
-      const formattedDateTimeOld = moment(OldestPlaybackActivity).format("YYYY-MM-DD HH:mm:ss");
-      query = query + ` WHERE DateCreated < '${formattedDateTimeOld}'`;
-      if (MaxPlaybackReportingPluginID) {
-        query = query + ` AND rowid > ${MaxPlaybackReportingPluginID}`;
-      }
-    }
-
-    if (!OldestPlaybackActivity && NewestPlaybackActivity) {
-      const formattedDateTimeNew = moment(NewestPlaybackActivity).format("YYYY-MM-DD HH:mm:ss");
-      query = query + ` WHERE DateCreated > '${formattedDateTimeNew}'`;
-      if (MaxPlaybackReportingPluginID) {
-        query = query + ` AND rowid > ${MaxPlaybackReportingPluginID}`;
-      }
-    }
-
-    if (!OldestPlaybackActivity && !NewestPlaybackActivity && MaxPlaybackReportingPluginID) {
-      query = query + ` WHERE rowid > ${MaxPlaybackReportingPluginID}`;
-    }
-
-    query += " order by rowid";
-
-    PlaybacksyncTask.loggedData.push({ color: "dodgerblue", Message: "Query built. Executing." });
-    //
-
-    const PlaybackData = await API.StatsSubmitCustomQuery(query);
-
-    let DataToInsert = await PlaybackData.map(mappingPlaybackReporting);
-
-    if (DataToInsert.length > 0) {
-      PlaybacksyncTask.loggedData.push({ color: "dodgerblue", Message: `Inserting ${DataToInsert.length} Rows.` });
-      let result = await db.insertBulk("jf_playback_reporting_plugin_data", DataToInsert, columnsPlaybackReporting);
-
-      if (result.Result === "SUCCESS") {
-        PlaybacksyncTask.loggedData.push({ color: "dodgerblue", Message: `${DataToInsert.length} Rows have been inserted.` });
-        PlaybacksyncTask.loggedData.push({
-          color: "yellow",
-          Message: "Running process to format data to be inserted into the Activity Table",
-        });
+      if (!hasPlaybackReportingPlugin || hasPlaybackReportingPlugin.length === 0) {
+        PlaybacksyncTask.loggedData.push({ color: "dodgerblue", Message: `No new data to insert.` });
       } else {
-        PlaybacksyncTask.loggedData.push({ color: "red", Message: "Error: " + result.message });
-        await logging.updateLog(PlaybacksyncTask.uuid, PlaybacksyncTask.loggedData, taskstate.FAILED);
+        PlaybacksyncTask.loggedData.push({
+          color: "lawngreen",
+          Message: "Playback Reporting Plugin not detected. Skipping step.",
+        });
       }
+    } else {
+      //
+
+      PlaybacksyncTask.loggedData.push({ color: "dodgerblue", Message: "Determining query constraints." });
+      const OldestPlaybackActivity = await db
+        .query('SELECT  MIN("ActivityDateInserted") "OldestPlaybackActivity" FROM public.jf_playback_activity')
+        .then((res) => res.rows[0]?.OldestPlaybackActivity);
+
+      const NewestPlaybackActivity = await db
+        .query('SELECT  MAX("ActivityDateInserted") "OldestPlaybackActivity" FROM public.jf_playback_activity')
+        .then((res) => res.rows[0]?.OldestPlaybackActivity);
+
+      const MaxPlaybackReportingPluginID = await db
+        .query('SELECT MAX(rowid) "MaxRowId" FROM jf_playback_reporting_plugin_data')
+        .then((res) => res.rows[0]?.MaxRowId);
+
+      //Query Builder
+      let query = `SELECT rowid, * FROM PlaybackActivity`;
+
+      if (OldestPlaybackActivity && NewestPlaybackActivity) {
+        const formattedDateTimeOld = moment(OldestPlaybackActivity).format("YYYY-MM-DD HH:mm:ss");
+        const formattedDateTimeNew = moment(NewestPlaybackActivity).format("YYYY-MM-DD HH:mm:ss");
+        query = query + ` WHERE (DateCreated < '${formattedDateTimeOld}' or DateCreated > '${formattedDateTimeNew}')`;
+      }
+
+      if (OldestPlaybackActivity && !NewestPlaybackActivity) {
+        const formattedDateTimeOld = moment(OldestPlaybackActivity).format("YYYY-MM-DD HH:mm:ss");
+        query = query + ` WHERE DateCreated < '${formattedDateTimeOld}'`;
+        if (MaxPlaybackReportingPluginID) {
+          query = query + ` AND rowid > ${MaxPlaybackReportingPluginID}`;
+        }
+      }
+
+      if (!OldestPlaybackActivity && NewestPlaybackActivity) {
+        const formattedDateTimeNew = moment(NewestPlaybackActivity).format("YYYY-MM-DD HH:mm:ss");
+        query = query + ` WHERE DateCreated > '${formattedDateTimeNew}'`;
+        if (MaxPlaybackReportingPluginID) {
+          query = query + ` AND rowid > ${MaxPlaybackReportingPluginID}`;
+        }
+      }
+
+      if (!OldestPlaybackActivity && !NewestPlaybackActivity && MaxPlaybackReportingPluginID) {
+        query = query + ` WHERE rowid > ${MaxPlaybackReportingPluginID}`;
+      }
+
+      query += " order by rowid";
+
+      PlaybacksyncTask.loggedData.push({ color: "dodgerblue", Message: "Query built. Executing." });
+      //
+
+      const PlaybackData = await API.StatsSubmitCustomQuery(query);
+
+      let DataToInsert = await PlaybackData.map(mappingPlaybackReporting);
+
+      if (DataToInsert.length > 0) {
+        PlaybacksyncTask.loggedData.push({ color: "dodgerblue", Message: `Inserting ${DataToInsert.length} Rows.` });
+        let result = await db.insertBulk("jf_playback_reporting_plugin_data", DataToInsert, columnsPlaybackReporting);
+
+        if (result.Result === "SUCCESS") {
+          PlaybacksyncTask.loggedData.push({ color: "dodgerblue", Message: `${DataToInsert.length} Rows have been inserted.` });
+          PlaybacksyncTask.loggedData.push({
+            color: "yellow",
+            Message: "Running process to format data to be inserted into the Activity Table",
+          });
+        } else {
+          PlaybacksyncTask.loggedData.push({ color: "red", Message: "Error: " + result.message });
+          await logging.updateLog(PlaybacksyncTask.uuid, PlaybacksyncTask.loggedData, taskstate.FAILED);
+        }
+      }
+
+      PlaybacksyncTask.loggedData.push({ color: "dodgerblue", Message: "Process complete. Data has been imported." });
     }
+    await db.query("CALL ji_insert_playback_plugin_data_to_activity_table()");
+    PlaybacksyncTask.loggedData.push({ color: "dodgerblue", Message: "Any imported data has been processed." });
 
-    PlaybacksyncTask.loggedData.push({ color: "dodgerblue", Message: "Process complete. Data has been imported." });
+    PlaybacksyncTask.loggedData.push({ color: "lawngreen", Message: `Playback Reporting Plugin Sync Complete` });
+    await logging.updateLog(PlaybacksyncTask.uuid, PlaybacksyncTask.loggedData, taskstate.SUCCESS);
+  } catch (error) {
+    PlaybacksyncTask.loggedData.push({ color: "red", Message: `Error: ${error}` });
+    await logging.updateLog(PlaybacksyncTask.uuid, PlaybacksyncTask.loggedData, taskstate.FAILED);
+    sendUpdate("PlaybackSyncTask", { type: "Error", message: "Error: Playback Plugin Sync failed" });
   }
-  await db.query("CALL ji_insert_playback_plugin_data_to_activity_table()");
-  PlaybacksyncTask.loggedData.push({ color: "dodgerblue", Message: "Any imported data has been processed." });
-
-  PlaybacksyncTask.loggedData.push({ color: "lawngreen", Message: `Playback Reporting Plugin Sync Complete` });
 }
 
 async function updateLibraryStatsData() {
@@ -970,63 +987,73 @@ async function partialSync(triggertype) {
 
 ///////////////////////////////////////Sync All
 router.get("/beginSync", async (req, res) => {
-  const config = await new configClass().getConfig();
+  try {
+    const taskManager = new TaskManager().getInstance();
+    const taskScheduler = new TaskScheduler().getInstance();
+    const success = taskManager.addTask({
+      task: taskManager.taskList.JellyfinSync,
+      onComplete: async () => {
+        console.log("Full Sync completed successfully");
+        await taskScheduler.getTaskHistory();
+        res.send("Full Sync completed successfully");
 
-  if (config.error) {
-    res.send({ error: "Config Details Not Found" });
-    return;
-  }
-
-  const last_execution = await db
-    .query(
-      `SELECT "Result"
-  FROM public.jf_logging
-  WHERE "Name"='${taskName.fullsync}'
-  ORDER BY "TimeRun" DESC
-  LIMIT 1`
-    )
-    .then((res) => res.rows);
-
-  if (last_execution.length !== 0) {
-    if (last_execution[0].Result === taskstate.RUNNING) {
-      sendUpdate("TaskError", "Error: Sync is already running");
-      res.send();
+        sendUpdate("FullSyncTask", { type: "Success", message: triggertype.Manual + " Full Sync Completed" });
+      },
+      onError: async (error) => {
+        console.error(error);
+        await taskScheduler.getTaskHistory();
+        res.status(500).send("Full Sync failed");
+        sendUpdate("FullSyncTask", { type: "Error", message: "Error: Full Sync failed" });
+      },
+    });
+    if (!success) {
+      res.status(500).send("Full Sync already running");
+      sendUpdate("FullSyncTask", { type: "Error", message: "Full Sync is already running" });
       return;
     }
-  }
 
-  await fullSync(triggertype.Manual);
-  res.send();
+    taskManager.startTask(taskManager.taskList.JellyfinSync, triggertype.Manual);
+  } catch (error) {
+    console.error(error);
+    res.status(500).send("Full Sync failed");
+  }
 });
 
 router.get("/beginPartialSync", async (req, res) => {
-  const config = await new configClass().getConfig();
+  try {
+    const taskManager = new TaskManager().getInstance();
+    const taskScheduler = new TaskScheduler().getInstance();
+    const success = taskManager.addTask({
+      task: taskManager.taskList.PartialJellyfinSync,
+      onComplete: async () => {
+        console.log("Recently Added Items Sync completed successfully");
+        await taskScheduler.getTaskHistory();
+        res.send("Recently Added Items Sync completed successfully");
 
-  if (config.error) {
-    res.send({ error: config.error });
-    return;
-  }
-
-  const last_execution = await db
-    .query(
-      `SELECT "Result"
-  FROM public.jf_logging
-  WHERE "Name"='${taskName.partialsync}'
-  ORDER BY "TimeRun" DESC
-  LIMIT 1`
-    )
-    .then((res) => res.rows);
-
-  if (last_execution.length !== 0) {
-    if (last_execution[0].Result === taskstate.RUNNING) {
-      sendUpdate("TaskError", "Error: Sync is already running");
-      res.send();
+        sendUpdate("PartialSyncTask", { type: "Success", message: triggertype.Manual + " Recently Added Items Sync Completed" });
+      },
+      onError: async (error) => {
+        await taskScheduler.getTaskHistory();
+        console.error(error);
+        res.status(500).send("Recently Added Items Sync failed");
+        sendUpdate("PartialSyncTask", { type: "Error", message: "Error: Recently Added Items Sync failed" });
+      },
+      onExit: async () => {
+        await taskScheduler.getTaskHistory();
+        sendUpdate("PartialSyncTask", { type: "Error", message: "Task Stopped" });
+      },
+    });
+    if (!success) {
+      res.status(500).send("Recently Added Items Sync already running");
+      sendUpdate("PartialSyncTask", { type: "Error", message: "Recently Added Items Sync is already running" });
       return;
     }
-  }
 
-  await partialSync(triggertype.Manual);
-  res.send();
+    taskManager.startTask(taskManager.taskList.PartialJellyfinSync, triggertype.Manual);
+  } catch (error) {
+    console.error(error);
+    res.status(500).send("Recently Added Items Sync failed");
+  }
 });
 
 ///////////////////////////////////////Write Users
@@ -1143,31 +1170,37 @@ router.post("/fetchItem", async (req, res) => {
 
 //////////////////////////////////////////////////////syncPlaybackPluginData
 router.get("/syncPlaybackPluginData", async (req, res) => {
-  const config = await new configClass().getConfig();
-
-  const uuid = randomUUID();
-  PlaybacksyncTask = { loggedData: [], uuid: uuid };
   try {
-    await logging.insertLog(uuid, triggertype.Manual, taskName.import);
-    sendUpdate("PlaybackSyncTask", { type: "Start", message: "Playback Plugin Sync Started" });
+    const taskManager = new TaskManager().getInstance();
+    const taskScheduler = new TaskScheduler().getInstance();
+    const success = taskManager.addTask({
+      task: taskManager.taskList.JellyfinPlaybackReportingPluginSync,
+      onComplete: async () => {
+        console.log("Playback Plugin Sync completed successfully");
 
-    if (config.error) {
-      res.send({ error: config.error });
-      PlaybacksyncTask.loggedData.push({ Message: config.error });
-      await logging.updateLog(uuid, PlaybacksyncTask.loggedData, taskstate.FAILED);
+        await taskScheduler.getTaskHistory();
+        res.send("Playback Plugin Sync completed successfully");
+      },
+      onError: async (error) => {
+        await taskScheduler.getTaskHistory();
+        console.error(error);
+        res.status(500).send("Playback Plugin Sync failed");
+      },
+      onExit: async () => {
+        await taskScheduler.getTaskHistory();
+        sendUpdate("PlaybackSyncTask", { type: "Error", message: "Task Stopped" });
+      },
+    });
+    if (!success) {
+      res.status(500).send("Playback Plugin Sync already running");
+      sendUpdate("PlaybackSyncTask", { type: "Error", message: "Playback Plugin Sync is already running" });
       return;
     }
 
-    await sleep(5000);
-    await syncPlaybackPluginData();
-
-    await logging.updateLog(PlaybacksyncTask.uuid, PlaybacksyncTask.loggedData, taskstate.SUCCESS);
-    sendUpdate("PlaybackSyncTask", { type: "Success", message: "Playback Plugin Sync Completed" });
-    res.send("syncPlaybackPluginData Complete");
+    taskManager.startTask(taskManager.taskList.JellyfinPlaybackReportingPluginSync, triggertype.Manual, PlaybacksyncTask);
   } catch (error) {
-    PlaybacksyncTask.loggedData.push({ color: "red", Message: getErrorLineNumber(error) + ": Error: " + error });
-    await logging.updateLog(PlaybacksyncTask.uuid, PlaybacksyncTask.loggedData, taskstate.FAILED);
-    res.send("syncPlaybackPluginData Halted with Errors");
+    console.error(error);
+    res.status(500).send("Playback Plugin Sync failed");
   }
 });
 
@@ -1188,4 +1221,5 @@ module.exports = {
   router,
   fullSync,
   partialSync,
+  syncPlaybackPluginData,
 };

backend/server.js

@@ -28,7 +28,9 @@ const utilsRouter = require("./routes/utils");
 
 // tasks
 const ActivityMonitor = require("./tasks/ActivityMonitor");
-const tasks = require("./tasks/tasks");
+const TaskManager = require("./classes/task-manager-singleton");
+const TaskScheduler = require("./classes/task-scheduler-singleton");
+// const tasks = require("./tasks/tasks");
 
 // websocket
 const { setupWebSocketServer } = require("./ws");
@@ -239,9 +241,8 @@ try {
       server.listen(PORT, LISTEN_IP, async () => {
         console.log(`[JELLYSTAT] Server listening on http://127.0.0.1:${PORT}`);
         ActivityMonitor.ActivityMonitor(1000);
-        tasks.FullSyncTask();
-        tasks.RecentlyAddedItemsSyncTask();
-        tasks.BackupTask();
+        new TaskManager();
+        new TaskScheduler();
       });
     });
   });

backend/socket-io-client.js

@@ -0,0 +1,30 @@
+const io = require("socket.io-client");
+
+class SocketIoClient {
+  constructor(serverUrl) {
+    this.serverUrl = serverUrl;
+    this.client = null;
+  }
+
+  connect() {
+    this.client = io(this.serverUrl);
+  }
+
+  waitForConnection() {
+    return new Promise((resolve) => {
+      if (this.client && this.client.connected) {
+        resolve();
+      } else {
+        this.client.on("connect", resolve);
+      }
+    });
+  }
+
+  sendMessage(message) {
+    if (this.client && this.client.connected) {
+      this.client.emit("message", JSON.stringify(message));
+    }
+  }
+}
+
+module.exports = SocketIoClient;

backend/tasks/BackupTask.js

@@ -1,128 +1,36 @@
-const db = require("../db");
+const { parentPort } = require("worker_threads");
 const Logging = require("../classes/logging");
-const configClass =require("../classes/config");
-
 const backup = require("../classes/backup");
-const moment = require('moment');
-const { randomUUID }  = require('crypto');
+const { randomUUID } = require("crypto");
 const taskstate = require("../logging/taskstate");
 const taskName = require("../logging/taskName");
 const triggertype = require("../logging/triggertype");
+const { sendUpdate } = require("../ws");
 
+async function runBackupTask(triggerType = triggertype.Automatic) {
+  try {
+    const uuid = randomUUID();
+    const refLog = { logData: [], uuid: uuid };
 
-async function BackupTask() {
-  try{
+    console.log("Running Scheduled Backup");
 
-    await db.query(
-          `UPDATE jf_logging SET "Result"='${taskstate.FAILED}' WHERE "Name"='${taskName.backup}' AND "Result"='${taskstate.RUNNING}'`
-         );
-    }
-    catch(error)
-    {
-      console.log('Error Cleaning up Backup Tasks: '+error);
-    }
-let interval=10000;
-let taskDelay=1440; // 1 day in minutes
+    Logging.insertLog(uuid, triggerType, taskName.backup);
 
+    await backup(refLog);
+    Logging.updateLog(uuid, refLog.logData, taskstate.SUCCESS);
+    sendUpdate("BackupTask", { type: "Success", message: `${triggerType} Backup Completed` });
+    console.log("Scheduled Backup Complete");
+    parentPort.postMessage({ status: "complete" });
+  } catch (error) {
+    parentPort.postMessage({ status: "error", message: error.message });
 
-try{//get interval from db
-
-  
-  const settingsjson = await db
-    .query('SELECT settings FROM app_config where "ID"=1')
-    .then((res) => res.rows);
-    
-  if (settingsjson.length > 0) {
-    const settings = settingsjson[0].settings || {};
-
-    let backuptasksettings = settings.Tasks?.Backup || {};
-
-    if (backuptasksettings.Interval) {
-      taskDelay=backuptasksettings.Interval;
-    } else {
-      backuptasksettings.Interval=taskDelay;
-    }
-
-    if(!settings.Tasks)
-    {
-      settings.Tasks = {};
-    }
-    if(!settings.Tasks.Backup)
-    {
-      settings.Tasks.Backup = {};
-    }
-    settings.Tasks.Backup = backuptasksettings;
-
-
-    let query = 'UPDATE app_config SET settings=$1 where "ID"=1';
-
-    await db.query(query, [settings]);
+    console.log(error);
+    return [];
   }
 }
-catch(error)
-{
-  console.log('Sync Task Settings Error: '+error);
-}
 
-async function intervalCallback() {
-    clearInterval(intervalTask); 
-    try{
-       let current_time = moment();
-       const config = await new configClass().getConfig();
-   
-       if (config.error) 
-       {
-           return;
-       }
-     
-     
-       const last_execution=await db.query( `SELECT "TimeRun","Result"
-                                             FROM public.jf_logging
-                                             WHERE "Name"='${taskName.backup}' AND "Result" in ('${taskstate.SUCCESS}','${taskstate.RUNNING}')
-                                             ORDER BY "TimeRun" DESC
-                                             LIMIT 1`).then((res) => res.rows);
-
-       if(last_execution.length!==0)
-       { 
-           let last_execution_time = moment(last_execution[0].TimeRun).add(taskDelay, 'minutes');
-
-           if(!current_time.isAfter(last_execution_time) || last_execution[0].Result ===taskstate.RUNNING)
-           {
-   
-               intervalTask = setInterval(intervalCallback, interval); 
-               return;
-           }
-       }
-
-      const uuid = randomUUID();
-      let refLog={logData:[],uuid:uuid};
-
-   
-       console.log('Running Scheduled Backup');
-
-       Logging.insertLog(uuid,triggertype.Automatic,taskName.backup);
-   
-
-       await backup(refLog);
-       Logging.updateLog(uuid,refLog.logData,taskstate.SUCCESS);
-
-
-       console.log('Scheduled Backup Complete');
-       
-       } catch (error) 
-       {
-           console.log(error);
-           return [];
-       }
-   
-       intervalTask = setInterval(intervalCallback, interval); 
-     }
-   
-   let intervalTask = setInterval(intervalCallback, interval);
-
-
-}
-
-module.exports = {
-  BackupTask,
-};
+parentPort.on("message", (message) => {
+  if (message.command === "start") {
+    runBackupTask(message.triggertype);
+  }
+});

backend/tasks/FullSyncTask.js

@@ -1,115 +1,22 @@
-const db = require("../db");
-const moment = require("moment");
-const sync = require("../routes/sync");
-const taskName = require("../logging/taskName");
-const taskstate = require("../logging/taskstate");
+const { parentPort } = require("worker_threads");
 const triggertype = require("../logging/triggertype");
+const sync = require("../routes/sync");
 
-async function FullSyncTask() {
+async function runFullSyncTask(triggerType = triggertype.Automatic) {
   try {
-    await db.query(
-      `UPDATE jf_logging SET "Result"='${taskstate.FAILED}' WHERE "Name"='${taskName.fullsync}' AND "Result"='${taskstate.RUNNING}'`
-    );
+    await sync.fullSync(triggerType);
+
+    parentPort.postMessage({ status: "complete" });
   } catch (error) {
-    console.log("Error Cleaning up Sync Tasks: " + error);
+    parentPort.postMessage({ status: "error", message: error.message });
+
+    console.log(error);
+    return [];
   }
-
-  let interval = 10000;
-
-  let taskDelay = 1440; //in minutes
-
-  async function fetchTaskSettings() {
-    try {
-      //get interval from db
-
-      const settingsjson = await db.query('SELECT settings FROM app_config where "ID"=1').then((res) => res.rows);
-
-      if (settingsjson.length > 0) {
-        const settings = settingsjson[0].settings || {};
-
-        let synctasksettings = settings.Tasks?.JellyfinSync || {};
-
-        if (synctasksettings.Interval) {
-          taskDelay = synctasksettings.Interval;
-        } else {
-          synctasksettings.Interval = taskDelay;
-
-          if (!settings.Tasks) {
-            settings.Tasks = {};
-          }
-          if (!settings.Tasks.JellyfinSync) {
-            settings.Tasks.JellyfinSync = {};
-          }
-          settings.Tasks.JellyfinSync = synctasksettings;
-
-          let query = 'UPDATE app_config SET settings=$1 where "ID"=1';
-
-          await db.query(query, [settings]);
-        }
-      }
-    } catch (error) {
-      console.log("Sync Task Settings Error: " + error);
-    }
-  }
-
-  async function intervalCallback() {
-    clearInterval(intervalTask);
-    try {
-      let current_time = moment();
-      const { rows: config } = await db.query('SELECT * FROM app_config where "ID"=1');
-
-      if (config.length === 0 || config[0].JF_HOST === null || config[0].JF_API_KEY === null) {
-        return;
-      }
-
-      const last_execution = await db
-        .query(
-          `SELECT "TimeRun","Result"
-                                          FROM public.jf_logging
-                                          WHERE "Name"='${taskName.fullsync}'
-                                          ORDER BY "TimeRun" DESC
-                                          LIMIT 1`
-        )
-        .then((res) => res.rows);
-
-      const last_execution_partialSync = await db
-        .query(
-          `SELECT "TimeRun","Result"
-                                            FROM public.jf_logging
-                                            WHERE "Name"='${taskName.partialsync}'
-                                            AND "Result"='${taskstate.RUNNING}'
-                                            ORDER BY "TimeRun" DESC
-                                            LIMIT 1`
-        )
-        .then((res) => res.rows);
-      if (last_execution.length !== 0) {
-        await fetchTaskSettings();
-        let last_execution_time = moment(last_execution[0].TimeRun).add(taskDelay, "minutes");
-
-        if (
-          !current_time.isAfter(last_execution_time) ||
-          last_execution[0].Result === taskstate.RUNNING ||
-          last_execution_partialSync.length > 0
-        ) {
-          intervalTask = setInterval(intervalCallback, interval);
-          return;
-        }
-      }
-
-      console.log("Running Scheduled Sync");
-      await sync.fullSync(triggertype.Automatic);
-      console.log("Scheduled Sync Complete");
-    } catch (error) {
-      console.log(error);
-      return [];
-    }
-
-    intervalTask = setInterval(intervalCallback, interval);
-  }
-
-  let intervalTask = setInterval(intervalCallback, interval);
 }
 
-module.exports = {
-  FullSyncTask,
-};
+parentPort.on("message", (message) => {
+  if (message.command === "start") {
+    runFullSyncTask(message.triggertype);
+  }
+});

backend/tasks/PlaybackReportingPluginSyncTask.js

@@ -0,0 +1,21 @@
+const { parentPort } = require("worker_threads");
+const sync = require("../routes/sync");
+
+async function runPlaybackReportingPluginSyncTask() {
+  try {
+    await sync.syncPlaybackPluginData();
+
+    parentPort.postMessage({ status: "complete" });
+  } catch (error) {
+    parentPort.postMessage({ status: "error", message: error.message });
+
+    console.log(error);
+    return [];
+  }
+}
+
+parentPort.on("message", (message) => {
+  if (message.command === "start") {
+    runPlaybackReportingPluginSyncTask();
+  }
+});

backend/tasks/RecentlyAddedItemsSyncTask.js

@@ -1,115 +1,22 @@
-const db = require("../db");
-const moment = require("moment");
-const sync = require("../routes/sync");
-const taskName = require("../logging/taskName");
-const taskstate = require("../logging/taskstate");
+const { parentPort } = require("worker_threads");
 const triggertype = require("../logging/triggertype");
+const sync = require("../routes/sync");
 
-async function RecentlyAddedItemsSyncTask() {
+async function runPartialSyncTask(triggerType = triggertype.Automatic) {
   try {
-    await db.query(
-      `UPDATE jf_logging SET "Result"='${taskstate.FAILED}' WHERE "Name"='${taskName.partialsync}' AND "Result"='${taskstate.RUNNING}'`
-    );
+    await sync.partialSync(triggerType);
+
+    parentPort.postMessage({ status: "complete" });
   } catch (error) {
-    console.log("Error Cleaning up Sync Tasks: " + error);
+    parentPort.postMessage({ status: "error", message: error.message });
+
+    console.log(error);
+    return [];
   }
-
-  let interval = 11000;
-
-  let taskDelay = 60; //in minutes
-
-  async function fetchTaskSettings() {
-    try {
-      //get interval from db
-
-      const settingsjson = await db.query('SELECT settings FROM app_config where "ID"=1').then((res) => res.rows);
-
-      if (settingsjson.length > 0) {
-        const settings = settingsjson[0].settings || {};
-
-        let synctasksettings = settings.Tasks?.PartialJellyfinSync || {};
-
-        if (synctasksettings.Interval) {
-          taskDelay = synctasksettings.Interval;
-        } else {
-          synctasksettings.Interval = taskDelay;
-
-          if (!settings.Tasks) {
-            settings.Tasks = {};
-          }
-          if (!settings.Tasks.PartialJellyfinSync) {
-            settings.Tasks.PartialJellyfinSync = {};
-          }
-          settings.Tasks.PartialJellyfinSync = synctasksettings;
-
-          let query = 'UPDATE app_config SET settings=$1 where "ID"=1';
-
-          await db.query(query, [settings]);
-        }
-      }
-    } catch (error) {
-      console.log("Sync Task Settings Error: " + error);
-    }
-  }
-
-  async function intervalCallback() {
-    clearInterval(intervalTask);
-    try {
-      let current_time = moment();
-      const { rows: config } = await db.query('SELECT * FROM app_config where "ID"=1');
-
-      if (!config || config.length === 0 || config[0].JF_HOST === null || config[0].JF_API_KEY === null) {
-        return;
-      }
-
-      const last_execution = await db
-        .query(
-          `SELECT "TimeRun","Result"
-                                          FROM public.jf_logging
-                                          WHERE "Name"='${taskName.partialsync}'
-                                          ORDER BY "TimeRun" DESC
-                                          LIMIT 1`
-        )
-        .then((res) => res.rows);
-
-      const last_execution_FullSync = await db
-        .query(
-          `SELECT "TimeRun","Result"
-                                            FROM public.jf_logging
-                                            WHERE "Name"='${taskName.fullsync}'
-                                            AND "Result"='${taskstate.RUNNING}'
-                                            ORDER BY "TimeRun" DESC
-                                            LIMIT 1`
-        )
-        .then((res) => res.rows);
-      if (last_execution.length !== 0) {
-        await fetchTaskSettings();
-        let last_execution_time = moment(last_execution[0].TimeRun).add(taskDelay, "minutes");
-
-        if (
-          !current_time.isAfter(last_execution_time) ||
-          last_execution[0].Result === taskstate.RUNNING ||
-          last_execution_FullSync.length > 0
-        ) {
-          intervalTask = setInterval(intervalCallback, interval);
-          return;
-        }
-      }
-
-      console.log("Running Recently Added Scheduled Sync");
-      await sync.partialSync(triggertype.Automatic);
-      console.log("Scheduled Recently Added Sync Complete");
-    } catch (error) {
-      console.log(error);
-      return [];
-    }
-
-    intervalTask = setInterval(intervalCallback, interval);
-  }
-
-  let intervalTask = setInterval(intervalCallback, interval);
 }
 
-module.exports = {
-  RecentlyAddedItemsSyncTask,
-};
+parentPort.on("message", (message) => {
+  if (message.command === "start") {
+    runPartialSyncTask(message.triggertype);
+  }
+});

backend/tasks/tasks.js

@@ -1,10 +0,0 @@
-const { BackupTask } = require("./BackupTask");
-const { RecentlyAddedItemsSyncTask } = require("./RecentlyAddedItemsSyncTask");
-const { FullSyncTask } = require("./FullSyncTask");
-
-const tasks = {
-    FullSyncTask:FullSyncTask,
-    RecentlyAddedItemsSyncTask:RecentlyAddedItemsSyncTask,
-    BackupTask:BackupTask,
-  };
-module.exports = tasks;

backend/ws-server-singleton.js

@@ -0,0 +1,17 @@
+class WebSocketServerSingleton {
+  constructor() {
+    if (!WebSocketServerSingleton.instance) {
+      WebSocketServerSingleton.instance = null;
+    }
+  }
+
+  setInstance(io) {
+    WebSocketServerSingleton.instance = io;
+  }
+
+  getInstance() {
+    return WebSocketServerSingleton.instance;
+  }
+}
+
+module.exports = new WebSocketServerSingleton();

backend/ws.js

@@ -1,29 +1,46 @@
 // ws.js
 const socketIO = require("socket.io");
+const webSocketServerSingleton = require("./ws-server-singleton.js");
+const SocketIoClient = require("./socket-io-client.js");
 
+const socketClient = new SocketIoClient("http://127.0.0.1:3000");
 let io; // Store the socket.io server instance
 
 const setupWebSocketServer = (server, namespacePath) => {
-  io = socketIO(server, { path: namespacePath + "/socket.io" }); // Create the socket.io server
+  io = socketIO(server, { path: namespacePath + "/socket.io" });
+
+  socketClient.connect();
 
   io.on("connection", (socket) => {
     // console.log("Client connected to namespace:", namespacePath);
 
     socket.on("message", (message) => {
-      console.log(`Received: ${message}`);
+      const payload = JSON.parse(message);
+      sendUpdate(payload.tag, payload.message);
     });
   });
+
+  webSocketServerSingleton.setInstance(io);
 };
 
 const sendToAllClients = (message) => {
-  if (io) {
-    io.emit("message", message);
+  const ioInstance = webSocketServerSingleton.getInstance();
+  if (ioInstance) {
+    ioInstance.emit("message", message);
   }
 };
 
-const sendUpdate = (tag, message) => {
-  if (io) {
-    io.emit(tag, message);
+const sendUpdate = async (tag, message) => {
+  const ioInstance = webSocketServerSingleton.getInstance();
+  if (ioInstance) {
+    ioInstance.emit(tag, message);
+  } else {
+    if (socketClient.client == null || socketClient.client.connected == false) {
+      socketClient.connect();
+      await socketClient.waitForConnection();
+    }
+
+    socketClient.sendMessage({ tag: tag, message: message });
   }
 };
 

package-lock.json

@@ -6,7 +6,7 @@
   "packages": {
     "": {
       "name": "jfstat",
-      "version": "1.1.3",
+      "version": "1.1.4",
       "dependencies": {
         "@emotion/react": "^11.14.0",
         "@emotion/styled": "^11.14.0",
@@ -70,7 +70,7 @@
         "swagger-autogen": "^2.23.5",
         "swagger-jsdoc": "^6.2.8",
         "swagger-ui-express": "^5.0.0",
-        "ws": "^8.13.0"
+        "ws": "^8.18.1"
       },
       "devDependencies": {
         "@types/react": "^18.2.15",
@@ -9326,6 +9326,26 @@
       "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz",
       "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA=="
     },
+    "node_modules/engine.io-client/node_modules/ws": {
+      "version": "8.17.1",
+      "resolved": "https://registry.npmjs.org/ws/-/ws-8.17.1.tgz",
+      "integrity": "sha512-6XQFvXTkbfUOZOKKILFG1PDK2NDQs4azKQl26T0YS5CxqWLgXajbPZ+h4gZekJyRqFU8pvnbAbbs/3TgRPy+GQ==",
+      "engines": {
+        "node": ">=10.0.0"
+      },
+      "peerDependencies": {
+        "bufferutil": "^4.0.1",
+        "utf-8-validate": ">=5.0.2"
+      },
+      "peerDependenciesMeta": {
+        "bufferutil": {
+          "optional": true
+        },
+        "utf-8-validate": {
+          "optional": true
+        }
+      }
+    },
     "node_modules/engine.io-parser": {
       "version": "5.2.1",
       "resolved": "https://registry.npmjs.org/engine.io-parser/-/engine.io-parser-5.2.1.tgz",
@@ -22647,9 +22667,9 @@
       }
     },
     "node_modules/ws": {
-      "version": "8.17.1",
-      "resolved": "https://registry.npmjs.org/ws/-/ws-8.17.1.tgz",
-      "integrity": "sha512-6XQFvXTkbfUOZOKKILFG1PDK2NDQs4azKQl26T0YS5CxqWLgXajbPZ+h4gZekJyRqFU8pvnbAbbs/3TgRPy+GQ==",
+      "version": "8.18.1",
+      "resolved": "https://registry.npmjs.org/ws/-/ws-8.18.1.tgz",
+      "integrity": "sha512-RKW2aJZMXeMxVpnZ6bck+RswznaxmzdULiBr6KY7XkTnW8uvt0iT9H5DkHUChXrc+uurzwa0rVI16n/Xzjdz1w==",
       "engines": {
         "node": ">=10.0.0"
       },

package.json

@@ -77,7 +77,7 @@
     "swagger-autogen": "^2.23.5",
     "swagger-jsdoc": "^6.2.8",
     "swagger-ui-express": "^5.0.0",
-    "ws": "^8.13.0"
+    "ws": "^8.18.1"
   },
   "devDependencies": {
     "@types/react": "^18.2.15",

src/lib/tasklist.jsx

@@ -17,7 +17,7 @@ export const taskList = [
   },
   {
     id: 2,
-    name: "Jellyfin Playback Reporting Plugin Sync",
+    name: "JellyfinPlaybackReportingPluginSync",
     description: <Trans i18nKey={"TASK_DESCRIPTION.Jellyfin_Playback_Reporting_Plugin_Sync"} />,
     type: "IMPORT",
     link: "/sync/syncPlaybackPluginData",

src/pages/components/settings/Task.jsx

@@ -8,57 +8,65 @@ import i18next from "i18next";
 import { Trans } from "react-i18next";
 import "../../css/settings/settings.css";
 
-function Task({ task, processing, taskIntervals, updateTask, onClick }) {
-    const intervals = [
-      { value: 15, display: i18next.t("SETTINGS_PAGE.INTERVALS.15_MIN") },
-      { value: 30, display: i18next.t("SETTINGS_PAGE.INTERVALS.30_MIN") },
-      { value: 60, display: i18next.t("SETTINGS_PAGE.INTERVALS.1_HOUR") },
-      { value: 720, display: i18next.t("SETTINGS_PAGE.INTERVALS.12_HOURS") },
-      { value: 1440, display: i18next.t("SETTINGS_PAGE.INTERVALS.1_DAY") },
-      { value: 10080, display: i18next.t("SETTINGS_PAGE.INTERVALS.1_WEEK") },
-    ];
-    return (
-      <TableRow key={task.id}>
-        <TableCell>{task.description}</TableCell>
-        <TableCell>
-          <Trans i18nKey={`TASK_TYPE.${task.type}`} />
-        </TableCell>
-  
-        <TableCell>
-          {task.type === "JOB" ? (
-            <Dropdown className="w-100" key={task.id}>
-              <Dropdown.Toggle variant="outline-primary" className="w-100 dropdown-basic">
-                {taskIntervals &&
-                  intervals.find((interval) => interval.value === (taskIntervals[task.name]?.Interval || 15)).display}
-              </Dropdown.Toggle>
-              <Dropdown.Menu className="w-100">
-                {taskIntervals &&
-                  intervals.map((interval) => (
-                    <Dropdown.Item
-                      onClick={() => updateTask(task.name, interval.value)}
-                      value={interval.value}
-                      key={interval.value}
-                    >
-                      {interval.display}
-                    </Dropdown.Item>
-                  ))}
-              </Dropdown.Menu>
-            </Dropdown>
-          ) : (
-            <></>
-          )}
-        </TableCell>
-        <TableCell className="d-flex justify-content-center">
-          <Button
-            variant={!processing ? "outline-primary" : "outline-light"}
-            disabled={processing}
-            onClick={() => onClick(task.link)}
-          >
-            <Trans i18nKey={"START"} />
-          </Button>
-        </TableCell>
-      </TableRow>
-    );
-  }
+function Task({ task, taskState, processing, taskIntervals, updateTask, onClick, stopTask }) {
+  const intervals = [
+    { value: 15, display: i18next.t("SETTINGS_PAGE.INTERVALS.15_MIN") },
+    { value: 30, display: i18next.t("SETTINGS_PAGE.INTERVALS.30_MIN") },
+    { value: 60, display: i18next.t("SETTINGS_PAGE.INTERVALS.1_HOUR") },
+    { value: 720, display: i18next.t("SETTINGS_PAGE.INTERVALS.12_HOURS") },
+    { value: 1440, display: i18next.t("SETTINGS_PAGE.INTERVALS.1_DAY") },
+    { value: 10080, display: i18next.t("SETTINGS_PAGE.INTERVALS.1_WEEK") },
+  ];
+  const state = taskState ? taskState.filter((state) => state.task === task.name)[0] : null;
 
-  export default Task
+  return (
+    <TableRow key={task.id}>
+      <TableCell>{task.description}</TableCell>
+      <TableCell>
+        <Trans i18nKey={`TASK_TYPE.${task.type}`} />
+      </TableCell>
+
+      <TableCell>
+        {task.type === "JOB" ? (
+          <Dropdown className="w-100" key={task.id}>
+            <Dropdown.Toggle variant="outline-primary" className="w-100 dropdown-basic">
+              {taskIntervals &&
+                intervals.find((interval) => interval.value === (taskIntervals[task.name]?.Interval || 15)).display}
+            </Dropdown.Toggle>
+            <Dropdown.Menu className="w-100">
+              {taskIntervals &&
+                intervals.map((interval) => (
+                  <Dropdown.Item
+                    onClick={() => updateTask(task.name, interval.value)}
+                    value={interval.value}
+                    key={interval.value}
+                  >
+                    {interval.display}
+                  </Dropdown.Item>
+                ))}
+            </Dropdown.Menu>
+          </Dropdown>
+        ) : (
+          <></>
+        )}
+      </TableCell>
+      <TableCell className="d-flex justify-content-center">
+        {state ? (
+          state.running == true ? (
+            <Button variant={"danger"} onClick={() => stopTask(task.name)}>
+              <Trans i18nKey={"STOP"} />
+            </Button>
+          ) : (
+            <Button variant={"outline-primary"} onClick={() => onClick(task.link)}>
+              <Trans i18nKey={"START"} />
+            </Button>
+          )
+        ) : (
+          <></>
+        )}
+      </TableCell>
+    </TableRow>
+  );
+}
+
+export default Task;

src/pages/components/settings/Tasks.jsx

@@ -1,4 +1,4 @@
-import { useState } from "react";
+import { useState, useEffect } from "react";
 import axios from "../../../lib/axios_instance";
 import Table from "@mui/material/Table";
 import TableBody from "@mui/material/TableBody";
@@ -6,8 +6,9 @@ import TableCell from "@mui/material/TableCell";
 import TableContainer from "@mui/material/TableContainer";
 import TableHead from "@mui/material/TableHead";
 import TableRow from "@mui/material/TableRow";
-import { taskList } from "../../../lib/tasklist";
+import { taskList } from "../../../lib/tasklist.jsx";
 import Task from "./Task";
+import socket from "../../../socket";
 
 import "../../css/settings/settings.css";
 import { Trans } from "react-i18next";
@@ -16,6 +17,18 @@ export default function Tasks() {
   const [processing, setProcessing] = useState(false);
   const [taskIntervals, setTaskIntervals] = useState([]);
   const token = localStorage.getItem("token");
+  const [taskStateList, setTaskStateList] = useState();
+
+  useEffect(() => {
+    socket.on("task-list", (data) => {
+      if (typeof data === "object" && Array.isArray(data)) {
+        setTaskStateList(data);
+      }
+    });
+    return () => {
+      socket.off("task-list");
+    };
+  }, [taskStateList]);
 
   async function executeTask(url) {
     setProcessing(true);
@@ -33,6 +46,19 @@ export default function Tasks() {
     setProcessing(false);
   }
 
+  async function stopTask(task) {
+    await axios
+      .get(`/api/stopTask?task=${task}`, {
+        headers: {
+          Authorization: `Bearer ${token}`,
+          "Content-Type": "application/json",
+        },
+      })
+      .catch((error) => {
+        console.log(error);
+      });
+  }
+
   async function updateTaskSettings(taskName, Interval) {
     taskName = taskName.replace(/ /g, "");
 
@@ -100,15 +126,17 @@ export default function Tasks() {
           </TableHead>
           <TableBody>
             {taskList.map((task) => (
-              <Task 
-                key={task.id} 
-                task={task} 
-                processing={processing} 
+              <Task
+                key={task.id}
+                task={task}
+                taskState={taskStateList}
+                processing={processing}
                 taskIntervals={taskIntervals}
-                updateTask={updateTaskSettings} 
-                onClick={executeTask} 
-                />
-              ))}
+                updateTask={updateTaskSettings}
+                onClick={executeTask}
+                stopTask={stopTask}
+              />
+            ))}
           </TableBody>
         </Table>
       </TableContainer>