import json import time from services.auth import login_required from services.presence import notify_message from services.rediscache import redis from services.schema import mutation from models.chat import Message @mutation.field("create_message") @login_required async def create_message(_, info, chat_id: 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_id}") 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 chat_id = chat_dict["id"] # Создание нового сообщения new_message: Message = { "chat_id": chat_id, "id": message_id, "created_by": author_id, "body": body, "created_at": int(time.time()), "updated_at": None, } # Если есть ответ, добавляем его в сообщение if reply_to: new_message["reply_to"] = reply_to # Обновление времени последнего обновления чата chat_dict["updated_at"] = new_message["created_at"] # Запись обновленных данных чата обратно в Redis await redis.execute("SET", f"chats/{chat_id}", json.dumps(chat_dict)) print(f"[inbox] creating message {new_message}") # Запись нового сообщения в Redis await redis.execute( "SET", f"chats/{chat_id}/messages/{message_id}", json.dumps(new_message), ) # Добавление ID нового сообщения в список ID сообщений чата await redis.execute("LPUSH", f"chats/{chat_id}/message_ids", str(message_id)) # Обновление ID следующего сообщения await redis.execute("SET", f"chats/{chat_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)) # Отправка уведомления о новом сообщении new_message["chat_id"] = chat_id await notify_message(new_message, "create") return {"message": new_message, "error": None} @mutation.field("update_message") @login_required async def update_message(_, info, message): author_id = info.context["author_id"] chat_id = message.get("chat_id") chat_str = "" if chat_id: chat_str = await redis.execute("GET", f"chats/{chat_id}") if not chat_str: return {"error": "chat not exist"} message_id = message.get("id") body = message.get("body") if message_id: 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["created_by"] != author_id: return {"error": "access denied"} if body: message["body"] = body message["updated_at"] = int(datetime.now(tz=timezone.utc).timestamp()) await redis.execute("SET", f"chats/{chat_id}/messages/{message_id}", json.dumps(message)) # Отправка уведомления message["chat_id"] = chat_id await notify_message(message, "update") return {"message": message, "error": None} else: return {"message": message, "error": "cannot update, no message_id"} @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)) message["chat_id"] = chat_id await notify_message(message, "delete") return {} @mutation.field("mark_as_read") @login_required async def mark_as_read(_, 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) members = set(chat["members"]) if author_id not in members: return {"error": "access denied"} await redis.execute("LREM", f"chats/{chat_id}/unread/{author_id}", 0, str(message_id)) 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) await notify_message(message, "seen") return {"error": None}