package kb

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net"
	"net/url"
	"regexp"
	"strings"
	"time"

	"gorm.io/gorm"

	"github.com/tidbcloud/AgentTeam/internal/core/secrets"
	"github.com/tidbcloud/AgentTeam/internal/observability/logging"
	"github.com/tidbcloud/AgentTeam/internal/service/resource/repos"
	"github.com/tidbcloud/AgentTeam/internal/store/automation"
	"github.com/tidbcloud/AgentTeam/internal/store/kb"
	reposstore "github.com/tidbcloud/AgentTeam/internal/store/resource/repos"
)

const (
	// KBJobTypeSyncSourceMirror enqueues a manual or scheduled external Git pull mirror sync.
	KBJobTypeSyncSourceMirror = "sync_source_mirror"

	KBSourceMirrorStageQueued      = "queued"
	KBSourceMirrorStageValidating  = "validating_remote"
	KBSourceMirrorStageSyncingAGS  = "syncing_ags_repo"
	KBSourceMirrorStageReconciling = "reconciling_kb_index"
	KBSourceMirrorStageReady       = "ready"
	KBSourceMirrorStageFailed      = "failed"

	KBSourceMirrorTriggerManual    = "manual"
	KBSourceMirrorTriggerScheduled = "scheduled"

	// Lower values are claimed first. Initial imports are user-blocking while
	// refreshes operate on already usable sources and may wait behind them.
	KBSourceMirrorJobPriorityInitialImport = 10
	KBSourceMirrorJobPriorityRefresh       = 100

	defaultSourceMirrorTimeout          = 30 * time.Minute
	defaultSourceMirrorPreflightTimeout = 30 * time.Second
	defaultSourceMirrorLeaseBuffer      = 15 * time.Minute
	defaultSourceMirrorGitMaxAttempts   = 3
	defaultSourceMirrorGitRetryDelay    = 2 * time.Second
	defaultGitOutputLimit               = 8 * 1024

	// kbMirrorCredentialSecretName is the fixed automation secret name used to
	// store the access token for a KB source mirror. One source mirror needs at
	// most one credential, so a fixed name avoids exposing secret naming to the
	// user.
	kbMirrorCredentialSecretName = "kb_mirror_credential"
)

// SourceMirrorSyncJobPayload is intentionally narrow: it identifies the source,
// requested ref/status, trigger, requester, and correlation id. Mirror secrets
// stay behind credential_ref in source mirror metadata and must not be copied
// into job payloads.
type SourceMirrorSyncJobPayload struct {
	SourceID                       string `json:"source_id"`
	Ref                            string `json:"ref,omitempty"`
	Status                         string `json:"status,omitempty"`
	Trigger                        string `json:"trigger,omitempty"`
	RequestedBy                    string `json:"requested_by,omitempty"`
	TraceID                        string `json:"trace_id,omitempty"`
	ExpectedCurrentSourceCommitSHA string `json:"expected_current_source_commit_sha,omitempty"`
	SourceRevisionFenceSet         bool   `json:"source_revision_fence_set,omitempty"`
}

type pendingMirrorAccessToken struct {
	secret automation.AutomationSecret
}

// prepareMirrorAccessToken encrypts an access token without changing durable
// state. Callers updating an existing mirror persist it only after locking the
// source and validating the mirror generation.
func (s *KBService) prepareMirrorAccessToken(workspaceID, repoOwner, repoName, accessToken, subject string, now time.Time) (pendingMirrorAccessToken, error) {
	if err := s.ensureMirrorAccessTokenStorageConfigured(); err != nil {
		return pendingMirrorAccessToken{}, err
	}
	codec, err := secrets.NewAESGCMCodec(s.automationCredentialsKey)
	if err != nil {
		return pendingMirrorAccessToken{}, invalidArgument("Automation secret storage is not configured on server")
	}
	ciphertext, err := codec.Seal(strings.TrimSpace(accessToken))
	if err != nil {
		return pendingMirrorAccessToken{}, fmt.Errorf("encrypt mirror access token: %w", err)
	}
	return pendingMirrorAccessToken{secret: automation.AutomationSecret{
		WorkspaceID:     strings.TrimSpace(workspaceID),
		RepositoryOwner: strings.TrimSpace(repoOwner),
		RepositoryName:  strings.TrimSpace(repoName),
		Name:            kbMirrorCredentialSecretName,
		ValueCiphertext: ciphertext,
		CreatedBy:       strings.TrimSpace(subject),
		UpdatedBy:       strings.TrimSpace(subject),
		CreatedAt:       now,
		UpdatedAt:       now,
	}}, nil
}

func persistMirrorAccessToken(ctx context.Context, store *automation.Store, pending pendingMirrorAccessToken) (string, error) {
	_, err := store.UpsertAutomationSecret(ctx, pending.secret)
	if err != nil {
		return "", fmt.Errorf("store mirror access token secret: %w", err)
	}
	return "automation-secret://" + kbMirrorCredentialSecretName, nil
}

// storeMirrorAccessToken is used while creating a source, before a source row
// exists to lock. Existing-mirror mutations instead persist the prepared token
// inside saveSourceMirrorUpdate's source-scoped transaction.
func (s *KBService) storeMirrorAccessToken(ctx context.Context, workspaceID, repoOwner, repoName, accessToken, subject string) (string, error) {
	pending, err := s.prepareMirrorAccessToken(workspaceID, repoOwner, repoName, accessToken, subject, s.now())
	if err != nil {
		return "", err
	}
	return persistMirrorAccessToken(ctx, s.automationSecretStore, pending)
}

func (s *KBService) ensureMirrorAccessTokenStorageConfigured() error {
	if s.automationSecretStore == nil {
		return invalidArgument("Automation secret storage is not configured on server")
	}
	if len(s.automationCredentialsKey) != 32 {
		return invalidArgument("Automation secret storage is not configured on server")
	}
	return nil
}

// findSourceMirrorRepository resolves the mirror resource repository for a KB
// source from the resource repository store.
func (s *KBService) findSourceMirrorRepository(ctx context.Context, workspaceID, sourceID string) (reposstore.ResourceRepository, error) {
	if s.repositoryStore == nil {
		return reposstore.ResourceRepository{}, invalidArgument("resource repository store is not configured")
	}
	repos, err := s.repositoryStore.ListForResource(ctx, workspaceID, reposstore.ResourceTypeKBSource, sourceID)
	if err != nil {
		return reposstore.ResourceRepository{}, fmt.Errorf("load source repositories: %w", err)
	}
	if mirrorRepo := findMirrorResourceRepository(repos); mirrorRepo != nil {
		return *mirrorRepo, nil
	}
	return reposstore.ResourceRepository{}, invalidArgument("mirror repository mapping was not found for source")
}

type sourceMirrorSyncPayload struct {
	Kind                           string `json:"kind,omitempty"`
	SourceID                       string `json:"source_id,omitempty"`
	Ref                            string `json:"ref,omitempty"`
	Stage                          string `json:"stage,omitempty"`
	Message                        string `json:"message,omitempty"`
	Trigger                        string `json:"trigger,omitempty"`
	RequestedBy                    string `json:"requested_by,omitempty"`
	TraceID                        string `json:"trace_id,omitempty"`
	ExpectedCurrentSourceCommitSHA string `json:"expected_current_source_commit_sha,omitempty"`
	SourceRevisionFenceSet         bool   `json:"source_revision_fence_set,omitempty"`
	FailedStage                    string `json:"failed_stage,omitempty"`
	ErrorCode                      string `json:"error_code,omitempty"`
	Retryable                      *bool  `json:"retryable,omitempty"`
	GitOperation                   string `json:"git_operation,omitempty"`
	GitAttempt                     int    `json:"git_attempt,omitempty"`
	GitMaxAttempts                 int    `json:"git_max_attempts,omitempty"`
}

// SourceMirrorSyncRequest describes one external Git pull mirror execution.
type SourceMirrorSyncRequest struct {
	WorkspaceID          string
	KnowledgeBaseID      string
	Source               kb.Source
	Mirror               kb.SourceMirror
	RemoteURL            string
	DefaultRef           string
	CredentialRef        string
	CredentialRepository repos.RepositoryIdentity
	Target               repos.RepositoryIdentity
	AllowPrivateHosts    bool
	OnGitProgress        func(SourceMirrorGitProgress) error
}

// SourceMirrorSyncResult captures safe, non-secret sync output.
type SourceMirrorSyncResult struct {
	MirroredCommit     string
	ResolvedDefaultRef string
	DefaultRefAuto     bool
}

// SourceMirrorCredential contains resolved process-only credentials for one
// source mirror sync. Values must never be persisted, logged, copied into job
// payloads, or returned by APIs.
type SourceMirrorCredential struct {
	Username            string
	Password            string
	AuthorizationHeader string
}

// SourceMirrorCredentialRequest scopes one runtime credential_ref lookup.
type SourceMirrorCredentialRequest struct {
	WorkspaceID     string
	KnowledgeBaseID string
	SourceID        string
	CredentialRef   string
	RemoteURL       string
	UpstreamType    string
	RepositoryOwner string
	RepositoryName  string
}

