// RegisterTaskDefinitionIfNeeded checks if a task definition has already been
// registered via the provided cache, and if so returns it.
// Otherwise, it registers a new one.
//
// This exists to avoid an explosion of task definitions for automatically
// registered inputs.
func (client *ecsClient) RegisterTaskDefinitionIfNeeded(
	request *ecs.RegisterTaskDefinitionInput,
	taskDefinitionCache cache.Cache) (*ecs.TaskDefinition, error) {
	if request.Family == nil {
		return nil, errors.New("invalid task definitions: family is required")
	}

	taskDefResp, err := client.DescribeTaskDefinition(aws.StringValue(request.Family))

	// If there are no task definitions for this family OR the task definition exists and is marked as 'INACTIVE',
	// register the task definition and create a cache entry
	if err != nil || *taskDefResp.Status == ecs.TaskDefinitionStatusInactive {
		return persistTaskDefinition(request, client, taskDefinitionCache)
	}

	tdHash := client.constructTaskDefinitionCacheHash(taskDefResp, request)

	td := &ecs.TaskDefinition{}
	err = taskDefinitionCache.Get(tdHash, td)
	if err != nil || !cachedTaskDefinitionRevisionIsActive(td, client) {
		log.WithFields(log.Fields{
			"taskDefHash": tdHash,
			"taskDef":     td,
		}).Debug("cache miss")
		return persistTaskDefinition(request, client, taskDefinitionCache)
	}

	log.WithFields(log.Fields{
		"taskDefHash": tdHash,
		"taskDef":     td,
	}).Debug("cache hit")
	return td, nil
}

// persistTaskDefinition registers the task definition with ECS and creates a new local cache entry
func persistTaskDefinition(request *ecs.RegisterTaskDefinitionInput, client *ecsClient, taskDefinitionCache cache.Cache) (*ecs.TaskDefinition, error) {
	resp, err := client.RegisterTaskDefinition(request)
	if err != nil {
		return nil, err
	}

	tdHash := client.constructTaskDefinitionCacheHash(resp, request)

	err = taskDefinitionCache.Put(tdHash, resp)
	if err != nil {
		log.WithFields(log.Fields{
			"error": err,
		}).Warn("Could not cache task definition; redundant task definitions might be created")
		// We can keep going even if we can't cache and operate mostly fine
	}
	return resp, err

}

// RegisterTaskDefinitionIfNeeded checks if a task definition has already been
// registered via the provided cache, and if so returns it.
// Otherwise, it regsiters a new one.
//
// This exists to avoid an explosion of task definitions for automatically
// registered inputs
func (client *ecsClient) RegisterTaskDefinitionIfNeeded(
	request *ecs.RegisterTaskDefinitionInput,
	taskDefinitionCache cache.Cache) (*ecs.TaskDefinition, error) {
	if request.Family == nil {
		return nil, errors.New("invalid task definitions: family is required")
	}

	// md5sum of the 'GoString' formatted request
	// Open questions: is the gostring output actually specified to be
	// deterministic? Is it good enough that we don't care?
	// TODO, describe the cache hit to make sure it's still ACTIVE
	tdHash := fmt.Sprintf("%x", md5.Sum([]byte(request.GoString())))
	td := &ecs.TaskDefinition{}
	if err := taskDefinitionCache.Get(tdHash, td); err == nil {
		log.WithFields(log.Fields{
			"taskDefHash": tdHash,
			"taskDef":     td,
		}).Debug("cache hit")
		return td, nil
	} else {
		log.WithFields(log.Fields{
			"taskDefHash": tdHash,
			"taskDef":     td,
		}).Debug("cache miss")
	}

	resp, err := client.RegisterTaskDefinition(request)
	if err != nil {
		return nil, err
	}
	tdErr := taskDefinitionCache.Put(tdHash, resp)
	if tdErr != nil {
		log.WithFields(log.Fields{
			"error": tdErr,
		}).Warn("Could not cache task definition; redundant TDs might be created")
		// We can keep going even if we can't cache and operate mostly fine
	}
	return resp, err
}