package setup

import (
	"errors"
	"fmt"

	"code.justin.tv/d8a/buddy/cmd/buddy-cli/data"
)

func AirflowSqlalchemyConnection() (string, error) {
	rdsInstance, err := data.RootData.LocateDataStore(false)
	if err != nil {
		return "", err
	}

	configFile := data.RootData.ConfigFile()
	secretPath := fmt.Sprintf("%s/rds-buddy-store/%s/airflow_password", configFile.SandstormTeam, configFile.Environment)
	password, err := data.RootData.SandstormClient().Manager().Get(secretPath)
	if err != nil {
		return "", err
	}

	if password == nil {
		return "", fmt.Errorf("secret %s does not exist in sandstorm", secretPath)
	}

	return fmt.Sprintf("postgresql://airflow:%s@%s:%d/airflow", string(password.Plaintext), *rdsInstance.Endpoint.Address, *rdsInstance.Endpoint.Port), nil
}

func AirflowRedisConnection() (string, error) {
	redisCluster, err := data.RootData.LocateRedis()
	if err != nil {
		return "", err
	}

	if len(redisCluster.CacheNodes) < 1 {
		return "", errors.New("could not locate any nodes in the airflow redis cluster")
	}

	return fmt.Sprintf("redis://%s:%d", *redisCluster.CacheNodes[0].Endpoint.Address, *redisCluster.CacheNodes[0].Endpoint.Port), nil
}
