func (a *App) UpdateParamsAndTemplate(changes map[string]string, template string) error {
	req := &cloudformation.UpdateStackInput{
		StackName:    aws.String(a.Name),
		Capabilities: []*string{aws.String("CAPABILITY_IAM")},
	}

	if template != "" {
		req.TemplateURL = aws.String(template)
	} else {
		req.UsePreviousTemplate = aws.Bool(true)
	}

	params := a.Parameters

	for key, val := range changes {
		params[key] = val
	}

	for key, val := range params {
		req.Parameters = append(req.Parameters, &cloudformation.Parameter{
			ParameterKey:   aws.String(key),
			ParameterValue: aws.String(val),
		})
	}

	_, err := CloudFormation().UpdateStack(req)

	return err
}

func ListReleases(app string) (Releases, error) {
	req := &dynamodb.QueryInput{
		KeyConditions: map[string]*dynamodb.Condition{
			"app": &dynamodb.Condition{
				AttributeValueList: []*dynamodb.AttributeValue{
					&dynamodb.AttributeValue{S: aws.String(app)},
				},
				ComparisonOperator: aws.String("EQ"),
			},
		},
		IndexName:        aws.String("app.created"),
		Limit:            aws.Int64(20),
		ScanIndexForward: aws.Bool(false),
		TableName:        aws.String(releasesTable(app)),
	}

	res, err := DynamoDB().Query(req)

	if err != nil {
		return nil, err
	}

	releases := make(Releases, len(res.Items))

	for i, item := range res.Items {
		releases[i] = *releaseFromItem(item)
	}

	return releases, nil
}

func (s *Service) UpdatePapertrail(arns map[string]string) error {
	input := struct {
		ARNs map[string]string
	}{
		arns,
	}

	formation, err := buildTemplate(fmt.Sprintf("service/%s", s.Type), "service", input)

	if err != nil {
		return err
	}

	// Update stack with all linked ARNs and EventSourceMappings
	_, err = CloudFormation().UpdateStack(&cloudformation.UpdateStackInput{
		StackName:    aws.String(s.Name),
		Capabilities: []*string{aws.String("CAPABILITY_IAM")},
		Parameters: []*cloudformation.Parameter{
			&cloudformation.Parameter{
				ParameterKey:   aws.String("Url"),
				ParameterValue: aws.String(s.Parameters["Url"]),
			},
		},
		TemplateBody: aws.String(formation),
	})

	return err
}

func (b *Build) log(line string) {
	b.Logs += fmt.Sprintf("%s\n", line)

	if b.kinesis == "" {
		app, err := GetApp(b.App)

		if err != nil {
			panic(err)
		}

		b.kinesis = app.Outputs["Kinesis"]
	}

	_, err := Kinesis().PutRecords(&kinesis.PutRecordsInput{
		StreamName: aws.String(b.kinesis),
		Records: []*kinesis.PutRecordsRequestEntry{
			&kinesis.PutRecordsRequestEntry{
				Data:         []byte(fmt.Sprintf("build: %s", line)),
				PartitionKey: aws.String(string(time.Now().UnixNano())),
			},
		},
	})

	if err != nil {
		fmt.Fprintf(os.Stderr, "ERROR: %s\n", err)
	}
}

// Retrieve generates a new set of temporary credentials using STS.
func (p *AssumeRoleProvider) Retrieve() (credentials.Value, error) {

	// Apply defaults where parameters are not set.
	if p.Client == nil {
		p.Client = sts.New(nil)
	}
	if p.RoleSessionName == "" {
		// Try to work out a role name that will hopefully end up unique.
		p.RoleSessionName = fmt.Sprintf("%d", time.Now().UTC().UnixNano())
	}
	if p.Duration == 0 {
		// Expire as often as AWS permits.
		p.Duration = 15 * time.Minute
	}

	roleOutput, err := p.Client.AssumeRole(&sts.AssumeRoleInput{
		DurationSeconds: aws.Int64(int64(p.Duration / time.Second)),
		RoleArn:         aws.String(p.RoleARN),
		RoleSessionName: aws.String(p.RoleSessionName),
		ExternalId:      p.ExternalID,
	})

	if err != nil {
		return credentials.Value{}, err
	}

	// We will proactively generate new credentials before they expire.
	p.SetExpiration(*roleOutput.Credentials.Expiration, p.ExpiryWindow)

	return credentials.Value{
		AccessKeyID:     *roleOutput.Credentials.AccessKeyId,
		SecretAccessKey: *roleOutput.Credentials.SecretAccessKey,
		SessionToken:    *roleOutput.Credentials.SessionToken,
	}, nil
}

