message.go 5.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199
  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 *sql.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, ts time.Time) *Message {
  51. query := messageSelect + " WHERE dc_chan_id=$1 AND dc_chan_receiver=$2 AND timestamp<=$3 ORDER BY timestamp DESC, dc_attachment_id DESC LIMIT 1"
  52. return mq.New().Scan(mq.db.QueryRow(query, key.ChannelID, key.Receiver, 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) DeleteAll(key PortalKey) {
  59. query := "DELETE FROM message WHERE dc_chan_id=$1 AND dc_chan_receiver=$2"
  60. _, err := mq.db.Exec(query, key.ChannelID, key.Receiver)
  61. if err != nil {
  62. mq.log.Warnfln("Failed to delete messages of %s: %v", key, err)
  63. panic(err)
  64. }
  65. }
  66. func (mq *MessageQuery) GetByMXID(key PortalKey, mxid id.EventID) *Message {
  67. query := messageSelect + " WHERE dc_chan_id=$1 AND dc_chan_receiver=$2 AND mxid=$3"
  68. row := mq.db.QueryRow(query, key.ChannelID, key.Receiver, mxid)
  69. if row == nil {
  70. return nil
  71. }
  72. return mq.New().Scan(row)
  73. }
  74. type Message struct {
  75. db *Database
  76. log log.Logger
  77. DiscordID string
  78. AttachmentID string
  79. EditIndex int
  80. Channel PortalKey
  81. SenderID string
  82. Timestamp time.Time
  83. ThreadID string
  84. MXID id.EventID
  85. }
  86. func (m *Message) DiscordProtoChannelID() string {
  87. if m.ThreadID != "" {
  88. return m.ThreadID
  89. } else {
  90. return m.Channel.ChannelID
  91. }
  92. }
  93. func (m *Message) Scan(row dbutil.Scannable) *Message {
  94. var ts int64
  95. var threadID sql.NullString
  96. err := row.Scan(&m.DiscordID, &m.AttachmentID, &m.EditIndex, &m.Channel.ChannelID, &m.Channel.Receiver, &m.SenderID, &ts, &threadID, &m.MXID)
  97. if err != nil {
  98. if !errors.Is(err, sql.ErrNoRows) {
  99. m.log.Errorln("Database scan failed:", err)
  100. panic(err)
  101. }
  102. return nil
  103. }
  104. if ts != 0 {
  105. m.Timestamp = time.UnixMilli(ts)
  106. }
  107. m.ThreadID = threadID.String
  108. return m
  109. }
  110. const messageInsertQuery = `
  111. INSERT INTO message (
  112. dcid, dc_attachment_id, dc_edit_index, dc_chan_id, dc_chan_receiver, dc_sender, timestamp, dc_thread_id, mxid
  113. )
  114. VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
  115. `
  116. var messageMassInsertTemplate = strings.Replace(messageInsertQuery, "($1, $2, $3, $4, $5, $6, $7, $8, $9)", "%s", 1)
  117. type MessagePart struct {
  118. AttachmentID string
  119. MXID id.EventID
  120. }
  121. func (m *Message) MassInsert(msgs []MessagePart) {
  122. if len(msgs) == 0 {
  123. return
  124. }
  125. valueStringFormat := "($1, $%d, $2, $3, $4, $5, $6, $7, $%d)"
  126. if m.db.Dialect == dbutil.SQLite {
  127. valueStringFormat = strings.ReplaceAll(valueStringFormat, "$", "?")
  128. }
  129. params := make([]interface{}, 7+len(msgs)*2)
  130. placeholders := make([]string, len(msgs))
  131. params[0] = m.DiscordID
  132. params[1] = m.EditIndex
  133. params[2] = m.Channel.ChannelID
  134. params[3] = m.Channel.Receiver
  135. params[4] = m.SenderID
  136. params[5] = m.Timestamp.UnixMilli()
  137. params[6] = m.ThreadID
  138. for i, msg := range msgs {
  139. params[7+i*2] = msg.AttachmentID
  140. params[7+i*2+1] = msg.MXID
  141. placeholders[i] = fmt.Sprintf(valueStringFormat, 7+i*2+1, 7+i*2+2)
  142. }
  143. _, err := m.db.Exec(fmt.Sprintf(messageMassInsertTemplate, strings.Join(placeholders, ", ")), params...)
  144. if err != nil {
  145. m.log.Warnfln("Failed to insert %d parts of %s@%s: %v", len(msgs), m.DiscordID, m.Channel, err)
  146. panic(err)
  147. }
  148. }
  149. func (m *Message) Insert() {
  150. _, err := m.db.Exec(messageInsertQuery,
  151. m.DiscordID, m.AttachmentID, m.EditIndex, m.Channel.ChannelID, m.Channel.Receiver, m.SenderID,
  152. m.Timestamp.UnixMilli(), strPtr(m.ThreadID), m.MXID)
  153. if err != nil {
  154. m.log.Warnfln("Failed to insert %s@%s: %v", m.DiscordID, m.Channel, err)
  155. panic(err)
  156. }
  157. }
  158. func (m *Message) Delete() {
  159. query := "DELETE FROM message WHERE dcid=$1 AND dc_chan_id=$2 AND dc_chan_receiver=$3 AND dc_attachment_id=$4"
  160. _, err := m.db.Exec(query, m.DiscordID, m.Channel.ChannelID, m.Channel.Receiver, m.AttachmentID)
  161. if err != nil {
  162. m.log.Warnfln("Failed to delete %q of %s@%s: %v", m.AttachmentID, m.DiscordID, m.Channel, err)
  163. panic(err)
  164. }
  165. }