1package webmail

3// todo: may want to add some json omitempty tags to MessageItem, or Message to reduce json size, or just have smaller types that send only the fields that are needed.

5import (

6 "compress/gzip"

7 "context"

8 cryptrand "crypto/rand"

9 "encoding/base64"

10 "encoding/json"

11 "errors"

12 "fmt"

13 "log/slog"

14 "net/http"

15 "path/filepath"

16 "reflect"

17 "runtime/debug"

18 "slices"

19 "strconv"

20 "strings"

21 "sync"

22 "time"

24 "github.com/mjl-/bstore"

25 "github.com/mjl-/sherpa"

27 "github.com/mjl-/mox/dns"

28 "github.com/mjl-/mox/message"

29 "github.com/mjl-/mox/metrics"

30 "github.com/mjl-/mox/mlog"

31 "github.com/mjl-/mox/mox-"

32 "github.com/mjl-/mox/moxvar"

33 "github.com/mjl-/mox/smtp"

34 "github.com/mjl-/mox/store"

35)

37// Request is a request to an SSE connection to send messages, either for a new

38// view, to continue with an existing view, or to a cancel an ongoing request.

39type Request struct {

40 ID int64

42 SSEID int64 // SSE connection.

44 // To indicate a request is a continuation (more results) of the previous view.

45 // Echoed in events, client checks if it is getting results for the latest request.

46 ViewID int64

48 // If set, this request and its view are canceled. A new view must be started.

49 Cancel bool

51 Query Query

52 Page Page

53}

55type ThreadMode string

57const (

58 ThreadOff ThreadMode = "off"

59 ThreadOn ThreadMode = "on"

60 ThreadUnread ThreadMode = "unread"

61)

63// Query is a request for messages that match filters, in a given order.

64type Query struct {

65 OrderAsc bool // Order by received ascending or desending.

66 Threading ThreadMode

67 Filter Filter

68 NotFilter NotFilter

69}

71// AttachmentType is for filtering by attachment type.

72type AttachmentType string

74const (

75 AttachmentIndifferent AttachmentType = ""

76 AttachmentNone AttachmentType = "none"

77 AttachmentAny AttachmentType = "any"

78 AttachmentImage AttachmentType = "image" // png, jpg, gif, ...

79 AttachmentPDF AttachmentType = "pdf"

80 AttachmentArchive AttachmentType = "archive" // zip files, tgz, ...

81 AttachmentSpreadsheet AttachmentType = "spreadsheet" // ods, xlsx, ...

82 AttachmentDocument AttachmentType = "document" // odt, docx, ...

83 AttachmentPresentation AttachmentType = "presentation" // odp, pptx, ...

84)

86// Filter selects the messages to return. Fields that are set must all match,

87// for slices each element by match ("and").

88type Filter struct {

89 // If -1, then all mailboxes except Trash/Junk/Rejects. Otherwise, only active if > 0.

90 MailboxID int64

92 // If true, also submailboxes are included in the search.

93 MailboxChildrenIncluded bool

95 // In case client doesn't know mailboxes and their IDs yet. Only used during sse

96 // connection setup, where it is turned into a MailboxID. Filtering only looks at

97 // MailboxID.

98 MailboxName string

100 Words []string // Case insensitive substring match for each string.

101 From []string

102 To []string // Including Cc and Bcc.

103 Oldest *time.Time

104 Newest *time.Time

105 Subject []string

106 Attachments AttachmentType

107 Labels []string

108 Headers [][2]string // Header values can be empty, it's a check if the header is present, regardless of value.

109 SizeMin int64

110 SizeMax int64

111}

112

113// NotFilter matches messages that don't match these fields.

114type NotFilter struct {

115 Words []string

116 From []string

117 To []string

118 Subject []string

119 Attachments AttachmentType

120 Labels []string

121}

122

123// Page holds pagination parameters for a request.

124type Page struct {

125 // Start returning messages after this ID, if > 0. For pagination, fetching the

126 // next set of messages.

127 AnchorMessageID int64

128

129 // Number of messages to return, must be >= 1, we never return more than 10000 for

130 // one request.

131 Count int

132

133 // If > 0, return messages until DestMessageID is found. More than Count messages

134 // can be returned. For long-running searches, it may take a while before this

135 // message if found.

136 DestMessageID int64

137}

138

139// todo: MessageAddress and MessageEnvelope into message.Address and message.Envelope.

140

141// MessageAddress is like message.Address, but with a dns.Domain, with unicode name

142// included.

143type MessageAddress struct {

144 Name string // Free-form name for display in mail applications.

145 User string // Localpart, encoded.

146 Domain dns.Domain

147}

148

149// MessageEnvelope is like message.Envelope, as used in message.Part, but including

150// unicode host names for IDNA names.

151type MessageEnvelope struct {

152 // todo: should get sherpadoc to understand type embeds and embed the non-MessageAddress fields from message.Envelope.

153 Date time.Time

154 Subject string

155 From []MessageAddress

156 Sender []MessageAddress

157 ReplyTo []MessageAddress

158 To []MessageAddress

159 CC []MessageAddress

160 BCC []MessageAddress

161 InReplyTo string

162 MessageID string

163}

164

165// MessageItem is sent by queries, it has derived information analyzed from

166// message.Part, made for the needs of the message items in the message list.

167// messages.

168type MessageItem struct {

169 Message store.Message // Without ParsedBuf and MsgPrefix, for size. With Preview, even if it isn't stored yet in the database.

170 Envelope MessageEnvelope

171 Attachments []Attachment

172 IsSigned bool

173 IsEncrypted bool

174 MatchQuery bool // If message does not match query, it can still be included because of threading.

175 MoreHeaders [][2]string // All headers from store.Settings.ShowHeaders that are present.

176}

177

178// ParsedMessage has more parsed/derived information about a message, intended

179// for rendering the (contents of the) message. Information from MessageItem is

180// not duplicated.

181type ParsedMessage struct {

182 ID int64

183 Part message.Part

184 Headers map[string][]string

185 ViewMode store.ViewMode

186

187 Texts []string // Contents of text parts, can be empty.

188

189 // Whether there is an HTML part. The webclient renders HTML message parts through

190 // an iframe and a separate request with strict CSP headers to prevent script

191 // execution and loading of external resources, which isn't possible when loading

192 // in iframe with inline HTML because not all browsers support the iframe csp

193 // attribute.

194 HasHTML bool

195

196 ListReplyAddress *MessageAddress // From List-Post.

197

198 TextPaths [][]int // Paths to text parts.

199 HTMLPath []int // Path to HTML part.

200

201 // Information used by MessageItem, not exported in this type.

202 envelope MessageEnvelope

203 attachments []Attachment

204 isSigned bool

205 isEncrypted bool

206}

207

208// EventStart is the first message sent on an SSE connection, giving the client

209// basic data to populate its UI. After this event, messages will follow quickly in

210// an EventViewMsgs event.

211type EventStart struct {

212 SSEID int64

213 LoginAddress MessageAddress

214 Addresses []MessageAddress

215 DomainAddressConfigs map[string]DomainAddressConfig // ASCII domain to address config.

216 MailboxName string

217 Mailboxes []store.Mailbox

218 Introbox string

219 RejectsMailbox string

220 Settings store.Settings

221 AccountPath string // If nonempty, the path on same host to webaccount interface.

222 Version string

223}

224

225// DomainAddressConfig has the address (localpart) configuration for a domain, so

226// the webmail client can decide if an address matches the addresses of the

227// account.

228type DomainAddressConfig struct {

229 LocalpartCatchallSeparators []string // Can be empty.

230 LocalpartCaseSensitive bool

231}

232

233// EventViewMsgs contains messages for a view, possibly a continuation of an

234// earlier list of messages.

235type EventViewMsgs struct {

236 ViewID int64

237 RequestID int64

238

239 // If empty, this was the last message for the request. If non-empty, a list of

240 // thread messages. Each with the first message being the reason this thread is

241 // included and can be used as AnchorID in followup requests. If the threading mode

242 // is "off" in the query, there will always be only a single message. If a thread

243 // is sent, all messages in the thread are sent, including those that don't match

244 // the query (e.g. from another mailbox). Threads can be displayed based on the

245 // ThreadParentIDs field, with possibly slightly different display based on field

246 // ThreadMissingLink.

247 MessageItems [][]MessageItem

248

249 // If set, will match the target page.DestMessageID from the request.

250 ParsedMessage *ParsedMessage

251

252 // If set, there are no more messages in this view at this moment. Messages can be

253 // added, typically via Change messages, e.g. for new deliveries.

254 ViewEnd bool

255}

256

257// EventViewErr indicates an error during a query for messages. The request is

258// aborted, no more request-related messages will be sent until the next request.

259type EventViewErr struct {

260 ViewID int64

261 RequestID int64

262 Err string // To be displayed in client.

263 err error // Original message, for checking against context.Canceled.

264}

265

266// EventViewReset indicates that a request for the next set of messages in a few

267// could not be fulfilled, e.g. because the anchor message does not exist anymore.

268// The client should clear its list of messages. This can happen before

269// EventViewMsgs events are sent.

270type EventViewReset struct {

271 ViewID int64

272 RequestID int64

273}

274

275// EventViewChanges contain one or more changes relevant for the client, either

276// with new mailbox total/unseen message counts, or messages added/removed/modified

277// (flags) for the current view.

278type EventViewChanges struct {

279 ViewID int64

280 Changes [][2]any // The first field of [2]any is a string, the second of the Change types below.

281}

282

283// ChangeMsgAdd adds a new message and possibly its thread to the view.

284type ChangeMsgAdd struct {

285 store.ChangeAddUID

286 MessageItems []MessageItem

287}

288

289// ChangeMsgRemove removes one or more messages from the view.

290type ChangeMsgRemove struct {

291 store.ChangeRemoveUIDs

292}

293

294// ChangeMsgFlags updates flags for one message.

