message.go 4.8 KB

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