1package main

2

3import (

4 "bufio"

5 "bytes"

6 "context"

7 "encoding/json"

8 "errors"

9 "fmt"

10 "io"

11 "log"

12 "log/slog"

13 "maps"

14 "net"

15 "os"

16 "path/filepath"

17 "runtime/debug"

18 "strconv"

19 "strings"

20 "time"

21

22 "github.com/mjl-/bstore"

23

24 "github.com/mjl-/mox/admin"

25 "github.com/mjl-/mox/config"

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

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

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/queue"

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

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

35 "github.com/mjl-/mox/webapi"

36 "slices"

37)

38

39// ctl represents a connection to the ctl unix domain socket of a running mox instance.

40// ctl provides functions to read/write commands/responses/data streams.

41type ctl struct {

42 cmd string // Set for server-side of commands.

43 conn net.Conn

44 r *bufio.Reader // Set for first reader.

45 x any // If set, errors are handled by calling panic(x) instead of log.Fatal.

46 log mlog.Log // If set, along with x, logging is done here.

47}

48

49// xctl opens a ctl connection.

50func xctl() *ctl {

51 p := mox.DataDirPath("ctl")

52 conn, err := net.Dial("unix", p)

53 if err != nil {

54 log.Fatalf("connecting to control socket at %q: %v", p, err)

55 }

56 ctl := &ctl{conn: conn}

57 version := ctl.xread()

58 if version != "ctlv0" {

59 log.Fatalf("ctl protocol mismatch, got %q, expected ctlv0", version)

60 }

61 return ctl

62}

63

64// Interpret msg as an error.

65// If ctl.x is set, the string is also written to the ctl to be interpreted as error by the other party.

66func (c *ctl) xerror(msg string) {

67 if c.x == nil {

68 log.Fatalln(msg)

69 }

70 c.log.Debugx("ctl error", fmt.Errorf("%s", msg), slog.String("cmd", c.cmd))

71 c.xwrite(msg)

72 panic(c.x)

73}

74

75// Check if err is not nil. If so, handle error through ctl.x or log.Fatal. If

76// ctl.x is set, the error string is written to ctl, to be interpreted as an error

77// by the command reading from ctl.

78func (c *ctl) xcheck(err error, msg string) {

79 if err == nil {

80 return

81 }

82 if c.x == nil {

83 log.Fatalf("%s: %s", msg, err)

84 }

85 c.log.Debugx(msg, err, slog.String("cmd", c.cmd))

86 fmt.Fprintf(c.conn, "%s: %s\n", msg, err)

87 panic(c.x)

88}

89

90// Read a line and return it without trailing newline.

91func (c *ctl) xread() string {

92 if c.r == nil {

93 c.r = bufio.NewReader(c.conn)

94 }

95 line, err := c.r.ReadString('\n')

96 c.xcheck(err, "read from ctl")

97 return strings.TrimSuffix(line, "\n")

98}

99

100// Read a line. If not "ok", the string is interpreted as an error.

101func (c *ctl) xreadok() {

102 line := c.xread()

103 if line != "ok" {

104 c.xerror(line)

105 }

106}

107

108// Write a string, typically a command or parameter.

109func (c *ctl) xwrite(text string) {

110 _, err := fmt.Fprintln(c.conn, text)

111 c.xcheck(err, "write")

112}

113

114// Write "ok" to indicate success.

115func (c *ctl) xwriteok() {

116 c.xwrite("ok")

117}

118

119// Copy data from a stream from ctl to dst.

120func (c *ctl) xstreamto(dst io.Writer) {

121 _, err := io.Copy(dst, c.reader())

122 c.xcheck(err, "reading message")

123}

124

125// Copy data from src to a stream to ctl.

126func (c *ctl) xstreamfrom(src io.Reader) {

127 xw := c.writer()

128 _, err := io.Copy(xw, src)

129 c.xcheck(err, "copying")

130 xw.xclose()

131}

132

133// Writer returns an io.Writer for a data stream to ctl.

134// When done writing, caller must call xclose to signal the end of the stream.

135// Behaviour of "x" is copied from ctl.

136func (c *ctl) writer() *ctlwriter {

137 return &ctlwriter{cmd: c.cmd, conn: c.conn, x: c.x, log: c.log}

138}

139

140// Reader returns an io.Reader for a data stream from ctl.

141// Behaviour of "x" is copied from ctl.

142func (c *ctl) reader() *ctlreader {

143 if c.r == nil {

144 c.r = bufio.NewReader(c.conn)

145 }

146 return &ctlreader{cmd: c.cmd, conn: c.conn, r: c.r, x: c.x, log: c.log}

147}

148

149/*

150Ctlwriter and ctlreader implement the writing and reading a data stream. They

151implement the io.Writer and io.Reader interface. In the protocol below each

152non-data message ends with a newline that is typically stripped when

153interpreting.

154

155Zero or more data transactions:

156

157 > "123" (for data size) or an error message

158 > data, 123 bytes

159 < "ok" or an error message

160

161Followed by a end of stream indicated by zero data bytes message:

162

163 > "0"

164*/

165

166type ctlwriter struct {

167 cmd string // Set for server-side of commands.

168 conn net.Conn // Ctl socket from which messages are read.

169 buf []byte // Scratch buffer, for reading response.

170 x any // If not nil, errors in Write and xcheckf are handled with panic(x), otherwise with a log.Fatal.

171 log mlog.Log

172}

173

174// Write implements io.Writer. Errors other than EOF are handled through behaviour

175// for s.x, either a panic or log.Fatal.

176func (s *ctlwriter) Write(buf []byte) (int, error) {

177 _, err := fmt.Fprintf(s.conn, "%d\n", len(buf))

178 s.xcheck(err, "write count")

179 _, err = s.conn.Write(buf)

180 s.xcheck(err, "write data")

181 if s.buf == nil {

182 s.buf = make([]byte, 512)

183 }

184 n, err := s.conn.Read(s.buf)

185 s.xcheck(err, "reading response to write")

186 line := strings.TrimSuffix(string(s.buf[:n]), "\n")

187 if line != "ok" {

188 s.xerror(line)

189 }

190 return len(buf), nil

191}

192

193func (s *ctlwriter) xerror(msg string) {

194 if s.x == nil {

195 log.Fatalln(msg)

196 } else {

197 s.log.Debugx("error", fmt.Errorf("%s", msg), slog.String("cmd", s.cmd))

198 panic(s.x)

199 }

200}

201

202func (s *ctlwriter) xcheck(err error, msg string) {

203 if err == nil {

204 return

205 }

206 if s.x == nil {

207 log.Fatalf("%s: %s", msg, err)

208 } else {

209 s.log.Debugx(msg, err, slog.String("cmd", s.cmd))

210 panic(s.x)

211 }

212}

213

214func (s *ctlwriter) xclose() {

215 _, err := fmt.Fprintf(s.conn, "0\n")

216 s.xcheck(err, "write eof")

217}

218

219type ctlreader struct {

220 cmd string // Set for server-side of command.

221 conn net.Conn // For writing "ok" after reading.

222 r *bufio.Reader // Buffered ctl socket.

223 err error // If set, returned for each read. can also be io.EOF.

224 npending int // Number of bytes that can still be read until a new count line must be read.

225 x any // If set, errors are handled with panic(x) instead of log.Fatal.

226 log mlog.Log // If x is set, logging goes to log.

227}

228

229// Read implements io.Reader. Errors other than EOF are handled through behaviour

230// for s.x, either a panic or log.Fatal.

