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

2package webops

3

4import (

5 "context"

6 "errors"

7 "fmt"

8 "io"

9 "os"

10 "sort"

11 "time"

12

13 "golang.org/x/exp/maps"

14

15 "github.com/mjl-/bstore"

16

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

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

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

20)

21

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

23

24type XOps struct {

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

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

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

28}

29

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

31 if mailboxID == 0 {

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

33 }

34 mb := store.Mailbox{ID: mailboxID}

35 err := tx.Get(&mb)

36 if err == bstore.ErrAbsent {

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

38 }

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

40 return mb

41}

42

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

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

45 if messageID == 0 {

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

47 }

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

49 err := tx.Get(&m)

50 if err == bstore.ErrAbsent {

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

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

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

54 }

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

56 return m

57}

58

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

60 acc.WithWLock(func() {

61 var changes []store.Change

62

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

64 _, changes = x.MessageDeleteTx(ctx, log, tx, acc, messageIDs, 0)

65 })

66

67 store.BroadcastChanges(acc, changes)

68 })

69

70 for _, mID := range messageIDs {

71 p := acc.MessagePath(mID)

72 err := os.Remove(p)

73 log.Check(err, "removing message file for expunge")

74 }

75}

76

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

78 removeChanges := map[int64]store.ChangeRemoveUIDs{}

79 changes := make([]store.Change, 0, len(messageIDs)+1) // n remove, 1 mailbox counts

80

81 var mb store.Mailbox

82 remove := make([]store.Message, 0, len(messageIDs))

83

84 var totalSize int64

85 for _, mid := range messageIDs {

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

87 totalSize += m.Size

88

89 if m.MailboxID != mb.ID {

90 if mb.ID != 0 {

91 err := tx.Update(&mb)

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

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

94 }

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

96 }

97

98 qmr := bstore.QueryTx[store.Recipient](tx)

99 qmr.FilterEqual("MessageID", m.ID)

100 _, err := qmr.Delete()

101 x.Checkf(ctx, err, "removing message recipients")

102

103 mb.Sub(m.MailboxCounts())

104

105 if modseq == 0 {

106 modseq, err = acc.NextModSeq(tx)

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

108 }

109 m.Expunged = true

110 m.ModSeq = modseq

111 err = tx.Update(&m)

112 x.Checkf(ctx, err, "marking message as expunged")

113

114 ch := removeChanges[m.MailboxID]

115 ch.UIDs = append(ch.UIDs, m.UID)

116 ch.MailboxID = m.MailboxID

117 ch.ModSeq = modseq

118 removeChanges[m.MailboxID] = ch

119 remove = append(remove, m)

120 }

121

122 if mb.ID != 0 {

123 err := tx.Update(&mb)

124 x.Checkf(ctx, err, "updating count in mailbox")

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

126 }

127

128 err := acc.AddMessageSize(log, tx, -totalSize)

129 x.Checkf(ctx, err, "updating disk usage")

130

131 // Mark removed messages as not needing training, then retrain them, so if they

132 // were trained, they get untrained.

133 for i := range remove {

134 remove[i].Junk = false

135 remove[i].Notjunk = false

136 }

137 err = acc.RetrainMessages(ctx, log, tx, remove, true)

138 x.Checkf(ctx, err, "untraining deleted messages")

139

140 for _, ch := range removeChanges {

141 sort.Slice(ch.UIDs, func(i, j int) bool {

142 return ch.UIDs[i] < ch.UIDs[j]

143 })

144 changes = append(changes, ch)

145 }

146

147 return modseq, changes

148}

149

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

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

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

153

