hostbbq-monitoring/server/src/ctrl/server-connector.class.ts

import axios, { AxiosError } from 'axios';
import moment from 'moment';
import { Pool } from 'mysql';

import defaults from '../../../common/defaults.module';
import { Logger } from '../../../common/util/logger.class';

import { FCMController } from './fcm-controller.class';
import { Timer } from '../timer.class';
import { MariaDBDatabase } from './mariadb-database.class';

export class ServerConnector {
  private subscriptions: Array<{ id: number; interval: number; server: Server }> = [];

  constructor(private pool: Pool) {
    const db = new MariaDBDatabase(pool);
    (async () => {
      try {
        await db.open();
        const serverList = await db.getAllServerConfigs();

        for (const server of serverList) {
          let interval = Number(server.config['syncInterval']);
          if (Number.isNaN(interval)) interval = defaults.serverSync.interval;

          Logger.info('[INFO] Starting Server Sync Connector for', server.title, 'with interval', interval, 'seconds ...');
          const id = Timer.instance.subscribe(interval, async () => await this.timerTick(server));
          this.subscriptions.push({ id, interval, server });

          Logger.info('[INFO] Initial Sync for', server.title, '...');
          await this.timerTick(server);
        }
      } catch (err) {
        Logger.error('[FATAL] Initializing ServerConnector failed:', err);
        Logger.error('[EXITING]');
        process.exit(1);
      } finally {
        try {
          await db.close();
        } catch (e) {}
      }
    })();
  }

  private async timerTick(server: Server & { errors?: number }) {
    Logger.debug('[DEBUG] TICK', new Date(), JSON.stringify(server));
    if (process.env.DEV_MODE) return Logger.warn('[WARN] DEV_MODE active - sync inactive.');

    let trxHdl: number | undefined = undefined;
    const db = new MariaDBDatabase(this.pool);
    try {
      await db.open();

      // Start Transaction, receiving Data and a Transaction Handle
      let response = await axios.get(`http://${server.fqdn}:8890/`, { responseType: 'json' });
      trxHdl = response.data.hdl;

      if (!trxHdl) return; // No data

      const data: ReducedData[] = response.data.data.map((entry: any) => ({ ...entry, time: new Date(entry.time) }));

      // Process data in DB
      await db.insertServerData(server.id, data);

      // Commit Transaction
      await axios.patch(`http://${server.fqdn}:8890/${trxHdl}`, null, { responseType: 'json' });

      if (server.errors) {
        // notify [RECOVERY]
        try {
          await FCMController.instance.sendNotificationToTopic(defaults.fcmTopics.serverData, {
            title: `[RECOVERY] ${moment().format('HH:mm')} Server '${server.title}': [OK]`,
            body: `[RECOVERY] Server '${server.title}': daemon OK`
          });
        } catch (err) {
          Logger.error('[ERROR] Notification failure:', err);
        }
      }
      server.errors = 0;
    } catch (err) {
      Logger.error('[ERROR] Server data sync failed:', err);

      if (err instanceof AxiosError) {
        if (['ECONNREFUSED', 'ECONNABORTED'].includes(err.code ?? '')) {
          try {
            server.errors = (server.errors ?? 0) + 1;
            await FCMController.instance.sendNotificationToTopic(defaults.fcmTopics.serverData, {
              title: `[WARN] ${moment().format('HH:mm')} Server '${server.title}': daemon unreachable (${err.code})`,
              body: new String(err).toString()
            });
          } catch (err) {
            Logger.error('[ERROR] Notification failure:', err);
          }
        }
      }

      if (!!trxHdl) {
        Logger.error(`[WARN] Rolling back transaction #${trxHdl} ... `);
        const response = await axios.delete(`http://${server.fqdn}:8890/${trxHdl}`, { responseType: 'json' });
        if (response.data.ok) {
          Logger.error(`[WARN] Rollback succeeded.`);
        }
      }

      // if (!!db) await db.close();
    } finally {
      try {
        await db.close();
      } catch (e) {}
    }
  }
}