231func (s *ctlreader) Read(buf []byte) (N int, Err error) {

232 if s.err != nil {

233 return 0, s.err

234 }

235 if s.npending == 0 {

236 line, err := s.r.ReadString('\n')

237 s.xcheck(err, "reading count")

238 line = strings.TrimSuffix(line, "\n")

239 n, err := strconv.ParseInt(line, 10, 32)

240 if err != nil {

241 s.xerror(line)

242 }

243 if n == 0 {

244 s.err = io.EOF

245 return 0, s.err

246 }

247 s.npending = int(n)

248 }

249 rn := min(len(buf), s.npending)

250 n, err := s.r.Read(buf[:rn])

251 s.xcheck(err, "read from ctl")

252 s.npending -= n

253 if s.npending == 0 {

254 _, err = fmt.Fprintln(s.conn, "ok")

255 s.xcheck(err, "writing ok after reading")

256 }

257 return n, err

258}

259

260func (s *ctlreader) xerror(msg string) {

261 if s.x == nil {

262 log.Fatalln(msg)

263 } else {

264 s.log.Debugx("error", fmt.Errorf("%s", msg), slog.String("cmd", s.cmd))

265 panic(s.x)

266 }

267}

268

269func (s *ctlreader) xcheck(err error, msg string) {

270 if err == nil {

271 return

272 }

273 if s.x == nil {

274 log.Fatalf("%s: %s", msg, err)

275 } else {

276 s.log.Debugx(msg, err, slog.String("cmd", s.cmd))

277 panic(s.x)

278 }

279}

280

281// servectl handles requests on the unix domain socket "ctl", e.g. for graceful shutdown, local mail delivery.

282func servectl(ctx context.Context, cid int64, log mlog.Log, conn net.Conn, shutdown func()) {

283 log.Debug("ctl connection")

284

285 var stop = struct{}{} // Sentinel value for panic and recover.

286 xctl := &ctl{conn: conn, x: stop, log: log}

287 defer func() {

288 x := recover()

289 if x == nil || x == stop {

290 return

291 }

292 log.Error("servectl panic", slog.Any("err", x), slog.String("cmd", xctl.cmd))

293 debug.PrintStack()

294 metrics.PanicInc(metrics.Ctl)

295 }()

296

297 defer func() {

298 err := conn.Close()

299 log.Check(err, "close ctl connection")

300 }()

301

302 xctl.xwrite("ctlv0")

303 for {

304 servectlcmd(ctx, xctl, cid, shutdown)

305 }

306}

307

308func xparseJSON(xctl *ctl, s string, v any) {

309 dec := json.NewDecoder(strings.NewReader(s))

310 dec.DisallowUnknownFields()

311 err := dec.Decode(v)

312 xctl.xcheck(err, "parsing from ctl as json")

313}

314

