mycdc/mycdc.js

require('require-yaml');
require('colors');
const fs = require('fs');
const path = require('path');
const ZongJi = require('./zongji');
const mysql = require('mysql2/promise');
const amqp = require('amqplib');

const allEvents = [
  'writerows',
  'updaterows',
  'deleterows'
];

module.exports = class MyCDC {
  constructor() {
    this.running = false;
    this.filename = null;
    this.position = null;
    this.schemaMap = new Map();
    this.queues = {};
  }

  async start() {
    const defaultConfig = require('./config.yml');
    const config = this.config = Object.assign({}, defaultConfig);
    const localPath = path.join(__dirname, 'config.local.yml');
    if (fs.existsSync(localPath)) {
      const localConfig = require(localPath);
      Object.assign(config, localConfig);
    }

    const queues = config.queues;
    for (const queueName in queues) {
      const includeSchema = queues[queueName].includeSchema;
      for (const schemaName in includeSchema) {
        let tableMap = this.schemaMap.get(schemaName);
        if (!tableMap) {
          tableMap = new Map();
          this.schemaMap.set(schemaName, tableMap);
        }

        const schema = includeSchema[schemaName];
        for (const tableName in schema) {
          const table = schema[tableName];
          //if (typeof table !== 'object') continue;

          let tableInfo = tableMap.get(tableName);
          if (!tableInfo) {
            tableInfo = {
              queues: new Map(),
              events: new Map(),
              columns: new Map(),
              fk: 'id'
            };
            tableMap.set(tableName, tableInfo);
          }
          tableInfo.queues.set(queueName, table);

          const events = table.events || allEvents;
          for (const event of events) {
            let eventInfo = tableInfo.events.get(event);
            if (!eventInfo) {
              eventInfo = [];
              tableInfo.events.set(event, eventInfo);
            }
            eventInfo.push(queueName);
          }

          const columns = table.columns;
          for (const column of columns) {
            let columnInfo = tableInfo.columns.get(column);
            if (!columnInfo) {
              columnInfo = [];
              tableInfo.columns.set(column, columnInfo);
            }
            columnInfo.push(queueName);
          }

          if (table.id)
            tableInfo.id = table.id;
        }

        this.schemaMap.set(schemaName, tableMap);
      }
    }

    const includeSchema = {};
    for (const [schemaName, tableMap] of this.schemaMap)
      includeSchema[schemaName] = Array.from(tableMap.keys());
  
    this.opts = {
      includeEvents: [
        'rotate',
        'tablemap',
        'writerows',
        'updaterows',
        'deleterows'
      ],
      includeSchema
    };

    if (config.testMode)
      console.log('Test mode enabled, just logging queries to console.');

    console.log('Starting process.');
    await this.init();
    console.log('Process started.');
  }

  async stop() {
    console.log('Stopping process.');
    await this.end();
    console.log('Process stopped.');
  }

  async init() {
    const config = this.config;
    this.debug('MyCDC', 'Initializing.');
    this.onErrorListener = err => this.onError(err);

    // DB connection

    this.db = await mysql.createConnection(config.db);
    this.db.on('error', this.onErrorListener);

    // RabbitMQ

    this.publisher = await amqp.connect(config.amqp);
    this.channel = await this.publisher.createChannel();

    for (const queueName in config.queues) {
      await this.channel.assertQueue(queueName, {
        durable: true
      });
    }

    // Zongji

    const zongji = new ZongJi(config.db);
    this.zongji = zongji;

    this.onBinlogListener = evt => this.onBinlog(evt);
    zongji.on('binlog', this.onBinlogListener);

    const [res] = await this.db.query(
      'SELECT `logName`, `position` FROM `binlogQueue` WHERE code = ?',
      [config.code]
    );
    if (res.length) {
      const [row] = res;
      this.filename = row.logName;
      this.position = row.position;
      Object.assign(this.opts, {
        filename: this.filename,
        position: this.position
      });
    } else
      this.opts.startAtEnd = true;

    this.debug('Zongji', 'Starting.');
    await new Promise((resolve, reject) => {
      const onReady = () => {
        zongji.off('error', onError);
        resolve();
      };
      const onError = err => {
        this.zongji = null;
        zongji.off('ready', onReady);
        zongji.off('binlog', this.onBinlogListener);
        reject(err);
      }

      zongji.once('ready', onReady);
      zongji.once('error',  onError);
      zongji.start(this.opts);
    });
    this.debug('Zongji', 'Started.');

    this.zongji.on('error', this.onErrorListener);

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

    // Summary

    this.running = true;
    this.debug('MyCDC', 'Initialized.');
  }

  async end(silent) {
    const zongji = this.zongji;
    if (!zongji) return;

    this.debug('MyCDC', 'Ending.');

    // Zongji

    clearInterval(this.flushInterval);
    clearInterval(this.pingInterval);
    zongji.off('binlog', this.onBinlogListener);
    zongji.off('error', this.onErrorListener);
    this.zongji = null;
    this.running = false;

    this.debug('Zongji', 'Stopping.');
    // FIXME: Cannot call Zongji.stop(), it doesn't wait to end connection
    zongji.connection.destroy(() => {
      console.log('zongji.connection.destroy');
    });
    await new Promise(resolve => {
      zongji.ctrlConnection.query('KILL ' + zongji.connection.threadId,
      err => {
        if (err && !silent)
          console.error(err);
        resolve();
      });
    });
    zongji.ctrlConnection.destroy(() => {
      console.log('zongji.ctrlConnection.destroy');
    });
    zongji.emit('stopped');
    this.debug('Zongji', 'Stopped.');

    // RabbitMQ

    await this.publisher.close();

    // DB connection

    this.db.off('error', this.onErrorListener);
    // FIXME: mysql2/promise bug, db.end() ends process
    this.db.on('error', () => {});
    try {
      await this.db.end();
    } catch (err) {
      if (!silent)
        console.error(err);
    }

    // Summary

    this.debug('MyCDC', 'Ended.');
  }

  async tryRestart() {
    try {
      await this.init();
      console.log('Process restarted.');
    } catch(err) {
      setTimeout(() => this.tryRestart(), 30);
    }
  }

  async onError(err) {
    console.log(`Error: ${err.code}: ${err.message}`);
    try {
      await this.end(true);
    } catch(e) {}

    switch (err.code) {
      case 'PROTOCOL_CONNECTION_LOST':
      case 'ECONNRESET':
      console.log('Trying to restart process.');
      await this.tryRestart();
      break;
    default:
      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;

    const queueNames = tableInfo.events.get(eventName);
    if (!queueNames) return;

    const rows = evt.rows;
    const queues = this.config.queues;
    const tableQueues = tableInfo.queues;

    const changes = new Map();
    for (const queueName of queueNames) {
      const change = {
        mode: queues[queueName].mode
      };
      changes.set(queueName, change);

      switch(change.mode)  {
      case 'fk':
        change.fks = new Set();
        break;
      case 'changes':
        change.rows = [];
        break;
      }
    }

    function addChange(queueNames, row, old) {
      for (const queueName of queueNames) {
        const queueInfo = tableQueues.get(queueName);
        const change = changes.get(queueName);

        const key = row[queueInfo.key];
        const oldKey = old ? old[queueInfo.key] : null;

        switch(change.mode)  {
        case 'fk':
          change.fks.add(key);
          if (old && !equals(oldKey, key))
            change.fks.add(oldKey);
          break;
        case 'changes':
          const queueRow = {};
          for (const column of queueInfo.columns)
            if (row[column] !== undefined)
              queueRow[column] = row[column];
          change.rows.push(queueRow);
          break;
        }
      }
    }

    const columnMap = tableInfo.columns;
    const columns = columnMap.keys();

    if (eventName === 'updaterows') {
      const changedQueues = new Set();
      for (const row of rows) {
        changedQueues.clear();
        const after = row.after;

        for (const col of columns) {
          if (after[col] === undefined
          || equals(after[col], row.before[col]))
            continue;

          for (const queue of columnMap.get(col))
            changedQueues.add(queue);

          if (changedQueues.size === queueNames.length)
            break;
        }

        if (changedQueues.size)
          addChange(changedQueues, after, row.before);
      }
    } else {
      for (const row of rows)
        addChange(queueNames, row);
    }

    for (const [queueName, change] of changes) {
      const jsonData = {
        eventName,
        table: table.tableName,
        schema: table.parentSchema,
        mode: change.mode
      };

      let nChanges;
      switch(change.mode)  {
      case 'fk':
        jsonData.fks = Array.from(change.fks);
        nChanges = change.fks.size;
        break;
      case 'changes':
        jsonData.rows = change.rows;
        nChanges = change.rows.length;
        break;
      }

      if (!nChanges) continue;

      const data = JSON.stringify(jsonData);
      this.channel.sendToQueue(queueName,
        Buffer.from(data), {persistent: true});

      console.debug('Queued:'.blue, `${queueName}:`.yellow, `${table.tableName}(${nChanges}) [${eventName}]`);
    }

    this.position = evt.nextPosition;
    this.flushed = false;
  }

  async flushQueue() {
    if (this.flushed) return;
    this.debug('Flush', `filename: ${this.filename}, position: ${this.position}`);
  
    const replaceQuery =
      'REPLACE INTO `binlogQueue` SET `code` = ?, `logName` = ?, `position` = ?';
    if (!this.config.testMode)
      await this.db.query(replaceQuery, [this.config.code, this.filename, this.position]);

    this.flushed = true;
  }

  async connectionPing() {
    this.debug('Ping', 'Sending ping to database.');

    // FIXME: Should Zongji.connection be pinged?
    await new Promise((resolve, reject) => {
      this.zongji.ctrlConnection.ping(err => {
        if (err) return reject(err);
        resolve();
      });
    })
    await this.db.ping();
  }

  debug(namespace, message) {
    if (this.config.debug)
      console.debug(`${namespace}:`.blue, message.yellow);
  }
}

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