Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
7.6 kB · 272 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273package 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 email-did worker reads. one-time, at cutover://// email-did-migrate <appview.db> <cf-api-token> <cf-account-id> <kv-namespace-id>//// phase 1 writes each verified email (lowercased — the worker looks up// lowercased keys) → did; phase 2 writes `__primary:<did>` → 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 <appview.db> <cf-api-token> <cf-account-id> <kv-namespace-id>\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)}