315func servectlcmd(ctx context.Context, xctl *ctl, cid int64, shutdown func()) {

316 log := xctl.log

317 cmd := xctl.xread()

318 xctl.cmd = cmd

319 log.Info("ctl command", slog.String("cmd", cmd))

320 switch cmd {

321 case "stop":

322 shutdown()

323 os.Exit(0)

324

325 case "deliver":

326 /* The protocol, double quoted are literals.

327

328 > "deliver"

329 > address

330 < "ok"

331 > stream

332 < "ok"

333 */

334

335 to := xctl.xread()

336 a, _, addr, err := store.OpenEmail(log, to, false)

337 xctl.xcheck(err, "lookup destination address")

338

339 msgFile, err := store.CreateMessageTemp(log, "ctl-deliver")

340 xctl.xcheck(err, "creating temporary message file")

341 defer store.CloseRemoveTempFile(log, msgFile, "deliver message")

342 mw := message.NewWriter(msgFile)

343 xctl.xwriteok()

344

345 xctl.xstreamto(mw)

346 err = msgFile.Sync()

347 xctl.xcheck(err, "syncing message to storage")

348

349 m := store.Message{

350 Received: time.Now(),

351 Size: mw.Size,

352 }

353

354 a.WithWLock(func() {

355 err := a.DeliverDestination(log, addr, &m, msgFile)

356 xctl.xcheck(err, "delivering message")

357 log.Info("message delivered through ctl", slog.Any("to", to))

358 })

359

360 err = a.Close()

361 xctl.xcheck(err, "closing account")

362 xctl.xwriteok()

363

364 case "setaccountpassword":

365 /* protocol:

366 > "setaccountpassword"

367 > account

368 > password

369 < "ok" or error

370 */

371

372 account := xctl.xread()

373 pw := xctl.xread()

374

375 acc, err := store.OpenAccount(log, account, false)

376 xctl.xcheck(err, "open account")

377 defer func() {

378 if acc != nil {

379 err := acc.Close()

380 log.Check(err, "closing account after setting password")

381 }

382 }()

383

384 err = acc.SetPassword(log, pw)

385 xctl.xcheck(err, "setting password")

386 err = acc.Close()

387 xctl.xcheck(err, "closing account")

388 acc = nil

389 xctl.xwriteok()

390

391 case "queueholdruleslist":

392 /* protocol:

393 > "queueholdruleslist"

394 < "ok"

395 < stream

396 */

397 l, err := queue.HoldRuleList(ctx)

398 xctl.xcheck(err, "listing hold rules")

399 xctl.xwriteok()

400 xw := xctl.writer()

401 fmt.Fprintln(xw, "hold rules:")

402 for _, hr := range l {

403 var elems []string

404 if hr.Account != "" {

405 elems = append(elems, fmt.Sprintf("account %q", hr.Account))

406 }

407 var zerodom dns.Domain

408 if hr.SenderDomain != zerodom {

409 elems = append(elems, fmt.Sprintf("sender domain %q", hr.SenderDomain.Name()))

410 }

411 if hr.RecipientDomain != zerodom {

412 elems = append(elems, fmt.Sprintf("sender domain %q", hr.RecipientDomain.Name()))

413 }

414 if len(elems) == 0 {

415 fmt.Fprintf(xw, "id %d: all messages\n", hr.ID)

416 } else {

417 fmt.Fprintf(xw, "id %d: %s\n", hr.ID, strings.Join(elems, ", "))

418 }

419 }

420 if len(l) == 0 {

421 fmt.Fprint(xw, "(none)\n")

422 }

423 xw.xclose()

424

425 case "queueholdrulesadd":

426 /* protocol:

427 > "queueholdrulesadd"

428 > account

429 > senderdomainstr

430 > recipientdomainstr

431 < "ok" or error

432 */

433 var hr queue.HoldRule

434 hr.Account = xctl.xread()

435 senderdomstr := xctl.xread()

436 rcptdomstr := xctl.xread()

437 var err error

438 hr.SenderDomain, err = dns.ParseDomain(senderdomstr)

439 xctl.xcheck(err, "parsing sender domain")

440 hr.RecipientDomain, err = dns.ParseDomain(rcptdomstr)

441 xctl.xcheck(err, "parsing recipient domain")

442 hr, err = queue.HoldRuleAdd(ctx, log, hr)

443 xctl.xcheck(err, "add hold rule")

444 xctl.xwriteok()

445

446 case "queueholdrulesremove":

447 /* protocol:

448 > "queueholdrulesremove"

449 > id

450 < "ok" or error

451 */

452 idstr := xctl.xread()

453 id, err := strconv.ParseInt(idstr, 10, 64)

454 xctl.xcheck(err, "parsing id")

455 err = queue.HoldRuleRemove(ctx, log, id)

456 xctl.xcheck(err, "remove hold rule")

457 xctl.xwriteok()

458

459 case "queuelist":

460 /* protocol:

461 > "queuelist"

462 > filters as json

463 > sort as json

464 < "ok"

465 < stream

466 */

467 filterline := xctl.xread()

468 sortline := xctl.xread()

469 var f queue.Filter

470 xparseJSON(xctl, filterline, &f)

471 var s queue.Sort

472 xparseJSON(xctl, sortline, &s)

473 qmsgs, err := queue.List(ctx, f, s)

474 xctl.xcheck(err, "listing queue")

475 xctl.xwriteok()

476

477 xw := xctl.writer()

478 fmt.Fprintln(xw, "messages:")

479 for _, qm := range qmsgs {

480 var lastAttempt string

481 if qm.LastAttempt != nil {

482 lastAttempt = time.Since(*qm.LastAttempt).Round(time.Second).String()

483 }

484 fmt.Fprintf(xw, "%5d %s from:%s to:%s next %s last %s error %q\n", qm.ID, qm.Queued.Format(time.RFC3339), qm.Sender().LogString(), qm.Recipient().LogString(), -time.Since(qm.NextAttempt).Round(time.Second), lastAttempt, qm.LastResult().Error)

485 }

486 if len(qmsgs) == 0 {

487 fmt.Fprint(xw, "(none)\n")

488 }

489 xw.xclose()

490

491 case "queueholdset":

492 /* protocol:

493 > "queueholdset"

494 > queuefilters as json

495 > "true" or "false"

496 < "ok" or error

497 < count

498 */

499

500 filterline := xctl.xread()

501 hold := xctl.xread() == "true"

502 var f queue.Filter

503 xparseJSON(xctl, filterline, &f)

504 count, err := queue.HoldSet(ctx, f, hold)

505 xctl.xcheck(err, "setting on hold status for messages")

506 xctl.xwriteok()

507 xctl.xwrite(fmt.Sprintf("%d", count))

508

509 case "queueschedule":

510 /* protocol:

511 > "queueschedule"

512 > queuefilters as json

513 > relative to now

514 > duration

515 < "ok" or error

516 < count

517 */

518

519 filterline := xctl.xread()

520 relnow := xctl.xread()

521 duration := xctl.xread()

522 var f queue.Filter

523 xparseJSON(xctl, filterline, &f)

524 d, err := time.ParseDuration(duration)

525 xctl.xcheck(err, "parsing duration for next delivery attempt")

526 var count int

527 if relnow == "" {

528 count, err = queue.NextAttemptAdd(ctx, f, d)

529 } else {

530 count, err = queue.NextAttemptSet(ctx, f, time.Now().Add(d))

531 }

532 xctl.xcheck(err, "setting next delivery attempts in queue")

533 xctl.xwriteok()

534 xctl.xwrite(fmt.Sprintf("%d", count))

535

536 case "queuetransport":

537 /* protocol:

538 > "queuetransport"

539 > queuefilters as json

540 > transport

541 < "ok" or error

542 < count

543 */

544

545 filterline := xctl.xread()

546 transport := xctl.xread()

547 var f queue.Filter

548 xparseJSON(xctl, filterline, &f)

549 count, err := queue.TransportSet(ctx, f, transport)

550 xctl.xcheck(err, "adding to next delivery attempts in queue")

551 xctl.xwriteok()

552 xctl.xwrite(fmt.Sprintf("%d", count))

553

554 case "queuerequiretls":

555 /* protocol:

556 > "queuerequiretls"

557 > queuefilters as json

558 > reqtls (empty string, "true" or "false")

559 < "ok" or error

560 < count

561 */

562

563 filterline := xctl.xread()

564 reqtls := xctl.xread()

565 var req *bool

566 switch reqtls {

567 case "":

568 case "true":

569 v := true

570 req = &v

571 case "false":

572 v := false

573 req = &v

574 default:

575 xctl.xcheck(fmt.Errorf("unknown value %q", reqtls), "parsing value")

576 }

577 var f queue.Filter

578 xparseJSON(xctl, filterline, &f)

579 count, err := queue.RequireTLSSet(ctx, f, req)

580 xctl.xcheck(err, "setting tls requirements on messages in queue")

581 xctl.xwriteok()

582 xctl.xwrite(fmt.Sprintf("%d", count))

583

584 case "queuefail":

585 /* protocol:

586 > "queuefail"

587 > queuefilters as json

588 < "ok" or error

589 < count

590 */

591

592 filterline := xctl.xread()

593 var f queue.Filter

594 xparseJSON(xctl, filterline, &f)

595 count, err := queue.Fail(ctx, log, f)

596 xctl.xcheck(err, "marking messages from queue as failed")

597 xctl.xwriteok()

598 xctl.xwrite(fmt.Sprintf("%d", count))

599

600 case "queuedrop":

601 /* protocol:

602 > "queuedrop"

603 > queuefilters as json

604 < "ok" or error

605 < count

606 */

607

608 filterline := xctl.xread()

609 var f queue.Filter

610 xparseJSON(xctl, filterline, &f)

611 count, err := queue.Drop(ctx, log, f)

612 xctl.xcheck(err, "dropping messages from queue")

613 xctl.xwriteok()

614 xctl.xwrite(fmt.Sprintf("%d", count))

615

616 case "queuedump":

617 /* protocol:

618 > "queuedump"

619 > id

620 < "ok" or error

621 < stream

622 */

623

624 idstr := xctl.xread()

625 id, err := strconv.ParseInt(idstr, 10, 64)

626 if err != nil {

627 xctl.xcheck(err, "parsing id")

628 }

629 mr, err := queue.OpenMessage(ctx, id)

630 xctl.xcheck(err, "opening message")

631 defer func() {

632 err := mr.Close()

633 log.Check(err, "closing message from queue")

634 }()

635 xctl.xwriteok()

636 xctl.xstreamfrom(mr)

637

638 case "queueretiredlist":

639 /* protocol:

640 > "queueretiredlist"

641 > filters as json

642 > sort as json

643 < "ok"

644 < stream

645 */

646 filterline := xctl.xread()

647 sortline := xctl.xread()

648 var f queue.RetiredFilter

649 xparseJSON(xctl, filterline, &f)

650 var s queue.RetiredSort

651 xparseJSON(xctl, sortline, &s)

652 qmsgs, err := queue.RetiredList(ctx, f, s)

653 xctl.xcheck(err, "listing retired queue")

654 xctl.xwriteok()

655

656 xw := xctl.writer()

657 fmt.Fprintln(xw, "retired messages:")

658 for _, qm := range qmsgs {

659 var lastAttempt string

660 if qm.LastAttempt != nil {

661 lastAttempt = time.Since(*qm.LastAttempt).Round(time.Second).String()

662 }

663 result := "failure"

664 if qm.Success {

665 result = "success"

666 }

667 sender, err := qm.Sender()

668 xcheckf(err, "parsing sender")

669 fmt.Fprintf(xw, "%5d %s %s from:%s to:%s last %s error %q\n", qm.ID, qm.Queued.Format(time.RFC3339), result, sender.LogString(), qm.Recipient().LogString(), lastAttempt, qm.LastResult().Error)

670 }

671 if len(qmsgs) == 0 {

672 fmt.Fprint(xw, "(none)\n")

673 }

674 xw.xclose()

675

676 case "queueretiredprint":

677 /* protocol:

678 > "queueretiredprint"

679 > id

680 < "ok"

681 < stream

682 */

683 idstr := xctl.xread()

684 id, err := strconv.ParseInt(idstr, 10, 64)

685 if err != nil {

686 xctl.xcheck(err, "parsing id")

687 }

688 l, err := queue.RetiredList(ctx, queue.RetiredFilter{IDs: []int64{id}}, queue.RetiredSort{})

689 xctl.xcheck(err, "getting retired messages")

690 if len(l) == 0 {

691 xctl.xcheck(errors.New("not found"), "getting retired message")

692 }

693 m := l[0]

694 xctl.xwriteok()

695 xw := xctl.writer()

696 enc := json.NewEncoder(xw)

697 enc.SetIndent("", "\t")

698 err = enc.Encode(m)

699 xctl.xcheck(err, "encode retired message")

700 xw.xclose()

701

702 case "queuehooklist":

703 /* protocol:

704 > "queuehooklist"

705 > filters as json

706 > sort as json

707 < "ok"

708 < stream

709 */

710 filterline := xctl.xread()

711 sortline := xctl.xread()

712 var f queue.HookFilter

713 xparseJSON(xctl, filterline, &f)

714 var s queue.HookSort

715 xparseJSON(xctl, sortline, &s)

716 hooks, err := queue.HookList(ctx, f, s)

717 xctl.xcheck(err, "listing webhooks")

718 xctl.xwriteok()

719

720 xw := xctl.writer()

721 fmt.Fprintln(xw, "webhooks:")

722 for _, h := range hooks {

723 var lastAttempt string

724 if len(h.Results) > 0 {

725 lastAttempt = time.Since(h.LastResult().Start).Round(time.Second).String()

726 }

727 fmt.Fprintf(xw, "%5d %s account:%s next %s last %s error %q url %s\n", h.ID, h.Submitted.Format(time.RFC3339), h.Account, time.Until(h.NextAttempt).Round(time.Second), lastAttempt, h.LastResult().Error, h.URL)

728 }

729 if len(hooks) == 0 {

730 fmt.Fprint(xw, "(none)\n")

731 }

732 xw.xclose()

733

734 case "queuehookschedule":

735 /* protocol:

736 > "queuehookschedule"

737 > hookfilters as json

738 > relative to now

739 > duration

740 < "ok" or error

741 < count

742 */

743

744 filterline := xctl.xread()

745 relnow := xctl.xread()

746 duration := xctl.xread()

747 var f queue.HookFilter

748 xparseJSON(xctl, filterline, &f)

749 d, err := time.ParseDuration(duration)

750 xctl.xcheck(err, "parsing duration for next delivery attempt")

751 var count int

752 if relnow == "" {

753 count, err = queue.HookNextAttemptAdd(ctx, f, d)

754 } else {

755 count, err = queue.HookNextAttemptSet(ctx, f, time.Now().Add(d))

756 }

757 xctl.xcheck(err, "setting next delivery attempts in queue")

758 xctl.xwriteok()

759 xctl.xwrite(fmt.Sprintf("%d", count))

760

761 case "queuehookcancel":

762 /* protocol:

763 > "queuehookcancel"

764 > hookfilters as json

765 < "ok" or error

766 < count

767 */

768

769 filterline := xctl.xread()

770 var f queue.HookFilter

771 xparseJSON(xctl, filterline, &f)

772 count, err := queue.HookCancel(ctx, log, f)

773 xctl.xcheck(err, "canceling webhooks in queue")

774 xctl.xwriteok()

775 xctl.xwrite(fmt.Sprintf("%d", count))

776

777 case "queuehookprint":

778 /* protocol:

779 > "queuehookprint"

780 > id

781 < "ok"

782 < stream

783 */

784 idstr := xctl.xread()

785 id, err := strconv.ParseInt(idstr, 10, 64)

786 if err != nil {

787 xctl.xcheck(err, "parsing id")

788 }

789 l, err := queue.HookList(ctx, queue.HookFilter{IDs: []int64{id}}, queue.HookSort{})

790 xctl.xcheck(err, "getting webhooks")

791 if len(l) == 0 {

792 xctl.xcheck(errors.New("not found"), "getting webhook")

793 }

794 h := l[0]

795 xctl.xwriteok()

796 xw := xctl.writer()

797 enc := json.NewEncoder(xw)

798 enc.SetIndent("", "\t")

799 err = enc.Encode(h)

800 xctl.xcheck(err, "encode webhook")

801 xw.xclose()

802

803 case "queuehookretiredlist":

804 /* protocol:

805 > "queuehookretiredlist"

806 > filters as json

807 > sort as json

808 < "ok"

809 < stream

810 */

811 filterline := xctl.xread()

812 sortline := xctl.xread()

813 var f queue.HookRetiredFilter

814 xparseJSON(xctl, filterline, &f)

815 var s queue.HookRetiredSort

816 xparseJSON(xctl, sortline, &s)

817 l, err := queue.HookRetiredList(ctx, f, s)

818 xctl.xcheck(err, "listing retired webhooks")

819 xctl.xwriteok()

820

821 xw := xctl.writer()

822 fmt.Fprintln(xw, "retired webhooks:")

823 for _, h := range l {

824 var lastAttempt string

825 if len(h.Results) > 0 {

826 lastAttempt = time.Since(h.LastResult().Start).Round(time.Second).String()

827 }

828 result := "success"

829 if !h.Success {

830 result = "failure"

831 }

832 fmt.Fprintf(xw, "%5d %s %s account:%s last %s error %q url %s\n", h.ID, h.Submitted.Format(time.RFC3339), result, h.Account, lastAttempt, h.LastResult().Error, h.URL)

833 }

834 if len(l) == 0 {

835 fmt.Fprint(xw, "(none)\n")

836 }

837 xw.xclose()

838

839 case "queuehookretiredprint":

840 /* protocol:

841 > "queuehookretiredprint"

842 > id

843 < "ok"

844 < stream

845 */

846 idstr := xctl.xread()

847 id, err := strconv.ParseInt(idstr, 10, 64)

848 if err != nil {

849 xctl.xcheck(err, "parsing id")

850 }

851 l, err := queue.HookRetiredList(ctx, queue.HookRetiredFilter{IDs: []int64{id}}, queue.HookRetiredSort{})

852 xctl.xcheck(err, "getting retired webhooks")

853 if len(l) == 0 {

854 xctl.xcheck(errors.New("not found"), "getting retired webhook")

855 }

856 h := l[0]

857 xctl.xwriteok()

858 xw := xctl.writer()

859 enc := json.NewEncoder(xw)

860 enc.SetIndent("", "\t")

861 err = enc.Encode(h)

862 xctl.xcheck(err, "encode retired webhook")

863 xw.xclose()

864

865 case "queuesuppresslist":

866 /* protocol:

867 > "queuesuppresslist"

868 > account (or empty)

869 < "ok" or error

870 < stream

871 */

872

873 account := xctl.xread()

874 l, err := queue.SuppressionList(ctx, account)

875 xctl.xcheck(err, "listing suppressions")

876 xctl.xwriteok()

877 xw := xctl.writer()

878 fmt.Fprintln(xw, "suppressions (account, address, manual, time added, base adddress, reason):")

879 for _, sup := range l {

880 manual := "No"

881 if sup.Manual {

882 manual = "Yes"

883 }

884 fmt.Fprintf(xw, "%q\t%q\t%s\t%s\t%q\t%q\n", sup.Account, sup.OriginalAddress, manual, sup.Created.Round(time.Second), sup.BaseAddress, sup.Reason)

885 }

886 if len(l) == 0 {

887 fmt.Fprintln(xw, "(none)")

888 }

889 xw.xclose()

890

891 case "queuesuppressadd":

892 /* protocol:

893 > "queuesuppressadd"

894 > account

895 > address

896 < "ok" or error

897 */

898

899 account := xctl.xread()

900 address := xctl.xread()

901 _, ok := mox.Conf.Account(account)

902 if !ok {

903 xctl.xcheck(errors.New("unknown account"), "looking up account")

904 }

905 addr, err := smtp.ParseAddress(address)

906 xctl.xcheck(err, "parsing address")

907 sup := webapi.Suppression{

908 Account: account,

909 Manual: true,

910 Reason: "added through mox cli",

911 }

912 err = queue.SuppressionAdd(ctx, addr.Path(), &sup)

913 xctl.xcheck(err, "adding suppression")

914 xctl.xwriteok()

915

916 case "queuesuppressremove":

917 /* protocol:

918 > "queuesuppressremove"

919 > account

920 > address

921 < "ok" or error

922 */

923

924 account := xctl.xread()

925 address := xctl.xread()

926 addr, err := smtp.ParseAddress(address)

927 xctl.xcheck(err, "parsing address")

928 err = queue.SuppressionRemove(ctx, account, addr.Path())

929 xctl.xcheck(err, "removing suppression")

930 xctl.xwriteok()

931

932 case "queuesuppresslookup":

933 /* protocol:

934 > "queuesuppresslookup"

935 > account or empty

936 > address

937 < "ok" or error

938 < stream

939 */

940

941 account := xctl.xread()

942 address := xctl.xread()

943 if account != "" {

944 _, ok := mox.Conf.Account(account)

945 if !ok {

946 xctl.xcheck(errors.New("unknown account"), "looking up account")

947 }

948 }

949 addr, err := smtp.ParseAddress(address)

950 xctl.xcheck(err, "parsing address")

951 sup, err := queue.SuppressionLookup(ctx, account, addr.Path())

952 xctl.xcheck(err, "looking up suppression")

953 xctl.xwriteok()

954 xw := xctl.writer()

955 if sup == nil {

956 fmt.Fprintln(xw, "not present")

957 } else {

958 manual := "no"

959 if sup.Manual {

960 manual = "yes"

961 }

962 fmt.Fprintf(xw, "present\nadded: %s\nmanual: %s\nbase address: %s\nreason: %q\n", sup.Created.Round(time.Second), manual, sup.BaseAddress, sup.Reason)

963 }

964 xw.xclose()

965

966 case "importmaildir", "importmbox":

967 mbox := cmd == "importmbox"

968 ximportctl(ctx, xctl, mbox)

969

970 case "domainadd":

971 /* protocol:

972 > "domainadd"

973 > disabled as "true" or "false"

974 > domain

975 > account

976 > localpart

977 < "ok" or error

978 */

979 var disabled bool

980 switch s := xctl.xread(); s {

981 case "true":

982 disabled = true

983 case "false":

984 disabled = false

985 default:

986 xctl.xcheck(fmt.Errorf("invalid value %q", s), "parsing disabled boolean")

987 }

988

989 domain := xctl.xread()

990 account := xctl.xread()

991 localpart := xctl.xread()

992 d, err := dns.ParseDomain(domain)

993 xctl.xcheck(err, "parsing domain")

994 err = admin.DomainAdd(ctx, disabled, d, account, smtp.Localpart(localpart))

995 xctl.xcheck(err, "adding domain")

996 xctl.xwriteok()

997

998 case "domainrm":

999 /* protocol:

1000 > "domainrm"

1001 > domain

1002 < "ok" or error

1003 */

1004 domain := xctl.xread()

1005 d, err := dns.ParseDomain(domain)

1006 xctl.xcheck(err, "parsing domain")

1007 err = admin.DomainRemove(ctx, d)

1008 xctl.xcheck(err, "removing domain")

1009 xctl.xwriteok()

1010

1011 case "domaindisabled":

1012 /* protocol:

1013 > "domaindisabled"

1014 > "true" or "false"

1015 > domain

1016 < "ok" or error

1017 */

1018 domain := xctl.xread()

1019 var disabled bool

1020 switch s := xctl.xread(); s {

1021 case "true":

1022 disabled = true

1023 case "false":

1024 disabled = false

1025 default:

1026 xctl.xerror("bad boolean value")

1027 }

1028 err := admin.DomainSave(ctx, domain, func(d *config.Domain) error {

1029 d.Disabled = disabled

1030 return nil

1031 })

1032 xctl.xcheck(err, "saving domain")

1033 xctl.xwriteok()

1034

1035 case "accountadd":

1036 /* protocol:

1037 > "accountadd"

1038 > account

1039 > address

1040 < "ok" or error

1041 */

1042 account := xctl.xread()

1043 address := xctl.xread()

1044 err := admin.AccountAdd(ctx, account, address)

1045 xctl.xcheck(err, "adding account")

1046 xctl.xwriteok()

1047

1048 case "accountrm":

1049 /* protocol:

1050 > "accountrm"

1051 > account

1052 < "ok" or error

1053 */

1054 account := xctl.xread()

1055 err := admin.AccountRemove(ctx, account)

1056 xctl.xcheck(err, "removing account")

1057 xctl.xwriteok()

1058

1059 case "accountdisabled":

1060 /* protocol:

1061 > "accountdisabled"

1062 > account

1063 > message (if empty, then enabled)

1064 < "ok" or error

1065 */

1066 account := xctl.xread()

1067 message := xctl.xread()

1068

1069 acc, err := store.OpenAccount(log, account, false)

1070 xctl.xcheck(err, "open account")

1071 defer func() {

1072 err := acc.Close()

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

1074 }()

1075

1076 err = admin.AccountSave(ctx, account, func(acc *config.Account) {

1077 acc.LoginDisabled = message

1078 })

1079 xctl.xcheck(err, "saving account")

1080

1081 err = acc.SessionsClear(ctx, xctl.log)

1082 xctl.xcheck(err, "clearing active web sessions")

1083

1084 xctl.xwriteok()

1085

1086 case "accountenable":

1087 /* protocol:

1088 > "accountenable"

1089 > account

1090 < "ok" or error

1091 */

1092 account := xctl.xread()

1093 err := admin.AccountSave(ctx, account, func(acc *config.Account) {

1094 acc.LoginDisabled = ""

1095 })

1096 xctl.xcheck(err, "enabling account")

1097 xctl.xwriteok()

1098

1099 case "tlspubkeylist":

1100 /* protocol:

1101 > "tlspubkeylist"

1102 > account (or empty)

1103 < "ok" or error

1104 < stream

1105 */

1106 accountOpt := xctl.xread()

1107 tlspubkeys, err := store.TLSPublicKeyList(ctx, accountOpt)

1108 xctl.xcheck(err, "list tls public keys")

1109 xctl.xwriteok()

1110 xw := xctl.writer()

1111 fmt.Fprintf(xw, "# fingerprint, type, name, account, login address, no imap preauth (%d)\n", len(tlspubkeys))

1112 for _, k := range tlspubkeys {

1113 fmt.Fprintf(xw, "%s\t%s\t%q\t%s\t%s\t%v\n", k.Fingerprint, k.Type, k.Name, k.Account, k.LoginAddress, k.NoIMAPPreauth)

1114 }

1115 xw.xclose()

1116

1117 case "tlspubkeyget":

1118 /* protocol:

1119 > "tlspubkeyget"

1120 > fingerprint

1121 < "ok" or error

1122 < type

1123 < name

1124 < account

1125 < address

1126 < noimappreauth (true/false)

1127 < stream (certder)

1128 */

1129 fp := xctl.xread()

1130 tlspubkey, err := store.TLSPublicKeyGet(ctx, fp)

1131 xctl.xcheck(err, "looking tls public key")

1132 xctl.xwriteok()

1133 xctl.xwrite(tlspubkey.Type)

1134 xctl.xwrite(tlspubkey.Name)

1135 xctl.xwrite(tlspubkey.Account)

1136 xctl.xwrite(tlspubkey.LoginAddress)

1137 xctl.xwrite(fmt.Sprintf("%v", tlspubkey.NoIMAPPreauth))

1138 xctl.xstreamfrom(bytes.NewReader(tlspubkey.CertDER))

1139

1140 case "tlspubkeyadd":

1141 /* protocol:

1142 > "tlspubkeyadd"

1143 > loginaddress

1144 > name (or empty)

1145 > noimappreauth (true/false)

1146 > stream (certder)

1147 < "ok" or error

1148 */

1149 loginAddress := xctl.xread()

1150 name := xctl.xread()

1151 noimappreauth := xctl.xread()

1152 if noimappreauth != "true" && noimappreauth != "false" {

1153 xctl.xcheck(fmt.Errorf("bad value %q", noimappreauth), "parsing noimappreauth")

1154 }

1155 var b bytes.Buffer

1156 xctl.xstreamto(&b)

1157 tlspubkey, err := store.ParseTLSPublicKeyCert(b.Bytes())

1158 xctl.xcheck(err, "parsing certificate")

1159 if name != "" {

1160 tlspubkey.Name = name

1161 }

1162 acc, _, _, err := store.OpenEmail(xctl.log, loginAddress, false)

1163 xctl.xcheck(err, "open account for address")

1164 defer func() {

1165 err := acc.Close()

1166 xctl.log.Check(err, "close account")

1167 }()

1168 tlspubkey.Account = acc.Name

1169 tlspubkey.LoginAddress = loginAddress

1170 tlspubkey.NoIMAPPreauth = noimappreauth == "true"

1171

1172 err = store.TLSPublicKeyAdd(ctx, &tlspubkey)

1173 xctl.xcheck(err, "adding tls public key")

1174 xctl.xwriteok()

1175

1176 case "tlspubkeyrm":

1177 /* protocol:

1178 > "tlspubkeyadd"

1179 > fingerprint

1180 < "ok" or error

1181 */

1182 fp := xctl.xread()

1183 err := store.TLSPublicKeyRemove(ctx, fp)

1184 xctl.xcheck(err, "removing tls public key")

1185 xctl.xwriteok()

1186

1187 case "addressadd":

1188 /* protocol:

1189 > "addressadd"

1190 > address

1191 > account

1192 < "ok" or error

1193 */

1194 address := xctl.xread()

1195 account := xctl.xread()

1196 err := admin.AddressAdd(ctx, address, account)

1197 xctl.xcheck(err, "adding address")

1198 xctl.xwriteok()

1199

1200 case "addressrm":

1201 /* protocol:

1202 > "addressrm"

1203 > address

1204 < "ok" or error

1205 */

1206 address := xctl.xread()

1207 err := admin.AddressRemove(ctx, address)

1208 xctl.xcheck(err, "removing address")

1209 xctl.xwriteok()

1210

1211 case "aliaslist":

1212 /* protocol:

1213 > "aliaslist"

1214 > domain

1215 < "ok" or error

1216 < stream

1217 */

1218 domain := xctl.xread()

1219 d, err := dns.ParseDomain(domain)

1220 xctl.xcheck(err, "parsing domain")

1221 dc, ok := mox.Conf.Domain(d)

1222 if !ok {

1223 xctl.xcheck(errors.New("no such domain"), "listing aliases")

1224 }

1225 xctl.xwriteok()

1226 xw := xctl.writer()

1227 for _, a := range dc.Aliases {

1228 lp, err := smtp.ParseLocalpart(a.LocalpartStr)

1229 xctl.xcheck(err, "parsing alias localpart")

1230 fmt.Fprintln(xw, smtp.NewAddress(lp, a.Domain).Pack(true))

1231 }

1232 xw.xclose()

1233

1234 case "aliasprint":

1235 /* protocol:

1236 > "aliasprint"

1237 > address

1238 < "ok" or error

1239 < stream

1240 */

1241 address := xctl.xread()

1242 _, alias, ok := mox.Conf.AccountDestination(address)

1243 if !ok {

1244 xctl.xcheck(errors.New("no such address"), "looking up alias")

1245 } else if alias == nil {

1246 xctl.xcheck(errors.New("address not an alias"), "looking up alias")

1247 }

1248 xctl.xwriteok()

1249 xw := xctl.writer()

1250 fmt.Fprintf(xw, "# postpublic %v\n", alias.PostPublic)

1251 fmt.Fprintf(xw, "# listmembers %v\n", alias.ListMembers)

1252 fmt.Fprintf(xw, "# allowmsgfrom %v\n", alias.AllowMsgFrom)

1253 fmt.Fprintln(xw, "# members:")

1254 for _, a := range alias.Addresses {

1255 fmt.Fprintln(xw, a)

1256 }

1257 xw.xclose()

1258

1259 case "aliasadd":

1260 /* protocol:

1261 > "aliasadd"

1262 > address

1263 > json alias

1264 < "ok" or error

1265 */

1266 address := xctl.xread()

1267 line := xctl.xread()

1268 addr, err := smtp.ParseAddress(address)

1269 xctl.xcheck(err, "parsing address")

1270 var alias config.Alias

1271 xparseJSON(xctl, line, &alias)

1272 err = admin.AliasAdd(ctx, addr, alias)

1273 xctl.xcheck(err, "adding alias")

1274 xctl.xwriteok()

1275

1276 case "aliasupdate":

1277 /* protocol:

1278 > "aliasupdate"

1279 > alias

1280 > "true" or "false" for postpublic

1281 > "true" or "false" for listmembers

1282 > "true" or "false" for allowmsgfrom

1283 < "ok" or error

1284 */

1285 address := xctl.xread()

1286 postpublic := xctl.xread()

1287 listmembers := xctl.xread()

1288 allowmsgfrom := xctl.xread()

1289 addr, err := smtp.ParseAddress(address)

1290 xctl.xcheck(err, "parsing address")

1291 err = admin.DomainSave(ctx, addr.Domain.Name(), func(d *config.Domain) error {

1292 a, ok := d.Aliases[addr.Localpart.String()]

1293 if !ok {

1294 return fmt.Errorf("alias does not exist")

1295 }

1296

1297 switch postpublic {

1298 case "false":

1299 a.PostPublic = false

1300 case "true":

1301 a.PostPublic = true

1302 }

1303 switch listmembers {

1304 case "false":

1305 a.ListMembers = false

1306 case "true":

1307 a.ListMembers = true

1308 }

1309 switch allowmsgfrom {

1310 case "false":

1311 a.AllowMsgFrom = false

1312 case "true":

1313 a.AllowMsgFrom = true

1314 }

1315

1316 d.Aliases = maps.Clone(d.Aliases)

1317 d.Aliases[addr.Localpart.String()] = a

1318 return nil

1319 })

1320 xctl.xcheck(err, "saving alias")

1321 xctl.xwriteok()

1322

1323 case "aliasrm":

1324 /* protocol:

1325 > "aliasrm"

1326 > alias

1327 < "ok" or error

1328 */

1329 address := xctl.xread()

1330 addr, err := smtp.ParseAddress(address)

1331 xctl.xcheck(err, "parsing address")

1332 err = admin.AliasRemove(ctx, addr)

1333 xctl.xcheck(err, "removing alias")

1334 xctl.xwriteok()

1335

1336 case "aliasaddaddr":

1337 /* protocol:

1338 > "aliasaddaddr"

1339 > alias

1340 > addresses as json

1341 < "ok" or error

1342 */

1343 address := xctl.xread()

1344 line := xctl.xread()

1345 addr, err := smtp.ParseAddress(address)

1346 xctl.xcheck(err, "parsing address")

1347 var addresses []string

1348 xparseJSON(xctl, line, &addresses)

1349 err = admin.AliasAddressesAdd(ctx, addr, addresses)

1350 xctl.xcheck(err, "adding addresses to alias")

1351 xctl.xwriteok()

1352

1353 case "aliasrmaddr":

1354 /* protocol:

1355 > "aliasrmaddr"

1356 > alias

1357 > addresses as json

1358 < "ok" or error

1359 */

1360 address := xctl.xread()

1361 line := xctl.xread()

1362 addr, err := smtp.ParseAddress(address)

1363 xctl.xcheck(err, "parsing address")

1364 var addresses []string

1365 xparseJSON(xctl, line, &addresses)

1366 err = admin.AliasAddressesRemove(ctx, addr, addresses)

1367 xctl.xcheck(err, "removing addresses to alias")

1368 xctl.xwriteok()

1369

1370 case "loglevels":

1371 /* protocol:

1372 > "loglevels"

1373 < "ok"

1374 < stream

1375 */

1376 xctl.xwriteok()

1377 l := mox.Conf.LogLevels()

1378 keys := []string{}

1379 for k := range l {

1380 keys = append(keys, k)

1381 }

1382 slices.Sort(keys)

1383 s := ""

1384 for _, k := range keys {

1385 ks := k

1386 if ks == "" {

1387 ks = "(default)"

1388 }

1389 s += ks + ": " + mlog.LevelStrings[l[k]] + "\n"

1390 }

1391 xctl.xstreamfrom(strings.NewReader(s))

1392

1393 case "setloglevels":

1394 /* protocol:

1395 > "setloglevels"

1396 > pkg

1397 > level (if empty, log level for pkg will be unset)

1398 < "ok" or error

1399 */

1400 pkg := xctl.xread()

1401 levelstr := xctl.xread()

1402 if levelstr == "" {

1403 mox.Conf.LogLevelRemove(log, pkg)

1404 } else {

1405 level, ok := mlog.Levels[levelstr]

1406 if !ok {

1407 xctl.xerror("bad level")

1408 }

1409 mox.Conf.LogLevelSet(log, pkg, level)

1410 }

1411 xctl.xwriteok()

1412

1413 case "retrain":

1414 /* protocol:

1415 > "retrain"

1416 > account or empty

1417 < "ok" or error

1418 */

1419 account := xctl.xread()

1420

1421 xretrain := func(name string) {

1422 acc, err := store.OpenAccount(log, name, false)

1423 xctl.xcheck(err, "open account")

1424 defer func() {

1425 if acc != nil {

1426 err := acc.Close()

1427 log.Check(err, "closing account after retraining")

1428 }

1429 }()

1430

1431 // todo: can we retrain an account without holding a write lock? perhaps by writing a junkfilter to a new location, and staying informed of message changes while we go through all messages in the account?

1432

1433 acc.WithWLock(func() {

1434 conf, _ := acc.Conf()

1435 if conf.JunkFilter == nil {

1436 xctl.xcheck(store.ErrNoJunkFilter, "looking for junk filter")

1437 }

1438

1439 // Remove existing junk filter files.

1440 basePath := mox.DataDirPath("accounts")

1441 dbPath := filepath.Join(basePath, acc.Name, "junkfilter.db")

1442 bloomPath := filepath.Join(basePath, acc.Name, "junkfilter.bloom")

1443 err := os.Remove(dbPath)

1444 log.Check(err, "removing old junkfilter database file", slog.String("path", dbPath))

1445 err = os.Remove(bloomPath)

1446 log.Check(err, "removing old junkfilter bloom filter file", slog.String("path", bloomPath))

1447

1448 // Open junk filter, this creates new files.

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

1450 xctl.xcheck(err, "open new junk filter")

1451 defer func() {

1452 if jf == nil {

1453 return

1454 }

1455 err := jf.CloseDiscard()

1456 log.Check(err, "closing junk filter during cleanup")

1457 }()

1458

1459 // Read through messages with either junk or nonjunk flag set, and train them.

1460 var total, trained int

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

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

1463 q.FilterEqual("Expunged", false)

1464 return q.ForEach(func(m store.Message) error {

1465 total++

1466 if m.Junk == m.Notjunk {

1467 return nil

1468 }

1469 ok, err := acc.TrainMessage(ctx, log, jf, m.Notjunk, m)

1470 if ok {

1471 trained++

1472 }

1473 if m.TrainedJunk == nil || *m.TrainedJunk != m.Junk {

1474 m.TrainedJunk = &m.Junk

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

1476 return fmt.Errorf("marking message as trained: %v", err)

1477 }

1478 }

1479 return err

1480 })

1481 })

1482 xctl.xcheck(err, "training messages")

1483 log.Info("retrained messages", slog.Int("total", total), slog.Int("trained", trained))

1484

1485 // Close junk filter, marking success.

1486 err = jf.Close()

1487 jf = nil

1488 xctl.xcheck(err, "closing junk filter")

1489 })

