package main import ( "bufio" "bytes" "context" "crypto/sha256" "encoding/base64" "encoding/csv" "encoding/hex" "encoding/json" "errors" "flag" "fmt" "io" "net/http" "net/url" "os" "path/filepath" "sort" "strconv" "strings" "sync" "time" "gopkg.in/yaml.v3" ) const migrationPlanVersion = 1 // migrationPlan is intentionally an operator-owned file rather than a server // database model. It can be code reviewed, checked into an operations repo, // and resumed by another operator without exposing source credentials. type migrationPlan struct { Version int `yaml:"version" json:"version"` Source migrationPlanSource `yaml:"source" json:"source"` Target migrationPlanTarget `yaml:"target" json:"target"` Execution migrationPlanExecution `yaml:"execution" json:"execution"` Cutover migrationPlanCutover `yaml:"cutover" json:"cutover"` Identity migrationPlanIdentity `yaml:"identity" json:"identity"` Repositories []migrationPlanRepo `yaml:"repositories" json:"repositories"` } type migrationPlanSource struct { Provider string `yaml:"provider" json:"provider"` APIURL string `yaml:"api_url" json:"api_url"` TokenEnv string `yaml:"token_env" json:"token_env"` } type migrationPlanTarget struct { APIURL string `yaml:"api_url" json:"api_url"` TokenEnv string `yaml:"token_env" json:"token_env"` } type migrationPlanExecution struct { RepositoryConcurrency int `yaml:"repository_concurrency" json:"repository_concurrency"` GitHubFetchConcurrency int `yaml:"github_fetch_concurrency" json:"github_fetch_concurrency"` ApplyBatchLimit int `yaml:"apply_batch_limit" json:"apply_batch_limit"` ReleaseAssetMaxBytes int64 `yaml:"release_asset_max_bytes" json:"release_asset_max_bytes"` } type migrationPlanCutover struct { RequireWiki bool `yaml:"require_wiki" json:"require_wiki"` RequireLFSPreflight bool `yaml:"require_lfs_preflight" json:"require_lfs_preflight"` RequireReleaseAssets bool `yaml:"require_release_assets" json:"require_release_assets"` BlockOnUnresolvedFindings bool `yaml:"block_on_unresolved_findings" json:"block_on_unresolved_findings"` } type migrationPlanIdentity struct { InvitationRequired bool `yaml:"invitation_required" json:"invitation_required"` MappingsCSV string `yaml:"mappings_csv" json:"mappings_csv"` } type migrationPlanRepo struct { Source string `yaml:"source" json:"source"` Target string `yaml:"target" json:"target"` TargetMode string `yaml:"target_mode" json:"target_mode"` Mode string `yaml:"mode" json:"mode"` Wiki string `yaml:"wiki" json:"wiki"` StackAPI string `yaml:"stack_api" json:"stack_api"` Access string `yaml:"access" json:"access"` } type migrationFinding struct { Severity string `json:"severity"` Code string `json:"code"` Message string `json:"message"` Action string `json:"action,omitempty"` Details map[string]any `json:"details,omitempty"` } type migrationPreflight struct { Source string `json:"source"` Target string `json:"target"` CheckedAt time.Time `json:"checked_at"` SourceUpdatedAt time.Time `json:"source_updated_at"` SourcePushedAt *time.Time `json:"source_pushed_at,omitempty"` SourceDefaultBranch string `json:"source_default_branch,omitempty"` SourceRepository migrationSourceRepository `json:"source_repository"` Counts map[string]int `json:"counts"` Bytes map[string]int64 `json:"bytes"` SourceWiki bool `json:"source_wiki"` LFSDetected bool `json:"lfs_detected"` Findings []migrationFinding `json:"findings"` } // migrationSourceRepository is non-secret source metadata captured by the // local executor. It allows a newly created AGS target to retain the source // settings without sending the GitHub token to AGS. type migrationSourceRepository struct { Description string `json:"description,omitempty"` Visibility string `json:"visibility,omitempty"` Private bool `json:"private"` DefaultBranch string `json:"default_branch,omitempty"` HasIssues bool `json:"has_issues"` HasWiki bool `json:"has_wiki"` } // migrationSourceFence is the source-side evidence captured after the operator // has frozen writes. A final capture must match it before cutover can succeed. type migrationSourceFence struct { CapturedAt time.Time `json:"captured_at"` GitRefsDigest string `json:"git_refs_digest"` InventoryDigest string `json:"inventory_digest"` ContentDigest string `json:"content_digest"` ExcludedObjects []string `json:"excluded_objects,omitempty"` SourceUpdatedAt time.Time `json:"source_updated_at"` SourcePushedAt *time.Time `json:"source_pushed_at,omitempty"` InventoryCounts map[string]int `json:"inventory_counts"` InventoryBytes map[string]int64 `json:"inventory_bytes"` } type migrationRepoState struct { Source string `json:"source"` Target string `json:"target"` Status string `json:"status"` Preflight *migrationPreflight `json:"preflight,omitempty"` SourceFence *migrationSourceFence `json:"source_fence,omitempty"` Postflight []migrationFinding `json:"postflight_findings,omitempty"` ControlVerified bool `json:"control_verified,omitempty"` ControlImportID uint `json:"control_import_id,omitempty"` MetadataImportID uint `json:"metadata_import_id,omitempty"` SnapshotDir string `json:"snapshot_dir,omitempty"` Verification json.RawMessage `json:"verification,omitempty"` StackVerification *migrationStackVerification `json:"stack_verification,omitempty"` IdentitySummary *migrationIdentitySummary `json:"identity_summary,omitempty"` RecoveryRequired string `json:"recovery_required,omitempty"` Error string `json:"error,omitempty"` UpdatedAt time.Time `json:"updated_at"` } type migrationPlanState struct { Version int `json:"version"` PlanDigest string `json:"plan_digest"` PlanPath string `json:"plan_path"` SourceFreezeConfirmedAt time.Time `json:"source_freeze_confirmed_at,omitempty"` StartedAt time.Time `json:"started_at"` UpdatedAt time.Time `json:"updated_at"` Repositories map[string]*migrationRepoState `json:"repositories"` } type migrationRepoReport struct { Source string `json:"source"` Target string `json:"target"` Status string `json:"status"` Preflight *migrationPreflight `json:"preflight,omitempty"` SourceFence *migrationSourceFence `json:"source_fence,omitempty"` SourceFreezeConfirmedAt time.Time `json:"source_freeze_confirmed_at,omitempty"` Postflight []migrationFinding `json:"postflight_findings,omitempty"` ControlImportID uint `json:"control_import_id,omitempty"` MetadataImportID uint `json:"metadata_import_id,omitempty"` Verification json.RawMessage `json:"verification,omitempty"` StackVerification *migrationStackVerification `json:"stack_verification,omitempty"` IdentitySummary *migrationIdentitySummary `json:"identity_summary,omitempty"` RecoveryRequired string `json:"recovery_required,omitempty"` Findings []migrationFinding `json:"findings,omitempty"` Error string `json:"error,omitempty"` CutoverReady bool `json:"cutover_ready"` GeneratedAt time.Time `json:"generated_at"` } type migrationControlImportResponse struct { ID uint `json:"id"` Status string `json:"status"` } type migrationSourceUser struct { ID uint `json:"id"` ImportID uint `json:"import_id"` SourceLogin string `json:"source_login"` SourceID string `json:"source_id"` State string `json:"state"` } type migrationIdentityCSVRow struct { SourceRepository string ImportID uint SourceUserID uint SourceLogin string GitHubNumericID string TargetAGSLogin string SkipInvitation bool } type migrationIdentitySummary struct { Total int `json:"total"` PendingClaim int `json:"pending_claim"` PendingInvitation int `json:"pending_invitation"` Claimed int `json:"claimed"` Rejected int `json:"rejected"` } type migrationImportVerification struct { PendingWrites map[string]int `json:"pending_writes"` FailedWrites map[string]int `json:"failed_writes"` } // migrationStackVerification is durable cutover evidence for the Stack // relationship graph. Generic batch verification proves a write ran; it does // not prove the Stack members, order, or deep link are correct. type migrationStackVerification struct { Skipped bool `json:"skipped,omitempty"` SkipReason string `json:"skip_reason,omitempty"` SourceCount int `json:"source_count"` TargetCount int `json:"target_count"` MappingCount int `json:"mapping_count"` VerifiedCount int `json:"verified_count"` Checks []migrationStackCheck `json:"checks,omitempty"` } type migrationStackCheck struct { SourceID string `json:"source_id,omitempty"` SourceNumber int `json:"source_number,omitempty"` SourceURL string `json:"source_url,omitempty"` TargetID string `json:"target_id,omitempty"` TargetNumber int `json:"target_number,omitempty"` TargetURL string `json:"target_url,omitempty"` TargetAPIURL string `json:"target_api_url,omitempty"` SourcePullRequestNumbers []int `json:"source_pull_request_numbers,omitempty"` TargetPullRequestNumbers []int `json:"target_pull_request_numbers,omitempty"` MembersMatch bool `json:"members_match"` OrderMatches bool `json:"order_matches"` URLMappingMatches bool `json:"url_mapping_matches"` Status string `json:"status"` } type migrationImportMapping struct { ID uint `json:"id"` SourceType string `json:"source_type"` SourceID string `json:"source_id"` SourceNumber *int `json:"source_number"` SourceURL string `json:"source_url"` TargetType string `json:"target_type"` TargetID string `json:"target_id"` TargetNumber *int `json:"target_number"` TargetURL string `json:"target_url"` } type migrationSnapshotStack struct { ID string Number int URL string PullRequestNumbers []int } type migrationTargetStack struct { ID uint `json:"id"` Number int `json:"number"` URL string `json:"url"` PullRequests []struct { Number int `json:"number"` } `json:"pull_requests"` } func migrationPlanCommand(ctx context.Context, args []string) error { if len(args) == 0 { return errors.New("usage: ags-migrate plan validate|preflight|run|report|identity-template|apply-identities ...") } switch args[0] { case "validate": return migrationPlanValidate(args[1:]) case "preflight": return migrationPlanPreflight(ctx, args[1:]) case "run": return migrationPlanRun(ctx, args[1:]) case "report": return migrationPlanReport(args[1:]) case "identity-template": return migrationPlanIdentityTemplate(ctx, args[1:]) case "apply-identities": return migrationPlanApplyIdentities(ctx, args[1:]) default: return errors.New("usage: ags-migrate plan validate|preflight|run|report|identity-template|apply-identities ...") } } func migrationPlanValidate(args []string) error { fs := flag.NewFlagSet("plan validate", flag.ContinueOnError) path := fs.String("file", "migration.yaml", "migration plan YAML") if err := fs.Parse(args); err != nil { return err } plan, digest, err := loadMigrationPlan(*path) if err != nil { return err } return writeJSON(os.Stdout, map[string]any{ "valid": true, "version": plan.Version, "repositories": len(plan.Repositories), "plan_digest": digest, }) } func migrationPlanPreflight(ctx context.Context, args []string) error { fs := flag.NewFlagSet("plan preflight", flag.ContinueOnError) path := fs.String("file", "migration.yaml", "migration plan YAML") outDir := fs.String("out", "migration-reports", "directory for preflight reports") failBlocking := fs.Bool("fail-on-blocking", false, "return an error when a repository has blocking findings") if err := fs.Parse(args); err != nil { return err } plan, _, err := loadMigrationPlan(*path) if err != nil { return err } token, err := requiredEnv(plan.Source.TokenEnv) if err != nil { return err } if err := os.MkdirAll(*outDir, 0o755); err != nil { return err } reports, err := runPlanPreflight(ctx, plan, token) if err != nil { return err } blocking := false for _, report := range reports { if hasBlockingFindings(report.Findings) { blocking = true } if err := writeMigrationPreflight(*outDir, report); err != nil { return err } } if err := writeJSONFile(filepath.Join(*outDir, "preflight-summary.json"), reports); err != nil { return err } if err := writeMigrationMarkdown(filepath.Join(*outDir, "preflight-summary.md"), reports); err != nil { return err } if err := writeJSON(os.Stdout, reports); err != nil { return err } if blocking && *failBlocking { return errors.New("migration preflight found blocking findings") } return nil } func migrationPlanRun(ctx context.Context, args []string) error { fs := flag.NewFlagSet("plan run", flag.ContinueOnError) path := fs.String("file", "migration.yaml", "migration plan YAML") statePath := fs.String("state", ".ags-migrate-state.json", "durable resume state JSON") workDir := fs.String("work-dir", ".ags-migrate", "snapshot and report directory") allowBlocking := fs.Bool("allow-blocking-findings", false, "continue despite preflight blockers") sourceFrozen := fs.Bool("confirm-source-frozen", false, "confirm that source writes are frozen for this cutover run") if err := fs.Parse(args); err != nil { return err } plan, digest, err := loadMigrationPlan(*path) if err != nil { return err } if !*sourceFrozen { return errors.New("refusing cutover run without --confirm-source-frozen; freeze source writes, then rerun") } sourceToken, err := requiredEnv(plan.Source.TokenEnv) if err != nil { return err } targetToken, err := requiredEnv(plan.Target.TokenEnv) if err != nil { return err } if err := os.MkdirAll(*workDir, 0o755); err != nil { return err } state, err := loadOrCreateMigrationState(*statePath, *path, digest, plan) if err != nil { return err } if err := saveMigrationState(*statePath, state); err != nil { return err } if state.SourceFreezeConfirmedAt.IsZero() { state.SourceFreezeConfirmedAt = time.Now().UTC() if err := saveMigrationState(*statePath, state); err != nil { return err } } concurrency := plan.Execution.RepositoryConcurrency if concurrency <= 0 { concurrency = 1 } sem := make(chan struct{}, concurrency) var wg sync.WaitGroup var mu sync.Mutex var errs []error for _, repo := range plan.Repositories { repo := repo wg.Add(1) go func() { defer wg.Done() sem <- struct{}{} defer func() { <-sem }() if err := runPlannedRepository(ctx, plan, repo, state, *statePath, *workDir, sourceToken, targetToken, *allowBlocking, &mu); err != nil { mu.Lock() if entry := state.Repositories[migrationPlanRepoKey(repo)]; entry != nil { entry.Status = "failed" entry.Error = err.Error() entry.UpdatedAt = time.Now().UTC() } mu.Unlock() if saveErr := saveMigrationStateLocked(*statePath, state, &mu); saveErr != nil { err = errors.Join(err, saveErr) } mu.Lock() errs = append(errs, fmt.Errorf("%s: %w", repo.Source, err)) mu.Unlock() } }() } wg.Wait() if err := saveMigrationState(*statePath, state); err != nil { return err } reportErr := migrationPlanReport([]string{"--file", *path, "--state", *statePath, "--out", filepath.Join(*workDir, "reports")}) if len(errs) > 0 { if reportErr != nil { errs = append(errs, reportErr) } return errors.Join(errs...) } return reportErr } func migrationPlanReport(args []string) error { fs := flag.NewFlagSet("plan report", flag.ContinueOnError) path := fs.String("file", "migration.yaml", "migration plan YAML") statePath := fs.String("state", ".ags-migrate-state.json", "durable resume state JSON") outDir := fs.String("out", "migration-reports", "directory for generated reports") if err := fs.Parse(args); err != nil { return err } plan, _, err := loadMigrationPlan(*path) if err != nil { return err } state, err := readMigrationState(*statePath) if err != nil { return err } if err := os.MkdirAll(*outDir, 0o755); err != nil { return err } // Reports are also used after invitation responses. When the target token is // available, refresh this mutable server-side state instead of presenting the // identity summary captured at the end of the original import run. targetToken := strings.TrimSpace(os.Getenv(plan.Target.TokenEnv)) refreshedIdentitySummary := false reports := make([]migrationRepoReport, 0, len(plan.Repositories)) for _, repo := range plan.Repositories { entry := state.Repositories[migrationPlanRepoKey(repo)] if entry == nil { entry = &migrationRepoState{Source: repo.Source, Target: repo.Target, Status: "not_started"} } if targetToken != "" && entry.MetadataImportID != 0 { summary, err := summarizeMigrationSourceUsers(context.Background(), plan.Target.APIURL, targetToken, entry.MetadataImportID) if err != nil { return fmt.Errorf("refresh identity summary for %s: %w", entry.Source, err) } entry.IdentitySummary = &summary entry.UpdatedAt = time.Now().UTC() refreshedIdentitySummary = true } report := migrationRepoReport{ Source: repo.Source, Target: repo.Target, Status: entry.Status, Preflight: entry.Preflight, SourceFence: entry.SourceFence, SourceFreezeConfirmedAt: state.SourceFreezeConfirmedAt, Postflight: entry.Postflight, ControlImportID: entry.ControlImportID, MetadataImportID: entry.MetadataImportID, StackVerification: entry.StackVerification, IdentitySummary: entry.IdentitySummary, RecoveryRequired: entry.RecoveryRequired, Error: entry.Error, GeneratedAt: time.Now().UTC(), } report.Verification = entry.Verification if entry.Preflight != nil { report.Findings = append(report.Findings, entry.Preflight.Findings...) } report.Findings = append(report.Findings, entry.Postflight...) if entry.Status == "succeeded" && (entry.SourceFence == nil || entry.SourceFence.ContentDigest == "" || state.SourceFreezeConfirmedAt.IsZero()) { report.Findings = append(report.Findings, migrationFinding{ Severity: "block", Code: "source_freeze_unverified", Message: "migration state has no verified source freeze fence", Action: "create a new migration state and rerun after freezing source writes", }) } if entry.RecoveryRequired != "" { report.Findings = append(report.Findings, migrationFinding{ Severity: "block", Code: "target_reset_required", Message: entry.RecoveryRequired, Action: "explicitly reset or replace the target repository and staged import job, then start with a new migration state", }) } report.CutoverReady = entry.Status == "succeeded" && !hasBlockingFindings(report.Findings) if plan.Cutover.BlockOnUnresolvedFindings && hasUnresolvedFindings(report.Findings) { report.CutoverReady = false } reports = append(reports, report) if err := writeJSONFile(filepath.Join(*outDir, reportFileName(repo.Source)+".json"), report); err != nil { return err } } if refreshedIdentitySummary { if err := saveMigrationState(*statePath, state); err != nil { return err } } if err := writeJSONFile(filepath.Join(*outDir, "summary.json"), reports); err != nil { return err } if err := writeMigrationReportMarkdown(filepath.Join(*outDir, "summary.md"), reports); err != nil { return err } return writeJSON(os.Stdout, reports) } func migrationPlanIdentityTemplate(ctx context.Context, args []string) error { fs := flag.NewFlagSet("plan identity-template", flag.ContinueOnError) path := fs.String("file", "migration.yaml", "migration plan YAML") statePath := fs.String("state", ".ags-migrate-state.json", "durable resume state JSON") outPath := fs.String("out", "identity-mappings.csv", "CSV mapping template") if err := fs.Parse(args); err != nil { return err } plan, _, err := loadMigrationPlan(*path) if err != nil { return err } targetToken, err := requiredEnv(plan.Target.TokenEnv) if err != nil { return err } state, err := readMigrationState(*statePath) if err != nil { return err } file, err := os.Create(*outPath) if err != nil { return err } defer file.Close() w := csv.NewWriter(file) defer w.Flush() if err := w.Write([]string{"source_repository", "import_id", "source_user_id", "source_login", "github_numeric_id", "target_ags_login", "skip_invitation"}); err != nil { return err } keys := make([]string, 0, len(state.Repositories)) for key := range state.Repositories { keys = append(keys, key) } sort.Strings(keys) for _, key := range keys { entry := state.Repositories[key] if entry.MetadataImportID == 0 { continue } users, err := listMigrationSourceUsers(ctx, plan.Target.APIURL, targetToken, entry.MetadataImportID) if err != nil { return fmt.Errorf("list source users for %s: %w", entry.Source, err) } for _, sourceUser := range users { if err := w.Write([]string{ entry.Source, fmt.Sprint(entry.MetadataImportID), fmt.Sprint(sourceUser.ID), sourceUser.SourceLogin, sourceUser.SourceID, "", "false", }); err != nil { return err } } } return w.Error() } // migrationPlanApplyIdentities applies a reviewed CSV in bounded API batches. // The server creates an attribution invitation by default; only deployments // that deliberately mark a row skip_invitation=true receive an immediate // reattribution. func migrationPlanApplyIdentities(ctx context.Context, args []string) error { fs := flag.NewFlagSet("plan apply-identities", flag.ContinueOnError) path := fs.String("file", "migration.yaml", "migration plan YAML") statePath := fs.String("state", ".ags-migrate-state.json", "durable resume state JSON") csvPath := fs.String("csv", "", "reviewed identity mapping CSV") if err := fs.Parse(args); err != nil { return err } plan, _, err := loadMigrationPlan(*path) if err != nil { return err } if strings.TrimSpace(*csvPath) == "" { *csvPath = plan.Identity.MappingsCSV } if strings.TrimSpace(*csvPath) == "" { return errors.New("identity mapping CSV is required via --csv or identity.mappings_csv") } targetToken, err := requiredEnv(plan.Target.TokenEnv) if err != nil { return err } state, err := readMigrationState(*statePath) if err != nil { return err } rows, err := readMigrationIdentityCSV(*csvPath) if err != nil { return err } byImportID := make(map[uint][]migrationIdentityCSVRow) for _, row := range rows { if strings.TrimSpace(row.TargetAGSLogin) == "" { continue } if plan.Identity.InvitationRequired && row.SkipInvitation { return fmt.Errorf("CSV row for %q source user %d skips a required attribution invitation", row.SourceRepository, row.SourceUserID) } entry, ok := migrationStateRepositoryBySource(state, row.SourceRepository) if !ok || entry.MetadataImportID != row.ImportID { return fmt.Errorf("CSV row for %q does not match this migration state", row.SourceRepository) } byImportID[row.ImportID] = append(byImportID[row.ImportID], row) } applied := 0 for importID, mappings := range byImportID { for _, group := range chunkIdentityRows(mappings, 500) { requestMappings := make([]map[string]any, 0, len(group)) for _, row := range group { requestMappings = append(requestMappings, map[string]any{ "source_user_id": row.SourceUserID, "target_user_login": row.TargetAGSLogin, "skip_invitation": row.SkipInvitation, }) } endpoint := strings.TrimRight(plan.Target.APIURL, "/") + "/api/ext/v1/imports/" + fmt.Sprint(importID) + "/source-users" if _, err := patchJSONResponse(ctx, endpoint, targetToken, map[string]any{"mappings": requestMappings}); err != nil { return fmt.Errorf("apply %d identity mappings for import %d: %w", len(group), importID, err) } applied += len(group) } } return writeJSON(os.Stdout, map[string]any{"csv": *csvPath, "applied": applied, "invitation_required": plan.Identity.InvitationRequired}) } func listMigrationSourceUsers(ctx context.Context, target, token string, importID uint) ([]migrationSourceUser, error) { endpoint := strings.TrimRight(target, "/") + "/api/ext/v1/imports/" + fmt.Sprint(importID) + "/source-users" users := make([]migrationSourceUser, 0) var afterID uint for { page, err := getMigrationJSON(ctx, endpoint+"?limit=500&after_id="+fmt.Sprint(afterID), token) if err != nil { return nil, err } var items []migrationSourceUser if err := json.Unmarshal(page, &items); err != nil { return nil, fmt.Errorf("decode source identity page: %w", err) } if len(items) == 0 { return users, nil } users = append(users, items...) last := items[len(items)-1].ID if last <= afterID { return nil, errors.New("source identity pagination did not advance") } afterID = last } } func summarizeMigrationSourceUsers(ctx context.Context, target, token string, importID uint) (migrationIdentitySummary, error) { users, err := listMigrationSourceUsers(ctx, target, token, importID) if err != nil { return migrationIdentitySummary{}, err } summary := migrationIdentitySummary{Total: len(users)} for _, user := range users { switch user.State { case "pending_claim": summary.PendingClaim++ case "pending_invitation": summary.PendingInvitation++ case "claimed": summary.Claimed++ case "rejected": summary.Rejected++ } } return summary, nil } func readMigrationIdentityCSV(path string) ([]migrationIdentityCSVRow, error) { file, err := os.Open(path) if err != nil { return nil, err } defer file.Close() reader := csv.NewReader(file) header, err := reader.Read() if err != nil { return nil, err } columns := make(map[string]int, len(header)) for i, name := range header { columns[strings.TrimSpace(name)] = i } required := []string{"source_repository", "import_id", "source_user_id", "target_ags_login", "skip_invitation"} for _, name := range required { if _, ok := columns[name]; !ok { return nil, fmt.Errorf("identity mapping CSV is missing %q", name) } } value := func(record []string, name string) string { index, ok := columns[name] if !ok || index >= len(record) { return "" } return strings.TrimSpace(record[index]) } var rows []migrationIdentityCSVRow for line := 2; ; line++ { record, err := reader.Read() if errors.Is(err, io.EOF) { break } if err != nil { return nil, fmt.Errorf("read identity mapping CSV line %d: %w", line, err) } if len(record) == 0 || strings.TrimSpace(value(record, "source_repository")) == "" { continue } importID, err := strconv.ParseUint(value(record, "import_id"), 10, 64) if err != nil || importID == 0 { return nil, fmt.Errorf("identity mapping CSV line %d has invalid import_id", line) } sourceUserID, err := strconv.ParseUint(value(record, "source_user_id"), 10, 64) if err != nil || sourceUserID == 0 { return nil, fmt.Errorf("identity mapping CSV line %d has invalid source_user_id", line) } skipInvitation, err := strconv.ParseBool(value(record, "skip_invitation")) if err != nil { return nil, fmt.Errorf("identity mapping CSV line %d has invalid skip_invitation", line) } rows = append(rows, migrationIdentityCSVRow{ SourceRepository: value(record, "source_repository"), ImportID: uint(importID), SourceUserID: uint(sourceUserID), SourceLogin: value(record, "source_login"), GitHubNumericID: value(record, "github_numeric_id"), TargetAGSLogin: value(record, "target_ags_login"), SkipInvitation: skipInvitation, }) } return rows, nil } func migrationStateRepositoryBySource(state *migrationPlanState, source string) (*migrationRepoState, bool) { for _, entry := range state.Repositories { if entry.Source == source { return entry, true } } return nil, false } func chunkIdentityRows(rows []migrationIdentityCSVRow, size int) [][]migrationIdentityCSVRow { if size <= 0 { size = 500 } chunks := make([][]migrationIdentityCSVRow, 0, (len(rows)+size-1)/size) for len(rows) > 0 { end := size if end > len(rows) { end = len(rows) } chunks = append(chunks, rows[:end]) rows = rows[end:] } return chunks } func loadMigrationPlan(path string) (migrationPlan, string, error) { data, err := os.ReadFile(path) if err != nil { return migrationPlan{}, "", err } var plan migrationPlan if err := yaml.Unmarshal(data, &plan); err != nil { return migrationPlan{}, "", fmt.Errorf("parse migration plan: %w", err) } if err := normalizeAndValidateMigrationPlan(&plan); err != nil { return migrationPlan{}, "", err } digest := sha256.Sum256(data) return plan, "sha256:" + hex.EncodeToString(digest[:]), nil } func normalizeAndValidateMigrationPlan(plan *migrationPlan) error { if plan.Version == 0 { plan.Version = migrationPlanVersion } if plan.Version != migrationPlanVersion { return fmt.Errorf("unsupported migration plan version %d", plan.Version) } plan.Source.Provider = strings.ToLower(strings.TrimSpace(plan.Source.Provider)) plan.Source.APIURL = strings.TrimRight(strings.TrimSpace(plan.Source.APIURL), "/") plan.Source.TokenEnv = strings.TrimSpace(plan.Source.TokenEnv) plan.Target.APIURL = strings.TrimRight(strings.TrimSpace(plan.Target.APIURL), "/") plan.Target.TokenEnv = strings.TrimSpace(plan.Target.TokenEnv) if plan.Source.Provider == "" { plan.Source.Provider = "github" } if plan.Source.Provider != "github" { return errors.New("migration plan source.provider must be github") } if plan.Source.APIURL == "" { plan.Source.APIURL = defaultGitHubAPI } if plan.Source.TokenEnv == "" || plan.Target.APIURL == "" || plan.Target.TokenEnv == "" { return errors.New("migration plan source.token_env, target.api_url, and target.token_env are required") } if len(plan.Repositories) == 0 { return errors.New("migration plan requires at least one repository") } seenSource := map[string]struct{}{} seenTarget := map[string]struct{}{} for i := range plan.Repositories { repo := &plan.Repositories[i] repo.Source = strings.TrimSpace(repo.Source) repo.Target = strings.TrimSpace(repo.Target) repo.TargetMode = strings.ToLower(strings.TrimSpace(repo.TargetMode)) repo.Mode = strings.ToLower(strings.TrimSpace(repo.Mode)) repo.Wiki = strings.ToLower(strings.TrimSpace(repo.Wiki)) repo.StackAPI = strings.ToLower(strings.TrimSpace(repo.StackAPI)) repo.Access = strings.ToLower(strings.TrimSpace(repo.Access)) if _, _, err := splitRepo(repo.Source); err != nil { return fmt.Errorf("repositories[%d].source: %w", i, err) } if _, _, err := splitRepo(repo.Target); err != nil { return fmt.Errorf("repositories[%d].target: %w", i, err) } if repo.Mode == "" { repo.Mode = "core" } if repo.TargetMode == "" { repo.TargetMode = "create" } if repo.TargetMode != "create" && repo.TargetMode != "existing_empty" { return fmt.Errorf("repositories[%d].target_mode must be create or existing_empty", i) } if repo.Mode != "core" && repo.Mode != "full" { return fmt.Errorf("repositories[%d].mode must be core or full", i) } if repo.Wiki == "" { repo.Wiki = "optional" } if repo.Wiki != "required" && repo.Wiki != "optional" && repo.Wiki != "disabled" { return fmt.Errorf("repositories[%d].wiki must be required, optional, or disabled", i) } if repo.StackAPI == "" { repo.StackAPI = "required" } if repo.StackAPI != "required" && repo.StackAPI != "skip_if_unreadable" { return fmt.Errorf("repositories[%d].stack_api must be required or skip_if_unreadable", i) } if repo.Access == "" { repo.Access = "handoff" } if repo.Access != "handoff" && repo.Access != "skip" { return fmt.Errorf("repositories[%d].access must be handoff or skip", i) } sourceKey := strings.ToLower(repo.Source) targetKey := strings.ToLower(repo.Target) if _, ok := seenSource[sourceKey]; ok { return fmt.Errorf("duplicate source repository %q", repo.Source) } if _, ok := seenTarget[targetKey]; ok { return fmt.Errorf("duplicate target repository %q", repo.Target) } seenSource[sourceKey] = struct{}{} seenTarget[targetKey] = struct{}{} } if plan.Execution.RepositoryConcurrency <= 0 { plan.Execution.RepositoryConcurrency = 1 } if plan.Execution.GitHubFetchConcurrency <= 0 { plan.Execution.GitHubFetchConcurrency = defaultFetchConcurrency } if plan.Execution.ApplyBatchLimit <= 0 { plan.Execution.ApplyBatchLimit = defaultApplyBatchLimit } if plan.Execution.ReleaseAssetMaxBytes <= 0 { plan.Execution.ReleaseAssetMaxBytes = defaultMaxAssetBytes } return nil } func runPlanPreflight(ctx context.Context, plan migrationPlan, token string) ([]migrationPreflight, error) { concurrency := plan.Execution.RepositoryConcurrency if concurrency <= 0 { concurrency = 1 } sem := make(chan struct{}, concurrency) reports := make([]migrationPreflight, len(plan.Repositories)) var wg sync.WaitGroup var mu sync.Mutex var errs []error for i, repo := range plan.Repositories { i, repo := i, repo wg.Add(1) go func() { defer wg.Done() sem <- struct{}{} defer func() { <-sem }() report, err := preflightMigrationRepository(ctx, plan, repo, token) if err != nil { mu.Lock() errs = append(errs, fmt.Errorf("%s: %w", repo.Source, err)) mu.Unlock() return } reports[i] = report }() } wg.Wait() if len(errs) > 0 { return nil, errors.Join(errs...) } return reports, nil } func preflightMigrationRepository(ctx context.Context, plan migrationPlan, repo migrationPlanRepo, token string) (migrationPreflight, error) { owner, name, _ := splitRepo(repo.Source) client := newGitHubClient(plan.Source.APIURL, token) repoRaw, err := client.getRaw(ctx, "/repos/"+owner+"/"+name, nil) if err != nil { return migrationPreflight{}, err } var source struct { Description string `json:"description"` Visibility string `json:"visibility"` Private bool `json:"private"` HasIssues bool `json:"has_issues"` HasWiki bool `json:"has_wiki"` Size int `json:"size"` UpdatedAt time.Time `json:"updated_at"` PushedAt *time.Time `json:"pushed_at"` DefaultBranch string `json:"default_branch"` } if err := json.Unmarshal(repoRaw, &source); err != nil { return migrationPreflight{}, err } report := migrationPreflight{ Source: repo.Source, Target: repo.Target, CheckedAt: time.Now().UTC(), SourceUpdatedAt: source.UpdatedAt, SourcePushedAt: source.PushedAt, SourceDefaultBranch: source.DefaultBranch, SourceWiki: source.HasWiki, SourceRepository: migrationSourceRepository{Description: source.Description, Visibility: source.Visibility, Private: source.Private, DefaultBranch: source.DefaultBranch, HasIssues: source.HasIssues, HasWiki: source.HasWiki}, Counts: map[string]int{"repository_size_kib": source.Size}, Bytes: map[string]int64{}, } stacks, stacksErr := listGitHubPullRequestStacks(ctx, client, owner, name) switch { case stacksErr == nil: report.Counts["pull_request_stacks"] = len(stacks) case isGitHubStacksUnavailable(stacksErr): // GitHub documents 404 here when the public-preview Stack feature is // not enabled for this repository. There is therefore no source Stack // relation to restore, rather than a product-wide export limitation. report.Counts["pull_request_stacks"] = 0 report.Findings = append(report.Findings, migrationFinding{ Severity: "info", Code: "github_stacks_not_enabled", Message: "GitHub Stacks API is not enabled for this source repository; no Stack relationships will be imported", Action: "enable GitHub Stacks before the source freeze if Stack fidelity is required", }) default: finding := migrationFinding{ Severity: "block", Code: "github_stacks_unreadable", Message: stacksErr.Error(), Action: "restore GitHub Stack API access before cutover; an unreadable Stack surface cannot be treated as empty", Details: map[string]any{"stack_api_policy": repo.StackAPI}, } if repo.StackAPI == "skip_if_unreadable" { finding.Severity = "warn" finding.Code = "github_stacks_unreadable_waived" finding.Action = "Stack migration is explicitly waived for this repository; document the relationship data that will remain on GitHub" } report.Findings = append(report.Findings, finding) } if repo.Wiki == "required" && !source.HasWiki { report.Findings = append(report.Findings, migrationFinding{Severity: "block", Code: "wiki_disabled", Message: "source repository has no enabled Wiki", Action: "enable or explicitly waive Wiki migration"}) } if repo.Wiki != "disabled" && source.HasWiki { report.Counts["wiki_repositories"] = 1 } releases, releaseErr := client.listRaw(ctx, repoPath(owner, name, "releases"), nil) if releaseErr != nil { report.Findings = append(report.Findings, migrationFinding{Severity: "warn", Code: "releases_unreadable", Message: releaseErr.Error(), Action: "verify source token release access"}) } else { report.Counts["releases"] = len(releases) assetCount := 0 var assetBytes int64 overLimit := make([]string, 0) for _, raw := range releases { var release struct { Assets []struct { Name string `json:"name"` Size int64 `json:"size"` } `json:"assets"` } if json.Unmarshal(raw, &release) != nil { continue } assetCount += len(release.Assets) for _, asset := range release.Assets { assetBytes += asset.Size if asset.Size > plan.Execution.ReleaseAssetMaxBytes { overLimit = append(overLimit, asset.Name) } } } report.Counts["release_assets"] = assetCount report.Bytes["release_assets"] = assetBytes if plan.Cutover.RequireReleaseAssets && assetCount > 0 && repo.Mode != "full" { report.Findings = append(report.Findings, migrationFinding{Severity: "block", Code: "release_assets_not_selected", Message: "release assets exist but repository mode is core", Action: "set mode: full or explicitly waive release assets", Details: map[string]any{"asset_count": assetCount, "asset_bytes": assetBytes}}) } if repo.Mode == "full" && len(overLimit) > 0 { report.Findings = append(report.Findings, migrationFinding{Severity: "block", Code: "release_assets_exceed_limit", Message: "release assets exceed the configured in-memory transfer limit", Action: "raise execution.release_asset_max_bytes only when AGS storage can safely hold the assets, or migrate them through a dedicated asset path", Details: map[string]any{"max_bytes": plan.Execution.ReleaseAssetMaxBytes, "asset_count": len(overLimit), "asset_names": limitMigrationRefSamples(overLimit)}}) } } if attrs, err := client.getRaw(ctx, repoPath(owner, name, "contents/.gitattributes"), nil); err == nil { var content struct { Content string `json:"content"` Encoding string `json:"encoding"` } if json.Unmarshal(attrs, &content) == nil && strings.EqualFold(content.Encoding, "base64") { decoded, decodeErr := base64.StdEncoding.DecodeString(strings.ReplaceAll(content.Content, "\n", "")) if decodeErr == nil && bytesContainLFSFilter(decoded) { report.LFSDetected = true report.Findings = append(report.Findings, migrationFinding{Severity: lfsFindingSeverity(plan), Code: "lfs_detected", Message: "Git LFS filter detected in .gitattributes", Action: "migrate LFS objects and verify target checkout before cutover"}) } } } issuePayloads, issueErr := client.listRaw(ctx, repoPath(owner, name, "issues"), mapStringValues("state", "all")) if issueErr != nil { report.Findings = append(report.Findings, migrationFinding{Severity: "warn", Code: "issues_unreadable", Message: issueErr.Error(), Action: "verify source token issue access"}) } else { issues, pulls, attachmentURLs := 0, 0, 0 for _, raw := range issuePayloads { var issue struct { PullRequest any `json:"pull_request"` Body *string `json:"body"` } if json.Unmarshal(raw, &issue) != nil { continue } if issue.PullRequest != nil { pulls++ } else { issues++ } if issue.Body != nil { attachmentURLs += likelyAttachmentURLCount(*issue.Body) } } report.Counts["issues"] = issues report.Counts["pull_requests"] = pulls commentURLs, commentCount, commentErr := preflightAttachmentURLs(ctx, client, repoPath(owner, name, "issues/comments")) if commentErr != nil { report.Findings = append(report.Findings, migrationFinding{Severity: "warn", Code: "issue_comments_unreadable", Message: commentErr.Error(), Action: "verify source token issue comment access"}) } else { report.Counts["issue_comments"] = commentCount attachmentURLs += commentURLs } reviewURLs, reviewCount, reviewErr := preflightAttachmentURLs(ctx, client, repoPath(owner, name, "pulls/comments")) if reviewErr != nil { report.Findings = append(report.Findings, migrationFinding{Severity: "warn", Code: "review_comments_unreadable", Message: reviewErr.Error(), Action: "verify source token review comment access"}) } else { report.Counts["review_comments"] = reviewCount attachmentURLs += reviewURLs } report.Counts["attachment_urls"] = attachmentURLs if attachmentURLs > 0 { report.Findings = append(report.Findings, migrationFinding{Severity: "warn", Code: "attachments_referenced", Message: "imported Issue, PR, or review bodies contain externally hosted attachment URLs", Action: "verify attachment retention or archive before cutover", Details: map[string]any{"url_count": attachmentURLs}}) } } if repo.Access == "handoff" { handoffs, err := snapshotAccessHandoffs(ctx, client, owner, name) if err != nil { report.Findings = append(report.Findings, migrationFinding{Severity: "block", Code: "access_handoff_unreadable", Message: err.Error(), Action: "grant the source token read access to collaborators, Teams, and Team membership, or set access: skip explicitly"}) } else { report.Counts["access_handoffs"] = len(handoffs) } } for _, probe := range []struct{ key, endpoint string }{ {"webhooks", "hooks"}, {"workflows", "actions/workflows"}, {"secret_names", "actions/secrets"}, } { raw, err := client.getRaw(ctx, repoPath(owner, name, probe.endpoint), nil) if err != nil { report.Findings = append(report.Findings, migrationFinding{Severity: "warn", Code: probe.key + "_unreadable", Message: err.Error(), Action: "verify source token scope and record whether this data is required"}) continue } report.Counts[probe.key] = jsonArrayOrWrappedCount(raw, probe.key) } return report, nil } func runPlannedRepository(ctx context.Context, plan migrationPlan, repo migrationPlanRepo, state *migrationPlanState, statePath, workDir, sourceToken, targetToken string, allowBlocking bool, mu *sync.Mutex) error { key := migrationPlanRepoKey(repo) mu.Lock() entry := state.Repositories[key] status := "" if entry != nil { status = entry.Status } mu.Unlock() if entry == nil { return fmt.Errorf("missing migration state for %s", repo.Source) } if entry.RecoveryRequired != "" { return fmt.Errorf("migration target requires explicit recovery: %s", entry.RecoveryRequired) } mu.Lock() preflight := entry.Preflight sourceFence := entry.SourceFence mu.Unlock() if status == "succeeded" { if sourceFence == nil || sourceFence.ContentDigest == "" { return errors.New("previously succeeded migration has no source content fence; use a new state path and rerun cutover") } return nil } if preflight == nil { computedPreflight, err := preflightMigrationRepository(ctx, plan, repo, sourceToken) if err != nil { return err } mu.Lock() entry.Preflight = &computedPreflight entry.UpdatedAt = time.Now().UTC() mu.Unlock() if err := saveMigrationStateLocked(statePath, state, mu); err != nil { return err } preflight = &computedPreflight } mu.Lock() preflight = entry.Preflight mu.Unlock() if hasBlockingFindings(preflight.Findings) && !allowBlocking { return errors.New("preflight has blocking findings") } if sourceFence == nil { captured, err := captureMigrationSourceFence(ctx, plan, repo, sourceToken) if err != nil { return err } mu.Lock() entry.SourceFence = &captured entry.UpdatedAt = time.Now().UTC() mu.Unlock() if err := saveMigrationStateLocked(statePath, state, mu); err != nil { return err } sourceFence = &captured } else if sourceFence.ContentDigest == "" { return errors.New("existing migration state has no source content fence; use a new state path and rerun cutover") } mu.Lock() snapshotDir := entry.SnapshotDir if snapshotDir == "" { snapshotDir = filepath.Join(workDir, "snapshots", reportFileName(repo.Source)) entry.SnapshotDir = snapshotDir entry.UpdatedAt = time.Now().UTC() } mu.Unlock() if err := saveMigrationStateLocked(statePath, state, mu); err != nil { return err } if _, err := os.Stat(filepath.Join(snapshotDir, "manifest.json")); errors.Is(err, os.ErrNotExist) { if err := githubSnapshot(ctx, []string{ "--repo", repo.Source, "--out", snapshotDir, "--api", plan.Source.APIURL, "--token-env", plan.Source.TokenEnv, "--objects", strings.Join(migrationSnapshotObjects(repo, sourceFence.ExcludedObjects), ","), "--fetch-concurrency", fmt.Sprint(plan.Execution.GitHubFetchConcurrency), "--max-asset-bytes", fmt.Sprint(plan.Execution.ReleaseAssetMaxBytes), }); err != nil { return err } } else if err != nil { return err } if err := validateMigrationSnapshotSource(snapshotDir, repo.Source); err != nil { return err } snapshotContentDigest, err := migrationSnapshotContentDigest(snapshotDir) if err != nil { return err } if sourceFence.ContentDigest == "" { mu.Lock() entry.SourceFence.ContentDigest = snapshotContentDigest entry.UpdatedAt = time.Now().UTC() sourceFence = entry.SourceFence mu.Unlock() if err := saveMigrationStateLocked(statePath, state, mu); err != nil { return err } } else if sourceFence.ContentDigest != snapshotContentDigest { return errors.New("snapshot content does not match the source fence; use a new state path and rerun cutover") } fenceFindings, err := verifyMigrationSourceFence(ctx, plan, repo, sourceToken, *sourceFence) if err != nil { return err } if len(fenceFindings) > 0 { mu.Lock() entry.Postflight = appendMigrationFindings(entry.Postflight, fenceFindings) entry.UpdatedAt = time.Now().UTC() mu.Unlock() if err := saveMigrationStateLocked(statePath, state, mu); err != nil { return err } return errors.New("source changed before target mutation; keep the source frozen and rerun with a new migration state") } mu.Lock() importID := entry.MetadataImportID mu.Unlock() if importID == 0 { id, err := createMigrationImport(ctx, plan, repo, preflight, targetToken) if err != nil { return err } mu.Lock() entry.ControlImportID = id // Kept for backward-compatible state/report readers. entry.MetadataImportID = id entry.Status = "target_created" entry.UpdatedAt = time.Now().UTC() mu.Unlock() if err := saveMigrationStateLocked(statePath, state, mu); err != nil { return err } importID = id } mu.Lock() gitMirrored := entry.ControlVerified postflight := entry.Postflight mu.Unlock() if !gitMirrored { if err := mirrorMigrationRepository(ctx, plan, repo, preflight, sourceToken, targetToken, workDir); err != nil { // A later git push --mirror converges the target to the same source // refs, so local transport failures are safe to retry with the same // durable ImportJob. mu.Lock() entry.Status = "target_created" entry.Error = "local Git or Wiki mirror failed; resolve the transport error and rerun to resume the same ImportJob" entry.UpdatedAt = time.Now().UTC() mu.Unlock() if saveErr := saveMigrationStateLocked(statePath, state, mu); saveErr != nil { return errors.Join(err, saveErr) } return err } postflight = verifyMigrationGitRefs(ctx, plan, repo, sourceToken, targetToken) mu.Lock() entry.Postflight = appendMigrationFindings(entry.Postflight, postflight) entry.ControlVerified = true entry.Status = "git_mirrored" entry.UpdatedAt = time.Now().UTC() mu.Unlock() if err := saveMigrationStateLocked(statePath, state, mu); err != nil { return err } } if hasBlockingFindings(postflight) && !allowBlocking { return errors.New("Git mirror verification has blocking findings") } metadataImportID := importID if err := uploadMigrationSnapshot(ctx, plan.Target.APIURL, targetToken, metadataImportID, snapshotDir, repo.Source); err != nil { if saveErr := markMigrationRecoveryRequired(state, statePath, mu, entry, "snapshot staging request failed; the staged import may contain partial uploaded data"); saveErr != nil { return errors.Join(err, saveErr) } return err } if _, err := applyMigrationBatches(ctx, plan.Target.APIURL, targetToken, metadataImportID, plan.Execution.ApplyBatchLimit); err != nil { if saveErr := markMigrationRecoveryRequired(state, statePath, mu, entry, "staged import apply request failed; target may contain partial metadata"); saveErr != nil { return errors.Join(err, saveErr) } return err } verification, verifyReport, err := getMigrationImportVerification(ctx, plan.Target.APIURL, targetToken, metadataImportID) if err != nil { if saveErr := markMigrationRecoveryRequired(state, statePath, mu, entry, "staged import verification could not complete after target mutation"); saveErr != nil { return errors.Join(err, saveErr) } return err } verificationFindings := migrationVerificationFindings(verifyReport) if len(verificationFindings) > 0 { mu.Lock() entry.Verification = verification entry.Postflight = appendMigrationFindings(entry.Postflight, verificationFindings) entry.RecoveryRequired = "staged import has failed or unapplied writes; target may contain partial metadata" entry.UpdatedAt = time.Now().UTC() mu.Unlock() if err := saveMigrationStateLocked(statePath, state, mu); err != nil { return err } return errors.New("staged import verification has pending or failed writes") } stackVerification, stackFindings, err := verifyMigrationStackImport(ctx, plan.Target.APIURL, targetToken, metadataImportID, snapshotDir, repo, *sourceFence) if err != nil { if saveErr := markMigrationRecoveryRequired(state, statePath, mu, entry, "post-apply Stack verification could not complete after target mutation"); saveErr != nil { return errors.Join(err, saveErr) } return err } mu.Lock() entry.Verification = verification entry.StackVerification = &stackVerification entry.UpdatedAt = time.Now().UTC() mu.Unlock() if len(stackFindings) > 0 { mu.Lock() entry.Postflight = appendMigrationFindings(entry.Postflight, stackFindings) entry.RecoveryRequired = "Stack verification found a source/target mismatch; target may contain an incomplete relationship graph" entry.UpdatedAt = time.Now().UTC() mu.Unlock() if err := saveMigrationStateLocked(statePath, state, mu); err != nil { return err } return errors.New("Stack verification failed after staged import") } if err := saveMigrationStateLocked(statePath, state, mu); err != nil { return err } fenceFindings, err = verifyMigrationSourceFence(ctx, plan, repo, sourceToken, *sourceFence) if err != nil { if saveErr := markMigrationRecoveryRequired(state, statePath, mu, entry, "post-apply source fence could not complete after target mutation"); saveErr != nil { return errors.Join(err, saveErr) } return err } if len(fenceFindings) > 0 { mu.Lock() entry.Verification = verification entry.Postflight = appendMigrationFindings(entry.Postflight, fenceFindings) entry.RecoveryRequired = "source drift was detected after target mutation; target may contain a mixed snapshot" entry.UpdatedAt = time.Now().UTC() mu.Unlock() if err := saveMigrationStateLocked(statePath, state, mu); err != nil { return err } return errors.New("source changed after the migration freeze fence was captured") } identitySummary, err := summarizeMigrationSourceUsers(ctx, plan.Target.APIURL, targetToken, metadataImportID) if err != nil { if saveErr := markMigrationRecoveryRequired(state, statePath, mu, entry, "post-apply identity summary could not complete after target mutation"); saveErr != nil { return errors.Join(err, saveErr) } return err } mu.Lock() entry.Status = "succeeded" entry.Error = "" entry.Verification = verification entry.IdentitySummary = &identitySummary entry.UpdatedAt = time.Now().UTC() mu.Unlock() return saveMigrationStateLocked(statePath, state, mu) } // markMigrationRecoveryRequired makes uncertain target state terminal for a // plan run. An interrupted request can have committed remotely after the client // lost its response, so automatic retry would risk a mixed migration. func markMigrationRecoveryRequired(state *migrationPlanState, statePath string, mu *sync.Mutex, entry *migrationRepoState, reason string) error { mu.Lock() entry.RecoveryRequired = reason entry.UpdatedAt = time.Now().UTC() mu.Unlock() return saveMigrationStateLocked(statePath, state, mu) } func createMigrationImport(ctx context.Context, plan migrationPlan, repo migrationPlanRepo, preflight *migrationPreflight, targetToken string) (uint, error) { sourceOwner, sourceName, _ := splitRepo(repo.Source) targetOwner, targetName, _ := splitRepo(repo.Target) request := map[string]any{ "source_owner": sourceOwner, "source_repo": sourceName, "source_api_url": plan.Source.APIURL, "target_owner": targetOwner, "target_repo": targetName, "target_mode": repo.TargetMode, "options": map[string]any{ "import_actions_metadata": true, }, } if preflight != nil { targetRepository := preflight.SourceRepository // State files from before repository metadata was added still have the // original default-branch field. Keep their target safe and usable on // resume rather than interpreting absent boolean fields as disabled. if targetRepository.DefaultBranch == "" { targetRepository = migrationSourceRepository{ Visibility: "private", Private: true, DefaultBranch: preflight.SourceDefaultBranch, HasIssues: true, HasWiki: preflight.SourceWiki, } } request["target_repository"] = targetRepository } body, err := postJSONResponse(ctx, strings.TrimRight(plan.Target.APIURL, "/")+"/api/ext/v1/imports/github", targetToken, request) if err != nil { return 0, err } var result migrationControlImportResponse if err := json.Unmarshal(body, &result); err != nil { return 0, fmt.Errorf("decode import response: %w", err) } if result.ID == 0 { return 0, errors.New("AGS import response has no id") } return result.ID, nil } func mirrorMigrationRepository(ctx context.Context, plan migrationPlan, repo migrationPlanRepo, preflight *migrationPreflight, sourceToken, targetToken, workDir string) error { sourceOwner, sourceName, _ := splitRepo(repo.Source) targetOwner, targetName, _ := splitRepo(repo.Target) sourceRemote, err := githubGitRemote(plan.Source.APIURL, sourceOwner, sourceName) if err != nil { return err } targetRemote, err := agsGitRemote(plan.Target.APIURL, targetOwner, targetName) if err != nil { return err } if err := mirrorGitRepository(ctx, sourceRemote, sourceToken, targetRemote, targetToken, workDir); err != nil { return err } if repo.Wiki == "disabled" || preflight == nil || !preflight.SourceWiki { return nil } if err := mirrorGitRepository(ctx, strings.TrimSuffix(sourceRemote, ".git")+".wiki.git", sourceToken, strings.TrimSuffix(targetRemote, ".git")+".wiki.git", targetToken, workDir); err != nil { if repo.Wiki == "optional" { return fmt.Errorf("optional source Wiki exists but could not be mirrored: %w", err) } return fmt.Errorf("required source Wiki could not be mirrored: %w", err) } return nil } func uploadMigrationSnapshot(ctx context.Context, target, token string, importID uint, snapshotDir, sourceRepo string) error { manifest, err := readManifest(snapshotDir) if err != nil { return err } if !strings.EqualFold(strings.TrimSpace(manifest.SourceRepo), strings.TrimSpace(sourceRepo)) { return fmt.Errorf("snapshot source %q does not match planned repository %q", manifest.SourceRepo, sourceRepo) } baseEndpoint := strings.TrimRight(target, "/") + "/api/ext/v1/imports/" + fmt.Sprint(importID) if err := uploadSnapshotAssetFiles(ctx, baseEndpoint, token, snapshotDir, manifest); err != nil { return err } requests := make([]importBatchRequest, 0, len(manifest.Shards)) for _, shard := range manifest.Shards { items, err := readBatchItems(filepath.Join(snapshotDir, shard.Path)) if err != nil { return err } if len(items) == 0 { continue } requests = append(requests, importBatchRequest{BatchID: strings.TrimSuffix(filepath.Base(shard.Path), filepath.Ext(shard.Path)), ObjectType: shard.ObjectType, SchemaVersion: manifest.SchemaVersion, Items: items}) } endpoint := baseEndpoint + "/batches/bulk" for _, chunk := range chunkUploadBatches(requests, defaultUploadBulkShards, defaultUploadBulkBytes) { if _, err := postJSONResponse(ctx, endpoint, token, importBatchesRequest{Batches: chunk}); err != nil { return err } } return nil } func validateMigrationSnapshotSource(snapshotDir, sourceRepo string) error { manifest, err := readManifest(snapshotDir) if err != nil { return err } if !strings.EqualFold(strings.TrimSpace(manifest.SourceRepo), strings.TrimSpace(sourceRepo)) { return fmt.Errorf("snapshot source %q does not match planned repository %q", manifest.SourceRepo, sourceRepo) } return nil } func applyMigrationBatches(ctx context.Context, target, token string, importID uint, batchLimit int) (importApplyResponse, error) { endpoint := strings.TrimRight(target, "/") + "/api/ext/v1/imports/" + fmt.Sprint(importID) + "/apply" for { body, err := postJSONResponse(ctx, endpoint, token, importApplyRequest{BatchLimit: batchLimit, SkipVerify: true}) if err != nil { return importApplyResponse{}, err } var report importApplyResponse if err := json.Unmarshal(body, &report); err != nil { return importApplyResponse{}, err } if report.RemainingSupportedBatches == 0 { return report, nil } if report.AppliedBatches == 0 { return report, fmt.Errorf("import %d apply made no progress", importID) } } } func getMigrationImportVerification(ctx context.Context, target, token string, importID uint) (json.RawMessage, migrationImportVerification, error) { body, err := getMigrationJSON(ctx, strings.TrimRight(target, "/")+"/api/ext/v1/imports/"+fmt.Sprint(importID)+"/verify", token) if err != nil { return nil, migrationImportVerification{}, err } var report migrationImportVerification if err := json.Unmarshal(body, &report); err != nil { return nil, migrationImportVerification{}, fmt.Errorf("decode staged import verification: %w", err) } return json.RawMessage(body), report, nil } func migrationVerificationFindings(report migrationImportVerification) []migrationFinding { findings := make([]migrationFinding, 0, 2) if count := migrationObjectCount(report.FailedWrites); count > 0 { findings = append(findings, migrationFinding{ Severity: "block", Code: "staged_import_failed_writes", Message: "one or more staged import writes failed", Action: "resolve the failed batches or reset the target import before resuming", Details: map[string]any{"count": count, "by_type": report.FailedWrites}, }) } if count := migrationObjectCount(report.PendingWrites); count > 0 { findings = append(findings, migrationFinding{ Severity: "block", Code: "staged_import_pending_writes", Message: "one or more staged import writes remain unapplied", Action: "apply every staged batch successfully before cutover", Details: map[string]any{"count": count, "by_type": report.PendingWrites}, }) } return findings } func migrationObjectCount(values map[string]int) int { total := 0 for _, value := range values { if value > 0 { total += value } } return total } func verifyMigrationGitRefs(ctx context.Context, plan migrationPlan, repo migrationPlanRepo, sourceToken, targetToken string) []migrationFinding { sourceOwner, sourceName, _ := splitRepo(repo.Source) targetOwner, targetName, _ := splitRepo(repo.Target) sourceRefs, err := migrationGitHubRefs(ctx, newGitHubClient(plan.Source.APIURL, sourceToken), sourceOwner, sourceName) if err != nil { return []migrationFinding{{Severity: "block", Code: "source_git_refs_unverified", Message: err.Error(), Action: "restore source Git API access and rerun verification before cutover"}} } targetRefs, err := migrationAGSRefs(ctx, plan.Target.APIURL, targetToken, targetOwner, targetName) if err != nil { return []migrationFinding{{Severity: "block", Code: "target_git_refs_unverified", Message: err.Error(), Action: "restore AGS Git API access and rerun verification before cutover"}} } missing, mismatched, extra := compareMigrationGitRefs(migrationComparableGitRefs(sourceRefs), migrationComparableGitRefs(targetRefs)) if len(missing) == 0 && len(mismatched) == 0 && len(extra) == 0 { return nil } return []migrationFinding{{ Severity: "block", Code: "git_refs_mismatch", Message: "source and target Git refs differ after mirror import", Action: "resolve ref mismatches and rerun verification before cutover", Details: map[string]any{ "source_ref_count": len(sourceRefs), "target_ref_count": len(targetRefs), "missing": missing, "mismatched": mismatched, "extra": extra, }, }} } func captureMigrationSourceFence(ctx context.Context, plan migrationPlan, repo migrationPlanRepo, sourceToken string) (migrationSourceFence, error) { preflight, err := preflightMigrationRepository(ctx, plan, repo, sourceToken) if err != nil { return migrationSourceFence{}, fmt.Errorf("capture source freeze inventory: %w", err) } owner, name, _ := splitRepo(repo.Source) refs, err := migrationGitHubRefs(ctx, newGitHubClient(plan.Source.APIURL, sourceToken), owner, name) if err != nil { return migrationSourceFence{}, fmt.Errorf("capture source freeze Git refs: %w", err) } inventoryDigest, err := migrationPreflightInventoryDigest(preflight) if err != nil { return migrationSourceFence{}, err } return migrationSourceFence{ CapturedAt: time.Now().UTC(), GitRefsDigest: migrationGitRefsDigest(migrationComparableGitRefs(refs)), InventoryDigest: inventoryDigest, ExcludedObjects: migrationExcludedObjects(repo, preflight), SourceUpdatedAt: preflight.SourceUpdatedAt, SourcePushedAt: preflight.SourcePushedAt, InventoryCounts: cloneMigrationCounts(preflight.Counts), InventoryBytes: cloneMigrationBytes(preflight.Bytes), }, nil } func verifyMigrationSourceFence(ctx context.Context, plan migrationPlan, repo migrationPlanRepo, sourceToken string, before migrationSourceFence) ([]migrationFinding, error) { if before.ContentDigest == "" { return nil, errors.New("source content fence is missing") } after, err := captureMigrationSourceFence(ctx, plan, repo, sourceToken) if err != nil { return nil, err } after.ContentDigest, err = migrationSourceContentDigest(ctx, plan, repo, sourceToken, before.ExcludedObjects) if err != nil { return nil, fmt.Errorf("capture source content fence: %w", err) } if before.GitRefsDigest == after.GitRefsDigest && before.InventoryDigest == after.InventoryDigest && before.ContentDigest == after.ContentDigest { return nil, nil } return []migrationFinding{{ Severity: "block", Code: "source_changed_during_cutover", Message: "source Git refs, imported content, or collaboration inventory changed after the freeze fence was captured", Action: "keep the source frozen, start a new migration state, and rerun the cutover", Details: map[string]any{ "before_git_refs_digest": before.GitRefsDigest, "after_git_refs_digest": after.GitRefsDigest, "before_inventory_digest": before.InventoryDigest, "after_inventory_digest": after.InventoryDigest, "before_content_digest": before.ContentDigest, "after_content_digest": after.ContentDigest, }, }}, nil } func migrationSourceContentDigest(ctx context.Context, plan migrationPlan, repo migrationPlanRepo, sourceToken string, excludedObjects []string) (string, error) { owner, name, err := splitRepo(repo.Source) if err != nil { return "", err } client := newGitHubClient(plan.Source.APIURL, sourceToken) cache := snapshotSourceCache{} payloadsByType := make(map[string][]json.RawMessage) for _, selected := range migrationSnapshotObjects(repo, excludedObjects) { objectType, payloads, err := snapshotObjectPayloads(ctx, client, owner, name, selected, &cache, plan.Execution.GitHubFetchConcurrency, plan.Execution.ReleaseAssetMaxBytes) if err != nil { return "", fmt.Errorf("read source %s for content fence: %w", objectType, err) } if len(payloads) > 0 { payloadsByType[objectType] = append(payloadsByType[objectType], payloads...) } } return migrationPayloadDigest(payloadsByType) } func migrationExcludedObjects(repo migrationPlanRepo, preflight migrationPreflight) []string { if repo.StackAPI != "skip_if_unreadable" || !migrationHasFindingCode(preflight.Findings, "github_stacks_unreadable_waived") { return nil } return []string{"pull_request_stacks"} } func migrationHasFindingCode(findings []migrationFinding, code string) bool { for _, finding := range findings { if finding.Code == code { return true } } return false } func migrationSnapshotObjects(repo migrationPlanRepo, excludedObjects []string) []string { excluded := make(map[string]struct{}, len(excludedObjects)) for _, objectType := range excludedObjects { excluded[strings.ToLower(strings.TrimSpace(objectType))] = struct{}{} } selected := expandSnapshotObjects(repo.Mode) filtered := make([]string, 0, len(selected)) for _, objectType := range selected { if _, skip := excluded[strings.ToLower(objectType)]; !skip { filtered = append(filtered, objectType) } } if repo.Access == "handoff" { if _, skip := excluded["access_handoffs"]; !skip && !migrationSnapshotObjectSelected(filtered, "access_handoffs") { filtered = append(filtered, "access_handoffs") } } return filtered } func migrationSnapshotObjectSelected(objects []string, wanted string) bool { for _, objectType := range objects { if strings.EqualFold(strings.TrimSpace(objectType), wanted) { return true } } return false } func migrationObjectIsExcluded(excludedObjects []string, objectType string) bool { for _, excluded := range excludedObjects { if strings.EqualFold(strings.TrimSpace(excluded), objectType) { return true } } return false } func migrationSnapshotContentDigest(snapshotDir string) (string, error) { manifest, err := readManifest(snapshotDir) if err != nil { return "", err } itemDigestsByType := make(map[string][]string) for _, asset := range manifest.AssetFiles { path, err := snapshotAssetFilePath(snapshotDir, asset.Path) if err != nil { return "", fmt.Errorf("invalid snapshot asset path for %q: %w", asset.SourceID, err) } if err := verifySnapshotAssetFile(path, asset.Size, asset.Checksum); err != nil { return "", fmt.Errorf("verify snapshot asset %q: %w", asset.SourceID, err) } } for _, shard := range manifest.Shards { file, err := os.Open(filepath.Join(snapshotDir, shard.Path)) if err != nil { return "", err } reader := bufio.NewReaderSize(file, 256*1024) for { line, readErr := reader.ReadBytes('\n') line = bytes.TrimSpace(line) if len(line) > 0 { itemDigest, err := migrationCanonicalPayloadDigest(line) if err != nil { file.Close() return "", fmt.Errorf("canonicalize %s snapshot payload: %w", shard.ObjectType, err) } itemDigestsByType[shard.ObjectType] = append(itemDigestsByType[shard.ObjectType], itemDigest) } if errors.Is(readErr, io.EOF) { break } if readErr != nil { file.Close() return "", readErr } } if err := file.Close(); err != nil { return "", err } } return migrationItemDigest(itemDigestsByType), nil } func migrationPayloadDigest(payloadsByType map[string][]json.RawMessage) (string, error) { itemDigestsByType := make(map[string][]string, len(payloadsByType)) for objectType, payloads := range payloadsByType { itemDigests := make([]string, 0, len(payloads)) for _, payload := range payloads { itemDigest, err := migrationCanonicalPayloadDigest(payload) if err != nil { return "", fmt.Errorf("canonicalize %s payload: %w", objectType, err) } itemDigests = append(itemDigests, itemDigest) } itemDigestsByType[objectType] = itemDigests } return migrationItemDigest(itemDigestsByType), nil } func migrationItemDigest(itemDigestsByType map[string][]string) string { types := make([]string, 0, len(itemDigestsByType)) for objectType := range itemDigestsByType { types = append(types, objectType) } sort.Strings(types) hash := sha256.New() for _, objectType := range types { itemDigests := itemDigestsByType[objectType] sort.Strings(itemDigests) _, _ = io.WriteString(hash, objectType) _, _ = io.WriteString(hash, "\x00") for _, itemDigest := range itemDigests { _, _ = io.WriteString(hash, itemDigest) _, _ = io.WriteString(hash, "\n") } } return "sha256:" + hex.EncodeToString(hash.Sum(nil)) } func migrationCanonicalPayloadDigest(payload json.RawMessage) (string, error) { decoder := json.NewDecoder(bytes.NewReader(payload)) decoder.UseNumber() var value any if err := decoder.Decode(&value); err != nil { return "", err } value = removeSnapshotTransportFields(value) canonical, err := json.Marshal(value) if err != nil { return "", err } sum := sha256.Sum256(canonical) return "sha256:" + hex.EncodeToString(sum[:]), nil } func removeSnapshotTransportFields(value any) any { switch typed := value.(type) { case map[string]any: out := make(map[string]any, len(typed)) for key, child := range typed { if key == "content_base64" { continue } out[key] = removeSnapshotTransportFields(child) } return out case []any: out := make([]any, len(typed)) for i, child := range typed { out[i] = removeSnapshotTransportFields(child) } return out default: return value } } func appendMigrationFindings(existing, additions []migrationFinding) []migrationFinding { for _, addition := range additions { duplicate := false for _, current := range existing { if current.Code == addition.Code && current.Message == addition.Message { duplicate = true break } } if !duplicate { existing = append(existing, addition) } } return existing } func migrationPreflightInventoryDigest(preflight migrationPreflight) (string, error) { payload := struct { SourceUpdatedAt time.Time `json:"source_updated_at"` SourcePushedAt *time.Time `json:"source_pushed_at,omitempty"` SourceDefaultBranch string `json:"source_default_branch"` SourceWiki bool `json:"source_wiki"` LFSDetected bool `json:"lfs_detected"` Counts map[string]int `json:"counts"` Bytes map[string]int64 `json:"bytes"` }{ SourceUpdatedAt: preflight.SourceUpdatedAt, SourcePushedAt: preflight.SourcePushedAt, SourceDefaultBranch: preflight.SourceDefaultBranch, SourceWiki: preflight.SourceWiki, LFSDetected: preflight.LFSDetected, Counts: preflight.Counts, Bytes: preflight.Bytes, } data, err := json.Marshal(payload) if err != nil { return "", fmt.Errorf("encode source freeze inventory: %w", err) } sum := sha256.Sum256(data) return "sha256:" + hex.EncodeToString(sum[:]), nil } func migrationGitRefsDigest(refs map[string]string) string { names := make([]string, 0, len(refs)) for name := range refs { names = append(names, name) } sort.Strings(names) hash := sha256.New() for _, name := range names { _, _ = io.WriteString(hash, name) _, _ = io.WriteString(hash, "\x00") _, _ = io.WriteString(hash, refs[name]) _, _ = io.WriteString(hash, "\n") } return "sha256:" + hex.EncodeToString(hash.Sum(nil)) } func cloneMigrationCounts(in map[string]int) map[string]int { out := make(map[string]int, len(in)) for key, value := range in { out[key] = value } return out } func cloneMigrationBytes(in map[string]int64) map[string]int64 { out := make(map[string]int64, len(in)) for key, value := range in { out[key] = value } return out } // migrationComparableGitRefs excludes GitHub's generated PR merge previews // and the temporary remote-tracking refs created while mirroring. Neither is // a stable repository ref that AGS can preserve or should require at cutover. func migrationComparableGitRefs(refs map[string]string) map[string]string { filtered := make(map[string]string, len(refs)) for name, sha := range refs { if strings.HasPrefix(name, "refs/remotes/") { continue } if strings.HasPrefix(name, "refs/pull/") && strings.HasSuffix(name, "/merge") { continue } filtered[name] = sha } return filtered } func migrationGitHubRefs(ctx context.Context, client *githubClient, owner, name string) (map[string]string, error) { raw, err := client.listRaw(ctx, repoPath(owner, name, "git/matching-refs/"), nil) if err != nil { return nil, err } return decodeMigrationGitRefs(raw) } func migrationAGSRefs(ctx context.Context, target, token, owner, name string) (map[string]string, error) { endpoint := strings.TrimRight(target, "/") + "/api/v3/repos/" + url.PathEscape(owner) + "/" + url.PathEscape(name) + "/git/matching-refs" body, err := getMigrationJSON(ctx, endpoint, token) if err != nil { return nil, err } var raw []json.RawMessage if err := json.Unmarshal(body, &raw); err != nil { return nil, fmt.Errorf("decode AGS Git refs: %w", err) } return decodeMigrationGitRefs(raw) } func decodeMigrationGitRefs(raw []json.RawMessage) (map[string]string, error) { refs := make(map[string]string, len(raw)) for _, item := range raw { var ref struct { Ref string `json:"ref"` Object struct { SHA string `json:"sha"` } `json:"object"` } if err := json.Unmarshal(item, &ref); err != nil { return nil, err } if strings.TrimSpace(ref.Ref) == "" || strings.TrimSpace(ref.Object.SHA) == "" { continue } refs[ref.Ref] = ref.Object.SHA } return refs, nil } func compareMigrationGitRefs(source, target map[string]string) (missing, mismatched, extra []string) { for ref, sourceSHA := range source { targetSHA, ok := target[ref] switch { case !ok: missing = append(missing, ref) case !strings.EqualFold(sourceSHA, targetSHA): mismatched = append(mismatched, ref) } } for ref := range target { if _, ok := source[ref]; !ok { extra = append(extra, ref) } } sort.Strings(missing) sort.Strings(mismatched) sort.Strings(extra) return limitMigrationRefSamples(missing), limitMigrationRefSamples(mismatched), limitMigrationRefSamples(extra) } func limitMigrationRefSamples(refs []string) []string { const maxSamples = 20 if len(refs) <= maxSamples { return refs } return append(refs[:maxSamples:maxSamples], fmt.Sprintf("... and %d more", len(refs)-maxSamples)) } func getMigrationImportMappings(ctx context.Context, target, token string, importID uint, targetType string) ([]migrationImportMapping, error) { endpoint := strings.TrimRight(target, "/") + "/api/ext/v1/imports/" + fmt.Sprint(importID) + "/mapping" mappings := make([]migrationImportMapping, 0) var afterID uint for { query := url.Values{"limit": {"500"}, "target_type": {targetType}} if afterID > 0 { query.Set("after_id", strconv.FormatUint(uint64(afterID), 10)) } body, err := getMigrationJSON(ctx, endpoint+"?"+query.Encode(), token) if err != nil { return nil, err } var page []migrationImportMapping if err := json.Unmarshal(body, &page); err != nil { return nil, fmt.Errorf("decode import mappings: %w", err) } if len(page) == 0 { return mappings, nil } for _, mapping := range page { if mapping.ID == 0 || mapping.ID <= afterID { return nil, errors.New("import mapping pagination did not advance by ID") } mappings = append(mappings, mapping) afterID = mapping.ID } if len(page) < 500 { return mappings, nil } } } func verifyMigrationStackImport(ctx context.Context, target, token string, importID uint, snapshotDir string, repo migrationPlanRepo, fence migrationSourceFence) (migrationStackVerification, []migrationFinding, error) { if migrationObjectIsExcluded(fence.ExcludedObjects, "pull_request_stacks") { return migrationStackVerification{Skipped: true, SkipReason: "Stack API access was explicitly waived before the source freeze"}, nil, nil } sourceStacks, err := migrationSnapshotStacks(snapshotDir) if err != nil { return migrationStackVerification{}, nil, err } targetStacks, err := listMigrationTargetStacks(ctx, target, token, repo.Target) if err != nil { return migrationStackVerification{}, nil, err } mappings, err := getMigrationImportMappings(ctx, target, token, importID, "pull_request_stack") if err != nil { return migrationStackVerification{}, nil, err } verification := migrationStackVerification{SourceCount: len(sourceStacks), TargetCount: len(targetStacks), MappingCount: len(mappings)} targetByID := make(map[string]migrationTargetStack, len(targetStacks)) for _, stack := range targetStacks { targetByID[strconv.FormatUint(uint64(stack.ID), 10)] = stack } mappingBySourceID := make(map[string]migrationImportMapping, len(mappings)) for _, mapping := range mappings { if mapping.SourceID == "" { continue } if _, exists := mappingBySourceID[mapping.SourceID]; !exists { mappingBySourceID[mapping.SourceID] = mapping } } sourceByID := make(map[string]struct{}, len(sourceStacks)) for _, source := range sourceStacks { sourceByID[source.ID] = struct{}{} check := migrationStackCheck{SourceID: source.ID, SourceNumber: source.Number, SourceURL: source.URL, SourcePullRequestNumbers: source.PullRequestNumbers} problems := make([]string, 0) mapping, found := mappingBySourceID[source.ID] if !found { problems = append(problems, "missing_mapping") } else { if mapping.SourceType != "pull_request_stack" { problems = append(problems, "wrong_source_type") } if mapping.SourceNumber == nil || *mapping.SourceNumber != source.Number { problems = append(problems, "source_number_mismatch") } if mapping.SourceURL != source.URL { problems = append(problems, "source_url_mismatch") } targetStack, exists := targetByID[mapping.TargetID] if !exists { problems = append(problems, "missing_target_stack") } else { check.TargetID = mapping.TargetID check.TargetNumber = targetStack.Number check.TargetURL = mapping.TargetURL check.TargetAPIURL = targetStack.URL check.TargetPullRequestNumbers = migrationTargetStackPullRequestNumbers(targetStack) check.MembersMatch = migrationSamePullRequestMembers(source.PullRequestNumbers, check.TargetPullRequestNumbers) check.OrderMatches = migrationSamePullRequestOrder(source.PullRequestNumbers, check.TargetPullRequestNumbers) check.URLMappingMatches = mapping.TargetNumber != nil && *mapping.TargetNumber == targetStack.Number && migrationStackTargetURLMatches(repo.Target, mapping.TargetURL, targetStack.Number) if !check.MembersMatch { problems = append(problems, "member_mismatch") } if !check.OrderMatches { problems = append(problems, "member_order_mismatch") } if mapping.TargetNumber == nil || *mapping.TargetNumber != targetStack.Number { problems = append(problems, "target_number_mismatch") } if !check.URLMappingMatches { problems = append(problems, "target_url_not_gh_web_stack_route") } } } if len(problems) == 0 { check.Status = "verified" verification.VerifiedCount++ } else { check.Status = strings.Join(problems, ",") } verification.Checks = append(verification.Checks, check) } for _, mapping := range mappings { if _, exists := sourceByID[mapping.SourceID]; !exists { verification.Checks = append(verification.Checks, migrationStackCheck{SourceID: mapping.SourceID, TargetID: mapping.TargetID, TargetURL: mapping.TargetURL, Status: "mapping_has_no_source_stack"}) } } sort.Slice(verification.Checks, func(i, j int) bool { if verification.Checks[i].SourceNumber != verification.Checks[j].SourceNumber { return verification.Checks[i].SourceNumber < verification.Checks[j].SourceNumber } return verification.Checks[i].SourceID < verification.Checks[j].SourceID }) failedChecks := len(verification.Checks) - verification.VerifiedCount if verification.SourceCount == verification.TargetCount && verification.SourceCount == verification.MappingCount && failedChecks == 0 { return verification, nil, nil } return verification, []migrationFinding{{ Severity: "block", Code: "stack_verification_failed", Message: "source and target pull request Stacks differ after staged import", Action: "reset the target Stack import, correct the source/target relationship mismatch, and rerun before cutover", Details: map[string]any{"source_count": verification.SourceCount, "target_count": verification.TargetCount, "mapping_count": verification.MappingCount, "verified_count": verification.VerifiedCount, "failed_check_count": failedChecks, "samples": migrationStackCheckSamples(verification.Checks)}, }}, nil } func migrationSnapshotStacks(snapshotDir string) ([]migrationSnapshotStack, error) { manifest, err := readManifest(snapshotDir) if err != nil { return nil, err } stacks := make([]migrationSnapshotStack, 0) seen := make(map[string]struct{}) for _, shard := range manifest.Shards { if shard.ObjectType != "pull_request_stacks" && shard.ObjectType != "pull_request_stack" { continue } items, err := readBatchItems(filepath.Join(snapshotDir, shard.Path)) if err != nil { return nil, err } for _, item := range items { var payload struct { ID int64 `json:"id"` Number int `json:"number"` URL string `json:"url"` PullRequests []struct { Number int `json:"number"` } `json:"pull_requests"` } if err := json.Unmarshal(item.Payload, &payload); err != nil { return nil, fmt.Errorf("decode Stack snapshot payload: %w", err) } stack := migrationSnapshotStack{ID: firstString(item.SourceID, strconv.FormatInt(payload.ID, 10)), Number: payload.Number, URL: firstString(item.SourceURL, payload.URL)} if item.SourceNumber != nil { stack.Number = *item.SourceNumber } for _, pullRequest := range payload.PullRequests { stack.PullRequestNumbers = append(stack.PullRequestNumbers, pullRequest.Number) } if stack.ID == "" || stack.Number <= 0 || stack.URL == "" { return nil, errors.New("Stack snapshot has no stable source ID, number, or URL") } if err := validateMigrationStackMembers(stack.PullRequestNumbers); err != nil { return nil, fmt.Errorf("Stack #%d: %w", stack.Number, err) } if _, exists := seen[stack.ID]; exists { return nil, fmt.Errorf("duplicate Stack source ID %q in snapshot", stack.ID) } seen[stack.ID] = struct{}{} stacks = append(stacks, stack) } } sort.Slice(stacks, func(i, j int) bool { if stacks[i].Number != stacks[j].Number { return stacks[i].Number < stacks[j].Number } return stacks[i].ID < stacks[j].ID }) return stacks, nil } func listMigrationTargetStacks(ctx context.Context, target, token, targetRepo string) ([]migrationTargetStack, error) { owner, name, err := splitRepo(targetRepo) if err != nil { return nil, err } endpoint := strings.TrimRight(target, "/") + "/api/v3/repos/" + url.PathEscape(owner) + "/" + url.PathEscape(name) + "/stacks" stacks := make([]migrationTargetStack, 0) for page := 1; ; page++ { query := url.Values{"per_page": {"100"}, "page": {strconv.Itoa(page)}} body, err := getMigrationJSON(ctx, endpoint+"?"+query.Encode(), token) if err != nil { return nil, err } var items []migrationTargetStack if err := json.Unmarshal(body, &items); err != nil { return nil, fmt.Errorf("decode target Stacks: %w", err) } stacks = append(stacks, items...) if len(items) < 100 { return stacks, nil } } } func migrationTargetStackPullRequestNumbers(stack migrationTargetStack) []int { numbers := make([]int, 0, len(stack.PullRequests)) for _, pullRequest := range stack.PullRequests { numbers = append(numbers, pullRequest.Number) } return numbers } func validateMigrationStackMembers(numbers []int) error { if len(numbers) < 2 { return errors.New("Stack must contain at least two pull requests") } seen := make(map[int]struct{}, len(numbers)) for _, number := range numbers { if number <= 0 { return errors.New("Stack contains an invalid pull request number") } if _, exists := seen[number]; exists { return fmt.Errorf("Stack contains duplicate pull request #%d", number) } seen[number] = struct{}{} } return nil } func migrationSamePullRequestMembers(source, target []int) bool { if len(source) != len(target) { return false } counts := make(map[int]int, len(source)) for _, number := range source { counts[number]++ } for _, number := range target { counts[number]-- } for _, count := range counts { if count != 0 { return false } } return true } func migrationSamePullRequestOrder(source, target []int) bool { if len(source) != len(target) { return false } for i := range source { if source[i] != target[i] { return false } } return true } func migrationStackTargetURLMatches(targetRepo, rawURL string, stackNumber int) bool { owner, name, err := splitRepo(targetRepo) if err != nil { return false } parsed, err := url.ParseRequestURI(rawURL) if err != nil || parsed.Scheme == "" || parsed.Host == "" { return false } expected := "/" + url.PathEscape(owner) + "/" + url.PathEscape(name) + "/stacks/" + strconv.Itoa(stackNumber) return strings.HasSuffix(strings.TrimRight(parsed.EscapedPath(), "/"), expected) } func migrationStackCheckSamples(checks []migrationStackCheck) []migrationStackCheck { const maxSamples = 20 samples := make([]migrationStackCheck, 0, maxSamples) for _, check := range checks { if check.Status == "verified" { continue } samples = append(samples, check) if len(samples) == maxSamples { break } } return samples } func getMigrationJSON(ctx context.Context, endpoint, token string) ([]byte, error) { req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) if err != nil { return nil, err } req.Header.Set("Authorization", "Bearer "+token) resp, err := http.DefaultClient.Do(req) if err != nil { return nil, err } defer resp.Body.Close() body, err := io.ReadAll(resp.Body) if err != nil { return nil, err } if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { return nil, fmt.Errorf("GET %s failed: %s: %s", endpoint, resp.Status, strings.TrimSpace(string(body))) } return body, nil } func loadOrCreateMigrationState(path, planPath, digest string, plan migrationPlan) (*migrationPlanState, error) { state, err := readMigrationState(path) if err == nil { if state.PlanDigest != digest { return nil, errors.New("migration plan changed; use a new --state path or intentionally reset state") } return state, nil } if !errors.Is(err, os.ErrNotExist) { return nil, err } state = &migrationPlanState{Version: migrationPlanVersion, PlanDigest: digest, PlanPath: planPath, StartedAt: time.Now().UTC(), Repositories: map[string]*migrationRepoState{}} for _, repo := range plan.Repositories { state.Repositories[migrationPlanRepoKey(repo)] = &migrationRepoState{Source: repo.Source, Target: repo.Target, Status: "pending", UpdatedAt: time.Now().UTC()} } return state, nil } func readMigrationState(path string) (*migrationPlanState, error) { data, err := os.ReadFile(path) if err != nil { return nil, err } var state migrationPlanState if err := json.Unmarshal(data, &state); err != nil { return nil, fmt.Errorf("parse migration state: %w", err) } if state.Repositories == nil { state.Repositories = map[string]*migrationRepoState{} } return &state, nil } func saveMigrationStateLocked(path string, state *migrationPlanState, mu *sync.Mutex) error { mu.Lock() defer mu.Unlock() return saveMigrationState(path, state) } func saveMigrationState(path string, state *migrationPlanState) error { state.UpdatedAt = time.Now().UTC() data, err := json.MarshalIndent(state, "", " ") if err != nil { return err } dir := filepath.Dir(path) if err := os.MkdirAll(dir, 0o755); err != nil { return err } tmp, err := os.CreateTemp(dir, ".ags-migrate-state-*") if err != nil { return err } tmpName := tmp.Name() defer os.Remove(tmpName) if _, err := tmp.Write(append(data, '\n')); err != nil { tmp.Close() return err } if err := tmp.Chmod(0o600); err != nil { tmp.Close() return err } if err := tmp.Close(); err != nil { return err } return os.Rename(tmpName, path) } func writeMigrationPreflight(outDir string, report migrationPreflight) error { base := reportFileName(report.Source) if err := writeJSONFile(filepath.Join(outDir, base+".json"), report); err != nil { return err } return writeMigrationMarkdown(filepath.Join(outDir, base+".md"), []migrationPreflight{report}) } func writeJSONFile(path string, value any) error { data, err := json.MarshalIndent(value, "", " ") if err != nil { return err } return os.WriteFile(path, append(data, '\n'), 0o644) } func writeMigrationMarkdown(path string, reports []migrationPreflight) error { var b strings.Builder b.WriteString("# AGS Migration Preflight\n\n") for _, report := range reports { fmt.Fprintf(&b, "## %s -> %s\n\n", report.Source, report.Target) fmt.Fprintf(&b, "- Checked: %s\n", report.CheckedAt.Format(time.RFC3339)) for _, finding := range report.Findings { fmt.Fprintf(&b, "- [%s] %s: %s\n", strings.ToUpper(finding.Severity), finding.Code, finding.Message) } b.WriteByte('\n') } return os.WriteFile(path, []byte(b.String()), 0o644) } func writeMigrationReportMarkdown(path string, reports []migrationRepoReport) error { var b strings.Builder b.WriteString("# AGS Migration Report\n\n") for _, report := range reports { fmt.Fprintf(&b, "## %s -> %s\n\n", report.Source, report.Target) fmt.Fprintf(&b, "- Status: %s\n- Cutover ready: %t\n", report.Status, report.CutoverReady) if report.Error != "" { fmt.Fprintf(&b, "- Error: %s\n", report.Error) } if stack := report.StackVerification; stack != nil { if stack.Skipped { fmt.Fprintf(&b, "- Stack verification: skipped (%s)\n", stack.SkipReason) } else { fmt.Fprintf(&b, "- Stack verification: source %d, target %d, mappings %d, verified %d\n", stack.SourceCount, stack.TargetCount, stack.MappingCount, stack.VerifiedCount) for _, check := range migrationStackCheckSamples(stack.Checks) { source := migrationMarkdownStackLink("source Stack #"+strconv.Itoa(check.SourceNumber), check.SourceURL) target := migrationMarkdownStackLink("target Stack #"+strconv.Itoa(check.TargetNumber), check.TargetURL) fmt.Fprintf(&b, "- Stack mismatch: %s -> %s (%s)\n", source, target, check.Status) } } } for _, finding := range report.Findings { fmt.Fprintf(&b, "- [%s] %s: %s\n", strings.ToUpper(finding.Severity), finding.Code, finding.Message) } b.WriteByte('\n') } return os.WriteFile(path, []byte(b.String()), 0o644) } func migrationMarkdownStackLink(label, rawURL string) string { parsed, err := url.ParseRequestURI(rawURL) if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Host == "" { return label } return "[" + label + "](<" + parsed.String() + ">)" } func migrationPlanRepoKey(repo migrationPlanRepo) string { return repo.Source + "->" + repo.Target } func reportFileName(repo string) string { readable := strings.NewReplacer("/", "-", "\\", "-", " ", "-").Replace(strings.ToLower(strings.TrimSpace(repo))) sum := sha256.Sum256([]byte(repo)) return readable + "-" + hex.EncodeToString(sum[:])[:12] } func requiredEnv(name string) (string, error) { name = strings.TrimSpace(name) if name == "" { return "", errors.New("token environment variable is empty") } value := strings.TrimSpace(os.Getenv(name)) if value == "" { return "", fmt.Errorf("%s is empty", name) } return value, nil } func hasBlockingFindings(findings []migrationFinding) bool { for _, finding := range findings { if finding.Severity == "block" { return true } } return false } func hasUnresolvedFindings(findings []migrationFinding) bool { for _, finding := range findings { if finding.Severity == "warn" || finding.Severity == "block" { return true } } return false } func lfsFindingSeverity(plan migrationPlan) string { if plan.Cutover.RequireLFSPreflight { return "block" } return "warn" } func bytesContainLFSFilter(data []byte) bool { return strings.Contains(strings.ToLower(string(data)), "filter=lfs") || strings.Contains(strings.ToLower(string(data)), "filter = lfs") } func likelyAttachmentURLCount(value string) int { lower := strings.ToLower(value) count := 0 for _, marker := range []string{"user-images.githubusercontent.com", "github.com/user-attachments", "private-user-images.githubusercontent.com"} { count += strings.Count(lower, marker) } return count } func preflightAttachmentURLs(ctx context.Context, client *githubClient, endpoint string) (int, int, error) { payloads, err := client.listRaw(ctx, endpoint, nil) if err != nil { return 0, 0, err } count := 0 for _, raw := range payloads { var item struct { Body *string `json:"body"` } if json.Unmarshal(raw, &item) == nil && item.Body != nil { count += likelyAttachmentURLCount(*item.Body) } } return count, len(payloads), nil } func mapStringValues(key, value string) url.Values { return url.Values{key: []string{value}} } func jsonArrayOrWrappedCount(raw []byte, key string) int { var array []json.RawMessage if json.Unmarshal(raw, &array) == nil { return len(array) } var wrapped map[string]json.RawMessage if json.Unmarshal(raw, &wrapped) != nil { return 0 } for _, candidate := range []string{key, "workflows", "secrets"} { if value, ok := wrapped[candidate]; ok && json.Unmarshal(value, &array) == nil { return len(array) } } return 0 }