package runtimeapi

import (
	"context"
	"database/sql/driver"
	"errors"
	"strings"
	"testing"

	"github.com/nexuscrm/kernel/pkg/telemetry"
	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/catalog/v1"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/testutil"
	"google.golang.org/grpc"
	"google.golang.org/grpc/metadata"
)

type metricsDriverConn struct{}

func (*metricsDriverConn) Prepare(string) (driver.Stmt, error) { return nil, errors.New("unused") }
func (*metricsDriverConn) Close() error                        { return nil }
func (*metricsDriverConn) Begin() (driver.Tx, error)           { return nil, errors.New("unused") }
func (*metricsDriverConn) QueryContext(context.Context, string, []driver.NamedValue) (driver.Rows, error) {
	return nil, errors.New("observed")
}

func observedSQLCounter(statements int) *telemetry.SQLStatementCounter {
	ctx, counter := telemetry.WithSQLStatementCounter(context.Background())
	conn := &telemetry.TracingConn{Conn: &metricsDriverConn{}}
	for range statements {
		_, _ = conn.QueryContext(ctx, "SELECT 1", nil)
	}
	return counter
}

func TestRuntimeRPCSQLObservationUsesClosedGeneratedMethodIdentity(t *testing.T) {
	metrics := newRecordsMetrics(nil)
	interceptor := metrics.sqlObservationInterceptor()
	_, _ = interceptor(context.Background(), nil, &grpc.UnaryServerInfo{
		FullMethod: runtimev1.RecordsService_EvaluateRecordAccess_FullMethodName,
	}, func(ctx context.Context, _ any) (any, error) {
		stageCtx, _ := telemetry.WithSQLStatementCounter(ctx)
		_, _ = (&telemetry.TracingConn{Conn: &metricsDriverConn{}}).QueryContext(stageCtx, "SELECT 1", nil)
		return nil, nil
	})
	_, _ = interceptor(context.Background(), nil, &grpc.UnaryServerInfo{
		FullMethod: runtimev1.EffectivePermissionService_ReadEffectivePermissions_FullMethodName,
	}, func(ctx context.Context, _ any) (any, error) {
		_, _ = (&telemetry.TracingConn{Conn: &metricsDriverConn{}}).QueryContext(ctx, "SELECT 1", nil)
		return nil, nil
	})

	registry := prometheus.NewPedanticRegistry()
	if err := registry.Register(metrics); err != nil {
		t.Fatal(err)
	}
	expected := `
# HELP nexuscrm_runtime_rpc_sql_statements_total Context-propagating SQL statement attempts across one complete admitted Runtime RPC.
# TYPE nexuscrm_runtime_rpc_sql_statements_total counter
nexuscrm_runtime_rpc_sql_statements_total{operation="evaluate_record_access"} 1
nexuscrm_runtime_rpc_sql_statements_total{operation="count_related_records"} 0
nexuscrm_runtime_rpc_sql_statements_total{operation="get_mutation_status"} 0
nexuscrm_runtime_rpc_sql_statements_total{operation="identity_exchange"} 0
nexuscrm_runtime_rpc_sql_statements_total{operation="list_recycle_bin"} 0
nexuscrm_runtime_rpc_sql_statements_total{operation="mutate_records"} 0
nexuscrm_runtime_rpc_sql_statements_total{operation="preview_create_defaults"} 0
nexuscrm_runtime_rpc_sql_statements_total{operation="purge_recycle_bin"} 0
nexuscrm_runtime_rpc_sql_statements_total{operation="query_history"} 0
nexuscrm_runtime_rpc_sql_statements_total{operation="query_records"} 0
nexuscrm_runtime_rpc_sql_statements_total{operation="read_effective_permissions"} 1
nexuscrm_runtime_rpc_sql_statements_total{operation="read_record_view"} 0
nexuscrm_runtime_rpc_sql_statements_total{operation="read_selected_root"} 0
nexuscrm_runtime_rpc_sql_statements_total{operation="resolve_attachment_sources"} 0
nexuscrm_runtime_rpc_sql_statements_total{operation="restore_recycle_bin"} 0
`
	if err := testutil.GatherAndCompare(registry, strings.NewReader(expected), "nexuscrm_runtime_rpc_sql_statements_total"); err != nil {
		t.Fatal(err)
	}
}