1490 }

1491

1492 if account == "" {

1493 for _, name := range mox.Conf.Accounts() {

1494 xretrain(name)

1495 }

1496 } else {

1497 xretrain(account)

1498 }

1499 xctl.xwriteok()

1500

1501 case "recalculatemailboxcounts":

1502 /* protocol:

1503 > "recalculatemailboxcounts"

1504 > account

1505 < "ok" or error

1506 < stream

1507 */

1508 account := xctl.xread()

1509 acc, err := store.OpenAccount(log, account, false)

1510 xctl.xcheck(err, "open account")

1511 defer func() {

1512 if acc != nil {

1513 err := acc.Close()

1514 log.Check(err, "closing account after recalculating mailbox counts")

1515 }

1516 }()

1517 xctl.xwriteok()

1518

1519 xw := xctl.writer()

1520

1521 acc.WithWLock(func() {

1522 var changes []store.Change

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

1524 var totalSize int64

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

1526 mc, err := mb.CalculateCounts(tx)

1527 if err != nil {

1528 return fmt.Errorf("calculating counts for mailbox %q: %w", mb.Name, err)

1529 }

1530 totalSize += mc.Size

1531

1532 if mc != mb.MailboxCounts {

1533 fmt.Fprintf(xw, "for %s setting new counts %s (was %s)\n", mb.Name, mc, mb.MailboxCounts)

1534 mb.HaveCounts = true

1535 mb.MailboxCounts = mc

1536 if err := tx.Update(&mb); err != nil {

1537 return fmt.Errorf("storing new counts for %q: %v", mb.Name, err)

1538 }

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

1540 }

1541 return nil

1542 })

