resource_sql_database_instance.go

// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
// ----------------------------------------------------------------------------
//
//	***     AUTO GENERATED CODE    ***    Type: Handwritten     ***
//
// ----------------------------------------------------------------------------
//
//	This code is generated by Magic Modules using the following:
//
//	Source file: https://github.com/GoogleCloudPlatform/magic-modules/tree/main/mmv1/third_party/terraform/services/sql/resource_sql_database_instance.go.tmpl
//
//	DO NOT EDIT this file directly. Any changes made to this file will be
//	overwritten during the next generation cycle.
//
// ----------------------------------------------------------------------------
package sql

import (
	"context"
	"errors"
	"fmt"
	"log"
	"reflect"
	"slices"
	"strconv"
	"strings"
	"time"

	"github.com/hashicorp/go-cty/cty"
	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/customdiff"
	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/id"
	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/validation"

	"github.com/hashicorp/terraform-provider-google/google/services/servicenetworking"
	"github.com/hashicorp/terraform-provider-google/google/tpgresource"
	transport_tpg "github.com/hashicorp/terraform-provider-google/google/transport"
	"github.com/hashicorp/terraform-provider-google/google/verify"

	"google.golang.org/api/googleapi"
	sqladmin "google.golang.org/api/sqladmin/v1beta4"
)

// Match fully-qualified or relative URLs
const privateNetworkLinkRegex = "^(?:http(?:s)?://.+/)?projects/(" + verify.ProjectRegex + ")/global/networks/((?:[a-z](?:[-a-z0-9]*[a-z0-9])?))$"

var sqlDatabaseAuthorizedNetWorkSchemaElem *schema.Resource = &schema.Resource{
	Schema: map[string]*schema.Schema{
		"expiration_time": {
			Type:     schema.TypeString,
			Optional: true,
		},
		"name": {
			Type:     schema.TypeString,
			Optional: true,
		},
		"value": {
			Type:     schema.TypeString,
			Required: true,
		},
	},
}

var sqlDatabaseFlagSchemaElem *schema.Resource = &schema.Resource{
	Schema: map[string]*schema.Schema{
		"value": {
			Type:        schema.TypeString,
			Required:    true,
			Description: `Value of the flag.`,
		},
		"name": {
			Type:        schema.TypeString,
			Required:    true,
			Description: `Name of the flag.`,
		},
	},
}

var (
	backupConfigurationKeys = []string{
		"settings.0.backup_configuration.0.binary_log_enabled",
		"settings.0.backup_configuration.0.enabled",
		"settings.0.backup_configuration.0.start_time",
		"settings.0.backup_configuration.0.location",
		"settings.0.backup_configuration.0.point_in_time_recovery_enabled",
		"settings.0.backup_configuration.0.backup_retention_settings",
		"settings.0.backup_configuration.0.transaction_log_retention_days",
		"settings.0.backup_configuration.0.backup_tier",
	}

	connectionPoolConfigKeys = []string{
		"settings.0.connection_pool_config.0.connection_pooling_enabled",
		"settings.0.connection_pool_config.0.flags",
	}

	ipConfigurationKeys = []string{
		"settings.0.ip_configuration.0.authorized_networks",
		"settings.0.ip_configuration.0.ipv4_enabled",
		"settings.0.ip_configuration.0.private_network",
		"settings.0.ip_configuration.0.allocated_ip_range",
		"settings.0.ip_configuration.0.enable_private_path_for_google_cloud_services",
		"settings.0.ip_configuration.0.psc_config",
		"settings.0.ip_configuration.0.ssl_mode",
		"settings.0.ip_configuration.0.server_ca_mode",
		"settings.0.ip_configuration.0.server_ca_pool",
		"settings.0.ip_configuration.0.custom_subject_alternative_names",
	}

	maintenanceWindowKeys = []string{
		"settings.0.maintenance_window.0.day",
		"settings.0.maintenance_window.0.hour",
		"settings.0.maintenance_window.0.update_track",
	}

	replicaConfigurationKeys = []string{
		"replica_configuration.0.cascadable_replica",
		"replica_configuration.0.ca_certificate",
		"replica_configuration.0.client_certificate",
		"replica_configuration.0.client_key",
		"replica_configuration.0.connect_retry_interval",
		"replica_configuration.0.dump_file_path",
		"replica_configuration.0.failover_target",
		"replica_configuration.0.master_heartbeat_period",
		"replica_configuration.0.password",
		"replica_configuration.0.ssl_cipher",
		"replica_configuration.0.username",
		"replica_configuration.0.verify_server_certificate",
	}

	insightsConfigKeys = []string{
		"settings.0.insights_config.0.query_insights_enabled",
		"settings.0.insights_config.0.query_string_length",
		"settings.0.insights_config.0.record_application_tags",
		"settings.0.insights_config.0.record_client_address",
		"settings.0.insights_config.0.query_plans_per_minute",
	}

	sqlServerAuditConfigurationKeys = []string{
		"settings.0.sql_server_audit_config.0.bucket",
		"settings.0.sql_server_audit_config.0.retention_interval",
		"settings.0.sql_server_audit_config.0.upload_interval",
	}

	readPoolAutoScaleConfigKeys = []string{
		"settings.0.read_pool_auto_scale_config.0.enabled",
		"settings.0.read_pool_auto_scale_config.0.min_node_count",
		"settings.0.read_pool_auto_scale_config.0.max_node_count",
		"settings.0.read_pool_auto_scale_config.0.target_metrics",
		"settings.0.read_pool_auto_scale_config.0.disable_scale_in",
		"settings.0.read_pool_auto_scale_config.0.scale_in_cooldown_seconds",
		"settings.0.read_pool_auto_scale_config.0.scale_out_cooldown_seconds",
	}
)

func nodeCountCustomDiff(ctx context.Context, d *schema.ResourceDiff, meta interface{}) error {
	autoScaleEnabled := d.Get("settings.0.read_pool_auto_scale_config.0.enabled").(bool)

	if !autoScaleEnabled {
		// Keep the diff
		return nil
	}

	currentNodeCountI, _ := d.GetChange("node_count")
	currentNodeCount := currentNodeCountI.(int)
	minNodeCount := d.Get("settings.0.read_pool_auto_scale_config.0.min_node_count").(int)
	maxNodeCount := d.Get("settings.0.read_pool_auto_scale_config.0.max_node_count").(int)
	if currentNodeCount < minNodeCount {
		// Node count will only be less than min node count if min node count is being increased.
		// Set node count to be new min node count.
		return d.SetNew("node_count", minNodeCount)
	}
	if currentNodeCount > maxNodeCount {
		// Node count will only be greater than max node count if max node count is being descreased.
		// Set node count to be new max node count.
		return d.SetNew("node_count", maxNodeCount)
	}

	// Otherwise when autoscaling is enabled, ignore the node count in config.
	return d.Clear("node_count")
}

func diskSizeCutomizeDiff(ctx context.Context, d *schema.ResourceDiff, meta interface{}) error {
	key := "settings.0.disk_size"

	old, new := d.GetChange(key)
	// It's okay to remove size entirely.
	if old == nil || new == nil {
		return nil
	}
	autoResizeI, exists := d.GetOkExists("settings.0.disk_autoresize")
	autoResize := !exists || autoResizeI.(bool)

	if old.(int) <= new.(int) {
		// If a resize up, always allow it - keep the diff.
		return nil
	}

	if autoResize {
		// Allow having disk size larger in the state than in config if auto resize is enabled.
		// Delete the diff.
		err := d.Clear(key)
		if err != nil {
			return err
		}
		return nil
	}

	// If we are here, we are trying to shrink the disk with auto resize disabled and no ignore changes on disk size.
	// This will force a new resource.
	if err := d.ForceNew(key); err != nil {
		return err
	}

	return nil
}

