package service

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"net/url"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/pingcap/agent-git-service/internal/db"
	"github.com/pingcap/agent-git-service/internal/randutil"

	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

const maxImportBatchItems = 10000
const maxBulkImportBatches = 100
const maxImportApplyBatchLimit = 100
const importBatchApplyClaimTTL = 30 * time.Minute

var errImportBatchClaimLost = errors.New("import batch apply claim lost")

var supportedImportBatchWriters = []string{
	"label",
	"labels",
	"milestone",
	"milestones",
	"issue",
	"issues",
	"issue_comment",
	"issue_comments",
	"pull_request",
	"pull_requests",
	"pull_request_stack",
	"pull_request_stacks",
	"pull_request_review",
	"pull_request_reviews",
	"pull_request_review_comment",
	"pull_request_review_comments",
	"release",
	"releases",
	"release_asset",
	"release_assets",
	"branch_protection",
	"branch_protections",
	"webhook",
	"webhooks",
	"workflow",
	"workflows",
	"secret_name",
	"secret_names",
	"access_handoff",
	"access_handoffs",
}

type ImportBatchItemInput struct {
	SourceID     string
	SourceNodeID string
	SourceNumber *int
	SourceURL    string
	PayloadJSON  string
}

type CreateImportBatchInput struct {
	ImportJobID   uint
	BatchID       string
	ObjectType    string
	SchemaVersion int
	SourceCursor  string
	Checksum      string
	Items         []ImportBatchItemInput
}

type ApplyImportBatchesOptions struct {
	BatchLimit int
	SkipVerify bool
}

type ImportApplyReport struct {
	ImportID                  uint           `json:"import_id"`
	BatchLimit                int            `json:"batch_limit,omitempty"`
	AppliedBatches            int            `json:"applied_batches"`
	AppliedItems              int            `json:"applied_items"`
	SkippedBatches            int            `json:"skipped_batches"`
	RemainingBatches          int            `json:"remaining_batches"`
	RemainingSupportedBatches int            `json:"remaining_supported_batches"`
	Unsupported               map[string]int `json:"unsupported"`
	NativeApplied             map[string]int `json:"native_applied"`
	PendingByType             map[string]int `json:"pending_by_type"`
	FailedByType              map[string]int `json:"failed_by_type"`
	SupportedWriter           []string       `json:"supported_writers"`
}

type ImportVerifyReport struct {
	ImportID      uint           `json:"import_id"`
	BatchCount    int64          `json:"batch_count"`
	StagingCount  int64          `json:"staging_count"`
	MappingCount  int64          `json:"mapping_count"`
	ByObjectType  map[string]any `json:"by_object_type"`
	PendingWrites map[string]int `json:"pending_writes"`
	FailedWrites  map[string]int `json:"failed_writes"`
}

type importCountRow struct {
	ObjectType string
	Status     string
	Count      int
}

func (s *Service) CreateImportBatch(ctx context.Context, in CreateImportBatchInput) (db.ImportBatch, error) {
	if err := prepareImportBatchInput(&in); err != nil {
		return db.ImportBatch{}, err
	}

	var out db.ImportBatch
	err := s.DBForCtx(ctx).Transaction(func(tx *gorm.DB) error {
		if err := lockImportJobForBatchTx(tx, in.ImportJobID); err != nil {
			return err
		}
		if err := markStagedImportJobRunningTx(tx, in.ImportJobID); err != nil {
			return err
		}
		batch, err := createImportBatchTx(tx, in, true)
		if err != nil {
			return err
		}
		out = batch
		return nil
	})
	if err != nil {
		return db.ImportBatch{}, wrapErr(err)
	}
	return out, nil
}

func (s *Service) CreateImportBatches(ctx context.Context, inputs []CreateImportBatchInput) ([]db.ImportBatch, error) {
	if len(inputs) == 0 {
		return nil, fmt.Errorf("%w: at least one import batch is required", ErrValidation)
	}
	if len(inputs) > maxBulkImportBatches {
		return nil, fmt.Errorf("%w: bulk import batch count exceeds %d", ErrValidation, maxBulkImportBatches)
	}
	prepared := make([]CreateImportBatchInput, len(inputs))
	jobID := inputs[0].ImportJobID
	for i, in := range inputs {
		if in.ImportJobID != jobID {
			return nil, fmt.Errorf("%w: bulk import batches must use one import job", ErrValidation)
		}
		if err := prepareImportBatchInput(&in); err != nil {
			return nil, err
		}
		prepared[i] = in
	}

	out := make([]db.ImportBatch, 0, len(prepared))
	err := s.DBForCtx(ctx).Transaction(func(tx *gorm.DB) error {
		if err := lockImportJobForBatchTx(tx, jobID); err != nil {
			return err
		}
		if err := markStagedImportJobRunningTx(tx, jobID); err != nil {
			return err
		}
		for _, in := range prepared {
			batch, err := createImportBatchTx(tx, in, false)
			if err != nil {
				return err
			}
			out = append(out, batch)
		}
		return createImportLogTx(tx, jobID, "info", ImportStageQueued, "import batches staged", map[string]any{
			"batch_count": len(prepared),
			"item_count":  totalImportBatchItems(prepared),
		})
	})
	if err != nil {
		return nil, wrapErr(err)
	}
	return out, nil
}

func (s *Service) ListImportBatches(ctx context.Context, jobID uint, limit int) ([]db.ImportBatch, error) {
	limit = clampImportLimit(limit)
	var batches []db.ImportBatch
	if err := s.DBForCtx(ctx).
		Where("import_job_id = ?", jobID).
		Order("id asc").
		Limit(limit).
		Find(&batches).Error; err != nil {
		return nil, wrapErr(err)
	}
	return batches, nil
}

func (s *Service) ApplyImportBatches(ctx context.Context, jobID uint) (ImportApplyReport, error) {
	return s.ApplyImportBatchesWithOptions(ctx, jobID, ApplyImportBatchesOptions{})
}

func (s *Service) ApplyImportBatchesWithOptions(ctx context.Context, jobID uint, opts ApplyImportBatchesOptions) (ImportApplyReport, error) {
	batchLimit := normalizeImportApplyBatchLimit(opts.BatchLimit)
	report := ImportApplyReport{
		ImportID:        jobID,
		BatchLimit:      batchLimit,
		Unsupported:     map[string]int{},
		NativeApplied:   map[string]int{},
		PendingByType:   map[string]int{},
		FailedByType:    map[string]int{},
		SupportedWriter: append([]string(nil), supportedImportBatchWriters...),
	}

	database := s.DBForCtx(ctx)
	var job db.ImportJob
	if err := database.Take(&job, "id = ?", jobID).Error; err != nil {
		return report, wrapErr(err)
	}
	if job.Status == ImportStatusCanceled {
		return report, fmt.Errorf("%w: import job is canceled", ErrInvalidState)
	}

	batches, err := s.claimImportBatches(ctx, jobID, batchLimit)
	if err != nil {
		return report, wrapErr(err)
	}
	for index, batch := range batches {
		if err := ctx.Err(); err != nil {
			_ = s.releaseImportBatchClaims(ctx, batches[index:])
			return report, err
		}
		count, supported, err := s.applyOneImportBatch(ctx, job, batch)
		if err != nil {
			if errors.Is(err, errImportBatchClaimLost) {
				continue
			}
			markErr := database.Transaction(func(tx *gorm.DB) error {
				return markImportBatchFailedTx(tx, batch, err)
			})
			if errors.Is(markErr, errImportBatchClaimLost) {
				continue
			}
			if markErr != nil {
				_ = s.releaseImportBatchClaims(ctx, batches[index+1:])
				return report, wrapErr(markErr)
			}
			_ = s.releaseImportBatchClaims(ctx, batches[index+1:])
			return report, wrapErr(err)
		}
		if supported {
			report.NativeApplied[batch.ObjectType] += count
			report.AppliedItems += count
			report.AppliedBatches++
		} else {
			report.Unsupported[batch.ObjectType] += batch.ItemCount
			report.PendingByType[batch.ObjectType] += batch.ItemCount
			report.SkippedBatches++
		}
	}
	remaining, err := countRemainingImportBatches(database, jobID, false)
	if err != nil {
		return report, wrapErr(err)
	}
	report.RemainingBatches = int(remaining)
	remainingSupported, err := countRemainingImportBatches(database, jobID, true)
	if err != nil {
		return report, wrapErr(err)
	}
	report.RemainingSupportedBatches = int(remainingSupported)
	if err := createImportLogTx(database, jobID, "info", ImportStageVerify, "batch apply completed", report); err != nil {
		return report, wrapErr(err)
	}
	// Intermediate plan iterations can skip the relatively expensive summary.
	// The final iteration must still verify before it makes a durable success
	// claim; otherwise no remaining batches alone could mark a broken import as
	// complete.
	shouldVerify := !opts.SkipVerify || (report.RemainingBatches == 0 && report.AppliedBatches+report.SkippedBatches > 0)
	if shouldVerify {
		verify, err := s.VerifyImportJob(ctx, jobID)
		if err != nil {
			return report, wrapErr(err)
		}
		for objectType, count := range verify.PendingWrites {
			report.PendingByType[objectType] = count
		}
		for objectType, count := range verify.FailedWrites {
			report.FailedByType[objectType] = count
		}
	}
	if report.RemainingBatches == 0 && report.AppliedBatches+report.SkippedBatches > 0 && len(report.FailedByType) == 0 && len(report.PendingByType) == 0 {
		if err := s.completeStagedImportJob(ctx, jobID); err != nil {
			return report, err
		}
	}
	return report, nil
}

func markStagedImportJobRunningTx(tx *gorm.DB, jobID uint) error {
	result := tx.Model(&db.ImportJob{}).
		Where("id = ? AND status = ?", jobID, ImportStatusQueued).
		Updates(map[string]any{"status": ImportStatusRunning, "stage": ImportStageBaseData})
	if result.Error != nil {
		return result.Error
	}
	if result.RowsAffected > 0 {
		return createImportLogTx(tx, jobID, "info", ImportStageBaseData, "staged import batches are being uploaded", nil)
	}
	return nil
}

func (s *Service) completeStagedImportJob(ctx context.Context, jobID uint) error {
	now := time.Now().UTC()
	return s.DBForCtx(ctx).Transaction(func(tx *gorm.DB) error {
		var job db.ImportJob
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Take(&job, "id = ?", jobID).Error; err != nil {
			return wrapErr(err)
		}
		if job.Status == ImportStatusCanceled || job.Status == ImportStatusSucceeded {
			return nil
		}
		var remaining int64
		if err := tx.Model(&db.ImportBatch{}).Where("import_job_id = ? AND status IN ?", jobID, []string{ImportBatchStatusStaged, ImportBatchStatusApplying, ImportBatchStatusFailed}).Count(&remaining).Error; err != nil {
			return wrapErr(err)
		}
		if remaining != 0 {
			return nil
		}
		if err := tx.Model(&db.ImportJob{}).Where("id = ?", jobID).Updates(map[string]any{
			"status":       ImportStatusSucceeded,
			"stage":        ImportStageVerify,
			"completed_at": &now,
		}).Error; err != nil {
			return err
		}
		if err := releaseImportTargetReservationTx(tx, jobID); err != nil {
			return err
		}
		return createImportLogTx(tx, jobID, "info", ImportStageVerify, "staged import job completed", nil)
	})
}

func normalizeImportApplyBatchLimit(limit int) int {
	if limit <= 0 {
		return 0
	}
	if limit > maxImportApplyBatchLimit {
		return maxImportApplyBatchLimit
	}
	return limit
}

func countRemainingImportBatches(database *gorm.DB, jobID uint, supportedOnly bool) (int64, error) {
	var count int64
	query := database.Model(&db.ImportBatch{}).Where("import_job_id = ? AND status IN ?", jobID, []string{ImportBatchStatusStaged, ImportBatchStatusApplying})
	if supportedOnly {
		query = query.Where("object_type IN ?", supportedImportBatchWriters)
	}
	if err := query.Count(&count).Error; err != nil {
		return 0, err
	}
	return count, nil
}

func (s *Service) applyOneImportBatch(ctx context.Context, job db.ImportJob, batch db.ImportBatch) (int, bool, error) {
	var count int
	supported := true
	err := s.DBForCtx(ctx).Transaction(func(tx *gorm.DB) error {
		var locked db.ImportBatch
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Take(&locked, "id = ? AND status = ? AND apply_claim_token = ?", batch.ID, ImportBatchStatusApplying, batch.ApplyClaimToken).Error; err != nil {
			if errors.Is(err, gorm.ErrRecordNotFound) {
				return errImportBatchClaimLost
			}
			return err
		}
		batch = locked
		var err error
		switch batch.ObjectType {
		case "label", "labels":
			count, err = applyImportLabelsTx(tx, job, batch)
		case "milestone", "milestones":
			count, err = applyImportMilestonesTx(tx, job, batch)
		case "issue", "issues":
			count, err = s.applyImportIssuesTx(tx, job, batch)
		case "issue_comment", "issue_comments":
			count, err = s.applyImportIssueCommentsTx(tx, job, batch)
		case "pull_request", "pull_requests":
			count, err = s.applyImportPullRequestsTx(tx, job, batch)
		case "pull_request_stack", "pull_request_stacks":
			count, err = s.applyImportPullRequestStacksTx(tx, job, batch)
		case "pull_request_review", "pull_request_reviews":
			count, err = s.applyImportPullRequestReviewsTx(tx, job, batch)
		case "pull_request_review_comment", "pull_request_review_comments":
			count, err = s.applyImportPullRequestReviewCommentsTx(tx, job, batch)
		case "release", "releases":
			count, err = s.applyImportReleasesTx(tx, job, batch)
		case "release_asset", "release_assets":
			count, err = s.applyImportReleaseAssetsTx(ctx, tx, job, batch)
		case "branch_protection", "branch_protections":
			count, err = s.applyImportBranchProtectionsTx(tx, job, batch)
		case "webhook", "webhooks":
			count, err = s.applyImportWebhooksTx(tx, job, batch)
		case "workflow", "workflows":
			count, err = s.applyImportWorkflowsTx(tx, job, batch)
		case "secret_name", "secret_names":
			count, err = s.applyImportSecretNamesTx(tx, job, batch)
		case "access_handoff", "access_handoffs":
			count, err = s.applyImportAccessHandoffsTx(tx, job, batch)
		default:
			supported = false
		}
		return err
	})
	return count, supported, err
}

// claimImportBatches changes a batch from staged to applying before any native
// writes begin. The conditional update is the ownership boundary: another
// worker that observed the same staged row cannot later mark this worker's
// successful batch failed.
func (s *Service) claimImportBatches(ctx context.Context, jobID uint, batchLimit int) ([]db.ImportBatch, error) {
	var claimed []db.ImportBatch
	now := time.Now().UTC()
	err := s.DBForCtx(ctx).Transaction(func(tx *gorm.DB) error {
		if err := tx.Model(&db.ImportBatch{}).
			Where("import_job_id = ? AND status = ? AND apply_claim_expires_at IS NOT NULL AND apply_claim_expires_at <= ?", jobID, ImportBatchStatusApplying, now).
			Updates(map[string]any{
				"status":                 ImportBatchStatusStaged,
				"apply_claim_token":      "",
				"apply_claim_expires_at": nil,
				"updated_at":             now,
			}).Error; err != nil {
			return err
		}

		var candidates []db.ImportBatch
		// Stack batches depend on the pull-request mappings created by the PR
		// writer. Snapshot upload order is intentionally not a correctness
		// requirement, so claim PR batches before Stack batches even when an
		// operator uploaded the Stack shard first.
		query := tx.Where("import_job_id = ? AND status = ? AND object_type IN ?", jobID, ImportBatchStatusStaged, supportedImportBatchWriters).
			Order("CASE WHEN object_type IN ('pull_request', 'pull_requests') THEN 0 WHEN object_type IN ('pull_request_stack', 'pull_request_stacks') THEN 2 ELSE 1 END ASC").
			Order("id asc")
		if batchLimit > 0 {
			query = query.Limit(batchLimit)
		}
		if err := query.Find(&candidates).Error; err != nil {
			return err
		}
		for _, candidate := range candidates {
			claimToken := randutil.Hex(32)
			expiresAt := now.Add(importBatchApplyClaimTTL)
			result := tx.Model(&db.ImportBatch{}).
				Where("id = ? AND status = ?", candidate.ID, ImportBatchStatusStaged).
				Updates(map[string]any{
					"status":                 ImportBatchStatusApplying,
					"apply_claim_token":      claimToken,
					"apply_claim_expires_at": &expiresAt,
					"updated_at":             now,
				})
			if result.Error != nil {
				return result.Error
			}
			if result.RowsAffected != 1 {
				continue
			}
			candidate.Status = ImportBatchStatusApplying
			candidate.ApplyClaimToken = claimToken
			candidate.ApplyClaimExpiresAt = &expiresAt
			claimed = append(claimed, candidate)
		}
		return nil
	})
	return claimed, err
}

// releaseImportBatchClaims returns work that this request did not start to the
// staged queue. Conditional tokens ensure it never releases another worker's
// claim after a lease was recovered or reassigned.
func (s *Service) releaseImportBatchClaims(ctx context.Context, batches []db.ImportBatch) error {
	if len(batches) == 0 {
		return nil
	}
	now := time.Now().UTC()
	return s.DBForCtx(ctx).Transaction(func(tx *gorm.DB) error {
		for _, batch := range batches {
			if strings.TrimSpace(batch.ApplyClaimToken) == "" {
				continue
			}
			result := tx.Model(&db.ImportBatch{}).
				Where("id = ? AND status = ? AND apply_claim_token = ?", batch.ID, ImportBatchStatusApplying, batch.ApplyClaimToken).
				Updates(map[string]any{
					"status":                 ImportBatchStatusStaged,
					"apply_claim_token":      "",
					"apply_claim_expires_at": nil,
					"updated_at":             now,
				})
			if result.Error != nil {
				return result.Error
			}
		}
		return nil
	})
}