154 acc.WithRLock(func() {

155 var changes []store.Change

156

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

158 var modseq store.ModSeq

159 var retrain []store.Message

160 var mb, origmb store.Mailbox

161

162 for _, mid := range messageIDs {

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

164

165 if mb.ID != m.MailboxID {

166 if mb.ID != 0 {

167 err := tx.Update(&mb)

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

169 if mb.MailboxCounts != origmb.MailboxCounts {

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

171 }

172 if mb.KeywordsChanged(origmb) {

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

174 }

175 }

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

177 origmb = mb

178 }

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

180

181 mb.Sub(m.MailboxCounts())

182 oflags := m.Flags

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

184 var kwChanged bool

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

186 mb.Add(m.MailboxCounts())

187

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

189 continue

190 }

191

192 if modseq == 0 {

193 modseq, err = acc.NextModSeq(tx)

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

195 }

196 m.ModSeq = modseq

197 err = tx.Update(&m)

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

199

200 changes = append(changes, m.ChangeFlags(oflags))

201 retrain = append(retrain, m)

202 }

203

204 if mb.ID != 0 {

205 err := tx.Update(&mb)

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

207 if mb.MailboxCounts != origmb.MailboxCounts {

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

209 }

210 if mb.KeywordsChanged(origmb) {

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

212 }

213 }

214

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

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

217 })

218

219 store.BroadcastChanges(acc, changes)

220 })

221}

222

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

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

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

226

227 acc.WithRLock(func() {

228 var retrain []store.Message

229 var changes []store.Change

230

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

232 var modseq store.ModSeq

233 var mb, origmb store.Mailbox

234

235 for _, mid := range messageIDs {

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

237

238 if mb.ID != m.MailboxID {

239 if mb.ID != 0 {

240 err := tx.Update(&mb)

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

242 if mb.MailboxCounts != origmb.MailboxCounts {

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

244 }

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

246 }

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

248 origmb = mb

249 }

250

251 oflags := m.Flags

252 mb.Sub(m.MailboxCounts())

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

254 var changed bool

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

256 mb.Add(m.MailboxCounts())

257

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

259 continue

260 }

261

262 if modseq == 0 {

263 modseq, err = acc.NextModSeq(tx)

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

265 }

266 m.ModSeq = modseq

267 err = tx.Update(&m)

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

269

270 changes = append(changes, m.ChangeFlags(oflags))

271 retrain = append(retrain, m)

272 }

273

274 if mb.ID != 0 {

275 err := tx.Update(&mb)

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

277 if mb.MailboxCounts != origmb.MailboxCounts {

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

279 }

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

281 }

282

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

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

285 })

286

287 store.BroadcastChanges(acc, changes)

288 })

289}

290

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

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

293// and the changes are propagated.

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

295 acc.WithRLock(func() {

296 var changes []store.Change

297

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

299 var modseq store.ModSeq

300

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

302

303 for _, mbID := range mailboxIDs {

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

305

306 // Find messages to update.

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

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

309 q.FilterEqual("Seen", false)

310 q.FilterEqual("Expunged", false)

311 q.SortAsc("UID")

312 var have bool

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

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

315

316 oflags := m.Flags

317 mb.Sub(m.MailboxCounts())

318 m.Seen = true

319 mb.Add(m.MailboxCounts())

320

321 if modseq == 0 {

322 var err error

323 modseq, err = acc.NextModSeq(tx)

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

325 }

326 m.ModSeq = modseq

327 err := tx.Update(&m)

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

329

330 changes = append(changes, m.ChangeFlags(oflags))

331 return nil

332 })

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

334

335 if have {

336 err := tx.Update(&mb)

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

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

339 }

340 }

341 })

342

343 store.BroadcastChanges(acc, changes)

344 })

345}

346

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

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

349 acc.WithRLock(func() {

350 var changes []store.Change

351

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

353 if mailboxName != "" {

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

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

356 if mb == nil {

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

358 } else {

359 mailboxID = mb.ID

360 }

361 }

362

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

364

365 if len(messageIDs) == 0 {

366 return

367 }

368

369 _, changes = x.MessageMoveTx(ctx, log, acc, tx, messageIDs, mbDst, 0)

370 })

371

372 store.BroadcastChanges(acc, changes)

373 })

374}

375

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

377 retrain := make([]store.Message, 0, len(messageIDs))

378 removeChanges := map[int64]store.ChangeRemoveUIDs{}

379 // n adds, 1 remove, 2 mailboxcounts, optimistic and at least for a single message.

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

381

382 var mbSrc store.Mailbox

383

384 keywords := map[string]struct{}{}

385 now := time.Now()

386