// SourceMirrorCredentialResolver resolves an opaque credential_ref at sync
// runtime. Implementations should scope lookups to the workspace and target
// repository and return only process-local Git credentials.
type SourceMirrorCredentialResolver interface {
	ResolveSourceMirrorCredential(context.Context, SourceMirrorCredentialRequest) (SourceMirrorCredential, error)
}

// KBSourceMirrorRepositorySyncer syncs a validated external Git remote into an
// AGS source repository. Implementations must treat CredentialRef only as an
// opaque secret reference and must not log or persist resolved secret values.
type KBSourceMirrorRepositorySyncer interface {
	SyncSourceMirror(context.Context, SourceMirrorSyncRequest) (SourceMirrorSyncResult, error)
}

// SourceMirrorPreflightRequest describes a lightweight preflight check to
// verify that the remote URL and credential can access the upstream repository
// before any AGS repos are provisioned.
type SourceMirrorPreflightRequest struct {
	WorkspaceID       string
	KnowledgeBaseID   string
	SourceID          string
	RemoteURL         string
	DefaultRef        string
	UpstreamType      string
	CredentialRef     string
	AccessToken       string
	RepositoryOwner   string
	RepositoryName    string
	AllowPrivateHosts bool
}

// KBSourceMirrorPreflighter performs a lightweight ls-remote level check
// against an external Git remote before provisioning AGS repositories.
// This catches token permission errors and unreachable remotes early,
// preventing orphan AGS repos from partial provisioning failures.
type KBSourceMirrorPreflighter interface {
	PreflightSourceMirror(context.Context, SourceMirrorPreflightRequest) error
}

type SourceMirrorRunOptions struct {
	Limit         int
	LeaseOwner    string
	LeaseDuration time.Duration
	WorkspaceIDs  []string
}

type SourceMirrorRunResult struct {
	Claimed   int
	Succeeded int
	Failed    int
}

type SourceMirrorScheduleOptions struct {
	Limit        int
	WorkspaceIDs []string
}

type SourceMirrorScheduleResult struct {
	Scanned  int
	Enqueued int
	Failed   int
}

type SourceMirrorSyncListResult struct {
	Mirror kb.SourceMirror
	Jobs   []kb.Job
}

type SourceMirrorRemoteValidationOptions struct {
	AllowPrivateHosts bool
	Resolver          sourceMirrorHostResolver
}

type sourceMirrorParsedRemote struct {
	URL       *url.URL
	Host      string
	Path      string
	Sanitized string
	SCPLike   bool
}

type sourceMirrorHostResolver interface {
	LookupIP(context.Context, string) ([]net.IP, error)
}

type defaultSourceMirrorHostResolver struct{}

func (defaultSourceMirrorHostResolver) LookupIP(ctx context.Context, host string) ([]net.IP, error) {
	return net.DefaultResolver.LookupIP(ctx, "ip", host)
}

func encodeSourceMirrorSyncJobPayload(payload SourceMirrorSyncJobPayload) (string, error) {
	payload.SourceID = strings.TrimSpace(payload.SourceID)
	payload.Ref = strings.TrimSpace(payload.Ref)
	payload.Status = strings.TrimSpace(payload.Status)
	payload.Trigger = strings.TrimSpace(payload.Trigger)
	payload.RequestedBy = strings.TrimSpace(payload.RequestedBy)
	payload.TraceID = strings.TrimSpace(payload.TraceID)
	payload.ExpectedCurrentSourceCommitSHA = strings.TrimSpace(payload.ExpectedCurrentSourceCommitSHA)
	encoded, err := json.Marshal(payload)
	if err != nil {
		return "", err
	}
	return string(encoded), nil
}

func parseSourceMirrorSyncJobPayload(raw string) (SourceMirrorSyncJobPayload, error) {
	if strings.TrimSpace(raw) == "" {
		return SourceMirrorSyncJobPayload{}, nil
	}
	var payload SourceMirrorSyncJobPayload
	if err := json.Unmarshal([]byte(raw), &payload); err != nil {
		return SourceMirrorSyncJobPayload{}, invalidArgument("source mirror sync job payload must be valid JSON")
	}
	payload.SourceID = strings.TrimSpace(payload.SourceID)
	payload.Ref = strings.TrimSpace(payload.Ref)
	payload.Status = strings.TrimSpace(payload.Status)
	payload.Trigger = strings.TrimSpace(payload.Trigger)
	payload.RequestedBy = strings.TrimSpace(payload.RequestedBy)
	payload.TraceID = strings.TrimSpace(payload.TraceID)
	payload.ExpectedCurrentSourceCommitSHA = strings.TrimSpace(payload.ExpectedCurrentSourceCommitSHA)
	return payload, nil
}

func encodeSourceMirrorProgress(payload sourceMirrorSyncPayload) string {
	payload.Kind = "source_mirror_sync"
	return kbJSONString(payload)
}

func parseSourceMirrorProgress(raw string) (sourceMirrorSyncPayload, error) {
	if strings.TrimSpace(raw) == "" {
		return sourceMirrorSyncPayload{}, nil
	}
	var payload sourceMirrorSyncPayload
	if err := json.Unmarshal([]byte(raw), &payload); err != nil {
		return sourceMirrorSyncPayload{}, invalidArgument("source mirror sync progress must be valid JSON")
	}
	payload.SourceID = strings.TrimSpace(payload.SourceID)
	payload.Ref = strings.TrimSpace(payload.Ref)
	payload.Stage = strings.TrimSpace(payload.Stage)
	payload.Message = strings.TrimSpace(payload.Message)
	payload.Trigger = strings.TrimSpace(payload.Trigger)
	payload.RequestedBy = strings.TrimSpace(payload.RequestedBy)
	payload.TraceID = strings.TrimSpace(payload.TraceID)
	payload.FailedStage = strings.TrimSpace(payload.FailedStage)
	payload.ErrorCode = strings.TrimSpace(payload.ErrorCode)
	payload.GitOperation = strings.TrimSpace(payload.GitOperation)
	return payload, nil
}

func (s *KBService) RequestSourceMirrorSync(ctx context.Context, workspaceID string, subject string, kbID string, sourceID string, ref string) (kb.Job, error) {
	base, err := s.GetKnowledgeBase(ctx, workspaceID, subject, kbID)
	if err != nil {
		return kb.Job{}, err
	}
	if !roleCanEdit(s.knowledgeBaseRole(ctx, base, subject)) {
		return kb.Job{}, forbidden("knowledge base requires editor access")
	}
	return s.enqueueSourceMirrorSync(ctx, base.WorkspaceID, base.ID, sourceID, ref, subject, KBSourceMirrorStageQueued, true)
}

