message.go 5.8 KB

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