feat(subscriptions): added polar as payment provider for subscriptions

* feature(dashboard): add polar / subscription

* wip(payments): manage subscription

* wip(payments): add free product, faq and some other improvements

* fix(root): change node to bundler in tsconfig

* wip(payments): display current subscription

* feat(dashboard): schedule project for deletion

* wip(payments): support custom products/subscriptions

* wip(payments): fix polar scripts

* wip(payments): add json package to dockerfiles

apps/api/src/controllers/live.controller.ts

@@ -1,72 +1,35 @@
-import type { FastifyReply, FastifyRequest } from 'fastify';
+import type { SocketStream } from '@fastify/websocket';
+import type { FastifyRequest } from 'fastify';
 import superjson from 'superjson';
-import type * as WebSocket from 'ws';
+import type { WebSocket } from 'ws';
 
-import { getSuperJson } from '@openpanel/common';
-import type { IServiceEvent, Notification } from '@openpanel/db';
 import {
-  TABLE_NAMES,
   eventBuffer,
-  getEvents,
   getProfileByIdCached,
   transformMinimalEvent,
 } from '@openpanel/db';
-import { getRedisCache, getRedisPub, getRedisSub } from '@openpanel/redis';
+import { setSuperJson } from '@openpanel/json';
+import {
+  psubscribeToPublishedEvent,
+  subscribeToPublishedEvent,
+} from '@openpanel/redis';
 import { getProjectAccess } from '@openpanel/trpc';
+import { getOrganizationAccess } from '@openpanel/trpc/src/access';
+
+type WebSocketConnection = SocketStream & {
+  socket: WebSocket & {
+    on(event: 'close', listener: () => void): void;
+    send(data: string): void;
+    close(): void;
+  };
+};
 
 export function getLiveEventInfo(key: string) {
   return key.split(':').slice(2) as [string, string];
 }
 
