// resourceAwsLambdaEventSourceMappingUpdate maps to:
// UpdateEventSourceMapping in the API / SDK
func resourceAwsLambdaEventSourceMappingUpdate(d *schema.ResourceData, meta interface{}) error {
	conn := meta.(*AWSClient).lambdaconn

	log.Printf("[DEBUG] Updating Lambda event source mapping: %s", d.Id())

	params := &lambda.UpdateEventSourceMappingInput{
		UUID:         aws.String(d.Id()),
		BatchSize:    aws.Int64(int64(d.Get("batch_size").(int))),
		FunctionName: aws.String(d.Get("function_name").(string)),
		Enabled:      aws.Bool(d.Get("enabled").(bool)),
	}

	err := resource.Retry(1*time.Minute, func() *resource.RetryError {
		_, err := conn.UpdateEventSourceMapping(params)
		if err != nil {
			if awserr, ok := err.(awserr.Error); ok {
				if awserr.Code() == "InvalidParameterValueException" {
					return resource.RetryableError(awserr)
				}
			}
			return resource.NonRetryableError(err)
		}
		return nil
	})

	if err != nil {
		return fmt.Errorf("Error updating Lambda event source mapping: %s", err)
	}

	return resourceAwsLambdaEventSourceMappingRead(d, meta)
}

func resourceAwsGlacierVaultRead(d *schema.ResourceData, meta interface{}) error {
	glacierconn := meta.(*AWSClient).glacierconn

	input := &glacier.DescribeVaultInput{
		VaultName: aws.String(d.Id()),
	}

	out, err := glacierconn.DescribeVault(input)
	if err != nil {
		return fmt.Errorf("Error reading Glacier Vault: %s", err.Error())
	}

	awsClient := meta.(*AWSClient)
	d.Set("name", out.VaultName)
	d.Set("arn", out.VaultARN)

	location, err := buildGlacierVaultLocation(awsClient.accountid, d.Id())
	if err != nil {
		return err
	}
	d.Set("location", location)

	tags, err := getGlacierVaultTags(glacierconn, d.Id())
	if err != nil {
		return err
	}
	d.Set("tags", tags)

	log.Printf("[DEBUG] Getting the access_policy for Vault %s", d.Id())
	pol, err := glacierconn.GetVaultAccessPolicy(&glacier.GetVaultAccessPolicyInput{
		VaultName: aws.String(d.Id()),
	})

	if awserr, ok := err.(awserr.Error); ok && awserr.Code() == "ResourceNotFoundException" {
		d.Set("access_policy", "")
	} else if pol != nil {
		policy, err := normalizeJsonString(*pol.Policy.Policy)
		if err != nil {
			return errwrap.Wrapf("access policy contains an invalid JSON: {{err}}", err)
		}
		d.Set("access_policy", policy)
	} else {
		return err
	}

	notifications, err := getGlacierVaultNotification(glacierconn, d.Id())
	if awserr, ok := err.(awserr.Error); ok && awserr.Code() == "ResourceNotFoundException" {
		d.Set("notification", "")
	} else if pol != nil {
		d.Set("notification", notifications)
	} else {
		return err
	}

	return nil
}

func getARNForQueueURL(config *aws.Config, queueURL string) (string, error) {
	svc := sqs.New(config)

	arnKey := "QueueArn"

	params := &sqs.GetQueueAttributesInput{
		QueueUrl: aws.String(string(queueURL)),
		AttributeNames: []*string{
			aws.String(arnKey),
		},
	}

	resp, err := svc.GetQueueAttributes(params)
	if awserr, ok := err.(awserr.Error); ok {
		return "", fmt.Errorf("aws error while getting ARN for SQS queue: %v %v", awserr.Code(), awserr.Message())
	} else if err != nil {
		return "", fmt.Errorf("error while getting ARN for SQS queue: %v", err)
	} else if resp == nil || resp.Attributes == nil {
		return "", nil
	}

	atts := resp.Attributes

	arnPtr, ok := atts[arnKey]
	if !ok || arnPtr == nil {
		return "", nil
	}

	return *arnPtr, nil
}