func (s *Service) VerifyImportJob(ctx context.Context, jobID uint) (ImportVerifyReport, error) {
	report := ImportVerifyReport{
		ImportID:      jobID,
		ByObjectType:  map[string]any{},
		PendingWrites: map[string]int{},
		FailedWrites:  map[string]int{},
	}
	database := s.DBForCtx(ctx)
	var job db.ImportJob
	if err := database.Select("id").Take(&job, "id = ?", jobID).Error; err != nil {
		return report, wrapErr(err)
	}
	if err := database.Model(&db.ImportBatch{}).Where("import_job_id = ?", jobID).Count(&report.BatchCount).Error; err != nil {
		return report, wrapErr(err)
	}
	if err := database.Model(&db.ImportStagingItem{}).Where("import_job_id = ?", jobID).Count(&report.StagingCount).Error; err != nil {
		return report, wrapErr(err)
	}
	if err := database.Model(&db.ImportItemMapping{}).Where("import_job_id = ?", jobID).Count(&report.MappingCount).Error; err != nil {
		return report, wrapErr(err)
	}

	var rows []importCountRow
	if err := database.Model(&db.ImportStagingItem{}).
		Select("object_type, status, count(*) as count").
		Where("import_job_id = ?", jobID).
		Group("object_type, status").
		Find(&rows).Error; err != nil {
		return report, wrapErr(err)
	}
	for _, row := range rows {
		byStatus, _ := report.ByObjectType[row.ObjectType].(map[string]int)
		if byStatus == nil {
			byStatus = map[string]int{}
			report.ByObjectType[row.ObjectType] = byStatus
		}
		byStatus[row.Status] = row.Count
		switch row.Status {
		case ImportStagingStatusStaged:
			report.PendingWrites[row.ObjectType] += row.Count
		case ImportStagingStatusFailed:
			report.FailedWrites[row.ObjectType] += row.Count
		}
	}
	return report, nil
}

func normalizeImportBatchInput(in *CreateImportBatchInput) {
	in.BatchID = strings.TrimSpace(in.BatchID)
	in.ObjectType = strings.ToLower(strings.TrimSpace(in.ObjectType))
	in.SourceCursor = strings.TrimSpace(in.SourceCursor)
	in.Checksum = strings.TrimSpace(in.Checksum)
	if in.SchemaVersion <= 0 {
		in.SchemaVersion = 1
	}
	for i := range in.Items {
		in.Items[i].SourceID = strings.TrimSpace(in.Items[i].SourceID)
		in.Items[i].SourceNodeID = strings.TrimSpace(in.Items[i].SourceNodeID)
		in.Items[i].SourceURL = strings.TrimSpace(in.Items[i].SourceURL)
		in.Items[i].PayloadJSON = strings.TrimSpace(in.Items[i].PayloadJSON)
	}
}

func prepareImportBatchInput(in *CreateImportBatchInput) error {
	normalizeImportBatchInput(in)
	if err := validateImportBatchInput(*in); err != nil {
		return err
	}
	computedChecksum := checksumImportBatchItems(in.Items)
	if in.Checksum != "" && !checksumEqual(in.Checksum, computedChecksum) {
		return fmt.Errorf("%w: batch checksum mismatch", ErrValidation)
	}
	if in.Checksum == "" {
		in.Checksum = computedChecksum
	}
	return nil
}

func validateImportBatchInput(in CreateImportBatchInput) error {
	switch {
	case in.ImportJobID == 0:
		return fmt.Errorf("%w: import_id is required", ErrValidation)
	case in.BatchID == "":
		return fmt.Errorf("%w: batch_id is required", ErrValidation)
	case in.ObjectType == "":
		return fmt.Errorf("%w: object_type is required", ErrValidation)
	case len(in.Items) == 0:
		return fmt.Errorf("%w: items are required", ErrValidation)
	case len(in.Items) > maxImportBatchItems:
		return fmt.Errorf("%w: batch contains %d items, max %d", ErrValidation, len(in.Items), maxImportBatchItems)
	}
	for i, item := range in.Items {
		if item.PayloadJSON == "" {
			return fmt.Errorf("%w: item %d payload is required", ErrValidation, i)
		}
		if !json.Valid([]byte(item.PayloadJSON)) {
			return fmt.Errorf("%w: item %d payload is invalid JSON", ErrValidation, i)
		}
	}
	return nil
}

func lockImportJobForBatchTx(tx *gorm.DB, jobID uint) error {
	var job db.ImportJob
	if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Take(&job, "id = ?", jobID).Error; err != nil {
		return err
	}
	if isTerminalImportStatus(job.Status) {
		return fmt.Errorf("%w: import job is already %s", ErrInvalidState, job.Status)
	}
	return nil
}

func createImportBatchTx(tx *gorm.DB, in CreateImportBatchInput, createBatchLog bool) (db.ImportBatch, error) {
	var existing db.ImportBatch
	err := tx.Take(&existing, "import_job_id = ? AND batch_id = ?", in.ImportJobID, in.BatchID).Error
	if err == nil {
		if !existingImportBatchMatches(existing, in) {
			return db.ImportBatch{}, fmt.Errorf("%w: import batch %q already exists with different metadata", ErrConflict, in.BatchID)
		}
		return existing, nil
	}
	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
		return db.ImportBatch{}, err
	}

	batch := db.ImportBatch{
		ImportJobID:   in.ImportJobID,
		BatchID:       in.BatchID,
		ObjectType:    in.ObjectType,
		SchemaVersion: in.SchemaVersion,
		SourceCursor:  in.SourceCursor,
		ItemCount:     len(in.Items),
		Checksum:      in.Checksum,
		Status:        ImportBatchStatusStaged,
	}
	if err := tx.Create(&batch).Error; err != nil {
		return db.ImportBatch{}, err
	}

	items := make([]db.ImportStagingItem, 0, len(in.Items))
	for idx, item := range in.Items {
		items = append(items, db.ImportStagingItem{
			ImportJobID:   in.ImportJobID,
			ImportBatchID: batch.ID,
			ObjectType:    in.ObjectType,
			ItemIndex:     idx,
			SourceID:      item.SourceID,
			SourceNodeID:  item.SourceNodeID,
			SourceNumber:  item.SourceNumber,
			SourceURL:     item.SourceURL,
			PayloadJSON:   db.LargeText(item.PayloadJSON),
			Checksum:      checksumString(item.PayloadJSON),
			Status:        ImportStagingStatusStaged,
		})
	}
	if len(items) > 0 {
		if err := tx.CreateInBatches(items, 1000).Error; err != nil {
			return db.ImportBatch{}, err
		}
	}
	if createBatchLog {
		if err := createImportLogTx(tx, in.ImportJobID, "info", ImportStageQueued, "import batch staged", map[string]any{
			"batch_id":    in.BatchID,
			"object_type": in.ObjectType,
			"item_count":  len(in.Items),
			"checksum":    in.Checksum,
		}); err != nil {
			return db.ImportBatch{}, err
		}
	}
	return batch, nil
}

func existingImportBatchMatches(existing db.ImportBatch, in CreateImportBatchInput) bool {
	return existing.ObjectType == in.ObjectType &&
		existing.SchemaVersion == in.SchemaVersion &&
		existing.ItemCount == len(in.Items) &&
		checksumEqual(existing.Checksum, in.Checksum)
}

func totalImportBatchItems(inputs []CreateImportBatchInput) int {
	var total int
	for _, in := range inputs {
		total += len(in.Items)
	}
	return total
}

func checksumImportBatchItems(items []ImportBatchItemInput) string {
	h := sha256.New()
	for _, item := range items {
		h.Write([]byte(item.PayloadJSON))
		h.Write([]byte{'\n'})
	}
	return "sha256:" + hex.EncodeToString(h.Sum(nil))
}

func checksumString(value string) string {
	sum := sha256.Sum256([]byte(value))
	return "sha256:" + hex.EncodeToString(sum[:])
}

func checksumEqual(left, right string) bool {
	return strings.EqualFold(strings.TrimSpace(left), strings.TrimSpace(right))
}

func markImportBatchFailedTx(tx *gorm.DB, batch db.ImportBatch, cause error) error {
	message := cause.Error()
	now := time.Now().UTC()
	result := tx.Model(&db.ImportBatch{}).Where("id = ? AND status = ? AND apply_claim_token = ?", batch.ID, ImportBatchStatusApplying, batch.ApplyClaimToken).Updates(map[string]any{
		"status":                 ImportBatchStatusFailed,
		"error_message":          db.LargeText(message),
		"apply_claim_token":      "",
		"apply_claim_expires_at": nil,
		"updated_at":             now,
	})
	if result.Error != nil {
		return result.Error
	}
	if result.RowsAffected != 1 {
		return errImportBatchClaimLost
	}
	if err := tx.Model(&db.ImportStagingItem{}).Where("import_batch_id = ?", batch.ID).Updates(map[string]any{
		"status":        ImportStagingStatusFailed,
		"error_message": db.LargeText(message),
		"updated_at":    now,
	}).Error; err != nil {
		return err
	}
	return nil
}

func markImportBatchAppliedTx(tx *gorm.DB, batch db.ImportBatch) error {
	now := time.Now().UTC()
	result := tx.Model(&db.ImportBatch{}).Where("id = ? AND status = ? AND apply_claim_token = ?", batch.ID, ImportBatchStatusApplying, batch.ApplyClaimToken).Updates(map[string]any{
		"status":                 ImportBatchStatusApplied,
		"apply_claim_token":      "",
		"apply_claim_expires_at": nil,
		"applied_at":             &now,
		"updated_at":             now,
	})
	if result.Error != nil {
		return result.Error
	}
	if result.RowsAffected != 1 {
		return errImportBatchClaimLost
	}
	return tx.Model(&db.ImportStagingItem{}).Where("import_batch_id = ?", batch.ID).Updates(map[string]any{
		"status":     ImportStagingStatusApplied,
		"applied_at": &now,
		"updated_at": now,
	}).Error
}

type importLabelPayload struct {
	ID          int64  `json:"id"`
	NodeID      string `json:"node_id"`
	Name        string `json:"name"`
	Color       string `json:"color"`
	Description string `json:"description"`
	Default     bool   `json:"default"`
	URL         string `json:"url"`
	HTMLURL     string `json:"html_url"`
}

func applyImportLabelsTx(tx *gorm.DB, job db.ImportJob, batch db.ImportBatch) (int, error) {
	targetRepo, err := importTargetRepoTx(tx, job)
	if err != nil {
		return 0, err
	}
	var items []db.ImportStagingItem
	if err := tx.Where("import_batch_id = ? AND status = ?", batch.ID, ImportStagingStatusStaged).Order("item_index asc").Find(&items).Error; err != nil {
		return 0, err
	}
	for _, item := range items {
		var payload importLabelPayload
		if err := json.Unmarshal([]byte(item.PayloadJSON), &payload); err != nil {
			return 0, fmt.Errorf("%w: decode label payload: %v", ErrValidation, err)
		}
		name := strings.TrimSpace(payload.Name)
		if name == "" {
			name = item.SourceID
		}
		if name == "" {
			return 0, fmt.Errorf("%w: label name is required", ErrValidation)
		}
		label := db.Label{
			RepositoryID: targetRepo.ID,
			Name:         name,
			Color:        normalizeHexColor(payload.Color),
			Description:  payload.Description,
			Default:      payload.Default,
		}
		if err := tx.Clauses(clause.OnConflict{
			Columns: []clause.Column{{Name: "repository_id"}, {Name: "name"}},
			DoUpdates: clause.AssignmentColumns([]string{
				"color",
				"description",
				"default",
				"updated_at",
			}),
		}).Create(&label).Error; err != nil {
			return 0, err
		}
		if err := tx.Take(&label, "repository_id = ? AND name = ?", targetRepo.ID, name).Error; err != nil {
			return 0, err
		}
		sourceID := importFirstNonEmpty(item.SourceID, strconv.FormatInt(payload.ID, 10), name)
		sourceURL := importFirstNonEmpty(item.SourceURL, payload.HTMLURL, payload.URL)
		if err := upsertImportMappingTx(tx, job.ID, "label", sourceID, importFirstNonEmpty(item.SourceNodeID, payload.NodeID), item.SourceNumber, sourceURL, "label", strconv.FormatUint(uint64(label.ID), 10), nil, "", map[string]any{"name": name}); err != nil {
			return 0, err
		}
	}
	if err := markImportBatchAppliedTx(tx, batch); err != nil {
		return 0, err
	}
	return len(items), nil
}

type importMilestonePayload struct {
	ID          int64      `json:"id"`
	NodeID      string     `json:"node_id"`
	Number      int        `json:"number"`
	Title       string     `json:"title"`
	Description string     `json:"description"`
	State       string     `json:"state"`
	Creator     *ghUser    `json:"creator"`
	URL         string     `json:"url"`
	HTMLURL     string     `json:"html_url"`
	DueOn       *time.Time `json:"due_on"`
	CreatedAt   time.Time  `json:"created_at"`
	UpdatedAt   time.Time  `json:"updated_at"`
	ClosedAt    *time.Time `json:"closed_at"`
}

func applyImportMilestonesTx(tx *gorm.DB, job db.ImportJob, batch db.ImportBatch) (int, error) {
	targetRepo, err := importTargetRepoTx(tx, job)
	if err != nil {
		return 0, err
	}
	var items []db.ImportStagingItem
	if err := tx.Where("import_batch_id = ? AND status = ?", batch.ID, ImportStagingStatusStaged).Order("item_index asc").Find(&items).Error; err != nil {
		return 0, err
	}
	for _, item := range items {
		var payload importMilestonePayload
		if err := json.Unmarshal([]byte(item.PayloadJSON), &payload); err != nil {
			return 0, fmt.Errorf("%w: decode milestone payload: %v", ErrValidation, err)
		}
		number := payload.Number
		if number == 0 && item.SourceNumber != nil {
			number = *item.SourceNumber
		}
		if number == 0 {
			return 0, fmt.Errorf("%w: milestone number is required", ErrValidation)
		}
		title := strings.TrimSpace(payload.Title)
		if title == "" {
			title = fmt.Sprintf("Milestone %d", number)
		}
		state := normalizeImportState(payload.State)
		if state == "" {
			state = "open"
		}
		creator, err := (&githubImportRunner{job: job}).ensureSourceUserTx(tx, payload.Creator)
		if err != nil {
			return 0, err
		}
		milestone := db.Milestone{
			RepositoryID: targetRepo.ID,
			Number:       number,
			Title:        title,
			Description:  payload.Description,
			State:        state,
			CreatorID:    creator.ID,
			DueOn:        payload.DueOn,
			CreatedAt:    nonZeroTime(payload.CreatedAt, time.Now().UTC()),
			UpdatedAt:    nonZeroTime(payload.UpdatedAt, time.Now().UTC()),
			ClosedAt:     payload.ClosedAt,
		}
		if err := tx.Clauses(clause.OnConflict{
			Columns: []clause.Column{{Name: "repository_id"}, {Name: "number"}},
			DoUpdates: clause.AssignmentColumns([]string{
				"title",
				"description",
				"state",
				"creator_id",
				"due_on",
				"updated_at",
				"closed_at",
			}),
		}).Create(&milestone).Error; err != nil {
			return 0, err
		}
		if err := tx.Take(&milestone, "repository_id = ? AND number = ?", targetRepo.ID, number).Error; err != nil {
			return 0, err
		}
		sourceID := importFirstNonEmpty(item.SourceID, strconv.FormatInt(payload.ID, 10), strconv.Itoa(number))
		sourceURL := importFirstNonEmpty(item.SourceURL, payload.HTMLURL, payload.URL)
		if err := upsertImportMappingTx(tx, job.ID, "milestone", sourceID, importFirstNonEmpty(item.SourceNodeID, payload.NodeID), &number, sourceURL, "milestone", strconv.FormatUint(uint64(milestone.ID), 10), &number, "", map[string]any{"title": title}); err != nil {
			return 0, err
		}
	}
	if err := markImportBatchAppliedTx(tx, batch); err != nil {
		return 0, err
	}
	return len(items), nil
}

