Add Cloudflare email and collaboration workflows

apps/server/internal/email/cloudflare.go

@@ -0,0 +1,160 @@
+package email
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+)
+
+const (
+	defaultCloudflareAPIURL = "https://api.cloudflare.com/client/v4"
+	cloudflareTimeout       = 15 * time.Second
+)
+
+// CloudflareSender sends transactional email through Cloudflare Email
+// Service's REST API. It is intended for deployments outside Workers.
+type CloudflareSender struct {
+	accountID string
+	apiToken  string
+	apiURL    string
+	from      string
+	client    *http.Client
+	logger    *slog.Logger
+}
+
+func NewCloudflareSender(accountID, apiToken, from, apiURL string, logger *slog.Logger) *CloudflareSender {
+	if apiURL == "" {
+		apiURL = defaultCloudflareAPIURL
+	}
+	if logger == nil {
+		logger = slog.Default()
+	}
+	return &CloudflareSender{
+		accountID: accountID,
+		apiToken:  apiToken,
+		apiURL:    strings.TrimRight(apiURL, "/"),
+		from:      from,
+		client: &http.Client{
+			Timeout: cloudflareTimeout,
+			CheckRedirect: func(req *http.Request, via []*http.Request) error {
+				return http.ErrUseLastResponse
+			},
+		},
+		logger: logger,
+	}
+}
+
+type cloudflareRequest struct {
+	From    string   `json:"from"`
+	To      []string `json:"to"`
+	Subject string   `json:"subject"`
+	Text    string   `json:"text"`
+}
+
+type cloudflareResponse struct {
+	Success bool `json:"success"`
+	Errors  []struct {
+		Code    int    `json:"code"`
+		Message string `json:"message"`
+	} `json:"errors"`
+	Result *struct {
+		Delivered        []string `json:"delivered"`
+		MessageID        string   `json:"message_id"`
+		PermanentBounces []string `json:"permanent_bounces"`
+		Queued           []string `json:"queued"`
+	} `json:"result"`
+}
+
+func (c *CloudflareSender) Send(ctx context.Context, to []string, subject, body string) error {
+	if c.accountID == "" {
+		return &PermanentError{Cause: fmt.Errorf("cloudflare account id not configured")}
+	}
+	if c.apiToken == "" {
+		return &PermanentError{Cause: fmt.Errorf("cloudflare API token not configured")}
+	}
+	if c.from == "" {
+		return &PermanentError{Cause: fmt.Errorf("cloudflare from address not configured")}
+	}
+	if len(to) == 0 {
+		return &PermanentError{Cause: fmt.Errorf("cloudflare send requires at least one recipient")}
+	}
+
+	payload, err := json.Marshal(cloudflareRequest{
+		From:    c.from,
+		To:      to,
+		Subject: subject,
+		Text:    body,
+	})
+	if err != nil {
+		return fmt.Errorf("marshal cloudflare email request: %w", err)
+	}
+
+	endpoint := fmt.Sprintf("%s/accounts/%s/email/sending/send", c.apiURL, url.PathEscape(c.accountID))
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(payload))
+	if err != nil {
+		return fmt.Errorf("build cloudflare email request: %w", err)
+	}
+	req.Header.Set("Authorization", "Bearer "+c.apiToken)
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json")
+
+	resp, err := c.client.Do(req)
+	if err != nil {
+		return &TemporaryError{Cause: fmt.Errorf("cloudflare email request: %w", err)}
+	}
+	defer resp.Body.Close()
+
+	responseBody, readErr := io.ReadAll(io.LimitReader(resp.Body, 1<<14))
+	if readErr != nil {
+		return &TemporaryError{Cause: fmt.Errorf("read cloudflare email response: %w", readErr)}
+	}
+	if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= http.StatusInternalServerError {
+		return &TemporaryError{Cause: fmt.Errorf("cloudflare email server error: status=%d body=%s", resp.StatusCode, truncate(string(responseBody)))}
+	}
+	if resp.StatusCode != http.StatusOK {
+		return &PermanentError{Cause: fmt.Errorf("cloudflare email rejected: status=%d body=%s", resp.StatusCode, truncate(string(responseBody)))}
+	}
+
+	var parsed cloudflareResponse
+	if err := json.Unmarshal(responseBody, &parsed); err != nil {
+		return fmt.Errorf("decode cloudflare email response: %w", err)
+	}
+	if !parsed.Success {
+		return &PermanentError{Cause: fmt.Errorf("cloudflare email rejected: %s", cloudflareErrors(parsed.Errors))}
+	}
+	if parsed.Result == nil {
+		return fmt.Errorf("cloudflare email response missing result")
+	}
+	if len(parsed.Result.PermanentBounces) > 0 {
+		return &PermanentError{Cause: fmt.Errorf("cloudflare email permanently bounced for %d recipient(s)", len(parsed.Result.PermanentBounces))}
+	}
+
+	c.logger.Info("cloudflare email accepted",
+		"message_id", parsed.Result.MessageID,
+		"delivered", len(parsed.Result.Delivered),
+		"queued", len(parsed.Result.Queued),
+		"subject", subject,
+	)
+	return nil
+}
+
+func cloudflareErrors(errors []struct {
+	Code    int    `json:"code"`
+	Message string `json:"message"`
+}) string {
+	if len(errors) == 0 {
+		return "unknown API error"
+	}
+	parts := make([]string, 0, len(errors))
+	for _, apiErr := range errors {
+		parts = append(parts, fmt.Sprintf("code=%d: %s", apiErr.Code, apiErr.Message))
+	}
+	return strings.Join(parts, "; ")
+}