func ResourceSqlDatabaseInstance() *schema.Resource {
	return &schema.Resource{
		Create: resourceSqlDatabaseInstanceCreate,
		Read:   resourceSqlDatabaseInstanceRead,
		Update: resourceSqlDatabaseInstanceUpdate,
		Delete: resourceSqlDatabaseInstanceDelete,
		Importer: &schema.ResourceImporter{
			State: resourceSqlDatabaseInstanceImport,
		},

		Timeouts: &schema.ResourceTimeout{
			Create: schema.DefaultTimeout(90 * time.Minute),
			Update: schema.DefaultTimeout(90 * time.Minute),
			Delete: schema.DefaultTimeout(90 * time.Minute),
		},

		CustomizeDiff: customdiff.All(
			tpgresource.DefaultProviderProject,
			diskSizeCutomizeDiff,
			customdiff.ForceNewIf("master_instance_name", func(_ context.Context, d *schema.ResourceDiff, meta interface{}) bool {
				// If we set master but this is not the new master of a switchover, require replacement and warn user.
				return !isSwitchoverFromOldPrimarySide(d)
			}),
			customdiff.ForceNewIf("replica_configuration.0.cascadable_replica", func(_ context.Context, d *schema.ResourceDiff, meta interface{}) bool {
				return !isSwitchoverFromOldPrimarySide(d)
			}),
			customdiff.IfValueChange("instance_type", isReplicaPromoteRequested, checkPromoteConfigurationsAndUpdateDiff),
			privateNetworkCustomizeDiff,
			pitrSupportDbCustomizeDiff,
			nodeCountCustomDiff,
		),

		Schema: map[string]*schema.Schema{
			"region": {
				Type:        schema.TypeString,
				Optional:    true,
				Computed:    true,
				ForceNew:    true,
				Description: `The region the instance will sit in. Note, Cloud SQL is not available in all regions. A valid region must be provided to use this resource. If a region is not provided in the resource definition, the provider region will be used instead, but this will be an apply-time error for instances if the provider region is not supported with Cloud SQL. If you choose not to provide the region argument for this resource, make sure you understand this.`,
			},
			"deletion_protection": {
				Type:        schema.TypeBool,
				Default:     true,
				Optional:    true,
				Description: `Used to block Terraform from deleting a SQL Instance. Defaults to true.`,
			},
			"final_backup_description": {
				Type:        schema.TypeString,
				Optional:    true,
				Description: `The description of final backup if instance enable create final backup during instance deletion. `,
			},
			"settings": {
				Type:         schema.TypeList,
				Optional:     true,
				Computed:     true,
				AtLeastOneOf: []string{"settings", "clone", "point_in_time_restore_context"},
				MaxItems:     1,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"version": {
							Type:        schema.TypeInt,
							Computed:    true,
							Description: `Used to make sure changes to the settings block are atomic.`,
						},
						"tier": {
							Type:        schema.TypeString,
							Required:    true,
							Description: `The machine type to use. See tiers for more details and supported versions. Postgres supports only shared-core machine types, and custom machine types such as db-custom-2-13312. See the Custom Machine Type Documentation to learn about specifying custom machine types.`,
						},
						"edition": {
							Type:         schema.TypeString,
							Optional:     true,
							Computed:     true,
							ValidateFunc: validation.StringInSlice([]string{"ENTERPRISE", "ENTERPRISE_PLUS"}, false),
							Description:  `The edition of the instance, can be ENTERPRISE or ENTERPRISE_PLUS.`,
						},
						"advanced_machine_features": {
							Type:     schema.TypeList,
							Optional: true,
							MaxItems: 1,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"threads_per_core": {
										Type:        schema.TypeInt,
										Optional:    true,
										Description: `The number of threads per physical core. Can be 1 or 2.`,
									},
								},
							},
						},
						"data_cache_config": {
							Type:        schema.TypeList,
							Optional:    true,
							Computed:    true,
							MaxItems:    1,
							Description: `Data cache configurations.`,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"data_cache_enabled": {
										Type:        schema.TypeBool,
										Optional:    true,
										Default:     false,
										Description: `Whether data cache is enabled for the instance.`,
									},
								},
							},
						},
						"activation_policy": {
							Type:        schema.TypeString,
							Optional:    true,
							Default:     "ALWAYS",
							Description: `This specifies when the instance should be active. Can be either ALWAYS, NEVER or ON_DEMAND.`,
						},
						"active_directory_config": {
							Type:     schema.TypeList,
							Optional: true,
							MaxItems: 1,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"domain": {
										Type:        schema.TypeString,
										Required:    true,
										Description: `Domain name of the Active Directory for SQL Server (e.g., mydomain.com).`,
									},
								},
							},
						},
						"deny_maintenance_period": {
							Type:     schema.TypeList,
							Optional: true,
							MaxItems: 1,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"end_date": {
										Type:        schema.TypeString,
										Required:    true,
										Description: `End date before which maintenance will not take place. The date is in format yyyy-mm-dd i.e., 2020-11-01, or mm-dd, i.e., 11-01`,
									},
									"start_date": {
										Type:        schema.TypeString,
										Required:    true,
										Description: `Start date after which maintenance will not take place. The date is in format yyyy-mm-dd i.e., 2020-11-01, or mm-dd, i.e., 11-01`,
									},
									"time": {
										Type:        schema.TypeString,
										Required:    true,
										Description: `Time in UTC when the "deny maintenance period" starts on start_date and ends on end_date. The time is in format: HH:mm:SS, i.e., 00:00:00`,
									},
								},
							},
						},
						"sql_server_audit_config": {
							Type:     schema.TypeList,
							Optional: true,
							MaxItems: 1,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"bucket": {
										Type:         schema.TypeString,
										Optional:     true,
										AtLeastOneOf: sqlServerAuditConfigurationKeys,
										Description:  `The name of the destination bucket (e.g., gs://mybucket).`,
									},
									"retention_interval": {
										Type:         schema.TypeString,
										Optional:     true,
										AtLeastOneOf: sqlServerAuditConfigurationKeys,
										Description:  `How long to keep generated audit files. A duration in seconds with up to nine fractional digits, terminated by 's'. Example: "3.5s"..`,
									},
									"upload_interval": {
										Type:         schema.TypeString,
										Optional:     true,
										AtLeastOneOf: sqlServerAuditConfigurationKeys,
										Description:  `How often to upload generated audit files. A duration in seconds with up to nine fractional digits, terminated by 's'. Example: "3.5s".`,
									},
								},
							},
						},
						"time_zone": {
							Type:        schema.TypeString,
							Optional:    true,
							Description: `The time_zone to be used by the database engine (supported only for SQL Server), in SQL Server timezone format.`,
						},
						"availability_type": {
							Type:         schema.TypeString,
							Optional:     true,
							Default:      "ZONAL",
							ValidateFunc: validation.StringInSlice([]string{"REGIONAL", "ZONAL"}, false),
							Description: `The availability type of the Cloud SQL instance, high availability
(REGIONAL) or single zone (ZONAL). For all instances, ensure that
settings.backup_configuration.enabled is set to true.
For MySQL instances, ensure that settings.backup_configuration.binary_log_enabled is set to true.
For Postgres instances, ensure that settings.backup_configuration.point_in_time_recovery_enabled
is set to true. Defaults to ZONAL.
For read pool instances, this field is read-only. The availability type is changed by specifying
the number of nodes (node_count).`,
						},
						"effective_availability_type": {
							Type:     schema.TypeString,
							Computed: true,
							Description: `The availability type of the Cloud SQL instance, high availability
(REGIONAL) or single zone (ZONAL). This field always contains the value that is reported by the
API (for read pools, effective_availability_type may differ from availability_type).`,
						},
						"backup_configuration": {
							Type:     schema.TypeList,
							Optional: true,
							Computed: true,
							MaxItems: 1,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"binary_log_enabled": {
										Type:             schema.TypeBool,
										Optional:         true,
										AtLeastOneOf:     backupConfigurationKeys,
										DiffSuppressFunc: EnhancedBackupManagerDiffSuppressFunc,
										Description:      `True if binary logging is enabled. If settings.backup_configuration.enabled is false, this must be as well. Can only be used with MySQL.`,
									},
									"enabled": {
										Type:             schema.TypeBool,
										Optional:         true,
										AtLeastOneOf:     backupConfigurationKeys,
										DiffSuppressFunc: EnhancedBackupManagerDiffSuppressFunc,
										Description:      `True if backup configuration is enabled.`,
									},
									"start_time": {
										Type:     schema.TypeString,
										Optional: true,
										// start_time is randomly assigned if not set
										Computed:         true,
										AtLeastOneOf:     backupConfigurationKeys,
										DiffSuppressFunc: EnhancedBackupManagerDiffSuppressFunc,
										Description:      `HH:MM format time indicating when backup configuration starts.`,
									},
									"location": {
										Type:         schema.TypeString,
										Optional:     true,
										AtLeastOneOf: backupConfigurationKeys,
										Description:  `Location of the backup configuration.`,
									},
									"point_in_time_recovery_enabled": {
										Type:             schema.TypeBool,
										Optional:         true,
										AtLeastOneOf:     backupConfigurationKeys,
										DiffSuppressFunc: EnhancedBackupManagerDiffSuppressFunc,
										Description:      `True if Point-in-time recovery is enabled.`,
									},
									"transaction_log_retention_days": {
										Type:             schema.TypeInt,
										Computed:         true,
										Optional:         true,
										AtLeastOneOf:     backupConfigurationKeys,
										DiffSuppressFunc: EnhancedBackupManagerDiffSuppressFunc,
										Description:      `The number of days of transaction logs we retain for point in time restore, from 1-7. (For PostgreSQL Enterprise Plus instances, from 1 to 35.)`,
									},
									"backup_retention_settings": {
										Type:             schema.TypeList,
										Optional:         true,
										AtLeastOneOf:     backupConfigurationKeys,
										Computed:         true,
										DiffSuppressFunc: EnhancedBackupManagerDiffSuppressFunc,
										MaxItems:         1,
										Elem: &schema.Resource{
											Schema: map[string]*schema.Schema{
												"retained_backups": {
													Type:        schema.TypeInt,
													Required:    true,
													Description: `Number of backups to retain.`,
												},
												"retention_unit": {
													Type:        schema.TypeString,
													Optional:    true,
													Default:     "COUNT",
													Description: `The unit that 'retainedBackups' represents. Defaults to COUNT`,
												},
											},
										},
									},
									"backup_tier": {
										Type:        schema.TypeString,
										Computed:    true,
										Description: `Backup tier that manages the backups for the instance.`,
									},
								},
							},
						},
						"collation": {
							Type:        schema.TypeString,
							Optional:    true,
							ForceNew:    true,
							Description: `The name of server instance collation.`,
						},
						"database_flags": {
							Type:     schema.TypeSet,
							Optional: true,
							Set:      schema.HashResource(sqlDatabaseFlagSchemaElem),
							Elem:     sqlDatabaseFlagSchemaElem,
						},
						"disk_autoresize": {
							Type:        schema.TypeBool,
							Optional:    true,
							Default:     true,
							Description: `Enables auto-resizing of the storage size. Defaults to true.`,
						},
						"disk_autoresize_limit": {
							Type:        schema.TypeInt,
							Optional:    true,
							Default:     0,
							Description: `The maximum size, in GB, to which storage capacity can be automatically increased. The default value is 0, which specifies that there is no limit.`,
						},
						"enable_google_ml_integration": {
							Type:        schema.TypeBool,
							Optional:    true,
							Description: `Enables Vertex AI Integration.`,
						},
						"enable_dataplex_integration": {
							Type:        schema.TypeBool,
							Optional:    true,
							Description: `Enables Dataplex Integration.`,
						},
						"disk_size": {
							Type:        schema.TypeInt,
							Optional:    true,
							Computed:    true,
							Description: `The size of data disk, in GB. Size of a running instance cannot be reduced but can be increased. The minimum value is 10GB for PD_SSD, PD_HDD and 20GB for HYPERDISK_BALANCED.`,
						},
						"disk_type": {
							Type:             schema.TypeString,
							Optional:         true,
							Computed:         true,
							ForceNew:         true,
							DiffSuppressFunc: caseDiffDashSuppress,
							Description:      `The type of supported data disk is tier dependent and can be PD_SSD or PD_HDD or HYPERDISK_BALANCED.`,
						},
						"connection_pool_config": {
							Type:        schema.TypeSet,
							Optional:    true,
							Computed:    true,
							Description: `The managed connection pool setting for a Cloud SQL instance.`,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"connection_pooling_enabled": {
										Type:        schema.TypeBool,
										Optional:    true,
										Description: `Whether Managed Connection Pool is enabled for this instance.`,
									},
									"flags": {
										Type:        schema.TypeSet,
										Optional:    true,
										Set:         schema.HashResource(sqlDatabaseFlagSchemaElem),
										Elem:        sqlDatabaseFlagSchemaElem,
										Description: `List of connection pool configuration flags`,
									},
								},
							},
						},
						"ip_configuration": {
							Type:     schema.TypeList,
							Optional: true,
							Computed: true,
							MaxItems: 1,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"authorized_networks": {
										Type:         schema.TypeSet,
										Optional:     true,
										Set:          schema.HashResource(sqlDatabaseAuthorizedNetWorkSchemaElem),
										Elem:         sqlDatabaseAuthorizedNetWorkSchemaElem,
										AtLeastOneOf: ipConfigurationKeys,
									},
									"ipv4_enabled": {
										Type:         schema.TypeBool,
										Optional:     true,
										Default:      true,
										AtLeastOneOf: ipConfigurationKeys,
										Description:  `Whether this Cloud SQL instance should be assigned a public IPV4 address. At least ipv4_enabled must be enabled or a private_network must be configured.`,
									},
									"private_network": {
										Type:             schema.TypeString,
										Optional:         true,
										ValidateFunc:     verify.OrEmpty(verify.ValidateRegexp(privateNetworkLinkRegex)),
										DiffSuppressFunc: tpgresource.CompareSelfLinkRelativePaths,
										AtLeastOneOf:     ipConfigurationKeys,
										Description:      `The VPC network from which the Cloud SQL instance is accessible for private IP. For example, projects/myProject/global/networks/default. Specifying a network enables private IP. At least ipv4_enabled must be enabled or a private_network must be configured. This setting can be updated, but it cannot be removed after it is set.`,
									},
									"allocated_ip_range": {
										Type:         schema.TypeString,
										Optional:     true,
										AtLeastOneOf: ipConfigurationKeys,
										Description:  `The name of the allocated ip range for the private ip CloudSQL instance. For example: "google-managed-services-default". If set, the instance ip will be created in the allocated range. The range name must comply with RFC 1035. Specifically, the name must be 1-63 characters long and match the regular expression [a-z]([-a-z0-9]*[a-z0-9])?.`,
									},
									"enable_private_path_for_google_cloud_services": {
										Type:         schema.TypeBool,
										Optional:     true,
										AtLeastOneOf: ipConfigurationKeys,
										Description:  `Whether Google Cloud services such as BigQuery are allowed to access data in this Cloud SQL instance over a private IP connection. SQLSERVER database type is not supported.`,
									},
									"psc_config": {
										Type:        schema.TypeSet,
										Optional:    true,
										Description: `PSC settings for a Cloud SQL instance.`,
										Elem: &schema.Resource{
											Schema: map[string]*schema.Schema{
												"psc_enabled": {
													Type:        schema.TypeBool,
													Optional:    true,
													Description: `Whether PSC connectivity is enabled for this instance.`,
												},
												"allowed_consumer_projects": {
													Type:     schema.TypeSet,
													Optional: true,
													Elem: &schema.Schema{
														Type: schema.TypeString,
													},
													Set:         schema.HashString,
													Description: `List of consumer projects that are allow-listed for PSC connections to this instance. This instance can be connected to with PSC from any network in these projects. Each consumer project in this list may be represented by a project number (numeric) or by a project id (alphanumeric).`,
												},
												"network_attachment_uri": {
													Type:        schema.TypeString,
													Optional:    true,
													Description: `Name of network attachment resource used to authorize a producer service to connect a PSC interface to the consumer's VPC. For example: "projects/myProject/regions/myRegion/networkAttachments/myNetworkAttachment". This is required to enable outbound connection on a PSC instance.`,
												},
												"psc_auto_connections": {
													Type:     schema.TypeList,
													Optional: true,
													Elem: &schema.Resource{
														Schema: map[string]*schema.Schema{
															"consumer_service_project_id": {
																Type:        schema.TypeString,
																Optional:    true,
																Description: `The project ID of consumer service project of this consumer endpoint.`,
															},
															"consumer_network": {
																Type:        schema.TypeString,
																Required:    true,
																Description: `The consumer network of this consumer endpoint. This must be a resource path that includes both the host project and the network name. The consumer host project of this network might be different from the consumer service project.`,
															},
															"consumer_network_status": {
																Type:        schema.TypeString,
																Computed:    true,
																Description: `The connection policy status of the consumer network.`,
															},
															"ip_address": {
																Type:        schema.TypeString,
																Computed:    true,
																Description: `The IP address of the consumer endpoint.`,
															},
															"status": {
																Type:        schema.TypeString,
																Computed:    true,
																Description: `The connection status of the consumer endpoint.`,
															},
														},
													},
													Description: `A comma-separated list of networks or a comma-separated list of network-project pairs. Each project in this list is represented by a project number (numeric) or by a project ID (alphanumeric). This allows Private Service Connect connections to be created automatically for the specified networks.`,
												},
											},
										},
									},
									"ssl_mode": {
										Type:         schema.TypeString,
										Optional:     true,
										Computed:     true,
										ValidateFunc: validation.StringInSlice([]string{"ALLOW_UNENCRYPTED_AND_ENCRYPTED", "ENCRYPTED_ONLY", "TRUSTED_CLIENT_CERTIFICATE_REQUIRED"}, false),
										Description:  `Specify how SSL connection should be enforced in DB connections.`,
										AtLeastOneOf: ipConfigurationKeys,
									},
									"server_ca_mode": {
										Type:         schema.TypeString,
										Optional:     true,
										Computed:     true,
										ValidateFunc: validation.StringInSlice([]string{"CA_MODE_UNSPECIFIED", "GOOGLE_MANAGED_INTERNAL_CA", "GOOGLE_MANAGED_CAS_CA", "CUSTOMER_MANAGED_CAS_CA"}, false),
										Description:  `Specify how the server certificate's Certificate Authority is hosted.`,
										AtLeastOneOf: ipConfigurationKeys,
									},
									"server_ca_pool": {
										Type:         schema.TypeString,
										Optional:     true,
										Description:  `The resource name of the server CA pool for an instance with "CUSTOMER_MANAGED_CAS_CA" as the "server_ca_mode".`,
										AtLeastOneOf: ipConfigurationKeys,
									},
									"custom_subject_alternative_names": {
										Type:     schema.TypeSet,
										Optional: true,
										Elem: &schema.Schema{
											Type: schema.TypeString,
										},
										Set:          schema.HashString,
										Description:  `The custom subject alternative names for an instance with "CUSTOMER_MANAGED_CAS_CA" as the "server_ca_mode".`,
										AtLeastOneOf: ipConfigurationKeys,
									},
								},
							},
						},
						"location_preference": {
							Type:     schema.TypeList,
							Optional: true,
							MaxItems: 1,
							Computed: true,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"follow_gae_application": {
										Type:         schema.TypeString,
										Optional:     true,
										AtLeastOneOf: []string{"settings.0.location_preference.0.follow_gae_application", "settings.0.location_preference.0.zone"},
										Description:  `A Google App Engine application whose zone to remain in. Must be in the same region as this instance.`,
									},
									"zone": {
										Type:         schema.TypeString,
										Optional:     true,
										AtLeastOneOf: []string{"settings.0.location_preference.0.follow_gae_application", "settings.0.location_preference.0.zone"},
										Description:  `The preferred compute engine zone.`,
									},
									"secondary_zone": {
										Type:        schema.TypeString,
										Optional:    true,
										Description: `The preferred Compute Engine zone for the secondary/failover`,
									},
								},
							},
						},
						"maintenance_window": {
							Type:     schema.TypeList,
							Optional: true,
							MaxItems: 1,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"day": {
										Type:         schema.TypeInt,
										Optional:     true,
										ValidateFunc: validation.IntBetween(1, 7),
										AtLeastOneOf: maintenanceWindowKeys,
										Description:  `Day of week (1-7), starting on Monday`,
									},
									"hour": {
										Type:         schema.TypeInt,
										Optional:     true,
										ValidateFunc: validation.IntBetween(0, 23),
										AtLeastOneOf: maintenanceWindowKeys,
										Description:  `Hour of day (0-23), ignored if day not set`,
									},
									"update_track": {
										Type:         schema.TypeString,
										Optional:     true,
										AtLeastOneOf: maintenanceWindowKeys,
										Description:  `Receive updates after one week (canary) or after two weeks (stable) or after five weeks (week5) of notification.`,
									},
								},
							},
							Description: `Declares a one-hour maintenance window when an Instance can automatically restart to apply updates. The maintenance window is specified in UTC time.`,
						},
						"pricing_plan": {
							Type:        schema.TypeString,
							Optional:    true,
							Default:     "PER_USE",
							Description: `Pricing plan for this instance, can only be PER_USE.`,
						},
						"user_labels": {
							Type:        schema.TypeMap,
							Optional:    true,
							Computed:    true,
							Elem:        &schema.Schema{Type: schema.TypeString},
							Description: `A set of key/value user label pairs to assign to the instance.`,
						},
						"insights_config": {
							Type:     schema.TypeList,
							Optional: true,
							Computed: true,
							MaxItems: 1,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"query_insights_enabled": {
										Type:         schema.TypeBool,
										Optional:     true,
										AtLeastOneOf: insightsConfigKeys,
										Description:  `True if Query Insights feature is enabled.`,
									},
									"query_string_length": {
										Type:         schema.TypeInt,
										Optional:     true,
										Default:      1024,
										ValidateFunc: validation.IntBetween(1, 1048576),
										AtLeastOneOf: insightsConfigKeys,
										Description:  `Maximum query length stored in bytes. Between 256 and 4500. Default to 1024. For Enterprise Plus instances, from 1 to 1048576.`,
									},
									"record_application_tags": {
										Type:         schema.TypeBool,
										Optional:     true,
										AtLeastOneOf: insightsConfigKeys,
										Description:  `True if Query Insights will record application tags from query when enabled.`,
									},
									"record_client_address": {
										Type:         schema.TypeBool,
										Optional:     true,
										AtLeastOneOf: insightsConfigKeys,
										Description:  `True if Query Insights will record client address when enabled.`,
									},
									"query_plans_per_minute": {
										Type:         schema.TypeInt,
										Optional:     true,
										Computed:     true,
										ValidateFunc: validation.IntBetween(0, 200),
										AtLeastOneOf: insightsConfigKeys,
										Description:  `Number of query execution plans captured by Insights per minute for all queries combined. Between 0 and 20. Default to 5. For Enterprise Plus instances, from 0 to 200.`,
									},
								},
							},
							Description: `Configuration of Query Insights.`,
						},
						"password_validation_policy": {
							Type:     schema.TypeList,
							Optional: true,
							MaxItems: 1,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"min_length": {
										Type:         schema.TypeInt,
										Optional:     true,
										ValidateFunc: validation.IntBetween(0, 2147483647),
										Description:  `Minimum number of characters allowed.`,
									},
									"complexity": {
										Type:         schema.TypeString,
										Optional:     true,
										ValidateFunc: validation.StringInSlice([]string{"COMPLEXITY_DEFAULT", "COMPLEXITY_UNSPECIFIED"}, false),
										Description:  `Password complexity.`,
									},
									"reuse_interval": {
										Type:         schema.TypeInt,
										Optional:     true,
										ValidateFunc: validation.IntBetween(0, 2147483647),
										Description:  `Number of previous passwords that cannot be reused.`,
									},
									"disallow_username_substring": {
										Type:        schema.TypeBool,
										Optional:    true,
										Description: `Disallow username as a part of the password.`,
									},
									"password_change_interval": {
										Type:        schema.TypeString,
										Optional:    true,
										Description: `Minimum interval after which the password can be changed. This flag is only supported for PostgresSQL.`,
									},
									"enable_password_policy": {
										Type:        schema.TypeBool,
										Required:    true,
										Description: `Whether the password policy is enabled or not.`,
									},
								},
							},
						},
						"connector_enforcement": {
							Type:         schema.TypeString,
							Optional:     true,
							Computed:     true,
							ValidateFunc: validation.StringInSlice([]string{"NOT_REQUIRED", "REQUIRED"}, false),
							Description:  `Enables the enforcement of Cloud SQL Auth Proxy or Cloud SQL connectors for all the connections. If enabled, all the direct connections are rejected.`,
						},
						"deletion_protection_enabled": {
							Type:        schema.TypeBool,
							Optional:    true,
							Description: `Configuration to protect against accidental instance deletion.`,
						},
						"retain_backups_on_delete": {
							Type:        schema.TypeBool,
							Optional:    true,
							Description: `When this parameter is set to true, Cloud SQL retains backups of the instance even after the instance is deleted. The ON_DEMAND backup will be retained until customer deletes the backup or the project. The AUTOMATED backup will be retained based on the backups retention setting.`,
						},
						"final_backup_config": {
							Type:     schema.TypeList,
							Optional: true,
							MaxItems: 1,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"enabled": {
										Type:        schema.TypeBool,
										Optional:    true,
										Description: `When this parameter is set to true, the final backup is enabled for the instance`,
									},
									"retention_days": {
										Type:         schema.TypeInt,
										Optional:     true,
										ValidateFunc: validation.IntBetween(1, 36135),
										Description:  `The number of days to retain the final backup after the instance deletion. The valid range is between 1 and 365. For instances managed by BackupDR, the valid range is between 1 day and 99 years. The final backup will be purged at (time_of_instance_deletion + retention_days).`,
									},
								},
							},
							Description: `Config used to determine the final backup settings for the instance`,
						},
						"read_pool_auto_scale_config": {
							Type:     schema.TypeList,
							Optional: true,
							Computed: true,
							MaxItems: 1,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"enabled": {
										Type:         schema.TypeBool,
										Optional:     true,
										Default:      false,
										AtLeastOneOf: readPoolAutoScaleConfigKeys,
										Description:  `True if Read Pool Auto Scale is enabled.`,
									},
									"max_node_count": {
										Type:         schema.TypeInt,
										Optional:     true,
										ValidateFunc: validation.IntBetween(1, 20),
										AtLeastOneOf: readPoolAutoScaleConfigKeys,
										Description:  `Maximum number of nodes in the read pool. If set to lower than current node count, node count will be updated.`,
									},
									"min_node_count": {
										Type:         schema.TypeInt,
										Optional:     true,
										ValidateFunc: validation.IntBetween(1, 20),
										AtLeastOneOf: readPoolAutoScaleConfigKeys,
										Description:  `Minimum number of nodes in the read pool. If set to higher than current node count, node count will be updated.`,
									},
									"target_metrics": {
										Type:         schema.TypeSet,
										Optional:     true,
										AtLeastOneOf: readPoolAutoScaleConfigKeys,
										Description:  `Target metrics for Read Pool Auto Scale.`,
										Elem: &schema.Resource{
											Schema: map[string]*schema.Schema{
												"metric": {
													Type:        schema.TypeString,
													Optional:    true,
													Description: `Metric name for Read Pool Auto Scale.`,
												},
												"target_value": {
													Type:        schema.TypeFloat,
													Optional:    true,
													Description: `Target value for Read Pool Auto Scale.`,
												},
											},
										},
									},
									"disable_scale_in": {
										Type:         schema.TypeBool,
										Optional:     true,
										AtLeastOneOf: readPoolAutoScaleConfigKeys,
										Description:  `True if auto scale in is disabled.`,
									},
									"scale_in_cooldown_seconds": {
										Type:         schema.TypeInt,
										Optional:     true,
										AtLeastOneOf: readPoolAutoScaleConfigKeys,
										Description:  `The cooldown period for scale in operations.`,
									},
									"scale_out_cooldown_seconds": {
										Type:         schema.TypeInt,
										Optional:     true,
										AtLeastOneOf: readPoolAutoScaleConfigKeys,
										Description:  `The cooldown period for scale out operations.`,
									},
								},
							},
							Description: `Configuration of Read Pool Auto Scale.`,
						},
					},
				},
				Description: `The settings to use for the database. The configuration is detailed below.`,
			},

			"connection_name": {
				Type:        schema.TypeString,
				Computed:    true,
				Description: `The connection name of the instance to be used in connection strings. For example, when connecting with Cloud SQL Proxy.`,
			},
			"maintenance_version": {
				Type:             schema.TypeString,
				Computed:         true,
				Optional:         true,
				Description:      `Maintenance version.`,
				DiffSuppressFunc: maintenanceVersionDiffSuppress,
			},
			"available_maintenance_versions": {
				Type:     schema.TypeList,
				Computed: true,
				Elem: &schema.Schema{
					Type: schema.TypeString,
				},
				Description: `Available Maintenance versions.`,
			},
			"database_version": {
				Type:             schema.TypeString,
				Required:         true,
				Description:      `The MySQL, PostgreSQL or SQL Server (beta) version to use. Supported values include MYSQL_5_6, MYSQL_5_7, MYSQL_8_0, MYSQL_8_4, POSTGRES_9_6, POSTGRES_10, POSTGRES_11, POSTGRES_12, POSTGRES_13, POSTGRES_14, POSTGRES_15, POSTGRES_16, POSTGRES_17, POSTGRES_18, POSTGRES_19,SQLSERVER_2017_STANDARD, SQLSERVER_2017_ENTERPRISE, SQLSERVER_2017_EXPRESS, SQLSERVER_2017_WEB. Database Version Policies includes an up-to-date reference of supported versions.`,
				DiffSuppressFunc: databaseVersionDiffSuppress,
			},
			"encryption_key_name": {
				Type:     schema.TypeString,
				Optional: true,
				Computed: true,
				ForceNew: true,
			},
			"root_password": {
				Type:          schema.TypeString,
				Optional:      true,
				Sensitive:     true,
				Description:   `Initial root password. Required for MS SQL Server.`,
				ConflictsWith: []string{"root_password_wo"},
			},
			"root_password_wo": {
				Type:     schema.TypeString,
				Optional: true,
				Description: `Initial root password. Required for MS SQL Server.
				Note: This property is write-only and will not be read from the API. For more info see [updating write-only arguments](/docs/providers/google/guides/using_write_only_arguments.html#updating-write-only-arguments)`,
				WriteOnly:     true,
				ConflictsWith: []string{"root_password"},
				RequiredWith:  []string{"root_password_wo_version"},
			},
			"root_password_wo_version": {
				Type:         schema.TypeString,
				Optional:     true,
				Description:  `Triggers update of root_password_wo write-only. For more info see [updating write-only arguments](/docs/providers/google/guides/using_write_only_arguments.html#updating-write-only-arguments)`,
				RequiredWith: []string{"root_password_wo"},
			},
			"ip_address": {
				Type:     schema.TypeList,
				Computed: true,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"ip_address": {
							Type:     schema.TypeString,
							Computed: true,
						},
						"type": {
							Type:     schema.TypeString,
							Computed: true,
						},
						"time_to_retire": {
							Type:     schema.TypeString,
							Computed: true,
						},
					},
				},
			},

			"first_ip_address": {
				Type:        schema.TypeString,
				Computed:    true,
				Description: `The first IPv4 address of any type assigned. This is to support accessing the first address in the list in a terraform output when the resource is configured with a count.`,
			},

			"public_ip_address": {
				Type:        schema.TypeString,
				Computed:    true,
				Description: `IPv4 address assigned. This is a workaround for an issue fixed in Terraform 0.12 but also provides a convenient way to access an IP of a specific type without performing filtering in a Terraform config.`,
			},

			"private_ip_address": {
				Type:        schema.TypeString,
				Computed:    true,
				Description: `IPv4 address assigned. This is a workaround for an issue fixed in Terraform 0.12 but also provides a convenient way to access an IP of a specific type without performing filtering in a Terraform config.`,
			},

			"name": {
				Type:        schema.TypeString,
				Optional:    true,
				Computed:    true,
				ForceNew:    true,
				Description: `The name of the instance. If the name is left blank, Terraform will randomly generate one when the instance is first created. This is done because after a name is used, it cannot be reused for up to one week.`,
			},

			"master_instance_name": {
				Type:        schema.TypeString,
				Optional:    true,
				Computed:    true,
				Description: `The name of the instance that will act as the master in the replication setup. Note, this requires the master to have binary_log_enabled set, as well as existing backups.`,
			},

			"project": {
				Type:        schema.TypeString,
				Optional:    true,
				Computed:    true,
				ForceNew:    true,
				Description: `The ID of the project in which the resource belongs. If it is not provided, the provider project is used.`,
			},

			"instance_type": {
				Type:        schema.TypeString,
				Computed:    true,
				Optional:    true,
				Description: `The type of the instance. See https://cloud.google.com/sql/docs/mysql/admin-api/rest/v1/instances#SqlInstanceType for supported values.`,
			},

			"node_count": {
				Type:        schema.TypeInt,
				Computed:    true,
				Optional:    true,
				Description: `For a read pool instance, the number of nodes in the read pool. For read pools with auto scaling enabled, this field is read only.`,
			},
			"replica_configuration": {
				Type:     schema.TypeList,
				Optional: true,
				MaxItems: 1,
				// Returned from API on all replicas
				Computed:  true,
				Sensitive: true,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"ca_certificate": {
							Type:         schema.TypeString,
							Optional:     true,
							ForceNew:     true,
							AtLeastOneOf: replicaConfigurationKeys,
							Description:  `PEM representation of the trusted CA's x509 certificate.`,
						},
						"cascadable_replica": {
							Type:         schema.TypeBool,
							Optional:     true,
							AtLeastOneOf: replicaConfigurationKeys,
							Description:  `Specifies if a SQL Server replica is a cascadable replica. A cascadable replica is a SQL Server cross region replica that supports replica(s) under it.`,
						},
						"client_certificate": {
							Type:         schema.TypeString,
							Optional:     true,
							ForceNew:     true,
							AtLeastOneOf: replicaConfigurationKeys,
							Description:  `PEM representation of the replica's x509 certificate.`,
						},
						"client_key": {
							Type:         schema.TypeString,
							Optional:     true,
							ForceNew:     true,
							AtLeastOneOf: replicaConfigurationKeys,
							Description:  `PEM representation of the replica's private key. The corresponding public key in encoded in the client_certificate.`,
						},
						"connect_retry_interval": {
							Type:         schema.TypeInt,
							Optional:     true,
							ForceNew:     true,
							AtLeastOneOf: replicaConfigurationKeys,
							Description:  `The number of seconds between connect retries. MySQL's default is 60 seconds.`,
						},
						"dump_file_path": {
							Type:         schema.TypeString,
							Optional:     true,
							ForceNew:     true,
							AtLeastOneOf: replicaConfigurationKeys,
							Description:  `Path to a SQL file in Google Cloud Storage from which replica instances are created. Format is gs://bucket/filename.`,
						},
						"failover_target": {
							Type:         schema.TypeBool,
							Optional:     true,
							ForceNew:     true,
							AtLeastOneOf: replicaConfigurationKeys,
							Description:  `Specifies if the replica is the failover target. If the field is set to true the replica will be designated as a failover replica. If the master instance fails, the replica instance will be promoted as the new master instance. Not supported for Postgres`,
						},
						"master_heartbeat_period": {
							Type:         schema.TypeInt,
							Optional:     true,
							ForceNew:     true,
							AtLeastOneOf: replicaConfigurationKeys,
							Description:  `Time in ms between replication heartbeats.`,
						},
						"password": {
							Type:         schema.TypeString,
							Optional:     true,
							ForceNew:     true,
							Sensitive:    true,
							AtLeastOneOf: replicaConfigurationKeys,
							Description:  `Password for the replication connection.`,
						},
						"ssl_cipher": {
							Type:         schema.TypeString,
							Optional:     true,
							ForceNew:     true,
							AtLeastOneOf: replicaConfigurationKeys,
							Description:  `Permissible ciphers for use in SSL encryption.`,
						},
						"username": {
							Type:         schema.TypeString,
							Optional:     true,
							ForceNew:     true,
							AtLeastOneOf: replicaConfigurationKeys,
							Description:  `Username for replication connection.`,
						},
						"verify_server_certificate": {
							Type:         schema.TypeBool,
							Optional:     true,
							ForceNew:     true,
							AtLeastOneOf: replicaConfigurationKeys,
							Description:  `True if the master's common name value is checked during the SSL handshake.`,
						},
					},
				},
				Description: `The configuration for replication.`,
			},
			"replica_names": {
				Type:     schema.TypeList,
				Computed: true,
				Optional: true,
				Elem: &schema.Schema{
					Type: schema.TypeString,
				},
				Description: `The replicas of the instance.`,
			},
			"replication_cluster": {
				Type:     schema.TypeList,
				Computed: true,
				Optional: true,
				MaxItems: 1,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"psa_write_endpoint": {
							Type:        schema.TypeString,
							Optional:    true,
							Computed:    true,
							Description: fmt.Sprintf(`If set, this field indicates this instance has a private service access (PSA) DNS endpoint that is pointing to the primary instance of the cluster. If this instance is the primary, then the DNS endpoint points to this instance. After a switchover or replica failover operation, this DNS endpoint points to the promoted instance. This is a read-only field, returned to the user as information. This field can exist even if a standalone instance doesn't have a DR replica yet or the DR replica is deleted.`),
						},
						"failover_dr_replica_name": {
							Type:             schema.TypeString,
							Optional:         true,
							Description:      fmt.Sprintf(`If the instance is a primary instance, then this field identifies the disaster recovery (DR) replica. The standard format of this field is "your-project:your-instance". You can also set this field to "your-instance", but cloud SQL backend will convert it to the aforementioned standard format.`),
							DiffSuppressFunc: areDRReplicaNamesEqual,
						},
						"dr_replica": {
							Type:        schema.TypeBool,
							Computed:    true,
							Description: `Read-only field that indicates whether the replica is a DR replica.`,
						},
					},
				},
				Description: "A primary instance and disaster recovery replica pair. Applicable to MySQL and PostgreSQL. This field can be set if the primary has psa_write_endpoint set or both the primary and replica are created.",
			},
			"server_ca_cert": {
				Type:      schema.TypeList,
				Computed:  true,
				Sensitive: true,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"cert": {
							Type:        schema.TypeString,
							Computed:    true,
							Description: `The CA Certificate used to connect to the SQL Instance via SSL.`,
						},
						"common_name": {
							Type:        schema.TypeString,
							Computed:    true,
							Description: `The CN valid for the CA Cert.`,
						},
						"create_time": {
							Type:        schema.TypeString,
							Computed:    true,
							Description: `Creation time of the CA Cert.`,
						},
						"expiration_time": {
							Type:        schema.TypeString,
							Computed:    true,
							Description: `Expiration time of the CA Cert.`,
						},
						"sha1_fingerprint": {
							Type:        schema.TypeString,
							Computed:    true,
							Description: `SHA Fingerprint of the CA Cert.`,
						},
					},
				},
			},
			"service_account_email_address": {
				Type:        schema.TypeString,
				Computed:    true,
				Description: `The service account email address assigned to the instance.`,
			},
			"self_link": {
				Type:        schema.TypeString,
				Computed:    true,
				Description: `The URI of the created resource.`,
			},
			"psc_service_attachment_link": {
				Type:        schema.TypeString,
				Computed:    true,
				Description: `The link to service attachment of PSC instance.`,
			},
			"dns_name": {
				Type:        schema.TypeString,
				Computed:    true,
				Description: `The instance-level dns name of the instance for PSC instances or public IP CAS instances.`,
			},
			"dns_names": {
				Type:     schema.TypeList,
				Computed: true,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"name": {
							Type:     schema.TypeString,
							Computed: true,
						},
						"connection_type": {
							Type:     schema.TypeString,
							Computed: true,
						},
						"dns_scope": {
							Type:     schema.TypeString,
							Computed: true,
						},
					},
				},
				Description: `The list of DNS names used by this instance. Different connection types for an instance may have different DNS names. DNS names can apply to an individual instance or a cluster of instances.`,
			},
			"restore_backup_context": {
				Type:     schema.TypeList,
				Optional: true,
				MaxItems: 1,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"backup_run_id": {
							Type:        schema.TypeInt,
							Required:    true,
							Description: `The ID of the backup run to restore from.`,
						},
						"instance_id": {
							Type:        schema.TypeString,
							Optional:    true,
							Description: `The ID of the instance that the backup was taken from.`,
						},
						"project": {
							Type:        schema.TypeString,
							Optional:    true,
							Description: `The full project ID of the source instance.`,
						},
					},
				},
			},
			"backupdr_backup": {
				Type:        schema.TypeString,
				Optional:    true,
				Description: `The name of the BackupDR backup to restore from.`,
			},
			"clone": {
				Type:         schema.TypeList,
				Optional:     true,
				Computed:     false,
				AtLeastOneOf: []string{"settings", "clone", "point_in_time_restore_context"},
				Description:  `Configuration for creating a new instance as a clone of another instance.`,
				MaxItems:     1,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"source_instance_name": {
							Type:        schema.TypeString,
							Required:    true,
							Description: `The name of the instance from which the point in time should be restored.`,
						},
						"point_in_time": {
							Type:             schema.TypeString,
							Optional:         true,
							DiffSuppressFunc: tpgresource.TimestampDiffSuppress(time.RFC3339Nano),
							Description:      `The timestamp of the point in time that should be restored.`,
						},
						"preferred_zone": {
							Type:        schema.TypeString,
							Optional:    true,
							Description: `(Point-in-time recovery for PostgreSQL only) Clone to an instance in the specified zone. If no zone is specified, clone to the same zone as the source instance.`,
						},
						"database_names": {
							Type:     schema.TypeList,
							Optional: true,
							Elem: &schema.Schema{
								Type: schema.TypeString,
							},
							Description: `(SQL Server only, use with point_in_time) clone only the specified databases from the source instance. Clone all databases if empty.`,
						},
						"allocated_ip_range": {
							Type:        schema.TypeString,
							Optional:    true,
							Description: `The name of the allocated ip range for the private ip CloudSQL instance. For example: "google-managed-services-default". If set, the cloned instance ip will be created in the allocated range. The range name must comply with [RFC 1035](https://tools.ietf.org/html/rfc1035). Specifically, the name must be 1-63 characters long and match the regular expression [a-z]([-a-z0-9]*[a-z0-9])?.`,
						},
						"source_instance_deletion_time": {
							Type:             schema.TypeString,
							Optional:         true,
							DiffSuppressFunc: tpgresource.TimestampDiffSuppress(time.RFC3339Nano),
							Description:      `The timestamp of when the source instance was deleted for a clone from a deleted instance.`,
						},
					},
				},
			},
			"point_in_time_restore_context": {
				Type:         schema.TypeList,
				Optional:     true,
				Computed:     false,
				AtLeastOneOf: []string{"settings", "clone", "point_in_time_restore_context"},
				Description:  `Configuration for creating a new instance using point-in-time-restore from backupdr backup.`,
				MaxItems:     1,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"datasource": {
							Type:        schema.TypeString,
							Required:    true,
							Description: `The Google Cloud Backup and Disaster Recovery Datasource URI. For example: "projects/my-project/locations/us-central1/datasources/my-datasource".`,
						},
						"point_in_time": {
							Type:             schema.TypeString,
							Optional:         true,
							DiffSuppressFunc: tpgresource.TimestampDiffSuppress(time.RFC3339Nano),
							Description:      `The date and time to which you want to restore the instance.`,
						},
						"preferred_zone": {
							Type:        schema.TypeString,
							Optional:    true,
							Description: `Point-in-time recovery of an instance to the specified zone. If no zone is specified, then clone to the same primary zone as the source instance.`,
						},
						"allocated_ip_range": {
							Type:        schema.TypeString,
							Optional:    true,
							Description: `The name of the allocated IP range for the internal IP Cloud SQL instance. For example: "google-managed-services-default". If you set this, then Cloud SQL creates the IP address for the cloned instance in the allocated range. This range must comply with [RFC 1035](https://tools.ietf.org/html/rfc1035) standards. Specifically, the name must be 1-63 characters long and match the regular expression [a-z]([-a-z0-9]*[a-z0-9])?.`,
						},
						"target_instance": {
							Type:        schema.TypeString,
							Optional:    true,
							Description: `The name of the target instance to restore to.`,
						},
					},
				},
			},
		},
		UseJSONNumber: true,
	}
}

