//nolint:gocritic // Catalog lifecycle boundaries deliberately receive immutable activation contracts by value.
package byoc

import (
	"context"
	"errors"
	"fmt"
	"hash/fnv"
	"strconv"
	"strings"
)

const (
	byocGlobalDatabase        = "nextgen_global_service"
	byocUIFeatureFlagName     = "byoc_enable"
	byocUIFeatureFlagArtifact = "ui_feature_flag"
)

func (d *AWSDriver) applyCatalog(ctx context.Context, input Input, journal *Journal) (*StageResult, error) {
	fingerprint, err := Fingerprint(input)
	if err != nil {
		return nil, err
	}
	selectorSlot, err := d.readSelectorSlot(ctx, input)
	if err != nil {
		return nil, fmt.Errorf("revalidate BYOC selector-slot reservation before catalog mutation: %w", err)
	}
	if !selectorSlotBoundToInput(selectorSlot, input, fingerprint) {
		return nil, errors.New("BYOC catalog mutation requires the exact live selector-slot reservation")
	}
	if err := d.validateCatalogSelectionBoundary(ctx, input, journal); err != nil {
		return nil, err
	}
	if err := d.validateCatalogIdentityBoundary(ctx, input); err != nil {
		return nil, err
	}
	artifacts := catalogArtifacts(input)
	sql := renderCatalogSQL(input, artifacts)
	if _, err := d.kubernetesRunner.Run(ctx, Command{
		Name: "kubectl",
		Args: kubeArgs(input,
			"-n", "cp9-system", "exec", "-i", "deployment/premium-metadb", "--",
			"env", "MYSQL_PWD=cp9-local", "mysql", "-uroot", byocGlobalDatabase,
		),
		Stdin: []byte(sql),
	}); err != nil {
		return nil, fmt.Errorf("register BYOC Region in the shared global catalog: %w", err)
	}
	record := &StageRecord{Artifacts: artifacts}
	if _, err := d.revalidateCatalog(ctx, input, record); err != nil {
		return nil, fmt.Errorf("verify BYOC Region global catalog: %w", err)
	}
	return &StageResult{Artifacts: artifacts}, nil
}

// validateCatalogSelectionBoundary protects the public API's actual Region
// selector. nextgen-global-service selects one Region by tenant, provider,
// physical region, and service plan; logical Region/account identity is not in
// that key. A second active BYOC Region in the same slot would therefore route
// creates nondeterministically even when it uses another customer account.
func (d *AWSDriver) validateCatalogSelectionBoundary(ctx context.Context, input Input, journal *Journal) error {
	artifacts := catalogArtifacts(input)
	query := fmt.Sprintf(`SELECT
COUNT(DISTINCT CASE WHEN r.id=%s THEN r.id END),
COUNT(DISTINCT CASE WHEN r.id<>%s THEN r.id END)
FROM regions r
JOIN providers p ON p.id=r.provider_id AND p.deleted_at IS NULL
JOIN region_service_plans rsp ON rsp.region_id=r.id AND rsp.deleted_at IS NULL
JOIN region_tenants rt ON rt.region_id=r.id AND rt.region_service_plan_id=rsp.id AND rt.deleted_at IS NULL
WHERE r.deleted_at IS NULL AND r.status='active' AND r.service_plan='BYOC'
  AND p.name='aws' AND p.service_plan='BYOC'
  AND rsp.service_plan='BYOC' AND r.region_name=%s AND rt.tenant_id=%s;`,
		artifacts["region_id"], artifacts["region_id"], sqlQuote(input.Region), input.TenantID)
	output, err := d.runGlobalCatalogQuery(ctx, input, query)
	if err != nil {
		return fmt.Errorf("observe BYOC catalog selector boundary: %w", err)
	}
	fields := strings.Fields(string(output))
	if len(fields) != 2 {
		return fmt.Errorf("observe BYOC catalog selector boundary: unexpected result %q", strings.TrimSpace(string(output)))
	}
	owned, ownedErr := strconv.Atoi(fields[0])
	foreign, foreignErr := strconv.Atoi(fields[1])
	if ownedErr != nil || foreignErr != nil || owned < 0 || owned > 1 || foreign < 0 {
		return fmt.Errorf("observe BYOC catalog selector boundary: invalid counts %q", strings.TrimSpace(string(output)))
	}
	if foreign != 0 {
		return fmt.Errorf(
			"BYOC catalog selector for tenant %s, aws/%s, and service plan BYOC already has %d active foreign Region(s)",
			input.TenantID,
			input.Region,
			foreign,
		)
	}
	if journal == nil && owned != 0 {
		return fmt.Errorf("BYOC catalog selector already contains the session Region without a matching activation journal; CP9 will not adopt it")
	}
	return nil
}

