package resource_limit

import (
	"errors"
	"sync/atomic"
	"testing"
	"time"

	"github.com/robfig/cron/v3"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/account/idl/pbgen/proto/account"
	"google.golang.org/protobuf/types/known/timestamppb"

	"github.com/pingcap-inc/clinic/pkg/auth/session"
	"github.com/pingcap-inc/clinic/pkg/db"
)

func TestMigrateOrganizationJobSkipsUnchangedOrganizations(t *testing.T) {
	updatedAt := time.Date(2026, time.August, 25, 3, 0, 0, 0, time.UTC)
	cloudOrgs := []*account.Org{
		newCloudOrg(101, updatedAt),
		newCloudOrg(102, updatedAt.Add(time.Minute)),
		newCloudOrg(103, updatedAt),
	}

	var stateCalls int
	var requestedIDs []uint
	var upsertedIDs []uint
	deps := organizationSyncDependencies{
		listOrganizations: func(*session.CtxUser) ([]*account.Org, int64, error) {
			return cloudOrgs, int64(len(cloudOrgs)), nil
		},
		getOrganization: func(*session.CtxUser, uint64) (*account.Org, error) {
			t.Fatal("scheduled sync must not fetch organization details")
			return nil, nil
		},
		loadUpdatedAt: func(ids []uint) (map[uint]time.Time, error) {
			stateCalls++
			requestedIDs = append(requestedIDs, ids...)
			return map[uint]time.Time{
				101: updatedAt,
				102: updatedAt,
			}, nil
		},
		upsertOrganization: func(org *db.Organization, force bool) error {
			assert.False(t, force)
			upsertedIDs = append(upsertedIDs, org.DbaasOrgID)
			return nil
		},
	}

	(&MigrateOrganizationJob{}).run(deps)

	assert.Equal(t, 1, stateCalls)
	assert.Equal(t, []uint{101, 102, 103}, requestedIDs)
	assert.Equal(t, []uint{102, 103}, upsertedIDs)
}

func TestMigrateOrganizationJobPreservesForcedRefresh(t *testing.T) {
	updatedAt := time.Date(2026, time.August, 25, 3, 0, 0, 0, time.UTC)
	cloudOrgs := []*account.Org{
		newCloudOrg(201, updatedAt),
		newCloudOrg(202, updatedAt),
	}

	var detailIDs []uint64
	var upsertedIDs []uint
	deps := organizationSyncDependencies{
		listOrganizations: func(*session.CtxUser) ([]*account.Org, int64, error) {
			return cloudOrgs, int64(len(cloudOrgs)), nil
		},
		getOrganization: func(_ *session.CtxUser, id uint64) (*account.Org, error) {
			detailIDs = append(detailIDs, id)
			return newCloudOrg(id, updatedAt), nil
		},
		loadUpdatedAt: func([]uint) (map[uint]time.Time, error) {
			t.Fatal("forced refresh must not prefilter organizations")
			return nil, nil
		},
		upsertOrganization: func(org *db.Organization, force bool) error {
			assert.True(t, force)
			upsertedIDs = append(upsertedIDs, org.DbaasOrgID)
			return nil
		},
	}

	(&MigrateOrganizationJob{UpdatePlan: true}).run(deps)

	assert.Equal(t, []uint64{201, 202}, detailIDs)
	assert.Equal(t, []uint{201, 202}, upsertedIDs)
}

func TestMigrateOrganizationJobStopsForcedRefreshAfterDetailError(t *testing.T) {
	updatedAt := time.Date(2026, time.August, 25, 3, 0, 0, 0, time.UTC)
	cloudOrgs := []*account.Org{
		newCloudOrg(301, updatedAt),
		newCloudOrg(302, updatedAt),
		newCloudOrg(303, updatedAt),
	}

	var detailIDs []uint64
	var upsertedIDs []uint
	deps := organizationSyncDependencies{
		listOrganizations: func(*session.CtxUser) ([]*account.Org, int64, error) {
			return cloudOrgs, int64(len(cloudOrgs)), nil
		},
		getOrganization: func(_ *session.CtxUser, id uint64) (*account.Org, error) {
			detailIDs = append(detailIDs, id)
			if id == 302 {
				return nil, errors.New("get organization")
			}
			return newCloudOrg(id, updatedAt), nil
		},
		loadUpdatedAt: func([]uint) (map[uint]time.Time, error) {
			t.Fatal("forced refresh must not prefilter organizations")
			return nil, nil
		},
		upsertOrganization: func(org *db.Organization, force bool) error {
			assert.True(t, force)
			upsertedIDs = append(upsertedIDs, org.DbaasOrgID)
			return nil
		},
	}

	(&MigrateOrganizationJob{UpdatePlan: true}).run(deps)

	assert.Equal(t, []uint64{301, 302}, detailIDs)
	assert.Equal(t, []uint{301}, upsertedIDs)
}

func TestCronJobCanSkipOverlappingRuns(t *testing.T) {
	c := &CronJob{
		cron: cron.New(),
		Jobs: make(map[cron.EntryID]cron.Job),
	}
	job := &blockingCronJob{
		started: make(chan struct{}),
		release: make(chan struct{}),
	}
	require.NoError(t, c.addJob(
		"* * * * *",
		job,
		cron.SkipIfStillRunning(cron.DiscardLogger),
	))
	require.Len(t, c.Jobs, 1)

	var entryID cron.EntryID
	for id := range c.Jobs {
		entryID = id
	}
	wrappedJob := c.cron.Entry(entryID).Job

	firstDone := make(chan struct{})
	go func() {
		wrappedJob.Run()
		close(firstDone)
	}()
	<-job.started

	secondDone := make(chan struct{})
	go func() {
		wrappedJob.Run()
		close(secondDone)
	}()

	select {
	case <-secondDone:
		assert.Equal(t, int32(1), job.starts.Load())
	case <-time.After(time.Second):
		close(job.release)
		<-firstDone
		<-secondDone
		t.Fatal("overlapping run was not skipped")
	}

	close(job.release)
	<-firstDone
}

func newCloudOrg(id uint64, updatedAt time.Time) *account.Org {
	return &account.Org{
		Id:        id,
		Name:      "test-org",
		Email:     "owner@example.com",
		CreatedAt: timestamppb.New(updatedAt.Add(-time.Hour)),
		UpdatedAt: timestamppb.New(updatedAt),
	}
}

type blockingCronJob struct {
	starts  atomic.Int32
	started chan struct{}
	release chan struct{}
}

func (j *blockingCronJob) Run() {
	if j.starts.Add(1) == 1 {
		close(j.started)
	}
	<-j.release
}