feat: billing status in workspace report

services/rotor/src/lib/pg-config-store.ts

@@ -1,7 +1,7 @@
 import { EnrichedConnectionConfig } from "@jitsu-internal/console/lib/server/fast-store";
 import { createInMemoryStore } from "./inmem-store";
 import { Pool, PoolClient } from "pg";
-import { assertDefined, getLog, namedParameters, newError, stopwatch } from "juava";
+import { assertDefined, getLog, namedParameters, newError } from "juava";
 import Cursor from "pg-cursor";
 import omit from "lodash/omit";
 import hash from "object-hash";

webapps/ee-api/lib/stripe.ts

@@ -268,3 +268,51 @@ export async function getOrCreatePortalConfiguration() {
   }
   return configurationId;
 }
+
+export async function listAllInvoices() {
+  const timer = Date.now();
+  let starting_after: string | undefined = undefined;
+  const allInvoices: Stripe.Invoice[] = [];
+  do {
+    const result = await stripe.invoices.list({
+      limit: 100,
+      starting_after: starting_after,
+      created: {
+        //invoices for past 90 days
+        gte: Math.floor(Date.now() / 1000 - 90 * 24 * 60 * 60),
+      },
+    });
+    starting_after = result?.data[result.data.length - 1]?.id;
+    if (result?.data) {
+      allInvoices.push(...result?.data);
+    }
+  } while (starting_after);
+  getLog()
+    .atInfo()
+    .log(`${allInvoices.length} invoices found. Took ${Date.now() - timer}ms`);
+  return allInvoices;
+}
+
+export function getInvoiceStartDate(invoice: Stripe.Invoice) {
+  return new Date(invoice.lines.data[0].period.start * 1000);
+}
+
+export function getInvoiceEndDate(invoice: Stripe.Invoice) {
+  return new Date(invoice.lines.data[0].period.end * 1000);
+}
+
+export async function listAllSubscriptions(): Promise<Stripe.Subscription[]> {
+  let starting_after: string | undefined = undefined;
+  const allSubscriptions: Stripe.Subscription[] = [];
+  do {
+    const result = await stripe.subscriptions.list({
+      limit: 100,
+      starting_after: starting_after,
+    });
+    starting_after = result?.data[result.data.length - 1]?.id;
+    if (result?.data) {
+      allSubscriptions.push(...result?.data);
+    }
+  } while (starting_after);
+  return allSubscriptions;
+}

webapps/ee-api/pages/api/billing/workspaces.ts

@@ -3,26 +3,10 @@ import { auth } from "../../../lib/auth";
 import { assertDefined, assertTrue, requireDefined } from "juava";
 import { withErrorHandler } from "../../../lib/error-handler";
 import { store } from "../../../lib/services";
-import { getStripeObjectTag, stripe, stripeDataTable, stripeLink } from "../../../lib/stripe";
+import { getStripeObjectTag, listAllSubscriptions, stripe, stripeDataTable, stripeLink } from "../../../lib/stripe";
 import Stripe from "stripe";
 import { omit } from "lodash";
 
-async function listAllSubscriptions(): Promise<Stripe.Subscription[]> {
-  let starting_after: string | undefined = undefined;
-  const allSubscriptions: Stripe.Subscription[] = [];
-  do {
-    const result = await stripe.subscriptions.list({
-      limit: 100,
-      starting_after: starting_after,
-    });
-    starting_after = result?.data[result.data.length - 1]?.id;
-    if (result?.data) {
-      allSubscriptions.push(...result?.data);
-    }
-  } while (starting_after);
-  return allSubscriptions;
-}
-
 const handler = async function handler(req: NextApiRequest, res: NextApiResponse) {
   const claims = await auth(req, res);
   const workspaceId: string | undefined = req.query.workspaceId ? (req.query.workspaceId as string) : undefined;

webapps/ee-api/pages/api/report/overage.ts

@@ -3,7 +3,15 @@ import { assertTrue, getLog } from "juava";
 import { withErrorHandler } from "../../../lib/error-handler";
 import { auth } from "../../../lib/auth";
 import { store } from "../../../lib/services";
-import { getAvailableProducts, stripe, stripeDataTable, stripeLink } from "../../../lib/stripe";
+import {
+  getAvailableProducts,
+  getInvoiceEndDate,
+  getInvoiceStartDate,
+  listAllInvoices,
+  stripe,
+  stripeDataTable,
+  stripeLink,
+} from "../../../lib/stripe";
 import Stripe from "stripe";
 import pick from "lodash/pick";
 import { buildWorkspaceReport } from "./workspace-stat";
@@ -29,36 +37,6 @@ function round(date: Date | string, granularity: "day" = "day"): { start: string
   }
 }
 