// validateCatalogIdentityBoundary prevents ON DUPLICATE KEY UPDATE from ever
// modifying a foreign row. Exact deterministic rows may be resumed or
// resurrected after a prior tombstone; any partial ID/unique-key match is a
// collision and blocks before the first customer-account mutation.
func (d *AWSDriver) validateCatalogIdentityBoundary(ctx context.Context, input Input) error {
	a := catalogArtifacts(input)
	provider := "(SELECT id FROM providers WHERE name='aws' AND service_plan='BYOC' AND deleted_at IS NULL LIMIT 1)"
	query := fmt.Sprintf(
		`SELECT (
(SELECT COUNT(*) FROM regions
 WHERE (id=%s OR name=%s) AND NOT (id=%s AND name=%s)) +
(SELECT COUNT(*) FROM region_service_plans
 WHERE (id=%s OR (region_id=%s AND service_plan='BYOC'))
   AND NOT (id=%s AND region_id=%s AND service_plan='BYOC')) +
(SELECT COUNT(*) FROM region_tenants
 WHERE (id=%s OR (region_id=%s AND tenant_id=%s))
   AND NOT (id=%s AND region_id=%s AND tenant_id=%s AND region_service_plan_id=%s)) +
(SELECT COUNT(*) FROM provider_identities
 WHERE (id=%s OR (region_id=%s AND identity=%s))
   AND NOT (id=%s AND region_id=%s AND identity=%s)) +
(SELECT COUNT(*) FROM default_provider_identities
 WHERE (id=%s OR (provider_id=%s AND service_plan='BYOC' AND region_id=%s AND tenant_id=%s))
   AND NOT (id=%s AND provider_id=%s AND service_plan='BYOC' AND region_id=%s AND tenant_id=%s AND identity=%s)) +
(SELECT COUNT(*) FROM rcu_quota
 WHERE (id=%s OR (tenant_id=%s AND service_plan='BYOC'))
   AND NOT (id=%s AND tenant_id=%s AND service_plan='BYOC')) +
(SELECT IF(COUNT(*)=1, 0, 1) FROM account.feature_flags
 WHERE flag_name=%s)
) AS catalog_identity_conflicts;`,
		a["region_id"],
		sqlQuote(a["logical_region_name"]),
		a["region_id"],
		sqlQuote(a["logical_region_name"]),
		a["region_service_plan_id"],
		a["region_id"],
		a["region_service_plan_id"],
		a["region_id"],
		a["region_tenant_id"],
		a["region_id"],
		input.TenantID,
		a["region_tenant_id"],
		a["region_id"],
		input.TenantID,
		a["region_service_plan_id"],
		a["provider_identity_id"],
		a["region_id"],
		sqlQuote(input.CustomerAccountID),
		a["provider_identity_id"],
		a["region_id"],
		sqlQuote(input.CustomerAccountID),
		a["default_identity_id"],
		provider,
		a["region_id"],
		input.TenantID,
		a["default_identity_id"],
		provider,
		a["region_id"],
		input.TenantID,
		sqlQuote(input.CustomerAccountID),
		a["rcu_quota_id"],
		input.TenantID,
		a["rcu_quota_id"],
		input.TenantID,
		sqlQuote(a[byocUIFeatureFlagArtifact]),
	)
	output, err := d.runGlobalCatalogQuery(ctx, input, query)
	if err != nil {
		return fmt.Errorf("observe BYOC catalog identity boundary: %w", err)
	}
	conflicts, err := strconv.Atoi(strings.TrimSpace(string(output)))
	if err != nil || conflicts < 0 {
		return fmt.Errorf("observe BYOC catalog identity boundary: unexpected result %q", strings.TrimSpace(string(output)))
	}
	if conflicts != 0 {
		return fmt.Errorf("BYOC catalog identity preflight found %d conflicting row(s); CP9 will not update or adopt them", conflicts)
	}
	return nil
}

func (d *AWSDriver) revalidateCatalog(ctx context.Context, input Input, record *StageRecord) (*StageResult, error) {
	if record == nil {
		return nil, ErrNeedsReconcile
	}
	want := catalogArtifacts(input)
	for key, value := range want {
		if record.Artifacts[key] != value {
			return nil, ErrNeedsReconcile
		}
	}
	query := renderCatalogPresenceQuery(input, want)
	output, err := d.kubernetesRunner.Run(ctx, Command{
		Name: "kubectl",
		Args: kubeArgs(input,
			"-n", "cp9-system", "exec", "deployment/premium-metadb", "--",
			"env", "MYSQL_PWD=cp9-local", "mysql", "-uroot", "-Nse", query, byocGlobalDatabase,
		),
	})
	if err != nil {
		return nil, ErrNeedsReconcile
	}
	if strings.Join(strings.Fields(string(output)), "\t") != "1\t1\t1\t1\t1\t1\t1\t1" {
		return nil, ErrNeedsReconcile
	}
	return &StageResult{Artifacts: cloneArtifacts(record.Artifacts)}, nil
}