func (r *Release) Promote() error {
	formation, err := r.Formation()

	if err != nil {
		return err
	}

	existing, err := formationParameters(formation)

	if err != nil {
		return err
	}

	app, err := GetApp(r.App)

	if err != nil {
		return err
	}

	manifest, err := LoadManifest(r.Manifest)

	if err != nil {
		return err
	}

	for _, me := range manifest {
		app.Parameters[fmt.Sprintf("%sCommand", UpperName(me.Name))] = me.CommandString()
		app.Parameters[fmt.Sprintf("%sImage", UpperName(me.Name))] = fmt.Sprintf("%s/%s-%s:%s", os.Getenv("REGISTRY_HOST"), r.App, me.Name, r.Build)
	}

	app.Parameters["Environment"] = r.EnvironmentUrl()
	app.Parameters["Kernel"] = CustomTopic
	app.Parameters["Release"] = r.Id
	app.Parameters["Version"] = os.Getenv("RELEASE")

	if os.Getenv("ENCRYPTION_KEY") != "" {
		app.Parameters["Key"] = os.Getenv("ENCRYPTION_KEY")
	}

	params := []*cloudformation.Parameter{}

	for key, value := range app.Parameters {
		if _, ok := existing[key]; ok {
			params = append(params, &cloudformation.Parameter{ParameterKey: aws.String(key), ParameterValue: aws.String(value)})
		}
	}

	req := &cloudformation.UpdateStackInput{
		Capabilities: []*string{aws.String("CAPABILITY_IAM")},
		StackName:    aws.String(r.App),
		TemplateBody: aws.String(formation),
		Parameters:   params,
	}

	_, err = CloudFormation().UpdateStack(req)

	return err
}

func Region(req *Request) *string {
	if req != nil {
		if region, ok := req.ResourceProperties["Region"].(string); ok && region != "" {
			return aws.String(region)
		}
	}

	return aws.String(os.Getenv("AWS_REGION"))
}

func s3Delete(bucket, key string) error {
	req := &s3.DeleteObjectInput{
		Bucket: aws.String(bucket),
		Key:    aws.String(key),
	}

	_, err := S3().DeleteObject(req)

	return err
}

func ListDeployments(app string) (Deployments, error) {
	a, err := GetApp(app)

	if err != nil {
		return nil, err
	}

	res, err := ECS().DescribeServices(&ecs.DescribeServicesInput{
		Cluster:  aws.String(os.Getenv("CLUSTER")),
		Services: []*string{aws.String(a.TaskDefinitionFamily())},
	})

	if err != nil {
		return nil, err
	}

	// no service yet, so no deployments
	if len(res.Services) != 1 {
		return Deployments{}, nil
	}

	service := res.Services[0]

	deployments := make(Deployments, len(service.Deployments))

	for i, d := range service.Deployments {
		deployments[i] = Deployment{
			Status:  *d.Status,
			Desired: *d.DesiredCount,
			Pending: *d.PendingCount,
			Running: *d.RunningCount,
			Created: *d.CreatedAt,
		}

		tres, err := ECS().DescribeTaskDefinition(&ecs.DescribeTaskDefinitionInput{
			TaskDefinition: d.TaskDefinition,
		})

		if err != nil {
			return nil, err
		}

		if len(tres.TaskDefinition.ContainerDefinitions) > 0 {
			for _, kp := range tres.TaskDefinition.ContainerDefinitions[0].Environment {
				if *kp.Name == "RELEASE" {
					deployments[i].Release = *kp.Value
					break
				}
			}
		}
	}

	return deployments, nil
}

