internal/db/message.go (view raw)
1package db
2
3import (
4 "context"
5 "database/sql"
6 "encoding/json"
7 "errors"
8 "fmt"
9 "log/slog"
10 "net/mail"
11 "time"
12
13 "postern/internal/model"
14
15 "github.com/emersion/go-imap/v2"
16)
17
18func (db *DB) GetAllMailboxMessages(ctx context.Context, userID, mailboxID int) ([]model.Message, error) {
19 query := `
20SELECT
21 mm.uid,
22 mm.modseq,
23 m.size,
24 m.blob_address,
25 mm.internal_date,
26 m.date,
27 m.subject,
28 m.message_id,
29 m.in_reply_to,
30 m.from_addr,
31 m.sender,
32 m.reply_to,
33 m.to_addr,
34 m.cc,
35 m.bcc,
36 coalesce(json_group_array(mf.flag), '[]') AS flags
37FROM (
38 SELECT
39 uid,
40 message_id,
41 modseq,
42 internal_date
43 FROM mailbox_messages
44 JOIN mailboxes ON mailboxes.id = mailbox_messages.mailbox_id
45 WHERE mailbox_messages.mailbox_id = ?
46 AND mailboxes.user_id = ?
47 AND mailbox_messages.expunged_modseq IS NULL
48) numbered
49JOIN mailbox_messages mm ON mm.mailbox_id = ? AND mm.uid = numbered.uid
50JOIN messages m ON m.id = numbered.message_id
51LEFT JOIN message_flags mf ON mf.mailbox_id = ? AND mf.uid = numbered.uid
52GROUP BY mm.uid
53ORDER BY mm.uid ASC
54`
55
56 rows, err := db.read.QueryContext(ctx, query, mailboxID, userID, mailboxID, mailboxID)
57 if err != nil {
58 return nil, err
59 }
60 defer func(rows *sql.Rows) {
61 err := rows.Close()
62 if err != nil {
63 slog.Error("closing rows", "error", err)
64 }
65 }(rows)
66
67 out := make([]model.Message, 0)
68 for rows.Next() {
69 var m model.Message
70 var flagsJSON []byte
71 err := rows.Scan(
72 &m.UID,
73 &m.ModSeq,
74 &m.RFC822Size,
75 &m.BlobHash,
76 &m.InternalDate,
77 &m.EnvelopeDate,
78 &m.EnvelopeSubject,
79 &m.EnvelopeMessageID,
80 &m.EnvelopeInReplyTo,
81 &m.EnvelopeFrom,
82 &m.EnvelopeSender,
83 &m.EnvelopeReplyTo,
84 &m.EnvelopeTo,
85 &m.EnvelopeCc,
86 &m.EnvelopeBcc,
87 &flagsJSON,
88 )
89 if err != nil {
90 return nil, fmt.Errorf("scanning message: %w", err)
91 }
92 m.ServerSeq = func(ctx context.Context) (uint32, error) {
93 serverSeq, err := db.UIDToServerSeq(ctx, mailboxID, m.UID)
94 if err != nil {
95 return 0, err
96 }
97 return serverSeq, nil
98 }
99 if len(flagsJSON) > 0 {
100 if err := json.Unmarshal(flagsJSON, &m.Flags); err != nil {
101 return nil, fmt.Errorf("unmarshaling flags: %w", err)
102 }
103 } else {
104 m.Flags = []string{}
105 }
106 out = append(out, m)
107 }
108
109 return out, rows.Err()
110}
111
112func (db *DB) GetMailboxMessagesByUID(ctx context.Context, userID, mailboxID int, uids []uint32) ([]model.Message, error) {
113 allMessages, err := db.GetAllMailboxMessages(ctx, userID, mailboxID)
114 if err != nil {
115 return nil, err
116 }
117
118 uidSet := make(map[uint32]bool, len(uids))
119 for _, uid := range uids {
120 uidSet[uid] = true
121 }
122
123 out := make([]model.Message, 0, len(uids))
124 for _, m := range allMessages {
125 if uidSet[m.UID] {
126 out = append(out, m)
127 }
128 }
129
130 return out, nil
131}
132
133func (db *DB) AppendMessage(ctx context.Context, mailboxID, userID int,
134 blobHash string, size int64, internalDate string, flags []string, msgHeader mail.Header) (uint32, error) {
135
136 tx, err := db.write.BeginTx(ctx, nil)
137 if err != nil {
138 return 0, err
139 }
140 defer txRollback(tx)
141
142 // 1. Allocate a UID and bump modseq, verifying mailbox ownership in one shot.
143 // RETURNING reflects post-update values, so uidnext-1 is the UID to assign.
144 var assignedUID uint32
145 var newModSeq int64
146 err = tx.QueryRowContext(ctx, `
147 UPDATE mailboxes
148 SET uidnext = uidnext + 1,
149 highest_modseq = highest_modseq + 1
150 WHERE id = ? AND user_id = ?
151 RETURNING uidnext - 1, highest_modseq
152 `, mailboxID, userID).Scan(&assignedUID, &newModSeq)
153 if errors.Is(err, sql.ErrNoRows) {
154 return 0, fmt.Errorf("mailbox %d not found for user %d", mailboxID, userID)
155 }
156 if err != nil {
157 return 0, fmt.Errorf("allocating uid: %w", err)
158 }
159
160 // 2. Insert the message, or reuse the existing row for identical content.
161 // The no-op DO UPDATE forces RETURNING to yield the existing id on conflict.
162 var internalMessageID int64
163 var subject, messageID, inReplyTo, fromAddr, sender, replyTo, toAddr, cc, bcc sql.NullString
164 var datestring sql.NullString
165 if msgHeader != nil {
166 date, err := msgHeader.Date()
167 if err == nil && !date.IsZero() {
168 datestring = sql.NullString{
169 Valid: true,
170 String: date.UTC().Format(time.RFC3339),
171 }
172 }
173
174 for headerKey, headerValue := range msgHeader {
175 switch headerKey {
176 case "Subject":
177 subject = sql.NullString{
178 Valid: true,
179 String: headerValue[0],
180 }
181 case "Message-Id":
182 messageID = sql.NullString{
183 Valid: true,
184 String: headerValue[0],
185 }
186 case "In-Reply-To":
187 inReplyTo = sql.NullString{
188 Valid: true,
189 String: headerValue[0],
190 }
191 case "From":
192 fromAddr = sql.NullString{
193 Valid: true,
194 String: headerValue[0],
195 }
196 case "Sender":
197 sender = sql.NullString{
198 Valid: true,
199 String: headerValue[0],
200 }
201 case "Reply-To":
202 replyTo = sql.NullString{
203 Valid: true,
204 String: headerValue[0],
205 }
206 case "To":
207 toAddr = sql.NullString{
208 Valid: true,
209 String: headerValue[0],
210 }
211 case "Cc":
212 cc = sql.NullString{
213 Valid: true,
214 String: headerValue[0],
215 }
216 case "Bcc":
217 bcc = sql.NullString{
218 Valid: true,
219 String: headerValue[0],
220 }
221 }
222 }
223 }
224
225 err = tx.QueryRowContext(ctx, `
226 INSERT INTO messages (blob_address, size, subject, message_id, in_reply_to, date, from_addr, sender, reply_to, to_addr, cc, bcc)
227 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
228 ON CONFLICT(blob_address) DO UPDATE SET blob_address = blob_address
229 RETURNING id
230 `, blobHash, size, subject, messageID, inReplyTo, datestring, fromAddr, sender, replyTo, toAddr, cc, bcc).Scan(&internalMessageID)
231 if err != nil {
232 return 0, fmt.Errorf("inserting message: %w", err)
233 }
234
235 // 3. Link the message into the mailbox at the allocated UID.
236 _, err = tx.ExecContext(ctx, `
237 INSERT INTO mailbox_messages (mailbox_id, uid, message_id, modseq, internal_date)
238 VALUES (?, ?, ?, ?, ?)
239 `, mailboxID, assignedUID, internalMessageID, newModSeq, internalDate)
240 if err != nil {
241 return 0, fmt.Errorf("linking message: %w", err)
242 }
243
244 // 4. Store flags for this mailbox/uid.
245 for _, flag := range flags {
246 _, err = tx.ExecContext(ctx, `
247 INSERT INTO message_flags (mailbox_id, uid, flag)
248 VALUES (?, ?, ?)
249 ON CONFLICT DO NOTHING
250 `, mailboxID, assignedUID, flag)
251 if err != nil {
252 return 0, fmt.Errorf("inserting flag %q: %w", flag, err)
253 }
254 }
255
256 if err := tx.Commit(); err != nil {
257 return 0, fmt.Errorf("committing append: %w", err)
258 }
259 return assignedUID, nil
260}
261
262// SetMessageFlags sets flags for a message and returns those flags.
263// Returning flags is for consistency with addMessageFlags and deleteMessageFlags.
264func (db *DB) SetMessageFlags(ctx context.Context, mailboxID int, messageID uint32, flags []string) ([]string, error) {
265 tx, err := db.write.BeginTx(ctx, nil)
266 if err != nil {
267 return nil, err
268 }
269 defer txRollback(tx)
270
271 _, err = tx.ExecContext(ctx, `DELETE FROM message_flags WHERE mailbox_id = ? and uid = ?`, mailboxID, messageID)
272 if err != nil {
273 return nil, err
274 }
275
276 for _, flag := range flags {
277 _, err = tx.ExecContext(ctx, `INSERT INTO message_flags (mailbox_id, uid, flag) VALUES (?, ?, ?)`, mailboxID, messageID, flag)
278 if err != nil {
279 return nil, err
280 }
281 }
282 return flags, tx.Commit()
283}
284
285func (db *DB) AddMessageFlags(ctx context.Context, mailboxID int, messageID uint32, flags []string) ([]string, error) {
286 tx, err := db.write.BeginTx(ctx, nil)
287 if err != nil {
288 return nil, err
289 }
290 defer txRollback(tx)
291
292 for _, flag := range flags {
293 _, err = tx.ExecContext(ctx, `INSERT INTO message_flags (mailbox_id, uid, flag) VALUES (?, ?, ?) ON CONFLICT DO NOTHING`, mailboxID, messageID, flag)
294 if err != nil {
295 return nil, err
296 }
297 }
298
299 flagSet, err := getMessageFlags(ctx, tx, mailboxID, messageID)
300 if err != nil {
301 return nil, err
302 }
303
304 return flagSet, tx.Commit()
305}
306
307func (db *DB) DeleteMessageFlags(ctx context.Context, mailboxID int, messageID uint32, flags []string) ([]string, error) {
308 tx, err := db.write.BeginTx(ctx, nil)
309 if err != nil {
310 return nil, err
311 }
312
313 for _, flag := range flags {
314 _, err = tx.ExecContext(ctx, `DELETE FROM message_flags WHERE mailbox_id = ? AND uid = ? AND flag = ?`, mailboxID, messageID, flag)
315 if err != nil {
316 return nil, errors.Join(err, tx.Rollback())
317 }
318 }
319
320 flagSet, err := getMessageFlags(ctx, tx, mailboxID, messageID)
321 if err != nil {
322 return nil, errors.Join(err, tx.Rollback())
323 }
324
325 return flagSet, tx.Commit()
326}
327
328// getMessageFlags is a helper for addMessageFlags and deleteMessageFlags
329func getMessageFlags(ctx context.Context, tx *sql.Tx, mailboxID int, messageID uint32) ([]string, error) {
330 var flags []string
331 rows, err := tx.QueryContext(ctx, `SELECT flag FROM message_flags WHERE mailbox_id = ? AND uid = ?`, mailboxID, messageID)
332 if err != nil {
333 return nil, err
334 }
335 defer func(rows *sql.Rows) {
336 _ = rows.Close()
337 }(rows)
338
339 for rows.Next() {
340 var flag string
341 if err := rows.Scan(&flag); err != nil {
342 return nil, err
343 }
344 flags = append(flags, flag)
345 }
346
347 return flags, rows.Err()
348}
349
350func (db *DB) CopyMessagesToMailbox(ctx context.Context, fromMailboxID, toMailboxID int, uids []uint32) ([]uint32, error) {
351 tx, err := db.write.BeginTx(ctx, nil)
352 if err != nil {
353 return nil, err
354 }
355 defer txRollback(tx)
356
357 destUIDs := make([]uint32, len(uids))
358 for i, uid := range uids {
359 nextUID, err := getAndIncreaseNextUID(ctx, tx, toMailboxID)
360 if err != nil {
361 return nil, err
362 }
363 destUIDs[i] = nextUID
364
365 _, err = tx.ExecContext(
366 ctx,
367 `
368INSERT INTO mailbox_messages (mailbox_id, uid, message_id, modseq, internal_date, expunged_modseq)
369SELECT ?, ?, message_id, modseq, internal_date, expunged_modseq
370FROM mailbox_messages
371WHERE mailbox_id = ? AND uid = ?`,
372 toMailboxID, nextUID, fromMailboxID, uid)
373 if err != nil {
374 return nil, err
375 }
376
377 _, err = tx.ExecContext(
378 ctx,
379 `
380INSERT INTO message_flags (mailbox_id, uid, flag)
381SELECT ?, ?, flag
382FROM message_flags
383WHERE mailbox_id = ? AND uid = ?`,
384 toMailboxID, nextUID, fromMailboxID, uid)
385 if err != nil {
386 return nil, err
387 }
388 }
389
390 return destUIDs, tx.Commit()
391}
392
393func (db *DB) MoveMessagesToMailbox(ctx context.Context, fromMailboxID, toMailboxID int, uids []uint32) ([]uint32, error) {
394 tx, err := db.write.BeginTx(ctx, nil)
395 if err != nil {
396 return nil, err
397 }
398 defer txRollback(tx)
399
400 destUIDs := make([]uint32, len(uids))
401 for i, uid := range uids {
402 nextUID, err := getAndIncreaseNextUID(ctx, tx, toMailboxID)
403 if err != nil {
404 return nil, err
405 }
406 destUIDs[i] = nextUID
407
408 _, err = tx.ExecContext(
409 ctx,
410 `
411UPDATE mailbox_messages
412SET mailbox_id = ?, uid = ?
413WHERE mailbox_id = ? AND uid = ?`,
414 toMailboxID, nextUID, fromMailboxID, uid)
415 if err != nil {
416 return nil, err
417 }
418 }
419
420 return destUIDs, tx.Commit()
421}
422
423func (db *DB) SetGatekeeperDecision(ctx context.Context, userID int, fromAddress, destination string) error {
424 _, err := db.write.ExecContext(ctx,
425 `
426INSERT INTO gatekeepers (user_id, from_address, destination, created_at)
427VALUES (?, ?, ?, ?) ON CONFLICT DO UPDATE SET destination = ?`,
428 userID, fromAddress, destination, time.Now().Unix(), destination)
429 if err != nil {
430 return err
431 }
432
433 slog.Info("set gatekeeper decision", "user_id", userID, "destination", destination, "from_address", fromAddress)
434 return nil
435}
436
437// getAndIncreaseNextUID returns the next UID to be used. This function increases the mailbox' UID on each call.
438// It should only be used as part of another transaction.
439// There are certainly more efficient ways, but this is by far the most readable.
440func getAndIncreaseNextUID(ctx context.Context, db *sql.Tx, mailboxID int) (uint32, error) {
441 res := db.QueryRowContext(
442 ctx,
443 `UPDATE mailboxes SET uidnext = uidnext + 1 WHERE id = ? RETURNING uidnext`, mailboxID)
444 if res.Err() != nil {
445 return 0, res.Err()
446 }
447
448 var uidnext uint32
449 err := res.Scan(&uidnext)
450 if err != nil {
451 return 0, err
452 }
453 return uidnext - 1, nil // The query returns the upcoming UID. We need to return the to-be-used UID
454}
455
456func (db *DB) GetAllFlaggedDeletedMessages(ctx context.Context, mailboxID int) ([]uint32, error) {
457 rows, err := db.read.QueryContext(ctx, `SELECT uid FROM message_flags WHERE flag = '\Deleted' AND mailbox_id = ?`, mailboxID)
458 if err != nil {
459 return nil, err
460 }
461
462 defer func(rows *sql.Rows) {
463 _ = rows.Close()
464 }(rows)
465
466 uids := make([]uint32, 0)
467 for rows.Next() {
468 var uid uint32
469 err := rows.Scan(&uid)
470 if err != nil {
471 return nil, err
472 }
473 uids = append(uids, uid)
474 }
475 return uids, rows.Err()
476}
477
478func (db *DB) DeleteMessageFromMailbox(ctx context.Context, mailboxID int, uid uint32) error {
479 // message_flags are automatically deleted via CASCADE
480 _, err := db.write.ExecContext(ctx, `DELETE FROM mailbox_messages WHERE mailbox_id = ? AND uid = ?`, mailboxID, uid)
481 return err
482}
483
484// UIDExists reports whether a user has a message with the given uid.
485// RFC 9051: A non-existent unique identifier is ignored without any error message generated. Thus, it is possible for a
486// UID FETCH command to return an OK without any data or a UID COPY, UID MOVE, or UID STORE to return an OK without
487// performing any operations.
488func (db *DB) UIDExists(ctx context.Context, userID int, uid uint32) (bool, error) {
489 query := `
490 SELECT EXISTS (
491 SELECT 1 FROM mailbox_messages mm
492 JOIN mailboxes m ON mm.mailbox_id = m.id
493 WHERE m.user_id = ? AND mm.uid = ?
494 )`
495
496 var exists bool
497 err := db.read.QueryRowContext(ctx, query, userID, uid).Scan(&exists)
498 if err != nil {
499 return false, fmt.Errorf("error checking uid existence: %w", err)
500 }
501 return exists, nil
502}
503
504func (db *DB) GetLastUID(ctx context.Context, userID, mailboxID int) (imap.UID, error) {
505 var uid imap.UID
506 err := db.read.QueryRowContext(ctx, `
507 SELECT COALESCE(MAX(mm.uid), 0)
508 FROM mailbox_messages mm
509 JOIN mailboxes mb ON mb.id = mm.mailbox_id
510 WHERE mm.mailbox_id = ?
511 AND mb.user_id = ?
512 AND mm.expunged_modseq IS NULL
513 `, mailboxID, userID).Scan(&uid)
514 return uid, err
515}
516
517// nextUIDValidity is db roll-back save. Even with a played-back backup, the counter will be increasing.
518// It will also be increasing if it is called more than once per second.
519func nextUIDValidity(tx *sql.Tx) (uint32, error) {
520 var v int64
521 err := tx.QueryRow(`
522 UPDATE uidvalidity_counter
523 SET counter = MAX(counter + 1, CAST(strftime('%s','now') AS INTEGER))
524 RETURNING counter
525 `).Scan(&v)
526 if err != nil {
527 return 0, err
528 }
529 return uint32(v), nil
530}