1// Package webops implements shared functionality between webapisrv and webmail.

2package webops

3

4import (

5 "context"

6 "errors"

7 "fmt"

8 "io"

9 "log/slog"

10 "os"

11 "path/filepath"

12 "slices"

13 "sort"

14 "time"

15

16 "github.com/mjl-/bstore"

17

18 "github.com/mjl-/mox/junk"

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

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

21 "github.com/mjl-/mox/moxio"

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

23)

24

25var ErrMessageNotFound = errors.New("no such message")

26

27type XOps struct {

28 DBWrite func(ctx context.Context, acc *store.Account, fn func(tx *bstore.Tx))

29 Checkf func(ctx context.Context, err error, format string, args ...any)

30 Checkuserf func(ctx context.Context, err error, format string, args ...any)

31}

32

33func (x XOps) mailboxID(ctx context.Context, tx *bstore.Tx, mailboxID int64) store.Mailbox {

34 if mailboxID == 0 {

35 x.Checkuserf(ctx, errors.New("invalid zero mailbox ID"), "getting mailbox")

36 }

37 mb, err := store.MailboxID(tx, mailboxID)

38 if err == bstore.ErrAbsent || err == store.ErrMailboxExpunged {

39 x.Checkuserf(ctx, err, "getting mailbox")

40 }

41 x.Checkf(ctx, err, "getting mailbox")

42 return mb

43}

44

45// messageID returns a non-expunged message or panics with a sherpa error.

46func (x XOps) messageID(ctx context.Context, tx *bstore.Tx, messageID int64) store.Message {

47 if messageID == 0 {

48 x.Checkuserf(ctx, errors.New("invalid zero message id"), "getting message")

49 }

50 m := store.Message{ID: messageID}

51 err := tx.Get(&m)

52 if err == bstore.ErrAbsent {

53 x.Checkuserf(ctx, ErrMessageNotFound, "getting message")

54 } else if err == nil && m.Expunged {

55 x.Checkuserf(ctx, errors.New("message was removed"), "getting message")

56 }

57 x.Checkf(ctx, err, "getting message")

58 return m

59}

60

61func (x XOps) MessageDelete(ctx context.Context, log mlog.Log, acc *store.Account, messageIDs []int64) {

62 acc.WithWLock(func() {

63 var changes []store.Change

64

65 x.DBWrite(ctx, acc, func(tx *bstore.Tx) {

66 var modseq store.ModSeq

67 changes = x.MessageDeleteTx(ctx, log, tx, acc, messageIDs, &modseq)

68 })

69

70 store.BroadcastChanges(acc, changes)

71 })

72}

73

