package aws

import (
	"context"
	"errors"
	"fmt"
	"net"
	"net/url"
	"strconv"
	"strings"
	"time"

	coremodel "security-center-backend/internal/model"
	"security-center-backend/internal/modules/caasm/collector"
	model "security-center-backend/internal/modules/caasm/model"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/apprunner"
	apprunnertypes "github.com/aws/aws-sdk-go-v2/service/apprunner/types"
	"github.com/aws/aws-sdk-go-v2/service/appsync"
	appsynctypes "github.com/aws/aws-sdk-go-v2/service/appsync/types"
	"github.com/aws/aws-sdk-go-v2/service/cognitoidentityprovider"
	cognitotypes "github.com/aws/aws-sdk-go-v2/service/cognitoidentityprovider/types"
	"github.com/aws/aws-sdk-go-v2/service/elasticbeanstalk"
	ebtypes "github.com/aws/aws-sdk-go-v2/service/elasticbeanstalk/types"
	"github.com/aws/aws-sdk-go-v2/service/elasticsearchservice"
	estypes "github.com/aws/aws-sdk-go-v2/service/elasticsearchservice/types"
	"github.com/aws/aws-sdk-go-v2/service/kafka"
	kafkatypes "github.com/aws/aws-sdk-go-v2/service/kafka/types"
	"github.com/aws/aws-sdk-go-v2/service/lambda"
	lambdatypes "github.com/aws/aws-sdk-go-v2/service/lambda/types"
	"github.com/aws/aws-sdk-go-v2/service/opensearch"
	opensearchtypes "github.com/aws/aws-sdk-go-v2/service/opensearch/types"
	"github.com/aws/aws-sdk-go-v2/service/transfer"
	transfertypes "github.com/aws/aws-sdk-go-v2/service/transfer/types"
)

const endpointOnlyPortUnknownReason = "port_not_exposed_by_service_api"

type awsEndpointObservation struct {
	AssetNativeID     string
	Service           string
	DomainName        string
	DomainClass       string
	Registrar         string
	IsPublic          bool
	Port              uint32
	Protocol          string
	ObservationSource model.PortObservationSource
	ServiceName       string
	PortUnknownReason string
}

func (c *AWSCollector) collectAppRunner(ctx context.Context, region string, result *collector.AssetResult) error {
	svc := apprunner.NewFromConfig(c.cfg, func(o *apprunner.Options) {
		o.Region = region
	})
	paginator := apprunner.NewListServicesPaginator(svc, &apprunner.ListServicesInput{})

	var errs []error
	for paginator.HasMorePages() {
		page, err := paginator.NextPage(ctx)
		if err != nil {
			return err
		}
		for _, summary := range page.ServiceSummaryList {
			serviceArn := aws.ToString(summary.ServiceArn)
			if serviceArn == "" {
				continue
			}
			desc, err := svc.DescribeService(ctx, &apprunner.DescribeServiceInput{ServiceArn: &serviceArn})
			if err != nil {
				appendAWSPartialFailure(result, "service", fmt.Sprintf("%s:apprunner:%s", region, serviceArn), fmt.Sprintf("failed to describe App Runner service %s: %v", serviceArn, err))
				errs = append(errs, fmt.Errorf("app runner %s: %w", serviceArn, err))
				continue
			}
			if desc.Service == nil {
				continue
			}
			c.appendAppRunnerService(region, desc.Service, result)
		}
	}
	return errors.Join(errs...)
}

func (c *AWSCollector) appendAppRunnerService(region string, service *apprunnertypes.Service, result *collector.AssetResult) {
	if service == nil {
		return
	}
	nativeID := aws.ToString(service.ServiceArn)
	if nativeID == "" {
		return
	}
	name := firstNonEmpty(aws.ToString(service.ServiceName), aws.ToString(service.ServiceId), nativeID)
	asset := c.newBaseAsset(nativeID, name, string(model.AssetTypeApplication), string(service.Status), region)
	c.setService(&asset, "app_runner")
	hostname := normalizeEndpointHostname(aws.ToString(service.ServiceUrl))
	asset.Metadata = coremodel.JSONB{
		"service_id":                aws.ToString(service.ServiceId),
		"service_url":               hostname,
		"status":                    string(service.Status),
		"port_observation_reason":   "",
		"endpoint_observation_type": "service_hostname",
	}
	appendPortUnknownReason(asset.Metadata, hostname, 443, endpointOnlyPortUnknownReason)
	result.Assets = append(result.Assets, asset)

	c.appendAWSEndpointObservation(result, awsEndpointObservation{
		AssetNativeID:     asset.NativeID,
		Service:           "app_runner",
		DomainName:        hostname,
		IsPublic:          true,
		Port:              443,
		Protocol:          "tcp",
		ObservationSource: model.PortObservationSourceInferredWeb,
		ServiceName:       "https",
	})
}

