package impl

import (
	"context"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline/planner"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/pkg/lake"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

// This file holds helpers shared by the lifecycle coordinator and the
// observation model. Keeping them here means neither of those files depends on
// the other, so orchestration and observation stay separable.

const (
	tsoPhysicalShiftBits = 18

	// lakeContextTypeDataPipeline identifies this resource to Lake's token
	// exchange. It is part of the wire contract, so keep one definition.
	lakeContextTypeDataPipeline = "data_pipeline"
	// Lake writes all three context fields into the exchanged token and then
	// requires the matching headers on every business call - a partial context
	// is a 403. Data Pipeline has no revision of its own to report, so the
	// version is pinned; the isolation that matters is per-Pipeline, and
	// ContextID already carries that.
	lakeContextVersion = uint64(1)
)

// lakeInvocation builds the per-request Lake identity. Nothing here is
// persisted: the Lake adapter exchanges a fresh token for every business call.
func lakeInvocation(pipeline *entity.DataPipeline) lake.Invocation {
	return lake.Invocation{
		TiDBOrgID:      strconv.FormatUint(pipeline.TenantID.Uint64(), 10),
		LakeOrgID:      pipeline.Params.Destination.LakeOrgID,
		ContextType:    lakeContextTypeDataPipeline,
		ContextID:      strconv.FormatUint(pipeline.ID.Uint64(), 10),
		ContextVersion: lakeContextVersion,
		InitiatedBy:    strconv.FormatUint(pipeline.CreatorID.Uint64(), 10),
	}
}

// tsoTime extracts the physical timestamp from a TSO. A zero or unrepresentable
// TSO yields the zero time so callers never publish a fabricated position.
func tsoTime(tso uint64) time.Time {
	if tso == 0 {
		return time.Time{}
	}
	physicalMillis := tso >> tsoPhysicalShiftBits
	if physicalMillis > uint64(^uint64(0)>>1) {
		return time.Time{}
	}
	return time.UnixMilli(int64(physicalMillis)).UTC()
}

// sanitizeChildMessage prepares a child service message for the persisted read
// model, which Get/List return verbatim.
//
// The caller-supplied values are not enough on their own: a TiCDC kernel error
// embeds the sink URI, and this service builds S3 sink URIs with the secret
// access key in the query string. The observer only holds that key as
// ciphertext, so it can never pass it here - the URI has to be redacted
// structurally instead.
func sanitizeChildMessage(message string, values ...string) string {
	const maxMessageBytes = 2048
	result := redactURICredentials(message)
	for _, value := range values {
		if value != "" {
			result = strings.ReplaceAll(result, value, "[REDACTED]")
		}
	}
	if len(result) > maxMessageBytes {
		result = result[:maxMessageBytes]
	}
	return result
}

// uriCredentialPattern matches the object-storage credential query parameters
// this service puts into sink URIs, wherever they appear inside a message.
var uriCredentialPattern = regexp.MustCompile(
	`(?i)\b(secret-access-key|access-key|sas-token|secret_access_key|access_key_id)=[^&\s"']*`)

func redactURICredentials(message string) string {
	return uriCredentialPattern.ReplaceAllString(message, "${1}=[REDACTED]")
}

func validateStagingInput(
	staging domain.DataPipelineStagingInput,
) (string, string, bizerr.BizError) {
	if staging.Type != domain.DataPipelineStagingTypeCustomerManagedS3 || staging.S3 == nil {
		return "", "", bizerr.New(bizerr.ErrCodeParams, "customer-managed S3 staging is required")
	}
	if strings.TrimSpace(staging.S3.Region) == "" {
		return "", "", bizerr.New(bizerr.ErrCodeParams, "S3 region is required")
	}
	if credentialErr := validateStagingAuth(*staging.S3); credentialErr != nil {
		return "", "", credentialErr
	}
	if queueErr := validateStagingNotification(*staging.S3); queueErr != nil {
		return "", "", queueErr
	}
	bucket, prefix, scheme, err := planner.ParseS3BaseURI(staging.S3.URI)
	if err != nil {
		return "", "", invalidPlanError(err)
	}
	if providerErr := validateStagingProvider(*staging.S3, scheme); providerErr != nil {
		return "", "", providerErr
	}
	return bucket, prefix, nil
}

// validateStagingProvider keeps the provider, the URI scheme, the endpoint and
// the queue consistent with each other. Each of these is separately valid and
// only wrong in combination, which single-field rules cannot express.
func validateStagingProvider(
	s3 domain.DataPipelineS3StagingInput, scheme planner.StagingScheme,
) bizerr.BizError {
	if want := planner.SchemeForProvider(s3.Provider); scheme != want {
		return bizerr.New(bizerr.ErrCodeParams,
			"staging.s3.uri must use the %q scheme for this provider, but it uses %q",
			want, scheme)
	}
	if s3.Provider != domain.DataPipelineS3ProviderAlibabaOSS {
		// An endpoint is only meaningful for a non-AWS provider. Accepting one
		// for AWS would send the children somewhere the snapshot cannot follow:
		// Export addresses AWS staging by bucket and region alone.
		if strings.TrimSpace(s3.Endpoint) != "" {
			return bizerr.New(bizerr.ErrCodeParams,
				"staging.s3.endpoint is only supported for the Alibaba OSS provider")
		}
		return nil
	}
	// The changefeed and the destination both reach OSS through this endpoint,
	// and the changefeed has no other way: without it TiCDC falls back to the
	// AWS bucket-region probe, which does not answer for OSS.
	//
	// The snapshot does not read it. cluster-service-ng derives its own from
	// the bucket's region and hardcodes the public form
	// (https://oss-<region>.aliyuncs.com), because OSSCloudStorage carries no
	// endpoint field to pass one through. A private or VPC endpoint therefore
	// applies to the incremental half and not to the snapshot - see the OSS
	// branch of buildExportOutputURI in that repository.
	if strings.TrimSpace(s3.Endpoint) == "" {
		return bizerr.New(bizerr.ErrCodeParams,
			"staging.s3.endpoint is required for the Alibaba OSS provider")
	}
	// Export refuses OSS with a role: nextgen-global-service answers
	// "oss role_arn is not supported yet" for that combination. Accepting it
	// here would defer the same refusal to the snapshot task, where it arrives
	// as a failed Pipeline rather than a rejected request.
	if s3.UsesRoleARN() {
		return bizerr.New(bizerr.ErrCodeParams,
			"staging.s3.role_arn is not supported for the Alibaba OSS provider yet; "+
				"use access key credentials")
	}
	// The destination cannot consume notifications for an OSS bucket yet, so a
	// queue here would be accepted and then never read. Rejecting it keeps the
	// delivery plan honest: an OSS staging area is polled.
	if strings.TrimSpace(s3.SQSQueueURL) != "" {
		return bizerr.New(bizerr.ErrCodeParams,
			"staging.s3.sqs_queue_url is not supported for the Alibaba OSS provider yet; "+
				"an OSS staging area is polled")
	}
	return nil
}

// validateStagingAuth enforces that exactly one credential shape is supplied.
// The two modes are mutually exclusive, and accepting both would leave it
// ambiguous which one the children were actually configured with.
func validateStagingAuth(s3 domain.DataPipelineS3StagingInput) bizerr.BizError {
	hasKeyPair := strings.TrimSpace(s3.AccessKeyID) != "" || s3.SecretAccessKey != ""
	hasRole := strings.TrimSpace(s3.RoleARN) != ""
	switch s3.AuthType {
	case domain.DataPipelineS3AuthTypeAccessKey:
		if strings.TrimSpace(s3.AccessKeyID) == "" || s3.SecretAccessKey == "" {
			return bizerr.New(bizerr.ErrCodeParams, "S3 access key id and secret key are required")
		}
		if hasRole {
			return bizerr.New(bizerr.ErrCodeParams, "S3 role ARN must be empty for access key auth")
		}
	case domain.DataPipelineS3AuthTypeRoleARN:
		if !hasRole {
			return bizerr.New(bizerr.ErrCodeParams, "S3 role ARN is required")
		}
		if hasKeyPair {
			return bizerr.New(bizerr.ErrCodeParams, "S3 access key must be empty for role ARN auth")
		}
	default:
		return bizerr.New(bizerr.ErrCodeParams, "S3 auth type is required")
	}
	return nil
}

// validateStagingNotification checks the optional notification queue.
//
// The queue type is deliberately not constrained. The destination's own
// documentation calls for a FIFO queue, but S3 event notifications cannot
// target one - S3 rejects the configuration outright, because it supplies
// neither of the two identifiers a FIFO queue requires per message. Enforcing
// FIFO here would therefore reject the only wiring a customer can set up
// directly, and enforcing standard would reject a queue fed through
// EventBridge. Whether the queue can actually be consumed is answered by the
// destination probing it during staging validation, which is where an answer
// can be based on more than the name.
func validateStagingNotification(s3 domain.DataPipelineS3StagingInput) bizerr.BizError {
	if strings.TrimSpace(s3.SQSQueueURL) != "" {
		return nil
	}
	// No queue is the default: the destination lists the staging prefix on the
	// planned interval. An SQS region without a queue is a configuration that
	// says nothing, so it is rejected rather than silently ignored.
	if strings.TrimSpace(s3.SQSRegion) != "" {
		return bizerr.New(bizerr.ErrCodeParams, "SQS region requires an SQS queue URL")
	}
	return nil
}

func invalidPlanError(err error) bizerr.BizError {
	return bizerr.New(bizerr.ErrCodeParams, "invalid data pipeline configuration: %s", err.Error())
}

// setPipelineStatus is the only way a Pipeline's status should change. It keeps
// StatusEnteredAt in step, which is what every "how long has it been like
// this?" question reads - the stuck-lifecycle alerts depend on it being
// written exactly when the status moves, and on nothing else touching it.
func setPipelineStatus(
	ctx context.Context,
	pipeline *entity.DataPipeline,
	status entity.DataPipelineStatus,
	now time.Time,
) {
	if pipeline == nil || pipeline.Status == status {
		return
	}
	entered := now.UTC()
	// Logged here rather than at the ten call sites because this is already the
	// only place a status may change, and the guard above means it fires on a
	// real transition and not on a no-op re-save. Without it the lifecycle has
	// to be reconstructed from which DAG actions ran, which says what was
	// attempted rather than what the Pipeline became.
	log.GetLogger(ctx).Info(
		"data pipeline status changed",
		zap.Uint64("clusterID", pipeline.ClusterID.Uint64()),
		zap.String("from", string(pipeline.Status)),
		zap.String("to", string(status)),
		zap.Duration("heldPreviousFor", now.UTC().Sub(statusSince(pipeline))),
	)
	pipeline.Status = status
	pipeline.StatusEnteredAt = &entered
}

// statusSince reports when a Pipeline entered its current status, falling back
// to creation for rows written before StatusEnteredAt existed.
func statusSince(pipeline *entity.DataPipeline) time.Time {
	if pipeline.StatusEnteredAt != nil {
		return *pipeline.StatusEnteredAt
	}
	return pipeline.CreatedAt
}