1543 if err != nil {

1544 return err

1545 }

1546

1547 du := store.DiskUsage{ID: 1}

1548 if err := tx.Get(&du); err != nil {

1549 return fmt.Errorf("get disk usage: %v", err)

1550 }

1551 if du.MessageSize != totalSize {

1552 fmt.Fprintf(xw, "setting new total message size %d (was %d)\n", totalSize, du.MessageSize)

1553 du.MessageSize = totalSize

1554 if err := tx.Update(&du); err != nil {

1555 return fmt.Errorf("update disk usage: %v", err)

1556 }

1557 }

1558 return nil

1559 })

1560 xctl.xcheck(err, "write transaction for mailbox counts")

1561

1562 store.BroadcastChanges(acc, changes)

1563 })

1564 xw.xclose()

1565

1566 case "fixmsgsize":

1567 /* protocol:

1568 > "fixmsgsize"

1569 > account or empty

1570 < "ok" or error

1571 < stream

1572 */

1573

1574 accountOpt := xctl.xread()

1575 xctl.xwriteok()

1576 xw := xctl.writer()

1577

1578 var foundProblem bool

1579 const batchSize = 10000

1580

1581 xfixmsgsize := func(accName string) {

1582 acc, err := store.OpenAccount(log, accName, false)

1583 xctl.xcheck(err, "open account")

1584 defer func() {

1585 err := acc.Close()

1586 log.Check(err, "closing account after fixing message sizes")

1587 }()

1588

1589 total := 0

1590 var lastID int64

1591 for {

1592 var n int

1593

1594 acc.WithRLock(func() {

1595 mailboxCounts := map[int64]store.Mailbox{} // For broadcasting.

1596

1597 // Don't process all message in one transaction, we could block the account for too long.

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

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

1600 q.FilterEqual("Expunged", false)

1601 q.FilterGreater("ID", lastID)

1602 q.Limit(batchSize)

1603 q.SortAsc("ID")

1604 return q.ForEach(func(m store.Message) error {

1605 lastID = m.ID

1606 n++

1607

1608 p := acc.MessagePath(m.ID)

1609 st, err := os.Stat(p)

1610 if err != nil {

1611 mb := store.Mailbox{ID: m.MailboxID}

1612 if xerr := tx.Get(&mb); xerr != nil {

1613 fmt.Fprintf(xw, "get mailbox id %d for message with file error: %v\n", mb.ID, xerr)

1614 }

1615 fmt.Fprintf(xw, "checking file %s for message %d in mailbox %q (id %d): %v (continuing)\n", p, m.ID, mb.Name, mb.ID, err)

1616 return nil

1617 }

1618 filesize := st.Size()

1619 correctSize := int64(len(m.MsgPrefix)) + filesize

1620 if m.Size == correctSize {

1621 return nil

1622 }

1623

1624 foundProblem = true

1625

1626 mb := store.Mailbox{ID: m.MailboxID}

1627 if err := tx.Get(&mb); err != nil {

1628 fmt.Fprintf(xw, "get mailbox id %d for message with file size mismatch: %v\n", mb.ID, err)

1629 return nil

1630 }

1631 if mb.Expunged {

1632 fmt.Fprintf(xw, "message %d is in expunged mailbox %q (id %d) (continuing)\n", m.ID, mb.Name, mb.ID)

1633 }

1634 fmt.Fprintf(xw, "fixing message %d in mailbox %q (id %d) with incorrect size %d, should be %d (len msg prefix %d + on-disk file %s size %d)\n", m.ID, mb.Name, mb.ID, m.Size, correctSize, len(m.MsgPrefix), p, filesize)

1635

1636 // We assume that the original message size was accounted as stored in the mailbox

1637 // total size. If this isn't correct, the user can always run

1638 // recalculatemailboxcounts.

1639 mb.Size -= m.Size

1640 mb.Size += correctSize

1641 if err := tx.Update(&mb); err != nil {

1642 return fmt.Errorf("update mailbox counts: %v", err)

1643 }

1644 mailboxCounts[mb.ID] = mb

1645

1646 m.Size = correctSize

1647

1648 mr := acc.MessageReader(m)

1649 part, err := message.EnsurePart(log.Logger, false, mr, m.Size)

1650 if err != nil {

1651 fmt.Fprintf(xw, "parsing message %d again: %v (continuing)\n", m.ID, err)

1652 }

1653 m.ParsedBuf, err = json.Marshal(part)

1654 if err != nil {

1655 return fmt.Errorf("marshal parsed message: %v", err)

1656 }

1657 total++

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

1659 return fmt.Errorf("update message: %v", err)

1660 }

1661 return nil

1662 })

1663 })

1664 xctl.xcheck(err, "find and fix wrong message sizes")

1665

1666 var changes []store.Change

1667 for _, mb := range mailboxCounts {

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

1669 }

1670 store.BroadcastChanges(acc, changes)

1671 })