func (s *Service) applyImportIssuesTx(tx *gorm.DB, job db.ImportJob, batch db.ImportBatch) (int, error) {
	target, err := importTargetRepoTx(tx, job)
	if err != nil {
		return 0, err
	}
	runner := importBatchRunner(s, job)
	items, err := stagedImportItemsTx(tx, batch)
	if err != nil {
		return 0, err
	}
	parsed, err := parseIssueBatchItems(items)
	if err != nil {
		return 0, err
	}
	if len(parsed) == 0 {
		if err := markImportBatchAppliedTx(tx, batch); err != nil {
			return 0, err
		}
		return 0, nil
	}
	users, err := runner.ensureSourceUsersTx(tx, issueAuthors(parsed))
	if err != nil {
		return 0, err
	}
	milestoneIDs, err := milestoneIDsByNumberTx(tx, target.ID, issueMilestones(parsed))
	if err != nil {
		return 0, err
	}
	now := time.Now().UTC()
	issueRows := make([]db.Issue, 0, len(parsed))
	maxNumber := 0
	for _, entry := range parsed {
		source := entry.source
		login := importSourceLogin(source.User)
		author, ok := users[login]
		if !ok || author.ID == 0 {
			return 0, fmt.Errorf("%w: source user %q was not created", ErrInvalidState, login)
		}
		var milestoneID *uint
		if source.Milestone != nil && source.Milestone.Number != 0 {
			if id, ok := milestoneIDs[source.Milestone.Number]; ok {
				milestoneID = &id
			}
		}
		title := strings.TrimSpace(source.Title)
		if title == "" {
			title = fmt.Sprintf("Issue %d", entry.number)
		}
		issueRows = append(issueRows, db.Issue{
			Number:           entry.number,
			RepositoryID:     target.ID,
			Title:            title,
			Body:             db.LargeText(valueOrEmpty(source.Body)),
			State:            normalizeImportState(source.State),
			AuthorID:         author.ID,
			AssigneeLogins:   joinUserLogins(source.Assignees),
			MilestoneID:      milestoneID,
			Locked:           source.Locked,
			ActiveLockReason: strings.ToUpper(source.ActiveLockReason),
			StateReason:      normalizeImportStateReason(source.StateReason),
			CreatedAt:        nonZeroTime(source.CreatedAt, now),
			UpdatedAt:        nonZeroTime(source.UpdatedAt, now),
			ClosedAt:         source.ClosedAt,
			ClosedByLogin:    loginOrEmpty(source.ClosedBy),
		})
		if entry.number > maxNumber {
			maxNumber = entry.number
		}
	}
	if err := tx.Clauses(clause.OnConflict{
		Columns: []clause.Column{{Name: "repository_id"}, {Name: "number"}},
		DoUpdates: clause.AssignmentColumns([]string{
			"title", "body", "state", "author_id", "assignee_logins", "milestone_id",
			"locked", "active_lock_reason", "state_reason", "updated_at", "closed_at", "closed_by_login",
		}),
	}).CreateInBatches(&issueRows, 1000).Error; err != nil {
		return 0, err
	}
	issuesByNumber, err := issuesByNumberTx(tx, target.ID, importIssueNumbers(parsed))
	if err != nil {
		return 0, err
	}
	labelsByIssueID := make(map[uint][]ghLabel, len(parsed))
	mappings := make([]db.ImportItemMapping, 0, len(parsed))
	for _, entry := range parsed {
		issue, ok := issuesByNumber[entry.number]
		if !ok {
			return 0, fmt.Errorf("%w: issue #%d was not created", ErrInvalidState, entry.number)
		}
		labelsByIssueID[issue.ID] = entry.source.Labels
		sourceNumber := entry.number
		targetNumber := issue.Number
		mappings = append(mappings, db.ImportItemMapping{
			ImportJobID:  job.ID,
			SourceType:   importTypeIssue,
			SourceID:     entry.sourceID,
			SourceNodeID: entry.sourceNodeID,
			SourceNumber: &sourceNumber,
			SourceURL:    entry.sourceURL,
			TargetType:   "issue",
			TargetID:     strconv.FormatUint(uint64(issue.ID), 10),
			TargetNumber: &targetNumber,
			TargetURL:    runner.targetIssueURL(issue.Number),
			MetadataJSON: "",
		})
	}
	if err := runner.replaceIssueLabelsBulkTx(tx, target.ID, labelsByIssueID); err != nil {
		return 0, err
	}
	if maxNumber > 0 {
		if err := ensureIssuePRCounterAtLeastTx(tx, target.ID, maxNumber+1); err != nil {
			return 0, err
		}
	}
	if err := bulkUpsertImportMappingsTx(tx, mappings); err != nil {
		return 0, err
	}
	if err := markImportBatchAppliedTx(tx, batch); err != nil {
		return 0, err
	}
	return len(parsed), nil
}

type parsedIssueImport struct {
	source       ghIssue
	number       int
	sourceID     string
	sourceNodeID string
	sourceURL    string
}

func parseIssueBatchItems(items []db.ImportStagingItem) ([]parsedIssueImport, error) {
	parsed := make([]parsedIssueImport, 0, len(items))
	seenSourceIDs := map[string]struct{}{}
	seenNumbers := map[int]struct{}{}
	for _, item := range items {
		var source ghIssue
		if err := json.Unmarshal([]byte(item.PayloadJSON), &source); err != nil {
			return nil, fmt.Errorf("%w: decode issue payload: %v", ErrValidation, err)
		}
		if source.PullRequest != nil {
			continue
		}
		number, err := importItemNumber(item, source.Number, "issue")
		if err != nil {
			return nil, err
		}
		source.Number = number
		if _, ok := seenNumbers[number]; ok {
			return nil, fmt.Errorf("%w: duplicate issue number %d in batch", ErrValidation, number)
		}
		seenNumbers[number] = struct{}{}
		sourceID := importItemSourceID(item, source.ID, strconv.Itoa(number))
		if sourceID == "" {
			return nil, fmt.Errorf("%w: issue source_id is required", ErrValidation)
		}
		if _, ok := seenSourceIDs[sourceID]; ok {
			return nil, fmt.Errorf("%w: duplicate issue source_id %q in batch", ErrValidation, sourceID)
		}
		seenSourceIDs[sourceID] = struct{}{}
		parsed = append(parsed, parsedIssueImport{
			source:       source,
			number:       number,
			sourceID:     sourceID,
			sourceNodeID: importFirstNonEmpty(item.SourceNodeID, source.NodeID),
			sourceURL:    importFirstNonEmpty(item.SourceURL, source.HTMLURL),
		})
	}
	return parsed, nil
}

func issueAuthors(entries []parsedIssueImport) []*ghUser {
	out := make([]*ghUser, 0, len(entries)*2)
	for _, entry := range entries {
		out = append(out, entry.source.User)
		if entry.source.ClosedBy != nil {
			out = append(out, entry.source.ClosedBy)
		}
	}
	return out
}

func issueMilestones(entries []parsedIssueImport) []*ghMilestone {
	out := make([]*ghMilestone, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.source.Milestone)
	}
	return out
}

func importIssueNumbers(entries []parsedIssueImport) []int {
	out := make([]int, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.number)
	}
	return out
}

func milestoneIDsByNumberTx(tx *gorm.DB, repoID uint, milestones []*ghMilestone) (map[int]uint, error) {
	numbers := make([]int, 0, len(milestones))
	for _, milestone := range milestones {
		if milestone != nil && milestone.Number > 0 {
			numbers = append(numbers, milestone.Number)
		}
	}
	numbers = uniqueInts(numbers)
	if len(numbers) == 0 {
		return map[int]uint{}, nil
	}
	var rows []db.Milestone
	if err := tx.Select("id", "number").Where("repository_id = ? AND number IN ?", repoID, numbers).Find(&rows).Error; err != nil {
		return nil, err
	}
	out := make(map[int]uint, len(rows))
	for _, row := range rows {
		out[row.Number] = row.ID
	}
	return out, nil
}

func issuesByNumberTx(tx *gorm.DB, repoID uint, numbers []int) (map[int]db.Issue, error) {
	numbers = uniqueInts(numbers)
	if len(numbers) == 0 {
		return map[int]db.Issue{}, nil
	}
	var rows []db.Issue
	if err := tx.Where("repository_id = ? AND number IN ?", repoID, numbers).Find(&rows).Error; err != nil {
		return nil, err
	}
	out := make(map[int]db.Issue, len(rows))
	for _, row := range rows {
		out[row.Number] = row
	}
	return out, nil
}

func pullRequestsByNumberTx(tx *gorm.DB, repoID uint, numbers []int) (map[int]db.PullRequest, error) {
	numbers = uniqueInts(numbers)
	if len(numbers) == 0 {
		return map[int]db.PullRequest{}, nil
	}
	var rows []db.PullRequest
	if err := tx.Where("repository_id = ? AND number IN ?", repoID, numbers).Find(&rows).Error; err != nil {
		return nil, err
	}
	out := make(map[int]db.PullRequest, len(rows))
	for _, row := range rows {
		out[row.Number] = row
	}
	return out, nil
}

func (s *Service) applyImportIssueCommentsTx(tx *gorm.DB, job db.ImportJob, batch db.ImportBatch) (int, error) {
	target, err := importTargetRepoTx(tx, job)
	if err != nil {
		return 0, err
	}
	runner := importBatchRunner(s, job)
	items, err := stagedImportItemsTx(tx, batch)
	if err != nil {
		return 0, err
	}
	parsed, err := parseIssueCommentBatchItems(items)
	if err != nil {
		return 0, err
	}
	authors, err := runner.ensureSourceUsersTx(tx, issueCommentAuthors(parsed))
	if err != nil {
		return 0, err
	}
	existingIDs, err := runner.mappingTargetIDsTx(tx, importTypeIssueComment, issueCommentSourceIDs(parsed))
	if err != nil {
		return 0, err
	}
	now := time.Now().UTC()
	comments := make([]db.IssueComment, 0, len(parsed))
	for _, entry := range parsed {
		login := importSourceLogin(entry.source.User)
		author, ok := authors[login]
		if !ok || author.ID == 0 {
			return 0, fmt.Errorf("%w: source user %q was not created", ErrInvalidState, login)
		}
		comment := db.IssueComment{
			RepositoryID: target.ID,
			IssueNumber:  entry.number,
			Body:         db.LargeText(entry.source.Body),
			AuthorID:     author.ID,
			CreatedAt:    nonZeroTime(entry.source.CreatedAt, now),
			UpdatedAt:    nonZeroTime(entry.source.UpdatedAt, now),
		}
		if existingID, ok := existingIDs[entry.sourceID]; ok {
			comment.ID = existingID
		}
		comments = append(comments, comment)
	}
	existingComments := make([]db.IssueComment, 0, len(comments))
	newComments := make([]db.IssueComment, 0, len(comments))
	newCommentIndexes := make([]int, 0, len(comments))
	for idx, comment := range comments {
		if comment.ID != 0 {
			existingComments = append(existingComments, comment)
			continue
		}
		newComments = append(newComments, comment)
		newCommentIndexes = append(newCommentIndexes, idx)
	}
	if len(existingComments) > 0 {
		if err := tx.Clauses(clause.OnConflict{
			Columns: []clause.Column{{Name: "id"}},
			DoUpdates: clause.AssignmentColumns([]string{
				"repository_id", "issue_number", "body", "author_id", "updated_at",
			}),
		}).CreateInBatches(&existingComments, 1000).Error; err != nil {
			return 0, err
		}
	}
	if len(newComments) > 0 {
		if err := tx.CreateInBatches(&newComments, 1000).Error; err != nil {
			return 0, err
		}
		for idx, comment := range newComments {
			comments[newCommentIndexes[idx]].ID = comment.ID
		}
	}
	mappings := make([]db.ImportItemMapping, 0, len(parsed))
	for idx, entry := range parsed {
		comment := comments[idx]
		if comment.ID == 0 {
			return 0, fmt.Errorf("%w: issue comment target id was not assigned", ErrInvalidState)
		}
		sourceNumber := entry.number
		mappings = append(mappings, db.ImportItemMapping{
			ImportJobID:  job.ID,
			SourceType:   importTypeIssueComment,
			SourceID:     entry.sourceID,
			SourceNodeID: entry.sourceNodeID,
			SourceNumber: &sourceNumber,
			SourceURL:    entry.sourceURL,
			TargetType:   "issue_comment",
			TargetID:     strconv.FormatUint(uint64(comment.ID), 10),
			TargetURL:    runner.targetIssueCommentURL(entry.number, comment.ID),
		})
	}
	if err := bulkUpsertImportMappingsTx(tx, mappings); err != nil {
		return 0, err
	}
	if err := markImportBatchAppliedTx(tx, batch); err != nil {
		return 0, err
	}
	return len(items), nil
}

type parsedIssueCommentImport struct {
	source       ghIssueComment
	number       int
	sourceID     string
	sourceNodeID string
	sourceURL    string
}

func parseIssueCommentBatchItems(items []db.ImportStagingItem) ([]parsedIssueCommentImport, error) {
	parsed := make([]parsedIssueCommentImport, 0, len(items))
	seenSourceIDs := map[string]struct{}{}
	for _, item := range items {
		var source ghIssueComment
		if err := json.Unmarshal([]byte(item.PayloadJSON), &source); err != nil {
			return nil, fmt.Errorf("%w: decode issue comment payload: %v", ErrValidation, err)
		}
		number, err := importItemNumber(item, 0, "issue_comment")
		if err != nil {
			return nil, err
		}
		sourceID := importItemSourceID(item, source.ID, "")
		if sourceID == "" {
			return nil, fmt.Errorf("%w: issue comment source_id is required", ErrValidation)
		}
		if _, ok := seenSourceIDs[sourceID]; ok {
			return nil, fmt.Errorf("%w: duplicate issue comment source_id %q in batch", ErrValidation, sourceID)
		}
		seenSourceIDs[sourceID] = struct{}{}
		parsed = append(parsed, parsedIssueCommentImport{
			source:       source,
			number:       number,
			sourceID:     sourceID,
			sourceNodeID: importFirstNonEmpty(item.SourceNodeID, source.NodeID),
			sourceURL:    importFirstNonEmpty(item.SourceURL, source.HTMLURL),
		})
	}
	return parsed, nil
}

func issueCommentAuthors(entries []parsedIssueCommentImport) []*ghUser {
	out := make([]*ghUser, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.source.User)
	}
	return out
}

func issueCommentSourceIDs(entries []parsedIssueCommentImport) []string {
	out := make([]string, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.sourceID)
	}
	return out
}

func (s *Service) applyImportPullRequestsTx(tx *gorm.DB, job db.ImportJob, batch db.ImportBatch) (int, error) {
	target, err := importTargetRepoTx(tx, job)
	if err != nil {
		return 0, err
	}
	runner := importBatchRunner(s, job)
	items, err := stagedImportItemsTx(tx, batch)
	if err != nil {
		return 0, err
	}
	parsed, err := parsePullRequestBatchItems(items)
	if err != nil {
		return 0, err
	}
	if len(parsed) == 0 {
		if err := markImportBatchAppliedTx(tx, batch); err != nil {
			return 0, err
		}
		return 0, nil
	}
	users, err := runner.ensureSourceUsersTx(tx, pullRequestAuthors(parsed))
	if err != nil {
		return 0, err
	}
	milestoneIDs, err := milestoneIDsByNumberTx(tx, target.ID, pullRequestMilestones(parsed))
	if err != nil {
		return 0, err
	}

	now := time.Now().UTC()
	prRows := make([]db.PullRequest, 0, len(parsed))
	maxNumber := 0
	for _, entry := range parsed {
		source := entry.source
		login := importSourceLogin(source.User)
		author, ok := users[login]
		if !ok || author.ID == 0 {
			return 0, fmt.Errorf("%w: source user %q was not created", ErrInvalidState, login)
		}
		var milestoneID *uint
		if source.Milestone != nil && source.Milestone.Number != 0 {
			if id, ok := milestoneIDs[source.Milestone.Number]; ok {
				milestoneID = &id
			}
		}
		title := strings.TrimSpace(source.Title)
		if title == "" {
			title = fmt.Sprintf("Pull request %d", entry.number)
		}
		pr := db.PullRequest{
			Number:              entry.number,
			RepositoryID:        target.ID,
			HeadRepositoryID:    target.ID,
			Title:               title,
			Body:                db.LargeText(valueOrEmpty(source.Body)),
			State:               normalizeImportState(source.State),
			AuthorID:            author.ID,
			AssigneeLogins:      joinUserLogins(source.Assignees),
			MilestoneID:         milestoneID,
			HeadRef:             source.Head.Ref,
			HeadSHA:             source.Head.SHA,
			BaseRef:             importFirstNonEmpty(source.Base.Ref, target.DefaultBranch),
			BaseSHA:             source.Base.SHA,
			Draft:               source.Draft,
			MaintainerCanModify: source.MaintainerCanModify,
			Merged:              source.Merged,
			MergeCommitSHA:      source.MergeCommitSHA,
			MergedByLogin:       loginOrEmpty(source.MergedBy),
			CreatedAt:           nonZeroTime(source.CreatedAt, now),
			UpdatedAt:           nonZeroTime(source.UpdatedAt, now),
			ClosedAt:            source.ClosedAt,
			MergedAt:            source.MergedAt,
		}
		if source.Merged {
			pr.State = db.StateClosed
		}
		prRows = append(prRows, pr)
		if entry.number > maxNumber {
			maxNumber = entry.number
		}
	}
	if err := tx.Clauses(clause.OnConflict{
		Columns: []clause.Column{{Name: "repository_id"}, {Name: "number"}},
		DoUpdates: clause.AssignmentColumns([]string{
			"head_repository_id", "title", "body", "state", "author_id", "assignee_logins",
			"milestone_id", "head_ref", "head_sha", "base_ref", "base_sha", "draft",
			"maintainer_can_modify", "merged", "merge_commit_sha", "merged_by_login",
			"updated_at", "closed_at", "merged_at",
		}),
	}).CreateInBatches(&prRows, 1000).Error; err != nil {
		return 0, err
	}
	prsByNumber, err := pullRequestsByNumberTx(tx, target.ID, pullRequestNumbers(parsed))
	if err != nil {
		return 0, err
	}
	labelsByPRID := make(map[uint][]ghLabel, len(parsed))
	mappings := make([]db.ImportItemMapping, 0, len(parsed))
	for _, entry := range parsed {
		pr, ok := prsByNumber[entry.number]
		if !ok {
			return 0, fmt.Errorf("%w: pull request #%d was not created", ErrInvalidState, entry.number)
		}
		labelsByPRID[pr.ID] = entry.source.Labels
		sourceNumber := entry.number
		targetNumber := pr.Number
		mapping, err := newImportItemMapping(job.ID, importTypePullRequest, entry.sourceID, entry.sourceNodeID, &sourceNumber, entry.sourceURL, "pull_request", strconv.FormatUint(uint64(pr.ID), 10), &targetNumber, runner.targetPullURL(pr.Number), map[string]any{
			"source_head_repo": entry.source.Head.repoFullName(),
			"source_base_repo": entry.source.Base.repoFullName(),
		})
		if err != nil {
			return 0, err
		}
		mappings = append(mappings, mapping)
	}
	if err := runner.replacePRLabelsBulkTx(tx, target.ID, labelsByPRID); err != nil {
		return 0, err
	}
	if maxNumber > 0 {
		if err := ensureIssuePRCounterAtLeastTx(tx, target.ID, maxNumber+1); err != nil {
			return 0, err
		}
	}
	if err := bulkUpsertImportMappingsTx(tx, mappings); err != nil {
		return 0, err
	}
	if err := markImportBatchAppliedTx(tx, batch); err != nil {
		return 0, err
	}
	return len(parsed), nil
}

type parsedPullRequestImport struct {
	source       ghPullRequest
	number       int
	sourceID     string
	sourceNodeID string
	sourceURL    string
}

