feat: make the worker answer every SEND_EMAIL it acks with exactly one per-task EMAIL_FAILED or EMAIL_SENT, raising the typed account events (auth, disabled, rate limited, server error) in addition with the EmailErrorEvent body the consumer expects instead of a SendEmailResult it could not parse; a mailbox that is not loaded yet or a storage blip is left for a few bus redeliveries before being reported, using the new Message.Attempt and Redelivers fields the NATS and Kafka buses now fill in, and an SMTP INVALID_CREDENTIALS send failure is classified as an auth error

internal/app/worker/event_send_email.go

@@ -3,6 +3,7 @@ package worker
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"time"
@@ -29,9 +30,13 @@ func (w *WorkerService) HandleSendEmail(ctx context.Context, sendEmail models.Se
 	w.mailManager.RUnlock()
 
 	if !exists {
+		// The mailbox is not loaded here: it is still being added (its
+		// ADD_EMAIL is queued behind this send), or this worker restarted and
+		// the reconciler has not re-shipped it yet. Leave the send for
+		// redelivery a few times so a queued ADD_EMAIL gets processed first,
+		// then report the failure so the control plane retries the step.
 		err := fmt.Errorf("email account %s not found in worker", sendEmail.EmailID.String())
-		w.sendEmailFailure(sendEmail.TaskID, sendEmail.EmailID, mail, err.Error())
-		return err
+		return w.failSend(ctx, sendEmail, err.Error(), true)
 	}
 
 	// Decrypt subject
@@ -50,8 +55,7 @@ func (w *WorkerService) HandleSendEmail(ctx context.Context, sendEmail models.Se
 	bodyPlain, bodyHTML, attachmentRefs, err := w.fetchEmailBody(ctx, sendEmail.OrgID, sendEmail.BodyS3Key)
 	if err != nil {
 		log.Error().Err(err).Str("s3_key", sendEmail.BodyS3Key).Msg("Failed to fetch email body from S3")
-		w.sendEmailFailure(sendEmail.TaskID, sendEmail.EmailID, mail, fmt.Sprintf("failed to fetch email body: %v", err))
-		return err
+		return w.failSend(ctx, sendEmail, fmt.Sprintf("failed to fetch email body: %v", err), true)
 	}
 
 	// Fetch each attachment's bytes from object storage by key. A fetch failure
@@ -59,8 +63,7 @@ func (w *WorkerService) HandleSendEmail(ctx context.Context, sendEmail models.Se
 	attachments, err := w.fetchAttachments(ctx, attachmentRefs)
 	if err != nil {
 		log.Error().Err(err).Str("task_id", sendEmail.TaskID.String()).Msg("Failed to fetch attachment bytes from S3")
-		w.sendEmailFailure(sendEmail.TaskID, sendEmail.EmailID, mail, fmt.Sprintf("failed to fetch attachment: %v", err))
-		return err
+		return w.failSend(ctx, sendEmail, fmt.Sprintf("failed to fetch attachment: %v", err), true)
 	}
 
 	// Use unified Send method
@@ -220,7 +223,30 @@ func (w *WorkerService) sendEmailSuccess(taskID uuid.UUID, messageID, providerMs
 	}
 }
 
-// sendEmailError sends a structured error result back to the jobs service
+// sendNotLoadedRedeliveries is how many bus deliveries a send may burn waiting
+// for a transient condition (mailbox not loaded yet, object storage blip)
+// before the worker reports it failed. Each redelivery is a second apart, and
+// the bus stops redelivering at ten, so this stays well inside that.
+const sendNotLoadedRedeliveries = 5
+
+// failSend reports a send the worker could not attempt. A retryable condition
+// is first left for bus redelivery (returning an error naks the message) so a
+// queued ADD_EMAIL or a storage blip can clear; once the redeliveries are
+// spent, when the bus does not redeliver, or when the condition is not
+// retryable, the failure result is produced and the message is acked, handing
+// the retry to the control plane.
+func (w *WorkerService) failSend(ctx context.Context, sendEmail models.SendEmail, reason string, retryable bool) error {
+	if d := deliveryOf(ctx); retryable && d.redelivers && d.attempt < sendNotLoadedRedeliveries {
+		return errors.New(reason)
+	}
+	w.sendEmailFailure(sendEmail.TaskID, sendEmail.EmailID, nil, reason)
+	return nil
+}
+
+// sendEmailError reports a failed send attempt. The per-task result is always
+// an EMAIL_FAILED so the control plane has one result channel to walk the
+// send back on; account-level conditions (auth, disabled, rate limit, server
+// error) additionally raise their own typed event carrying the full context.
 func (w *WorkerService) sendEmailError(taskID uuid.UUID, emailID uuid.UUID, mail *wmail.WMail, mailErr *errx.MailError) {
 	// Determine the appropriate event type based on error
 	eventType := wmail.DetermineErrorEventType(mailErr)
@@ -236,14 +262,15 @@ func (w *WorkerService) sendEmailError(taskID uuid.UUID, emailID uuid.UUID, mail
 		SentAt:         time.Now(),
 	}
 
-	if err := w.Produce(eventType, taskID.String(), result); err != nil {
-		log.Error().Err(err).Str("task_id", taskID.String()).Msg("Failed to produce email error event")
+	if err := w.Produce(models.JobEventTypeEmailFailed, taskID.String(), result); err != nil {
+		log.Error().Err(err).Str("task_id", taskID.String()).Msg("Failed to produce email failed event")
 	}
 
-	// For critical auth/disabled errors, also send a separate error event with full context
+	// Account-level conditions also raise their typed event with full context
 	if eventType == models.JobEventTypeEmailAuthError ||
 		eventType == models.JobEventTypeEmailDisabled ||
-		eventType == models.JobEventTypeEmailRateLimited {
+		eventType == models.JobEventTypeEmailRateLimited ||
+		eventType == models.JobEventTypeEmailServerError {
 
 		userInfo := mailErr.GetUserErrorInfo()
 		errorEvent := models.EmailErrorEvent{

internal/app/worker/receive.go

@@ -8,6 +8,21 @@ import (
 	"github.com/warmbly/warmbly/internal/models"
 )
 
+type deliveryKey struct{}
+
+// delivery is what a handler may know about the bus message it is processing.
+type delivery struct {
+	attempt    int  // 1-based delivery count; 0 when unknown
+	redelivers bool // a handler error gets the message delivered again
+}
+
+// deliveryOf returns the bus delivery details for the message a handler is
+// processing (zero value when the handler was not invoked from Receive).
+func deliveryOf(ctx context.Context) delivery {
+	d, _ := ctx.Value(deliveryKey{}).(delivery)
+	return d
+}
+
 // Receive is the eventbus.Handler that drives the worker's event loop. It
 // decodes the wire payload via the injected codec.Codec and dispatches to
 // HandleEvent.
@@ -17,7 +32,7 @@ func (w *WorkerService) Receive(ctx context.Context, msg eventbus.Message) error
 		return err
 	}
 
-	hctx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	hctx, cancel := context.WithTimeout(context.WithValue(ctx, deliveryKey{}, delivery{attempt: msg.Attempt, redelivers: msg.Redelivers}), 30*time.Second)
 	defer cancel()
 	return w.HandleEvent(hctx, &event)
 }

internal/app/worker/wmail/send.go

@@ -422,7 +422,7 @@ func DetermineErrorEventType(err *errx.MailError) models.JobEventType {
 	}
 
 	switch err.Code {
-	case errx.MailErrorCodeGoogleAuth, errx.MailErrorCodeAuthenticationFailed:
+	case errx.MailErrorCodeGoogleAuth, errx.MailErrorCodeAuthenticationFailed, errx.MailErrorCodeInvalidCredentials:
 		return models.JobEventTypeEmailAuthError
 
 	case errx.MailErrorCodeAccountSuspended, errx.MailErrorCodeAuthorizationFailed:

internal/infrastructure/eventbus/bus.go

@@ -76,6 +76,15 @@ type Message struct {
 	Topic   string
 	Key     string
 	Payload []byte
+	// Attempt is the 1-based delivery count of this message (NATS reports it
+	// from the consumer's redelivery metadata; Kafka always reports 1). A
+	// handler that retries by returning an error can read it to know when the
+	// broker is about to stop redelivering and give up cleanly instead.
+	Attempt int
+	// Redelivers is true when a handler error leaves the message for another
+	// delivery (NATS). Kafka commits regardless, so a handler must not count
+	// on a retry there and should finish what it can on this delivery.
+	Redelivers bool
 }
 
 // Subject normalises a topic name to the dot-separated form by replacing any

internal/infrastructure/eventbus/kafka.go

@@ -140,6 +140,7 @@ func (b *KafkaBus) Subscribe(ctx context.Context, topics []string, group string,
 			Topic:   topic,
 			Key:     string(msg.Key),
 			Payload: msg.Value,
+			Attempt: 1,
 		}); err != nil {
 			log.Error().Err(err).Str("topic", topic).Msg("eventbus kafka handler error")
 			return err

internal/infrastructure/eventbus/nats.go

@@ -249,10 +249,16 @@ func (b *NATSBus) Subscribe(ctx context.Context, topics []string, group string,
 		// on the exact separator and should use Subject() if it needs to
 		// compare against a Kafka-style topic name.
 		topic := strings.TrimPrefix(m.Subject(), b.prefix+".")
+		attempt := 1
+		if meta, merr := m.Metadata(); merr == nil && meta.NumDelivered > 0 {
+			attempt = int(meta.NumDelivered)
+		}
 		if err := invokeHandler(hctx, handler, Message{
-			Topic:   topic,
-			Key:     key,
-			Payload: m.Data(),
+			Topic:      topic,
+			Key:        key,
+			Payload:    m.Data(),
+			Attempt:    attempt,
+			Redelivers: true,
 		}); err != nil {
 			log.Error().Err(err).Str("subject", m.Subject()).Msg("eventbus nats handler error")
 			// Nak with a short delay so transient errors don't hot-loop.