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({ ...bill, 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.");
    }
    return res.status(200).json({ message: "OpenSearch index updated successfully." });
  } catch (error) {
    console.log(error);
    return res.status(500).json({ message: "Error updating OpenSearch index.", error: error.message });
  }
}

OpenSearchUpdateHandler();