package repository

import (
	"context"
	"database/sql"
	"fmt"
	"path/filepath"
	"testing"

	"github.com/pingcap/log"
	"github.com/tj/assert"
)

func newTestCacheRepo(t *testing.T, sources []string) *CacheClusterBranchRepo {
	t.Helper()
	cacheFile := filepath.Join(t.TempDir(), "cache.db")
	db, err := sql.Open("duckdb", cacheFile)
	assert.Nil(t, err)
	_, err = db.Exec(createClusterBranchTableSQL)
	assert.Nil(t, err)
	t.Cleanup(func() { db.Close() })
	return &CacheClusterBranchRepo{ttlMinutes: 30, db: db, sources: sources}
}

func recordingFns(names []string, cluster *ClusterBranch, err error, calls *[]string) map[string]func(context.Context, string) (*ClusterBranch, error) {
	fns := make(map[string]func(context.Context, string) (*ClusterBranch, error), len(names))
	for _, name := range names {
		fns[name] = func(_ context.Context, _ string) (*ClusterBranch, error) {
			*calls = append(*calls, name)
			return cluster, err
		}
	}
	return fns
}

func TestDoQueryFallsThroughOnNotFound(t *testing.T) {
	var calls []string
	notFound := fmt.Errorf("cluster not found")
	repo := newTestCacheRepo(t, []string{"local", "ddb", "infra"})
	fns := recordingFns([]string{"local", "ddb", "infra"}, nil, notFound, &calls)

	_, err := repo.doQuery(context.TODO(), log.L(), "1001", fns)
	assert.NotNil(t, err)
	assert.Equal(t, []string{"local", "ddb", "infra"}, calls)
}

func TestDoQueryStopsAtFirstHitAndCachesLocally(t *testing.T) {
	cluster := &ClusterBranch{Type: CBTypeCluster, PrimaryID: "1001", KeyspaceID: 1}
	var calls []string
	notFound := fmt.Errorf("cluster not found")
	repo := newTestCacheRepo(t, []string{"local", "ddb", "infra"})
	fns := recordingFns([]string{"local", "ddb", "infra"}, cluster, nil, &calls)
	fns["local"] = func(_ context.Context, _ string) (*ClusterBranch, error) {
		calls = append(calls, "local")
		return nil, notFound
	}

	got, err := repo.doQuery(context.TODO(), log.L(), "1001", fns)
	assert.Nil(t, err)
	assert.Equal(t, cluster, got)
	assert.Equal(t, []string{"local", "ddb"}, calls)

	// the hit from ddb must be written to the local duckdb cache
	cached, err := repo.getItemByPrimaryId(context.TODO(), "1001")
	assert.Nil(t, err)
	assert.Equal(t, cluster, cached)

	// the cached item carries both keys
	cachedByKeyspace, err := repo.getItemByKeyspaceId(context.TODO(), "1")
	assert.Nil(t, err)
	assert.Equal(t, cluster, cachedByKeyspace)
}

func TestDoQueryLocalHitSkipsCacheWrite(t *testing.T) {
	cluster := &ClusterBranch{Type: CBTypeCluster, PrimaryID: "1001", KeyspaceID: 1}
	var calls []string
	repo := newTestCacheRepo(t, []string{"local", "ddb"})
	fns := recordingFns([]string{"local", "ddb"}, cluster, nil, &calls)

	got, err := repo.doQuery(context.TODO(), log.L(), "1001", fns)
	assert.Nil(t, err)
	assert.Equal(t, cluster, got)
	assert.Equal(t, []string{"local"}, calls)
}

func TestDoQueryReturnsNonNotFoundErrorImmediately(t *testing.T) {
	var calls []string
	boom := fmt.Errorf("boom")
	repo := newTestCacheRepo(t, []string{"local", "ddb"})
	fns := recordingFns([]string{"local", "ddb"}, nil, boom, &calls)

	_, err := repo.doQuery(context.TODO(), log.L(), "1001", fns)
	assert.Equal(t, boom, err)
	assert.Equal(t, []string{"local"}, calls)
}