// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: BUSL-1.1

package reaper

import (
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"

	svctest "github.com/hashicorp/consul/agent/grpc-external/services/resource/testing"
	"github.com/hashicorp/consul/internal/controller"
	"github.com/hashicorp/consul/internal/resource"
	"github.com/hashicorp/consul/internal/resource/demo"
	"github.com/hashicorp/consul/internal/resource/resourcetest"
	rtest "github.com/hashicorp/consul/internal/resource/resourcetest"
	"github.com/hashicorp/consul/proto-public/pbresource"
	"github.com/hashicorp/consul/sdk/testutil"
)

func TestReconcile_ResourceWithNoChildren(t *testing.T) {
	client := setupResourceService(t)
	runReaperTestCaseWithTenancies(func(tenancy *pbresource.Tenancy) {
		// Seed the database with an artist.
		res, err := demo.GenerateV2Artist()

		// set resource tenancy from default to test tenancy
		res.Id.Tenancy = tenancy

		require.NoError(t, err)
		ctx := testutil.TestContext(t)
		writeRsp, err := client.Write(ctx, &pbresource.WriteRequest{Resource: res})
		require.NoError(t, err)

		// Delete the artist to create a tombstone
		_, err = client.Delete(ctx, &pbresource.DeleteRequest{Id: writeRsp.Resource.Id})
		require.NoError(t, err)

		// Retrieve tombstone
		listRsp, err := client.List(ctx, &pbresource.ListRequest{
			Type:    resource.TypeV1Tombstone,
			Tenancy: tenancy,
		})
		require.NoError(t, err)
		require.Len(t, listRsp.Resources, 1)
		tombstone := listRsp.Resources[0]

		// Verify reconcile does first pass and queues up for a second pass
		rec := newReconciler()
		runtime := controller.Runtime{
			Client: client,
			Logger: testutil.Logger(t),
		}
		req := controller.Request{ID: tombstone.Id}
		require.ErrorIs(t, controller.RequeueAfterError(secondPassDelay), rec.Reconcile(ctx, runtime, req))

		// Verify condition FirstPassCompleted is true
		readRsp, err := client.Read(ctx, &pbresource.ReadRequest{Id: tombstone.Id})
		require.NoError(t, err)
		tombstone = readRsp.Resource
		condition := tombstone.Status[statusKeyReaperController].Conditions[0]
		require.Equal(t, conditionTypeFirstPassCompleted, condition.Type)
		require.Equal(t, pbresource.Condition_STATE_TRUE, condition.State)

		// Verify reconcile does second pass and tombstone is deleted
		// Fake out time so elapsed time > secondPassDelay
		rec.timeNow = func() time.Time { return time.Now().Add(secondPassDelay + time.Second) }
		require.NoError(t, rec.Reconcile(ctx, runtime, req))
		_, err = client.Read(ctx, &pbresource.ReadRequest{Id: tombstone.Id})
		require.Error(t, err)
		require.Equal(t, codes.NotFound.String(), status.Code(err).String())

		// Reconcile again to verify no-op on an already deleted tombstone
		require.NoError(t, rec.Reconcile(ctx, runtime, req))
	})
}