// Makes private_network ForceNew if it is changing from set to nil. The API returns an error
// if this change is attempted in-place.
func privateNetworkCustomizeDiff(_ context.Context, d *schema.ResourceDiff, meta interface{}) error {
	old, new := d.GetChange("settings.0.ip_configuration.0.private_network")

	if old != "" && new == "" {
		if err := d.ForceNew("settings.0.ip_configuration.0.private_network"); err != nil {
			return err
		}
	}

	return nil
}

// helper function to see if string within list contains a particular substring
func stringContainsSlice(arr []string, str string) bool {
	for _, i := range arr {
		if strings.Contains(str, i) {
			return true
		}
	}
	return false
}

// Point in time recovery for MySQL database instances needs binary_log_enabled set to true and
// not point_in_time_recovery_enabled, which is confusing to users. This checks for
// point_in_time_recovery_enabled being set to a non-PostgreSQL and non-SQLServer database instances and suggests
// binary_log_enabled.
func pitrSupportDbCustomizeDiff(_ context.Context, diff *schema.ResourceDiff, v interface{}) error {
	pitr := diff.Get("settings.0.backup_configuration.0.point_in_time_recovery_enabled").(bool)
	dbVersion := diff.Get("database_version").(string)
	dbVersionPitrValid := []string{"POSTGRES", "SQLSERVER"}
	if pitr && !stringContainsSlice(dbVersionPitrValid, dbVersion) {
		return fmt.Errorf("point_in_time_recovery_enabled is only available for the following %v. You may want to consider using binary_log_enabled instead and remove point_in_time_recovery_enabled (removing point_in_time_recovery_enabled and adding binary_log_enabled will enable pitr for MYSQL)", dbVersionPitrValid)
	}
	return nil
}

