traxxx/src/utils/upsert.js

58 lines
1.7 KiB
JavaScript
Raw Normal View History

'use strict';
const knex = require('../knex');
const logger = require('../logger')(__filename);
async function upsert(table, items, identifier = ['id'], _knex) {
const identifiers = Array.isArray(identifier) ? identifier : [identifier];
const duplicates = await knex(table).whereIn(identifiers, items.map((item) => identifiers.map((identifierX) => item[identifierX])));
const duplicatesByIdentifiers = duplicates.reduce((acc, duplicate) => {
const duplicateIdentifier = identifiers.map((identifierX) => duplicate[identifierX]).toString();
return { ...acc, [duplicateIdentifier]: duplicate };
}, {});
const { insert, update } = items.reduce((acc, item) => {
const itemIdentifier = identifiers.map((identifierX) => item[identifierX]).toString();
if (duplicatesByIdentifiers[itemIdentifier]) {
acc.update.push(item);
return acc;
}
acc.insert.push(item);
return acc;
}, {
insert: [],
update: [],
});
if (knex) {
logger.debug(`${table}: Inserting ${insert.length}`);
logger.debug(`${table}: Updating ${update.length}`);
const [inserted, updated] = await Promise.all([
insert.length > 0 ? knex(table).returning('*').insert(insert) : [],
update.length > 0 ? knex.transaction(async (trx) => Promise.all(update.map((item) => {
const clause = identifiers.reduce((acc, identifierX) => ({ ...acc, [identifierX]: item[identifierX] }), {});
return trx
.where(clause)
.update(item)
.into(table)
.returning('*');
}))) : [],
]);
return {
inserted: Array.isArray(inserted) ? inserted : [],
updated: updated.reduce((acc, updatedItems) => acc.concat(updatedItems), []),
};
}
return { insert, update };
}
module.exports = upsert;