bodyshop/os-loader.js

const Dinero = require("dinero.js");

//const client = require("../graphql-client/graphql-client").client;
const _ = require("lodash");
const GraphQLClient = require("graphql-request").GraphQLClient;
const logger = require("./server/utils/logger");

const path = require("path");
const client = require("./server/graphql-client/graphql-client").client;
require("dotenv").config({
  path: path.resolve(
    process.cwd(),
    `.env.${process.env.NODE_ENV || "development"}`
  ),
});
const { Client, Connection } = require("@opensearch-project/opensearch");
const { defaultProvider } = require("@aws-sdk/credential-provider-node");
const aws4 = require("aws4");
const { gql } = require("graphql-request");
const gqlclient = require("./server/graphql-client/graphql-client").client;
// const osClient = new Client({
//   node: `https://imex:Wl0d8k@!@search-imexonline-search-ixp2stfvwp6qocjsowzjzyreoy.ca-central-1.es.amazonaws.com/`,
// });

var host = process.env.OPEN_SEARCH_HOST; // e.g. https://my-domain.region.es.amazonaws.com
const createAwsConnector = (credentials, region) => {
  class AmazonConnection extends Connection {
    buildRequestObject(params) {
      const request = super.buildRequestObject(params);
      request.service = "es";
      request.region = region;
      request.headers = request.headers || {};
      request.headers["host"] = request.hostname;

      return aws4.sign(request, credentials);
    }
  }
  return {
    Connection: AmazonConnection,
  };
};

const getClient = async () => {
  const credentials = await defaultProvider()();
  return new Client({
    ...createAwsConnector(credentials, "ca-central-1"),
    node: host,
  });
};

async function OpenSearchUpdateHandler(req, res) {
  try {
    var osClient = await getClient();
    // const osClient = new Client({
    //   node: `https://imex:password@search-imexonline-search-ixp2stfvwp6qocjsowzjzyreoy.ca-central-1.es.amazonaws.com`,
    // });

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

    // return;

    var batchSize = 1000;
    var promiseQueue = [];

    //Jobs Load.
    const jobsData = await gqlclient.request(`query{jobs{
          id
          bodyshopid:shopid
          clm_no
          clm_total
          comment
          ins_co_nm
          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();