1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
|
package main
import (
"database/sql"
"encoding/json"
"fmt"
"github.com/emersion/go-message/mail"
"io"
"log"
"strings"
"time"
"github.com/emersion/go-imap"
"github.com/emersion/go-imap/client"
)
// ============================================================================
// IMAP helpers
// ============================================================================
// imapCreds holds the connection parameters for an IMAP server.
type imapCreds struct {
host string
port int
user string
pass string
}
// Mailbox groups a mailbox name with its UIDVALIDITY. These two values must
// always travel together — a UID is only meaningful within a specific
// (name, uidvalidity) pair.
type Mailbox struct {
Name string
UIDValidity uint32
}
// connectIMAP dials the IMAP server, logs in, and returns the ready client.
// The caller is responsible for calling c.Logout().
func connectIMAP(creds imapCreds) (*client.Client, error) {
addr := fmt.Sprintf("%s:%d", creds.host, creds.port)
log.Printf("imap: connecting to %s ...", addr)
c, err := client.DialTLS(addr, nil)
if err != nil {
return nil, fmt.Errorf("dial: %w", err)
}
if err := c.Login(creds.user, creds.pass); err != nil {
c.Logout()
return nil, fmt.Errorf("login: %w", err)
}
log.Printf("imap: logged in as %s", creds.user)
return c, nil
}
// ensureMailboxExists creates the named IMAP mailbox if it does not already
// exist. Errors from the CREATE command are logged but not returned when the
// mailbox already exists (the server returns NO in that case).
func ensureMailboxExists(creds imapCreds, name string) error {
c, err := connectIMAP(creds)
if err != nil {
return err
}
defer c.Logout()
if err := c.Create(name); err != nil {
// Servers return an error if the mailbox already exists; ignore it.
log.Printf("imap: CREATE %q: %v (may already exist)", name, err)
}
return nil
}
// selectMailbox selects a mailbox and returns a Mailbox with its current
// UIDVALIDITY, recording/warning via recordUIDValidity.
func selectMailbox(c *client.Client, db *sql.DB, name string) (Mailbox, uint32, error) {
status, err := c.Select(name, true /* readonly */)
if err != nil {
return Mailbox{}, 0, fmt.Errorf("select %s: %w", name, err)
}
mbox := Mailbox{Name: name, UIDValidity: status.UidValidity}
if err := recordUIDValidity(db, mbox); err != nil {
return Mailbox{}, 0, fmt.Errorf("record uidvalidity: %w", err)
}
log.Printf("imap: selected %s (messages=%d uidvalidity=%d uidnext=%d)",
name, status.Messages, status.UidValidity, status.UidNext)
return mbox, status.Messages, nil
}
// recordUIDValidity inserts a new (mailbox, uidvalidity) pair into the log if
// it hasn't been seen before. If this mailbox already has a different
// uidvalidity on record, a warning is logged — the mailbox was reconstructed.
// Old messages from previous epochs remain in the DB but are invisible to
// queries because they carry a different uidvalidity in their composite key.
func recordUIDValidity(db *sql.DB, mbox Mailbox) error {
tx, err := db.Begin()
if err != nil {
return fmt.Errorf("begin: %w", err)
}
defer tx.Rollback()
// Check whether we've seen any other uidvalidity for this mailbox.
var count int
err = tx.QueryRow(
`SELECT COUNT(*) FROM mailbox_uidvalidity_log
WHERE mailbox = ? AND uidvalidity != ?`,
mbox.Name, mbox.UIDValidity,
).Scan(&count)
if err != nil {
return fmt.Errorf("query uidvalidity log: %w", err)
}
if count > 0 {
log.Printf("imap: WARNING: UIDVALIDITY for %s changed to %d "+
"(mailbox was reconstructed; %d old epoch(s) in log). "+
"Old messages remain in DB but are no longer visible.",
mbox.Name, mbox.UIDValidity, count)
}
if _, err = tx.Exec(
`INSERT OR IGNORE INTO mailbox_uidvalidity_log(mailbox, uidvalidity, first_seen)
VALUES (?, ?, ?)`,
mbox.Name, mbox.UIDValidity, time.Now().Unix(),
); err != nil {
return fmt.Errorf("insert uidvalidity: %w", err)
}
return tx.Commit()
}
// dbMaxUID returns the highest UID stored for the given mailbox+uidvalidity, or 0.
func dbMaxUID(db *sql.DB, mbox Mailbox) (uint32, error) {
var uid sql.NullInt64
err := db.QueryRow(
`SELECT MAX(uid) FROM messages WHERE mailbox = ? AND uidvalidity = ?`,
mbox.Name, mbox.UIDValidity,
).Scan(&uid)
if err != nil {
return 0, err
}
if !uid.Valid {
return 0, nil
}
return uint32(uid.Int64), nil
}
// addr is the JSON-serialisable form of a single email address.
type addr struct {
Name string `json:"name,omitempty"`
Address string `json:"address"`
}
// imapAddrsToJSON converts a slice of IMAP addresses to a JSON string.
// Returns nil if the slice is empty.
func imapAddrsToJSON(addrs []*imap.Address) []byte {
if len(addrs) == 0 {
return nil
}
out := make([]addr, 0, len(addrs))
for _, a := range addrs {
out = append(out, addr{
Name: a.PersonalName,
Address: a.MailboxName + "@" + a.HostName,
})
}
b, _ := json.Marshal(out)
return b
}
// referencesFromHeader parses the References header from raw RFC 5322 bytes.
// Returns a space-separated list of message-ids, or "" if absent.
func referencesFromHeader(rawHeader []byte) string {
if rawHeader == nil {
return ""
}
mr, err := mail.CreateReader(strings.NewReader(string(rawHeader)))
if err != nil {
return ""
}
refs, err := mr.Header.MsgIDList("References")
if err != nil || len(refs) == 0 {
return ""
}
return strings.Join(refs, " ")
}
// readMsgBody reads all body section literals from a message into a map keyed
// by the FetchItem string (e.g. "RFC822.HEADER", "RFC822").
// This avoids the Peek flag mismatch in BodySectionName.Equal().
func readMsgBody(msg *imap.Message) (map[string][]byte, error) {
out := make(map[string][]byte, len(msg.Body))
for k, lit := range msg.Body {
b, err := io.ReadAll(lit)
if err != nil {
return nil, fmt.Errorf("read section %s: %w", k.FetchItem(), err)
}
out[string(k.FetchItem())] = b
}
return out, nil
}
// insertMessage inserts a single IMAP message into the database.
// rawHeader is the RFC822.HEADER bytes (from header-only fetches).
// rawBody is the full RFC822 bytes (from full fetches), may be nil.
// Must be called within a transaction.
func insertMessage(tx *sql.Tx, mbox Mailbox, msg *imap.Message, rawHeader, rawBody []byte) (bool, error) {
if rawHeader == nil && rawBody == nil {
return false, fmt.Errorf("no body bytes for uid=%d", msg.Uid)
}
subject := ""
fromAddr := ""
var date int64
var messageID, inReplyTo string
var toAddrs, ccAddrs, bccAddrs []byte
if env := msg.Envelope; env != nil {
subject = env.Subject
if len(env.From) > 0 {
a := env.From[0]
if a.PersonalName != "" {
fromAddr = fmt.Sprintf("%s <%s@%s>", a.PersonalName, a.MailboxName, a.HostName)
} else {
fromAddr = fmt.Sprintf("%s@%s", a.MailboxName, a.HostName)
}
}
date = env.Date.Unix()
messageID = env.MessageId
inReplyTo = env.InReplyTo
toAddrs = imapAddrsToJSON(env.To)
ccAddrs = imapAddrsToJSON(env.Cc)
bccAddrs = imapAddrsToJSON(env.Bcc)
}
// References is not in the IMAP envelope — parse from header_raw.
references := referencesFromHeader(rawHeader)
// A server that did not report RFC822.SIZE leaves this at zero; store NULL
// so the message carries no size hint rather than a false one of 0 bytes.
var rfc822Size any
if msg.Size > 0 {
rfc822Size = msg.Size
}
res, err := tx.Exec(
`INSERT OR IGNORE INTO messages
(mailbox, uid, uidvalidity, subject, from_addr, date,
header_raw, display_part,
message_id, in_reply_to, references_, to_addrs, cc_addrs, bcc_addrs,
rfc822_size)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
mbox.Name, msg.Uid, mbox.UIDValidity, subject, fromAddr, date,
rawHeader, rawBody,
messageID, inReplyTo, references, toAddrs, ccAddrs, bccAddrs,
rfc822Size,
)
if err != nil {
return false, err
}
n, _ := res.RowsAffected()
if n == 0 {
return false, nil
}
msgDBID, err := res.LastInsertId()
if err != nil {
return false, fmt.Errorf("last insert id: %w", err)
}
if err := parseAndInsertHeaders(tx, msgDBID, rawHeader); err != nil {
return false, fmt.Errorf("insert headers uid=%d: %w", msg.Uid, err)
}
return true, nil
}
// parseAndInsertHeaders parses all RFC 5322 headers from rawHeader and inserts
// them into message_headers as (message_id, name, value, ord) rows.
// name is lowercased; value is unfolded and encoded-word decoded via go-message.
// Multi-valued headers (e.g. Received) each get their own row with ord preserving order.
func parseAndInsertHeaders(tx *sql.Tx, msgID int64, rawHeader []byte) error {
if rawHeader == nil {
return nil
}
mr, err := mail.CreateReader(strings.NewReader(string(rawHeader)))
if err != nil {
// Fall back to raw line-by-line parse if go-message chokes.
return insertHeadersRaw(tx, msgID, rawHeader)
}
ord := 0
fields := mr.Header.Fields()
for fields.Next() {
name := strings.ToLower(fields.Key())
value := fields.Value()
if _, err := tx.Exec(
`INSERT INTO message_headers(message_id, name, value, ord) VALUES (?,?,?,?)`,
msgID, name, value, ord,
); err != nil {
return err
}
ord++
}
return nil
}
// insertHeadersRaw is a fallback parser that handles headers line-by-line
// when go-message fails to parse the header block. It correctly unfolds
// continuation lines (lines starting with whitespace).
func insertHeadersRaw(tx *sql.Tx, msgID int64, rawHeader []byte) error {
lines := strings.Split(string(rawHeader), "\n")
ord := 0
var curName, curValue string
flush := func() error {
if curName == "" {
return nil
}
_, err := tx.Exec(
`INSERT INTO message_headers(message_id, name, value, ord) VALUES (?,?,?,?)`,
msgID, strings.ToLower(curName), strings.TrimSpace(curValue), ord,
)
ord++
curName, curValue = "", ""
return err
}
for _, line := range lines {
line = strings.TrimRight(line, "\r")
if line == "" {
break // end of headers
}
if len(line) > 0 && (line[0] == ' ' || line[0] == '\t') {
// Folded continuation line.
curValue += " " + strings.TrimSpace(line)
continue
}
if err := flush(); err != nil {
return err
}
if i := strings.IndexByte(line, ':'); i > 0 {
curName = line[:i]
curValue = strings.TrimSpace(line[i+1:])
}
}
return flush()
}
// fetchItemsHeaderOnly fetches envelope + headers only (no body).
// Body is fetched lazily on first open via fetchSingleBody.
//
// RFC822.SIZE rides along because the server has it precomputed and it is a
// single integer: it costs no extra round-trip on a fetch that is happening
// anyway, and it is what lets a listing tell a client how expensive a message
// will be to read before it fetches it.
var fetchItemsHeaderOnly = []imap.FetchItem{
imap.FetchUid,
imap.FetchEnvelope,
imap.FetchRFC822Header,
imap.FetchRFC822Size,
}
// fetchNew fetches all messages with UID > afterUID for the given mailbox
// and inserts them into the database.
// Returns the new maximum UID (unchanged if nothing new).
func fetchNew(c *client.Client, db *sql.DB, mbox Mailbox, afterUID uint32) (uint32, error) {
seqset := new(imap.SeqSet)
seqset.AddRange(afterUID+1, 0) // 0 = "*" in go-imap
messages := make(chan *imap.Message, 10)
done := make(chan error, 1)
go func() {
done <- c.UidFetch(seqset, fetchItemsHeaderOnly, messages)
}()
// Collect all messages first, then insert in one transaction.
var fetched []*imap.Message
for msg := range messages {
fetched = append(fetched, msg)
}
if err := <-done; err != nil {
return afterUID, fmt.Errorf("uid fetch: %w", err)
}
if len(fetched) == 0 {
return afterUID, nil
}
tx, err := db.Begin()
if err != nil {
return afterUID, fmt.Errorf("begin tx: %w", err)
}
defer tx.Rollback()
inserted := 0
maxUID := afterUID
for _, msg := range fetched {
body, err := readMsgBody(msg)
if err != nil {
log.Printf("warn: read uid %d: %v", msg.Uid, err)
continue
}
ok, err := insertMessage(tx, mbox, msg, body[string(imap.FetchRFC822Header)], nil)
if err != nil {
log.Printf("warn: insert uid %d: %v", msg.Uid, err)
continue
}
if ok {
inserted++
}
if msg.Uid > maxUID {
maxUID = msg.Uid
}
}
if err := tx.Commit(); err != nil {
return afterUID, fmt.Errorf("commit: %w", err)
}
if inserted > 0 {
log.Printf("imap: inserted %d new messages (maxUID now %d)", inserted, maxUID)
}
return maxUID, nil
}
// ============================================================================
// Backfill sync
// ============================================================================
const backfillBatchSize = 100
// backfillMailbox fetches all messages in the given mailbox since the cutoff
// date that are not already in the database, using the global pool for parallel
// IMAP connections. Uses a UID SEARCH to diff against local state first.
func backfillMailbox(searchConn *client.Client, db *sql.DB, pool *globalPool, mbox Mailbox, since time.Time) error {
// 1. Ask server for all UIDs since cutoff (on the existing search connection).
criteria := imap.NewSearchCriteria()
criteria.Since = since
serverUIDs, err := searchConn.UidSearch(criteria)
if err != nil {
return fmt.Errorf("uid search: %w", err)
}
log.Printf("imap: backfill %s: server has %d UIDs since %s",
mbox.Name, len(serverUIDs), since.Format("2006-01-02"))
if len(serverUIDs) == 0 {
return nil
}
// 2. Load the UIDs we already have locally into a set.
rows, err := db.Query(
`SELECT uid FROM messages WHERE mailbox = ? AND uidvalidity = ?`,
mbox.Name, mbox.UIDValidity,
)
if err != nil {
return fmt.Errorf("query local uids: %w", err)
}
localUIDs := make(map[uint32]struct{})
for rows.Next() {
var uid uint32
if err := rows.Scan(&uid); err != nil {
rows.Close()
return fmt.Errorf("scan uid: %w", err)
}
localUIDs[uid] = struct{}{}
}
rows.Close()
if err := rows.Err(); err != nil {
return fmt.Errorf("rows: %w", err)
}
// 3. Compute the set difference.
var missing []uint32
for _, uid := range serverUIDs {
if _, have := localUIDs[uid]; !have {
missing = append(missing, uid)
}
}
log.Printf("imap: backfill %s: %d already local, fetching %d missing",
mbox.Name, len(localUIDs), len(missing))
if len(missing) == 0 {
return nil
}
// 4. Distribute batches across pool workers using the global pool.
numBatches := (len(missing) + backfillBatchSize - 1) / backfillBatchSize
log.Printf("imap: backfill %s: fetching %d batches via global pool",
mbox.Name, numBatches)
type batchResult struct {
inserted int
err error
batchNum int
}
results := make(chan batchResult, numBatches)
for i := 0; i < len(missing); i += backfillBatchSize {
end := min(i+backfillBatchSize, len(missing))
batch := missing[i:end]
batchNum := i/backfillBatchSize + 1
// Acquire a connection from the global pool (blocks until one is free).
c, err := pool.acquire(mbox.Name)
if err != nil {
results <- batchResult{batchNum: batchNum, err: err}
continue
}
go func(c *client.Client, batch []uint32, batchNum int) {
defer pool.release(c, mbox.Name)
seqset := new(imap.SeqSet)
for _, uid := range batch {
seqset.AddNum(uid)
}
msgs := make(chan *imap.Message, backfillBatchSize)
done := make(chan error, 1)
go func() { done <- c.UidFetch(seqset, fetchItemsHeaderOnly, msgs) }()
var fetched []*imap.Message
for msg := range msgs {
fetched = append(fetched, msg)
}
if err := <-done; err != nil {
results <- batchResult{batchNum: batchNum, err: fmt.Errorf("uid fetch batch %d: %w", batchNum, err)}
return
}
tx, err := db.Begin()
if err != nil {
results <- batchResult{batchNum: batchNum, err: fmt.Errorf("begin tx batch %d: %w", batchNum, err)}
return
}
inserted := 0
for _, msg := range fetched {
body, err := readMsgBody(msg)
if err != nil {
log.Printf("warn: read uid %d: %v", msg.Uid, err)
continue
}
ok, err := insertMessage(tx, mbox, msg, body[string(imap.FetchRFC822Header)], nil)
if err != nil {
log.Printf("warn: insert uid %d: %v", msg.Uid, err)
continue
}
if ok {
inserted++
}
}
if err := tx.Commit(); err != nil {
tx.Rollback()
results <- batchResult{batchNum: batchNum, err: fmt.Errorf("commit batch %d: %w", batchNum, err)}
return
}
results <- batchResult{batchNum: batchNum, inserted: inserted}
}(c, batch, batchNum)
}
// 6. Collect results.
totalInserted := 0
for range numBatches {
r := <-results
if r.err != nil {
log.Printf("warn: backfill batch %d: %v", r.batchNum, r.err)
continue
}
totalInserted += r.inserted
if r.inserted > 0 {
log.Printf("imap: backfill %s: batch %d done, inserted %d",
mbox.Name, r.batchNum, r.inserted)
}
}
log.Printf("imap: backfill %s: done, inserted %d new messages",
mbox.Name, totalInserted)
return nil
}
// ============================================================================
// Expunge reconciliation
// ============================================================================
// Deleting rows is the one irreversible thing mailweb does to its own mirror,
// and a re-sync only restores the backfill window — anything older is gone for
// good. So a diff that would remove an implausible share of a mailbox is
// treated as a bug (a truncated SEARCH response, the wrong mailbox selected)
// rather than as news, and nothing is deleted.
//
// Both conditions must hold before the refusal fires: the fraction guards large
// mailboxes, where 20% is thousands of rows, and the absolute count keeps a
// small mailbox from tripping it when a handful of messages legitimately go —
// two messages leaving a mailbox of five is 40% and entirely normal.
const (
pruneMaxFraction = 0.2
pruneMinAbsolute = 50
)
// pruneMissing deletes the rows of mbox whose UID the server no longer lists.
//
// serverUIDs must come from a SEARCH over the whole mailbox, not a windowed
// one: a UID absent from the answer is taken as deleted, so a window that
// merely failed to mention a message would delete it.
//
// The scope is (mailbox, uidvalidity). Rows from earlier epochs carry a
// different uidvalidity, are invisible to every query already, and say nothing
// about what the current epoch contains — a UID set from this epoch is no
// evidence about them either way, so they are left alone.
//
// The scope is bounded above as well, by uidNext — the mailbox's UIDNEXT as
// the searching connection knew it. A message the server assigns after that
// point has a UID at or above the mark (RFC 3501 §2.3.1.1), so a local row
// there is necessarily newer than the search, and its absence from the answer
// is a statement the search never made. Two things put such a row there: a
// SEARCH answered by a connection whose session predates the message, and an
// ordinary race, since fetchLoop inserts on its own connection and may do so
// between the search and this prune. The watermark answers both without
// depending on the timing of either.
//
// UIDNEXT is the bound rather than the highest UID the answer happened to
// contain, because the two differ in exactly the case that matters. If the
// newest messages in the mailbox were the ones expunged, the highest surviving
// UID sits below them, and a mark taken from the answer would shield the very
// rows that ought to go — the prune would then never remove them, since
// nothing later makes the mark rise past them. UIDNEXT does not move when
// messages leave, only when they arrive, so it separates "newer than the
// search" from "deleted at the end of the mailbox".
//
// A uidNext of 0 means the caller could not determine it. That is treated as
// no upper bound rather than as a bound of zero: the latter would silently
// disable pruning altogether.
//
// Returns the number of messages deleted.
func pruneMissing(db *sql.DB, mbox Mailbox, serverUIDs []uint32, uidNext uint32) (int, error) {
onServer := make(map[uint32]struct{}, len(serverUIDs))
for _, uid := range serverUIDs {
onServer[uid] = struct{}{}
}
rows, err := db.Query(
`SELECT id, uid FROM messages WHERE mailbox = ? AND uidvalidity = ?`,
mbox.Name, mbox.UIDValidity,
)
if err != nil {
return 0, fmt.Errorf("query local uids: %w", err)
}
var goneIDs []int64
localCount := 0
aboveWatermark := 0
for rows.Next() {
var id int64
var uid uint32
if err := rows.Scan(&id, &uid); err != nil {
rows.Close()
return 0, fmt.Errorf("scan uid: %w", err)
}
localCount++
if uidNext > 0 && uid >= uidNext {
aboveWatermark++
continue
}
if _, have := onServer[uid]; !have {
goneIDs = append(goneIDs, id)
}
}
rows.Close()
if err := rows.Err(); err != nil {
return 0, fmt.Errorf("rows: %w", err)
}
if aboveWatermark > 0 {
log.Printf("imap: reconcile %s: %d message(s) newer than the search "+
"(uid >= uidnext %d) held back from the prune",
mbox.Name, aboveWatermark, uidNext)
}
if len(goneIDs) == 0 {
return 0, nil
}
if len(goneIDs) > pruneMinAbsolute &&
float64(len(goneIDs)) > pruneMaxFraction*float64(localCount) {
return 0, fmt.Errorf(
"refusing to prune %d of %d messages from %s (uidvalidity=%d): "+
"more than %.0f%% of the mailbox would be deleted, which is more "+
"likely a bug than an expunge; nothing was deleted",
len(goneIDs), localCount, mbox.Name, mbox.UIDValidity,
pruneMaxFraction*100)
}
tx, err := db.Begin()
if err != nil {
return 0, fmt.Errorf("begin: %w", err)
}
defer tx.Rollback()
// message_headers rows must be deleted explicitly: the foreign key has no
// ON DELETE CASCADE, and SQLite does not enforce foreign keys at all unless
// the connection asks it to, so nothing would remove them for us. Orphaned
// header rows are not merely wasted space — the contacts and forge views
// join through them, so a stale row would keep a deleted message visible
// there after it had gone from the message table.
for _, id := range goneIDs {
if _, err := tx.Exec(`DELETE FROM message_headers WHERE message_id = ?`, id); err != nil {
return 0, fmt.Errorf("delete headers for msg %d: %w", id, err)
}
if _, err := tx.Exec(`DELETE FROM messages WHERE id = ?`, id); err != nil {
return 0, fmt.Errorf("delete msg %d: %w", id, err)
}
}
if err := tx.Commit(); err != nil {
return 0, fmt.Errorf("commit: %w", err)
}
log.Printf("imap: reconcile %s: pruned %d of %d messages no longer on server",
mbox.Name, len(goneIDs), localCount)
return len(goneIDs), nil
}
// reconcileMailbox asks the server for every UID in the mailbox and prunes the
// local rows that are missing from the answer. c must already be selected on
// mbox.
//
// This is the counterpart of backfillMailbox's set difference, run in the other
// direction: backfill asks what the server has that we lack, reconcile asks
// what we have that the server lacks. It exists because an expunge cannot be
// applied directly — an EXPUNGE response carries a sequence number, and mailweb
// never holds a message list to resolve one against. But it does not need to be:
// the update only has to say that something vanished, and this diff says what.
func reconcileMailbox(c *client.Client, db *sql.DB, mbox Mailbox) error {
// Bring the connection up to date before asking it what exists. A pooled
// connection may have been idle since before the newest mail arrived, and
// a server need only announce new mail while running a command, so an
// unrefreshed SEARCH can answer with a set that omits it. Here that is not
// a failed read but a deletion: pruneMissing takes a UID absent from the
// answer as gone. One round-trip, on a path that already settles for five
// seconds before it runs.
if err := c.Noop(); err != nil {
return fmt.Errorf("refresh session before search: %w", err)
}
// SEARCH ALL, not the backfill's SEARCH SINCE. The window is matched
// against INTERNALDATE while the local date column holds the Date: header,
// so the two disagree for any message delivered well after it was sent, and
// a windowed diff would read that disagreement as a deletion.
serverUIDs, err := c.UidSearch(imap.NewSearchCriteria())
if err != nil {
return fmt.Errorf("uid search: %w", err)
}
// Read UIDNEXT after the search, so the watermark cannot sit below a
// message the search already knew about. It is taken from the connection's
// own mailbox state, which is what makes it the right bound: it describes
// the same session that answered the search, rather than some later truth
// about the mailbox that this answer was never evidence for.
var uidNext uint32
if status := c.Mailbox(); status != nil {
uidNext = status.UidNext
}
if uidNext == 0 {
// Not fatal: pruneMissing treats an unknown watermark as no upper
// bound, which is how this worked before the bound existed. Worth
// logging, because it means the guard is not in force this time round.
log.Printf("imap: reconcile %s: server did not report UIDNEXT; "+
"pruning without an upper bound", mbox.Name)
}
// An empty answer for a mailbox we hold messages for is indistinguishable
// here from every message having been deleted. pruneMissing's threshold
// would catch it, but only after building the full delete list, and the
// case is worth naming rather than leaving to a generic guard.
if len(serverUIDs) == 0 {
log.Printf("imap: reconcile %s: server reports an empty mailbox, skipping prune",
mbox.Name)
return nil
}
if _, err := pruneMissing(db, mbox, serverUIDs, uidNext); err != nil {
return err
}
return nil
}
// initialSync runs backfill for all mailboxes in parallel using the global pool.
func initialSync(db *sql.DB, creds imapCreds, pool *globalPool, mailboxes []string) error {
since := time.Now().AddDate(-2, 0, 0)
errs := make(chan error, len(mailboxes))
for _, name := range mailboxes {
go func(name string) {
c, err := connectIMAP(creds)
if err != nil {
errs <- err
return
}
mbox, _, err := selectMailbox(c, db, name)
if err != nil {
c.Logout()
errs <- err
return
}
if err := backfillMailbox(c, db, pool, mbox, since); err != nil {
c.Logout()
errs <- err
return
}
// Reconcile after backfilling, on the same connection: the two
// diff the same UID set in opposite directions, so running them
// in this order means the prune sees everything the backfill just
// inserted rather than racing it and deleting it again.
//
// A failure here is not fatal the way a failed backfill is. A
// missed prune leaves rows that are stale, which is what mailweb
// did for its whole existence; refusing to start over it would
// trade a cosmetic fault for an outage.
if err := reconcileMailbox(c, db, mbox); err != nil {
log.Printf("warn: reconcile %s: %v", mbox.Name, err)
}
c.Logout()
errs <- nil
}(name)
}
for range mailboxes {
if err := <-errs; err != nil {
return err
}
}
return nil
}
// ============================================================================
// IDLE listener
// ============================================================================
const reconnectDelay = 5 * time.Second
// idleLoop maintains a persistent IMAP connection in IDLE mode for one mailbox.
// When the server signals new mail it sends the current Mailbox on fetchCh.
// On any error it logs, waits, and reconnects.
func idleLoop(creds imapCreds, db *sql.DB, mailboxName string, fetchCh, reconcileCh chan<- Mailbox) {
for {
if err := runIdle(creds, db, mailboxName, fetchCh, reconcileCh); err != nil {
log.Printf("idle[%s]: error: %v — reconnecting in %s", mailboxName, err, reconnectDelay)
time.Sleep(reconnectDelay)
}
}
}
func runIdle(creds imapCreds, db *sql.DB, mailboxName string, fetchCh, reconcileCh chan<- Mailbox) error {
updates := make(chan client.Update, 4)
c, err := connectIMAP(creds)
if err != nil {
return err
}
defer c.Logout()
c.Updates = updates
mbox, _, err := selectMailbox(c, db, mailboxName)
if err != nil {
return err
}
log.Printf("idle: watching %s", mbox.Name)
// Both signals are deliberately lossy: the receiver's work is a set
// difference against the server, so it does not matter how many times it
// was asked, only that it runs once afterwards. Dropping a signal when one
// is already queued keeps a burst of updates from queueing a burst of
// identical work.
signalFetch := func() {
select {
case fetchCh <- mbox:
default:
}
}
signalReconcile := func() {
select {
case reconcileCh <- mbox:
default:
}
}
// Loop over IDLE restarts on the same connection, reconnecting only on error.
for {
stop := make(chan struct{})
idleDone := make(chan error, 1)
go func() {
idleDone <- c.Idle(stop, &client.IdleOptions{
LogoutTimeout: 29 * time.Minute,
})
}()
idleErr := func() error {
for {
select {
case update := <-updates:
switch u := update.(type) {
case *client.MailboxUpdate:
log.Printf("idle: MailboxUpdate messages=%d", u.Mailbox.Messages)
// Signal a fetch on any mailbox change. fetchNew uses
// UidFetch((maxUID+1):*) which is idempotent — if nothing
// new exists it returns zero messages harmlessly.
signalFetch()
case *client.ExpungeUpdate:
log.Printf("idle: ExpungeUpdate SeqNum=%d", u.SeqNum)
// The sequence number is not actionable — mailweb holds
// no message list to resolve one against — but the fact
// that something was expunged is. reconcileLoop turns it
// into a UID set difference, which needs no such state.
signalReconcile()
case *client.MessageUpdate:
log.Printf("idle: MessageUpdate uid=%d", u.Message.Uid)
case *client.StatusUpdate:
// Server keepalives — ignore.
default:
log.Printf("idle: unknown update: %T", update)
}
case err := <-idleDone:
return err
}
}
}()
if idleErr != nil {
return fmt.Errorf("idle: %w", idleErr)
}
// Idle returned cleanly (LogoutTimeout) — restart on the same connection.
log.Printf("idle: restarting IDLE on %s", mbox.Name)
}
}
// fetchLoop waits for Mailbox signals from idleLoop and fetches new messages.
// It tracks maxUID per (mailbox, uidvalidity) in memory, loaded from the DB
// at startup. On any error it logs, waits, and retries.
func fetchLoop(creds imapCreds, db *sql.DB, fetchCh <-chan Mailbox) {
// maxUID is keyed by Mailbox so it works correctly across uidvalidity epochs.
maxUIDs := make(map[Mailbox]uint32)
for mbox := range fetchCh {
if _, seen := maxUIDs[mbox]; !seen {
uid, err := dbMaxUID(db, mbox)
if err != nil {
log.Printf("fetch: could not load maxUID for %+v: %v", mbox, err)
}
maxUIDs[mbox] = uid
log.Printf("fetch: %s uidvalidity=%d starting maxUID=%d",
mbox.Name, mbox.UIDValidity, uid)
}
for {
newMax, err := runFetch(creds, db, mbox, maxUIDs[mbox])
if err != nil {
log.Printf("fetch: error: %v — retrying in %s", err, reconnectDelay)
time.Sleep(reconnectDelay)
continue
}
maxUIDs[mbox] = newMax
break
}
}
}
func runFetch(creds imapCreds, db *sql.DB, mbox Mailbox, afterUID uint32) (uint32, error) {
c, err := connectIMAP(creds)
if err != nil {
return afterUID, err
}
defer c.Logout()
// Select the mailbox but trust the mbox we received from idleLoop —
// we only re-record uidvalidity to keep the log up to date.
if err := recordUIDValidity(db, mbox); err != nil {
return afterUID, fmt.Errorf("record uidvalidity: %w", err)
}
if _, err := c.Select(mbox.Name, true); err != nil {
return afterUID, fmt.Errorf("select: %w", err)
}
return fetchNew(c, db, mbox, afterUID)
}
// reconcileSettle is how long reconcileLoop waits after the first expunge
// signal before acting, collecting whatever else arrives in the meantime.
//
// A filter run or a bulk delete expunges many messages at once — a single burst
// in the logs held 163 — and each one produces its own untagged EXPUNGE. They
// all describe the same mailbox and are answered by the same SEARCH, so acting
// on the first would mean doing that work once per message deleted.
const reconcileSettle = 5 * time.Second
// reconcileLoop waits for expunge signals from idleLoop and prunes the messages
// the server no longer has.
//
// Signals are coalesced twice over: once by the non-blocking send in idleLoop,
// which drops a signal when one is already queued, and once here by settling
// for reconcileSettle and folding everything that arrived into a set of
// mailboxes. A burst of expunges in one mailbox therefore costs one SEARCH, and
// a burst spanning several costs one per mailbox.
//
// Errors are logged and dropped rather than retried. Unlike a fetch, whose
// failure loses a message until the next signal, a failed prune leaves rows
// that are merely stale, and the next expunge or restart will try again.
func reconcileLoop(db *sql.DB, pool *globalPool, reconcileCh <-chan Mailbox) {
for first := range reconcileCh {
pending := map[Mailbox]struct{}{first: {}}
// Settle: absorb the rest of the burst.
timer := time.NewTimer(reconcileSettle)
collect:
for {
select {
case mbox := <-reconcileCh:
pending[mbox] = struct{}{}
case <-timer.C:
break collect
}
}
for mbox := range pending {
if err := runReconcile(db, pool, mbox); err != nil {
log.Printf("reconcile: %s: %v", mbox.Name, err)
}
}
}
}
// runReconcile borrows a connection from the pool, which already knows how to
// hand out one selected on the wanted mailbox, and reconciles it. The pool is
// used rather than a fresh connection because reconciling is rare and short:
// opening a connection would cost more than the SEARCH it carries.
func runReconcile(db *sql.DB, pool *globalPool, mbox Mailbox) error {
c, err := pool.acquire(mbox.Name)
if err != nil {
return err
}
defer pool.release(c, mbox.Name)
return reconcileMailbox(c, db, mbox)
}
|