func resourceSqlDatabaseInstanceCreate(d *schema.ResourceData, meta interface{}) error {
	config := meta.(*transport_tpg.Config)
	userAgent, err := tpgresource.GenerateUserAgentString(d, config.UserAgent)
	if err != nil {
		return err
	}

	project, err := tpgresource.GetProject(d, config)
	if err != nil {
		return err
	}

	region, err := tpgresource.GetRegion(d, config)
	if err != nil {
		return err
	}

	var name string
	if v, ok := d.GetOk("name"); ok {
		name = v.(string)
	} else {
		name = id.UniqueId()
	}

	if err := d.Set("name", name); err != nil {
		return fmt.Errorf("Error setting name: %s", err)
	}

	// SQL Instances that fail to create are expensive- see https://github.com/hashicorp/terraform-provider-google/issues/7154
	// We can fail fast to stop instance names from getting reserved.
	network := d.Get("settings.0.ip_configuration.0.private_network").(string)
	if network != "" {
		err = sqlDatabaseInstanceServiceNetworkPrecheck(d, config, userAgent, network)
		if err != nil {
			return err
		}
	}

	databaseVersion := d.Get("database_version").(string)

	instance := &sqladmin.DatabaseInstance{
		Name:                 name,
		Region:               region,
		DatabaseVersion:      databaseVersion,
		MasterInstanceName:   d.Get("master_instance_name").(string),
		ReplicaConfiguration: expandReplicaConfiguration(d.Get("replica_configuration").([]interface{})),
	}

	cloneContext, cloneSource := expandCloneContext(d.Get("clone").([]interface{}))
	pointInTimeRestoreContext := expandPointInTimeRestoreContext(d.Get("point_in_time_restore_context").([]interface{}))

	s, ok := d.GetOk("settings")
	desiredSettings := expandSqlDatabaseInstanceSettings(s.([]interface{}), databaseVersion)
	if ok {
		instance.Settings = desiredSettings
	}

	if _, ok := d.GetOk("maintenance_version"); ok {
		instance.MaintenanceVersion = d.Get("maintenance_version").(string)
	}

	if _, ok := d.GetOk("instance_type"); ok {
		instance.InstanceType = d.Get("instance_type").(string)
	}

	if _, ok := d.GetOk("node_count"); ok {
		instance.NodeCount = int64(d.Get("node_count").(int))
	}
	if _, ok := d.GetOk("root_password_wo_version"); ok {
		instance.RootPassword = tpgresource.GetRawConfigAttributeAsString(d, "root_password_wo")
	} else if _, ok := d.GetOk("root_password"); ok {
		instance.RootPassword = d.Get("root_password").(string)
	}

	// Modifying a replica during Create can cause problems if the master is
	// modified at the same time. Lock the master until we're done in order
	// to prevent that.
	if !sqlDatabaseIsMaster(d) {
		transport_tpg.MutexStore.Lock(instanceMutexKey(project, instance.MasterInstanceName))
		defer transport_tpg.MutexStore.Unlock(instanceMutexKey(project, instance.MasterInstanceName))
	}

	if k, ok := d.GetOk("encryption_key_name"); ok {
		instance.DiskEncryptionConfiguration = &sqladmin.DiskEncryptionConfiguration{
			KmsKeyName: k.(string),
		}
	}

	var patchData *sqladmin.DatabaseInstance

	// BinaryLogging can be enabled on replica instances but only after creation.
	if instance.MasterInstanceName != "" && instance.Settings != nil && instance.Settings.BackupConfiguration != nil && instance.Settings.BackupConfiguration.BinaryLogEnabled {
		settingsCopy := expandSqlDatabaseInstanceSettings(s.([]interface{}), databaseVersion)
		bc := settingsCopy.BackupConfiguration
		patchData = &sqladmin.DatabaseInstance{Settings: &sqladmin.Settings{BackupConfiguration: bc}}

		instance.Settings.BackupConfiguration.BinaryLogEnabled = false
	}

	var op *sqladmin.Operation
	err = transport_tpg.Retry(transport_tpg.RetryOptions{
		RetryFunc: func() (operr error) {
			if cloneContext != nil {
				cloneContext.DestinationInstanceName = name
				clodeReq := sqladmin.InstancesCloneRequest{CloneContext: cloneContext}
				op, operr = config.NewSqlAdminClient(userAgent).Instances.Clone(project, cloneSource, &clodeReq).Do()
			} else if pointInTimeRestoreContext != nil {
				parent := fmt.Sprintf("projects/%s", project)
				op, operr = config.NewSqlAdminClient(userAgent).Instances.PointInTimeRestore(parent, pointInTimeRestoreContext).Do()
			} else {
				op, operr = config.NewSqlAdminClient(userAgent).Instances.Insert(project, instance).Do()
			}
			return operr
		},
		Timeout:              d.Timeout(schema.TimeoutCreate),
		ErrorRetryPredicates: []transport_tpg.RetryErrorPredicateFunc{transport_tpg.IsSqlOperationInProgressError},
	})
	if err != nil {
		return fmt.Errorf("Error, failed to create instance %s: %s", instance.Name, err)
	}

	id, err := tpgresource.ReplaceVars(d, config, "projects/{{project}}/instances/{{name}}")
	if err != nil {
		return fmt.Errorf("Error constructing id: %s", err)
	}
	d.SetId(id)

	err = SqlAdminOperationWaitTime(config, op, project, "Create Instance", userAgent, d.Timeout(schema.TimeoutCreate))
	if err != nil {
		d.SetId("")
		return err
	}

	// If a default root user was created with a wildcard ('%') hostname, delete it. Note it
	// appears to only be created for certain types of databases, like MySQL.
	// Users in a replica instance are inherited from the master instance and should be left alone.
	// This deletion is done immediately after the instance is created, in order to minimize the
	// risk of it being left on the instance, which would present a security concern.
	if sqlDatabaseIsMaster(d) {
		var users *sqladmin.UsersListResponse
		err = transport_tpg.Retry(transport_tpg.RetryOptions{
			RetryFunc: func() error {
				users, err = config.NewSqlAdminClient(userAgent).Users.List(project, instance.Name).Do()
				return err
			},
			Timeout:              d.Timeout(schema.TimeoutRead),
			ErrorRetryPredicates: []transport_tpg.RetryErrorPredicateFunc{transport_tpg.IsSqlOperationInProgressError},
		})
		if err != nil {
			return fmt.Errorf("Error, attempting to list users associated with instance %s: %s", instance.Name, err)
		}
		for _, u := range users.Items {
			if u.Name == "root" && u.Host == "%" {
				err = transport_tpg.Retry(transport_tpg.RetryOptions{
					RetryFunc: func() error {
						op, err = config.NewSqlAdminClient(userAgent).Users.Delete(project, instance.Name).Host(u.Host).Name(u.Name).Do()
						if err == nil {
							err = SqlAdminOperationWaitTime(config, op, project, "Delete default root User", userAgent, d.Timeout(schema.TimeoutCreate))
						}
						return err
					},
				})
				if err != nil {
					return fmt.Errorf("Error, failed to delete default 'root'@'*' u, but the database was created successfully: %s", err)
				}
			}
		}
	}

	// patch any fields that need to be sent postcreation
	if patchData != nil {
		err = transport_tpg.Retry(transport_tpg.RetryOptions{
			RetryFunc: func() (rerr error) {
				op, rerr = config.NewSqlAdminClient(userAgent).Instances.Patch(project, instance.Name, patchData).Do()
				return rerr
			},
			Timeout:              d.Timeout(schema.TimeoutUpdate),
			ErrorRetryPredicates: []transport_tpg.RetryErrorPredicateFunc{transport_tpg.IsSqlOperationInProgressError},
		})
		if err != nil {
			return fmt.Errorf("Error, failed to update instance settings for %s: %s", instance.Name, err)
		}
		err = SqlAdminOperationWaitTime(config, op, project, "Patch Instance", userAgent, d.Timeout(schema.TimeoutUpdate))
		if err != nil {
			return err
		}
	}

	err = resourceSqlDatabaseInstanceRead(d, meta)
	if err != nil {
		return err
	}

	// Refresh settings from read as they may have defaulted from the API
	s = d.Get("settings")
	// If we've created an instance as a clone or pitr, we need to update it to set any user defined settings
	if len(s.([]interface{})) != 0 && (cloneContext != nil || pointInTimeRestoreContext != nil) && desiredSettings != nil {
		instanceUpdate := &sqladmin.DatabaseInstance{
			Settings: desiredSettings,
		}
		_settings := s.([]interface{})[0].(map[string]interface{})
		instanceUpdate.Settings.SettingsVersion = int64(_settings["version"].(int))
		var op *sqladmin.Operation
		err = transport_tpg.Retry(transport_tpg.RetryOptions{
			RetryFunc: func() (rerr error) {
				op, rerr = config.NewSqlAdminClient(userAgent).Instances.Update(project, name, instanceUpdate).Do()
				return rerr
			},
			Timeout:              d.Timeout(schema.TimeoutUpdate),
			ErrorRetryPredicates: []transport_tpg.RetryErrorPredicateFunc{transport_tpg.IsSqlOperationInProgressError},
		})
		if err != nil {
			return fmt.Errorf("Error, failed to update instance settings for %s: %s", instance.Name, err)
		}

		err = SqlAdminOperationWaitTime(config, op, project, "Update Instance", userAgent, d.Timeout(schema.TimeoutUpdate))
		if err != nil {
			return err
		}

		// Refresh the state of the instance after updating the settings
		err = resourceSqlDatabaseInstanceRead(d, meta)
		if err != nil {
			return err
		}
	}

	// Perform a backup restore if the backup context exists
	if r, ok := d.GetOk("restore_backup_context"); ok {
		log.Printf("[DEBUG] Restoring instance %s from backup context: %v", name, r)
		err = sqlDatabaseInstanceRestoreFromBackup(d, config, userAgent, project, name, r, "")
		if err != nil {
			return err
		}
	} else if b, ok := d.GetOk("backupdr_backup"); ok && b.(string) != "" {
		log.Printf("[DEBUG] Restoring instance %s from BackupDR backup: %s", name, b.(string))
		err = sqlDatabaseInstanceRestoreFromBackup(d, config, userAgent, project, name, nil, b)
		if err != nil {
			return err
		}
	}

	return nil
}

// Available fields for settings vary between database versions.
func expandSqlDatabaseInstanceSettings(configured []interface{}, databaseVersion string) *sqladmin.Settings {
	if len(configured) == 0 || configured[0] == nil {
		return nil
	}

	_settings := configured[0].(map[string]interface{})
	settings := &sqladmin.Settings{
		// Version is unset in Create but is set during update
		SettingsVersion:           int64(_settings["version"].(int)),
		DataCacheConfig:           expandDataCacheConfig(_settings["data_cache_config"].([]interface{})),
		Tier:                      _settings["tier"].(string),
		Edition:                   _settings["edition"].(string),
		AdvancedMachineFeatures:   expandSqlServerAdvancedMachineFeatures(_settings["advanced_machine_features"].([]interface{})),
		ForceSendFields:           []string{"StorageAutoResize", "EnableGoogleMlIntegration", "EnableDataplexIntegration", "RetainBackupsOnDelete"},
		ActivationPolicy:          _settings["activation_policy"].(string),
		ActiveDirectoryConfig:     expandActiveDirectoryConfig(_settings["active_directory_config"].([]interface{})),
		DenyMaintenancePeriods:    expandDenyMaintenancePeriod(_settings["deny_maintenance_period"].([]interface{})),
		SqlServerAuditConfig:      expandSqlServerAuditConfig(_settings["sql_server_audit_config"].([]interface{})),
		TimeZone:                  _settings["time_zone"].(string),
		AvailabilityType:          _settings["availability_type"].(string),
		ConnectorEnforcement:      _settings["connector_enforcement"].(string),
		Collation:                 _settings["collation"].(string),
		DataDiskSizeGb:            int64(_settings["disk_size"].(int)),
		DataDiskType:              _settings["disk_type"].(string),
		PricingPlan:               _settings["pricing_plan"].(string),
		DeletionProtectionEnabled: _settings["deletion_protection_enabled"].(bool),
		EnableGoogleMlIntegration: _settings["enable_google_ml_integration"].(bool),
		EnableDataplexIntegration: _settings["enable_dataplex_integration"].(bool),
		RetainBackupsOnDelete:     _settings["retain_backups_on_delete"].(bool),
		FinalBackupConfig:         expandFinalBackupConfig(_settings["final_backup_config"].([]interface{})),
		UserLabels:                tpgresource.ConvertStringMap(_settings["user_labels"].(map[string]interface{})),
		BackupConfiguration:       expandBackupConfiguration(_settings["backup_configuration"].([]interface{})),
		DatabaseFlags:             expandDatabaseFlags(_settings["database_flags"].(*schema.Set).List()),
		IpConfiguration:           expandIpConfiguration(_settings["ip_configuration"].([]interface{}), databaseVersion),
		LocationPreference:        expandLocationPreference(_settings["location_preference"].([]interface{})),
		MaintenanceWindow:         expandMaintenanceWindow(_settings["maintenance_window"].([]interface{})),
		InsightsConfig:            expandInsightsConfig(_settings["insights_config"].([]interface{})),
		PasswordValidationPolicy:  expandPasswordValidationPolicy(_settings["password_validation_policy"].([]interface{})),
		ConnectionPoolConfig:      expandConnectionPoolConfig(_settings["connection_pool_config"].(*schema.Set).List()),
		ReadPoolAutoScaleConfig:   expandReadPoolAutoScaleConfig(_settings["read_pool_auto_scale_config"].([]interface{})),
	}

	resize := _settings["disk_autoresize"].(bool)
	settings.StorageAutoResize = &resize
	settings.StorageAutoResizeLimit = int64(_settings["disk_autoresize_limit"].(int))

	return settings
}

func expandReplicaConfiguration(configured []interface{}) *sqladmin.ReplicaConfiguration {
	if len(configured) == 0 || configured[0] == nil {
		return nil
	}

	_replicaConfiguration := configured[0].(map[string]interface{})
	return &sqladmin.ReplicaConfiguration{
		CascadableReplica: _replicaConfiguration["cascadable_replica"].(bool),
		FailoverTarget:    _replicaConfiguration["failover_target"].(bool),

		// MysqlReplicaConfiguration has been flattened in the TF schema, so
		// we'll keep it flat here instead of another expand method.
		MysqlReplicaConfiguration: &sqladmin.MySqlReplicaConfiguration{
			CaCertificate:           _replicaConfiguration["ca_certificate"].(string),
			ClientCertificate:       _replicaConfiguration["client_certificate"].(string),
			ClientKey:               _replicaConfiguration["client_key"].(string),
			ConnectRetryInterval:    int64(_replicaConfiguration["connect_retry_interval"].(int)),
			DumpFilePath:            _replicaConfiguration["dump_file_path"].(string),
			MasterHeartbeatPeriod:   int64(_replicaConfiguration["master_heartbeat_period"].(int)),
			Password:                _replicaConfiguration["password"].(string),
			SslCipher:               _replicaConfiguration["ssl_cipher"].(string),
			Username:                _replicaConfiguration["username"].(string),
			VerifyServerCertificate: _replicaConfiguration["verify_server_certificate"].(bool),
		},
	}
}

func expandCloneContext(configured []interface{}) (*sqladmin.CloneContext, string) {
	if len(configured) == 0 || configured[0] == nil {
		return nil, ""
	}

	_cloneConfiguration := configured[0].(map[string]interface{})

	databaseNames := []string{}
	rawDatabaseNames := _cloneConfiguration["database_names"].([]interface{})
	for _, db := range rawDatabaseNames {
		databaseNames = append(databaseNames, db.(string))
	}

	return &sqladmin.CloneContext{
		PointInTime:                _cloneConfiguration["point_in_time"].(string),
		PreferredZone:              _cloneConfiguration["preferred_zone"].(string),
		DatabaseNames:              databaseNames,
		AllocatedIpRange:           _cloneConfiguration["allocated_ip_range"].(string),
		SourceInstanceDeletionTime: _cloneConfiguration["source_instance_deletion_time"].(string),
	}, _cloneConfiguration["source_instance_name"].(string)
}

func expandMaintenanceWindow(configured []interface{}) *sqladmin.MaintenanceWindow {
	if len(configured) == 0 || configured[0] == nil {
		return nil
	}

	window := configured[0].(map[string]interface{})
	return &sqladmin.MaintenanceWindow{
		Day:             int64(window["day"].(int)),
		Hour:            int64(window["hour"].(int)),
		UpdateTrack:     window["update_track"].(string),
		ForceSendFields: []string{"Hour"},
	}
}

