(20241213) Mail Module Reworked!
This commit is contained in:
+85
-65
@@ -129,6 +129,13 @@ class MessageController(BaseModel):
|
||||
message: CoreMessageModel
|
||||
) -> ObjectId:
|
||||
|
||||
"""
|
||||
Simply insert one message document into the database.
|
||||
:param mongo_conn: The instance of the database connector to use for the operation.
|
||||
:param message: The message to save into the database.
|
||||
:return: The object id of the inserted document.
|
||||
"""
|
||||
|
||||
# Simply insert the document:
|
||||
return await mongo_conn.insert_one(
|
||||
collection = self.MESSAGES_COLLECTION,
|
||||
@@ -139,9 +146,17 @@ class MessageController(BaseModel):
|
||||
async def bulk_write(
|
||||
self,
|
||||
mongo_conn: AsyncMongo,
|
||||
mongo_operations
|
||||
mongo_operations: list
|
||||
) -> int:
|
||||
|
||||
"""
|
||||
Needed in cases like forcing re-sync of mails where you need to perform actions like bulk replacements of
|
||||
existing documents. Not recommended to use. Please use very carefully to ensure document integrity.
|
||||
:param mongo_conn: The instance of the database connector to use for the operation.
|
||||
:param mongo_operations: The list operations that are supported by MongoDB's Bulk Write system.
|
||||
:return: The no. of documents affected.
|
||||
"""
|
||||
|
||||
return await mongo_conn.bulk_write(
|
||||
collection = self.MESSAGES_COLLECTION,
|
||||
requests = mongo_operations,
|
||||
@@ -195,7 +210,7 @@ class MessageController(BaseModel):
|
||||
) -> List[CoreMessageModel] | None:
|
||||
|
||||
"""
|
||||
Fetches many messages in one call, but leaves out the full payloads. This does not mark messages as read.
|
||||
Fetches many messages in one call, but leaves out the full payloads.
|
||||
:param mongo_conn: The instance of the database connector to use for the operation.
|
||||
:param token_ids: The token ids of the accounts from which these messages must be fetched.
|
||||
:param limit: The max. no. of messages to retrieve in this call.
|
||||
@@ -223,8 +238,8 @@ class MessageController(BaseModel):
|
||||
projection = {
|
||||
"_id": True,
|
||||
"ts": True,
|
||||
"syncTs": True,
|
||||
"tokenId": True,
|
||||
"markedAsUnread": True,
|
||||
"serviceType": True,
|
||||
"client": True,
|
||||
"clientMessageId": True,
|
||||
@@ -232,10 +247,11 @@ class MessageController(BaseModel):
|
||||
"isSent": True,
|
||||
"isBroadcast": True,
|
||||
"sentSuccessfully": True,
|
||||
"sender": True,
|
||||
"chat": True,
|
||||
"snippet": True,
|
||||
"aiSnippet": True,
|
||||
"preview": True,
|
||||
"tags": True,
|
||||
"usedAi": True
|
||||
"tags": True
|
||||
},
|
||||
raise_exception = True
|
||||
)
|
||||
@@ -263,9 +279,6 @@ class MessageController(BaseModel):
|
||||
:return: The list of messages (as the message model). This list can be empty.
|
||||
"""
|
||||
|
||||
# Note down the timestamp at which this event occurred:
|
||||
request_ts = date_time.get_current_utc_date_time(as_string = False)
|
||||
|
||||
# Prepare the filter:
|
||||
if not isinstance(token_ids, list): token_ids = [token_ids]
|
||||
token_ids = [ObjectId(t) for t in token_ids]
|
||||
@@ -285,75 +298,31 @@ class MessageController(BaseModel):
|
||||
raise_exception = True
|
||||
)
|
||||
|
||||
# We now mark these fetched messages as read through a bulk-write operation:
|
||||
operations = [
|
||||
UpdateOne(
|
||||
filter = {"_id": record["_id"]},
|
||||
update = [{
|
||||
"$set": {
|
||||
"readTs": {
|
||||
"$cond": {
|
||||
"if": {
|
||||
"$or": [
|
||||
{"$eq": ["$readTs", None]},
|
||||
{"$eq": [{"$type": "$readTs"}, "missing"]}
|
||||
]
|
||||
},
|
||||
"then": request_ts,
|
||||
"else": "$readTs"
|
||||
}
|
||||
}
|
||||
}
|
||||
}],
|
||||
upsert = False
|
||||
) for record in records
|
||||
]
|
||||
updated_count = await mongo_conn.bulk_write(
|
||||
collection = self.MESSAGES_COLLECTION,
|
||||
requests = operations,
|
||||
raise_exception = True
|
||||
)
|
||||
|
||||
# Convert the fetched records to instances of the data model and return:
|
||||
return [CoreMessageModel(**record) for record in records]
|
||||
|
||||
async def get_message(
|
||||
self,
|
||||
mongo_conn: AsyncMongo,
|
||||
token_id: ObjectId | str,
|
||||
message_id: ObjectId | str,
|
||||
) -> CoreMessageModel | None:
|
||||
|
||||
"""
|
||||
Gets one message if you know its message id. Marks that message as read.
|
||||
:param mongo_conn:
|
||||
:param message_id:
|
||||
:return:
|
||||
Gets one message if you know its message id.
|
||||
:param mongo_conn: The instance of the database connector to use for the operation.
|
||||
:param token_id: The id of the auth-token associated with the message. Needed for security.
|
||||
:param message_id: The id of the message that needs to be read.
|
||||
:return: The contents of that one message in a structured format.
|
||||
"""
|
||||
|
||||
# Note down the timestamp at which this event occurred:
|
||||
request_ts = date_time.get_current_utc_date_time(as_string = False)
|
||||
|
||||
# We fetch the whole payload of that one message
|
||||
# while also marking it as read if not already marked:
|
||||
record = await mongo_conn.find_one_and_update(
|
||||
# We fetch the whole payload of that one message:
|
||||
record = await mongo_conn.find_one(
|
||||
collection = self.MESSAGES_COLLECTION,
|
||||
filter = {"_id": ObjectId(message_id)},
|
||||
update = [{
|
||||
"$set": {
|
||||
"readTs": {
|
||||
"$cond": {
|
||||
"if": {
|
||||
"$or": [
|
||||
{"$eq": ["$readTs", None]},
|
||||
{"$eq": [{"$type": "$readTs"}, "missing"]}
|
||||
]
|
||||
},
|
||||
"then": request_ts,
|
||||
"else": "$readTs"
|
||||
}
|
||||
}
|
||||
}
|
||||
}],
|
||||
filter = {
|
||||
"_id": ObjectId(message_id),
|
||||
"tokenId": ObjectId(token_id)
|
||||
},
|
||||
raise_exception = True
|
||||
)
|
||||
|
||||
@@ -372,6 +341,57 @@ class MessageController(BaseModel):
|
||||
# We don't support updating messages themselves,
|
||||
# but we will allow updating fields like tags, marking as read or unread, etc.
|
||||
|
||||
async def update_tags(
|
||||
self,
|
||||
mongo_conn: AsyncMongo,
|
||||
token_id: ObjectId | str,
|
||||
message_id: ObjectId | str,
|
||||
unset_tags: List[str] = None,
|
||||
set_tags: List[str] = None
|
||||
) -> bool:
|
||||
|
||||
"""
|
||||
Updates the tags on one message. The tags to remove are processed first, the ones to add are processed later.
|
||||
:param mongo_conn: The instance of the database connector to use for the operation.
|
||||
:param token_id: The id of the auth-token associated with the message. Needed for security.
|
||||
:param message_id: The id of the message that needs to be read.
|
||||
:param unset_tags: The tags to remove from the message.
|
||||
:param set_tags: The tags to add to the message.
|
||||
:return: True if the update was successful, else False.
|
||||
"""
|
||||
|
||||
# Update the tags:
|
||||
return await mongo_conn.update_one(
|
||||
collection = self.MESSAGES_COLLECTION,
|
||||
filter = {
|
||||
"_id": ObjectId(message_id),
|
||||
"tokenId": ObjectId(token_id)
|
||||
},
|
||||
update = [{
|
||||
"$set": {
|
||||
"tags": {
|
||||
"$let": {
|
||||
"vars": {
|
||||
"removed_tags": {
|
||||
"$setDifference": [
|
||||
"$tags",
|
||||
unset_tags
|
||||
]
|
||||
}
|
||||
},
|
||||
"in": {
|
||||
"$setUnion": [
|
||||
"$$removed_tags",
|
||||
set_tags
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}],
|
||||
raise_exception = True
|
||||
)
|
||||
|
||||
# ┏┓┳┓┳┳┳┓ ┳┓ ┓
|
||||
# ┃ ┣┫┃┃┃┃ ━━ ┃┃┏┓┃┏┓╋┏┓
|
||||
# ┗┛┛┗┗┛┻┛ ┻┛┗ ┗┗ ┗┗
|
||||
|
||||
Reference in New Issue
Block a user