func TestReconcile_ResourceWithChildren(t *testing.T) {
	client := setupResourceService(t)
	runReaperTestCaseWithTenancies(func(tenancy *pbresource.Tenancy) {
		// Seed the database with an artist
		res, err := demo.GenerateV2Artist()

		// set resource tenancy from default to test tenancy
		res.Id.Tenancy = tenancy

		require.NoError(t, err)
		ctx := testutil.TestContext(t)
		writeRsp, err := client.Write(ctx, &pbresource.WriteRequest{Resource: res})
		require.NoError(t, err)
		artist := writeRsp.Resource

		// Create 3 albums owned by the artist
		numAlbums := 3
		for i := 0; i < numAlbums; i++ {
			res, err = demo.GenerateV2Album(artist.Id)
			require.NoError(t, err)
			_, err := client.Write(ctx, &pbresource.WriteRequest{Resource: res})
			require.NoError(t, err)
		}

		// Delete the artist to create a tombstone
		_, err = client.Delete(ctx, &pbresource.DeleteRequest{Id: writeRsp.Resource.Id})
		require.NoError(t, err)

		// Retrieve the tombstone
		listRsp, err := client.List(ctx, &pbresource.ListRequest{
			Type:    resource.TypeV1Tombstone,
			Tenancy: writeRsp.Resource.Id.Tenancy,
		})
		require.NoError(t, err)
		require.Len(t, listRsp.Resources, 1)
		tombstone := listRsp.Resources[0]

		// Verify reconcile does first pass delete and queues up for a second pass
		rec := newReconciler()
		runtime := controller.Runtime{
			Client: client,
			Logger: testutil.Logger(t),
		}
		req := controller.Request{ID: tombstone.Id}
		require.ErrorIs(t, controller.RequeueAfterError(secondPassDelay), rec.Reconcile(ctx, runtime, req))

		// Verify 3 albums deleted
		listRsp, err = client.List(ctx, &pbresource.ListRequest{
			Type:    demo.TypeV2Album,
			Tenancy: artist.Id.Tenancy,
		})
		require.NoError(t, err)
		require.Empty(t, listRsp.Resources)

		// Verify condition FirstPassCompleted is true
		readRsp, err := client.Read(ctx, &pbresource.ReadRequest{Id: tombstone.Id})
		require.NoError(t, err)
		tombstone = readRsp.Resource
		condition := tombstone.Status[statusKeyReaperController].Conditions[0]
		require.Equal(t, conditionTypeFirstPassCompleted, condition.Type)
		require.Equal(t, pbresource.Condition_STATE_TRUE, condition.State)

		// Verify reconcile does second pass
		// Fake out time so elapsed time > secondPassDelay
		rec.timeNow = func() time.Time { return time.Now().Add(secondPassDelay + time.Second) }
		require.NoError(t, rec.Reconcile(ctx, runtime, req))

		// Verify artist tombstone deleted
		_, err = client.Read(ctx, &pbresource.ReadRequest{Id: tombstone.Id})
		require.Error(t, err)
		require.Equal(t, codes.NotFound.String(), status.Code(err).String())

		// Verify tombstones for 3 albums created
		listRsp, err = client.List(ctx, &pbresource.ListRequest{
			Type:    resource.TypeV1Tombstone,
			Tenancy: artist.Id.Tenancy,
		})
		require.NoError(t, err)
		require.Len(t, listRsp.Resources, numAlbums)
	})
}

func TestReconcile_RequeueWithDelayWhenSecondPassDelayNotElapsed(t *testing.T) {
	client := setupResourceService(t)
	runReaperTestCaseWithTenancies(func(tenancy *pbresource.Tenancy) {
		// Seed the database with an artist.
		res, err := demo.GenerateV2Artist()

		// set resource tenancy from default to test tenancy
		res.Id.Tenancy = tenancy

		require.NoError(t, err)
		ctx := testutil.TestContext(t)
		writeRsp, err := client.Write(ctx, &pbresource.WriteRequest{Resource: res})
		require.NoError(t, err)

		// Delete the artist to create a tombstone
		_, err = client.Delete(ctx, &pbresource.DeleteRequest{Id: writeRsp.Resource.Id})
		require.NoError(t, err)

		// Retrieve tombstone
		listRsp, err := client.List(ctx, &pbresource.ListRequest{
			Type:    resource.TypeV1Tombstone,
			Tenancy: writeRsp.Resource.Id.Tenancy,
		})
		require.NoError(t, err)
		require.Len(t, listRsp.Resources, 1)
		tombstone := listRsp.Resources[0]

		// Verify reconcile does first pass and queues up for a second pass
		rec := newReconciler()
		runtime := controller.Runtime{
			Client: client,
			Logger: testutil.Logger(t),
		}
		req := controller.Request{ID: tombstone.Id}
		require.ErrorIs(t, controller.RequeueAfterError(secondPassDelay), rec.Reconcile(ctx, runtime, req))

		// Verify condition FirstPassCompleted is true
		readRsp, err := client.Read(ctx, &pbresource.ReadRequest{Id: tombstone.Id})
		require.NoError(t, err)
		tombstone = readRsp.Resource
		condition := tombstone.Status[statusKeyReaperController].Conditions[0]
		require.Equal(t, conditionTypeFirstPassCompleted, condition.Type)
		require.Equal(t, pbresource.Condition_STATE_TRUE, condition.State)

		// Verify requeued for second pass since secondPassDelay time has not elapsed
		require.ErrorIs(t, controller.RequeueAfterError(secondPassDelay), rec.Reconcile(ctx, runtime, req))
	})
}

func runReaperTestCaseWithTenancies(testCase func(tenancy *pbresource.Tenancy)) {
	for _, tenancy := range resourcetest.TestTenancies() {
		testCase(tenancy)
	}
}

func setupResourceService(t *testing.T) pbresource.ResourceServiceClient {
	return svctest.NewResourceServiceBuilder().
		WithTenancies(rtest.TestTenancies()...).
		WithRegisterFns(demo.RegisterTypes).
		Run(t)
}