1 files changed, 152 insertions, 0 deletions
diff --git a/internal/controlplane/outbox.go b/internal/controlplane/outbox.go
new file mode 100644
index 0000000..b54a064
--- /dev/null
+++ b/internal/controlplane/outbox.go
@@ -0,0 +1,152 @@
+package controlplane
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/url"
+	"strings"
+	"time"
+
+	"aigw/internal/mailer"
+
+	"github.com/jackc/pgx/v5"
+)
+
+const (
+	actionVerifyEmail   = "verify_email"
+	actionPasswordReset = "password_reset"
+	actionInvite        = "invite"
+)
+
+func (s *Store) issueActionToken(ctx context.Context, tx pgx.Tx, userID, email, displayName, purpose, publicURL, remoteIP string, ttl time.Duration) (string, error) {
+	token, tokenHash, err := randomCredential("act-aigw-")
+	if err != nil {
+		return "", err
+	}
+	if _, err := tx.Exec(ctx, `UPDATE console_action_tokens SET consumed_at=now()
+		WHERE user_id=$1 AND purpose=$2 AND consumed_at IS NULL`, userID, purpose); err != nil {
+		return "", fmt.Errorf("expire previous account action: %w", err)
+	}
+	if _, err := tx.Exec(ctx, `INSERT INTO console_action_tokens
+		(user_id,purpose,token_hash,expires_at,requested_ip) VALUES ($1,$2,$3,$4,NULLIF($5,'')::inet)`,
+		userID, purpose, tokenHash, time.Now().UTC().Add(ttl), remoteIP); err != nil {
+		return "", fmt.Errorf("create account action: %w", err)
+	}
+	link, err := accountActionURL(publicURL, purpose, token)
+	if err != nil {
+		return "", err
+	}
+	subject, body := actionMail(displayName, purpose, link, ttl)
+	ciphertext, err := s.cipher.Encrypt(body)
+	if err != nil {
+		return "", fmt.Errorf("encrypt account email: %w", err)
+	}
+	if _, err := tx.Exec(ctx, `INSERT INTO console_mail_outbox
+		(recipient,template,subject,body_ciphertext) VALUES ($1,$2,$3,$4)`, email, purpose, subject, ciphertext); err != nil {
+		return "", fmt.Errorf("queue account email: %w", err)
+	}
+	return token, nil
+}
+
+func accountActionURL(publicURL, purpose, token string) (string, error) {
+	parsed, err := url.Parse(publicURL)
+	if err != nil || parsed.Host == "" {
+		return "", errors.New("console public URL is invalid")
+	}
+	query := parsed.Query()
+	switch purpose {
+	case actionVerifyEmail:
+		query.Set("action", "verify-email")
+	case actionPasswordReset:
+		query.Set("action", "reset-password")
+	case actionInvite:
+		query.Set("action", "accept-invite")
+	default:
+		return "", errors.New("unsupported account action")
+	}
+	query.Set("token", token)
+	parsed.RawQuery = query.Encode()
+	return parsed.String(), nil
+}
+
+func actionMail(displayName, purpose, link string, ttl time.Duration) (string, string) {
+	name := strings.TrimSpace(displayName)
+	if name == "" {
+		name = "there"
+	}
+	minutes := int(ttl.Minutes())
+	expires := fmt.Sprintf("%d minutes", minutes)
+	if minutes%1440 == 0 {
+		expires = durationLabel(minutes/1440, "day")
+	} else if minutes%60 == 0 {
+		expires = durationLabel(minutes/60, "hour")
+	}
+	switch purpose {
+	case actionVerifyEmail:
+		return "Verify your AIGW email", fmt.Sprintf("Hi %s,\n\nVerify your email to activate your AIGW workspace:\n\n%s\n\nThis link expires in %s and can be used once. If you did not create this account, ignore this email.\n", name, link, expires)
+	case actionPasswordReset:
+		return "Reset your AIGW password", fmt.Sprintf("Hi %s,\n\nReset your AIGW password:\n\n%s\n\nThis link expires in %s and can be used once. If you did not request this, your password has not changed.\n", name, link, expires)
+	default:
+		return "You have been invited to AIGW", fmt.Sprintf("Hi %s,\n\nAccept your AIGW invitation and choose a password:\n\n%s\n\nThis link expires in %s and can be used once.\n", name, link, expires)
+	}
+}
+
+func durationLabel(value int, unit string) string {
+	if value != 1 {
+		unit += "s"
+	}
+	return fmt.Sprintf("%d %s", value, unit)
+}
+
+func (s *Store) ClaimMail(ctx context.Context) (mailer.Message, bool, error) {
+	tx, err := s.db.Begin(ctx)
+	if err != nil {
+		return mailer.Message{}, false, err
+	}
+	defer tx.Rollback(ctx)
+	var message mailer.Message
+	var ciphertext []byte
+	err = tx.QueryRow(ctx, `WITH candidate AS (
+		SELECT id FROM console_mail_outbox
+		WHERE ((status IN ('pending','failed') AND available_at <= now())
+		   OR (status='sending' AND claimed_at < now()-interval '5 minutes'))
+		ORDER BY available_at,created_at FOR UPDATE SKIP LOCKED LIMIT 1
+	) UPDATE console_mail_outbox o SET status='sending',claimed_at=now(),attempts=attempts+1,last_error=''
+	FROM candidate WHERE o.id=candidate.id
+	RETURNING o.id::text,o.recipient,o.subject,o.body_ciphertext`,
+	).Scan(&message.ID, &message.Recipient, &message.Subject, &ciphertext)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return mailer.Message{}, false, nil
+	}
+	if err != nil {
+		return mailer.Message{}, false, fmt.Errorf("claim mail outbox item: %w", err)
+	}
+	message.Body, err = s.cipher.Decrypt(ciphertext)
+	if err != nil {
+		return mailer.Message{}, false, fmt.Errorf("decrypt mail outbox item: %w", err)
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return mailer.Message{}, false, err
+	}
+	return message, true, nil
+}
+
+func (s *Store) MarkMailSent(ctx context.Context, id string) error {
+	_, err := s.db.Exec(ctx, `UPDATE console_mail_outbox SET status='sent',sent_at=now(),claimed_at=NULL,last_error=''
+		WHERE id=$1 AND status='sending'`, id)
+	return err
+}
+
+func (s *Store) MarkMailFailed(ctx context.Context, id string, deliveryErr error) error {
+	message := "delivery failed"
+	if deliveryErr != nil {
+		message = deliveryErr.Error()
+	}
+	if len(message) > 1000 {
+		message = message[:1000]
+	}
+	_, err := s.db.Exec(ctx, `UPDATE console_mail_outbox SET status='failed',claimed_at=NULL,last_error=$2,
+		available_at=now()+make_interval(secs => LEAST(300, 5 * attempts)) WHERE id=$1 AND status='sending'`, id, message)
+	return err
+}