func TestRuntimeRPCProfileSplitsTheExistingSecurityAdmissionBoundary(t *testing.T) {
	metrics := newRecordsMetrics(nil)
	info := &grpc.UnaryServerInfo{
		FullMethod: runtimev1.EffectivePermissionService_ReadEffectivePermissions_FullMethodName,
	}
	outer := metrics.sqlObservationInterceptor()
	postSecurity := metrics.postSecurityObservationInterceptor()
	_, err := outer(context.Background(), nil, info, func(ctx context.Context, request any) (any, error) {
		securityCtx, finishSecurity := beginRuntimeRPCProfilePhase(ctx, runtimeRPCPhaseSecurityState)
		_, _ = (&telemetry.TracingConn{Conn: &metricsDriverConn{}}).QueryContext(securityCtx, "SELECT security admission", nil)
		finishSecurity()
		return postSecurity(ctx, request, info, func(handlerCtx context.Context, _ any) (any, error) {
			_, _ = (&telemetry.TracingConn{Conn: &metricsDriverConn{}}).QueryContext(handlerCtx, "SELECT handler one", nil)
			_, _ = (&telemetry.TracingConn{Conn: &metricsDriverConn{}}).QueryContext(handlerCtx, "SELECT handler two", nil)
			return nil, nil
		})
	})
	if err != nil {
		t.Fatal(err)
	}
	profile := metrics.rpcProfile(runtimeOperationEffectivePermissions)
	totalProfile := profile.phases[runtimeRPCPhaseTotal]
	postProfile := profile.phases[runtimeRPCPhasePostSecurityAdmission]
	securityProfile := profile.phases[runtimeRPCPhaseSecurityState]
	if got := totalProfile.observations.Load(); got != 1 {
		t.Fatalf("total observations = %d, want 1", got)
	}
	if got := postProfile.observations.Load(); got != 1 {
		t.Fatalf("post-security observations = %d, want 1", got)
	}
	if total, post := totalProfile.nanos.Load(), postProfile.nanos.Load(); total < post {
		t.Fatalf("total duration = %d, post-security duration = %d", total, post)
	}
	if total, post := totalProfile.sqlStatements.Load(), postProfile.sqlStatements.Load(); total != 3 || post != 2 {
		t.Fatalf("SQL statements total=%d post-security=%d, want 3 and 2", total, post)
	}
	if total, post := totalProfile.sqlNanos.Load(), postProfile.sqlNanos.Load(); total < post || post == 0 {
		t.Fatalf("SQL duration total=%d post-security=%d", total, post)
	}
	if got := securityProfile.sqlStatements.Load(); got != 1 {
		t.Fatalf("security-state SQL statements = %d, want 1", got)
	}

	registry := prometheus.NewPedanticRegistry()
	if err := registry.Register(metrics); err != nil {
		t.Fatal(err)
	}
	expected := `
# HELP nexuscrm_runtime_rpc_profile_observations_total Runtime RPC observations split around the Runtime SecurityAdmission boundary.
# TYPE nexuscrm_runtime_rpc_profile_observations_total counter
nexuscrm_runtime_rpc_profile_observations_total{operation="read_effective_permissions",phase="post_security_admission"} 1
nexuscrm_runtime_rpc_profile_observations_total{operation="read_effective_permissions",phase="security_state"} 1
nexuscrm_runtime_rpc_profile_observations_total{operation="read_effective_permissions",phase="total"} 1
`
	if err := testutil.GatherAndCompare(
		registry, strings.NewReader(expected), "nexuscrm_runtime_rpc_profile_observations_total",
	); err != nil {
		t.Fatal(err)
	}
	sqlExpected := `
# HELP nexuscrm_runtime_rpc_profile_sql_statements_total Context-propagating SQL statement attempts split around the Runtime SecurityAdmission boundary.
# TYPE nexuscrm_runtime_rpc_profile_sql_statements_total counter
nexuscrm_runtime_rpc_profile_sql_statements_total{operation="read_effective_permissions",phase="post_security_admission"} 2
nexuscrm_runtime_rpc_profile_sql_statements_total{operation="read_effective_permissions",phase="security_state"} 1
nexuscrm_runtime_rpc_profile_sql_statements_total{operation="read_effective_permissions",phase="total"} 3
`
	if err := testutil.GatherAndCompare(
		registry, strings.NewReader(sqlExpected), "nexuscrm_runtime_rpc_profile_sql_statements_total",
	); err != nil {
		t.Fatal(err)
	}
}

