package matcher

import (
	"context"
	"fmt"

	"github.com/pingcap/log"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/agent"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	importcfg "github.com/tidbcloud/dataflow-service/pkg/import-config"
	storagepkg "github.com/tidbcloud/dataflow-service/pkg/storage"
	"go.uber.org/zap"
)

type Matcher interface {
	Match(ctx context.Context, targetSource *agent.SearchImportFilesReq_TargetSource) (matchedUri string, err error)
}

type DefaultMatcher struct {
	client     storagepkg.StorageClient
	bucket     string
	uriPrefix  string
	dataFormat uint32
}

func NewDefaultMatcher(client storagepkg.StorageClient, bucket, uriPrefix string, dataFormat uint32) *DefaultMatcher {
	return &DefaultMatcher{
		client:     client,
		bucket:     bucket,
		uriPrefix:  uriPrefix,
		dataFormat: dataFormat,
	}
}

func (m *DefaultMatcher) Match(ctx context.Context, targetSource *agent.SearchImportFilesReq_TargetSource) (string, error) {
	schema, table := targetSource.TargetTable.Schema, targetSource.TargetTable.Table
	prefixes := []string{
		fmt.Sprintf("%s.%s.", schema, table),
		fmt.Sprintf("%s.%s_", schema, table),
	}

	if m.dataFormat == uint32(dataflow.ImportDataFormat_AURORA_SNAPSHOT) {
		prefixes = []string{fmt.Sprintf("%s/%s.%s/", schema, schema, table)}
	}

	for _, prefix := range prefixes {
		objectUris, err := m.client.ListObjects(ctx, m.bucket, m.uriPrefix+prefix)
		if err != nil {
			return "", err
		}
		for _, uri := range objectUris {
			match := importcfg.MatchDefaultNamingRule(uri, schema, table, m.dataFormat)

			if match {
				log.L().Info("matched object", zap.String("key", uri), zap.Any("table", targetSource.TargetTable))
				return uri, nil
			}

			log.L().Info("mismatched object", zap.String("key", uri), zap.Any("table", targetSource.TargetTable))
		}
	}

	return "", nil
}