func OpsworksInstanceStateRefreshFunc(conn *opsworks.OpsWorks, instanceID string) resource.StateRefreshFunc {
	return func() (interface{}, string, error) {
		resp, err := conn.DescribeInstances(&opsworks.DescribeInstancesInput{
			InstanceIds: []*string{aws.String(instanceID)},
		})
		if err != nil {
			if awserr, ok := err.(awserr.Error); ok && awserr.Code() == "ResourceNotFoundException" {
				// Set this to nil as if we didn't find anything.
				resp = nil
			} else {
				log.Printf("Error on OpsworksInstanceStateRefresh: %s", err)
				return nil, "", err
			}
		}

		if resp == nil || len(resp.Instances) == 0 {
			// Sometimes AWS just has consistency issues and doesn't see
			// our instance yet. Return an empty state.
			return nil, "", nil
		}

		i := resp.Instances[0]
		return i, *i.Status, nil
	}
}

func findExistingTopicARN(config *aws.Config, topic string) (string, error) {
	svc := sns.New(nil)

	params := &sns.ListTopicsInput{
		NextToken: nil,
	}

	for {
		resp, err := svc.ListTopics(params)
		if awserr, ok := err.(awserr.Error); ok {
			return "", fmt.Errorf("aws error while listing SNS topics: %v %v", awserr.Code(), awserr.Message())
		} else if err != nil {
			return "", fmt.Errorf("error while listing SNS topics: %v", err)
		} else if resp == nil || resp.Topics == nil {
			break
		}

		for _, topicPtr := range resp.Topics {
			if topicPtr != nil && topicPtr.TopicArn != nil && strings.HasSuffix(*topicPtr.TopicArn, topic) {
				return *topicPtr.TopicArn, nil
			}
		}

		if resp.NextToken != nil {
			params.NextToken = resp.NextToken
		} else {
			break
		}
	}

	return "", nil
}

func testAccCheckAwsOpsworksRdsDbDestroy(s *terraform.State) error {
	client := testAccProvider.Meta().(*AWSClient).opsworksconn

	for _, rs := range s.RootModule().Resources {
		if rs.Type != "aws_opsworks_rds_db_instance" {
			continue
		}

		req := &opsworks.DescribeRdsDbInstancesInput{
			StackId: aws.String(rs.Primary.Attributes["stack_id"]),
		}

		resp, err := client.DescribeRdsDbInstances(req)
		if err == nil {
			if len(resp.RdsDbInstances) > 0 {
				return fmt.Errorf("OpsWorks Rds db instances  still exist.")
			}
		}

		if awserr, ok := err.(awserr.Error); ok {
			if awserr.Code() != "ResourceNotFoundException" {
				return err
			}
		}
	}
	return nil
}

func testAccCheckAWSEcsServiceDestroy(s *terraform.State) error {
	conn := testAccProvider.Meta().(*AWSClient).ecsconn

	for _, rs := range s.RootModule().Resources {
		if rs.Type != "aws_ecs_service" {
			continue
		}

		out, err := conn.DescribeServices(&ecs.DescribeServicesInput{
			Services: []*string{aws.String(rs.Primary.ID)},
		})

		if awserr, ok := err.(awserr.Error); ok && awserr.Code() == "ClusterNotFoundException" {
			continue
		}

		if err == nil {
			if len(out.Services) > 0 {
				return fmt.Errorf("ECS service still exists:\n%#v", out.Services)
			}
		}

		return err
	}

	return nil
}

