// GetSessions returns the current session from the SessionProvider.
func (dsp *ConfigurableSessionProvider) GetSession() *session.Session {
	dsp.sessionMutex.Lock()
	defer dsp.sessionMutex.Unlock()

	if dsp._session != nil {
		return dsp._session
	}

	cfgs := []*aws.Config{
		aws.NewConfig().WithRegion(dsp.ConfigRegion),
	}

	if dsp.ConfigAccessKey != "" {
		cfgs = append(cfgs, aws.NewConfig().WithCredentials(credentials.NewStaticCredentials(
			dsp.ConfigAccessKey,
			dsp.ConfigSecretKey,
			dsp.ConfigSessionToken,
		)))
	} else if !dsp.ConfigDisableENVCredentials {
		// NOTE: We may want to init all credential providers in one config, as they might overwrite each other
		cfgs = append(cfgs, aws.NewConfig().WithCredentials(credentials.NewCredentials(&credentials.EnvProvider{})))
	} else {
		panic("no valid configuration parameters for aws credentials found.")
	}

	dsp._session = session.New(cfgs...)
	return dsp._session
}

func (u *User) Run(args []string) int {
	cli := iam.New(session.New(aws.NewConfig().WithRegion(u.config.Region)))

	u.ListUsers(cli)

	return 0
}

// Start a test
func (t *Test) Start() <-chan queue.RegionsAggData {
	awsConfig := aws.NewConfig().WithRegion(t.config.Regions[0])

	if t.config.AwsProfile != "" {
		creds := credentials.NewSharedCredentials("", t.config.AwsProfile)
		if _, err := creds.Get(); err != nil {
			log.Fatal(err)
		}
		awsConfig.WithCredentials(creds)
	}

	infra, err := infrastructure.New(t.config.Regions, awsConfig)
	if err != nil {
		log.Fatal(err)
	}

	t.invokeLambdas(awsConfig, infra.QueueURL())

	results := make(chan queue.RegionsAggData)

	go func() {
		for result := range queue.Aggregate(awsConfig, infra.QueueURL(), t.config.TotalRequests) {
			results <- result
		}
		infra.Clean()
		close(results)
	}()

	return results
}

func ListBuckets(region string) {
	config := aws.NewConfig().WithRegion(region)
	sess := session.New(config)
	svc := s3.New(sess)

	var params *s3.ListBucketsInput
	resp, err := svc.ListBuckets(params)

	if err != nil {
		if awsErr, ok := err.(awserr.Error); ok {
			// Generic AWS error with Code, Message, and original error (if any)
			fmt.Println(awsErr.Code(), awsErr.Message(), awsErr.OrigErr())
			if reqErr, ok := err.(awserr.RequestFailure); ok {
				// A service error occurred
				fmt.Println(reqErr.Code(), reqErr.Message(), reqErr.StatusCode(), reqErr.RequestID())
			}
		} else {
			// This case should never be hit, the SDK should always return an
			// error which satisfies the awserr.Error interface.
			fmt.Println(err.Error())
		}
	}

	// Pretty-print the response data.
	fmt.Println(awsutil.Prettify(resp))
}

func listS3(wg *sync.WaitGroup, index int, results chan<- []string) {

	var creds *credentials.Credentials
	if *accessFlag != "" && *secretFlag != "" {
		creds = credentials.NewStaticCredentials(*accessFlag, *secretFlag, "")
	} else {
		creds = credentials.AnonymousCredentials
	}
	sess := session.New(aws.NewConfig().WithCredentials(creds).WithRegion(*regionFlag).WithEndpoint(*endpointFlag).WithS3ForcePathStyle(true))

	prefix := fmt.Sprintf("%s%x", *prefixFlag, index)

	svc := s3.New(sess)
	inputparams := &s3.ListObjectsInput{
		Bucket: aws.String(*bucketFlag),
		Prefix: aws.String(prefix),
	}

	result := make([]string, 0, 1000)

	svc.ListObjectsPages(inputparams, func(page *s3.ListObjectsOutput, lastPage bool) bool {

		for _, value := range page.Contents {
			result = append(result, *value.Key)
		}

		if lastPage {
			results <- result
			wg.Done()
			return false
		} else {
			return true
		}
	})
}

