feature(dashboard): add integrations and notifications

packages/common/src/object.ts

@@ -47,17 +47,16 @@ export function getSafeJson<T>(str: string): T | null {
 
 export function getSuperJson<T>(str: string): T | null {
   const json = getSafeJson<T>(str);
-  if (
-    typeof json === 'object' &&
-    json !== null &&
-    'json' in json &&
-    'meta' in json
-  ) {
+  if (typeof json === 'object' && json !== null && 'json' in json) {
     return superjson.parse<T>(str);
   }
   return json;
 }
 
+export function setSuperJson(str: Record<string, unknown>): string {
+  return superjson.stringify(str);
+}
+
 type AnyObject = Record<string, any>;
 
 export function deepMergeObjects<T>(target: AnyObject, source: AnyObject): T {

packages/common/src/string.ts

@@ -1,3 +1,7 @@
 export function stripTrailingSlash(url: string) {
   return url.replace(/\/+$/, '');
 }
+
+export function stripLeadingAndTrailingSlashes(url: string) {
+  return url.replace(/^[/]+|[/]+$/g, '');
+}

packages/db/index.ts

@@ -15,4 +15,6 @@ export * from './src/services/user.service';
 export * from './src/services/reference.service';
 export * from './src/services/id.service';
 export * from './src/services/retention.service';
+export * from './src/services/notification.service';
 export * from './src/buffers';
+export * from './src/types';

packages/db/package.json

@@ -18,9 +18,11 @@
     "@openpanel/common": "workspace:*",
     "@openpanel/constants": "workspace:*",
     "@openpanel/logger": "workspace:*",
+    "@openpanel/queue": "workspace:^",
     "@openpanel/redis": "workspace:*",
     "@openpanel/validation": "workspace:*",
     "@prisma/client": "^5.1.1",
+    "prisma-json-types-generator": "^3.1.1",
     "ramda": "^0.29.1",
     "sqlstring": "^2.3.3",
     "superjson": "^1.13.3",

packages/db/prisma/migrations/20240925202841_notifications/migration.sql

@@ -0,0 +1,71 @@
+-- CreateEnum
+CREATE TYPE "IntegrationType" AS ENUM ('app', 'mail', 'custom');
+
+-- CreateTable
+CREATE TABLE "notification_settings" (
+    "id" UUID NOT NULL DEFAULT gen_random_uuid(),
+    "projectId" TEXT NOT NULL,
+    "settings" JSONB NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "notification_settings_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "notifications" (
+    "id" UUID NOT NULL DEFAULT gen_random_uuid(),
+    "projectId" TEXT NOT NULL,
+    "title" TEXT NOT NULL,
+    "message" TEXT NOT NULL,
+    "isReadAt" TIMESTAMP(3),
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "integrationId" UUID,
+    "integrationType" "IntegrationType" NOT NULL,
+
+    CONSTRAINT "notifications_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "integrations" (
+    "id" UUID NOT NULL DEFAULT gen_random_uuid(),
+    "name" TEXT NOT NULL,
+    "type" TEXT NOT NULL,
+    "config" JSONB NOT NULL,
+    "organizationId" TEXT NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "integrations_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "_IntegrationToNotificationControl" (
+    "A" UUID NOT NULL,
+    "B" UUID NOT NULL
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "_IntegrationToNotificationControl_AB_unique" ON "_IntegrationToNotificationControl"("A", "B");
+
+-- CreateIndex
+CREATE INDEX "_IntegrationToNotificationControl_B_index" ON "_IntegrationToNotificationControl"("B");
+
+-- AddForeignKey
+ALTER TABLE "notification_settings" ADD CONSTRAINT "notification_settings_projectId_fkey" FOREIGN KEY ("projectId") REFERENCES "projects"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "notifications" ADD CONSTRAINT "notifications_projectId_fkey" FOREIGN KEY ("projectId") REFERENCES "projects"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "notifications" ADD CONSTRAINT "notifications_integrationId_fkey" FOREIGN KEY ("integrationId") REFERENCES "integrations"("id") ON DELETE SET NULL ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "integrations" ADD CONSTRAINT "integrations_organizationId_fkey" FOREIGN KEY ("organizationId") REFERENCES "organizations"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "_IntegrationToNotificationControl" ADD CONSTRAINT "_IntegrationToNotificationControl_A_fkey" FOREIGN KEY ("A") REFERENCES "integrations"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "_IntegrationToNotificationControl" ADD CONSTRAINT "_IntegrationToNotificationControl_B_fkey" FOREIGN KEY ("B") REFERENCES "notification_settings"("id") ON DELETE CASCADE ON UPDATE CASCADE;

packages/db/prisma/migrations/20240925204316_notification_send_to_app_and_email/migration.sql

@@ -0,0 +1,14 @@
+/*
+  Warnings:
+
+  - You are about to drop the column `integrationType` on the `notifications` table. All the data in the column will be lost.
+
+*/
+-- AlterTable
+ALTER TABLE "notification_settings" ADD COLUMN     "sendToApp" BOOLEAN NOT NULL DEFAULT false,
+ADD COLUMN     "sendToEmail" BOOLEAN NOT NULL DEFAULT false;
+
+-- AlterTable
+ALTER TABLE "notifications" DROP COLUMN "integrationType",
+ADD COLUMN     "sendToApp" BOOLEAN NOT NULL DEFAULT false,
+ADD COLUMN     "sendToEmail" BOOLEAN NOT NULL DEFAULT false;

packages/db/prisma/migrations/20240926175415_renaming/migration.sql

@@ -0,0 +1,14 @@
+/*
+  Warnings:
+
+  - You are about to drop the column `type` on the `integrations` table. All the data in the column will be lost.
+  - You are about to drop the column `settings` on the `notification_settings` table. All the data in the column will be lost.
+  - Added the required column `config` to the `notification_settings` table without a default value. This is not possible if the table is not empty.
+
+*/
+-- AlterTable
+ALTER TABLE "integrations" DROP COLUMN "type";
+
+-- AlterTable
+ALTER TABLE "notification_settings" DROP COLUMN "settings",
+ADD COLUMN     "config" JSONB NOT NULL;

packages/db/prisma/migrations/20240927190558_rename_notification_controls/migration.sql

@@ -0,0 +1,55 @@
+/*
+  Warnings:
+
+  - You are about to drop the `_IntegrationToNotificationControl` table. If the table is not empty, all the data it contains will be lost.
+  - You are about to drop the `notification_settings` table. If the table is not empty, all the data it contains will be lost.
+
+*/
+-- DropForeignKey
+ALTER TABLE "_IntegrationToNotificationControl" DROP CONSTRAINT "_IntegrationToNotificationControl_A_fkey";
+
+-- DropForeignKey
+ALTER TABLE "_IntegrationToNotificationControl" DROP CONSTRAINT "_IntegrationToNotificationControl_B_fkey";
+
+-- DropForeignKey
+ALTER TABLE "notification_settings" DROP CONSTRAINT "notification_settings_projectId_fkey";
+
+-- DropTable
+DROP TABLE "_IntegrationToNotificationControl";
+
+-- DropTable
+DROP TABLE "notification_settings";
+
+-- CreateTable
+CREATE TABLE "notification_rules" (
+    "id" UUID NOT NULL DEFAULT gen_random_uuid(),
+    "projectId" TEXT NOT NULL,
+    "sendToApp" BOOLEAN NOT NULL DEFAULT false,
+    "sendToEmail" BOOLEAN NOT NULL DEFAULT false,
+    "config" JSONB NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "notification_rules_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "_IntegrationToNotificationRule" (
+    "A" UUID NOT NULL,
+    "B" UUID NOT NULL
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "_IntegrationToNotificationRule_AB_unique" ON "_IntegrationToNotificationRule"("A", "B");
+
+-- CreateIndex
+CREATE INDEX "_IntegrationToNotificationRule_B_index" ON "_IntegrationToNotificationRule"("B");
+
+-- AddForeignKey
+ALTER TABLE "notification_rules" ADD CONSTRAINT "notification_rules_projectId_fkey" FOREIGN KEY ("projectId") REFERENCES "projects"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "_IntegrationToNotificationRule" ADD CONSTRAINT "_IntegrationToNotificationRule_A_fkey" FOREIGN KEY ("A") REFERENCES "integrations"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "_IntegrationToNotificationRule" ADD CONSTRAINT "_IntegrationToNotificationRule_B_fkey" FOREIGN KEY ("B") REFERENCES "notification_rules"("id") ON DELETE CASCADE ON UPDATE CASCADE;

packages/db/prisma/migrations/20240927195752_add_name_to_rules/migration.sql

@@ -0,0 +1,8 @@
+/*
+  Warnings:
+
+  - Added the required column `name` to the `notification_rules` table without a default value. This is not possible if the table is not empty.
+
+*/
+-- AlterTable
+ALTER TABLE "notification_rules" ADD COLUMN     "name" TEXT NOT NULL;

packages/db/prisma/migrations/20241001215748_add_payload_to_notifications/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "notifications" ADD COLUMN     "payload" JSONB;

packages/db/prisma/schema.prisma

@@ -5,6 +5,10 @@ generator client {
   provider = "prisma-client-js"
 }
 
+generator json {
+  provider = "prisma-json-types-generator"
+}
+
 datasource db {
   provider  = "postgresql"
   url       = env("DATABASE_URL")
@@ -30,6 +34,7 @@ model Organization {
   Client          Client[]
   Dashboard       Dashboard[]
   ShareOverview   ShareOverview[]
+  integrations    Integration[]
 
   @@map("organizations")
 }
@@ -87,6 +92,9 @@ model Project {
   references Reference[]
   access     ProjectAccess[]
 
+  notificationRules NotificationRule[]
+  notifications     Notification[]
+
   createdAt DateTime @default(now())
   updatedAt DateTime @default(now()) @updatedAt
 
@@ -299,3 +307,59 @@ model Reference {
 
   @@map("references")
 }
+
+enum IntegrationType {
+  app
+  mail
+  custom
+}
+
+model NotificationRule {
+  id           String        @id @default(dbgenerated("gen_random_uuid()")) @db.Uuid
+  name         String
+  projectId    String
+  project      Project       @relation(fields: [projectId], references: [id])
+  integrations Integration[]
+  sendToApp    Boolean       @default(false)
+  sendToEmail  Boolean       @default(false)
+  /// [IPrismaNotificationRuleConfig]
+  config       Json
+  createdAt    DateTime      @default(now())
+  updatedAt    DateTime      @default(now()) @updatedAt
+
+  @@map("notification_rules")
+}
+
+model Notification {
+  id            String       @id @default(dbgenerated("gen_random_uuid()")) @db.Uuid
+  projectId     String
+  project       Project      @relation(fields: [projectId], references: [id])
+  title         String
+  message       String
+  isReadAt      DateTime?
+  createdAt     DateTime     @default(now())
+  updatedAt     DateTime     @default(now()) @updatedAt
+  sendToApp     Boolean      @default(false)
+  sendToEmail   Boolean      @default(false)
+  integration   Integration? @relation(fields: [integrationId], references: [id])
+  integrationId String?      @db.Uuid
+  /// [IPrismaNotificationPayload]
+  payload       Json?
+
+  @@map("notifications")
+}
+
+model Integration {
+  id                String             @id @default(dbgenerated("gen_random_uuid()")) @db.Uuid
+  name              String
+  /// [IPrismaIntegrationConfig]
+  config            Json
+  organization      Organization       @relation(fields: [organizationId], references: [id])
+  organizationId    String
+  notificationRules NotificationRule[]
+  notifications     Notification[]
+  createdAt         DateTime           @default(now())
+  updatedAt         DateTime           @default(now()) @updatedAt
+
+  @@map("integrations")
+}

packages/db/src/services/chart.service.ts

@@ -1,6 +1,9 @@
 import { escape } from 'sqlstring';
 
-import { getTimezoneFromDateString } from '@openpanel/common';
+import {
+  getTimezoneFromDateString,
+  stripLeadingAndTrailingSlashes,
+} from '@openpanel/common';
 import type {
   IChartEventFilter,
   IGetChartDataInput,
@@ -328,7 +331,10 @@ export function getEventFiltersWhereClause(filters: IChartEventFilter[]) {
         }
         case 'regex': {
           where[id] = value
-            .map((val) => `match(${name}, ${escape(String(val).trim())})`)
+            .map(
+              (val) =>
+                `match(${name}, ${escape(stripLeadingAndTrailingSlashes(String(val)).trim())})`,
+            )
             .join(' OR ');
           break;
         }

packages/db/src/services/event.service.ts

@@ -17,8 +17,8 @@ import type { EventMeta, Prisma } from '../prisma-client';
 import { db } from '../prisma-client';
 import { createSqlBuilder } from '../sql-builder';
 import { getEventFiltersWhereClause } from './chart.service';
-import { getProfiles, upsertProfile } from './profile.service';
 import type { IServiceProfile } from './profile.service';
+import { getProfiles, upsertProfile } from './profile.service';
 
 export type IImportedEvent = Omit<
   IClickhouseEvent,

packages/db/src/services/notification.service.ts

@@ -0,0 +1,314 @@
+import { stripLeadingAndTrailingSlashes } from '@openpanel/common';
+import { notificationQueue } from '@openpanel/queue';
+import { cacheable } from '@openpanel/redis';
+import type { IChartEvent, IChartEventFilter } from '@openpanel/validation';
+import { pathOr } from 'ramda';
+import {
+  type Integration,
+  type Notification,
+  Prisma,
+  db,
+} from '../prisma-client';
+import type {
+  IServiceCreateEventPayload,
+  IServiceEvent,
+} from './event.service';
+import { getProjectByIdCached } from './project.service';
+
+type ICreateNotification = Pick<
+  Notification,
+  'projectId' | 'title' | 'message' | 'integrationId' | 'payload'
+>;
+
+export type INotificationPayload =
+  | {
+      type: 'event';
+      event: IServiceCreateEventPayload;
+    }
+  | {
+      type: 'funnel';
+      funnel: IServiceEvent[];
+    };
+
+export const APP_NOTIFICATION_INTEGRATION_ID = 'app';
+export const EMAIL_NOTIFICATION_INTEGRATION_ID = 'email';
+
+export const BASE_INTEGRATIONS: Integration[] = [
+  {
+    id: APP_NOTIFICATION_INTEGRATION_ID,
+    name: 'App',
+    createdAt: new Date(),
+    updatedAt: new Date(),
+    config: {
+      type: APP_NOTIFICATION_INTEGRATION_ID,
+    },
+    organizationId: '',
+  },
+  {
+    id: EMAIL_NOTIFICATION_INTEGRATION_ID,
+    name: 'Email',
+    createdAt: new Date(),
+    updatedAt: new Date(),
+    config: {
+      type: EMAIL_NOTIFICATION_INTEGRATION_ID,
+    },
+    organizationId: '',
+  },
+];
+
+export const isBaseIntegration = (id: string) =>
+  BASE_INTEGRATIONS.find((i) => i.id === id);
+
+export const getNotificationRulesByProjectId = cacheable(
+  function getNotificationRulesByProjectId(projectId: string) {
+    return db.notificationRule.findMany({
+      where: {
+        projectId,
+      },
+      select: {
+        id: true,
+        name: true,
+        sendToApp: true,
+        sendToEmail: true,
+        config: true,
+        integrations: {
+          select: {
+            id: true,
+          },
+        },
+      },
+    });
+  },
+  60 * 24,
+);
+
+function getIntegration(integrationId: string | null) {
+  if (integrationId === APP_NOTIFICATION_INTEGRATION_ID) {
+    return {
+      integrationId: null,
+      sendToApp: true,
+      sendToEmail: false,
+    };
+  }
+
+  if (integrationId === EMAIL_NOTIFICATION_INTEGRATION_ID) {
+    return {
+      integrationId: null,
+      sendToApp: false,
+      sendToEmail: true,
+    };
+  }
+
+  return {
+    sendToApp: false,
+    sendToEmail: false,
+    integrationId,
+  };
+}
+
+export async function createNotification(notification: ICreateNotification) {
+  const res = await db.notification.create({
+    data: {
+      title: notification.title,
+      message: notification.message,
+      projectId: notification.projectId,
+      payload: notification.payload || Prisma.DbNull,
+      ...getIntegration(notification.integrationId),
+    },
+  });
+
+  return triggerNotification(res);
+}
+
+export function triggerNotification(notification: Notification) {
+  return notificationQueue.add('sendNotification', {
+    type: 'sendNotification',
+    payload: {
+      notification,
+    },
+  });
+}
+
+function matchEventFilters(
+  event: IServiceCreateEventPayload,
+  filters: IChartEventFilter[],
+) {
+  return filters.every((filter) => {
+    const { name, value, operator } = filter;
+
+    if (value.length === 0) return true;
+
+    if (name === 'has_profile') {
+      if (value.includes('true')) {
+        return event.profileId !== event.deviceId;
+      }
+      return event.profileId === event.deviceId;
+    }
+
+    const propertyValue = (
+      name.startsWith('properties.')
+        ? pathOr('', name.split('.'), event)
+        : pathOr('', [name], event)
+    ).trim();
+
+    switch (operator) {
+      case 'is':
+        return value.includes(propertyValue);
+      case 'isNot':
+        return !value.includes(propertyValue);
+      case 'contains':
+        return value.some((val) => propertyValue.includes(String(val)));
+      case 'doesNotContain':
+        return !value.some((val) => propertyValue.includes(String(val)));
+      case 'startsWith':
+        return value.some((val) => propertyValue.startsWith(String(val)));
+      case 'endsWith':
+        return value.some((val) => propertyValue.endsWith(String(val)));
+      case 'regex': {
+        return value
+          .map((val) => stripLeadingAndTrailingSlashes(String(val)))
+          .some((val) => new RegExp(val).test(propertyValue));
+      }
+      default:
+        return false;
+    }
+  });
+}
+
+export function matchEvent(
+  event: IServiceCreateEventPayload,
+  chartEvent: IChartEvent,
+) {
+  if (event.name !== chartEvent.name) {
+    return false;
+  }
+
+  if (chartEvent.filters.length > 0) {
+    return matchEventFilters(event, chartEvent.filters);
+  }
+
+  return true;
+}
+
+export async function checkNotificationRulesForEvent(
+  payload: IServiceCreateEventPayload,
+) {
+  const project = await getProjectByIdCached(payload.projectId);
+  const rules = await getNotificationRulesByProjectId(payload.projectId);
+  await Promise.all(
+    rules.flatMap((rule) => {
+      if (rule.config.type === 'events') {
+        const match = rule.config.events.find((event) => {
+          return matchEvent(payload, event);
+        });
+
+        if (!match) {
+          return [];
+        }
+
+        const notification = {
+          title: `You received a new "${payload.name}" event`,
+          message: project?.name ? `Project: ${project?.name}` : '',
+          projectId: payload.projectId,
+          payload: {
+            type: 'event',
+            event: payload,
+          },
+        } as const;
+
+        const promises = rule.integrations.map((integration) =>
+          createNotification({
+            ...notification,
+            integrationId: integration.id,
+          }),
+        );
+
+        if (rule.sendToApp) {
+          promises.push(
+            createNotification({
+              ...notification,
+              integrationId: APP_NOTIFICATION_INTEGRATION_ID,
+            }),
+          );
+        }
+
+        if (rule.sendToEmail) {
+          promises.push(
+            createNotification({
+              ...notification,
+              integrationId: EMAIL_NOTIFICATION_INTEGRATION_ID,
+            }),
+          );
+        }
+
+        return promises;
+      }
+    }),
+  );
+}
+
+export async function checkNotificationRulesForSessionEnd(
+  events: IServiceEvent[],
+) {
+  const sortedEvents = events.sort(
+    (a, b) => a.createdAt.getTime() - b.createdAt.getTime(),
+  );
+  const projectId = sortedEvents[0]?.projectId;
+  if (!projectId) return null;
+
+  const [project, rules] = await Promise.all([
+    getProjectByIdCached(projectId),
+    getNotificationRulesByProjectId(projectId),
+  ]);
+
+  const funnelRules = rules.filter((rule) => rule.config.type === 'funnel');
+
+  const notificationPromises = funnelRules.flatMap((rule) => {
+    // Match funnel events
+    let funnelIndex = 0;
+    const matchedEvents = [];
+    for (const event of sortedEvents) {
+      if (matchEvent(event, rule.config.events[funnelIndex]!)) {
+        matchedEvents.push(event);
+        funnelIndex++;
+        if (funnelIndex === rule.config.events.length) break;
+      }
+    }
+
+    // If funnel not completed, skip this rule
+    if (funnelIndex < rule.config.events.length) return [];
+
+    // Create notification object
+    const notification = {
+      title: `Funnel "${rule.name}" completed`,
+      message: project?.name ? `Project: ${project?.name}` : '',
+      projectId,
+      payload: { type: 'funnel', funnel: matchedEvents } as const,
+    };
+
+    // Generate notification promises
+    return [
+      ...rule.integrations.map((integration) =>
+        createNotification({ ...notification, integrationId: integration.id }),
+      ),
+      ...(rule.sendToApp
+        ? [
+            createNotification({
+              ...notification,
+              integrationId: APP_NOTIFICATION_INTEGRATION_ID,
+            }),
+          ]
+        : []),
+      ...(rule.sendToEmail
+        ? [
+            createNotification({
+              ...notification,
+              integrationId: EMAIL_NOTIFICATION_INTEGRATION_ID,
+            }),
+          ]
+        : []),
+    ];
+  });
+
+  await Promise.all(notificationPromises);
+}

packages/db/src/services/project.service.ts

@@ -1,5 +1,6 @@
 import { auth } from '@clerk/nextjs/server';
 
+import { cacheable } from '@openpanel/redis';
 import type { Prisma, Project } from '../prisma-client';
 import { db } from '../prisma-client';
 
@@ -24,6 +25,8 @@ export async function getProjectById(id: string) {
   return res;
 }
 
+export const getProjectByIdCached = cacheable(getProjectById, 60 * 60 * 24);
+
 export async function getProjectWithClients(id: string) {
   const res = await db.project.findUnique({
     where: {

packages/db/src/types.ts

@@ -0,0 +1,13 @@
+import type {
+  IIntegrationConfig,
+  INotificationRuleConfig,
+} from '@openpanel/validation';
+import type { INotificationPayload } from './services/notification.service';
+
+declare global {
+  namespace PrismaJson {
+    type IPrismaNotificationRuleConfig = INotificationRuleConfig;
+    type IPrismaIntegrationConfig = IIntegrationConfig;
+    type IPrismaNotificationPayload = INotificationPayload;
+  }
+}

packages/db/tsconfig.json

@@ -7,6 +7,6 @@
     },
     "tsBuildInfoFile": "node_modules/.cache/tsbuildinfo.json"
   },
-  "include": ["."],
+  "include": [".", "./src/types.ts"],
   "exclude": ["node_modules"]
 }

packages/integrations/index.ts

@@ -0,0 +1,2 @@
+// Empty, import directly from src/
+export {};

packages/integrations/package.json

@@ -0,0 +1,18 @@
+{
+  "name": "@openpanel/integrations",
+  "version": "0.0.1",
+  "main": "index.ts",
+  "scripts": {
+    "typecheck": "tsc --noEmit"
+  },
+  "dependencies": {
+    "@slack/bolt": "^3.18.0",
+    "@slack/oauth": "^3.0.0"
+  },
+  "devDependencies": {
+    "@openpanel/tsconfig": "workspace:*",
+    "@openpanel/validation": "workspace:*",
+    "@types/node": "^18.16.0",
+    "typescript": "^5.2.2"
+  }
+}

packages/integrations/src/discord.ts

@@ -0,0 +1,34 @@
+// Cred to (@OpenStatusHQ) https://github.com/openstatusHQ/openstatus/blob/main/packages/notifications/discord/src/index.ts
+
+export function sendDiscordNotification({
+  webhookUrl,
+  message,
+}: {
+  webhookUrl: string;
+  message: string;
+}) {
+  return fetch(webhookUrl, {
+    method: 'POST',
+    headers: {
+      'Content-Type': 'application/json',
+    },
+    body: JSON.stringify({
+      content: message,
+      avatar_url: 'https://openpanel.dev/logo.jpg',
+      username: 'OpenPanel Notifications',
+    }),
+  }).catch((err) => {
+    return {
+      ok: false,
+      json: () => Promise.resolve({}),
+    };
+  });
+}
+
+export function sendTestDiscordNotification(webhookUrl: string) {
+  return sendDiscordNotification({
+    webhookUrl,
+    message:
+      '**🧪 Test [OpenPanel.dev](<https://openpanel.dev/>)**\nIf you can read this, your Slack webhook is functioning correctly!\n',
+  });
+}

packages/integrations/src/slack.ts

@@ -0,0 +1,50 @@
+// Cred to (@c_alares) https://github.com/christianalares/seventy-seven/blob/main/packages/integrations/src/slack/index.ts
+
+import { LogLevel, App as SlackApp } from '@slack/bolt';
+import { InstallProvider } from '@slack/oauth';
+
+const SLACK_CLIENT_ID = process.env.SLACK_CLIENT_ID;
+const SLACK_CLIENT_SECRET = process.env.SLACK_CLIENT_SECRET;
+const SLACK_OAUTH_REDIRECT_URL = process.env.SLACK_OAUTH_REDIRECT_URL;
+const SLACK_STATE_SECRET = process.env.SLACK_STATE_SECRET;
+
+export const slackInstaller = new InstallProvider({
+  clientId: SLACK_CLIENT_ID!,
+  clientSecret: SLACK_CLIENT_SECRET!,
+  stateSecret: SLACK_STATE_SECRET,
+  logLevel: process.env.NODE_ENV === 'development' ? LogLevel.DEBUG : undefined,
+});
+
+export const getSlackInstallUrl = ({
+  integrationId,
+  organizationId,
+}: { integrationId: string; organizationId: string }) => {
+  return slackInstaller.generateInstallUrl({
+    scopes: [
+      'incoming-webhook',
+      'chat:write',
+      'chat:write.public',
+      'team:read',
+    ],
+    redirectUri: SLACK_OAUTH_REDIRECT_URL,
+    metadata: JSON.stringify({ integrationId, organizationId }),
+  });
+};
+
+export function sendSlackNotification({
+  webhookUrl,
+  message,
+}: {
+  webhookUrl: string;
+  message: string;
+}) {
+  return fetch(webhookUrl, {
+    method: 'POST',
+    headers: {
+      'Content-Type': 'application/json',
+    },
+    body: JSON.stringify({
+      text: message,
+    }),
+  });
+}

packages/integrations/tsconfig.json

@@ -0,0 +1,12 @@
+{
+  "extends": "@openpanel/tsconfig/base.json",
+  "compilerOptions": {
+    "baseUrl": ".",
+    "paths": {
+      "@/*": ["./src/*"],
+    },
+    "tsBuildInfoFile": "node_modules/.cache/tsbuildinfo.json"
+  },
+  "include": ["."],
+  "exclude": ["node_modules"]
+}

packages/queue/index.ts

@@ -1,9 +1,4 @@
-export {
-  eventsQueue,
-  cronQueue,
-  sessionsQueue,
-  sessionsQueueEvents,
-} from './src/queues';
+export * from './src/queues';
 export type * from './src/queues';
 export { findJobByPrefix } from './src/utils';
 export type { JobsOptions } from 'bullmq';

packages/queue/src/queues.ts

@@ -1,6 +1,6 @@
 import { Queue, QueueEvents } from 'bullmq';
 
-import type { IServiceEvent } from '@openpanel/db';
+import type { IServiceEvent, Notification } from '@openpanel/db';
 import { getRedisQueue } from '@openpanel/redis';
 import type { TrackPayload } from '@openpanel/sdk';
 
@@ -90,3 +90,20 @@ export const cronQueue = new Queue<CronQueuePayload>('cron', {
     removeOnComplete: 10,
   },
 });
+
+export type NotificationQueuePayload = {
+  type: 'sendNotification';
+  payload: {
+    notification: Notification;
+  };
+};
+
+export const notificationQueue = new Queue<NotificationQueuePayload>(
+  'notification',
+  {
+    connection: getRedisQueue(),
+    defaultJobOptions: {
+      removeOnComplete: 10,
+    },
+  },
+);

packages/trpc/package.json

@@ -10,8 +10,9 @@
     "@openpanel/common": "workspace:*",
     "@openpanel/constants": "workspace:*",
     "@openpanel/db": "workspace:*",
-    "@openpanel/validation": "workspace:*",
+    "@openpanel/integrations": "workspace:^",
     "@openpanel/redis": "workspace:*",
+    "@openpanel/validation": "workspace:*",
     "@seventy-seven/sdk": "0.0.0-beta.2",
     "@trpc/server": "^10.45.1",
     "date-fns": "^3.3.1",

packages/trpc/src/root.ts

@@ -2,6 +2,8 @@ import { chartRouter } from './routers/chart';
 import { clientRouter } from './routers/client';
 import { dashboardRouter } from './routers/dashboard';
 import { eventRouter } from './routers/event';
+import { integrationRouter } from './routers/integration';
+import { notificationRouter } from './routers/notification';
 import { onboardingRouter } from './routers/onboarding';
 import { organizationRouter } from './routers/organization';
 import { profileRouter } from './routers/profile';
@@ -32,6 +34,8 @@ export const appRouter = createTRPCRouter({
   onboarding: onboardingRouter,
   reference: referenceRouter,
   ticket: ticketRouter,
+  notification: notificationRouter,
+  integration: integrationRouter,
 });
 
 // export type definition of API

packages/trpc/src/routers/integration.ts

@@ -0,0 +1,137 @@
+import { z } from 'zod';
+
+import { BASE_INTEGRATIONS, db } from '@openpanel/db';
+
+import { getSlackInstallUrl } from '@openpanel/integrations/src/slack';
+import {
+  type ISlackConfig,
+  zCreateDiscordIntegration,
+  zCreateSlackIntegration,
+  zCreateWebhookIntegration,
+} from '@openpanel/validation';
+import { getOrganizationAccessCached } from '../access';
+import { TRPCAccessError } from '../errors';
+import { createTRPCRouter, protectedProcedure } from '../trpc';
+
+export const integrationRouter = createTRPCRouter({
+  get: protectedProcedure
+    .input(z.object({ id: z.string() }))
+    .query(async ({ input, ctx }) => {
+      const integration = await db.integration.findUniqueOrThrow({
+        where: {
+          id: input.id,
+        },
+      });
+
+      const access = await getOrganizationAccessCached({
+        userId: ctx.session.userId,
+        organizationId: integration.organizationId,
+      });
+
+      if (!access) {
+        throw TRPCAccessError('You do not have access to this project');
+      }
+
+      return integration;
+    }),
+  list: protectedProcedure
+    .input(z.object({ organizationId: z.string() }))
+    .query(async ({ input }) => {
+      const integrations = await db.integration.findMany({
+        where: {
+          organizationId: input.organizationId,
+        },
+      });
+
+      return [...BASE_INTEGRATIONS, ...integrations];
+    }),
+  createOrUpdateSlack: protectedProcedure
+    .input(zCreateSlackIntegration)
+    .mutation(async ({ input }) => {
+      if (input.id) {
+        const res = await db.integration.update({
+          where: {
+            id: input.id,
+            organizationId: input.organizationId,
+          },
+          data: {
+            name: input.name,
+            // This is empty and will be filled by the webhook
+            config: {} as ISlackConfig,
+          },
+        });
+
+        return {
+          ...res,
+          slackInstallUrl: await getSlackInstallUrl({
+            integrationId: res.id,
+            organizationId: input.organizationId,
+          }),
+        };
+      }
+
+      const res = await db.integration.create({
+        data: {
+          name: input.name,
+          organizationId: input.organizationId,
+          // This is empty and will be filled by the webhook
+          config: {} as ISlackConfig,
+        },
+      });
+
+      return {
+        ...res,
+        slackInstallUrl: await getSlackInstallUrl({
+          integrationId: res.id,
+          organizationId: input.organizationId,
+        }),
+      };
+    }),
+  createOrUpdate: protectedProcedure
+    .input(z.union([zCreateDiscordIntegration, zCreateWebhookIntegration]))
+    .mutation(async ({ input }) => {
+      if (input.id) {
+        return db.integration.update({
+          where: {
+            id: input.id,
+            organizationId: input.organizationId,
+          },
+          data: {
+            name: input.name,
+            config: input.config,
+          },
+        });
+      }
+      return db.integration.create({
+        data: {
+          name: input.name,
+          organizationId: input.organizationId,
+          config: input.config,
+        },
+      });
+    }),
+  delete: protectedProcedure
+    .input(z.object({ id: z.string() }))
+    .mutation(async ({ input: { id }, ctx }) => {
+      const integration = await db.integration.findUniqueOrThrow({
+        where: {
+          id,
+        },
+      });
+
+      const access = await getOrganizationAccessCached({
+        userId: ctx.session.userId,
+        organizationId: integration.organizationId,
+      });
+
+      if (!access) {
+        throw TRPCAccessError('You do not have access to this project');
+      }
+
+      return db.integration.delete({
+        where: {
+          id,
+        },
+      });
+    }),
+});

packages/trpc/src/routers/notification.ts

@@ -0,0 +1,150 @@
+import { z } from 'zod';
+
+import {
+  APP_NOTIFICATION_INTEGRATION_ID,
+  BASE_INTEGRATIONS,
+  EMAIL_NOTIFICATION_INTEGRATION_ID,
+  db,
+  getNotificationRulesByProjectId,
+  isBaseIntegration,
+} from '@openpanel/db';
+import { zCreateNotificationRule } from '@openpanel/validation';
+
+import { getProjectAccess } from '../access';
+import { TRPCAccessError } from '../errors';
+import { createTRPCRouter, protectedProcedure } from '../trpc';
+
+export const notificationRouter = createTRPCRouter({
+  list: protectedProcedure
+    .input(z.object({ projectId: z.string() }))
+    .query(async ({ input }) => {
+      return db.notification.findMany({
+        where: {
+          projectId: input.projectId,
+          sendToApp: true,
+        },
+        orderBy: {
+          createdAt: 'desc',
+        },
+        include: {
+          integration: {
+            include: {
+              notificationRules: {
+                select: {
+                  name: true,
+                },
+              },
+            },
+          },
+        },
+        take: 100,
+      });
+    }),
+  rules: protectedProcedure
+    .input(z.object({ projectId: z.string() }))
+    .query(async ({ input }) => {
+      return db.notificationRule
+        .findMany({
+          where: {
+            projectId: input.projectId,
+          },
+          include: {
+            integrations: true,
+          },
+          orderBy: {
+            createdAt: 'desc',
+          },
+        })
+        .then((rules) => {
+          return rules.map((rule) => {
+            return {
+              ...rule,
+              integrations: [
+                ...BASE_INTEGRATIONS.filter((integration) => {
+                  return (
+                    (integration.id === APP_NOTIFICATION_INTEGRATION_ID &&
+                      rule.sendToApp) ||
+                    (integration.id === EMAIL_NOTIFICATION_INTEGRATION_ID &&
+                      rule.sendToEmail)
+                  );
+                }),
+                ...rule.integrations,
+              ],
+            };
+          });
+        });
+    }),
+  createOrUpdateRule: protectedProcedure
+    .input(zCreateNotificationRule)
+    .mutation(async ({ input }) => {
+      // Clear the cache for the project
+      await getNotificationRulesByProjectId.clear(input.projectId);
+
+      if (input.id) {
+        return db.notificationRule.update({
+          where: {
+            id: input.id,
+          },
+          data: {
+            name: input.name,
+            projectId: input.projectId,
+            sendToApp: !!input.integrations.find(
+              (id) => id === APP_NOTIFICATION_INTEGRATION_ID,
+            ),
+            sendToEmail: !!input.integrations.find(
+              (id) => id === EMAIL_NOTIFICATION_INTEGRATION_ID,
+            ),
+            integrations: {
+              set: input.integrations
+                .filter((id) => !isBaseIntegration(id))
+                .map((id) => ({ id })),
+            },
+            config: input.config,
+          },
+        });
+      }
+
+      return db.notificationRule.create({
+        data: {
+          name: input.name,
+          projectId: input.projectId,
+          sendToApp: !!input.integrations.find(
+            (id) => id === APP_NOTIFICATION_INTEGRATION_ID,
+          ),
+          sendToEmail: !!input.integrations.find(
+            (id) => id === EMAIL_NOTIFICATION_INTEGRATION_ID,
+          ),
+          integrations: {
+            connect: input.integrations
+              .filter((id) => !isBaseIntegration(id))
+              .map((id) => ({ id })),
+          },
+          config: input.config,
+        },
+      });
+    }),
+  deleteRule: protectedProcedure
+    .input(z.object({ id: z.string() }))
+    .mutation(async ({ input: { id }, ctx }) => {
+      const rule = await db.notificationRule.findUniqueOrThrow({
+        where: {
+          id,
+        },
+      });
+
+      const access = await getProjectAccess({
+        userId: ctx.session.userId,
+        projectId: rule.projectId,
+      });
+
+      if (!access) {
+        throw TRPCAccessError('You do not have access to this project');
+      }
+
+      return db.notificationRule.delete({
+        where: {
+          id,
+        },
+      });
+    }),
+});

packages/trpc/src/routers/project.ts

@@ -1,6 +1,11 @@
 import { z } from 'zod';
 
-import { db, getId, getProjectsByOrganizationSlug } from '@openpanel/db';
+import {
+  db,
+  getId,
+  getProjectByIdCached,
+  getProjectsByOrganizationSlug,
+} from '@openpanel/db';
 
 import { getProjectAccess } from '../access';
 import { TRPCAccessError } from '../errors';
@@ -34,8 +39,7 @@ export const projectRouter = createTRPCRouter({
       if (!access) {
         throw TRPCAccessError('You do not have access to this project');
       }
-
-      return db.project.update({
+      const res = await db.project.update({
         where: {
           id: input.id,
         },
@@ -43,6 +47,8 @@ export const projectRouter = createTRPCRouter({
           name: input.name,
         },
       });
+      await getProjectByIdCached.clear(input.id);
+      return res;
     }),
   create: protectedProcedure
     .input(

packages/validation/src/index.ts

@@ -148,3 +148,119 @@ export const zOnboardingProject = z
       }
     }
   });
+
+export const zSlackAuthResponse = z.object({
+  ok: z.literal(true),
+  app_id: z.string(),
+  authed_user: z.object({
+    id: z.string(),
+  }),
+  scope: z.string(),
+  token_type: z.literal('bot'),
+  access_token: z.string(),
+  bot_user_id: z.string(),
+  team: z.object({
+    id: z.string(),
+    name: z.string(),
+  }),
+  incoming_webhook: z.object({
+    channel: z.string(),
+    channel_id: z.string(),
+    configuration_url: z.string().url(),
+    url: z.string().url(),
+  }),
+});
+
+export const zSlackConfig = z
+  .object({
+    type: z.literal('slack'),
+  })
+  .merge(zSlackAuthResponse);
+
+export type ISlackConfig = z.infer<typeof zSlackConfig>;
+
+export const zWebhookConfig = z.object({
+  type: z.literal('webhook'),
+  url: z.string().url(),
+  headers: z.record(z.string()),
+  payload: z.record(z.string(), z.unknown()),
+});
+export type IWebhookConfig = z.infer<typeof zWebhookConfig>;
+
+export const zDiscordConfig = z.object({
+  type: z.literal('discord'),
+  url: z.string().url(),
+});
+export type IDiscordConfig = z.infer<typeof zDiscordConfig>;
+
+export const zAppConfig = z.object({
+  type: z.literal('app'),
+});
+export type IAppConfig = z.infer<typeof zAppConfig>;
+
+export const zEmailConfig = z.object({
+  type: z.literal('email'),
+});
+export type IEmailConfig = z.infer<typeof zEmailConfig>;
+
+export type IIntegrationConfig =
+  | ISlackConfig
+  | IDiscordConfig
+  | IWebhookConfig
+  | IAppConfig
+  | IEmailConfig;
+
+const zCreateIntegration = z.object({
+  id: z.string().optional(),
+  name: z.string().min(1),
+  organizationId: z.string().min(1),
+});
+
+export const zCreateSlackIntegration = zCreateIntegration;
+
+export const zCreateWebhookIntegration = zCreateIntegration.merge(
+  z.object({
+    config: zWebhookConfig,
+  }),
+);
+
+export const zCreateDiscordIntegration = zCreateIntegration.merge(
+  z.object({
+    config: zDiscordConfig,
+  }),
+);
+
+export const zNotificationRuleEventConfig = z.object({
+  type: z.literal('events'),
+  events: z.array(zChartEvent),
+});
+
+export type INotificationRuleEventConfig = z.infer<
+  typeof zNotificationRuleEventConfig
+>;
+
+export const zNotificationRuleFunnelConfig = z.object({
+  type: z.literal('funnel'),
+  events: z.array(zChartEvent).min(1),
+});
+
+export type INotificationRuleFunnelConfig = z.infer<
+  typeof zNotificationRuleFunnelConfig
+>;
+
+export const zNotificationRuleConfig = z.discriminatedUnion('type', [
+  zNotificationRuleEventConfig,
+  zNotificationRuleFunnelConfig,
+]);
+
+export type INotificationRuleConfig = z.infer<typeof zNotificationRuleConfig>;
+
+export const zCreateNotificationRule = z.object({
+  id: z.string().optional(),
+  name: z.string().min(1),
+  config: zNotificationRuleConfig,
+  integrations: z.array(z.string()),
+  sendToApp: z.boolean(),
+  sendToEmail: z.boolean(),
+  projectId: z.string(),
+});