apps/server/internal/email/cloudflare_test.go

@@ -0,0 +1,128 @@
+package email
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+func TestCloudflareSenderSend(t *testing.T) {
+	var got cloudflareRequest
+	var gotAuthorization string
+	var gotPath string
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotAuthorization = r.Header.Get("Authorization")
+		gotPath = r.URL.Path
+		body, _ := io.ReadAll(r.Body)
+		_ = json.Unmarshal(body, &got)
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{"success":true,"errors":[],"messages":[],"result":{"delivered":["alice@example.com"],"message_id":"message-1","permanent_bounces":[],"queued":[]}}`))
+	}))
+	defer server.Close()
+
+	sender := NewCloudflareSender("account-1", "token-1", "notifications@example.com", server.URL, testLogger(t))
+	if err := sender.Send(context.Background(), []string{"alice@example.com"}, "Hello", "World"); err != nil {
+		t.Fatalf("Send: %v", err)
+	}
+	if gotAuthorization != "Bearer token-1" {
+		t.Errorf("authorization = %q", gotAuthorization)
+	}
+	if gotPath != "/accounts/account-1/email/sending/send" {
+		t.Errorf("path = %q", gotPath)
+	}
+	if got.From != "notifications@example.com" || got.Subject != "Hello" || got.Text != "World" {
+		t.Errorf("request = %#v", got)
+	}
+	if len(got.To) != 1 || got.To[0] != "alice@example.com" {
+		t.Errorf("to = %#v", got.To)
+	}
+}
+
+func TestCloudflareSenderRetryableResponses(t *testing.T) {
+	for _, status := range []int{http.StatusTooManyRequests, http.StatusInternalServerError} {
+		t.Run(http.StatusText(status), func(t *testing.T) {
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(status)
+				_, _ = w.Write([]byte(`{"success":false}`))
+			}))
+			defer server.Close()
+
+			sender := NewCloudflareSender("account", "token", "from@example.com", server.URL, testLogger(t))
+			err := sender.Send(context.Background(), []string{"to@example.com"}, "subject", "body")
+			var temporary *TemporaryError
+			if !errors.As(err, &temporary) {
+				t.Fatalf("error = %T %v, want TemporaryError", err, err)
+			}
+		})
+	}
+}
+
+func TestCloudflareSenderPermanentRejection(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusBadRequest)
+		_, _ = w.Write([]byte(`{"success":false,"errors":[{"code":1000,"message":"Sender domain not verified"}]}`))
+	}))
+	defer server.Close()
+
+	sender := NewCloudflareSender("account", "token", "from@example.com", server.URL, testLogger(t))
+	err := sender.Send(context.Background(), []string{"to@example.com"}, "subject", "body")
+	if err == nil || !strings.Contains(err.Error(), "400") {
+		t.Fatalf("error = %v, want permanent 400 rejection", err)
+	}
+	var temporary *TemporaryError
+	if errors.As(err, &temporary) {
+		t.Fatalf("error = %T, should not be retryable", err)
+	}
+}
+
+func TestCloudflareSenderAPIErrorAndBounce(t *testing.T) {
+	tests := []struct {
+		name string
+		body string
+		want string
+	}{
+		{"api error", `{"success":false,"errors":[{"code":1000,"message":"Sender domain not verified"}],"result":null}`, "Sender domain not verified"},
+		{"bounce", `{"success":true,"errors":[],"result":{"delivered":[],"message_id":"message-1","permanent_bounces":["to@example.com"],"queued":[]}}`, "permanently bounced"},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				_, _ = w.Write([]byte(test.body))
+			}))
+			defer server.Close()
+			sender := NewCloudflareSender("account", "token", "from@example.com", server.URL, testLogger(t))
+			err := sender.Send(context.Background(), []string{"to@example.com"}, "subject", "body")
+			if err == nil || !strings.Contains(err.Error(), test.want) {
+				t.Fatalf("error = %v, want substring %q", err, test.want)
+			}
+		})
+	}
+}
+
+func TestCloudflareSenderValidation(t *testing.T) {
+	tests := []struct {
+		name      string
+		accountID string
+		token     string
+		from      string
+		to        []string
+	}{
+		{"account", "", "token", "from@example.com", []string{"to@example.com"}},
+		{"token", "account", "", "from@example.com", []string{"to@example.com"}},
+		{"from", "account", "token", "", []string{"to@example.com"}},
+		{"recipient", "account", "token", "from@example.com", nil},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			sender := NewCloudflareSender(test.accountID, test.token, test.from, "", testLogger(t))
+			if err := sender.Send(context.Background(), test.to, "subject", "body"); err == nil {
+				t.Fatal("expected validation error")
+			}
+		})
+	}
+}

apps/server/internal/email/postmark.go

@@ -36,16 +36,21 @@ func NewPostmarkSender(serverToken, from, apiURL string, logger *slog.Logger) *P
 		serverToken: serverToken,
 		apiURL:      strings.TrimRight(apiURL, "/"),
 		from:        from,
-		client:      &http.Client{Timeout: postmarkTimeout},
-		logger:      logger,
+		client: &http.Client{
+			Timeout: postmarkTimeout,
+			CheckRedirect: func(req *http.Request, via []*http.Request) error {
+				return http.ErrUseLastResponse
+			},
+		},
+		logger: logger,
 	}
 }
 
 type postmarkRequest struct {
-	From     string `json:"From"`
-	To       string `json:"To"`
-	Subject  string `json:"Subject"`
-	TextBody string `json:"TextBody"`
+	From     string           `json:"From"`
+	To       string           `json:"To"`
+	Subject  string           `json:"Subject"`
+	TextBody string           `json:"TextBody"`
 	Headers  []postmarkHeader `json:"Headers,omitempty"`
 }
 
@@ -66,13 +71,13 @@ type postmarkResponse struct {
 // send if they need different reply-to addresses.
 func (p *PostmarkSender) Send(ctx context.Context, to []string, subject, body string) error {
 	if p.serverToken == "" {
-		return fmt.Errorf("postmark server token not configured")
+		return &PermanentError{Cause: fmt.Errorf("postmark server token not configured")}
 	}
 	if p.from == "" {
-		return fmt.Errorf("postmark from address not configured")
+		return &PermanentError{Cause: fmt.Errorf("postmark from address not configured")}
 	}
 	if len(to) == 0 {
-		return fmt.Errorf("postmark send requires at least one recipient")
+		return &PermanentError{Cause: fmt.Errorf("postmark send requires at least one recipient")}
 	}
 
 	payload := postmarkRequest{
@@ -96,7 +101,7 @@ func (p *PostmarkSender) Send(ctx context.Context, to []string, subject, body st
 
 	resp, err := p.client.Do(req)
 	if err != nil {
-		return fmt.Errorf("postmark request: %w", err)
+		return &TemporaryError{Cause: fmt.Errorf("postmark request: %w", err)}
 	}
 	defer resp.Body.Close()
 
@@ -107,7 +112,7 @@ func (p *PostmarkSender) Send(ctx context.Context, to []string, subject, body st
 	}
 
 	if resp.StatusCode != http.StatusOK {
-		return fmt.Errorf("postmark rejected: status=%d body=%s", resp.StatusCode, truncate(string(respBytes)))
+		return &PermanentError{Cause: fmt.Errorf("postmark rejected: status=%d body=%s", resp.StatusCode, truncate(string(respBytes)))}
 	}
 
 	var parsed postmarkResponse
@@ -117,7 +122,7 @@ func (p *PostmarkSender) Send(ctx context.Context, to []string, subject, body st
 	if parsed.ErrorCode != 0 {
 		// Postmark distinguishes "inactive recipient" (406) and other errors. Treat
 		// 406 as non-retryable; others bubble up as generic failures.
-		return fmt.Errorf("postmark error code=%d: %s", parsed.ErrorCode, parsed.Message)
+		return &PermanentError{Cause: fmt.Errorf("postmark error code=%d: %s", parsed.ErrorCode, parsed.Message)}
 	}
 
 	p.logger.Info("postmark email sent", "to", parsed.To, "subject", subject)
@@ -139,3 +144,11 @@ type TemporaryError struct {
 
 func (t *TemporaryError) Error() string { return t.Cause.Error() }
 func (t *TemporaryError) Unwrap() error { return t.Cause }
+
+// PermanentError marks a provider rejection that cannot succeed on retry.
+type PermanentError struct {
+	Cause error
+}
+
+func (p *PermanentError) Error() string { return p.Cause.Error() }
+func (p *PermanentError) Unwrap() error { return p.Cause }

apps/server/internal/email/queue.go

@@ -3,6 +3,7 @@ package email
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"fmt"
 	"log/slog"
 	"math"
@@ -211,17 +212,28 @@ func (q *Queue) deliver(ctx context.Context, row QueueRow) error {
 	}
 
 	// Failure: retry or give up.
-	if row.Attempts >= row.MaxAttempts {
+	var permanent *PermanentError
+	if errors.As(err, &permanent) {
+		if _, ferr := q.db.ExecContext(ctx, `
+			UPDATE email_queue SET status = 'failed', last_error = ? WHERE id = ?
+		`, truncateErr(err.Error()), row.ID); ferr != nil {
+			return fmt.Errorf("mark permanent failure: %w", ferr)
+		}
+		q.logger.Warn("email permanently rejected", "id", row.ID, "attempts", row.Attempts+1, "error", err)
+		return nil
+	}
+	currentAttempt := row.Attempts + 1
+	if currentAttempt >= row.MaxAttempts {
 		if _, ferr := q.db.ExecContext(ctx, `
 			UPDATE email_queue SET status = 'failed', last_error = ? WHERE id = ?
 		`, truncateErr(err.Error()), row.ID); ferr != nil {
 			return fmt.Errorf("mark failed: %w", ferr)
 		}
-		q.logger.Warn("email permanently failed", "id", row.ID, "attempts", row.Attempts, "error", err)
+		q.logger.Warn("email permanently failed", "id", row.ID, "attempts", currentAttempt, "error", err)
 		return nil
 	}
 
-	backoff := q.backoff(row.Attempts)
+	backoff := q.backoff(currentAttempt)
 	next := time.Now().UTC().Add(backoff)
 	if _, err := q.db.ExecContext(ctx, `
 		UPDATE email_queue SET status = 'pending', next_attempt_at = ?, last_error = ? WHERE id = ?
@@ -229,9 +241,9 @@ func (q *Queue) deliver(ctx context.Context, row QueueRow) error {
 		return fmt.Errorf("schedule retry: %w", err)
 	}
 
-	// TemporaryError (Postmark 5xx) is retried with the same backoff as
-	// other errors; the max-attempts cap still bounds total work.
-	q.logger.Info("email retry scheduled", "id", row.ID, "attempt", row.Attempts, "backoff", backoff, "error", err)
+	// Temporary provider failures and unclassified sender failures are retried;
+	// PermanentError is handled above without wasting additional attempts.
+	q.logger.Info("email retry scheduled", "id", row.ID, "attempt", currentAttempt, "backoff", backoff, "error", err)
 	return nil
 }
 

