9 "github.com/mjl-/bstore"
11 "github.com/mjl-/mox/message"
12 "github.com/mjl-/mox/mlog"
13 "github.com/mjl-/mox/store"
16// Replace relaces a message for another, atomically, possibly in another mailbox,
17// without needing a sequence of: append message, store \deleted flag, expunge.
20func (c *conn) cmdxReplace(isUID bool, tag, cmd string, p *parser) {
35 var storeFlags store.Flags
38 // Error must be a syntax error, to properly abort the connection due to literal.
40 storeFlags, keywords, err = store.ParseFlagsKeywords(p.xflagList())
42 xsyntaxErrorf("parsing flags: %v", err)
55 // todo: only with utf8 should we we accept message headers with utf-8. we currently always accept them.
56 // todo: this is only relevant if we also support the CATENATE extension?
58 utf8 := p.take("UTF8 (")
63 // For utf8, we already consumed the required ~ above.
64 size, synclit := p.xliteralSize(!utf8, false)
66 // Check the request, including old message in database, whether the message fits
67 // in quota. If a non-nil func is returned, an error was found. Calling the
68 // function aborts handling this command.
70 checkMessage := func(tx *bstore.Tx) func() {
72 return func() { xuserErrorf("mailbox open in read-only mode") }
75 mb, err := c.account.MailboxFind(tx, name)
77 return func() { xserverErrorf("finding mailbox: %v", err) }
80 return func() { xusercodeErrorf("TRYCREATE", "%w", store.ErrUnknownMailbox) }
83 // Resolve "*" for UID or message sequence.
86 q := bstore.QueryTx[store.Message](tx)
87 q.FilterNonzero(store.Message{MailboxID: c.mailboxID})
88 q.FilterEqual("Expunged", false)
89 q.FilterLess("UID", c.uidnext)
93 if err == bstore.ErrAbsent {
94 return func() { xsyntaxErrorf("cannot use * on empty mailbox") }
96 xcheckf(err, "get last message in mailbox")
98 } else if c.exists == 0 {
99 return func() { xsyntaxErrorf("cannot use * on empty mailbox") }
101 num = uint32(c.uids[c.exists-1])
103 num = uint32(c.exists)
108 // Find or verify UID of message to replace.
110 uidOld = store.UID(num)
111 } else if num > c.exists {
112 return func() { xuserErrorf("invalid msgseq") }
114 uidOld = c.uids[int(num)-1]
117 // Check the message still exists in the database. If it doesn't, it may have been
118 // deleted just now and we won't check the quota. We'll raise an error later on,
119 // when we are not possibly reading a sync literal and can respond with unsolicited
121 q := bstore.QueryTx[store.Message](tx)
122 q.FilterNonzero(store.Message{MailboxID: c.mailboxID, UID: uidOld})
123 q.FilterEqual("Expunged", false)
124 q.FilterLess("UID", c.uidnext)
126 if err == bstore.ErrAbsent {
130 return func() { xserverErrorf("get message to replace: %v", err) }
133 // Check if we can add size bytes. We can't necessarily remove the current message yet.
134 ok, maxSize, err := c.account.CanAddMessageSize(tx, size)
136 return func() { xserverErrorf("check quota: %v", err) }
140 return func() { xusercodeErrorf("OVERQUOTA", "account over maximum total message size %d", maxSize) }
147 // Check request, if it cannot succeed, fail it now before client is sending the data.
149 name = xcheckmailboxname(name, true)
151 c.account.WithRLock(func() {
152 c.xdbread(func(tx *bstore.Tx) {
153 errfn = checkMessage(tx)
163 name, _, err = store.CheckMailboxName(name, true)
165 errfn = func() { xusercodeErrorf("CANNOT", "%s", err) }
167 c.account.WithRLock(func() {
168 c.xdbread(func(tx *bstore.Tx) {
169 errfn = checkMessage(tx)
176 var newID int64 // Delivered message ID, file removed on error.
181 // We got a non-sync literal, we will consume some data, but abort if there's too
182 // much. We draw the line at 1mb. Client should have used synchronizing literal.
183 if size > 1000*1000 {
185 err := errors.New("error condition and non-synchronizing literal too big")
186 bye := "* BYE [ALERT] " + err.Error()
187 panic(syntaxError{bye, "TOOBIG", err.Error(), err})
189 // Message will not be accepted.
192 // Read the message into a temporary file.
194 file, err = store.CreateMessageTemp(c.log, "imap-replace")
195 xcheckf(err, "creating temp file for message")
196 defer store.CloseRemoveTempFile(c.log, file, "temporary message file")
200 if !commit && newID != 0 {
201 p := c.account.MessagePath(newID)
203 c.xsanity(err, "remove message file for replace after error")
208 // Read the message data.
209 defer c.xtraceread(mlog.LevelTracedata)()
210 mw := message.NewWriter(f)
211 msize, err := io.Copy(mw, io.LimitReader(c.br, size))
212 c.xtraceread(mlog.LevelTrace) // Restore.
214 // Cannot use xcheckf due to %w handling of errIO.
215 c.xbrokenf("reading literal message: %s (%w)", err, errIO)
218 c.xbrokenf("read %d bytes for message, expected %d (%w)", msize, size, errIO)
221 // Finish reading the command.
222 line := c.xreadline(false)
223 p = newParser(line, c)
229 // If an error was found earlier, abort the command now that we've read the message.
234 var oldMsgExpunged bool
236 var om, nm store.Message
237 var mbSrc, mbDst store.Mailbox // Src and dst mailboxes can be different.
../rfc/8508:263
239 var pendingChanges []store.Change
242 c.flushChanges(pendingChanges)
245 c.account.WithWLock(func() {
246 var changes []store.Change
248 c.xdbwrite(func(tx *bstore.Tx) {
249 mbSrc = c.xmailboxID(tx, c.mailboxID)
251 // Get old message. If it has been expunged, we should have a pending change for
252 // it. We'll send untagged responses and fail the command.
254 qom := bstore.QueryTx[store.Message](tx)
255 qom.FilterNonzero(store.Message{MailboxID: mbSrc.ID, UID: uidOld})
257 xcheckf(err, "get old message to replace from database")
259 oldMsgExpunged = true
263 // Check quota for addition of new message. We can't necessarily yet remove the old message.
264 ok, maxSize, err := c.account.CanAddMessageSize(tx, mw.Size)
265 xcheckf(err, "checking quota")
268 xusercodeErrorf("OVERQUOTA", "account over maximum total message size %d", maxSize)
271 modseq, err := c.account.NextModSeq(tx)
272 xcheckf(err, "get next mod seq")
274 chremuids, _, err := c.account.MessageRemove(c.log, tx, modseq, &mbSrc, store.RemoveOpts{}, om)
275 xcheckf(err, "expunge old message")
276 changes = append(changes, chremuids)
277 // Note: we only add a mbSrc counts change later on, if it is not equal to mbDst.
279 err = tx.Update(&mbSrc)
280 xcheckf(err, "updating source mailbox counts")
282 mbDst = c.xmailbox(tx, name, "TRYCREATE")
283 mbDst.ModSeq = modseq
285 nkeywords := len(mbDst.Keywords)
287 // Make new message to deliver.
290 MailboxOrigID: mbDst.ID,
299 err = c.account.MessageAdd(c.log, tx, &mbDst, &nm, file, store.AddOpts{})
300 xcheckf(err, "delivering message")
303 changes = append(changes, nm.ChangeAddUID(mbDst), mbDst.ChangeCounts())
304 if nkeywords != len(mbDst.Keywords) {
305 changes = append(changes, mbDst.ChangeKeywords())
308 err = tx.Update(&mbDst)
309 xcheckf(err, "updating destination mailbox")
312 // Fetch pending changes, possibly with new UIDs, so we can apply them before adding our own new UID.
313 overflow, pendingChanges = c.comm.Get()
319 // Success, make sure messages aren't cleaned up anymore.
322 // Broadcast the change to other connections.
323 if mbSrc.ID != mbDst.ID {
324 changes = append(changes, mbSrc.ChangeCounts())
329 // Must update our msgseq/uids tracking with latest pending changes.
332 c.xapplyChanges(overflow, l, false)
334 // If we couldn't find the message, send a NO response. We've just applied pending
335 // changes, which should have expunged the absent message.
337 xuserErrorf("message to be replaced has been expunged")
340 // If the destination mailbox is our currently selected mailbox, we register and
341 // announce the new message.
342 if mbDst.ID == c.mailboxID {
344 // We send an untagged OK with APPENDUID, for sane bookkeeping in clients.
../rfc/8508:401
345 c.xbwritelinef("* OK [APPENDUID %d %d] ", mbDst.UIDValidity, nm.UID)
346 c.xbwritelinef("* %d EXISTS", c.exists)
349 // We must return vanished instead of expunge, and also highestmodseq, when qresync
351 qresync := c.enabled[capQresync]
353 // Now that we are in sync with msgseq, we can find our old msgseq and say it is
359 oseq = c.xsequence(om.UID)
360 c.sequenceRemove(oseq, om.UID)
362 if qresync || c.uidonly {
363 c.xbwritelinef("* VANISHED %d", om.UID)
366 c.xbwritelinef("* %d EXPUNGE", oseq)
368 c.xwriteresultf("%s OK [HIGHESTMODSEQ %d] replaced", tag, nm.ModSeq.Client())