// CreateRecord for hosted zone id
func CreateRecord(zone string, record string, dest string) error {
	svc := route53.New(session.New(), aws.NewConfig())
	_, err := svc.ChangeResourceRecordSets(&route53.ChangeResourceRecordSetsInput{
		ChangeBatch: &route53.ChangeBatch{
			Changes: []*route53.Change{
				{
					Action: aws.String("CREATE"),
					ResourceRecordSet: &route53.ResourceRecordSet{
						Name: aws.String(record),
						Type: aws.String("A"),
						ResourceRecords: []*route53.ResourceRecord{
							{
								Value: aws.String(dest),
							},
						},
						TTL: aws.Int64(60),
					},
				},
			},
			Comment: aws.String("ResourceDescription"),
		},
		HostedZoneId: aws.String(zone),
	})

	return err
}

func init() {
	gucumber.Before("@efs", func() {
		// FIXME remove custom region
		gucumber.World["client"] = efs.New(smoke.Session,
			aws.NewConfig().WithRegion("us-west-2"))
	})
}

// PreRun sets up global tasks used for most commands, some use PreRunNoop
// to remove this default behaviour.
func preRun(c *cobra.Command, args []string) error {
	if l, err := log.ParseLevel(logLevel); err == nil {
		log.SetLevel(l)
	}

	config := aws.NewConfig()

	if profile != "" {
		config = config.WithCredentials(credentials.NewSharedCredentials("", profile))
	}

	Session = session.New(config)

	Project = &project.Project{
		Log:  log.Log,
		Path: ".",
	}

	if dryRun {
		log.SetLevel(log.WarnLevel)
		Project.Service = dryrun.New(Session)
		Project.Concurrency = 1
	} else {
		Project.Service = lambda.New(Session)
	}

	if chdir != "" {
		if err := os.Chdir(chdir); err != nil {
			return err
		}
	}

	return Project.Open()
}

func createConf() *aws.Config {
	conf := aws.NewConfig()

	if config.Log.LogDebug == config.Log_Flag_OFF {
		conf.WithLogLevel(aws.LogOff)
	} else {

		loglevel := aws.LogDebug

		if config.Log.LogSigning == config.Log_Flag_ON {
			loglevel |= aws.LogDebugWithSigning
		}

		if config.Log.LogHTTPBody == config.Log_Flag_ON {
			loglevel |= aws.LogDebugWithHTTPBody
		}

		if config.Log.LogRequestRetries == config.Log_Flag_ON {
			loglevel |= aws.LogDebugWithRequestRetries
		}

		if config.Log.LogRequestErrors == config.Log_Flag_ON {
			loglevel |= aws.LogDebugWithRequestErrors
		}

		conf.WithLogLevel(loglevel)
	}
	return conf
}

func (pv *PersistentValues) preRun(c *cobra.Command, args []string) {
	if l, err := log.ParseLevel(pv.LogLevel); err == nil {
		log.SetLevel(l)
	}

	pv.session = session.New(aws.NewConfig())

	pv.project = &project.Project{
		Log:  log.Log,
		Path: ".",
	}

	if pv.DryRun {
		log.SetLevel(log.WarnLevel)
		pv.project.Service = dryrun.New(pv.session)
		pv.project.Concurrency = 1
	} else {
		pv.project.Service = lambda.New(pv.session)
	}

	if pv.Chdir != "" {
		if err := os.Chdir(pv.Chdir); err != nil {
			log.Fatalf("error: %s", err)
		}
	}

	if err := pv.project.Open(); err != nil {
		log.Fatalf("error opening project: %s", err)
	}
}

