#!/usr/bin/env node
var pkg = require('./package.json');
var Mqtt = require('mqtt');
var Influx = require('influx');
var log = require('yalm');
var config = require('./config.js');
log.setLevel(config.verbosity);
log.info(config);

var mqttConnected;
var influxDBConnected;

var buffer = [];
var bufferCount = 0;

var influx;

log.info(pkg.name + ' ' + pkg.version + ' starting');
log.info('mqtt trying to connect', config.url);
var mqtt = Mqtt.connect(config.url, { will: { topic: config.name + '/connected', payload: '0', retain: true } });

mqtt.on('connect', function () {
  mqttConnected = true;

  log.info('mqtt connected', config.url);
  mqtt.publish(config.name + '/connected', '1', { retain: true });

  log.info('mqtt subscribe', config.name + '/set/#');
  mqtt.subscribe(config.name + '/set/#');

  influx = new Influx.InfluxDB({
    host: config['influxHost'],
    port: config['influxPort'],
    database: config['influxDb'],
    pool: {
      requestTimeout: 5000
    }
  });
  influx.ping(5000).then(hosts => {
    hosts.forEach(host => {
      if (host.online) {
        influxDBConnected = true;
        log.log(`${host.url.host} responded in ${host.rtt}ms running ${host.version})`);
        mqtt.publish(config.name + '/connected', '2', { retain: true });
      } else {
        mqtt.publish(config.name + '/connected', '1', { retain: true });
        log.error(`${host.url.host} is offline :(`);
      }
    });
  });
});

mqtt.on('close', function () {
  if (mqttConnected) {
    mqttConnected = false;
    log.info('mqtt closed ' + config.url);
  }
});

mqtt.on('error', function (err) {
  log.error('mqtt', err);
});

mqtt.on('message', (topic, message) => {
  const topicPrefix = config.name + '/set/';
  if (topic.startsWith(topicPrefix)) {
    const measurement = topic.substring(topicPrefix.length);
    var payload = JSON.parse(message.toString());
    var timestamp = new Date();
    if (payload.ts) { // convert to ms timestamp
      if (payload.ts < 15120042550) {
        payload.ts = payload.ts * 1000;
      }
      timestamp = new Date(payload.ts);
    }
    var point = {
      measurement: measurement,
      tags: payload.tags,
      timestamp: timestamp,
      fields: {}
    };

    if (payload.fields !== undefined) {
      point.fields = payload.fields;
    }
    if (payload.val !== undefined) {
      point.fields.value = payload.val;
    }
    // log.debug(point)
    buffer.push(point);
    bufferCount += 1;
    if (bufferCount >= 1000) writeInflux();
  }
});

function writeInflux () {
  if (!bufferCount) return;
  if (!influxDBConnected) return;
  log.debug('write', bufferCount);
  influx.writePoints(buffer).then(() => {
    log.debug('wrote');
  }).catch(err => {
    log.error(`Error saving data to InfluxDB! ${err.stack}`);
  });
  influx.ping(5000).then(hosts => {
    hosts.forEach(host => {
      if (host.online) {
        influxDBConnected = true;
        log.log(`${host.url.host} responded in ${host.rtt}ms running ${host.version})`);
        mqtt.publish(config.name + '/connected', '2', { retain: true });
      } else {
        mqtt.publish(config.name + '/connected', '1', { retain: true });
        log.error(`${host.url.host} is offline :(`);
      }
    });
  });

  buffer = [];
  bufferCount = 0;
}

setInterval(writeInflux, 60000);

@@ -988,456 +988,8 @@
 %7D);%0A
-  influx.ping(5000).then(hosts =%3E %7B%0A    hosts.forEach(host =%3E %7B%0A      if (host.online) %7B%0A        influxDBConnected = true;%0A        log.log(%60$%7Bhost.url.host%7D responded in $%7Bhost.rtt%7Dms running $%7Bhost.version%7D)%60);%0A        mqtt.publish(config.name + '/connected', '2', %7B retain: true %7D);%0A      %7D else %7B%0A        mqtt.publish(config.name + '/connected', '1', %7B retain: true %7D);%0A        log.error(%60$%7Bhost.url.host%7D is offline :(%60);%0A      %7D%0A    %7D);%0A  %7D);%0A
 %7D);%0A
@@ -2243,16 +2243,54 @@
 rote');%0A
+    buffer = %5B%5D;%0A    bufferCount = 0;%0A
   %7D).cat
@@ -2820,43 +2820,8 @@
  %7D);
-%0A%0A  buffer = %5B%5D;%0A  bufferCount = 0;
 %0A%7D%0A%0A

remove ping after connect