1package webaccount
2
3import (
4 "archive/tar"
5 "archive/zip"
6 "bufio"
7 "bytes"
8 "compress/gzip"
9 "context"
10 cryptrand "crypto/rand"
11 "encoding/json"
12 "errors"
13 "fmt"
14 "io"
15 "log/slog"
16 "maps"
17 "os"
18 "path"
19 "runtime/debug"
20 "slices"
21 "strconv"
22 "strings"
23 "time"
24
25 "golang.org/x/text/unicode/norm"
26
27 "github.com/mjl-/bstore"
28
29 "github.com/mjl-/mox/message"
30 "github.com/mjl-/mox/metrics"
31 "github.com/mjl-/mox/mlog"
32 "github.com/mjl-/mox/mox-"
33 "github.com/mjl-/mox/store"
34)
35
36type importListener struct {
37 Token string
38 Events chan importEvent
39 Register chan bool // Whether register is successful.
40}
41
42type importEvent struct {
43 Token string
44 SSEMsg []byte // Full SSE message, including event: ... and data: ... \n\n
45 Event any // nil, importCount, importProblem, importDone, importAborted
46 Cancel func() // For cancelling the context causing abort of the import. Set in first, import-registering, event.
47}
48
49type importAbortRequest struct {
50 Token string
51 Response chan error
52}
53
54var importers = struct {
55 Register chan *importListener
56 Unregister chan *importListener
57 Events chan importEvent
58 Abort chan importAbortRequest
59 Stop chan struct{}
60}{
61 make(chan *importListener, 1),
62 make(chan *importListener, 1),
63 make(chan importEvent),
64 make(chan importAbortRequest),
65 make(chan struct{}),
66}
67
68// ImportManage should be run as a goroutine, it manages imports of mboxes/maildirs, propagating progress over SSE connections.
69func ImportManage() {
70 log := mlog.New("httpimport", nil)
71 defer func() {
72 if x := recover(); x != nil {
73 log.Error("import manage panic", slog.Any("err", x))
74 debug.PrintStack()
75 metrics.PanicInc(metrics.Importmanage)
76 }
77 }()
78
79 type state struct {
80 MailboxCounts map[string]int
81 Problems []string
82 Done *time.Time
83 Aborted *time.Time
84 Listeners map[*importListener]struct{}
85 Cancel func()
86 }
87
88 imports := map[string]state{} // Token to state.
89 for {
90 select {
91 case l := <-importers.Register:
92 // If we have state, send it so the client is up to date.
93 s, ok := imports[l.Token]
94 l.Register <- ok
95 if !ok {
96 break
97 }
98 s.Listeners[l] = struct{}{}
99
100 sendEvent := func(kind string, v any) {
101 buf, err := json.Marshal(v)
102 if err != nil {
103 log.Errorx("marshal event", err, slog.String("kind", kind), slog.Any("event", v))
104 return
105 }
106 ssemsg := fmt.Sprintf("event: %s\ndata: %s\n\n", kind, buf)
107
108 select {
109 case l.Events <- importEvent{kind, []byte(ssemsg), nil, nil}:
110 default:
111 log.Debug("dropped initial import event to slow consumer")
112 }
113 }
114
115 for m, c := range s.MailboxCounts {
116 sendEvent("count", importCount{m, c})
117 }
118 for _, p := range s.Problems {
119 sendEvent("problem", importProblem{p})
120 }
121 if s.Done != nil {
122 sendEvent("done", importDone{})
123 } else if s.Aborted != nil {
124 sendEvent("aborted", importAborted{})
125 }
126
127 case l := <-importers.Unregister:
128 delete(imports[l.Token].Listeners, l)
129
130 case e := <-importers.Events:
131 s, ok := imports[e.Token]
132 if !ok {
133 s = state{
134 MailboxCounts: map[string]int{},
135 Listeners: map[*importListener]struct{}{},
136 Cancel: e.Cancel,
137 }
138 imports[e.Token] = s
139 }
140 for l := range s.Listeners {
141 select {
142 case l.Events <- e:
143 default:
144 log.Debug("dropped import event to slow consumer")
145 }
146 }
147 if e.Event != nil {
148 s := imports[e.Token]
149 switch x := e.Event.(type) {
150 case importCount:
151 s.MailboxCounts[x.Mailbox] = x.Count
152 case importProblem:
153 s.Problems = append(s.Problems, x.Message)
154 case importDone:
155 now := time.Now()
156 s.Done = &now
157 case importAborted:
158 now := time.Now()
159 s.Aborted = &now
160 }
161 imports[e.Token] = s
162 }
163
164 case a := <-importers.Abort:
165 s, ok := imports[a.Token]
166 if !ok {
167 a.Response <- errors.New("import not found")
168 return
169 }
170 if s.Done != nil {
171 a.Response <- errors.New("import already finished")
172 return
173 }
174 s.Cancel()
175 a.Response <- nil
176
177 case <-importers.Stop:
178 return
179 }
180
181 // Cleanup old state.
182 for t, s := range imports {
183 if len(s.Listeners) > 0 {
184 continue
185 }
186 if s.Done != nil && time.Since(*s.Done) > time.Minute || s.Aborted != nil && time.Since(*s.Aborted) > time.Minute {
187 delete(imports, t)
188 }
189 }
190 }
191}
192
193type importCount struct {
194 Mailbox string
195 Count int
196}
197type importProblem struct {
198 Message string
199}
200type importDone struct{}
201type importAborted struct{}
202type importStep struct {
203 Title string
204}
205
206// importStart prepare the import and launches the goroutine to actually import.
207// importStart is responsible for closing f and removing f.
208func importStart(log mlog.Log, accName string, f *os.File, skipMailboxPrefix string) (string, bool, error) {
209 defer func() {
210 if f != nil {
211 store.CloseRemoveTempFile(log, f, "upload for import")
212 }
213 }()
214
215 var buf [16]byte
216 cryptrand.Read(buf[:])
217 token := fmt.Sprintf("%x", buf)
218
219 if _, err := f.Seek(0, 0); err != nil {
220 return "", false, fmt.Errorf("seek to start of file: %v", err)
221 }
222
223 // Recognize file format.
224 var iszip bool
225 magicZip := []byte{0x50, 0x4b, 0x03, 0x04}
226 magicGzip := []byte{0x1f, 0x8b}
227 magic := make([]byte, 4)
228 if _, err := f.ReadAt(magic, 0); err != nil {
229 return "", true, fmt.Errorf("detecting file format: %v", err)
230 }
231 if bytes.Equal(magic, magicZip) {
232 iszip = true
233 } else if !bytes.Equal(magic[:2], magicGzip) {
234 return "", true, fmt.Errorf("file is not a zip or gzip file")
235 }
236
237 var zr *zip.Reader
238 var tr *tar.Reader
239 if iszip {
240 fi, err := f.Stat()
241 if err != nil {
242 return "", false, fmt.Errorf("stat temporary import zip file: %v", err)
243 }
244 zr, err = zip.NewReader(f, fi.Size())
245 if err != nil {
246 return "", true, fmt.Errorf("opening zip file: %v", err)
247 }
248 } else {
249 gzr, err := gzip.NewReader(f)
250 if err != nil {
251 return "", true, fmt.Errorf("gunzip: %v", err)
252 }
253 tr = tar.NewReader(gzr)
254 }
255
256 acc, err := store.OpenAccount(log, accName, false)
257 if err != nil {
258 return "", false, fmt.Errorf("open acount: %v", err)
259 }
260 acc.Lock() // Not using WithWLock because importMessage is responsible for unlocking.
261
262 tx, err := acc.DB.Begin(context.Background(), true)
263 if err != nil {
264 acc.Unlock()
265 xerr := acc.Close()
266 log.Check(xerr, "closing account")
267 return "", false, fmt.Errorf("start transaction: %v", err)
268 }
269
270 // Ensure token is registered before returning, with context that can be canceled.
271 ctx, cancel := context.WithCancel(mox.Shutdown)
272 importers.Events <- importEvent{token, []byte(": keepalive\n\n"), nil, cancel}
273
274 log.Info("starting import")
275 go importMessages(ctx, log.WithCid(mox.Cid()), token, acc, tx, zr, tr, f, skipMailboxPrefix)
276 f = nil // importMessages is now responsible for closing and removing.
277
278 return token, false, nil
279}
280
281// importMessages imports the messages from zip/tgz file f.
282// importMessages is responsible for unlocking and closing acc, and closing tx and f.
283func importMessages(ctx context.Context, log mlog.Log, token string, acc *store.Account, tx *bstore.Tx, zr *zip.Reader, tr *tar.Reader, f *os.File, skipMailboxPrefix string) {
284 // If a fatal processing error occurs, we panic with this type.
285 type importError struct{ Err error }
286
287 // During import we collect all changes and broadcast them at the end, when successful.
288 var changes []store.Change
289
290 // ID's of delivered messages. If we have to rollback, we have to remove this files.
291 var newIDs []int64
292
293 sendEvent := func(kind string, v any) {
294 buf, err := json.Marshal(v)
295 if err != nil {
296 log.Errorx("marshal event", err, slog.String("kind", kind), slog.Any("event", v))
297 return
298 }
299 ssemsg := fmt.Sprintf("event: %s\ndata: %s\n\n", kind, buf)
300 importers.Events <- importEvent{token, []byte(ssemsg), v, nil}
301 }
302
303 canceled := func() bool {
304 select {
305 case <-ctx.Done():
306 sendEvent("aborted", importAborted{})
307 return true
308 default:
309 return false
310 }
311 }
312
313 problemf := func(format string, args ...any) {
314 msg := fmt.Sprintf(format, args...)
315 sendEvent("problem", importProblem{Message: msg})
316 }
317
318 defer func() {
319 store.CloseRemoveTempFile(log, f, "uploaded messages")
320
321 for _, id := range newIDs {
322 p := acc.MessagePath(id)
323 err := os.Remove(p)
324 log.Check(err, "closing message file after import error", slog.String("path", p))
325 }
326 if tx != nil {
327 err := tx.Rollback()
328 log.Check(err, "rolling back transaction")
329 }
330 if acc != nil {
331 acc.Unlock()
332 err := acc.Close()
333 log.Check(err, "closing account")
334 }
335
336 x := recover()
337 if x == nil {
338 return
339 }
340 if err, ok := x.(importError); ok {
341 log.Errorx("import error", err.Err)
342 problemf("%s (aborting)", err.Err)
343 sendEvent("aborted", importAborted{})
344 } else {
345 log.Error("import panic", slog.Any("err", x))
346 debug.PrintStack()
347 metrics.PanicInc(metrics.Importmessages)
348 }
349 }()
350
351 ximportcheckf := func(err error, format string, args ...any) {
352 if err != nil {
353 panic(importError{fmt.Errorf("%s: %s", fmt.Sprintf(format, args...), err)})
354 }
355 }
356
357 err := acc.ThreadingWait(log)
358 ximportcheckf(err, "waiting for account thread upgrade")
359
360 conf, _ := acc.Conf()
361
362 jf, _, err := acc.OpenJunkFilter(ctx, log)
363 if err != nil && !errors.Is(err, store.ErrNoJunkFilter) {
364 ximportcheckf(err, "open junk filter")
365 }
366 defer func() {
367 if jf != nil {
368 err := jf.CloseDiscard()
369 log.Check(err, "closing junk filter")
370 }
371 }()
372
373 // Mailboxes we imported, and message counts.
374 mailboxNames := map[string]*store.Mailbox{}
375 mailboxIDs := map[int64]*store.Mailbox{}
376 mailboxKeywordCounts := map[int64]int{}
377 messages := map[string]int{}
378
379 maxSize := acc.QuotaMessageSize()
380 du := store.DiskUsage{ID: 1}
381 err = tx.Get(&du)
382 ximportcheckf(err, "get disk usage")
383 var addSize int64
384
385 // For maildirs, we are likely to get a possible dovecot-keywords file after having
386 // imported the messages. Once we see the keywords, we use them. But before that
387 // time we remember which messages miss keywords. Once the keywords become
388 // available, we'll fix up the flags for the unknown messages
389 mailboxKeywords := map[string]map[rune]string{} // Mailbox to 'a'-'z' to flag name.
390 mailboxMissingKeywordMessages := map[string]map[int64]string{} // Mailbox to message id to string consisting of the unrecognized flags.
391
392 // Previous mailbox an event was sent for. We send an event for new mailboxes, when
393 // another 100 messages were added, when adding a message to another mailbox, and
394 // finally at the end as a closing statement.
395 var prevMailbox string
396
397 var modseq store.ModSeq // Assigned on first message, used for all messages.
398
399 trainMessage := func(m *store.Message, p message.Part, pos string) {
400 words, err := jf.ParseMessage(p)
401 if err != nil {
402 problemf("parsing message %s for updating junk filter: %v (continuing)", pos, err)
403 return
404 }
405 err = jf.Train(ctx, !m.Junk, words)
406 if err != nil {
407 problemf("training junk filter for message %s: %v (continuing)", pos, err)
408 return
409 }
410 m.TrainedJunk = &m.Junk
411 }
412
413 openTrainMessage := func(m *store.Message) {
414 path := acc.MessagePath(m.ID)
415 f, err := os.Open(path)
416 if err != nil {
417 problemf("opening message again for training junk filter: %v (continuing)", err)
418 return
419 }
420 defer func() {
421 err := f.Close()
422 log.Check(err, "closing file after training junkfilter")
423 }()
424 p, err := m.LoadPart(f)
425 if err != nil {
426 problemf("loading parsed message again for training junk filter: %v (continuing)", err)
427 return
428 }
429 trainMessage(m, p, fmt.Sprintf("message id %d", m.ID))
430 }
431
432 xensureMailbox := func(name string) *store.Mailbox {
433 // Ensure name is normalized.
434 name = norm.NFC.String(name)
435 name, _, err := store.CheckMailboxName(name, true)
436 ximportcheckf(err, "checking mailbox name")
437
438 if mb, ok := mailboxNames[name]; ok {
439 return mb
440 }
441
442 var p string
443 var mb *store.Mailbox
444 var parent store.Mailbox
445 for i, e := range strings.Split(name, "/") {
446 if i == 0 {
447 p = e
448 } else {
449 p = path.Join(p, e)
450 }
451 if _, ok := mailboxNames[p]; ok {
452 continue
453 }
454
455 mb, err = acc.MailboxFind(tx, p)
456 ximportcheckf(err, "looking up mailbox %s to import to (aborting)", p)
457 if mb == nil {
458 uidvalidity, err := acc.NextUIDValidity(tx)
459 ximportcheckf(err, "finding next uid validity")
460
461 if modseq == 0 {
462 var err error
463 modseq, err = acc.NextModSeq(tx)
464 ximportcheckf(err, "assigning next modseq")
465 }
466
467 mb = &store.Mailbox{
468 CreateSeq: modseq,
469 ModSeq: modseq,
470 ParentID: parent.ID,
471 Name: p,
472 UIDValidity: uidvalidity,
473 UIDNext: 1,
474 HaveCounts: true,
475 // Do not assign special-use flags. This existing account probably already has such mailboxes.
476 }
477 err = tx.Insert(mb)
478 ximportcheckf(err, "inserting mailbox in database")
479 parent = *mb
480
481 if tx.Get(&store.Subscription{Name: p}) != nil {
482 err := tx.Insert(&store.Subscription{Name: p})
483 ximportcheckf(err, "subscribing to imported mailbox")
484 }
485 changes = append(changes, store.ChangeAddMailbox{Mailbox: *mb, Flags: []string{`\Subscribed`}})
486 }
487 if prevMailbox != "" && mb.Name != prevMailbox {
488 sendEvent("count", importCount{prevMailbox, messages[prevMailbox]})
489 }
490 mailboxKeywordCounts[mb.ID] = len(mb.Keywords)
491 mailboxNames[mb.Name] = mb
492 mailboxIDs[mb.ID] = mb
493 sendEvent("count", importCount{mb.Name, 0})
494 prevMailbox = mb.Name
495 }
496 return mb
497 }
498
499 xdeliver := func(mb *store.Mailbox, m *store.Message, f *os.File, pos string) {
500 defer store.CloseRemoveTempFile(log, f, "message file for import")
501 m.MailboxID = mb.ID
502 m.MailboxOrigID = mb.ID
503
504 addSize += m.Size
505 if maxSize > 0 && du.MessageSize+addSize > maxSize {
506 ximportcheckf(fmt.Errorf("account over maximum total size %d", maxSize), "checking quota")
507 }
508
509 if modseq == 0 {
510 var err error
511 modseq, err = acc.NextModSeq(tx)
512 ximportcheckf(err, "assigning next modseq")
513 }
514 m.CreateSeq = modseq
515 m.ModSeq = modseq
516
517 // Parse message and store parsed information for later fast retrieval.
518 p, err := message.EnsurePart(log.Logger, false, f, m.Size)
519 if err != nil {
520 problemf("parsing message %s: %s (continuing)", pos, err)
521 }
522 m.ParsedBuf, err = json.Marshal(p)
523 ximportcheckf(err, "marshal parsed message structure")
524
525 // Set fields needed for future threading. By doing it now, MessageAdd won't
526 // have to parse the Part again.
527 p.SetReaderAt(store.FileMsgReader(m.MsgPrefix, f))
528 m.PrepareThreading(log, &p)
529
530 if m.Received.IsZero() {
531 if p.Envelope != nil && !p.Envelope.Date.IsZero() {
532 m.Received = p.Envelope.Date
533 } else {
534 m.Received = time.Now()
535 }
536 }
537
538 // We set the flags that Deliver would set now and train ourselves. This prevents
539 // Deliver from training, which would open the junk filter, change it, and write it
540 // back to disk, for each message (slow).
541 m.JunkFlagsForMailbox(*mb, conf)
542 if jf != nil && m.NeedsTraining() {
543 trainMessage(m, p, pos)
544 }
545
546 opts := store.AddOpts{
547 SkipDirSync: true,
548 SkipTraining: true,
549 SkipThreads: true,
550 SkipUpdateDiskUsage: true,
551 SkipCheckQuota: true,
552 SkipPreview: true,
553 }
554 if err := acc.MessageAdd(log, tx, mb, m, f, opts); err != nil {
555 problemf("delivering message %s: %s (continuing)", pos, err)
556 return
557 }
558 newIDs = append(newIDs, m.ID)
559 changes = append(changes, m.ChangeAddUID(*mb))
560 messages[mb.Name]++
561 if messages[mb.Name]%100 == 0 || prevMailbox != mb.Name {
562 prevMailbox = mb.Name
563 sendEvent("count", importCount{mb.Name, messages[mb.Name]})
564 }
565 }
566
567 ximportMbox := func(mailbox, filename string, r io.Reader) {
568 if mailbox == "" {
569 problemf("empty mailbox name for mbox file %s (skipping)", filename)
570 return
571 }
572 mb := xensureMailbox(mailbox)
573
574 mr := store.NewMboxReader(log, store.CreateMessageTemp, filename, r)
575 for {
576 m, mf, pos, err := mr.Next()
577 if err == io.EOF {
578 break
579 } else if err != nil {
580 ximportcheckf(err, "next message in mbox file")
581 }
582
583 xdeliver(mb, m, mf, pos)
584 }
585 }
586
587 ximportMaildir := func(mailbox, filename string, r io.Reader) {
588 if mailbox == "" {
589 problemf("empty mailbox name for maildir file %s (skipping)", filename)
590 return
591 }
592 mb := xensureMailbox(mailbox)
593
594 f, err := store.CreateMessageTemp(log, "import")
595 ximportcheckf(err, "creating temp message")
596 defer func() {
597 if f != nil {
598 store.CloseRemoveTempFile(log, f, "message to import")
599 }
600 }()
601
602 // Copy data, changing bare \n into \r\n.
603 br := bufio.NewReader(r)
604 w := bufio.NewWriter(f)
605 var size int64
606 for {
607 line, err := br.ReadBytes('\n')
608 if err != nil && err != io.EOF {
609 ximportcheckf(err, "reading message")
610 }
611 if len(line) > 0 {
612 if !bytes.HasSuffix(line, []byte("\r\n")) {
613 line = append(line[:len(line)-1], "\r\n"...)
614 }
615
616 n, err := w.Write(line)
617 ximportcheckf(err, "writing message")
618 size += int64(n)
619 }
620 if err == io.EOF {
621 break
622 }
623 }
624 err = w.Flush()
625 ximportcheckf(err, "writing message")
626
627 var received time.Time
628 t := strings.SplitN(path.Base(filename), ".", 2)
629 if v, err := strconv.ParseInt(t[0], 10, 64); err == nil {
630 received = time.Unix(v, 0)
631 }
632
633 // Parse flags. See https://cr.yp.to/proto/maildir.html.
634 var keepFlags string
635 var flags store.Flags
636 keywords := map[string]bool{}
637 t = strings.SplitN(path.Base(filename), ":2,", 2)
638 if len(t) == 2 {
639 for _, c := range t[1] {
640 switch c {
641 case 'P':
642 // Passed, doesn't map to a common IMAP flag.
643 case 'R':
644 flags.Answered = true
645 case 'S':
646 flags.Seen = true
647 case 'T':
648 flags.Deleted = true
649 case 'D':
650 flags.Draft = true
651 case 'F':
652 flags.Flagged = true
653 default:
654 if c >= 'a' && c <= 'z' {
655 dovecotKeywords, ok := mailboxKeywords[mailbox]
656 if !ok {
657 // No keywords file seen yet, we'll try later if it comes in.
658 keepFlags += string(c)
659 } else if kw, ok := dovecotKeywords[c]; ok {
660 flagSet(&flags, keywords, kw)
661 }
662 }
663 }
664 }
665 }
666
667 m := store.Message{
668 Received: received,
669 Flags: flags,
670 Keywords: slices.Sorted(maps.Keys(keywords)),
671 Size: size,
672 }
673 xdeliver(mb, &m, f, filename)
674 f = nil
675 if keepFlags != "" {
676 if _, ok := mailboxMissingKeywordMessages[mailbox]; !ok {
677 mailboxMissingKeywordMessages[mailbox] = map[int64]string{}
678 }
679 mailboxMissingKeywordMessages[mailbox][m.ID] = keepFlags
680 }
681 }
682
683 importFile := func(name string, r io.Reader) {
684 origName := name
685
686 if strings.HasPrefix(name, skipMailboxPrefix) {
687 name = strings.TrimPrefix(name[len(skipMailboxPrefix):], "/")
688 }
689
690 if strings.HasSuffix(name, "/") {
691 name = strings.TrimSuffix(name, "/")
692 dir := path.Dir(name)
693 switch path.Base(dir) {
694 case "new", "cur", "tmp":
695 // Maildir, ensure it exists.
696 mailbox := path.Dir(dir)
697 xensureMailbox(mailbox)
698 }
699 // Otherwise, this is just a directory that probably holds mbox files and maildirs.
700 return
701 }
702
703 if strings.HasSuffix(path.Base(name), ".mbox") {
704 mailbox := name[:len(name)-len(".mbox")]
705 ximportMbox(mailbox, origName, r)
706 return
707 }
708 dir := path.Dir(name)
709 dirbase := path.Base(dir)
710 switch dirbase {
711 case "new", "cur", "tmp":
712 mailbox := path.Dir(dir)
713 ximportMaildir(mailbox, origName, r)
714 return
715 }
716
717 if path.Base(name) != "dovecot-keywords" {
718 problemf("unrecognized file %s (skipping)", origName)
719 return
720 }
721
722 // Handle dovecot-keywords.
723 mailbox := path.Dir(name)
724 dovecotKeywords := map[rune]string{}
725 words, err := store.ParseDovecotKeywordsFlags(r, log)
726 log.Check(err, "parsing dovecot keywords for mailbox", slog.String("mailbox", mailbox))
727 for i, kw := range words {
728 dovecotKeywords['a'+rune(i)] = kw
729 }
730 mailboxKeywords[mailbox] = dovecotKeywords
731
732 for id, chars := range mailboxMissingKeywordMessages[mailbox] {
733 var flags, zeroflags store.Flags
734 keywords := map[string]bool{}
735 for _, c := range chars {
736 kw, ok := dovecotKeywords[c]
737 if !ok {
738 problemf("unspecified dovecot message flag %c for message id %d (continuing)", c, id)
739 continue
740 }
741 flagSet(&flags, keywords, kw)
742 }
743 if flags == zeroflags && len(keywords) == 0 {
744 continue
745 }
746
747 m := store.Message{ID: id}
748 err := tx.Get(&m)
749 ximportcheckf(err, "get imported message for flag update")
750
751 mb := mailboxIDs[m.MailboxID]
752 mb.Sub(m.MailboxCounts())
753
754 oflags := m.Flags
755 m.Flags = m.Flags.Set(flags, flags)
756 m.Keywords = slices.Sorted(maps.Keys(keywords))
757
758 mb.Add(m.MailboxCounts())
759
760 mb.Keywords, _ = store.MergeKeywords(mb.Keywords, m.Keywords)
761
762 // We train before updating, training may set m.TrainedJunk.
763 if jf != nil && m.NeedsTraining() {
764 openTrainMessage(&m)
765 }
766 err = tx.Update(&m)
767 ximportcheckf(err, "updating message after flag update")
768 changes = append(changes, m.ChangeFlags(oflags, *mb))
769 }
770 delete(mailboxMissingKeywordMessages, mailbox)
771 }
772
773 if zr != nil {
774 for _, f := range zr.File {
775 if canceled() {
776 return
777 }
778 zf, err := f.Open()
779 if err != nil {
780 problemf("opening file %s in zip: %v", f.Name, err)
781 continue
782 }
783 importFile(f.Name, zf)
784 err = zf.Close()
785 log.Check(err, "closing file from zip")
786 }
787 } else {
788 for {
789 if canceled() {
790 return
791 }
792 h, err := tr.Next()
793 if err == io.EOF {
794 break
795 } else if err != nil {
796 problemf("reading next tar header: %v (aborting)", err)
797 return
798 }
799 importFile(h.Name, tr)
800 }
801 }
802
803 total := 0
804 for _, count := range messages {
805 total += count
806 }
807 log.Debug("messages imported", slog.Int("total", total))
808
809 // Send final update for count of last-imported mailbox.
810 if prevMailbox != "" {
811 sendEvent("count", importCount{prevMailbox, messages[prevMailbox]})
812 }
813
814 // Match threads.
815 if len(newIDs) > 0 {
816 sendEvent("step", importStep{"matching messages with threads"})
817 err = acc.AssignThreads(ctx, log, tx, newIDs[0], 0, io.Discard)
818 ximportcheckf(err, "assigning messages to threads")
819 }
820
821 // Update mailboxes with counts and keywords.
822 for _, mb := range mailboxIDs {
823 err = tx.Update(mb)
824 ximportcheckf(err, "updating mailbox count and keywords")
825
826 changes = append(changes, mb.ChangeCounts())
827 if len(mb.Keywords) != mailboxKeywordCounts[mb.ID] {
828 changes = append(changes, mb.ChangeKeywords())
829 }
830 }
831
832 err = acc.AddMessageSize(log, tx, addSize)
833 ximportcheckf(err, "updating disk usage after import")
834
835 err = tx.Commit()
836 tx = nil
837 ximportcheckf(err, "commit")
838 newIDs = nil
839
840 if jf != nil {
841 if err := jf.Close(); err != nil {
842 problemf("saving changes of training junk filter: %v (continuing)", err)
843 log.Errorx("saving changes of training junk filter", err)
844 }
845 jf = nil
846 }
847
848 store.BroadcastChanges(acc, changes)
849 acc.Unlock()
850 err = acc.Close()
851 log.Check(err, "closing account after import")
852 acc = nil
853
854 sendEvent("done", importDone{})
855}
856
857func flagSet(flags *store.Flags, keywords map[string]bool, word string) {
858 switch word {
859 case "forwarded", "$forwarded":
860 flags.Forwarded = true
861 case "junk", "$junk":
862 flags.Junk = true
863 case "notjunk", "$notjunk", "nonjunk", "$nonjunk":
864 flags.Notjunk = true
865 case "phishing", "$phishing":
866 flags.Phishing = true
867 case "mdnsent", "$mdnsent":
868 flags.MDNSent = true
869 default:
870 if err := store.CheckKeyword(word); err == nil {
871 keywords[word] = true
872 }
873 }
874}
875