func NewAwsAsgEbs(maxRetries int) *AwsAsgEbs {
	awsAsgEbs := &AwsAsgEbs{}

	metadata := ec2metadata.New(session.New())

	region, err := metadata.Region()
	if err != nil {
		log.WithFields(log.Fields{"error": err}).Fatal("Failed to get region from instance metadata")
	}
	log.WithFields(log.Fields{"region": region}).Info("Setting region")
	awsAsgEbs.Region = region

	availabilityZone, err := metadata.GetMetadata("placement/availability-zone")
	if err != nil {
		log.WithFields(log.Fields{"error": err}).Fatal("Failed to get availability zone from instance metadata")
	}
	log.WithFields(log.Fields{"az": availabilityZone}).Info("Setting availability zone")
	awsAsgEbs.AvailabilityZone = availabilityZone

	instanceId, err := metadata.GetMetadata("instance-id")
	if err != nil {
		log.WithFields(log.Fields{"error": err}).Fatal("Failed to get instance id from instance metadata")
	}
	log.WithFields(log.Fields{"instance_id": instanceId}).Info("Setting instance id")
	awsAsgEbs.InstanceId = instanceId

	awsAsgEbs.AwsConfig = aws.NewConfig().
		WithRegion(region).
		WithCredentials(ec2rolecreds.NewCredentials(session.New())).
		WithMaxRetries(maxRetries)

	return awsAsgEbs
}

func (r *Route53Provider) Init(rootDomainName string) error {
	var region, accessKey, secretKey string
	if region = os.Getenv("AWS_REGION"); len(region) == 0 {
		return fmt.Errorf("AWS_REGION is not set")
	}

	if accessKey = os.Getenv("AWS_ACCESS_KEY"); len(accessKey) == 0 {
		return fmt.Errorf("AWS_ACCESS_KEY is not set")
	}

	if secretKey = os.Getenv("AWS_SECRET_KEY"); len(secretKey) == 0 {
		return fmt.Errorf("AWS_SECRET_KEY is not set")
	}

	// Comply with the API's 5 req/s rate limit. If there are other
	// clients using the same account the AWS SDK will throttle the
	// requests automatically if the global rate limit is exhausted.
	r.limiter = ratelimit.NewBucketWithRate(5.0, 1)

	creds := credentials.NewStaticCredentials(accessKey, secretKey, "")
	config := aws.NewConfig().WithMaxRetries(route53MaxRetries).
		WithCredentials(creds).
		WithRegion(region)

	r.client = awsRoute53.New(session.New(config))

	if err := r.setHostedZone(rootDomainName); err != nil {
		return err
	}

	logrus.Infof("Configured %s with hosted zone %s in region %s",
		r.GetName(), rootDomainName, region)

	return nil
}

// presignURL will presign the request by using SoureRegion to sign with. SourceRegion is not
// sent to the service, and is only used to not have the SDKs parsing ARNs.
func presignURL(r *request.Request, sourceRegion *string, newParams interface{}) *string {
	cfg := r.Config.Copy(aws.NewConfig().
		WithEndpoint("").
		WithRegion(aws.StringValue(sourceRegion)))

	clientInfo := r.ClientInfo
	resolved, err := r.Config.EndpointResolver.EndpointFor(
		clientInfo.ServiceName, aws.StringValue(cfg.Region),
		func(opt *endpoints.Options) {
			opt.DisableSSL = aws.BoolValue(cfg.DisableSSL)
			opt.UseDualStack = aws.BoolValue(cfg.UseDualStack)
		},
	)
	if err != nil {
		r.Error = err
		return nil
	}

	clientInfo.Endpoint = resolved.URL
	clientInfo.SigningRegion = resolved.SigningRegion

	// Presign a request with modified params
	req := request.New(*cfg, clientInfo, r.Handlers, r.Retryer, r.Operation, newParams, r.Data)
	req.Operation.HTTPMethod = "GET"
	uri, err := req.Presign(5 * time.Minute) // 5 minutes should be enough.
	if err != nil {                          // bubble error back up to original request
		r.Error = err
		return nil
	}

	// We have our URL, set it on params
	return &uri
}