func expandLocationPreference(configured []interface{}) *sqladmin.LocationPreference {
	if len(configured) == 0 || configured[0] == nil {
		return nil
	}

	_locationPreference := configured[0].(map[string]interface{})
	return &sqladmin.LocationPreference{
		FollowGaeApplication: _locationPreference["follow_gae_application"].(string),
		Zone:                 _locationPreference["zone"].(string),
		SecondaryZone:        _locationPreference["secondary_zone"].(string),
	}
}

func expandIpConfiguration(configured []interface{}, databaseVersion string) *sqladmin.IpConfiguration {
	if len(configured) == 0 || configured[0] == nil {
		return nil
	}

	_ipConfiguration := configured[0].(map[string]interface{})

	forceSendFields := []string{"Ipv4Enabled"}
	nullFields := []string{"RequireSsl"}

	if !strings.HasPrefix(databaseVersion, "SQLSERVER") {
		forceSendFields = append(forceSendFields, "EnablePrivatePathForGoogleCloudServices")
	}

	return &sqladmin.IpConfiguration{
		Ipv4Enabled:                             _ipConfiguration["ipv4_enabled"].(bool),
		PrivateNetwork:                          _ipConfiguration["private_network"].(string),
		AllocatedIpRange:                        _ipConfiguration["allocated_ip_range"].(string),
		AuthorizedNetworks:                      expandAuthorizedNetworks(_ipConfiguration["authorized_networks"].(*schema.Set).List()),
		EnablePrivatePathForGoogleCloudServices: _ipConfiguration["enable_private_path_for_google_cloud_services"].(bool),
		ForceSendFields:                         forceSendFields,
		NullFields:                              nullFields,
		PscConfig:                               expandPscConfig(_ipConfiguration["psc_config"].(*schema.Set).List()),
		SslMode:                                 _ipConfiguration["ssl_mode"].(string),
		ServerCaMode:                            _ipConfiguration["server_ca_mode"].(string),
		ServerCaPool:                            _ipConfiguration["server_ca_pool"].(string),
		CustomSubjectAlternativeNames:           tpgresource.ConvertStringArr(_ipConfiguration["custom_subject_alternative_names"].(*schema.Set).List()),
	}
}

func expandPscAutoConnectionConfig(configured []interface{}) []*sqladmin.PscAutoConnectionConfig {
	pscAutoConnections := make([]*sqladmin.PscAutoConnectionConfig, 0, len(configured))

	for _, _flag := range configured {
		if _flag == nil {
			continue
		}
		_entry := _flag.(map[string]interface{})

		pscAutoConnections = append(pscAutoConnections, &sqladmin.PscAutoConnectionConfig{
			ConsumerNetwork: _entry["consumer_network"].(string),
			ConsumerProject: _entry["consumer_service_project_id"].(string),
		})
	}
	return pscAutoConnections
}

func expandPscConfig(configured []interface{}) *sqladmin.PscConfig {
	for _, _pscConfig := range configured {
		_entry := _pscConfig.(map[string]interface{})
		return &sqladmin.PscConfig{
			PscEnabled:              _entry["psc_enabled"].(bool),
			AllowedConsumerProjects: tpgresource.ConvertStringArr(_entry["allowed_consumer_projects"].(*schema.Set).List()),
			NetworkAttachmentUri:    _entry["network_attachment_uri"].(string),
			PscAutoConnections:      expandPscAutoConnectionConfig(_entry["psc_auto_connections"].([]interface{})),
		}
	}

	return nil
}

func expandFlags(configured []interface{}) []*sqladmin.ConnectionPoolFlags {
	connectionPoolFlags := make([]*sqladmin.ConnectionPoolFlags, 0, len(configured))
	for _, _flag := range configured {
		if _flag == nil {
			continue
		}
		_entry := _flag.(map[string]interface{})

		connectionPoolFlags = append(connectionPoolFlags, &sqladmin.ConnectionPoolFlags{
			Name:  _entry["name"].(string),
			Value: _entry["value"].(string),
		})
	}
	return connectionPoolFlags
}

func expandConnectionPoolConfig(configured []interface{}) *sqladmin.ConnectionPoolConfig {
	if len(configured) == 0 || configured[0] == nil {
		return nil
	}

	_connectionPoolConfig := configured[0].(map[string]interface{})

	return &sqladmin.ConnectionPoolConfig{
		ConnectionPoolingEnabled: _connectionPoolConfig["connection_pooling_enabled"].(bool),
		Flags:                    expandFlags(_connectionPoolConfig["flags"].(*schema.Set).List()),
	}
}

func expandAuthorizedNetworks(configured []interface{}) []*sqladmin.AclEntry {
	an := make([]*sqladmin.AclEntry, 0, len(configured))
	for _, _acl := range configured {
		_entry := _acl.(map[string]interface{})
		an = append(an, &sqladmin.AclEntry{
			ExpirationTime: _entry["expiration_time"].(string),
			Name:           _entry["name"].(string),
			Value:          _entry["value"].(string),
		})
	}

	return an
}

func expandDatabaseFlags(configured []interface{}) []*sqladmin.DatabaseFlags {
	databaseFlags := make([]*sqladmin.DatabaseFlags, 0, len(configured))
	for _, _flag := range configured {
		if _flag == nil {
			continue
		}
		_entry := _flag.(map[string]interface{})

		databaseFlags = append(databaseFlags, &sqladmin.DatabaseFlags{
			Name:  _entry["name"].(string),
			Value: _entry["value"].(string),
		})
	}
	return databaseFlags
}

func expandDataCacheConfig(configured interface{}) *sqladmin.DataCacheConfig {
	l := configured.([]interface{})
	if len(l) == 0 {
		return nil
	}
	config := l[0].(map[string]interface{})
	return &sqladmin.DataCacheConfig{
		DataCacheEnabled: config["data_cache_enabled"].(bool),
	}
}

func expandBackupConfiguration(configured []interface{}) *sqladmin.BackupConfiguration {
	if len(configured) == 0 || configured[0] == nil {
		return nil
	}

	_backupConfiguration := configured[0].(map[string]interface{})
	return &sqladmin.BackupConfiguration{
		BinaryLogEnabled:            _backupConfiguration["binary_log_enabled"].(bool),
		BackupRetentionSettings:     expandBackupRetentionSettings(_backupConfiguration["backup_retention_settings"]),
		Enabled:                     _backupConfiguration["enabled"].(bool),
		StartTime:                   _backupConfiguration["start_time"].(string),
		Location:                    _backupConfiguration["location"].(string),
		TransactionLogRetentionDays: int64(_backupConfiguration["transaction_log_retention_days"].(int)),
		PointInTimeRecoveryEnabled:  _backupConfiguration["point_in_time_recovery_enabled"].(bool),
		BackupTier:                  _backupConfiguration["backup_tier"].(string),
		ForceSendFields:             []string{"BinaryLogEnabled", "Enabled", "PointInTimeRecoveryEnabled"},
	}
}

func expandFinalBackupConfig(configured []interface{}) *sqladmin.FinalBackupConfig {
	if len(configured) == 0 || configured[0] == nil {
		return nil
	}

	_finalBackupConfig := configured[0].(map[string]interface{})
	return &sqladmin.FinalBackupConfig{
		Enabled:         _finalBackupConfig["enabled"].(bool),
		RetentionDays:   int64(_finalBackupConfig["retention_days"].(int)),
		ForceSendFields: []string{"Enabled"},
	}
}

func expandBackupRetentionSettings(configured interface{}) *sqladmin.BackupRetentionSettings {
	l := configured.([]interface{})
	if len(l) == 0 {
		return nil
	}
	config := l[0].(map[string]interface{})
	return &sqladmin.BackupRetentionSettings{
		RetainedBackups: int64(config["retained_backups"].(int)),
		RetentionUnit:   config["retention_unit"].(string),
	}
}

func expandActiveDirectoryConfig(configured interface{}) *sqladmin.SqlActiveDirectoryConfig {
	l := configured.([]interface{})
	if len(l) == 0 {
		return nil
	}

	config := l[0].(map[string]interface{})
	return &sqladmin.SqlActiveDirectoryConfig{
		Domain: config["domain"].(string),
	}
}

func expandDenyMaintenancePeriod(configured []interface{}) []*sqladmin.DenyMaintenancePeriod {
	denyMaintenancePeriod := make([]*sqladmin.DenyMaintenancePeriod, 0, len(configured))

	for _, _flag := range configured {
		if _flag == nil {
			continue
		}
		_entry := _flag.(map[string]interface{})

		denyMaintenancePeriod = append(denyMaintenancePeriod, &sqladmin.DenyMaintenancePeriod{
			EndDate:   _entry["end_date"].(string),
			StartDate: _entry["start_date"].(string),
			Time:      _entry["time"].(string),
		})
	}
	return denyMaintenancePeriod

}

func expandSqlServerAdvancedMachineFeatures(configured interface{}) *sqladmin.AdvancedMachineFeatures {
	l := configured.([]interface{})
	if len(l) == 0 {
		return nil
	}

	config := l[0].(map[string]interface{})
	return &sqladmin.AdvancedMachineFeatures{
		ThreadsPerCore: int64(config["threads_per_core"].(int)),
	}
}

func expandSqlServerAuditConfig(configured interface{}) *sqladmin.SqlServerAuditConfig {
	l := configured.([]interface{})
	if len(l) == 0 || l[0] == nil {
		return nil
	}

	config := l[0].(map[string]interface{})
	return &sqladmin.SqlServerAuditConfig{
		Bucket:            config["bucket"].(string),
		RetentionInterval: config["retention_interval"].(string),
		UploadInterval:    config["upload_interval"].(string),
	}
}

func expandInsightsConfig(configured []interface{}) *sqladmin.InsightsConfig {
	if len(configured) == 0 || configured[0] == nil {
		return nil
	}

	_insightsConfig := configured[0].(map[string]interface{})
	return &sqladmin.InsightsConfig{
		QueryInsightsEnabled:  _insightsConfig["query_insights_enabled"].(bool),
		QueryStringLength:     int64(_insightsConfig["query_string_length"].(int)),
		RecordApplicationTags: _insightsConfig["record_application_tags"].(bool),
		RecordClientAddress:   _insightsConfig["record_client_address"].(bool),
		QueryPlansPerMinute:   int64(_insightsConfig["query_plans_per_minute"].(int)),
	}
}

func expandPasswordValidationPolicy(configured []interface{}) *sqladmin.PasswordValidationPolicy {
	if len(configured) == 0 || configured[0] == nil {
		return nil
	}

	_passwordValidationPolicy := configured[0].(map[string]interface{})
	return &sqladmin.PasswordValidationPolicy{
		MinLength:                 int64(_passwordValidationPolicy["min_length"].(int)),
		Complexity:                _passwordValidationPolicy["complexity"].(string),
		ReuseInterval:             int64(_passwordValidationPolicy["reuse_interval"].(int)),
		DisallowUsernameSubstring: _passwordValidationPolicy["disallow_username_substring"].(bool),
		PasswordChangeInterval:    _passwordValidationPolicy["password_change_interval"].(string),
		EnablePasswordPolicy:      _passwordValidationPolicy["enable_password_policy"].(bool),
	}
}

func expandTargetMetrics(configured []interface{}) []*sqladmin.TargetMetric {
	targetMetrics := make([]*sqladmin.TargetMetric, 0, len(configured))
	for _, _metric := range configured {
		if _metric == nil {
			continue
		}
		_entry := _metric.(map[string]interface{})

		targetMetric := &sqladmin.TargetMetric{}

		if v, ok := _entry["metric"]; ok && v != nil {
			targetMetric.Metric = v.(string)
		}

		if v, ok := _entry["target_value"]; ok && v != nil {
			targetMetric.TargetValue = v.(float64)
		}

		targetMetrics = append(targetMetrics, targetMetric)
	}
	return targetMetrics
}

func expandReadPoolAutoScaleConfig(configured []interface{}) *sqladmin.ReadPoolAutoScaleConfig {
	if len(configured) == 0 || configured[0] == nil {
		return nil
	}

	_readPoolAutoScaleConfig := configured[0].(map[string]interface{})
	if !_readPoolAutoScaleConfig["enabled"].(bool) {
		return nil
	}
	return &sqladmin.ReadPoolAutoScaleConfig{
		Enabled:                 _readPoolAutoScaleConfig["enabled"].(bool),
		MaxNodeCount:            int64(_readPoolAutoScaleConfig["max_node_count"].(int)),
		MinNodeCount:            int64(_readPoolAutoScaleConfig["min_node_count"].(int)),
		TargetMetrics:           expandTargetMetrics(_readPoolAutoScaleConfig["target_metrics"].(*schema.Set).List()),
		DisableScaleIn:          _readPoolAutoScaleConfig["disable_scale_in"].(bool),
		ScaleInCooldownSeconds:  int64(_readPoolAutoScaleConfig["scale_in_cooldown_seconds"].(int)),
		ScaleOutCooldownSeconds: int64(_readPoolAutoScaleConfig["scale_out_cooldown_seconds"].(int)),
	}

}

func resourceSqlDatabaseInstanceRead(d *schema.ResourceData, meta interface{}) error {
	config := meta.(*transport_tpg.Config)
	userAgent, err := tpgresource.GenerateUserAgentString(d, config.UserAgent)
	if err != nil {
		return err
	}

	project, err := tpgresource.GetProject(d, config)
	if err != nil {
		return err
	}

	var instance *sqladmin.DatabaseInstance
	err = transport_tpg.Retry(transport_tpg.RetryOptions{
		RetryFunc: func() (rerr error) {
			instance, rerr = config.NewSqlAdminClient(userAgent).Instances.Get(project, d.Get("name").(string)).Do()
			return rerr
		},
		Timeout:              d.Timeout(schema.TimeoutRead),
		ErrorRetryPredicates: []transport_tpg.RetryErrorPredicateFunc{transport_tpg.IsSqlOperationInProgressError},
	})
	if err != nil {
		return transport_tpg.HandleNotFoundError(err, d, fmt.Sprintf("SQL Database Instance %q", d.Get("name").(string)))
	}

	if err := d.Set("name", instance.Name); err != nil {
		return fmt.Errorf("Error setting name: %s", err)
	}
	if err := d.Set("region", instance.Region); err != nil {
		return fmt.Errorf("Error setting region: %s", err)
	}
	if err := d.Set("database_version", instance.DatabaseVersion); err != nil {
		return fmt.Errorf("Error setting database_version: %s", err)
	}
	if err := d.Set("connection_name", instance.ConnectionName); err != nil {
		return fmt.Errorf("Error setting connection_name: %s", err)
	}
	if err := d.Set("maintenance_version", instance.MaintenanceVersion); err != nil {
		return fmt.Errorf("Error setting maintenance_version: %s", err)
	}
	if err := d.Set("psc_service_attachment_link", instance.PscServiceAttachmentLink); err != nil {
		return fmt.Errorf("Error setting psc_service_attachment_link: %s", err)
	}
	if err := d.Set("available_maintenance_versions", instance.AvailableMaintenanceVersions); err != nil {
		return fmt.Errorf("Error setting available_maintenance_version: %s", err)
	}
	if err := d.Set("service_account_email_address", instance.ServiceAccountEmailAddress); err != nil {
		return fmt.Errorf("Error setting service_account_email_address: %s", err)
	}
	if err := d.Set("instance_type", instance.InstanceType); err != nil {
		return fmt.Errorf("Error setting instance_type: %s", err)
	}
	if err := d.Set("node_count", instance.NodeCount); err != nil {
		return fmt.Errorf("Error setting node_count: %s", err)
	}
	if err := d.Set("settings", flattenSettings(instance.Settings, instance.InstanceType, d)); err != nil {
		log.Printf("[WARN] Failed to set SQL Database Instance Settings")
	}
	if instance.DiskEncryptionConfiguration != nil {
		if err := d.Set("encryption_key_name", instance.DiskEncryptionConfiguration.KmsKeyName); err != nil {
			return fmt.Errorf("Error setting encryption_key_name: %s", err)
		}
	}

	if err := d.Set("replica_configuration", flattenReplicaConfiguration(instance.ReplicaConfiguration, d)); err != nil {
		log.Printf("[WARN] Failed to set SQL Database Instance Replica Configuration")
	}

	if err := d.Set("replica_names", instance.ReplicaNames); err != nil {
		return fmt.Errorf("Error setting replica_names: %w", err)
	}

	// We always set replication_cluster because it is computed+optional.
	if err := d.Set("replication_cluster", flattenReplicationCluster(instance.ReplicationCluster, d)); err != nil {
		return fmt.Errorf("Error setting replication_cluster: %w", err)
	}
	ipAddresses := flattenIpAddresses(instance.IpAddresses)
	if err := d.Set("ip_address", ipAddresses); err != nil {
		log.Printf("[WARN] Failed to set SQL Database Instance IP Addresses")
	}

	if len(ipAddresses) > 0 {
		if err := d.Set("first_ip_address", ipAddresses[0]["ip_address"]); err != nil {
			return fmt.Errorf("Error setting first_ip_address: %s", err)
		}
	}

	publicIpAddress := ""
	privateIpAddress := ""
	for _, ip := range instance.IpAddresses {
		if publicIpAddress == "" && ip.Type == "PRIMARY" {
			publicIpAddress = ip.IpAddress
		}

		if privateIpAddress == "" && ip.Type == "PRIVATE" {
			privateIpAddress = ip.IpAddress
		}
	}

	if err := d.Set("public_ip_address", publicIpAddress); err != nil {
		return fmt.Errorf("Error setting public_ip_address: %s", err)
	}
	if err := d.Set("private_ip_address", privateIpAddress); err != nil {
		return fmt.Errorf("Error setting private_ip_address: %s", err)
	}

	if err := d.Set("server_ca_cert", flattenServerCaCerts([]*sqladmin.SslCert{instance.ServerCaCert})); err != nil {
		log.Printf("[WARN] Failed to set SQL Database CA Certificate")
	}

	if err := d.Set("master_instance_name", strings.TrimPrefix(instance.MasterInstanceName, project+":")); err != nil {
		return fmt.Errorf("Error setting master_instance_name: %s", err)
	}
	if err := d.Set("project", project); err != nil {
		return fmt.Errorf("Error setting project: %s", err)
	}
	if err := d.Set("self_link", instance.SelfLink); err != nil {
		return fmt.Errorf("Error setting self_link: %s", err)
	}
	if err := d.Set("psc_service_attachment_link", instance.PscServiceAttachmentLink); err != nil {
		return fmt.Errorf("Error setting psc_service_attachment_link: %s", err)
	}
	if err := d.Set("dns_name", instance.DnsName); err != nil {
		return fmt.Errorf("Error setting dns_name: %s", err)
	}
	if err := d.Set("dns_names", flattenDnsNames(instance.DnsNames)); err != nil {
		return fmt.Errorf("Error setting dns_names: %s", err)
	}
	d.SetId(instance.Name)

	return nil
}