74func (x XOps) MessageDeleteTx(ctx context.Context, log mlog.Log, tx *bstore.Tx, acc *store.Account, messageIDs []int64, modseq *store.ModSeq) []store.Change {

75 changes := make([]store.Change, 0, 1+1) // 1 remove, 1 mailbox counts, optimistic that all messages are in 1 mailbox.

76

77 var jf *junk.Filter

78 defer func() {

79 if jf != nil {

80 err := jf.CloseDiscard()

81 log.Check(err, "close junk filter")

82 }

83 }()

84

85 conf, _ := acc.Conf()

86

87 var mb store.Mailbox

88 var changeRemoveUIDs store.ChangeRemoveUIDs

89 xflushMailbox := func() {

90 err := tx.Update(&mb)

91 x.Checkf(ctx, err, "updating mailbox counts")

92 slices.Sort(changeRemoveUIDs.UIDs)

93 changeRemoveUIDs.UIDNext = mb.UIDNext

94 changeRemoveUIDs.MessageCountIMAP = mb.MessageCountIMAP()

95 changeRemoveUIDs.Unseen = uint32(mb.MailboxCounts.Unseen)

96 changes = append(changes, mb.ChangeCounts(), changeRemoveUIDs)

97 }

98

99 for _, id := range messageIDs {

100 m := x.messageID(ctx, tx, id)

101

102 if *modseq == 0 {

103 var err error

104 *modseq, err = acc.NextModSeq(tx)

105 x.Checkf(ctx, err, "assigning next modseq")

106 }

107

108 if m.MailboxID != mb.ID {

109 if mb.ID != 0 {

110 xflushMailbox()

111 }

112 mb = x.mailboxID(ctx, tx, m.MailboxID)

113 mb.ModSeq = *modseq

114 changeRemoveUIDs = store.ChangeRemoveUIDs{MailboxID: mb.ID, ModSeq: *modseq}

115 }

116

117 if m.Junk != m.Notjunk && jf == nil && conf.JunkFilter != nil {

118 var err error

119 jf, _, err = acc.OpenJunkFilter(ctx, log)

120 x.Checkf(ctx, err, "open junk filter")

121 }

122

123 opts := store.RemoveOpts{JunkFilter: jf}

124 _, _, err := acc.MessageRemove(log, tx, *modseq, &mb, opts, m)

125 x.Checkf(ctx, err, "expunge message")

126

127 changeRemoveUIDs.UIDs = append(changeRemoveUIDs.UIDs, m.UID)

128 changeRemoveUIDs.MsgIDs = append(changeRemoveUIDs.MsgIDs, m.ID)

129 }

130

131 xflushMailbox()

132

133 if jf != nil {

134 err := jf.Close()

135 jf = nil

136 x.Checkf(ctx, err, "close junk filter")

137 }

138

139 return changes

140}

141

142func (x XOps) MessageFlagsAdd(ctx context.Context, log mlog.Log, acc *store.Account, messageIDs []int64, flaglist []string) {

143 flags, keywords, err := store.ParseFlagsKeywords(flaglist)

144 x.Checkuserf(ctx, err, "parsing flags")

145

146 acc.WithRLock(func() {

147 var changes []store.Change

148

149 x.DBWrite(ctx, acc, func(tx *bstore.Tx) {

150 var modseq store.ModSeq

151 var retrain []store.Message

152 var mb, origmb store.Mailbox

153

154 for _, mid := range messageIDs {

155 m := x.messageID(ctx, tx, mid)

156

157 if modseq == 0 {

158 modseq, err = acc.NextModSeq(tx)

159 x.Checkf(ctx, err, "assigning next modseq")

160 }

161

162 if mb.ID != m.MailboxID {

163 if mb.ID != 0 {

164 mb.ModSeq = modseq

165 err := tx.Update(&mb)

166 x.Checkf(ctx, err, "updating mailbox")

167 if mb.MailboxCounts != origmb.MailboxCounts {

168 changes = append(changes, mb.ChangeCounts())

169 }

170 if mb.KeywordsChanged(origmb) {

171 changes = append(changes, mb.ChangeKeywords())

172 }

173 }

174 mb = x.mailboxID(ctx, tx, m.MailboxID)

175 origmb = mb

176 }

177 mb.Keywords, _ = store.MergeKeywords(mb.Keywords, keywords)

178

179 mb.Sub(m.MailboxCounts())

180 oflags := m.Flags

181 m.Flags = m.Flags.Set(flags, flags)

182 var kwChanged bool

183 m.Keywords, kwChanged = store.MergeKeywords(m.Keywords, keywords)

184 mb.Add(m.MailboxCounts())

185

186 if m.Flags == oflags && !kwChanged {

187 continue

188 }

189

190 m.ModSeq = modseq

191 err = tx.Update(&m)

192 x.Checkf(ctx, err, "updating message")

193

194 changes = append(changes, m.ChangeFlags(oflags, mb))

195 retrain = append(retrain, m)

196 }

197

198 if mb.ID != 0 {

199 mb.ModSeq = modseq

200 err := tx.Update(&mb)

201 x.Checkf(ctx, err, "updating mailbox")

202 if mb.MailboxCounts != origmb.MailboxCounts {

203 changes = append(changes, mb.ChangeCounts())

204 }

205 if mb.KeywordsChanged(origmb) {

206 changes = append(changes, mb.ChangeKeywords())

207 }

208 }

209

210 err = acc.RetrainMessages(ctx, log, tx, retrain)

211 x.Checkf(ctx, err, "retraining messages")

212 })

213

214 store.BroadcastChanges(acc, changes)

215 })

216}

