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 }));
  }

  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 (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 (?);';
          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],
            ...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 
        HealthCheckDataEntry.*
      FROM HealthCheckDataEntry
      JOIN HealthCheckConfig ON HealthCheckConfig.ID = HealthCheckDataEntry.ConfigID
      WHERE HealthCheckConfig.ServerID = ?
        AND HealthCheckDataEntry.ConfigID = ?
        AND HealthCheckDataEntry.Timestamp BETWEEN ? AND ?
      ORDER BY Timestamp, ID;
    `,
      [serverID, confID, from.getTime(), to.getTime()]
    );

    const mapByTimestamp = result.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[]>;

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

  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,
      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;
  }
}