func KMSKeyCreate(req Request) (string, map[string]string, error) {
	res, err := KMS(req).CreateKey(&kms.CreateKeyInput{
		Description: aws.String(req.ResourceProperties["Description"].(string)),
		KeyUsage:    aws.String(req.ResourceProperties["KeyUsage"].(string)),
	})

	if err != nil {
		return "", nil, err
	}

	return *res.KeyMetadata.Arn, nil, nil
}

func cleanupBucketObject(bucket, key, version string, S3 *s3.S3) {
	req := &s3.DeleteObjectInput{
		Bucket:    aws.String(bucket),
		Key:       aws.String(key),
		VersionId: aws.String(version),
	}

	_, err := S3.DeleteObject(req)

	if err != nil {
		fmt.Printf("error: %s\n", err)
	}
}

func (p *Process) Stop() error {
	req := &ecs.StopTaskInput{
		Cluster: aws.String(os.Getenv("CLUSTER")),
		Task:    aws.String(p.taskArn),
	}

	_, err := ECS().StopTask(req)

	if err != nil {
		return err
	}

	return nil
}

func s3Get(bucket, key string) ([]byte, error) {
	req := &s3.GetObjectInput{
		Bucket: aws.String(bucket),
		Key:    aws.String(key),
	}

	res, err := S3().GetObject(req)

	if err != nil {
		return nil, err
	}

	return ioutil.ReadAll(res.Body)
}

func dequeueMessage() ([]Message, error) {
	req := &sqs.ReceiveMessageInput{
		MaxNumberOfMessages: aws.Int64(10),
		QueueUrl:            aws.String(MessageQueueUrl),
		WaitTimeSeconds:     aws.Int64(10),
	}

	res, err := SQS().ReceiveMessage(req)

	if err != nil {
		return nil, err
	}

	messages := make([]Message, len(res.Messages))

	var message Message

	for i, m := range res.Messages {
		err = json.Unmarshal([]byte(*m.Body), &message)

		if err != nil {
			return nil, err
		}

		message.MessageID = m.MessageId
		message.ReceiptHandle = m.ReceiptHandle

		messages[i] = message
	}

	return messages, nil
}

func (s *Service) LinkPapertrail(app App) error {
	// build map of app name -> arn of all linked services
	arns := map[string]string{}

	for k, v := range s.Outputs {
		if strings.HasSuffix(k, "Link") {
			n := DashName(k)
			arns[n[:len(n)-5]] = v
		}
	}

	// get full Kinesis ARN for app
	req, err := Kinesis().DescribeStream(&kinesis.DescribeStreamInput{
		StreamName: aws.String(app.Outputs["Kinesis"]),
	})

	arn := *req.StreamDescription.StreamARN

	if err != nil {
		return err
	}

	// append new ARN and update
	arns[app.Name] = arn
	return s.UpdatePapertrail(arns)
}

func subscribeKinesis(stream string, output chan []byte, quit chan bool) {
	sreq := &kinesis.DescribeStreamInput{
		StreamName: aws.String(stream),
	}
	sres, err := Kinesis().DescribeStream(sreq)

	if err != nil {
		fmt.Printf("err1 %+v\n", err)
		// panic(err)
		return
	}

	shards := make([]string, len(sres.StreamDescription.Shards))

	for i, s := range sres.StreamDescription.Shards {
		shards[i] = *s.ShardId
	}

	done := make([](chan bool), len(shards))

	for i, shard := range shards {
		done[i] = make(chan bool)
		go subscribeKinesisShard(stream, shard, output, done[i])
	}
}

func S3Put(bucket, key string, data []byte, public bool) error {
	req := &s3.PutObjectInput{
		Body:          bytes.NewReader(data),
		Bucket:        aws.String(bucket),
		ContentLength: aws.Int64(int64(len(data))),
		Key:           aws.String(key),
	}

	if public {
		req.ACL = aws.String("public-read")
	}

	_, err := S3().PutObject(req)

	return err
}