387 for _, mid := range messageIDs {

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

389

390 // We may have loaded this mailbox in the previous iteration of this loop.

391 if m.MailboxID != mbSrc.ID {

392 if mbSrc.ID != 0 {

393 err := tx.Update(&mbSrc)

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

395 changes = append(changes, mbSrc.ChangeCounts())

396 }

397 mbSrc = x.mailboxID(ctx, tx, m.MailboxID)

398 }

399

400 if mbSrc.ID == mbDst.ID {

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

402 x.Checkuserf(ctx, errors.New("already in destination mailbox"), "moving message")

403 }

404

405 var err error

406 if modseq == 0 {

407 modseq, err = acc.NextModSeq(tx)

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

409 }

410

411 ch := removeChanges[m.MailboxID]

412 ch.UIDs = append(ch.UIDs, m.UID)

413 ch.ModSeq = modseq

414 ch.MailboxID = m.MailboxID

415 removeChanges[m.MailboxID] = ch

416

417 // Copy of message record that we'll insert when UID is freed up.

418 om := m

419 om.PrepareExpunge()

420 om.ID = 0 // Assign new ID.

421 om.ModSeq = modseq

422

423 mbSrc.Sub(m.MailboxCounts())

424

425 if mbDst.Trash {

426 m.Seen = true

427 }

428 conf, _ := acc.Conf()

429 m.MailboxID = mbDst.ID

430 if m.IsReject && m.MailboxDestinedID != 0 {

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

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

433 m.MailboxOrigID = m.MailboxDestinedID

434 m.IsReject = false

435 m.Seen = false

436 }

437 m.UID = mbDst.UIDNext

438 m.ModSeq = modseq

439 mbDst.UIDNext++

440 m.JunkFlagsForMailbox(mbDst, conf)

441 m.SaveDate = &now

442 err = tx.Update(&m)

443 x.Checkf(ctx, err, "updating moved message in database")

444

445 // Now that UID is unused, we can insert the old record again.

446 err = tx.Insert(&om)

447 x.Checkf(ctx, err, "inserting record for expunge after moving message")

448

449 mbDst.Add(m.MailboxCounts())

450

451 changes = append(changes, m.ChangeAddUID())

452 retrain = append(retrain, m)

453

454 for _, kw := range m.Keywords {

455 keywords[kw] = struct{}{}

456 }

457 }

458

459 err := tx.Update(&mbSrc)

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

461

462 changes = append(changes, mbSrc.ChangeCounts(), mbDst.ChangeCounts())

463

464 // Ensure destination mailbox has keywords of the moved messages.

465 var mbKwChanged bool

466 mbDst.Keywords, mbKwChanged = store.MergeKeywords(mbDst.Keywords, maps.Keys(keywords))

467 if mbKwChanged {

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

469 }

470

471 err = tx.Update(&mbDst)

472 x.Checkf(ctx, err, "updating mailbox with uidnext")

473

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

475 x.Checkf(ctx, err, "retraining messages after move")

476

477 // Ensure UIDs of the removed message are in increasing order. It is quite common

478 // for all messages to be from a single source mailbox, meaning this is just one

479 // change, for which we preallocated space.

480 for _, ch := range removeChanges {

481 sort.Slice(ch.UIDs, func(i, j int) bool {

482 return ch.UIDs[i] < ch.UIDs[j]

483 })

484 changes = append(changes, ch)

485 }

486

487 return modseq, changes

488}

489

490func isText(p message.Part) bool {

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

492}

493

494func isHTML(p message.Part) bool {

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

496}

497

498func isAlternative(p message.Part) bool {

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

500}

501

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

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

504 if err != nil {

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

506 }

507 return string(buf), nil

508}

509

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

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

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

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

514

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

516 if isText(p) {

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

518 return data, "", true, err

519 } else if isHTML(p) {

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

521 return "", data, true, err

522 }

523

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

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

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

527 var haveText, haveHTML bool

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

529 if isText(pp) {

530 haveText = true

531 text, err = readPart(pp, maxSize)

532 if !isAlternative(p) {

533 break

534 }

535 } else if isHTML(pp) {

536 haveHTML = true

537 html, err = readPart(pp, maxSize)

538 if !isAlternative(p) {

539 break

540 }

541 }

542 }

543 if haveText || haveHTML {

544 return text, html, true, err

545 }

546

547 // Descend into the subparts.

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

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

550 if found {

551 break

552 }

553 }

554 return

555}

556