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

import fs from 'fs';
import fsp from 'fs/promises';
import moment from 'moment';
import path from 'path';
import { Database as SQLiteDB, OPEN_CREATE, OPEN_READWRITE } from 'sqlite3';

import defaults from '../../../common/defaults.module';
import { ServiceConfig, validateParamType } from '../../../common/interfaces/service-config.interface';
import { HttpCheckData, HttpCheckStatus, ServiceCheckData, ServiceCheckDataEntry } from '../../../common/lib/http-check-data.module';
import { Logger } from '../../../common/util/logger.class';

import { ValidationException } from '../lib/validation-exception.class';
import { DBMigration } from './db-migration.class';
import { SQLiteController } from './sqlite-controller.base';

export enum ServiceChangedStatus {
  None,
  Created,
  Activated,
  Deactivated,
  Rescheduled
}

export class Database extends SQLiteController {
  public set onError(listener: (error: any) => void) {
    this._onError = listener;
  }

  private _onError: (error: any) => void = err => console.error('[DB.ONERROR]', err);

  public async open(migrate = false) {
    try {
      const DATA_DIR = process.env.DATA_DIR || 'data';

      if (!fs.existsSync(DATA_DIR)) await fsp.mkdir(DATA_DIR);

      const DATA_FILE = path.resolve(DATA_DIR, 'data.db');
      const exists = fs.existsSync(DATA_FILE);

      await new Promise<void>((res, rej) => {
        this.db = new SQLiteDB(DATA_FILE, OPEN_READWRITE | OPEN_CREATE, err => (err ? rej(err) : res()));
        this.db.on('error', this._onError);
      });
      Logger.info('[INFO]', exists ? 'Opened' : 'Created', 'SQLite3 Database file', DATA_FILE);

      if (!this.db) throw new Error('Database not opened.');

      if (!exists) {
        // INITIAL TABLE SETUP
        await this.run(
          `CREATE TABLE Server (
            ID INTEGER PRIMARY KEY AUTOINCREMENT,
            Title TEXT NOT NULL UNIQUE,
            FQDN TEXT NOT NULL UNIQUE
           );`,
          []
        );

        await this.run(
          `CREATE TABLE ServerConfig (
            ID INTEGER PRIMARY KEY AUTOINCREMENT,
            ServerID INTEGER NOT NULL,
            Key TEXT NOT NULL,
            Value TEXT NOT NULL,
            FOREIGN KEY(ServerID) REFERENCES Server(ID),
            UNIQUE(ServerID, Key)
          )`,
          []
        );

        await this.run(
          `CREATE TABLE ServerDataEntry (
            ID INTEGER PRIMARY KEY AUTOINCREMENT,
            ServerID INTEGER NOT NULL,
            Timestamp INTEGER NOT NULL,
            FOREIGN KEY(ServerID) REFERENCES Server(ID),
            UNIQUE(ServerID, Timestamp)
           );`,
          []
        );

        await this.run(
          `CREATE TABLE ServerDataValue (
            ID INTEGER PRIMARY KEY AUTOINCREMENT,
            EntryID INTEGER NOT NULL,
            Type Text NOT NULL,
            Key TEXT NOT NULL,
            Value REAL NOT NULL,
            FOREIGN KEY(EntryID) REFERENCES ServerDataEntry(ID),
            UNIQUE(EntryID, Type, Key)
          );`,
          []
        );

        let result = await this.run(`INSERT INTO Server(Title, FQDN) VALUES(?, ?);`, ['Raspi4', '10.8.0.10']);
        const serverID = result.lastID;
        Logger.debug(`[DEBUG] Created Server #${serverID}`);
        result = await this.run(`INSERT INTO ServerConfig(ServerID, Key, Value) VALUES(?, ?, ?);`, [serverID, 'syncInterval', 300]);
      }

      if (migrate) {
        // RUN DB MIGRATIONS
        const mig = new DBMigration(this.db);
        await mig.update();
      }

      // DB RUNTIME SETTINGS
      await this.exec('PRAGMA foreign_keys=on;');
    } catch (err) {
      Logger.error('[FATAL] Initializing Database failed:', err);
      Logger.error('[EXITING]');
      process.exit(1);
    }
  }