func catalogArtifacts(input Input) map[string]string {
	artifacts := map[string]string{
		"service_plan":            "BYOC",
		"logical_region_name":     byocLogicalRegionName(input),
		"api_region_id":           "aws-" + strings.ToLower(strings.ReplaceAll(input.Region, "_", "-")),
		"region_id":               strconv.FormatUint(stableCatalogID("region:"+input.Session+":"+input.CustomerAccountID), 10),
		"region_service_plan_id":  strconv.FormatUint(stableCatalogID("plan:"+input.Session+":"+input.CustomerAccountID), 10),
		"region_tenant_id":        strconv.FormatUint(stableCatalogID("tenant:"+input.Session+":"+input.CustomerAccountID), 10),
		"provider_identity_id":    strconv.FormatUint(stableCatalogID("identity:"+input.Session+":"+input.CustomerAccountID), 10),
		"default_identity_id":     strconv.FormatUint(stableCatalogID("default-identity:"+input.Session+":"+input.CustomerAccountID), 10),
		"rcu_quota_id":            strconv.FormatUint(stableCatalogID("quota:BYOC:"+input.TenantID), 10),
		"service_address":         byocRegionalAPIServiceAddress,
		byocUIFeatureFlagArtifact: byocUIFeatureFlagName,
	}
	return artifacts
}

func byocLogicalRegionName(input Input) string {
	return "byoc-" + input.Session
}

func stableCatalogID(value string) uint64 {
	h := fnv.New64a()
	_, _ = h.Write([]byte(value))
	return 8_000_000_000_000_000_000 + h.Sum64()%1_000_000_000_000_000_000
}

func renderCatalogSQL(input Input, a map[string]string) string {
	provider := "(SELECT id FROM providers WHERE name='aws' AND service_plan='BYOC' AND deleted_at IS NULL LIMIT 1)"
	return fmt.Sprintf(`START TRANSACTION;
INSERT INTO providers (created_at, updated_at, deleted_at, name, display_name, service_plan, is_public, status)
VALUES (NOW(3), NOW(3), NULL, 'aws', 'AWS', 'BYOC', 1, 'active')
ON DUPLICATE KEY UPDATE updated_at=VALUES(updated_at), deleted_at=NULL, display_name=VALUES(display_name), is_public=1, status='active';
INSERT INTO regions (id, created_at, updated_at, deleted_at, name, region_name, provider_id, display_name, country_code, area, priority, default_backup_time, service_address, service_plan, is_public, status, shared_bucket_url)
VALUES (%s, NOW(3), NOW(3), NULL, %s, %s, %s, %s, 'US', 'Americas', 100, '00:00', %s, 'BYOC', 0, 'active', %s)
ON DUPLICATE KEY UPDATE updated_at=VALUES(updated_at), deleted_at=NULL, region_name=VALUES(region_name), provider_id=VALUES(provider_id), display_name=VALUES(display_name), service_address=VALUES(service_address), service_plan='BYOC', is_public=0, status='active', shared_bucket_url=VALUES(shared_bucket_url);
INSERT INTO region_service_plans (id, created_at, updated_at, deleted_at, region_id, service_plan, is_public)
VALUES (%s, NOW(3), NOW(3), NULL, %s, 'BYOC', 0)
ON DUPLICATE KEY UPDATE updated_at=VALUES(updated_at), deleted_at=NULL, is_public=0;
INSERT INTO region_tenants (id, created_at, updated_at, deleted_at, region_id, tenant_id, region_service_plan_id)
VALUES (%s, NOW(3), NOW(3), NULL, %s, %s, %s)
ON DUPLICATE KEY UPDATE updated_at=VALUES(updated_at), deleted_at=NULL, region_id=VALUES(region_id), tenant_id=VALUES(tenant_id), region_service_plan_id=VALUES(region_service_plan_id);
INSERT INTO provider_identities (id, created_at, updated_at, deleted_at, region_id, identity)
VALUES (%s, NOW(3), NOW(3), NULL, %s, %s)
ON DUPLICATE KEY UPDATE updated_at=VALUES(updated_at), deleted_at=NULL, identity=VALUES(identity);
INSERT INTO default_provider_identities (id, created_at, updated_at, deleted_at, provider_id, service_plan, region_id, tenant_id, identity)
VALUES (%s, NOW(3), NOW(3), NULL, %s, 'BYOC', %s, %s, %s)
ON DUPLICATE KEY UPDATE updated_at=VALUES(updated_at), deleted_at=NULL, identity=VALUES(identity);
INSERT INTO rcu_quota (id, created_at, updated_at, deleted_at, tenant_id, quota, total_max_rcu, burst_allowance_percent, service_plan, idle_resource_pool_ttl_in_hours, vcpus_limit_per_resource_pool, minimal_rcu_baseline, minimal_rcu_coefficient, scale_down_cooldown_minutes)
VALUES (%s, NOW(3), NOW(3), NULL, %s, 1000000, 1000000, 0, 'BYOC', 1, 0, 5000, 0.25, 720)
ON DUPLICATE KEY UPDATE updated_at=VALUES(updated_at), deleted_at=NULL, quota=VALUES(quota), total_max_rcu=VALUES(total_max_rcu), burst_allowance_percent=VALUES(burst_allowance_percent), idle_resource_pool_ttl_in_hours=VALUES(idle_resource_pool_ttl_in_hours), vcpus_limit_per_resource_pool=VALUES(vcpus_limit_per_resource_pool), minimal_rcu_baseline=VALUES(minimal_rcu_baseline), minimal_rcu_coefficient=VALUES(minimal_rcu_coefficient), scale_down_cooldown_minutes=VALUES(scale_down_cooldown_minutes);
INSERT INTO account.feature_flag_orgs (flag_id, org_id)
SELECT id, %s FROM account.feature_flags WHERE flag_name=%s
ON DUPLICATE KEY UPDATE updated_at=NOW();
COMMIT;
	`, a["region_id"], sqlQuote(a["logical_region_name"]), sqlQuote(input.Region), provider,
		sqlQuote("CP9 BYOC "+input.Region), sqlQuote(a["service_address"]),
		sqlQuote(""),
		a["region_service_plan_id"], a["region_id"],
		a["region_tenant_id"], a["region_id"], input.TenantID, a["region_service_plan_id"],
		a["provider_identity_id"], a["region_id"], sqlQuote(input.CustomerAccountID),
		a["default_identity_id"], provider, a["region_id"], input.TenantID, sqlQuote(input.CustomerAccountID),
		a["rcu_quota_id"], input.TenantID,
		input.TenantID, sqlQuote(a[byocUIFeatureFlagArtifact]),
	)
}

