dtp-base/app/services/host-cache.js

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

'use strict';

import dgram from 'node:dgram';
import { v4 as uuidv4 } from 'uuid';

import { SiteService, SiteError } from '../../lib/site-lib.js';

export default class HostCacheService extends SiteService {

  static get name ( ) { return 'HostCacheService'; }
  static get slug ( ) { return 'hostCache'; }

  constructor (dtp) {
    super(dtp, HostCacheService);
    this.transactions = { };
  }

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

    this.log.info('creating UDP host-cache socket');
    this.hostCache = dgram.createSocket('udp4', this.onMessage.bind(this));
    this.hostCache.on('error', this.onError.bind(this));

    this.log.info('connecting UDP host-cache socket');
    // this.hostCache.bind(0, '127.0.0.1');
    this.hostCache.connect(
      parseInt(process.env.HOST_CACHE_PORT || '8000', 10),
      process.env.HOST_CACHE_HOST || '127.0.0.1',
    );
  }

  async stop ( ) {
    if (this.hostCache) {
      this.log.info('disconnecting UDP host-cache socket');
      this.hostCache.disconnect();
      delete this.hostCache;
    }
  }

  async getFile (bucket, key) {
    return new Promise((resolve, reject) => {
      const transaction = { tid: uuidv4(), bucket, key, resolve, reject };
      this.transactions[transaction.tid] = transaction;
      const message = JSON.stringify({
        tid: transaction.tid,
        cmd: 'getFile',
        params: { bucket, key },
      });
      this.hostCache.send(message);
    });
  }

  async fetchUrl (url) {
    return new Promise((resolve, reject) => {
      const transaction = { tid: uuidv4(), url, resolve, reject };
      this.transactions[transaction.tid] = transaction;
      const message = JSON.stringify({
        tid: transaction.tid,
        cmd: 'fetchUrl',
        params: { url },
      });
      this.hostCache.send(message);
    });
  }

  async onMessage (message, rinfo) {
    message = message.toString('utf8');
    message = JSON.parse(message);
    switch (message.res.cmd) {
      case 'getFile':
        return this.onGetFile(message, rinfo);
      case 'fetchUrl':
        return this.onFetchUrl(message, rinfo);
    }
  }

  async onGetFile (message) {
    const transaction = this.transactions[message.tid];
    if (!transaction) {
      this.log.error('getFile response received with no matching transaction', { tid: message.tid });
      return;
    }

    delete this.transactions[message.tid];

    if (!message.res.success) {
      transaction.reject(new SiteError(message.res.statusCode, message.res.message));
      return;
    }

    transaction.resolve({
      success: message.res.success,
      message: message.res.message,
      file: message.res.file,
      flags: message.flags,
      duration: message.duration,
    });
  }

  async onFetchUrl (message) {
    const transaction = this.transactions[message.tid];
    if (!transaction) {
      this.log.error('fetchUrl response received with no matching transaction', { tid: message.tid });
      return;
    }

    delete this.transactions[message.tid];

    if (!message.res.success) {
      transaction.reject(new SiteError(message.res.statusCode || 500, message.res.message));
      return;
    }

    transaction.resolve({
      success: message.res.success,
      message: message.res.message,
      file: message.res.file,
      flags: message.flags,
      duration: message.duration,
    });
  }

  async onError (error) {
    this.log.error('onError', { error });
    if ((error.errno !== -111) || (error.code !== 'ECONNREFUSED')) {
      return;
    }
    if (!this.transactions) {
      return;
    }
    if (this.transactions) {
      for (const key in this.transactions) {
        this.log.alert('destroying host cache transaction', { key });
        const transaction = this.transactions[key];
        transaction.reject(error);
        delete this.transactions[key];
      }
    }
  }
}