func (c *AWSCollector) collectOpenSearch(ctx context.Context, region string, result *collector.AssetResult) error {
	var errs []error
	if err := c.collectOpenSearchDomains(ctx, region, result); err != nil {
		errs = append(errs, fmt.Errorf("opensearch domains: %w", err))
	}
	if err := c.collectElasticsearchDomains(ctx, region, result); err != nil {
		errs = append(errs, fmt.Errorf("elasticsearch domains: %w", err))
	}
	return errors.Join(errs...)
}

func (c *AWSCollector) collectOpenSearchDomains(ctx context.Context, region string, result *collector.AssetResult) error {
	svc := opensearch.NewFromConfig(c.cfg, func(o *opensearch.Options) {
		o.Region = region
	})
	list, err := svc.ListDomainNames(ctx, &opensearch.ListDomainNamesInput{})
	if err != nil {
		return err
	}
	names := make([]string, 0, len(list.DomainNames))
	for _, domain := range list.DomainNames {
		if name := aws.ToString(domain.DomainName); name != "" {
			names = append(names, name)
		}
	}
	for _, batch := range chunkStrings(names, 5) {
		desc, err := svc.DescribeDomains(ctx, &opensearch.DescribeDomainsInput{DomainNames: batch})
		if err != nil {
			appendAWSPartialFailure(result, "service", fmt.Sprintf("%s:opensearch:%s", region, strings.Join(batch, ",")), fmt.Sprintf("failed to describe OpenSearch domains: %v", err))
			continue
		}
		for _, domain := range desc.DomainStatusList {
			c.appendOpenSearchDomain(region, domain, result)
		}
	}
	return nil
}

func (c *AWSCollector) appendOpenSearchDomain(region string, domain opensearchtypes.DomainStatus, result *collector.AssetResult) {
	nativeID := aws.ToString(domain.ARN)
	if nativeID == "" {
		nativeID = aws.ToString(domain.DomainId)
	}
	if nativeID == "" {
		return
	}
	name := firstNonEmpty(aws.ToString(domain.DomainName), nativeID)
	endpoints := openSearchEndpointList(domain.Endpoint, domain.EndpointV2, domain.Endpoints)
	isPublic := len(domain.Endpoints) == 0 && len(endpoints) > 0

	asset := c.newBaseAsset(nativeID, name, string(model.AssetTypeDatabase), openSearchStatus(domain.Created, domain.Deleted), region)
	c.setService(&asset, "opensearch")
	asset.Metadata = coremodel.JSONB{
		"domain_id":                 aws.ToString(domain.DomainId),
		"domain_name":               name,
		"engine_version":            aws.ToString(domain.EngineVersion),
		"endpoint_count":            len(endpoints),
		"public_endpoint":           isPublic,
		"endpoint_observation_type": "service_hostname",
	}
	appendPortUnknownReason(asset.Metadata, strings.Join(endpoints, ","), 443, endpointOnlyPortUnknownReason)
	result.Assets = append(result.Assets, asset)

	for _, endpoint := range endpoints {
		c.appendAWSEndpointObservation(result, awsEndpointObservation{
			AssetNativeID:     asset.NativeID,
			Service:           "opensearch",
			DomainName:        endpoint,
			IsPublic:          isPublic,
			Port:              443,
			Protocol:          "tcp",
			ObservationSource: model.PortObservationSourceServiceEndpoint,
			ServiceName:       "https",
		})
	}
}

func (c *AWSCollector) collectElasticsearchDomains(ctx context.Context, region string, result *collector.AssetResult) error {
	svc := elasticsearchservice.NewFromConfig(c.cfg, func(o *elasticsearchservice.Options) {
		o.Region = region
	})
	list, err := svc.ListDomainNames(ctx, &elasticsearchservice.ListDomainNamesInput{})
	if err != nil {
		return err
	}
	names := make([]string, 0, len(list.DomainNames))
	for _, domain := range list.DomainNames {
		if name := aws.ToString(domain.DomainName); name != "" {
			names = append(names, name)
		}
	}
	for _, batch := range chunkStrings(names, 5) {
		desc, err := svc.DescribeElasticsearchDomains(ctx, &elasticsearchservice.DescribeElasticsearchDomainsInput{DomainNames: batch})
		if err != nil {
			appendAWSPartialFailure(result, "service", fmt.Sprintf("%s:elasticsearch:%s", region, strings.Join(batch, ",")), fmt.Sprintf("failed to describe Elasticsearch domains: %v", err))
			continue
		}
		for _, domain := range desc.DomainStatusList {
			c.appendElasticsearchDomain(region, domain, result)
		}
	}
	return nil
}

