package runtimegrpc

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync/atomic"
	"time"

	identitysemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/identityv1"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"google.golang.org/grpc"
	"google.golang.org/grpc/connectivity"
	"google.golang.org/grpc/credentials"
)

// RuntimeRecordMutationClient composes the existing trusted identity exchange
// with the neutral mutation and status methods. The ordinary path is one
// mutation RPC. Only an outcome-ambiguous failure enters reconciliation, and a
// replay reuses the exact immutable command identity and content.
type RuntimeRecordMutationClient struct {
	connection           *grpc.ClientConn
	identity             *IdentityExchangeClient
	identityCoordinator  *identityExchangeCoordinator
	mutation             *MutationClient
	appInstallationID    string
	workloadToken        string
	attemptTimeout       time.Duration
	admissionRetryBudget time.Duration
	identityCalls        atomic.Uint64
	recordsCalls         atomic.Uint64
	admissionRetries     atomic.Uint64
	admissionExhausted   atomic.Uint64
}

func DialRecordMutationRuntime(ctx context.Context, config RuntimeClientConfig) (*RuntimeRecordMutationClient, error) {
	return dialRecordMutationRuntime(ctx, config, runtimeClientOptions{}, newIdentityExchangeCoordinator(config))
}

// DialRecordMutationRuntimeWithOptions adds reviewed behavior without changing
// the released RuntimeClientConfig layout.
func DialRecordMutationRuntimeWithOptions(ctx context.Context, config RuntimeClientConfig, values ...RuntimeClientOption) (*RuntimeRecordMutationClient, error) {
	options, err := resolveRuntimeClientOptions(values...)
	if err != nil {
		return nil, err
	}
	return dialRecordMutationRuntime(ctx, config, options, newIdentityExchangeCoordinator(config))
}

func (group *RuntimeClientGroup) DialRecordMutationRuntime(ctx context.Context) (*RuntimeRecordMutationClient, error) {
	if group == nil {
		return nil, errors.New("runtime client group: configuration is required")
	}
	return dialRecordMutationRuntime(ctx, group.config, group.options, group.identityCoordinator)
}

func dialRecordMutationRuntime(ctx context.Context, config RuntimeClientConfig, options runtimeClientOptions, coordinator *identityExchangeCoordinator) (*RuntimeRecordMutationClient, error) {
	if ctx == nil {
		return nil, errors.New("runtime mutation client: startup context is required")
	}
	if strings.TrimSpace(config.Target) == "" || strings.TrimSpace(config.Target) != config.Target ||
		strings.TrimSpace(config.AppInstallationID) == "" || strings.TrimSpace(config.AppInstallationID) != config.AppInstallationID ||
		strings.TrimSpace(config.WorkloadToken) == "" || strings.TrimSpace(config.WorkloadToken) != config.WorkloadToken ||
		config.AttemptTimeout <= 0 {
		return nil, errors.New("runtime mutation client: target, AppInstallation, workload token and attempt timeout are required")
	}
	transport, err := runtimeClientTransportCredentials(config)
	if err != nil {
		return nil, err
	}
	return dialRecordMutationRuntimeWithCredentialsAndCoordinator(ctx, config, options, transport, coordinator)
}

func dialRecordMutationRuntimeWithCredentials(ctx context.Context, config RuntimeClientConfig, transport credentials.TransportCredentials) (*RuntimeRecordMutationClient, error) {
	return dialRecordMutationRuntimeWithCredentialsAndCoordinator(ctx, config, runtimeClientOptions{}, transport, newIdentityExchangeCoordinator(config))
}