func parsePullRequestBatchItems(items []db.ImportStagingItem) ([]parsedPullRequestImport, error) {
	parsed := make([]parsedPullRequestImport, 0, len(items))
	seenSourceIDs := map[string]struct{}{}
	seenNumbers := map[int]struct{}{}
	for _, item := range items {
		var source ghPullRequest
		if err := json.Unmarshal([]byte(item.PayloadJSON), &source); err != nil {
			return nil, fmt.Errorf("%w: decode pull request payload: %v", ErrValidation, err)
		}
		number, err := importItemNumber(item, source.Number, "pull_request")
		if err != nil {
			return nil, err
		}
		source.Number = number
		if _, ok := seenNumbers[number]; ok {
			return nil, fmt.Errorf("%w: duplicate pull request number %d in batch", ErrValidation, number)
		}
		seenNumbers[number] = struct{}{}
		sourceID := importItemSourceID(item, source.ID, strconv.Itoa(number))
		if sourceID == "" {
			return nil, fmt.Errorf("%w: pull request source_id is required", ErrValidation)
		}
		if _, ok := seenSourceIDs[sourceID]; ok {
			return nil, fmt.Errorf("%w: duplicate pull request source_id %q in batch", ErrValidation, sourceID)
		}
		seenSourceIDs[sourceID] = struct{}{}
		parsed = append(parsed, parsedPullRequestImport{
			source:       source,
			number:       number,
			sourceID:     sourceID,
			sourceNodeID: importFirstNonEmpty(item.SourceNodeID, source.NodeID),
			sourceURL:    importFirstNonEmpty(item.SourceURL, source.HTMLURL),
		})
	}
	return parsed, nil
}

func pullRequestAuthors(entries []parsedPullRequestImport) []*ghUser {
	out := make([]*ghUser, 0, len(entries)*2)
	for _, entry := range entries {
		out = append(out, entry.source.User)
		if entry.source.MergedBy != nil {
			out = append(out, entry.source.MergedBy)
		}
	}
	return out
}

func pullRequestMilestones(entries []parsedPullRequestImport) []*ghMilestone {
	out := make([]*ghMilestone, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.source.Milestone)
	}
	return out
}

func pullRequestNumbers(entries []parsedPullRequestImport) []int {
	out := make([]int, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.number)
	}
	return out
}

func (s *Service) applyImportPullRequestStacksTx(tx *gorm.DB, job db.ImportJob, batch db.ImportBatch) (int, error) {
	target, err := importTargetRepoTx(tx, job)
	if err != nil {
		return 0, err
	}
	runner := importBatchRunner(s, job)
	items, err := stagedImportItemsTx(tx, batch)
	if err != nil {
		return 0, err
	}
	parsed, err := parsePullRequestStackBatchItems(items)
	if err != nil {
		return 0, err
	}
	for _, entry := range parsed {
		if _, err := importPullRequestStackTx(tx, job, target, entry.source, runner.targetPullRequestStackURL); err != nil {
			return 0, err
		}
	}
	if err := markImportBatchAppliedTx(tx, batch); err != nil {
		return 0, err
	}
	return len(parsed), nil
}

type parsedPullRequestStackImport struct {
	source ghPullRequestStack
}

func parsePullRequestStackBatchItems(items []db.ImportStagingItem) ([]parsedPullRequestStackImport, error) {
	parsed := make([]parsedPullRequestStackImport, 0, len(items))
	seenSourceIDs := make(map[string]struct{}, len(items))
	for _, item := range items {
		var source ghPullRequestStack
		if err := json.Unmarshal([]byte(item.PayloadJSON), &source); err != nil {
			return nil, fmt.Errorf("%w: decode pull request Stack payload: %v", ErrValidation, err)
		}
		number, err := importItemNumber(item, source.Number, "pull_request_stack")
		if err != nil {
			return nil, err
		}
		source.Number = number
		sourceID := importItemSourceID(item, source.ID, "")
		if sourceID == "" {
			return nil, fmt.Errorf("%w: pull request Stack source_id is required", ErrValidation)
		}
		if _, exists := seenSourceIDs[sourceID]; exists {
			return nil, fmt.Errorf("%w: duplicate pull request Stack source_id %q", ErrValidation, sourceID)
		}
		seenSourceIDs[sourceID] = struct{}{}
		if source.ID == 0 {
			id, err := strconv.ParseInt(sourceID, 10, 64)
			if err != nil || id <= 0 {
				return nil, fmt.Errorf("%w: pull request Stack id is required", ErrValidation)
			}
			source.ID = id
		}
		parsed = append(parsed, parsedPullRequestStackImport{source: source})
	}
	return parsed, nil
}

// importPullRequestStackTx is shared by the batch importer and the older
// server-side GitHub worker. It restores a Stack as historical metadata, so
// merged members are valid: GitHub can return a retained Stack whose older PRs
// are already merged. The public CreatePullRequestStack path rightly rejects
// that live-user operation, but an importer must preserve the source record.
func importPullRequestStackTx(tx *gorm.DB, job db.ImportJob, target db.Repository, source ghPullRequestStack, targetURL func(int) string) (db.PullRequestStack, error) {
	sourceID := sourceIDFromInt64(source.ID)
	if sourceID == "" || source.Number <= 0 {
		return db.PullRequestStack{}, fmt.Errorf("%w: pull request Stack id and number are required", ErrValidation)
	}
	targetPRs, err := importedStackPullRequestsTx(tx, job.ID, target.ID, source.PullRequests)
	if err != nil {
		return db.PullRequestStack{}, err
	}

	var mapping db.ImportItemMapping
	err = tx.First(&mapping, "import_job_id = ? AND source_type = ? AND source_id = ?", job.ID, importTypePullRequestStack, sourceID).Error
	if err == nil {
		stackID, parseErr := strconv.ParseUint(mapping.TargetID, 10, 64)
		if parseErr != nil || stackID == 0 {
			return db.PullRequestStack{}, fmt.Errorf("%w: imported Stack mapping has an invalid target", ErrInvalidState)
		}
		var existing db.PullRequestStack
		if err := tx.First(&existing, "id = ? AND repository_id = ?", uint(stackID), target.ID).Error; err != nil {
			return db.PullRequestStack{}, fmt.Errorf("%w: imported Stack mapping target is missing", ErrInvalidState)
		}
		if err := verifyImportedPullRequestStackTx(tx, existing, targetPRs); err != nil {
			return db.PullRequestStack{}, err
		}
		return existing, nil
	}
	if !errors.Is(err, gorm.ErrRecordNotFound) {
		return db.PullRequestStack{}, err
	}

	if err := validateImportedPullRequestStackTx(tx, target, source, targetPRs); err != nil {
		return db.PullRequestStack{}, err
	}
	if err := lockRepoForNumbering(tx, target.ID); err != nil {
		return db.PullRequestStack{}, err
	}
	targetNumber, err := nextPullRequestStackNumberTx(tx, target.ID)
	if err != nil {
		return db.PullRequestStack{}, err
	}
	baseRef := importFirstNonEmpty(source.Base.Ref, targetPRs[0].BaseRef)
	stack := db.PullRequestStack{
		RepositoryID: target.ID,
		Number:       targetNumber,
		BaseRef:      baseRef,
		BaseSHA:      targetPRs[0].BaseSHA,
		CreatedAt:    nonZeroTime(source.CreatedAt, time.Now().UTC()),
		UpdatedAt:    nonZeroTime(source.CreatedAt, time.Now().UTC()),
	}
	if err := tx.Create(&stack).Error; err != nil {
		return db.PullRequestStack{}, err
	}
	entries := make([]db.PullRequestStackEntry, len(targetPRs))
	for i := range targetPRs {
		entries[i] = db.PullRequestStackEntry{StackID: stack.ID, PullRequestID: targetPRs[i].ID, Position: i + 1}
	}
	if err := tx.CreateInBatches(&entries, 1000).Error; err != nil {
		return db.PullRequestStack{}, err
	}
	sourceNumber := source.Number
	mapping, err = newImportItemMapping(job.ID, importTypePullRequestStack, sourceID, source.NodeID, &sourceNumber, source.URL, "pull_request_stack", strconv.FormatUint(uint64(stack.ID), 10), &stack.Number, targetURL(stack.Number), map[string]any{
		"source_stack_number":         source.Number,
		"source_pull_request_numbers": stackSourcePullRequestNumbers(source.PullRequests),
		"source_base_ref":             source.Base.Ref,
		"source_open":                 source.Open,
	})
	if err != nil {
		return db.PullRequestStack{}, err
	}
	if err := bulkUpsertImportMappingsTx(tx, []db.ImportItemMapping{mapping}); err != nil {
		return db.PullRequestStack{}, err
	}
	return stack, nil
}

func importedStackPullRequestsTx(tx *gorm.DB, jobID, repoID uint, members []ghPullRequestStackPR) ([]db.PullRequest, error) {
	if len(members) < MinPullRequestStackSize || len(members) > MaxPullRequestStackSize {
		return nil, fmt.Errorf("%w: pull request Stack must contain between %d and %d pull requests", ErrValidation, MinPullRequestStackSize, MaxPullRequestStackSize)
	}
	numbers := make([]int, 0, len(members))
	seen := make(map[int]struct{}, len(members))
	for _, member := range members {
		if member.Number <= 0 {
			return nil, fmt.Errorf("%w: Stack pull request number is required", ErrValidation)
		}
		if _, exists := seen[member.Number]; exists {
			return nil, fmt.Errorf("%w: duplicate Stack pull request #%d", ErrValidation, member.Number)
		}
		seen[member.Number] = struct{}{}
		numbers = append(numbers, member.Number)
	}
	var mappings []db.ImportItemMapping
	if err := tx.Where("import_job_id = ? AND source_type = ? AND source_number IN ?", jobID, importTypePullRequest, numbers).Find(&mappings).Error; err != nil {
		return nil, err
	}
	bySourceNumber := make(map[int]db.ImportItemMapping, len(mappings))
	for _, mapping := range mappings {
		if mapping.SourceNumber != nil {
			bySourceNumber[*mapping.SourceNumber] = mapping
		}
	}
	targetIDs := make([]uint, 0, len(numbers))
	for _, number := range numbers {
		mapping, ok := bySourceNumber[number]
		if !ok {
			return nil, fmt.Errorf("%w: pull request #%d must be imported before its Stack", ErrInvalidState, number)
		}
		id, err := strconv.ParseUint(mapping.TargetID, 10, 64)
		if err != nil || id == 0 {
			return nil, fmt.Errorf("%w: pull request #%d has an invalid import mapping", ErrInvalidState, number)
		}
		targetIDs = append(targetIDs, uint(id))
	}
	var rows []db.PullRequest
	if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("repository_id = ? AND id IN ?", repoID, targetIDs).Find(&rows).Error; err != nil {
		return nil, err
	}
	byID := make(map[uint]db.PullRequest, len(rows))
	for _, row := range rows {
		byID[row.ID] = row
	}
	ordered := make([]db.PullRequest, 0, len(targetIDs))
	for i, id := range targetIDs {
		pr, ok := byID[id]
		if !ok {
			return nil, fmt.Errorf("%w: mapped pull request #%d is missing from the target repository", ErrInvalidState, numbers[i])
		}
		ordered = append(ordered, pr)
	}
	return ordered, nil
}

func validateImportedPullRequestStackTx(tx *gorm.DB, target db.Repository, source ghPullRequestStack, prs []db.PullRequest) error {
	if len(prs) < MinPullRequestStackSize {
		return fmt.Errorf("%w: pull request Stack must contain at least %d pull requests", ErrValidation, MinPullRequestStackSize)
	}
	ids := make([]uint, len(prs))
	expectedBaseRef := source.Base.Ref
	for i, pr := range prs {
		if pr.RepositoryID != target.ID {
			return fmt.Errorf("%w: Stack pull request is outside the target repository", ErrInvalidState)
		}
		if expectedBaseRef != "" && pr.BaseRef != expectedBaseRef {
			return fmt.Errorf("%w: Stack pull request #%d targets %q, want %q", ErrValidation, pr.Number, pr.BaseRef, expectedBaseRef)
		}
		ids[i] = pr.ID
		// GitHub retargets the next open pull request to the Stack root when a
		// lower member has already merged. Preserve the source member order
		// instead of requiring a branch that no longer exists.
		if pr.Merged {
			expectedBaseRef = source.Base.Ref
		} else {
			expectedBaseRef = pr.HeadRef
		}
	}
	var memberships int64
	if err := tx.Model(&db.PullRequestStackEntry{}).Where("pull_request_id IN ?", ids).Count(&memberships).Error; err != nil {
		return err
	}
	if memberships != 0 {
		return fmt.Errorf("%w: a Stack pull request already belongs to another Stack", ErrConflict)
	}
	return nil
}

func verifyImportedPullRequestStackTx(tx *gorm.DB, stack db.PullRequestStack, expected []db.PullRequest) error {
	var entries []db.PullRequestStackEntry
	if err := tx.Where("stack_id = ?", stack.ID).Order("position ASC").Find(&entries).Error; err != nil {
		return err
	}
	if len(entries) != len(expected) {
		return fmt.Errorf("%w: imported Stack mapping does not match source member count", ErrConflict)
	}
	for i, entry := range entries {
		if entry.Position != i+1 || entry.PullRequestID != expected[i].ID {
			return fmt.Errorf("%w: imported Stack mapping does not match source member order", ErrConflict)
		}
	}
	return nil
}

func stackSourcePullRequestNumbers(members []ghPullRequestStackPR) []int {
	numbers := make([]int, 0, len(members))
	for _, member := range members {
		numbers = append(numbers, member.Number)
	}
	return numbers
}

func (s *Service) applyImportPullRequestReviewsTx(tx *gorm.DB, job db.ImportJob, batch db.ImportBatch) (int, error) {
	target, err := importTargetRepoTx(tx, job)
	if err != nil {
		return 0, err
	}
	runner := importBatchRunner(s, job)
	items, err := stagedImportItemsTx(tx, batch)
	if err != nil {
		return 0, err
	}
	parsed, err := parsePullRequestReviewBatchItems(items)
	if err != nil {
		return 0, err
	}
	if len(parsed) == 0 {
		if err := markImportBatchAppliedTx(tx, batch); err != nil {
			return 0, err
		}
		return 0, nil
	}
	prsByNumber, err := pullRequestsByNumberTx(tx, target.ID, pullRequestReviewNumbers(parsed))
	if err != nil {
		return 0, err
	}
	if _, err := runner.ensureSourceUsersTx(tx, pullRequestReviewAuthors(parsed)); err != nil {
		return 0, err
	}
	existingIDs, err := runner.mappingTargetIDsTx(tx, importTypePullReview, pullRequestReviewSourceIDs(parsed))
	if err != nil {
		return 0, err
	}

	now := time.Now().UTC()
	reviews := make([]db.PullRequestReview, 0, len(parsed))
	for _, entry := range parsed {
		pr, ok := prsByNumber[entry.number]
		if !ok {
			return 0, fmt.Errorf("%w: pull request #%d must be imported before reviews/comments", ErrInvalidState, entry.number)
		}
		createdAt := entry.source.CreatedAt
		if entry.source.SubmittedAt != nil {
			createdAt = *entry.source.SubmittedAt
		}
		review := db.PullRequestReview{
			PullRequestID: pr.ID,
			AuthorLogin:   loginOrEmpty(entry.source.User),
			State:         normalizeImportReviewState(entry.source.State),
			Body:          db.LargeText(entry.source.Body),
			CommitSHA:     entry.source.CommitID,
			SubmittedAt:   entry.source.SubmittedAt,
			CreatedAt:     nonZeroTime(createdAt, now),
			UpdatedAt:     nonZeroTime(entry.source.UpdatedAt, now),
		}
		if existingID, ok := existingIDs[entry.sourceID]; ok {
			review.ID = existingID
		}
		reviews = append(reviews, review)
	}
	existingReviews := make([]db.PullRequestReview, 0, len(reviews))
	newReviews := make([]db.PullRequestReview, 0, len(reviews))
	newReviewIndexes := make([]int, 0, len(reviews))
	for idx, review := range reviews {
		if review.ID != 0 {
			existingReviews = append(existingReviews, review)
			continue
		}
		newReviews = append(newReviews, review)
		newReviewIndexes = append(newReviewIndexes, idx)
	}
	if len(existingReviews) > 0 {
		if err := tx.Clauses(clause.OnConflict{
			Columns: []clause.Column{{Name: "id"}},
			DoUpdates: clause.AssignmentColumns([]string{
				"pull_request_id", "author_login", "state", "body", "commit_sha", "submitted_at", "updated_at",
			}),
		}).CreateInBatches(&existingReviews, 1000).Error; err != nil {
			return 0, err
		}
	}
	if len(newReviews) > 0 {
		if err := tx.CreateInBatches(&newReviews, 1000).Error; err != nil {
			return 0, err
		}
		for idx, review := range newReviews {
			reviews[newReviewIndexes[idx]].ID = review.ID
		}
	}
	mappings := make([]db.ImportItemMapping, 0, len(parsed))
	for idx, entry := range parsed {
		review := reviews[idx]
		if review.ID == 0 {
			return 0, fmt.Errorf("%w: pull request review target id was not assigned", ErrInvalidState)
		}
		sourceNumber := entry.number
		mappings = append(mappings, db.ImportItemMapping{
			ImportJobID:  job.ID,
			SourceType:   importTypePullReview,
			SourceID:     entry.sourceID,
			SourceNodeID: entry.sourceNodeID,
			SourceNumber: &sourceNumber,
			SourceURL:    entry.sourceURL,
			TargetType:   "pull_request_review",
			TargetID:     strconv.FormatUint(uint64(review.ID), 10),
			TargetURL:    runner.targetPullURL(entry.number) + "#pullrequestreview-" + strconv.FormatUint(uint64(review.ID), 10),
		})
	}
	if err := bulkUpsertImportMappingsTx(tx, mappings); err != nil {
		return 0, err
	}
	if err := markImportBatchAppliedTx(tx, batch); err != nil {
		return 0, err
	}
	return len(parsed), nil
}

type parsedPullRequestReviewImport struct {
	source       ghReview
	number       int
	sourceID     string
	sourceNodeID string
	sourceURL    string
}

