dtp-base/app/workers/tracker-monitor.js

// tracker-monitor.js
// Copyright (C) 2024 DTP Technologies, LLC
// All Rights Reserved

'use strict';

import 'dotenv/config';

import path, { dirname } from 'path';
import dayjs from 'dayjs';

import { SiteRuntime } from '../../lib/site-lib.js';

import { CronJob } from 'cron';
const CRON_TIMEZONE = 'America/New_York';

class TrackerMonitorWorker extends SiteRuntime {

  static get name ( ) { return 'TrackerMonitorWorker'; }
  static get slug ( ) { return 'trackerMonitor'; }

  constructor (rootPath) {
    super(TrackerMonitorWorker, rootPath);
  }

  async start ( ) {
    await super.start();

    const mongoose = await import('mongoose');
    this.TaskSession = mongoose.model('TaskSession');

    this.viewModel = { };
    await this.populateViewModel(this.viewModel);

    /*
     * Cron jobs
     */

    const sessionExpireSchedule = '* */5 * * * *'; // Every 5 minutes
    this.cronJob = new CronJob(
      sessionExpireSchedule,
      this.expireTaskSessions.bind(this),
      null,
      true,
      CRON_TIMEZONE,
    );
    await this.expireTaskSessions();

    /*
     * Bull Queue job processors
     */

    // this.log.info('registering queue job processor', { config: this.config.jobQueues.links });
    // this.linksProcessingQueue = this.services.jobQueue.getJobQueue('links', this.config.jobQueues.links);
    // this.linksProcessingQueue.process('link-ingest', 1, this.ingestLink.bind(this));

    this.log.info('Tracker Monitor online');
  }

  async shutdown ( ) {
    this.log.alert('ChatLinksWorker shutting down');
    await super.shutdown();
  }

  async expireTaskSessions ( ) {
    const { task: taskService } = this.services;

    const NOW = new Date();
    const oldestDate = dayjs(NOW).subtract(10, 'minute');

    this.log.debug('scanning for defunct sessions');
    await this.TaskSession
      .find({
        $and: [
          { status: { $in: ['active', 'reconnecting'] } },
          { lastUpdated: { $lt: oldestDate } },
        ],
      })
      .cursor()
      .eachAsync(async (session) => {
        this.log.info('expiring defunct task work session', {
          session: {
            _id: session._id,
            created: session.created,
            lastUpdated: session.lastUpdated,
          },
        });
        taskService.closeTaskSession(session, 'expired');
      });
  }
}

(async ( ) => {
  
  try {
    const { fileURLToPath } = await import('node:url');
    const __dirname = dirname(fileURLToPath(import.meta.url)); // jshint ignore:line

    const worker = new TrackerMonitorWorker(path.resolve(__dirname, '..', '..'));
    await worker.start();

  } catch (error) {
    console.error('failed to start tracker monitor worker', { error });
    process.exit(-1);
  }

})();