package aws

import (
	"context"
	"fmt"
	"slices"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/iam"
	iamtypes "github.com/aws/aws-sdk-go-v2/service/iam/types"

	"github.com/tidbcloud/controlplane9/internal/verbose"
)

// IAMInventory is a point-in-time snapshot of an account's cp9-taggable IAM
// entities and their tags.
//
// It exists because IAM is global and Resource Groups Tagging API is regional:
// GetResources never returns an IAM entity, so tag-index discovery cannot see
// IAM at all. Two consequences followed from that, and one snapshot fixes both.
//
// Discovery gap: a session whose only remaining resources are IAM entities was
// never returned as an expired candidate, so its roles and policies survived
// every scheduled run. Reading IAM directly here makes those sessions
// discoverable (see ExpiredScopes).
//
// Cost: the per-session scan issues one ListRoleTags per role and one
// ListPolicyTags per local policy in the whole account, and repeats that for
// every candidate. An account with ~100 expired candidates and ~80 roles
// therefore made thousands of tag calls to answer one question per session,
// which throttles and then exhausts its own timeout budget. Scanning once and
// filtering in memory makes the account cost independent of candidate count.
//
// A snapshot is for discovery only. Deletion re-reads each entity's tags
// directly (VerifyKnownResource), so a snapshot that has gone stale cannot
// widen what gets deleted: an entity created after the scan is simply not seen
// this run, and one deleted after the scan fails its own targeted re-read.
// Treat the value as read-only once built; purge shares one across concurrent
// per-session workers.
type IAMInventory struct {
	roles            []iamInventoryEntry
	instanceProfiles []iamInventoryProfile
	policies         []iamInventoryEntry
	degraded         bool
}

// Complete reports whether every IAM section was readable. Broad cleanup may
// reuse a partial snapshot for discovery, but it must not prove a scope stale.
func (inv *IAMInventory) Complete() bool {
	return inv != nil && !inv.degraded
}

type iamInventoryEntry struct {
	arn  string
	name string
	tags map[string]string
}

// iamInventoryProfile keeps the whole instance profile because the tagged-IAM
// walk follows profile.Roles to reach roles a session owns by name.
type iamInventoryProfile struct {
	profile iamtypes.InstanceProfile
	tags    map[string]string
}

// ScanIAMInventory reads every role, instance profile and local policy in the
// account once, together with its tags. It is read-only.
//
// A listing the credentials may not read degrades to an empty section rather
// than failing the caller, matching the per-session scan it replaces: cleanup
// must still proceed on the resources it can see. The caller owns the timeout;
// exceeding it returns what was collected so far, because a partial snapshot
// still narrows discovery correctly and a hard failure here would block every
// candidate in the account.
func ScanIAMInventory(ctx context.Context, lister IAMListTagAPI, tagReader IAMTagAPI) (*IAMInventory, error) {
	if lister == nil {
		return nil, fmt.Errorf("iam list client must not be nil")
	}
	inv := &IAMInventory{}
	reader := &iamInventoryTagReader{lister: lister, tagReader: tagReader}

	if err := inv.scanRoles(ctx, reader); err != nil {
		return inv, err
	}
	if err := inv.scanInstanceProfiles(ctx, lister); err != nil {
		return inv, err
	}
	if err := inv.scanPolicies(ctx, reader); err != nil {
		return inv, err
	}
	verbose.Logf(
		"AWS iam inventory scanned roles=%d instance_profiles=%d policies=%d",
		len(inv.roles), len(inv.instanceProfiles), len(inv.policies),
	)
	return inv, nil
}

// iamInventoryTagReader resolves an entity's tags, preferring the paginating
// tag reader the tracker already uses so the inventory and the per-session scan
// agree on what an entity's tags are.
type iamInventoryTagReader struct {
	lister    IAMListTagAPI
	tagReader IAMTagAPI
}

func (r *iamInventoryTagReader) roleTags(ctx context.Context, name string) (map[string]string, error) {
	if r.tagReader == nil {
		return map[string]string{}, nil
	}
	tags := map[string]string{}
	input := &iam.ListRoleTagsInput{RoleName: aws.String(name)}
	for {
		out, err := r.tagReader.ListRoleTags(ctx, input)
		if err != nil {
			return nil, err
		}
		copyIAMTags(tags, out.Tags)
		if !out.IsTruncated || out.Marker == nil {
			return tags, nil
		}
		input.Marker = out.Marker
	}
}

func (r *iamInventoryTagReader) policyTags(ctx context.Context, arn string) (map[string]string, error) {
	if r.tagReader == nil {
		return map[string]string{}, nil
	}
	tags := map[string]string{}
	input := &iam.ListPolicyTagsInput{PolicyArn: aws.String(arn)}
	for {
		out, err := r.tagReader.ListPolicyTags(ctx, input)
		if err != nil {
			return nil, err
		}
		copyIAMTags(tags, out.Tags)
		if !out.IsTruncated || out.Marker == nil {
			return tags, nil
		}
		input.Marker = out.Marker
	}
}