func testAccCheckAwsOpsworksApplicationDestroy(s *terraform.State) error {
	client := testAccProvider.Meta().(*AWSClient).opsworksconn

	for _, rs := range s.RootModule().Resources {
		if rs.Type != "aws_opsworks_application" {
			continue
		}

		req := &opsworks.DescribeAppsInput{
			AppIds: []*string{
				aws.String(rs.Primary.ID),
			},
		}

		resp, err := client.DescribeApps(req)
		if err == nil {
			if len(resp.Apps) > 0 {
				return fmt.Errorf("OpsWorks App still exist.")
			}
		}

		if awserr, ok := err.(awserr.Error); ok {
			if awserr.Code() != "ResourceNotFoundException" {
				return err
			}
		}
	}

	return nil
}

func setPolicy(key, secret, region, queueARN, queueURL string, topicARNs []string) error {
	svc := sqs.New(getConfig(key, secret, region))

	bs, err := json.Marshal(topicARNs)
	if err != nil {
		return fmt.Errorf("error while creating policy for SQS queue: %v", err)
	}

	policy := fmt.Sprintf(policyFormat, queueARN, string(bs))

	params := &sqs.SetQueueAttributesInput{
		Attributes: map[string]*string{
			"Policy": aws.String(policy),
		},
		QueueUrl: aws.String(queueURL),
	}

	_, err = svc.SetQueueAttributes(params)
	if awserr, ok := err.(awserr.Error); ok {
		return fmt.Errorf("aws error while setting policy for SQS queue: %v %v", awserr.Code(), awserr.Message())
	} else if err != nil {
		return fmt.Errorf("error while setting policy for SQS queue: %v", err)
	}

	return nil
}

func testAccCheckVaultNotificationsMissing(name string) resource.TestCheckFunc {
	return func(s *terraform.State) error {
		rs, ok := s.RootModule().Resources[name]
		if !ok {
			return fmt.Errorf("Not found: %s", name)
		}

		if rs.Primary.ID == "" {
			return fmt.Errorf("No ID is set")
		}

		glacierconn := testAccProvider.Meta().(*AWSClient).glacierconn
		out, err := glacierconn.GetVaultNotifications(&glacier.GetVaultNotificationsInput{
			VaultName: aws.String(rs.Primary.ID),
		})

		if awserr, ok := err.(awserr.Error); ok && awserr.Code() != "ResourceNotFoundException" {
			return fmt.Errorf("Expected ResourceNotFoundException for Vault %s Notification Block but got %s", rs.Primary.ID, awserr.Code())
		}

		if out.VaultNotificationConfig != nil {
			return fmt.Errorf("Vault Notification Block has been found for %s", rs.Primary.ID)
		}

		return nil
	}

}

func testAccCheckAwsOpsworksStackDestroy(s *terraform.State) error {
	opsworksconn := testAccProvider.Meta().(*AWSClient).opsworksconn
	for _, rs := range s.RootModule().Resources {
		if rs.Type != "aws_opsworks_stack" {
			continue
		}

		req := &opsworks.DescribeStacksInput{
			StackIds: []*string{
				aws.String(rs.Primary.ID),
			},
		}

		_, err := opsworksconn.DescribeStacks(req)
		if err != nil {
			if awserr, ok := err.(awserr.Error); ok {
				if awserr.Code() == "ResourceNotFoundException" {
					// not found, all good
					return nil
				}
			}
			return err
		}
	}
	return fmt.Errorf("Fall through error for OpsWorks stack test")
}

// Log submits a batch of logs to the LogStream.
func (s *LogStream) Log(logs []*cloudwatchlogs.InputLogEvent) error {
	params := &cloudwatchlogs.PutLogEventsInput{
		LogEvents:     logs,
		LogGroupName:  s.Group,
		LogStreamName: s.Stream,
		SequenceToken: s.Token,
	}

	resp, err := s.service.PutLogEvents(params)
	awserr, _ := err.(awserr.Error)

	if awserr != nil {
		switch awserr.Code() {
		case "InvalidSequenceTokenException":
			log.Infof("Retrying log upload with new token - length %d, error, %v", len(logs), err)
			return s.retryBatchWithNewToken(logs)
		default:
			log.Errorf("Log upload failed - length: %d, error: %v", len(logs), err)
			return awserr
		}
	}

	if resp.RejectedLogEventsInfo != nil {
		log.Warnf("Log upload succeeded with rejected events - length: %d", len(logs))
	} else {
		log.Debugf("Log upload succeeded - length: %d", len(logs))
	}

	s.Token = resp.NextSequenceToken

	return nil
}