func (s *KBService) UpdateSourceMirror(ctx context.Context, workspaceID string, subject string, kbID string, sourceID string, input KBSourceMirrorMutation) (kb.SourceMirror, error) {
	base, err := s.GetKnowledgeBase(ctx, workspaceID, subject, kbID)
	if err != nil {
		return kb.SourceMirror{}, err
	}
	if !roleCanEdit(s.knowledgeBaseRole(ctx, base, subject)) {
		return kb.SourceMirror{}, forbidden("knowledge base requires editor access")
	}
	source, err := s.store.GetSource(ctx, base.WorkspaceID, base.ID, sourceID)
	if err != nil {
		return kb.SourceMirror{}, mapStoreNotFound(err)
	}
	native, err := s.kbSourceNativeActive(ctx, base.WorkspaceID, source.ID)
	if err != nil {
		return kb.SourceMirror{}, err
	}
	if native {
		return kb.SourceMirror{}, nativeKBSourceLegacyOperationConflict()
	}
	existing, err := s.store.GetSourceMirror(ctx, base.WorkspaceID, base.ID, source.ID)
	if err != nil {
		return kb.SourceMirror{}, mapStoreNotFound(err)
	}
	now := s.now()
	nextEnabled := existing.Enabled
	if input.Enabled != nil {
		nextEnabled = *input.Enabled
	}
	credentialInputProvided := strings.TrimSpace(input.CredentialRef) != "" ||
		strings.TrimSpace(input.AccessToken) != ""
	updatedDefaultRef := existing.DefaultRef
	if input.DefaultRefSet || strings.TrimSpace(input.DefaultRef) != "" {
		updatedDefaultRef = input.DefaultRef
	}
	updated, err := s.prepareSourceMirrorMutation(ctx, source, KBSourceMirrorMutation{
		UpstreamType:    firstNonEmptyString(input.UpstreamType, existing.UpstreamType),
		RemoteURL:       firstNonEmptyString(input.RemoteURL, existing.RemoteURL),
		CredentialRef:   input.CredentialRef,
		AccessToken:     input.AccessToken,
		DefaultRef:      updatedDefaultRef,
		DefaultRefSet:   input.DefaultRefSet,
		IntervalSeconds: input.IntervalSeconds,
		Enabled:         input.Enabled,
	}, subject, now)
	if err != nil {
		return kb.SourceMirror{}, err
	}
	if !input.DefaultRefSet {
		updated.DefaultRefExplicit = existing.DefaultRefExplicit
	}
	if strings.TrimSpace(input.CredentialRef) == "" && strings.TrimSpace(input.AccessToken) == "" {
		updated.CredentialRef = existing.CredentialRef
	}
	var pendingAccessToken *pendingMirrorAccessToken
	if strings.TrimSpace(input.AccessToken) != "" {
		mirrorRepo, err := s.findSourceMirrorRepository(ctx, base.WorkspaceID, source.ID)
		if err != nil {
			return kb.SourceMirror{}, err
		}
		pending, err := s.prepareMirrorAccessToken(base.WorkspaceID, mirrorRepo.RepoOwner, mirrorRepo.RepoName, input.AccessToken, subject, now)
		if err != nil {
			return kb.SourceMirror{}, err
		}
		pendingAccessToken = &pending
		updated.CredentialRef = "automation-secret://" + kbMirrorCredentialSecretName
	}
	shouldValidateCurrentCredential := nextEnabled || credentialInputProvided
	if strings.TrimSpace(input.AccessToken) == "" &&
		strings.TrimSpace(updated.CredentialRef) != "" &&
		shouldValidateCurrentCredential {
		requireOwner := strings.TrimSpace(input.CredentialRef) != ""
		if err := s.validatePersonalAccessTokenForMirrorMutation(ctx, source, updated, subject, requireOwner); err != nil {
			return kb.SourceMirror{}, err
		}
	}
	if input.IntervalSeconds == nil {
		updated.IntervalSeconds = existing.IntervalSeconds
		if input.Enabled == nil {
			updated.NextSyncAt = existing.NextSyncAt
		} else {
			updated.NextSyncAt = nextSourceMirrorSyncAt(updated, now)
		}
	}
	updated.SourceID = existing.SourceID
	updated.CreatedBy = existing.CreatedBy
	updated.CreatedAt = existing.CreatedAt
	updated.LastSyncAt = existing.LastSyncAt
	updated.LastSuccessAt = existing.LastSuccessAt
	updated.LastError = existing.LastError
	updated.LastMirroredCommit = existing.LastMirroredCommit
	disabling := input.Enabled != nil && !*input.Enabled
	// When the remote URL or default ref actually changes, the previously
	// synced commit and success timestamp belong to a different repository
	// or branch. Clear them so the UI does not show stale success state for
	// a remote that has not been synced yet.
	remoteChanged := strings.TrimSpace(updated.RemoteURL) != strings.TrimSpace(existing.RemoteURL)
	refChanged := strings.TrimSpace(updated.DefaultRef) != strings.TrimSpace(existing.DefaultRef)
	upstreamChanged := strings.TrimSpace(updated.UpstreamType) != strings.TrimSpace(existing.UpstreamType)
	credentialChanged := strings.TrimSpace(input.AccessToken) != "" ||
		strings.TrimSpace(updated.CredentialRef) != strings.TrimSpace(existing.CredentialRef)
	configChanged := remoteChanged || refChanged
	executionConfigChanged := configChanged || upstreamChanged || credentialChanged
	if configChanged && !disabling {
		updated.LastSuccessAt = nil
		updated.LastMirroredCommit = ""
		updated.SyncStatus = kb.SourceMirrorSyncStatusIdle
		updated.NextSyncAt = nextSourceMirrorSyncAt(updated, now)
	}
	if input.Enabled == nil {
		updated.Enabled = existing.Enabled
		if !configChanged {
			updated.SyncStatus = existing.SyncStatus
			updated.NextSyncAt = existing.NextSyncAt
		}
	}
	saved, err := s.saveSourceMirrorUpdate(ctx, base, source, existing, updated, input.ExpectedDefaultRef, pendingAccessToken, subject, now, disabling || executionConfigChanged)
	if err == nil {
		s.decorateSourceMirrorCredentialSummaries(ctx, []*kb.SourceMirror{&saved})
	}
	return saved, err
}

func (s *KBService) saveSourceMirrorUpdate(ctx context.Context, base kb.KnowledgeBase, source kb.Source, expected, updated kb.SourceMirror, expectedDefaultRef *string, pendingAccessToken *pendingMirrorAccessToken, subject string, now time.Time, cancelActiveJobs bool) (kb.SourceMirror, error) {
	var saved kb.SourceMirror
	err := s.store.Transaction(ctx, func(tx *kb.Store) error {
		lockedSource, err := tx.LockSource(ctx, base.WorkspaceID, base.ID, source.ID)
		if err != nil {
			return err
		}
		source = lockedSource
		current, err := tx.GetSourceMirror(ctx, base.WorkspaceID, base.ID, source.ID)
		if err != nil {
			return mapStoreNotFound(err)
		}
		if !sameSourceMirrorMutationGeneration(expected, current) {
			return conflict("source mirror configuration changed while updating; retry")
		}
		if expectedDefaultRef != nil && strings.TrimSpace(current.DefaultRef) != strings.TrimSpace(*expectedDefaultRef) {
			return conflict("source mirror default branch changed while updating; refresh and retry")
		}
		if pendingAccessToken != nil {
			if _, err := persistMirrorAccessToken(ctx, s.automationSecretStore.WithDB(tx.DB()), *pendingAccessToken); err != nil {
				return err
			}
		}
		saved, err = tx.UpdateSourceMirror(ctx, updated)
		if err != nil {
			return err
		}
		source.DefaultRef = firstNonEmptyString(saved.DefaultRef, source.DefaultRef)
		source.SyncStatus = saved.SyncStatus
		source.LastError = saved.LastError
		source.UpdatedBy = strings.TrimSpace(subject)
		source.UpdatedAt = now
		if _, err := tx.UpdateSource(ctx, source); err != nil {
			return err
		}
		if !cancelActiveJobs {
			return nil
		}
		_, err = tx.CancelSourceJobs(ctx, base.WorkspaceID, base.ID, source.ID, KBJobTypeSyncSourceMirror, now)
		return err
	})
	return saved, err
}

func sameSourceMirrorMutationGeneration(expected, current kb.SourceMirror) bool {
	return expected.SourceID == current.SourceID &&
		expected.WorkspaceID == current.WorkspaceID &&
		expected.KnowledgeBaseID == current.KnowledgeBaseID &&
		expected.UpstreamType == current.UpstreamType &&
		expected.RemoteURL == current.RemoteURL &&
		expected.CredentialRef == current.CredentialRef &&
		expected.DefaultRef == current.DefaultRef &&
		sameOptionalBool(expected.DefaultRefExplicit, current.DefaultRefExplicit) &&
		expected.IntervalSeconds == current.IntervalSeconds &&
		expected.Enabled == current.Enabled &&
		expected.SyncStatus == current.SyncStatus &&
		sameOptionalTime(expected.LastSyncAt, current.LastSyncAt) &&
		sameOptionalTime(expected.LastSuccessAt, current.LastSuccessAt) &&
		sameOptionalTime(expected.NextSyncAt, current.NextSyncAt) &&
		expected.LastError == current.LastError &&
		expected.LastMirroredCommit == current.LastMirroredCommit &&
		expected.UpdatedBy == current.UpdatedBy &&
		expected.UpdatedAt.Equal(current.UpdatedAt)
}

func sameOptionalBool(left, right *bool) bool {
	if left == nil || right == nil {
		return left == nil && right == nil
	}
	return *left == *right
}

func sameOptionalTime(left, right *time.Time) bool {
	if left == nil || right == nil {
		return left == nil && right == nil
	}
	return left.Equal(*right)
}

func (s *KBService) ListSourceMirrorSyncs(ctx context.Context, workspaceID string, subject string, kbID string, sourceID string, page int, perPage int) (SourceMirrorSyncListResult, error) {
	base, err := s.GetKnowledgeBase(ctx, workspaceID, subject, kbID)
	if err != nil {
		return SourceMirrorSyncListResult{}, err
	}
	if _, err := s.store.GetSource(ctx, base.WorkspaceID, base.ID, sourceID); err != nil {
		return SourceMirrorSyncListResult{}, mapStoreNotFound(err)
	}
	mirror, err := s.store.GetSourceMirror(ctx, base.WorkspaceID, base.ID, sourceID)
	if err != nil {
		return SourceMirrorSyncListResult{}, mapStoreNotFound(err)
	}
	s.decorateSourceMirrorCredentialSummaries(ctx, []*kb.SourceMirror{&mirror})
	page, perPage = normalizeKBPagination(page, perPage)
	jobs, err := s.store.ListSourceJobs(ctx, base.WorkspaceID, base.ID, sourceID, KBJobTypeSyncSourceMirror, page, perPage)
	if err != nil {
		return SourceMirrorSyncListResult{}, err
	}
	return SourceMirrorSyncListResult{Mirror: mirror, Jobs: jobs}, nil
}

