From db4d67a2a6fb43da1f64b632a7d0c5a0e8b0e421 Mon Sep 17 00:00:00 2001
From: Robin Jarry <robin@jarry.cc>
Date: Sun, 12 Jan 2025 21:28:33 +0100
Subject: [PATCH] email: fix empty emails sent on retries

io.Reader objects can only be consumed once. On retry, re-reading from
them will return 0 bytes all the time.

Change NewTask to take in a bytes array instead of a reader. Create
a new reader on each retry.

Also remove SendRaw which has an inherent issue related to retry as
well. message.Entity is virtually a reader that can only be consumed
once.

Update EnqueueRaw to take a bytes array directly and have it call
NewTask() like EnqueueStd does.

Fixes: 78cfc5d87a7d ("email: allow sending raw emails")
Signed-off-by: Robin Jarry <robin@jarry.cc>
---
 email/send.go   | 46 ----------------------------------------------
 email/worker.go | 39 ++++++++++++++++-----------------------
 2 files changed, 16 insertions(+), 69 deletions(-)

diff --git a/email/send.go b/email/send.go
index 48b021a91fafe99c37de3fe4ba16e82b44a72b1d..27fe52ca2824c7c916712c9fb6686b711ec739eb 100644
--- a/email/send.go
+++ b/email/send.go
@@ -6,7 +6,6 @@ import (
 	"io"
 	"strconv"
 
-	"github.com/emersion/go-message"
 	_ "github.com/emersion/go-message/charset"
 	"github.com/emersion/go-message/mail"
 	"github.com/emersion/go-sasl"
@@ -132,48 +131,3 @@ func Send(ctx context.Context, msg io.Reader, rcpts []string) error {
 	}
 	return err
 }
-
-// Sends a raw email. Blocks until it's sent or an error occurs.
-func SendRaw(ctx context.Context, email *message.Entity, to []string) error {
-	worker := ForContext(ctx)
-
-	if worker.client == nil || worker.client.Noop() != nil {
-		worker.client = nil
-
-		c, sender, err := mailSetup(ctx)
-		if err != nil {
-			return err
-		}
-		worker.client = c
-		worker.sender = sender
-	}
-
-	var writer io.WriteCloser
-	var err error
-
-	err = worker.client.Mail(worker.sender.Address, nil)
-	if err != nil {
-		goto end
-	}
-	for _, rcpt := range to {
-		if err := worker.client.Rcpt(rcpt, nil); err != nil {
-			return err
-		}
-	}
-	writer, err = worker.client.Data()
-	if err != nil {
-		goto end
-	}
-	err = email.WriteTo(writer)
-	if err != nil {
-		writer.Close()
-		goto end
-	}
-	err = writer.Close()
-end:
-	if err != nil {
-		worker.client.Close()
-		worker.client = nil
-	}
-	return err
-}
diff --git a/email/worker.go b/email/worker.go
index a35f17216d054528b32ba867762b8804d9b21d90..5437090297c65eea3cd0ae0c6c3341709c61d7de 100644
--- a/email/worker.go
+++ b/email/worker.go
@@ -15,7 +15,6 @@ import (
 	"git.sr.ht/~sircmpwn/core-go/config"
 	work "git.sr.ht/~sircmpwn/dowork"
 	"github.com/ProtonMail/go-crypto/openpgp"
-	"github.com/emersion/go-message"
 	_ "github.com/emersion/go-message/charset"
 	"github.com/emersion/go-message/mail"
 	"github.com/emersion/go-pgpmail"
@@ -38,20 +37,26 @@ type workerContext struct {
 // Returns a task which will send this email for the work queue. If the caller
 // does not need to customize the task parameters, the Enqueue function may be
 // more desirable.
-func NewTask(msg *bytes.Buffer, rcpts []string) *work.Task {
+func NewTask(data []byte, rcpts []string) *work.Task {
 	return work.NewTask(func(ctx context.Context) error {
-		err := Send(ctx, msg, rcpts)
+		err := Send(ctx, bytes.NewReader(data), rcpts)
 		if err != nil {
 			log.Printf("Error sending mail: %v", err)
 		}
 		return err
 	}).Retries(10).After(func(ctx context.Context, task *work.Task) {
+		var to string
+		if len(rcpts) == 1 {
+			to = rcpts[0]
+		} else {
+			to = fmt.Sprintf("%d recipients", len(rcpts))
+		}
 		if task.Result() == nil {
 			log.Printf("Mail to %s sent after %d attempts",
-				strings.Join(rcpts, ", "), task.Attempts())
+				to, task.Attempts())
 		} else {
 			log.Printf("Mail to %s failed after %d attempts: %v",
-				strings.Join(rcpts, ", "), task.Attempts(), task.Result())
+				to, task.Attempts(), task.Result())
 		}
 	})
 }
@@ -170,27 +175,15 @@ func EnqueueStd(ctx context.Context, header mail.Header,
 		log.Fatal(err)
 	}
 
-	return queue.Enqueue(NewTask(&buf, rcpts))
+	return queue.Enqueue(NewTask(buf.Bytes(), rcpts))
 }
 
-func EnqueueRaw(ctx context.Context, email *message.Entity, to []string) error {
+func EnqueueRaw(ctx context.Context, data []byte, to []string) error {
+	if len(to) == 0 {
+		panic("recipients list cannot be empty")
+	}
 	queue := ForContext(ctx).queue
-	return queue.Enqueue(work.NewTask(func(ctx context.Context) error {
-		err := SendRaw(ctx, email, to)
-		if err != nil {
-			log.Printf("Error sending mail: %v", err)
-		}
-		return err
-	}).Retries(10).After(func(ctx context.Context, task *work.Task) {
-		msgID := email.Header.Get("Message-ID")
-		if task.Result() == nil {
-			log.Printf("Mail %s sent to %d recipients after %d attempts",
-				msgID, len(to), task.Attempts())
-		} else {
-			log.Printf("Mail %s failed after %d attempts: %v",
-				msgID, task.Attempts(), task.Result())
-		}
-	}))
+	return queue.Enqueue(NewTask(data, to))
 }
 
 type nopWriteCloser struct {