import json from datetime import datetime, timezone from typing import List from services.auth import login_required from services.presence import notify_message from services.rediscache import redis from services.schema import mutation from validators.chat import Message @mutation.field("createMessage") @login_required async def create_message(_, info, chat: str, body: str, reply_to=None): """Создание сообщения с телом :body для чата :chat_id с возможным ответом на :reply_to""" author_id = info.context["author_id"] # Получение данных чата из Redis chat_data = await redis.execute("GET", f"chats/{chat}") print(f"[resolvers.messages] debug chat data: {chat_data}") # Если данных чата нет, возвращаем ошибку if not chat_data: return {"error": "chat is not exist"} else: # Преобразование данных чата из строки JSON в словарь chat_dict = json.loads(chat_data) print(chat_dict) # Получение ID следующего сообщения message_id = await redis.execute("GET", f"chats/{chat_dict['id']}/next_message_id") message_id = int(message_id) if message_id else 0 # Создание нового сообщения new_message: Message = { "chat": chat_dict["id"], "id": message_id, "author": author_id, "body": body, "createdAt": int(datetime.now(tz=timezone.utc).timestamp()), "updatedAt": None, } # Если есть ответ, добавляем его в сообщение if reply_to: new_message["replyTo"] = reply_to # Обновление времени последнего обновления чата chat_dict["updatedAt"] = new_message["createdAt"] # Запись обновленных данных чата обратно в Redis await redis.execute("SET", f"chats/{chat_dict['id']}", json.dumps(chat_dict)) print(f"[inbox] creating message {new_message}") # Запись нового сообщения в Redis await redis.execute( "SET", f"chats/{chat_dict['id']}/messages/{message_id}", json.dumps(new_message), ) # Добавление ID нового сообщения в список ID сообщений чата await redis.execute("LPUSH", f"chats/{chat_dict['id']}/message_ids", str(message_id)) # Обновление ID следующего сообщения await redis.execute("SET", f"chats/{chat_dict['id']}/next_message_id", str(message_id + 1)) # Добавление нового сообщения в список непрочитанных сообщений для каждого участника чата members = chat_dict["members"] for member_id in members: await redis.execute("LPUSH", f"chats/{chat_dict['id']}/unread/{member_id}", str(message_id)) # Отправка уведомления о новом сообщении await notify_message(new_message, chat_dict["id"]) return {"message": new_message, "error": None} @mutation.field("updateMessage") @login_required async def update_message(_, info, chat_id: str, message_id: int, body: str): author_id = info.context["author_id"] chat_str = await redis.execute("GET", f"chats/{chat_id}") if not chat_str: return {"error": "chat not exist"} message = await redis.execute("GET", f"chats/{chat_id}/messages/{message_id}") if not message: return {"error": "message not exist"} message = json.loads(message) if message["author"] != author_id: return {"error": "access denied"} message["body"] = body message["updatedAt"] = int(datetime.now(tz=timezone.utc).timestamp()) await redis.execute("SET", f"chats/{chat_id}/messages/{message_id}", json.dumps(message)) # TODO: use presence service to notify about updated message return {"message": message, "error": None} @mutation.field("deleteMessage") @login_required async def delete_message(_, info, chat_id: str, message_id: int): author_id = info.context["author_id"] chat_str = await redis.execute("GET", f"chats/{chat_id}") if not chat_str: return {"error": "chat not exist"} chat = json.loads(chat_str) message_data = await redis.execute("GET", f"chats/{chat_id}/messages/{str(message_id)}") if not message_data: return {"error": "message not exist"} message: Message = json.loads(message_data) if message["author"] != author_id: return {"error": "access denied"} await redis.execute("LREM", f"chats/{chat_id}/message_ids", 0, str(message_id)) await redis.execute("DEL", f"chats/{chat_id}/messages/{str(message_id)}") members = chat["members"] for member_id in members: await redis.execute("LREM", f"chats/{chat_id}/unread/{member_id}", 0, str(message_id)) # TODO: use presence service to notify about deleted message return {} @mutation.field("markAsRead") @login_required async def mark_as_read(_, info, chat_id: str, messages: List[int]): author_id = info.context["author_id"] chat_str = await redis.execute("GET", f"chats/{chat_id}") if not chat_str: return {"error": "chat not exist"} chat = json.loads(chat_str) members = set(chat["members"]) if author_id not in members: return {"error": "access denied"} for message_id in messages: await redis.execute("LREM", f"chats/{chat_id}/unread/{author_id}", 0, str(message_id)) return {"error": None}