  public async getAllServerConfigs(): Promise<Server[]> {
    const res = await this.stmt(
      `SELECT 
        Server.*,
        ServerConfig.Key,
        ServerConfig.Value
        FROM Server
        LEFT OUTER JOIN ServerConfig ON Server.ID = ServerConfig.ServerID
        ORDER BY Server.Title, ServerConfig.Key`,
      []
    );

    return res.rows.reduce((res: Server[], line, i) => {
      const serverID = line['ID'];
      let server: Server;
      if (i === 0 || res[res.length - 1].id !== serverID) {
        server = { id: serverID, title: line['Title'], fqdn: line['FQDN'], config: {} };
        res.push(server);
      } else {
        server = res[res.length - 1];
      }

      if (!!line['Key']) {
        server.config[line['Key']] = line['Value'];
      }

      return res;
    }, [] as Server[]);
  }

  public async insertServerData(serverID: number, data: ReducedData[]) {
    if (!data.length) return;

    await this.beginTransaction();
    try {
      for (const entry of data) {
        const result = await this.run('INSERT INTO ServerDataEntry(ServerID, Timestamp) VALUES(?, ?);', [serverID, entry.time.getTime()]);
        let entryID = result.lastID;

        for (const type of Object.keys(entry).filter(t => !['time', 'hdd'].includes(t))) {
          for (const key of Object.keys((entry as any)[type])) {
            await this.run('INSERT INTO ServerDataValue(EntryID, Type, Key, Value) VALUES(?, ?, ?, ?);', [
              entryID,
              type,
              key,
              (entry as any)[type][key]
            ]);
          }
        }

        if (entry.hdd) {
          for (const mount of Object.keys(entry.hdd)) {
            for (const key of Object.keys(entry.hdd[mount])) {
              await this.run('INSERT INTO ServerDataValue(EntryID, Type, Key, Value) VALUES(?, ?, ?, ?);', [
                entryID,
                `hdd:${mount}`,
                key,
                (entry.hdd[mount] as any)[key]
              ]);
            }
          }
        }
      }
      await this.commit();
    } catch (err) {
      await this.rollback();
      throw err;
    }
  }

  public async getServerDataTypes(serverID: number) {
    const results = await this.stmt(
      `
      SELECT
        ServerDataValue.Type 
      FROM ServerDataEntry
      JOIN ServerDataValue ON ServerDataEntry.ID = ServerDataValue.EntryID
      WHERE ServerDataEntry.ServerID = ?
      GROUP BY ServerDataValue.Type
      ORDER BY ServerDataValue.Type;
    `,
      [serverID]
    );

    return results.rows.reduce((res: Array<ServerDataTypesConfig>, { Type: type }) => {
      if (!type.startsWith('hdd:')) {
        res.push({ type });
      } else {
        let hdd = res.find(c => c.type === 'hdd');
        if (!hdd) {
          hdd = { type: 'hdd', subtypes: [] };
          res.push(hdd);
        }
        hdd.subtypes?.push({ type: type.substring(4) });
      }
      return res;
    }, []) as Array<ServerDataTypesConfig>;
  }

  public async queryServerData(serverID: number, type: ServerDataType, from: Date, to: Date): Promise<ServerData[]> {
    const diffMs = moment(to).diff(moment(from));
    const sectionMs = Math.floor(diffMs / 100);
    const select_max = type !== 'cpu';
    const select_types = select_max ? [type, type, type] : [type, type];
    const result = await this.stmt(
      `
      SELECT 
        CEIL(Timestamp / ?) * ? as 'Timegroup',
        AVG(VALUE_AVG.Value) as 'avg',
        MAX(VALUE_PEAK.Value) as 'peak'${
          select_max
            ? `,
        MAX(VALUE_MAX.Value) as 'max'`
            : ''
        }
      FROM ServerDataEntry
      JOIN ServerDataValue AS VALUE_AVG ON ServerDataEntry.ID = VALUE_AVG.EntryID AND VALUE_AVG.Type = ? AND VALUE_AVG.Key = 'avg'
      JOIN ServerDataValue AS VALUE_PEAK ON ServerDataEntry.ID = VALUE_PEAK.EntryID AND VALUE_PEAK.Type = ? AND VALUE_PEAK.Key = 'peak'
      ${
        select_max
          ? "JOIN ServerDataValue AS VALUE_MAX ON ServerDataEntry.ID = VALUE_MAX.EntryID AND VALUE_MAX.Type = ? AND VALUE_MAX.Key = 'max'"
          : ''
      }
      WHERE ServerDataEntry.ServerID = ?
      AND ServerDataEntry.Timestamp BETWEEN ? AND ?
      GROUP BY Timegroup
      ORDER BY Timegroup;
    `,
      [sectionMs, sectionMs, ...select_types, serverID, from.getTime(), to.getTime()]
    );

    return result.rows.map(r => ({ time: new Date(r.Timegroup), avg: r.avg, peak: r.peak, max: r.max }));
  }

