447 أسطر
16 KiB
TypeScript
447 أسطر
16 KiB
TypeScript
import { BadRequestException, ForbiddenException, Injectable, Logger, NotFoundException } from '@nestjs/common';
|
|
import { Types } from 'mongoose';
|
|
import { extname } from 'path';
|
|
import { ReactionType } from '../../common/enums/reaction-type.enum';
|
|
import { decodeOffsetCursor, encodeOffsetCursor } from '../../common/utils/cursor.util';
|
|
import { buildPaginatedResponse } from '../../common/utils/pagination.util';
|
|
import { resolveMongoSortDirection } from '../../common/utils/sort.util';
|
|
import { ManagedStorageService } from '../../infrastructure/storage/managed-storage.service';
|
|
import { NotificationsService } from '../notifications/notifications.service';
|
|
import { UsersRepository } from '../users/users.repository';
|
|
import { CreateConversationDto } from './dto/create-conversation.dto';
|
|
import { MessageQueryDto } from './dto/message-query.dto';
|
|
import { SendMessageDto } from './dto/send-message.dto';
|
|
import { ChatRealtimeService } from './chat-realtime.service';
|
|
import { ChatRepository } from './chat.repository';
|
|
|
|
@Injectable()
|
|
export class ChatService {
|
|
private readonly logger = new Logger(ChatService.name);
|
|
|
|
constructor(
|
|
private readonly chatRepository: ChatRepository,
|
|
private readonly usersRepository: UsersRepository,
|
|
private readonly notificationsService: NotificationsService,
|
|
private readonly storageService: ManagedStorageService,
|
|
private readonly chatRealtimeService: ChatRealtimeService,
|
|
) {}
|
|
|
|
async createConversation(currentUserId: string, dto: CreateConversationDto) {
|
|
const uniqueParticipantIds = Array.from(new Set([currentUserId, ...dto.participantIds]));
|
|
if (uniqueParticipantIds.length < 2) {
|
|
throw new BadRequestException('Conversation must include at least 2 participants');
|
|
}
|
|
|
|
for (const participantId of uniqueParticipantIds) {
|
|
if (!Types.ObjectId.isValid(participantId)) {
|
|
throw new BadRequestException('Invalid participant id');
|
|
}
|
|
}
|
|
|
|
const users = await Promise.all(uniqueParticipantIds.map((id) => this.usersRepository.findById(id)));
|
|
if (users.some((u) => !u || u.isDisabled)) {
|
|
throw new BadRequestException('One or more participants are invalid or disabled');
|
|
}
|
|
|
|
const isGroup = dto.isGroup ?? uniqueParticipantIds.length > 2;
|
|
if (!isGroup && uniqueParticipantIds.length !== 2) {
|
|
throw new BadRequestException('Direct conversation must contain exactly 2 participants');
|
|
}
|
|
|
|
if (!isGroup) {
|
|
const otherId = uniqueParticipantIds.find((id) => id !== currentUserId) as string;
|
|
const block = await this.chatRepository.findAnyBlockBetween(currentUserId, otherId);
|
|
if (block) {
|
|
throw new ForbiddenException('You cannot start chat with this user');
|
|
}
|
|
const existing = await this.chatRepository.findDirectConversation(currentUserId, otherId);
|
|
if (existing) {
|
|
return existing;
|
|
}
|
|
}
|
|
|
|
return this.chatRepository.createConversation({
|
|
participantIds: uniqueParticipantIds,
|
|
isGroup,
|
|
title: dto.title,
|
|
createdBy: currentUserId,
|
|
});
|
|
}
|
|
|
|
async getMyConversations(currentUserId: string, query: MessageQueryDto) {
|
|
const page = query.page ?? 1;
|
|
const limit = query.limit ?? 20;
|
|
const cursorOffset = decodeOffsetCursor(query.cursor);
|
|
const skip = cursorOffset ?? (page - 1) * limit;
|
|
const direction = resolveMongoSortDirection(query.sortOrder);
|
|
|
|
const [items, total] = await Promise.all([
|
|
this.chatRepository.findConversationsForUser(currentUserId, skip, limit, {
|
|
lastMessageAt: direction,
|
|
updatedAt: direction,
|
|
}),
|
|
this.chatRepository.countConversationsForUser(currentUserId),
|
|
]);
|
|
|
|
const mappedItems = items.map((conversation) => {
|
|
const conversationObject = conversation.toObject();
|
|
const unreadMap = this.normalizeUnreadCountByUser(conversationObject.unreadCountByUser);
|
|
return {
|
|
...conversationObject,
|
|
unreadCountByUser: unreadMap,
|
|
unreadCount: unreadMap[currentUserId] ?? 0,
|
|
lastMessageAt: conversation.lastMessageAt ?? null,
|
|
};
|
|
});
|
|
const nextOffset = skip + mappedItems.length;
|
|
const nextCursor = nextOffset < total ? encodeOffsetCursor(nextOffset) : null;
|
|
|
|
return buildPaginatedResponse(mappedItems, {
|
|
page,
|
|
limit,
|
|
total,
|
|
offset: skip,
|
|
currentCursor: query.cursor ?? null,
|
|
nextCursor,
|
|
mode: 'cursor',
|
|
});
|
|
}
|
|
|
|
async getMessages(currentUserId: string, conversationId: string, query: MessageQueryDto) {
|
|
if (!conversationId?.trim()) {
|
|
throw new BadRequestException('conversationId is required');
|
|
}
|
|
|
|
const conversation = await this.assertConversationMember(currentUserId, conversationId);
|
|
const page = query.page ?? 1;
|
|
const limit = query.limit ?? 20;
|
|
const cursorOffset = decodeOffsetCursor(query.cursor);
|
|
const skip = cursorOffset ?? (page - 1) * limit;
|
|
const sort = { createdAt: resolveMongoSortDirection(query.sortOrder) } as Record<string, 1 | -1>;
|
|
|
|
const [items, total] = await Promise.all([
|
|
this.chatRepository.findMessages(conversation.id, currentUserId, skip, limit, sort),
|
|
this.chatRepository.countMessages(conversation.id, currentUserId),
|
|
]);
|
|
|
|
await this.chatRepository.clearConversationUnreadForUser(conversation.id, currentUserId);
|
|
const nextOffset = skip + items.length;
|
|
const nextCursor = nextOffset < total ? encodeOffsetCursor(nextOffset) : null;
|
|
|
|
return buildPaginatedResponse(items, {
|
|
page,
|
|
limit,
|
|
total,
|
|
offset: skip,
|
|
currentCursor: query.cursor ?? null,
|
|
nextCursor,
|
|
mode: 'cursor',
|
|
});
|
|
}
|
|
|
|
async sendMessage(currentUserId: string, dto: SendMessageDto) {
|
|
const conversation = await this.assertConversationMember(currentUserId, dto.conversationId);
|
|
await this.assertNoChatBlockInConversation(currentUserId, conversation.participantIds.map((id) => id.toString()));
|
|
const messageType = dto.messageType ?? 'text';
|
|
const content = dto.content?.trim() ?? '';
|
|
const mediaUrl = dto.mediaUrl?.trim() ?? '';
|
|
|
|
if (messageType === 'text' && !content) {
|
|
throw new BadRequestException('Text message content is required');
|
|
}
|
|
|
|
if (messageType !== 'text' && !mediaUrl) {
|
|
throw new BadRequestException('mediaUrl is required for non-text messages');
|
|
}
|
|
|
|
const replyToMessageId = await this.resolveReplyToMessageId(dto.replyToMessageId, conversation.id);
|
|
const message = await this.chatRepository.createMessage({
|
|
conversationId: conversation.id,
|
|
senderId: currentUserId,
|
|
content,
|
|
messageType,
|
|
mediaUrl,
|
|
replyToMessageId,
|
|
});
|
|
|
|
const preview = messageType === 'text' ? content : `${messageType} message`;
|
|
await this.chatRepository.updateConversationAfterNewMessage(
|
|
conversation.id,
|
|
message.id,
|
|
currentUserId,
|
|
preview,
|
|
);
|
|
this.chatRealtimeService.emitNewMessage(message.conversationId.toString(), message);
|
|
await this.dispatchMessageNotifications(
|
|
currentUserId,
|
|
conversation.participantIds.map((id) => id.toString()),
|
|
conversation.id,
|
|
preview,
|
|
);
|
|
|
|
return message;
|
|
}
|
|
|
|
async sendMessageWithUpload(
|
|
currentUserId: string,
|
|
dto: SendMessageDto,
|
|
file?: { mimetype?: string; size: number; buffer: Buffer; originalname?: string },
|
|
) {
|
|
if (!file) {
|
|
throw new BadRequestException('mediaFile is required');
|
|
}
|
|
|
|
const mediaType = this.resolveUploadedMessageType(file);
|
|
const mediaUrl = await this.storageService.saveFile({
|
|
folderSegments: ['chat', 'media'],
|
|
extension: this.resolveMediaExtension(mediaType, file),
|
|
buffer: file.buffer,
|
|
contentType: file.mimetype,
|
|
fileNamePrefix: 'message',
|
|
});
|
|
|
|
try {
|
|
return await this.sendMessage(currentUserId, {
|
|
...dto,
|
|
messageType: mediaType,
|
|
mediaUrl,
|
|
});
|
|
} catch (error) {
|
|
await this.storageService.deleteFile(mediaUrl);
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async markMessageSeen(currentUserId: string, messageId: string) {
|
|
const message = await this.chatRepository.findMessageById(messageId);
|
|
if (!message) {
|
|
throw new NotFoundException('Message not found');
|
|
}
|
|
|
|
await this.assertConversationMember(currentUserId, message.conversationId.toString());
|
|
await this.chatRepository.markMessageSeen(message.id, currentUserId);
|
|
await this.chatRepository.clearConversationUnreadForUser(message.conversationId.toString(), currentUserId);
|
|
|
|
return { success: true };
|
|
}
|
|
|
|
async unsendMessage(currentUserId: string, messageId: string) {
|
|
const message = await this.chatRepository.findMessageById(messageId);
|
|
if (!message) {
|
|
throw new NotFoundException('Message not found');
|
|
}
|
|
if (message.senderId.toString() !== currentUserId) {
|
|
throw new ForbiddenException('You can only unsend your own messages');
|
|
}
|
|
|
|
const updated = await this.chatRepository.unsendMessage(messageId, currentUserId);
|
|
if (!updated) {
|
|
throw new NotFoundException('Message not found');
|
|
}
|
|
return updated;
|
|
}
|
|
|
|
async reactToMessage(currentUserId: string, messageId: string, reactionType: ReactionType) {
|
|
const message = await this.chatRepository.findMessageById(messageId);
|
|
if (!message) {
|
|
throw new NotFoundException('Message not found');
|
|
}
|
|
await this.assertConversationMember(currentUserId, message.conversationId.toString());
|
|
const updated = await this.chatRepository.setMessageReaction(messageId, currentUserId, reactionType);
|
|
if (!updated) {
|
|
throw new NotFoundException('Message not found');
|
|
}
|
|
return updated;
|
|
}
|
|
|
|
async deleteMessageForMe(currentUserId: string, messageId: string) {
|
|
const message = await this.chatRepository.findMessageById(messageId);
|
|
if (!message) {
|
|
throw new NotFoundException('Message not found');
|
|
}
|
|
await this.assertConversationMember(currentUserId, message.conversationId.toString());
|
|
await this.chatRepository.deleteMessageForUser(messageId, currentUserId);
|
|
return { success: true };
|
|
}
|
|
|
|
async blockUser(currentUserId: string, targetUserId: string) {
|
|
if (!Types.ObjectId.isValid(targetUserId)) {
|
|
throw new BadRequestException('Invalid target user id');
|
|
}
|
|
if (currentUserId === targetUserId) {
|
|
throw new BadRequestException('You cannot block yourself');
|
|
}
|
|
|
|
const target = await this.usersRepository.findById(targetUserId);
|
|
if (!target) {
|
|
throw new NotFoundException('Target user not found');
|
|
}
|
|
|
|
await this.chatRepository.createBlock(currentUserId, targetUserId);
|
|
return { blocked: true, targetUserId };
|
|
}
|
|
|
|
async unblockUser(currentUserId: string, targetUserId: string) {
|
|
if (!Types.ObjectId.isValid(targetUserId)) {
|
|
throw new BadRequestException('Invalid target user id');
|
|
}
|
|
await this.chatRepository.removeBlock(currentUserId, targetUserId);
|
|
return { blocked: false, targetUserId };
|
|
}
|
|
|
|
async getBlockStatus(currentUserId: string, targetUserId: string) {
|
|
if (!Types.ObjectId.isValid(targetUserId)) {
|
|
throw new BadRequestException('Invalid target user id');
|
|
}
|
|
|
|
const iBlocked = !!(await this.chatRepository.findBlock(currentUserId, targetUserId));
|
|
const blockedMe = !!(await this.chatRepository.findBlock(targetUserId, currentUserId));
|
|
|
|
return { targetUserId, iBlocked, blockedMe };
|
|
}
|
|
|
|
async getMyBlockedUsers(currentUserId: string) {
|
|
const items = await this.chatRepository.findBlocksByBlocker(currentUserId);
|
|
return { items };
|
|
}
|
|
|
|
async assertConversationMember(userId: string, conversationId: string) {
|
|
if (!Types.ObjectId.isValid(conversationId)) {
|
|
throw new BadRequestException('Invalid conversation id');
|
|
}
|
|
|
|
const conversation = await this.chatRepository.findConversationById(conversationId);
|
|
if (!conversation) {
|
|
throw new NotFoundException('Conversation not found');
|
|
}
|
|
|
|
const isMember = conversation.participantIds.some((id) => id.toString() === userId);
|
|
if (!isMember) {
|
|
throw new ForbiddenException('You are not a member of this conversation');
|
|
}
|
|
|
|
return conversation;
|
|
}
|
|
|
|
private async assertNoChatBlockInConversation(currentUserId: string, participantIds: string[]) {
|
|
for (const participantId of participantIds) {
|
|
if (participantId === currentUserId) {
|
|
continue;
|
|
}
|
|
const block = await this.chatRepository.findAnyBlockBetween(currentUserId, participantId);
|
|
if (block) {
|
|
throw new ForbiddenException('Cannot send message because one of participants is blocked');
|
|
}
|
|
}
|
|
}
|
|
|
|
private normalizeUnreadCountByUser(value: unknown): Record<string, number> {
|
|
if (!value) {
|
|
return {};
|
|
}
|
|
|
|
if (value instanceof Map) {
|
|
return Object.fromEntries(
|
|
Array.from(value.entries()).map(([key, count]) => [key.toString(), Number(count) || 0]),
|
|
);
|
|
}
|
|
|
|
if (typeof value === 'object') {
|
|
return Object.fromEntries(
|
|
Object.entries(value as Record<string, unknown>)
|
|
.filter(([key]) => !key.startsWith('$__') && key !== '$isMongooseMap')
|
|
.map(([key, count]) => [key, Number(count) || 0]),
|
|
);
|
|
}
|
|
|
|
return {};
|
|
}
|
|
|
|
private async resolveReplyToMessageId(replyToMessageId: string | undefined, conversationId: string) {
|
|
if (!replyToMessageId) {
|
|
return null;
|
|
}
|
|
if (!Types.ObjectId.isValid(replyToMessageId)) {
|
|
throw new BadRequestException('Invalid replyToMessageId');
|
|
}
|
|
const replyMessage = await this.chatRepository.findMessageById(replyToMessageId);
|
|
if (!replyMessage || replyMessage.conversationId.toString() !== conversationId) {
|
|
throw new BadRequestException('Reply message must belong to the same conversation');
|
|
}
|
|
return replyMessage.id;
|
|
}
|
|
|
|
private resolveUploadedMessageType(file: { mimetype?: string; originalname?: string }) {
|
|
if (file.mimetype?.startsWith('image/')) {
|
|
return 'image' as const;
|
|
}
|
|
if (file.mimetype?.startsWith('video/')) {
|
|
return 'video' as const;
|
|
}
|
|
if (file.mimetype?.startsWith('audio/')) {
|
|
return 'audio' as const;
|
|
}
|
|
const extension = extname(file.originalname ?? '').toLowerCase();
|
|
if (['.jpg', '.jpeg', '.png', '.webp', '.gif'].includes(extension)) {
|
|
return 'image' as const;
|
|
}
|
|
if (['.mp4', '.mov', '.webm', '.mkv', '.avi'].includes(extension)) {
|
|
return 'video' as const;
|
|
}
|
|
if (['.mp3', '.wav', '.m4a', '.aac', '.ogg'].includes(extension)) {
|
|
return 'audio' as const;
|
|
}
|
|
throw new BadRequestException('mediaFile must be image, video, or audio');
|
|
}
|
|
|
|
private resolveMediaExtension(
|
|
mediaType: 'image' | 'video' | 'audio',
|
|
file: { mimetype?: string; originalname?: string },
|
|
): string {
|
|
const extension = extname(file.originalname ?? '').toLowerCase();
|
|
const allowed = {
|
|
image: new Set(['.jpg', '.jpeg', '.png', '.webp', '.gif']),
|
|
video: new Set(['.mp4', '.mov', '.webm', '.mkv', '.avi']),
|
|
audio: new Set(['.mp3', '.wav', '.m4a', '.aac', '.ogg', '.webm']),
|
|
}[mediaType];
|
|
if (allowed.has(extension)) {
|
|
return extension;
|
|
}
|
|
if (mediaType === 'image') {
|
|
return file.mimetype === 'image/png' ? '.png' : file.mimetype === 'image/webp' ? '.webp' : '.jpg';
|
|
}
|
|
if (mediaType === 'video') {
|
|
return file.mimetype === 'video/quicktime' ? '.mov' : file.mimetype === 'video/webm' ? '.webm' : '.mp4';
|
|
}
|
|
return file.mimetype === 'audio/webm' ? '.webm' : file.mimetype === 'audio/ogg' ? '.ogg' : '.mp3';
|
|
}
|
|
|
|
private async dispatchMessageNotifications(
|
|
actorId: string,
|
|
participantIds: string[],
|
|
conversationId: string,
|
|
previewText: string,
|
|
): Promise<void> {
|
|
for (const recipientId of participantIds) {
|
|
if (recipientId === actorId) {
|
|
continue;
|
|
}
|
|
|
|
try {
|
|
await this.notificationsService.createMessageNotification(
|
|
actorId,
|
|
recipientId,
|
|
conversationId,
|
|
previewText.slice(0, 160),
|
|
);
|
|
} catch (error) {
|
|
this.logger.warn(
|
|
`Message notification failed for actor=${actorId} recipient=${recipientId}: ${
|
|
error instanceof Error ? error.message : 'unknown error'
|
|
}`,
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|