messages.py 8.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277
  1. import json
  2. import time
  3. import uuid
  4. from typing import Optional
  5. from open_webui.internal.db import Base, get_db
  6. from open_webui.models.tags import TagModel, Tag, Tags
  7. from pydantic import BaseModel, ConfigDict
  8. from sqlalchemy import BigInteger, Boolean, Column, String, Text, JSON
  9. from sqlalchemy import or_, func, select, and_, text
  10. from sqlalchemy.sql import exists
  11. ####################
  12. # Message DB Schema
  13. ####################
  14. class MessageReaction(Base):
  15. __tablename__ = "message_reaction"
  16. id = Column(Text, primary_key=True)
  17. user_id = Column(Text)
  18. message_id = Column(Text)
  19. name = Column(Text)
  20. created_at = Column(BigInteger)
  21. class MessageReactionModel(BaseModel):
  22. model_config = ConfigDict(from_attributes=True)
  23. id: str
  24. user_id: str
  25. message_id: str
  26. name: str
  27. created_at: int # timestamp in epoch
  28. class Message(Base):
  29. __tablename__ = "message"
  30. id = Column(Text, primary_key=True)
  31. user_id = Column(Text)
  32. channel_id = Column(Text, nullable=True)
  33. parent_id = Column(Text, nullable=True)
  34. content = Column(Text)
  35. data = Column(JSON, nullable=True)
  36. meta = Column(JSON, nullable=True)
  37. created_at = Column(BigInteger) # time_ns
  38. updated_at = Column(BigInteger) # time_ns
  39. class MessageModel(BaseModel):
  40. model_config = ConfigDict(from_attributes=True)
  41. id: str
  42. user_id: str
  43. channel_id: Optional[str] = None
  44. parent_id: Optional[str] = None
  45. content: str
  46. data: Optional[dict] = None
  47. meta: Optional[dict] = None
  48. created_at: int # timestamp in epoch
  49. updated_at: int # timestamp in epoch
  50. ####################
  51. # Forms
  52. ####################
  53. class MessageForm(BaseModel):
  54. content: str
  55. parent_id: Optional[str] = None
  56. data: Optional[dict] = None
  57. meta: Optional[dict] = None
  58. class Reactions(BaseModel):
  59. name: str
  60. user_ids: list[str]
  61. count: int
  62. class MessageResponse(MessageModel):
  63. latest_reply_at: Optional[int]
  64. reply_count: int
  65. reactions: list[Reactions]
  66. class MessageTable:
  67. def insert_new_message(
  68. self, form_data: MessageForm, channel_id: str, user_id: str
  69. ) -> Optional[MessageModel]:
  70. with get_db() as db:
  71. id = str(uuid.uuid4())
  72. ts = int(time.time_ns())
  73. message = MessageModel(
  74. **{
  75. "id": id,
  76. "user_id": user_id,
  77. "channel_id": channel_id,
  78. "parent_id": form_data.parent_id,
  79. "content": form_data.content,
  80. "data": form_data.data,
  81. "meta": form_data.meta,
  82. "created_at": ts,
  83. "updated_at": ts,
  84. }
  85. )
  86. result = Message(**message.model_dump())
  87. db.add(result)
  88. db.commit()
  89. db.refresh(result)
  90. return MessageModel.model_validate(result) if result else None
  91. def get_message_by_id(self, id: str) -> Optional[MessageResponse]:
  92. with get_db() as db:
  93. message = db.get(Message, id)
  94. if not message:
  95. return None
  96. reactions = self.get_reactions_by_message_id(id)
  97. replies = self.get_replies_by_message_id(id)
  98. return MessageResponse(
  99. **{
  100. **MessageModel.model_validate(message).model_dump(),
  101. "latest_reply_at": replies[0].created_at if replies else None,
  102. "reply_count": len(replies),
  103. "reactions": reactions,
  104. }
  105. )
  106. def get_replies_by_message_id(self, id: str) -> list[MessageModel]:
  107. with get_db() as db:
  108. all_messages = (
  109. db.query(Message)
  110. .filter_by(parent_id=id)
  111. .order_by(Message.created_at.desc())
  112. .all()
  113. )
  114. return [MessageModel.model_validate(message) for message in all_messages]
  115. def get_reply_user_ids_by_message_id(self, id: str) -> list[str]:
  116. with get_db() as db:
  117. return [
  118. message.user_id
  119. for message in db.query(Message).filter_by(parent_id=id).all()
  120. ]
  121. def get_messages_by_channel_id(
  122. self, channel_id: str, skip: int = 0, limit: int = 50
  123. ) -> list[MessageModel]:
  124. with get_db() as db:
  125. all_messages = (
  126. db.query(Message)
  127. .filter_by(channel_id=channel_id, parent_id=None)
  128. .order_by(Message.created_at.desc())
  129. .offset(skip)
  130. .limit(limit)
  131. .all()
  132. )
  133. return [MessageModel.model_validate(message) for message in all_messages]
  134. def get_messages_by_parent_id(
  135. self, channel_id: str, parent_id: str, skip: int = 0, limit: int = 50
  136. ) -> list[MessageModel]:
  137. with get_db() as db:
  138. message = db.get(Message, parent_id)
  139. if not message:
  140. return []
  141. all_messages = (
  142. db.query(Message)
  143. .filter_by(channel_id=channel_id, parent_id=parent_id)
  144. .order_by(Message.created_at.desc())
  145. .offset(skip)
  146. .limit(limit)
  147. .all()
  148. )
  149. return [
  150. MessageModel.model_validate(message) for message in all_messages
  151. ] + [MessageModel.model_validate(message)]
  152. def update_message_by_id(
  153. self, id: str, form_data: MessageForm
  154. ) -> Optional[MessageModel]:
  155. with get_db() as db:
  156. message = db.get(Message, id)
  157. message.content = form_data.content
  158. message.data = form_data.data
  159. message.meta = form_data.meta
  160. message.updated_at = int(time.time_ns())
  161. db.commit()
  162. db.refresh(message)
  163. return MessageModel.model_validate(message) if message else None
  164. def add_reaction_to_message(
  165. self, id: str, user_id: str, name: str
  166. ) -> Optional[MessageReactionModel]:
  167. with get_db() as db:
  168. reaction_id = str(uuid.uuid4())
  169. reaction = MessageReactionModel(
  170. id=reaction_id,
  171. user_id=user_id,
  172. message_id=id,
  173. name=name,
  174. created_at=int(time.time_ns()),
  175. )
  176. result = MessageReaction(**reaction.model_dump())
  177. db.add(result)
  178. db.commit()
  179. db.refresh(result)
  180. return MessageReactionModel.model_validate(result) if result else None
  181. def get_reactions_by_message_id(self, id: str) -> list[Reactions]:
  182. with get_db() as db:
  183. all_reactions = db.query(MessageReaction).filter_by(message_id=id).all()
  184. reactions = {}
  185. for reaction in all_reactions:
  186. if reaction.name not in reactions:
  187. reactions[reaction.name] = {
  188. "name": reaction.name,
  189. "user_ids": [],
  190. "count": 0,
  191. }
  192. reactions[reaction.name]["user_ids"].append(reaction.user_id)
  193. reactions[reaction.name]["count"] += 1
  194. return [Reactions(**reaction) for reaction in reactions.values()]
  195. def remove_reaction_by_id_and_user_id_and_name(
  196. self, id: str, user_id: str, name: str
  197. ) -> bool:
  198. with get_db() as db:
  199. db.query(MessageReaction).filter_by(
  200. message_id=id, user_id=user_id, name=name
  201. ).delete()
  202. db.commit()
  203. return True
  204. def delete_reactions_by_id(self, id: str) -> bool:
  205. with get_db() as db:
  206. db.query(MessageReaction).filter_by(message_id=id).delete()
  207. db.commit()
  208. return True
  209. def delete_replies_by_id(self, id: str) -> bool:
  210. with get_db() as db:
  211. db.query(Message).filter_by(parent_id=id).delete()
  212. db.commit()
  213. return True
  214. def delete_message_by_id(self, id: str) -> bool:
  215. with get_db() as db:
  216. db.query(Message).filter_by(id=id).delete()
  217. # Delete all reactions to this message
  218. db.query(MessageReaction).filter_by(message_id=id).delete()
  219. db.commit()
  220. return True
  221. Messages = MessageTable()