package runtimegrpc

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"errors"
	"fmt"
	"os"
	"strings"
	"sync/atomic"
	"time"

	identitysemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/identityv1"
	accesssemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordaccessv1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/envelope"
	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"
	"google.golang.org/grpc/metadata"
)

// RuntimeClientConfig is app-blind transport configuration shared by the
// neutral Runtime clients. Product code supplies only an AppInstallation
// selector and semantic request values.
type RuntimeClientConfig struct {
	Target                string
	AppInstallationID     string
	WorkloadToken         string
	CAFile                string
	ServerName            string
	ClientCertificateFile string
	ClientKeyFile         string
	// AttemptTimeout bounds each Runtime RPC independently. The parent context
	// remains the owner of the complete L1 route budget.
	AttemptTimeout time.Duration
}

type RuntimeClientMetrics struct {
	IdentityCalls uint64
	RecordsCalls  uint64
}

type RuntimeRecordAccessClient struct {
	connection          *grpc.ClientConn
	identity            *IdentityExchangeClient
	identityCoordinator *identityExchangeCoordinator
	records             *RecordAccessClient
	appInstallationID   string
	workloadToken       string
	attemptTimeout      time.Duration
	identityCalls       atomic.Uint64
	recordsCalls        atomic.Uint64
}

func DialRecordAccessRuntime(ctx context.Context, config RuntimeClientConfig) (*RuntimeRecordAccessClient, error) {
	return dialRecordAccessRuntime(ctx, config, newIdentityExchangeCoordinator(config))
}

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

func dialRecordAccessRuntime(ctx context.Context, config RuntimeClientConfig, coordinator *identityExchangeCoordinator) (*RuntimeRecordAccessClient, error) {
	if ctx == nil {
		return nil, errors.New("runtime 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 client: target, AppInstallation, workload token and attempt timeout are required")
	}
	transport, err := runtimeClientTransportCredentials(config)
	if err != nil {
		return nil, err
	}
	return dialRecordAccessRuntimeWithCredentialsAndCoordinator(ctx, config, transport, coordinator)
}

func dialRecordAccessRuntimeWithCredentials(ctx context.Context, config RuntimeClientConfig, transport credentials.TransportCredentials) (*RuntimeRecordAccessClient, error) {
	return dialRecordAccessRuntimeWithCredentialsAndCoordinator(ctx, config, transport, newIdentityExchangeCoordinator(config))
}

func dialRecordAccessRuntimeWithCredentialsAndCoordinator(ctx context.Context, config RuntimeClientConfig, transport credentials.TransportCredentials, coordinator *identityExchangeCoordinator) (*RuntimeRecordAccessClient, error) {
	if transport == nil {
		return nil, errors.New("runtime client: transport credentials are required")
	}
	connection, err := runtimeClientConnection(config.Target, transport, coordinator)
	if err != nil {
		return nil, fmt.Errorf("runtime client: create connection: %w", err)
	}
	generatedIdentity := runtimev1.NewIdentityContextServiceClient(connection)
	generatedRecords := runtimev1.NewRecordsServiceClient(connection)
	client := &RuntimeRecordAccessClient{
		connection: connection, appInstallationID: config.AppInstallationID, workloadToken: config.WorkloadToken,
		attemptTimeout: config.AttemptTimeout, 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.records, err = NewRecordAccessClient(func(callCtx context.Context, request *runtimev1.EvaluateRecordAccessRequest, options ...grpc.CallOption) (*runtimev1.EvaluateRecordAccessResponse, error) {
		client.recordsCalls.Add(1)
		return generatedRecords.EvaluateRecordAccess(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 client: connection shut down during startup")
		}
		if !connection.WaitForStateChange(ctx, state) {
			_ = connection.Close()
			return nil, fmt.Errorf("runtime client: Runtime not ready: %w", ctx.Err())
		}
	}
}

func (c *RuntimeRecordAccessClient) Evaluate(ctx context.Context, bearer string, request accesssemantic.Request) (accesssemantic.Response, error) {
	if c == nil || c.identity == nil || c.records == nil || c.connection == nil {
		return accesssemantic.Response{}, &accesssemantic.Failure{Category: accesssemantic.FailureUnavailable}
	}
	delegation, delegated := outgoingRuntimeDelegation(ctx, c.identityCoordinator, c.appInstallationID)
	if !delegated {
		identityCtx, cancelIdentity := context.WithTimeout(ctx, c.attemptTimeout)
		exchanged, err := c.identityCoordinator.exchange(identityCtx, bearer, identitysemantic.Request{
			AppInstallationID: c.appInstallationID, RequestedCapabilities: []string{accesssemantic.CapabilityName},
		}, c.identity.Exchange)
		cancelIdentity()
		if err != nil {
			return accesssemantic.Response{}, err
		}
		delegation = exchanged.OpaqueDelegation
	}
	recordsCtx, cancelRecords := context.WithTimeout(ctx, c.attemptTimeout)
	defer cancelRecords()
	return c.records.EvaluateRecordAccess(withRuntimeDelegation(recordsCtx, delegation), request)
}

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

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

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

func withWorkloadAuthorization(ctx context.Context, token string) context.Context {
	values, _ := metadata.FromOutgoingContext(ctx)
	values = values.Copy()
	values.Delete("authorization")
	values.Set("authorization", "Bearer "+token)
	return metadata.NewOutgoingContext(ctx, values)
}

func withRuntimeDelegation(ctx context.Context, delegation string) context.Context {
	values, _ := metadata.FromOutgoingContext(ctx)
	values = values.Copy()
	// The end-user bearer is consumed only by IdentityContext.Exchange. Records
	// receives the resulting Runtime delegation and must never inherit a bearer
	// from a reused caller context.
	values.Delete(EndUserAuthorizationMetadataKey)
	values.Delete(envelope.RuntimeDelegationMetadataKey)
	values.Set(envelope.RuntimeDelegationMetadataKey, delegation)
	return metadata.NewOutgoingContext(ctx, values)
}

func runtimeClientTransportCredentials(config RuntimeClientConfig) (credentials.TransportCredentials, error) {
	if config.CAFile == "" || config.ServerName == "" {
		return nil, errors.New("runtime client: CA file and server name are required for TLS")
	}
	caPEM, err := os.ReadFile(config.CAFile)
	if err != nil {
		return nil, fmt.Errorf("runtime client: read CA: %w", err)
	}
	roots := x509.NewCertPool()
	if !roots.AppendCertsFromPEM(caPEM) {
		return nil, errors.New("runtime client: CA contains no certificates")
	}
	tlsConfig := &tls.Config{MinVersion: tls.VersionTLS13, RootCAs: roots, ServerName: config.ServerName}
	if (config.ClientCertificateFile == "") != (config.ClientKeyFile == "") {
		return nil, errors.New("runtime client: client certificate and key must be supplied together")
	}
	if config.ClientCertificateFile != "" {
		certificate, err := tls.LoadX509KeyPair(config.ClientCertificateFile, config.ClientKeyFile)
		if err != nil {
			return nil, fmt.Errorf("runtime client: load client identity: %w", err)
		}
		tlsConfig.Certificates = []tls.Certificate{certificate}
	}
	return credentials.NewTLS(tlsConfig), nil
}