1672 if n < batchSize {

1673 break

1674 }

1675 }

1676 fmt.Fprintf(xw, "%d message size(s) fixed for account %s\n", total, accName)

1677 }

1678

1679 if accountOpt != "" {

1680 xfixmsgsize(accountOpt)

1681 } else {

1682 for i, accName := range mox.Conf.Accounts() {

1683 var line string

1684 if i > 0 {

1685 line = "\n"

1686 }

1687 fmt.Fprintf(xw, "%sFixing message sizes in account %s...\n", line, accName)

1688 xfixmsgsize(accName)

1689 }

1690 }

1691 if foundProblem {

1692 fmt.Fprintf(xw, "\nProblems were found and fixed. You should invalidate messages stored at imap clients with the \"mox bumpuidvalidity account [mailbox]\" command.\n")

1693 }

1694

1695 xw.xclose()

1696

1697 case "reparse":

1698 /* protocol:

1699 > "reparse"

1700 > account or empty

1701 < "ok" or error

1702 < stream

1703 */

1704

1705 accountOpt := xctl.xread()

1706 xctl.xwriteok()

1707 xw := xctl.writer()

1708

1709 const batchSize = 100

1710

1711 xreparseAccount := func(accName string) {

1712 acc, err := store.OpenAccount(log, accName, false)

1713 xctl.xcheck(err, "open account")

1714 defer func() {

1715 err := acc.Close()

1716 log.Check(err, "closing account after reparsing messages")

1717 }()

1718

1719 total := 0

1720 var lastID int64

1721 for {

1722 var n int

1723 // Don't process all message in one transaction, we could block the account for too long.

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

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

1726 q.FilterEqual("Expunged", false)

1727 q.FilterGreater("ID", lastID)

1728 q.Limit(batchSize)

1729 q.SortAsc("ID")

1730 return q.ForEach(func(m store.Message) error {

1731 lastID = m.ID

1732 mr := acc.MessageReader(m)

1733 p, err := message.EnsurePart(log.Logger, false, mr, m.Size)

1734 if err != nil {

1735 fmt.Fprintf(xw, "parsing message %d: %v (continuing)\n", m.ID, err)

1736 }

1737 m.ParsedBuf, err = json.Marshal(p)

1738 if err != nil {

1739 return fmt.Errorf("marshal parsed message: %v", err)

1740 }

1741 total++

1742 n++

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

1744 return fmt.Errorf("update message: %v", err)

1745 }

1746 return nil

1747 })

1748

1749 })