func parsePullRequestReviewBatchItems(items []db.ImportStagingItem) ([]parsedPullRequestReviewImport, error) {
	parsed := make([]parsedPullRequestReviewImport, 0, len(items))
	seenSourceIDs := map[string]struct{}{}
	for _, item := range items {
		var source ghReview
		if err := json.Unmarshal([]byte(item.PayloadJSON), &source); err != nil {
			return nil, fmt.Errorf("%w: decode pull request review payload: %v", ErrValidation, err)
		}
		number, err := importItemNumber(item, 0, "pull_request_review")
		if err != nil {
			return nil, err
		}
		sourceID := importItemSourceID(item, source.ID, "")
		if sourceID == "" {
			return nil, fmt.Errorf("%w: pull request review source_id is required", ErrValidation)
		}
		if _, ok := seenSourceIDs[sourceID]; ok {
			return nil, fmt.Errorf("%w: duplicate pull request review source_id %q in batch", ErrValidation, sourceID)
		}
		seenSourceIDs[sourceID] = struct{}{}
		parsed = append(parsed, parsedPullRequestReviewImport{
			source:       source,
			number:       number,
			sourceID:     sourceID,
			sourceNodeID: importFirstNonEmpty(item.SourceNodeID, source.NodeID),
			sourceURL:    importFirstNonEmpty(item.SourceURL, source.HTMLURL),
		})
	}
	return parsed, nil
}

func pullRequestReviewAuthors(entries []parsedPullRequestReviewImport) []*ghUser {
	out := make([]*ghUser, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.source.User)
	}
	return out
}

func pullRequestReviewNumbers(entries []parsedPullRequestReviewImport) []int {
	out := make([]int, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.number)
	}
	return out
}

func pullRequestReviewSourceIDs(entries []parsedPullRequestReviewImport) []string {
	out := make([]string, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.sourceID)
	}
	return out
}

func (s *Service) applyImportPullRequestReviewCommentsTx(tx *gorm.DB, job db.ImportJob, batch db.ImportBatch) (int, error) {
	target, err := importTargetRepoTx(tx, job)
	if err != nil {
		return 0, err
	}
	runner := importBatchRunner(s, job)
	items, err := stagedImportItemsTx(tx, batch)
	if err != nil {
		return 0, err
	}
	parsed, err := parsePullRequestReviewCommentBatchItems(items)
	if err != nil {
		return 0, err
	}
	if len(parsed) == 0 {
		if err := markImportBatchAppliedTx(tx, batch); err != nil {
			return 0, err
		}
		return 0, nil
	}
	prsByNumber, err := pullRequestsByNumberTx(tx, target.ID, pullRequestReviewCommentNumbers(parsed))
	if err != nil {
		return 0, err
	}
	if _, err := runner.ensureSourceUsersTx(tx, pullRequestReviewCommentAuthors(parsed)); err != nil {
		return 0, err
	}
	reviewIDs, err := runner.mappingTargetIDsTx(tx, importTypePullReview, pullRequestReviewCommentReviewSourceIDs(parsed))
	if err != nil {
		return 0, err
	}
	existingIDs, err := runner.mappingTargetIDsTx(tx, importTypePullReviewComment, pullRequestReviewCommentSourceIDs(parsed))
	if err != nil {
		return 0, err
	}

	now := time.Now().UTC()
	comments := make([]db.PRReviewComment, 0, len(parsed))
	for _, entry := range parsed {
		pr, ok := prsByNumber[entry.number]
		if !ok {
			return 0, fmt.Errorf("%w: pull request #%d must be imported before reviews/comments", ErrInvalidState, entry.number)
		}
		var reviewID *uint
		if entry.source.PullRequestReviewID != 0 {
			if id, ok := reviewIDs[strconv.FormatInt(entry.source.PullRequestReviewID, 10)]; ok {
				reviewID = &id
			}
		}
		comment := db.PRReviewComment{
			PullRequestReviewID: reviewID,
			PullRequestID:       pr.ID,
			AuthorLogin:         loginOrEmpty(entry.source.User),
			Body:                db.LargeText(entry.source.Body),
			CommitID:            entry.source.CommitID,
			Path:                entry.source.Path,
			Line:                entry.source.Line,
			OriginalLine:        entry.source.OriginalLine,
			StartLine:           entry.source.StartLine,
			Side:                strings.ToUpper(entry.source.Side),
			SubjectType:         importFirstNonEmpty(strings.ToLower(entry.source.SubjectType), "line"),
			DiffHunk:            entry.source.DiffHunk,
			CreatedAt:           nonZeroTime(entry.source.CreatedAt, now),
			UpdatedAt:           nonZeroTime(entry.source.UpdatedAt, now),
		}
		if existingID, ok := existingIDs[entry.sourceID]; ok {
			comment.ID = existingID
		}
		comments = append(comments, comment)
	}
	existingComments := make([]db.PRReviewComment, 0, len(comments))
	newComments := make([]db.PRReviewComment, 0, len(comments))
	newCommentIndexes := make([]int, 0, len(comments))
	for idx, comment := range comments {
		if comment.ID != 0 {
			existingComments = append(existingComments, comment)
			continue
		}
		newComments = append(newComments, comment)
		newCommentIndexes = append(newCommentIndexes, idx)
	}
	if len(existingComments) > 0 {
		if err := tx.Clauses(clause.OnConflict{
			Columns: []clause.Column{{Name: "id"}},
			DoUpdates: clause.AssignmentColumns([]string{
				"pull_request_review_id", "pull_request_id", "author_login", "body", "commit_id",
				"path", "line", "original_line", "start_line", "side", "subject_type", "diff_hunk", "updated_at",
			}),
		}).CreateInBatches(&existingComments, 1000).Error; err != nil {
			return 0, err
		}
	}
	if len(newComments) > 0 {
		if err := tx.CreateInBatches(&newComments, 1000).Error; err != nil {
			return 0, err
		}
		for idx, comment := range newComments {
			comments[newCommentIndexes[idx]].ID = comment.ID
		}
	}

	mappings := make([]db.ImportItemMapping, 0, len(parsed))
	for idx, entry := range parsed {
		comment := comments[idx]
		if comment.ID == 0 {
			return 0, fmt.Errorf("%w: pull request review comment target id was not assigned", ErrInvalidState)
		}
		sourceNumber := entry.number
		mapping, err := newImportItemMapping(job.ID, importTypePullReviewComment, entry.sourceID, entry.sourceNodeID, &sourceNumber, entry.sourceURL, "pull_request_review_comment", strconv.FormatUint(uint64(comment.ID), 10), nil, runner.targetPullReviewCommentURL(entry.number, comment.ID), map[string]any{
			"source_in_reply_to_id": entry.source.InReplyToID,
		})
		if err != nil {
			return 0, err
		}
		mappings = append(mappings, mapping)
	}
	if err := bulkUpsertImportMappingsTx(tx, mappings); err != nil {
		return 0, err
	}
	if err := resolveReviewCommentRepliesTx(tx, runner, parsed); err != nil {
		return 0, err
	}
	if err := markImportBatchAppliedTx(tx, batch); err != nil {
		return 0, err
	}
	return len(parsed), nil
}

type parsedPullRequestReviewCommentImport struct {
	source       ghReviewComment
	number       int
	sourceID     string
	sourceNodeID string
	sourceURL    string
}

func parsePullRequestReviewCommentBatchItems(items []db.ImportStagingItem) ([]parsedPullRequestReviewCommentImport, error) {
	parsed := make([]parsedPullRequestReviewCommentImport, 0, len(items))
	seenSourceIDs := map[string]struct{}{}
	for _, item := range items {
		var source ghReviewComment
		if err := json.Unmarshal([]byte(item.PayloadJSON), &source); err != nil {
			return nil, fmt.Errorf("%w: decode pull request review comment payload: %v", ErrValidation, err)
		}
		number, err := importItemNumber(item, 0, "pull_request_review_comment")
		if err != nil {
			return nil, err
		}
		sourceID := importItemSourceID(item, source.ID, "")
		if sourceID == "" {
			return nil, fmt.Errorf("%w: pull request review comment source_id is required", ErrValidation)
		}
		if _, ok := seenSourceIDs[sourceID]; ok {
			return nil, fmt.Errorf("%w: duplicate pull request review comment source_id %q in batch", ErrValidation, sourceID)
		}
		seenSourceIDs[sourceID] = struct{}{}
		parsed = append(parsed, parsedPullRequestReviewCommentImport{
			source:       source,
			number:       number,
			sourceID:     sourceID,
			sourceNodeID: importFirstNonEmpty(item.SourceNodeID, source.NodeID),
			sourceURL:    importFirstNonEmpty(item.SourceURL, source.HTMLURL),
		})
	}
	return parsed, nil
}

func pullRequestReviewCommentAuthors(entries []parsedPullRequestReviewCommentImport) []*ghUser {
	out := make([]*ghUser, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.source.User)
	}
	return out
}

func pullRequestReviewCommentNumbers(entries []parsedPullRequestReviewCommentImport) []int {
	out := make([]int, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.number)
	}
	return out
}

func pullRequestReviewCommentSourceIDs(entries []parsedPullRequestReviewCommentImport) []string {
	out := make([]string, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.sourceID)
	}
	return out
}

func pullRequestReviewCommentReviewSourceIDs(entries []parsedPullRequestReviewCommentImport) []string {
	out := make([]string, 0, len(entries))
	for _, entry := range entries {
		if entry.source.PullRequestReviewID != 0 {
			out = append(out, strconv.FormatInt(entry.source.PullRequestReviewID, 10))
		}
	}
	return out
}

func (s *Service) applyImportReleasesTx(tx *gorm.DB, job db.ImportJob, batch db.ImportBatch) (int, error) {
	target, err := importTargetRepoTx(tx, job)
	if err != nil {
		return 0, err
	}
	runner := importBatchRunner(s, job)
	items, err := stagedImportItemsTx(tx, batch)
	if err != nil {
		return 0, err
	}
	parsed, err := parseReleaseBatchItems(items)
	if err != nil {
		return 0, err
	}
	if len(parsed) == 0 {
		if err := markImportBatchAppliedTx(tx, batch); err != nil {
			return 0, err
		}
		return 0, nil
	}
	authors, err := runner.ensureSourceUsersTx(tx, releaseAuthors(parsed))
	if err != nil {
		return 0, err
	}
	now := time.Now().UTC()
	rows := make([]db.Release, 0, len(parsed))
	for _, entry := range parsed {
		login := importSourceLogin(entry.source.Author)
		author, ok := authors[login]
		if !ok || author.ID == 0 {
			return 0, fmt.Errorf("%w: source user %q was not created", ErrInvalidState, login)
		}
		rows = append(rows, db.Release{
			RepositoryID: target.ID,
			TagName:      entry.tagName,
			Name:         entry.source.Name,
			Body:         db.LargeText(entry.source.Body),
			Draft:        entry.source.Draft,
			PreRelease:   entry.source.Prerelease,
			AuthorID:     author.ID,
			CreatedAt:    nonZeroTime(entry.source.CreatedAt, now),
			PublishedAt:  entry.source.PublishedAt,
		})
	}
	if err := tx.Clauses(clause.OnConflict{
		Columns: []clause.Column{{Name: "repository_id"}, {Name: "tag_name"}},
		DoUpdates: clause.AssignmentColumns([]string{
			"name", "body", "draft", "pre_release", "author_id", "published_at",
		}),
	}).CreateInBatches(&rows, 1000).Error; err != nil {
		return 0, err
	}
	releasesByTagName, err := releasesByTagNameTx(tx, target.ID, releaseTagNames(parsed))
	if err != nil {
		return 0, err
	}
	mappings := make([]db.ImportItemMapping, 0, len(parsed))
	for _, entry := range parsed {
		release, ok := releasesByTagName[entry.tagName]
		if !ok {
			return 0, fmt.Errorf("%w: release %q was not created", ErrInvalidState, entry.tagName)
		}
		mapping, err := newImportItemMapping(
			job.ID,
			importTypeRelease,
			entry.sourceID,
			entry.sourceNodeID,
			nil,
			entry.sourceURL,
			"release",
			strconv.FormatUint(uint64(release.ID), 10),
			nil,
			runner.targetReleaseURL(release.ID),
			map[string]any{"tag_name": entry.tagName},
		)
		if err != nil {
			return 0, err
		}
		mappings = append(mappings, mapping)
	}
	if err := bulkUpsertImportMappingsTx(tx, mappings); err != nil {
		return 0, err
	}
	if err := markImportBatchAppliedTx(tx, batch); err != nil {
		return 0, err
	}
	return len(items), nil
}

func (s *Service) applyImportReleaseAssetsTx(ctx context.Context, tx *gorm.DB, job db.ImportJob, batch db.ImportBatch) (int, error) {
	target, err := importTargetRepoTx(tx, job)
	if err != nil {
		return 0, err
	}
	runner := importBatchRunner(s, job)
	items, err := stagedImportItemsTx(tx, batch)
	if err != nil {
		return 0, err
	}
	entries, err := parseReleaseAssetBatchItems(items)
	if err != nil {
		return 0, err
	}
	if len(entries) == 0 {
		if err := markImportBatchAppliedTx(tx, batch); err != nil {
			return 0, err
		}
		return 0, nil
	}
	releasesBySourceID, err := releaseTargetsBySourceIDTx(tx, runner, releaseAssetReleaseSourceIDs(entries))
	if err != nil {
		return 0, err
	}
	releasesByTagName, err := releasesByTagNameTx(tx, target.ID, releaseAssetTagNames(entries))
	if err != nil {
		return 0, err
	}
	applyEntries := make([]releaseAssetApplyEntry, 0, len(entries))
	for _, entry := range entries {
		release, ok := db.Release{}, false
		if entry.source.ReleaseID != 0 {
			release, ok = releasesBySourceID[strconv.FormatInt(entry.source.ReleaseID, 10)]
		}
		if !ok && strings.TrimSpace(entry.source.ReleaseTagName) != "" {
			release, ok = releasesByTagName[strings.TrimSpace(entry.source.ReleaseTagName)]
		}
		if !ok {
			return 0, fmt.Errorf("%w: release asset requires release_id or release_tag_name imported first", ErrValidation)
		}
		applyEntries = append(applyEntries, releaseAssetApplyEntry{
			release: release,
			source:  entry.source,
		})
	}
	stagedAssets, err := stagedImportAssetsBySourceIDTx(tx, job.ID, releaseAssetSourceIDs(entries))
	if err != nil {
		return 0, err
	}
	if err := s.applyReleaseAssetsBulkTx(ctx, tx, runner, applyEntries, stagedAssets); err != nil {
		return 0, err
	}
	if err := markImportBatchAppliedTx(tx, batch); err != nil {
		return 0, err
	}
	return len(items), nil
}

func (s *Service) applyImportBranchProtectionsTx(tx *gorm.DB, job db.ImportJob, batch db.ImportBatch) (int, error) {
	target, err := importTargetRepoTx(tx, job)
	if err != nil {
		return 0, err
	}
	runner := importBatchRunner(s, job)
	items, err := stagedImportItemsTx(tx, batch)
	if err != nil {
		return 0, err
	}
	parsed, err := parseBranchProtectionBatchItems(items)
	if err != nil {
		return 0, err
	}
	if len(parsed) == 0 {
		if err := markImportBatchAppliedTx(tx, batch); err != nil {
			return 0, err
		}
		return 0, nil
	}
	rows := make([]db.BranchProtection, 0, len(parsed))
	for _, entry := range parsed {
		rows = append(rows, db.BranchProtection{
			RepositoryID:             target.ID,
			BranchName:               entry.branchName,
			RequiredStatusChecksJSON: mustJSON(entry.protection["required_status_checks"]),
			EnforceAdmins:            protectionEnabled(entry.protection["enforce_admins"]),
			RequiredSignatures:       protectionEnabled(entry.protection["required_signatures"]),
			RequiredPullRequestJSON:  mustJSON(entry.protection["required_pull_request_reviews"]),
			RestrictionsJSON:         mustJSON(entry.protection["restrictions"]),
		})
	}
	if err := tx.Clauses(clause.OnConflict{
		Columns: []clause.Column{{Name: "repository_id"}, {Name: "branch_name"}},
		DoUpdates: clause.AssignmentColumns([]string{
			"required_status_checks_json",
			"enforce_admins",
			"required_signatures",
			"required_pull_request_json",
			"restrictions_json",
			"updated_at",
		}),
	}).CreateInBatches(&rows, 1000).Error; err != nil {
		return 0, err
	}
	rowsByBranchName, err := branchProtectionsByNameTx(tx, target.ID, branchProtectionNames(parsed))
	if err != nil {
		return 0, err
	}
	mappings := make([]db.ImportItemMapping, 0, len(parsed))
	for _, entry := range parsed {
		row, ok := rowsByBranchName[entry.branchName]
		if !ok {
			return 0, fmt.Errorf("%w: branch protection %q was not created", ErrInvalidState, entry.branchName)
		}
		mapping, err := newImportItemMapping(
			job.ID,
			importTypeBranchProtection,
			entry.branchName,
			entry.sourceNodeID,
			nil,
			entry.sourceURL,
			"branch_protection",
			strconv.FormatUint(uint64(row.ID), 10),
			nil,
			runner.targetRepoURL()+"/settings/branches",
			entry.protection,
		)
		if err != nil {
			return 0, err
		}
		mappings = append(mappings, mapping)
	}
	if err := bulkUpsertImportMappingsTx(tx, mappings); err != nil {
		return 0, err
	}
	if err := markImportBatchAppliedTx(tx, batch); err != nil {
		return 0, err
	}
	return len(items), nil
}