func (s *KBService) EnqueueDueSourceMirrorSyncs(ctx context.Context, options SourceMirrorScheduleOptions) (SourceMirrorScheduleResult, error) {
	if s == nil || s.store == nil {
		return SourceMirrorScheduleResult{}, invalidArgument("kb service is not configured")
	}
	limit := options.Limit
	if limit <= 0 {
		limit = 10
	}
	now := s.now()
	mirrors, err := s.store.ListDueSourceMirrors(ctx, now, limit, options.WorkspaceIDs)
	if err != nil {
		return SourceMirrorScheduleResult{}, err
	}
	result := SourceMirrorScheduleResult{Scanned: len(mirrors)}
	var scheduleErr error
	for _, mirror := range mirrors {
		if _, err := s.enqueueSourceMirrorSync(ctx, mirror.WorkspaceID, mirror.KnowledgeBaseID, mirror.SourceID, mirror.DefaultRef, "meshx-kb-scheduler", KBSourceMirrorStageQueued, false); err != nil {
			result.Failed++
			diagnostic := classifySourceMirrorError(KBSourceMirrorStageQueued, err)
			s.logger.ErrorContext(ctx, "kb source mirror scheduler enqueue failed",
				"workspace_id", mirror.WorkspaceID,
				"kb_id", mirror.KnowledgeBaseID,
				"source_id", mirror.SourceID,
				"stage", KBSourceMirrorStageQueued,
				"error_class", diagnostic.Code,
				"retryable", diagnostic.Retryable,
				"error", SanitizeKBProgressMessage(err.Error()),
			)
			scheduleErr = errors.Join(scheduleErr, fmt.Errorf("enqueue source mirror %s: %s", mirror.SourceID, SanitizeKBProgressMessage(err.Error())))
			continue
		}
		result.Enqueued++
	}
	return result, scheduleErr
}

func (s *KBService) enqueueSourceMirrorSync(ctx context.Context, workspaceID, kbID, sourceID, ref, actor, stage string, failIfExisting bool) (kb.Job, error) {
	if s == nil || s.store == nil {
		return kb.Job{}, invalidArgument("kb service is not configured")
	}
	native, err := s.kbSourceNativeActive(ctx, workspaceID, sourceID)
	if err != nil {
		return kb.Job{}, err
	}
	if native {
		return kb.Job{}, nativeKBSourceLegacyOperationConflict()
	}
	// Perform external/DNS and credential validation before opening the
	// transaction. The source lock below re-reads and compares the fields that
	// define this sync generation before it creates any job.
	validatedMirror, validatedRemoteURL, err := s.validateSourceMirrorForSync(ctx, workspaceID, kbID, sourceID)
	if err != nil {
		return kb.Job{}, err
	}
	trigger := sourceMirrorTrigger(actor)
	requestedBy := strings.TrimSpace(actor)
	if trigger == KBSourceMirrorTriggerScheduled {
		requestedBy = ""
	}
	now := s.now()
	traceID := newKBID("trace")
	var created kb.Job
	var effectiveRef string
	err = s.store.Transaction(ctx, func(tx *kb.Store) error {
		source, err := tx.LockSource(ctx, workspaceID, kbID, sourceID)
		if err != nil {
			return mapStoreNotFound(err)
		}
		mirror, err := tx.GetSourceMirror(ctx, workspaceID, kbID, sourceID)
		if err != nil {
			return mapStoreNotFound(err)
		}
		if !mirror.Enabled {
			return invalidArgument("source mirror is disabled")
		}
		if !sameSourceMirrorSyncGeneration(validatedMirror, mirror) {
			return conflict("source mirror configuration changed while queuing sync; retry")
		}
		mirror.RemoteURL = validatedRemoteURL
		effectiveRef = firstNonEmptyString(strings.TrimSpace(ref), strings.TrimSpace(mirror.DefaultRef))
		expectedCurrentSourceCommitSHA := ""
		if current, currentErr := tx.GetCurrentSourceRevision(ctx, workspaceID, kbID, sourceID); currentErr == nil {
			expectedCurrentSourceCommitSHA = strings.TrimSpace(current.SourceCommitSHA)
		} else if !errors.Is(currentErr, gorm.ErrRecordNotFound) {
			return currentErr
		}
		payload, err := encodeSourceMirrorSyncJobPayload(SourceMirrorSyncJobPayload{
			SourceID: sourceID, Ref: effectiveRef, Status: stage, Trigger: trigger,
			RequestedBy: requestedBy, TraceID: traceID,
			ExpectedCurrentSourceCommitSHA: expectedCurrentSourceCommitSHA, SourceRevisionFenceSet: true,
		})
		if err != nil {
			return err
		}
		priority := KBSourceMirrorJobPriorityRefresh
		if mirror.LastSuccessAt == nil && strings.TrimSpace(mirror.LastMirroredCommit) == "" {
			priority = KBSourceMirrorJobPriorityInitialImport
		}
		job := kb.Job{
			ID: newKBID("job"), WorkspaceID: strings.TrimSpace(workspaceID), KnowledgeBaseID: strings.TrimSpace(kbID),
			SourceID: strings.TrimSpace(sourceID), JobType: KBJobTypeSyncSourceMirror, Status: KBJobStatusQueued,
			Priority: priority, Progress: payload, CreatedAt: now, UpdatedAt: now,
		}
		created, err = tx.CreateUniqueSourceJob(ctx, job)
		if err != nil {
			if errors.Is(err, kb.ErrSourceJobAlreadyActive) {
				if failIfExisting {
					return conflict("a source mirror sync is already queued or running")
				}
				// Scheduler path: created already holds the existing job;
				// return nil without mutating mirror/source status.
				return nil
			}
			return err
		}
		mirror.SyncStatus = kb.SourceMirrorSyncStatusQueued
		mirror.LastError = ""
		mirror.UpdatedBy = strings.TrimSpace(actor)
		mirror.UpdatedAt = now
		if _, err := tx.UpdateSourceMirror(ctx, mirror); err != nil {
			return err
		}
		// For background syncs of an already-imported mirror source, do not
		// regress source.Status to queued. The initial import is complete;
		// changing source.Status to queued here would cause importProgress to
		// fall through to the provisioning-queued branch and show a misleading
		// "importing 10%" banner. Only the initial import (no prior success)
		// should drive source.Status to queued.
		initiallyImported := mirror.LastSuccessAt != nil || strings.TrimSpace(mirror.LastMirroredCommit) != ""
		source.LastError = ""
		source.UpdatedBy = strings.TrimSpace(actor)
		source.UpdatedAt = now
		if initiallyImported {
			source.SyncStatus = kb.SourceMirrorSyncStatusQueued
		} else {
			source.SyncStatus = kb.SourceMirrorSyncStatusQueued
			source.Status = KBSourceStatusQueued
		}
		if _, err := tx.UpdateSource(ctx, source); err != nil {
			return err
		}
		return nil
	})
	if err == nil && created.ID != "" && strings.TrimSpace(created.Progress) != "" {
		s.logSourceMirrorLifecycle(ctx, "enqueued", created, sourceMirrorSyncPayload{
			SourceID:    sourceID,
			Ref:         effectiveRef,
			Stage:       stage,
			Trigger:     trigger,
			RequestedBy: requestedBy,
			TraceID:     traceID,
		}, slog.LevelInfo)
	}
	return created, err
}

func (s *KBService) validateSourceMirrorForSync(ctx context.Context, workspaceID, kbID, sourceID string) (kb.SourceMirror, string, error) {
	mirror, err := s.store.GetSourceMirror(ctx, workspaceID, kbID, sourceID)
	if err != nil {
		return kb.SourceMirror{}, "", mapStoreNotFound(err)
	}
	if !mirror.Enabled {
		return kb.SourceMirror{}, "", invalidArgument("source mirror is disabled")
	}
	remoteURL, err := ValidateSourceMirrorRemoteForUpstream(ctx, mirror.UpstreamType, mirror.RemoteURL, s.sourceMirrorRemoteValidationOptions())
	if err != nil {
		return kb.SourceMirror{}, "", err
	}
	mirror.RemoteURL = remoteURL
	if err := s.validatePersonalAccessTokenForMirrorSync(ctx, mirror); err != nil {
		return kb.SourceMirror{}, "", err
	}
	return mirror, remoteURL, nil
}

func sameSourceMirrorSyncGeneration(validated, current kb.SourceMirror) bool {
	return validated.Enabled == current.Enabled &&
		strings.TrimSpace(validated.UpstreamType) == strings.TrimSpace(current.UpstreamType) &&
		strings.TrimSpace(validated.RemoteURL) == strings.TrimSpace(current.RemoteURL) &&
		strings.TrimSpace(validated.DefaultRef) == strings.TrimSpace(current.DefaultRef) &&
		strings.TrimSpace(validated.CredentialRef) == strings.TrimSpace(current.CredentialRef)
}