217

218func (x XOps) MessageFlagsClear(ctx context.Context, log mlog.Log, acc *store.Account, messageIDs []int64, flaglist []string) {

219 flags, keywords, err := store.ParseFlagsKeywords(flaglist)

220 x.Checkuserf(ctx, err, "parsing flags")

221

222 acc.WithRLock(func() {

223 var retrain []store.Message

224 var changes []store.Change

225

226 x.DBWrite(ctx, acc, func(tx *bstore.Tx) {

227 var modseq store.ModSeq

228 var mb, origmb store.Mailbox

229

230 for _, mid := range messageIDs {

231 m := x.messageID(ctx, tx, mid)

232

233 if modseq == 0 {

234 modseq, err = acc.NextModSeq(tx)

235 x.Checkf(ctx, err, "assigning next modseq")

236 }

237

238 if mb.ID != m.MailboxID {

239 if mb.ID != 0 {

240 mb.ModSeq = modseq

241 err := tx.Update(&mb)

242 x.Checkf(ctx, err, "updating counts for mailbox")

243 if mb.MailboxCounts != origmb.MailboxCounts {

244 changes = append(changes, mb.ChangeCounts())

245 }

246 // note: cannot remove keywords from mailbox by removing keywords from message.

247 }

248 mb = x.mailboxID(ctx, tx, m.MailboxID)

249 origmb = mb

250 }

251

252 oflags := m.Flags

253 mb.Sub(m.MailboxCounts())

254 m.Flags = m.Flags.Set(flags, store.Flags{})

255 var changed bool

256 m.Keywords, changed = store.RemoveKeywords(m.Keywords, keywords)

257 mb.Add(m.MailboxCounts())

258

259 if m.Flags == oflags && !changed {

260 continue

261 }

262

263 m.ModSeq = modseq

264 err = tx.Update(&m)

265 x.Checkf(ctx, err, "updating message")

266

267 changes = append(changes, m.ChangeFlags(oflags, mb))

268 retrain = append(retrain, m)

269 }

270

271 if mb.ID != 0 {

272 mb.ModSeq = modseq

273 err := tx.Update(&mb)

274 x.Checkf(ctx, err, "updating keywords in mailbox")

275 if mb.MailboxCounts != origmb.MailboxCounts {

276 changes = append(changes, mb.ChangeCounts())

277 }

278 // note: cannot remove keywords from mailbox by removing keywords from message.

279 }

280

281 err = acc.RetrainMessages(ctx, log, tx, retrain)

282 x.Checkf(ctx, err, "retraining messages")

283 })

284

285 store.BroadcastChanges(acc, changes)

286 })

287}

288

289// MailboxesMarkRead updates all messages in the referenced mailboxes as seen when

290// they aren't yet. The mailboxes are updated with their unread messages counts,

291// and the changes are propagated.

292func (x XOps) MailboxesMarkRead(ctx context.Context, log mlog.Log, acc *store.Account, mailboxIDs []int64) {

293 acc.WithRLock(func() {

294 var changes []store.Change

295

296 x.DBWrite(ctx, acc, func(tx *bstore.Tx) {

297 var modseq store.ModSeq

298

299 // Note: we don't need to retrain, changing the "seen" flag is not relevant.

300

301 for _, mbID := range mailboxIDs {

302 mb := x.mailboxID(ctx, tx, mbID)

303

304 // Find messages to update.

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

306 q.FilterNonzero(store.Message{MailboxID: mb.ID})

307 q.FilterEqual("Seen", false)

308 q.FilterEqual("Expunged", false)

309 q.SortAsc("UID")

310 var have bool

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

312 have = true // We need to update mailbox.

313

314 oflags := m.Flags

315 mb.Sub(m.MailboxCounts())

316 m.Seen = true

317 mb.Add(m.MailboxCounts())

318

319 if modseq == 0 {

320 var err error

321 modseq, err = acc.NextModSeq(tx)

322 x.Checkf(ctx, err, "assigning next modseq")

323 }

324 m.ModSeq = modseq

325 err := tx.Update(&m)

326 x.Checkf(ctx, err, "updating message")

327

328 changes = append(changes, m.ChangeFlags(oflags, mb))

329 return nil

330 })

331 x.Checkf(ctx, err, "listing messages to mark as read")

332

333 if have {

334 mb.ModSeq = modseq

335 err := tx.Update(&mb)

336 x.Checkf(ctx, err, "updating mailbox")

337 changes = append(changes, mb.ChangeCounts())

338 }

339 }

340 })

341

342 store.BroadcastChanges(acc, changes)

343 })