  public async queryServerStats(serverID: number, type: ServerDataType, from: Date, to: Date): Promise<ReducedValuesPerc> {
    const select_max = type !== 'cpu';
    const select_types = select_max ? [type, type, type] : [type, type];
    const result = await this.stmt(
      `
      SELECT 
        AVG(VALUE_AVG.Value) as 'avg',
        AVG(VALUE_PEAK.Value) as 'peak'${
          select_max
            ? `,
        MAX(VALUE_MAX.Value) as 'max'`
            : ''
        }
      FROM ServerDataEntry
      JOIN ServerDataValue AS VALUE_AVG ON ServerDataEntry.ID = VALUE_AVG.EntryID AND VALUE_AVG.Type = ? AND VALUE_AVG.Key = 'avg'
      JOIN ServerDataValue AS VALUE_PEAK ON ServerDataEntry.ID = VALUE_PEAK.EntryID AND VALUE_PEAK.Type = ? AND VALUE_PEAK.Key = 'peak'
      ${
        select_max
          ? "JOIN ServerDataValue AS VALUE_MAX ON ServerDataEntry.ID = VALUE_MAX.EntryID AND VALUE_MAX.Type = ? AND VALUE_MAX.Key = 'max'"
          : ''
      }
      WHERE ServerDataEntry.ServerID = ?
      AND ServerDataEntry.Timestamp BETWEEN ? AND ?;
    `,
      [...select_types, serverID, from.getTime(), to.getTime()]
    );

    const row = result.rows[0];
    if (Object.keys(row).includes('max')) {
      return {
        avg: ((row['avg'] as number) / (row['max'] as number)) * 100,
        peak: ((row['peak'] as number) / (row['peak'] as number)) * 100
      };
    } else {
      return {
        avg: row['avg'] as number,
        peak: row['peak'] as number
      };
    }
  }

  private async getHealthCheckConfigs(serverID?: number, type = 'http') {
    const res = await this.stmt(
      `SELECT 
        HealthCheckConfig.*,
        HealthCheckParams.Type as '_ParamType',
        HealthCheckParams.Key as '_ParamKey',
        HealthCheckParams.Value as '_ParamValue'
        FROM HealthCheckConfig
        LEFT OUTER JOIN HealthCheckParams ON HealthCheckConfig.ID = HealthCheckParams.ConfigID
        WHERE HealthCheckConfig.Type = ?
        ${!!serverID ? 'AND HealthCheckConfig.ServerID = ?' : ''}
        ORDER BY HealthCheckConfig.Title, _ParamType, _ParamKey`,
      [type, serverID]
    );

    return this.configFromResultRows(res.rows);
  }

  public async getHttpCheckConfigs(serverID?: number) {
    return (await this.getHealthCheckConfigs(serverID)).map(this.httpCheckConfigFrom);
  }

  private async getHealthCheckConfigByID(serverID: number, configID: number) {
    if (!serverID && !configID) return null;

    const res = await this.stmt(
      `SELECT 
        HealthCheckConfig.*,
        HealthCheckParams.Type as '_ParamType',
        HealthCheckParams.Key as '_ParamKey',
        HealthCheckParams.Value as '_ParamValue'
        FROM HealthCheckConfig
        LEFT OUTER JOIN HealthCheckParams ON HealthCheckConfig.ID = HealthCheckParams.ConfigID
        WHERE HealthCheckConfig.ID = ?
        AND HealthCheckConfig.ServerID = ?
        ORDER BY HealthCheckConfig.Title, _ParamType, _ParamKey`,
      [configID, serverID]
    );

    if (!res.rows.length) return null;

    const configs = this.configFromResultRows(res.rows);

    return configs[0];
  }