func (c *AWSCollector) appendElasticsearchDomain(region string, domain estypes.ElasticsearchDomainStatus, result *collector.AssetResult) {
	nativeID := aws.ToString(domain.ARN)
	if nativeID == "" {
		nativeID = aws.ToString(domain.DomainId)
	}
	if nativeID == "" {
		return
	}
	name := firstNonEmpty(aws.ToString(domain.DomainName), nativeID)
	endpoints := openSearchEndpointList(domain.Endpoint, nil, domain.Endpoints)
	isPublic := len(domain.Endpoints) == 0 && len(endpoints) > 0

	asset := c.newBaseAsset(nativeID, name, string(model.AssetTypeDatabase), elasticsearchStatus(domain.Created, domain.Deleted), region)
	c.setService(&asset, "elasticsearch")
	asset.Metadata = coremodel.JSONB{
		"domain_id":                 aws.ToString(domain.DomainId),
		"domain_name":               name,
		"engine_version":            aws.ToString(domain.ElasticsearchVersion),
		"endpoint_count":            len(endpoints),
		"public_endpoint":           isPublic,
		"endpoint_observation_type": "service_hostname",
	}
	appendPortUnknownReason(asset.Metadata, strings.Join(endpoints, ","), 443, endpointOnlyPortUnknownReason)
	result.Assets = append(result.Assets, asset)

	for _, endpoint := range endpoints {
		c.appendAWSEndpointObservation(result, awsEndpointObservation{
			AssetNativeID:     asset.NativeID,
			Service:           "elasticsearch",
			DomainName:        endpoint,
			IsPublic:          isPublic,
			Port:              443,
			Protocol:          "tcp",
			ObservationSource: model.PortObservationSourceServiceEndpoint,
			ServiceName:       "https",
		})
	}
}

func (c *AWSCollector) collectLambdaFunctionURLs(ctx context.Context, region string, result *collector.AssetResult) error {
	svc := lambda.NewFromConfig(c.cfg, func(o *lambda.Options) {
		o.Region = region
	})
	paginator := lambda.NewListFunctionsPaginator(svc, &lambda.ListFunctionsInput{})

	var errs []error
	for paginator.HasMorePages() {
		page, err := paginator.NextPage(ctx)
		if err != nil {
			return err
		}
		for _, fn := range page.Functions {
			functionName := aws.ToString(fn.FunctionName)
			if functionName == "" {
				continue
			}
			urlCfg, err := svc.GetFunctionUrlConfig(ctx, &lambda.GetFunctionUrlConfigInput{FunctionName: &functionName})
			if err != nil {
				if isAWSNotFound(err) {
					continue
				}
				appendAWSPartialFailure(result, "service", fmt.Sprintf("%s:lambda_function_url:%s", region, functionName), fmt.Sprintf("failed to collect Lambda function URL for %s: %v", functionName, err))
				errs = append(errs, fmt.Errorf("lambda function url %s: %w", functionName, err))
				continue
			}
			c.appendLambdaFunctionURL(region, fn, urlCfg, result)
		}
	}
	return errors.Join(errs...)
}

func (c *AWSCollector) appendLambdaFunctionURL(region string, fn lambdatypes.FunctionConfiguration, urlCfg *lambda.GetFunctionUrlConfigOutput, result *collector.AssetResult) {
	if urlCfg == nil {
		return
	}
	nativeID := firstNonEmpty(aws.ToString(fn.FunctionArn), aws.ToString(urlCfg.FunctionArn), aws.ToString(fn.FunctionName))
	if nativeID == "" {
		return
	}
	name := firstNonEmpty(aws.ToString(fn.FunctionName), nativeID)
	hostname := normalizeEndpointHostname(aws.ToString(urlCfg.FunctionUrl))
	asset := c.newBaseAsset(nativeID, name, string(model.AssetTypeCompute), string(fn.State), region)
	c.setService(&asset, "lambda")
	asset.Metadata = coremodel.JSONB{
		"function_arn":              nativeID,
		"runtime":                   string(fn.Runtime),
		"function_url":              hostname,
		"function_url_auth_type":    string(urlCfg.AuthType),
		"function_url_invoke_mode":  string(urlCfg.InvokeMode),
		"endpoint_observation_type": "service_hostname",
	}
	appendPortUnknownReason(asset.Metadata, hostname, 443, endpointOnlyPortUnknownReason)
	result.Assets = append(result.Assets, asset)

	c.appendAWSEndpointObservation(result, awsEndpointObservation{
		AssetNativeID:     asset.NativeID,
		Service:           "lambda_function_url",
		DomainName:        hostname,
		IsPublic:          urlCfg.AuthType == lambdatypes.FunctionUrlAuthTypeNone,
		Port:              443,
		Protocol:          "tcp",
		ObservationSource: model.PortObservationSourceInferredWeb,
		ServiceName:       "https",
	})
}