344}

345

346// MessageMove moves messages to the mailbox represented by mailboxName, or to mailboxID if mailboxName is empty.

347//

348// If markSeen is true, the messages are marked as seen.

349func (x XOps) MessageMove(ctx context.Context, log mlog.Log, acc *store.Account, messageIDs []int64, mailboxName string, mailboxID int64, markSeen bool) {

350 acc.WithWLock(func() {

351 var changes []store.Change

352

353 var newIDs []int64

354 defer func() {

355 for _, id := range newIDs {

356 p := acc.MessagePath(id)

357 err := os.Remove(p)

358 log.Check(err, "removing delivered message after failure", slog.String("path", p))

359 }

360 }()

361

362 x.DBWrite(ctx, acc, func(tx *bstore.Tx) {

363 if mailboxName != "" {

364 mb, err := acc.MailboxFind(tx, mailboxName)

365 x.Checkf(ctx, err, "looking up mailbox name")

366 if mb == nil {

367 x.Checkuserf(ctx, errors.New("not found"), "looking up mailbox name")

368 } else {

369 mailboxID = mb.ID

370 }

371 }

372

373 mbDst := x.mailboxID(ctx, tx, mailboxID)

374

375 if len(messageIDs) == 0 {

376 return

377 }

378

379 var modseq store.ModSeq

380 newIDs, changes = x.MessageMoveTx(ctx, log, acc, tx, messageIDs, mbDst, &modseq, markSeen)

381 })

382 newIDs = nil

383

384 store.BroadcastChanges(acc, changes)

385 })

386}

387

388// MessageMoveTx moves message to a new mailbox, which must be different than their

389// current mailbox. Moving a message is done by changing the MailboxID and

390// assigning an appropriate new UID, and then inserting a replacement Message record

391// with new ID that is marked expunged in the original mailbox, along with a

392// MessageErase record so the message gets erased when all sessions stopped

393// referencing the message.

394//

395// If markSeen is true, the messages are marked as seen.