func (s *Service) applyImportWebhooksTx(tx *gorm.DB, job db.ImportJob, batch db.ImportBatch) (int, error) {
	target, err := importTargetRepoTx(tx, job)
	if err != nil {
		return 0, err
	}
	runner := importBatchRunner(s, job)
	items, err := stagedImportItemsTx(tx, batch)
	if err != nil {
		return 0, err
	}
	for _, item := range items {
		var source ghHook
		if err := json.Unmarshal([]byte(item.PayloadJSON), &source); err != nil {
			return 0, fmt.Errorf("%w: decode webhook payload: %v", ErrValidation, err)
		}
		sourceID := importItemSourceID(item, source.ID, "")
		if sourceID == "" {
			return 0, fmt.Errorf("%w: webhook source_id is required", ErrValidation)
		}
		existingID, ok, err := runner.mappingTargetIDTx(tx, importTypeWebhook, sourceID)
		if err != nil {
			return 0, err
		}
		config := cloneStringMap(source.Config)
		config["secret"] = ""
		config["import_note"] = "GitHub webhook secret is not readable; configure a new secret before enabling."
		configJSON, err := marshalJSON(config)
		if err != nil {
			return 0, err
		}
		eventsJSON, err := marshalJSON(source.Events)
		if err != nil {
			return 0, err
		}
		hook := db.Webhook{
			RepositoryID: target.ID,
			Name:         importFirstNonEmpty(source.Name, "web"),
			Active:       false,
			EventsJSON:   eventsJSON,
			ConfigJSON:   configJSON,
			CreatedAt:    nonZeroTime(source.CreatedAt, time.Now().UTC()),
			UpdatedAt:    nonZeroTime(source.UpdatedAt, time.Now().UTC()),
		}
		if ok {
			hook.ID = existingID
			if err := tx.Model(&db.Webhook{}).Where("id = ?", existingID).Updates(map[string]any{
				"repository_id": hook.RepositoryID,
				"name":          hook.Name,
				"active":        hook.Active,
				"events_json":   hook.EventsJSON,
				"config_json":   hook.ConfigJSON,
				"updated_at":    hook.UpdatedAt,
			}).Error; err != nil {
				return 0, err
			}
		} else if err := tx.Create(&hook).Error; err != nil {
			return 0, err
		}
		sourceURL := importFirstNonEmpty(item.SourceURL, source.URL)
		if err := upsertImportMappingTx(tx, job.ID, importTypeWebhook, sourceID, item.SourceNodeID, nil, sourceURL, "webhook", strconv.FormatUint(uint64(hook.ID), 10), nil, runner.targetRepoURL()+"/settings/hooks", map[string]any{
			"active": false,
			"note":   "imported inactive because GitHub does not expose webhook secrets",
		}); err != nil {
			return 0, err
		}
	}
	if err := markImportBatchAppliedTx(tx, batch); err != nil {
		return 0, err
	}
	return len(items), nil
}

func (s *Service) applyImportWorkflowsTx(tx *gorm.DB, job db.ImportJob, batch db.ImportBatch) (int, error) {
	target, err := importTargetRepoTx(tx, job)
	if err != nil {
		return 0, err
	}
	runner := importBatchRunner(s, job)
	items, err := stagedImportItemsTx(tx, batch)
	if err != nil {
		return 0, err
	}
	parsed, err := parseWorkflowBatchItems(items)
	if err != nil {
		return 0, err
	}
	if len(parsed) == 0 {
		if err := markImportBatchAppliedTx(tx, batch); err != nil {
			return 0, err
		}
		return 0, nil
	}
	now := time.Now().UTC()
	rows := make([]db.Workflow, 0, len(parsed))
	for _, entry := range parsed {
		rows = append(rows, db.Workflow{
			RepositoryID: target.ID,
			Name:         entry.source.Name,
			Path:         entry.path,
			State:        normalizeWorkflowState(entry.source.State),
			CreatedAt:    nonZeroTime(entry.source.CreatedAt, now),
			UpdatedAt:    nonZeroTime(entry.source.UpdatedAt, now),
		})
	}
	if err := tx.Clauses(clause.OnConflict{
		Columns:   []clause.Column{{Name: "repository_id"}, {Name: "path"}},
		DoUpdates: clause.AssignmentColumns([]string{"name", "state", "updated_at"}),
	}).CreateInBatches(&rows, 1000).Error; err != nil {
		return 0, err
	}
	rowsByPath, err := workflowsByPathTx(tx, target.ID, workflowPaths(parsed))
	if err != nil {
		return 0, err
	}
	mappings := make([]db.ImportItemMapping, 0, len(parsed))
	for _, entry := range parsed {
		row, ok := rowsByPath[entry.path]
		if !ok {
			return 0, fmt.Errorf("%w: workflow %q was not created", ErrInvalidState, entry.path)
		}
		mapping, err := newImportItemMapping(
			job.ID,
			importTypeWorkflow,
			entry.sourceID,
			entry.sourceNodeID,
			nil,
			entry.sourceURL,
			"workflow",
			strconv.FormatUint(uint64(row.ID), 10),
			nil,
			runner.targetRepoURL()+"/actions/workflows/"+url.PathEscape(entry.path),
			map[string]any{"path": entry.path},
		)
		if err != nil {
			return 0, err
		}
		mappings = append(mappings, mapping)
	}
	if err := bulkUpsertImportMappingsTx(tx, mappings); err != nil {
		return 0, err
	}
	if err := markImportBatchAppliedTx(tx, batch); err != nil {
		return 0, err
	}
	return len(items), nil
}

func (s *Service) applyImportSecretNamesTx(tx *gorm.DB, job db.ImportJob, batch db.ImportBatch) (int, error) {
	items, err := stagedImportItemsTx(tx, batch)
	if err != nil {
		return 0, err
	}
	for _, item := range items {
		var source ghSecretName
		if err := json.Unmarshal([]byte(item.PayloadJSON), &source); err != nil {
			return 0, fmt.Errorf("%w: decode secret name payload: %v", ErrValidation, err)
		}
		name := importFirstNonEmpty(source.Name, item.SourceID)
		if name == "" {
			return 0, fmt.Errorf("%w: secret name is required", ErrValidation)
		}
		if err := upsertImportMappingTx(tx, job.ID, importTypeSecretName, name, item.SourceNodeID, nil, item.SourceURL, "secret_name", name, nil, "", map[string]any{
			"name":       name,
			"created_at": source.CreatedAt,
			"updated_at": source.UpdatedAt,
			"note":       "GitHub does not expose secret values; recreate this secret in AGS.",
		}); err != nil {
			return 0, err
		}
	}
	if err := markImportBatchAppliedTx(tx, batch); err != nil {
		return 0, err
	}
	return len(items), nil
}

// importAccessHandoffPayload records source collaboration for an operator to
// review after import. It is intentionally not an instruction to create AGS
// collaborators, Teams, or repository grants.
type importAccessHandoffPayload struct {
	Kind         string `json:"kind"`
	SourceID     string `json:"source_id"`
	SourceNodeID string `json:"source_node_id"`
	SourceLogin  string `json:"source_login"`
	GitHubID     int64  `json:"github_id"`
	SourceURL    string `json:"source_url"`
	Team         string `json:"team"`
	TeamName     string `json:"team_name"`
	Permission   string `json:"permission"`
}

func (s *Service) applyImportAccessHandoffsTx(tx *gorm.DB, job db.ImportJob, batch db.ImportBatch) (int, error) {
	items, err := stagedImportItemsTx(tx, batch)
	if err != nil {
		return 0, err
	}
	runner := importBatchRunner(s, job)
	mappings := make([]db.ImportItemMapping, 0, len(items))
	seen := make(map[string]struct{}, len(items))
	for _, item := range items {
		var source importAccessHandoffPayload
		if err := json.Unmarshal([]byte(item.PayloadJSON), &source); err != nil {
			return 0, fmt.Errorf("%w: decode access handoff payload: %v", ErrValidation, err)
		}
		source.Kind = strings.ToLower(strings.TrimSpace(source.Kind))
		source.SourceID = importFirstNonEmpty(strings.TrimSpace(source.SourceID), strings.TrimSpace(item.SourceID))
		source.SourceNodeID = importFirstNonEmpty(strings.TrimSpace(source.SourceNodeID), strings.TrimSpace(item.SourceNodeID))
		source.SourceURL = importFirstNonEmpty(strings.TrimSpace(source.SourceURL), strings.TrimSpace(item.SourceURL))
		if source.SourceID == "" {
			return 0, fmt.Errorf("%w: access handoff source_id is required", ErrValidation)
		}
		key := source.Kind + "\x00" + source.SourceID
		if _, exists := seen[key]; exists {
			return 0, fmt.Errorf("%w: duplicate access handoff %q", ErrValidation, key)
		}
		seen[key] = struct{}{}

		metadata := map[string]any{
			"permission": normalizeImportPermission(source.Permission),
			"reason":     "requires a separate, explicit destination access grant",
		}
		var sourceType string
		switch source.Kind {
		case "collaborator", "team_member":
			if source.GitHubID <= 0 || strings.TrimSpace(source.SourceLogin) == "" {
				return 0, fmt.Errorf("%w: %s handoff requires a stable GitHub ID and login", ErrValidation, source.Kind)
			}
			user, err := runner.ensureSourceUserTx(tx, &ghUser{ID: source.GitHubID, NodeID: source.SourceNodeID, Login: source.SourceLogin, HTMLURL: source.SourceURL})
			if err != nil {
				return 0, err
			}
			if user.Status == db.UserStatusActive {
				metadata["matched_ags_login"] = user.Login
			}
			if source.Kind == "collaborator" {
				sourceType = importTypeCollaborator
			} else {
				sourceType = importTypeTeamMember
				metadata["team"] = source.Team
				metadata["team_name"] = source.TeamName
				metadata["reason"] = "requires a separate, explicit destination team membership and repository access grant"
			}
		case "team":
			if strings.TrimSpace(source.Team) == "" {
				return 0, fmt.Errorf("%w: team handoff requires a team slug", ErrValidation)
			}
			sourceType = importTypeTeam
			metadata["team"] = source.Team
			metadata["team_name"] = importFirstNonEmpty(source.TeamName, source.Team)
		case "":
			return 0, fmt.Errorf("%w: access handoff kind is required", ErrValidation)
		default:
			return 0, fmt.Errorf("%w: unsupported access handoff kind %q", ErrValidation, source.Kind)
		}
		mapping, err := newImportItemMapping(job.ID, sourceType, source.SourceID, source.SourceNodeID, nil, source.SourceURL, "access_handoff", "", nil, "", metadata)
		if err != nil {
			return 0, err
		}
		mappings = append(mappings, mapping)
	}
	if err := bulkUpsertImportMappingsTx(tx, mappings); err != nil {
		return 0, err
	}
	if err := markImportBatchAppliedTx(tx, batch); err != nil {
		return 0, err
	}
	return len(items), nil
}

type importReleaseAssetPayload struct {
	ghReleaseAsset
	ReleaseID      int64  `json:"release_id"`
	ReleaseTagName string `json:"release_tag_name"`
	ContentSHA256  string `json:"content_sha256"`
	ContentSize    *int64 `json:"content_size"`
}

type parsedReleaseImport struct {
	source       ghRelease
	tagName      string
	sourceID     string
	sourceNodeID string
	sourceURL    string
}

func parseReleaseBatchItems(items []db.ImportStagingItem) ([]parsedReleaseImport, error) {
	parsed := make([]parsedReleaseImport, 0, len(items))
	seenSourceIDs := map[string]struct{}{}
	seenTags := map[string]struct{}{}
	for _, item := range items {
		var source ghRelease
		if err := json.Unmarshal([]byte(item.PayloadJSON), &source); err != nil {
			return nil, fmt.Errorf("%w: decode release payload: %v", ErrValidation, err)
		}
		tagName := strings.TrimSpace(source.TagName)
		if tagName == "" {
			return nil, fmt.Errorf("%w: release tag_name is required", ErrValidation)
		}
		if _, ok := seenTags[tagName]; ok {
			return nil, fmt.Errorf("%w: duplicate release tag_name %q in batch", ErrValidation, tagName)
		}
		seenTags[tagName] = struct{}{}
		sourceID := importItemSourceID(item, source.ID, tagName)
		if sourceID == "" {
			return nil, fmt.Errorf("%w: release source_id is required", ErrValidation)
		}
		if _, ok := seenSourceIDs[sourceID]; ok {
			return nil, fmt.Errorf("%w: duplicate release source_id %q in batch", ErrValidation, sourceID)
		}
		seenSourceIDs[sourceID] = struct{}{}
		parsed = append(parsed, parsedReleaseImport{
			source:       source,
			tagName:      tagName,
			sourceID:     sourceID,
			sourceNodeID: importFirstNonEmpty(item.SourceNodeID, source.NodeID),
			sourceURL:    importFirstNonEmpty(item.SourceURL, source.HTMLURL),
		})
	}
	return parsed, nil
}

func releaseAuthors(entries []parsedReleaseImport) []*ghUser {
	out := make([]*ghUser, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.source.Author)
	}
	return out
}

func releaseTagNames(entries []parsedReleaseImport) []string {
	out := make([]string, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.tagName)
	}
	return out
}

type parsedReleaseAssetImport struct {
	source   importReleaseAssetPayload
	sourceID string
}

func parseReleaseAssetBatchItems(items []db.ImportStagingItem) ([]parsedReleaseAssetImport, error) {
	parsed := make([]parsedReleaseAssetImport, 0, len(items))
	seenSourceIDs := map[string]struct{}{}
	for _, item := range items {
		var raw map[string]json.RawMessage
		if err := json.Unmarshal([]byte(item.PayloadJSON), &raw); err != nil {
			return nil, fmt.Errorf("%w: decode release asset payload: %v", ErrValidation, err)
		}
		if _, legacy := raw["content_base64"]; legacy {
			return nil, fmt.Errorf("%w: release assets must be uploaded through raw-byte staging, not content_base64", ErrValidation)
		}
		var source importReleaseAssetPayload
		if err := json.Unmarshal([]byte(item.PayloadJSON), &source); err != nil {
			return nil, fmt.Errorf("%w: decode release asset payload: %v", ErrValidation, err)
		}
		sourceID := releaseAssetSourceID(source)
		if sourceID == "" {
			return nil, fmt.Errorf("%w: release asset source_id is required", ErrValidation)
		}
		if source.ContentSize == nil {
			return nil, fmt.Errorf("%w: release asset %q has no content_size", ErrValidation, source.Name)
		}
		if *source.ContentSize < 0 {
			return nil, fmt.Errorf("%w: release asset %q has a negative content_size", ErrValidation, source.Name)
		}
		if _, err := normalizeImportAssetChecksum(source.ContentSHA256); err != nil {
			return nil, fmt.Errorf("release asset %q: %w", source.Name, err)
		}
		if _, ok := seenSourceIDs[sourceID]; ok {
			return nil, fmt.Errorf("%w: duplicate release asset source_id %q in batch", ErrValidation, sourceID)
		}
		seenSourceIDs[sourceID] = struct{}{}
		parsed = append(parsed, parsedReleaseAssetImport{source: source, sourceID: sourceID})
	}
	return parsed, nil
}

func releaseAssetSourceID(source importReleaseAssetPayload) string {
	return importFirstNonEmpty(sourceIDFromInt64(source.ID), source.Name, source.BrowserDownloadURL, source.URL)
}

func releaseAssetReleaseSourceIDs(entries []parsedReleaseAssetImport) []string {
	out := make([]string, 0, len(entries))
	for _, entry := range entries {
		if entry.source.ReleaseID != 0 {
			out = append(out, strconv.FormatInt(entry.source.ReleaseID, 10))
		}
	}
	return out
}

func releaseAssetSourceIDs(entries []parsedReleaseAssetImport) []string {
	out := make([]string, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.sourceID)
	}
	return out
}

func releaseAssetTagNames(entries []parsedReleaseAssetImport) []string {
	out := make([]string, 0, len(entries))
	for _, entry := range entries {
		if strings.TrimSpace(entry.source.ReleaseTagName) != "" {
			out = append(out, strings.TrimSpace(entry.source.ReleaseTagName))
		}
	}
	return out
}

type releaseAssetApplyEntry struct {
	release db.Release
	source  importReleaseAssetPayload
}

type preparedReleaseAsset struct {
	release  db.Release
	source   importReleaseAssetPayload
	sourceID string
	asset    db.ReleaseAsset
}