295type ChangeMsgFlags struct {

296 store.ChangeFlags

297}

298

299// ChangeMsgThread updates muted/collapsed fields for one message.

300type ChangeMsgThread struct {

301 store.ChangeThread

302}

303

304// ChangeMailboxRemove indicates a mailbox was removed, including all its messages.

305type ChangeMailboxRemove struct {

306 store.ChangeRemoveMailbox

307}

308

309// ChangeMailboxAdd indicates a new mailbox was added, initially without any messages.

310type ChangeMailboxAdd struct {

311 Mailbox store.Mailbox

312}

313

314// ChangeMailboxRename indicates a mailbox was renamed. Its ID stays the same.

315// It could be under a new parent.

316type ChangeMailboxRename struct {

317 store.ChangeRenameMailbox

318}

319

320// ChangeMailboxCounts set new total and unseen message counts for a mailbox.

321type ChangeMailboxCounts struct {

322 store.ChangeMailboxCounts

323}

324

325// ChangeMailboxSpecialUse has updated special-use flags for a mailbox.

326type ChangeMailboxSpecialUse struct {

327 store.ChangeMailboxSpecialUse

328}

329

330// ChangeMailboxKeywords has an updated list of keywords for a mailbox, e.g. after

331// a message was added with a keyword that wasn't in the mailbox yet.

332type ChangeMailboxKeywords struct {

333 store.ChangeMailboxKeywords

334}

335

336// View holds the information about the returned data for a query. It is used to

337// determine whether mailbox changes should be sent to the client, we only send

338// addition/removal/flag-changes of messages that are in view, or would extend it

339// if the view is at the end of the results.

340type view struct {

341 Request Request

342

343 // Received of last message we sent to the client. We use it to decide if a newly

344 // delivered message is within the view and the client should get a notification.

345 LastMessageReceived time.Time

346

347 // If set, the last message in the query view has been sent. There is no need to do

348 // another query, it will not return more data. Used to decide if an event for a

349 // new message should be sent.

350 End bool

351

352 // Whether message must or must not match mailboxIDs.

353 matchMailboxIDs bool

354 // Mailboxes to match, can be multiple, for matching children. If empty, there is

355 // no filter on mailboxes.

356 mailboxIDs map[int64]bool

357

358 // Threads sent to client. New messages for this thread are also sent, regardless

359 // of regular query matching, so also for other mailboxes. If the user (re)moved

360 // all messages of a thread, they may still receive events for the thread. Only

361 // filled when query with threading not off.

362 threadIDs map[int64]struct{}

363}

364

365// sses tracks all sse connections, and access to them.

366var sses = struct {

367 sync.Mutex

368 gen int64

369 m map[int64]sse

370}{m: map[int64]sse{}}

371

372// sse represents an sse connection.

373type sse struct {

374 ID int64 // Also returned in EventStart and used in Request to identify the request.

375 AccountName string // Used to check the authenticated user has access to the SSE connection.

376 Request chan Request // Goroutine will receive requests from here, coming from API calls.

377}

378

379// called by the goroutine when the connection is closed or breaks.

380func (sse sse) unregister() {

381 sses.Lock()

382 defer sses.Unlock()

383 delete(sses.m, sse.ID)

384

385 // Drain any pending requests, preventing blocked goroutines from API calls.

386 for {

387 select {

388 case <-sse.Request:

389 default:

390 return

391 }

392 }

393}

394

395func sseRegister(accountName string) sse {

396 sses.Lock()

397 defer sses.Unlock()

398 sses.gen++

399 v := sse{sses.gen, accountName, make(chan Request, 1)}

400 sses.m[v.ID] = v

401 return v

402}

403

404// sseGet returns a reference to an existing connection if it exists and user

405// has access.

406func sseGet(id int64, accountName string) (sse, bool) {

407 sses.Lock()

408 defer sses.Unlock()

409 s := sses.m[id]

410 if s.AccountName != accountName {

411 return sse{}, false

412 }

413 return s, true

414}

415

416// ssetoken is a temporary token that has not yet been used to start an SSE

417// connection. Created by Token, consumed by a new SSE connection.

418type ssetoken struct {

419 token string // Uniquely generated.

420 accName string

421 address string // Address used to authenticate in call that created the token.

422 sessionToken store.SessionToken // SessionToken that created this token, checked before sending updates.

423 validUntil time.Time

424}

425

426// ssetokens maintains unused tokens. We have just one, but it's a type so we

427// can define methods.

428type ssetokens struct {

429 sync.Mutex

430 accountTokens map[string][]ssetoken // Account to max 10 most recent tokens, from old to new.

431 tokens map[string]ssetoken // Token to details, for finding account for a token.

432}

433

434var sseTokens = ssetokens{

435 accountTokens: map[string][]ssetoken{},

436 tokens: map[string]ssetoken{},

437}

438

439// xgenerate creates and saves a new token. It ensures no more than 10 tokens

440// per account exist, removing old ones if needed.

441func (x *ssetokens) xgenerate(ctx context.Context, accName, address string, sessionToken store.SessionToken) string {

442 var buf [16]byte

443 cryptrand.Read(buf[:])

444 st := ssetoken{base64.RawURLEncoding.EncodeToString(buf[:]), accName, address, sessionToken, time.Now().Add(time.Minute)}

445

446 x.Lock()

447 defer x.Unlock()

448 n := len(x.accountTokens[accName])

449 if n >= 10 {

450 for _, ost := range x.accountTokens[accName][:n-9] {

451 delete(x.tokens, ost.token)

452 }

453 copy(x.accountTokens[accName], x.accountTokens[accName][n-9:])

454 x.accountTokens[accName] = x.accountTokens[accName][:9]

455 }

456 x.accountTokens[accName] = append(x.accountTokens[accName], st)

457 x.tokens[st.token] = st

458 return st.token

459}

460

461// check verifies a token, and consumes it if valid.

462func (x *ssetokens) check(token string) (string, string, store.SessionToken, bool, error) {

463 x.Lock()

464 defer x.Unlock()

465

466 st, ok := x.tokens[token]

467 if !ok {

468 return "", "", "", false, nil

469 }

470 delete(x.tokens, token)

471 if i := slices.Index(x.accountTokens[st.accName], st); i < 0 {

472 return "", "", "", false, errors.New("internal error, could not find token in account")

473 } else {

474 copy(x.accountTokens[st.accName][i:], x.accountTokens[st.accName][i+1:])

475 x.accountTokens[st.accName] = x.accountTokens[st.accName][:len(x.accountTokens[st.accName])-1]

476 if len(x.accountTokens[st.accName]) == 0 {

477 delete(x.accountTokens, st.accName)

478 }

479 }

480 if time.Now().After(st.validUntil) {

481 return "", "", "", false, nil

482 }

483 return st.accName, st.address, st.sessionToken, true, nil

484}

485

486// ioErr is panicked on i/o errors in serveEvents and handled in a defer.

487type ioErr struct {

488 err error

489}

490

491// ensure we have a non-nil moreHeaders, taking it from Settings.

492func ensureMoreHeaders(tx *bstore.Tx, moreHeaders []string) ([]string, error) {

493 if moreHeaders != nil {

494 return moreHeaders, nil

495 }

496

497 s := store.Settings{ID: 1}

498 if err := tx.Get(&s); err != nil {

499 return nil, fmt.Errorf("get settings: %v", err)

500 }

501 moreHeaders = s.ShowHeaders

502 if moreHeaders == nil {

503 moreHeaders = []string{} // Ensure we won't get Settings again next call.

504 }

505 return moreHeaders, nil

506}

507

508// serveEvents serves an SSE connection. Authentication is done through a query

509// string parameter "singleUseToken", a one-time-use token returned by the Token

510// API call.