-async function listAllInvoices() {
-  const timer = Date.now();
-  let starting_after: string | undefined = undefined;
-  const allInvoices: Stripe.Invoice[] = [];
-  do {
-    const result = await stripe.invoices.list({
-      limit: 100,
-      starting_after: starting_after,
-      created: {
-        //invoices for past 90 days
-        gte: Math.floor(Date.now() / 1000 - 90 * 24 * 60 * 60),
-      },
-    });
-    starting_after = result?.data[result.data.length - 1]?.id;
-    if (result?.data) {
-      allInvoices.push(...result?.data);
-    }
-  } while (starting_after);
-  log.atInfo().log(`${allInvoices.length} invoices found. Took ${Date.now() - timer}ms`);
-  return allInvoices;
-}
-
-function getInvoiceStartDate(invoice: Stripe.Invoice) {
-  return new Date(invoice.lines.data[0].period.start * 1000);
-}
-
-function getInvoiceEndDate(invoice: Stripe.Invoice) {
-  return new Date(invoice.lines.data[0].period.end * 1000);
-}
-
 const msPerHour = 1000 * 60 * 60;
 const handler = async function handler(req: NextApiRequest, res: NextApiResponse) {
   const claims = await auth(req, res);

webapps/ee-api/pages/api/report/workspace-stat.ts

@@ -1,10 +1,12 @@
 import type { NextApiRequest, NextApiResponse } from "next";
-import { getLog, namedParameters, SqlQueryParameters, unrollParams } from "juava";
+import { assertDefined, assertTrue, namedParameters, requireDefined, SqlQueryParameters, unrollParams } from "juava";
 import { withErrorHandler } from "../../../lib/error-handler";
 import { auth } from "../../../lib/auth";
-import { clickhouse, pg } from "../../../lib/services";
+import { clickhouse, pg, store } from "../../../lib/services";
 import * as PG from "pg";
 import { getServerLog } from "../../../lib/log";
+import { listAllSubscriptions, stripe, stripeDataTable } from "../../../lib/stripe";
+import Stripe from "stripe";
 
 const log = getServerLog("/api/report");
 
@@ -77,9 +79,10 @@ async function getClickhousePart({
     },
     query_params: queryParams,
   });
-  log.atInfo().log(`Clickhouse query took ${Date.now() - timer}ms`);
+  const rows: any[] = ((await resultSet.json()) as any).data;
+  log.atInfo().log(`Clickhouse query took ${Date.now() - timer}ms. Rows in result: ${rows.length}`);
 
-  return ((await resultSet.json()) as any).data.map(({ events, period, ...rest }) => ({
+  return rows.map(({ events, period, ...rest }) => ({
     events: Number(events),
     period: period.replace(" ", "T") + ".000Z",
     ...rest,
@@ -150,7 +153,56 @@ const handler = async function handler(req: NextApiRequest, res: NextApiResponse
   const end = getDate(req.query.end as string, new Date().toISOString()).toISOString();
   const granularity = "day"; // = ((req.query.granularity as string) || "day").toLowerCase();
   const reportResult = await buildWorkspaceReport(start, end, granularity, workspaceId);
-  const records = extended ? await extend(reportResult) : reportResult;
+  let records = extended ? await extend(reportResult) : reportResult;
+  if (req.query.billing === "true") {
+    const allWorkspaces = workspaceId
+      ? [
+          {
+            id: workspaceId,
+            obj: await store.getTable(stripeDataTable).get(workspaceId),
+          },
+        ]
+      : await store.getTable(stripeDataTable).list();
+
+    const subscriptions: Record<string, Stripe.Subscription[]> = (await listAllSubscriptions()).reduce((acc, sub) => {
+      let customerId = typeof sub.customer === "string" ? sub.customer : sub.customer.id;
+      acc[customerId] = [...(acc[customerId] || []), sub];
+      return acc;
+    }, {});
+
+    const sub2product = new Map<string, Stripe.Product>();
+    for (const sub of Object.values(subscriptions).flat()) {
+      const productId = sub.items.data[0].price.product;
+      assertDefined(productId, `Can't get product from subscription ${sub.id}`);
+      assertTrue(typeof productId === "string", `Subscription ${sub.id} should have a string product id`);
+      const product = await stripe.products.retrieve(productId as string);
+      assertDefined(product, `Can't get product ${productId} from subscription ${sub.id}. Product doesn't exist`);
+      sub2product.set(sub.id, product);
+    }
+    const workspacePlans: Record<string, string> = {};
+
+    for (const { id: workspaceId, obj } of allWorkspaces) {
+      const { stripeCustomerId } = obj;
+      const customerSubscriptions = subscriptions[stripeCustomerId];
+      if (!customerSubscriptions) {
+        continue;
+      }
+      const products = customerSubscriptions.map(sub =>
+        requireDefined(sub2product.get(sub.id), `Can't find product for subscription ${sub.id}`)
+      );
+      const product = products.find(p => !!p.metadata.jitsu_plan_id);
+      if (product) {
+        if (customerSubscriptions.find(sub => sub.status === "active" && !sub.cancel_at_period_end)) {
+          workspacePlans[workspaceId] = "paying";
+        } else if (customerSubscriptions.find(sub => sub.status === "active" || sub.cancel_at_period_end)) {
+          workspacePlans[workspaceId] = "cancelling";
+        } else {
+          workspacePlans[workspaceId] = "free";
+        }
+      }
+      records = records.map(r => ({ ...r, billingStatus: workspacePlans[r.workspaceId] || "free" }));
+    }
+  }
   res.send(req.query.format === "array" ? records : { data: records });
 };