func (inv *IAMInventory) scanRoles(ctx context.Context, reader *iamInventoryTagReader) error {
	input := &iam.ListRolesInput{}
	for {
		out, err := reader.lister.ListRoles(ctx, input)
		if err != nil {
			if isAuthorizationErr(err) {
				inv.degraded = true
				verbose.Logf("AWS iam inventory role list unavailable: %v", err)
				return nil
			}
			return fmt.Errorf("iam inventory list roles: %w", err)
		}
		for _, role := range out.Roles {
			name := aws.ToString(role.RoleName)
			arn := aws.ToString(role.Arn)
			if name == "" || arn == "" {
				continue
			}
			tags, err := reader.roleTags(ctx, name)
			if err != nil {
				if isAuthorizationErr(err) {
					inv.degraded = true
					continue
				}
				if isNotFoundErr(err) {
					continue
				}
				if ctx.Err() != nil {
					inv.degraded = true
					verbose.Logf("AWS iam inventory role tag scan stopped early: %v", err)
					return nil
				}
				return fmt.Errorf("iam inventory role tags %s: %w", name, err)
			}
			inv.roles = append(inv.roles, iamInventoryEntry{arn: arn, name: name, tags: tags})
		}
		if !out.IsTruncated || out.Marker == nil {
			return nil
		}
		input.Marker = out.Marker
	}
}

func (inv *IAMInventory) scanInstanceProfiles(ctx context.Context, lister IAMListTagAPI) error {
	input := &iam.ListInstanceProfilesInput{}
	for {
		out, err := lister.ListInstanceProfiles(ctx, input)
		if err != nil {
			if isAuthorizationErr(err) {
				inv.degraded = true
				verbose.Logf("AWS iam inventory instance profile list unavailable: %v", err)
				return nil
			}
			return fmt.Errorf("iam inventory list instance profiles: %w", err)
		}
		for i := range out.InstanceProfiles {
			profile := out.InstanceProfiles[i]
			if aws.ToString(profile.Arn) == "" {
				continue
			}
			// ListInstanceProfiles returns tags inline, so this section costs one
			// call per page rather than one per entity.
			tags := map[string]string{}
			copyIAMTags(tags, profile.Tags)
			inv.instanceProfiles = append(inv.instanceProfiles, iamInventoryProfile{profile: profile, tags: tags})
		}
		if !out.IsTruncated || out.Marker == nil {
			return nil
		}
		input.Marker = out.Marker
	}
}

func (inv *IAMInventory) scanPolicies(ctx context.Context, reader *iamInventoryTagReader) error {
	input := &iam.ListPoliciesInput{Scope: iamtypes.PolicyScopeTypeLocal}
	for {
		out, err := reader.lister.ListPolicies(ctx, input)
		if err != nil {
			if isAuthorizationErr(err) {
				inv.degraded = true
				verbose.Logf("AWS iam inventory policy list unavailable: %v", err)
				return nil
			}
			return fmt.Errorf("iam inventory list policies: %w", err)
		}
		for _, policy := range out.Policies {
			arn := aws.ToString(policy.Arn)
			if arn == "" {
				continue
			}
			tags, err := reader.policyTags(ctx, arn)
			if err != nil {
				if isAuthorizationErr(err) {
					inv.degraded = true
					continue
				}
				if isNotFoundErr(err) {
					continue
				}
				if ctx.Err() != nil {
					inv.degraded = true
					verbose.Logf("AWS iam inventory policy tag scan stopped early: %v", err)
					return nil
				}
				return fmt.Errorf("iam inventory policy tags %s: %w", arn, err)
			}
			inv.policies = append(inv.policies, iamInventoryEntry{
				arn: arn, name: aws.ToString(policy.PolicyName), tags: tags,
			})
		}
		if !out.IsTruncated || out.Marker == nil {
			return nil
		}
		input.Marker = out.Marker
	}
}

// ExpiredScopes returns the owner/session candidates that IAM alone proves are
// expired, applying exactly the tag contract broad discovery already requires:
// cp9-managed in {true, terraform-substrate} plus cp9-owner, cp9-session and a
// parseable cp9-ttl at or before now. An entity missing any of those is skipped
// rather than guessed at, which is what keeps account-level infrastructure such
// as cp9-developer-power out of the result: it carries no session and no TTL.
func (inv *IAMInventory) ExpiredScopes(now time.Time) []ExpiredManagedScope {
	return FilterExpiredManagedScopes(inv.ManagedScopes(), now)
}