511func serveEvents(ctx context.Context, log mlog.Log, accountPath string, w http.ResponseWriter, r *http.Request) {

512 if r.Method != "GET" {

513 http.Error(w, "405 - method not allowed - use get", http.StatusMethodNotAllowed)

514 return

515 }

516

517 flusher, ok := w.(http.Flusher)

518 if !ok {

519 log.Error("internal error: ResponseWriter not a http.Flusher")

520 http.Error(w, "500 - internal error - cannot sync to http connection", 500)

521 return

522 }

523

524 q := r.URL.Query()

525 token := q.Get("singleUseToken")

526 if token == "" {

527 http.Error(w, "400 - bad request - missing credentials", http.StatusBadRequest)

528 return

529 }

530 accName, address, sessionToken, ok, err := sseTokens.check(token)

531 if err != nil {

532 http.Error(w, "500 - internal server error - "+err.Error(), http.StatusInternalServerError)

533 return

534 }

535 if !ok {

536 http.Error(w, "400 - bad request - bad token", http.StatusBadRequest)

537 return

538 }

539 if _, err := store.SessionUse(ctx, log, accName, sessionToken, ""); err != nil {

540 http.Error(w, "400 - bad request - bad session token", http.StatusBadRequest)

541 return

542 }

543

544 // We can simulate a slow SSE connection. It seems firefox doesn't slow down

545 // incoming responses with its slow-network similation.

546 var waitMin, waitMax time.Duration

547 waitMinMsec := q.Get("waitMinMsec")

548 waitMaxMsec := q.Get("waitMaxMsec")

549 if waitMinMsec != "" && waitMaxMsec != "" {

550 if v, err := strconv.ParseInt(waitMinMsec, 10, 64); err != nil {

551 http.Error(w, "400 - bad request - parsing waitMinMsec: "+err.Error(), http.StatusBadRequest)

552 return

553 } else {

554 waitMin = time.Duration(v) * time.Millisecond

555 }

556

557 if v, err := strconv.ParseInt(waitMaxMsec, 10, 64); err != nil {

558 http.Error(w, "400 - bad request - parsing waitMaxMsec: "+err.Error(), http.StatusBadRequest)

559 return

560 } else {

561 waitMax = time.Duration(v) * time.Millisecond

562 }

563 }

564

565 // Parse the request with initial mailbox/search criteria.

566 var req Request

567 dec := json.NewDecoder(strings.NewReader(q.Get("request")))

568 dec.DisallowUnknownFields()

569 if err := dec.Decode(&req); err != nil {

570 http.Error(w, "400 - bad request - bad request query string parameter: "+err.Error(), http.StatusBadRequest)

571 return

572 } else if req.Page.Count <= 0 {

573 http.Error(w, "400 - bad request - request cannot have Page.Count 0", http.StatusBadRequest)

574 return

575 }

576 if req.Query.Threading == "" {

577 req.Query.Threading = ThreadOff

578 }

579

580 var writer *eventWriter

581

582 metricSSEConnections.Inc()

583 defer metricSSEConnections.Dec()

584

585 // Below here, error handling cause through xcheckf, which panics with

586 // *sherpa.Error, after which we send an error event to the client. We can also get

587 // an *ioErr when the connection is broken.

588 defer func() {

589 x := recover()

590 if x == nil {

591 return

592 }

593 if err, ok := x.(*sherpa.Error); ok {

594 writer.xsendEvent(ctx, log, "fatalErr", err.Message)

595 } else if _, ok := x.(ioErr); ok {

596 return

597 } else {

598 log.WithContext(ctx).Error("serveEvents panic", slog.Any("err", x))

599 debug.PrintStack()

600 metrics.PanicInc(metrics.Webmail)

601 panic(x)

602 }

603 }()

604

605 h := w.Header()

606 h.Set("Content-Type", "text/event-stream")

607 h.Set("Cache-Control", "no-cache")

608

609 // We'll be sending quite a bit of message data (text) in JSON (plenty duplicate

610 // keys), so should be quite compressible.

611 var out writeFlusher

612 gz := mox.AcceptsGzip(r)

613 if gz {

614 h.Set("Content-Encoding", "gzip")

615 out, _ = gzip.NewWriterLevel(w, gzip.BestSpeed)

616 } else {

617 out = nopFlusher{w}

618 }

619 out = httpFlusher{out, flusher}

620

621 // We'll be writing outgoing SSE events through writer.

622 writer = newEventWriter(out, waitMin, waitMax, accName, sessionToken)

623 defer writer.close()

624

625 // Fetch initial data.

626 acc, err := store.OpenAccount(log, accName, true)

627 xcheckf(ctx, err, "open account")

628 defer func() {

629 err := acc.Close()

630 log.Check(err, "closing account")

631 }()

632 comm := store.RegisterComm(acc)

633 defer comm.Unregister()

634

635 // List addresses that the client can use to send email from.

636 accConf, _ := acc.Conf()

637 loginAddr, err := smtp.ParseAddress(address)

638 xcheckf(ctx, err, "parsing login address")

639 _, _, _, dest, err := mox.LookupAddress(loginAddr.Localpart, loginAddr.Domain, false, false, false)

640 xcheckf(ctx, err, "looking up destination for login address")

641 loginName := accConf.FullName

642 if dest.FullName != "" {

643 loginName = dest.FullName

644 }

645 loginAddress := MessageAddress{Name: loginName, User: loginAddr.Localpart.String(), Domain: loginAddr.Domain}

646 var addresses []MessageAddress

647 for a, dest := range accConf.Destinations {

648 name := dest.FullName

649 if name == "" {

650 name = accConf.FullName

651 }

652 var ma MessageAddress

653 if strings.HasPrefix(a, "@") {

654 dom, err := dns.ParseDomain(a[1:])

655 xcheckf(ctx, err, "parsing destination address for account")

656 ma = MessageAddress{Domain: dom}

657 } else {

658 addr, err := smtp.ParseAddress(a)

659 xcheckf(ctx, err, "parsing destination address for account")

660 ma = MessageAddress{Name: name, User: addr.Localpart.String(), Domain: addr.Domain}

661 }

662 addresses = append(addresses, ma)

663 }

664 // User is allowed to send using alias address as message From address. Webmail

665 // will choose it when replying to a message sent to that address.

666 aliasAddrs := map[MessageAddress]bool{}

667 for _, a := range accConf.Aliases {

668 if a.Alias.AllowMsgFrom {

669 ma := MessageAddress{User: a.Alias.LocalpartStr, Domain: a.Alias.Domain}

670 if !aliasAddrs[ma] {

671 addresses = append(addresses, ma)

672 }

673 aliasAddrs[ma] = true

674 }

675 }

676

677 // We implicitly start a query. We use the reqctx for the transaction, because the

678 // transaction is passed to the query, which can be canceled.

679 reqctx, reqctxcancel := context.WithCancel(ctx)

680 defer func() {

681 // We also cancel in cancelDrain later on, but there is a brief window where the

682 // context wouldn't be canceled.

683 if reqctxcancel != nil {

684 reqctxcancel()

685 reqctxcancel = nil

686 }

687 }()

688

689 // qtx is kept around during connection initialization, until we pass it off to the

690 // goroutine that starts querying for messages.

691 var qtx *bstore.Tx

692 defer func() {

693 if qtx != nil {

694 err := qtx.Rollback()

695 log.Check(err, "rolling back")

696 }

697 }()

698

699 var mbl []store.Mailbox

700 settings := store.Settings{ID: 1}

701

702 // We only take the rlock when getting the tx.

703 acc.WithRLock(func() {

704 // Now a read-only transaction we'll use during the query.

705 qtx, err = acc.DB.Begin(reqctx, false)

706 xcheckf(ctx, err, "begin transaction")

707

708 mbl, err = bstore.QueryTx[store.Mailbox](qtx).FilterEqual("Expunged", false).List()

709 xcheckf(ctx, err, "list mailboxes")

710

711 err = qtx.Get(&settings)

712 xcheckf(ctx, err, "get settings")

713 })

714

715 // Find the designated mailbox if a mailbox name is set, or there are no filters at all.

716 var zerofilter Filter

717 var zeronotfilter NotFilter

718 var mailbox store.Mailbox

719 var mailboxPrefixes []string

720 var matchMailboxes bool

721 mailboxIDs := map[int64]bool{}

722 mailboxName := req.Query.Filter.MailboxName

723 if mailboxName != "" || reflect.DeepEqual(req.Query.Filter, zerofilter) && reflect.DeepEqual(req.Query.NotFilter, zeronotfilter) {

724 if mailboxName == "" {

725 mailboxName = "Inbox"

726 }

727

728 var inbox store.Mailbox

729 for _, e := range mbl {

730 if e.Name == mailboxName {

731 mailbox = e

732 }

733 if e.Name == "Inbox" {

734 inbox = e

735 }

736 }

737 if mailbox.ID == 0 {

738 mailbox = inbox

739 }

740 if mailbox.ID == 0 {

741 xcheckf(ctx, errors.New("inbox not found"), "setting initial mailbox")

742 }

743 req.Query.Filter.MailboxID = mailbox.ID

744 req.Query.Filter.MailboxName = ""

745 mailboxPrefixes = []string{mailbox.Name + "/"}

746 matchMailboxes = true

747 mailboxIDs[mailbox.ID] = true

748 } else {

749 matchMailboxes, mailboxIDs, mailboxPrefixes = xprepareMailboxIDs(ctx, qtx, req.Query.Filter, accConf.RejectsMailbox)

750 }

751 if req.Query.Filter.MailboxChildrenIncluded {

752 xgatherMailboxIDs(ctx, qtx, mailboxIDs, mailboxPrefixes)

753 }

754

755 // todo: write a last-event-id based on modseq? if last-event-id is present, we would have to send changes to mailboxes, messages, hopefully reducing the amount of data sent.

756

757 sse := sseRegister(acc.Name)

758 defer sse.unregister()

759

760 // Per-domain localpart config so webclient can decide if an address belongs to the account.

761 domainAddressConfigs := map[string]DomainAddressConfig{}

762 for _, a := range addresses {

763 dom, _ := mox.Conf.Domain(a.Domain)

764 domainAddressConfigs[a.Domain.ASCII] = DomainAddressConfig{dom.LocalpartCatchallSeparatorsEffective, dom.LocalpartCaseSensitive}

765 }

766

767 // Write first event, allowing client to fill its UI with mailboxes.

768 start := EventStart{

769 SSEID: sse.ID,

770 LoginAddress: loginAddress,

771 Addresses: addresses,

772 DomainAddressConfigs: domainAddressConfigs,

773 MailboxName: mailbox.Name,

774 Mailboxes: mbl,

775 Introbox: accConf.Introbox,

776 RejectsMailbox: accConf.RejectsMailbox,

777 Settings: settings,

778 AccountPath: accountPath,

779 Version: moxvar.Version,

780 }

781 writer.xsendEvent(ctx, log, "start", start)

782

783 // The goroutine doing the querying will send messages on these channels, which

784 // result in an event being written on the SSE connection.

785 viewMsgsc := make(chan EventViewMsgs)

786 viewErrc := make(chan EventViewErr)

787 viewResetc := make(chan EventViewReset)

788 donec := make(chan int64) // When request is done.

789

790 // Start a view, it determines if we send a change to the client. And start an

791 // implicit query for messages, we'll send the messages to the client which can

792 // fill its ui with messages.

793 v := view{req, time.Time{}, false, matchMailboxes, mailboxIDs, map[int64]struct{}{}}

794 go viewRequestTx(reqctx, log, acc, qtx, v, viewMsgsc, viewErrc, viewResetc, donec)

795 qtx = nil // viewRequestTx closes qtx

796

797 // When canceling a query, we must drain its messages until it says it is done.

798 // Otherwise the sending goroutine would hang indefinitely on a channel send.

799 cancelDrain := func() {

800 if reqctxcancel != nil {

801 // Cancel the goroutine doing the querying.

802 reqctxcancel()

803 reqctx = nil

804 reqctxcancel = nil

805 } else {

806 return

807 }

808

809 // Drain events until done.

810 for {

811 select {

812 case <-viewMsgsc:

813 case <-viewErrc:

814 case <-viewResetc:

815 case <-donec:

816 return

817 }

818 }

819 }

820

821 // If we stop and a query is in progress, we must drain the channel it will send on.

822 defer cancelDrain()

823

824 // Changes broadcasted by other connections on this account. If applicable for the

825 // connection/view, we send events.

826 xprocessChanges := func(changes []store.Change) {

827 taggedChanges := [][2]any{}

828

829 newPreviews := map[int64]string{}

830 defer storeNewPreviews(ctx, log, acc, newPreviews)

831

832 // We get a transaction first time we need it.

833 var xtx *bstore.Tx

834 defer func() {

835 if xtx != nil {

836 err := xtx.Rollback()

837 log.Check(err, "rolling back transaction")

838 }

839 }()

840 ensureTx := func() error {

841 if xtx != nil {

842 return nil

843 }

844 acc.RLock()

845 defer acc.RUnlock()

846 var err error

847 xtx, err = acc.DB.Begin(ctx, false)

848 return err

849 }

850 // This getmsg will now only be called mailboxID+UID, not with messageID set.

851 // todo jmap: change store.Change* to include MessageID's? would mean duplication of information resulting in possible mismatch.

852 getmsg := func(messageID int64, mailboxID int64, uid store.UID) (store.Message, error) {

853 if err := ensureTx(); err != nil {

854 return store.Message{}, fmt.Errorf("transaction: %v", err)

855 }

856 return bstore.QueryTx[store.Message](xtx).FilterEqual("Expunged", false).FilterNonzero(store.Message{MailboxID: mailboxID, UID: uid}).Get()

857 }

858

859 // Additional headers from settings to add to MessageItems.

860 var moreHeaders []string

861 xmoreHeaders := func() []string {

862 err := ensureTx()

863 xcheckf(ctx, err, "transaction")

864

865 moreHeaders, err = ensureMoreHeaders(xtx, moreHeaders)

866 xcheckf(ctx, err, "ensuring more headers")

867 return moreHeaders

868 }

869

870 // Return uids that are within range in view. Because the end has been reached, or

871 // because the UID is not after the last message.

872 xchangedUIDs := func(mailboxID int64, uids []store.UID, isRemove bool) (changedUIDs []store.UID) {

873 uidsAny := make([]any, len(uids))

874 for i, uid := range uids {

875 uidsAny[i] = uid

876 }

877 err := ensureTx()

878 xcheckf(ctx, err, "transaction")

879 q := bstore.QueryTx[store.Message](xtx)

880 q.FilterNonzero(store.Message{MailboxID: mailboxID})

881 q.FilterEqual("UID", uidsAny...)

882 mbOK := v.matchesMailbox(mailboxID)

883 err = q.ForEach(func(m store.Message) error {

884 _, thread := v.threadIDs[m.ThreadID]

885 if thread || mbOK && (v.inRange(m) || isRemove && m.Expunged) {

886 changedUIDs = append(changedUIDs, m.UID)

887 }

888 return nil

889 })

890 xcheckf(ctx, err, "fetching messages for change")

891 return changedUIDs

892 }

893

894 // Forward changes that are relevant to the current view.

895 for _, change := range changes {

896 switch c := change.(type) {

897 case store.ChangeAddUID:

898 ok, err := v.matches(log, acc, true, 0, c.MailboxID, c.UID, c.Flags, c.Keywords, getmsg)

899 xcheckf(ctx, err, "matching new message against view")

900 m, err := getmsg(0, c.MailboxID, c.UID)

901 xcheckf(ctx, err, "get message")

902 _, thread := v.threadIDs[m.ThreadID]

903 if !ok && !thread {

904 continue

905 }

906

907 state := msgState{acc: acc, log: log, newPreviews: newPreviews}

908 mi, err := messageItem(log, m, &state, xmoreHeaders())

909 state.clear()

910 xcheckf(ctx, err, "make messageitem")

911 mi.MatchQuery = ok

912

913 mil := []MessageItem{mi}

914 if !thread && req.Query.Threading != ThreadOff {

915 err := ensureTx()

916 xcheckf(ctx, err, "transaction")

917 more, _, err := gatherThread(log, xtx, acc, v, m, 0, false, xmoreHeaders(), newPreviews)

918 xcheckf(ctx, err, "gathering thread messages for id %d, thread %d", m.ID, m.ThreadID)

919 mil = append(mil, more...)

920 v.threadIDs[m.ThreadID] = struct{}{}

921 }

922

923 taggedChanges = append(taggedChanges, [2]any{"ChangeMsgAdd", ChangeMsgAdd{c, mil}})

924

925 // If message extends the view, store it as such.

926 if !v.Request.Query.OrderAsc && m.Received.Before(v.LastMessageReceived) || v.Request.Query.OrderAsc && m.Received.After(v.LastMessageReceived) {

927 v.LastMessageReceived = m.Received

928 }

929

930 case store.ChangeRemoveUIDs:

931 comm.RemovalSeen(c)

932

933 // We may send changes for uids the client doesn't know, that's fine.

934 changedUIDs := xchangedUIDs(c.MailboxID, c.UIDs, true)

935 if len(changedUIDs) == 0 {

936 continue

937 }

938 ch := ChangeMsgRemove{c}

939 ch.UIDs = changedUIDs

940 taggedChanges = append(taggedChanges, [2]any{"ChangeMsgRemove", ch})

941

942 case store.ChangeFlags:

943 // We may send changes for uids the client doesn't know, that's fine.

944 changedUIDs := xchangedUIDs(c.MailboxID, []store.UID{c.UID}, false)

945 if len(changedUIDs) == 0 {

946 continue

947 }

948 ch := ChangeMsgFlags{c}

949 ch.UID = changedUIDs[0]

950 taggedChanges = append(taggedChanges, [2]any{"ChangeMsgFlags", ch})

951

952 case store.ChangeThread:

953 // Change in muted/collaped state, just always ship it.

954 taggedChanges = append(taggedChanges, [2]any{"ChangeMsgThread", ChangeMsgThread{c}})

955

956 case store.ChangeRemoveMailbox:

957 taggedChanges = append(taggedChanges, [2]any{"ChangeMailboxRemove", ChangeMailboxRemove{c}})

958

959 case store.ChangeAddMailbox:

960 taggedChanges = append(taggedChanges, [2]any{"ChangeMailboxAdd", ChangeMailboxAdd{c.Mailbox}})

961

962 case store.ChangeRenameMailbox:

963 taggedChanges = append(taggedChanges, [2]any{"ChangeMailboxRename", ChangeMailboxRename{c}})

964

965 case store.ChangeMailboxCounts:

966 taggedChanges = append(taggedChanges, [2]any{"ChangeMailboxCounts", ChangeMailboxCounts{c}})

967

968 case store.ChangeMailboxSpecialUse:

969 taggedChanges = append(taggedChanges, [2]any{"ChangeMailboxSpecialUse", ChangeMailboxSpecialUse{c}})

970

971 case store.ChangeMailboxKeywords:

972 taggedChanges = append(taggedChanges, [2]any{"ChangeMailboxKeywords", ChangeMailboxKeywords{c}})

973

974 case store.ChangeAddSubscription, store.ChangeRemoveSubscription:

975 // Webmail does not care about subscriptions.

976

977 case store.ChangeAnnotation:

978 // Nothing.

979

980 default:

981 panic(fmt.Sprintf("missing case for change %T", c))

982 }

983 }

984

985 if len(taggedChanges) > 0 {

986 viewChanges := EventViewChanges{v.Request.ViewID, taggedChanges}

987 writer.xsendEvent(ctx, log, "viewChanges", viewChanges)

988 }

989 }

990

991 timer := time.NewTimer(5 * time.Minute) // For keepalives.

992 defer timer.Stop()

993 for {

994 if writer.wrote {

995 timer.Reset(5 * time.Minute)

996 writer.wrote = false

997 }

998

999 pending := comm.Pending

1000 if reqctx != nil {

1001 pending = nil

1002 }

1003

1004 select {

1005 case <-mox.Shutdown.Done():

1006 writer.xsendEvent(ctx, log, "serverShutdown", "server is shutting down")

1007 // Work around go vet, it doesn't see defer cancelDrain.

1008 if reqctxcancel != nil {

1009 reqctxcancel()

1010 }

1011 return

1012

1013 case <-timer.C:

1014 _, err := fmt.Fprintf(out, ": keepalive\n\n")

1015 if err == nil {

1016 err = out.Flush()

1017 }

1018 if err != nil {

1019 log.Errorx("write keepalive", err)

1020 // Work around go vet, it doesn't see defer cancelDrain.

1021 if reqctxcancel != nil {

1022 reqctxcancel()

1023 }

1024 return

1025 }

1026 writer.wrote = true

1027

1028 case vm := <-viewMsgsc:

1029 if vm.RequestID != v.Request.ID || vm.ViewID != v.Request.ViewID {

1030 panic(fmt.Sprintf("received msgs for view,request id %d,%d instead of %d,%d", vm.ViewID, vm.RequestID, v.Request.ViewID, v.Request.ID))

1031 }

1032 if vm.ViewEnd {

1033 v.End = true

1034 }

1035 if len(vm.MessageItems) > 0 {

1036 v.LastMessageReceived = vm.MessageItems[len(vm.MessageItems)-1][0].Message.Received

1037 }

1038 writer.xsendEvent(ctx, log, "viewMsgs", vm)

1039

1040 case ve := <-viewErrc:

1041 if ve.RequestID != v.Request.ID || ve.ViewID != v.Request.ViewID {

1042 panic(fmt.Sprintf("received err for view,request id %d,%d instead of %d,%d", ve.ViewID, ve.RequestID, v.Request.ViewID, v.Request.ID))

1043 }

1044 if errors.Is(ve.err, context.Canceled) || mlog.IsClosed(ve.err) {

1045 // Work around go vet, it doesn't see defer cancelDrain.

1046 if reqctxcancel != nil {

1047 reqctxcancel()

1048 }

1049 return

1050 }

1051 writer.xsendEvent(ctx, log, "viewErr", ve)

1052

1053 case vr := <-viewResetc:

1054 if vr.RequestID != v.Request.ID || vr.ViewID != v.Request.ViewID {

1055 panic(fmt.Sprintf("received reset for view,request id %d,%d instead of %d,%d", vr.ViewID, vr.RequestID, v.Request.ViewID, v.Request.ID))

1056 }

1057 writer.xsendEvent(ctx, log, "viewReset", vr)

1058

1059 case id := <-donec:

1060 if id != v.Request.ID {

1061 panic(fmt.Sprintf("received done for request id %d instead of %d", id, v.Request.ID))

1062 }

1063 if reqctxcancel != nil {

1064 reqctxcancel()

1065 }

1066 reqctx = nil

1067 reqctxcancel = nil

1068

1069 case req := <-sse.Request:

1070 if reqctx != nil {

1071 cancelDrain()

1072 }

1073 if req.Cancel {

1074 v = view{req, time.Time{}, false, false, nil, nil}

1075 continue

1076 }

1077

1078 reqctx, reqctxcancel = context.WithCancel(ctx)

1079

1080 stop := func() (stop bool) {

1081 // rtx is handed off viewRequestTx below, but we must clean it up in case of errors.

1082 var rtx *bstore.Tx

1083 var err error

1084 defer func() {

1085 if rtx != nil {

1086 err = rtx.Rollback()

1087 log.Check(err, "rolling back transaction")

1088 }

1089 }()

1090 acc.WithRLock(func() {

1091 rtx, err = acc.DB.Begin(reqctx, false)

1092 })

1093 if err != nil {

1094 reqctxcancel()

1095 reqctx = nil

1096 reqctxcancel = nil

1097

1098 if errors.Is(err, context.Canceled) {

1099 return true

1100 }

1101 err := fmt.Errorf("begin transaction: %v", err)

1102 viewErr := EventViewErr{v.Request.ViewID, v.Request.ID, err.Error(), err}

1103 writer.xsendEvent(ctx, log, "viewErr", viewErr)

1104 return false

1105 }

1106

1107 // Reset view state for new query.

1108 if req.ViewID != v.Request.ViewID {

1109 matchMailboxes, mailboxIDs, mailboxPrefixes := xprepareMailboxIDs(ctx, rtx, req.Query.Filter, accConf.RejectsMailbox)

1110 if req.Query.Filter.MailboxChildrenIncluded {

1111 xgatherMailboxIDs(ctx, rtx, mailboxIDs, mailboxPrefixes)

1112 }

1113 v = view{req, time.Time{}, false, matchMailboxes, mailboxIDs, map[int64]struct{}{}}

1114 } else {

1115 v.Request = req

1116 }

1117 go viewRequestTx(reqctx, log, acc, rtx, v, viewMsgsc, viewErrc, viewResetc, donec)

1118 rtx = nil

1119 return false

1120 }()

1121 if stop {

1122 return

1123 }

1124

1125 case <-pending:

1126 overflow, changes := comm.Get()

1127 if overflow {

1128 writer.xsendEvent(ctx, log, "fatalErr", "out of sync, too many pending changes")

1129 return

1130 }

1131 xprocessChanges(changes)

1132

1133 case <-ctx.Done():

1134 // Work around go vet, it doesn't see defer cancelDrain.

1135 if reqctxcancel != nil {

1136 reqctxcancel()

1137 }

1138 return

1139 }

1140 }

1141}