func (s *KBService) RunPendingSourceMirrorJobs(ctx context.Context, options SourceMirrorRunOptions) (SourceMirrorRunResult, error) {
	if s == nil || s.store == nil {
		return SourceMirrorRunResult{}, invalidArgument("kb service is not configured")
	}
	if s.sourceMirrorSyncer == nil {
		return SourceMirrorRunResult{}, invalidArgument("source mirror syncer is not configured")
	}
	limit := options.Limit
	if limit <= 0 {
		limit = 1
	}
	leaseOwner := strings.TrimSpace(options.LeaseOwner)
	if leaseOwner == "" {
		leaseOwner = "meshx-kb-mirror-worker"
	}
	leaseDuration := options.LeaseDuration
	if leaseDuration <= 0 {
		leaseDuration = SourceMirrorLeaseDuration(defaultSourceMirrorTimeout)
	}
	type completedJob struct {
		id  string
		err error
	}
	completed := make(chan completedJob, limit)
	result := SourceMirrorRunResult{}
	active := 0
	var runErr error
	claimMore := true
	claimAndStart := func(available int) (int, error) {
		jobs, err := s.store.ClaimQueuedJobsFair(ctx, KBJobTypeSyncSourceMirror, leaseOwner, leaseDuration, available, s.now(), options.WorkspaceIDs)
		if err != nil {
			return 0, err
		}
		for _, job := range jobs {
			result.Claimed++
			active++
			logging.Go(nil, func() {
				completed <- completedJob{id: job.ID, err: s.runSourceMirrorSyncJob(ctx, job)}
			})
		}
		return len(jobs), nil
	}

	// Treat Limit as the worker pool size, not a one-shot batch. Refill a slot
	// as soon as its job completes so a slow or failing peer cannot keep older
	// queued work waiting for the next periodic tick.
	// Claim the initial pool as one batch before starting any jobs. Besides
	// preserving fair selection, this avoids overlapping a second claim
	// transaction with DB writes from a job that was just started (notably on
	// SQLite, where that overlap can fail with "database is locked").
	if _, err := claimAndStart(limit); err != nil {
		runErr = errors.Join(runErr, err)
		claimMore = false
	}
	for active > 0 {
		job := <-completed
		active--
		if job.err != nil {
			result.Failed++
			runErr = errors.Join(runErr, fmt.Errorf("source mirror job %s: %w", job.id, job.err))
		} else {
			result.Succeeded++
		}
		if !claimMore {
			continue
		}
		if _, err := claimAndStart(1); err != nil {
			runErr = errors.Join(runErr, err)
			claimMore = false
		}
	}
	return result, runErr
}

// SourceMirrorLeaseDuration returns the job lease used for a source mirror sync.
// The lease covers the bounded git sync plus a buffer for reconciliation and DB state updates.
func SourceMirrorLeaseDuration(syncTimeout time.Duration) time.Duration {
	if syncTimeout <= 0 {
		syncTimeout = defaultSourceMirrorTimeout
	}
	return syncTimeout + defaultSourceMirrorLeaseBuffer
}

// ReclaimStaleSourceMirrorJobs resets running mirror sync jobs whose leases
// have expired back to queued. It should be called once at startup to recover
// jobs left in "running" state by a previous process crash or restart, so they
// are immediately eligible for the next worker tick instead of waiting for the
// full lease duration to elapse.
func (s *KBService) ReclaimStaleSourceMirrorJobs(ctx context.Context, workspaceScope ...[]string) (int64, error) {
	if s == nil || s.store == nil {
		return 0, invalidArgument("kb service is not configured")
	}
	return s.store.ReclaimStaleRunningJobs(ctx, KBJobTypeSyncSourceMirror, s.now(), workspaceScope...)
}

func (s *KBService) runSourceMirrorSyncJob(ctx context.Context, job kb.Job) error {
	leaseOwner := strings.TrimSpace(job.LeaseOwner)
	payload, err := parseSourceMirrorSyncJobPayload(job.Progress)
	if err != nil {
		return s.failSourceMirrorSyncJob(ctx, job, sourceMirrorSyncPayload{}, err)
	}
	progress := sourceMirrorSyncPayload{
		SourceID:                       firstNonEmptyString(payload.SourceID, job.SourceID),
		Ref:                            payload.Ref,
		Stage:                          firstNonEmptyString(payload.Status, KBSourceMirrorStageQueued),
		Trigger:                        payload.Trigger,
		RequestedBy:                    payload.RequestedBy,
		TraceID:                        firstNonEmptyString(payload.TraceID, job.ID),
		ExpectedCurrentSourceCommitSHA: payload.ExpectedCurrentSourceCommitSHA,
		SourceRevisionFenceSet:         payload.SourceRevisionFenceSet,
	}
	s.logSourceMirrorLifecycle(ctx, "started", job, progress, slog.LevelInfo)
	if progress.SourceID == "" {
		return s.failSourceMirrorSyncJob(ctx, job, progress, invalidArgument("source mirror sync source_id is required"))
	}
	if err := s.updateSourceMirrorJobProgress(ctx, &job, leaseOwner, progress, KBSourceMirrorStageValidating, "validating source mirror remote"); err != nil {
		return err
	}
	source, err := s.store.GetSource(ctx, job.WorkspaceID, job.KnowledgeBaseID, progress.SourceID)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return s.cancelSourceMirrorSyncJob(ctx, job, progress, "source was deleted")
		}
		return s.failSourceMirrorSyncJob(ctx, job, progress, mapStoreNotFound(err))
	}
	native, authorityErr := s.kbSourceNativeActive(ctx, job.WorkspaceID, progress.SourceID)
	if authorityErr != nil {
		return s.failSourceMirrorSyncJob(ctx, job, progress, authorityErr)
	}
	if native {
		return s.cancelSourceMirrorSyncJob(ctx, job, progress, "source native authority is active")
	}
	mirror, err := s.store.GetSourceMirror(ctx, job.WorkspaceID, job.KnowledgeBaseID, progress.SourceID)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return s.cancelSourceMirrorSyncJob(ctx, job, progress, "source mirror was deleted")
		}
		return s.failSourceMirrorSyncJob(ctx, job, progress, mapStoreNotFound(err))
	}
	if !mirror.Enabled {
		return s.cancelSourceMirrorSyncJob(ctx, job, progress, "source mirror is disabled")
	}
	// Jobs created before the revision fence was introduced take their fence
	// exactly once when execution starts. Newly enqueued jobs already carry it.
	if !progress.SourceRevisionFenceSet {
		if current, currentErr := s.store.GetCurrentSourceRevision(ctx, job.WorkspaceID, job.KnowledgeBaseID, progress.SourceID); currentErr == nil {
			progress.ExpectedCurrentSourceCommitSHA = strings.TrimSpace(current.SourceCommitSHA)
		} else if !errors.Is(currentErr, gorm.ErrRecordNotFound) {
			return s.failSourceMirrorSyncJob(ctx, job, progress, currentErr)
		}
		progress.SourceRevisionFenceSet = true
	}
	remoteURL, err := ValidateSourceMirrorRemoteForUpstream(ctx, mirror.UpstreamType, mirror.RemoteURL, s.sourceMirrorRemoteValidationOptions())
	if err != nil {
		return s.failSourceMirrorSyncJob(ctx, job, progress, err)
	}
	mirror.RemoteURL = remoteURL
	if err := s.validatePersonalAccessTokenForMirrorSync(ctx, mirror); err != nil {
		return s.failSourceMirrorSyncJob(ctx, job, progress, err)
	}
	target, err := s.sourceMirrorTargetRepository(ctx, source)
	if err != nil {
		return s.failSourceMirrorSyncJob(ctx, job, progress, err)
	}
	credentialRepository, err := s.sourcePrimaryRepository(ctx, source)
	if err != nil {
		return s.failSourceMirrorSyncJob(ctx, job, progress, err)
	}
	ref := firstNonEmptyString(progress.Ref, mirror.DefaultRef, source.DefaultRef)
	if err := s.markSourceMirrorRunning(ctx, &job, leaseOwner, source, mirror, ref, remoteURL); err != nil {
		return err
	}
	if err := s.updateSourceMirrorJobProgress(ctx, &job, leaseOwner, progress, KBSourceMirrorStageSyncingAGS, "syncing external git remote into AGS source repository"); err != nil {
		return err
	}
	result, err := s.sourceMirrorSyncer.SyncSourceMirror(ctx, SourceMirrorSyncRequest{
		WorkspaceID:          job.WorkspaceID,
		KnowledgeBaseID:      job.KnowledgeBaseID,
		Source:               source,
		Mirror:               mirror,
		RemoteURL:            remoteURL,
		DefaultRef:           ref,
		CredentialRef:        mirror.CredentialRef,
		CredentialRepository: credentialRepository,
		Target:               target,
		AllowPrivateHosts:    s.allowPrivateSourceMirrorHosts,
		OnGitProgress: func(gitProgress SourceMirrorGitProgress) error {
			return s.recordSourceMirrorGitProgress(ctx, &job, leaseOwner, gitProgress)
		},
	})
	if err != nil {
		return s.failSourceMirrorSyncJob(ctx, job, progress, fmt.Errorf("sync source mirror: %w", err))
	}
	if err := s.updateSourceMirrorJobProgress(ctx, &job, leaseOwner, progress, KBSourceMirrorStageReconciling, "reconciling mirrored AGS source repository"); err != nil {
		return err
	}
	if s.wikiStore != nil {
		if _, err := s.wikiStore.Reconcile(ctx, target); err != nil {
			return s.failSourceMirrorSyncJob(ctx, job, progress, fmt.Errorf("reconcile mirrored source repository: %w", err))
		}
	}
	return s.completeSourceMirrorSyncJob(ctx, job, leaseOwner, progress, source, mirror, result)
}