type replicaDRKind int

const (
	replicaDRNone replicaDRKind = iota
	replicaDRByPromote
	replicaDRBySwitchover
)

func resourceSqlDatabaseInstanceUpdate(d *schema.ResourceData, meta interface{}) error {
	config := meta.(*transport_tpg.Config)
	userAgent, err := tpgresource.GenerateUserAgentString(d, config.UserAgent)
	if err != nil {
		return err
	}

	project, err := tpgresource.GetProject(d, config)
	if err != nil {
		return err
	}
	var maintenance_version string
	if v, ok := d.GetOk("maintenance_version"); ok {
		maintenance_version = v.(string)
	}

	replicaDRKind := replicaDRNone
	if d.HasChange("instance_type") {
		oldInstanceType, newInstanceType := d.GetChange("instance_type")

		if isReplicaPromoteRequested(nil, oldInstanceType, newInstanceType, nil) {
			if isSwitchoverRequested(d) {
				replicaDRKind = replicaDRBySwitchover
			} else {
				err = checkPromoteConfigurations(d)
				if err != nil {
					return err
				}
				replicaDRKind = replicaDRByPromote
			}
		}
	}

	desiredSetting := d.Get("settings")
	var op *sqladmin.Operation
	var instance *sqladmin.DatabaseInstance

	databaseVersion := d.Get("database_version").(string)

	// Check if the activation policy is being updated. If it is being changed to ALWAYS this should be done first.
	if d.HasChange("settings.0.activation_policy") && d.Get("settings.0.activation_policy").(string) == "ALWAYS" {
		instance = &sqladmin.DatabaseInstance{Settings: &sqladmin.Settings{ActivationPolicy: "ALWAYS"}}
		err = transport_tpg.Retry(transport_tpg.RetryOptions{
			RetryFunc: func() (rerr error) {
				op, rerr = config.NewSqlAdminClient(userAgent).Instances.Patch(project, d.Get("name").(string), instance).Do()
				return rerr
			},
			Timeout:              d.Timeout(schema.TimeoutUpdate),
			ErrorRetryPredicates: []transport_tpg.RetryErrorPredicateFunc{transport_tpg.IsSqlOperationInProgressError},
		})
		if err != nil {
			return fmt.Errorf("Error, failed to patch instance settings for %s: %s", instance.Name, err)
		}
		err = SqlAdminOperationWaitTime(config, op, project, "Patch Instance", userAgent, d.Timeout(schema.TimeoutUpdate))
		if err != nil {
			return err
		}
		err = resourceSqlDatabaseInstanceRead(d, meta)
		if err != nil {
			return err
		}
	}

	// Check if the database version is being updated, because patching database version is an atomic operation and can not be
	// performed with other fields, we first patch database version before updating the rest of the fields.
	if d.HasChange("database_version") {
		instance = &sqladmin.DatabaseInstance{DatabaseVersion: databaseVersion}
		err = transport_tpg.Retry(transport_tpg.RetryOptions{
			RetryFunc: func() (rerr error) {
				op, rerr = config.NewSqlAdminClient(userAgent).Instances.Patch(project, d.Get("name").(string), instance).Do()
				return rerr
			},
			Timeout:              d.Timeout(schema.TimeoutUpdate),
			ErrorRetryPredicates: []transport_tpg.RetryErrorPredicateFunc{transport_tpg.IsSqlOperationInProgressError},
		})
		if err != nil {
			return fmt.Errorf("Error, failed to patch instance settings for %s: %s", instance.Name, err)
		}
		err = SqlAdminOperationWaitTime(config, op, project, "Patch Instance", userAgent, d.Timeout(schema.TimeoutUpdate))
		if err != nil {
			return err
		}
		err = resourceSqlDatabaseInstanceRead(d, meta)
		if err != nil {
			return err
		}
	}

	// Check if the root_password is being updated, because updating root_password is an atomic operation and can not be
	// performed with other fields, we first update root password before updating the rest of the fields.
	if d.HasChange("root_password") || d.HasChange("root_password_wo_version") {
		var oldPwd, newPwd interface{}
		if d.HasChange("root_password_wo_version") {
			oldPwd = ""
			newPwd = tpgresource.GetRawConfigAttributeAsString(d, "root_password_wo")
		} else {
			oldPwd, newPwd = d.GetChange("root_password")
		}
		password := newPwd.(string)
		dv := d.Get("database_version").(string)
		name := ""
		host := ""
		if strings.Contains(dv, "MYSQL") {
			name = "root"
			host = "%"
		} else if strings.Contains(dv, "POSTGRES") {
			name = "postgres"
		} else if strings.Contains(dv, "SQLSERVER") {
			name = "sqlserver"
			if len(password) == 0 {
				if err := d.Set("root_password", oldPwd.(string)); err != nil {
					return fmt.Errorf("Error re-setting root_password: %s", err)
				}
				return fmt.Errorf("Error, root password cannot be empty for SQL Server instance.")
			}
		} else {
			if err := d.Set("root_password", oldPwd.(string)); err != nil {
				return fmt.Errorf("Error re-setting root_password: %s", err)
			}
			return fmt.Errorf("Error, invalid database version")
		}
		instance := d.Get("name").(string)

		user := &sqladmin.User{
			Name:     name,
			Instance: instance,
			Password: password,
		}

		transport_tpg.MutexStore.Lock(instanceMutexKey(project, instance))
		defer transport_tpg.MutexStore.Unlock(instanceMutexKey(project, instance))
		var op *sqladmin.Operation
		updateFunc := func() error {
			op, err = config.NewSqlAdminClient(userAgent).Users.Update(project, instance, user).Host(host).Name(name).Do()
			return err
		}
		err = transport_tpg.Retry(transport_tpg.RetryOptions{
			RetryFunc: updateFunc,
			Timeout:   d.Timeout(schema.TimeoutUpdate),
		})

		if err != nil {
			if err := d.Set("root_password", oldPwd.(string)); err != nil {
				return fmt.Errorf("Error re-setting root_password: %s", err)
			}
			return fmt.Errorf("Error, failed to update root_password : %s", err)
		}

		err = SqlAdminOperationWaitTime(config, op, project, "Insert User", userAgent, d.Timeout(schema.TimeoutUpdate))

		if err != nil {
			if err := d.Set("root_password", oldPwd.(string)); err != nil {
				return fmt.Errorf("Error re-setting root_password: %s", err)
			}
			return fmt.Errorf("Error, failed to update root_password : %s", err)
		}
	}

	// Check if the maintenance version is being updated, because patching maintenance version is an atomic operation and can not be
	// performed with other fields, we first patch maintenance version before updating the rest of the fields.
	if d.HasChange("maintenance_version") {
		instance = &sqladmin.DatabaseInstance{MaintenanceVersion: maintenance_version}
		err = transport_tpg.Retry(transport_tpg.RetryOptions{
			RetryFunc: func() (rerr error) {
				op, rerr = config.NewSqlAdminClient(userAgent).Instances.Patch(project, d.Get("name").(string), instance).Do()
				return rerr
			},
			Timeout:              d.Timeout(schema.TimeoutUpdate),
			ErrorRetryPredicates: []transport_tpg.RetryErrorPredicateFunc{transport_tpg.IsSqlOperationInProgressError},
		})
		if err != nil {
			return fmt.Errorf("Error, failed to patch instance settings for %s: %s", instance.Name, err)
		}
		err = SqlAdminOperationWaitTime(config, op, project, "Patch Instance", userAgent, d.Timeout(schema.TimeoutUpdate))
		if err != nil {
			return err
		}
		err = resourceSqlDatabaseInstanceRead(d, meta)
		if err != nil {
			return err
		}
	}

	if replicaDRKind != replicaDRNone {
		var retryFunc func() (rerr error)
		switch replicaDRKind {
		case replicaDRByPromote:
			retryFunc = func() (rerr error) {
				op, rerr = config.NewSqlAdminClient(userAgent).Instances.PromoteReplica(project, d.Get("name").(string)).Do()
				return rerr
			}
		case replicaDRBySwitchover:
			retryFunc = func() (rerr error) {
				op, rerr = config.NewSqlAdminClient(userAgent).Instances.Switchover(project, d.Get("name").(string)).Do()
				return rerr
			}
		default:
			return fmt.Errorf("unknown replica DR scenario: %v", replicaDRKind)
		}

		err = transport_tpg.Retry(transport_tpg.RetryOptions{
			RetryFunc:            retryFunc,
			Timeout:              d.Timeout(schema.TimeoutUpdate),
			ErrorRetryPredicates: []transport_tpg.RetryErrorPredicateFunc{transport_tpg.IsSqlOperationInProgressError},
		})
		if err != nil {
			return fmt.Errorf("Error, failed to promote read replica instance as primary stand-alone %s: %s", d.Get("name"), err)
		}
		err = SqlAdminOperationWaitTime(config, op, project, "Promote Instance", userAgent, d.Timeout(schema.TimeoutUpdate))
		if err != nil {
			return err
		}
		err = resourceSqlDatabaseInstanceRead(d, meta)
		if err != nil {
			return err
		}
	}

	// Check if the edition is being updated, because patching edition is an atomic operation and can not be
	// performed with other fields, we first patch edition, tier and data cache config before updating the rest of the fields.
	if d.HasChange("settings.0.edition") {
		edition := d.Get("settings.0.edition").(string)
		tier := d.Get("settings.0.tier").(string)
		dataCacheConfig := expandDataCacheConfig(d.Get("settings.0.data_cache_config").([]interface{}))
		instance = &sqladmin.DatabaseInstance{Settings: &sqladmin.Settings{Edition: edition, Tier: tier, DataCacheConfig: dataCacheConfig}}
		err = transport_tpg.Retry(transport_tpg.RetryOptions{
			RetryFunc: func() (rerr error) {
				op, rerr = config.NewSqlAdminClient(userAgent).Instances.Patch(project, d.Get("name").(string), instance).Do()
				return rerr
			},
			Timeout:              d.Timeout(schema.TimeoutUpdate),
			ErrorRetryPredicates: []transport_tpg.RetryErrorPredicateFunc{transport_tpg.IsSqlOperationInProgressError},
		})
		if err != nil {
			return fmt.Errorf("Error, failed to patch instance settings for %s: %s", instance.Name, err)
		}
		err = SqlAdminOperationWaitTime(config, op, project, "Patch Instance", userAgent, d.Timeout(schema.TimeoutUpdate))
		if err != nil {
			return err
		}
		err = resourceSqlDatabaseInstanceRead(d, meta)
		if err != nil {
			return err
		}
	}

	instance = &sqladmin.DatabaseInstance{
		Settings: expandSqlDatabaseInstanceSettings(desiredSetting.([]interface{}), databaseVersion),
	}

	if d.HasChange("settings.0.disk_size") {
		autoResize := true
		_, autoResizeI := d.GetChange("settings.0.disk_autoresize")
		if autoResizeI != nil {
			autoResize = autoResizeI.(bool)
		}
		oldDiskSizeI, newDiskSizeI := d.GetChange("settings.0.disk_size")
		if !autoResize || newDiskSizeI.(int) > oldDiskSizeI.(int) {
			// If auto resize is not enabled - set the disk size as requested, even if it's a decrease - let it fail.
			// Otherwise, allow increasing even if auto resize is enabled.
			instance.Settings.DataDiskSizeGb = int64(newDiskSizeI.(int))
		}
	}

	s := d.Get("settings")
	_settings := s.([]interface{})[0].(map[string]interface{})
	// Instance.Patch operation on completion updates the settings proto version by +8. As terraform does not know this it tries
	// to make an update call with the proto version before patch and fails. To resolve this issue we update the setting version
	// before making the update call.
	instance.Settings.SettingsVersion = int64(_settings["version"].(int))
	// Collation cannot be included in the update request
	instance.Settings.Collation = ""
	instance.Settings.DataCacheConfig = expandDataCacheConfig(_settings["data_cache_config"].([]interface{}))

	// Lock on the master_instance_name just in case updating any replica
	// settings causes operations on the master.
	if v, ok := d.GetOk("master_instance_name"); ok {
		transport_tpg.MutexStore.Lock(instanceMutexKey(project, v.(string)))
		defer transport_tpg.MutexStore.Unlock(instanceMutexKey(project, v.(string)))
	}

	if _, ok := d.GetOk("instance_type"); ok {
		instance.InstanceType = d.Get("instance_type").(string)
	}

	if _, ok := d.GetOk("node_count"); ok {
		instance.NodeCount = int64(d.Get("node_count").(int))
	}

	if _, ok := d.GetOk("psc_service_attachment_link"); ok {
		instance.PscServiceAttachmentLink = d.Get("psc_service_attachment_link").(string)
	}

	// Database Version is required for all calls with Google ML integration enabled or it will be rejected by the API.
	if d.Get("settings.0.enable_google_ml_integration").(bool) || len(_settings["connection_pool_config"].(*schema.Set).List()) > 0 {
		instance.DatabaseVersion = databaseVersion
	}

	failoverDrReplicaName := d.Get("replication_cluster.0.failover_dr_replica_name").(string)
	psaWriteEndpoint := d.Get("replication_cluster.0.psa_write_endpoint").(string)
	if failoverDrReplicaName != "" || psaWriteEndpoint != "" {
		if failoverDrReplicaName != "" && psaWriteEndpoint != "" {
			instance.ReplicationCluster = &sqladmin.ReplicationCluster{
				FailoverDrReplicaName: failoverDrReplicaName,
				PsaWriteEndpoint:      psaWriteEndpoint,
			}
		} else if failoverDrReplicaName != "" {
			instance.ReplicationCluster = &sqladmin.ReplicationCluster{
				FailoverDrReplicaName: failoverDrReplicaName,
			}
		} else {
			instance.ReplicationCluster = &sqladmin.ReplicationCluster{
				PsaWriteEndpoint: psaWriteEndpoint,
			}
		}
	}

	err = transport_tpg.Retry(transport_tpg.RetryOptions{
		RetryFunc: func() (rerr error) {
			op, rerr = config.NewSqlAdminClient(userAgent).Instances.Update(project, d.Get("name").(string), instance).Do()
			return rerr
		},
		Timeout:              d.Timeout(schema.TimeoutUpdate),
		ErrorRetryPredicates: []transport_tpg.RetryErrorPredicateFunc{transport_tpg.IsSqlOperationInProgressError},
	})
	if err != nil {
		return fmt.Errorf("Error, failed to update instance settings for %s: %s", instance.Name, err)
	}

	err = SqlAdminOperationWaitTime(config, op, project, "Update Instance", userAgent, d.Timeout(schema.TimeoutUpdate))
	if err != nil {
		return err
	}

	// Perform a backup restore if the backup context exists and has changed
	if r, ok := d.GetOk("restore_backup_context"); ok {
		if d.HasChange("restore_backup_context") {
			err = sqlDatabaseInstanceRestoreFromBackup(d, config, userAgent, project, d.Get("name").(string), r, "")
			if err != nil {
				return err
			}
		}
	} else if b, ok := d.GetOk("backupdr_backup"); ok && b.(string) != "" {
		if d.HasChange("backupdr_backup") {
			err = sqlDatabaseInstanceRestoreFromBackup(d, config, userAgent, project, d.Get("name").(string), nil, b)
			if err != nil {
				return err
			}
		}
	}

	// Check if timezone is updated
	if d.HasChange("settings.0.time_zone") {
		timezone := d.Get("settings.0.time_zone").(string)
		instance = &sqladmin.DatabaseInstance{Settings: &sqladmin.Settings{TimeZone: timezone}}
		err = transport_tpg.Retry(transport_tpg.RetryOptions{
			RetryFunc: func() (rerr error) {
				op, rerr = config.NewSqlAdminClient(userAgent).Instances.Patch(project, d.Get("name").(string), instance).Do()
				return err
			},
			Timeout:              d.Timeout(schema.TimeoutUpdate),
			ErrorRetryPredicates: []transport_tpg.RetryErrorPredicateFunc{transport_tpg.IsSqlOperationInProgressError},
		})
		if err != nil {
			return fmt.Errorf("Error, failed to patch instance settings for %s: %s", instance.Name, err)
		}
		err = SqlAdminOperationWaitTime(config, op, project, "Patch Instance", userAgent, d.Timeout(schema.TimeoutUpdate))
		if err != nil {
			return err
		}
		err = resourceSqlDatabaseInstanceRead(d, meta)
		if err != nil {
			return err
		}
	}

	return resourceSqlDatabaseInstanceRead(d, meta)
}