-export async function testVisitors(
-  req: FastifyRequest<{
-    Params: {
-      projectId: string;
-    };
-  }>,
-  reply: FastifyReply,
-) {
-  const events = await getEvents(
-    `SELECT * FROM ${TABLE_NAMES.events} LIMIT 500`,
-  );
-  const event = events[Math.floor(Math.random() * events.length)];
-  if (!event) {
-    return reply.status(404).send('No event found');
-  }
-  event.projectId = req.params.projectId;
-  getRedisPub().publish('event:received', superjson.stringify(event));
-  getRedisCache().set(
-    `live:event:${event.projectId}:${Math.random() * 1000}`,
-    '',
-    'EX',
-    10,
-  );
-  reply.status(202).send(event);
-}
-
-export async function testEvents(
-  req: FastifyRequest<{
-    Params: {
-      projectId: string;
-    };
-  }>,
-  reply: FastifyReply,
-) {
-  const events = await getEvents(
-    `SELECT * FROM ${TABLE_NAMES.events} LIMIT 500`,
-  );
-  const event = events[Math.floor(Math.random() * events.length)];
-  if (!event) {
-    return reply.status(404).send('No event found');
-  }
-  getRedisPub().publish('event:saved', superjson.stringify(event));
-  reply.status(202).send(event);
-}
-
 export function wsVisitors(
-  connection: {
-    socket: WebSocket;
-  },
+  connection: WebSocketConnection,
   req: FastifyRequest<{
     Params: {
       projectId: string;
@@ -75,60 +38,46 @@ export function wsVisitors(
 ) {
   const { params } = req;
 
-  getRedisSub().subscribe('event:received');
-  getRedisSub().psubscribe('__keyevent@0__:expired');
-
-  const message = (channel: string, message: string) => {
-    if (channel === 'event:received') {
-      const event = getSuperJson<IServiceEvent>(message);
-      if (event?.projectId === params.projectId) {
-        eventBuffer.getActiveVisitorCount(params.projectId).then((count) => {
-          connection.socket.send(String(count));
-        });
-      }
-    }
-  };
-  const pmessage = (pattern: string, channel: string, message: string) => {
-    if (!message.startsWith('live:visitor:')) {
-      return null;
-    }
-
-    const [projectId] = getLiveEventInfo(message);
-    if (projectId && projectId === params.projectId) {
+  const unsubscribe = subscribeToPublishedEvent('events', 'saved', (event) => {
+    if (event?.projectId === params.projectId) {
       eventBuffer.getActiveVisitorCount(params.projectId).then((count) => {
         connection.socket.send(String(count));
       });
     }
-  };
+  });
 
-  getRedisSub().on('message', message);
-  getRedisSub().on('pmessage', pmessage);
+  const punsubscribe = psubscribeToPublishedEvent(
+    '__keyevent@0__:expired',
+    (key) => {
+      const [projectId] = getLiveEventInfo(key);
+      if (projectId && projectId === params.projectId) {
+        eventBuffer.getActiveVisitorCount(params.projectId).then((count) => {
+          connection.socket.send(String(count));
+        });
+      }
+    },
+  );
 
   connection.socket.on('close', () => {
-    getRedisSub().unsubscribe('event:saved');
-    getRedisSub().punsubscribe('__keyevent@0__:expired');
-    getRedisSub().off('message', message);
-    getRedisSub().off('pmessage', pmessage);
+    unsubscribe();
+    punsubscribe();
   });
 }
 
 export async function wsProjectEvents(
-  connection: {
-    socket: WebSocket;
-  },
+  connection: WebSocketConnection,
   req: FastifyRequest<{
     Params: {
       projectId: string;
     };
     Querystring: {
       token?: string;
-      type?: string;
+      type?: 'saved' | 'received';
     };
   }>,
 ) {
   const { params, query } = req;
   const type = query.type || 'saved';
-  const subscribeToEvent = `event:${type}`;
 
   if (!['saved', 'received'].includes(type)) {
     connection.socket.send('Invalid type');
@@ -148,12 +97,11 @@ export async function wsProjectEvents(
     projectId: params.projectId,
   });
 
-  getRedisSub().subscribe(subscribeToEvent);
-
-  const message = async (channel: string, message: string) => {
-    if (channel === subscribeToEvent) {
-      const event = getSuperJson<IServiceEvent>(message);
-      if (event?.projectId === params.projectId) {
+  const unsubscribe = subscribeToPublishedEvent(
+    'events',
+    type,
+    async (event) => {
+      if (event.projectId === params.projectId) {
         const profile = await getProfileByIdCached(
           event.profileId,
           event.projectId,
@@ -169,31 +117,21 @@ export async function wsProjectEvents(
           ),
         );
       }
-    }
-  };
+    },
+  );
 
-  getRedisSub().on('message', message as any);
-
-  connection.socket.on('close', () => {
-    getRedisSub().unsubscribe(subscribeToEvent);
-    getRedisSub().off('message', message as any);
-  });
+  connection.socket.on('close', () => unsubscribe());
 }
 
 export async function wsProjectNotifications(
-  connection: {
-    socket: WebSocket;
-  },
+  connection: WebSocketConnection,
   req: FastifyRequest<{
     Params: {
       projectId: string;
     };
-    Querystring: {
-      token?: string;
-    };
   }>,
 ) {
-  const { params, query } = req;
+  const { params } = req;
   const userId = req.session?.userId;
 
   if (!userId) {
@@ -202,8 +140,6 @@ export async function wsProjectNotifications(
     return;
   }
 
-  const subscribeToEvent = 'notification';
-
   const access = await getProjectAccess({
     userId,
     projectId: params.projectId,
@@ -215,21 +151,54 @@ export async function wsProjectNotifications(
     return;
   }
 
-  getRedisSub().subscribe(subscribeToEvent);
-
-  const message = async (channel: string, message: string) => {
-    if (channel === subscribeToEvent) {
-      const notification = getSuperJson<Notification>(message);
-      if (notification?.projectId === params.projectId) {
+  const unsubscribe = subscribeToPublishedEvent(
+    'notification',
+    'created',
+    (notification) => {
+      if (notification.projectId === params.projectId) {
         connection.socket.send(superjson.stringify(notification));
       }
-    }
-  };
+    },
+  );
 
-  getRedisSub().on('message', message as any);
-
-  connection.socket.on('close', () => {
-    getRedisSub().unsubscribe(subscribeToEvent);
-    getRedisSub().off('message', message as any);
-  });
+  connection.socket.on('close', () => unsubscribe());
+}
+
+export async function wsOrganizationEvents(
+  connection: WebSocketConnection,
+  req: FastifyRequest<{
+    Params: {
+      organizationId: string;
+    };
+  }>,
+) {
+  const { params } = req;
+  const userId = req.session?.userId;
+
+  if (!userId) {
+    connection.socket.send('No active session');
+    connection.socket.close();
+    return;
+  }
+
+  const access = await getOrganizationAccess({
+    userId,
+    organizationId: params.organizationId,
+  });
+
+  if (!access) {
+    connection.socket.send('No access');
+    connection.socket.close();
+    return;
+  }
+
+  const unsubscribe = subscribeToPublishedEvent(
+    'organization',
+    'subscription_updated',
+    (message) => {
+      connection.socket.send(setSuperJson(message));
+    },
+  );
+
+  connection.socket.on('close', () => unsubscribe());
 }

apps/api/src/controllers/webhook.controller.ts

@@ -5,6 +5,12 @@ import {
   sendSlackNotification,
   slackInstaller,
 } from '@openpanel/integrations/src/slack';
+import {
+  PolarWebhookVerificationError,
+  getProduct,
+  validatePolarEvent,
+} from '@openpanel/payments';
+import { publishEvent } from '@openpanel/redis';
 import { zSlackAuthResponse } from '@openpanel/validation';
 import type { FastifyReply, FastifyRequest } from 'fastify';
 import { z } from 'zod';
@@ -102,3 +108,123 @@ export async function slackWebhook(
     return reply.status(500).header('Content-Type', 'text/html').send(html);
   }
 }
+
+export async function polarWebhook(
+  request: FastifyRequest<{
+    Querystring: unknown;
+  }>,
+  reply: FastifyReply,
+) {
+  try {
+    const event = validatePolarEvent(
+      request.rawBody!,
+      request.headers as Record<string, string>,
+      process.env.POLAR_WEBHOOK_SECRET ?? '',
+    );
+
+    switch (event.type) {
+      case 'order.created': {
+        const metadata = z
+          .object({
+            organizationId: z.string(),
+          })
+          .parse(event.data.metadata);
+
+        if (event.data.billingReason === 'subscription_cycle') {
+          await db.organization.update({
+            where: {
+              id: metadata.organizationId,
+            },
+            data: {
+              subscriptionPeriodEventsCount: 0,
+            },
+          });
+        }
+        break;
+      }
+      case 'subscription.updated': {
+        const metadata = z
+          .object({
+            organizationId: z.string(),
+            userId: z.string(),
+          })
+          .parse(event.data.metadata);
+
+        const product = await getProduct(event.data.productId);
+        const eventsLimit = product.metadata?.eventsLimit;
+        const subscriptionPeriodEventsLimit =
+          typeof eventsLimit === 'number' ? eventsLimit : undefined;
+
+        if (!subscriptionPeriodEventsLimit) {
+          request.log.warn('No events limit found for product', { product });
+        }
+
+        // If we get a cancel event and we cant find it we should ignore it
+        // Since we only have one subscription per organization but you can have several in polar
+        // we dont want to override the existing subscription with a canceled one
+        // TODO: might be other events that we should handle like this?!
+        if (event.data.status === 'canceled') {
+          const orgSubscription = await db.organization.findFirst({
+            where: {
+              subscriptionCustomerId: event.data.customer.id,
+              subscriptionId: event.data.id,
+              subscriptionStatus: 'active',
+            },
+          });
+
+          if (!orgSubscription) {
+            return reply.status(202).send('OK');
+          }
+        }
+
+        await db.organization.update({
+          where: {
+            id: metadata.organizationId,
+          },
+          data: {
+            subscriptionId: event.data.id,
+            subscriptionCustomerId: event.data.customer.id,
+            subscriptionPriceId: event.data.priceId,
+            subscriptionProductId: event.data.productId,
+            subscriptionStatus: event.data.status,
+            subscriptionStartsAt: event.data.currentPeriodStart,
+            subscriptionCanceledAt: event.data.canceledAt,
+            subscriptionEndsAt:
+              event.data.status === 'canceled'
+                ? event.data.cancelAtPeriodEnd
+                  ? event.data.currentPeriodEnd
+                  : event.data.canceledAt
+                : event.data.currentPeriodEnd,
+            subscriptionCreatedByUserId: metadata.userId,
+            subscriptionInterval: event.data.recurringInterval,
+            subscriptionPeriodEventsLimit,
+          },
+        });
+
+        await publishEvent('organization', 'subscription_updated', {
+          organizationId: metadata.organizationId,
+        });
+
+        break;
+      }
+    }
+
+    reply.status(202).send('OK');
+  } catch (error) {
+    if (error instanceof PolarWebhookVerificationError) {
+      request.log.error('Polar webhook error', { error });
+      reply.status(403).send('');
+    }
+
+    throw error;
+  }
+}
+
+function isToday(date: Date) {
+  const today = new Date();
+  return (
+    date.getDate() === today.getDate() &&
+    date.getMonth() === today.getMonth() &&
+    date.getFullYear() === today.getFullYear()
+  );
+}

apps/api/src/index.ts

@@ -92,6 +92,10 @@ const startServer = async () => {
       };
     });
 
+    await fastify.register(import('fastify-raw-body'), {
+      global: false,
+    });
+
     fastify.addHook('preHandler', ipHook);
     fastify.addHook('preHandler', timestampHook);
     fastify.addHook('preHandler', fixHook);

apps/api/src/routes/live.router.ts

@@ -2,35 +2,31 @@ import * as controller from '@/controllers/live.controller';
 import fastifyWS from '@fastify/websocket';
 import type { FastifyPluginCallback } from 'fastify';
 
+// TODO: `as any` is a workaround since it starts to break after changed module resolution to bundler
+// which is needed for @polar/sdk (dont have time to resolve this now)
 const liveRouter: FastifyPluginCallback = (fastify, opts, done) => {
-  fastify.route({
-    method: 'GET',
-    url: '/visitors/test/:projectId',
-    handler: controller.testVisitors,
-  });
-  fastify.route({
-    method: 'GET',
-    url: '/events/test/:projectId',
-    handler: controller.testEvents,
-  });
-
   fastify.register(fastifyWS);
 
   fastify.register((fastify, _, done) => {
+    fastify.get(
+      '/organization/:organizationId',
+      { websocket: true },
+      controller.wsOrganizationEvents as any,
+    );
     fastify.get(
       '/visitors/:projectId',
       { websocket: true },
-      controller.wsVisitors,
+      controller.wsVisitors as any,
     );
     fastify.get(
       '/events/:projectId',
       { websocket: true },
-      controller.wsProjectEvents,
+      controller.wsProjectEvents as any,
     );
     fastify.get(
       '/notifications/:projectId',
       { websocket: true },
-      controller.wsProjectNotifications,
+      controller.wsProjectNotifications as any,
     );
     done();
   });

apps/api/src/routes/webhook.router.ts

@@ -7,6 +7,14 @@ const webhookRouter: FastifyPluginCallback = (fastify, opts, done) => {
     url: '/slack',
     handler: controller.slackWebhook,
   });
+  fastify.route({
+    method: 'POST',
+    url: '/polar',
+    handler: controller.polarWebhook,
+    config: {
+      rawBody: true,
+    },
+  });
   done();
 };
 

apps/api/src/utils/parse-zod-query-string.ts

@@ -1,4 +1,4 @@
-import { getSafeJson } from '@openpanel/common';
+import { getSafeJson } from '@openpanel/json';
 
 export const parseQueryString = (obj: Record<string, any>): any => {
   return Object.fromEntries(