func resourceAwsAutoscalingGroupDelete(d *schema.ResourceData, meta interface{}) error {
	conn := meta.(*AWSClient).autoscalingconn

	// Read the autoscaling group first. If it doesn't exist, we're done.
	// We need the group in order to check if there are instances attached.
	// If so, we need to remove those first.
	g, err := getAwsAutoscalingGroup(d.Id(), conn)
	if err != nil {
		return err
	}
	if g == nil {
		log.Printf("[INFO] Autoscaling Group %q not found", d.Id())
		d.SetId("")
		return nil
	}
	if len(g.Instances) > 0 || *g.DesiredCapacity > 0 {
		if err := resourceAwsAutoscalingGroupDrain(d, meta); err != nil {
			return err
		}
	}

	log.Printf("[DEBUG] AutoScaling Group destroy: %v", d.Id())
	deleteopts := autoscaling.DeleteAutoScalingGroupInput{
		AutoScalingGroupName: aws.String(d.Id()),
		ForceDelete:          aws.Bool(d.Get("force_delete").(bool)),
	}

	// We retry the delete operation to handle InUse/InProgress errors coming
	// from scaling operations. We should be able to sneak in a delete in between
	// scaling operations within 5m.
	err = resource.Retry(5*time.Minute, func() *resource.RetryError {
		if _, err := conn.DeleteAutoScalingGroup(&deleteopts); err != nil {
			if awserr, ok := err.(awserr.Error); ok {
				switch awserr.Code() {
				case "InvalidGroup.NotFound":
					// Already gone? Sure!
					return nil
				case "ResourceInUse", "ScalingActivityInProgress":
					// These are retryable
					return resource.RetryableError(awserr)
				}
			}
			// Didn't recognize the error, so shouldn't retry.
			return resource.NonRetryableError(err)
		}
		// Successful delete
		return nil
	})
	if err != nil {
		return err
	}

	return resource.Retry(5*time.Minute, func() *resource.RetryError {
		if g, _ = getAwsAutoscalingGroup(d.Id(), conn); g != nil {
			return resource.RetryableError(
				fmt.Errorf("Auto Scaling Group still exists"))
		}
		return nil
	})
}

// Delete functions.
func (p *Project) Delete(names []string) error {
	p.Log.Debugf("deleting %d functions", len(names))

	for _, name := range names {
		fn, err := p.FunctionByName(name)

		if err == ErrNotFound {
			p.Log.Warnf("function %q does not exist in project", name)
			continue
		}

		if _, err := fn.GetConfig(); err != nil {
			if awserr, ok := err.(awserr.Error); ok && awserr.Code() == "ResourceNotFoundException" {
				p.Log.Infof("function %q hasn't been deployed yet or has been deleted manually on AWS Lambda", name)
				continue
			}
			return err
		}

		if err := fn.Delete(); err != nil {
			return err
		}
	}

	return nil
}

// outputList format.
func outputList() {
	fmt.Println()
	for _, fn := range root.Project.Functions {
		config, err := fn.GetConfigCurrent()

		if awserr, ok := err.(awserr.Error); ok && awserr.Code() == "ResourceNotFoundException" {
			fmt.Printf("  \033[%dm%s\033[0m (not deployed) \n", colors.Blue, fn.Name)
		} else {
			fmt.Printf("  \033[%dm%s\033[0m\n", colors.Blue, fn.Name)
		}

		if fn.Description != "" {
			fmt.Printf("    description: %v\n", fn.Description)
		}
		fmt.Printf("    runtime: %v\n", fn.Runtime)
		fmt.Printf("    memory: %vmb\n", fn.Memory)
		fmt.Printf("    timeout: %vs\n", fn.Timeout)
		fmt.Printf("    role: %v\n", fn.Role)
		fmt.Printf("    handler: %v\n", fn.Handler)

		if err != nil {
			fmt.Println()
			continue // ignore
		}

		fmt.Printf("    current version: %s\n", *config.Configuration.Version)

		fmt.Println()
	}
}

