Почему при получении данных с помощью async.parallel & async.forEach получаются дублированные значения, записываемые в БД?

Question

Я пытаюсь построить веб-скребок с узлом, в настоящее время мне нужно получить данные истории транзакций для каждого продукта. Я использую библиотеки node-postgres и async.

Это моя схема базы данных:

id [PK](bigint)   |   data (json[])
------------------------------------

Если я запускаю свой код, он делает правильную уникальную запись для каждого идентификатора. Однако данные иногда дублируются для нескольких продуктов, например:

123   |   {date: 2018-11-01, number: 2}
345   |   {date: 2018-11-03, number: 5}
567   |   {date: 2018-11-03, number: 5}
789   |   {date: 2018-11-03, number: 5}
912   |   {date: 2018-11-04, number: 24}
---------------------------------------

Таким образом, продукты 345, 567 и 789 имеют одни и те же данные, но они должны отличаться. Это мой код:

var getTx = async function(product, callback){
  getTransactionsParallel(product.id, daysOfHistory)
  .then(function(transactions) {
    client.query('INSERT INTO tx (id, data) VALUES ($1, $2) ON CONFLICT DO NOTHING;', [product.id, transactions], (err, res) => {
      if(!err){
        logger.info('|| Inserted TX into DB ' + product.id);
        callback();
      } else {
        logger.error('|| Error while inserting product ' + product.id)
        callback();
      }
    })
  })
}

client.query("select * from products", (err, res) => {
  if(!err){
    products = res.rows;
  } else {
    logger.error(err.stack)
  }

  _.each(products, function(product) {
    work.push(function(callback){
      getTx(product, callback)
    })
  })

  async.parallelLimit(work, 10, function(){
    logger.info(' || Finished. || ');
  })
})

А вот функции для получения самих данных:

const getTransactionPage = async function(offset, id, daysOfHistory, callback ){
  try {
      await rp({ uri: url, forever: true, agent: false })
      .then(($) => {
        body = $.replace('result(', '');
        body = body.replace(');', '');
        let data;

        if(data = JSON.parse(body)){
          let transactions = data.records;
          _.each(transactions, function(transaction){
            // push to array
          });
        }
      })
      callback();
  }
};

const getTransactionsParallel = function(id, daysOfHistory){
  return new Promise((resolve, reject) => {
    // async request to fetch maxpages (async function [...] await rp(url) [...])
    getMaxPages(id, function(maxPages){
      let pages = Array.from({length: maxPages}, (v, k) => k+1);

      async.forEach(pages, function(page, callback){
        getTransactionPage(page, id, daysOfHistory, callback)
      }, function done(err, res){
        resolve(dataArray)
      });
    })
  })
}

Итак, что я тут не так сделал?