message_cache.go 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735
  1. package server
  2. import (
  3. "database/sql"
  4. "encoding/json"
  5. "errors"
  6. "fmt"
  7. "net/netip"
  8. "strings"
  9. "time"
  10. _ "github.com/mattn/go-sqlite3" // SQLite driver
  11. "heckel.io/ntfy/log"
  12. "heckel.io/ntfy/util"
  13. )
  14. var (
  15. errUnexpectedMessageType = errors.New("unexpected message type")
  16. )
  17. // Messages cache
  18. const (
  19. createMessagesTableQuery = `
  20. BEGIN;
  21. CREATE TABLE IF NOT EXISTS messages (
  22. id INTEGER PRIMARY KEY AUTOINCREMENT,
  23. mid TEXT NOT NULL,
  24. time INT NOT NULL,
  25. topic TEXT NOT NULL,
  26. message TEXT NOT NULL,
  27. title TEXT NOT NULL,
  28. priority INT NOT NULL,
  29. tags TEXT NOT NULL,
  30. click TEXT NOT NULL,
  31. icon TEXT NOT NULL,
  32. actions TEXT NOT NULL,
  33. attachment_name TEXT NOT NULL,
  34. attachment_type TEXT NOT NULL,
  35. attachment_size INT NOT NULL,
  36. attachment_expires INT NOT NULL,
  37. attachment_url TEXT NOT NULL,
  38. sender TEXT NOT NULL,
  39. user TEXT NOT NULL,
  40. encoding TEXT NOT NULL,
  41. published INT NOT NULL
  42. );
  43. CREATE INDEX IF NOT EXISTS idx_mid ON messages (mid);
  44. CREATE INDEX IF NOT EXISTS idx_time ON messages (time);
  45. CREATE INDEX IF NOT EXISTS idx_topic ON messages (topic);
  46. COMMIT;
  47. `
  48. insertMessageQuery = `
  49. INSERT INTO messages (mid, time, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, sender, user, encoding, published)
  50. VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
  51. `
  52. pruneMessagesQuery = `DELETE FROM messages WHERE time < ? AND published = 1`
  53. selectRowIDFromMessageID = `SELECT id FROM messages WHERE mid = ?` // Do not include topic, see #336 and TestServer_PollSinceID_MultipleTopics
  54. selectMessagesSinceTimeQuery = `
  55. SELECT mid, time, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, sender, user, encoding
  56. FROM messages
  57. WHERE topic = ? AND time >= ? AND published = 1
  58. ORDER BY time, id
  59. `
  60. selectMessagesSinceTimeIncludeScheduledQuery = `
  61. SELECT mid, time, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, sender, user, encoding
  62. FROM messages
  63. WHERE topic = ? AND time >= ?
  64. ORDER BY time, id
  65. `
  66. selectMessagesSinceIDQuery = `
  67. SELECT mid, time, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, sender, user, encoding
  68. FROM messages
  69. WHERE topic = ? AND id > ? AND published = 1
  70. ORDER BY time, id
  71. `
  72. selectMessagesSinceIDIncludeScheduledQuery = `
  73. SELECT mid, time, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, sender, user, encoding
  74. FROM messages
  75. WHERE topic = ? AND (id > ? OR published = 0)
  76. ORDER BY time, id
  77. `
  78. selectMessagesDueQuery = `
  79. SELECT mid, time, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, sender, user, encoding
  80. FROM messages
  81. WHERE time <= ? AND published = 0
  82. ORDER BY time, id
  83. `
  84. updateMessagePublishedQuery = `UPDATE messages SET published = 1 WHERE mid = ?`
  85. selectMessagesCountQuery = `SELECT COUNT(*) FROM messages`
  86. selectMessageCountPerTopicQuery = `SELECT topic, COUNT(*) FROM messages GROUP BY topic`
  87. selectTopicsQuery = `SELECT topic FROM messages GROUP BY topic`
  88. selectAttachmentsSizeBySenderQuery = `SELECT IFNULL(SUM(attachment_size), 0) FROM messages WHERE sender = ? AND attachment_expires >= ?`
  89. selectAttachmentsSizeByUserQuery = `SELECT IFNULL(SUM(attachment_size), 0) FROM messages WHERE user = ? AND attachment_expires >= ?`
  90. )
  91. // Schema management queries
  92. const (
  93. currentSchemaVersion = 9
  94. createSchemaVersionTableQuery = `
  95. CREATE TABLE IF NOT EXISTS schemaVersion (
  96. id INT PRIMARY KEY,
  97. version INT NOT NULL
  98. );
  99. `
  100. insertSchemaVersion = `INSERT INTO schemaVersion VALUES (1, ?)`
  101. updateSchemaVersion = `UPDATE schemaVersion SET version = ? WHERE id = 1`
  102. selectSchemaVersionQuery = `SELECT version FROM schemaVersion WHERE id = 1`
  103. // 0 -> 1
  104. migrate0To1AlterMessagesTableQuery = `
  105. BEGIN;
  106. ALTER TABLE messages ADD COLUMN title TEXT NOT NULL DEFAULT('');
  107. ALTER TABLE messages ADD COLUMN priority INT NOT NULL DEFAULT(0);
  108. ALTER TABLE messages ADD COLUMN tags TEXT NOT NULL DEFAULT('');
  109. COMMIT;
  110. `
  111. // 1 -> 2
  112. migrate1To2AlterMessagesTableQuery = `
  113. ALTER TABLE messages ADD COLUMN published INT NOT NULL DEFAULT(1);
  114. `
  115. // 2 -> 3
  116. migrate2To3AlterMessagesTableQuery = `
  117. BEGIN;
  118. ALTER TABLE messages ADD COLUMN click TEXT NOT NULL DEFAULT('');
  119. ALTER TABLE messages ADD COLUMN attachment_name TEXT NOT NULL DEFAULT('');
  120. ALTER TABLE messages ADD COLUMN attachment_type TEXT NOT NULL DEFAULT('');
  121. ALTER TABLE messages ADD COLUMN attachment_size INT NOT NULL DEFAULT('0');
  122. ALTER TABLE messages ADD COLUMN attachment_expires INT NOT NULL DEFAULT('0');
  123. ALTER TABLE messages ADD COLUMN attachment_owner TEXT NOT NULL DEFAULT('');
  124. ALTER TABLE messages ADD COLUMN attachment_url TEXT NOT NULL DEFAULT('');
  125. COMMIT;
  126. `
  127. // 3 -> 4
  128. migrate3To4AlterMessagesTableQuery = `
  129. ALTER TABLE messages ADD COLUMN encoding TEXT NOT NULL DEFAULT('');
  130. `
  131. // 4 -> 5
  132. migrate4To5AlterMessagesTableQuery = `
  133. BEGIN;
  134. CREATE TABLE IF NOT EXISTS messages_new (
  135. id INTEGER PRIMARY KEY AUTOINCREMENT,
  136. mid TEXT NOT NULL,
  137. time INT NOT NULL,
  138. topic TEXT NOT NULL,
  139. message TEXT NOT NULL,
  140. title TEXT NOT NULL,
  141. priority INT NOT NULL,
  142. tags TEXT NOT NULL,
  143. click TEXT NOT NULL,
  144. attachment_name TEXT NOT NULL,
  145. attachment_type TEXT NOT NULL,
  146. attachment_size INT NOT NULL,
  147. attachment_expires INT NOT NULL,
  148. attachment_url TEXT NOT NULL,
  149. attachment_owner TEXT NOT NULL,
  150. encoding TEXT NOT NULL,
  151. published INT NOT NULL
  152. );
  153. CREATE INDEX IF NOT EXISTS idx_mid ON messages_new (mid);
  154. CREATE INDEX IF NOT EXISTS idx_topic ON messages_new (topic);
  155. INSERT
  156. INTO messages_new (
  157. mid, time, topic, message, title, priority, tags, click, attachment_name, attachment_type,
  158. attachment_size, attachment_expires, attachment_url, attachment_owner, encoding, published)
  159. SELECT
  160. id, time, topic, message, title, priority, tags, click, attachment_name, attachment_type,
  161. attachment_size, attachment_expires, attachment_url, attachment_owner, encoding, published
  162. FROM messages;
  163. DROP TABLE messages;
  164. ALTER TABLE messages_new RENAME TO messages;
  165. COMMIT;
  166. `
  167. // 5 -> 6
  168. migrate5To6AlterMessagesTableQuery = `
  169. ALTER TABLE messages ADD COLUMN actions TEXT NOT NULL DEFAULT('');
  170. `
  171. // 6 -> 7
  172. migrate6To7AlterMessagesTableQuery = `
  173. ALTER TABLE messages RENAME COLUMN attachment_owner TO sender;
  174. `
  175. // 7 -> 8
  176. migrate7To8AlterMessagesTableQuery = `
  177. ALTER TABLE messages ADD COLUMN icon TEXT NOT NULL DEFAULT('');
  178. `
  179. // 8 -> 9
  180. migrate8To9AlterMessagesTableQuery = `
  181. CREATE INDEX IF NOT EXISTS idx_time ON messages (time);
  182. `
  183. )
  184. type messageCache struct {
  185. db *sql.DB
  186. queue *util.BatchingQueue[*message]
  187. nop bool
  188. }
  189. // newSqliteCache creates a SQLite file-backed cache
  190. func newSqliteCache(filename, startupQueries string, batchSize int, batchTimeout time.Duration, nop bool) (*messageCache, error) {
  191. db, err := sql.Open("sqlite3", filename)
  192. if err != nil {
  193. return nil, err
  194. }
  195. if err := setupCacheDB(db, startupQueries); err != nil {
  196. return nil, err
  197. }
  198. var queue *util.BatchingQueue[*message]
  199. if batchSize > 0 || batchTimeout > 0 {
  200. queue = util.NewBatchingQueue[*message](batchSize, batchTimeout)
  201. }
  202. cache := &messageCache{
  203. db: db,
  204. queue: queue,
  205. nop: nop,
  206. }
  207. go cache.processMessageBatches()
  208. return cache, nil
  209. }
  210. // newMemCache creates an in-memory cache
  211. func newMemCache() (*messageCache, error) {
  212. return newSqliteCache(createMemoryFilename(), "", 0, 0, false)
  213. }
  214. // newNopCache creates an in-memory cache that discards all messages;
  215. // it is always empty and can be used if caching is entirely disabled
  216. func newNopCache() (*messageCache, error) {
  217. return newSqliteCache(createMemoryFilename(), "", 0, 0, true)
  218. }
  219. // createMemoryFilename creates a unique memory filename to use for the SQLite backend.
  220. // From mattn/go-sqlite3: "Each connection to ":memory:" opens a brand new in-memory
  221. // sql database, so if the stdlib's sql engine happens to open another connection and
  222. // you've only specified ":memory:", that connection will see a brand new database.
  223. // A workaround is to use "file::memory:?cache=shared" (or "file:foobar?mode=memory&cache=shared").
  224. // Every connection to this string will point to the same in-memory database."
  225. func createMemoryFilename() string {
  226. return fmt.Sprintf("file:%s?mode=memory&cache=shared", util.RandomString(10))
  227. }
  228. // AddMessage stores a message to the message cache synchronously, or queues it to be stored at a later date asyncronously.
  229. // The message is queued only if "batchSize" or "batchTimeout" are passed to the constructor.
  230. func (c *messageCache) AddMessage(m *message) error {
  231. if c.queue != nil {
  232. c.queue.Enqueue(m)
  233. return nil
  234. }
  235. return c.addMessages([]*message{m})
  236. }
  237. // addMessages synchronously stores a match of messages. If the database is locked, the transaction waits until
  238. // SQLite's busy_timeout is exceeded before erroring out.
  239. func (c *messageCache) addMessages(ms []*message) error {
  240. if c.nop {
  241. return nil
  242. }
  243. if len(ms) == 0 {
  244. return nil
  245. }
  246. start := time.Now()
  247. tx, err := c.db.Begin()
  248. if err != nil {
  249. return err
  250. }
  251. defer tx.Rollback()
  252. stmt, err := tx.Prepare(insertMessageQuery)
  253. if err != nil {
  254. return err
  255. }
  256. defer stmt.Close()
  257. for _, m := range ms {
  258. if m.Event != messageEvent {
  259. return errUnexpectedMessageType
  260. }
  261. published := m.Time <= time.Now().Unix()
  262. tags := strings.Join(m.Tags, ",")
  263. var attachmentName, attachmentType, attachmentURL string
  264. var attachmentSize, attachmentExpires int64
  265. if m.Attachment != nil {
  266. attachmentName = m.Attachment.Name
  267. attachmentType = m.Attachment.Type
  268. attachmentSize = m.Attachment.Size
  269. attachmentExpires = m.Attachment.Expires
  270. attachmentURL = m.Attachment.URL
  271. }
  272. var actionsStr string
  273. if len(m.Actions) > 0 {
  274. actionsBytes, err := json.Marshal(m.Actions)
  275. if err != nil {
  276. return err
  277. }
  278. actionsStr = string(actionsBytes)
  279. }
  280. var sender string
  281. if m.Sender.IsValid() {
  282. sender = m.Sender.String()
  283. }
  284. _, err := stmt.Exec(
  285. m.ID,
  286. m.Time,
  287. m.Topic,
  288. m.Message,
  289. m.Title,
  290. m.Priority,
  291. tags,
  292. m.Click,
  293. m.Icon,
  294. actionsStr,
  295. attachmentName,
  296. attachmentType,
  297. attachmentSize,
  298. attachmentExpires,
  299. attachmentURL,
  300. sender,
  301. m.User,
  302. m.Encoding,
  303. published,
  304. )
  305. if err != nil {
  306. return err
  307. }
  308. }
  309. if err := tx.Commit(); err != nil {
  310. log.Error("Cache: Writing %d message(s) failed (took %v)", len(ms), time.Since(start))
  311. return err
  312. }
  313. log.Debug("Cache: Wrote %d message(s) in %v", len(ms), time.Since(start))
  314. return nil
  315. }
  316. func (c *messageCache) Messages(topic string, since sinceMarker, scheduled bool) ([]*message, error) {
  317. if since.IsNone() {
  318. return make([]*message, 0), nil
  319. } else if since.IsID() {
  320. return c.messagesSinceID(topic, since, scheduled)
  321. }
  322. return c.messagesSinceTime(topic, since, scheduled)
  323. }
  324. func (c *messageCache) messagesSinceTime(topic string, since sinceMarker, scheduled bool) ([]*message, error) {
  325. var rows *sql.Rows
  326. var err error
  327. if scheduled {
  328. rows, err = c.db.Query(selectMessagesSinceTimeIncludeScheduledQuery, topic, since.Time().Unix())
  329. } else {
  330. rows, err = c.db.Query(selectMessagesSinceTimeQuery, topic, since.Time().Unix())
  331. }
  332. if err != nil {
  333. return nil, err
  334. }
  335. return readMessages(rows)
  336. }
  337. func (c *messageCache) messagesSinceID(topic string, since sinceMarker, scheduled bool) ([]*message, error) {
  338. idrows, err := c.db.Query(selectRowIDFromMessageID, since.ID())
  339. if err != nil {
  340. return nil, err
  341. }
  342. defer idrows.Close()
  343. if !idrows.Next() {
  344. return c.messagesSinceTime(topic, sinceAllMessages, scheduled)
  345. }
  346. var rowID int64
  347. if err := idrows.Scan(&rowID); err != nil {
  348. return nil, err
  349. }
  350. idrows.Close()
  351. var rows *sql.Rows
  352. if scheduled {
  353. rows, err = c.db.Query(selectMessagesSinceIDIncludeScheduledQuery, topic, rowID)
  354. } else {
  355. rows, err = c.db.Query(selectMessagesSinceIDQuery, topic, rowID)
  356. }
  357. if err != nil {
  358. return nil, err
  359. }
  360. return readMessages(rows)
  361. }
  362. func (c *messageCache) MessagesDue() ([]*message, error) {
  363. rows, err := c.db.Query(selectMessagesDueQuery, time.Now().Unix())
  364. if err != nil {
  365. return nil, err
  366. }
  367. return readMessages(rows)
  368. }
  369. func (c *messageCache) MarkPublished(m *message) error {
  370. _, err := c.db.Exec(updateMessagePublishedQuery, m.ID)
  371. return err
  372. }
  373. func (c *messageCache) MessageCounts() (map[string]int, error) {
  374. rows, err := c.db.Query(selectMessageCountPerTopicQuery)
  375. if err != nil {
  376. return nil, err
  377. }
  378. defer rows.Close()
  379. var topic string
  380. var count int
  381. counts := make(map[string]int)
  382. for rows.Next() {
  383. if err := rows.Scan(&topic, &count); err != nil {
  384. return nil, err
  385. } else if err := rows.Err(); err != nil {
  386. return nil, err
  387. }
  388. counts[topic] = count
  389. }
  390. return counts, nil
  391. }
  392. func (c *messageCache) Topics() (map[string]*topic, error) {
  393. rows, err := c.db.Query(selectTopicsQuery)
  394. if err != nil {
  395. return nil, err
  396. }
  397. defer rows.Close()
  398. topics := make(map[string]*topic)
  399. for rows.Next() {
  400. var id string
  401. if err := rows.Scan(&id); err != nil {
  402. return nil, err
  403. }
  404. topics[id] = newTopic(id)
  405. }
  406. if err := rows.Err(); err != nil {
  407. return nil, err
  408. }
  409. return topics, nil
  410. }
  411. func (c *messageCache) Prune(olderThan time.Time) error {
  412. start := time.Now()
  413. if _, err := c.db.Exec(pruneMessagesQuery, olderThan.Unix()); err != nil {
  414. log.Warn("Cache: Pruning failed (after %v): %s", time.Since(start), err.Error())
  415. }
  416. log.Debug("Cache: Pruning successful (took %v)", time.Since(start))
  417. return nil
  418. }
  419. func (c *messageCache) AttachmentBytesUsedBySender(sender string) (int64, error) {
  420. rows, err := c.db.Query(selectAttachmentsSizeBySenderQuery, sender, time.Now().Unix())
  421. if err != nil {
  422. return 0, err
  423. }
  424. return c.readAttachmentBytesUsed(rows)
  425. }
  426. func (c *messageCache) AttachmentBytesUsedByUser(user string) (int64, error) {
  427. rows, err := c.db.Query(selectAttachmentsSizeByUserQuery, user, time.Now().Unix())
  428. if err != nil {
  429. return 0, err
  430. }
  431. return c.readAttachmentBytesUsed(rows)
  432. }
  433. func (c *messageCache) readAttachmentBytesUsed(rows *sql.Rows) (int64, error) {
  434. defer rows.Close()
  435. var size int64
  436. if !rows.Next() {
  437. return 0, errors.New("no rows found")
  438. }
  439. if err := rows.Scan(&size); err != nil {
  440. return 0, err
  441. } else if err := rows.Err(); err != nil {
  442. return 0, err
  443. }
  444. return size, nil
  445. }
  446. func (c *messageCache) processMessageBatches() {
  447. if c.queue == nil {
  448. return
  449. }
  450. for messages := range c.queue.Dequeue() {
  451. if err := c.addMessages(messages); err != nil {
  452. log.Error("Cache: %s", err.Error())
  453. }
  454. }
  455. }
  456. func readMessages(rows *sql.Rows) ([]*message, error) {
  457. defer rows.Close()
  458. messages := make([]*message, 0)
  459. for rows.Next() {
  460. var timestamp, attachmentSize, attachmentExpires int64
  461. var priority int
  462. var id, topic, msg, title, tagsStr, click, icon, actionsStr, attachmentName, attachmentType, attachmentURL, sender, user, encoding string
  463. err := rows.Scan(
  464. &id,
  465. &timestamp,
  466. &topic,
  467. &msg,
  468. &title,
  469. &priority,
  470. &tagsStr,
  471. &click,
  472. &icon,
  473. &actionsStr,
  474. &attachmentName,
  475. &attachmentType,
  476. &attachmentSize,
  477. &attachmentExpires,
  478. &attachmentURL,
  479. &sender,
  480. &user,
  481. &encoding,
  482. )
  483. if err != nil {
  484. return nil, err
  485. }
  486. var tags []string
  487. if tagsStr != "" {
  488. tags = strings.Split(tagsStr, ",")
  489. }
  490. var actions []*action
  491. if actionsStr != "" {
  492. if err := json.Unmarshal([]byte(actionsStr), &actions); err != nil {
  493. return nil, err
  494. }
  495. }
  496. senderIP, err := netip.ParseAddr(sender)
  497. if err != nil {
  498. senderIP = netip.Addr{} // if no IP stored in database, return invalid address
  499. }
  500. var att *attachment
  501. if attachmentName != "" && attachmentURL != "" {
  502. att = &attachment{
  503. Name: attachmentName,
  504. Type: attachmentType,
  505. Size: attachmentSize,
  506. Expires: attachmentExpires,
  507. URL: attachmentURL,
  508. }
  509. }
  510. messages = append(messages, &message{
  511. ID: id,
  512. Time: timestamp,
  513. Event: messageEvent,
  514. Topic: topic,
  515. Message: msg,
  516. Title: title,
  517. Priority: priority,
  518. Tags: tags,
  519. Click: click,
  520. Icon: icon,
  521. Actions: actions,
  522. Attachment: att,
  523. Sender: senderIP, // Must parse assuming database must be correct
  524. User: user,
  525. Encoding: encoding,
  526. })
  527. }
  528. if err := rows.Err(); err != nil {
  529. return nil, err
  530. }
  531. return messages, nil
  532. }
  533. func setupCacheDB(db *sql.DB, startupQueries string) error {
  534. // Run startup queries
  535. if startupQueries != "" {
  536. if _, err := db.Exec(startupQueries); err != nil {
  537. return err
  538. }
  539. }
  540. // If 'messages' table does not exist, this must be a new database
  541. rowsMC, err := db.Query(selectMessagesCountQuery)
  542. if err != nil {
  543. return setupNewCacheDB(db)
  544. }
  545. rowsMC.Close()
  546. // If 'messages' table exists, check 'schemaVersion' table
  547. schemaVersion := 0
  548. rowsSV, err := db.Query(selectSchemaVersionQuery)
  549. if err == nil {
  550. defer rowsSV.Close()
  551. if !rowsSV.Next() {
  552. return errors.New("cannot determine schema version: cache file may be corrupt")
  553. }
  554. if err := rowsSV.Scan(&schemaVersion); err != nil {
  555. return err
  556. }
  557. rowsSV.Close()
  558. }
  559. // Do migrations
  560. if schemaVersion == currentSchemaVersion {
  561. return nil
  562. } else if schemaVersion == 0 {
  563. return migrateFrom0(db)
  564. } else if schemaVersion == 1 {
  565. return migrateFrom1(db)
  566. } else if schemaVersion == 2 {
  567. return migrateFrom2(db)
  568. } else if schemaVersion == 3 {
  569. return migrateFrom3(db)
  570. } else if schemaVersion == 4 {
  571. return migrateFrom4(db)
  572. } else if schemaVersion == 5 {
  573. return migrateFrom5(db)
  574. } else if schemaVersion == 6 {
  575. return migrateFrom6(db)
  576. } else if schemaVersion == 7 {
  577. return migrateFrom7(db)
  578. } else if schemaVersion == 8 {
  579. return migrateFrom8(db)
  580. }
  581. // TODO add user column
  582. return fmt.Errorf("unexpected schema version found: %d", schemaVersion)
  583. }
  584. func setupNewCacheDB(db *sql.DB) error {
  585. if _, err := db.Exec(createMessagesTableQuery); err != nil {
  586. return err
  587. }
  588. if _, err := db.Exec(createSchemaVersionTableQuery); err != nil {
  589. return err
  590. }
  591. if _, err := db.Exec(insertSchemaVersion, currentSchemaVersion); err != nil {
  592. return err
  593. }
  594. return nil
  595. }
  596. func migrateFrom0(db *sql.DB) error {
  597. log.Info("Migrating cache database schema: from 0 to 1")
  598. if _, err := db.Exec(migrate0To1AlterMessagesTableQuery); err != nil {
  599. return err
  600. }
  601. if _, err := db.Exec(createSchemaVersionTableQuery); err != nil {
  602. return err
  603. }
  604. if _, err := db.Exec(insertSchemaVersion, 1); err != nil {
  605. return err
  606. }
  607. return migrateFrom1(db)
  608. }
  609. func migrateFrom1(db *sql.DB) error {
  610. log.Info("Migrating cache database schema: from 1 to 2")
  611. if _, err := db.Exec(migrate1To2AlterMessagesTableQuery); err != nil {
  612. return err
  613. }
  614. if _, err := db.Exec(updateSchemaVersion, 2); err != nil {
  615. return err
  616. }
  617. return migrateFrom2(db)
  618. }
  619. func migrateFrom2(db *sql.DB) error {
  620. log.Info("Migrating cache database schema: from 2 to 3")
  621. if _, err := db.Exec(migrate2To3AlterMessagesTableQuery); err != nil {
  622. return err
  623. }
  624. if _, err := db.Exec(updateSchemaVersion, 3); err != nil {
  625. return err
  626. }
  627. return migrateFrom3(db)
  628. }
  629. func migrateFrom3(db *sql.DB) error {
  630. log.Info("Migrating cache database schema: from 3 to 4")
  631. if _, err := db.Exec(migrate3To4AlterMessagesTableQuery); err != nil {
  632. return err
  633. }
  634. if _, err := db.Exec(updateSchemaVersion, 4); err != nil {
  635. return err
  636. }
  637. return migrateFrom4(db)
  638. }
  639. func migrateFrom4(db *sql.DB) error {
  640. log.Info("Migrating cache database schema: from 4 to 5")
  641. if _, err := db.Exec(migrate4To5AlterMessagesTableQuery); err != nil {
  642. return err
  643. }
  644. if _, err := db.Exec(updateSchemaVersion, 5); err != nil {
  645. return err
  646. }
  647. return migrateFrom5(db)
  648. }
  649. func migrateFrom5(db *sql.DB) error {
  650. log.Info("Migrating cache database schema: from 5 to 6")
  651. if _, err := db.Exec(migrate5To6AlterMessagesTableQuery); err != nil {
  652. return err
  653. }
  654. if _, err := db.Exec(updateSchemaVersion, 6); err != nil {
  655. return err
  656. }
  657. return migrateFrom6(db)
  658. }
  659. func migrateFrom6(db *sql.DB) error {
  660. log.Info("Migrating cache database schema: from 6 to 7")
  661. if _, err := db.Exec(migrate6To7AlterMessagesTableQuery); err != nil {
  662. return err
  663. }
  664. if _, err := db.Exec(updateSchemaVersion, 7); err != nil {
  665. return err
  666. }
  667. return migrateFrom7(db)
  668. }
  669. func migrateFrom7(db *sql.DB) error {
  670. log.Info("Migrating cache database schema: from 7 to 8")
  671. if _, err := db.Exec(migrate7To8AlterMessagesTableQuery); err != nil {
  672. return err
  673. }
  674. if _, err := db.Exec(updateSchemaVersion, 8); err != nil {
  675. return err
  676. }
  677. return migrateFrom8(db)
  678. }
  679. func migrateFrom8(db *sql.DB) error {
  680. log.Info("Migrating cache database schema: from 8 to 9")
  681. if _, err := db.Exec(migrate8To9AlterMessagesTableQuery); err != nil {
  682. return err
  683. }
  684. if _, err := db.Exec(updateSchemaVersion, 9); err != nil {
  685. return err
  686. }
  687. return nil // Update this when a new version is added
  688. }