apps/server/internal/email/queue_test.go

@@ -5,6 +5,7 @@ import (
 	"database/sql"
 	"errors"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -152,9 +153,9 @@ func TestQueue_RetryThenSucceed(t *testing.T) {
 	go q.Run(ctx)
 
 	if err := waitFor(ctx, func() bool {
-		return sender.successCount >= 1
+		return sender.successCount.Load() >= 1
 	}); err != nil {
-		t.Fatalf("email never succeeded: successCount=%d attempts=%d: %v", sender.successCount, sender.attempts, err)
+		t.Fatalf("email never succeeded: successCount=%d attempts=%d: %v", sender.successCount.Load(), sender.attempts.Load(), err)
 	}
 
 	// Verify the row is marked sent.
@@ -196,6 +197,32 @@ func TestQueue_PermanentFailure(t *testing.T) {
 	}
 }
 
+func TestQueue_ProviderRejectionFailsImmediately(t *testing.T) {
+	db := newTestDB(t)
+	sender := &recordingSender{err: &PermanentError{Cause: errors.New("sender domain not verified")}}
+	q := NewQueue(db, sender, testLogger(t))
+
+	id, err := q.Enqueue(context.Background(), "notif:test-1", "alice@example.com", "Fail", "Body")
+	if err != nil {
+		t.Fatalf("Enqueue: %v", err)
+	}
+	rows, err := q.claimBatch(context.Background(), 1)
+	if err != nil || len(rows) != 1 {
+		t.Fatalf("claimBatch = %#v, %v", rows, err)
+	}
+	if err := q.deliver(context.Background(), rows[0]); err != nil {
+		t.Fatalf("deliver: %v", err)
+	}
+	var status string
+	var attempts int
+	if err := db.QueryRow(`SELECT status, attempts FROM email_queue WHERE id = ?`, id).Scan(&status, &attempts); err != nil {
+		t.Fatalf("query row: %v", err)
+	}
+	if status != "failed" || attempts != 1 {
+		t.Fatalf("status=%q attempts=%d, want failed after one attempt", status, attempts)
+	}
+}
+
 func TestQueue_NilSenderLeavesPending(t *testing.T) {
 	db := newTestDB(t)
 	q := NewQueue(db, nil, testLogger(t))
@@ -218,17 +245,17 @@ func TestQueue_NilSenderLeavesPending(t *testing.T) {
 
 // flakySender fails the first N attempts then succeeds.
 type flakySender struct {
-	failTimes    int
-	attempts     int
-	successCount int
+	failTimes    int32
+	attempts     atomic.Int32
+	successCount atomic.Int32
 }
 
 func (f *flakySender) Send(ctx context.Context, to []string, subject, body string) error {
-	f.attempts++
-	if f.attempts <= f.failTimes {
+	attempt := f.attempts.Add(1)
+	if attempt <= f.failTimes {
 		return errors.New("transient failure")
 	}
-	f.successCount++
+	f.successCount.Add(1)
 	return nil
 }
 

apps/server/internal/email/templates.go

@@ -9,16 +9,16 @@ import (
 // TemplateData carries the context required to render any notification email.
 // Fields are optional per template; callers should populate what they have.
 type TemplateData struct {
-	ActorName    string
-	ActorEmail   string
-	DocumentPath string
+	ActorName     string
+	ActorEmail    string
+	DocumentPath  string
 	DocumentTitle string
-	CommentBody  string
-	MentionText  string
-	ConflictDesc string
-	ViewURL      string
-	UnsubURL     string
-	InstanceName string
+	CommentBody   string
+	MentionText   string
+	ConflictDesc  string
+	ViewURL       string
+	UnsubURL      string
+	InstanceName  string
 }
 
 // Render returns a plain-text subject and body for the given notification
@@ -101,7 +101,7 @@ func renderComment(data TemplateData) (string, string, error) {
 	} else {
 		b.WriteString("> (no body)\n")
 	}
-	b.WriteString("\nReply to this email to respond.\n")
+	b.WriteString("\nOpen the document to reply.\n")
 	b.WriteString(footer(data))
 	return subject, b.String(), nil
 }

apps/server/internal/email/templates_test.go

@@ -7,12 +7,12 @@ import (
 
 func TestRender_FileChanged(t *testing.T) {
 	subject, body, err := Render("file_changed", TemplateData{
-		ActorName:    "Alice",
-		DocumentPath: "docs/api.md",
+		ActorName:     "Alice",
+		DocumentPath:  "docs/api.md",
 		DocumentTitle: "API Reference",
-		ViewURL:      "https://example.com/docs/api.md",
-		UnsubURL:     "https://example.com/unsub/123",
-		InstanceName: "Cairnquire",
+		ViewURL:       "https://example.com/docs/api.md",
+		UnsubURL:      "https://example.com/unsub/123",
+		InstanceName:  "Cairnquire",
 	})
 	if err != nil {
 		t.Fatalf("Render: %v", err)
@@ -39,10 +39,10 @@ func TestRender_FileChanged(t *testing.T) {
 
 func TestRender_Comment(t *testing.T) {
 	subject, body, err := Render("comment", TemplateData{
-		ActorName:    "Bob",
+		ActorName:     "Bob",
 		DocumentTitle: "Getting Started",
-		CommentBody:  "This step is unclear.\nCan we add an example?",
-		ViewURL:      "https://example.com/docs/getting-started",
+		CommentBody:   "This step is unclear.\nCan we add an example?",
+		ViewURL:       "https://example.com/docs/getting-started",
 	})
 	if err != nil {
 		t.Fatalf("Render: %v", err)
@@ -56,16 +56,16 @@ func TestRender_Comment(t *testing.T) {
 	if !strings.Contains(body, "> Can we add an example?") {
 		t.Errorf("body should quote second line: %q", body)
 	}
-	if !strings.Contains(body, "Reply to this email") {
-		t.Errorf("body should mention reply-to-respond: %q", body)
+	if !strings.Contains(body, "Open the document to reply") {
+		t.Errorf("body should link users to the supported reply flow: %q", body)
 	}
 }
 
 func TestRender_Mention(t *testing.T) {
 	subject, body, err := Render("mention", TemplateData{
-		ActorName:    "Carol",
+		ActorName:     "Carol",
 		DocumentTitle: "Spec",
-		MentionText:  "@dan please review",
+		MentionText:   "@dan please review",
 	})
 	if err != nil {
 		t.Fatalf("Render: %v", err)