package br

import (
	"context"
	"encoding/json"
	"fmt"
	"net/url"
	"path"
	"time"

	"github.com/tidbcloud/dataflow-service/app/models/domain"
	bucketRepo "github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/internal/model/util"
	"github.com/tidbcloud/dataflow-service/internal/utils"
	"github.com/tidbcloud/dataflow-service/pkg/blob"
	"github.com/tidbcloud/pkgv2/log"
	"go.uber.org/zap"
)

const (
	CRPiTRResumeStateFileDir = "crr-checkpoint"
	CRPiTRResumeStateFile    = "resume-state.json"
)

// PersistentState mirrors the checkpoint file written by the CRPiTR replicator.
// Refer to br/pkg/task/stream.go
type PersistentState struct {
	LastCheckpoint uint64 `json:"last_checkpoint"`
}

type BRCommon struct {
	bucketRepo bucketRepo.BucketRepo
}

func NewBRCommon(bucketRepo bucketRepo.BucketRepo) *BRCommon {
	return &BRCommon{
		bucketRepo: bucketRepo,
	}
}

// GetCheckpointForCRPiTR reads resume-state.json from the given bucket/region
// and returns the checkpoint as Unix seconds.
// Returns (0, false, nil) if the region is not covered by any bucket snapshot or
// if the checkpoint file is not found; the caller should fall back to the primary
// region's end timestamp in that case.
// Returns (0, false, err) on any other error.
func (b *BRCommon) GetCheckpointForCRPiTR(
	ctx context.Context,
	logBackup entity.LogBackup,
	region string,
	getStorageFunc blob.GetStorageFunc,
) (int64, bool, error) {
	logger := log.WithContextL(ctx).
		With(zap.Uint64("logBackupID", logBackup.ID.Uint64()), zap.String("region", region))
	if logBackup.BucketSettingSnapshot == nil {
		return 0, false, nil
	}
	bucketSnapshot, found := logBackup.BucketSettingSnapshot.GetBucketByRegion(region)
	if !found {
		return 0, false, nil
	}
	bucket, loadErr := b.bucketRepo.Load(ctx, bucketSnapshot.BucketId)
	if loadErr != nil {
		return 0, false, fmt.Errorf("failed to load bucket: %w", loadErr)
	}

	storageURI, err := crPiTRCheckpointStorageURI(bucket, logBackup)
	if err != nil {
		return 0, false, err
	}

	opt := BuildStorageOption(bucket, region)
	storage, err := getStorageFunc(ctx, storageURI, opt)
	if err != nil {
		if blob.IsNotFoundError(err) {
			return 0, false, nil
		}
		logger.Warn("GetCheckpointForCRPiTR: failed to open storage",
			zap.String("storageURI", utils.RedactSourceURI(storageURI)),
			zap.Error(err))
		return 0, false, fmt.Errorf("failed to open storage: %w", err)
	}

	data, err := storage.ReadFile(ctx, CRPiTRResumeStateFile)
	if err != nil {
		if blob.IsNotFoundError(err) {
			return 0, false, nil
		}
		logger.Warn("GetCheckpointForCRPiTR: failed to read resume-state.json",
			zap.String("storageURI", utils.RedactSourceURI(storageURI)),
			zap.Error(err))
		return 0, false, fmt.Errorf("failed to read resume-state.json: %w", err)
	}

	var state PersistentState
	if err := json.Unmarshal(data, &state); err != nil {
		logger.Warn("GetCheckpointForCRPiTR: failed to parse resume-state.json",
			zap.String("storageURI", utils.RedactSourceURI(storageURI)),
			zap.Error(err))
		return 0, false, fmt.Errorf("failed to parse resume-state.json: %w", err)
	}

	// A zero checkpoint means the replicator has not yet written a valid
	// checkpoint; treat it as not-found so the caller falls back gracefully.
	if state.LastCheckpoint == 0 {
		return 0, false, nil
	}
	ts := util.TransToTS(state.LastCheckpoint)
	logger.Info("GetCheckpointForCRPiTR: successfully got checkpoint",
		zap.Time("lastCheckpoint", time.Unix(ts, 0)))
	return ts, true, nil
}

// crPiTRCheckpointStorageURI returns the replica storage URI pointing at the
// CRPiTR checkpoint sub-directory inside the given bucket. It replaces the
// bucket host in sourceURI (for AWS) so reads land in the replica, then
// appends the checkpoint sub-directory.
func crPiTRCheckpointStorageURI(bucket *entity.Bucket, logBackup entity.LogBackup) (string, error) {
	uri, err := bucket.URIInBucket(logBackup.Storage)
	if err != nil {
		return "", fmt.Errorf("failed to build replica storage URI: %w", err)
	}
	parsed, err := url.Parse(uri)
	if err != nil {
		return "", fmt.Errorf("failed to parse replica storage URI: %w", err)
	}
	parsed.Path = path.Join(parsed.Path, CRPiTRResumeStateFileDir)
	return parsed.String(), nil
}

// BuildStorageOption constructs a blob.StorageOption for accessing the given
// bucket in the specified region, including any provider-specific credentials.
func BuildStorageOption(bucket *entity.Bucket, region string) *blob.StorageOption {
	opt := &blob.StorageOption{
		ClusterRegion: region,
		ProjectID:     bucket.Uint64ProjectID(),
	}
	if bucket.InfraProviderType == entity.InfraProviderTypeAWS ||
		bucket.InfraProviderType == entity.InfraProviderTypeEKS {
		if bucket.CloudAccountID != nil {
			opt.AWSRoleARN = domain.BuildDataflowServiceInfraRoleARN(*bucket.CloudAccountID)
		}
	}
	return opt
}