1142

1143// xprepareMailboxIDs prepare the first half of filters for mailboxes, based on

1144// f.MailboxID (-1 is special). matchMailboxes indicates whether the IDs in

1145// mailboxIDs must or must not match. mailboxPrefixes is for use with

1146// xgatherMailboxIDs to gather children of the mailboxIDs.

1147func xprepareMailboxIDs(ctx context.Context, tx *bstore.Tx, f Filter, rejectsMailbox string) (matchMailboxes bool, mailboxIDs map[int64]bool, mailboxPrefixes []string) {

1148 matchMailboxes = true

1149 mailboxIDs = map[int64]bool{}

1150 if f.MailboxID == -1 {

1151 matchMailboxes = false

1152 // Add the trash, junk and account rejects mailbox.

1153 err := bstore.QueryTx[store.Mailbox](tx).FilterEqual("Expunged", false).ForEach(func(mb store.Mailbox) error {

1154 if mb.Trash || mb.Junk || mb.Name == rejectsMailbox {

1155 mailboxPrefixes = append(mailboxPrefixes, mb.Name+"/")

1156 mailboxIDs[mb.ID] = true

1157 }

1158 return nil

1159 })

1160 xcheckf(ctx, err, "finding trash/junk/rejects mailbox")

1161 } else if f.MailboxID > 0 {

1162 mb, err := store.MailboxID(tx, f.MailboxID)

1163 xcheckf(ctx, err, "get mailbox")

1164 mailboxIDs[f.MailboxID] = true

1165 mailboxPrefixes = []string{mb.Name + "/"}

1166 }

1167 return

1168}