  public async getHttpCheckConfigByID(serverID: number, configID: number) {
    return this.httpCheckConfigFrom(await this.getHealthCheckConfigByID(serverID, configID));
  }

  public async saveHttpCheckConfig(serverID: number, conf: HttpCheckConfig) {
    const validationErrors = this.validateHttpCheckConfig(conf);
    if (validationErrors) throw new ValidationException('Validation of HttpCheckConfig object failed', validationErrors);

    conf.serverId = serverID;
    let status = ServiceChangedStatus.None;

    const oldConf = await this.getHttpCheckConfigByID(serverID, conf.id);
    await this.beginTransaction();
    try {
      if (oldConf) {
        // UPDATE
        if (oldConf.title !== conf.title) {
          await this.stmt('UPDATE HealthCheckConfig SET Title = ?', [conf.title]);
        }

        let updValues: any[][] = [];
        if (oldConf.url !== conf.url) updValues.push([conf.url, conf.id, 'url']);
        if (oldConf.interval !== conf.interval) {
          updValues.push([conf.interval, conf.id, 'interval']);
          status = ServiceChangedStatus.Rescheduled;
        }
        if (oldConf.timeout !== conf.timeout) updValues.push([conf.timeout ?? defaults.serviceChecks.httpTimeout, conf.id, 'timeout']);
        if (oldConf.active !== conf.active) {
          updValues.push([conf.active ?? defaults.serviceChecks.active ? 1 : 0, conf.id, 'active']);
          status = conf.active ?? defaults.serviceChecks.active ? ServiceChangedStatus.Activated : ServiceChangedStatus.Deactivated;
        }
        if (oldConf.notify !== conf.notify) updValues.push([conf.notify ?? defaults.serviceChecks.notify ? 1 : 0, conf.id, 'notify']);
        if (oldConf.notifyThreshold !== conf.notifyThreshold)
          updValues.push([conf.notifyThreshold ?? defaults.serviceChecks.notifyThreshold, conf.id, 'notifyThreshold']);
        if (updValues.length) {
          for (const data of updValues) {
            await this.run(`UPDATE HealthCheckParams SET Value = ? WHERE ConfigID = ? AND Key = ?;`, data);
          }
        }

        const res = await this.stmt('SELECT * FROM HealthCheckParams WHERE ConfigID = ? and Key = "check";', [conf.id]);
        updValues = [];
        const delIDs: number[] = [];
        res.rows.forEach((row, i) => {
          if (i < conf.checks.length) {
            updValues.push([conf.checks[i], row['ID']]);
          } else {
            delIDs.push(row['ID']);
          }
        });

        if (delIDs.length) {
          const delSql = `DELETE FROM HealthCheckParams WHERE ID IN (${delIDs.map(() => '?').join(',')});`;
          await this.run(delSql, delIDs);
        }

        if (updValues.length) {
          for (const data of updValues) {
            await this.run('UPDATE HealthCheckParams SET Value = ? WHERE ID = ?;', data);
          }
        }
        const insValues = conf.checks.filter((c, i) => i > res.rows.length - 1).map(c => [conf.id, 'regexp', 'check', c]);
        if (insValues.length) {
          for (const data of insValues) {
            await this.run('INSERT INTO HealthCheckParams(ConfigID, Type, Key, Value) VALUES(?, ?, ?, ?);', data);
          }
        }
      } else {
        // INSERT
        const res = await this.run('INSERT INTO HealthCheckConfig(ServerID, Type, Title) VALUES(?, ?, ?);', [serverID, 'http', conf.title]);
        conf.id = res.lastID;
        if (conf.active ?? defaults.serviceChecks.active) {
          status = ServiceChangedStatus.Created;
        }

        const insCheckValues = conf.checks.map(c => [res.lastID, 'regexp', 'check', c]);
        await this.run(
          `INSERT INTO HealthCheckParams(ConfigID, Type, Key, Value) VALUES
          (?, ?, ?, ?),
          (?, ?, ?, ?),
          (?, ?, ?, ?),
          (?, ?, ?, ?),
          (?, ?, ?, ?),
          (?, ?, ?, ?)${conf.checks.length ? `,${insCheckValues.map(() => '(?, ?, ?, ?)').join(',')}` : ''}`,
          [
            ...[res.lastID, 'text', 'url', conf.url],
            ...[res.lastID, 'boolean', 'active', conf.active ?? defaults.serviceChecks.active ? 1 : 0],
            ...[res.lastID, 'number', 'interval', conf.interval],
            ...[res.lastID, 'number', 'timeout', conf.timeout ?? defaults.serviceChecks.httpTimeout],
            ...[res.lastID, 'boolean', 'notify', conf.notify ?? defaults.serviceChecks.notify],
            ...[res.lastID, 'number', 'notifyThreshold', conf.notifyThreshold ?? defaults.serviceChecks.notifyThreshold],
            ...conf.checks.reduce((ret, check) => [...ret, res.lastID, 'regexp', 'check', check], [] as any[])
          ]
        );
      }

      await this.commit();
      return { status, result: conf };
    } catch (err) {
      await this.rollback();
      throw err;
    }
  }

