message.go 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202
  1. package database
  2. import (
  3. "database/sql"
  4. "errors"
  5. "fmt"
  6. "strings"
  7. "time"
  8. log "maunium.net/go/maulogger/v2"
  9. "maunium.net/go/mautrix/id"
  10. "maunium.net/go/mautrix/util/dbutil"
  11. )
  12. type MessageQuery struct {
  13. db *Database
  14. log log.Logger
  15. }
  16. const (
  17. messageSelect = "SELECT dcid, dc_attachment_id, dc_edit_index, dc_chan_id, dc_chan_receiver, dc_sender, timestamp, dc_thread_id, mxid FROM message"
  18. )
  19. func (mq *MessageQuery) New() *Message {
  20. return &Message{
  21. db: mq.db,
  22. log: mq.log,
  23. }
  24. }
  25. func (mq *MessageQuery) scanAll(rows dbutil.Rows, err error) []*Message {
  26. if err != nil {
  27. mq.log.Warnfln("Failed to query many messages: %v", err)
  28. panic(err)
  29. } else if rows == nil {
  30. return nil
  31. }
  32. var messages []*Message
  33. for rows.Next() {
  34. messages = append(messages, mq.New().Scan(rows))
  35. }
  36. return messages
  37. }
  38. func (mq *MessageQuery) GetByDiscordID(key PortalKey, discordID string) []*Message {
  39. query := messageSelect + " WHERE dc_chan_id=$1 AND dc_chan_receiver=$2 AND dcid=$3 AND dc_edit_index=0 ORDER BY dc_attachment_id ASC"
  40. return mq.scanAll(mq.db.Query(query, key.ChannelID, key.Receiver, discordID))
  41. }
  42. func (mq *MessageQuery) GetFirstByDiscordID(key PortalKey, discordID string) *Message {
  43. query := messageSelect + " WHERE dc_chan_id=$1 AND dc_chan_receiver=$2 AND dcid=$3 AND dc_edit_index=0 ORDER BY dc_attachment_id ASC LIMIT 1"
  44. return mq.New().Scan(mq.db.QueryRow(query, key.ChannelID, key.Receiver, discordID))
  45. }
  46. func (mq *MessageQuery) GetLastByDiscordID(key PortalKey, discordID string) *Message {
  47. query := messageSelect + " WHERE dc_chan_id=$1 AND dc_chan_receiver=$2 AND dcid=$3 AND dc_edit_index=0 ORDER BY dc_attachment_id DESC LIMIT 1"
  48. return mq.New().Scan(mq.db.QueryRow(query, key.ChannelID, key.Receiver, discordID))
  49. }
  50. func (mq *MessageQuery) GetClosestBefore(key PortalKey, threadID string, ts time.Time) *Message {
  51. query := messageSelect + " WHERE dc_chan_id=$1 AND dc_chan_receiver=$2 AND dc_thread_id=$3 AND timestamp<=$4 ORDER BY timestamp DESC, dc_attachment_id DESC LIMIT 1"
  52. return mq.New().Scan(mq.db.QueryRow(query, key.ChannelID, key.Receiver, threadID, ts.UnixMilli()))
  53. }
  54. func (mq *MessageQuery) GetLastInThread(key PortalKey, threadID string) *Message {
  55. query := messageSelect + " WHERE dc_chan_id=$1 AND dc_chan_receiver=$2 AND dc_thread_id=$3 AND dc_edit_index=0 ORDER BY timestamp DESC, dc_attachment_id DESC LIMIT 1"
  56. return mq.New().Scan(mq.db.QueryRow(query, key.ChannelID, key.Receiver, threadID))
  57. }
  58. func (mq *MessageQuery) GetLast(key PortalKey) *Message {
  59. query := messageSelect + " WHERE dc_chan_id=$1 AND dc_chan_receiver=$2 AND dc_edit_index=0 ORDER BY timestamp DESC LIMIT 1"
  60. return mq.New().Scan(mq.db.QueryRow(query, key.ChannelID, key.Receiver))
  61. }
  62. func (mq *MessageQuery) DeleteAll(key PortalKey) {
  63. query := "DELETE FROM message WHERE dc_chan_id=$1 AND dc_chan_receiver=$2"
  64. _, err := mq.db.Exec(query, key.ChannelID, key.Receiver)
  65. if err != nil {
  66. mq.log.Warnfln("Failed to delete messages of %s: %v", key, err)
  67. panic(err)
  68. }
  69. }
  70. func (mq *MessageQuery) GetByMXID(key PortalKey, mxid id.EventID) *Message {
  71. query := messageSelect + " WHERE dc_chan_id=$1 AND dc_chan_receiver=$2 AND mxid=$3"
  72. row := mq.db.QueryRow(query, key.ChannelID, key.Receiver, mxid)
  73. if row == nil {
  74. return nil
  75. }
  76. return mq.New().Scan(row)
  77. }
  78. type Message struct {
  79. db *Database
  80. log log.Logger
  81. DiscordID string
  82. AttachmentID string
  83. EditIndex int
  84. Channel PortalKey
  85. SenderID string
  86. Timestamp time.Time
  87. ThreadID string
  88. MXID id.EventID
  89. }
  90. func (m *Message) DiscordProtoChannelID() string {
  91. if m.ThreadID != "" {
  92. return m.ThreadID
  93. } else {
  94. return m.Channel.ChannelID
  95. }
  96. }
  97. func (m *Message) Scan(row dbutil.Scannable) *Message {
  98. var ts int64
  99. err := row.Scan(&m.DiscordID, &m.AttachmentID, &m.EditIndex, &m.Channel.ChannelID, &m.Channel.Receiver, &m.SenderID, &ts, &m.ThreadID, &m.MXID)
  100. if err != nil {
  101. if !errors.Is(err, sql.ErrNoRows) {
  102. m.log.Errorln("Database scan failed:", err)
  103. panic(err)
  104. }
  105. return nil
  106. }
  107. if ts != 0 {
  108. m.Timestamp = time.UnixMilli(ts)
  109. }
  110. return m
  111. }
  112. const messageInsertQuery = `
  113. INSERT INTO message (
  114. dcid, dc_attachment_id, dc_edit_index, dc_chan_id, dc_chan_receiver, dc_sender, timestamp, dc_thread_id, mxid
  115. )
  116. VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
  117. `
  118. var messageMassInsertTemplate = strings.Replace(messageInsertQuery, "($1, $2, $3, $4, $5, $6, $7, $8, $9)", "%s", 1)
  119. type MessagePart struct {
  120. AttachmentID string
  121. MXID id.EventID
  122. }
  123. func (m *Message) MassInsert(msgs []MessagePart) {
  124. if len(msgs) == 0 {
  125. return
  126. }
  127. valueStringFormat := "($1, $%d, $2, $3, $4, $5, $6, $7, $%d)"
  128. if m.db.Dialect == dbutil.SQLite {
  129. valueStringFormat = strings.ReplaceAll(valueStringFormat, "$", "?")
  130. }
  131. params := make([]interface{}, 7+len(msgs)*2)
  132. placeholders := make([]string, len(msgs))
  133. params[0] = m.DiscordID
  134. params[1] = m.EditIndex
  135. params[2] = m.Channel.ChannelID
  136. params[3] = m.Channel.Receiver
  137. params[4] = m.SenderID
  138. params[5] = m.Timestamp.UnixMilli()
  139. params[6] = m.ThreadID
  140. for i, msg := range msgs {
  141. params[7+i*2] = msg.AttachmentID
  142. params[7+i*2+1] = msg.MXID
  143. placeholders[i] = fmt.Sprintf(valueStringFormat, 7+i*2+1, 7+i*2+2)
  144. }
  145. _, err := m.db.Exec(fmt.Sprintf(messageMassInsertTemplate, strings.Join(placeholders, ", ")), params...)
  146. if err != nil {
  147. m.log.Warnfln("Failed to insert %d parts of %s@%s: %v", len(msgs), m.DiscordID, m.Channel, err)
  148. panic(err)
  149. }
  150. }
  151. func (m *Message) Insert() {
  152. _, err := m.db.Exec(messageInsertQuery,
  153. m.DiscordID, m.AttachmentID, m.EditIndex, m.Channel.ChannelID, m.Channel.Receiver, m.SenderID,
  154. m.Timestamp.UnixMilli(), m.ThreadID, m.MXID)
  155. if err != nil {
  156. m.log.Warnfln("Failed to insert %s@%s: %v", m.DiscordID, m.Channel, err)
  157. panic(err)
  158. }
  159. }
  160. func (m *Message) Delete() {
  161. query := "DELETE FROM message WHERE dcid=$1 AND dc_chan_id=$2 AND dc_chan_receiver=$3 AND dc_attachment_id=$4"
  162. _, err := m.db.Exec(query, m.DiscordID, m.Channel.ChannelID, m.Channel.Receiver, m.AttachmentID)
  163. if err != nil {
  164. m.log.Warnfln("Failed to delete %q of %s@%s: %v", m.AttachmentID, m.DiscordID, m.Channel, err)
  165. panic(err)
  166. }
  167. }