func (c *AWSCollector) collectTransferFamily(ctx context.Context, region string, result *collector.AssetResult) error {
	svc := transfer.NewFromConfig(c.cfg, func(o *transfer.Options) {
		o.Region = region
	})
	paginator := transfer.NewListServersPaginator(svc, &transfer.ListServersInput{})

	var errs []error
	for paginator.HasMorePages() {
		page, err := paginator.NextPage(ctx)
		if err != nil {
			return err
		}
		for _, summary := range page.Servers {
			serverID := aws.ToString(summary.ServerId)
			if serverID == "" {
				continue
			}
			desc, err := svc.DescribeServer(ctx, &transfer.DescribeServerInput{ServerId: &serverID})
			if err != nil {
				appendAWSPartialFailure(result, "service", fmt.Sprintf("%s:transfer:%s", region, serverID), fmt.Sprintf("failed to describe Transfer Family server %s: %v", serverID, err))
				errs = append(errs, fmt.Errorf("transfer server %s: %w", serverID, err))
				continue
			}
			if desc.Server == nil {
				continue
			}
			c.appendTransferServer(region, desc.Server, result)
		}
	}
	return errors.Join(errs...)
}

func (c *AWSCollector) appendTransferServer(region string, server *transfertypes.DescribedServer, result *collector.AssetResult) {
	if server == nil {
		return
	}
	nativeID := aws.ToString(server.Arn)
	if nativeID == "" {
		return
	}
	serverID := transferServerIDFromARN(nativeID)
	hostname := fmt.Sprintf("%s.server.transfer.%s.amazonaws.com", serverID, region)
	isPublic := server.EndpointType == transfertypes.EndpointTypePublic
	protocols := server.Protocols

	asset := c.newBaseAsset(nativeID, firstNonEmpty(serverID, nativeID), string(model.AssetTypeNetwork), string(server.State), region)
	c.setService(&asset, "transfer_family")
	asset.Metadata = coremodel.JSONB{
		"server_id":                 serverID,
		"endpoint_type":             string(server.EndpointType),
		"protocols":                 protocols,
		"identity_provider_type":    string(server.IdentityProviderType),
		"endpoint_observation_type": "service_hostname",
	}
	if len(protocols) == 0 {
		appendPortUnknownReason(asset.Metadata, hostname, 0, "transfer_protocols_not_returned")
	}
	result.Assets = append(result.Assets, asset)

	if len(protocols) == 0 {
		c.appendAWSEndpointObservation(result, awsEndpointObservation{
			AssetNativeID:     asset.NativeID,
			Service:           "transfer_family",
			DomainName:        hostname,
			IsPublic:          isPublic,
			PortUnknownReason: "transfer_protocols_not_returned",
		})
		return
	}
	for _, protocol := range protocols {
		port, name := transferProtocolPort(protocol)
		c.appendAWSEndpointObservation(result, awsEndpointObservation{
			AssetNativeID:     asset.NativeID,
			Service:           "transfer_family",
			DomainName:        hostname,
			IsPublic:          isPublic,
			Port:              port,
			Protocol:          "tcp",
			ObservationSource: model.PortObservationSourceServiceEndpoint,
			ServiceName:       name,
		})
	}
}

func (c *AWSCollector) collectElasticBeanstalk(ctx context.Context, region string, result *collector.AssetResult) error {
	svc := elasticbeanstalk.NewFromConfig(c.cfg, func(o *elasticbeanstalk.Options) {
		o.Region = region
	})
	var nextToken *string
	for {
		page, err := svc.DescribeEnvironments(ctx, &elasticbeanstalk.DescribeEnvironmentsInput{
			IncludeDeleted: aws.Bool(false),
			NextToken:      nextToken,
		})
		if err != nil {
			return err
		}
		for _, env := range page.Environments {
			c.appendElasticBeanstalkEnvironment(region, env, result)
		}
		if page.NextToken == nil || aws.ToString(page.NextToken) == "" {
			break
		}
		nextToken = page.NextToken
	}
	return nil
}

func (c *AWSCollector) appendElasticBeanstalkEnvironment(region string, env ebtypes.EnvironmentDescription, result *collector.AssetResult) {
	nativeID := firstNonEmpty(aws.ToString(env.EnvironmentArn), aws.ToString(env.EnvironmentId))
	if nativeID == "" {
		return
	}
	name := firstNonEmpty(aws.ToString(env.EnvironmentName), nativeID)
	hostname := normalizeEndpointHostname(firstNonEmpty(aws.ToString(env.CNAME), aws.ToString(env.EndpointURL)))
	asset := c.newBaseAsset(nativeID, name, string(model.AssetTypeApplication), string(env.Status), region)
	c.setService(&asset, "elastic_beanstalk")
	asset.Metadata = coremodel.JSONB{
		"application_name":          aws.ToString(env.ApplicationName),
		"environment_id":            aws.ToString(env.EnvironmentId),
		"environment_name":          name,
		"cname":                     normalizeEndpointHostname(aws.ToString(env.CNAME)),
		"endpoint_url":              normalizeEndpointHostname(aws.ToString(env.EndpointURL)),
		"health":                    string(env.Health),
		"health_status":             string(env.HealthStatus),
		"endpoint_observation_type": "service_hostname",
	}
	appendPortUnknownReason(asset.Metadata, hostname, 0, endpointOnlyPortUnknownReason)
	result.Assets = append(result.Assets, asset)

	c.appendAWSEndpointObservation(result, awsEndpointObservation{
		AssetNativeID:     asset.NativeID,
		Service:           "elastic_beanstalk",
		DomainName:        hostname,
		IsPublic:          hostname != "",
		PortUnknownReason: endpointOnlyPortUnknownReason,
	})
}