1750 xctl.xcheck(err, "update messages with parsed mime structure")

1751 if n < batchSize {

1752 break

1753 }

1754 }

1755 fmt.Fprintf(xw, "%d message(s) reparsed for account %s\n", total, accName)

1756 }

1757

1758 if accountOpt != "" {

1759 xreparseAccount(accountOpt)

1760 } else {

1761 for i, accName := range mox.Conf.Accounts() {

1762 var line string

1763 if i > 0 {

1764 line = "\n"

1765 }

1766 fmt.Fprintf(xw, "%sReparsing account %s...\n", line, accName)

1767 xreparseAccount(accName)

1768 }

1769 }

1770 xw.xclose()

1771

1772 case "reassignthreads":

1773 /* protocol:

1774 > "reassignthreads"

1775 > account or empty

1776 < "ok" or error

1777 < stream

1778 */

1779

1780 accountOpt := xctl.xread()

1781 xctl.xwriteok()

1782 xw := xctl.writer()

1783

1784 xreassignThreads := func(accName string) {

1785 acc, err := store.OpenAccount(log, accName, false)

1786 xctl.xcheck(err, "open account")

1787 defer func() {

1788 err := acc.Close()

1789 log.Check(err, "closing account after reassigning threads")

1790 }()

1791

1792 // We don't want to step on an existing upgrade process.

1793 err = acc.ThreadingWait(log)

1794 xctl.xcheck(err, "waiting for threading upgrade to finish")

1795 // todo: should we try to continue if the threading upgrade failed? only if there is a chance it will succeed this time...

1796

1797 // todo: reassigning isn't atomic (in a single transaction), ideally it would be (bstore would need to be able to handle large updates).

1798 const batchSize = 50000

1799 total, err := acc.ResetThreading(ctx, log, batchSize, true)

1800 xctl.xcheck(err, "resetting threading fields")

1801 fmt.Fprintf(xw, "New thread base subject assigned to %d message(s), starting to reassign threads...\n", total)

1802

1803 // Assign threads again. Ideally we would do this in a single transaction, but

1804 // bstore/boltdb cannot handle so many pending changes, so we set a high batchsize.

1805 err = acc.AssignThreads(ctx, log, nil, 0, 50000, xw)

1806 xctl.xcheck(err, "reassign threads")

1807

1808 fmt.Fprintf(xw, "Threads reassigned. You should invalidate messages stored at imap clients with the \"mox bumpuidvalidity account [mailbox]\" command.\n")

1809 }

1810

1811 if accountOpt != "" {

1812 xreassignThreads(accountOpt)

1813 } else {

1814 for i, accName := range mox.Conf.Accounts() {

1815 var line string

1816 if i > 0 {

1817 line = "\n"

1818 }

1819 fmt.Fprintf(xw, "%sReassigning threads for account %s...\n", line, accName)

1820 xreassignThreads(accName)

1821 }

1822 }

1823 xw.xclose()

1824

1825 case "backup":

1826 xbackupctl(ctx, xctl)

1827

1828 case "imapserve":

1829 /* protocol:

1830 > "imapserve"

1831 > address

1832 < "ok or error"

1833 imap protocol

1834 */

1835 address := xctl.xread()

1836 xctl.xwriteok()

1837 imapserver.ServeConnPreauth("(imapserve)", cid, xctl.conn, address)

1838 xctl.log.Debug("imap connection finished")

1839

1840 default:

1841 log.Info("unrecognized command", slog.String("cmd", cmd))

1842 xctl.xwrite("unrecognized command")

1843 return

1844 }

1845}

1846