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 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)
}