From 555241df5cd6fc3c6ec84400aa019f27d6b16fcf Mon Sep 17 00:00:00 2001 From: Anirudh Oppiliappan Date: Tue, 25 Aug 2026 13:48:38 +0300 Subject: [PATCH] cmd/email-did-migrate: bulk-load verified emails from existing appview.db Signed-off-by: Anirudh Oppiliappan --- cmd/email-did-migrate/main.go | 272 +++++++++++++++++++++++++++++ cmd/email-did-migrate/main_test.go | 207 ++++++++++++++++++++++ 2 files changed, 479 insertions(+) create mode 100644 cmd/email-did-migrate/main.go create mode 100644 cmd/email-did-migrate/main_test.go diff --git a/cmd/email-did-migrate/main.go b/cmd/email-did-migrate/main.go new file mode 100644 index 000000000..f784340da --- /dev/null +++ b/cmd/email-did-migrate/main.go @@ -0,0 +1,272 @@ +package main + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "net/http" + "net/url" + "os" + "sort" + "strings" + "time" + + _ "github.com/mattn/go-sqlite3" +) + +// email-did-migrate bulk-loads the appview's verified emails into the +// EMAIL_DID KV namespace that the web worker reads. one-time, at cutover: +// +// email-did-migrate +// +// phase 1 writes each verified email (lowercased — the worker looks up +// lowercased keys) → did; phase 2 writes `__primary:` → primary email. +// writes are idempotent, so a failed run can simply be re-run. + +var httpClient = &http.Client{Timeout: 30 * time.Second} + +// ~20 writes/s keeps us comfortably inside the KV REST write quota +// (~1200/min) over a multi-minute run. +const writeDelay = 50 * time.Millisecond + +func main() { + flag.Parse() + if flag.NArg() != 4 { + fmt.Fprintf(os.Stderr, "usage: %s \n", os.Args[0]) + os.Exit(1) + } + dbPath, apiToken, accountID, nsID := flag.Arg(0), flag.Arg(1), flag.Arg(2), flag.Arg(3) + + // read-only: the appview keeps serving while the mapping is copied out + db, err := sql.Open("sqlite3", dbPath+"?_mode=ro") + if err != nil { + fatal("open source: %v", err) + } + defer db.Close() + + baseURL := fmt.Sprintf( + "https://api.cloudflare.com/client/v4/accounts/%s/storage/kv/namespaces/%s/", + accountID, nsID, + ) + + start := time.Now() + if err := run(context.Background(), db, apiToken, baseURL); err != nil { + fatal("%v", err) + } + fmt.Printf("migration complete in %s\n", time.Since(start).Round(time.Second)) +} + +type emailRow struct { + email string + did string + primary bool +} + +func run(ctx context.Context, db *sql.DB, apiToken, baseURL string) error { + verified, err := queryEmails(ctx, db, `select email, did, is_primary from emails where verified = true order by email, did`) + if err != nil { + return fmt.Errorf("phase 1 query: %w", err) + } + primary, err := queryEmails(ctx, db, `select email, did, is_primary from emails where verified = true and is_primary = true order by email, did`) + if err != nil { + return fmt.Errorf("phase 2 query: %w", err) + } + + // the worker reads lowercased keys, but the appview schema only forbids + // exact-case duplicates per did — `A@x.com` and `a@x.com` can both be + // verified for different accounts. one address only ever answers with one + // did, so resolve collisions up front: the account that marked the address + // primary wins, else the first row in (email, did) order, and warn. + byKey := make(map[string]*emailRow, len(verified)) + for i := range verified { + r := verified[i] + key := strings.ToLower(r.email) + if prev, ok := byKey[key]; ok { + if r.primary && !prev.primary { + byKey[key] = &r + } + fmt.Printf(" warning: %q verified by %s and %s, keeping %s\n", key, prev.did, r.did, byKey[key].did) + continue + } + byKey[key] = &r + } + + emailKeys := orderKeys(byKey) + primaryByKey := make(map[string]string, len(primary)) + for _, r := range primary { + primaryByKey["__primary:"+r.did] = r.email + } + primaryKeys := orderKeys(primaryByKey) + + fmt.Printf(" %d verified emails (%d distinct keys), %d primary\n", len(verified), len(emailKeys), len(primaryKeys)) + + for _, key := range emailKeys { + if err := putValue(ctx, apiToken, baseURL, key, byKey[key].did); err != nil { + return fmt.Errorf("phase 1: %w", err) + } + select { + case <-time.After(writeDelay): + case <-ctx.Done(): + return ctx.Err() + } + } + fmt.Printf(" wrote %d email → did\n", len(emailKeys)) + + for _, key := range primaryKeys { + if err := putValue(ctx, apiToken, baseURL, key, primaryByKey[key]); err != nil { + return fmt.Errorf("phase 2: %w", err) + } + select { + case <-time.After(writeDelay): + case <-ctx.Done(): + return ctx.Err() + } + } + fmt.Printf(" wrote %d __primary:did → email\n", len(primaryKeys)) + + count, err := listKeyCount(ctx, apiToken, baseURL) + if err != nil { + return fmt.Errorf("verify: %w", err) + } + // idempotent reruns and stale keys only ever make this count bigger + expected := len(emailKeys) + len(primaryKeys) + if count < expected { + return fmt.Errorf("verify: listed %d keys, expected >= %d", count, expected) + } + fmt.Printf(" verified %d keys (expected >= %d)\n", count, expected) + return nil +} + +func orderKeys[V any](m map[string]V) []string { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + sort.Strings(keys) + return keys +} + +func queryEmails(ctx context.Context, db *sql.DB, query string) ([]emailRow, error) { + rows, err := db.QueryContext(ctx, query) + if err != nil { + return nil, err + } + defer rows.Close() + + out := make([]emailRow, 0) + for rows.Next() { + var r emailRow + if err := rows.Scan(&r.email, &r.did, &r.primary); err != nil { + return nil, err + } + out = append(out, r) + } + return out, rows.Err() +} + +// retryableError marks transient failures (429/5xx, network) that warrant a +// small backoff retry; anything else fails the run immediately. +type retryableError struct{ err error } + +func (e *retryableError) Error() string { return e.err.Error() } +func (e *retryableError) Unwrap() error { return e.err } + +func attempt(ctx context.Context, fn func() error) error { + var err error + for i := range 4 { + if i > 0 { + select { + case <-time.After(time.Duration(1<<(i-1)) * 500 * time.Millisecond): + case <-ctx.Done(): + return ctx.Err() + } + } + if err = fn(); err == nil { + return nil + } + var retry *retryableError + if !errors.As(err, &retry) { + return err + } + } + return err +} + +func putValue(ctx context.Context, apiToken, baseURL, key, value string) error { + target := baseURL + "values/" + url.PathEscape(key) + return attempt(ctx, func() error { + req, err := http.NewRequestWithContext(ctx, http.MethodPut, target, strings.NewReader(value)) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+apiToken) + req.Header.Set("Content-Type", "text/plain") + resp, err := httpClient.Do(req) + if err != nil { + return &retryableError{err} + } + defer resp.Body.Close() + body, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + return nil + } + err = fmt.Errorf("put %q: %s: %s", key, resp.Status, body) + if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= 500 { + return &retryableError{err} + } + return err + }) +} + +func listKeyCount(ctx context.Context, apiToken, baseURL string) (int, error) { + cursor := "" + count := 0 + for { + target := baseURL + "keys?limit=1000" + if cursor != "" { + target += "&cursor=" + url.QueryEscape(cursor) + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil) + if err != nil { + return 0, err + } + req.Header.Set("Authorization", "Bearer "+apiToken) + resp, err := httpClient.Do(req) + if err != nil { + return 0, err + } + body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + resp.Body.Close() + if err != nil { + return 0, err + } + if resp.StatusCode != http.StatusOK { + return 0, fmt.Errorf("list keys: %s: %s", resp.Status, body) + } + var out struct { + Result []struct { + Name string `json:"name"` + } `json:"result"` + ResultInfo struct { + Cursor string `json:"cursor"` + } `json:"result_info"` + } + if err := json.Unmarshal(body, &out); err != nil { + return 0, err + } + count += len(out.Result) + if out.ResultInfo.Cursor == "" { + return count, nil + } + cursor = out.ResultInfo.Cursor + } +} + +func fatal(format string, args ...any) { + fmt.Fprintf(os.Stderr, format+"\n", args...) + os.Exit(1) +} diff --git a/cmd/email-did-migrate/main_test.go b/cmd/email-did-migrate/main_test.go new file mode 100644 index 000000000..550211f95 --- /dev/null +++ b/cmd/email-did-migrate/main_test.go @@ -0,0 +1,207 @@ +package main + +import ( + "context" + "database/sql" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "sort" + "strings" + "sync" + "testing" +) + +// fakeKV is a minimal in-memory stand-in for the Cloudflare KV REST API: +// PUT /values/ stores, GET /keys lists (with cursor pagination). +type fakeKV struct { + mu sync.Mutex + values map[string]string + fail429 int // number of PUTs to fail with 429 before succeeding + puts int // total PUTs seen +} + +func newFakeKV(fail429 int) *fakeKV { + return &fakeKV{values: map[string]string{}, fail429: fail429} +} + +func (f *fakeKV) handler() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPut && strings.HasPrefix(r.URL.Path, "/values/") { + f.mu.Lock() + f.puts++ + if f.puts <= f.fail429 { + f.mu.Unlock() + http.Error(w, `{"success":false,"errors":[{"code":8000013,"message":"rate limited"}]}`, http.StatusTooManyRequests) + return + } + key := strings.TrimPrefix(r.URL.Path, "/values/") + body, _ := io.ReadAll(r.Body) + f.values[key] = string(body) + f.mu.Unlock() + io.WriteString(w, `{"success":true}`) + return + } + if r.Method == http.MethodGet && r.URL.Path == "/keys" { + f.mu.Lock() + keys := make([]string, 0, len(f.values)) + for k := range f.values { + keys = append(keys, k) + } + sort.Strings(keys) + f.mu.Unlock() + + page := keys + cursor := "" + switch r.URL.Query().Get("cursor") { + case "": + if len(keys) > 2 { + page, cursor = keys[:2], "next-page" + } + case "next-page": + page = keys[2:] + } + names := make([]map[string]string, 0, len(page)) + for _, k := range page { + names = append(names, map[string]string{"name": k}) + } + encoded, err := json.Marshal(names) + if err != nil { + panic(err) + } + cursorField := "null" + if cursor != "" { + cursorField = `"` + cursor + `"` + } + io.WriteString(w, `{"success":true,"result":`+string(encoded)+`,"result_info":{"cursor":`+cursorField+`}}`) + return + } + http.NotFound(w, r) + }) +} + +// openTestDB creates an appview-shaped emails table and returns the handle. +func openTestDB(t *testing.T, rows [][4]any) *sql.DB { + t.Helper() + db, err := sql.Open("sqlite3", t.TempDir()+"/appview.db") + if err != nil { + t.Fatalf("open db: %v", err) + } + t.Cleanup(func() { db.Close() }) + if _, err := db.Exec(` + create table emails ( + did text not null, + email text not null, + verified integer not null default 0, + is_primary integer not null default 0, + unique(did, email) + )`); err != nil { + t.Fatalf("create table: %v", err) + } + stmt, err := db.Prepare(`insert into emails (did, email, verified, is_primary) values (?, ?, ?, ?)`) + if err != nil { + t.Fatalf("prepare: %v", err) + } + defer stmt.Close() + for _, row := range rows { + if _, err := stmt.Exec(row[0], row[1], row[2], row[3]); err != nil { + t.Fatalf("insert %v: %v", row, err) + } + } + return db +} + +func runAgainst(t *testing.T, kv *fakeKV, db *sql.DB) { + t.Helper() + srv := httptest.NewServer(kv.handler()) + t.Cleanup(srv.Close) + if err := run(context.Background(), db, "test-token", srv.URL+"/"); err != nil { + t.Fatalf("run: %v", err) + } +} + +func TestMigratePhasesAndVerification(t *testing.T) { + db := openTestDB(t, [][4]any{ + {"did:plc:alice", "alice@example.com", 1, 1}, + // uppercase on the way in — keys are stored lowercased + {"did:plc:bob", "Bob@Example.com", 1, 0}, + // unverified signups are never written + {"did:plc:carol", "carol@example.com", 0, 0}, + {"did:plc:dave", "dave@example.com", 1, 0}, + }) + kv := newFakeKV(0) + runAgainst(t, kv, db) + + want := map[string]string{ + "alice@example.com": "did:plc:alice", + "bob@example.com": "did:plc:bob", // lowercased key + "dave@example.com": "did:plc:dave", + "__primary:did:plc:alice": "alice@example.com", + } + kv.mu.Lock() + defer kv.mu.Unlock() + if len(kv.values) != len(want) { + t.Fatalf("stored %d keys, want %d: %v", len(kv.values), len(want), kv.values) + } + for key, value := range want { + if got := kv.values[key]; got != value { + t.Errorf("key %q = %q, want %q", key, got, value) + } + } + if _, ok := kv.values["carol@example.com"]; ok { + t.Error("unverified carol@example.com was written") + } +} + +func TestCollidingEmailKeysPreferPrimary(t *testing.T) { + // the appview schema allows the same address (case variants included) to + // be verified for several dids; they collapse onto one lowercased kv key. + // the primary account must win and the verification count stays exact. + db := openTestDB(t, [][4]any{ + {"did:plc:nonprimary", "A@Example.com", 1, 0}, + {"did:plc:primary", "a@example.com", 1, 1}, + {"did:plc:solo", "solo@example.com", 1, 0}, + }) + kv := newFakeKV(0) + runAgainst(t, kv, db) + + want := map[string]string{ + "a@example.com": "did:plc:primary", // primary wins the shared key + "solo@example.com": "did:plc:solo", + "__primary:did:plc:primary": "a@example.com", + } + kv.mu.Lock() + defer kv.mu.Unlock() + if len(kv.values) != len(want) { + t.Fatalf("stored %d keys, want %d: %v", len(kv.values), len(want), kv.values) + } + for key, value := range want { + if got := kv.values[key]; got != value { + t.Errorf("key %q = %q, want %q", key, got, value) + } + } + if _, ok := kv.values["A@example.com"]; ok { + t.Error("non-normalized duplicate key was written") + } + if _, ok := kv.values["__primary:did:plc:nonprimary"]; ok { + t.Error("non-primary did must not get a __primary key") + } +} + +func TestPutRetriesTransientFailure(t *testing.T) { + db := openTestDB(t, [][4]any{ + {"did:plc:alice", "alice@example.com", 1, 1}, + }) + kv := newFakeKV(1) // first PUT 429s, then succeeds + runAgainst(t, kv, db) + + kv.mu.Lock() + defer kv.mu.Unlock() + if got := kv.values["alice@example.com"]; got != "did:plc:alice" { + t.Errorf("key after retry = %q, want did:plc:alice", got) + } + if got := kv.values["__primary:did:plc:alice"]; got != "alice@example.com" { + t.Errorf("primary after retry = %q, want alice@example.com", got) + } +} -- 2.51.2