chore: migrate Invoice to pg

package.json

@@ -104,7 +104,7 @@
     "husky": "^7.0.4",
     "jscodeshift": "^0.14.0",
     "kysely": "^0.27.2",
-    "kysely-codegen": "^0.11.0",
+    "kysely-codegen": "0.12.0",
     "lerna": "^6.4.1",
     "mini-css-extract-plugin": "^2.7.2",
     "minimist": "^1.2.5",

packages/server/billing/helpers/generateInvoice.ts

@@ -1,5 +1,6 @@
 import {InvoiceItemType} from 'parabol-client/types/constEnums'
 import Stripe from 'stripe'
+import getKysely from '../../postgres/getKysely'
 import getRethink from '../../database/rethinkDriver'
 import Coupon from '../../database/types/Coupon'
 import Invoice, {InvoiceStatusEnum} from '../../database/types/Invoice'
@@ -303,6 +304,7 @@ export default async function generateInvoice(
   dataLoader: DataLoaderWorker
 ) {
   const r = await getRethink()
+  const pg = getKysely()
   const now = new Date()
 
   const {itemDict, nextPeriodCharges, unknownLineItems} = makeItemDict(stripeLineItems)
@@ -376,31 +378,35 @@ export default async function generateInvoice(
       })) ||
     null
 
-  const dbInvoice = new Invoice({
-    id: invoiceId,
-    amountDue: invoice.amount_due,
-    createdAt: now,
-    coupon,
-    total: invoice.total,
-    billingLeaderEmails,
-    creditCard: organization.creditCard,
-    endAt: fromEpochSeconds(invoice.period_end),
-    invoiceDate: fromEpochSeconds(invoice.due_date!),
-    lines: invoiceLineItems,
-    nextPeriodCharges,
-    orgId,
-    orgName: organization.name,
-    paidAt,
-    payUrl: invoice.hosted_invoice_url,
-    picture: organization.picture,
-    startAt: fromEpochSeconds(invoice.period_start),
-    startingBalance: invoice.starting_balance,
-    status,
-    tier: nextPeriodCharges.interval === 'year' ? 'enterprise' : 'team'
-  })
-
-  return r
-    .table('Invoice')
-    .insert(dbInvoice, {conflict: 'replace', returnChanges: true})('changes')(0)('new_val')
-    .run()
+  const dbInvoice = Invoice.toPg(
+    new Invoice({
+      id: invoiceId,
+      amountDue: invoice.amount_due,
+      createdAt: now,
+      coupon,
+      total: invoice.total,
+      billingLeaderEmails,
+      creditCard: organization.creditCard,
+      endAt: fromEpochSeconds(invoice.period_end),
+      invoiceDate: fromEpochSeconds(invoice.due_date!),
+      lines: invoiceLineItems,
+      nextPeriodCharges,
+      orgId,
+      orgName: organization.name,
+      paidAt,
+      payUrl: invoice.hosted_invoice_url,
+      picture: organization.picture,
+      startAt: fromEpochSeconds(invoice.period_start),
+      startingBalance: invoice.starting_balance,
+      status,
+      tier: nextPeriodCharges.interval === 'year' ? 'enterprise' : 'team'
+    })
+  )
+
+  return pg
+    .insertInto('Invoice')
+    .values(dbInvoice)
+    .onConflict((oc) => oc.column('id').doUpdateSet(dbInvoice))
+    .returningAll()
+    .executeTakeFirst()
 }

packages/server/database/types/Invoice.ts

@@ -96,4 +96,14 @@ export default class Invoice {
     this.status = status
     this.tier = tier
   }
+
+  static toPg(invoice: Invoice) {
+    return {
+      ...invoice,
+      coupon: JSON.stringify(invoice.coupon),
+      creditCard: JSON.stringify(invoice.creditCard),
+      lines: invoice.lines.map((line) => JSON.stringify(line)),
+      nextPeriodCharges: JSON.stringify(invoice.nextPeriodCharges)
+    }
+  }
 }

packages/server/graphql/private/mutations/backupOrganization.ts