  async deleteHealthCheckConfig(serverID: number, confID: number) {
    const conf = await this.getHealthCheckConfigByID(serverID, confID);
    if (!conf) return false;

    await this.run('DELETE FROM HealthCheckConfig WHERE ID = ?;', [confID]);
    return true;
  }

  async insertHealthCheckData(confID: number, time: Date, status: HttpCheckStatus, message: string) {
    const res = await this.run('INSERT INTO HealthCheckDataEntry(ConfigID, Timestamp, Status, Message) VALUES(?, ?, ?, ?);', [
      confID,
      time.getTime(),
      status,
      message
    ]);

    return {
      id: res.lastID,
      configId: confID,
      time,
      status,
      message
    } as HttpCheckData;
  }

  async queryServiceCheckData(serverID: number, confID: number, from: Date, to: Date) {
    const result = await this.stmt(
      `
      SELECT DataEntryChanges.*
      FROM HealthCheckConfig
      JOIN (
        SELECT * FROM (
          SELECT 
            *
          FROM HealthCheckDataEntry
          WHERE ConfigID = ?
          AND Timestamp BETWEEN ? AND ?
          ORDER BY ID
          LIMIT 0, 1
        ) AS FIRST_STATE

        UNION --+--+--

        SELECT 
          ID,
          ConfigID,
          Timestamp,
          Status,
          Message
        FROM
        (
          SELECT 
            HealthCheckDataEntry.*, 
            LAG(Status) OVER (ORDER BY ConfigID, Timestamp) AS previous_state, 
            LAG(Message) OVER (ORDER BY ConfigID, Timestamp) AS previous_msg
          FROM HealthCheckDataEntry
          WHERE ConfigID = ?
        )
        WHERE Status <> previous_state
        AND Message <> previous_msg
        
        UNION --+--+--
        
        SELECT * FROM (
          SELECT 
            *
          FROM HealthCheckDataEntry
          WHERE ConfigID = ?
          AND Timestamp BETWEEN ? AND ?
          ORDER BY ID DESC
          LIMIT 0, 1
        ) AS LAST_STATE
        ORDER BY ConfigID, Timestamp
      ) AS DataEntryChanges ON DataEntryChanges.ConfigID = HealthCheckConfig.ID
      WHERE HealthCheckConfig.ServerID = ?
        AND DataEntryChanges.Timestamp BETWEEN ? AND ?
      ORDER BY Timestamp, ID;`,
      [confID, from.getTime(), to.getTime(), confID, confID, from.getTime(), to.getTime(), serverID, from.getTime(), to.getTime()]
    );

    const mapByTimestamp = this.mapServiceCheckDataByTimestamp(result.rows);

    const arr: ServiceCheckData[] = [];
    for (const entry of mapByTimestamp.entries()) {
      arr.push({
        time: new Date(entry[0]),
        data: entry[1]
      });
    }
    return arr;
  }

