Serialize email verification and resend to preserve registration state

internal/auth/email.go

@@ -112,6 +112,9 @@ func (s *service) registerEmail(w http.ResponseWriter, r *http.Request, project,
 		return err
 	}
 	defer tx.Rollback()
+	if _, err = tx.ExecContext(r.Context(), "SELECT pg_advisory_xact_lock(hashtextextended($1,0))", "email:"+project+":"+email); err != nil {
+		return err
+	}
 	verification := token()
 	result, err := tx.ExecContext(r.Context(), `
 INSERT INTO email_registrations(project,email,password_hash,token_hash,expires_at)
@@ -167,6 +170,16 @@ func (s *service) verifyEmail(w http.ResponseWriter, r *http.Request, project st
 	}
 	defer tx.Rollback()
 	var email, hash string
+	err = tx.QueryRowContext(r.Context(), "SELECT email FROM email_registrations WHERE project=$1 AND token_hash=$2 AND expires_at>now()", project, digest(input.Token)).Scan(&email)
+	if errors.Is(err, sql.ErrNoRows) {
+		return reject(w, 400, "invalid_verification")
+	}
+	if err != nil {
+		return err
+	}
+	if _, err = tx.ExecContext(r.Context(), "SELECT pg_advisory_xact_lock(hashtextextended($1,0))", "email:"+project+":"+email); err != nil {
+		return err
+	}
 	err = tx.QueryRowContext(r.Context(), `
 DELETE FROM email_registrations WHERE project=$1 AND token_hash=$2 AND expires_at>now() RETURNING email,password_hash
 `, project, digest(input.Token)).Scan(&email, &hash)

internal/auth/email_test.go

@@ -0,0 +1,72 @@
+package auth
+
+import (
+	"crypto/rand"
+	"database/sql"
+	"encoding/hex"
+	"net/http/httptest"
+	"net/url"
+	"os"
+	"regexp"
+	"strings"
+	"testing"
+)
+
+func TestConcurrentRegistrationAndVerification(t *testing.T) {
+	dsn := os.Getenv("PLATFORM_TEST_DATABASE_URL")
+	if dsn == "" {
+		t.Skip("requires isolated integration database")
+	}
+	db, err := sql.Open("pgx", dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	var raw [12]byte
+	rand.Read(raw[:])
+	project := "email-" + hex.EncodeToString(raw[:])
+	frontend, _ := url.Parse("https://app.example.com/auth/callback")
+	s := &service{db: db, projects: map[string]*url.URL{project: frontend}}
+	defer func() {
+		_, _ = db.Exec("DELETE FROM project_accounts WHERE project=$1", project)
+		_, _ = db.Exec("DELETE FROM email_registrations WHERE project=$1", project)
+		_, _ = db.Exec("DELETE FROM email_outbox WHERE project=$1", project)
+	}()
+	request := httptest.NewRequest("POST", "https://api.example.com/register", nil)
+	out := httptest.NewRecorder()
+	if err = s.registerEmail(out, request, project, "fixture@example.invalid", "long-fixture-password"); err != nil {
+		t.Fatal(err)
+	}
+	var body string
+	if err = db.QueryRow("SELECT body FROM email_outbox WHERE project=$1", project).Scan(&body); err != nil {
+		t.Fatal(err)
+	}
+	match := regexp.MustCompile(`verify_email=([A-Za-z0-9_-]{43})`).FindStringSubmatch(body)
+	if len(match) != 2 {
+		t.Fatal("missing verification link")
+	}
+	done := make(chan error, 2)
+	go func() {
+		r := httptest.NewRequest("POST", "https://api.example.com/register", nil)
+		done <- s.registerEmail(httptest.NewRecorder(), r, project, "fixture@example.invalid", "another-fixture-password")
+	}()
+	go func() {
+		r := httptest.NewRequest("POST", "https://api.example.com/verify-email", strings.NewReader(`{"token":"`+match[1]+`"}`))
+		done <- s.verifyEmail(httptest.NewRecorder(), r, project)
+	}()
+	for range 2 {
+		if err := <-done; err != nil {
+			t.Fatal(err)
+		}
+	}
+	var accounts, pending int
+	if err = db.QueryRow("SELECT count(*) FROM email_credentials WHERE project=$1", project).Scan(&accounts); err != nil {
+		t.Fatal(err)
+	}
+	if err = db.QueryRow("SELECT count(*) FROM email_registrations WHERE project=$1", project).Scan(&pending); err != nil {
+		t.Fatal(err)
+	}
+	if accounts+pending != 1 {
+		t.Fatalf("concurrent registration left accounts=%d pending=%d", accounts, pending)
+	}
+}