feat: 接入消息系统 (#11)

功能介绍：https://jgf29kqp7z.feishu.cn/wiki/WJYKwvLVyidOK2kuUovcittTn6e
配置介绍：https://jgf29kqp7z.feishu.cn/wiki/ODXAwD3xYiND6mkUBHwcdG8Ense

.changeset/friendly-donkeys-play.md

@@ -0,0 +1,17 @@
+---
+"@scow/notification": minor
+"@scow/notification-protos": patch
+"@scow/lib-operation-log": patch
+"@scow/lib-notification": patch
+"@scow/scowd-protos": patch
+"@scow/mis-server": patch
+"@scow/portal-web": patch
+"@scow/gateway": patch
+"@scow/mis-web": patch
+"@scow/config": patch
+"@scow/cli": patch
+"@scow/lib-web": patch
+"@scow/grpc-api": patch
+---
+
+新增消息系统

apps/cli/src/cmd/updateCli.ts

@@ -30,7 +30,7 @@ interface Options {
 }
 
 const owner = "PKUHPC";
-const repo = "SCOW";
+const repo = "private-scow";
 const workflow_id = "test-build-publish.yaml";
 
 const allowedArch = ["x64", "arm64"];

apps/cli/src/compose/index.ts

@@ -48,6 +48,9 @@ export const createComposeSpec = (config: InstallConfigSchema) => {
   const AI_PATH = config.ai?.basePath || "/ai";
   checkPathFormat("ai.basePath", AI_PATH);
 
+  const NOTIFICATION_PATH = config.notification?.basePath || "/notification";
+  checkPathFormat("notification.basePath", NOTIFICATION_PATH);
+
   const serviceLogEnv = {
     LOG_LEVEL: config.log.level,
     LOG_PRETTY: String(config.log.pretty),
@@ -132,6 +135,7 @@ export const createComposeSpec = (config: InstallConfigSchema) => {
       "PORTAL_PATH": PORTAL_PATH,
       "MIS_PATH": MIS_PATH,
       "AI_PATH": AI_PATH,
+      "NOTIFICATION_PATH": NOTIFICATION_PATH,
       "CLIENT_MAX_BODY_SIZE": config.gateway.uploadFileSizeLimit,
       "PROXY_READ_TIMEOUT": config.gateway.proxyReadTimeout,
       "PUBLIC_PATH": publicPath,
@@ -443,5 +447,29 @@ export const createComposeSpec = (config: InstallConfigSchema) => {
     });
   }
 
+  if (config.notification) {
+    addService("notification", {
+      image: scowImage,
+      ports: {},
+      environment: {
+        "SCOW_LAUNCH_APP": "notification",
+        "NEXT_PUBLIC_BASE_PATH": join(BASE_PATH, NOTIFICATION_PATH),
+        "MIS_SERVER_URL": config.mis ? "mis-server:5000" : "",
+        "DB_PASSWORD": config.mis?.dbPassword ?? "",
+        "AUTH_EXTERNAL_URL": config.auth.custom?.external?.url || join(BASE_PATH, "/auth"),
+        "AUTH_INTERNAL_URL": authUrl || "http://auth:5000",
+        "PUBLIC_PATH": join(BASE_PATH, publicPath),
+        "PROTOCOL": config.gateway.protocol,
+        ...serviceLogEnv,
+        ...nodeOptions ? { NODE_OPTIONS: nodeOptions } : {},
+      },
+      volumes: {
+        "/etc/hosts": "/etc/hosts",
+        "./config": "/etc/scow",
+        "~/.ssh": "/root/.ssh",
+      },
+    });
+  }
+
   return composeSpec;
 };

apps/cli/src/config/install.ts

@@ -30,12 +30,12 @@ export const InstallConfigSchema = Type.Object({
     ssl: Type.Optional(Type.Object({
       enabled: Type.Boolean({ description: "到 SCOWD 的连接是否启动SSL", default: false }),
       caCertPath: Type.String({ description: "SCOWD CA根证书路径, 相对 config 的默认目录", default: "./scowd/certs/ca.crt" }),
-      scowCertPath: Type.String({ 
-        description: "SCOWD CA签名的 SCOW 证书路径， 相对 config 的默认目录", 
+      scowCertPath: Type.String({
+        description: "SCOWD CA签名的 SCOW 证书路径， 相对 config 的默认目录",
         default: "./scowd/certs/scow.crt",
       }),
-      scowPrivateKeyPath: Type.String({ 
-        description: "SCOWD CA签名的 SCOW 私钥路径， 相对 config 的默认目录", 
+      scowPrivateKeyPath: Type.String({
+        description: "SCOWD CA签名的 SCOW 私钥路径， 相对 config 的默认目录",
         default: "./scowd/certs/scow.key",
       }),
     }, { description: "scowd  全局 ssl 相关配置" })),
@@ -166,6 +166,10 @@ export const InstallConfigSchema = Type.Object({
     })),
   })),
 
+  notification: Type.Optional(Type.Object({
+    basePath: Type.String({ description: "消息系统的部署路径，相对于整个系统的basePath", default: "/notification" }),
+  })),
+
   misc: Type.Optional(Type.Object({
     nodeOptions:  Type.Optional(Type.String({ description: "传递给node服务的参数" })),
   }, { description: "多个不好分类的配置参数参数" })),

apps/gateway/assets/nginx.conf

@@ -39,6 +39,14 @@ server {
     include includes/websocket;
   }
 
+  location ${BASE_PATH}${NOTIFICATION_PATH} {
+    set $notification_path_url ${NOTIFICATION_PATH_INTERNAL_URL};
+    proxy_pass $notification_path_url;
+
+    include includes/headers;
+    include includes/websocket;
+  }
+
   location ${BASE_PATH}${VNC_PATH} {
     proxy_pass ${NOVNC_INTERNAL_URL};
 

apps/gateway/src/env.ts

@@ -27,7 +27,10 @@ export const config = envConfig({
   MIS_PATH_INTERNAL_URL: str({ desc: "管理系统内部路径", default: "http://mis-web:3000" }),
 
   AI_PATH: str({ desc: "AI系统路径", default: "/ai" }),
-  AI_PATH_INTERNAL_URL: str({ desc: "门户系统内部路径", default: "http://ai:3000" }),
+  AI_PATH_INTERNAL_URL: str({ desc: "AI系统内部路径", default: "http://ai:3000" }),
+
+  NOTIFICATION_PATH: str({ desc: "消息系统路径", default: "/notification" }),
+  NOTIFICATION_PATH_INTERNAL_URL: str({ desc: "消息系统内部路径", default: "http://notification:3000" }),
 
   VNC_PATH: str({ desc: "VNC客户端路径", default: "/vnc/" }),
   NOVNC_INTERNAL_URL: str({ desc: "NOVNC内部地址", default: "http://novnc:80/" }),

apps/mis-server/package.json

@@ -43,6 +43,8 @@
     "@scow/lib-ssh": "workspace:*",
     "@scow/protos": "workspace:*",
     "@scow/scheduler-adapter-protos": "workspace:*",
+    "@scow/lib-notification": "workspace:*",
+    "@scow/notification-protos": "workspace:*",
     "@scow/utils": "workspace:*",
     "@sinclair/typebox": "0.32.34",
     "dayjs": "1.11.11",
@@ -51,7 +53,8 @@
     "pino": "9.3.2",
     "pino-pretty": "10.3.1",
     "uuid": "10.0.0",
-    "wait-on": "8.0.0"
+    "wait-on": "8.0.0",
+    "@bufbuild/protobuf": "1.8.0"
   },
   "devDependencies": {
     "@types/google-protobuf": "3.15.12",

apps/mis-server/src/bl/charging.ts

@@ -15,15 +15,19 @@ import { Loaded } from "@mikro-orm/core";
 import { SqlEntityManager } from "@mikro-orm/mysql";
 import { ClusterConfigSchema } from "@scow/config/build/cluster";
 import { Decimal, decimalToMoney } from "@scow/lib-decimal";
+import { TargetType } from "@scow/notification-protos/build/message_common_pb";
 import { blockAccount, blockUserInAccount, unblockAccount, unblockUserInAccount } from "src/bl/block";
 import { Account } from "src/entities/Account";
 import { ChargeRecord } from "src/entities/ChargeRecord";
 import { PayRecord } from "src/entities/PayRecord";
 import { Tenant } from "src/entities/Tenant";
 import { UserAccount } from "src/entities/UserAccount";
+import { InternalMessageType } from "src/models/messageType";
 import { ClusterPlugin } from "src/plugins/clusters";
 import { callHook } from "src/plugins/hookClient";
 import { getAccountStateInfo, getUserStateInfo } from "src/utils/accountUserState";
+import { getAccountOwnerAndAdmin } from "src/utils/getAccountOwnerAndAdmin";
+import { sendMessage } from "src/utils/sendMessage";
 import { AnyJson } from "src/utils/types";
 
 interface PayRequest {
@@ -85,6 +89,19 @@ export async function pay(
   if (target instanceof Account) {
     await callHook("accountPaid", {
       accountName: target.accountName, amount: decimalToMoney(amount), type, comment }, logger);
+
+    // 给账户充值时发送消息
+    const ownerAndAdmin = await getAccountOwnerAndAdmin(target.accountName, logger, em);
+    await sendMessage({
+      messageType: InternalMessageType.AccountRechargeSuccess,
+      targetType: TargetType.USER, targetIds: ownerAndAdmin.map((x) => x.userId),
+      metadata: {
+        time: (new Date()).toISOString(),
+        accountName: target.accountName,
+        chargeAmount: amount.toString(),
+        amount: target.balance.toString(),
+      },
+    }, logger);
   } else {
     await callHook("tenantPaid", { tenantName: target.name, amount: decimalToMoney(amount), type, comment }, logger);
   }
@@ -142,6 +159,43 @@ export async function charge(
   const prevBalance = target.balance;
   target.balance = target.balance.minus(amount);
 
+  if (target instanceof Account) {
+
+    const ownerAndAdmin = await getAccountOwnerAndAdmin(target.accountName, logger, em);
+    await sendMessage({
+      messageType: InternalMessageType.AccountBalance,
+      targetType: TargetType.USER, targetIds: ownerAndAdmin.map((x) => x.userId),
+      metadata: {
+        accountName: target.accountName,
+        amount: amount.toString(),
+        balance: target.balance.toString(),
+      },
+    }, logger);
+
+    if (target.balance.lt(target.blockThresholdAmount ?? 0)) {
+      await sendMessage({
+        messageType: InternalMessageType.AccountOverdue,
+        targetType: TargetType.USER, targetIds: ownerAndAdmin.map((x) => x.userId),
+        metadata: {
+          time: (new Date()).toISOString(),
+          accountName: target.accountName,
+          amount: target.balance.minus(target.blockThresholdAmount ?? 0).abs().toString(),
+        },
+      }, logger);
+    }
+
+    if (target.balance.lt(20)) {
+      await sendMessage({
+        messageType: InternalMessageType.AccountLowBalance,
+        targetType: TargetType.USER, targetIds: ownerAndAdmin.map((x) => x.userId),
+        metadata: {
+          time: (new Date()).toISOString(),
+          accountName: target.accountName,
+        },
+      }, logger);
+    }
+  }
+
   if (
     target instanceof Account
     && checkShouldBlockAccount(target)

apps/mis-server/src/config/notification.ts

@@ -0,0 +1,20 @@
+/**
+ * Copyright (c) 2022 Peking University and Peking University Institute for Computing and Digital Economy
+ * SCOW is licensed under Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ *          http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ */
+
+import { getNotificationNodeClient } from "@scow/lib-notification/build/index";
+
+import { commonConfig } from "./common";
+
+const notifConfig = commonConfig.notification;
+
+export const notifClient = notifConfig?.enabled
+  ? getNotificationNodeClient(notifConfig.address) : undefined;

apps/mis-server/src/models/messageType.ts

@@ -0,0 +1,23 @@
+/**
+ * Copyright (c) 2022 Peking University and Peking University Institute for Computing and Digital Economy
+ * SCOW is licensed under Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ *          http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ */
+
+export enum InternalMessageType {
+  AccountOverdue = "AccountOverdue",
+  AccountRechargeSuccess = "AccountRechargeSuccess",
+  AccountLowBalance = "AccountLowBalance",
+  AccountBalance = "AccountBalance",
+  AccountLocked = "AccountLocked",
+  AccountRestored = "AccountRestored",
+  JobStarted = "JobStarted",
+  JobFinished = "JobFinished",
+  JobAbnormalTermination = "JobAbnormalTermination",
+}