1169

1170// xgatherMailboxIDs adds all mailboxes with a prefix matching any of

1171// mailboxPrefixes to mailboxIDs, to expand filtering to children of mailboxes.

1172func xgatherMailboxIDs(ctx context.Context, tx *bstore.Tx, mailboxIDs map[int64]bool, mailboxPrefixes []string) {

1173 // Gather more mailboxes to filter on, based on mailboxPrefixes.

1174 if len(mailboxPrefixes) == 0 {

1175 return

1176 }

1177 err := bstore.QueryTx[store.Mailbox](tx).FilterEqual("Expunged", false).ForEach(func(mb store.Mailbox) error {

1178 for _, p := range mailboxPrefixes {

1179 if strings.HasPrefix(mb.Name, p) {

1180 mailboxIDs[mb.ID] = true

1181 break

1182 }

1183 }

1184 return nil

1185 })

1186 xcheckf(ctx, err, "gathering mailboxes")

1187}

1188

1189// matchesMailbox returns whether a mailbox matches the view.

1190func (v view) matchesMailbox(mailboxID int64) bool {

1191 return len(v.mailboxIDs) == 0 || v.matchMailboxIDs && v.mailboxIDs[mailboxID] || !v.matchMailboxIDs && !v.mailboxIDs[mailboxID]

1192}

1193

1194// inRange returns whether m is within the range for the view, whether a change for

1195// this message should be sent to the client so it can update its state.

1196func (v view) inRange(m store.Message) bool {

1197 return v.End || !v.Request.Query.OrderAsc && !m.Received.Before(v.LastMessageReceived) || v.Request.Query.OrderAsc && !m.Received.After(v.LastMessageReceived)

1198}

1199

1200// matches checks if the message, identified by either messageID or mailboxID+UID,

1201// is in the current "view" (i.e. passing the filters, and if checkRange is set

1202// also if within the range of sent messages based on sort order and the last seen

1203// message). getmsg retrieves the message, which may be necessary depending on the

1204// active filters. Used to determine if a store.Change with a new message should be

1205// sent, and for the destination and anchor messages in view requests.

1206func (v view) matches(log mlog.Log, acc *store.Account, checkRange bool, messageID int64, mailboxID int64, uid store.UID, flags store.Flags, keywords []string, getmsg func(int64, int64, store.UID) (store.Message, error)) (match bool, rerr error) {

1207 var m store.Message

1208 ensureMessage := func() bool {

1209 if m.ID == 0 && rerr == nil {

1210 m, rerr = getmsg(messageID, mailboxID, uid)

1211 }

1212 return rerr == nil

1213 }

1214

1215 q := v.Request.Query

1216

1217 // Warning: Filters must be kept in sync between queryMessage and view.matches.

1218

1219 // Check filters.

1220 if len(v.mailboxIDs) > 0 && (!ensureMessage() || v.matchMailboxIDs && !v.mailboxIDs[m.MailboxID] || !v.matchMailboxIDs && v.mailboxIDs[m.MailboxID]) {

1221 return false, rerr

1222 }

1223 // note: anchorMessageID is not relevant for matching.

1224 flagfilter := q.flagFilterFn()

1225 if flagfilter != nil && !flagfilter(flags, keywords) {

1226 return false, rerr

1227 }

1228

1229 if q.Filter.Oldest != nil && (!ensureMessage() || m.Received.Before(*q.Filter.Oldest)) {

1230 return false, rerr

1231 }

1232 if q.Filter.Newest != nil && (!ensureMessage() || !m.Received.Before(*q.Filter.Newest)) {

1233 return false, rerr

1234 }

1235

1236 if q.Filter.SizeMin > 0 && (!ensureMessage() || m.Size < q.Filter.SizeMin) {

1237 return false, rerr

1238 }

1239 if q.Filter.SizeMax > 0 && (!ensureMessage() || m.Size > q.Filter.SizeMax) {

1240 return false, rerr

1241 }

1242

1243 state := msgState{acc: acc, log: log}

1244 defer func() {

1245 if rerr == nil && state.err != nil {

1246 rerr = state.err

1247 }

1248 state.clear()

1249 }()

1250

1251 attachmentFilter := q.attachmentFilterFn(log, acc, &state)

1252 if attachmentFilter != nil && (!ensureMessage() || !attachmentFilter(m)) {

1253 return false, rerr

1254 }

1255

1256 envFilter := q.envFilterFn(log, &state)

1257 if envFilter != nil && (!ensureMessage() || !envFilter(m)) {

1258 return false, rerr

1259 }

1260

1261 headerFilter := q.headerFilterFn(log, &state)

1262 if headerFilter != nil && (!ensureMessage() || !headerFilter(m)) {

1263 return false, rerr

1264 }

1265

1266 wordsFilter := q.wordsFilterFn(log, &state)

1267 if wordsFilter != nil && (!ensureMessage() || !wordsFilter(m)) {

1268 return false, rerr

1269 }

1270

1271 // Now check that we are either within the sorting order, or "last" was sent.

1272 if !checkRange || v.End || ensureMessage() && v.inRange(m) {

1273 return true, rerr

1274 }

1275 return false, rerr

1276}

