import { BadRequestException, HttpException, HttpStatus, Injectable, Inject, forwardRef } from '@nestjs/common'; import { CommonService } from 'src/common/common.service'; import { ModelsService } from 'src/models/models.service'; import { chatHistory, CreateConnection, CreateGroupDto, GroupOperations, GroupOperationsDto, RequestDto, SendMessageFromPushDto } from './dto/chat.dto'; import { Types } from 'mongoose'; import { ChatAggregations } from './chat.aggregation'; import * as moment from 'moment'; import { SessionDocument } from 'src/user/schema/session.schema'; import { chatDisappearing, ConnectionTypes } from './schema/connections.schema'; import { log } from 'node:console'; import { EditMessageDto, SendMessageDto } from './dto/socket.dto'; import { connectionType } from './schema/messages.schema'; import { Cron } from "@nestjs/schedule"; import { ChatGateway } from './chat.gateway'; @Injectable() export class ChatService { private projection = { __v: 0 } as const; private lean_options = { lean: true }; private new_lean_options = { new: true, lean: true }; private new_options = { new: true }; constructor( private readonly common: CommonService, private readonly model: ModelsService, private readonly aggregate: ChatAggregations, @Inject(forwardRef(() => ChatGateway)) private readonly gatewayService: ChatGateway, ) { } async checkConnection(sent_by: string, sent_to: string) { try { let query = { $or: [ { $and: [ { sent_by: new Types.ObjectId(sent_by) }, { sent_to: new Types.ObjectId(sent_to) }, { connection_type: "NORMAL" } ] }, { $and: [ { sent_by: new Types.ObjectId(sent_to) }, { sent_to: new Types.ObjectId(sent_by) }, { connection_type: "NORMAL" }] }, ], }; const projection = { __v: 0 } as const; let options = { lean: true }; let fetch_conn: any = await this.model.Connections.find( query, projection, options ); return fetch_conn; } catch (error) { return error; } } async createConnections(user_id: string, payload: CreateConnection) { try { let { sent_to } = payload; if (sent_to && sent_to == user_id) { throw new HttpException({ message: "user id and sent to id can't be same" }, HttpStatus.BAD_REQUEST); } else { const check_connection: any = await this.checkConnection( user_id, sent_to ); let connection = check_connection[0]; if (!check_connection?.length) { connection = await this.saveConnetions(user_id, sent_to); } // Fetch comprehensive connection data using aggregation pipeline let groupConnectionIds = await this.model.GroupMembers.distinct("connection_id", { user_id: new Types.ObjectId(user_id), is_exit_from_group: false }).lean(); let query: any = [ await this.aggregate.match({ $and: [ { _id: new Types.ObjectId(connection._id) }, { $or: [ { sent_by: new Types.ObjectId(user_id) }, { sent_to: new Types.ObjectId(user_id) }, { _id: { $in: groupConnectionIds } }, ], }, { connection_deleted_by: { $nin: [new Types.ObjectId(user_id)] } }, ] }), await this.aggregate.lookup_messages(user_id), await this.aggregate.count_total_messages(), await this.aggregate.setData(user_id), await this.aggregate.lookupForCheckMutedConnection(user_id), await this.aggregate.lookupUser(), await this.aggregate.unwindData("$fetch_users"), await this.aggregate.lookupUnreadChat(user_id), await this.aggregate.countMessageData(), await this.aggregate.findOtherUserId(user_id), await this.aggregate.lookupBlockedUsers(user_id), await this.aggregate.addBlockedStatus(), await this.aggregate.addLockedStatus(user_id), await this.aggregate.fetchMessages(user_id), await this.aggregate.unwindData("$fetch_messages"), await this.aggregate.setLastMsg(user_id), await this.aggregate.groupData(), await this.aggregate.addPinnedStatus(user_id), ]; let fetch_data: any = await this.model.Connections.aggregate(query); let response = { count: fetch_data?.length ?? 0, data: fetch_data?.[0] ?? null, }; return response } } catch (error) { throw error // this.server.to(socket_id).emit("create_connection", { error: error }); } } async saveConnetions(sent_by: string, sent_to: string) { try { let data_to_save = { sent_by: new Types.ObjectId(sent_by), sent_to: new Types.ObjectId(sent_to), created_at: +new Date(), connection_type: "NORMAL", updated_id: +new Date(), connection_locked_by: [], connection_pinned_by: [], }; let save_conn = await this.model.Connections.create(data_to_save); return save_conn; } catch (error) { return error; } } listChatUsers = async (_id: string, search: string, type: string, conn_type: string, pagination?: number, limit?: number) => { try { let options = await this.common.set_options(pagination, limit); let groupConnectionIds = await this.model.GroupMembers.distinct("connection_id", { user_id: new Types.ObjectId(_id), is_exit_from_group: false }).lean(); console.log(conn_type, "connection_type+++++++++++++"); let connQuery: any = { connection_type: "NORMAL" } if (conn_type == ConnectionTypes.GROUP) { connQuery = { $and: [{ connection_type: "GROUP" }, { _id: { $in: groupConnectionIds } } ] } } let archivedConnections = await this.model.Connections.countDocuments({ ...connQuery, connection_archived_by: new Types.ObjectId(_id), connection_deleted_by: { $nin: [new Types.ObjectId(_id)] } }); let lockedConnections = await this.model.Connections.countDocuments({ ...connQuery, connection_locked_by: new Types.ObjectId(_id), connection_archived_by: { $nin: [new Types.ObjectId(_id)] }, connection_deleted_by: { $nin: [new Types.ObjectId(_id)] } }); let query: any = [ await this.aggregate.matchData(_id, groupConnectionIds, type, conn_type), await this.aggregate.lookup_messages(_id), await this.aggregate.count_total_messages(), await this.aggregate.setData(_id), await this.aggregate.lookupForCheckMutedConnection(_id), await this.aggregate.lookupUser(), await this.aggregate.unwindData("$fetch_users"), await this.aggregate.filterUsersByName(search), await this.aggregate.lookupUnreadChat(_id), await this.aggregate.countMessageData(), await this.aggregate.findOtherUserId(_id), await this.aggregate.lookupBlockedUsers(_id), await this.aggregate.addBlockedStatus(), await this.aggregate.addLockedStatus(_id), await this.aggregate.fetchMessages(_id), await this.aggregate.unwindData("$fetch_messages"), await this.aggregate.setLastMsg(_id), await this.aggregate.firstMessage(_id), await this.aggregate.setFirstMessage(_id), await this.aggregate.groupData(), await this.aggregate.filterWithoutMessage(_id), await this.aggregate.addPinnedStatus(_id), // await this.aggregate.removeEmptyDoc(), // uncomment this when live in the case of no message found await this.aggregate.facetData(options.skip, options.limit) ]; let fetch_data: any = await this.model.Connections.aggregate(query); console.log(fetch_data, "fetch++++++++++++++++"); let response = { count: fetch_data[0]?.count[0]?.count ?? 0, data: fetch_data[0]?.data ?? [], total_archived_connections: archivedConnections ?? 0, total_locked_connections: lockedConnections ?? 0, }; return response } catch (err) { throw err; } }; message_history = async (user_id: string, req_query: chatHistory) => { try { let { connection_id } = req_query; let query: any = { $and: [ { connection_id: new Types.ObjectId(connection_id) }, { deleted_for: { $nin: [new Types.ObjectId(user_id)] } }, { is_deleted: false }, ] }; // if (start_ind_id && start_ind_id !== '') { // query.$and.push({ // _id: { $gt: new Types.ObjectId(start_ind_id) } // }); // } // if (last_ind_id && last_ind_id !== '') { // query.$and.push({ // _id: { $lt: new Types.ObjectId(last_ind_id) } // }); // } let qry: any = await this.message_history_query(query, user_id); let data = await this.model.Messages.aggregate(qry); return { data: data[0]?.data?.length ? data[0]?.data : [] }; } catch (error) { throw error } } message_history_query = async (query, user_id) => { try { let { page, limit } = query; let options = await this.common.set_options(page, limit); return [ await this.aggregate.match(query), await this.aggregate.lookupSentBy(), await this.aggregate.unwindSentBy(), await this.aggregate.lookupSentTo(), await this.aggregate.unwindSentTo(), // await this.aggregate.lookupProduct(), // await this.aggregate.unwindProduct(), // await this.aggregate.lookupBlockUser(), // await this.aggregate.unwindBlockUser(), // await this.aggregate.lookupStealthUser(), // await this.aggregate.unwindStealthUser(), // await this.aggregate.lookupCall(user_id), // await this.aggregate.unwindCall(), await this.aggregate.lookupReplyTo(), await this.aggregate.unwindReplyTo(), await this.aggregate.lookupChatEmoji(), await this.aggregate.addFieldsIsStared(user_id), await this.aggregate.groupDataForMessages(user_id), await this.aggregate.facetDataForMessage(options.skip, options.limit), await this.aggregate.projectData(), ]; } catch (error) { throw error; } } async messageDetails(message_id: string) { try { console.log("message_id", message_id) let populate_to = [ { path: "sent_by", select: "profile_pic name" }, { path: "sent_to", select: "profile_pic name" }, { path: "connection_id", select: "updated_at name image" }, { path: "reply_to", populate: [ { path: "sent_by", select: "profile_pic name" }, { path: "sent_to", select: "profile_pic name" }, ] } ]; let message = await this.model.Messages.findOne({ _id: new Types.ObjectId(message_id) }, { __v: 0 }, { lean: true }).populate(populate_to); return message } catch (error) { throw error } } async getUser(user_id: string, projection: any) { try { let get_user = await this.model.UserModel.findOne( { _id: new Types.ObjectId(user_id), }, projection, { lean: true } ); return get_user; } catch (error) { throw error; } } async saveSingleMessage(sent_by: string, payload: any, user_connection_id?: string, sent_to_user_id?: string) { try { let { sent_to, message, type, message_type, media_url, media, reply_msg_id, connection_id, message_id, // only for frontend } = payload; if (user_connection_id) { connection_id = user_connection_id; } if (sent_to_user_id) sent_to = sent_to_user_id; let connection: any = await this.model.Connections.findOne( { _id: connection_id }, this.projection, { lean: true } ); // ------------check messge disappearing is on or not--------------- let { is_disappeared_msg, chat_deleted_at: msg_disappearing_date, message_disapear_by_single_user } = await this.checkMsgDisAppearing(connection, sent_by, sent_to); let data_to_save: any = { ...payload, connection_type: connection.connection_type, connection_id: connection_id, sent_by, reply_msg_id: reply_msg_id ? new Types.ObjectId(reply_msg_id) : null, sent_to: !sent_to ? null : sent_to, type, message: message_type == "CURRENT_LOCATION" ? null : message, message_id, is_disappeared_msg, msg_disappearing_date, chat_disappear_by: message_disapear_by_single_user, deleted_for: [], message_status: "SENT", is_already_disappeared: false, message_type, ...(media_url && { media_url: media_url }), ...(media && { media: media }), delivered_to: [{ user_id: sent_by, delivered_at: moment().utc().valueOf() }], read_by: [{ user_id: sent_by, read_at: moment().utc().valueOf() }], // is_deleted: true, created_at: moment().utc().valueOf(), updated_at: moment().utc().valueOf(), }; let saved_message; saved_message = await this.model.Messages.create(data_to_save); if (message_type == "DOCUMENT" || message_type == "LINK") { let data_to_save = { connection_id: connection_id, message_id: saved_message._id, sender_id: sent_by, type: message_type, media_url: message_type == "DOCUMENT" ? media_url : message, caption: null, colors: [], thumbnail: null, deleted_for: [], created_at: moment().utc().valueOf(), updated_at: moment().utc().valueOf() }; await this.model.ChatMedias.create(data_to_save) } let { _id: new_msg_id } = saved_message; let response: any = await this.makeMsgResponse(new_msg_id); response[0].isStared = false return response[0]; } catch (error) { return error; } } async checkMsgDisAppearing(connection: any, sent_by: string, sent_to: string) { try { let is_disappeared_msg = false; let chat_deleted_at = 0; const current_date = moment().utc().valueOf(); const disappear_values = [ chatDisappearing.DAYS_7, chatDisappearing.HOURS_24, chatDisappearing.DAYS_90 ]; const durationMap = { [chatDisappearing.DAYS_7]: moment(current_date).utc().add(7, "days").valueOf(), [chatDisappearing.HOURS_24]: moment(current_date).utc().add(24, "hours").valueOf(), [chatDisappearing.DAYS_90]: moment(current_date).utc().add(90, "days").valueOf() }; let data: any[] = []; // ------------------------------------------------------------ // 1️⃣ Sender Disappearing Settings Check // ------------------------------------------------------------ let senderSetting = connection?.chat_disappear_by?.find( (res: any) => res?.user_id?.toString() === sent_by?.toString() ); if (senderSetting && senderSetting.chat_disappearing !== chatDisappearing.OFF) { chat_deleted_at = durationMap[senderSetting.chat_disappearing]; data.push({ user_id: new Types.ObjectId(sent_by), is_already_disappeared: false, is_disappeared_msg: true, msg_disappearing_date: chat_deleted_at }); is_disappeared_msg = true; } // ------------------------------------------------------------ // 2️⃣ GROUP CHAT: Check for all group member settings // ------------------------------------------------------------ if (connection.connection_type === connectionType.GROUP) { const groupMembers = await this.model.GroupMembers.distinct( "user_id", { connection_id: connection._id, is_exit_from_group: false, user_id: { $ne: sent_by } } ); for (const member of groupMembers) { const result = connection?.chat_disappear_by?.find( (res: any) => res?.user_id?.toString() === member.toString() ); if (result && result.chat_disappearing !== chatDisappearing.OFF) { chat_deleted_at = durationMap[result.chat_disappearing]; data.push({ user_id: new Types.ObjectId(member), is_already_disappeared: false, is_disappeared_msg: true, msg_disappearing_date: chat_deleted_at }); is_disappeared_msg = true; } } } // ------------------------------------------------------------ // 3️⃣ NORMAL CHAT: Check receiver’s disappearing setting // ------------------------------------------------------------ else { const receiverSetting = connection?.chat_disappear_by?.find( (res: any) => res?.user_id?.toString() === sent_to?.toString() ); if (receiverSetting && receiverSetting.chat_disappearing !== chatDisappearing.OFF) { chat_deleted_at = durationMap[receiverSetting.chat_disappearing]; data.push({ user_id: new Types.ObjectId(sent_to), is_already_disappeared: false, is_disappeared_msg: true, msg_disappearing_date: chat_deleted_at }); is_disappeared_msg = true; } } // ------------------------------------------------------------ // 4️⃣ Check if connection-level disappearing is enabled // ------------------------------------------------------------ if (disappear_values.includes(connection?.chat_disappearing)) { is_disappeared_msg = true; chat_deleted_at = durationMap[connection.chat_disappearing]; } // ------------------------------------------------------------ // Final Return // ------------------------------------------------------------ return { chat_deleted_at, is_disappeared_msg, message_disapear_by_single_user: data }; } catch (error) { throw error; } } async saveMediasMessage(sent_by: string, payload: any, media: any, is_another_user_blocket_me: boolean) { try { let { sent_to, message, type, message_type, media_url, message_id, // media, // reply_to, front_img, connection_id, reply_msg_id, product_id } = payload; let connection: any = await this.model.Connections.findOne( { _id: connection_id }, {}, { lean: true } ); //--------------function for check disappearing------------------- let { is_disappeared_msg, chat_deleted_at: msg_disappearing_date, message_disapear_by_single_user } = await this.checkMsgDisAppearing(connection, sent_by, sent_to); // --------------end--------------- let data_to_save = { connection_id: connection_id, sent_by, reply_msg_id: reply_msg_id ? new Types.ObjectId(reply_msg_id) : null, sent_to: !sent_to ? null : sent_to, type, message, message_id: media?.message_id, deleted_for: is_another_user_blocket_me ? [new Types.ObjectId(sent_to)] : [], message_status: "SENT", message_type, ...(media_url && { media_url: media_url }), ...(media && { media: [media] }), // is_deleted: true, is_already_disappeared: false, // msg_disappearing_date, is_disappeared_msg, msg_disappearing_date, chat_disappear_by: message_disapear_by_single_user, created_at: +new Date(), delivered_to: [{ user_id: sent_by, delivered_at: moment().utc().valueOf() }], read_by: [{ user_id: sent_by, read_at: moment().utc().valueOf() }], }; data_to_save.media = [media]; let saved_message = await this.model.Messages.create(data_to_save); let data_to_save_in_media = { message_type: message_type == "ONE_TIME_IMAGE" ? "ONE_TIME_MEDIA" : "NORMAL", connection_id: connection_id, message_id: saved_message._id, sent_by, sent_to: !sent_to ? null : sent_to, type: media?.type, media_url: media?.url, duration: media?.duration, caption: media?.caption, thumbnail: media?.thumbnail, colors: [], deleted_for: is_another_user_blocket_me ? [new Types.ObjectId(sent_to)] : [], created_at: moment().utc().valueOf(), updated_at: moment().utc().valueOf() }; await this.model.ChatMedias.create(data_to_save_in_media) let { _id: new_msg_id } = saved_message; let response = await this.makeMsgResponse(new_msg_id.toString()); return response; } catch (error) { return error; } } async saveForwardMessage(connection_id: string, sent_by: any, messageData: any, is_another_user_blocket_me: boolean) { try { let connection: any = await this.model.Connections.findOne( { _id: connection_id }, {}, { lean: true } ); let { is_disappeared_msg, chat_deleted_at: msg_disappearing_date, message_disapear_by_single_user } = await this.checkMsgDisAppearing(connection, sent_by, messageData?.sent_to?.toString()) let data_to_save = { connection_id: connection_id, sent_by, reply_msg_id: null, sent_to: messageData?.sent_to ? new Types.ObjectId(messageData?.sent_to) : null, type: 'FORWARDED', message: messageData?.message, message_id: messageData?.message_id, deleted_for: is_another_user_blocket_me ? [new Types.ObjectId(messageData?.sent_to)] : [], message_status: "SENT", message_type: messageData?.message_type, ...(messageData?.media_url && { media_url: messageData?.media_url }), ...(messageData?.media && { media: [messageData?.media] }), // is_deleted: true, msg_disappearing_date, is_already_disappeared: false, chat_disappear_by: message_disapear_by_single_user, is_disappeared_msg, created_at: +new Date(), updated_at: +new Date(), delivered_to: [{ user_id: sent_by, delivered_at: moment().utc().valueOf() }], read_by: [{ user_id: sent_by, read_at: moment().utc().valueOf() }], }; data_to_save.media = [messageData.media]; let saved_message = await this.model.Messages.create(data_to_save); if (data_to_save.media) { let data_to_save_in_media = { message_type: messageData.message_type == "ONE_TIME_IMAGE" ? "ONE_TIME_MEDIA" : "NORMAL", connection_id: connection_id, message_id: saved_message._id, sent_by, sent_to: !messageData?.sent_to ? null : messageData?.sent_to, type: messageData?.media?.type, media_url: messageData?.media?.url, duration: messageData?.media?.duration, caption: messageData?.media?.caption, thumbnail: messageData?.media?.thumbnail, colors: [], deleted_for: is_another_user_blocket_me ? [new Types.ObjectId(messageData?.sent_to)] : [], created_at: moment().utc().valueOf() }; await this.model.ChatMedias.create(data_to_save_in_media) } if (messageData.message_type == "DOCUMENT" || messageData.message_type == "LINK") { let data_to_save = { connection_id: connection_id, message_id: saved_message._id, sender_id: sent_by, type: messageData.message_type, media_url: messageData.message_type == "DOCUMENT" ? messageData.media_url : messageData.message, caption: null, colors: [], thumbnail: null, deleted_for: [], created_at: moment().utc().valueOf() }; await this.model.ChatMedias.create(data_to_save) } let { _id: new_msg_id } = saved_message; let response = await this.makeMsgResponse(new_msg_id.toString()); return response; } catch (error) { return error; } } async checkMuteOrUnmute(connection_id: string, sent_to: string) { try { let checkMuteOrUnmute = await this.model.Connections.findOne({ _id: new Types.ObjectId(connection_id), "connection_muted_by.user_id": new Types.ObjectId(sent_to) }); let is_muted = false; if (checkMuteOrUnmute) { let { connection_muted_by } = checkMuteOrUnmute; let find = connection_muted_by.filter((val) => val.user_id.toString() == sent_to?.toString()); let currentTime = moment().utc().valueOf(); if (find[0]?.time > currentTime) { is_muted = true; } } return is_muted; } catch (error) { throw error } } async makeMsgResponse(_id: string) { try { let query = { _id: _id }; let populate_to = [ { path: "sent_by", select: "profile_pic name" }, { path: "sent_to", select: "profile_pic name" }, { path: "connection_id", select: "updated_at name image" }, { path: "reply_msg_id", populate: [ { path: "sent_by", select: "profile_pic name" }, { path: "sent_to", select: "profile_pic name" }, ] } ]; let response = await this.model.Messages.find(query, this.projection, this.lean_options) .populate(populate_to) .exec(); if (response?.length) { let fetchEmoji = await this.model.MsgReactions.find({ message_id: response[0]?._id }).populate({ path: "user_id", model: 'users', select: "name profile_pic phone_no country_code" }); let data: any = []; if (fetchEmoji.length) { for (let i = 0; i < fetchEmoji.length; i++) { const element = fetchEmoji[i]; data.push({ total_users: element?.user_id?.length, emoji: element.emoji, users: element.user_id }) } } response[0]['message_emoji'] = data } return response; } catch (error) { throw error; } } sendMessageNotification = async (sent_to: string, notification_data: any, is_muted: boolean, is_arvhived: boolean) => { try { let actual_is_archived = is_arvhived; if (notification_data?.connection_id) { let connection = await this.model.Connections.findOne({ _id: new Types.ObjectId(notification_data.connection_id) }); if (connection && connection.connection_archived_by) { actual_is_archived = actual_is_archived || connection.connection_archived_by.some(id => id.toString() === sent_to.toString()); } } let query = { user_id: new Types.ObjectId(sent_to), fcm_token: { $ne: null }, }; console.log("query" , query); let sessions = await this.model.SessionModel.find(query, 'fcm_token device_type'); console.log(sessions, "+++++++++++++sessions"); if (sessions && sessions?.length > 0) { // Separate iOS and Android tokens let iosTokens: string[] = []; let androidTokens: string[] = []; sessions.forEach(session => { if (session.device_type === 'IOS') { iosTokens.push(session.fcm_token); } else if (session.device_type === 'ANDROID' || session.device_type === "WEB") { androidTokens.push(session.fcm_token); } }); notification_data.show_notification = is_muted || actual_is_archived ? true : false; // Handle iOS notifications if (iosTokens.length > 0) { if (notification_data.show_notification) { // Silent notification for iOS await this.sendSilentIOSNotification(iosTokens, notification_data); } else { // Normal notification for iOS await this.sendNormalIOSNotification(iosTokens, notification_data); } } // Handle Android notifications if (androidTokens.length > 0) { if (notification_data.show_notification) { await this.common.sendSilentPushNotification(androidTokens, notification_data); } else { await this.common.sendPushNotification(androidTokens, notification_data); } } } } catch (err) { throw err; } }; private async sendNormalIOSNotification(tokens: string[], notification_data: any) { try { // Use FCM for iOS notifications with iOS-specific payload await this.common.sendPushNotification(tokens, { ...notification_data, apns: { payload: { aps: { alert: { title: notification_data?.title, body: notification_data?.message }, "badge": 1, "sound": "default", "content-available": 1 } }, "headers": { "apns-priority": "10" } } }); } catch (err) { console.error('Error sending normal iOS notification:', err); throw err; } } private async sendSilentIOSNotification(tokens: string[], notification_data: any) { try { await this.common.sendSilentPushNotification(tokens, { ...notification_data, apns: { payload: { aps: { "badge": 1, "sound": "default", "content-available": 1 } }, "headers": { "apns-priority": "10" } } }); } catch (err) { console.error('Error sending silent iOS notification:', err); throw err; } } starred_messages_query = async (query, user_id) => { try { return [ await this.aggregate.match(query), await this.aggregate.lookupSentBy(), await this.aggregate.unwindSentBy(), await this.aggregate.lookupSentTo(), await this.aggregate.unwindSentTo(), await this.aggregate.lookupReplyTo(), await this.aggregate.unwindReplyTo(), await this.aggregate.lookupChatEmoji(), await this.aggregate.addFieldsIsStared(user_id), await this.aggregate.groupDataForMessages(user_id), await this.aggregate.facetDataForMessage(0, 1000), // Get all starred messages, limit to 1000 await this.aggregate.projectData(), ]; } catch (error) { throw error; } } async getStarredMessages(connection_id: string, user_id: string) { try { let query: any = { $and: [ { connection_id: new Types.ObjectId(connection_id) }, { starred_by: { $in: [new Types.ObjectId(user_id)] } }, { deleted_for: { $nin: [new Types.ObjectId(user_id)] } }, { is_deleted: false }, ] }; let qry: any = await this.starred_messages_query(query, user_id); let data = await this.model.Messages.aggregate(qry); return { message: "Starred messages retrieved successfully", data: data[0]?.data?.length ? data[0]?.data : [], count: data[0]?.data?.length || 0 }; } catch (error) { console.log("Error in getStarredMessages:", error); throw error; } } async getMediaByConnection(connection_id: string, user_id: string, media_type?: string, pagination?: number, limit?: number) { try { // Build query let query: any = { connection_id: new Types.ObjectId(connection_id), deleted_for: { $nin: [new Types.ObjectId(user_id)] } // Exclude media deleted for this user }; // Add media type filter if provided if (media_type) { query.type = media_type; } // Set pagination options const options = await this.common.set_options(pagination, limit); // Get media with message and sender population const mediaList = await this.model.ChatMedias.find(query, {}, options) .populate({ path: 'message_id', select: 'message created_at message_type' }) .populate({ path: 'sender_id', select: 'name profile_pic' }) .sort({ created_at: -1 }) .lean(); return { message: "Media retrieved successfully", data: mediaList, count: mediaList.length, pagination: { page: pagination || 0, limit: limit || 10 } }; } catch (error) { console.log("Error in getMediaByConnection:", error); throw error; } } async muteOrUnmuteConnection(user_id: string, dto: any) { try { let { connection_id, type, no_of_days_or_hours } = dto; const userObjectId = new Types.ObjectId(user_id); let query = { _id: new Types.ObjectId(connection_id), "connection_muted_by.user_id": userObjectId }; let checkConnection = await this.model.Connections.findOne(query); let message; if (checkConnection) { let update = { $pull: { connection_muted_by: { user_id: userObjectId } }, last_updated_at: moment().utc().valueOf() }; let newQuery = { _id: checkConnection._id }; await this.model.Connections.findOneAndUpdate(newQuery, update, this.new_options); message = "Notification unmuted for this connection successfully."; } else { // Mute: Add user with mute type and expiry time let time = moment().utc().valueOf(); // Default to current timestamp if (type === 'WEEK') { time = moment().utc().add(no_of_days_or_hours, 'weeks').valueOf(); } else if (type === 'HOUR') { time = moment().utc().add(no_of_days_or_hours, 'hours').valueOf(); } else if (type === 'ALWAYS') { time = moment().utc().add(10, 'years').valueOf(); } let update = { $push: { connection_muted_by: { user_id: userObjectId, type: type, time: time } } }; let queryToFind = { _id: new Types.ObjectId(connection_id) }; await this.model.Connections.findOneAndUpdate(queryToFind, { ...update, last_updated_at: moment().utc().valueOf() }, this.new_lean_options); message = "Notification muted for this connection successfully."; } return { message }; } catch (error) { throw error; } } async saveDisappearingMessage(connection: any, userId: string, otherUserId: string, message: string, deleted_for: Array) { try { let data_to_save = { connection_id: connection?._id, sent_by: new Types.ObjectId(userId), reply_to: null, read_by: [new Types.ObjectId(userId)], sent_to: new Types.ObjectId(otherUserId), product_id: null, type: "DISAPPEARING", message, message_status: "SENT", deleted_for: deleted_for, message_type: "TEXT", created_at: +new Date(), updated_at: +new Date(), }; let saved_message: any = await this.model.Messages.create({ ...data_to_save }); let response = await this.makeMsgResponse(saved_message._id); return response } catch (error) { throw error } } async starUnstarMessage( connection_id: Types.ObjectId, message_ids: Types.ObjectId[], user_id: string, type: string ) { try { let message = ""; if (type === "STARRED_MESSAGES") { const newQuery = { _id: { $in: message_ids ?? [] }, starred_by: { $nin: [new Types.ObjectId(user_id)] } }; const data = await this.model.Messages.find(newQuery); if (data?.length) { const tasks = data.map((msg) => this.model.Messages.findOneAndUpdate( { _id: msg._id }, { $push: { starred_by: new Types.ObjectId(user_id) } }, { new: true } ) ); await Promise.all(tasks); } message = "Messages starred successfully."; } else if (type === "UNSTARRED_MESSAGES") { const newQuery = { _id: { $in: message_ids ?? [] }, starred_by: { $in: [new Types.ObjectId(user_id)] } }; const data = await this.model.Messages.find(newQuery); if (data?.length) { const tasks = data.map((msg) => { const newIds = msg.starred_by .map((id) => id.toString()) .filter((id) => id !== user_id) .map((id) => new Types.ObjectId(id)); return this.model.Messages.findOneAndUpdate( { _id: msg._id }, { starred_by: newIds }, { new: true, lean: true } ); } ); await Promise.all(tasks); } message = "Messages unstarred successfully."; } else if (type === "UNSTARRED_ALL_MESSAGES") { const newQuery = { connection_id: new Types.ObjectId(connection_id), starred_by: { $in: [new Types.ObjectId(user_id)] } }; await this.model.Messages.updateMany( newQuery, { $pull: { starred_by: new Types.ObjectId(user_id) } }, { new: true, lean: true } ); message = "Messages unstarred successfully."; } return { message }; } catch (error) { throw error; } } createGroup = async (req: RequestDto, body: CreateGroupDto) => { let { image, name, members } = body; const user_id = req?.user_data?._id; try { let getConnections = await this.model.GroupMembers.distinct("connection_id", { user_id: user_id }); let checkGroup = await this.model.Connections.findOne({ group_name: name, $or: [ { group_creator_id: new Types.ObjectId(user_id) }, { _id: { $in: getConnections } } ] }); console.log(checkGroup); if (checkGroup) throw new HttpException("Group already exists", HttpStatus.BAD_REQUEST); let connection = await this.model.Connections.create({ group_name: name, group_creator_id: new Types.ObjectId(user_id), group_image_url: image, connection_type: "GROUP", connection_locked_by: [], connection_pinned_by: [], connection_deleted_by: [], connection_muted_by: [], }); members.push(user_id.toString()); let groupMembers = [...new Set(members)]; await Promise.all(groupMembers?.map(async (member) => { return this.model.GroupMembers.create({ connection_id: connection?._id, user_id: new Types.ObjectId(member), role: member == user_id.toString() ? "GROUP_ADMIN" : "MEMBER", joined_at: moment().utc().valueOf(), }); })); let data = await this.model.Connections.findOne({ _id: connection?._id }).populate({ path: "group_creator_id", select: "name profile_pic" }); return { message: "New group created successfully.", data: data }; } catch (err) { console.log(` err new_group `, err); throw err; } }; groupDetails = async (req: RequestDto, connection_id: string) => { try { let data = await this.model.Connections.findOne({ _id: new Types.ObjectId(connection_id) }).populate({ path: "group_creator_id", select: "name profile_pic" }).lean(); let groupMembers = await this.model.GroupMembers.find({ connection_id: connection_id, is_exit_from_group: false }).populate({ path: 'user_id', select: 'name profile_pic' }); return { data: { ...data, memers: groupMembers } }; } catch (err) { console.log(` err group Deatils `, err); throw err; } }; connectionDetails = async (req: RequestDto, connection_id: string) => { try { const loginUserId = req.user_data._id.toString(); // Fetch connection let connection = await this.model.Connections.findOne({ _id: new Types.ObjectId(connection_id) }) .populate({ path: "group_creator_id", select: "name profile_pic" }) .lean(); if (!connection) { return { message: "Connection not found", data: {} }; } // Fetch group members only for group chat let groupMembers: any = []; if (connection.connection_type === "GROUP") { groupMembers = await this.model.GroupMembers.find({ connection_id, is_exit_from_group: false }).populate({ path: "user_id", select: "name profile_pic is_online" }); } // Get other user details for normal connections let otherUser: any = null; if (connection.connection_type === "NORMAL") { const otherUserId = connection.sent_by.toString() === loginUserId ? connection.sent_to : connection.sent_by; otherUser = await this.model.UserModel.findOne({ _id: otherUserId }).select("name profile_pic country_code phone_no is_online").lean(); } // ----------------------------------------- // ONLY LOGIN USER DISAPPEARING CHECK // ----------------------------------------- let userDisappear : any = { is_user_disappear: false, user_disappear_type: chatDisappearing.OFF }; if (Array.isArray(connection.chat_disappear_by)) { let loginUserSetting: any = connection.chat_disappear_by.find( u => u.user_id.toString() === loginUserId ); if (loginUserSetting) { userDisappear.is_user_disappear = loginUserSetting.chat_disappearing !== chatDisappearing.OFF; userDisappear.user_disappear_type = loginUserSetting.chat_disappearing; userDisappear.chat_disapear_for = loginUserSetting.chat_disapear_for; } } // FINAL RESPONSE return { data: { ...connection, members: groupMembers, user: otherUser, ...userDisappear } }; } catch (err) { console.log(`Error in groupDetails: `, err); throw err; } }; exitFromGroup = async (user_id: string, connection_id: string) => { try { // find from members console.log("user_id", user_id); console.log("connection_id", connection_id); let groupMembers = await this.model.GroupMembers.findOne({ connection_id: new Types.ObjectId(connection_id), user_id: new Types.ObjectId(user_id) }); // check if user is a member of this group console.log("groupMembers", groupMembers); if (!groupMembers) throw new HttpException("Sorry, you are not a member of this group, so you can't perform this action.", HttpStatus.BAD_REQUEST); // check if user is already exit from this group if (groupMembers && groupMembers?.is_exit_from_group == true) throw new HttpException("You are already exit from this group.", HttpStatus.BAD_REQUEST); // check if user is the last member of this group let totalUsers = await this.model.GroupMembers.countDocuments({ connection_id: new Types.ObjectId(connection_id), is_exit_from_group: false }); if (totalUsers == 1) throw new HttpException("You cannot exit the group because you are the last member.", HttpStatus.BAD_REQUEST); // check if user is admin if (groupMembers && groupMembers?.role == "GROUP_ADMIN") { let update = { $set: { is_exit_from_group: true, exit_group_at: moment().utc().valueOf() } }; const options = { new: true }; // remove from members await this.model.GroupMembers.findOneAndUpdate({ connection_id: new Types.ObjectId(connection_id), user_id: new Types.ObjectId(user_id) }, update, options); // find next admin let findNextAdmin = await this.model.GroupMembers.findOne({ connection_id: new Types.ObjectId(connection_id), role: "MEMBER", is_exit_from_group: false }).lean(); // update next admin await this.model.GroupMembers.findOneAndUpdate({ _id: findNextAdmin?._id }, { role: "GROUP_ADMIN" }, options); return { message: "You have successfully exit from this group." }; } let update = { $set: { is_exit_from_group: true, exit_group_at: moment().utc().valueOf() } }; const options = { new: true }; // remove from members await this.model.GroupMembers.findOneAndUpdate({ connection_id: new Types.ObjectId(connection_id), user_id: new Types.ObjectId(user_id) }, update, options); return { message: "You have successfully exit from this group." }; } catch (err) { console.log(` err group Deatils `, err); throw err; } }; exitFromGroupAndDeleteChatForMe = async (user_id: string, connection_id: string) => { try { // find from members let groupMembers = await this.model.GroupMembers.findOne({ connection_id: connection_id, user_id: new Types.ObjectId(user_id) }) // check if user is a member of this group if (!groupMembers) throw new HttpException("Sorry, you are not a member of this group, so you can't perform this action.", HttpStatus.BAD_REQUEST); // check if user is already exit from this group if (groupMembers && groupMembers?.is_exit_from_group == true) throw new HttpException("You are already exit from this group.", HttpStatus.BAD_REQUEST); // check if user is the last member of this group let totalUsers = await this.model.GroupMembers.countDocuments({ connection_id: connection_id, is_exit_from_group: false }); if (totalUsers == 1) throw new HttpException("You cannot exit the group because you are the last member.", HttpStatus.BAD_REQUEST); // check if user is admin if (groupMembers && groupMembers?.role == "GROUP_ADMIN") { let update = { $set: { is_exit_from_group: true, exit_group_at: moment().utc().valueOf() } }; const options = { new: true }; // remove from members await this.model.GroupMembers.findOneAndUpdate({ connection_id: connection_id, user_id: new Types.ObjectId(user_id) }, update, options); // find next admin let findNextAdmin = await this.model.GroupMembers.findOne({ connection_id: connection_id, role: "MEMBER", is_exit_from_group: false }).lean(); // update next admin await this.model.GroupMembers.findOneAndUpdate({ _id: findNextAdmin?._id }, { role: "GROUP_ADMIN" }, options); // delete connection await this.model.Connections.findOneAndUpdate({ _id: connection_id }, { $addToSet: { connection_deleted_by: new Types.ObjectId(user_id) } }) // delete messages await this.model.Messages.updateMany({ connection_id: connection_id }, { $addToSet: { deleted_for: new Types.ObjectId(user_id) } }) return { message: "You have successfully exit from this group." }; } let update = { $set: { is_exit_from_group: true, exit_group_at: moment().utc().valueOf() } }; const options = { new: true }; // remove from members await this.model.GroupMembers.findOneAndUpdate({ connection_id: connection_id, user_id: new Types.ObjectId(user_id) }, update, options); //delete there connections await this.model.Connections.findOneAndUpdate({ _id: connection_id }, { $addToSet: { connection_deleted_by: new Types.ObjectId(user_id) } }); //delete there messages await this.model.Messages.updateMany({ connection_id: connection_id }, { $addToSet: { deleted_for: new Types.ObjectId(user_id) } }) return { message: "You have successfully exit from this group." }; } catch (err) { console.log(` err group Deatils `, err); throw err; } }; clearChatForParticularConnections = async (user_id: string, connection_id: string) => { try { let checkConnection = await this.model.Connections.findOne({ _id: connection_id }) if (!checkConnection) throw new HttpException("Connection not found", HttpStatus.BAD_REQUEST); // check if user is a member of this group if (checkConnection.connection_type == ConnectionTypes.GROUP) { let groupMembers = await this.model.GroupMembers.findOne({ connection_id: connection_id, user_id: new Types.ObjectId(user_id) }) if (!groupMembers) throw new HttpException("Sorry, you are not a member of this group, so you can't perform this action.", HttpStatus.BAD_REQUEST); if (groupMembers && groupMembers?.is_exit_from_group == true) throw new HttpException("You are already exit from this group.", HttpStatus.BAD_REQUEST); } await this.model.Messages.updateMany({ connection_id: connection_id }, { $addToSet: { deleted_for: new Types.ObjectId(user_id) } }) return { message: "Chat history clear successfully." }; } catch (err) { console.log(` err group Deatils `, err); throw err; } }; lockMultipleConnections = async (user_id: string, connection_ids: string[]) => { try { // convert connection_ids to objectIds let objectIds = connection_ids.map((id) => new Types.ObjectId(id)); // check if user is authorized to lock this connection let connections = await this.model.Connections.find({ _id: { $in: objectIds }, connection_locked_by: { $nin: [new Types.ObjectId(user_id)] } }) // check if user is authorized to lock this connection if (connection_ids?.length !== connections?.length) throw new HttpException("You are not authorized to lock this connection.", HttpStatus.BAD_REQUEST); let update = { $addToSet: { connection_locked_by: new Types.ObjectId(user_id) } }; const options = { new: true }; // lock connections await this.model.Connections.updateMany({ _id: { $in: objectIds } }, update, options); // use this line for clear GC objectIds = []; return { message: "Connection locked successfully" }; } catch (err) { console.log(` err group Deatils `, err); throw err; } }; unLockMultipleConnections = async (user_id: string, connection_ids: string[]) => { try { // convert connection_ids to objectIds let objectIds = connection_ids.map((id) => new Types.ObjectId(id)); // check if user is authorized to lock this connection let connections = await this.model.Connections.find({ _id: { $in: objectIds }, connection_locked_by: { $in: [new Types.ObjectId(user_id)] } }); // check if user is authorized to lock this connection if (connection_ids?.length !== connections?.length) throw new HttpException("You are not authorized to lock this connection.", HttpStatus.BAD_REQUEST); // unlock connections let update = { $pull: { connection_locked_by: new Types.ObjectId(user_id) } }; const options = { new: true }; await this.model.Connections.updateMany({ _id: { $in: objectIds } }, update, options); // use this line for clear GC objectIds = []; return { message: "Connection unlocked successfully" }; } catch (err) { console.log(` err group Deatils `, err); throw err; } }; pinMultipleConnections = async (user_id: string, connection_ids: string[]) => { try { // convert connection_ids to objectIds let objectIds = connection_ids.map((id) => new Types.ObjectId(id)); // check current pinned count (max 5 allowed) const MAX_PINNED_CONNECTIONS = 5; const currentPinnedCount = await this.model.Connections.countDocuments({ connection_pinned_by: { $in: [new Types.ObjectId(user_id)] } }); // find connections that are NOT already pinned by this user (these are the new ones to pin) let connectionsToPin = await this.model.Connections.find({ _id: { $in: objectIds }, connection_pinned_by: { $nin: [new Types.ObjectId(user_id)] } }); // if all connections are already pinned, just return success (skip silently) if (connectionsToPin.length === 0) { return { message: "Connection(s) already pinned" }; } // check if pinning these new connections would exceed the limit if (currentPinnedCount + connectionsToPin.length > MAX_PINNED_CONNECTIONS) { const remainingSlots = MAX_PINNED_CONNECTIONS - currentPinnedCount; throw new HttpException( `You can only pin up connectionsToPinnversations. You currently have ${currentPinnedCount} pinned. You can pin ${remainingSlots} more.`, HttpStatus.BAD_REQUEST ); } // get the IDs of connections that will actually be pinned let connectionIdsToPinArray = connectionsToPin.map(conn => conn._id); // add user to connection_pinned_by array ($addToSet prevents duplicates) let update = { $addToSet: { connection_pinned_by: new Types.ObjectId(user_id) } }; const options = { new: true }; await this.model.Connections.updateMany( { _id: { $in: connectionIdsToPinArray } }, update, options ); // clear memory objectIds = []; connectionIdsToPinArray = []; return { message: `Successfully pinned ${connectionsToPin.length} connection(s)`, pinned_count: connectionsToPin.length, total_pinned: currentPinnedCount + connectionsToPin.length }; } catch (err) { console.log(`Error in pinMultipleConnections:`, err); throw err; } }; unPinMultipleConnections = async (user_id: string, connection_ids: string[]) => { try { // convert connection_ids to objectIds let objectIds = connection_ids.map((id) => new Types.ObjectId(id)); // check if connections are actually pinned by this user let connections = await this.model.Connections.find({ _id: { $in: objectIds }, connection_pinned_by: { $in: [new Types.ObjectId(user_id)] } }); // validate: ensure all connections are currently pinned if (connection_ids?.length !== connections?.length) { throw new HttpException( "One or more connections are not pinned or do not exist.", HttpStatus.BAD_REQUEST ); } // remove user from connection_pinned_by array let update = { $pull: { connection_pinned_by: new Types.ObjectId(user_id) } }; const options = { new: true }; await this.model.Connections.updateMany( { _id: { $in: objectIds } }, update, options ); // clear memory objectIds = []; return { message: "Connection(s) unpinned successfully" }; } catch (err) { console.log(`Error in unPinMultipleConnections:`, err); throw err; } }; /** * Pin a message in a conversation with sliding window logic (max 3 pinned messages) * When user pins a 4th message, the oldest pinned message automatically gets unpinned * @param user_id - User ID who is pinning the message * @param connection_id - Connection/Conversation ID * @param message_id - Message ID to pin */ pinMessage = async (user_id: string, connection_id: string, message_id: string) => { try { const MAX_PINNED_MESSAGES = 3; const userId = new Types.ObjectId(user_id); const messageIdObj = new Types.ObjectId(message_id); const connectionIdObj = new Types.ObjectId(connection_id); // Step 1: Check if message exists and belongs to this connection const message = await this.model.Messages.findOne({ _id: messageIdObj, connection_id: connectionIdObj }); if (!message) { throw new HttpException( "Message not found or does not belong to this conversation.", HttpStatus.NOT_FOUND ); } // Step 2: Check if message is already pinned by this user const isAlreadyPinned = message.chat_pinned_by?.some( (pin) => pin.user_id.toString() === user_id ); if (isAlreadyPinned) { return { message: "Message is already pinned", pinned: false }; } // Step 3: Get currently pinned messages for this user in this connection const pinnedMessages = await this.model.Messages.find({ connection_id: connectionIdObj, 'chat_pinned_by.user_id': userId }) .sort({ 'chat_pinned_by.pinned_at': 1 }) // Sort by oldest first .select('_id chat_pinned_by'); // Step 4: Sliding window logic - if already 3 pinned, remove the oldest if (pinnedMessages.length >= MAX_PINNED_MESSAGES) { // Get the oldest pinned message const oldestPinnedMessage = pinnedMessages[0]; // Remove this user from the oldest message's chat_pinned_by array await this.model.Messages.updateOne( { _id: oldestPinnedMessage._id }, { $pull: { chat_pinned_by: { user_id: userId } } } ); } // Step 5: Pin the new message const pinData = { user_id: userId, pinned_for: connection_id, pinned_at: moment().utc().valueOf(), pinned_to: 0 // Can be used for expiry if needed in future }; await this.model.Messages.updateOne( { _id: messageIdObj }, { $addToSet: { chat_pinned_by: pinData } } ); // Step 6: Get updated count of pinned messages const updatedPinnedCount = await this.model.Messages.countDocuments({ connection_id: connectionIdObj, 'chat_pinned_by.user_id': userId }); return { message: "Message pinned successfully", pinned: true, total_pinned: updatedPinnedCount, removed_oldest: pinnedMessages.length >= MAX_PINNED_MESSAGES }; } catch (err) { console.log(`Error in pinMessage:`, err); throw err; } }; /** * Unpin a specific message * @param user_id - User ID who is unpinning the message * @param connection_id - Connection/Conversation ID * @param message_id - Message ID to unpin */ unpinMessage = async (user_id: string, connection_id: string, message_id: string) => { try { const userId = new Types.ObjectId(user_id); const messageIdObj = new Types.ObjectId(message_id); const connectionIdObj = new Types.ObjectId(connection_id); // Check if message exists and is pinned by this user const message = await this.model.Messages.findOne({ _id: messageIdObj, connection_id: connectionIdObj, 'chat_pinned_by.user_id': userId }); if (!message) { throw new HttpException( "Message not found, does not belong to this conversation, or is not pinned.", HttpStatus.NOT_FOUND ); } // Remove user from chat_pinned_by array await this.model.Messages.updateOne( { _id: messageIdObj }, { $pull: { chat_pinned_by: { user_id: userId } } } ); // Get remaining pinned count const remainingPinnedCount = await this.model.Messages.countDocuments({ connection_id: connectionIdObj, 'chat_pinned_by.user_id': userId }); return { message: "Message unpinned successfully", unpinned: true, total_pinned: remainingPinnedCount }; } catch (err) { console.log(`Error in unpinMessage:`, err); throw err; } }; /** * Get all pinned messages for a user in a specific conversation * @param user_id - User ID * @param connection_id - Connection/Conversation ID */ getPinnedMessages = async (user_id: string, connection_id: string) => { try { const userId = new Types.ObjectId(user_id); const connectionIdObj = new Types.ObjectId(connection_id); // Get all pinned messages for this user in this connection const pinnedMessages = await this.model.Messages.find({ connection_id: connectionIdObj, 'chat_pinned_by.user_id': userId, deleted_for: { $nin: [userId] }, // Exclude deleted messages is_deleted: false }) .populate('sent_by', 'name profile_image') .sort({ 'chat_pinned_by.pinned_at': -1 }) // Most recently pinned first .select('message message_type media_url media type sent_by created_at chat_pinned_by'); // Filter and format the response to only include relevant pin data const formattedMessages = pinnedMessages.map(msg => { const userPin = msg.chat_pinned_by?.find( pin => pin.user_id.toString() === user_id ); return { _id: msg._id, message: msg.message, message_type: msg.message_type, media_url: msg.media_url, media: msg.media, type: msg.type, sent_by: msg.sent_by, created_at: msg.created_at, pinned_at: userPin?.pinned_at || 0 }; }); return { pinned_messages: formattedMessages, total_pinned: formattedMessages.length, max_allowed: 3 }; } catch (err) { console.log(`Error in getPinnedMessages:`, err); throw err; } }; archiveMultipleConnections = async (user_id: string, connection_ids: string[]) => { try { // convert connection_ids to objectIds let objectIds = connection_ids.map((id) => new Types.ObjectId(id)); // check if user is authorized to lock this connection let connections = await this.model.Connections.find({ _id: { $in: objectIds }, connection_archived_by: { $nin: [new Types.ObjectId(user_id)] } }) // check if user is authorized to lock this connection if (connection_ids?.length !== connections?.length) throw new HttpException("You are not authorized to lock this connection.", HttpStatus.BAD_REQUEST); // archive connections let update = { $addToSet: { connection_archived_by: new Types.ObjectId(user_id) } }; const options = { new: true }; await this.model.Connections.updateMany({ _id: { $in: objectIds } }, update, options); return { message: "Connection locked successfully" }; } catch (err) { console.log(` err group Deatils `, err); throw err; } }; unArchiveMultipleConnections = async (user_id: string, connection_ids: string[]) => { try { // convert connection_ids to objectIds let objectIds = connection_ids.map((id) => new Types.ObjectId(id)); // check if user is authorized to lock this connection let connections = await this.model.Connections.find({ _id: { $in: objectIds }, connection_archived_by: { $in: [new Types.ObjectId(user_id)] } }) // check if user is authorized to lock this connection if (connection_ids?.length !== connections?.length) throw new HttpException("You are not authorized to lock this connection.", HttpStatus.BAD_REQUEST); // unlock connections let update = { $pull: { connection_archived_by: new Types.ObjectId(user_id) } }; const options = { new: true }; // unlock connections await this.model.Connections.updateMany({ _id: { $in: objectIds } }, update, options); return { message: "Connection unlocked successfully" }; } catch (err) { console.log(` err group Deatils `, err); throw err; } }; deleteMultipleConnections = async (user_id: string, connection_ids: string[], is_delete_chat: boolean) => { try { // convert connection_ids to objectIds let objectIds = connection_ids.map((id) => new Types.ObjectId(id)); // check if user is authorized to lock this connection let connections = await this.model.Connections.find({ _id: { $in: objectIds }, connection_deleted_by: { $in: [new Types.ObjectId(user_id)] } }) // check if user is authorized to lock this connection if (connection_ids?.length !== connections?.length) throw new HttpException("You are not authorized to lock this connection.", HttpStatus.BAD_REQUEST); // delete connections let update = { $addToSet: { connection_deleted_by: new Types.ObjectId(user_id) }, $pull: { connection_archived_by: new Types.ObjectId(user_id) } }; const options = { new: true }; if (is_delete_chat) { await this.model.Messages.updateMany({ connection_id: { $in: objectIds } }, { $addToSet: { deleted_for: new Types.ObjectId(user_id) } }) } // delete connections await this.model.Connections.updateMany({ _id: { $in: objectIds } }, update, options); return { message: "Connection deleted successfully" }; } catch (err) { console.log(` err group Deatils `, err); throw err; } }; deleteMessageForMe = async (user_id: string, message_ids: string[]) => { try { // convert message_ids to objectIds let objectIds = message_ids.map((id) => new Types.ObjectId(id)); // check if user is authorized to delete the messages let messages = await this.model.Messages.find({ _id: { $in: objectIds }, deleted_for: { $nin: [new Types.ObjectId(user_id)] } }) // check if user is authorized to lock this connection if (message_ids?.length !== messages?.length) throw new HttpException("No Need to provide already deleted message ids.", HttpStatus.BAD_REQUEST); // delete messages await this.model.Messages.updateMany({ _id: { $in: objectIds } }, { $addToSet: { deleted_for: new Types.ObjectId(user_id) } }) return { message: "Message deleted successfully" }; } catch (err) { console.log(` err group Deatils `, err); throw err; } }; deleteMessageForEveryOne = async (user_id: string, message_ids: string[]) => { try { // convert message_ids to objectIds let objectIds = message_ids.map((id) => new Types.ObjectId(id)); // check if user is authorized to delete the messages let messages = await this.model.Messages.find({ _id: { $in: objectIds }, sent_by: new Types.ObjectId(user_id) }) // check if user is authorized to lock this connection if (message_ids?.length !== messages?.length) throw new HttpException("No Need to provide already deleted message ids.", HttpStatus.BAD_REQUEST); let deletedUserIds: Array = [] if (!messages[0].sent_to) { let reciverIds = await this.model.GroupMembers.distinct('user_id', { connection_id: messages[0].connection_id, is_exit_from_group: false }); deletedUserIds = reciverIds; } else { deletedUserIds = [messages[0].sent_to, messages[0].sent_by] } // delete messages await this.model.Messages.updateMany({ _id: { $in: objectIds } }, { deleted_for: deletedUserIds }) return { message: "Messages deleted successfully", connection_id: messages[0].connection_id }; } catch (err) { console.log(` err group Deatils `, err); throw err; } }; removeMemberFromGroup = async (user_id: string, user_ids: string[], connection_id: string) => { try { // convert message_ids to objectIds let objectIds = user_ids.map((id) => new Types.ObjectId(id)); //first find group members let currentUser = await this.model.GroupMembers.findOne({ connection_id: new Types.ObjectId(connection_id), user_id: new Types.ObjectId(user_id), is_exit_from_group: false, role: "GROUP_ADMIN" }); //check if user is authorized to remove members from group if (!currentUser) throw new HttpException("Sorry, Only group admin can remove members from group.", HttpStatus.BAD_REQUEST); //find all members let membersIds = await this.model.GroupMembers.distinct('_id', { connection_id: new Types.ObjectId(connection_id), user_id: { $in: objectIds }, is_exit_from_group: false }); if (membersIds?.length !== user_ids?.length) throw new HttpException("No Need to provide already deleted userIds ids.", HttpStatus.BAD_REQUEST); // unlock connections await this.model.GroupMembers.updateMany({ _id: { $in: membersIds } }, { $set: { is_exit_from_group: true, exit_group_at: moment().utc().valueOf() } }); return { message: "Member removed from group successfully" }; } catch (err) { console.log(` err group Deatils `, err); throw err; } }; addMemberInGroup = async (user_id: string, user_ids: string[], connection_id: string) => { try { // Step 1: Check admin access const currentUser = await this.model.GroupMembers.findOne({ connection_id: new Types.ObjectId(connection_id), user_id: new Types.ObjectId(user_id), is_exit_from_group: false, role: "GROUP_ADMIN" }); if (!currentUser) { throw new HttpException( "Sorry, only group admin can add new members in the group.", HttpStatus.BAD_REQUEST ); } // Step 3: Function to add single member const addSingleMember = (uid: string) => { return this.model.GroupMembers.findOneAndUpdate( { connection_id: new Types.ObjectId(connection_id), user_id: new Types.ObjectId(uid), }, { $set: { updated_at: moment().utc().valueOf(), is_exit_from_group: false, exit_group_at: 0 }, $setOnInsert: { role: "MEMBER", joined_at: moment().utc().valueOf(), created_at: moment().utc().valueOf() } }, { upsert: true, new: true } ); }; // Step 4: Apply limiter on each DB task const tasks = user_ids.map(uid => addSingleMember(uid)); // Step 5: Execute all safely await Promise.all(tasks); // Clear array tasks.length = 0; // Now GC can free the objects, because there are no references left in the tasks array return { message: `${user_ids.length} member(s) added to group successfully` }; } catch (err) { console.log("error addMemberInGroup:", err); throw err; } }; blockUnBlockUser = async (loggerUserId: string, connectionId: string) => { try { // _id is connection ID, find the connection to get the other user's ID const connection = await this.model.Connections.findOne({ _id: new Types.ObjectId(connectionId), connection_type: 'NORMAL' // Only allow blocking in 1-on-1 chats }); if (!connection) { throw new HttpException('Connection not found or not a 1-on-1 chat', HttpStatus.NOT_FOUND); } // Determine the other user ID from the connection let otherUserId: string; if (connection.sent_by.toString() === loggerUserId.toString()) { otherUserId = connection.sent_to.toString(); } else if (connection.sent_to.toString() === loggerUserId.toString()) { otherUserId = connection.sent_by.toString(); } else { throw new HttpException('You are not part of this connection', HttpStatus.FORBIDDEN); } // Check if already blocked let fetchBlockedUser = await this.model.blockUsers.findOne({ blocked_by: new Types.ObjectId(loggerUserId), blocked_to: new Types.ObjectId(otherUserId) }); if (fetchBlockedUser) { // Unblock the user await this.model.blockUsers.deleteOne({ blocked_by: new Types.ObjectId(loggerUserId), blocked_to: new Types.ObjectId(otherUserId) }); return { message: "User unblocked successfully" }; } else { // Block the user await this.model.blockUsers.create({ blocked_by: new Types.ObjectId(loggerUserId), blocked_to: new Types.ObjectId(otherUserId) }); return { message: "User blocked successfully" }; } } catch (error) { throw error; } } groupOperations = async (req: RequestDto, body: GroupOperationsDto) => { try { let loggerUserId = req?.user_data?._id; let { operation_type, _ids, _id, message_id } = body; console.log("operation_type",operation_type) let message = ""; switch (operation_type) { case GroupOperations.EXIT_GROUP: await this.exitFromGroup(loggerUserId, _id); message = "You have successfully exit from this group."; break; case GroupOperations.EXIT_GROUP_AND_DELETE_FOR_ME: await this.exitFromGroupAndDeleteChatForMe(loggerUserId, _id); message = "You have successfully left the group and your chat has been cleared."; break; case GroupOperations.CLEAR_CHAT: await this.clearChatForParticularConnections(loggerUserId, _id); message = "You have successfully cleared the chat."; break; case GroupOperations.LOCK_CONNECTION: await this.lockMultipleConnections(loggerUserId, _ids); message = "You have successfully locked connections."; break; case GroupOperations.UNLOCK_CONNECTION: await this.unLockMultipleConnections(loggerUserId, _ids); message = "You have successfully unlocked connections."; break; case GroupOperations.PIN_CONNECTIONS: let pinResult = await this.pinMultipleConnections(loggerUserId, _ids); message = pinResult.message; break; case GroupOperations.UNPIN_CONNECTIONS: await this.unPinMultipleConnections(loggerUserId, _ids); message = "You have successfully unpinned connections."; break; case GroupOperations.ARCHIVE_CONNECTIONS: await this.archiveMultipleConnections(loggerUserId, _ids); message = "You have successfully archived connections."; break; case GroupOperations.UNARCHIVE_CONNECTIONS: await this.unArchiveMultipleConnections(loggerUserId, _ids); message = "You have successfully unarchived connections."; break; case GroupOperations.DELETE_CONNECTIONS: await this.deleteMultipleConnections(loggerUserId, _ids, false); message = "You have successfully deleted connections."; break; case GroupOperations.DELETE_CONNECTIONS_WITH_CHAT: await this.deleteMultipleConnections(loggerUserId, _ids, true); message = "You have successfully deleted connections with chat."; break; case GroupOperations.DELETE_MESSAGES_FOR_ME: await this.deleteMessageForMe(loggerUserId, _ids); message = "You have successfully deleted messages."; break; case GroupOperations.DELETE_MESSAGES_FOR_EVERYONE: await this.deleteMessageForEveryOne(loggerUserId, _ids); message = "You have successfully deleted messages for everyone."; break; case GroupOperations.REMOVE_MEMBER_FROM_GROUP: // in this _ids count as usersIds and _id is connectionId await this.removeMemberFromGroup(loggerUserId, _ids, _id); message = "You have successfully removed member from group."; break; case GroupOperations.ADD_MEMBER_TO_GROUP: // in this _ids count as usersIds and _id is connectionId await this.addMemberInGroup(loggerUserId, _ids, _id); message = "You have successfully added member to group."; break; case GroupOperations.BLOCK_OR_UNBLOCK_USERS: let { message: retunMessage } = await this.blockUnBlockUser(loggerUserId, _id); // _id is connectionId message = retunMessage; break; case GroupOperations.STARRED_MESSAGES: case GroupOperations.UNSTARRED_MESSAGES: case GroupOperations.UNSTARRED_ALL_MESSAGES: let userIds = _ids.map((id: string) => new Types.ObjectId(id)); let { message: retunMessages } = await this.starUnstarMessage(new Types.ObjectId(_id), userIds, loggerUserId, operation_type); message = retunMessages; break; case GroupOperations.PIN_MESSAGES: // _id is used as connection_id for message pin operations if (!_id || !message_id) { throw new HttpException( "_id (connection_id) and message_id are required for PIN_MESSAGES operation", HttpStatus.BAD_REQUEST ); } return await this.pinMessage(loggerUserId, _id, message_id); case GroupOperations.UNPIN_MESSAGES: // _id is used as connection_id for message unpin operations if (!_id || !message_id) { throw new HttpException( "_id (connection_id) and message_id are required for UNPIN_MESSAGES operation", HttpStatus.BAD_REQUEST ); } return await this.unpinMessage(loggerUserId, _id, message_id); default: break; } return message } catch (error) { throw error; } } async messageNotifications(title: string, payload: any, from_user: any, connection_id: string, user_id: string, sent_to: string, message, fetchConnectionType, response_data) { try { console.log("response_data>>"); console.dir(response_data, { depth: null }); let notification_data: any = { type: "NEW_MESSAGE", title: title, message_id: payload?.message_id, message: message, sent_by: user_id, sent_to: sent_to, sent_by_name: from_user?.name, sent_by_profile_pic: from_user?.profile_pic, sent_by_user: JSON.stringify({ name: from_user?.name, profile_pic: from_user?.profile_pic }), connection_id: connection_id, connection_type: fetchConnectionType ? fetchConnectionType.connection_type : "NORMAL", chat_id: response_data[0]?._id ?? response_data?._id, }; let is_muted = await this.checkMuteOrUnmute(connection_id, sent_to); // let is_archived = await this.chatService.checkConnectionArchivedOrNot(payload?.connection_id, sent_to); notification_data.is_muted = is_muted ? true : false; if (user_id?.toString() !== sent_to?.toString()) this.sendMessageNotification(sent_to, notification_data, is_muted, false) } catch (error) { throw error; } } async editMessage(payload: EditMessageDto, user_id: string) { try { let { message_id, message } = payload; let query = { _id: new Types.ObjectId(message_id), sent_by: new Types.ObjectId(user_id), }; let get_message = await this.model.Messages.findOne(query); if (!get_message) throw new BadRequestException(`You can't edit this messsage.`); let update = { message: message, is_edited: true, last_updated_at: moment().utc().valueOf(), updated_at: +new Date(), }; let response = await this.model.Messages.findOneAndUpdate( query, update, this.new_lean_options ); return response; } catch (err) { throw err; } } handleDisconnect = async (token: string, server?: any) => { try { let payload = await this.common.verify_token(token); await this.model.UserModel.findOneAndUpdate({ _id: payload._id }, { is_online: false }); } catch (error) { console.log(error, "error will be occured while disconnecting socket.........."); } } async sendMessageWithPush(userId: string, payload: SendMessageFromPushDto) { try { let user_id: any = new Types.ObjectId(userId); let { sent_to, connection_id } = payload; let firstCheckConnections = await this.model.Connections.findOne({ _id: new Types.ObjectId(connection_id) }); console.log(firstCheckConnections, "firstCheckConnections"); const from_user: any = await this.getUser(user_id, this.projection); let is_another_user_blocket_me = false; if (firstCheckConnections?.connection_type === "NORMAL") { let checkBlocked = await this.model.blockUsers.find({ blocked_by: user_id, blocked_to: sent_to }); if (checkBlocked.length) throw new HttpException({ message: "Please first unblock the user" }, HttpStatus.BAD_REQUEST); let checkAnotherUseerBlocked = await this.model.blockUsers.countDocuments({ blocked_by: sent_to, blocked_to: user_id }); is_another_user_blocket_me = checkAnotherUseerBlocked > 0; } return await this.gatewayService.singleMessageSave(user_id, payload, connection_id, from_user); } catch (error) { throw error; } } @Cron('* * * * *') // Runs every minute async messageDisappearing() { try { let currentTime = moment().utc().valueOf(); // Fetch all messages that are set to disappear but not yet disappeared let messages = await this.model.Messages.find({ is_disappeared_msg: true, is_already_disappeared: false }); if (!messages.length) return; let updatePromises = messages.map(async (msg) => { let { _id, connection_id, sent_by, sent_to, chat_disappear_by = [], msg_disappearing_date, deleted_for = [] } = msg; let deleteForUsers: any[] = []; let alreadyDeleted = new Set(deleted_for.map(id => id.toString())); // ------------------------------------------------------------------- // 1️⃣ HANDLE USER-LEVEL DISAPPEARING (per user logic) // ------------------------------------------------------------------- for (let setting of chat_disappear_by) { let { user_id, is_already_disappeared, msg_disappearing_date } = setting; let disappearTime = moment(msg_disappearing_date).utc().valueOf(); if (!is_already_disappeared && disappearTime <= currentTime) { if (!alreadyDeleted.has(user_id?.toString())) { deleteForUsers.push(new Types.ObjectId(user_id)); } setting.is_already_disappeared = true; setting.is_disappeared_msg = true; } } // ------------------------------------------------------------------- // 2️⃣ HANDLE MESSAGE-LEVEL DISAPPEARING (connection-level setting) // ------------------------------------------------------------------- if (msg_disappearing_date && msg_disappearing_date !== 0) { let disappearTime = moment(msg_disappearing_date).utc().valueOf(); if (disappearTime <= currentTime) { // NORMAL CHAT if (msg.connection_type === "NORMAL") { if (!alreadyDeleted.has(sent_by?.toString())) { deleteForUsers.push(sent_by); } if (!alreadyDeleted.has(sent_to?.toString())) { deleteForUsers.push(sent_to); } } // GROUP CHAT if (msg.connection_type === "GROUP") { let groupMembers = await this.model.GroupMembers.distinct( "user_id", { connection_id, is_exit_from_group: false } ); groupMembers.forEach((user) => { if (!alreadyDeleted.has(user?.toString())) { deleteForUsers.push(user); } }); } msg.is_already_disappeared = true; } } // If no users need deletion, skip update if (!deleteForUsers.length) return; // Final DB update deleteForUsers = deleteForUsers.filter((id) => id !== null); return this.model.Messages.updateOne( { _id }, { $addToSet: { deleted_for: { $each: deleteForUsers } }, chat_disappear_by, is_already_disappeared: msg.is_already_disappeared } ); }); await Promise.all(updatePromises); } catch (error) { console.error("Error in messageDisappearing:", error); throw error; } } }