func (c Cluster) stackConfig(opts StackTemplateOptions, compressUserData bool) (*stackConfig, error) {
	assets, err := ReadTLSAssets(opts.TLSAssetsDir)
	if err != nil {
		return nil, err
	}
	stackConfig := stackConfig{}

	if stackConfig.Config, err = c.Config(); err != nil {
		return nil, err
	}

	awsConfig := aws.NewConfig()
	awsConfig = awsConfig.WithRegion(stackConfig.Config.Region)
	kmsSvc := kms.New(session.New(awsConfig))

	compactAssets, err := assets.compact(stackConfig.Config, kmsSvc)
	if err != nil {
		return nil, fmt.Errorf("failed to compress TLS assets: %v", err)
	}

	stackConfig.Config.TLSConfig = compactAssets

	if stackConfig.UserDataWorker, err = execute(opts.WorkerTmplFile, stackConfig.Config, compressUserData); err != nil {
		return nil, fmt.Errorf("failed to render worker cloud config: %v", err)
	}
	if stackConfig.UserDataController, err = execute(opts.ControllerTmplFile, stackConfig.Config, compressUserData); err != nil {
		return nil, fmt.Errorf("failed to render controller cloud config: %v", err)
	}

	return &stackConfig, nil
}

// ValidateRegion checks that an AWS region name is valid
func ValidateRegion(region string) error {
	glog.V(2).Infof("Querying EC2 for all valid regions")

	request := &ec2.DescribeRegionsInput{}
	config := aws.NewConfig().WithRegion("us-east-1")
	client := ec2.New(session.New(), config)

	response, err := client.DescribeRegions(request)

	if err != nil {
		return fmt.Errorf("Got an error while querying for valid regions (verify your AWS credentials?)")
	}
	for _, r := range response.Regions {
		name := aws.StringValue(r.RegionName)
		if name == region {
			return nil
		}
	}

	if os.Getenv("SKIP_REGION_CHECK") != "" {
		glog.Infof("AWS region does not appear to be valid, but skipping because SKIP_REGION_CHECK is set")
		return nil
	}

	return fmt.Errorf("Region is not a recognized EC2 region: %q (check you have specified valid zones?)", region)
}

// SendMessage sends an SNS message to an SNS region.
// See http://docs.aws.amazon.com/sdk-for-go/api/service/sns.html#type-PublishInput
func SendMessage(region string, topic string, subject string, message string) {
	if region == "" {
		log.Fatal("SNS region is required")
	}

	if topic == "" {
		log.Fatal("SNS topic is required")
	}

	if subject == "" {
		log.Fatal("SNS subject is required")
	}

	if message == "" {
		log.Fatal("SNS message is required")
	}

	params := &sns.PublishInput{
		Subject:   aws.String(subject),
		Message:   aws.String(message),
		TargetArn: aws.String(topic),
	}

	sns := sns.New(session.New(), aws.NewConfig().WithRegion(region))

	_, err := sns.Publish(params)
	awserror.HandleError(err)
}

// Config returns a valid aws.Config object for access to AWS services, or
// an error if the authentication and region couldn't be resolved
func (c *AccessConfig) Config() (*aws.Config, error) {
	var creds *credentials.Credentials

	region, err := c.Region()
	if err != nil {
		return nil, err
	}
	config := aws.NewConfig().WithRegion(region).WithMaxRetries(11)
	if c.ProfileName != "" {
		profile, err := NewFromProfile(c.ProfileName)
		if err != nil {
			return nil, err
		}
		creds, err = profile.CredentialsFromProfile(config)
		if err != nil {
			return nil, err
		}
	} else {
		sess := session.New(config)
		creds = credentials.NewChainCredentials([]credentials.Provider{
			&credentials.StaticProvider{Value: credentials.Value{
				AccessKeyID:     c.AccessKey,
				SecretAccessKey: c.SecretKey,
				SessionToken:    c.Token,
			}},
			&credentials.EnvProvider{},
			&credentials.SharedCredentialsProvider{Filename: "", Profile: ""},
			&ec2rolecreds.EC2RoleProvider{
				Client: ec2metadata.New(sess),
			},
		})
	}
	return config.WithCredentials(creds), nil
}