func maintenanceVersionDiffSuppress(_, old, new string, _ *schema.ResourceData) bool {
	// Ignore the database version part and only compare the last part of the maintenance version which represents the release date of the version.
	if len(old) > 14 && len(new) > 14 && old[len(old)-14:] >= new[len(new)-14:] {
		log.Printf("[DEBUG] Maintenance version in configuration [%s] is older than current maintenance version [%s] on instance. Suppressing diff", new, old)
		return true
	} else {
		return false
	}
}

// enhancedBackupManagerDiffSuppressFunc suppresses diff changes to settings.backup_configuration
// when the SQL instance is managed by Google Cloud Backup and Disaster Recovery (DR) Service.
func EnhancedBackupManagerDiffSuppressFunc(k, old, new string, d *schema.ResourceData) bool {
	// If the API marks this instance as ENHANCED-managed, suppress diffs for backup config fields.
	tier, _ := d.Get("settings.0.backup_configuration.0.backup_tier").(string)
	if tier == "" {
		return false
	}
	if strings.EqualFold(tier, "ENHANCED") {
		log.Printf(
			"[INFO] Instance %s is managed by Google Cloud Backup and Disaster Recovery (BackupDR). "+
				"Terraform will not manage the '%s' field. "+
				"Any changes to this field in your Terraform configuration will be ignored.", d.Get("name"), k,
		)
		return true
	}
	return false
}

func databaseVersionDiffSuppress(_, oldVersion, newVersion string, _ *schema.ResourceData) bool {
	// Suppress diff when newVersion is MYSQL_8_0 and oldVersion is >= MYSQL_8_0_35 for MySQL version auto-upgrade cases.
	if newVersion == "MYSQL_8_0" && strings.HasPrefix(oldVersion, "MYSQL_8_0_") {
		if oldMinorVersion, err := strconv.Atoi(oldVersion[len("MYSQL_8_0_"):]); err == nil && oldMinorVersion >= 35 {
			log.Printf("[DEBUG] Database version is updated by auto version upgrade for MySQL 8.0, version updated from [%s] to [%s]. Suppressing diff.", oldVersion, newVersion)
			return true
		}
	}

	return false
}

func resourceSqlDatabaseInstanceDelete(d *schema.ResourceData, meta interface{}) error {
	config := meta.(*transport_tpg.Config)
	userAgent, err := tpgresource.GenerateUserAgentString(d, config.UserAgent)
	if err != nil {
		return err
	}

	project, err := tpgresource.GetProject(d, config)
	if err != nil {
		return err
	}

	// Check if deletion protection is enabled.

	if d.Get("deletion_protection").(bool) {
		return fmt.Errorf("Error, failed to delete instance because deletion_protection is set to true. Set it to false to proceed with instance deletion")
	}

	// Lock on the master_instance_name just in case deleting a replica causes
	// operations on the master.
	if v, ok := d.GetOk("master_instance_name"); ok {
		transport_tpg.MutexStore.Lock(instanceMutexKey(project, v.(string)))
		defer transport_tpg.MutexStore.Unlock(instanceMutexKey(project, v.(string)))
	}

	var op *sqladmin.Operation
	finalBackupDescription := ""
	if v, ok := d.GetOk("final_backup_description"); ok {
		finalBackupDescription = v.(string)
	}
	err = transport_tpg.Retry(transport_tpg.RetryOptions{
		RetryFunc: func() (rerr error) {
			op, rerr = config.NewSqlAdminClient(userAgent).Instances.Delete(project, d.Get("name").(string)).FinalBackupDescription(finalBackupDescription).Do()
			if rerr != nil {
				return rerr
			}
			err = SqlAdminOperationWaitTime(config, op, project, "Delete Instance", userAgent, d.Timeout(schema.TimeoutDelete))
			if err != nil {
				return err
			}
			return nil
		},
		Timeout:              d.Timeout(schema.TimeoutDelete),
		ErrorRetryPredicates: []transport_tpg.RetryErrorPredicateFunc{transport_tpg.IsSqlOperationInProgressError, IsSqlInternalError},
	})
	if err != nil {
		return fmt.Errorf("Error, failed to delete instance %s: %s", d.Get("name").(string), err)
	}
	return nil
}

func resourceSqlDatabaseInstanceImport(d *schema.ResourceData, meta interface{}) ([]*schema.ResourceData, error) {
	config := meta.(*transport_tpg.Config)
	if err := tpgresource.ParseImportId([]string{
		"^projects/(?P<project>[^/]+)/instances/(?P<name>[^/]+)$",
		"^(?P<project>[^/]+)/(?P<name>[^/]+)$",
		"^(?P<name>[^/]+)$"}, d, config); err != nil {
		return nil, err
	}

	if err := d.Set("deletion_protection", true); err != nil {
		return nil, fmt.Errorf("Error setting deletion_protection: %s", err)
	}

	// Replace import id for the resource id
	id, err := tpgresource.ReplaceVars(d, config, "projects/{{project}}/instances/{{name}}")
	if err != nil {
		return nil, fmt.Errorf("Error constructing id: %s", err)
	}
	d.SetId(id)

	return []*schema.ResourceData{d}, nil
}

func flattenSettings(settings *sqladmin.Settings, iType string, d *schema.ResourceData) []map[string]interface{} {
	if settings == nil {
		return nil
	}

	data := map[string]interface{}{
		"version":                     settings.SettingsVersion,
		"tier":                        settings.Tier,
		"edition":                     flattenEdition(settings.Edition),
		"activation_policy":           settings.ActivationPolicy,
		"availability_type":           d.Get("settings.0.availability_type"),
		"effective_availability_type": settings.AvailabilityType,
		"collation":                   settings.Collation,
		"connector_enforcement":       settings.ConnectorEnforcement,
		"disk_type":                   settings.DataDiskType,
		"disk_size":                   settings.DataDiskSizeGb,
		"pricing_plan":                settings.PricingPlan,
		"user_labels":                 settings.UserLabels,
		"password_validation_policy":  settings.PasswordValidationPolicy,
		"time_zone":                   settings.TimeZone,
		"deletion_protection_enabled": settings.DeletionProtectionEnabled,
		"retain_backups_on_delete":    settings.RetainBackupsOnDelete,
		"final_backup_config":         settings.FinalBackupConfig,
	}

	if data["availability_type"] == "" {
		data["availability_type"] = "ZONAL"
	}
	// For read pools, availability type is server managed. Above, we
	// pull it from the old TF resource so that it never shows a
	// diff. Now, here, for non-pool instances, we overwrite it with the
	// value obtained from the API (which would be the typical way to
	// populate the field).
	if iType != "READ_POOL_INSTANCE" {
		data["availability_type"] = settings.AvailabilityType
	}

	if settings.ActiveDirectoryConfig != nil {
		data["active_directory_config"] = flattenActiveDirectoryConfig(settings.ActiveDirectoryConfig)
	}

	if settings.DenyMaintenancePeriods != nil {
		data["deny_maintenance_period"] = flattenDenyMaintenancePeriod(settings.DenyMaintenancePeriods)
	}

	if settings.SqlServerAuditConfig != nil {
		data["sql_server_audit_config"] = flattenSqlServerAuditConfig(settings.SqlServerAuditConfig)
	}

	if settings.BackupConfiguration != nil {
		data["backup_configuration"] = flattenBackupConfiguration(settings.BackupConfiguration)
	}

	if settings.FinalBackupConfig != nil {
		data["final_backup_config"] = flattenFinalBackupConfig(settings.FinalBackupConfig)
	}

	if settings.DatabaseFlags != nil {
		data["database_flags"] = flattenDatabaseFlags(settings.DatabaseFlags)
	}

	if settings.ConnectionPoolConfig != nil {
		data["connection_pool_config"] = flattenConnectionPoolConfig(settings.ConnectionPoolConfig)
	}

	if settings.IpConfiguration != nil {
		data["ip_configuration"] = flattenIpConfiguration(settings.IpConfiguration, d)
	}

	if settings.LocationPreference != nil {
		data["location_preference"] = flattenLocationPreference(settings.LocationPreference)
	}

	if settings.MaintenanceWindow != nil {
		data["maintenance_window"] = flattenMaintenanceWindow(settings.MaintenanceWindow)
	}

	if settings.InsightsConfig != nil {
		data["insights_config"] = flattenInsightsConfig(settings.InsightsConfig)
	}

	if settings.ReadPoolAutoScaleConfig != nil {
		data["read_pool_auto_scale_config"] = flattenReadPoolAutoScaleConfig(settings.ReadPoolAutoScaleConfig)
	} else {
		data["read_pool_auto_scale_config"] = []map[string]interface{}{
			{
				"enabled": false,
			},
		}
	}

	data["disk_autoresize"] = settings.StorageAutoResize
	data["disk_autoresize_limit"] = settings.StorageAutoResizeLimit

	data["enable_google_ml_integration"] = settings.EnableGoogleMlIntegration
	data["enable_dataplex_integration"] = settings.EnableDataplexIntegration

	if settings.UserLabels != nil {
		data["user_labels"] = settings.UserLabels
	}

	if settings.PasswordValidationPolicy != nil {
		data["password_validation_policy"] = flattenPasswordValidationPolicy(settings.PasswordValidationPolicy)
	}

	if settings.DataCacheConfig != nil {
		data["data_cache_config"] = flattenDataCacheConfig(settings.DataCacheConfig)
	} else {
		data["data_cache_config"] = []map[string]interface{}{
			{
				"data_cache_enabled": false,
			},
		}
	}

	if settings.AdvancedMachineFeatures != nil {
		data["advanced_machine_features"] = flattenSqlServerAdvancedMachineFeatures(settings.AdvancedMachineFeatures)
	}

	return []map[string]interface{}{data}
}

func flattenDataCacheConfig(d *sqladmin.DataCacheConfig) []map[string]interface{} {
	if d == nil {
		return []map[string]interface{}{
			{
				"data_cache_enabled": false,
			},
		}
	}
	return []map[string]interface{}{
		{
			"data_cache_enabled": d.DataCacheEnabled,
		},
	}
}

func flattenBackupConfiguration(backupConfiguration *sqladmin.BackupConfiguration) []map[string]interface{} {
	data := map[string]interface{}{
		"binary_log_enabled":             backupConfiguration.BinaryLogEnabled,
		"enabled":                        backupConfiguration.Enabled,
		"start_time":                     backupConfiguration.StartTime,
		"location":                       backupConfiguration.Location,
		"point_in_time_recovery_enabled": backupConfiguration.PointInTimeRecoveryEnabled,
		"backup_retention_settings":      flattenBackupRetentionSettings(backupConfiguration.BackupRetentionSettings),
		"transaction_log_retention_days": backupConfiguration.TransactionLogRetentionDays,
		"backup_tier":                    backupConfiguration.BackupTier,
	}

	return []map[string]interface{}{data}
}

func flattenFinalBackupConfig(finalBackupConfig *sqladmin.FinalBackupConfig) []map[string]interface{} {
	data := map[string]interface{}{
		"enabled":        finalBackupConfig.Enabled,
		"retention_days": finalBackupConfig.RetentionDays,
	}

	return []map[string]interface{}{data}
}

func flattenBackupRetentionSettings(b *sqladmin.BackupRetentionSettings) []map[string]interface{} {
	if b == nil {
		return nil
	}
	return []map[string]interface{}{
		{
			"retained_backups": b.RetainedBackups,
			"retention_unit":   b.RetentionUnit,
		},
	}
}

func flattenActiveDirectoryConfig(sqlActiveDirectoryConfig *sqladmin.SqlActiveDirectoryConfig) []map[string]interface{} {
	if sqlActiveDirectoryConfig == nil {
		return nil
	}
	return []map[string]interface{}{
		{
			"domain": sqlActiveDirectoryConfig.Domain,
		},
	}
}

func flattenDenyMaintenancePeriod(denyMaintenancePeriod []*sqladmin.DenyMaintenancePeriod) []map[string]interface{} {
	flags := make([]map[string]interface{}, 0, len(denyMaintenancePeriod))

	for _, flag := range denyMaintenancePeriod {
		data := map[string]interface{}{
			"end_date":   flag.EndDate,
			"start_date": flag.StartDate,
			"time":       flag.Time,
		}

		flags = append(flags, data)
	}

	return flags
}

func flattenSqlServerAdvancedMachineFeatures(advancedMachineFeatures *sqladmin.AdvancedMachineFeatures) []map[string]interface{} {
	if advancedMachineFeatures == nil {
		return nil
	}
	return []map[string]interface{}{
		{
			"threads_per_core": advancedMachineFeatures.ThreadsPerCore,
		},
	}
}

func flattenSqlServerAuditConfig(sqlServerAuditConfig *sqladmin.SqlServerAuditConfig) []map[string]interface{} {
	if sqlServerAuditConfig == nil {
		return nil
	}
	return []map[string]interface{}{
		{
			"bucket":             sqlServerAuditConfig.Bucket,
			"retention_interval": sqlServerAuditConfig.RetentionInterval,
			"upload_interval":    sqlServerAuditConfig.UploadInterval,
		},
	}
}

func flattenDatabaseFlags(databaseFlags []*sqladmin.DatabaseFlags) []map[string]interface{} {
	flags := make([]map[string]interface{}, 0, len(databaseFlags))

	for _, flag := range databaseFlags {
		data := map[string]interface{}{
			"name":  flag.Name,
			"value": flag.Value,
		}

		flags = append(flags, data)
	}

	return flags
}

// flattenReplicationCluster converts cloud SQL backend ReplicationCluster (proto) to
// terraform replication_cluster. We explicitly allow the case when ReplicationCluster
// is nil since replication_cluster is computed+optional.
func flattenReplicationCluster(replicationCluster *sqladmin.ReplicationCluster, d *schema.ResourceData) []map[string]interface{} {
	data := make(map[string]interface{})
	data["psa_write_endpoint"] = ""
	if replicationCluster != nil && replicationCluster.PsaWriteEndpoint != "" {
		data["psa_write_endpoint"] = replicationCluster.PsaWriteEndpoint
	}
	data["failover_dr_replica_name"] = ""
	if replicationCluster != nil && replicationCluster.FailoverDrReplicaName != "" {
		data["failover_dr_replica_name"] = replicationCluster.FailoverDrReplicaName
	}
	data["dr_replica"] = false
	if replicationCluster != nil {
		data["dr_replica"] = replicationCluster.DrReplica
	}
	return []map[string]interface{}{data}
}

func flattenConnectionPoolFlags(connectionPoolFlags []*sqladmin.ConnectionPoolFlags) []interface{} {
	if len(connectionPoolFlags) == 0 { // Handles nil or empty slice
		return make([]interface{}, 0) // Explicitly return empty slice
	}

	mcpflags := make([]interface{}, len(connectionPoolFlags)) // Pre-allocate for efficiency
	for i, mcpflag := range connectionPoolFlags {
		data := map[string]interface{}{
			"name":  mcpflag.Name,
			"value": mcpflag.Value,
		}
		mcpflags[i] = data
	}
	return mcpflags
}

func flattenConnectionPoolConfig(connectionPoolConfig *sqladmin.ConnectionPoolConfig) []interface{} {
	if connectionPoolConfig == nil {
		return []interface{}{
			map[string]interface{}{
				"connection_pooling_enabled": false,
				"flags":                      make([]interface{}, 0), // Default to empty flags
			},
		}
	}
	data := map[string]interface{}{
		"connection_pooling_enabled": connectionPoolConfig.ConnectionPoolingEnabled,          // Corrected key
		"flags":                      flattenConnectionPoolFlags(connectionPoolConfig.Flags), // Corrected key
	}
	return []interface{}{data}
}

func flattenIpConfiguration(ipConfiguration *sqladmin.IpConfiguration, d *schema.ResourceData) interface{} {
	data := map[string]interface{}{
		"ipv4_enabled":       ipConfiguration.Ipv4Enabled,
		"private_network":    ipConfiguration.PrivateNetwork,
		"allocated_ip_range": ipConfiguration.AllocatedIpRange,
		"enable_private_path_for_google_cloud_services": ipConfiguration.EnablePrivatePathForGoogleCloudServices,
		"ssl_mode":                         ipConfiguration.SslMode,
		"server_ca_mode":                   ipConfiguration.ServerCaMode,
		"server_ca_pool":                   ipConfiguration.ServerCaPool,
		"custom_subject_alternative_names": ipConfiguration.CustomSubjectAlternativeNames,
	}

	if ipConfiguration.AuthorizedNetworks != nil {
		data["authorized_networks"] = flattenAuthorizedNetworks(ipConfiguration.AuthorizedNetworks)
	}

	if ipConfiguration.PscConfig != nil {
		data["psc_config"] = flattenPscConfigs(ipConfiguration.PscConfig)
	}

	return []map[string]interface{}{data}
}

func flattenPscAutoConnections(pscAutoConnections []*sqladmin.PscAutoConnectionConfig) []map[string]interface{} {
	flags := make([]map[string]interface{}, 0, len(pscAutoConnections))

	for _, flag := range pscAutoConnections {
		data := map[string]interface{}{
			"consumer_network":            flag.ConsumerNetwork,
			"consumer_network_status":     flag.ConsumerNetworkStatus,
			"consumer_service_project_id": flag.ConsumerProject,
			"ip_address":                  flag.IpAddress,
			"status":                      flag.Status,
		}

		flags = append(flags, data)
	}

	return flags
}

func flattenPscConfigs(pscConfig *sqladmin.PscConfig) interface{} {
	data := map[string]interface{}{
		"psc_enabled":               pscConfig.PscEnabled,
		"allowed_consumer_projects": schema.NewSet(schema.HashString, tpgresource.ConvertStringArrToInterface(pscConfig.AllowedConsumerProjects)),
		"network_attachment_uri":    pscConfig.NetworkAttachmentUri,
		"psc_auto_connections":      flattenPscAutoConnections(pscConfig.PscAutoConnections),
	}

	return []map[string]interface{}{data}
}