1277

1278type msgResp struct {

1279 err error // If set, an error happened and fields below are not set.

1280 reset bool // If set, the anchor message does not exist (anymore?) and we are sending messages from the start, fields below not set.

1281 viewEnd bool // If set, the last message for the view was seen, no more should be requested, fields below not set.

1282 mil []MessageItem // If none of the cases above apply, the messages that was found matching the query. First message was reason the thread is returned, for use as AnchorID in followup request.

1283 pm *ParsedMessage // If m was the target page.DestMessageID, or this is the first match, this is the parsed message of mi.

1284}

1285

1286func storeNewPreviews(ctx context.Context, log mlog.Log, acc *store.Account, newPreviews map[int64]string) {

1287 if len(newPreviews) == 0 {

1288 return

1289 }

1290

1291 defer func() {

1292 x := recover()

1293 if x != nil {

1294 log.Error("unhandled panic in storeNewPreviews", slog.Any("err", x))

1295 debug.PrintStack()

1296 metrics.PanicInc(metrics.Store)

1297 }

1298 }()

1299

1300 err := acc.DB.Write(ctx, func(tx *bstore.Tx) error {

1301 for id, preview := range newPreviews {

1302 m := store.Message{ID: id}

1303 if err := tx.Get(&m); err != nil {

1304 return fmt.Errorf("get message with id %d to store preview: %w", id, err)

1305 } else if !m.Expunged {

1306 m.Preview = &preview

1307 if err := tx.Update(&m); err != nil {

1308 return fmt.Errorf("updating message with id %d: %v", m.ID, err)

1309 }

1310 }

1311 }

1312 return nil

1313 })

1314 log.Check(err, "saving new previews with messages")

1315}

1316

1317// viewRequestTx executes a request (query with filters, pagination) by

1318// launching a new goroutine with queryMessages, receiving results as msgResp,

1319// and sending Event* to the SSE connection.

1320//

1321// It always closes tx.

1322func viewRequestTx(ctx context.Context, log mlog.Log, acc *store.Account, tx *bstore.Tx, v view, msgc chan EventViewMsgs, errc chan EventViewErr, resetc chan EventViewReset, donec chan int64) {

1323 // Newly generated previews which we'll save when the operation is done.

1324 newPreviews := map[int64]string{}

1325

1326 defer func() {

1327 err := tx.Rollback()

1328 log.Check(err, "rolling back query transaction")

1329

1330 donec <- v.Request.ID

1331

1332 // ctx can be canceled, we still want to store the previews.

1333 storeNewPreviews(context.Background(), log, acc, newPreviews)

1334

1335 x := recover() // Should not happen, but don't take program down if it does.

1336 if x != nil {

1337 log.WithContext(ctx).Error("viewRequestTx panic", slog.Any("err", x))

1338 debug.PrintStack()

1339 metrics.PanicInc(metrics.Webmailrequest)

1340 }

1341 }()

1342

1343 var msgitems [][]MessageItem // Gathering for 300ms, then flushing.

1344 var parsedMessage *ParsedMessage

1345 var viewEnd bool

1346

1347 var immediate bool // No waiting, flush immediate.

1348 t := time.NewTimer(300 * time.Millisecond)

1349 defer t.Stop()

1350

1351 sendViewMsgs := func(force bool) {

1352 if len(msgitems) == 0 && !force {

1353 return

1354 }

1355

1356 immediate = false

1357 msgc <- EventViewMsgs{v.Request.ViewID, v.Request.ID, msgitems, parsedMessage, viewEnd}

1358 msgitems = nil

1359 parsedMessage = nil

1360 t.Reset(300 * time.Millisecond)

1361 }

1362

1363 // todo: should probably rewrite code so we don't start yet another goroutine, but instead handle the query responses directly (through a struct that keeps state?) in the sse connection goroutine.

1364

1365 mrc := make(chan msgResp, 1)

1366 go queryMessages(ctx, log, acc, tx, v, mrc, newPreviews)

1367

1368 for {

1369 select {

1370 case mr, ok := <-mrc:

1371 if !ok {

1372 sendViewMsgs(false)

1373 // Empty message list signals this query is done.

1374 msgc <- EventViewMsgs{v.Request.ViewID, v.Request.ID, nil, nil, false}

1375 return

1376 }

1377 if mr.err != nil {

1378 sendViewMsgs(false)

1379 errc <- EventViewErr{v.Request.ViewID, v.Request.ID, mr.err.Error(), mr.err}

1380 return

1381 }

1382 if mr.reset {

1383 resetc <- EventViewReset{v.Request.ViewID, v.Request.ID}

1384 continue

1385 }

1386 if mr.viewEnd {

1387 viewEnd = true

1388 sendViewMsgs(true)

1389 return

1390 }

1391

1392 msgitems = append(msgitems, mr.mil)

1393 if mr.pm != nil {

1394 parsedMessage = mr.pm

1395 }

1396 if immediate {

1397 sendViewMsgs(true)

1398 }

1399

1400 case <-t.C:

1401 if len(msgitems) == 0 {

1402 // Nothing to send yet. We'll send immediately when the next message comes in.

1403 immediate = true

1404 } else {

1405 sendViewMsgs(false)

1406 }

1407 }

1408 }

1409}

1410

1411// queryMessages executes a query, with filter, pagination, destination message id

1412// to fetch (the message that the client had in view and wants to display again).

1413// It sends on msgc, with several types of messages: errors, whether the view is

1414// reset due to missing AnchorMessageID, and when the end of the view was reached

1415// and/or for a message.

1416// newPreviews is filled with previews, the caller must save them.

