LambdaMail/services/protocols/internal/application/usecase/process_outbound_email.go

package usecase

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"strings"
	"time"

	"github.com/alexedwards/argon2id"
	"github.com/google/uuid"

	"lambdamail/protocols/internal/application/port"
	"lambdamail/protocols/internal/domain/entity"
)

var (
	// ErrSubmissionAuthFailed is returned for both an unknown mailbox and a
	// wrong password: telling the two apart would turn submission into an
	// account-enumeration oracle.
	ErrSubmissionAuthFailed = errors.New("process outbound email: authentication failed")

	// ErrSendLimitExceeded is returned when a mailbox has queued more
	// recipients in the last hour than its policy allows.
	ErrSendLimitExceeded = errors.New("process outbound email: sender hourly recipient limit exceeded")

	// ErrSenderNotOwned is returned when the authenticated mailbox tries to
	// send as somebody else.
	ErrSenderNotOwned = errors.New("process outbound email: envelope sender does not belong to the authenticated mailbox")
)

// deliveryWindow is how long a job may keep retrying before it bounces
// (PLAN.md section 6.3).
const deliveryWindow = 5 * 24 * time.Hour

// sendLimitWindow is the period the per-mailbox recipient limit applies over.
const sendLimitWindow = time.Hour

// ProcessOutboundEmailInput is one authenticated submission.
type ProcessOutboundEmailInput struct {
	MailboxID  uuid.UUID
	SenderAddr string
	DomainName string
	Recipients []string
	Body       io.Reader
	// MaxRecipientsPerHour is the sender's policy, carried from the
	// authenticated session.
	MaxRecipientsPerHour int
}

// ProcessOutboundEmailUseCase implements the submission half of PLAN.md
// section 3: authenticate, apply send limits, sign with DKIM, enqueue.
type ProcessOutboundEmailUseCase struct {
	auth     port.AuthRepository
	outbound port.OutboundRepository
	blobs    port.BlobStorage
	signer   port.DkimSigner
}

func NewProcessOutboundEmailUseCase(
	auth port.AuthRepository,
	outbound port.OutboundRepository,
	blobs port.BlobStorage,
	signer port.DkimSigner,
) *ProcessOutboundEmailUseCase {
	return &ProcessOutboundEmailUseCase{auth: auth, outbound: outbound, blobs: blobs, signer: signer}
}

// Authenticate verifies a submission credential and returns the mailbox the
// session may then send as.
func (uc *ProcessOutboundEmailUseCase) Authenticate(ctx context.Context, address, password string) (*port.MailboxAuth, error) {
	rec, err := uc.auth.FindByAddress(ctx, address)
	if err != nil {
		return nil, err
	}
	if rec == nil {
		// Still spend the time a real comparison would, so that a missing
		// mailbox is not distinguishable by response time.
		_, _ = argon2id.ComparePasswordAndHash(password, decoyHash)
		return nil, ErrSubmissionAuthFailed
	}

	match, err := argon2id.ComparePasswordAndHash(password, rec.PasswordHash)
	if err != nil || !match {
		return nil, ErrSubmissionAuthFailed
	}
	return rec, nil
}

// decoyHash is a well-formed Argon2id hash of a value nobody knows. It exists
// only to keep the timing of a failed lookup close to that of a real check.
const decoyHash = "$argon2id$v=19$m=65536,t=3,p=2$YWJjZGVmZ2hpamtsbW5vcA$Zm9yY2VkLWRlY295LWNvbXBhcmlzb24taGFzaA"

// Submit signs the message and queues one delivery job per recipient. The
// message is stored once and every job references the same blob, matching the
// deduplication design of PLAN.md section 9.
func (uc *ProcessOutboundEmailUseCase) Submit(ctx context.Context, input ProcessOutboundEmailInput) error {
	if len(input.Recipients) == 0 {
		return fmt.Errorf("process outbound email: no recipients")
	}

	if err := uc.checkSendLimit(ctx, input); err != nil {
		return err
	}

	payload, err := io.ReadAll(input.Body)
	if err != nil {
		return fmt.Errorf("read submitted message: %w", err)
	}

	if uc.signer != nil {
		signed, err := uc.signer.Sign(ctx, input.DomainName, payload)
		if err != nil {
			return fmt.Errorf("sign submitted message: %w", err)
		}
		payload = signed
	}

	blob, err := uc.blobs.Store(ctx, bytes.NewReader(payload))
	if err != nil {
		return fmt.Errorf("store submitted message: %w", err)
	}

	now := time.Now()
	mailboxID := input.MailboxID

	for _, recipient := range input.Recipients {
		job := &entity.OutboundJob{
			MailboxID:         &mailboxID,
			BlobID:            blob.ID,
			EnvelopeFrom:      input.SenderAddr,
			EnvelopeTo:        recipient,
			DestinationDomain: recipientDomain(recipient),
			Status:            entity.OutboundJobStatusQueued,
			Attempt:           0,
			NextAttemptAt:     now,
			ExpiresAt:         now.Add(deliveryWindow),
		}
		if err := uc.outbound.Enqueue(ctx, job); err != nil {
			return fmt.Errorf("enqueue delivery to %s: %w", recipient, err)
		}
	}

	return nil
}

// checkSendLimit refuses the whole submission when accepting it would take the
// mailbox past its hourly recipient budget. Refusing up front is what contains
// a compromised account; accepting and dropping later would not.
func (uc *ProcessOutboundEmailUseCase) checkSendLimit(ctx context.Context, input ProcessOutboundEmailInput) error {
	if input.MaxRecipientsPerHour <= 0 {
		return nil
	}

	alreadySent, err := uc.outbound.CountRecipientsSince(ctx, input.MailboxID, time.Now().Add(-sendLimitWindow))
	if err != nil {
		return fmt.Errorf("check send limit: %w", err)
	}

	if alreadySent+len(input.Recipients) > input.MaxRecipientsPerHour {
		return fmt.Errorf("%w: %d already sent plus %d requested exceeds %d",
			ErrSendLimitExceeded, alreadySent, len(input.Recipients), input.MaxRecipientsPerHour)
	}
	return nil
}

// AuthorizeSender enforces that a session sends only as itself. Without this
// any account could forge any other address on the server.
func AuthorizeSender(session *port.MailboxAuth, envelopeFrom string) error {
	if session == nil {
		return ErrSubmissionAuthFailed
	}
	if !strings.EqualFold(strings.TrimSpace(envelopeFrom), session.EmailAddress) {
		return ErrSenderNotOwned
	}
	return nil
}

func recipientDomain(address string) string {
	if idx := strings.LastIndex(address, "@"); idx >= 0 && idx < len(address)-1 {
		return strings.ToLower(address[idx+1:])
	}
	return ""
}