396func (x XOps) MessageMoveTx(ctx context.Context, log mlog.Log, acc *store.Account, tx *bstore.Tx, messageIDs []int64, mbDst store.Mailbox, modseq *store.ModSeq, markSeen bool) ([]int64, []store.Change) {

397 var newIDs []int64

398 var commit bool

399 defer func() {

400 if commit {

401 return

402 }

403 for _, id := range newIDs {

404 p := acc.MessagePath(id)

405 err := os.Remove(p)

406 log.Check(err, "removing delivered message after failure", slog.String("path", p))

407 }

408 newIDs = nil

409 }()

410

411 // n adds, 1 remove, 2 mailboxcounts, 1 mailboxkeywords, optimistic that messages are in a single source mailbox.

412 changes := make([]store.Change, 0, len(messageIDs)+4)

413

414 var err error

415 if *modseq == 0 {

416 *modseq, err = acc.NextModSeq(tx)

417 x.Checkf(ctx, err, "assigning next modseq")

418 }

419

420 mbDst.ModSeq = *modseq

421

422 // Get messages.

423 l := make([]store.Message, len(messageIDs))

424 for i, id := range messageIDs {

425 l[i] = x.messageID(ctx, tx, id)

426 if l[i].MailboxID == mbDst.ID {

427 // Client should filter out messages that are already in mailbox.

428 x.Checkuserf(ctx, fmt.Errorf("message %d already in destination mailbox", l[i].ID), "moving message")

429 }

430 }

431

432 // Sort (group) by mailbox, sort by UID.

433 sort.Slice(l, func(i, j int) bool {

434 if l[i].MailboxID != l[j].MailboxID {

435 return l[i].MailboxID < l[j].MailboxID

436 }

437 return l[i].UID < l[j].UID

438 })

439

440 var jf *junk.Filter

441 defer func() {

442 if jf != nil {

443 err := jf.CloseDiscard()

444 log.Check(err, "close junk filter")

445 }

446 }()

447

448 accConf, _ := acc.Conf()

449

450 var mbSrc store.Mailbox

451 var changeRemoveUIDs store.ChangeRemoveUIDs

452 xflushMailbox := func() {

453 changeRemoveUIDs.UIDNext = mbSrc.UIDNext

454 changeRemoveUIDs.MessageCountIMAP = mbSrc.MessageCountIMAP()

455 changeRemoveUIDs.Unseen = uint32(mbSrc.MailboxCounts.Unseen)

456 changes = append(changes, changeRemoveUIDs, mbSrc.ChangeCounts())

457

458 err = tx.Update(&mbSrc)

459 x.Checkf(ctx, err, "updating source mailbox counts")

460 }

461

462 nkeywords := len(mbDst.Keywords)

463 now := time.Now()

464

465 syncDirs := map[string]struct{}{}

466

467 for _, om := range l {

468 if om.MailboxID != mbSrc.ID {

469 if mbSrc.ID != 0 {

470 xflushMailbox()

471 }

472 mbSrc = x.mailboxID(ctx, tx, om.MailboxID)

473 mbSrc.ModSeq = *modseq

474 changeRemoveUIDs = store.ChangeRemoveUIDs{MailboxID: mbSrc.ID, ModSeq: *modseq}

475 }

476

477 nm := om

478 nm.MailboxID = mbDst.ID

479 nm.UID = mbDst.UIDNext

480 err := mbDst.UIDNextAdd(1)

481 x.Checkf(ctx, err, "adding uid")

482 nm.ModSeq = *modseq

483 nm.CreateSeq = *modseq

484 nm.SaveDate = &now

485 if nm.IsReject && nm.MailboxDestinedID != 0 {

486 // Incorrectly delivered to Rejects mailbox. Adjust MailboxOrigID so this message

487 // is used for reputation calculation during future deliveries.

488 nm.MailboxOrigID = nm.MailboxDestinedID

489 nm.IsReject = false

490 nm.Seen = false

491 }

492 if markSeen {

493 nm.Seen = true

494 }

495

496 nm.JunkFlagsForMailboxMove(mbSrc, mbDst, accConf)

497

498 err = tx.Update(&nm)

499 x.Checkf(ctx, err, "updating message with new mailbox")

500

501 mbDst.Add(nm.MailboxCounts())

502

503 mbSrc.Sub(om.MailboxCounts())

504 om.ID = 0

505 om.Expunged = true

506 om.ModSeq = *modseq

507 om.TrainedJunk = nil

508 err = tx.Insert(&om)

509 x.Checkf(ctx, err, "inserting expunged message in old mailbox")

510

511 dstPath := acc.MessagePath(om.ID)

512 dstDir := filepath.Dir(dstPath)

513 if _, ok := syncDirs[dstDir]; !ok {

514 os.MkdirAll(dstDir, 0770)

515 syncDirs[dstDir] = struct{}{}

516 }

517

518 err = moxio.LinkOrCopy(log, dstPath, acc.MessagePath(nm.ID), nil, false)

519 x.Checkf(ctx, err, "duplicating message in old mailbox for current sessions")

520 newIDs = append(newIDs, nm.ID)

521 // We don't sync the directory. In case of a crash and files disappearing, the

522 // eraser will simply not find the file at next startup.

523

524 err = tx.Insert(&store.MessageErase{ID: om.ID, SkipUpdateDiskUsage: true})

525 x.Checkf(ctx, err, "insert message erase")

526

527 mbDst.Keywords, _ = store.MergeKeywords(mbDst.Keywords, nm.Keywords)

528

529 if accConf.JunkFilter != nil && nm.NeedsTraining() {

530 // Lazily open junk filter.

531 if jf == nil {

532 jf, _, err = acc.OpenJunkFilter(ctx, log)

533 x.Checkf(ctx, err, "open junk filter")

534 }

535 err := acc.RetrainMessage(ctx, log, tx, jf, &nm)

536 x.Checkf(ctx, err, "retrain message after moving")

537 }

538

539 changeRemoveUIDs.UIDs = append(changeRemoveUIDs.UIDs, om.UID)

540 changeRemoveUIDs.MsgIDs = append(changeRemoveUIDs.MsgIDs, om.ID)

541 changes = append(changes, nm.ChangeAddUID(mbDst))

542 }

543

544 for dir := range syncDirs {

545 err := moxio.SyncDir(log, dir)

546 x.Checkf(ctx, err, "sync directory")

547 }

548

549 xflushMailbox()

550

551 changes = append(changes, mbDst.ChangeCounts())

552 if nkeywords > len(mbDst.Keywords) {

553 changes = append(changes, mbDst.ChangeKeywords())

554 }

555

556 err = tx.Update(&mbDst)

557 x.Checkf(ctx, err, "updating destination mailbox with uidnext and modseq")

558

559 if jf != nil {

560 err := jf.Close()

561 x.Checkf(ctx, err, "saving junk filter")

562 jf = nil

563 }

564

565 commit = true

566 return newIDs, changes

567}