func dialRecordMutationRuntimeWithCredentialsAndCoordinator(ctx context.Context, config RuntimeClientConfig, options runtimeClientOptions, transport credentials.TransportCredentials, coordinator *identityExchangeCoordinator) (*RuntimeRecordMutationClient, error) {
	if transport == nil {
		return nil, errors.New("runtime mutation client: transport credentials are required")
	}
	connection, err := runtimeClientConnection(config.Target, transport, coordinator)
	if err != nil {
		return nil, fmt.Errorf("runtime mutation client: create connection: %w", err)
	}
	generatedIdentity := runtimev1.NewIdentityContextServiceClient(connection)
	generatedMutation := runtimev1.NewRecordMutationsServiceClient(connection)
	client := &RuntimeRecordMutationClient{
		connection: connection, appInstallationID: config.AppInstallationID,
		workloadToken: config.WorkloadToken, attemptTimeout: config.AttemptTimeout,
		admissionRetryBudget: options.recordMutationAdmissionRetryBudget,
		identityCoordinator:  coordinator,
	}
	client.identity, err = NewIdentityExchangeClient(func(callCtx context.Context, request *runtimev1.ExchangeRequest, options ...grpc.CallOption) (*runtimev1.ExchangeResponse, error) {
		client.identityCalls.Add(1)
		return generatedIdentity.Exchange(withWorkloadAuthorization(callCtx, client.workloadToken), request, options...)
	})
	if err != nil {
		_ = connection.Close()
		return nil, err
	}
	client.mutation, err = NewMutationClient(
		func(callCtx context.Context, options ...grpc.CallOption) (grpc.ClientStreamingClient[runtimev1.MutateRecordsRequest, runtimev1.MutateRecordsResponse], error) {
			client.recordsCalls.Add(1)
			return generatedMutation.MutateRecords(withWorkloadAuthorization(callCtx, client.workloadToken), options...)
		},
		func(callCtx context.Context, request *runtimev1.GetMutationStatusRequest, options ...grpc.CallOption) (*runtimev1.GetMutationStatusResponse, error) {
			client.recordsCalls.Add(1)
			return generatedMutation.GetMutationStatus(withWorkloadAuthorization(callCtx, client.workloadToken), request, options...)
		},
	)
	if err != nil {
		_ = connection.Close()
		return nil, err
	}
	connection.Connect()
	for {
		state := connection.GetState()
		if state == connectivity.Ready {
			return client, nil
		}
		if state == connectivity.Shutdown {
			_ = connection.Close()
			return nil, errors.New("runtime mutation client: connection shut down during startup")
		}
		if !connection.WaitForStateChange(ctx, state) {
			_ = connection.Close()
			return nil, fmt.Errorf("runtime mutation client: Runtime not ready: %w", ctx.Err())
		}
	}
}