func resourceAwsS3BucketPolicyPut(d *schema.ResourceData, meta interface{}) error {
	s3conn := meta.(*AWSClient).s3conn

	bucket := d.Get("bucket").(string)
	policy := d.Get("policy").(string)

	d.SetId(bucket)

	log.Printf("[DEBUG] S3 bucket: %s, put policy: %s", bucket, policy)

	params := &s3.PutBucketPolicyInput{
		Bucket: aws.String(bucket),
		Policy: aws.String(policy),
	}

	err := resource.Retry(1*time.Minute, func() *resource.RetryError {
		if _, err := s3conn.PutBucketPolicy(params); err != nil {
			if awserr, ok := err.(awserr.Error); ok {
				if awserr.Code() == "MalformedPolicy" {
					return resource.RetryableError(awserr)
				}
			}
			return resource.NonRetryableError(err)
		}
		return nil
	})

	if err != nil {
		return fmt.Errorf("Error putting S3 policy: %s", err)
	}

	return nil
}

func resourceAwsOpsworksUserProfileRead(d *schema.ResourceData, meta interface{}) error {
	client := meta.(*AWSClient).opsworksconn

	req := &opsworks.DescribeUserProfilesInput{
		IamUserArns: []*string{
			aws.String(d.Id()),
		},
	}

	log.Printf("[DEBUG] Reading OpsWorks user profile: %s", d.Id())

	resp, err := client.DescribeUserProfiles(req)
	if err != nil {
		if awserr, ok := err.(awserr.Error); ok {
			if awserr.Code() == "ResourceNotFoundException" {
				log.Printf("[DEBUG] OpsWorks user profile (%s) not found", d.Id())
				d.SetId("")
				return nil
			}
		}
		return err
	}

	for _, profile := range resp.UserProfiles {
		d.Set("allow_self_management", profile.AllowSelfManagement)
		d.Set("user_arn", profile.IamUserArn)
		d.Set("ssh_public_key", profile.SshPublicKey)
		d.Set("ssh_username", profile.SshUsername)
		break
	}

	return nil
}

func testAccCheckAwsOpsworksUserProfileDestroy(s *terraform.State) error {
	client := testAccProvider.Meta().(*AWSClient).opsworksconn

	for _, rs := range s.RootModule().Resources {
		if rs.Type != "aws_opsworks_user_profile" {
			continue
		}

		req := &opsworks.DescribeUserProfilesInput{
			IamUserArns: []*string{aws.String(rs.Primary.Attributes["user_arn"])},
		}
		resp, err := client.DescribeUserProfiles(req)

		if err == nil {
			if len(resp.UserProfiles) > 0 {
				return fmt.Errorf("OpsWorks User Profiles still exist.")
			}
		}

		if awserr, ok := err.(awserr.Error); ok {
			if awserr.Code() != "ResourceNotFoundException" {
				return err
			}
		}
	}
	return nil
}