func (s *KBService) markSourceMirrorRunning(ctx context.Context, job *kb.Job, leaseOwner string, source kb.Source, mirror kb.SourceMirror, ref string, remoteURL string) error {
	now := s.now()
	job.UpdatedAt = now
	return s.store.Transaction(ctx, func(tx *kb.Store) error {
		currentSource, err := tx.LockSource(ctx, job.WorkspaceID, job.KnowledgeBaseID, source.ID)
		if err != nil {
			return err
		}
		currentMirror, err := tx.GetSourceMirror(ctx, job.WorkspaceID, job.KnowledgeBaseID, source.ID)
		if err != nil {
			return err
		}
		if !sameSourceMirrorSyncGeneration(mirror, currentMirror) {
			return conflict("source mirror configuration changed before marking source running")
		}
		updated, ok, err := tx.UpdateLeasedJob(ctx, *job, leaseOwner, job.AttemptCount)
		if err != nil {
			return err
		}
		if !ok {
			return conflict("source mirror sync job lease was lost before marking source running")
		}
		*job = updated
		currentMirror.SyncStatus = kb.SourceMirrorSyncStatusRunning
		currentMirror.LastSyncAt = &now
		currentMirror.LastError = ""
		currentMirror.UpdatedAt = now
		if _, err := tx.UpdateSourceMirror(ctx, currentMirror); err != nil {
			return err
		}
		currentSource.SyncStatus = kb.SourceMirrorSyncStatusRunning
		currentSource.Status = KBSourceMirrorStageSyncingAGS
		currentSource.LastSyncAt = &now
		currentSource.LastError = ""
		currentSource.UpdatedAt = now
		_, err = tx.UpdateSource(ctx, currentSource)
		return err
	})
}

func (s *KBService) updateSourceMirrorJobProgress(ctx context.Context, job *kb.Job, leaseOwner string, payload sourceMirrorSyncPayload, stage string, message string) error {
	now := s.now()
	payload = carrySourceMirrorProgress(job.Progress, payload)
	payload.Stage = stage
	payload.Message = message
	job.Progress = encodeSourceMirrorProgress(payload)
	job.UpdatedAt = now
	updated, ok, err := s.store.UpdateLeasedJob(ctx, *job, leaseOwner, job.AttemptCount)
	if err != nil {
		return err
	}
	if !ok {
		return conflict("source mirror sync job lease was lost before progress update")
	}
	*job = updated
	s.logSourceMirrorLifecycle(ctx, "stage_changed", *job, payload, slog.LevelInfo)
	return nil
}

func (s *KBService) completeSourceMirrorSyncJob(ctx context.Context, job kb.Job, leaseOwner string, payload sourceMirrorSyncPayload, source kb.Source, mirror kb.SourceMirror, result SourceMirrorSyncResult) error {
	now := s.now()
	payload = carrySourceMirrorProgress(job.Progress, payload)
	payload.Stage = KBSourceMirrorStageReady
	payload.Message = "source mirror sync completed"
	job.Status = KBJobStatusSucceeded
	job.Error = ""
	job.Progress = encodeSourceMirrorProgress(payload)
	job.LeaseOwner = ""
	job.LeaseUntil = nil
	job.FinishedAt = &now
	job.UpdatedAt = now
	mirroredCommit := strings.TrimSpace(result.MirroredCommit)
	resolvedDefaultRef := strings.TrimSpace(result.ResolvedDefaultRef)
	err := s.store.Transaction(ctx, func(tx *kb.Store) error {
		currentSource, err := tx.LockSource(ctx, job.WorkspaceID, job.KnowledgeBaseID, source.ID)
		if err != nil {
			return err
		}
		currentMirror, err := tx.GetSourceMirror(ctx, job.WorkspaceID, job.KnowledgeBaseID, source.ID)
		if err != nil {
			return err
		}
		if !sameSourceMirrorSyncGeneration(mirror, currentMirror) {
			return conflict("source mirror configuration changed before completion")
		}
		if _, ok, err := tx.UpdateLeasedJob(ctx, job, leaseOwner, job.AttemptCount); err != nil {
			return err
		} else if !ok {
			return conflict("source mirror sync job lease was lost before completion")
		}
		if err := s.writeKnowledgeBaseJobActivityWithStore(ctx, s.knowledgeBaseActivityStore(tx), job); err != nil {
			return err
		}
		currentMirror.SyncStatus = kb.SourceMirrorSyncStatusSuccess
		currentMirror.LastSyncAt = &now
		currentMirror.LastSuccessAt = &now
		currentMirror.LastError = ""
		currentMirror.LastMirroredCommit = mirroredCommit
		if result.DefaultRefAuto && resolvedDefaultRef != "" {
			currentMirror.DefaultRef = resolvedDefaultRef
			auto := false
			currentMirror.DefaultRefExplicit = &auto
		}
		currentMirror.NextSyncAt = nextSourceMirrorSyncAt(currentMirror, now)
		currentMirror.UpdatedAt = now
		if _, err := tx.UpdateSourceMirror(ctx, currentMirror); err != nil {
			return err
		}
		currentSource.Status = KBSourceStatusReady
		currentSource.SyncStatus = KBSourceStatusReady
		currentSource.LastError = ""
		currentSource.LastSyncAt = &now
		if result.DefaultRefAuto && resolvedDefaultRef != "" {
			currentSource.DefaultRef = resolvedDefaultRef
		}
		currentSource.UpdatedAt = now
		if _, err := tx.UpdateSource(ctx, currentSource); err != nil {
			return err
		}
		if !isCommitSHA(mirroredCommit) {
			return conflict("source mirror sync completed without an immutable commit")
		}
		advanced, _, err := tx.AdvanceSourceRevision(ctx, kb.AdvanceSourceRevisionParams{
			ExpectedCurrentSourceCommitSHA: payload.ExpectedCurrentSourceCommitSHA,
			EnforceExpectedCurrent:         true,
			Revision: kb.SourceRevision{
				ID: newKBID("rev"), WorkspaceID: source.WorkspaceID, KnowledgeBaseID: source.KnowledgeBaseID,
				SourceID: source.ID, SourceCommitSHA: mirroredCommit, FreshnessStatus: "current",
				CreatedAt: now, UpdatedAt: now,
			},
		})
		if err != nil {
			return err
		}
		if strings.TrimSpace(advanced.SourceCommitSHA) != mirroredCommit {
			return conflict("source mirror sync completion was superseded by a newer source revision")
		}
		if resolvedDefaultRef != "" && s.repositoryStore != nil {
			if err := s.repositoryStore.WithDB(tx.DB()).UpdateDefaultRef(ctx, currentSource.WorkspaceID, reposstore.ResourceTypeKBSource, currentSource.ID, reposstore.RepoRoleMirror, resolvedDefaultRef, now); err != nil {
				return err
			}
		}
		if err := s.enqueueKnowledgeBaseAccessChangesForSource(ctx, tx, currentSource, now); err != nil {
			return err
		}
		return nil
	})
	if err == nil {
		s.logSourceMirrorLifecycle(ctx, "succeeded", job, payload, slog.LevelInfo,
			"mirrored_commit", mirroredCommit,
		)
	}
	return err
}

func (s *KBService) failSourceMirrorSyncJob(ctx context.Context, job kb.Job, payload sourceMirrorSyncPayload, cause error) error {
	leaseOwner := strings.TrimSpace(job.LeaseOwner)
	now := s.now()
	payload = carrySourceMirrorProgress(job.Progress, payload)
	failedStage := strings.TrimSpace(payload.Stage)
	if current, err := parseSourceMirrorProgress(job.Progress); err == nil {
		failedStage = firstNonEmptyString(current.Stage, failedStage)
	}
	diagnostic := classifySourceMirrorError(failedStage, cause)
	message := SanitizeKBProgressMessage(errorString(cause))
	if message == "" {
		message = "source mirror sync failed"
	}
	safeCause := errors.New(message)
	payload.FailedStage = failedStage
	payload.ErrorCode = diagnostic.Code
	payload.Retryable = &diagnostic.Retryable
	payload.Stage = KBSourceMirrorStageFailed
	payload.Message = message
	job.Status = KBJobStatusFailed
	job.Error = message
	job.Progress = encodeSourceMirrorProgress(payload)
	job.LeaseOwner = ""
	job.LeaseUntil = nil
	job.FinishedAt = &now
	job.UpdatedAt = now
	persisted := false
	persistErr := s.store.Transaction(ctx, func(tx *kb.Store) error {
		// Only update the job and mirror/source if we still hold the lease.
		// If the lease was lost (leaseOwner empty or UpdateLeasedJob returns
		// !ok), another worker may have already reclaimed and started a new
		// attempt, or the job may have been cancelled by a disable/delete.
		// In either case we must not overwrite the job or mark the
		// mirror/source as failed.
		leaseHeld := false
		if leaseOwner != "" {
			_, ok, err := tx.UpdateLeasedJob(ctx, job, leaseOwner, job.AttemptCount)
			if err != nil {
				return err
			}
			leaseHeld = ok
		}
		if !leaseHeld {
			return nil
		}
		persisted = true
		if err := s.writeKnowledgeBaseJobActivityWithStore(ctx, s.knowledgeBaseActivityStore(tx), job); err != nil {
			return err
		}

		if sourceID := firstNonEmptyString(payload.SourceID, job.SourceID); sourceID != "" {
			if mirror, err := tx.GetSourceMirror(ctx, job.WorkspaceID, job.KnowledgeBaseID, sourceID); err == nil {
				mirror.SyncStatus = kb.SourceMirrorSyncStatusFailed
				mirror.LastSyncAt = &now
				mirror.LastError = message
				mirror.NextSyncAt = nextSourceMirrorSyncAt(mirror, now)
				mirror.UpdatedAt = now
				if _, err := tx.UpdateSourceMirror(ctx, mirror); err != nil {
					return err
				}
			} else if !errors.Is(err, gorm.ErrRecordNotFound) {
				return err
			}
			if source, err := tx.GetSource(ctx, job.WorkspaceID, job.KnowledgeBaseID, sourceID); err == nil {
				source.Status = KBSourceStatusFailed
				source.SyncStatus = KBSourceStatusFailed
				source.LastError = message
				source.LastSyncAt = &now
				source.UpdatedAt = now
				if _, err := tx.UpdateSource(ctx, source); err != nil {
					return err
				}
			} else if !errors.Is(err, gorm.ErrRecordNotFound) {
				return err
			}
		}
		return nil
	})
	if persistErr != nil {
		s.logger.ErrorContext(ctx, "kb source mirror failure state persistence failed",
			"job_id", job.ID,
			"workspace_id", job.WorkspaceID,
			"kb_id", job.KnowledgeBaseID,
			"source_id", firstNonEmptyString(payload.SourceID, job.SourceID),
			"stage", failedStage,
			"attempt", job.AttemptCount,
			"error_class", diagnostic.Code,
			"retryable", diagnostic.Retryable,
			"trace_id", firstNonEmptyString(payload.TraceID, job.ID),
			"error", SanitizeKBProgressMessage(persistErr.Error()),
		)
		return errors.Join(safeCause, fmt.Errorf("persist source mirror failure state: %s", SanitizeKBProgressMessage(persistErr.Error())))
	}
	if persisted {
		s.logSourceMirrorLifecycle(ctx, "failed", job, payload, slog.LevelError,
			"error_class", diagnostic.Code,
			"retryable", diagnostic.Retryable,
			"error", message,
		)
	}
	return safeCause
}

// cancelSourceMirrorSyncJob marks a claimed mirror sync job as cancelled without
// treating the source or mirror as failed. It is used when a mirror is disabled
// or deleted while a job is still in-flight. The mirror and source sync status
// are set to disabled/idle rather than failed so the UI reflects the user's
// intent to stop syncing, not an error.
func (s *KBService) cancelSourceMirrorSyncJob(ctx context.Context, job kb.Job, payload sourceMirrorSyncPayload, reason string) error {
	leaseOwner := strings.TrimSpace(job.LeaseOwner)
	now := s.now()
	message := strings.TrimSpace(reason)
	if message == "" {
		message = "source mirror sync cancelled"
	}
	payload.Stage = KBSourceMirrorStageFailed
	payload.Message = message
	job.Status = KBJobStatusCancelled
	job.Error = ""
	job.Progress = encodeSourceMirrorProgress(payload)
	job.LeaseOwner = ""
	job.LeaseUntil = nil
	job.FinishedAt = &now
	job.UpdatedAt = now
	_ = s.store.Transaction(ctx, func(tx *kb.Store) error {
		if leaseOwner != "" {
			if _, ok, err := tx.UpdateLeasedJob(ctx, job, leaseOwner, job.AttemptCount); err != nil || !ok {
				return nil
			}
		} else {
			_, _ = tx.UpdateJob(ctx, job)
		}
		if sourceID := firstNonEmptyString(payload.SourceID, job.SourceID); sourceID != "" {
			if mirror, err := tx.GetSourceMirror(ctx, job.WorkspaceID, job.KnowledgeBaseID, sourceID); err == nil {
				if !mirror.Enabled {
					mirror.SyncStatus = kb.SourceMirrorSyncStatusDisabled
				} else {
					mirror.SyncStatus = kb.SourceMirrorSyncStatusIdle
				}
				mirror.LastError = ""
				mirror.UpdatedAt = now
				_, _ = tx.UpdateSourceMirror(ctx, mirror)
			}
			if source, err := tx.GetSource(ctx, job.WorkspaceID, job.KnowledgeBaseID, sourceID); err == nil {
				if mirror, mErr := tx.GetSourceMirror(ctx, job.WorkspaceID, job.KnowledgeBaseID, sourceID); mErr == nil && !mirror.Enabled {
					source.SyncStatus = kb.SourceMirrorSyncStatusDisabled
				} else {
					source.SyncStatus = kb.SourceMirrorSyncStatusIdle
				}
				source.LastError = ""
				source.UpdatedAt = now
				_, _ = tx.UpdateSource(ctx, source)
			}
		}
		return nil
	})
	return nil
}

func nextSourceMirrorSyncAt(mirror kb.SourceMirror, now time.Time) *time.Time {
	if !mirror.Enabled || mirror.IntervalSeconds <= 0 {
		return nil
	}
	next := now.Add(time.Duration(mirror.IntervalSeconds) * time.Second)
	return &next
}

func (s *KBService) sourceMirrorTargetRepository(ctx context.Context, source kb.Source) (repos.RepositoryIdentity, error) {
	return s.sourceRepositoryByRole(ctx, source, reposstore.RepoRoleMirror)
}

// sourcePrimaryRepository returns the user-visible repository that holds the
// source's primary content. In mirror mode this is the mirror repository; in
// default mode it is the primary repository. A source is in mirror mode when
// it has a SourceMirror record.
func (s *KBService) sourcePrimaryRepository(ctx context.Context, source kb.Source) (repos.RepositoryIdentity, error) {
	_, err := s.store.GetSourceMirror(ctx, source.WorkspaceID, source.KnowledgeBaseID, source.ID)
	if err == nil {
		return s.sourceRepositoryByRole(ctx, source, reposstore.RepoRoleMirror)
	}
	if !errors.Is(err, gorm.ErrRecordNotFound) {
		return repos.RepositoryIdentity{}, err
	}
	return s.sourceRepositoryByRole(ctx, source, reposstore.RepoRolePrimary)
}

func (s *KBService) sourceRepositoryByRole(ctx context.Context, source kb.Source, role string) (repos.RepositoryIdentity, error) {
	if s.repositoryStore == nil {
		return repos.RepositoryIdentity{}, invalidArgument("source repository mapping store is not configured")
	}
	repo, err := s.repositoryStore.Get(ctx, source.WorkspaceID, reposstore.ResourceTypeKBSource, source.ID, role)
	if err != nil {
		return repos.RepositoryIdentity{}, err
	}
	return repos.RepositoryIdentity{Owner: repo.RepoOwner, Name: repo.RepoName, FullName: repo.RepoFullName, ID: strings.TrimSpace(repo.AGSRepoID), DefaultBranch: firstNonEmptyString(repo.DefaultRef, source.DefaultRef)}, nil
}

func (s *KBService) sourceMirrorRemoteValidationOptions() SourceMirrorRemoteValidationOptions {
	return SourceMirrorRemoteValidationOptions{AllowPrivateHosts: s != nil && s.allowPrivateSourceMirrorHosts}
}

func ValidateSourceMirrorRemoteURL(ctx context.Context, raw string, options SourceMirrorRemoteValidationOptions) (string, error) {
	parsed, err := validateSourceMirrorRemoteURL(ctx, raw, options)
	if err != nil {
		return "", err
	}
	return parsed.Sanitized, nil
}

func ValidateSourceMirrorRemoteForUpstream(ctx context.Context, upstreamType string, raw string, options SourceMirrorRemoteValidationOptions) (string, error) {
	parsed, err := validateSourceMirrorRemoteURL(ctx, raw, options)
	if err != nil {
		return "", err
	}
	upstreamType = kb.NormalizeSourceMirrorUpstreamType(upstreamType)
	switch upstreamType {
	case kb.SourceMirrorUpstreamTypeGitHub:
		if err := validateGitHubSourceMirrorRemote(parsed); err != nil {
			return "", err
		}
	case kb.SourceMirrorUpstreamTypeGitLab:
		if err := validateSelfHostedGitMirrorRemote(parsed, "gitlab"); err != nil {
			return "", err
		}
	case kb.SourceMirrorUpstreamTypeGitea:
		if err := validateSelfHostedGitMirrorRemote(parsed, "gitea"); err != nil {
			return "", err
		}
	}
	return parsed.Sanitized, nil
}

func validateSourceMirrorRemoteURL(ctx context.Context, raw string, options SourceMirrorRemoteValidationOptions) (sourceMirrorParsedRemote, error) {
	raw = strings.TrimSpace(raw)
	if raw == "" {
		return sourceMirrorParsedRemote{}, invalidArgument("source mirror remote_url is required")
	}
	parsed, err := parseSourceMirrorRemoteURL(raw)
	if err != nil {
		return sourceMirrorParsedRemote{}, err
	}
	if parsed.URL != nil && parsed.URL.User != nil {
		return sourceMirrorParsedRemote{}, invalidArgument("source mirror remote_url must not include user info or plaintext credentials")
	}
	host := strings.TrimSpace(strings.Trim(parsed.Host, "[]"))
	if host == "" {
		return sourceMirrorParsedRemote{}, invalidArgument("source mirror remote_url must include a host")
	}
	parsed.Host = host
	if !options.AllowPrivateHosts {
		if err := rejectUnsafeSourceMirrorHost(ctx, host, options.Resolver); err != nil {
			return sourceMirrorParsedRemote{}, err
		}
	}
	return parsed, nil
}

var scpLikeGitURL = regexp.MustCompile(`^[A-Za-z0-9._~%-]+@([^:/]+):(.+)$`)

func parseSourceMirrorRemoteURL(raw string) (sourceMirrorParsedRemote, error) {
	if match := scpLikeGitURL.FindStringSubmatch(raw); match != nil {
		host := match[1]
		path := strings.TrimSpace(match[2])
		if path == "" || strings.HasPrefix(path, "-") {
			return sourceMirrorParsedRemote{}, invalidArgument("source mirror remote_url path is invalid")
		}
		return sourceMirrorParsedRemote{Host: host, Path: path, Sanitized: raw, SCPLike: true}, nil
	}
	parsed, err := url.Parse(raw)
	if err != nil {
		return sourceMirrorParsedRemote{}, invalidArgument("source mirror remote_url must be a valid URL")
	}
	scheme := strings.ToLower(strings.TrimSpace(parsed.Scheme))
	switch scheme {
	case "https", "http", "ssh":
	case "git":
		return sourceMirrorParsedRemote{}, invalidArgument("source mirror remote_url must not use unauthenticated git:// transport")
	case "file":
		return sourceMirrorParsedRemote{}, invalidArgument("source mirror remote_url must not use file:// transport")
	default:
		return sourceMirrorParsedRemote{}, invalidArgument("source mirror remote_url must use https, http, or ssh")
	}
	if parsed.RawQuery != "" || parsed.Fragment != "" {
		return sourceMirrorParsedRemote{}, invalidArgument("source mirror remote_url must not include query or fragment")
	}
	if strings.TrimSpace(parsed.Host) == "" || strings.TrimSpace(parsed.Path) == "" || parsed.Path == "/" {
		return sourceMirrorParsedRemote{}, invalidArgument("source mirror remote_url must include host and path")
	}
	if parsed.User != nil {
		return sourceMirrorParsedRemote{}, invalidArgument("source mirror remote_url must not include user info or plaintext credentials")
	}
	return sourceMirrorParsedRemote{URL: parsed, Host: parsed.Hostname(), Path: parsed.EscapedPath(), Sanitized: parsed.String()}, nil
}

func validateGitHubSourceMirrorRemote(remote sourceMirrorParsedRemote) error {
	if remote.SCPLike {
		return invalidArgument("github source mirror remote_url must use https://github.com/<owner>/<repo>.git")
	}
	if remote.URL == nil {
		return invalidArgument("github source mirror remote_url must use https://github.com/<owner>/<repo>.git")
	}
	if !strings.EqualFold(strings.TrimSpace(remote.URL.Scheme), "https") {
		return invalidArgument("github source mirror remote_url must use https")
	}
	if !strings.EqualFold(strings.TrimSuffix(strings.TrimSpace(remote.Host), "."), "github.com") {
		return invalidArgument("github source mirror remote_url host must be github.com")
	}
	segments := sourceMirrorRemotePathSegments(remote.URL.EscapedPath())
	if len(segments) != 2 {
		return invalidArgument("github source mirror remote_url must be https://github.com/<owner>/<repo>.git")
	}
	owner := segments[0]
	repo := strings.TrimSuffix(segments[1], ".git")
	if !validGitHubOwnerSegment(owner) || !validGitHubRepositorySegment(repo) {
		return invalidArgument("github source mirror remote_url owner and repo are invalid")
	}
	return nil
}

func sourceMirrorRemotePathSegments(escapedPath string) []string {
	rawParts := strings.Split(strings.Trim(escapedPath, "/"), "/")
	parts := make([]string, 0, len(rawParts))
	for _, part := range rawParts {
		part, err := url.PathUnescape(part)
		if err != nil {
			return nil
		}
		part = strings.TrimSpace(part)
		if part != "" {
			parts = append(parts, part)
		}
	}
	return parts
}

func validGitHubOwnerSegment(value string) bool {
	if value == "" || strings.HasPrefix(value, "-") || strings.HasSuffix(value, "-") {
		return false
	}
	for _, r := range value {
		switch {
		case r >= 'a' && r <= 'z':
		case r >= 'A' && r <= 'Z':
		case r >= '0' && r <= '9':
		case r == '-':
		default:
			return false
		}
	}
	return true
}

func validGitHubRepositorySegment(value string) bool {
	if value == "" || value == "." || value == ".." || strings.HasSuffix(value, ".") {
		return false
	}
	for _, r := range value {
		switch {
		case r >= 'a' && r <= 'z':
		case r >= 'A' && r <= 'Z':
		case r >= '0' && r <= '9':
		case r == '-' || r == '_' || r == '.':
		default:
			return false
		}
	}
	return true
}

// validateSelfHostedGitMirrorRemote validates a self-hosted Git platform remote
// (e.g. GitLab, Gitea). Unlike GitHub, the host is variable, so validation
// focuses on scheme and path depth: https is required for the URL form, and the
// path must contain at least owner/repo (GitLab subgroups are allowed, so there
// is no upper bound). SCP-like ssh remotes are accepted with the same path
// depth requirement.
func validateSelfHostedGitMirrorRemote(remote sourceMirrorParsedRemote, platform string) error {
	const minPathSegments = 2
	if remote.SCPLike {
		segments := sourceMirrorRemotePathSegments(remote.Path)
		if len(segments) < minPathSegments {
			return invalidArgument(fmt.Sprintf("%s source mirror remote_url must include at least %d path segments (owner/repo)", platform, minPathSegments))
		}
		return nil
	}
	if remote.URL == nil {
		return invalidArgument(fmt.Sprintf("%s source mirror remote_url must use https://<host>/<owner>/<repo>.git", platform))
	}
	if !strings.EqualFold(strings.TrimSpace(remote.URL.Scheme), "https") {
		return invalidArgument(fmt.Sprintf("%s source mirror remote_url must use https", platform))
	}
	segments := sourceMirrorRemotePathSegments(remote.URL.EscapedPath())
	if len(segments) < minPathSegments {
		return invalidArgument(fmt.Sprintf("%s source mirror remote_url must include at least %d path segments (owner/repo)", platform, minPathSegments))
	}
	return nil
}

func rejectUnsafeSourceMirrorHost(ctx context.Context, host string, resolver interface {
	LookupIP(context.Context, string) ([]net.IP, error)
}) error {
	lower := strings.ToLower(strings.TrimSuffix(host, "."))
	if lower == "localhost" || strings.HasSuffix(lower, ".localhost") {
		return invalidArgument("source mirror remote_url host must not be localhost")
	}
	if ip := net.ParseIP(lower); ip != nil {
		if isUnsafeSourceMirrorIP(ip) {
			return invalidArgument("source mirror remote_url host resolves to a private or local address")
		}
		return nil
	}
	if resolver == nil {
		resolver = defaultSourceMirrorHostResolver{}
	}
	ips, err := resolver.LookupIP(ctx, lower)
	if err != nil {
		return invalidArgument("source mirror remote_url host must resolve before git access")
	}
	if len(ips) == 0 {
		return invalidArgument("source mirror remote_url host must resolve before git access")
	}
	for _, ip := range ips {
		if isUnsafeSourceMirrorIP(ip) {
			return invalidArgument("source mirror remote_url host resolves to a private or local address")
		}
	}
	return nil
}

func isUnsafeSourceMirrorIP(ip net.IP) bool {
	return ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || ip.IsUnspecified() || ip.IsMulticast()
}