1417func queryMessages(ctx context.Context, log mlog.Log, acc *store.Account, tx *bstore.Tx, v view, mrc chan msgResp, newPreviews map[int64]string) {

1418 defer func() {

1419 x := recover() // Should not happen, but don't take program down if it does.

1420 if x != nil {

1421 log.WithContext(ctx).Error("queryMessages panic", slog.Any("err", x))

1422 debug.PrintStack()

1423 mrc <- msgResp{err: fmt.Errorf("query failed")}

1424 metrics.PanicInc(metrics.Webmailquery)

1425 }

1426

1427 close(mrc)

1428 }()

1429

1430 query := v.Request.Query

1431 page := v.Request.Page

1432

1433 // Warning: Filters must be kept in sync between queryMessage and view.matches.

1434

1435 checkMessage := func(id int64) (valid bool, rerr error) {

1436 m := store.Message{ID: id}

1437 err := tx.Get(&m)

1438 if err == bstore.ErrAbsent || err == nil && m.Expunged {

1439 return false, nil

1440 } else if err != nil {

1441 return false, err

1442 } else {

1443 return v.matches(log, acc, false, m.ID, m.MailboxID, m.UID, m.Flags, m.Keywords, func(int64, int64, store.UID) (store.Message, error) {

1444 return m, nil

1445 })

1446 }

1447 }

1448

1449 // Check if AnchorMessageID exists and matches filter. If not, we will reset the view.

1450 if page.AnchorMessageID > 0 {

1451 // Check if message exists and (still) matches the filter.

1452 // todo: if AnchorMessageID exists but no longer matches the filter, we are resetting the view, but could handle it more gracefully in the future. if the message is in a different mailbox, we cannot query as efficiently, we'll have to read through more messages.

1453 if valid, err := checkMessage(page.AnchorMessageID); err != nil {

1454 mrc <- msgResp{err: fmt.Errorf("querying AnchorMessageID: %v", err)}

1455 return

1456 } else if !valid {

1457 mrc <- msgResp{reset: true}

1458 page.AnchorMessageID = 0

1459 }

1460 }

1461

1462 // Check if page.DestMessageID exists and matches filter. If not, we will ignore

1463 // it instead of continuing to send message till the end of the view.

1464 if page.DestMessageID > 0 {

1465 if valid, err := checkMessage(page.DestMessageID); err != nil {

1466 mrc <- msgResp{err: fmt.Errorf("querying requested message: %v", err)}

1467 return

1468 } else if !valid {

1469 page.DestMessageID = 0

1470 }

1471 }

1472

1473 // todo optimize: we would like to have more filters directly on the database if they can use an index. eg if there is a keyword filter and no mailbox filter.

1474

1475 q := bstore.QueryTx[store.Message](tx)

1476 q.FilterEqual("Expunged", false)

1477 if len(v.mailboxIDs) > 0 {

1478 if len(v.mailboxIDs) == 1 && v.matchMailboxIDs {

1479 // Should result in fast indexed query.

1480 for mbID := range v.mailboxIDs {

1481 q.FilterNonzero(store.Message{MailboxID: mbID})

1482 }

1483 } else {

1484 idsAny := make([]any, 0, len(v.mailboxIDs))

1485 for mbID := range v.mailboxIDs {

1486 idsAny = append(idsAny, mbID)

1487 }

1488 if v.matchMailboxIDs {

1489 q.FilterEqual("MailboxID", idsAny...)

1490 } else {

1491 q.FilterNotEqual("MailboxID", idsAny...)

1492 }

1493 }

1494 }

1495

1496 // If we are looking for an anchor, keep skipping message early (cheaply) until we've seen it.

1497 if page.AnchorMessageID > 0 {

1498 var seen = false

1499 q.FilterFn(func(m store.Message) bool {

1500 if seen {

1501 return true

1502 }

1503 seen = m.ID == page.AnchorMessageID

1504 return false

1505 })

1506 }

1507

1508 // We may be added filters the the query below. The FilterFn signature does not

1509 // implement reporting errors, or anything else, just a bool. So when making the

1510 // filter functions, we give them a place to store parsed message state, and an

1511 // error. We check the error during and after query execution.

1512 state := msgState{acc: acc, log: log, newPreviews: newPreviews}

1513 defer state.clear()

1514

1515 flagfilter := query.flagFilterFn()

1516 if flagfilter != nil {

1517 q.FilterFn(func(m store.Message) bool {

1518 return flagfilter(m.Flags, m.Keywords)

1519 })

1520 }

1521

1522 if query.Filter.Oldest != nil {

1523 q.FilterGreaterEqual("Received", *query.Filter.Oldest)

1524 }

1525 if query.Filter.Newest != nil {

1526 q.FilterLessEqual("Received", *query.Filter.Newest)

1527 }

1528

1529 if query.Filter.SizeMin > 0 {

1530 q.FilterGreaterEqual("Size", query.Filter.SizeMin)

1531 }

1532 if query.Filter.SizeMax > 0 {

1533 q.FilterLessEqual("Size", query.Filter.SizeMax)

1534 }

1535

1536 attachmentFilter := query.attachmentFilterFn(log, acc, &state)

1537 if attachmentFilter != nil {

1538 q.FilterFn(attachmentFilter)

1539 }

1540

1541 envFilter := query.envFilterFn(log, &state)

1542 if envFilter != nil {

1543 q.FilterFn(envFilter)

1544 }

1545

1546 headerFilter := query.headerFilterFn(log, &state)

1547 if headerFilter != nil {

1548 q.FilterFn(headerFilter)

1549 }

1550

1551 wordsFilter := query.wordsFilterFn(log, &state)

1552 if wordsFilter != nil {

1553 q.FilterFn(wordsFilter)

1554 }

1555

1556 var moreHeaders []string // From store.Settings.ShowHeaders

1557

1558 if query.OrderAsc {

1559 q.SortAsc("Received")

1560 } else {

1561 q.SortDesc("Received")

1562 }

1563 found := page.DestMessageID <= 0

1564 end := true

1565 have := 0

1566 err := q.ForEach(func(m store.Message) error {

1567 // Check for an error in one of the filters, propagate it.

1568 if state.err != nil {

1569 return state.err

1570 }

1571

1572 if have >= page.Count && found || have > 10000 {

1573 end = false

1574 return bstore.StopForEach

1575 }

1576

1577 if _, ok := v.threadIDs[m.ThreadID]; ok {

1578 // Message was already returned as part of a thread.

1579 return nil

1580 }

1581

1582 var pm *ParsedMessage

1583 if m.ID == page.DestMessageID || page.DestMessageID == 0 && have == 0 && page.AnchorMessageID == 0 {

1584 // For threads, if there was no DestMessageID, we may be getting the newest

1585 // message. For an initial view, this isn't necessarily the first the user is

1586 // expected to read first, that would be the first unread, which we'll get below

1587 // when gathering the thread.

1588 found = true

1589 xpm, err := parsedMessage(log, &m, &state, true, false, false)

1590 if err != nil && errors.Is(err, message.ErrHeader) {

1591 log.Debug("not returning parsed message due to invalid headers", slog.Int64("msgid", m.ID), slog.Any("err", err))

1592 } else if err != nil {

1593 return fmt.Errorf("parsing message %d: %v", m.ID, err)

1594 } else {

1595 pm = &xpm

1596 }

1597 }

1598

1599 var err error

1600 moreHeaders, err = ensureMoreHeaders(tx, moreHeaders)

1601 if err != nil {

1602 return fmt.Errorf("ensuring more headers: %v", err)

1603 }

1604

1605 mi, err := messageItem(log, m, &state, moreHeaders)

1606 if err != nil {

1607 return fmt.Errorf("making messageitem for message %d: %v", m.ID, err)

1608 }

1609 mil := []MessageItem{mi}

1610 if query.Threading != ThreadOff {

1611 more, xpm, err := gatherThread(log, tx, acc, v, m, page.DestMessageID, page.AnchorMessageID == 0 && have == 0, moreHeaders, state.newPreviews)

1612 if err != nil {

1613 return fmt.Errorf("gathering thread messages for id %d, thread %d: %v", m.ID, m.ThreadID, err)

1614 }

1615 if xpm != nil {

1616 pm = xpm

1617 found = true

1618 }

1619 mil = append(mil, more...)

1620 v.threadIDs[m.ThreadID] = struct{}{}

1621

1622 // Calculate how many messages the frontend is going to show, and only count those as returned.

1623 collapsed := map[int64]bool{}

1624 for _, mi := range mil {

1625 collapsed[mi.Message.ID] = mi.Message.ThreadCollapsed

1626 }

1627 unread := map[int64]bool{} // Propagated to thread root.

1628 if query.Threading == ThreadUnread {

1629 for _, mi := range mil {

1630 mm := mi.Message

1631 if mm.Seen {

1632 continue

1633 }

1634 unread[mm.ID] = true

1635 for _, id := range mm.ThreadParentIDs {

1636 unread[id] = true

1637 }

1638 }

1639 }

1640 for _, mi := range mil {

1641 mm := mi.Message

1642 threadRoot := true

1643 rootID := mm.ID

1644 for _, id := range mm.ThreadParentIDs {

1645 if _, ok := collapsed[id]; ok {

1646 threadRoot = false

1647 rootID = id

1648 }

1649 }

1650 if threadRoot || (query.Threading == ThreadOn && !collapsed[rootID] || query.Threading == ThreadUnread && unread[rootID]) {

1651 have++

1652 }

1653 }

1654 } else {

1655 have++

1656 }

1657 if pm != nil && len(pm.envelope.From) == 1 {

1658 pm.ViewMode, err = fromAddrViewMode(tx, pm.envelope.From[0])

1659 if err != nil {

1660 return fmt.Errorf("gathering view mode for id %d: %v", m.ID, err)

1661 }

1662 }

1663 mrc <- msgResp{mil: mil, pm: pm}

1664 return nil

1665 })

1666 // Check for an error in one of the filters again. Check in ForEach would not

1667 // trigger if the last message has the error.

1668 if err == nil && state.err != nil {

1669 err = state.err

1670 }

1671 if err != nil {

1672 mrc <- msgResp{err: fmt.Errorf("querying messages: %v", err)}

1673 return

1674 }

1675 if end {

1676 mrc <- msgResp{viewEnd: true}

1677 }

1678}

1679

1680func gatherThread(log mlog.Log, tx *bstore.Tx, acc *store.Account, v view, m store.Message, destMessageID int64, first bool, moreHeaders []string, newPreviews map[int64]string) ([]MessageItem, *ParsedMessage, error) {

1681 if m.ThreadID == 0 {

1682 // If we would continue, FilterNonzero would fail because there are no non-zero fields.

1683 return nil, nil, fmt.Errorf("message has threadid 0, account is probably still being upgraded, try turning threading off until the upgrade is done")

1684 }

1685

1686 // Fetch other messages for this thread.

1687 qt := bstore.QueryTx[store.Message](tx)

1688 qt.FilterNonzero(store.Message{ThreadID: m.ThreadID})

1689 qt.FilterEqual("Expunged", false)

1690 qt.FilterNotEqual("ID", m.ID)

1691 qt.SortAsc("ID")

1692 tml, err := qt.List()

1693 if err != nil {

1694 return nil, nil, fmt.Errorf("listing other messages in thread for message %d, thread %d: %v", m.ID, m.ThreadID, err)

1695 }

1696

1697 var mil []MessageItem

1698 var pm *ParsedMessage

1699 var firstUnread bool

1700 for _, tm := range tml {

1701 err := func() error {

1702 xstate := msgState{acc: acc, log: log, newPreviews: newPreviews}

1703 defer xstate.clear()

1704

1705 mi, err := messageItem(log, tm, &xstate, moreHeaders)

1706 if err != nil {

1707 return fmt.Errorf("making messageitem for message %d, for thread %d: %v", tm.ID, m.ThreadID, err)

1708 }

1709 mi.MatchQuery, err = v.matches(log, acc, false, tm.ID, tm.MailboxID, tm.UID, tm.Flags, tm.Keywords, func(int64, int64, store.UID) (store.Message, error) {

1710 return tm, nil

1711 })

1712 if err != nil {

1713 return fmt.Errorf("matching thread message %d against view query: %v", tm.ID, err)

1714 }

1715 mil = append(mil, mi)

1716

1717 if tm.ID == destMessageID || destMessageID == 0 && first && (pm == nil || !firstUnread && !tm.Seen) {

1718 firstUnread = !tm.Seen

1719 xpm, err := parsedMessage(log, &tm, &xstate, true, false, false)

1720 if err != nil && errors.Is(err, message.ErrHeader) {

1721 log.Debug("not returning parsed message due to invalid headers", slog.Int64("msgid", m.ID), slog.Any("err", err))

1722 } else if err != nil {

1723 return fmt.Errorf("parsing thread message %d: %v", tm.ID, err)

1724 } else {

1725 pm = &xpm

1726 }

1727 }

1728 return nil

1729 }()

1730 if err != nil {

1731 return nil, nil, err

1732 }

1733 }

1734

1735 // Finally, the message that caused us to gather this thread (which is likely the

1736 // most recent message in the thread) could be the only unread message.

1737 if destMessageID == 0 && first && !m.Seen && !firstUnread {

1738 xstate := msgState{acc: acc, log: log}

1739 defer xstate.clear()

1740 xpm, err := parsedMessage(log, &m, &xstate, true, false, false)

1741 if err != nil && errors.Is(err, message.ErrHeader) {

1742 log.Debug("not returning parsed message due to invalid headers", slog.Int64("msgid", m.ID), slog.Any("err", err))

1743 } else if err != nil {

1744 return nil, nil, fmt.Errorf("parsing thread message %d: %v", m.ID, err)

1745 } else {

1746 pm = &xpm

1747 }

1748 }

1749

1750 return mil, pm, nil

1751}

