bodyshop/os-loader.js

require("dotenv").config({
    path: require("path").resolve(
        process.cwd(),
        `.env.${process.env.NODE_ENV || "development"}`
    ),
});

const {omit} = require("lodash");
const gqlClient = require("./server/graphql-client/graphql-client").client;
const getClient = require('./libs/awsUtils');

async function OpenSearchUpdateHandler(req, res) {
    try {
        const osClient = await getClient();

        //Clear out all current documents
        // const deleteResult = await osClient.deleteByQuery({
        //   index: ["*"], // ["jobs", "payments", "bills", "vehicles", "owners"],
        //   body: {
        //     query: {
        //       match_all: {},
        //     },
        //   },
        // });

        // return;

        const batchSize = 1000;
        const promiseQueue = [];

        //Jobs Load.
        const jobsData = await gqlClient.request(`query{jobs{
          id
          bodyshopid:shopid
          clm_no
          clm_total
          comment
          ins_co_nm
          owner_owing
          ownr_co_nm
          ownr_fn
          ownr_ln
          ownr_ph1
          ownr_ph2
          plate_no
          ro_number
          status
          v_model_yr
          v_make_desc
          v_model_desc
    }}`);
        for (let i = 0; i <= jobsData.jobs.length / batchSize; i++) {
            const slicedArray = jobsData.jobs.slice(
                i * batchSize,
                i * batchSize + batchSize
            );
            const bulkOperation = [];
            slicedArray.forEach((job) => {
                bulkOperation.push({index: {_index: "jobs", _id: job.id}});
                bulkOperation.push(job);
            });
            promiseQueue.push(bulkOperation);
        }

        //Owner Load
        const ownersData = await gqlClient.request(`{
      owners {
        id
        bodyshopid: shopid
        ownr_fn
        ownr_ln
        ownr_co_nm
        ownr_ph1
        ownr_ph2
      }
    }
    `);
        for (let i = 0; i <= ownersData.owners.length / batchSize; i++) {
            const slicedArray = ownersData.owners.slice(
                i * batchSize,
                i * batchSize + batchSize
            );
            const bulkOperation = [];
            slicedArray.forEach((owner) => {
                bulkOperation.push({index: {_index: "owners", _id: owner.id}});
                bulkOperation.push(owner);
            });
            promiseQueue.push(bulkOperation);
        }

        //Vehicles
        const vehiclesData = await gqlClient.request(`{
      vehicles {
        id
        bodyshopid: shopid
        plate_no
        v_model_yr
        v_model_desc
        v_make_desc
        v_color
        v_vin
      }
    }
    `);
        for (let i = 0; i <= vehiclesData.vehicles.length / batchSize; i++) {
            const slicedArray = vehiclesData.vehicles.slice(
                i * batchSize,
                i * batchSize + batchSize
            );
            const bulkOperation = [];
            slicedArray.forEach((vehicle) => {
                bulkOperation.push({index: {_index: "vehicles", _id: vehicle.id}});
                bulkOperation.push(vehicle);
            });
            promiseQueue.push(bulkOperation);
        }

        //payments
        const paymentsData = await gqlClient.request(`{
      payments {
        id
        amount
        created_at
        date
        exportedat
        memo
        payer
        paymentnum
        transactionid
        type
        job {
          id
          ownerid
          ownr_co_nm
          ownr_fn
          ownr_ln
          owner {
            id
            ownr_co_nm
            ownr_fn
            ownr_ln
          }
          ro_number
          bodyshopid: shopid
        }
      }
    }
    
    `);
        for (let i = 0; i <= paymentsData.payments.length / batchSize; i++) {
            const slicedArray = paymentsData.payments.slice(
                i * batchSize,
                i * batchSize + batchSize
            );
            const bulkOperation = [];
            slicedArray.forEach((payment) => {
                bulkOperation.push({index: {_index: "payments", _id: payment.id}});
                bulkOperation.push({
                    ...omit(payment, ["job"]),
                    bodyshopid: payment.job.bodyshopid,
                });
            });
            promiseQueue.push(bulkOperation);
        }

        //bills
        const billsData = await gqlClient.request(`{
      bills {
        id
        date
        exported
        exported_at
        invoice_number
        is_credit_memo
        total
        vendor {
          name
          id
        }
        job {
          ro_number
          id
          bodyshopid: shopid
        }
      }
    }`);
        for (let i = 0; i <= billsData.bills.length / batchSize; i++) {
            const slicedArray = billsData.bills.slice(
                i * batchSize,
                i * batchSize + batchSize
            );
            const bulkOperation = [];
            slicedArray.forEach((bill) => {
                bulkOperation.push({index: {_index: "bills", _id: bill.id}});
                bulkOperation.push({
                    ...omit(bill, ["job"]),
                    bodyshopid: bill.job.bodyshopid,
                });
            });
            promiseQueue.push(bulkOperation);
        }

        //Load the entire queue.
        for (const queueItem of promiseQueue) {
            const insertJobsBulk = await osClient.bulk({body: queueItem});

            console.log(
                ` ${insertJobsBulk.body.items.length} Records inserted in ${insertJobsBulk.body.took}.`
            );
            if (insertJobsBulk.body.errors)
                console.error("*** Error while inserting.");
        }
    } catch (error) {
        console.log(error);
    }
}

OpenSearchUpdateHandler();