func (c *AWSCollector) collectAppSync(ctx context.Context, region string, result *collector.AssetResult) error {
	svc := appsync.NewFromConfig(c.cfg, func(o *appsync.Options) {
		o.Region = region
	})
	paginator := appsync.NewListGraphqlApisPaginator(svc, &appsync.ListGraphqlApisInput{})
	for paginator.HasMorePages() {
		page, err := paginator.NextPage(ctx)
		if err != nil {
			return err
		}
		for _, api := range page.GraphqlApis {
			c.appendAppSyncAPI(region, api, result)
		}
	}
	return nil
}

func (c *AWSCollector) appendAppSyncAPI(region string, api appsynctypes.GraphqlApi, result *collector.AssetResult) {
	nativeID := firstNonEmpty(aws.ToString(api.Arn), aws.ToString(api.ApiId))
	if nativeID == "" {
		return
	}
	name := firstNonEmpty(aws.ToString(api.Name), aws.ToString(api.ApiId), nativeID)
	isPublic := api.Visibility != appsynctypes.GraphQLApiVisibilityPrivate
	asset := c.newBaseAsset(nativeID, name, string(model.AssetTypeApplication), "active", region)
	c.setService(&asset, "appsync")
	asset.Metadata = coremodel.JSONB{
		"api_id":                    aws.ToString(api.ApiId),
		"api_type":                  string(api.ApiType),
		"authentication_type":       string(api.AuthenticationType),
		"visibility":                string(api.Visibility),
		"uris":                      api.Uris,
		"dns":                       api.Dns,
		"endpoint_observation_type": "service_hostname",
	}
	result.Assets = append(result.Assets, asset)

	for _, endpoint := range uniqueStrings(mapValues(api.Uris), mapValues(api.Dns)) {
		c.appendAWSEndpointObservation(result, awsEndpointObservation{
			AssetNativeID:     asset.NativeID,
			Service:           "appsync",
			DomainName:        normalizeEndpointHostname(endpoint),
			IsPublic:          isPublic,
			Port:              443,
			Protocol:          "tcp",
			ObservationSource: model.PortObservationSourceInferredWeb,
			ServiceName:       "https",
		})
	}
}

func (c *AWSCollector) collectMSK(ctx context.Context, region string, result *collector.AssetResult) error {
	svc := kafka.NewFromConfig(c.cfg, func(o *kafka.Options) {
		o.Region = region
	})
	paginator := kafka.NewListClustersV2Paginator(svc, &kafka.ListClustersV2Input{})

	var errs []error
	for paginator.HasMorePages() {
		page, err := paginator.NextPage(ctx)
		if err != nil {
			return err
		}
		for _, cluster := range page.ClusterInfoList {
			clusterARN := aws.ToString(cluster.ClusterArn)
			if clusterARN == "" {
				continue
			}
			bootstrap, err := svc.GetBootstrapBrokers(ctx, &kafka.GetBootstrapBrokersInput{ClusterArn: &clusterARN})
			if err != nil {
				appendAWSPartialFailure(result, "service", fmt.Sprintf("%s:msk:%s", region, clusterARN), fmt.Sprintf("failed to collect MSK bootstrap brokers for %s: %v", clusterARN, err))
				errs = append(errs, fmt.Errorf("msk bootstrap brokers %s: %w", clusterARN, err))
				c.appendMSKCluster(region, cluster, nil, result)
				continue
			}
			c.appendMSKCluster(region, cluster, bootstrap, result)
		}
	}
	return errors.Join(errs...)
}