1752

1753// While checking the filters on a message, we may need to get more message

1754// details as each filter passes. We check the filters that need the basic

1755// information first, and load and cache more details for the next filters.

1756// msgState holds parsed details for a message, it is updated while filtering,

1757// with more information or reset for a next message.

1758type msgState struct {

1759 acc *store.Account // Never changes during lifetime.

1760 err error // Once set, doesn't get cleared.

1761 m store.Message

1762 part *message.Part // Will be without Reader when msgr is nil.

1763 msgr *store.MsgReader

1764 log mlog.Log

1765

1766 // If not nil, messages will get their Preview field filled when nil, and message

1767 // id and preview added to newPreviews, and saved in a separate write transaction

1768 // when the operation is done.

1769 newPreviews map[int64]string

1770}

1771

1772func (ms *msgState) clear() {

1773 if ms.msgr != nil {

1774 err := ms.msgr.Close()

1775 ms.log.Check(err, "closing message reader from state")

1776 ms.msgr = nil

1777 }

1778 *ms = msgState{acc: ms.acc, err: ms.err, log: ms.log, newPreviews: ms.newPreviews}

1779}

1780

1781func (ms *msgState) ensureMsg(m store.Message) {

1782 if m.ID != ms.m.ID {

1783 ms.clear()

1784 }

1785 ms.m = m

1786}

1787

1788func (ms *msgState) ensurePart(m store.Message, withMsgReader bool) bool {

1789 ms.ensureMsg(m)

1790

1791 if ms.err == nil {

1792 if ms.part == nil {

1793 if m.ParsedBuf == nil {

1794 ms.err = fmt.Errorf("message %d not parsed", m.ID)

1795 return false

1796 }

1797 var p message.Part

1798 if err := json.Unmarshal(m.ParsedBuf, &p); err != nil {

1799 ms.err = fmt.Errorf("load part for message %d: %w", m.ID, err)

1800 return false

1801 }

1802 ms.part = &p

1803 }

1804 if withMsgReader && ms.msgr == nil {

1805 ms.msgr = ms.acc.MessageReader(m)

1806 ms.part.SetReaderAt(ms.msgr)

1807 }

1808 }

1809 return ms.part != nil

1810}

1811

1812// flagFilterFn returns a function that applies the flag/keyword/"label"-related

1813// filters for a query. A nil function is returned if there are no flags to filter

1814// on.

1815func (q Query) flagFilterFn() func(store.Flags, []string) bool {

1816 labels := map[string]bool{}

1817 for _, k := range q.Filter.Labels {

1818 labels[k] = true

1819 }

1820 for _, k := range q.NotFilter.Labels {

1821 labels[k] = false

1822 }

1823

1824 if len(labels) == 0 {

1825 return nil

1826 }

1827

1828 var mask, flags store.Flags

1829 systemflags := map[string][]*bool{

1830 `\answered`: {&mask.Answered, &flags.Answered},

1831 `\flagged`: {&mask.Flagged, &flags.Flagged},

1832 `\deleted`: {&mask.Deleted, &flags.Deleted},

1833 `\seen`: {&mask.Seen, &flags.Seen},

1834 `\draft`: {&mask.Draft, &flags.Draft},

1835 `$junk`: {&mask.Junk, &flags.Junk},

1836 `$notjunk`: {&mask.Notjunk, &flags.Notjunk},

1837 `$forwarded`: {&mask.Forwarded, &flags.Forwarded},

1838 `$phishing`: {&mask.Phishing, &flags.Phishing},

1839 `$mdnsent`: {&mask.MDNSent, &flags.MDNSent},

1840 }

1841 keywords := map[string]bool{}

1842 for k, v := range labels {

1843 k = strings.ToLower(k)

1844 if mf, ok := systemflags[k]; ok {

1845 *mf[0] = true

1846 *mf[1] = v

1847 } else {

1848 keywords[k] = v

1849 }

1850 }

1851 return func(msgFlags store.Flags, msgKeywords []string) bool {

1852 var f store.Flags

1853 if f.Set(mask, msgFlags) != flags {

1854 return false

1855 }

1856 for k, v := range keywords {

1857 if slices.Contains(msgKeywords, k) != v {

1858 return false

1859 }

1860 }

1861 return true

1862 }

1863}

1864

1865// attachmentFilterFn returns a function that filters for the attachment-related

1866// filter from the query. A nil function is returned if there are attachment

1867// filters.

1868func (q Query) attachmentFilterFn(log mlog.Log, acc *store.Account, state *msgState) func(m store.Message) bool {

1869 if q.Filter.Attachments == AttachmentIndifferent && q.NotFilter.Attachments == AttachmentIndifferent {

1870 return nil

1871 }

1872

1873 return func(m store.Message) bool {

1874 if !state.ensurePart(m, true) {

1875 return false

1876 }

1877 types, err := attachmentTypes(log, m, state)

1878 if err != nil {

1879 state.err = err

1880 return false

1881 }

1882 return (q.Filter.Attachments == AttachmentIndifferent || types[q.Filter.Attachments]) && (q.NotFilter.Attachments == AttachmentIndifferent || !types[q.NotFilter.Attachments])

1883 }

1884}

1885

1886var attachmentMimetypes = map[string]AttachmentType{

1887 "application/pdf": AttachmentPDF,

1888 "application/zip": AttachmentArchive,

1889 "application/x-rar-compressed": AttachmentArchive,

1890 "application/vnd.oasis.opendocument.spreadsheet": AttachmentSpreadsheet,

1891 "application/vnd.ms-excel": AttachmentSpreadsheet,

1892 "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": AttachmentSpreadsheet,

1893 "application/vnd.oasis.opendocument.text": AttachmentDocument,

1894 "application/vnd.oasis.opendocument.presentation": AttachmentPresentation,

1895 "application/vnd.ms-powerpoint": AttachmentPresentation,

1896 "application/vnd.openxmlformats-officedocument.presentationml.presentation": AttachmentPresentation,

1897}

1898var attachmentExtensions = map[string]AttachmentType{

1899 ".pdf": AttachmentPDF,

1900 ".zip": AttachmentArchive,

1901 ".tar": AttachmentArchive,

1902 ".tgz": AttachmentArchive,

1903 ".tar.gz": AttachmentArchive,

1904 ".tbz2": AttachmentArchive,

1905 ".tar.bz2": AttachmentArchive,

1906 ".tar.lz": AttachmentArchive,

1907 ".tlz": AttachmentArchive,

1908 ".tar.xz": AttachmentArchive,

1909 ".txz": AttachmentArchive,

1910 ".tar.zst": AttachmentArchive,

1911 ".tar.lz4": AttachmentArchive,

1912 ".7z": AttachmentArchive,

1913 ".rar": AttachmentArchive,

1914 ".ods": AttachmentSpreadsheet,

1915 ".xls": AttachmentSpreadsheet,

1916 ".xlsx": AttachmentSpreadsheet,

1917 ".odt": AttachmentDocument,

1918 ".doc": AttachmentDocument,

1919 ".docx": AttachmentDocument,

1920 ".odp": AttachmentPresentation,

1921 ".ppt": AttachmentPresentation,

1922 ".pptx": AttachmentPresentation,

1923}

1924

1925func attachmentTypes(log mlog.Log, m store.Message, state *msgState) (map[AttachmentType]bool, error) {

1926 types := map[AttachmentType]bool{}

1927

1928 pm, err := parsedMessage(log, &m, state, false, false, false)

1929 if err != nil {

1930 return nil, fmt.Errorf("parsing message for attachments: %w", err)

1931 }

1932 for _, a := range pm.attachments {

1933 if a.Part.MediaType == "IMAGE" {

1934 types[AttachmentImage] = true

1935 continue

1936 }

1937 mt := strings.ToLower(a.Part.MediaType + "/" + a.Part.MediaSubType)

1938 if t, ok := attachmentMimetypes[mt]; ok {

1939 types[t] = true

1940 continue

1941 }

1942 _, filename, err := a.Part.DispositionFilename()

1943 if err != nil && (errors.Is(err, message.ErrParamEncoding) || errors.Is(err, message.ErrHeader)) {

1944 log.Debugx("parsing disposition/filename", err)

1945 } else if err != nil {

1946 return nil, fmt.Errorf("reading disposition/filename: %v", err)

1947 }

1948 if ext := filepath.Ext(filename); ext != "" {

1949 if t, ok := attachmentExtensions[strings.ToLower(ext)]; ok {

1950 types[t] = true

1951 }

1952 }

1953 }

1954

1955 if len(types) == 0 {

1956 types[AttachmentNone] = true

1957 } else {

1958 types[AttachmentAny] = true

1959 }

1960 return types, nil

1961}

1962

1963// envFilterFn returns a filter function for the "envelope" headers ("envelope" as

1964// used by IMAP, i.e. basic message headers from/to/subject, an unfortunate name

1965// clash with SMTP envelope) for the query. A nil function is returned if no

1966// filtering is needed.

1967func (q Query) envFilterFn(log mlog.Log, state *msgState) func(m store.Message) bool {

1968 if len(q.Filter.From) == 0 && len(q.Filter.To) == 0 && len(q.Filter.Subject) == 0 && len(q.NotFilter.From) == 0 && len(q.NotFilter.To) == 0 && len(q.NotFilter.Subject) == 0 {

1969 return nil

1970 }

1971

1972 lower := func(l []string) []string {

1973 if len(l) == 0 {

1974 return nil

1975 }

1976 r := make([]string, len(l))

1977 for i, s := range l {

1978 r[i] = strings.ToLower(s)

1979 }

1980 return r