// ManagedScopes returns every valid owner/session TTL observed in the IAM
// snapshot, keeping the latest TTL for each scope.
func (inv *IAMInventory) ManagedScopes() []ExpiredManagedScope {
	if inv == nil {
		return nil
	}
	var scopes []ExpiredManagedScope
	add := func(resource Resource, tags map[string]string) {
		owner := strings.TrimSpace(tags[TagOwner])
		session := strings.TrimSpace(tags[TagSession])
		ttlRaw := strings.TrimSpace(tags[TagTTL])
		if owner == "" || session == "" || ttlRaw == "" {
			return
		}
		switch strings.TrimSpace(tags[TagManaged]) {
		case TagManagedTerraformSubstrate, "true":
		default:
			return
		}
		ttl, err := parseCP9TTL(ttlRaw)
		if err != nil {
			return
		}
		resource.Owner = owner
		resource.SessionID = session
		scopes = appendManagedScope(scopes, &ExpiredManagedScope{
			Owner:     owner,
			Session:   session,
			TTL:       ttl,
			Resources: []Resource{resource},
		})
	}
	for i := range inv.roles {
		add(Resource{ARN: inv.roles[i].arn, Type: ResourceIAMRole}, inv.roles[i].tags)
	}
	for i := range inv.instanceProfiles {
		add(Resource{
			ARN:  aws.ToString(inv.instanceProfiles[i].profile.Arn),
			Type: ResourceIAMInstanceProfile,
		}, inv.instanceProfiles[i].tags)
	}
	for i := range inv.policies {
		add(Resource{ARN: inv.policies[i].arn, Type: ResourceIAMPolicy}, inv.policies[i].tags)
	}
	return scopes
}

// resourcesForScope returns the snapshot's entities that carry this exact
// owner/session scope, in the same shape the live per-session scan produced.
func (inv *IAMInventory) resourcesForScope(
	ctx context.Context,
	t *ResourceTracker,
	lister IAMListTagAPI,
	owner, sessionID string,
) ([]Resource, error) {
	if inv == nil {
		return nil, nil
	}
	var resources []Resource
	for i := range inv.instanceProfiles {
		entry := &inv.instanceProfiles[i]
		if !tagsMatchScope(entry.tags, owner, sessionID) {
			continue
		}
		resources = append(resources, Resource{
			ARN:       aws.ToString(entry.profile.Arn),
			Type:      ResourceIAMInstanceProfile,
			Region:    t.region,
			SessionID: entry.tags[TagSession],
			Owner:     entry.tags[TagOwner],
		})
		// Only matching profiles reach this call, so following the profile's roles
		// stays proportional to the session rather than to the account.
		profileResources, err := t.findIAMResourcesForTaggedInstanceProfile(ctx, lister, &entry.profile, entry.tags)
		if err != nil {
			return nil, err
		}
		resources = append(resources, profileResources...)
	}
	for i := range inv.roles {
		entry := &inv.roles[i]
		if !tagsMatchScope(entry.tags, owner, sessionID) {
			continue
		}
		resources = append(resources, Resource{
			ARN:       entry.arn,
			Type:      ResourceIAMRole,
			Region:    t.region,
			SessionID: entry.tags[TagSession],
			Owner:     entry.tags[TagOwner],
		})
	}
	for i := range inv.policies {
		entry := &inv.policies[i]
		if !tagsMatchScope(entry.tags, owner, sessionID) {
			continue
		}
		resources = append(resources, Resource{
			ARN:       entry.arn,
			Type:      ResourceIAMPolicy,
			Region:    t.region,
			SessionID: entry.tags[TagSession],
			Owner:     entry.tags[TagOwner],
		})
	}
	return resources, nil
}

// appendManagedScope de-duplicates by (owner, session) and keeps the latest TTL
// so any live same-name successor fails closed.
func appendManagedScope(scopes []ExpiredManagedScope, candidate *ExpiredManagedScope) []ExpiredManagedScope {
	// ponytail: account-local lists are small; add a session-keyed index only if live scans prove this pass expensive.
	index := slices.IndexFunc(scopes, func(scope ExpiredManagedScope) bool {
		return scope.Session == candidate.Session &&
			(ownerTagMatches(scope.Owner, candidate.Owner) || ownerTagMatches(candidate.Owner, scope.Owner))
	})
	if index < 0 {
		return append(scopes, *candidate)
	}
	if ownerTagMatches(candidate.Owner, scopes[index].Owner) {
		scopes[index].Owner = candidate.Owner
	}
	if candidate.TTL.After(scopes[index].TTL) {
		scopes[index].TTL = candidate.TTL
	}
	for _, resource := range candidate.Resources {
		if strings.TrimSpace(resource.ARN) == "" {
			continue
		}
		if slices.ContainsFunc(scopes[index].Resources, func(existing Resource) bool {
			return existing.ARN == resource.ARN && existing.Type == resource.Type
		}) {
			continue
		}
		scopes[index].Resources = append(scopes[index].Resources, resource)
	}
	return scopes
}

// MergeManagedScopes combines regional and account-global TTL evidence before
// expiration is evaluated.
func MergeManagedScopes(tagged, iamDerived []ExpiredManagedScope) []ExpiredManagedScope {
	merged := append([]ExpiredManagedScope(nil), tagged...)
	for i := range iamDerived {
		merged = appendManagedScope(merged, &iamDerived[i])
	}
	slices.SortFunc(merged, func(a, b ExpiredManagedScope) int {
		if a.Owner != b.Owner {
			return strings.Compare(a.Owner, b.Owner)
		}
		return strings.Compare(a.Session, b.Session)
	})
	return merged
}