func (c *RuntimeRecordMutationClient) Mutate(ctx context.Context, bearer string, request semantic.Request) (semantic.Response, error) {
	if c == nil || c.identity == nil || c.mutation == nil || c.connection == nil {
		return semantic.Response{}, &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	shape, err := semantic.CommandShapeFor(request)
	if err != nil {
		return semantic.Response{}, err
	}
	statusRequest, err := semantic.StatusRequestFor(request)
	if err != nil {
		return semantic.Response{}, err
	}
	retry := c.admissionRetryScope(ctx)
	delegation, err := retryExplicitBackoff(retry, func() (identitysemantic.Response, error) {
		return c.exchange(ctx, bearer)
	})
	if err != nil {
		return semantic.Response{}, err
	}
	response, err := retryExplicitBackoff(retry, func() (semantic.Response, error) {
		return c.mutateAttempt(ctx, delegation, request)
	})
	if err == nil || !requiresMutationReconciliation(err) {
		return response, err
	}
	statusResponse, statusErr := retryExplicitBackoff(retry, func() (semantic.StatusResponse, error) {
		return c.statusAttempt(ctx, delegation, shape, statusRequest)
	})
	if statusErr == nil && statusResponse.State == semantic.CommandCompleted {
		return mutationResponseFromStatus(request, statusResponse)
	}
	if statusErr != nil {
		// MISSING deliberately masks never-accepted, expired, and digest-mismatched
		// command bindings. It is therefore not durable proof that the first
		// attempt did not commit. Preserve the ambiguous original failure; only a
		// positively observed, non-terminal binding authorizes the exact replay.
		return semantic.Response{}, err
	}
	// Reconciliation positively observed the exact durable non-terminal command
	// binding. Replaying the exact command is now safe; the Runtime idempotency
	// binding resolves concurrent execution without duplicating effects. This is
	// the single recovery replay, not a generic retry loop.
	response, retryErr := retryExplicitBackoff(retry, func() (semantic.Response, error) {
		return c.mutateAttempt(ctx, delegation, request)
	})
	if retryErr == nil || !requiresMutationReconciliation(retryErr) {
		return response, retryErr
	}
	statusResponse, statusErr = retryExplicitBackoff(retry, func() (semantic.StatusResponse, error) {
		return c.statusAttempt(ctx, delegation, shape, statusRequest)
	})
	if statusErr == nil && statusResponse.State == semantic.CommandCompleted {
		return mutationResponseFromStatus(request, statusResponse)
	}
	return semantic.Response{}, retryErr
}

func (c *RuntimeRecordMutationClient) Status(ctx context.Context, bearer string, request semantic.Request) (semantic.StatusResponse, error) {
	if c == nil || c.identity == nil || c.mutation == nil || c.connection == nil {
		return semantic.StatusResponse{}, &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	shape, err := semantic.CommandShapeFor(request)
	if err != nil {
		return semantic.StatusResponse{}, err
	}
	statusRequest, err := semantic.StatusRequestFor(request)
	if err != nil {
		return semantic.StatusResponse{}, err
	}
	retry := c.admissionRetryScope(ctx)
	delegation, err := retryExplicitBackoff(retry, func() (identitysemantic.Response, error) {
		return c.exchange(ctx, bearer)
	})
	if err != nil {
		return semantic.StatusResponse{}, err
	}
	return retryExplicitBackoff(retry, func() (semantic.StatusResponse, error) {
		return c.statusAttempt(ctx, delegation, shape, statusRequest)
	})
}

func (c *RuntimeRecordMutationClient) admissionRetryScope(ctx context.Context) *admissionRetryScope {
	return newAdmissionRetryScope(
		ctx,
		c.admissionRetryBudget,
		func() { c.admissionRetries.Add(1) },
		func() { c.admissionExhausted.Add(1) },
	)
}

func (c *RuntimeRecordMutationClient) exchange(ctx context.Context, bearer string) (identitysemantic.Response, error) {
	attemptCtx, cancel := context.WithTimeout(ctx, c.attemptTimeout)
	defer cancel()
	return c.identityCoordinator.exchange(attemptCtx, bearer, identitysemantic.Request{
		AppInstallationID: c.appInstallationID, RequestedCapabilities: []string{semantic.CapabilityName},
	}, c.identity.Exchange)
}

func (c *RuntimeRecordMutationClient) mutateAttempt(ctx context.Context, delegation identitysemantic.Response, request semantic.Request) (semantic.Response, error) {
	attemptCtx, cancel := context.WithTimeout(ctx, c.attemptTimeout)
	defer cancel()
	return c.mutation.MutateRecords(withRuntimeDelegation(attemptCtx, delegation.OpaqueDelegation), request)
}

func (c *RuntimeRecordMutationClient) statusAttempt(ctx context.Context, delegation identitysemantic.Response, shape semantic.CommandShape, request semantic.StatusRequest) (semantic.StatusResponse, error) {
	attemptCtx, cancel := context.WithTimeout(ctx, c.attemptTimeout)
	defer cancel()
	return c.mutation.GetMutationStatus(withRuntimeDelegation(attemptCtx, delegation.OpaqueDelegation), shape, request)
}

func requiresMutationReconciliation(err error) bool {
	var failure *semantic.Failure
	return errors.As(err, &failure) && semantic.RetryDispositionFor(failure) == semantic.RetryReconcile
}

func mutationResponseFromStatus(request semantic.Request, status semantic.StatusResponse) (semantic.Response, error) {
	if status.State != semantic.CommandCompleted || status.Receipts == nil {
		return semantic.Response{}, &semantic.Failure{Category: semantic.FailureInternal}
	}
	response := semantic.Response{
		CommandID: status.CommandID, CommandDigest: status.CommandDigest,
		Results: status.Results, Receipts: *status.Receipts, OutcomeExpiresAtUTC: status.OutcomeExpiresAtUTC,
	}
	if err := semantic.ValidateResponse(request, response); err != nil {
		return semantic.Response{}, err
	}
	return response, nil
}

func (c *RuntimeRecordMutationClient) Ready() bool {
	return c != nil && runtimeConnectionReady(c.connection)
}

func (c *RuntimeRecordMutationClient) Metrics() RuntimeClientMetrics {
	if c == nil {
		return RuntimeClientMetrics{}
	}
	return RuntimeClientMetrics{IdentityCalls: c.identityCalls.Load(), RecordsCalls: c.recordsCalls.Load()}
}

// RuntimeAdmissionRetryMetrics is additive observability for the bounded
// admission retry policy. It is separate from the released RuntimeClientMetrics
// layout so existing unkeyed literals remain source-compatible.
type RuntimeAdmissionRetryMetrics struct {
	Retries   uint64
	Exhausted uint64
}

func (c *RuntimeRecordMutationClient) AdmissionRetryMetrics() RuntimeAdmissionRetryMetrics {
	if c == nil {
		return RuntimeAdmissionRetryMetrics{}
	}
	return RuntimeAdmissionRetryMetrics{Retries: c.admissionRetries.Load(), Exhausted: c.admissionExhausted.Load()}
}

func (c *RuntimeRecordMutationClient) Close() error {
	if c == nil || c.connection == nil {
		return nil
	}
	return c.connection.Close()
}