feat(msg): 修复通知服务并添加ES降级与Push Gateway推送
database.ts 导出db常量替代getDb()函数 env.ts JWT_SECRET/ES_URL改optional加DEV_MODE/PUSH_GATEWAY_URL elasticsearch.ts ES降级: ES_URL未设置时esClient=null notifications.service.ts 加createBatch+分页查询+Push Gateway推送调用 新建msg-init.sql创建2张表
This commit is contained in:
SpecialX
@@ -1,32 +1,102 @@
|
||||
import { Body, Controller, Get, Param, Post, Query, Req } from '@nestjs/common';
|
||||
import { NotificationsService } from './notifications.service.js';
|
||||
import type { SendNotificationDto } from './notifications.service.js';
|
||||
import {
|
||||
Body,
|
||||
Controller,
|
||||
Get,
|
||||
HttpException,
|
||||
HttpStatus,
|
||||
Param,
|
||||
Post,
|
||||
Put,
|
||||
Query,
|
||||
Req,
|
||||
} from "@nestjs/common";
|
||||
import { NotificationsService } from "./notifications.service.js";
|
||||
import type { SendNotificationDto } from "./notifications.service.js";
|
||||
|
||||
@Controller('notifications')
|
||||
interface AuthedRequest {
|
||||
headers: Record<string, string | string[] | undefined>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 从请求头读取用户身份。
|
||||
* Gateway 在通过鉴权后会注入 x-user-id；如未注入（开发模式或匿名请求）返回 null。
|
||||
*/
|
||||
function getUserIdFromRequest(req: AuthedRequest): string | null {
|
||||
const raw = req.headers["x-user-id"];
|
||||
if (typeof raw === "string" && raw.length > 0) return raw;
|
||||
return null;
|
||||
}
|
||||
|
||||
@Controller("notifications")
|
||||
export class NotificationsController {
|
||||
constructor(private readonly service: NotificationsService) {}
|
||||
|
||||
@Post()
|
||||
async send(@Body() body: unknown) {
|
||||
async send(@Body() body: unknown): Promise<{ success: true; data: unknown }> {
|
||||
const result = await this.service.send(body as SendNotificationDto);
|
||||
return { success: true, data: result };
|
||||
}
|
||||
|
||||
@Get()
|
||||
async list(@Req() req: { userId: string }, @Query('unread') unread: string) {
|
||||
const result = await this.service.listByUser(req.userId, unread === 'true');
|
||||
@Post("batch")
|
||||
async createBatch(
|
||||
@Body() body: unknown,
|
||||
): Promise<{ success: true; data: unknown }> {
|
||||
const dtos = (body as SendNotificationDto[]) ?? [];
|
||||
const result = await this.service.createBatch(dtos);
|
||||
return { success: true, data: result };
|
||||
}
|
||||
|
||||
@Post(':id/read')
|
||||
async markAsRead(@Param('id') id: string) {
|
||||
@Get("user/:userId")
|
||||
async listByUser(
|
||||
@Param("userId") userId: string,
|
||||
@Query("unread") unread: string,
|
||||
): Promise<{ success: true; data: unknown }> {
|
||||
const result = await this.service.listByUser(userId, unread === "true");
|
||||
return { success: true, data: result };
|
||||
}
|
||||
|
||||
@Get("user/:userId/page")
|
||||
async listByUserPaginated(
|
||||
@Param("userId") userId: string,
|
||||
@Query("page") page: string,
|
||||
@Query("pageSize") pageSize: string,
|
||||
): Promise<{ success: true; data: unknown }> {
|
||||
const pageNum = Number(page) > 0 ? Number(page) : 1;
|
||||
const pageSizeNum = Number(pageSize) > 0 ? Number(pageSize) : 20;
|
||||
const result = await this.service.listByUserWithPagination(
|
||||
userId,
|
||||
pageNum,
|
||||
pageSizeNum,
|
||||
);
|
||||
return { success: true, data: result };
|
||||
}
|
||||
|
||||
@Put(":id/read")
|
||||
async markAsRead(@Param("id") id: string): Promise<{ success: true }> {
|
||||
await this.service.markAsRead(id);
|
||||
return { success: true };
|
||||
}
|
||||
|
||||
@Get('search')
|
||||
async search(@Req() req: { userId: string }, @Query('q') q: string) {
|
||||
const result = await this.service.search(req.userId, q);
|
||||
@Get("search")
|
||||
async search(
|
||||
@Req() req: AuthedRequest,
|
||||
@Query("q") q: string,
|
||||
@Query("userId") userIdParam: string,
|
||||
): Promise<{ success: true; data: unknown }> {
|
||||
const userId = getUserIdFromRequest(req) ?? userIdParam;
|
||||
if (!userId) {
|
||||
throw new HttpException(
|
||||
{
|
||||
success: false,
|
||||
error: {
|
||||
code: "MSG_PERMISSION_DENIED",
|
||||
message: "Missing user identity",
|
||||
},
|
||||
},
|
||||
HttpStatus.FORBIDDEN,
|
||||
);
|
||||
}
|
||||
const result = await this.service.search(userId, q);
|
||||
return { success: true, data: result };
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,11 +1,18 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
import { getDb } from '../config/database.js';
|
||||
import { notifications, notificationPreferences } from './notifications.schema.js';
|
||||
import type { NotificationPreference } from './notifications.schema.js';
|
||||
import { v4 as uuidv4 } from 'uuid';
|
||||
import { eq, and } from 'drizzle-orm';
|
||||
import { esClient } from '../config/elasticsearch.js';
|
||||
import { logger } from '../shared/observability/logger.js';
|
||||
import { Injectable } from "@nestjs/common";
|
||||
import { eq, and, desc, sql } from "drizzle-orm";
|
||||
import { v4 as uuidv4 } from "uuid";
|
||||
import { db } from "../config/database.js";
|
||||
import {
|
||||
notifications,
|
||||
notificationPreferences,
|
||||
} from "./notifications.schema.js";
|
||||
import type {
|
||||
Notification,
|
||||
NotificationPreference,
|
||||
} from "./notifications.schema.js";
|
||||
import { safeIndex, safeSearch } from "../config/elasticsearch.js";
|
||||
import { env } from "../config/env.js";
|
||||
import { logger } from "../shared/observability/logger.js";
|
||||
|
||||
export interface SendNotificationDto {
|
||||
userId: string;
|
||||
@@ -16,22 +23,38 @@ export interface SendNotificationDto {
|
||||
metadata?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export interface SendResult {
|
||||
id: string;
|
||||
skipped: boolean;
|
||||
pushed: boolean;
|
||||
}
|
||||
|
||||
export interface PaginatedResult {
|
||||
items: Notification[];
|
||||
total: number;
|
||||
page: number;
|
||||
pageSize: number;
|
||||
}
|
||||
|
||||
@Injectable()
|
||||
export class NotificationsService {
|
||||
async send(dto: SendNotificationDto) {
|
||||
async send(dto: SendNotificationDto): Promise<SendResult> {
|
||||
const id = uuidv4();
|
||||
const db = getDb();
|
||||
const channel = dto.channel ?? "in_app";
|
||||
|
||||
// 检查用户偏好
|
||||
const [pref] = await db.select().from(notificationPreferences).where(eq(notificationPreferences.userId, dto.userId));
|
||||
const [pref] = await db
|
||||
.select()
|
||||
.from(notificationPreferences)
|
||||
.where(eq(notificationPreferences.userId, dto.userId));
|
||||
|
||||
const channel = dto.channel || 'in_app';
|
||||
if (pref && !this.isChannelEnabled(pref, channel)) {
|
||||
logger.info({ userId: dto.userId, channel }, 'Notification skipped by preference');
|
||||
return { skipped: true };
|
||||
logger.info(
|
||||
{ userId: dto.userId, channel },
|
||||
"Notification skipped by preference",
|
||||
);
|
||||
return { id, skipped: true, pushed: false };
|
||||
}
|
||||
|
||||
// 写入 DB
|
||||
await db.insert(notifications).values({
|
||||
id,
|
||||
userId: dto.userId,
|
||||
@@ -42,69 +65,148 @@ export class NotificationsService {
|
||||
metadata: dto.metadata,
|
||||
});
|
||||
|
||||
// 索引到 ES（全文检索）
|
||||
await safeIndex({
|
||||
index: "notifications",
|
||||
id,
|
||||
document: {
|
||||
userId: dto.userId,
|
||||
type: dto.type,
|
||||
title: dto.title,
|
||||
content: dto.content,
|
||||
channel,
|
||||
createdAt: new Date().toISOString(),
|
||||
},
|
||||
});
|
||||
|
||||
const pushed = await this.notifyPushGateway(dto, channel);
|
||||
|
||||
return { id, skipped: false, pushed };
|
||||
}
|
||||
|
||||
async createBatch(dtos: SendNotificationDto[]): Promise<SendResult[]> {
|
||||
const results: SendResult[] = [];
|
||||
for (const dto of dtos) {
|
||||
const result = await this.send(dto);
|
||||
results.push(result);
|
||||
}
|
||||
return results;
|
||||
}
|
||||
|
||||
async listByUser(
|
||||
userId: string,
|
||||
onlyUnread: boolean = false,
|
||||
): Promise<Notification[]> {
|
||||
const conditions = onlyUnread
|
||||
? and(eq(notifications.userId, userId), eq(notifications.isRead, false))
|
||||
: eq(notifications.userId, userId);
|
||||
return db
|
||||
.select()
|
||||
.from(notifications)
|
||||
.where(conditions)
|
||||
.orderBy(desc(notifications.createdAt));
|
||||
}
|
||||
|
||||
async listByUserWithPagination(
|
||||
userId: string,
|
||||
page: number,
|
||||
pageSize: number,
|
||||
): Promise<PaginatedResult> {
|
||||
const offset = (page - 1) * pageSize;
|
||||
const items = await db
|
||||
.select()
|
||||
.from(notifications)
|
||||
.where(eq(notifications.userId, userId))
|
||||
.orderBy(desc(notifications.createdAt))
|
||||
.limit(pageSize)
|
||||
.offset(offset);
|
||||
|
||||
const [countRow] = await db
|
||||
.select({ count: sql<number>`count(*)` })
|
||||
.from(notifications)
|
||||
.where(eq(notifications.userId, userId));
|
||||
|
||||
const total = countRow ? Number(countRow.count) : 0;
|
||||
|
||||
return { items, total, page, pageSize };
|
||||
}
|
||||
|
||||
async markAsRead(id: string): Promise<void> {
|
||||
await db
|
||||
.update(notifications)
|
||||
.set({ isRead: true })
|
||||
.where(eq(notifications.id, id));
|
||||
}
|
||||
|
||||
async search(userId: string, query: string): Promise<unknown[]> {
|
||||
const result = await safeSearch("notifications", {
|
||||
bool: {
|
||||
must: [
|
||||
{ term: { userId } },
|
||||
{
|
||||
multi_match: {
|
||||
query,
|
||||
fields: ["title", "content"],
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
return result.hits;
|
||||
}
|
||||
|
||||
private isChannelEnabled(
|
||||
pref: NotificationPreference,
|
||||
channel: string,
|
||||
): boolean {
|
||||
switch (channel) {
|
||||
case "email":
|
||||
return pref.emailEnabled;
|
||||
case "sms":
|
||||
return pref.smsEnabled;
|
||||
case "push":
|
||||
return pref.pushEnabled;
|
||||
case "in_app":
|
||||
return pref.inAppEnabled;
|
||||
default:
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 调用 Push Gateway 推送通知。Push Gateway 不可用时 try/catch 跳过（降级模式）。
|
||||
* 仅在 PUSH_GATEWAY_URL 配置且 channel 为 push 或 in_app 时触发。
|
||||
*/
|
||||
private async notifyPushGateway(
|
||||
dto: SendNotificationDto,
|
||||
channel: string,
|
||||
): Promise<boolean> {
|
||||
if (!env.PUSH_GATEWAY_URL) return false;
|
||||
if (channel !== "push" && channel !== "in_app") return false;
|
||||
|
||||
try {
|
||||
await esClient.index({
|
||||
index: 'notifications',
|
||||
id,
|
||||
document: {
|
||||
const res = await fetch(`${env.PUSH_GATEWAY_URL}/internal/push`, {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({
|
||||
userId: dto.userId,
|
||||
type: dto.type,
|
||||
title: dto.title,
|
||||
content: dto.content,
|
||||
channel,
|
||||
createdAt: new Date().toISOString(),
|
||||
},
|
||||
metadata: dto.metadata,
|
||||
}),
|
||||
});
|
||||
if (!res.ok) {
|
||||
logger.warn(
|
||||
{ status: res.status },
|
||||
"Push gateway returned non-ok status",
|
||||
);
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
} catch (err) {
|
||||
logger.error({ err }, 'Failed to index notification in ES');
|
||||
}
|
||||
|
||||
// TODO: 触发 Push Gateway 推送（P5 后期）
|
||||
|
||||
return { id, skipped: false };
|
||||
}
|
||||
|
||||
async listByUser(userId: string, onlyUnread: boolean = false) {
|
||||
const db = getDb();
|
||||
const conditions = onlyUnread
|
||||
? and(eq(notifications.userId, userId), eq(notifications.isRead, false))
|
||||
: eq(notifications.userId, userId);
|
||||
return db.select().from(notifications).where(conditions);
|
||||
}
|
||||
|
||||
async markAsRead(id: string) {
|
||||
const db = getDb();
|
||||
await db.update(notifications).set({ isRead: true }).where(eq(notifications.id, id));
|
||||
}
|
||||
|
||||
async search(userId: string, query: string) {
|
||||
const result = await esClient.search({
|
||||
index: 'notifications',
|
||||
query: {
|
||||
bool: {
|
||||
must: [
|
||||
{ term: { userId } },
|
||||
{
|
||||
multi_match: {
|
||||
query,
|
||||
fields: ['title', 'content'],
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
});
|
||||
return result.hits.hits;
|
||||
}
|
||||
|
||||
private isChannelEnabled(pref: NotificationPreference, channel: string): boolean {
|
||||
switch (channel) {
|
||||
case 'email': return pref.emailEnabled;
|
||||
case 'sms': return pref.smsEnabled;
|
||||
case 'push': return pref.pushEnabled;
|
||||
case 'in_app': return pref.inAppEnabled;
|
||||
default: return true;
|
||||
logger.warn({ err }, "Push gateway unavailable, skipping push");
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user