feat: add customer event (#7029)

packages/backend/server/src/config/affine.env.ts

@@ -26,6 +26,7 @@ AFFiNE.ENV_MAP = {
   MAILER_SECURE: ['mailer.secure', 'boolean'],
   THROTTLE_TTL: ['rateLimiter.ttl', 'int'],
   THROTTLE_LIMIT: ['rateLimiter.limit', 'int'],
+  METRICS_CUSTOMER_IO_TOKEN: ['metrics.customerIo.token', 'string'],
   COPILOT_OPENAI_API_KEY: 'plugins.copilot.openai.apiKey',
   COPILOT_FAL_API_KEY: 'plugins.copilot.fal.apiKey',
   COPILOT_UNSPLASH_API_KEY: 'plugins.copilot.unsplashKey',

packages/backend/server/src/core/auth/service.ts

@@ -377,7 +377,10 @@ export class AuthService implements OnApplicationBootstrap {
     });
   }
 
-  async changePassword(id: string, newPassword: string): Promise<User> {
+  async changePassword(
+    id: string,
+    newPassword: string
+  ): Promise<Omit<User, 'password'>> {
     const user = await this.user.findUserById(id);
 
     if (!user) {
@@ -386,46 +389,31 @@ export class AuthService implements OnApplicationBootstrap {
 
     const hashedPassword = await this.crypto.encryptPassword(newPassword);
 
-    return this.db.user.update({
-      where: {
-        id: user.id,
-      },
-      data: {
-        password: hashedPassword,
-      },
-    });
+    return this.user.updateUser(user.id, { password: hashedPassword });
   }
 
-  async changeEmail(id: string, newEmail: string): Promise<User> {
+  async changeEmail(
+    id: string,
+    newEmail: string
+  ): Promise<Omit<User, 'password'>> {
     const user = await this.user.findUserById(id);
 
     if (!user) {
       throw new BadRequestException('Invalid email');
     }
 
-    return this.db.user.update({
-      where: {
-        id,
-      },
-      data: {
-        email: newEmail,
-        emailVerifiedAt: new Date(),
-      },
+    return this.user.updateUser(id, {
+      email: newEmail,
+      emailVerifiedAt: new Date(),
     });
   }
 
   async setEmailVerified(id: string) {
-    return await this.db.user.update({
-      where: {
-        id,
-      },
-      data: {
-        emailVerifiedAt: new Date(),
-      },
-      select: {
-        emailVerifiedAt: true,
-      },
-    });
+    return await this.user.updateUser(
+      id,
+      { emailVerifiedAt: new Date() },
+      { emailVerifiedAt: true }
+    );
   }
 
   async sendChangePasswordEmail(email: string, callbackUrl: string) {

packages/backend/server/src/core/user/resolver.ts

@@ -117,7 +117,7 @@ export class UserResolver {
       throw new BadRequestException(`User not found`);
     }
 
-    const link = await this.storage.put(
+    const avatarUrl = await this.storage.put(
       `${user.id}-avatar`,
       avatar.createReadStream(),
       {
@@ -125,12 +125,7 @@ export class UserResolver {
       }
     );
 
-    return this.prisma.user.update({
-      where: { id: user.id },
-      data: {
-        avatarUrl: link,
-      },
-    });
+    return this.users.updateUser(user.id, { avatarUrl });
   }
 
   @Mutation(() => UserType, {
@@ -146,12 +141,7 @@ export class UserResolver {
       return user;
     }
 
-    return sessionUser(
-      await this.prisma.user.update({
-        where: { id: user.id },
-        data: input,
-      })
-    );
+    return sessionUser(await this.users.updateUser(user.id, input));
   }
 
   @Mutation(() => RemoveAvatar, {
@@ -162,10 +152,7 @@ export class UserResolver {
     if (!user) {
       throw new BadRequestException(`User not found`);
     }
-    await this.prisma.user.update({
-      where: { id: user.id },
-      data: { avatarUrl: null },
-    });
+    await this.users.updateUser(user.id, { avatarUrl: null });
     return { success: true };
   }
 

packages/backend/server/src/core/user/service.ts

@@ -1,10 +1,18 @@
-import { BadRequestException, Injectable } from '@nestjs/common';
+import { BadRequestException, Injectable, Logger } from '@nestjs/common';
 import { Prisma, PrismaClient } from '@prisma/client';
 
+import {
+  Config,
+  EventEmitter,
+  type EventPayload,
+  OnEvent,
+} from '../../fundamentals';
 import { Quota_FreePlanV1_1 } from '../quota/schema';
 
 @Injectable()
 export class UserService {
+  private readonly logger = new Logger(UserService.name);
+
   defaultUserSelect = {
     id: true,
     name: true,
@@ -12,9 +20,14 @@ export class UserService {
     emailVerifiedAt: true,
     avatarUrl: true,
     registered: true,
+    createdAt: true,
   } satisfies Prisma.UserSelect;
 
-  constructor(private readonly prisma: PrismaClient) {}
+  constructor(
+    private readonly config: Config,
+    private readonly prisma: PrismaClient,
+    private readonly emitter: EventEmitter
+  ) {}
 
   get userCreatingData() {
     return {
@@ -139,10 +152,75 @@ export class UserService {
       }
     }
 
+    this.emitter.emit('user.updated', user);
+
+    return user;
+  }
+
+  async updateUser(
+    id: string,
+    data: Prisma.UserUpdateInput,
+    select: Prisma.UserSelect = this.defaultUserSelect
+  ) {
+    const user = await this.prisma.user.update({ where: { id }, data, select });
+
+    this.emitter.emit('user.updated', user);
+
     return user;
   }
 
   async deleteUser(id: string) {
     return this.prisma.user.delete({ where: { id } });
   }
+
+  @OnEvent('user.updated')
+  async onUserUpdated(user: EventPayload<'user.deleted'>) {
+    const { enabled, customerIo } = this.config.metrics;
+    if (enabled && customerIo?.token) {
+      const payload = {
+        name: user.name,
+        email: user.email,
+        created_at: Number(user.createdAt),
+      };
+      try {
+        await fetch(`https://track.customer.io/api/v1/customers/${user.id}`, {
+          method: 'PUT',
+          headers: {
+            Authorization: `Basic ${customerIo.token}`,
+            'Content-Type': 'application/json',
+          },
+          body: JSON.stringify(payload),
+        });
+      } catch (e) {
+        this.logger.error('Failed to publish user update event:', e);
+      }
+    }
+  }
+
+  @OnEvent('user.deleted')
+  async onUserDeleted(user: EventPayload<'user.deleted'>) {
+    const { enabled, customerIo } = this.config.metrics;
+    if (enabled && customerIo?.token) {
+      try {
+        if (user.emailVerifiedAt) {
+          // suppress email if email is verified
+          await fetch(
+            `https://track.customer.io/api/v1/customers/${user.email}/suppress`,
+            {
+              method: 'POST',
+              headers: {
+                Authorization: `Basic ${customerIo.token}`,
+              },
+            }
+          );
+        }
+        await fetch(`https://track.customer.io/api/v1/customers/${user.id}`, {
+          method: 'DELETE',
+          headers: { Authorization: `Basic ${customerIo.token}` },
+        });
+      } catch (e) {
+        this.logger.error('Failed to publish user delete event:', e);
+      }
+    }
+  }
 }

packages/backend/server/src/data/app.ts

@@ -15,6 +15,7 @@ import { RevertCommand, RunCommand } from './commands/run';
       },
       metrics: {
         enabled: false,
+        customerIo: {},
       },
     }),
     BusinessAppModule,

packages/backend/server/src/fundamentals/config/def.ts

@@ -340,6 +340,9 @@ export interface AFFiNEConfig {
 
   metrics: {
     enabled: boolean;
+    customerIo: {
+      token: string;
+    };
   };
 
   telemetry: {

packages/backend/server/src/fundamentals/config/default.ts

@@ -188,6 +188,9 @@ export const getDefaultAFFiNEConfig: () => AFFiNEConfig = () => {
     },
     metrics: {
       enabled: false,
+      customerIo: {
+        token: '',
+      },
     },
     telemetry: {
       enabled: isSelfhosted,

packages/backend/server/src/fundamentals/event/def.ts

@@ -22,6 +22,7 @@ export interface DocEvents {
 }
 
 export interface UserEvents {
+  updated: Payload<Omit<User, 'password'>>;
   deleted: Payload<User>;
 }