func flattenAuthorizedNetworks(entries []*sqladmin.AclEntry) interface{} {
	networks := schema.NewSet(schema.HashResource(sqlDatabaseAuthorizedNetWorkSchemaElem), []interface{}{})

	for _, entry := range entries {
		data := map[string]interface{}{
			"expiration_time": entry.ExpirationTime,
			"name":            entry.Name,
			"value":           entry.Value,
		}

		networks.Add(data)
	}

	return networks
}

func flattenLocationPreference(locationPreference *sqladmin.LocationPreference) interface{} {
	data := map[string]interface{}{
		"follow_gae_application": locationPreference.FollowGaeApplication,
		"zone":                   locationPreference.Zone,
		"secondary_zone":         locationPreference.SecondaryZone,
	}

	return []map[string]interface{}{data}
}

func flattenMaintenanceWindow(maintenanceWindow *sqladmin.MaintenanceWindow) interface{} {
	data := map[string]interface{}{
		"day":          maintenanceWindow.Day,
		"hour":         maintenanceWindow.Hour,
		"update_track": maintenanceWindow.UpdateTrack,
	}

	return []map[string]interface{}{data}
}

func flattenReplicaConfiguration(replicaConfiguration *sqladmin.ReplicaConfiguration, d *schema.ResourceData) []map[string]interface{} {
	rc := []map[string]interface{}{}

	if replicaConfiguration != nil {
		data := map[string]interface{}{
			"cascadable_replica": replicaConfiguration.CascadableReplica,
			"failover_target":    replicaConfiguration.FailoverTarget,

			// Don't attempt to assign anything from replicaConfiguration.MysqlReplicaConfiguration,
			// since those fields are set on create and then not stored. See description at
			// https://cloud.google.com/sql/docs/mysql/admin-api/v1beta4/instances.
			// Instead, set them to the values they previously had so we don't set them all to zero.
			"ca_certificate":            d.Get("replica_configuration.0.ca_certificate"),
			"client_certificate":        d.Get("replica_configuration.0.client_certificate"),
			"client_key":                d.Get("replica_configuration.0.client_key"),
			"connect_retry_interval":    d.Get("replica_configuration.0.connect_retry_interval"),
			"dump_file_path":            d.Get("replica_configuration.0.dump_file_path"),
			"master_heartbeat_period":   d.Get("replica_configuration.0.master_heartbeat_period"),
			"password":                  d.Get("replica_configuration.0.password"),
			"ssl_cipher":                d.Get("replica_configuration.0.ssl_cipher"),
			"username":                  d.Get("replica_configuration.0.username"),
			"verify_server_certificate": d.Get("replica_configuration.0.verify_server_certificate"),
		}
		rc = append(rc, data)
	}

	return rc
}

func flattenIpAddresses(ipAddresses []*sqladmin.IpMapping) []map[string]interface{} {
	var ips []map[string]interface{}

	for _, ip := range ipAddresses {
		data := map[string]interface{}{
			"ip_address":     ip.IpAddress,
			"type":           ip.Type,
			"time_to_retire": ip.TimeToRetire,
		}

		ips = append(ips, data)
	}

	return ips
}

func flattenDnsNames(dnsNames []*sqladmin.DnsNameMapping) []map[string]interface{} {
	var dns []map[string]interface{}

	for _, mapping := range dnsNames {
		data := map[string]interface{}{
			"name":            mapping.Name,
			"connection_type": mapping.ConnectionType,
			"dns_scope":       mapping.DnsScope,
		}

		dns = append(dns, data)
	}

	return dns
}

func flattenServerCaCerts(caCerts []*sqladmin.SslCert) []map[string]interface{} {
	var certs []map[string]interface{}

	for _, caCert := range caCerts {
		if caCert != nil {
			data := map[string]interface{}{
				"cert":             caCert.Cert,
				"common_name":      caCert.CommonName,
				"create_time":      caCert.CreateTime,
				"expiration_time":  caCert.ExpirationTime,
				"sha1_fingerprint": caCert.Sha1Fingerprint,
			}

			certs = append(certs, data)
		}
	}

	return certs
}

func flattenInsightsConfig(insightsConfig *sqladmin.InsightsConfig) interface{} {
	data := map[string]interface{}{
		"query_insights_enabled":  insightsConfig.QueryInsightsEnabled,
		"query_string_length":     insightsConfig.QueryStringLength,
		"record_application_tags": insightsConfig.RecordApplicationTags,
		"record_client_address":   insightsConfig.RecordClientAddress,
		"query_plans_per_minute":  insightsConfig.QueryPlansPerMinute,
	}

	return []map[string]interface{}{data}
}

func flattenTargetMetrics(targetMetrics []*sqladmin.TargetMetric) []interface{} {
	if len(targetMetrics) == 0 { // Handles nil or empty slice
		return make([]interface{}, 0) // Explicitly return empty slice
	}

	metrics := make([]interface{}, len(targetMetrics)) // Pre-allocate for efficiency
	for i, metric := range targetMetrics {
		data := map[string]interface{}{
			"metric":       metric.Metric,
			"target_value": metric.TargetValue,
		}
		metrics[i] = data
	}
	return metrics
}

func flattenReadPoolAutoScaleConfig(readPoolAutoScaleConfig *sqladmin.ReadPoolAutoScaleConfig) interface{} {
	data := map[string]interface{}{
		"enabled":                    readPoolAutoScaleConfig.Enabled,
		"min_node_count":             readPoolAutoScaleConfig.MinNodeCount,
		"max_node_count":             readPoolAutoScaleConfig.MaxNodeCount,
		"target_metrics":             flattenTargetMetrics(readPoolAutoScaleConfig.TargetMetrics),
		"disable_scale_in":           readPoolAutoScaleConfig.DisableScaleIn,
		"scale_in_cooldown_seconds":  readPoolAutoScaleConfig.ScaleInCooldownSeconds,
		"scale_out_cooldown_seconds": readPoolAutoScaleConfig.ScaleOutCooldownSeconds,
	}

	return []map[string]interface{}{data}
}

func flattenPasswordValidationPolicy(passwordValidationPolicy *sqladmin.PasswordValidationPolicy) interface{} {
	data := map[string]interface{}{
		"min_length":                  passwordValidationPolicy.MinLength,
		"complexity":                  passwordValidationPolicy.Complexity,
		"reuse_interval":              passwordValidationPolicy.ReuseInterval,
		"disallow_username_substring": passwordValidationPolicy.DisallowUsernameSubstring,
		"password_change_interval":    passwordValidationPolicy.PasswordChangeInterval,
		"enable_password_policy":      passwordValidationPolicy.EnablePasswordPolicy,
	}
	return []map[string]interface{}{data}
}

func flattenEdition(v interface{}) string {
	if v == nil || tpgresource.IsEmptyValue(reflect.ValueOf(v)) {
		return "ENTERPRISE"
	}

	return v.(string)
}

func instanceMutexKey(project, instance_name string) string {
	return fmt.Sprintf("google-sql-database-instance-%s-%s", project, instance_name)
}

// sqlDatabaseIsMaster returns true if the provided schema.ResourceData represents a
// master SQL Instance, and false if it is a replica.
func sqlDatabaseIsMaster(d *schema.ResourceData) bool {
	_, ok := d.GetOk("master_instance_name")
	return !ok
}

func sqlDatabaseInstanceServiceNetworkPrecheck(d *schema.ResourceData, config *transport_tpg.Config, userAgent, network string) error {
	log.Printf("[DEBUG] checking network %q for at least one service networking connection", network)
	// This call requires projects.get permissions, which may not have been granted to the Terraform actor,
	// particularly in shared VPC setups. Most will! But it's not strictly required.
	serviceNetworkingNetworkName, err := servicenetworking.RetrieveServiceNetworkingNetworkName(d, config, network, userAgent)
	if err != nil {
		var gerr *googleapi.Error
		if errors.As(err, &gerr) {
			log.Printf("[DEBUG] retrieved googleapi error while creating sn name for %q. precheck skipped. code %v and message: %s", network, gerr.Code, gerr.Body)
			return nil
		}

		return err
	}

	response, err := config.NewServiceNetworkingClient(userAgent).Services.Connections.List("services/servicenetworking.googleapis.com").Network(serviceNetworkingNetworkName).Do()
	if err != nil {
		// It is possible that the actor creating the SQL Instance might not have permissions to call servicenetworking.services.connections.list
		log.Printf("[WARNING] Failed to list Service Networking of the project. Skipped Service Networking precheck.")
		return nil
	}

	if len(response.Connections) < 1 {
		return fmt.Errorf("Error, failed to create instance because the network doesn't have at least 1 private services connection. Please see https://cloud.google.com/sql/docs/mysql/private-ip#network_requirements for how to create this connection.")
	}

	return nil
}

func expandRestoreBackupContext(configured []interface{}) *sqladmin.RestoreBackupContext {
	if len(configured) == 0 || configured[0] == nil {
		return nil
	}

	_rc := configured[0].(map[string]interface{})
	return &sqladmin.RestoreBackupContext{
		BackupRunId: int64(_rc["backup_run_id"].(int)),
		InstanceId:  _rc["instance_id"].(string),
		Project:     _rc["project"].(string),
	}
}

func sqlDatabaseInstanceRestoreFromBackup(d *schema.ResourceData, config *transport_tpg.Config, userAgent, project, instanceId string, r interface{}, backupdrBackup interface{}) error {
	log.Printf("[DEBUG] Initiating SQL database instance backup restore")

	backupRequest := &sqladmin.InstancesRestoreBackupRequest{}

	if r != nil {
		restoreContext := r.([]interface{})
		backupRequest.RestoreBackupContext = expandRestoreBackupContext(restoreContext)
	} else if backupdrBackup != nil && backupdrBackup.(string) != "" {
		backupRequest.BackupdrBackup = backupdrBackup.(string)
	}

	var op *sqladmin.Operation
	err := transport_tpg.Retry(transport_tpg.RetryOptions{
		RetryFunc: func() (operr error) {
			op, operr = config.NewSqlAdminClient(userAgent).Instances.RestoreBackup(project, instanceId, backupRequest).Do()
			return operr
		},
		Timeout:              d.Timeout(schema.TimeoutUpdate),
		ErrorRetryPredicates: []transport_tpg.RetryErrorPredicateFunc{transport_tpg.IsSqlOperationInProgressError},
	})
	if err != nil {
		return fmt.Errorf("Error, failed to restore instance from backup %s: %s", instanceId, err)
	}

	err = SqlAdminOperationWaitTime(config, op, project, "Restore Backup", userAgent, d.Timeout(schema.TimeoutUpdate))
	if err != nil {
		return err
	}

	return nil
}

func expandPointInTimeRestoreContext(configured []interface{}) *sqladmin.PointInTimeRestoreContext {
	if len(configured) == 0 || configured[0] == nil {
		return nil
	}

	_pitrc := configured[0].(map[string]interface{})
	return &sqladmin.PointInTimeRestoreContext{
		PointInTime:      _pitrc["point_in_time"].(string),
		PreferredZone:    _pitrc["preferred_zone"].(string),
		Datasource:       _pitrc["datasource"].(string),
		AllocatedIpRange: _pitrc["allocated_ip_range"].(string),
		TargetInstance:   _pitrc["target_instance"].(string),
	}
}

func sqlDatabaseInstancePointInTimeRestore(d *schema.ResourceData, config *transport_tpg.Config, userAgent, project, instanceId string, r interface{}) error {
	log.Printf("[DEBUG] Initiating GCBDR managed SQL database instance backup point in time restore")
	pointInTimeRestoreContext := r.([]interface{})
	parent := fmt.Sprintf("projects/%s", project)

	var op *sqladmin.Operation
	err := transport_tpg.Retry(transport_tpg.RetryOptions{
		RetryFunc: func() (operr error) {
			op, operr = config.NewSqlAdminClient(userAgent).Instances.PointInTimeRestore(parent, expandPointInTimeRestoreContext(pointInTimeRestoreContext)).Do()
			return operr
		},
		Timeout:              d.Timeout(schema.TimeoutUpdate),
		ErrorRetryPredicates: []transport_tpg.RetryErrorPredicateFunc{transport_tpg.IsSqlOperationInProgressError},
	})
	if err != nil {
		return fmt.Errorf("Error, failed to point in restore an instance %s: %s", instanceId, err)
	}

	err = SqlAdminOperationWaitTime(config, op, project, "Point in Time Restore", userAgent, d.Timeout(schema.TimeoutUpdate))
	if err != nil {
		return err
	}

	return nil
}

func caseDiffDashSuppress(_, old, new string, _ *schema.ResourceData) bool {
	postReplaceNew := strings.Replace(new, "-", "_", -1)
	return strings.ToUpper(postReplaceNew) == strings.ToUpper(old)
}

func isMasterInstanceNameSet(_ context.Context, oldMasterInstanceName interface{}, newMasterInstanceName interface{}, _ interface{}) bool {
	new := newMasterInstanceName.(string)
	if new == "" {
		return false
	}

	return true
}

func isSwitchoverRequested(d *schema.ResourceData) bool {
	originalPrimaryName, _ := d.GetChange("master_instance_name")
	_, newReplicaNames := d.GetChange("replica_names")
	if !slices.Contains(newReplicaNames.([]interface{}), originalPrimaryName) {
		return false
	}
	return true
}

func isReplicaPromoteRequested(_ context.Context, oldInstanceType interface{}, newInstanceType interface{}, _ interface{}) bool {
	oldInstanceType = oldInstanceType.(string)
	newInstanceType = newInstanceType.(string)

	if newInstanceType == "CLOUD_SQL_INSTANCE" && oldInstanceType == "READ_REPLICA_INSTANCE" {
		return true
	}

	return false
}

// Check if this resource change is the manual update done on old primary after a switchover. If true, no replacement is needed.
func isSwitchoverFromOldPrimarySide(d *schema.ResourceDiff) bool {
	dbVersion := d.Get("database_version")
	oldInstanceType, newInstanceType := d.GetChange("instance_type")
	oldReplicaNames, newReplicaNames := d.GetChange("replica_names")
	_, newMasterInstanceName := d.GetChange("master_instance_name")
	_, newReplicaConfiguration := d.GetChange("replica_configuration")
	if len(newReplicaConfiguration.([]interface{})) != 1 || newReplicaConfiguration.([]interface{})[0] == nil {
		return false
	}
	replicaConfiguration := newReplicaConfiguration.([]interface{})[0].(map[string]interface{})
	cascadableReplica, cascadableReplicaFieldExists := replicaConfiguration["cascadable_replica"]

	instanceTypeChangedFromPrimaryToReplica := oldInstanceType.(string) == "CLOUD_SQL_INSTANCE" && newInstanceType.(string) == "READ_REPLICA_INSTANCE"
	newMasterInOldReplicaNames := slices.Contains(oldReplicaNames.([]interface{}), newMasterInstanceName)
	newMasterNotInNewReplicaNames := !slices.Contains(newReplicaNames.([]interface{}), newMasterInstanceName)
	isCascadableReplica := cascadableReplicaFieldExists && cascadableReplica.(bool)
	isSQLServer := strings.HasPrefix(dbVersion.(string), "SQLSERVER")

	return newMasterInstanceName != nil &&
		instanceTypeChangedFromPrimaryToReplica &&
		newMasterInOldReplicaNames && newMasterNotInNewReplicaNames && (!isSQLServer ||
		isCascadableReplica)
}

func checkPromoteConfigurations(d *schema.ResourceData) error {
	masterInstanceName := d.GetRawConfig().GetAttr("master_instance_name")
	replicaConfiguration := d.GetRawConfig().GetAttr("replica_configuration").AsValueSlice()

	return validatePromoteConfigurations(masterInstanceName, replicaConfiguration)
}

func checkPromoteConfigurationsAndUpdateDiff(_ context.Context, diff *schema.ResourceDiff, _ interface{}) error {
	masterInstanceName := diff.GetRawConfig().GetAttr("master_instance_name")
	replicaConfiguration := diff.GetRawConfig().GetAttr("replica_configuration").AsValueSlice()

	err := validatePromoteConfigurations(masterInstanceName, replicaConfiguration)
	if err != nil {
		return err
	}

	err = diff.SetNew("master_instance_name", nil)
	if err != nil {
		return err
	}

	err = diff.SetNew("replica_configuration", nil)
	if err != nil {
		return err
	}
	return nil
}

func validatePromoteConfigurations(masterInstanceName cty.Value, replicaConfigurations []cty.Value) error {
	if !masterInstanceName.IsNull() {
		return fmt.Errorf("Replica promote configuration check failed. Please remove master_instance_name and try again.")
	}

	if len(replicaConfigurations) != 0 {
		return fmt.Errorf("Replica promote configuration check failed. Please remove replica_configuration and try again.")
	}
	return nil
}

// areDRReplicaNamesEqual compares 2 DR replica names.
func areDRReplicaNamesEqual(_, oldValue, newValue string, d *schema.ResourceData) bool {
	project := d.Get("project").(string)
	// When we can't find project, just fallback to naive comparison.
	if project == "" {
		return oldValue == newValue
	}
	return normalizeDRReplicaName(oldValue, project) == normalizeDRReplicaName(newValue, project)
}

// normalizeDRReplicaName normalizes DR replica name by prefixing project name when it's missing.
func normalizeDRReplicaName(drReplicaName, project string) string {
	if strings.Contains(drReplicaName, ":") {
		return drReplicaName
	}
	return fmt.Sprintf("%s:%s", project, drReplicaName)
}