func (c *AWSCollector) appendMSKCluster(region string, cluster kafkatypes.Cluster, bootstrap *kafka.GetBootstrapBrokersOutput, result *collector.AssetResult) {
	nativeID := aws.ToString(cluster.ClusterArn)
	if nativeID == "" {
		return
	}
	name := firstNonEmpty(aws.ToString(cluster.ClusterName), nativeID)
	publicAccess := mskPublicAccess(cluster)
	brokers := mskPublicBrokerEndpoints(bootstrap)
	asset := c.newBaseAsset(nativeID, name, string(model.AssetTypeDatabase), string(cluster.State), region)
	c.setService(&asset, "msk")
	asset.Metadata = coremodel.JSONB{
		"cluster_name":              name,
		"cluster_type":              string(cluster.ClusterType),
		"public_access":             publicAccess,
		"public_broker_count":       len(brokers),
		"endpoint_observation_type": "service_hostname",
	}
	if len(brokers) == 0 {
		appendPortUnknownReason(asset.Metadata, "", 0, "msk_public_bootstrap_brokers_not_returned")
	}
	result.Assets = append(result.Assets, asset)

	for _, broker := range brokers {
		c.appendAWSEndpointObservation(result, awsEndpointObservation{
			AssetNativeID:     asset.NativeID,
			Service:           "msk_public_access",
			DomainName:        broker.Host,
			IsPublic:          true,
			Port:              broker.Port,
			Protocol:          "tcp",
			ObservationSource: model.PortObservationSourceServiceEndpoint,
			ServiceName:       broker.ServiceName,
		})
	}
}

func (c *AWSCollector) collectCognitoHostedDomains(ctx context.Context, region string, result *collector.AssetResult) error {
	svc := cognitoidentityprovider.NewFromConfig(c.cfg, func(o *cognitoidentityprovider.Options) {
		o.Region = region
	})
	paginator := cognitoidentityprovider.NewListUserPoolsPaginator(svc, &cognitoidentityprovider.ListUserPoolsInput{MaxResults: aws.Int32(60)})

	var errs []error
	for paginator.HasMorePages() {
		page, err := paginator.NextPage(ctx)
		if err != nil {
			return err
		}
		for _, summary := range page.UserPools {
			userPoolID := aws.ToString(summary.Id)
			if userPoolID == "" {
				continue
			}
			desc, err := svc.DescribeUserPool(ctx, &cognitoidentityprovider.DescribeUserPoolInput{UserPoolId: &userPoolID})
			if err != nil {
				appendAWSPartialFailure(result, "service", fmt.Sprintf("%s:cognito:%s", region, userPoolID), fmt.Sprintf("failed to describe Cognito user pool %s: %v", userPoolID, err))
				errs = append(errs, fmt.Errorf("cognito user pool %s: %w", userPoolID, err))
				continue
			}
			if desc.UserPool == nil {
				continue
			}
			c.appendCognitoUserPool(region, desc.UserPool, result)
		}
	}
	return errors.Join(errs...)
}

func (c *AWSCollector) appendCognitoUserPool(region string, userPool *cognitotypes.UserPoolType, result *collector.AssetResult) {
	if userPool == nil {
		return
	}
	nativeID := firstNonEmpty(aws.ToString(userPool.Arn), aws.ToString(userPool.Id))
	if nativeID == "" {
		return
	}
	name := firstNonEmpty(aws.ToString(userPool.Name), aws.ToString(userPool.Id), nativeID)
	asset := c.newBaseAsset(nativeID, name, string(model.AssetTypeIdentity), "active", region)
	c.setService(&asset, "cognito")
	prefixDomain := cognitoHostedDomain(aws.ToString(userPool.Domain), region)
	customDomain := normalizeEndpointHostname(aws.ToString(userPool.CustomDomain))
	asset.Metadata = coremodel.JSONB{
		"user_pool_id":              aws.ToString(userPool.Id),
		"domain_prefix":             aws.ToString(userPool.Domain),
		"hosted_domain":             prefixDomain,
		"custom_domain":             customDomain,
		"endpoint_observation_type": "service_hostname",
	}
	result.Assets = append(result.Assets, asset)

	for _, domain := range uniqueStrings([]string{prefixDomain, customDomain}) {
		c.appendAWSEndpointObservation(result, awsEndpointObservation{
			AssetNativeID:     asset.NativeID,
			Service:           "cognito_hosted_domain",
			DomainName:        domain,
			IsPublic:          true,
			Port:              443,
			Protocol:          "tcp",
			ObservationSource: model.PortObservationSourceInferredWeb,
			ServiceName:       "https",
		})
	}
}