func renderCatalogPresenceQuery(input Input, a map[string]string) string {
	return fmt.Sprintf(`SELECT
(SELECT COUNT(*) FROM providers WHERE name='aws' AND service_plan='BYOC' AND status='active' AND deleted_at IS NULL),
(SELECT COUNT(*) FROM regions WHERE id=%s AND name=%s AND region_name=%s AND service_plan='BYOC' AND service_address=%s AND status='active' AND deleted_at IS NULL),
(SELECT COUNT(*) FROM region_service_plans WHERE id=%s AND region_id=%s AND service_plan='BYOC' AND deleted_at IS NULL),
(SELECT COUNT(*) FROM region_tenants WHERE id=%s AND region_id=%s AND tenant_id=%s AND region_service_plan_id=%s AND deleted_at IS NULL),
(SELECT COUNT(*) FROM provider_identities WHERE id=%s AND region_id=%s AND identity=%s AND deleted_at IS NULL),
(SELECT COUNT(*) FROM default_provider_identities WHERE id=%s AND service_plan='BYOC' AND region_id=%s AND tenant_id=%s AND identity=%s AND deleted_at IS NULL),
(SELECT COUNT(*) FROM rcu_quota WHERE id=%s AND tenant_id=%s AND service_plan='BYOC' AND deleted_at IS NULL),
(SELECT COUNT(*) FROM account.feature_flags f
 JOIN account.feature_flag_orgs o ON o.flag_id=f.id
 WHERE f.flag_name=%s AND o.org_id=%s);`,
		a["region_id"], sqlQuote(a["logical_region_name"]), sqlQuote(input.Region), sqlQuote(a["service_address"]),
		a["region_service_plan_id"], a["region_id"],
		a["region_tenant_id"], a["region_id"], input.TenantID, a["region_service_plan_id"],
		a["provider_identity_id"], a["region_id"], sqlQuote(input.CustomerAccountID),
		a["default_identity_id"], a["region_id"], input.TenantID, sqlQuote(input.CustomerAccountID),
		a["rcu_quota_id"], input.TenantID,
		sqlQuote(a[byocUIFeatureFlagArtifact]), input.TenantID)
}

func sqlQuote(value string) string {
	return "'" + strings.ReplaceAll(value, "'", "''") + "'"
}