package grpc

import (
	"context"
	"errors"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
)

func TestImportAsyncTaskManagerCompletesAndStoresResult(t *testing.T) {
	manager := newImportAsyncTaskManager()

	taskID := manager.Create(context.Background(), func(context.Context) (any, error) {
		time.Sleep(10 * time.Millisecond)
		return "done", nil
	})

	task, ok := manager.Get(taskID)
	require.True(t, ok)
	require.Equal(t, importAsyncTaskStatusRunning, task.Status)

	require.Eventually(t, func() bool {
		task, ok = manager.Get(taskID)
		require.True(t, ok)
		return task.Status == importAsyncTaskStatusSucceeded
	}, time.Second, 10*time.Millisecond)

	require.Equal(t, "done", task.Result)
	require.Empty(t, task.Error)
}

func TestImportAsyncTaskManagerStoresError(t *testing.T) {
	manager := newImportAsyncTaskManager()

	taskID := manager.Create(context.Background(), func(context.Context) (any, error) {
		return nil, errors.New("boom")
	})

	require.Eventually(t, func() bool {
		task, ok := manager.Get(taskID)
		require.True(t, ok)
		return task.Status == importAsyncTaskStatusFailed
	}, time.Second, 10*time.Millisecond)

	task, ok := manager.Get(taskID)
	require.True(t, ok)
	require.Nil(t, task.Result)
	require.ErrorContains(t, task.Error, "boom")
}

func TestImportAsyncTaskManagerCleansExpiredCompletedTasks(t *testing.T) {
	manager := newImportAsyncTaskManagerWithRetention(100 * time.Millisecond)

	taskID, created := manager.CreateIfAbsent(context.Background(), "task-1", func(context.Context) (any, error) {
		return "done", nil
	})
	require.True(t, created)
	require.Equal(t, "task-1", taskID)

	require.Eventually(t, func() bool {
		task, ok := manager.Get(taskID)
		require.True(t, ok)
		return task.Status == importAsyncTaskStatusSucceeded
	}, time.Second, 10*time.Millisecond)

	require.Eventually(t, func() bool {
		_, ok := manager.Get(taskID)
		return !ok
	}, time.Second, 10*time.Millisecond)
}

func TestImportAsyncTaskManagerKeepsRunningTasks(t *testing.T) {
	manager := newImportAsyncTaskManagerWithRetention(time.Nanosecond)

	blocked := make(chan struct{})
	taskID, created := manager.CreateIfAbsent(context.Background(), "task-1", func(context.Context) (any, error) {
		<-blocked
		return "done", nil
	})
	require.True(t, created)

	require.Eventually(t, func() bool {
		task, ok := manager.Get(taskID)
		return ok && task.Status == importAsyncTaskStatusRunning
	}, time.Second, 10*time.Millisecond)

	time.Sleep(10 * time.Millisecond)
	task, ok := manager.Get(taskID)
	require.True(t, ok)
	require.Equal(t, importAsyncTaskStatusRunning, task.Status)

	close(blocked)
}