fix: write to database more frequently during manual sync (#20)

Author: ecktoteckto

packages/orb-sync-lib/src/database/postgres.ts

@@ -1,6 +1,7 @@
 import pg from 'pg';
 import { pg as sql } from 'yesql';
 import type { JsonSchema } from '../schemas/types';
+import { randomUUID } from 'node:crypto';
 
 type PostgresConfig = {
   databaseUrl: string;
@@ -25,6 +26,8 @@ export class PostgresClient {
     const chunkSize = 5;
     const results: pg.QueryResult<T>[] = [];
 
+    const timerLoggingLabel = `upsert-many-${randomUUID()}`;
+    console.time(timerLoggingLabel);
     for (let i = 0; i < entries.length; i += chunkSize) {
       const chunk = entries.slice(i, i + chunkSize);
 
@@ -43,6 +46,7 @@ export class PostgresClient {
 
       results.push(...(await Promise.all(queries)));
     }
+    console.timeEnd(timerLoggingLabel);
 
     return results.flatMap((it) => it.rows);
   }

packages/orb-sync-lib/src/sync/credit_notes.ts

@@ -5,6 +5,7 @@ import type Orb from 'orb-billing';
 import { CreditNotesFetchParams } from '../types';
 
 const TABLE = 'credit_notes';
+const TIMER_LOGGING_LABEL = 'fetch-credit-notes';
 
 export async function syncCreditNotes(postgresClient: PostgresClient, creditNotes: CreditNote[]) {
   return postgresClient.upsertMany(
@@ -22,19 +23,27 @@ export async function fetchAndSyncCreditNotes(
   orbClient: Orb,
   params: CreditNotesFetchParams
 ): Promise<number> {
-  const creditNotes = [];
+  let numberOfCreditNotes = 0;
 
+  console.time(TIMER_LOGGING_LABEL);
   let creditNotesPage = await orbClient.creditNotes.list({ limit: params.limit || 100 });
-  creditNotes.push(...creditNotesPage.data);
+  console.timeEnd(TIMER_LOGGING_LABEL);
+
+  numberOfCreditNotes += creditNotesPage.data.length;
+
+  await syncCreditNotes(postgresClient, creditNotesPage.data);
 
   while (creditNotesPage.hasNextPage()) {
+    console.time(TIMER_LOGGING_LABEL);
     creditNotesPage = await creditNotesPage.getNextPage();
-    creditNotes.push(...creditNotesPage.data);
-  }
+    console.timeEnd(TIMER_LOGGING_LABEL);
 
-  await syncCreditNotes(postgresClient, creditNotes);
+    numberOfCreditNotes += creditNotesPage.data.length;
+
+    await syncCreditNotes(postgresClient, creditNotesPage.data);
+  }
 
-  return creditNotes.length;
+  return numberOfCreditNotes;
 }
 
 export async function fetchAndSyncCreditNote(postgresClient: PostgresClient, orbClient: Orb, creditNoteId: string) {

packages/orb-sync-lib/src/sync/customers.ts

@@ -5,6 +5,7 @@ import { customerSchema } from '../schemas/customer';
 import { CustomersFetchParams } from '../types';
 
 const TABLE = 'customers';
+const TIMER_LOGGING_LABEL = 'fetch-customers';
 
 export async function syncCustomers(postgresClient: PostgresClient, customers: Customer[]) {
   return postgresClient.upsertMany(customers, TABLE, customerSchema);
@@ -15,25 +16,33 @@ export async function fetchAndSyncCustomers(
   orbClient: Orb,
   params: CustomersFetchParams
 ): Promise<number> {
-  const customers = [];
+  let numberOfCustomers = 0;
 
+  console.time(TIMER_LOGGING_LABEL);
   let customersPage = await orbClient.customers.list({
     limit: params.limit || 100,
     'created_at[gt]': params.createdAtGt,
     'created_at[gte]': params.createdAtGte,
     'created_at[lt]': params.createdAtLt,
     'created_at[lte]': params.createdAtLte,
   });
-  customers.push(...customersPage.data);
+  console.timeEnd(TIMER_LOGGING_LABEL);
+
+  numberOfCustomers += customersPage.data.length;
+
+  await syncCustomers(postgresClient, customersPage.data);
 
   while (customersPage.hasNextPage()) {
+    console.time(TIMER_LOGGING_LABEL);
     customersPage = await customersPage.getNextPage();
-    customers.push(...customersPage.data);
-  }
+    console.timeEnd(TIMER_LOGGING_LABEL);
 
-  await syncCustomers(postgresClient, customers);
+    numberOfCustomers += customersPage.data.length;
+
+    await syncCustomers(postgresClient, customersPage.data);
+  }
 
-  return customers.length;
+  return numberOfCustomers;
 }
 
 export async function fetchAndSyncCustomer(postgresClient: PostgresClient, orbClient: Orb, customerId: string) {

packages/orb-sync-lib/src/sync/invoices.ts

@@ -5,6 +5,7 @@ import { invoiceSchema } from '../schemas/invoice';
 import { InvoicesFetchParams } from '../types';
 
 const TABLE = 'invoices';
+const TIMER_LOGGING_LABEL = 'fetch-invoices';
 
 export async function syncInvoices(postgresClient: PostgresClient, invoices: Invoice[]) {
   return postgresClient.upsertMany(
@@ -23,25 +24,33 @@ export async function fetchAndSyncInvoices(
   orbClient: Orb,
   params: InvoicesFetchParams
 ): Promise<number> {
-  const invoices = [];
+  let numberOfInvoices = 0;
 
+  console.time(TIMER_LOGGING_LABEL);
   let invoicesPage = await orbClient.invoices.list({
     limit: params.limit || 100,
     'invoice_date[gt]': params.createdAtGt,
     'invoice_date[gte]': params.createdAtGte,
     'invoice_date[lt]': params.createdAtLt,
     'invoice_date[lte]': params.createdAtLte,
   });
-  invoices.push(...invoicesPage.data);
+  console.timeEnd(TIMER_LOGGING_LABEL);
+
+  numberOfInvoices += invoicesPage.data.length;
+
+  await syncInvoices(postgresClient, invoicesPage.data);
 
   while (invoicesPage.hasNextPage()) {
+    console.time(TIMER_LOGGING_LABEL);
     invoicesPage = await invoicesPage.getNextPage();
-    invoices.push(...invoicesPage.data);
-  }
+    console.timeEnd(TIMER_LOGGING_LABEL);
 
-  await syncInvoices(postgresClient, invoices);
+    numberOfInvoices += invoicesPage.data.length;
+
+    await syncInvoices(postgresClient, invoicesPage.data);
+  }
 
-  return invoices.length;
+  return numberOfInvoices;
 }
 
 export async function fetchAndSyncInvoice(postgresClient: PostgresClient, orbClient: Orb, invoiceId: string) {

packages/orb-sync-lib/src/sync/plans.ts

@@ -5,6 +5,7 @@ import { PlansFetchParams } from '../types';
 import { planSchema } from '../schemas/plan';
 
 const TABLE = 'plans';
+const TIMER_LOGGING_LABEL = 'fetch-plans';
 
 export async function syncPlans(postgresClient: PostgresClient, plans: Plan[]) {
   return postgresClient.upsertMany(plans, TABLE, planSchema);
@@ -15,26 +16,33 @@ export async function fetchAndSyncPlans(
   orbClient: Orb,
   params: PlansFetchParams
 ): Promise<number> {
-  const plans = [];
+  let numberOfPlans = 0;
 
+  console.time(TIMER_LOGGING_LABEL);
   let plansPage = await orbClient.plans.list({
     limit: params.limit || 100,
     'created_at[gt]': params.createdAtGt,
     'created_at[gte]': params.createdAtGte,
     'created_at[lt]': params.createdAtLt,
     'created_at[lte]': params.createdAtLte,
   });
+  console.timeEnd(TIMER_LOGGING_LABEL);
 
-  plans.push(...plansPage.data);
+  numberOfPlans += plansPage.data.length;
+
+  await syncPlans(postgresClient, plansPage.data);
 
   while (plansPage.hasNextPage()) {
+    console.time(TIMER_LOGGING_LABEL);
     plansPage = await plansPage.getNextPage();
-    plans.push(...plansPage.data);
-  }
+    console.timeEnd(TIMER_LOGGING_LABEL);
 
-  await syncPlans(postgresClient, plans);
+    numberOfPlans += plansPage.data.length;
+
+    await syncPlans(postgresClient, plansPage.data);
+  }
 
-  return plans.length;
+  return numberOfPlans;
 }
 
 export async function fetchAndSyncPlan(postgresClient: PostgresClient, orbClient: Orb, planId: string) {

packages/orb-sync-lib/src/sync/subscriptions.ts

@@ -5,6 +5,7 @@ import { subscriptionSchema } from '../schemas/subscription';
 import { SubscriptionsFetchParams } from '../types';
 
 const TABLE = 'subscriptions';
+const TIMER_LOGGING_LABEL = 'fetch-subscriptions';
 
 export async function syncSubscriptions(postgresClient: PostgresClient, subscriptions: Subscription[]) {
   return postgresClient.upsertMany(
@@ -23,25 +24,33 @@ export async function fetchAndSyncSubscriptions(
   orbClient: Orb,
   params: SubscriptionsFetchParams
 ): Promise<number> {
-  const subscriptions = [];
+  let numberOfSubscriptions = 0;
 
+  console.time(TIMER_LOGGING_LABEL);
   let subscriptionsPage = await orbClient.subscriptions.list({
     limit: params.limit || 100,
     'created_at[gt]': params.createdAtGt,
     'created_at[gte]': params.createdAtGte,
     'created_at[lt]': params.createdAtLt,
     'created_at[lte]': params.createdAtLte,
   });
-  subscriptions.push(...subscriptionsPage.data);
+  console.timeEnd(TIMER_LOGGING_LABEL);
+
+  numberOfSubscriptions += subscriptionsPage.data.length;
+
+  await syncSubscriptions(postgresClient, subscriptionsPage.data);
 
   while (subscriptionsPage.hasNextPage()) {
+    console.time(TIMER_LOGGING_LABEL);
     subscriptionsPage = await subscriptionsPage.getNextPage();
-    subscriptions.push(...subscriptionsPage.data);
-  }
+    console.timeEnd(TIMER_LOGGING_LABEL);
 
-  await syncSubscriptions(postgresClient, subscriptions);
+    numberOfSubscriptions += subscriptionsPage.data.length;
+
+    await syncSubscriptions(postgresClient, subscriptionsPage.data);
+  }
 
-  return subscriptions.length;
+  return numberOfSubscriptions;
 }
 
 export async function fetchAndSyncSubscription(postgresClient: PostgresClient, orbClient: Orb, subscriptionId: string) {