创建由壹点统一通知中心
This commit is contained in:
WuYongJie
commit
cfeaafb8f3
7
.dockerignore
Normal file
7
.dockerignore
Normal file
@ -0,0 +1,7 @@
|
||||
node_modules
|
||||
dist
|
||||
.env
|
||||
data
|
||||
*.db
|
||||
*.log
|
||||
.git
|
||||
16
.env.example
Normal file
16
.env.example
Normal file
@ -0,0 +1,16 @@
|
||||
PORT=3088
|
||||
# Windows 本机路径包含空格时，Prisma SQLite 请使用完整绝对路径；Docker 会覆盖为 /app/data。
|
||||
DATABASE_URL=file:D:/AI Code/YouYiDianNotificationCenter/data/notifications.db
|
||||
|
||||
# “由壹点”机器人所属企业内部应用的凭证，仅允许配置在通知中心。
|
||||
DINGTALK_APP_KEY=
|
||||
DINGTALK_APP_SECRET=
|
||||
DINGTALK_ROBOT_CODE=
|
||||
|
||||
# 每个项目必须使用不同密钥；allowedHosts 用于限制卡片跳转域名。
|
||||
# 生产环境请使用足够长的随机密钥，不要把真实值提交到 Git。
|
||||
YYD_NOTIFICATION_PROJECTS_JSON=[{"code":"material_library","name":"素材库","secret":"replace-with-random-secret","allowedHosts":["sc.example.com"]},{"code":"fixed_asset","name":"固定资产","secret":"replace-with-another-random-secret","allowedHosts":["asset.example.com"]}]
|
||||
|
||||
WORKER_ENABLED=true
|
||||
WORKER_INTERVAL_MS=1000
|
||||
READ_STATUS_INTERVAL_MS=5000
|
||||
1
.gitattributes
vendored
Normal file
1
.gitattributes
vendored
Normal file
@ -0,0 +1 @@
|
||||
*.sh text eol=lf
|
||||
8
.gitignore
vendored
Normal file
8
.gitignore
vendored
Normal file
@ -0,0 +1,8 @@
|
||||
node_modules/
|
||||
dist/
|
||||
.env
|
||||
data/
|
||||
*.log
|
||||
*.db
|
||||
*.db-journal
|
||||
*.db-wal
|
||||
24
Dockerfile
Normal file
24
Dockerfile
Normal file
@ -0,0 +1,24 @@
|
||||
FROM node:22-alpine AS build
|
||||
WORKDIR /app
|
||||
RUN apk add --no-cache openssl
|
||||
COPY package*.json ./
|
||||
RUN npm ci
|
||||
COPY prisma ./prisma
|
||||
RUN npx prisma generate
|
||||
COPY tsconfig.json ./
|
||||
COPY src ./src
|
||||
RUN npm run build
|
||||
|
||||
FROM node:22-alpine
|
||||
WORKDIR /app
|
||||
ENV NODE_ENV=production
|
||||
RUN apk add --no-cache openssl
|
||||
COPY package*.json ./
|
||||
RUN npm ci --omit=dev
|
||||
COPY --from=build /app/node_modules/.prisma ./node_modules/.prisma
|
||||
COPY --from=build /app/node_modules/@prisma/client ./node_modules/@prisma/client
|
||||
COPY --from=build /app/dist ./dist
|
||||
COPY prisma ./prisma
|
||||
RUN mkdir -p /app/data
|
||||
EXPOSE 3088
|
||||
CMD ["sh", "-c", "npx prisma migrate deploy && node dist/index.js"]
|
||||
53
README.md
Normal file
53
README.md
Normal file
@ -0,0 +1,53 @@
|
||||
# 由壹点通知中心
|
||||
|
||||
由壹点内部项目统一钉钉机器人单聊出口。只有本服务保存钉钉应用密钥和 `RobotCode`；素材库、固定资产、产品开发平台等业务系统使用各自的项目密钥调用本服务。
|
||||
|
||||
## 安全边界
|
||||
|
||||
- 每个项目使用独立 HMAC 密钥。
|
||||
- 请求签名包含时间、nonce、HTTP 方法、路径和正文摘要。
|
||||
- nonce 五分钟内只能使用一次。
|
||||
- 每个项目只能跳转到配置的 HTTPS 域名。
|
||||
- `projectCode + idempotencyKey` 唯一，防止业务重试造成重复通知。
|
||||
- 项目无法读取其它项目的投递记录。
|
||||
|
||||
## 本地启动
|
||||
|
||||
1. 复制 `.env.example` 为 `.env` 并填写真实配置；不要提交 `.env`。
|
||||
Windows 下项目目录包含空格，`DATABASE_URL` 请保留为完整绝对路径。
|
||||
2. 执行 `npm install`。
|
||||
3. 执行 `npm run prisma:generate`。
|
||||
4. 执行 `npm run prisma:migrate`。
|
||||
5. 执行 `npm run dev`。
|
||||
|
||||
健康检查：`GET /health`。
|
||||
|
||||
生产环境建议只把容器端口绑定到 `127.0.0.1`，再由 Nginx 提供独立 HTTPS 域名。素材库和固定资产可能位于不同主机，因此不能把业务项目配置成通知中心的容器内网地址。
|
||||
|
||||
当前生产部署脚本位于：
|
||||
|
||||
- `scripts/deploy-material-host.sh`：只管理通知中心独立容器、配置和独立 FRP 服务；
|
||||
- `scripts/deploy-public-nginx.sh`：备份并扩展素材库专用 FRP 端口白名单，然后为素材库公网域名增加 `/notification-center/` 反向代理；脚本会验证主站、S3 和通知中心，失败时回滚。
|
||||
|
||||
脚本不包含任何密钥。统一机器人凭证从素材库生产环境安全复制到通知中心独立 `.env`，项目 HMAC 密钥在服务器生成并以 `0600` 权限保存。
|
||||
|
||||
素材库生产 NAS 无法稳定访问 Docker Hub。正式发布应先在构建机或维护工作站构建 `youyidian-notification-center:latest`，将镜像安全传入生产机，再执行部署脚本；脚本检测到预载镜像后会使用 `--no-build` 启动。
|
||||
|
||||
## 签名算法
|
||||
|
||||
请求头：
|
||||
|
||||
- `x-yyd-project-code`
|
||||
- `x-yyd-timestamp`：Unix 秒
|
||||
- `x-yyd-nonce`：每次请求生成新的随机字符串
|
||||
- `x-yyd-signature`：HMAC-SHA256 十六进制字符串
|
||||
|
||||
待签名文本：
|
||||
|
||||
```text
|
||||
timestamp\nnonce\nHTTP_METHOD\nPATH\nSHA256(rawBody)
|
||||
```
|
||||
|
||||
发送接口：`POST /v1/notifications`。查询接口：`GET /v1/notifications/:id`。
|
||||
|
||||
通知中心只接受任务并异步发送，业务系统通过查询接口获取 `pending / sending / sent / unread / read / retrying / failed` 状态。
|
||||
21
docker-compose.yml
Normal file
21
docker-compose.yml
Normal file
@ -0,0 +1,21 @@
|
||||
services:
|
||||
notification-center:
|
||||
image: youyidian-notification-center:${YYD_NOTIFICATION_IMAGE_TAG:-latest}
|
||||
build: .
|
||||
container_name: youyidian-notification-center
|
||||
restart: unless-stopped
|
||||
ports:
|
||||
- "127.0.0.1:${PORT:-3088}:3088"
|
||||
environment:
|
||||
PORT: 3088
|
||||
DATABASE_URL: file:/app/data/notifications.db
|
||||
DINGTALK_APP_KEY: ${DINGTALK_APP_KEY}
|
||||
DINGTALK_APP_SECRET: ${DINGTALK_APP_SECRET}
|
||||
DINGTALK_ROBOT_CODE: ${DINGTALK_ROBOT_CODE}
|
||||
YYD_NOTIFICATION_PROJECTS_JSON: ${YYD_NOTIFICATION_PROJECTS_JSON}
|
||||
WORKER_ENABLED: ${WORKER_ENABLED:-true}
|
||||
volumes:
|
||||
- notification_data:/app/data
|
||||
|
||||
volumes:
|
||||
notification_data:
|
||||
1587
package-lock.json
generated
Normal file
1587
package-lock.json
generated
Normal file
File diff suppressed because it is too large
Load Diff
26
package.json
Normal file
26
package.json
Normal file
@ -0,0 +1,26 @@
|
||||
{
|
||||
"name": "youyidian-notification-center",
|
||||
"version": "1.0.0",
|
||||
"private": true,
|
||||
"description": "由壹点内部项目统一钉钉单聊通知中心",
|
||||
"scripts": {
|
||||
"dev": "tsx watch src/index.ts",
|
||||
"build": "prisma generate && tsc",
|
||||
"start": "node dist/index.js",
|
||||
"test": "tsx --test src/__tests__/*.test.ts",
|
||||
"prisma:generate": "prisma generate",
|
||||
"prisma:migrate": "prisma migrate deploy"
|
||||
},
|
||||
"dependencies": {
|
||||
"@prisma/client": "^5.22.0",
|
||||
"dotenv": "^16.4.7",
|
||||
"express": "^4.21.2",
|
||||
"prisma": "^5.22.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@types/express": "^4.17.21",
|
||||
"@types/node": "^22.10.2",
|
||||
"tsx": "^4.19.2",
|
||||
"typescript": "^5.7.2"
|
||||
}
|
||||
}
|
||||
41
prisma/migrations/20260715_init/migration.sql
Normal file
41
prisma/migrations/20260715_init/migration.sql
Normal file
@ -0,0 +1,41 @@
|
||||
-- CreateTable
|
||||
CREATE TABLE "NotificationDelivery" (
|
||||
"id" TEXT NOT NULL PRIMARY KEY,
|
||||
"projectCode" TEXT NOT NULL,
|
||||
"idempotencyKey" TEXT NOT NULL,
|
||||
"requestHash" TEXT NOT NULL,
|
||||
"recipientUserId" TEXT NOT NULL,
|
||||
"businessTitle" TEXT NOT NULL,
|
||||
"renderedTitle" TEXT NOT NULL,
|
||||
"content" TEXT NOT NULL,
|
||||
"actionTitle" TEXT NOT NULL,
|
||||
"actionUrl" TEXT NOT NULL,
|
||||
"status" TEXT NOT NULL DEFAULT 'pending',
|
||||
"attemptCount" INTEGER NOT NULL DEFAULT 0,
|
||||
"maxAttempts" INTEGER NOT NULL DEFAULT 4,
|
||||
"nextAttemptAt" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
"leaseOwner" TEXT,
|
||||
"leaseExpiresAt" DATETIME,
|
||||
"processQueryKey" TEXT,
|
||||
"lastErrorCode" TEXT,
|
||||
"lastErrorMessage" TEXT,
|
||||
"sentAt" DATETIME,
|
||||
"readAt" DATETIME,
|
||||
"createdAt" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
"updatedAt" DATETIME NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE "RequestNonce" (
|
||||
"id" TEXT NOT NULL PRIMARY KEY,
|
||||
"projectCode" TEXT NOT NULL,
|
||||
"nonce" TEXT NOT NULL,
|
||||
"expiresAt" DATETIME NOT NULL,
|
||||
"createdAt" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
|
||||
CREATE UNIQUE INDEX "NotificationDelivery_projectCode_idempotencyKey_key" ON "NotificationDelivery"("projectCode", "idempotencyKey");
|
||||
CREATE INDEX "NotificationDelivery_status_nextAttemptAt_idx" ON "NotificationDelivery"("status", "nextAttemptAt");
|
||||
CREATE INDEX "NotificationDelivery_leaseExpiresAt_idx" ON "NotificationDelivery"("leaseExpiresAt");
|
||||
CREATE INDEX "NotificationDelivery_processQueryKey_idx" ON "NotificationDelivery"("processQueryKey");
|
||||
CREATE UNIQUE INDEX "RequestNonce_projectCode_nonce_key" ON "RequestNonce"("projectCode", "nonce");
|
||||
CREATE INDEX "RequestNonce_expiresAt_idx" ON "RequestNonce"("expiresAt");
|
||||
3
prisma/migrations/migration_lock.toml
Normal file
3
prisma/migrations/migration_lock.toml
Normal file
@ -0,0 +1,3 @@
|
||||
# Please do not edit this file manually
|
||||
# It should be added in your version-control system (i.e. Git)
|
||||
provider = "sqlite"
|
||||
50
prisma/schema.prisma
Normal file
50
prisma/schema.prisma
Normal file
@ -0,0 +1,50 @@
|
||||
generator client {
|
||||
provider = "prisma-client-js"
|
||||
}
|
||||
|
||||
datasource db {
|
||||
provider = "sqlite"
|
||||
url = env("DATABASE_URL")
|
||||
}
|
||||
|
||||
model NotificationDelivery {
|
||||
id String @id @default(uuid())
|
||||
projectCode String
|
||||
idempotencyKey String
|
||||
requestHash String
|
||||
recipientUserId String
|
||||
businessTitle String
|
||||
renderedTitle String
|
||||
content String
|
||||
actionTitle String
|
||||
actionUrl String
|
||||
status String @default("pending")
|
||||
attemptCount Int @default(0)
|
||||
maxAttempts Int @default(4)
|
||||
nextAttemptAt DateTime @default(now())
|
||||
leaseOwner String?
|
||||
leaseExpiresAt DateTime?
|
||||
processQueryKey String?
|
||||
lastErrorCode String?
|
||||
lastErrorMessage String?
|
||||
sentAt DateTime?
|
||||
readAt DateTime?
|
||||
createdAt DateTime @default(now())
|
||||
updatedAt DateTime @updatedAt
|
||||
|
||||
@@unique([projectCode, idempotencyKey])
|
||||
@@index([status, nextAttemptAt])
|
||||
@@index([leaseExpiresAt])
|
||||
@@index([processQueryKey])
|
||||
}
|
||||
|
||||
model RequestNonce {
|
||||
id String @id @default(uuid())
|
||||
projectCode String
|
||||
nonce String
|
||||
expiresAt DateTime
|
||||
createdAt DateTime @default(now())
|
||||
|
||||
@@unique([projectCode, nonce])
|
||||
@@index([expiresAt])
|
||||
}
|
||||
235
scripts/deploy-material-host.sh
Normal file
235
scripts/deploy-material-host.sh
Normal file
@ -0,0 +1,235 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
target=/mnt/PoolOne/apps/youyidian-notification-center
|
||||
archive=/tmp/youyidian-notification-center-release.tar.gz
|
||||
source_env=/mnt/PoolOne/apps/youyidian-3d/.env
|
||||
backup_root=/mnt/PoolOne/apps/youyidian-notification-center-backups
|
||||
|
||||
test -f "$archive"
|
||||
test -f "$source_env"
|
||||
if tar -tzf "$archive" | grep -Eq '(^/|(^|/)\.\.(/|$))'; then
|
||||
echo '安装包包含不安全路径' >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
mkdir -p "$target" "$backup_root"
|
||||
if test -f "$target/package.json"; then
|
||||
stamp="$(date +%Y%m%d%H%M%S)"
|
||||
tar -czf "$backup_root/source-$stamp.tar.gz" --exclude=.env --exclude=data -C "$target" .
|
||||
fi
|
||||
tar -xzf "$archive" -C "$target"
|
||||
|
||||
python3 - <<'PY'
|
||||
from pathlib import Path
|
||||
import json
|
||||
import os
|
||||
import secrets
|
||||
import shutil
|
||||
import time
|
||||
|
||||
|
||||
def parse_env(path: Path):
|
||||
result = {}
|
||||
if not path.exists():
|
||||
return result
|
||||
for raw in path.read_text(encoding='utf-8').splitlines():
|
||||
line = raw.strip()
|
||||
if not line or line.startswith('#') or '=' not in line:
|
||||
continue
|
||||
key, value = line.split('=', 1)
|
||||
key = key.strip()
|
||||
value = value.strip()
|
||||
if value.startswith('"') and value.endswith('"'):
|
||||
try:
|
||||
value = json.loads(value)
|
||||
except Exception:
|
||||
value = value[1:-1]
|
||||
elif value.startswith("'") and value.endswith("'"):
|
||||
value = value[1:-1]
|
||||
result[key] = value
|
||||
return result
|
||||
|
||||
|
||||
def quoted(value):
|
||||
return json.dumps(str(value), ensure_ascii=False)
|
||||
|
||||
|
||||
source_path = Path('/mnt/PoolOne/apps/youyidian-3d/.env')
|
||||
target_path = Path('/mnt/PoolOne/apps/youyidian-notification-center/.env')
|
||||
source = parse_env(source_path)
|
||||
existing = parse_env(target_path)
|
||||
app_key = (
|
||||
source.get('DINGTALK_WORK_NOTICE_CLIENT_ID')
|
||||
or source.get('DINGTALK_WORK_NOTICE_APP_KEY')
|
||||
or source.get('DINGTALK_CLIENT_ID')
|
||||
or source.get('DINGTALK_APP_KEY')
|
||||
)
|
||||
app_secret = (
|
||||
source.get('DINGTALK_WORK_NOTICE_CLIENT_SECRET')
|
||||
or source.get('DINGTALK_WORK_NOTICE_APP_SECRET')
|
||||
or source.get('DINGTALK_CLIENT_SECRET')
|
||||
or source.get('DINGTALK_APP_SECRET')
|
||||
)
|
||||
robot_code = source.get('DINGTALK_ROBOT_CODE')
|
||||
if not app_key or not app_secret or not robot_code:
|
||||
raise SystemExit('素材库生产环境缺少统一机器人配置')
|
||||
|
||||
try:
|
||||
old_projects = json.loads(existing.get('YYD_NOTIFICATION_PROJECTS_JSON', '[]'))
|
||||
except Exception:
|
||||
old_projects = []
|
||||
by_code = {
|
||||
str(item.get('code')): item
|
||||
for item in old_projects
|
||||
if isinstance(item, dict)
|
||||
}
|
||||
|
||||
|
||||
def secret_for(code):
|
||||
old = str(by_code.get(code, {}).get('secret') or '')
|
||||
return old if len(old) >= 24 else secrets.token_hex(32)
|
||||
|
||||
|
||||
projects = [
|
||||
{
|
||||
'code': 'material_library',
|
||||
'name': '素材库',
|
||||
'secret': secret_for('material_library'),
|
||||
'allowedHosts': ['sc.jiansun.vip'],
|
||||
},
|
||||
{
|
||||
'code': 'fixed_asset',
|
||||
'name': '固定资产',
|
||||
'secret': secret_for('fixed_asset'),
|
||||
'allowedHosts': ['www.jsgdzc.cn', 'jsgdzc.cn'],
|
||||
},
|
||||
]
|
||||
center_lines = [
|
||||
'PORT=3088',
|
||||
'DINGTALK_APP_KEY=' + quoted(app_key),
|
||||
'DINGTALK_APP_SECRET=' + quoted(app_secret),
|
||||
'DINGTALK_ROBOT_CODE=' + quoted(robot_code),
|
||||
'YYD_NOTIFICATION_PROJECTS_JSON=' + quoted(
|
||||
json.dumps(projects, ensure_ascii=False, separators=(',', ':'))
|
||||
),
|
||||
'WORKER_ENABLED=true',
|
||||
'WORKER_INTERVAL_MS=1000',
|
||||
'READ_STATUS_INTERVAL_MS=5000',
|
||||
]
|
||||
target_path.write_text('\n'.join(center_lines) + '\n', encoding='utf-8')
|
||||
os.chmod(target_path, 0o600)
|
||||
|
||||
material_secret = next(
|
||||
item['secret'] for item in projects if item['code'] == 'material_library'
|
||||
)
|
||||
updates = {
|
||||
'YYD_NOTIFICATION_CENTER_URL': 'https://sc.jiansun.vip/notification-center',
|
||||
'YYD_NOTIFICATION_CENTER_PROJECT_CODE': 'material_library',
|
||||
'YYD_NOTIFICATION_CENTER_PROJECT_SECRET': material_secret,
|
||||
'YYD_NOTIFICATION_CENTER_TEST_USER_IDS': '',
|
||||
}
|
||||
backup = source_path.with_name(
|
||||
'.env.before-yyd-notification-' + time.strftime('%Y%m%d%H%M%S')
|
||||
)
|
||||
shutil.copy2(source_path, backup)
|
||||
os.chmod(backup, 0o600)
|
||||
lines = source_path.read_text(encoding='utf-8').splitlines()
|
||||
seen = set()
|
||||
output = []
|
||||
for line in lines:
|
||||
if '=' in line and not line.lstrip().startswith('#'):
|
||||
key = line.split('=', 1)[0].strip()
|
||||
if key in updates:
|
||||
output.append(key + '=' + quoted(updates[key]))
|
||||
seen.add(key)
|
||||
continue
|
||||
output.append(line)
|
||||
for key, value in updates.items():
|
||||
if key not in seen:
|
||||
output.append(key + '=' + quoted(value))
|
||||
source_path.write_text('\n'.join(output) + '\n', encoding='utf-8')
|
||||
os.chmod(source_path, 0o600)
|
||||
print('通知中心与素材库适配配置已写入')
|
||||
PY
|
||||
|
||||
cd "$target"
|
||||
image="youyidian-notification-center:${YYD_NOTIFICATION_IMAGE_TAG:-latest}"
|
||||
if docker image inspect "$image" >/dev/null 2>&1; then
|
||||
docker compose up -d --no-build
|
||||
else
|
||||
docker compose up -d --build
|
||||
fi
|
||||
|
||||
healthy=0
|
||||
for _ in $(seq 1 30); do
|
||||
if curl -fsS --max-time 3 http://127.0.0.1:3088/health >/tmp/yyd-notification-health.json; then
|
||||
healthy=1
|
||||
break
|
||||
fi
|
||||
sleep 2
|
||||
done
|
||||
if test "$healthy" != 1; then
|
||||
docker compose logs --tail 120
|
||||
exit 1
|
||||
fi
|
||||
|
||||
python3 - <<'PY'
|
||||
from pathlib import Path
|
||||
|
||||
source = Path('/etc/frp/frpc-sc-entry.toml').read_text(encoding='utf-8')
|
||||
header = source.split('[[proxies]]', 1)[0]
|
||||
header = header.replace(
|
||||
'/root/frp/frpc-sc-entry.log',
|
||||
'/root/frp/frpc-notify-entry.log',
|
||||
)
|
||||
proxy = '''[[proxies]]
|
||||
name = "youyidian-notification-center"
|
||||
type = "tcp"
|
||||
localIP = "127.0.0.1"
|
||||
localPort = 3088
|
||||
remotePort = 2326
|
||||
healthCheck.type = "http"
|
||||
healthCheck.path = "/health"
|
||||
healthCheck.intervalSeconds = 15
|
||||
healthCheck.maxFailed = 3
|
||||
healthCheck.timeoutSeconds = 5
|
||||
'''
|
||||
Path('/etc/frp/frpc-notify-entry.toml').write_text(
|
||||
header.rstrip() + '\n\n' + proxy,
|
||||
encoding='utf-8',
|
||||
)
|
||||
PY
|
||||
chmod 600 /etc/frp/frpc-notify-entry.toml
|
||||
|
||||
cat > /etc/systemd/system/frpc-notify-entry.service <<'EOF'
|
||||
[Unit]
|
||||
Description=FRP Client for Youyidian notification center
|
||||
After=network-online.target docker.service
|
||||
Wants=network-online.target
|
||||
Requires=docker.service
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
Restart=always
|
||||
RestartSec=10s
|
||||
LimitNOFILE=65535
|
||||
ExecStart=/root/frp-upgrade/frp_0.67.0_linux_amd64/frpc -c /etc/frp/frpc-notify-entry.toml
|
||||
StandardOutput=append:/root/frp/frpc-notify-entry.service.log
|
||||
StandardError=append:/root/frp/frpc-notify-entry.service.log
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
EOF
|
||||
systemctl daemon-reload
|
||||
systemctl enable --now frpc-notify-entry.service
|
||||
sleep 3
|
||||
systemctl is-active --quiet frpc-notify-entry.service
|
||||
|
||||
rm -f "$archive"
|
||||
curl -fsS http://127.0.0.1:3088/health
|
||||
echo
|
||||
printf 'CENTER_CONTAINER='
|
||||
docker inspect -f '{{.State.Status}}' youyidian-notification-center
|
||||
printf 'FRP_SERVICE='
|
||||
systemctl is-active frpc-notify-entry.service
|
||||
124
scripts/deploy-public-nginx.sh
Normal file
124
scripts/deploy-public-nginx.sh
Normal file
@ -0,0 +1,124 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
config=/etc/nginx/conf.d/sc.jiansun.vip.conf
|
||||
frps_config=/etc/frp/frps-sucai.toml
|
||||
frps_service=frps-sucai.service
|
||||
test -f "$config"
|
||||
test -f "$frps_config"
|
||||
backup="${config}.before-notification-center-$(date +%Y%m%d%H%M%S)"
|
||||
frps_backup="${frps_config}.before-notification-center-$(date +%Y%m%d%H%M%S)"
|
||||
cp -a "$config" "$backup"
|
||||
cp -a "$frps_config" "$frps_backup"
|
||||
|
||||
FRPS_CONFIG="$frps_config" python3 - <<'PY'
|
||||
from pathlib import Path
|
||||
import os
|
||||
import re
|
||||
|
||||
path = Path(os.environ['FRPS_CONFIG'])
|
||||
text = path.read_text(encoding='utf-8')
|
||||
match = re.search(
|
||||
r'allowPorts\s*=\s*\[\s*\{\s*start\s*=\s*2321\s*,\s*end\s*=\s*(\d+)\s*\}\s*,?\s*\]',
|
||||
text,
|
||||
re.S,
|
||||
)
|
||||
if not match:
|
||||
raise SystemExit('未识别素材库 FRP 端口白名单，拒绝修改')
|
||||
if int(match.group(1)) < 2326:
|
||||
text = text[:match.start(1)] + '2326' + text[match.end(1):]
|
||||
path.write_text(text, encoding='utf-8')
|
||||
PY
|
||||
|
||||
if ! /opt/frp-furniture-server/frps verify -c "$frps_config"; then
|
||||
cp -a "$frps_backup" "$frps_config"
|
||||
echo 'FRP 服务端配置验证失败，已回滚' >&2
|
||||
exit 1
|
||||
fi
|
||||
if ! systemctl restart "$frps_service"; then
|
||||
cp -a "$frps_backup" "$frps_config"
|
||||
systemctl restart "$frps_service"
|
||||
echo 'FRP 服务端重启失败，已回滚' >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
frp_ready=0
|
||||
for _ in $(seq 1 20); do
|
||||
if ss -lnt | grep -q ':2321 ' \
|
||||
&& ss -lnt | grep -q ':2322 ' \
|
||||
&& ss -lnt | grep -q ':2323 ' \
|
||||
&& ss -lnt | grep -q ':2326 '; then
|
||||
frp_ready=1
|
||||
break
|
||||
fi
|
||||
sleep 2
|
||||
done
|
||||
if test "$frp_ready" != 1; then
|
||||
cp -a "$frps_backup" "$frps_config"
|
||||
systemctl restart "$frps_service"
|
||||
echo 'FRP 隧道未全部恢复，已回滚' >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
CONFIG="$config" python3 - <<'PY'
|
||||
from pathlib import Path
|
||||
import os
|
||||
|
||||
path = Path(os.environ['CONFIG'])
|
||||
text = path.read_text(encoding='utf-8')
|
||||
if 'location ^~ /notification-center/' in text:
|
||||
raise SystemExit(0)
|
||||
needle = ' location / {\n'
|
||||
if needle not in text:
|
||||
raise SystemExit('未找到主站 location，拒绝修改')
|
||||
block = ''' location ^~ /notification-center/ {
|
||||
proxy_pass http://127.0.0.1:2326/;
|
||||
proxy_http_version 1.1;
|
||||
proxy_set_header Host $host;
|
||||
proxy_set_header X-Real-IP $remote_addr;
|
||||
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
|
||||
proxy_set_header X-Forwarded-Proto $scheme;
|
||||
proxy_connect_timeout 5s;
|
||||
proxy_send_timeout 30s;
|
||||
proxy_read_timeout 30s;
|
||||
client_max_body_size 128k;
|
||||
}
|
||||
|
||||
'''
|
||||
path.write_text(text.replace(needle, block + needle, 1), encoding='utf-8')
|
||||
PY
|
||||
|
||||
if ! nginx -t; then
|
||||
cp -a "$backup" "$config"
|
||||
nginx -t
|
||||
echo 'Nginx 配置验证失败，已回滚' >&2
|
||||
exit 1
|
||||
fi
|
||||
systemctl reload nginx
|
||||
|
||||
curl -fsS --max-time 10 --resolve sc.jiansun.vip:443:127.0.0.1 \
|
||||
https://sc.jiansun.vip/api/health >/tmp/yyd-material-public-health.json
|
||||
curl -fsS --max-time 10 --resolve s3.jiansun.vip:443:127.0.0.1 \
|
||||
https://s3.jiansun.vip/minio/health/live >/tmp/yyd-s3-public-health.txt
|
||||
|
||||
healthy=0
|
||||
for _ in $(seq 1 20); do
|
||||
if curl -fsS --max-time 5 --resolve sc.jiansun.vip:443:127.0.0.1 \
|
||||
https://sc.jiansun.vip/notification-center/health \
|
||||
>/tmp/yyd-notification-public-health.json; then
|
||||
healthy=1
|
||||
break
|
||||
fi
|
||||
sleep 2
|
||||
done
|
||||
if test "$healthy" != 1; then
|
||||
cp -a "$backup" "$config"
|
||||
nginx -t
|
||||
systemctl reload nginx
|
||||
echo '通知中心公网健康检查失败，已回滚 Nginx 配置' >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
cat /tmp/yyd-notification-public-health.json
|
||||
echo
|
||||
echo 'PUBLIC_NOTIFICATION_CENTER_OK'
|
||||
60
src/__tests__/server.test.ts
Normal file
60
src/__tests__/server.test.ts
Normal file
@ -0,0 +1,60 @@
|
||||
import assert from 'node:assert/strict'
|
||||
import crypto from 'node:crypto'
|
||||
import test from 'node:test'
|
||||
import { AddressInfo } from 'node:net'
|
||||
import { loadProjectConfigs } from '../config'
|
||||
import prisma from '../prisma'
|
||||
import { createServer } from '../server'
|
||||
import { createCanonicalSignature } from '../signature'
|
||||
|
||||
const projectSecret = '123456789012345678901234'
|
||||
const projects = loadProjectConfigs(JSON.stringify([
|
||||
{ code: 'fixed_asset', name: '固定资产', secret: projectSecret, allowedHosts: ['asset.example.com'] },
|
||||
]))
|
||||
|
||||
const headers = (method: string, path: string, body: string, nonce = crypto.randomBytes(18).toString('base64url')) => {
|
||||
const timestamp = String(Math.floor(Date.now() / 1000))
|
||||
return {
|
||||
'Content-Type': 'application/json',
|
||||
'x-yyd-project-code': 'fixed_asset',
|
||||
'x-yyd-timestamp': timestamp,
|
||||
'x-yyd-nonce': nonce,
|
||||
'x-yyd-signature': createCanonicalSignature({ secret: projectSecret, timestamp, nonce, method, path, body }),
|
||||
}
|
||||
}
|
||||
|
||||
test('通知中心验证签名、项目域名和幂等键', async () => {
|
||||
await prisma.requestNonce.deleteMany()
|
||||
await prisma.notificationDelivery.deleteMany()
|
||||
const server = createServer(projects).listen(0)
|
||||
await new Promise<void>((resolve) => server.once('listening', resolve))
|
||||
const port = (server.address() as AddressInfo).port
|
||||
const path = '/v1/notifications'
|
||||
const body = JSON.stringify({
|
||||
idempotencyKey: 'disposal_applied:53:1',
|
||||
recipientUserId: 'ding-user-1',
|
||||
title: '待审批报废',
|
||||
content: '有一项报废申请等待审批',
|
||||
actionTitle: '查看详情',
|
||||
actionUrl: 'https://asset.example.com/workflows/disposals/53',
|
||||
})
|
||||
|
||||
try {
|
||||
const first = await fetch(`http://127.0.0.1:${port}${path}`, { method: 'POST', headers: headers('POST', path, body), body })
|
||||
assert.equal(first.status, 202)
|
||||
const firstPayload = await first.json() as any
|
||||
assert.equal(firstPayload.title, '【由壹点·固定资产】待审批报废')
|
||||
|
||||
const second = await fetch(`http://127.0.0.1:${port}${path}`, { method: 'POST', headers: headers('POST', path, body), body })
|
||||
assert.equal(second.status, 200)
|
||||
const secondPayload = await second.json() as any
|
||||
assert.equal(secondPayload.id, firstPayload.id)
|
||||
|
||||
const badBody = JSON.stringify({ ...JSON.parse(body), actionUrl: 'https://evil.example.com/path' })
|
||||
const rejected = await fetch(`http://127.0.0.1:${port}${path}`, { method: 'POST', headers: headers('POST', path, badBody), body: badBody })
|
||||
assert.equal(rejected.status, 422)
|
||||
} finally {
|
||||
await new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()))
|
||||
await prisma.$disconnect()
|
||||
}
|
||||
})
|
||||
29
src/__tests__/signature.test.ts
Normal file
29
src/__tests__/signature.test.ts
Normal file
@ -0,0 +1,29 @@
|
||||
import assert from 'node:assert/strict'
|
||||
import test from 'node:test'
|
||||
import { loadProjectConfigs } from '../config'
|
||||
import { createCanonicalSignature, safeEqual } from '../signature'
|
||||
|
||||
test('每个项目必须使用独立且足够长的密钥', () => {
|
||||
const projects = loadProjectConfigs(JSON.stringify([
|
||||
{ code: 'fixed_asset', name: '固定资产', secret: '123456789012345678901234', allowedHosts: ['asset.example.com'] },
|
||||
]))
|
||||
assert.equal(projects.get('fixed_asset')?.name, '固定资产')
|
||||
assert.throws(() => loadProjectConfigs(JSON.stringify([
|
||||
{ code: 'bad', name: '错误项目', secret: 'short', allowedHosts: ['example.com'] },
|
||||
])))
|
||||
})
|
||||
|
||||
test('签名包含项目密钥、时间、nonce、方法、路径和正文', () => {
|
||||
const input = {
|
||||
secret: '123456789012345678901234',
|
||||
timestamp: '1784102400',
|
||||
nonce: 'nonce_1234567890abcdef',
|
||||
method: 'POST',
|
||||
path: '/v1/notifications',
|
||||
body: JSON.stringify({ title: '测试' }),
|
||||
}
|
||||
const signature = createCanonicalSignature(input)
|
||||
assert.equal(signature.length, 64)
|
||||
assert.equal(safeEqual(signature, createCanonicalSignature(input)), true)
|
||||
assert.equal(safeEqual(signature, createCanonicalSignature({ ...input, path: '/v1/other' })), false)
|
||||
})
|
||||
52
src/config.ts
Normal file
52
src/config.ts
Normal file
@ -0,0 +1,52 @@
|
||||
export type ProjectConfig = {
|
||||
code: string
|
||||
name: string
|
||||
secret: string
|
||||
allowedHosts: string[]
|
||||
}
|
||||
|
||||
const clean = (value: unknown) => String(value || '').trim()
|
||||
|
||||
export const loadProjectConfigs = (raw = process.env.YYD_NOTIFICATION_PROJECTS_JSON || '') => {
|
||||
if (!raw.trim()) return new Map<string, ProjectConfig>()
|
||||
let parsed: unknown
|
||||
try {
|
||||
parsed = JSON.parse(raw)
|
||||
} catch {
|
||||
throw new Error('YYD_NOTIFICATION_PROJECTS_JSON 不是有效 JSON')
|
||||
}
|
||||
if (!Array.isArray(parsed)) {
|
||||
throw new Error('YYD_NOTIFICATION_PROJECTS_JSON 必须是数组')
|
||||
}
|
||||
|
||||
const projects = new Map<string, ProjectConfig>()
|
||||
for (const row of parsed) {
|
||||
const item = row as Record<string, unknown>
|
||||
const code = clean(item.code)
|
||||
const name = clean(item.name)
|
||||
const secret = clean(item.secret)
|
||||
const allowedHosts = Array.isArray(item.allowedHosts)
|
||||
? item.allowedHosts.map(clean).filter(Boolean).map((host) => host.toLowerCase())
|
||||
: []
|
||||
if (!/^[a-z0-9_-]{2,40}$/.test(code)) throw new Error(`项目 code 不合法: ${code || '(空)'}`)
|
||||
if (!name) throw new Error(`项目 ${code} 缺少 name`)
|
||||
if (secret.length < 24) throw new Error(`项目 ${code} 的 secret 至少需要 24 个字符`)
|
||||
if (allowedHosts.length === 0) throw new Error(`项目 ${code} 至少需要一个 allowedHosts`)
|
||||
if (projects.has(code)) throw new Error(`项目 code 重复: ${code}`)
|
||||
projects.set(code, { code, name, secret, allowedHosts })
|
||||
}
|
||||
return projects
|
||||
}
|
||||
|
||||
export const getDingTalkConfig = () => ({
|
||||
appKey: clean(process.env.DINGTALK_APP_KEY),
|
||||
appSecret: clean(process.env.DINGTALK_APP_SECRET),
|
||||
robotCode: clean(process.env.DINGTALK_ROBOT_CODE),
|
||||
})
|
||||
|
||||
export const assertRuntimeConfig = (projects: Map<string, ProjectConfig>) => {
|
||||
if (projects.size === 0) throw new Error('未配置任何通知项目')
|
||||
const dingTalk = getDingTalkConfig()
|
||||
const missing = Object.entries(dingTalk).filter(([, value]) => !value).map(([key]) => key)
|
||||
if (missing.length > 0) throw new Error(`钉钉配置缺失: ${missing.join(', ')}`)
|
||||
}
|
||||
126
src/dingtalk.ts
Normal file
126
src/dingtalk.ts
Normal file
@ -0,0 +1,126 @@
|
||||
import { getDingTalkConfig } from './config'
|
||||
|
||||
type DingTalkPayload = Record<string, any>
|
||||
|
||||
export class DingTalkError extends Error {
|
||||
constructor(
|
||||
message: string,
|
||||
public readonly code: string,
|
||||
public readonly deterministic = false,
|
||||
) {
|
||||
super(message)
|
||||
this.name = 'DingTalkError'
|
||||
}
|
||||
}
|
||||
|
||||
let tokenCache: { token: string; expiresAt: number } | null = null
|
||||
let tokenPromise: Promise<string> | null = null
|
||||
|
||||
const requestJson = async (url: string, init?: RequestInit): Promise<DingTalkPayload> => {
|
||||
const controller = new AbortController()
|
||||
const timeout = setTimeout(() => controller.abort(), 10_000)
|
||||
try {
|
||||
const response = await fetch(url, {
|
||||
...init,
|
||||
signal: controller.signal,
|
||||
headers: { 'Content-Type': 'application/json', ...(init?.headers || {}) },
|
||||
})
|
||||
const data = await response.json().catch(() => ({})) as DingTalkPayload
|
||||
if (!response.ok) {
|
||||
const code = clean(data.code) || `HTTP_${response.status}`
|
||||
const message = clean(data.message) || `钉钉接口返回 HTTP ${response.status}`
|
||||
throw new DingTalkError(message, code, response.status >= 400 && response.status < 500 && response.status !== 429)
|
||||
}
|
||||
if (data.errcode !== undefined && Number(data.errcode) !== 0) {
|
||||
throw new DingTalkError(clean(data.errmsg) || '钉钉接口调用失败', String(data.errcode), isDeterministic(data.errcode, data.errmsg))
|
||||
}
|
||||
return data
|
||||
} catch (error: any) {
|
||||
if (error instanceof DingTalkError) throw error
|
||||
if (error?.name === 'AbortError') throw new DingTalkError('钉钉请求超时', 'TIMEOUT')
|
||||
throw new DingTalkError(clean(error?.message) || '钉钉网络请求失败', 'NETWORK')
|
||||
} finally {
|
||||
clearTimeout(timeout)
|
||||
}
|
||||
}
|
||||
|
||||
const clean = (value: unknown) => String(value || '').trim()
|
||||
const stringList = (value: unknown) => Array.isArray(value) ? value.map(clean).filter(Boolean) : []
|
||||
const isDeterministic = (code: unknown, message: unknown) =>
|
||||
['40014', '40056', '40057', '40078', '41001', '41009', '60111', '60121', '70020'].includes(String(code)) ||
|
||||
/不存在|不在可见范围|无权限|invalid user|forbidden user|userid/i.test(clean(message))
|
||||
|
||||
const getToken = async () => {
|
||||
if (tokenCache && tokenCache.expiresAt > Date.now() + 60_000) return tokenCache.token
|
||||
if (tokenPromise) return tokenPromise
|
||||
tokenPromise = (async () => {
|
||||
const { appKey, appSecret } = getDingTalkConfig()
|
||||
const data = await requestJson(
|
||||
`https://oapi.dingtalk.com/gettoken?appkey=${encodeURIComponent(appKey)}&appsecret=${encodeURIComponent(appSecret)}`,
|
||||
)
|
||||
const token = clean(data.access_token)
|
||||
if (!token) throw new DingTalkError('钉钉未返回 access_token', 'TOKEN_MISSING')
|
||||
tokenCache = { token, expiresAt: Date.now() + Math.max(300, Number(data.expires_in) || 7200) * 1000 }
|
||||
return token
|
||||
})()
|
||||
try {
|
||||
return await tokenPromise
|
||||
} finally {
|
||||
tokenPromise = null
|
||||
}
|
||||
}
|
||||
|
||||
export const sendRobotMessage = async (input: {
|
||||
recipientUserId: string
|
||||
title: string
|
||||
content: string
|
||||
actionTitle: string
|
||||
actionUrl: string
|
||||
}) => {
|
||||
const { robotCode } = getDingTalkConfig()
|
||||
const token = await getToken()
|
||||
const data = await requestJson('https://api.dingtalk.com/v1.0/robot/oToMessages/batchSend', {
|
||||
method: 'POST',
|
||||
headers: { 'x-acs-dingtalk-access-token': token },
|
||||
body: JSON.stringify({
|
||||
robotCode,
|
||||
userIds: [input.recipientUserId],
|
||||
msgKey: input.actionUrl ? 'sampleActionCard' : 'sampleMarkdown',
|
||||
msgParam: JSON.stringify(input.actionUrl ? {
|
||||
title: input.title.slice(0, 120),
|
||||
text: input.content.slice(0, 4800),
|
||||
singleTitle: input.actionTitle.slice(0, 20),
|
||||
singleURL: input.actionUrl,
|
||||
} : {
|
||||
title: input.title.slice(0, 120),
|
||||
text: input.content.slice(0, 4800),
|
||||
}),
|
||||
}),
|
||||
})
|
||||
|
||||
const invalid = stringList(data.invalidStaffIdList)
|
||||
const filtered = stringList(data.filteredStaffIdList)
|
||||
const flowControlled = stringList(data.flowControlledStaffIdList)
|
||||
if (invalid.includes(input.recipientUserId)) throw new DingTalkError('收件人钉钉 userId 无效', 'INVALID_USER', true)
|
||||
if (filtered.includes(input.recipientUserId)) throw new DingTalkError('收件人不在机器人可见范围内', 'FILTERED_USER', true)
|
||||
if (flowControlled.includes(input.recipientUserId)) throw new DingTalkError('钉钉机器人发送被限流', 'FLOW_CONTROLLED')
|
||||
const processQueryKey = clean(data.processQueryKey)
|
||||
if (!processQueryKey) throw new DingTalkError('钉钉未返回 processQueryKey', 'PROCESS_QUERY_KEY_MISSING')
|
||||
return processQueryKey
|
||||
}
|
||||
|
||||
export const queryRobotReadStatus = async (processQueryKey: string) => {
|
||||
const { robotCode } = getDingTalkConfig()
|
||||
const token = await getToken()
|
||||
const url = new URL('https://api.dingtalk.com/v1.0/robot/oToMessages/readStatus')
|
||||
url.searchParams.set('robotCode', robotCode)
|
||||
url.searchParams.set('processQueryKey', processQueryKey)
|
||||
const data = await requestJson(url.toString(), {
|
||||
headers: { 'x-acs-dingtalk-access-token': token },
|
||||
})
|
||||
const rows = Array.isArray(data.messageReadInfoList) ? data.messageReadInfoList : []
|
||||
const read = rows.some((row) => clean(row?.readStatus).toLowerCase() === 'read')
|
||||
const unread = rows.some((row) => clean(row?.readStatus).toLowerCase() === 'unread')
|
||||
const failed = clean(data.sendStatus).toLowerCase().includes('fail')
|
||||
return { read, unread, failed }
|
||||
}
|
||||
13
src/index.ts
Normal file
13
src/index.ts
Normal file
@ -0,0 +1,13 @@
|
||||
import 'dotenv/config'
|
||||
import { assertRuntimeConfig, loadProjectConfigs } from './config'
|
||||
import { createServer } from './server'
|
||||
import { startWorker } from './worker'
|
||||
|
||||
const projects = loadProjectConfigs()
|
||||
assertRuntimeConfig(projects)
|
||||
|
||||
const port = Math.max(1, Number(process.env.PORT) || 3088)
|
||||
createServer(projects).listen(port, () => {
|
||||
console.log(`[NotificationCenter] listening on ${port}, projects=${projects.size}`)
|
||||
})
|
||||
startWorker()
|
||||
5
src/prisma.ts
Normal file
5
src/prisma.ts
Normal file
@ -0,0 +1,5 @@
|
||||
import { PrismaClient } from '@prisma/client'
|
||||
|
||||
const prisma = new PrismaClient()
|
||||
|
||||
export default prisma
|
||||
157
src/server.ts
Normal file
157
src/server.ts
Normal file
@ -0,0 +1,157 @@
|
||||
import crypto from 'crypto'
|
||||
import express, { NextFunction, Request, Response } from 'express'
|
||||
import type { ProjectConfig } from './config'
|
||||
import prisma from './prisma'
|
||||
import { createCanonicalSignature, safeEqual, sha256 } from './signature'
|
||||
|
||||
type SignedRequest = Request & { rawBody?: Buffer; project?: ProjectConfig }
|
||||
|
||||
const clean = (value: unknown) => String(value || '').trim()
|
||||
const deliveryView = (row: any) => ({
|
||||
id: row.id,
|
||||
projectCode: row.projectCode,
|
||||
status: row.status,
|
||||
recipientUserId: row.recipientUserId,
|
||||
title: row.renderedTitle,
|
||||
processQueryKey: row.processQueryKey,
|
||||
lastErrorCode: row.lastErrorCode,
|
||||
lastErrorMessage: row.lastErrorMessage,
|
||||
sentAt: row.sentAt,
|
||||
readAt: row.readAt,
|
||||
createdAt: row.createdAt,
|
||||
updatedAt: row.updatedAt,
|
||||
})
|
||||
|
||||
const validateActionUrl = (value: string, project: ProjectConfig) => {
|
||||
let url: URL
|
||||
try {
|
||||
url = new URL(value)
|
||||
} catch {
|
||||
throw new Error('actionUrl 必须是完整 HTTPS 地址')
|
||||
}
|
||||
if (url.protocol !== 'https:') throw new Error('actionUrl 只允许 HTTPS')
|
||||
if (!project.allowedHosts.includes(url.hostname.toLowerCase())) {
|
||||
throw new Error(`actionUrl 域名不在项目白名单内: ${url.hostname}`)
|
||||
}
|
||||
return url.toString()
|
||||
}
|
||||
|
||||
export const createServer = (projects: Map<string, ProjectConfig>) => {
|
||||
const app = express()
|
||||
app.disable('x-powered-by')
|
||||
app.use(express.json({
|
||||
limit: '64kb',
|
||||
verify: (req, _res, buffer) => { (req as SignedRequest).rawBody = Buffer.from(buffer) },
|
||||
}))
|
||||
|
||||
app.get('/health', async (_req, res) => {
|
||||
try {
|
||||
await prisma.$queryRaw`SELECT 1`
|
||||
res.json({ ok: true, service: 'youyidian-notification-center', database: 'ready' })
|
||||
} catch {
|
||||
res.status(503).json({ ok: false, service: 'youyidian-notification-center', database: 'unavailable' })
|
||||
}
|
||||
})
|
||||
|
||||
const authenticate = async (req: SignedRequest, res: Response, next: NextFunction) => {
|
||||
try {
|
||||
const projectCode = clean(req.header('x-yyd-project-code'))
|
||||
const timestamp = clean(req.header('x-yyd-timestamp'))
|
||||
const nonce = clean(req.header('x-yyd-nonce'))
|
||||
const signature = clean(req.header('x-yyd-signature'))
|
||||
const project = projects.get(projectCode)
|
||||
if (!project || !timestamp || !nonce || !signature) return res.status(401).json({ error: 'AUTH_REQUIRED' })
|
||||
const timestampMs = Number(timestamp) * 1000
|
||||
if (!Number.isFinite(timestampMs) || Math.abs(Date.now() - timestampMs) > 5 * 60_000) {
|
||||
return res.status(401).json({ error: 'TIMESTAMP_EXPIRED' })
|
||||
}
|
||||
if (!/^[a-zA-Z0-9_-]{16,100}$/.test(nonce)) return res.status(401).json({ error: 'NONCE_INVALID' })
|
||||
const expected = createCanonicalSignature({
|
||||
secret: project.secret,
|
||||
timestamp,
|
||||
nonce,
|
||||
method: req.method,
|
||||
path: req.path,
|
||||
body: req.rawBody || Buffer.alloc(0),
|
||||
})
|
||||
if (!safeEqual(expected, signature)) return res.status(401).json({ error: 'SIGNATURE_INVALID' })
|
||||
|
||||
await prisma.requestNonce.deleteMany({ where: { expiresAt: { lt: new Date() } } })
|
||||
try {
|
||||
await prisma.requestNonce.create({ data: { projectCode, nonce, expiresAt: new Date(timestampMs + 5 * 60_000) } })
|
||||
} catch {
|
||||
return res.status(409).json({ error: 'NONCE_REPLAYED' })
|
||||
}
|
||||
req.project = project
|
||||
next()
|
||||
} catch (error) {
|
||||
next(error)
|
||||
}
|
||||
}
|
||||
|
||||
app.post('/v1/notifications', authenticate, async (req: SignedRequest, res, next) => {
|
||||
try {
|
||||
const project = req.project!
|
||||
const idempotencyKey = clean(req.body?.idempotencyKey)
|
||||
const recipientUserId = clean(req.body?.recipientUserId)
|
||||
const title = clean(req.body?.title)
|
||||
const content = clean(req.body?.content)
|
||||
const actionTitle = clean(req.body?.actionTitle) || '查看详情'
|
||||
let actionUrl: string
|
||||
try {
|
||||
actionUrl = validateActionUrl(clean(req.body?.actionUrl), project)
|
||||
} catch (error: any) {
|
||||
return res.status(422).json({ error: 'ACTION_URL_INVALID', message: clean(error?.message) })
|
||||
}
|
||||
if (!idempotencyKey || idempotencyKey.length > 160) return res.status(422).json({ error: 'IDEMPOTENCY_KEY_INVALID' })
|
||||
if (!recipientUserId || !title || !content) return res.status(422).json({ error: 'MESSAGE_FIELDS_REQUIRED' })
|
||||
if (title.length > 100 || content.length > 4800 || actionTitle.length > 20) return res.status(422).json({ error: 'MESSAGE_TOO_LONG' })
|
||||
if (title.startsWith('【由壹点')) return res.status(422).json({ error: 'TITLE_BRAND_RESERVED' })
|
||||
|
||||
const renderedTitle = `【由壹点·${project.name}】${title}`
|
||||
const normalized = { recipientUserId, title, content, actionTitle, actionUrl }
|
||||
const requestHash = sha256(JSON.stringify(normalized))
|
||||
const existing = await prisma.notificationDelivery.findUnique({
|
||||
where: { projectCode_idempotencyKey: { projectCode: project.code, idempotencyKey } },
|
||||
})
|
||||
if (existing) {
|
||||
if (existing.requestHash !== requestHash) return res.status(409).json({ error: 'IDEMPOTENCY_CONFLICT' })
|
||||
return res.status(200).json(deliveryView(existing))
|
||||
}
|
||||
|
||||
const created = await prisma.notificationDelivery.create({
|
||||
data: {
|
||||
id: crypto.randomUUID(),
|
||||
projectCode: project.code,
|
||||
idempotencyKey,
|
||||
requestHash,
|
||||
recipientUserId,
|
||||
businessTitle: title,
|
||||
renderedTitle,
|
||||
content,
|
||||
actionTitle,
|
||||
actionUrl,
|
||||
},
|
||||
})
|
||||
res.status(202).json(deliveryView(created))
|
||||
} catch (error) {
|
||||
next(error)
|
||||
}
|
||||
})
|
||||
|
||||
app.get('/v1/notifications/:id', authenticate, async (req: SignedRequest, res, next) => {
|
||||
try {
|
||||
const delivery = await prisma.notificationDelivery.findFirst({ where: { id: String(req.params.id), projectCode: req.project!.code } })
|
||||
if (!delivery) return res.status(404).json({ error: 'NOT_FOUND' })
|
||||
res.json(deliveryView(delivery))
|
||||
} catch (error) {
|
||||
next(error)
|
||||
}
|
||||
})
|
||||
|
||||
app.use((error: any, _req: Request, res: Response, _next: NextFunction) => {
|
||||
console.error('[NotificationCenter]', error)
|
||||
res.status(500).json({ error: 'INTERNAL_ERROR', message: clean(error?.message) || '服务异常' })
|
||||
})
|
||||
return app
|
||||
}
|
||||
27
src/signature.ts
Normal file
27
src/signature.ts
Normal file
@ -0,0 +1,27 @@
|
||||
import crypto from 'crypto'
|
||||
|
||||
export const sha256 = (value: string | Buffer) => crypto.createHash('sha256').update(value).digest('hex')
|
||||
|
||||
export const createCanonicalSignature = (input: {
|
||||
secret: string
|
||||
timestamp: string
|
||||
nonce: string
|
||||
method: string
|
||||
path: string
|
||||
body: string | Buffer
|
||||
}) => {
|
||||
const canonical = [
|
||||
input.timestamp,
|
||||
input.nonce,
|
||||
input.method.toUpperCase(),
|
||||
input.path,
|
||||
sha256(input.body),
|
||||
].join('\n')
|
||||
return crypto.createHmac('sha256', input.secret).update(canonical).digest('hex')
|
||||
}
|
||||
|
||||
export const safeEqual = (left: string, right: string) => {
|
||||
const a = Buffer.from(left)
|
||||
const b = Buffer.from(right)
|
||||
return a.length === b.length && crypto.timingSafeEqual(a, b)
|
||||
}
|
||||
123
src/worker.ts
Normal file
123
src/worker.ts
Normal file
@ -0,0 +1,123 @@
|
||||
import os from 'os'
|
||||
import prisma from './prisma'
|
||||
import { DingTalkError, queryRobotReadStatus, sendRobotMessage } from './dingtalk'
|
||||
|
||||
const workerId = `${os.hostname()}:${process.pid}:${Math.random().toString(36).slice(2, 10)}`
|
||||
const backoffMinutes = [1, 5, 15, 30]
|
||||
let busy = false
|
||||
let readBusy = false
|
||||
|
||||
const claimNext = async () => {
|
||||
const now = new Date()
|
||||
const claimable = {
|
||||
OR: [
|
||||
{ status: { in: ['pending', 'retrying'] }, nextAttemptAt: { lte: now } },
|
||||
{ status: 'sending', leaseExpiresAt: { lte: now } },
|
||||
],
|
||||
}
|
||||
const candidate = await prisma.notificationDelivery.findFirst({ where: claimable, orderBy: [{ nextAttemptAt: 'asc' }, { createdAt: 'asc' }] })
|
||||
if (!candidate) return null
|
||||
const claimed = await prisma.notificationDelivery.updateMany({
|
||||
where: { id: candidate.id, ...claimable },
|
||||
data: {
|
||||
status: 'sending',
|
||||
leaseOwner: workerId,
|
||||
leaseExpiresAt: new Date(Date.now() + 120_000),
|
||||
attemptCount: { increment: 1 },
|
||||
},
|
||||
})
|
||||
if (claimed.count !== 1) return null
|
||||
return prisma.notificationDelivery.findUnique({ where: { id: candidate.id } })
|
||||
}
|
||||
|
||||
const processOne = async () => {
|
||||
const delivery = await claimNext()
|
||||
if (!delivery) return
|
||||
try {
|
||||
const processQueryKey = await sendRobotMessage({
|
||||
recipientUserId: delivery.recipientUserId,
|
||||
title: delivery.renderedTitle,
|
||||
content: delivery.content,
|
||||
actionTitle: delivery.actionTitle,
|
||||
actionUrl: delivery.actionUrl,
|
||||
})
|
||||
await prisma.notificationDelivery.updateMany({
|
||||
where: { id: delivery.id, status: 'sending', leaseOwner: workerId },
|
||||
data: {
|
||||
status: 'sent',
|
||||
processQueryKey,
|
||||
sentAt: new Date(),
|
||||
nextAttemptAt: new Date(Date.now() + 60_000),
|
||||
leaseOwner: null,
|
||||
leaseExpiresAt: null,
|
||||
lastErrorCode: null,
|
||||
lastErrorMessage: null,
|
||||
},
|
||||
})
|
||||
} catch (error: any) {
|
||||
const deterministic = error instanceof DingTalkError && error.deterministic
|
||||
const exhausted = delivery.attemptCount >= delivery.maxAttempts
|
||||
const failed = deterministic || exhausted
|
||||
const delay = backoffMinutes[Math.min(Math.max(0, delivery.attemptCount - 1), backoffMinutes.length - 1)]
|
||||
await prisma.notificationDelivery.updateMany({
|
||||
where: { id: delivery.id, status: 'sending', leaseOwner: workerId },
|
||||
data: {
|
||||
status: failed ? 'failed' : 'retrying',
|
||||
nextAttemptAt: failed ? delivery.nextAttemptAt : new Date(Date.now() + delay * 60_000),
|
||||
leaseOwner: null,
|
||||
leaseExpiresAt: null,
|
||||
lastErrorCode: error instanceof DingTalkError ? error.code : 'UNKNOWN',
|
||||
lastErrorMessage: String(error?.message || '通知发送失败').slice(0, 2000),
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
const refreshOne = async () => {
|
||||
const delivery = await prisma.notificationDelivery.findFirst({
|
||||
where: {
|
||||
status: { in: ['sent', 'unread'] },
|
||||
processQueryKey: { not: null },
|
||||
nextAttemptAt: { lte: new Date() },
|
||||
},
|
||||
orderBy: { updatedAt: 'asc' },
|
||||
})
|
||||
if (!delivery?.processQueryKey) return
|
||||
try {
|
||||
const result = await queryRobotReadStatus(delivery.processQueryKey)
|
||||
if (result.read) {
|
||||
await prisma.notificationDelivery.update({ where: { id: delivery.id }, data: { status: 'read', readAt: new Date() } })
|
||||
} else if (result.failed) {
|
||||
await prisma.notificationDelivery.update({ where: { id: delivery.id }, data: { status: 'failed', lastErrorCode: 'SEND_RESULT_FAILED', lastErrorMessage: '钉钉返回发送失败' } })
|
||||
} else {
|
||||
await prisma.notificationDelivery.update({ where: { id: delivery.id }, data: { status: result.unread ? 'unread' : delivery.status, nextAttemptAt: new Date(Date.now() + 5 * 60_000) } })
|
||||
}
|
||||
} catch (error: any) {
|
||||
await prisma.notificationDelivery.update({
|
||||
where: { id: delivery.id },
|
||||
data: { lastErrorCode: error instanceof DingTalkError ? error.code : 'READ_STATUS_FAILED', lastErrorMessage: String(error?.message || '查询已读状态失败').slice(0, 2000), nextAttemptAt: new Date(Date.now() + 5 * 60_000) },
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
export const startWorker = () => {
|
||||
if (process.env.WORKER_ENABLED === 'false') return
|
||||
const interval = Math.max(500, Number(process.env.WORKER_INTERVAL_MS) || 1000)
|
||||
const readInterval = Math.max(2000, Number(process.env.READ_STATUS_INTERVAL_MS) || 5000)
|
||||
const tick = async () => {
|
||||
if (busy) return
|
||||
busy = true
|
||||
try {
|
||||
await processOne()
|
||||
} finally {
|
||||
busy = false
|
||||
}
|
||||
}
|
||||
setInterval(() => void tick(), interval).unref()
|
||||
setInterval(() => {
|
||||
if (readBusy) return
|
||||
readBusy = true
|
||||
void refreshOne().finally(() => { readBusy = false })
|
||||
}, readInterval).unref()
|
||||
setTimeout(() => void tick(), 500).unref()
|
||||
}
|
||||
17
tsconfig.json
Normal file
17
tsconfig.json
Normal file
@ -0,0 +1,17 @@
|
||||
{
|
||||
"compilerOptions": {
|
||||
"target": "ES2022",
|
||||
"module": "commonjs",
|
||||
"moduleResolution": "node",
|
||||
"outDir": "dist",
|
||||
"rootDir": "src",
|
||||
"strict": true,
|
||||
"esModuleInterop": true,
|
||||
"skipLibCheck": true,
|
||||
"forceConsistentCasingInFileNames": true,
|
||||
"resolveJsonModule": true,
|
||||
"types": ["node"]
|
||||
},
|
||||
"include": ["src/**/*.ts"],
|
||||
"exclude": ["node_modules", "dist"]
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user