func (c *AWSCollector) appendAWSEndpointObservation(result *collector.AssetResult, obs awsEndpointObservation) {
	if result == nil {
		return
	}
	domainName := normalizeEndpointHostname(obs.DomainName)
	if domainName == "" || obs.AssetNativeID == "" {
		return
	}
	service := strings.TrimSpace(obs.Service)
	domainClass := firstNonEmpty(obs.DomainClass, model.DomainClassServiceHostname)
	registrar := firstNonEmpty(obs.Registrar, service, "aws")

	domainAsset := model.DomainAsset{
		Asset:       model.Asset{NativeID: obs.AssetNativeID},
		DomainName:  domainName,
		DomainClass: domainClass,
		Registrar:   registrar,
	}
	setServiceEndpointDomainDefaults(&domainAsset, service, obs.IsPublic)
	c.setService(&domainAsset.Asset, service)
	result.DomainAssets = append(result.DomainAssets, domainAsset)

	if obs.Port == 0 {
		return
	}
	protocol := firstNonEmpty(obs.Protocol, "tcp")
	source := obs.ObservationSource
	if source == "" {
		source = model.PortObservationSourceServiceEndpoint
	}
	result.Ports = append(result.Ports, model.IPPortIndex{
		Asset:             model.Asset{NativeID: obs.AssetNativeID},
		Endpoint:          domainName,
		Port:              obs.Port,
		Protocol:          protocol,
		ObservationSource: source,
		ServiceName:       firstNonEmpty(obs.ServiceName, getServiceName(obs.Port)),
		Exposure:          endpointExposure(obs.IsPublic),
		Status:            "open",
		Evidence:          awsPortEvidence(service, obs.AssetNativeID, string(source), "service_managed"),
		Confidence:        awsConfidence("low", "AWS service endpoint metadata implies this service port."),
	})
}

func newAWSServiceDomainAsset(assetNativeID, domainName, domainClass, registrar string, observedAt time.Time) model.DomainAsset {
	domain := model.DomainAsset{
		Asset:       model.Asset{NativeID: assetNativeID},
		DomainName:  normalizeEndpointHostname(domainName),
		DomainClass: firstNonEmpty(domainClass, model.DomainClassServiceHostname),
		Registrar:   registrar,
	}
	setServiceEndpointDomainDefaults(&domain, registrar, true)
	if !observedAt.IsZero() {
		domain.LastSyncedAt = &observedAt
		domain.FirstSeenAt = &observedAt
		domain.LastSeenAt = &observedAt
	}
	return domain
}

func newAWSServiceEndpointPort(assetNativeID, endpoint string, port uint32, protocol string, source model.PortObservationSource, exposure, serviceName string, evidence coremodel.JSONB) model.IPPortIndex {
	if protocol == "" {
		protocol = "tcp"
	}
	if source == "" {
		source = model.PortObservationSourceServiceEndpoint
	}
	normalizedEndpoint := normalizeEndpointHostname(endpoint)
	ipAddress := ""
	if net.ParseIP(normalizedEndpoint) != nil {
		ipAddress = normalizedEndpoint
	}
	return model.IPPortIndex{
		Asset:             model.Asset{NativeID: assetNativeID},
		Endpoint:          normalizedEndpoint,
		IPAddress:         ipAddress,
		Port:              port,
		Protocol:          protocol,
		ObservationSource: source,
		ServiceName:       firstNonEmpty(serviceName, getServiceName(port)),
		Exposure:          firstNonEmpty(exposure, "public"),
		Status:            "open",
		Evidence:          evidence,
	}
}

func awsPortEvidence(sourceService, resourceID, evidenceSource, ownershipStatus string) coremodel.JSONB {
	return coremodel.JSONB{
		"source_service":   sourceService,
		"resource_id":      resourceID,
		"evidence_source":  evidenceSource,
		"ownership_status": ownershipStatus,
	}
}

func awsConfidence(level, reason string) coremodel.JSONB {
	return coremodel.JSONB{
		"level":  level,
		"reason": reason,
	}
}

func normalizeEndpointHostname(raw string) string {
	raw = strings.TrimSpace(raw)
	if raw == "" {
		return ""
	}
	if strings.Contains(raw, "://") {
		if parsed, err := url.Parse(raw); err == nil {
			raw = parsed.Host
		}
	}
	raw = strings.TrimSpace(raw)
	raw = strings.TrimSuffix(raw, ".")
	if host, _, err := net.SplitHostPort(raw); err == nil {
		raw = host
	}
	raw = strings.Trim(raw, "[]")
	return strings.ToLower(raw)
}

func appendPortUnknownReason(metadata coremodel.JSONB, endpoint string, port uint32, reason string) {
	if metadata == nil || reason == "" || (endpoint != "" && port > 0) {
		return
	}
	metadata["port_observation_reason"] = reason
}

func firstNonEmpty(values ...string) string {
	for _, value := range values {
		if trimmed := strings.TrimSpace(value); trimmed != "" {
			return trimmed
		}
	}
	return ""
}

func chunkStrings(values []string, size int) [][]string {
	if size <= 0 || len(values) == 0 {
		return nil
	}
	chunks := make([][]string, 0, (len(values)+size-1)/size)
	for start := 0; start < len(values); start += size {
		end := start + size
		if end > len(values) {
			end = len(values)
		}
		chunks = append(chunks, values[start:end])
	}
	return chunks
}

func openSearchEndpointList(endpoint, endpointV2 *string, endpoints map[string]string) []string {
	values := make([]string, 0, 2+len(endpoints))
	values = append(values, aws.ToString(endpoint), aws.ToString(endpointV2))
	for _, endpoint := range endpoints {
		values = append(values, endpoint)
	}
	return uniqueStrings(values)
}