@@ -71,11 +71,11 @@ const dumpPgDataToOrgBackupSchema = async (orgIds: string[]) => {
       [teamIds]
     )
     await client.query(
-      `CREATE TABLE "orgBackup"."TaskEstimate" AS (SELECT * FROM "TaskEstimate" WHERE "userId" = ANY ($1));`,
-      [userIds]
+      `CREATE TABLE "orgBackup"."Invoice" AS (SELECT * FROM "Invoice" WHERE "orgId" = ANY ($1));`,
+      [orgIds]
     )
     await client.query(
-      `CREATE TABLE "orgBackup"."User" AS (SELECT * FROM "User" WHERE "id" = ANY ($1));`,
+      `CREATE TABLE "orgBackup"."TaskEstimate" AS (SELECT * FROM "TaskEstimate" WHERE "userId" = ANY ($1));`,
       [userIds]
     )
     await client.query(
@@ -86,6 +86,10 @@ const dumpPgDataToOrgBackupSchema = async (orgIds: string[]) => {
       `CREATE TABLE "orgBackup"."TemplateScaleRef" AS (SELECT * FROM "TemplateScaleRef" WHERE "id" = ANY ($1));`,
       [templateScaleRefIds]
     )
+    await client.query(
+      `CREATE TABLE "orgBackup"."User" AS (SELECT * FROM "User" WHERE "id" = ANY ($1));`,
+      [userIds]
+    )
     await client.query('COMMIT')
   } catch (e) {
     await client.query('ROLLBACK')
@@ -174,9 +178,6 @@ const backupOrganization: MutationResolvers['backupOrganization'] = async (_sour
     atlassianAuth: (r.table('AtlassianAuth').getAll(r.args(teamIds), {index: 'teamId'}) as any)
       .coerceTo('array')
       .do((items: RValue) => r.db(DESTINATION).table('AtlassianAuth').insert(items)),
-    invoice: (r.table('Invoice').filter((row: RDatum) => r(orgIds).contains(row('orgId'))) as any)
-      .coerceTo('array')
-      .do((items: RValue) => r.db(DESTINATION).table('Invoice').insert(items)),
     invoiceItemHook: (
       r.table('InvoiceItemHook').filter((row: RDatum) => r(orgIds).contains(row('orgId'))) as any
     )

packages/server/graphql/private/mutations/changeEmailDomain.ts

@@ -72,23 +72,31 @@ const changeEmailDomain: MutationResolvers['changeEmailDomain'] = async (
       .set({domain: normalizedNewDomain})
       .where('domain', '=', normalizedOldDomain)
       .execute(),
-    r
-      .table('Invoice')
-      .filter((row: RDatum) =>
-        row('billingLeaderEmails').contains((email: RValue) =>
-          email.split('@').nth(1).eq(normalizedOldDomain)
+    pg
+      .updateTable('Invoice')
+      .set({
+        billingLeaderEmails: sql<string[]>`
+        array(
+          SELECT
+            CASE
+                WHEN strpos(email, '@') > 0 AND split_part(email, '@', 2) = '${normalizedOldDomain}' THEN
+                    split_part(email, '@', 1) || '@${normalizedNewDomain}'
+                ELSE
+                    email
+            END
+          FROM unnest(billingLeaderEmails) AS t(email)
+        )`
+      })
+      .where((eb) =>
+        eb.exists(sql<boolean[]>`
+        (
+          SELECT 1
+          FROM unnest(billingLeaderEmails) AS t(email)
+          WHERE strpos(email, '@') > 0 AND split_part(email, '@', 2) = '${normalizedOldDomain}'
         )
+      `)
       )
-      .update((row: RDatum) => ({
-        billingLeaderEmails: row('billingLeaderEmails').map((email: RValue) =>
-          r.branch(
-            email.split('@').nth(1).eq(normalizedOldDomain),
-            email.split('@').nth(0).add(`@${normalizedNewDomain}`),
-            email
-          )
-        )
-      }))
-      .run()
+      .execute()
   ])
 
   const usersUpdatedIds = updatedUserRes.map(({id}) => id)

packages/server/postgres/migrations/1677272969994_meetingTemplatesMove.ts

@@ -58,7 +58,7 @@ export async function up() {
         'batchSize is smaller than the number of items that share the same cursor. Increase batchSize'
       )
     }
-    return nextBatch.slice(0, lastMatchingUpdatedAt + 1)
+    return nextBatch.slice(0, lastMatchingUpdatedAt)
   }
 
   await pg.tx('meetingTemplateTable', (task) => {

packages/server/postgres/migrations/1709927241000_addInvoice.ts

@@ -0,0 +1,52 @@
+import {Client} from 'pg'
+import getPgConfig from '../getPgConfig'
+
+export async function up() {
+  const client = new Client(getPgConfig())
+  await client.connect()
+  await client.query(`
+    DO $$
+    BEGIN
+      IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'InvoiceStatusEnum') THEN
+            EXECUTE 'CREATE TYPE "InvoiceStatusEnum" AS ENUM (''FAILED'', ''PAID'', ''PENDING'', ''UPCOMING'')';
+      END IF;
+    END $$;
+
+    CREATE TABLE "Invoice" (
+      "id" VARCHAR(128) PRIMARY KEY, -- id is managed by our server app, values like upcoming_$orgId
+      "amountDue" NUMERIC(10,2) NOT NULL,
+      "createdAt" TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP NOT NULL,
+      "coupon" JSONB,
+      "total" NUMERIC(10,2) NOT NULL,
+      "billingLeaderEmails" "citext"[] NOT NULL,
+      "creditCard" JSONB,
+      "endAt" TIMESTAMP WITH TIME ZONE NOT NULL,
+      "invoiceDate" TIMESTAMP WITH TIME ZONE NOT NULL,
+      "lines" JSONB[] NOT NULL DEFAULT '{}',
+      "nextPeriodCharges" JSONB NOT NULL,
+      "orgId" VARCHAR(100),
+      "orgName" VARCHAR(100), -- Max len is 50 in EditableOrgName.tsx
+      "paidAt" TIMESTAMP WITH TIME ZONE,
+      "payUrl" VARCHAR(2056),
+      "picture" VARCHAR(2056),
+      "startAt" TIMESTAMP WITH TIME ZONE NOT NULL,
+      "startingBalance" NUMERIC(10,2) NOT NULL,
+      "status" "InvoiceStatusEnum" NOT NULL,
+      "tier" "TierEnum" NOT NULL
+    );
+
+    CREATE INDEX IF NOT EXISTS "idx_Invoice_orgId" ON "Invoice"("orgId");
+    CREATE INDEX IF NOT EXISTS "idx_Invoice_startAt" ON "Invoice"("startAt");
+  `)
+  await client.end()
+}
+
+export async function down() {
+  const client = new Client(getPgConfig())
+  await client.connect()
+  await client.query(`
+    DROP TABLE IF EXISTS "Invoice";
+    DROP TYPE IF EXISTS "InvoiceStatusEnum";
+  `)
+  await client.end()
+}

packages/server/postgres/migrations/1709927276000_moveInvoice.ts

@@ -0,0 +1,89 @@
+import {FirstParam} from 'parabol-client/types/generics'
+import {Client} from 'pg'
+import {r} from 'rethinkdb-ts'
+import getPgConfig from '../getPgConfig'
+import connectRethinkDB from '../../database/connectRethinkDB'
+import getPgp from '../getPgp'
+
+export async function up() {
+  await connectRethinkDB()
+  const {pgp, pg} = getPgp()
+  const batchSize = 1000
+
+  try {
+    await r.table('Invoice').indexCreate('startAt').run()
+    await r.table('Invoice').indexWait().run()
+  } catch {}
+
+  const columnSet = new pgp.helpers.ColumnSet(
+    [
+      'id',
+      'amountDue',
+      'createdAt',
+      {name: 'coupon', def: null},
+      'total',
+      'billingLeaderEmails',
+      {name: 'creditCard', def: null},
+      'endAt',
+      'invoiceDate',
+      {name: 'lines', cast: 'jsonb[]'},
+      'nextPeriodCharges',
+      'orgId',
+      'orgName',
+      {name: 'payUrl', def: null},
+      {name: 'picture', def: null},
+      'startAt',
+      'startingBalance',
+      'status',
+      'tier'
+    ],
+    {table: 'Invoice'}
+  )
+
+  const getNextData = async (leftBoundCursor: Date | undefined) => {
+    const startAt = leftBoundCursor || r.minval
+    const nextBatch = await r
+      .table('Invoice')
+      .between(startAt, r.maxval, {index: 'startAt', leftBound: 'open'})
+      .orderBy({index: 'startAt'})
+      .limit(batchSize)
+      .run()
+    if (nextBatch.length === 0) return null
+    if (nextBatch.length < batchSize) return nextBatch
+    const lastItem = nextBatch.pop()
+    const lastMatchingStartAt = nextBatch.findLastIndex(
+      (item) => item.startAt !== lastItem!.startAt
+    )
+    if (lastMatchingStartAt === -1) {
+      throw new Error(
+        'batchSize is smaller than the number of items that share the same cursor. Increase batchSize'
+      )
+    }
+    return nextBatch.slice(0, lastMatchingStartAt)
+  }
+
+  await pg.tx('Invoice', (task) => {
+    const fetchAndProcess: FirstParam<typeof task.sequence> = async (
+      _index,
+      leftBoundCursor: undefined | Date
+    ) => {
+      const nextData = await getNextData(leftBoundCursor)
+      if (!nextData) return undefined
+      const insert = pgp.helpers.insert(nextData, columnSet)
+      await task.none(insert)
+      return nextData.at(-1)!.startAt
+    }
+    return task.sequence(fetchAndProcess)
+  })
+  await r.getPoolMaster()?.drain()
+}
+
+export async function down() {
+  const client = new Client(getPgConfig())
+  await client.connect()
+  await client.query(`DELETE FROM "Invoice"`)
+  await client.end()
+  try {
+    await r.table('Invoice').indexDrop('startAt').run()
+  } catch {}
+}

yarn.lock

@@ -11192,11 +11192,16 @@ dotenv@8.0.0:
   resolved "https://registry.yarnpkg.com/dotenv/-/dotenv-8.0.0.tgz#ed310c165b4e8a97bb745b0a9d99c31bda566440"
   integrity sha512-30xVGqjLjiUOArT4+M5q9sYdvuR4riM6yK9wMcas9Vbp6zZa+ocC9dp6QoftuhTPhFAiLK/0C5Ni2nou/Bk8lg==
 
-dotenv@^16.0.0, dotenv@^16.0.3:
+dotenv@^16.0.0:
   version "16.0.3"
   resolved "https://registry.yarnpkg.com/dotenv/-/dotenv-16.0.3.tgz#115aec42bac5053db3c456db30cc243a5a836a07"
   integrity sha512-7GO6HghkA5fYG9TYnNxi14/7K9f5occMlp3zXAuSxn7CKCxt9xbNWG7yF8hTCSUchlfWSe3uLmlPfigevRItzQ==
 
+dotenv@^16.3.1:
+  version "16.4.5"
+  resolved "https://registry.yarnpkg.com/dotenv/-/dotenv-16.4.5.tgz#cdd3b3b604cb327e286b4762e13502f717cb099f"
+  integrity sha512-ZmdL2rui+eB2YwhsWzjInR8LldtZHGDoQ1ugH85ppHKwpUHL7j7rN0Ti9NCnGiQbhaZ11FpR+7ao1dNsmduNUg==
+
 dotenv@~10.0.0:
   version "10.0.0"
   resolved "https://registry.yarnpkg.com/dotenv/-/dotenv-10.0.0.tgz#3d4227b8fb95f81096cdd2b66653fb2c7085ba81"
@@ -11223,6 +11228,7 @@ draft-js-utils@^1.4.0:
 
 "draft-js@https://github.com/mattkrick/draft-js/tarball/559a21968370c4944511657817d601a6c4ade0f6":
   version "0.10.5"
+  uid "025fddba56f21aaf3383aee778e0b17025c9a7bc"
   resolved "https://github.com/mattkrick/draft-js/tarball/559a21968370c4944511657817d601a6c4ade0f6#025fddba56f21aaf3383aee778e0b17025c9a7bc"
   dependencies:
     fbjs "^0.8.15"
@@ -14905,13 +14911,13 @@ koalas@^1.0.2:
   resolved "https://registry.yarnpkg.com/koalas/-/koalas-1.0.2.tgz#318433f074235db78fae5661a02a8ca53ee295cd"
   integrity sha1-MYQz8HQjXbePrlZhoCqMpT7ilc0=
 
-kysely-codegen@^0.11.0:
-  version "0.11.0"
-  resolved "https://registry.yarnpkg.com/kysely-codegen/-/kysely-codegen-0.11.0.tgz#7506955c4c09201b571d528b42ffec8a1869160b"
-  integrity sha512-8aklzXygjANshk5BoGSQ0BWukKIoPL4/k1iFWyteGUQ/VtB1GlyrELBZv1GglydjLGECSSVDpsOgEXyWQmuksg==
+kysely-codegen@0.12.0:
+  version "0.12.0"
+  resolved "https://registry.yarnpkg.com/kysely-codegen/-/kysely-codegen-0.12.0.tgz#750a84387c916ecd0e204004796231b07805d60e"
+  integrity sha512-jTlN99kIp+igPZOck1P+WB0ATjO9ET0cymou0pBE0kpQfwhYsUj8YeCWCa5OfDna2VlMMfVolDYbzQhqBk5tiA==
   dependencies:
     chalk "4.1.2"
-    dotenv "^16.0.3"
+    dotenv "^16.3.1"
     git-diff "^2.0.6"
     micromatch "^4.0.5"
     minimist "^1.2.8"