568

569func isText(p message.Part) bool {

570 return p.MediaType == "" && p.MediaSubType == "" || p.MediaType == "TEXT" && p.MediaSubType == "PLAIN"

571}

572

573func isHTML(p message.Part) bool {

574 return p.MediaType == "" && p.MediaSubType == "" || p.MediaType == "TEXT" && p.MediaSubType == "HTML"

575}

576

577func isAlternative(p message.Part) bool {

578 return p.MediaType == "MULTIPART" && p.MediaSubType == "ALTERNATIVE"

579}

580

581func readPart(p message.Part, maxSize int64) (string, error) {

582 buf, err := io.ReadAll(io.LimitReader(p.ReaderUTF8OrBinary(), maxSize))

583 if err != nil {

584 return "", fmt.Errorf("reading part contents: %v", err)

585 }

586 return string(buf), nil

587}

588

589// ReadableParts returns the contents of the first text and/or html parts,

590// descending into multiparts, truncated to maxSize bytes if longer.

591func ReadableParts(p message.Part, maxSize int64) (text string, html string, found bool, err error) {

592 // todo: may want to merge this logic with webmail's message parsing.

593

594 // For non-multipart messages, top-level part.

595 if isText(p) {

596 data, err := readPart(p, maxSize)

597 return data, "", true, err

598 } else if isHTML(p) {

599 data, err := readPart(p, maxSize)

600 return "", data, true, err

601 }

602

603 // Look in sub-parts. Stop when we have a readable part, don't continue with other

604 // subparts unless we have a multipart/alternative.

605 // todo: we may have to look at disposition "inline".

606 var haveText, haveHTML bool

607 for _, pp := range p.Parts {

608 if isText(pp) {

609 haveText = true

610 text, err = readPart(pp, maxSize)

611 if !isAlternative(p) {

612 break

613 }

614 } else if isHTML(pp) {

615 haveHTML = true

616 html, err = readPart(pp, maxSize)

617 if !isAlternative(p) {

618 break

619 }

620 }

621 }

622 if haveText || haveHTML {

623 return text, html, true, err

624 }

625

626 // Descend into the subparts.

627 for _, pp := range p.Parts {

628 text, html, found, err = ReadableParts(pp, maxSize)

629 if found {

630 break

631 }

632 }

633 return

634}

635