func (s *Service) applyReleaseAssetsBulkTx(ctx context.Context, tx *gorm.DB, runner *githubImportRunner, entries []releaseAssetApplyEntry, stagedAssets map[string]db.ImportStagedAsset) error {
	if len(entries) == 0 {
		return nil
	}
	seenSourceIDs := map[string]struct{}{}
	prepared := make([]preparedReleaseAsset, 0, len(entries))
	mappings := make([]db.ImportItemMapping, 0, len(entries))
	now := time.Now().UTC()
	for _, entry := range entries {
		sourceID := releaseAssetSourceID(entry.source)
		if sourceID == "" {
			return fmt.Errorf("%w: release asset source_id is required", ErrValidation)
		}
		if _, ok := seenSourceIDs[sourceID]; ok {
			return fmt.Errorf("%w: duplicate release asset source_id %q in batch", ErrValidation, sourceID)
		}
		seenSourceIDs[sourceID] = struct{}{}
		staged, ok := stagedAssets[sourceID]
		if !ok {
			return fmt.Errorf("%w: release asset %q has no staged binary", ErrValidation, entry.source.Name)
		}
		expectedChecksum, err := normalizeImportAssetChecksum(entry.source.ContentSHA256)
		if err != nil {
			return err
		}
		if entry.source.ContentSize == nil || staged.Size != *entry.source.ContentSize || !checksumEqual(staged.Checksum, expectedChecksum) {
			return fmt.Errorf("%w: staged release asset %q does not match its metadata", ErrValidation, entry.source.Name)
		}
		if entry.source.Size != staged.Size {
			return fmt.Errorf("%w: staged release asset %q does not match GitHub's declared size", ErrValidation, entry.source.Name)
		}
		content, err := s.readImportStagedAsset(ctx, staged)
		if err != nil {
			return fmt.Errorf("read staged release asset %q: %w", entry.source.Name, err)
		}
		prepared = append(prepared, preparedReleaseAsset{
			release:  entry.release,
			source:   entry.source,
			sourceID: sourceID,
			asset: db.ReleaseAsset{
				ReleaseID:   entry.release.ID,
				Name:        entry.source.Name,
				Label:       entry.source.Label,
				ContentType: importFirstNonEmpty(entry.source.ContentType, "application/octet-stream"),
				Size:        int64(len(content)),
				Content:     content,
				CreatedAt:   nonZeroTime(entry.source.CreatedAt, now),
				UpdatedAt:   nonZeroTime(entry.source.UpdatedAt, now),
			},
		})
	}
	if len(prepared) > 0 {
		existingIDs, err := runner.mappingTargetIDsTx(tx, importTypeReleaseAsset, preparedReleaseAssetSourceIDs(prepared))
		if err != nil {
			return err
		}
		existingByReleaseName, err := releaseAssetsByReleaseAndNameTx(tx, prepared)
		if err != nil {
			return err
		}
		existingAssets := make([]db.ReleaseAsset, 0)
		newAssets := make([]db.ReleaseAsset, 0)
		newIndexes := make([]int, 0)
		for idx := range prepared {
			if existingID, ok := existingIDs[prepared[idx].sourceID]; ok {
				prepared[idx].asset.ID = existingID
			} else if existing, ok := existingByReleaseName[releaseAssetNameKey(prepared[idx].release.ID, prepared[idx].source.Name)]; ok {
				prepared[idx].asset.ID = existing.ID
			}
			if prepared[idx].asset.ID != 0 {
				existingAssets = append(existingAssets, prepared[idx].asset)
			} else {
				newAssets = append(newAssets, prepared[idx].asset)
				newIndexes = append(newIndexes, idx)
			}
		}
		if len(existingAssets) > 0 {
			if err := tx.Clauses(clause.OnConflict{
				Columns: []clause.Column{{Name: "id"}},
				DoUpdates: clause.AssignmentColumns([]string{
					"release_id", "name", "label", "content_type", "size", "content", "updated_at",
				}),
			}).CreateInBatches(&existingAssets, 1000).Error; err != nil {
				return err
			}
		}
		if len(newAssets) > 0 {
			if err := tx.CreateInBatches(&newAssets, 1000).Error; err != nil {
				return err
			}
			for idx, asset := range newAssets {
				prepared[newIndexes[idx]].asset.ID = asset.ID
			}
		}
		for _, entry := range prepared {
			if entry.asset.ID == 0 {
				return fmt.Errorf("%w: release asset target id was not assigned", ErrInvalidState)
			}
			mapping, err := newImportItemMapping(
				runner.job.ID,
				importTypeReleaseAsset,
				entry.sourceID,
				entry.source.NodeID,
				nil,
				entry.source.BrowserDownloadURL,
				"release_asset",
				strconv.FormatUint(uint64(entry.asset.ID), 10),
				nil,
				runner.targetReleaseAssetURL(entry.asset.ID),
				map[string]any{
					"name": entry.source.Name,
					"size": entry.asset.Size,
				},
			)
			if err != nil {
				return err
			}
			mappings = append(mappings, mapping)
		}
	}
	return bulkUpsertImportMappingsTx(tx, mappings)
}

func stagedImportAssetsBySourceIDTx(tx *gorm.DB, jobID uint, sourceIDs []string) (map[string]db.ImportStagedAsset, error) {
	sourceIDs = uniqueStrings(sourceIDs)
	if len(sourceIDs) == 0 {
		return map[string]db.ImportStagedAsset{}, nil
	}
	var rows []db.ImportStagedAsset
	if err := tx.Where("import_job_id = ? AND source_id IN ?", jobID, sourceIDs).Find(&rows).Error; err != nil {
		return nil, err
	}
	out := make(map[string]db.ImportStagedAsset, len(rows))
	for _, row := range rows {
		out[row.SourceID] = row
	}
	return out, nil
}

func preparedReleaseAssetSourceIDs(entries []preparedReleaseAsset) []string {
	out := make([]string, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.sourceID)
	}
	return out
}

func releaseAssetNameKey(releaseID uint, name string) string {
	return strconv.FormatUint(uint64(releaseID), 10) + "\x00" + strings.TrimSpace(name)
}

func releaseAssetsByReleaseAndNameTx(tx *gorm.DB, entries []preparedReleaseAsset) (map[string]db.ReleaseAsset, error) {
	releaseIDs := make([]uint, 0, len(entries))
	names := make([]string, 0, len(entries))
	for _, entry := range entries {
		name := strings.TrimSpace(entry.source.Name)
		if entry.release.ID == 0 || name == "" {
			continue
		}
		releaseIDs = append(releaseIDs, entry.release.ID)
		names = append(names, name)
	}
	releaseIDs = uniqueUintIDs(releaseIDs)
	names = uniqueStrings(names)
	if len(releaseIDs) == 0 || len(names) == 0 {
		return map[string]db.ReleaseAsset{}, nil
	}
	var rows []db.ReleaseAsset
	if err := tx.Where("release_id IN ? AND name IN ?", releaseIDs, names).Find(&rows).Error; err != nil {
		return nil, err
	}
	out := make(map[string]db.ReleaseAsset, len(rows))
	for _, row := range rows {
		out[releaseAssetNameKey(row.ReleaseID, row.Name)] = row
	}
	return out, nil
}

func releaseTargetsBySourceIDTx(tx *gorm.DB, runner *githubImportRunner, sourceIDs []string) (map[string]db.Release, error) {
	targetIDs, err := runner.mappingTargetIDsTx(tx, importTypeRelease, sourceIDs)
	if err != nil {
		return nil, err
	}
	if len(targetIDs) == 0 {
		return map[string]db.Release{}, nil
	}
	ids := make([]uint, 0, len(targetIDs))
	for _, id := range targetIDs {
		ids = append(ids, id)
	}
	releasesByID, err := releasesByIDTx(tx, ids)
	if err != nil {
		return nil, err
	}
	out := make(map[string]db.Release, len(targetIDs))
	for sourceID, id := range targetIDs {
		release, ok := releasesByID[id]
		if ok {
			out[sourceID] = release
		}
	}
	return out, nil
}

func releasesByIDTx(tx *gorm.DB, ids []uint) (map[uint]db.Release, error) {
	ids = uniqueUintIDs(ids)
	if len(ids) == 0 {
		return map[uint]db.Release{}, nil
	}
	var rows []db.Release
	if err := tx.Where("id IN ?", ids).Find(&rows).Error; err != nil {
		return nil, err
	}
	out := make(map[uint]db.Release, len(rows))
	for _, row := range rows {
		out[row.ID] = row
	}
	return out, nil
}

func releasesByTagNameTx(tx *gorm.DB, repoID uint, tagNames []string) (map[string]db.Release, error) {
	tagNames = uniqueStrings(tagNames)
	if len(tagNames) == 0 {
		return map[string]db.Release{}, nil
	}
	var rows []db.Release
	if err := tx.Where("repository_id = ? AND tag_name IN ?", repoID, tagNames).Find(&rows).Error; err != nil {
		return nil, err
	}
	out := make(map[string]db.Release, len(rows))
	for _, row := range rows {
		out[row.TagName] = row
	}
	return out, nil
}

type parsedBranchProtectionImport struct {
	branchName   string
	sourceNodeID string
	sourceURL    string
	protection   map[string]any
}

func parseBranchProtectionBatchItems(items []db.ImportStagingItem) ([]parsedBranchProtectionImport, error) {
	parsed := make([]parsedBranchProtectionImport, 0, len(items))
	seenBranches := map[string]struct{}{}
	for _, item := range items {
		var payload map[string]any
		if err := json.Unmarshal([]byte(item.PayloadJSON), &payload); err != nil {
			return nil, fmt.Errorf("%w: decode branch protection payload: %v", ErrValidation, err)
		}
		branchName := importFirstNonEmpty(item.SourceID, stringFromAny(payload["branch_name"]), stringFromAny(payload["name"]))
		if branchName == "" {
			return nil, fmt.Errorf("%w: branch protection branch_name is required", ErrValidation)
		}
		if _, ok := seenBranches[branchName]; ok {
			return nil, fmt.Errorf("%w: duplicate branch protection branch_name %q in batch", ErrValidation, branchName)
		}
		seenBranches[branchName] = struct{}{}
		protection := payload
		if nested, ok := payload["protection"].(map[string]any); ok {
			protection = nested
		}
		parsed = append(parsed, parsedBranchProtectionImport{
			branchName:   branchName,
			sourceNodeID: item.SourceNodeID,
			sourceURL:    importFirstNonEmpty(item.SourceURL, stringFromAny(payload["url"])),
			protection:   protection,
		})
	}
	return parsed, nil
}

func branchProtectionNames(entries []parsedBranchProtectionImport) []string {
	out := make([]string, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.branchName)
	}
	return out
}

func branchProtectionsByNameTx(tx *gorm.DB, repoID uint, branchNames []string) (map[string]db.BranchProtection, error) {
	branchNames = uniqueStrings(branchNames)
	if len(branchNames) == 0 {
		return map[string]db.BranchProtection{}, nil
	}
	var rows []db.BranchProtection
	if err := tx.Where("repository_id = ? AND branch_name IN ?", repoID, branchNames).Find(&rows).Error; err != nil {
		return nil, err
	}
	out := make(map[string]db.BranchProtection, len(rows))
	for _, row := range rows {
		out[row.BranchName] = row
	}
	return out, nil
}

type parsedWorkflowImport struct {
	source       ghWorkflow
	path         string
	sourceID     string
	sourceNodeID string
	sourceURL    string
}

func parseWorkflowBatchItems(items []db.ImportStagingItem) ([]parsedWorkflowImport, error) {
	parsed := make([]parsedWorkflowImport, 0, len(items))
	seenSourceIDs := map[string]struct{}{}
	seenPaths := map[string]struct{}{}
	for _, item := range items {
		var source ghWorkflow
		if err := json.Unmarshal([]byte(item.PayloadJSON), &source); err != nil {
			return nil, fmt.Errorf("%w: decode workflow payload: %v", ErrValidation, err)
		}
		path := strings.TrimSpace(source.Path)
		if path == "" {
			return nil, fmt.Errorf("%w: workflow path is required", ErrValidation)
		}
		if _, ok := seenPaths[path]; ok {
			return nil, fmt.Errorf("%w: duplicate workflow path %q in batch", ErrValidation, path)
		}
		seenPaths[path] = struct{}{}
		sourceID := importItemSourceID(item, source.ID, path)
		if sourceID == "" {
			return nil, fmt.Errorf("%w: workflow source_id is required", ErrValidation)
		}
		if _, ok := seenSourceIDs[sourceID]; ok {
			return nil, fmt.Errorf("%w: duplicate workflow source_id %q in batch", ErrValidation, sourceID)
		}
		seenSourceIDs[sourceID] = struct{}{}
		parsed = append(parsed, parsedWorkflowImport{
			source:       source,
			path:         path,
			sourceID:     sourceID,
			sourceNodeID: importFirstNonEmpty(item.SourceNodeID, source.NodeID),
			sourceURL:    importFirstNonEmpty(item.SourceURL, source.HTMLURL),
		})
	}
	return parsed, nil
}

func workflowPaths(entries []parsedWorkflowImport) []string {
	out := make([]string, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.path)
	}
	return out
}

func workflowsByPathTx(tx *gorm.DB, repoID uint, paths []string) (map[string]db.Workflow, error) {
	paths = uniqueStrings(paths)
	if len(paths) == 0 {
		return map[string]db.Workflow{}, nil
	}
	var rows []db.Workflow
	if err := tx.Where("repository_id = ? AND path IN ?", repoID, paths).Find(&rows).Error; err != nil {
		return nil, err
	}
	out := make(map[string]db.Workflow, len(rows))
	for _, row := range rows {
		out[row.Path] = row
	}
	return out, nil
}

func applyOneImportReleaseTx(tx *gorm.DB, runner *githubImportRunner, target db.Repository, item db.ImportStagingItem, source ghRelease) (db.Release, error) {
	author, err := runner.ensureSourceUserTx(tx, source.Author)
	if err != nil {
		return db.Release{}, err
	}
	tagName := strings.TrimSpace(source.TagName)
	if tagName == "" {
		return db.Release{}, fmt.Errorf("%w: release tag_name is required", ErrValidation)
	}
	release := db.Release{
		RepositoryID: target.ID,
		TagName:      tagName,
		Name:         source.Name,
		Body:         db.LargeText(source.Body),
		Draft:        source.Draft,
		PreRelease:   source.Prerelease,
		AuthorID:     author.ID,
		CreatedAt:    nonZeroTime(source.CreatedAt, time.Now().UTC()),
		PublishedAt:  source.PublishedAt,
	}
	if err := tx.Clauses(clause.OnConflict{
		Columns:   []clause.Column{{Name: "repository_id"}, {Name: "tag_name"}},
		DoUpdates: clause.AssignmentColumns([]string{"name", "body", "draft", "pre_release", "author_id", "published_at"}),
	}).Create(&release).Error; err != nil {
		return db.Release{}, err
	}
	if err := tx.First(&release, "repository_id = ? AND tag_name = ?", target.ID, tagName).Error; err != nil {
		return db.Release{}, err
	}
	sourceID := importItemSourceID(item, source.ID, tagName)
	sourceURL := importFirstNonEmpty(item.SourceURL, source.HTMLURL)
	if err := upsertImportMappingTx(tx, runner.job.ID, importTypeRelease, sourceID, importFirstNonEmpty(item.SourceNodeID, source.NodeID), nil, sourceURL, "release", strconv.FormatUint(uint64(release.ID), 10), nil, runner.targetReleaseURL(release.ID), map[string]any{
		"tag_name": tagName,
	}); err != nil {
		return db.Release{}, err
	}
	return release, nil
}

func resolveReviewCommentRepliesTx(tx *gorm.DB, runner *githubImportRunner, comments []parsedPullRequestReviewCommentImport) error {
	sourceIDs := make([]string, 0, len(comments)*2)
	for _, entry := range comments {
		if entry.source.InReplyToID == 0 {
			continue
		}
		sourceIDs = append(sourceIDs, entry.sourceID, strconv.FormatInt(entry.source.InReplyToID, 10))
	}
	if len(sourceIDs) == 0 {
		return nil
	}
	targetIDs, err := runner.mappingTargetIDsTx(tx, importTypePullReviewComment, sourceIDs)
	if err != nil {
		return err
	}
	parentByChild := map[uint]uint{}
	for _, entry := range comments {
		if entry.source.InReplyToID == 0 {
			continue
		}
		childID, childOK := targetIDs[entry.sourceID]
		parentID, parentOK := targetIDs[strconv.FormatInt(entry.source.InReplyToID, 10)]
		if !childOK || !parentOK {
			continue
		}
		parentByChild[childID] = parentID
	}
	if len(parentByChild) == 0 {
		return nil
	}
	childIDs := make([]uint, 0, len(parentByChild))
	for childID := range parentByChild {
		childIDs = append(childIDs, childID)
	}
	sort.Slice(childIDs, func(i, j int) bool { return childIDs[i] < childIDs[j] })

	var sql strings.Builder
	args := make([]any, 0, len(childIDs)*2+1)
	sql.WriteString("UPDATE pr_review_comments SET in_reply_to_id = CASE id ")
	for _, childID := range childIDs {
		sql.WriteString("WHEN ? THEN ? ")
		args = append(args, childID, parentByChild[childID])
	}
	sql.WriteString("END WHERE id IN ?")
	args = append(args, childIDs)
	return tx.Exec(sql.String(), args...).Error
}

func stagedImportItemsTx(tx *gorm.DB, batch db.ImportBatch) ([]db.ImportStagingItem, error) {
	var items []db.ImportStagingItem
	err := tx.Where("import_batch_id = ? AND status = ?", batch.ID, ImportStagingStatusStaged).
		Order("item_index asc").
		Find(&items).Error
	return items, err
}

func importBatchRunner(s *Service, job db.ImportJob) *githubImportRunner {
	return &githubImportRunner{s: s, job: job}
}

func importSourceLogin(source *ghUser) string {
	if source != nil && strings.TrimSpace(source.Login) != "" {
		return strings.TrimSpace(source.Login)
	}
	return "github-ghost"
}

// githubIdentityProvider scopes a GitHub numeric user ID to its source host.
// A GitHub Enterprise account and a GitHub.com account can have the same
// numeric subject and must never be considered the same person.
func (r *githubImportRunner) githubIdentityProvider() string {
	return githubIdentityProviderForAPIURL(r.job.SourceAPIURL)
}

func githubIdentityProviderForAPIURL(sourceAPIURL string) string {
	parsed, err := url.Parse(strings.TrimSpace(sourceAPIURL))
	host := ""
	if err == nil {
		host = strings.ToLower(strings.TrimSpace(parsed.Hostname()))
	}
	if host == "" || host == "api.github.com" || host == "github.com" {
		return "github"
	}
	digest := sha256.Sum256([]byte(host))
	return "github-" + hex.EncodeToString(digest[:])[:16]
}

func importSourceIdentitySubject(source *ghUser, login string) string {
	if source != nil && source.ID > 0 {
		return strconv.FormatInt(source.ID, 10)
	}
	// GitHub ghost users do not have a durable numeric subject. Keep a stable
	// placeholder for attribution, but never make it eligible for a claim.
	return "login:" + strings.ToLower(strings.TrimSpace(login))
}

func importPlaceholderLogin(provider, subject string) string {
	digest := sha256.Sum256([]byte(provider + "\x00" + subject))
	return "imported-github-" + hex.EncodeToString(digest[:])[:20]
}

