Kutt/server/migration/04_links.ts

require("dotenv").config();
import { v1 as NEO4J } from "neo4j-driver";
import knex from "knex";
import PQueue from "p-queue";
import { startOfHour } from "date-fns";

const queue = new PQueue({ concurrency: 1 });

// 1. Connect to Neo4j database
const neo4j = NEO4J.driver(
  process.env.NEO4J_DB_URI,
  NEO4J.auth.basic(process.env.NEO4J_DB_USERNAME, process.env.NEO4J_DB_PASSWORD)
);
// 2. Connect to Postgres database
const postgres = knex({
  client: "postgres",
  connection: {
    host: process.env.DB_HOST,
    database: process.env.DB_NAME,
    user: process.env.DB_USER,
    password: process.env.DB_PASSWORD
  }
});

(async function() {
  const startTime = Date.now();

  // 3. [NEO4J] Get all links
  const session = neo4j.session();
  session
    .run(
      "MATCH (l:URL) " +
        "OPTIONAL MATCH (l)-[:USES]->(d) " +
        "OPTIONAL MATCH (l)<-[:CREATED]-(u) " +
        "OPTIONAL MATCH (v)-[:VISITED]->(l) " +
        "OPTIONAL MATCH (v)-[:BROWSED_BY]->(b) " +
        "OPTIONAL MATCH (v)-[:OS]->(o) " +
        "OPTIONAL MATCH (v)-[:LOCATED_IN]->(c) " +
        "OPTIONAL MATCH (v)-[:REFERRED_BY]->(r) " +
        "OPTIONAL MATCH (v)-[:VISITED_IN]->(dd) " +
        "WITH l, u, d, COLLECT([b.browser, o.os, c.country, r.referrer, dd.date]) as stats " +
        "LIMIT 100 RETURN l, u.email as email, d.name as domain, stats"
    )
    .subscribe({
      onNext(record) {
        queue.add(async () => {
          const link = record.get("l").properties;
          const email = record.get("email");
          const address = record.get("domain");
          const stats = record.get("stats");

          // 4. Merge and normalize stats based on hour
          const visits: Record<
            string,
            Record<string, number | Record<string, number>>
          > = {} as any;

          stats.forEach(([b, o, country, referrer, date]) => {
            const dateHour = startOfHour(date).toISOString();
            const browser = b.toLowerCase();
            const os = o === "Mac Os X" ? "macos" : o.toLowerCase();
            visits[dateHour] = {
              ...visits[dateHour],
              total:
                (((visits[dateHour] && visits[dateHour].total) as number) ||
                  0) + 1,
              [`br_${browser}`]:
                (((visits[dateHour] &&
                  visits[dateHour][`br_${browser}`]) as number) || 0) + 1,
              [`os_${os}`]:
                (((visits[dateHour] &&
                  visits[dateHour][`os_${os}`]) as number) || 0) + 1,
              countries: {
                ...((visits[dateHour] || {}).countries as {}),
                [country.toLowerCase()]:
                  ((visits[dateHour] &&
                    visits[dateHour].countries[country.toLowerCase()]) ||
                    0) + 1
              },
              referrers: {
                ...((visits[dateHour] || {}).referrers as {}),
                [referrer.toLowerCase()]:
                  ((visits[dateHour] &&
                    visits[dateHour].countries[referrer.toLowerCase()]) ||
                    0) + 1
              }
            };
          });

          // 5. [Postgres] Find matching user and or domain
          const [user, domain] = await Promise.all([
            email &&
              postgres<User>("users")
                .where({ email })
                .first(),
            address &&
              postgres<Domain>("domains")
                .where({ address })
                .first()
          ]);

          // 6. [Postgres] Create link
          const data = {
            address: link.id,
            banned: !!link.banned,
            domain_id: domain ? domain.id : null,
            password: link.password,
            target: link.target,
            user_id: user ? user.id : null,
            ...(link.count && { visit_count: link.count.toNumber() }),
            ...(link.createdAt && { created_at: link.createdAt })
          };

          const exists = await postgres<Link>("links")
            .where({ address: link.id })
            .first();

          let link_id: number;
          if (exists) {
            const res = await postgres<Link>("links")
              .where("id", exists.id)
              .update(data, "id");
            link_id = res[0];
          } else {
            const res = await postgres<Link>("links").insert(data, "id");
            link_id = res[0];
          }

          // 7. [Postgres] Create visits
          for await (const [date, details] of Object.entries(visits)) {
            const data = {
              link_id,
              created_at: date,
              countries: details.countries as Record<string, number>,
              referrers: details.referrers as Record<string, number>,
              total: details.total as number,
              br_chrome: details.br_chrome as number,
              br_edge: details.br_edge as number,
              br_firefox: details.br_firefox as number,
              br_ie: details.br_ie as number,
              br_opera: details.br_opera as number,
              br_other: details.br_other as number,
              br_safari: details.br_safari as number,
              os_android: details.os_android as number,
              os_ios: details.os_ios as number,
              os_linux: details.os_linux as number,
              os_macos: details.os_macos as number,
              os_other: details.os_other as number,
              os_windows: details.os_windows as number
            };

            const visitExists = await postgres<Visit>("visits")
              .where({ link_id, created_at: data.created_at })
              .first();

            if (visitExists) {
              await postgres<Visit>("visits")
                .where("id", visitExists.id)
                .update(data);
            } else {
              await postgres<Visit>("visits").insert(data);
            }
          }
        });
      },
      onCompleted() {
        session.close();
        queue.add(() => {
          const endTime = Date.now();
          console.log(
            `✅ Done! It took ${(endTime - startTime) / 1000} seconds.`
          );
        });
      },
      onError(error) {
        console.log(error);
        session.close();
        throw error;
      }
    });
})();