func resourceAwsOpsworksStackRead(d *schema.ResourceData, meta interface{}) error {
	client := meta.(*AWSClient).opsworksconn

	req := &opsworks.DescribeStacksInput{
		StackIds: []*string{
			aws.String(d.Id()),
		},
	}

	log.Printf("[DEBUG] Reading OpsWorks stack: %s", d.Id())

	resp, err := client.DescribeStacks(req)
	if err != nil {
		if awserr, ok := err.(awserr.Error); ok {
			if awserr.Code() == "ResourceNotFoundException" {
				log.Printf("[DEBUG] OpsWorks stack (%s) not found", d.Id())
				d.SetId("")
				return nil
			}
		}
		return err
	}

	stack := resp.Stacks[0]
	d.Set("agent_version", stack.AgentVersion)
	d.Set("name", stack.Name)
	d.Set("region", stack.Region)
	d.Set("default_instance_profile_arn", stack.DefaultInstanceProfileArn)
	d.Set("service_role_arn", stack.ServiceRoleArn)
	d.Set("default_availability_zone", stack.DefaultAvailabilityZone)
	d.Set("default_os", stack.DefaultOs)
	d.Set("default_root_device_type", stack.DefaultRootDeviceType)
	d.Set("default_ssh_key_name", stack.DefaultSshKeyName)
	d.Set("default_subnet_id", stack.DefaultSubnetId)
	d.Set("hostname_theme", stack.HostnameTheme)
	d.Set("use_custom_cookbooks", stack.UseCustomCookbooks)
	if stack.CustomJson != nil {
		d.Set("custom_json", stack.CustomJson)
	}
	d.Set("use_opsworks_security_groups", stack.UseOpsworksSecurityGroups)
	d.Set("vpc_id", stack.VpcId)
	if color, ok := stack.Attributes["Color"]; ok {
		d.Set("color", color)
	}
	if stack.ConfigurationManager != nil {
		d.Set("configuration_manager_name", stack.ConfigurationManager.Name)
		d.Set("configuration_manager_version", stack.ConfigurationManager.Version)
	}
	if stack.ChefConfiguration != nil {
		d.Set("berkshelf_version", stack.ChefConfiguration.BerkshelfVersion)
		d.Set("manage_berkshelf", stack.ChefConfiguration.ManageBerkshelf)
	}
	resourceAwsOpsworksSetStackCustomCookbooksSource(d, stack.CustomCookbooksSource)

	return nil
}

func testDecryptPasswordAndTest(nProfile, nAccessKey, key string) resource.TestCheckFunc {
	return func(s *terraform.State) error {
		profileResource, ok := s.RootModule().Resources[nProfile]
		if !ok {
			return fmt.Errorf("Not found: %s", nProfile)
		}

		password, ok := profileResource.Primary.Attributes["encrypted_password"]
		if !ok {
			return errors.New("No password in state")
		}

		accessKeyResource, ok := s.RootModule().Resources[nAccessKey]
		if !ok {
			return fmt.Errorf("Not found: %s", nAccessKey)
		}

		accessKeyId := accessKeyResource.Primary.ID
		secretAccessKey, ok := accessKeyResource.Primary.Attributes["secret"]
		if !ok {
			return errors.New("No secret access key in state")
		}

		decryptedPassword, err := pgpkeys.DecryptBytes(password, key)
		if err != nil {
			return fmt.Errorf("Error decrypting password: %s", err)
		}

		iamAsCreatedUserSession := session.New(&aws.Config{
			Region:      aws.String("us-west-2"),
			Credentials: credentials.NewStaticCredentials(accessKeyId, secretAccessKey, ""),
		})
		_, err = iamAsCreatedUserSession.Config.Credentials.Get()
		if err != nil {
			return fmt.Errorf("Error getting session credentials: %s", err)
		}

		return resource.Retry(2*time.Minute, func() *resource.RetryError {
			iamAsCreatedUser := iam.New(iamAsCreatedUserSession)
			_, err = iamAsCreatedUser.ChangePassword(&iam.ChangePasswordInput{
				OldPassword: aws.String(decryptedPassword.String()),
				NewPassword: aws.String(generatePassword(20)),
			})
			if err != nil {
				if awserr, ok := err.(awserr.Error); ok && awserr.Code() == "InvalidClientTokenId" {
					return resource.RetryableError(err)
				}

				return resource.NonRetryableError(fmt.Errorf("Error changing decrypted password: %s", err))
			}

			return nil
		})
	}
}