func subscribeKinesisShard(stream, shard string, output chan []byte, quit chan bool) {
	ireq := &kinesis.GetShardIteratorInput{
		ShardId:           aws.String(shard),
		ShardIteratorType: aws.String("LATEST"),
		StreamName:        aws.String(stream),
	}

	ires, err := Kinesis().GetShardIterator(ireq)

	if err != nil {
		fmt.Printf("err2 %+v\n", err)
		// panic(err)
		return
	}

	iter := *ires.ShardIterator

	for {
		select {
		case <-quit:
			fmt.Println("quitting")
			return
		default:
			greq := &kinesis.GetRecordsInput{
				ShardIterator: aws.String(iter),
			}
			gres, err := Kinesis().GetRecords(greq)

			if err != nil {
				fmt.Printf("err3 %+v\n", err)
				// panic(err)
				return
			}

			iter = *gres.NextShardIterator

			for _, record := range gres.Records {
				output <- []byte(fmt.Sprintf("%s\n", string(record.Data)))
			}

			time.Sleep(500 * time.Millisecond)
		}
	}
}

func DockerHost() (string, error) {
	ares, err := ECS().ListContainerInstances(&ecs.ListContainerInstancesInput{
		Cluster: aws.String(os.Getenv("CLUSTER")),
	})

	if len(ares.ContainerInstanceArns) == 0 {
		return "", fmt.Errorf("no container instances")
	}

	cres, err := ECS().DescribeContainerInstances(&ecs.DescribeContainerInstancesInput{
		Cluster:            aws.String(os.Getenv("CLUSTER")),
		ContainerInstances: ares.ContainerInstanceArns,
	})

	if err != nil {
		return "", err
	}

	if len(cres.ContainerInstances) == 0 {
		return "", fmt.Errorf("no container instances")
	}

	id := *cres.ContainerInstances[rand.Intn(len(cres.ContainerInstances))].Ec2InstanceId

	ires, err := EC2().DescribeInstances(&ec2.DescribeInstancesInput{
		Filters: []*ec2.Filter{
			&ec2.Filter{Name: aws.String("instance-id"), Values: []*string{&id}},
		},
	})

	if len(ires.Reservations) != 1 || len(ires.Reservations[0].Instances) != 1 {
		return "", fmt.Errorf("could not describe container instance")
	}

	ip := *ires.Reservations[0].Instances[0].PrivateIpAddress

	if os.Getenv("DEVELOPMENT") == "true" {
		ip = *ires.Reservations[0].Instances[0].PublicIpAddress
	}

	return fmt.Sprintf("http://%s:2376", ip), nil
}

func LambdaFunctionCreate(req Request) (string, error) {
	bres, err := http.Get(req.ResourceProperties["ZipFile"].(string))

	if err != nil {
		return "", err
	}

	defer bres.Body.Close()

	body, err := ioutil.ReadAll(bres.Body)

	memory := 128
	timeout := 5

	if m, ok := req.ResourceProperties["Memory"].(string); ok && m != "" {
		memory, _ = strconv.Atoi(m)
	}

	if t, ok := req.ResourceProperties["Timeout"].(string); ok && t != "" {
		timeout, _ = strconv.Atoi(t)
	}

	role := fmt.Sprintf("arn:aws:iam::%s:role/%s", req.ResourceProperties["AccountId"].(string), req.ResourceProperties["Role"].(string))

	res, err := Lambda(req).CreateFunction(&lambda.CreateFunctionInput{
		FunctionName: aws.String(req.ResourceProperties["Name"].(string)),
		Handler:      aws.String(req.ResourceProperties["Handler"].(string)),
		MemorySize:   aws.Int64(int64(memory)),
		Timeout:      aws.Int64(int64(timeout)),
		Role:         aws.String(role),
		Runtime:      aws.String(req.ResourceProperties["Runtime"].(string)),
		Code: &lambda.FunctionCode{
			ZipFile: body,
		},
	})

	if err != nil {
		return "", err
	}

	return *res.FunctionArn, nil
}