  public async getLastErrors(confID: number, threshold: number) {
    const result = await this.stmt(
      `SELECT * FROM HealthCheckDataEntry
        WHERE ConfigID = ?
        AND Timestamp IN (
          SELECT Timestamp 
          FROM HealthCheckDataEntry
          WHERE ConfigID = ?
          GROUP BY Timestamp
          ORDER BY Timestamp DESC
          LIMIT 0, ?
        )
        ORDER BY Timestamp DESC, ID DESC`,
      [confID, confID, threshold]
    );

    const mapByTimestamp = this.mapServiceCheckDataByTimestamp(result.rows);
    const errors: ServiceCheckData[] = [];
    for (const entry of mapByTimestamp.entries()) {
      const time = entry[0];
      const data = entry[1];

      const errorData = data.filter(d => d.status !== HttpCheckStatus.OK);
      if (!errorData.length) break;

      errors.push({
        time: new Date(time),
        data: errorData
      });
    }

    return errors;
  }

  private mapServiceCheckDataByTimestamp(rows: any[]) {
    return rows.reduce((res: Map<number, ServiceCheckDataEntry[]>, row) => {
      const time: number = row['Timestamp'];
      if (!res.has(time)) res.set(time, []);
      res.get(time)?.push({
        status: row['Status'] as number,
        message: row['Message']
      });
      return res;
    }, new Map()) as Map<number, ServiceCheckDataEntry[]>;
  }

  private configFromResultRows(rows: any[]) {
    return rows.reduce((res: ServiceConfig[], line, i) => {
      const configID = line['ID'];
      let config: ServiceConfig;
      if (i === 0 || res[res.length - 1].id !== configID) {
        config = {
          id: configID,
          title: line['Title'],
          type: line['Type'],
          serverId: line['ServerID'],
          params: []
        };
        res.push(config);
      } else {
        config = res[res.length - 1];
      }

      if (!!line['_ParamKey']) {
        const type = validateParamType(line['_ParamType']);
        const key = line['_ParamKey'];
        if (key === 'check') {
          let checkParam = config.params.find(c => c.key === 'check');
          if (!checkParam) {
            config.params.push(
              (checkParam = {
                key: 'check',
                type: 'regexp',
                value: []
              })
            );
          }
          (checkParam.value as string[]).push(line['_ParamValue']);
        } else {
          config.params.push({
            type,
            key,
            value: type === 'number' ? Number(line['_ParamValue']) : type === 'boolean' ? Boolean(Number(line['_ParamValue'])) : line['_ParamValue']
          });
        }
      }

      return res;
    }, [] as ServiceConfig[]);
  }

  private httpCheckConfigFrom(hcConf: ServiceConfig | null): HttpCheckConfig | null {
    if (!hcConf) return null;
    const params = {
      url: hcConf.params?.find(p => p.key === 'url')?.value as string,
      active: (hcConf.params?.find(p => p.key === 'active')?.value as boolean) ?? defaults.serviceChecks.active,
      interval: hcConf.params?.find(p => p.key === 'interval')?.value as number,
      timeout: (hcConf.params?.find(p => p.key === 'timeout')?.value as number) ?? defaults.serviceChecks.httpTimeout,
      notify: (hcConf.params?.find(p => p.key === 'notify')?.value as boolean) ?? defaults.serviceChecks.notify,
      notifyThreshold: (hcConf.params?.find(p => p.key === 'notifyThreshold')?.value as number) ?? defaults.serviceChecks.notifyThreshold,
      checks: hcConf.params?.reduce((res, p) => (p.key === 'check' && Array.isArray(p.value) ? [...res, ...p.value] : res), [] as string[])
    };
    return {
      id: hcConf.id,
      title: hcConf.title,
      type: hcConf.type,
      serverId: hcConf.serverId,
      ...params
    };
  }

  private validateHttpCheckConfig(conf: Partial<HttpCheckConfig>): { [key: string]: string } | null {
    const errors = {} as any;
    if (!conf) return { null: 'Object was null or undefined' };
    if (!conf.title?.trim().length) errors['required|title'] = `Field 'title' is required.`;
    if (!conf.url?.trim().length) errors['required|url'] = `Field 'url' is required.`;
    if ((!conf.interval && conf.interval !== 0) || Number.isNaN(Number(conf.interval))) errors['required|interval'] = `Field 'interval' is required.`;

    if (!conf.checks || !Array.isArray(conf.checks))
      errors['required|checks'] = `Field 'checks' is required and must be an array of check expressions.`;

    return Object.keys(errors).length ? errors : null;
  }
}