mycdc/db-async.js

const ZongJi = require('./zongji');
const mysql = require('mysql2/promise');

const allEvents = new Set([
    'writerows',
    'updaterows',
    'deleterows'
  ]);
const fks = new Set();

module.exports = class DbAsync {
    constructor(config) {
      this.config = config;
      this.running = false;
      this.filename = null;
      this.position = null;
      this.schemaMap = new Map();
    }
  
    async start() {
      if (this.config.testMode)
        console.debug('Test mode enabled, just logging queries to console.');
  
      console.log('Starting process.');

      const db = await mysql.createConnection(this.config.db);
      this.db = db;
  
      const includeSchema = {};
      for (const schemaName in this.config.includeSchema) {
        const schema = this.config.includeSchema[schemaName];
        const tables = [];
        const tableMap = new Map();
    
        for (const tableName in schema) {
          const table = schema[tableName];
          tables.push(tableName);
    
          const tableInfo = {
            events: allEvents,
            columns: true,
            fk: 'id'
          };
          tableMap.set(tableName, tableInfo);
      
          if (typeof table === 'object') {
            if (Array.isArray(table.events))
              tableInfo.events = new Set(table.events);
            if (Array.isArray(table.columns))
              tableInfo.columns = new Set(table.columns);
            if (table.fk)
              tableInfo.fk = table.fk;
          }
        }
    
        includeSchema[schemaName] = tables;
        this.schemaMap.set(schemaName, tableMap);
      }
    
      const opts = {
        includeEvents: this.config.includeEvents,
        includeSchema
      };
      this.opts = opts;
    
      const [res] = await this.db.query(
        'SELECT `logName`, `position` FROM `binlogQueue` WHERE code = ?',
        [this.config.queue]
      );
      if (res.length) {
        const [row] = res;
        this.filename = row.logName;
        this.position = row.position;
        Object.assign(opts, {
          filename: this.filename,
          position: this.position
        });
      } else
        opts.startAtEnd = true;

      await this.startZongji();
    }

    async stop() {
      await this.stopZongji();
      await this.db.end();
    }
  
    async startZongji() {
      const zongji = new ZongJi(this.config.db);
      this.zongji = zongji;

      zongji.on('ready', () => this.onReady());
      zongji.on('stopped', () => this.onStopped());
      zongji.on('error', err => this.onError(err));
      zongji.on('binlog', evt => this.onBinlog(evt));
      zongji.start(this.opts);
    }
  
    async stopZongji() {
      console.debug('Stopping Zongji.');
      this.running = false;
      clearInterval(this.flushInterval);
      clearInterval(this.pingInterval);
      this.zongji.stop();
    }

    async restartZongji() {
      console.debug('Restaring Zongji.');
      await this.stopZongji();
      setTimeout(() => this.startZongji(this.opts), 1000);
    }
  
    onReady() {
      this.running = true;

      this.flushInterval = setInterval(
        () => this.flushQueue(), this.config.flushInterval);
      this.pingInterval = setInterval(
        () => this.connectionPing(), this.config.pingInterval * 1000);

      console.debug('Zongji ready.');
    }
  
    onStopped() {
      console.debug('Zongji stopped.');
    }

    async onError(err) {
      console.log(`Error: ${err.code}: ${err.message}`);
      switch (err.code) {
        case 'PROTOCOL_CONNECTION_LOST':
        case 'ECONNRESET':
        await this.restartZongji();
        break;
      default:
        await this.stop();
        process.exit();
      }
    }
  
    onBinlog(evt) {
      //evt.dump();
      const eventName = evt.getEventName();
      if (eventName === 'tablemap') return;
  
      if (eventName === 'rotate') {
        this.filename = evt.binlogName;
        this.position = evt.position;
        console.log(`[${eventName}] filename: ${this.filename}`, `position: ${this.position}`);
        return;
      }
  
      const table = evt.tableMap[evt.tableId];
      const tableMap = this.schemaMap.get(table.parentSchema);
      if (!tableMap) return;
  
      const tableInfo = tableMap.get(table.tableName);
      if (!tableInfo) return;
  
      if (!tableInfo.events.has(eventName)) return;
  
      let column;
      const rows = evt.rows;
  
      if (eventName === 'updaterows') {
        if (tableInfo.columns !== true) {
          let changes = false;
          for (const row of rows) {
            const after = row.after;
            for (const col in after) {
              if (tableInfo.columns.has(col) && !equals(after[col], row.before[col])) {
                fks.add(after[tableInfo.fk]);
                changes = true;
                if (!column) column = col;
                break;
              }
            }
          }
          if (!changes) return;
        } else {
          for (const row of rows)
            fks.add(row.after[tableInfo.fk]);
        }
      } else {
        for (const row of rows)
          fks.add(row[tableInfo.fk]);
      }
  
      const row = eventName === 'updaterows'
        ? rows[0].after
        : rows[0];
  
      if (this.config.debug) {
        console.debug(`[${eventName}] ${table.tableName}: ${rows.length}`);
        console.debug(`  ${tableInfo.fk}: ${row[tableInfo.fk]}`);
        if (column) {
          let before = formatValue(rows[0].before[column]);
          let after = formatValue(rows[0].after[column]);
          console.debug(`  ${column}: ${before} <- ${after}`);
        }
      }
  
      this.position = evt.nextPosition;
    }
  
    async flushQueue() {
      if (!this.running) return;
      console.log('==========================================================');
      console.log('Flush:', `filename: ${this.filename}`, `position: ${this.position}`);
      console.log(fks);
      if (!fks.size) return;
    
      const ids = [];
      for (const fk of fks) ids.push([fk]);
    
      const replaceQuery =
        'REPLACE INTO `binlogQueue`  SET `code` = ?, `logName` = ?, `position` = ?';
    
      if (this.config.testMode) {
        console.debug(this.config.addQuery);
        console.debug(replaceQuery);
      } else {
        await this.db.query(this.config.addQuery, [ids]);
        await this.db.query(replaceQuery, [this.config.queue, this.filename, this.position]);
      }
      fks.clear();
      console.log('==========================================================');
    }
  
    async connectionPing() {
      if (!this.running) return;
      if (this.config.debug)
        console.debug('Sending ping to database.')
      this.zongji.connection.ping();
      this.zongji.ctrlConnection.ping();
      await this.db.ping();
    }
  }
  
  function equals(a, b) {
    if (a === b)
      return true;
    const type = typeof a;
    if (a == null || b == null || type !== typeof b)
      return false;
    if (type === 'object' && a.constructor === b.constructor) {
      if (a instanceof Date)
        return a.getTime() === b.getTime();
    }
    return false;
  }
  
  function formatValue(value) {
    if (value instanceof Date)
      return value.toJSON();
    return value;
  }