func (r *githubImportRunner) ensureSourceUsersTx(tx *gorm.DB, sources []*ghUser) (map[string]db.User, error) {
	sourceByLogin := map[string]*ghUser{}
	for _, source := range sources {
		login := importSourceLogin(source)
		if _, ok := sourceByLogin[login]; !ok {
			sourceByLogin[login] = source
		}
	}
	if len(sourceByLogin) == 0 {
		return map[string]db.User{}, nil
	}
	logins := make([]string, 0, len(sourceByLogin))
	for login := range sourceByLogin {
		logins = append(logins, login)
	}
	sort.Strings(logins)

	provider := r.githubIdentityProvider()
	subjectByLogin := make(map[string]string, len(logins))
	sourceIDByLogin := make(map[string]string, len(logins))
	sourceURLByLogin := make(map[string]string, len(logins))
	claimableSubjects := make([]string, 0, len(logins))
	for _, login := range logins {
		source := sourceByLogin[login]
		subject := importSourceIdentitySubject(source, login)
		subjectByLogin[login] = subject
		if source != nil && source.ID > 0 {
			sourceIDByLogin[login] = strconv.FormatInt(source.ID, 10)
			sourceURLByLogin[login] = source.HTMLURL
			claimableSubjects = append(claimableSubjects, subject)
		}
	}

	var identities []db.UserIdentity
	if len(claimableSubjects) > 0 {
		if err := tx.Where("provider = ? AND subject IN ?", provider, claimableSubjects).Find(&identities).Error; err != nil {
			return nil, err
		}
	}
	identityBySubject := make(map[string]db.UserIdentity, len(identities))
	userIDs := make([]uint, 0, len(identities))
	for _, identity := range identities {
		identityBySubject[identity.Subject] = identity
		userIDs = append(userIDs, identity.UserID)
	}

	var existingSources []db.ImportSourceUser
	if err := tx.Where("import_job_id = ? AND source_login IN ?", r.job.ID, logins).Find(&existingSources).Error; err != nil {
		return nil, err
	}
	sourceRecordByLogin := make(map[string]db.ImportSourceUser, len(existingSources))
	for _, source := range existingSources {
		sourceRecordByLogin[source.SourceLogin] = source
		if source.TargetUserID != nil {
			userIDs = append(userIDs, *source.TargetUserID)
		}
	}

	usersByID := map[uint]db.User{}
	if len(userIDs) > 0 {
		var users []db.User
		if err := tx.Where("id IN ?", userIDs).Find(&users).Error; err != nil {
			return nil, err
		}
		for _, user := range users {
			usersByID[user.ID] = user
		}
	}
	eligibleUserIDs, err := eligibleImportAttributionUserIDsTx(tx, r.job, userIDs)
	if err != nil {
		return nil, err
	}

	now := time.Now().UTC()
	placeholderByLogin := make(map[string]string, len(logins))
	missing := make([]db.User, 0, len(logins))
	for _, login := range logins {
		if identity, ok := identityBySubject[subjectByLogin[login]]; ok {
			if user, found := usersByID[identity.UserID]; found && user.Status == db.UserStatusActive && eligibleUserIDs[user.ID] {
				continue
			}
		}
		if prior, ok := sourceRecordByLogin[login]; ok && prior.State == "claimed" && prior.TargetUserID != nil {
			if user, found := usersByID[*prior.TargetUserID]; found && user.Status == db.UserStatusActive {
				continue
			}
		}
		placeholder := importPlaceholderLogin(provider, subjectByLogin[login])
		placeholderByLogin[login] = placeholder
		missing = append(missing, db.User{
			Login:     placeholder,
			Name:      login,
			Type:      importUserType(sourceByLogin[login]),
			Status:    db.UserStatusSuspended,
			UserKind:  db.UserKindHuman,
			CreatedAt: now,
			UpdatedAt: now,
		})
	}
	if len(missing) > 0 {
		if err := tx.Clauses(clause.OnConflict{
			Columns:   []clause.Column{{Name: "login"}},
			DoNothing: true,
		}).CreateInBatches(&missing, 1000).Error; err != nil {
			return nil, err
		}
		placeholderLogins := make([]string, 0, len(placeholderByLogin))
		for _, placeholder := range placeholderByLogin {
			placeholderLogins = append(placeholderLogins, placeholder)
		}
		var placeholders []db.User
		if err := tx.Where("login IN ?", placeholderLogins).Find(&placeholders).Error; err != nil {
			return nil, err
		}
		for _, user := range placeholders {
			usersByID[user.ID] = user
		}
	}

	usersByLogin := make(map[string]db.User, len(logins))
	sourceUsers := make([]db.ImportSourceUser, 0, len(logins))
	for _, login := range logins {
		source := sourceByLogin[login]
		metadata, err := marshalJSON(map[string]any{
			"source_type": typeOrEmpty(source),
			"provider":    provider,
			"subject":     subjectByLogin[login],
		})
		if err != nil {
			return nil, err
		}
		prior := sourceRecordByLogin[login]
		state := "pending_claim"
		invitationState := "none"
		var targetUserID *uint
		placeholderLogin := placeholderByLogin[login]
		if placeholderLogin == "" {
			placeholderLogin = prior.PlaceholderLogin
		}
		if identity, ok := identityBySubject[subjectByLogin[login]]; ok {
			if user, found := usersByID[identity.UserID]; found && user.Status == db.UserStatusActive && eligibleUserIDs[user.ID] {
				usersByLogin[login] = user
				targetUserID = &user.ID
				state = "claimed"
			}
		}
		if targetUserID == nil && prior.State == "claimed" && prior.TargetUserID != nil {
			if user, found := usersByID[*prior.TargetUserID]; found && user.Status == db.UserStatusActive {
				usersByLogin[login] = user
				targetUserID = &user.ID
				state = "claimed"
			}
		}
		if targetUserID == nil {
			if prior.InvitationState == "pending" {
				state = "pending_invitation"
				invitationState = "pending"
			}
			if prior.InvitationState == "rejected" {
				state = "rejected"
				invitationState = "rejected"
			}
			var placeholder db.User
			if err := tx.Where("login = ?", placeholderLogin).Take(&placeholder).Error; err != nil {
				return nil, err
			}
			usersByLogin[login] = placeholder
		}
		if targetUserID != nil {
			invitationState = "accepted"
		}
		sourceUsers = append(sourceUsers, db.ImportSourceUser{
			ImportJobID:      r.job.ID,
			SourceLogin:      login,
			SourceID:         sourceIDByLogin[login],
			SourceURL:        sourceURLByLogin[login],
			TargetUserID:     targetUserID,
			PlaceholderLogin: placeholderLogin,
			State:            state,
			InvitationState:  invitationState,
			MetadataJSON:     db.LargeText(metadata),
		})
	}
	if len(sourceUsers) > 0 {
		if err := tx.Clauses(clause.OnConflict{
			Columns: []clause.Column{{Name: "import_job_id"}, {Name: "source_login"}},
			DoUpdates: clause.AssignmentColumns([]string{
				"source_id", "source_url", "target_user_id", "placeholder_login", "state", "invitation_state", "metadata_json", "updated_at",
			}),
		}).CreateInBatches(&sourceUsers, 1000).Error; err != nil {
			return nil, err
		}
	}
	return usersByLogin, nil
}

func (r *githubImportRunner) mappingTargetIDsTx(tx *gorm.DB, sourceType string, sourceIDs []string) (map[string]uint, error) {
	unique := make([]string, 0, len(sourceIDs))
	seen := map[string]struct{}{}
	for _, sourceID := range sourceIDs {
		sourceID = strings.TrimSpace(sourceID)
		if sourceID == "" {
			continue
		}
		if _, ok := seen[sourceID]; ok {
			continue
		}
		seen[sourceID] = struct{}{}
		unique = append(unique, sourceID)
	}
	if len(unique) == 0 {
		return map[string]uint{}, nil
	}
	var mappings []db.ImportItemMapping
	if err := tx.Select("source_id", "target_id").
		Where("import_job_id = ? AND source_type = ? AND source_id IN ?", r.job.ID, sourceType, unique).
		Find(&mappings).Error; err != nil {
		return nil, err
	}
	out := map[string]uint{}
	for _, mapping := range mappings {
		id, err := strconv.ParseUint(mapping.TargetID, 10, 64)
		if err != nil || id == 0 {
			continue
		}
		out[mapping.SourceID] = uint(id)
	}
	return out, nil
}

type issueLabelJoin struct {
	IssueID uint `gorm:"column:issue_id"`
	LabelID uint `gorm:"column:label_id"`
}

func (issueLabelJoin) TableName() string {
	return "issue_labels"
}

type prLabelJoin struct {
	PullRequestID uint `gorm:"column:pull_request_id"`
	LabelID       uint `gorm:"column:label_id"`
}

func (prLabelJoin) TableName() string {
	return "pr_labels"
}

func (r *githubImportRunner) ensureLabelsTx(tx *gorm.DB, repoID uint, sourceLabels []ghLabel) (map[string]db.Label, error) {
	sourceByName := map[string]ghLabel{}
	for _, source := range sourceLabels {
		name := strings.TrimSpace(source.Name)
		if name == "" {
			continue
		}
		source.Name = name
		sourceByName[name] = source
	}
	if len(sourceByName) == 0 {
		return map[string]db.Label{}, nil
	}
	names := make([]string, 0, len(sourceByName))
	for name := range sourceByName {
		names = append(names, name)
	}
	sort.Strings(names)

	rows := make([]db.Label, 0, len(names))
	for _, name := range names {
		source := sourceByName[name]
		rows = append(rows, db.Label{
			RepositoryID: repoID,
			Name:         name,
			Color:        normalizeHexColor(source.Color),
			Description:  source.Description,
			Default:      source.Default,
		})
	}
	if err := tx.Clauses(clause.OnConflict{
		Columns:   []clause.Column{{Name: "repository_id"}, {Name: "name"}},
		DoUpdates: clause.AssignmentColumns([]string{"color", "description", "default", "updated_at"}),
	}).CreateInBatches(&rows, 1000).Error; err != nil {
		return nil, err
	}

	var labels []db.Label
	if err := tx.Where("repository_id = ? AND name IN ?", repoID, names).Find(&labels).Error; err != nil {
		return nil, err
	}
	out := make(map[string]db.Label, len(labels))
	for _, label := range labels {
		out[label.Name] = label
	}

	mappings := make([]db.ImportItemMapping, 0, len(names))
	for _, name := range names {
		label, ok := out[name]
		if !ok {
			return nil, fmt.Errorf("%w: label %q was not created", ErrInvalidState, name)
		}
		source := sourceByName[name]
		sourceID := importFirstNonEmpty(sourceIDFromInt64(source.ID), name)
		mappings = append(mappings, db.ImportItemMapping{
			ImportJobID:  r.job.ID,
			SourceType:   importTypeLabel,
			SourceID:     sourceID,
			SourceNodeID: source.NodeID,
			SourceURL:    source.URL,
			TargetType:   "label",
			TargetID:     strconv.FormatUint(uint64(label.ID), 10),
			TargetURL:    r.targetRepoURL() + "/labels/" + url.PathEscape(label.Name),
		})
	}
	if err := bulkUpsertImportMappingsTx(tx, mappings); err != nil {
		return nil, err
	}
	return out, nil
}

func (r *githubImportRunner) replaceIssueLabelsBulkTx(tx *gorm.DB, repoID uint, labelsByIssueID map[uint][]ghLabel) error {
	if len(labelsByIssueID) == 0 {
		return nil
	}
	issueIDs := make([]uint, 0, len(labelsByIssueID))
	allLabels := make([]ghLabel, 0)
	for issueID, labels := range labelsByIssueID {
		issueIDs = append(issueIDs, issueID)
		allLabels = append(allLabels, labels...)
	}
	sort.Slice(issueIDs, func(i, j int) bool { return issueIDs[i] < issueIDs[j] })
	if err := tx.Where("issue_id IN ?", issueIDs).Delete(&issueLabelJoin{}).Error; err != nil {
		return err
	}
	labelsByName, err := r.ensureLabelsTx(tx, repoID, allLabels)
	if err != nil {
		return err
	}
	joins := make([]issueLabelJoin, 0)
	seen := map[string]struct{}{}
	for _, issueID := range issueIDs {
		for _, source := range labelsByIssueID[issueID] {
			label, ok := labelsByName[strings.TrimSpace(source.Name)]
			if !ok {
				continue
			}
			key := strconv.FormatUint(uint64(issueID), 10) + ":" + strconv.FormatUint(uint64(label.ID), 10)
			if _, ok := seen[key]; ok {
				continue
			}
			seen[key] = struct{}{}
			joins = append(joins, issueLabelJoin{IssueID: issueID, LabelID: label.ID})
		}
	}
	if len(joins) == 0 {
		return nil
	}
	return tx.CreateInBatches(&joins, 1000).Error
}

func (r *githubImportRunner) replacePRLabelsBulkTx(tx *gorm.DB, repoID uint, labelsByPRID map[uint][]ghLabel) error {
	if len(labelsByPRID) == 0 {
		return nil
	}
	prIDs := make([]uint, 0, len(labelsByPRID))
	allLabels := make([]ghLabel, 0)
	for prID, labels := range labelsByPRID {
		prIDs = append(prIDs, prID)
		allLabels = append(allLabels, labels...)
	}
	sort.Slice(prIDs, func(i, j int) bool { return prIDs[i] < prIDs[j] })
	if err := tx.Where("pull_request_id IN ?", prIDs).Delete(&prLabelJoin{}).Error; err != nil {
		return err
	}
	labelsByName, err := r.ensureLabelsTx(tx, repoID, allLabels)
	if err != nil {
		return err
	}
	joins := make([]prLabelJoin, 0)
	seen := map[string]struct{}{}
	for _, prID := range prIDs {
		for _, source := range labelsByPRID[prID] {
			label, ok := labelsByName[strings.TrimSpace(source.Name)]
			if !ok {
				continue
			}
			key := strconv.FormatUint(uint64(prID), 10) + ":" + strconv.FormatUint(uint64(label.ID), 10)
			if _, ok := seen[key]; ok {
				continue
			}
			seen[key] = struct{}{}
			joins = append(joins, prLabelJoin{PullRequestID: prID, LabelID: label.ID})
		}
	}
	if len(joins) == 0 {
		return nil
	}
	return tx.CreateInBatches(&joins, 1000).Error
}

func newImportItemMapping(jobID uint, sourceType, sourceID, sourceNodeID string, sourceNumber *int, sourceURL, targetType, targetID string, targetNumber *int, targetURL string, metadata any) (db.ImportItemMapping, error) {
	metadataJSON := ""
	if metadata != nil {
		encoded, err := marshalJSON(metadata)
		if err != nil {
			return db.ImportItemMapping{}, err
		}
		metadataJSON = encoded
	}
	return db.ImportItemMapping{
		ImportJobID:  jobID,
		SourceType:   sourceType,
		SourceID:     sourceID,
		SourceNodeID: sourceNodeID,
		SourceNumber: sourceNumber,
		SourceURL:    sourceURL,
		TargetType:   targetType,
		TargetID:     targetID,
		TargetNumber: targetNumber,
		TargetURL:    targetURL,
		MetadataJSON: db.LargeText(metadataJSON),
	}, nil
}

func bulkUpsertImportMappingsTx(tx *gorm.DB, mappings []db.ImportItemMapping) error {
	if len(mappings) == 0 {
		return nil
	}
	return tx.Clauses(clause.OnConflict{
		Columns: []clause.Column{{Name: "import_job_id"}, {Name: "source_type"}, {Name: "source_id"}},
		DoUpdates: clause.AssignmentColumns([]string{
			"source_node_id", "source_number", "source_url", "target_type", "target_id", "target_number", "target_url", "metadata_json", "updated_at",
		}),
	}).CreateInBatches(&mappings, 1000).Error
}

func uniqueInts(values []int) []int {
	seen := map[int]struct{}{}
	out := make([]int, 0, len(values))
	for _, value := range values {
		if value <= 0 {
			continue
		}
		if _, ok := seen[value]; ok {
			continue
		}
		seen[value] = struct{}{}
		out = append(out, value)
	}
	sort.Ints(out)
	return out
}

func importPullRequestByNumberTx(tx *gorm.DB, repoID uint, cache map[int]db.PullRequest, number int) (db.PullRequest, error) {
	if number <= 0 {
		return db.PullRequest{}, fmt.Errorf("%w: pull request number is required", ErrValidation)
	}
	if pr, ok := cache[number]; ok {
		return pr, nil
	}
	var pr db.PullRequest
	if err := tx.First(&pr, "repository_id = ? AND number = ?", repoID, number).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return db.PullRequest{}, fmt.Errorf("%w: pull request #%d must be imported before reviews/comments", ErrInvalidState, number)
		}
		return db.PullRequest{}, err
	}
	cache[number] = pr
	return pr, nil
}

func importItemNumber(item db.ImportStagingItem, fallback int, objectType string) (int, error) {
	if fallback > 0 {
		return fallback, nil
	}
	if item.SourceNumber != nil && *item.SourceNumber > 0 {
		return *item.SourceNumber, nil
	}
	return 0, fmt.Errorf("%w: %s source_number is required", ErrValidation, objectType)
}

func importItemSourceID(item db.ImportStagingItem, id int64, fallback string) string {
	return importFirstNonEmpty(item.SourceID, sourceIDFromInt64(id), fallback)
}

func sourceIDFromInt64(id int64) string {
	if id == 0 {
		return ""
	}
	return strconv.FormatInt(id, 10)
}

func stringFromAny(value any) string {
	switch v := value.(type) {
	case string:
		return strings.TrimSpace(v)
	case float64:
		if v == float64(int64(v)) {
			return strconv.FormatInt(int64(v), 10)
		}
		return strconv.FormatFloat(v, 'f', -1, 64)
	default:
		return ""
	}
}

func importTargetRepoTx(tx *gorm.DB, job db.ImportJob) (db.Repository, error) {
	var repo db.Repository
	if job.TargetRepoID != nil {
		if err := tx.Take(&repo, "id = ?", *job.TargetRepoID).Error; err == nil {
			return repo, nil
		} else if !errors.Is(err, gorm.ErrRecordNotFound) {
			return db.Repository{}, err
		}
	}
	fullName := job.TargetOwner + "/" + job.TargetRepo
	if err := tx.Take(&repo, "full_name = ?", fullName).Error; err != nil {
		return db.Repository{}, wrapErr(err)
	}
	if job.TargetRepoID == nil || *job.TargetRepoID != repo.ID {
		if err := tx.Model(&db.ImportJob{}).Where("id = ?", job.ID).Update("target_repo_id", repo.ID).Error; err != nil {
			return db.Repository{}, err
		}
	}
	return repo, nil
}