func uniqueStrings(groups ...[]string) []string {
	seen := map[string]struct{}{}
	out := []string{}
	for _, values := range groups {
		for _, value := range values {
			normalized := normalizeEndpointHostname(value)
			if normalized == "" {
				continue
			}
			if _, ok := seen[normalized]; ok {
				continue
			}
			seen[normalized] = struct{}{}
			out = append(out, normalized)
		}
	}
	return out
}

func mapValues(values map[string]string) []string {
	out := make([]string, 0, len(values))
	for _, value := range values {
		out = append(out, value)
	}
	return out
}

func openSearchStatus(created, deleted *bool) string {
	switch {
	case aws.ToBool(deleted):
		return "inactive"
	case aws.ToBool(created):
		return "active"
	default:
		return "pending"
	}
}

func elasticsearchStatus(created, deleted *bool) string {
	return openSearchStatus(created, deleted)
}

func isAWSNotFound(err error) bool {
	code := strings.ToLower(awsAPIErrorCode(err))
	return strings.Contains(code, "notfound") || strings.Contains(code, "notfoundexception") || strings.Contains(code, "resourcenotfound")
}

func transferServerIDFromARN(arn string) string {
	parts := strings.Split(arn, "/")
	if len(parts) > 1 {
		return strings.TrimSpace(parts[len(parts)-1])
	}
	return ""
}

func transferProtocolPort(protocol transfertypes.Protocol) (uint32, string) {
	switch protocol {
	case transfertypes.ProtocolSftp:
		return 22, "sftp"
	case transfertypes.ProtocolFtps:
		return 990, "ftps"
	case transfertypes.ProtocolFtp:
		return 21, "ftp"
	case transfertypes.ProtocolAs2:
		return 443, "as2"
	default:
		return 0, strings.ToLower(string(protocol))
	}
}

type mskBrokerEndpoint struct {
	Host        string
	Port        uint32
	ServiceName string
}

func mskPublicAccess(cluster kafkatypes.Cluster) string {
	if cluster.Provisioned == nil ||
		cluster.Provisioned.BrokerNodeGroupInfo == nil ||
		cluster.Provisioned.BrokerNodeGroupInfo.ConnectivityInfo == nil ||
		cluster.Provisioned.BrokerNodeGroupInfo.ConnectivityInfo.PublicAccess == nil {
		return ""
	}
	return aws.ToString(cluster.Provisioned.BrokerNodeGroupInfo.ConnectivityInfo.PublicAccess.Type)
}

func mskPublicBrokerEndpoints(bootstrap *kafka.GetBootstrapBrokersOutput) []mskBrokerEndpoint {
	if bootstrap == nil {
		return nil
	}
	return uniqueMSKBrokers(
		parseMSKBrokerString(aws.ToString(bootstrap.BootstrapBrokerStringPublicTls), "kafka-tls"),
		parseMSKBrokerString(aws.ToString(bootstrap.BootstrapBrokerStringPublicSaslIam), "kafka-sasl-iam"),
		parseMSKBrokerString(aws.ToString(bootstrap.BootstrapBrokerStringPublicSaslScram), "kafka-sasl-scram"),
	)
}

func parseMSKBrokerString(raw, serviceName string) []mskBrokerEndpoint {
	raw = strings.TrimSpace(raw)
	if raw == "" {
		return nil
	}
	var out []mskBrokerEndpoint
	for _, item := range strings.Split(raw, ",") {
		item = strings.TrimSpace(item)
		host, portRaw, err := net.SplitHostPort(item)
		if err != nil {
			continue
		}
		port, err := strconv.ParseUint(portRaw, 10, 32)
		if err != nil || port == 0 {
			continue
		}
		out = append(out, mskBrokerEndpoint{
			Host:        normalizeEndpointHostname(host),
			Port:        uint32(port),
			ServiceName: serviceName,
		})
	}
	return out
}

func uniqueMSKBrokers(groups ...[]mskBrokerEndpoint) []mskBrokerEndpoint {
	seen := map[string]struct{}{}
	var out []mskBrokerEndpoint
	for _, group := range groups {
		for _, broker := range group {
			if broker.Host == "" || broker.Port == 0 {
				continue
			}
			key := fmt.Sprintf("%s:%d:%s", broker.Host, broker.Port, broker.ServiceName)
			if _, ok := seen[key]; ok {
				continue
			}
			seen[key] = struct{}{}
			out = append(out, broker)
		}
	}
	return out
}

func cognitoHostedDomain(prefix, region string) string {
	prefix = normalizeEndpointHostname(prefix)
	region = strings.TrimSpace(region)
	if prefix == "" || region == "" {
		return ""
	}
	if strings.Contains(prefix, ".") {
		return prefix
	}
	return fmt.Sprintf("%s.auth.%s.amazoncognito.com", prefix, region)
}