func TestRuntimeRPCProfileUsesClosedPageAndActionMethodIdentities(t *testing.T) {
	tests := map[string]string{
		catalogv1.AppMetadataViewV3Service_GetAppMetadataViewV3_FullMethodName:              runtimeOperationAppMetadataV3,
		runtimev1.RecordSearchService_ReadSearchScopeOrder_FullMethodName:                   runtimeOperationSearchScopeOrder,
		runtimev1.FlowExecutionService_ExecuteFlow_FullMethodName:                           runtimeOperationExecuteFlow,
		runtimev1.ApprovalService_ReadApproval_FullMethodName:                               runtimeOperationReadApproval,
		runtimev1.ApprovalService_ExecuteApproval_FullMethodName:                            runtimeOperationExecuteApproval,
		runtimev1.PermissionAuthoringService_ReadPermissionAuthoringSnapshot_FullMethodName: runtimeOperationPermissionRead,
		runtimev1.PermissionAuthoringService_SavePermissionAuthoring_FullMethodName:         runtimeOperationPermissionSave,
		runtimev1.PermissionAuthoringService_GetPermissionAuthoringStatus_FullMethodName:    runtimeOperationPermissionStatus,
	}
	metrics := newRecordsMetrics(nil)
	for fullMethod, want := range tests {
		if got, ok := runtimeMetricOperationName(fullMethod); !ok || got != want {
			t.Fatalf("runtimeMetricOperationName(%q) = %q, %v; want %q, true", fullMethod, got, ok, want)
		}
		if profile := metrics.rpcProfile(want); profile == nil {
			t.Fatalf("runtime metric operation %q has no registered profile", want)
		}
	}
	if got, ok := runtimeMetricOperationName("/unknown.Service/Method"); ok || got != "" {
		t.Fatalf("unknown method = %q, %v; want closed rejection", got, ok)
	}
}

func TestSecurityAdmissionPublishesClosedInnerRPCProfilePhases(t *testing.T) {
	metrics := newRecordsMetrics(nil)
	info := &grpc.UnaryServerInfo{FullMethod: runtimev1.RecordsService_ReadSelectedRoot_FullMethodName}
	_, err := metrics.sqlObservationInterceptor()(context.Background(), nil, info, func(profileCtx context.Context, _ any) (any, error) {
		return nil, invokeSecurityChainWithContext(
			t, profileCtx, testSecurityAdmission(t), testWorkloadID, testDelegationToken, 1, func(context.Context) {},
		)
	})
	if err != nil {
		t.Fatal(err)
	}
	profile := metrics.rpcProfile(recordsOperationSelectedRoot)
	for _, phase := range []string{
		runtimeRPCPhaseWorkloadBinding,
		runtimeRPCPhaseDelegationVerification,
		runtimeRPCPhaseSecurityState,
		runtimeRPCPhaseSession,
	} {
		if got := profile.phases[phase].observations.Load(); got != 1 {
			t.Fatalf("phase %s observations = %d, want 1", phase, got)
		}
	}
}

func TestRuntimeMutationStreamSQLObservationCoversWholeHandler(t *testing.T) {
	metrics := newRecordsMetrics(nil)
	stream := &metricsServerStream{ctx: context.Background()}
	err := metrics.sqlObservationStreamInterceptor()(nil, stream, &grpc.StreamServerInfo{
		FullMethod:     runtimev1.RecordMutationsService_MutateRecords_FullMethodName,
		IsClientStream: true,
	}, func(_ any, admitted grpc.ServerStream) error {
		_, _ = (&telemetry.TracingConn{Conn: &metricsDriverConn{}}).QueryContext(admitted.Context(), "SELECT 1", nil)
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if got := metrics.mutateRPCSQLStatements.Load(); got != 1 {
		t.Fatalf("mutation RPC SQL statements = %d, want 1", got)
	}
	if got := metrics.mutationStatusRPCSQLStatements.Load(); got != 0 {
		t.Fatalf("mutation status SQL statements = %d, want 0", got)
	}
}

type metricsServerStream struct {
	ctx context.Context
}

func (s *metricsServerStream) SetHeader(metadata.MD) error  { return nil }
func (s *metricsServerStream) SendHeader(metadata.MD) error { return nil }
func (s *metricsServerStream) SetTrailer(metadata.MD)       {}
func (s *metricsServerStream) Context() context.Context     { return s.ctx }
func (s *metricsServerStream) SendMsg(any) error            { return nil }
func (s *metricsServerStream) RecvMsg(any) error            { return nil }