// AllSecurityGroups describes every instance in the requested regions
// *SecurityGroups are created for each *ec2.SecurityGroup
// and are passed to a channel
func AllSecurityGroups() chan *SecurityGroup {
	ch := make(chan *SecurityGroup, len(config.Regions))
	// waitgroup for all regions
	wg := sync.WaitGroup{}
	for _, region := range config.Regions {
		wg.Add(1)
		go func(region string) {
			defer wg.Done()
			// add region to waitgroup
			api := ec2.New(sess, aws.NewConfig().WithRegion(region))
			resp, err := api.DescribeSecurityGroups(&ec2.DescribeSecurityGroupsInput{})
			for _, sg := range resp.SecurityGroups {
				ch <- NewSecurityGroup(region, sg)
			}
			if err != nil {
				// probably should do something here...
				log.Error(err.Error())
			}
		}(region)
	}
	go func() {
		// in a separate goroutine, wait for all regions to finish
		// when they finish, close the chan
		wg.Wait()
		close(ch)

	}()
	return ch
}

// test that retries occur for request timeouts when response.Body can be nil
func TestRequestRecoverTimeoutWithNilBody(t *testing.T) {
	reqNum := 0
	reqs := []*http.Response{
		{StatusCode: 0, Body: nil}, // body can be nil when requests time out
		{StatusCode: 200, Body: body(`{"data":"valid"}`)},
	}
	errors := []error{
		errors.New("timeout"), nil,
	}

	s := awstesting.NewClient(aws.NewConfig().WithMaxRetries(10))
	s.Handlers.Validate.Clear()
	s.Handlers.Unmarshal.PushBack(unmarshal)
	s.Handlers.UnmarshalError.PushBack(unmarshalError)
	s.Handlers.AfterRetry.Clear() // force retry on all errors
	s.Handlers.AfterRetry.PushBack(func(r *request.Request) {
		if r.Error != nil {
			r.Error = nil
			r.Retryable = aws.Bool(true)
			r.RetryCount++
		}
	})
	s.Handlers.Send.Clear() // mock sending
	s.Handlers.Send.PushBack(func(r *request.Request) {
		r.HTTPResponse = reqs[reqNum]
		r.Error = errors[reqNum]
		reqNum++
	})
	out := &testData{}
	r := s.NewRequest(&request.Operation{Name: "Operation"}, nil, out)
	err := r.Send()
	assert.Nil(t, err)
	assert.Equal(t, 1, int(r.RetryCount))
	assert.Equal(t, "valid", out.Data)
}

func (cc ClientConfig) GetConfig() *aws.Config {

	//Explicitley set
	region := cc.Region

	// Environment var
	if region == "" {
		region = os.Getenv("AWS_REGION")
	}

	// Default
	if region == "" {
		region = "us-west-2"
	}

	// Create AWS config
	config := aws.NewConfig().WithRegion(region)

	// Check to see if DynamoDB is running locally
	endpoint := cc.Endpoint
	if endpoint == "" {
		// Env variable must be this format for Docker Compose to work, and also must replace tcp with http
		endpoint = os.Getenv("DYNAMODB_PORT")
		endpoint = strings.Replace(endpoint, "tcp", "http", 1)
	}

	if endpoint == "" {
		return config
	} else {
		return config.WithEndpoint(endpoint)
	}

}