Commit bfb60fbd by Xianquan Committed by GitHub

feat: add duplicate chat cleanup API (#7223)

parent ffc2d67d
import { NextAPI } from '@/service/middleware/entry';
import { parseApiInput } from '@fastgpt/service/common/zod/requestParseError';
import { MongoChat } from '@fastgpt/service/core/chat/chatSchema';
import { authCert } from '@fastgpt/service/support/permission/auth/common';
import type { ApiRequestProps } from '@fastgpt/service/type/next';
import { BoolSchema, IntSchema } from '@fastgpt/global/common/zod';
import z from 'zod';
/* ============================================================================
* API: 清理重复 Chat 会话头
* Route: POST /api/admin/dataClean/cleanupDuplicateChats
* Method: POST
* Description: 管理员数据清洗接口,按 appId + chatId 查找重复 chats 记录,保留 updateTime 最新的一条并可选择删除其余会话头。
* Tags: ['Admin', 'DataClean', 'Chat', 'Delete']
* ============================================================================ */
const DEFAULT_SAMPLE_LIMIT = 20;
const CleanupDuplicateChatsBodySchema = z
.object({
dryRun: BoolSchema.optional().meta({
example: true,
description: '是否只扫描统计不删除'
}),
dryrun: BoolSchema.optional().meta({
example: true,
description: '是否只扫描统计不删除,兼容小写参数'
}),
sampleLimit: IntSchema.refine((value) => value >= 0 && value <= 100)
.optional()
.meta({
example: DEFAULT_SAMPLE_LIMIT,
description: '返回重复组样本数量,范围 0~100'
})
})
.transform((body) => ({
dryRun: body.dryRun ?? body.dryrun ?? true,
sampleLimit: body.sampleLimit ?? DEFAULT_SAMPLE_LIMIT
}));
export type CleanupDuplicateChatsBodyType = z.infer<typeof CleanupDuplicateChatsBodySchema>;
const DuplicateChatGroupSampleSchema = z.object({
appId: z.string().meta({ description: '应用 ID 或历史 sourceId' }),
chatId: z.string().meta({ description: '会话 ID' }),
totalCount: z.number().int().nonnegative().meta({ description: '该组会话头总数' }),
duplicateCount: z.number().int().nonnegative().meta({ description: '该组预计删除数量' }),
keepId: z.string().optional().meta({ description: '保留的 chats 记录 ID' }),
deleteIds: z.array(z.string()).meta({ description: '预计删除的 chats 记录 ID 样本' })
});
export type DuplicateChatGroupSampleType = z.infer<typeof DuplicateChatGroupSampleSchema>;
const CleanupDuplicateChatsResponseSchema = z.object({
dryRun: z.boolean().meta({ description: '是否 dryRun' }),
scannedDuplicateGroupCount: z
.number()
.int()
.nonnegative()
.meta({ description: '扫描到的重复 appId + chatId 组数' }),
duplicateDocumentCount: z
.number()
.int()
.nonnegative()
.meta({ description: '预计删除的重复 chats 记录数' }),
deletedDocumentCount: z
.number()
.int()
.nonnegative()
.meta({ description: '实际删除的 chats 记录数,dryRun 时为 0' }),
sampleLimit: z.number().int().nonnegative().meta({ description: '返回样本数量限制' }),
samples: z.array(DuplicateChatGroupSampleSchema).meta({ description: '重复组样本' })
});
export type CleanupDuplicateChatsResponseType = z.infer<
typeof CleanupDuplicateChatsResponseSchema
>;
type DuplicateKeyGroup = {
_id: {
appId: unknown;
chatId: string;
};
count: number;
};
type DuplicateChatDoc = {
_id: unknown;
};
const stringifyId = (value: unknown) => {
if (value == null) return '';
if (typeof value === 'object' && 'toString' in value && typeof value.toString === 'function') {
return value.toString();
}
return String(value);
};
const findDuplicateChatGroups = () =>
MongoChat.aggregate<DuplicateKeyGroup>(
[
{
$group: {
_id: { appId: '$appId', chatId: '$chatId' },
count: { $sum: 1 }
}
},
{
$match: {
count: { $gt: 1 },
'_id.appId': { $exists: true, $ne: null },
'_id.chatId': { $exists: true, $type: 'string', $ne: '' }
}
},
{ $sort: { count: -1 } }
],
{ allowDiskUse: true }
);
const findDuplicateChatDocs = (group: DuplicateKeyGroup) =>
MongoChat.find({ appId: group._id.appId, chatId: group._id.chatId }, '_id')
.sort({ updateTime: -1, _id: -1 })
.lean<DuplicateChatDoc[]>();
/**
* 清理 `chats` 中历史重复会话头。
*
* 只删除 `chats` 元数据,不删除 `chatitems` 和 `chat_item_responses` 中的消息内容。
* 对每组重复的 `appId + chatId`,保留 `updateTime` 最新的一条;若时间相同,用 `_id`
* 倒序作为稳定兜底,避免多次 dry-run 与正式执行选择不同记录。
*/
export async function runCleanupDuplicateChatsMigration(
params: CleanupDuplicateChatsBodyType
): Promise<CleanupDuplicateChatsResponseType> {
const duplicateGroups = await findDuplicateChatGroups();
let duplicateDocumentCount = 0;
let deletedDocumentCount = 0;
const samples: DuplicateChatGroupSampleType[] = [];
for (const group of duplicateGroups) {
const docs = await findDuplicateChatDocs(group);
const keepDoc = docs[0];
const duplicateDocs = docs.slice(1);
if (!keepDoc || duplicateDocs.length === 0) {
continue;
}
const deleteIds = duplicateDocs.map((doc) => doc._id);
duplicateDocumentCount += deleteIds.length;
if (samples.length < params.sampleLimit) {
samples.push({
appId: stringifyId(group._id.appId),
chatId: group._id.chatId,
totalCount: docs.length,
duplicateCount: deleteIds.length,
keepId: stringifyId(keepDoc._id),
deleteIds: deleteIds.map(stringifyId)
});
}
if (!params.dryRun) {
const result = await MongoChat.deleteMany({ _id: { $in: deleteIds } });
deletedDocumentCount += result.deletedCount;
}
}
return CleanupDuplicateChatsResponseSchema.parse({
dryRun: params.dryRun,
scannedDuplicateGroupCount: duplicateGroups.length,
duplicateDocumentCount,
deletedDocumentCount,
sampleLimit: params.sampleLimit,
samples
});
}
/**
* 管理员重复会话头清理接口。
*
* 默认 dryRun。正式执行时只删除重复的 `chats` 会话头,消息明细保留不动。
*/
async function handler(req: ApiRequestProps): Promise<CleanupDuplicateChatsResponseType> {
await authCert({ req, authRoot: true });
const { body } = parseApiInput({
req,
bodySchema: CleanupDuplicateChatsBodySchema
});
return runCleanupDuplicateChatsMigration(body);
}
export default NextAPI(handler);
import { beforeEach, describe, expect, it } from 'vitest';
import {
ChatRoleEnum,
ChatSourceEnum,
ChatSourceTypeEnum
} from '@fastgpt/global/core/chat/constants';
import { MongoChat } from '@fastgpt/service/core/chat/chatSchema';
import { MongoChatItem } from '@fastgpt/service/core/chat/chatItemSchema';
import { Types } from '@fastgpt/service/common/mongo';
import { runCleanupDuplicateChatsMigration } from '@/pages/api/admin/dataClean/cleanupDuplicateChats';
const teamId = '65f000000000000000000061';
const tmbId = '65f000000000000000000062';
const appId = '65f000000000000000000063';
const otherAppId = '65f000000000000000000064';
const legacyUniqueIndexNames = ['appId_1_chatId_1', 'sourceType_1_appId_1_chatId_1'] as const;
const ensureLegacyDuplicateWritableCollection = async () => {
for (const indexName of legacyUniqueIndexNames) {
try {
await MongoChat.collection.dropIndex(indexName);
} catch (error) {
const codeName = (error as { codeName?: string }).codeName;
if (codeName !== 'IndexNotFound' && codeName !== 'NamespaceNotFound') {
throw error;
}
}
}
};
const createChatHeader = ({
id,
sourceId = appId,
chatId,
updateTime
}: {
id: string;
sourceId?: string;
chatId: string;
updateTime: Date;
}) => ({
_id: new Types.ObjectId(id),
teamId: new Types.ObjectId(teamId),
tmbId: new Types.ObjectId(tmbId),
sourceType: ChatSourceTypeEnum.app,
appId: new Types.ObjectId(sourceId),
chatId,
source: ChatSourceEnum.online,
title: `chat-${chatId}`,
createTime: new Date('2026-01-01T00:00:00.000Z'),
updateTime
});
const createChatUniqueIndexes = async () => {
await MongoChat.collection.createIndex({ appId: 1, chatId: 1 }, { unique: true });
await MongoChat.collection.createIndex(
{ sourceType: 1, appId: 1, chatId: 1 },
{ unique: true, name: 'sourceType_1_appId_1_chatId_1' }
);
};
describe('cleanupDuplicateChats data clean API', () => {
beforeEach(async () => {
await ensureLegacyDuplicateWritableCollection();
});
it('dry-runs duplicate chat headers without deleting data', async () => {
await MongoChat.collection.insertMany([
createChatHeader({
id: '65f000000000000000000101',
chatId: 'duplicate-chat',
updateTime: new Date('2026-01-01T00:00:00.000Z')
}),
createChatHeader({
id: '65f000000000000000000102',
chatId: 'duplicate-chat',
updateTime: new Date('2026-01-02T00:00:00.000Z')
}),
createChatHeader({
id: '65f000000000000000000103',
chatId: 'unique-chat',
updateTime: new Date('2026-01-03T00:00:00.000Z')
})
]);
const result = await runCleanupDuplicateChatsMigration({
dryRun: true,
sampleLimit: 10
});
expect(result).toMatchObject({
dryRun: true,
scannedDuplicateGroupCount: 1,
duplicateDocumentCount: 1,
deletedDocumentCount: 0,
samples: [
{
appId,
chatId: 'duplicate-chat',
totalCount: 2,
duplicateCount: 1,
keepId: '65f000000000000000000102',
deleteIds: ['65f000000000000000000101']
}
]
});
expect(await MongoChat.countDocuments({ appId, chatId: 'duplicate-chat' })).toBe(2);
});
it('deletes only duplicate chat headers and keeps chat items untouched', async () => {
await MongoChat.collection.insertMany([
createChatHeader({
id: '65f000000000000000000201',
chatId: 'duplicate-chat',
updateTime: new Date('2026-01-01T00:00:00.000Z')
}),
createChatHeader({
id: '65f000000000000000000202',
chatId: 'duplicate-chat',
updateTime: new Date('2026-01-02T00:00:00.000Z')
}),
createChatHeader({
id: '65f000000000000000000203',
chatId: 'duplicate-chat',
updateTime: new Date('2026-01-02T00:00:00.000Z')
}),
createChatHeader({
id: '65f000000000000000000204',
sourceId: otherAppId,
chatId: 'duplicate-chat',
updateTime: new Date('2026-01-03T00:00:00.000Z')
})
]);
await MongoChatItem.create({
teamId,
tmbId,
appId,
chatId: 'duplicate-chat',
dataId: 'item-1',
obj: ChatRoleEnum.AI,
value: [{ type: 'text', text: { content: 'answer' } }]
});
const result = await runCleanupDuplicateChatsMigration({
dryRun: false,
sampleLimit: 10
});
expect(result).toMatchObject({
dryRun: false,
scannedDuplicateGroupCount: 1,
duplicateDocumentCount: 2,
deletedDocumentCount: 2,
samples: [
{
appId,
chatId: 'duplicate-chat',
totalCount: 3,
duplicateCount: 2,
keepId: '65f000000000000000000203',
deleteIds: ['65f000000000000000000202', '65f000000000000000000201']
}
]
});
const keptChats = await MongoChat.find({ chatId: 'duplicate-chat' }).sort({ appId: 1 }).lean();
expect(keptChats.map((chat) => String(chat._id)).sort()).toEqual([
'65f000000000000000000203',
'65f000000000000000000204'
]);
expect(await MongoChatItem.countDocuments({ appId, chatId: 'duplicate-chat' })).toBe(1);
});
it('allows unique chat indexes to be created after cleanup', async () => {
await MongoChat.collection.insertMany([
createChatHeader({
id: '65f000000000000000000401',
chatId: 'duplicate-chat',
updateTime: new Date('2026-01-01T00:00:00.000Z')
}),
createChatHeader({
id: '65f000000000000000000402',
chatId: 'duplicate-chat',
updateTime: new Date('2026-01-02T00:00:00.000Z')
}),
createChatHeader({
id: '65f000000000000000000403',
chatId: 'unique-chat',
updateTime: new Date('2026-01-03T00:00:00.000Z')
})
]);
await expect(createChatUniqueIndexes()).rejects.toThrow(/duplicate key/i);
await ensureLegacyDuplicateWritableCollection();
await runCleanupDuplicateChatsMigration({
dryRun: false,
sampleLimit: 10
});
await expect(createChatUniqueIndexes()).resolves.toBeUndefined();
const indexes = await MongoChat.collection.indexes();
expect(
indexes.find((index) => index.name === 'appId_1_chatId_1' && index.unique === true)
).toBeTruthy();
expect(
indexes.find(
(index) => index.name === 'sourceType_1_appId_1_chatId_1' && index.unique === true
)
).toBeTruthy();
});
it('ignores invalid duplicate keys with empty chatId', async () => {
await MongoChat.collection.insertMany([
createChatHeader({
id: '65f000000000000000000301',
chatId: '',
updateTime: new Date('2026-01-01T00:00:00.000Z')
}),
createChatHeader({
id: '65f000000000000000000302',
chatId: '',
updateTime: new Date('2026-01-02T00:00:00.000Z')
})
]);
const result = await runCleanupDuplicateChatsMigration({
dryRun: false,
sampleLimit: 10
});
expect(result).toMatchObject({
scannedDuplicateGroupCount: 0,
duplicateDocumentCount: 0,
deletedDocumentCount: 0,
samples: []
});
expect(await MongoChat.countDocuments({ appId, chatId: '' })).toBe(2);
});
});
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment