package worker

import (
	"context"
	"fmt"
	"sort"
	"sync"
	"testing"
	"time"

	"github.com/pingcap/agent-git-service/internal/db"
	"github.com/pingcap/agent-git-service/internal/service"
	reposync "github.com/pingcap/agent-git-service/internal/sync"
	"github.com/pingcap/agent-git-service/internal/sync/canonical"
	"github.com/pingcap/agent-git-service/internal/testharness"
)

type integrationConnectionValidator struct{}

func (integrationConnectionValidator) ValidateRepository(context.Context, string, string, string) error {
	return nil
}

type integrationProviderFactory struct {
	providers map[string]*integrationProvider
}

func (f integrationProviderFactory) Open(_ context.Context, endpoint reposync.Endpoint) (reposync.Provider, error) {
	provider := f.providers[endpoint.Side]
	if provider == nil {
		return nil, fmt.Errorf("missing integration provider for side %s", endpoint.Side)
	}
	provider.setEndpoint(endpoint)
	return provider, nil
}

type integrationProvider struct {
	mu          sync.Mutex
	endpoint    reposync.Endpoint
	objects     map[string]reposync.InventoryObject
	revision    int
	applyCounts map[string]int
}

func newIntegrationProvider(objects ...reposync.InventoryObject) *integrationProvider {
	provider := &integrationProvider{objects: make(map[string]reposync.InventoryObject), applyCounts: make(map[string]int)}
	provider.replace(objects...)
	return provider
}

func (p *integrationProvider) setEndpoint(endpoint reposync.Endpoint) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.endpoint = endpoint
}

func (p *integrationProvider) Endpoint() reposync.Endpoint {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.endpoint
}

func (p *integrationProvider) Inventory(context.Context) ([]reposync.InventoryObject, string, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	objects := p.snapshotLocked()
	return objects, inventoryDigest(objects), nil
}

func (p *integrationProvider) InventoryIndex(ctx context.Context) ([]reposync.InventoryObject, string, error) {
	objects, digest, err := p.Inventory(ctx)
	for index := range objects {
		objects[index].Payload = nil
		objects[index].PayloadBytes = 0
	}
	return objects, digest, err
}

func (p *integrationProvider) IsEmpty(context.Context) (bool, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	return len(p.objects) == 0, nil
}

func (p *integrationProvider) Read(_ context.Context, objectType, identity string) (reposync.InventoryObject, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	object, exists := p.objects[objectType+"\x00"+identity]
	if !exists {
		return reposync.InventoryObject{}, reposync.ErrObjectNotFound
	}
	return cloneIntegrationObject(object), nil
}

func (p *integrationProvider) ReadBatch(_ context.Context, objectType string, identities []string) (map[string]reposync.InventoryObject, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	result := make(map[string]reposync.InventoryObject, len(identities))
	for _, identity := range identities {
		if object, exists := p.objects[objectType+"\x00"+identity]; exists {
			result[identity] = cloneIntegrationObject(object)
		}
	}
	return result, nil
}

func (p *integrationProvider) Recover(_ context.Context, item reposync.PlannedItem) (reposync.TargetState, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.recoverLocked(item), nil
}

func (p *integrationProvider) RecoverBatch(_ context.Context, items []reposync.PlannedItem) (map[string]reposync.TargetState, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	result := make(map[string]reposync.TargetState, len(items))
	for _, item := range items {
		result[reposync.ExecutionKey(item)] = p.recoverLocked(item)
	}
	return result, nil
}

func (p *integrationProvider) recoverLocked(item reposync.PlannedItem) reposync.TargetState {
	if item.TargetIdentity != "" {
		if object, exists := p.objects[item.ObjectType+"\x00"+item.TargetIdentity]; exists {
			return integrationTargetState(object)
		}
		return reposync.TargetState{Exists: false}
	}
	for _, object := range p.objects {
		if object.ObjectType == item.ObjectType && object.ObjectKey == item.ObjectKey {
			return integrationTargetState(object)
		}
	}
	return reposync.TargetState{Exists: false}
}

func (p *integrationProvider) ApplyBatch(_ context.Context, batch reposync.PhysicalBatch, _ reposync.Provider) ([]reposync.ApplyResult, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	results := make([]reposync.ApplyResult, 0, len(batch.Items))
	for _, item := range batch.Items {
		p.applyCounts[item.Action]++
		identity := item.TargetIdentity
		key := item.ObjectType + "\x00" + identity
		if item.Action == reposync.ActionDelete {
			delete(p.objects, key)
			results = append(results, reposync.ApplyResult{ObjectType: item.ObjectType, ObjectKey: item.ObjectKey, TargetIdentity: identity})
			continue
		}
		if identity == "" {
			identity = "target-" + item.SourceIdentity
			key = item.ObjectType + "\x00" + identity
		}
		p.revision++
		uniqueKey := ""
		if record, err := canonical.Decode(item.Payload); err == nil {
			uniqueKey = record.UniqueKey
		}
		managedLinkID := uint(0)
		if item.ObjectLinkID != nil {
			managedLinkID = *item.ObjectLinkID
		}
		p.objects[key] = reposync.InventoryObject{
			ObjectType: item.ObjectType, Identity: identity, ObjectKey: item.ObjectKey,
			UniqueKey: uniqueKey, Revision: fmt.Sprintf("target-r%d", p.revision), ContentHash: item.DesiredHash,
			PayloadBytes: int64(len(item.Payload)), Payload: append([]byte(nil), item.Payload...),
			Representation: item.Representation, ManagedLinkID: managedLinkID,
		}
		object := p.objects[key]
		results = append(results, reposync.ApplyResult{
			ObjectType: item.ObjectType, ObjectKey: item.ObjectKey, TargetIdentity: identity,
			TargetRevision: object.Revision, TargetHash: object.ContentHash,
		})
	}
	return results, nil
}

func (p *integrationProvider) Verify(_ context.Context, item reposync.PlannedItem, result reposync.ApplyResult) (reposync.TargetState, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	if item.Action == reposync.ActionDelete {
		_, exists := p.objects[item.ObjectType+"\x00"+item.TargetIdentity]
		return reposync.TargetState{Exists: exists, Identity: item.TargetIdentity, Managed: true}, nil
	}
	object, exists := p.objects[item.ObjectType+"\x00"+result.TargetIdentity]
	if !exists {
		return reposync.TargetState{Exists: false}, nil
	}
	return integrationTargetState(object), nil
}

func (*integrationProvider) Close() error { return nil }

func (p *integrationProvider) replace(objects ...reposync.InventoryObject) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.objects = make(map[string]reposync.InventoryObject, len(objects))
	for _, object := range objects {
		p.objects[object.ObjectType+"\x00"+object.Identity] = cloneIntegrationObject(object)
	}
}

func (p *integrationProvider) tamper(identity, hash string) {
	p.mu.Lock()
	defer p.mu.Unlock()
	key := canonical.TypeLabel + "\x00" + identity
	object := p.objects[key]
	p.revision++
	object.Revision = fmt.Sprintf("tampered-r%d", p.revision)
	object.ContentHash = hash
	p.objects[key] = object
}

func (p *integrationProvider) snapshotLocked() []reposync.InventoryObject {
	objects := make([]reposync.InventoryObject, 0, len(p.objects))
	for _, object := range p.objects {
		objects = append(objects, cloneIntegrationObject(object))
	}
	sort.Slice(objects, func(i, j int) bool {
		if objects[i].ObjectType != objects[j].ObjectType {
			return objects[i].ObjectType < objects[j].ObjectType
		}
		return objects[i].Identity < objects[j].Identity
	})
	return objects
}

func integrationTargetState(object reposync.InventoryObject) reposync.TargetState {
	return reposync.TargetState{
		Exists: true, Managed: object.ManagedLinkID != 0, Identity: object.Identity,
		Revision: object.Revision, Hash: object.ContentHash,
	}
}

func cloneIntegrationObject(object reposync.InventoryObject) reposync.InventoryObject {
	object.Payload = append([]byte(nil), object.Payload...)
	return object
}

func integrationLabel(t *testing.T, identity, name, revision, color string) reposync.InventoryObject {
	t.Helper()
	record := canonical.NewRecord(canonical.TypeLabel, identity, "label:"+identity, map[string]any{
		"name": name, "color": color, "description": "",
	})
	record.UniqueKey = name
	object, err := record.Inventory(revision, "native")
	if err != nil {
		t.Fatal(err)
	}
	return object
}

func TestWorkerBulkBootstrapIncrementalCheckpointAndFullAudit(t *testing.T) {
	svc, tenantHandle, cleanup := testharness.NewService(t, testharness.ServiceConfig{})
	defer cleanup()
	ctx := tenantHandle.Context()
	svc.SetRepositorySyncConnectionValidator(tenantHandle.Info().ID, integrationConnectionValidator{})

	owner := db.User{Login: "sync-worker-owner", Name: "sync-worker-owner", Type: db.TypeUser, UserKind: db.UserKindHuman, Status: db.UserStatusActive}
	if err := svc.DBForCtx(ctx).Create(&owner).Error; err != nil {
		t.Fatalf("create owner: %v", err)
	}
	ownerCtx, err := service.ContextWithUserForTest(ctx, owner)
	if err != nil {
		t.Fatal(err)
	}
	repository, err := svc.CreateRepo(ownerCtx, service.CreateRepoInput{OwnerLogin: owner.Login, Name: "worker-source"})
	if err != nil {
		t.Fatalf("create repository: %v", err)
	}
	binding, err := svc.CreateRepositorySyncBinding(ownerCtx, service.CreateRepositorySyncBindingInput{
		AGSRepositoryID: repository.ID, GitHubRepositoryID: "R_worker", GitHubRepositoryFullName: "example/worker",
		GitHubConnectionID: "github-app-installation:4", InitialAuthority: service.SyncProviderAGS,
	})
	if err != nil {
		t.Fatalf("create binding: %v", err)
	}

	labelOne := integrationLabel(t, "source-1", "bug", "source-r1", "ff0000")
	labelTwo := integrationLabel(t, "source-2", "feature", "source-r1", "00ff00")
	source := newIntegrationProvider(labelOne, labelTwo)
	target := newIntegrationProvider()
	worker, err := New(svc, integrationProviderFactory{providers: map[string]*integrationProvider{
		service.SyncSideA: source, service.SyncSideB: target,
	}}, Options{WorkerID: "integration-worker", LeaseTTL: time.Minute, BatchLimits: reposync.BatchLimits{MaxItems: 1, MaxBytes: 1 << 20}})
	if err != nil {
		t.Fatal(err)
	}

	run, err := svc.CreateRepositorySyncRun(ownerCtx, binding.ID, service.CreateRepositorySyncRunInput{Mode: service.SyncRunModeApply})
	if err != nil {
		t.Fatal(err)
	}
	if err := worker.DrainOnce(ctx); err != nil {
		t.Fatalf("bulk bootstrap: %v", err)
	}
	completed, err := svc.GetRepositorySyncRun(ownerCtx, run.ID)
	if err != nil {
		t.Fatal(err)
	}
	if completed.Status != service.SyncRunStatusCompleted || completed.PlannerMode != reposync.PlannerModeBootstrap {
		t.Fatalf("bootstrap run=%#v", completed)
	}
	if len(target.objects) != 2 || target.applyCounts[reposync.ActionCreate] != 2 {
		t.Fatalf("bootstrap target=%d create applies=%d", len(target.objects), target.applyCounts[reposync.ActionCreate])
	}

	updatedOne := integrationLabel(t, "source-1", "bug", "source-r2", "aa0000")
	labelThree := integrationLabel(t, "source-3", "docs", "source-r1", "0000ff")
	source.replace(updatedOne, labelThree)
	incremental, err := svc.CreateRepositorySyncRun(ownerCtx, binding.ID, service.CreateRepositorySyncRunInput{Mode: service.SyncRunModeApply})
	if err != nil {
		t.Fatal(err)
	}
	if err := worker.DrainOnce(ctx); err != nil {
		t.Fatalf("incremental checkpoint: %v", err)
	}
	incremental, err = svc.GetRepositorySyncRun(ownerCtx, incremental.ID)
	if err != nil {
		t.Fatal(err)
	}
	if incremental.Status != service.SyncRunStatusCompleted || incremental.PlannerMode != reposync.PlannerModeCheckpoint {
		t.Fatalf("incremental run=%#v", incremental)
	}
	if len(target.objects) != 2 || target.applyCounts[reposync.ActionCreate] != 3 || target.applyCounts[reposync.ActionUpdate] != 1 || target.applyCounts[reposync.ActionDelete] != 1 {
		t.Fatalf("incremental target=%d applies=%#v", len(target.objects), target.applyCounts)
	}

	target.tamper("target-source-1", "standby-tampered")
	checkpoint, err := svc.CreateRepositorySyncRun(ownerCtx, binding.ID, service.CreateRepositorySyncRunInput{Mode: service.SyncRunModeApply})
	if err != nil {
		t.Fatal(err)
	}
	if err := worker.DrainOnce(ctx); err != nil {
		t.Fatalf("unchanged checkpoint: %v", err)
	}
	var checkpointItems int64
	if err := svc.DBForCtx(ctx).Model(&db.SyncRunItem{}).Where("run_id = ?", checkpoint.ID).Count(&checkpointItems).Error; err != nil {
		t.Fatal(err)
	}
	if checkpointItems != 0 || target.objects[canonical.TypeLabel+"\x00target-source-1"].ContentHash != "standby-tampered" {
		t.Fatalf("checkpoint items=%d; checkpoint unexpectedly scanned unchanged target", checkpointItems)
	}

	fullAudit, err := svc.CreateRepositorySyncRun(ownerCtx, binding.ID, service.CreateRepositorySyncRunInput{
		Mode: service.SyncRunModeApply, AuditMode: service.SyncAuditModeFull,
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := worker.DrainOnce(ctx); err != nil {
		t.Fatalf("full audit repair: %v", err)
	}
	fullAudit, err = svc.GetRepositorySyncRun(ownerCtx, fullAudit.ID)
	if err != nil {
		t.Fatal(err)
	}
	if fullAudit.Status != service.SyncRunStatusCompleted || fullAudit.AuditMode != service.SyncAuditModeFull {
		t.Fatalf("full audit run=%#v", fullAudit)
	}
	mappedUpdated, err := canonical.MapObjectForTarget(updatedOne, service.SyncProviderAGS, service.SyncProviderGitHub)
	if err != nil {
		t.Fatal(err)
	}
	if target.objects[canonical.TypeLabel+"\x00target-source-1"].ContentHash != mappedUpdated.ContentHash || target.applyCounts[reposync.ActionUpdate] != 2 {
		t.Fatalf("full audit did not repair target; applies=%#v", target.applyCounts)
	}
	var refreshedBinding db.SyncBinding
	if err := svc.DBForCtx(ctx).First(&refreshedBinding, binding.ID).Error; err != nil {
		t.Fatal(err)
	}
	if refreshedBinding.LastFullAuditAt == nil {
		t.Fatal("successful full audit did not update last_full_audit_at")
	}
}

func TestWorkerNonEmptyBaselineUsesFullDiffAndPreservesUnmanagedTarget(t *testing.T) {
	svc, tenantHandle, cleanup := testharness.NewService(t, testharness.ServiceConfig{})
	defer cleanup()
	ctx := tenantHandle.Context()
	svc.SetRepositorySyncConnectionValidator(tenantHandle.Info().ID, integrationConnectionValidator{})

	owner := db.User{Login: "sync-nonempty-owner", Name: "sync-nonempty-owner", Type: db.TypeUser, UserKind: db.UserKindHuman, Status: db.UserStatusActive}
	if err := svc.DBForCtx(ctx).Create(&owner).Error; err != nil {
		t.Fatalf("create owner: %v", err)
	}
	ownerCtx, err := service.ContextWithUserForTest(ctx, owner)
	if err != nil {
		t.Fatal(err)
	}
	repository, err := svc.CreateRepo(ownerCtx, service.CreateRepoInput{OwnerLogin: owner.Login, Name: "nonempty-source"})
	if err != nil {
		t.Fatalf("create repository: %v", err)
	}
	binding, err := svc.CreateRepositorySyncBinding(ownerCtx, service.CreateRepositorySyncBindingInput{
		AGSRepositoryID: repository.ID, GitHubRepositoryID: "R_nonempty", GitHubRepositoryFullName: "example/nonempty",
		GitHubConnectionID: "github-app-installation:5", InitialAuthority: service.SyncProviderAGS,
	})
	if err != nil {
		t.Fatalf("create binding: %v", err)
	}

	source := newIntegrationProvider(integrationLabel(t, "source-docs", "docs", "source-r1", "0000ff"))
	unmanaged := integrationLabel(t, "unmanaged", "keep-me", "target-r1", "eeeeee")
	target := newIntegrationProvider(unmanaged)
	worker, err := New(svc, integrationProviderFactory{providers: map[string]*integrationProvider{
		service.SyncSideA: source, service.SyncSideB: target,
	}}, Options{WorkerID: "nonempty-worker", LeaseTTL: time.Minute})
	if err != nil {
		t.Fatal(err)
	}
	run, err := svc.CreateRepositorySyncRun(ownerCtx, binding.ID, service.CreateRepositorySyncRunInput{Mode: service.SyncRunModeApply})
	if err != nil {
		t.Fatal(err)
	}
	if err := worker.DrainOnce(ctx); err != nil {
		t.Fatalf("non-empty baseline: %v", err)
	}
	run, err = svc.GetRepositorySyncRun(ownerCtx, run.ID)
	if err != nil {
		t.Fatal(err)
	}
	if run.Status != service.SyncRunStatusCompleted || run.PlannerMode != reposync.PlannerModeCurrentState {
		t.Fatalf("non-empty baseline run=%#v", run)
	}
	if len(target.objects) != 2 || target.objects[canonical.TypeLabel+"\x00unmanaged"].ContentHash != unmanaged.ContentHash {
		t.Fatalf("unmanaged target was not preserved: %#v", target.objects)
	}
	var items []db.SyncRunItem
	if err := svc.DBForCtx(ctx).Where("run_id = ?", run.ID).Order("id").Find(&items).Error; err != nil {
		t.Fatal(err)
	}
	actions := map[string]int{}
	for _, item := range items {
		actions[item.Action]++
	}
	if actions[reposync.ActionCreate] != 1 || actions[reposync.ActionReport] != 1 {
		t.Fatalf("non-empty baseline actions=%#v", actions)
	}
}