Nodejs (Express) script to connect to Firestore and push clickstream data via endpoints from client

Firestore Collection

const express = require("express");
const cors = require("cors");
const { Firestore } = require("@google-cloud/firestore");
const { PubSub } = require("@google-cloud/pubsub");
const firebase = require("firebase");
const bodyParser = require("body-parser");
const productsJson = require("./products.json");

const app = express();
app.use(cors());
app.use(bodyParser.json());

process.env["GOOGLE_APPLICATION_CREDENTIALS"] = "./firestore-credentials.json";

const firestore = new Firestore();
const projectId = "kkrish-corp-ce-demos";
const pubsub_client = new PubSub({ projectId });
let pubsub_topic = null;
let products_collection = firestore.collection("products");
let clickstream_collection = firestore.collection("clickstream");
let categories_collection = firestore.collection("categories");
let cart_collection = firestore.collection("cart");

// /**************** Allow CORS headers ***************/

// app.all("/*", function (req, res, next) {
//   res.header("Access-Control-Allow-Origin", "*");
//   res.header("Access-Control-Allow-Headers", "X-Requested-With");
//   res.header("Access-Control-Allow-Methods", "GET", "PUT", "POST, OPTIONS");
//   next();
// });

/**************** Firebase Auth Initialization ***************/
var firebaseConfig = {
  apiKey: "AIzaSyCZXwGqgfjRbte1gYTMbPE8z30Y5leZZQQ",
  authDomain: "kkrish-corp-ce-demos.firebaseapp.com",
  projectId: "kkrish-corp-ce-demos",
  storageBucket: "kkrish-corp-ce-demos.appspot.com",
  messagingSenderId: "1085876812319",
  appId: "1:1085876812319:web:03bdc3b4f0306572d6b32a",
  measurementId: "G-C4GPJN87ZF",
};
firebase.initializeApp(firebaseConfig);

/************** User Session/Management **************/
app.post("/signup", async (req, res) => {
  firebase
    .auth()
    .createUserWithEmailAndPassword(req.body.email, req.body.password)
    .then((userCredential) => {
      // Signed in
      var user = userCredential.user;
      res.status(200).send({
        status: "success",
        email: user.email,
        uid: user.uid,
        token: user.refreshToken,
      });
    })
    .catch((error) => {
      var errorCode = error.code;
      var errorMessage = error.message;
      console.log(errorCode);
      console.log(errorMessage);
      res
        .status(404)
        .send({ status: "failed", code: errorCode, message: errorMessage });
    });
});

app.post("/login", async (req, res) => {
  firebase
    .auth()
    .signInWithEmailAndPassword(req.body.email, req.body.password)
    .then((userCredential) => {
      // Signed in
      var user = userCredential.user;
      res.status(200).send({
        status: "success",
        email: user.email,
        uid: user.uid,
        token: user.refreshToken,
      });
    })
    .catch((error) => {
      var errorCode = error.code;
      var errorMessage = error.message;
      console.log(errorCode);
      console.log(errorMessage);
      res
        .status(404)
        .send({ status: "failed", code: errorCode, message: errorMessage });
    });
});

app.get("/logout/:id", async (req, res) => {
  let response = firebase.auth().signOut();
  console.log(`user signed out successfully : ${response}`);
  if (response) res.status(200).send({ status: "success" });
  else res.status(200).send({ status: "failed" });
});

/************** Initialize PubSub Connection **************/

async function InitPubSub() {
  console.log("Pub Sub initialization...");
  const [topics] = await pubsub_client.getTopics();

  topics.forEach(async (topic) => {
    try {
      if (topic.name == `projects/${projectId}/topics/clickstream`) {
        pubsub_topic = topic;
        console.log(topic.name);
        return;
      } else {
        pubsub_topic = await pubsub_client.createTopic("clickstream");
      }
    } catch (error) {
      console.log("Pub-Sub " + error);
    }
  });
}

InitPubSub();
/************** Category Endpoints **************/
app.post("/category", async (req, res) => {
  let result = await categories_collection.add(req.body);
  if (result != null) res.status(200).send({ id: result.id });
});

app.get("/categories", async (req, res) => {
  let categories = [];
  let categories_doc = await categories_collection.listDocuments();
  for (const c of categories_doc) {
    let category = await c.get();
    categories.push({ id: c.id, data: category.data() });
  }
  res.status(200).send(categories);
});

app.get("/products/filter/:category", async (req, res) => {
  let products_doc = await products_collection
    .where("category", "==", req.params.category)
    .get();
  let products = [];
  products_doc.forEach((p) => {
    products.push({ id: p.id, data: p.data() });
  });
  res.status(200).send(products);
});

/************** Product Endpoints **************/

app.post("/product", async (req, res) => {
  let result = await products_collection.add(req.body);
  if (result != null) res.status(200).send({ id: result.id });
});

app.get("/products", async (req, res) => {
  let products = [];
  let products_doc = await products_collection.listDocuments();
  for (const p of products_doc) {
    let product = await p.get();
    products.push({ id: p.id, data: product.data() });
  }
  res.status(200).send(products);
});

app.get("/product/:id", async (req, res) => {
  let docRef = products_collection.doc(req.params.id);
  let product = await docRef.get();
  res.status(200).send({ data: product.data() });
});

/************** Cart Endpoints **************/
app.post("/cart/:id", async (req, res) => {
  const batch = firestore.batch();
  req.body.forEach(async (item) => {
    let payload = {
      product_data: item.data,
      product_id: item.id,
      user_id: req.params.id,
    };
    var docRefId = cart_collection.doc().id;
    batch.set(cart_collection.doc(docRefId), payload);
  });
  let committedTrans = await batch.commit();
  if (committedTrans != null) res.status(200).send({ status: "success" });
  else res.status(400).send({ status: "failed" });
});

app.get("/cart/:id", async (req, res) => {
  let cartItems = [];
  let cart_doc = await cart_collection
    .where("user_id", "==", req.params.id)
    .get();
  cart_doc.forEach((d) => {
    let cartItem = d.data();
    cartItems.push(cartItem);
  });
  res.status(200).send(cartItems);
});

/************** Clickstream Endpoints **************/

app.post("/clickstream", async (req, res) => {
  let result = await clickstream_collection.add(req.body);
  pubsub_topic.publish(Buffer.from(JSON.stringify(req.body)));
  if (result != null) res.status(200).send({ id: result.id });
});

/************** Dummy Connection Test **************/

app.post("/local-to-remote", async (req, res) => {
  let products = productsJson;
  products.forEach(async (p) => {
    let result = await products_collection.add(p);
    console.log(`Posted product#${p.id} return result #${result.id}`);
  });
  res.status(200).send({ status: "ok" });
});

app.get("*", async (req, res) => {
  res.status(200).send("Cannot GET/");
});

const PORT = process.env.PORT || 8080;
app.listen(PORT, () => {
  console.log(
    `The REST API started successfully and is listening for HTTP requests on ${PORT}`
  );
});