resource "kubernetes_namespace" "spark_jobs_namespace" {
  metadata {
    name = "${var.namespace_prefix}spark-jobs"
  }
}


# resource "kubernetes_service_account_v1" "spark_sa" {
#   depends_on = [kubernetes_namespace.spark_jobs_namespace]
#   metadata {
#     name      = "spark"
#     namespace = "${var.namespace_prefix}spark-jobs"
#   }
# }
#
resource "kubernetes_role" "airflow_spark_role" {
  depends_on = [kubernetes_namespace.spark_jobs_namespace]
  metadata {
    name      = "airflow-spark-role"
    namespace = "${var.namespace_prefix}spark-jobs"
  }

  rule {
    api_groups = ["sparkoperator.k8s.io"]
    resources  = ["sparkapplications", "sparkapplications/status",
        "scheduledsparkapplications", "scheduledsparkapplications/status"]
    verbs      = ["*"]
  }

  rule {
    api_groups = [""]
    resources  = ["pods/log"]
    verbs      = ["*"]
  }
}

resource "kubernetes_role_binding_v1" "airflow_spark_role_binding" {
  depends_on = [kubernetes_namespace.spark_jobs_namespace]
  metadata {
    name      = "airflow-spark-role-binding"
    namespace = "${var.namespace_prefix}spark-jobs"
  }

  subject {
    kind      = "ServiceAccount"
    name      = "airflow-worker"
    namespace = "${var.namespace_prefix}airflow"
  }

  role_ref {
    api_group = "rbac.authorization.k8s.io"
    kind      = "Role"
    name      = "airflow-spark-role"
  }
}

resource "kubernetes_role_binding_v1" "airflow_spark_role_binding2" {
 depends_on = [kubernetes_namespace.spark_jobs_namespace]
 metadata {
   name      = "airflow-spark-role-binding2"
   namespace = "${var.namespace_prefix}spark-jobs"
 }

 subject {
   kind      = "ServiceAccount"
   name      = "airflow-worker"
   namespace = "${var.namespace_prefix}airflow"
 }

 role_ref {
   api_group = "rbac.authorization.k8s.io"
   kind      = "Role"
   name      = "spark-role"
 }
}
#
#
# resource "kubernetes_role_binding_v1" "spark_role_binding" {
#   depends_on = [kubernetes_namespace.spark_jobs_namespace]
#   metadata {
#     name      = "spark-role-binding"
#     namespace = "${var.namespace_prefix}spark-jobs"
#   }
#
#   subject {
#     kind      = "ServiceAccount"
#     name      = "spark"
#     namespace = "${var.namespace_prefix}spark-jobs"
#   }
#
#   role_ref {
#     api_group = "rbac.authorization.k8s.io"
#     kind      = "Role"
#     name      = "spark-role"
#   }
# }
#

resource "helm_release" "gcp_spark_operator" {
  depends_on        = [kubernetes_namespace.spark_jobs_namespace]
  name              = "gcp-spark-operator"
  chart             = "spark-operator"
  repository        = "https://kubeflow.github.io/spark-operator"
  create_namespace  = "true"
  namespace         = "${var.namespace_prefix}gcp-spark-operator"
  dependency_update = "true"
  version           = "2.0.2"

  # set {
  #   name  = "image.repository"
  #   value = "kubeflow/spark-operator"
  # }

  # set {
  #   name  = "image.tag"
  #   value = "v1beta2-1.4.5-3.5.0"
  # }

  set {
    name  = "sparkJobNamespaces"
    value = "{${var.namespace_prefix}spark-jobs}"
  }

  set {
    name  = "serviceAccounts.spark.name"
    value = "spark"
  }

  set {
    name  = "webhook.enabled"
    value = "true"
  }

  set {
    name  = "ingressUrlFormat"
    value = "\\{\\{$appName\\}\\}.\\{\\{$appNamespace\\}\\}.${var.domain}"
    type  = "string"
  }
}

resource "kubernetes_namespace" "airflow" {
  metadata {
    name = "${var.namespace_prefix}airflow"
  }
}

resource "kubernetes_secret" "s3_conn_secrets" {
  depends_on = [kubernetes_namespace.airflow]
  metadata {
    name = "s3-conn-secrets"
    namespace = "${var.namespace_prefix}airflow"
  }

  data = {
    username = var.s3_key
    password = var.s3_secret
    AIRFLOW_CONN_S3_CONN = <<EOT
{
  "conn_type": "aws",
  "extra": {
    "aws_access_key_id": "${var.s3_key}",
    "aws_secret_access_key": "${var.s3_secret}",
    "endpoint_url": "${var.s3_endpoint}",
    "verify": false
  }
}
EOT
  }

  type = "Opaque"
}



resource "helm_release" "airflow" {
  depends_on = [kubernetes_secret.s3_conn_secrets]

  name              = "airflow"
  chart             = "airflow"
  repository        = "https://airflow.apache.org"
  namespace         = "${var.namespace_prefix}airflow"
  dependency_update = "true"
  version           = "1.15.0"

  values = [
    file("./envs/${var.env}/airflow.yaml")
  ]

  set {
    name  = "fernetkey"
    value = "TG9mVjJvVEpoREVYdmdTRWlHdENXQ05zOU5OU2VGY0U="
  }

  set {
    name  = "webserver.defaultUser.password"
    value = var.admin_password
  }

  set {
    name  = "spec.values.env"
    value = yamlencode([
      {
        name  = "AIRFLOW__WEBSERVER__BASE_URL",
        value = "https://airflow.${var.domain}"
      },
      {
        name  = "AIRFLOW__WEBSERVER__ENABLE_PROXY_FIX",
        value = "True"
      }
    ])
  }

  set {
    name ="dags.gitSync.repo"
    value = var.repo_url
  }

  set {
    name ="dags.gitSync.branch"
    value = var.branch_name
  }

  set {
    name ="dags.gitSync.subPath"
    value = var.dag_path
  }

  # set {
  #   name  = "images.airflow.repository"
  #   value = "gbloisi/airflow"
  # }

  # set {
  #   name  = "images.airflow.tag"
  #   value = "2.8.3rc1-python3.11"
  # }

  set {
    name  = "ingress.web.host"
    value = "airflow.${var.domain}"
  }
  set {
    name  = "ingress.flower.host"
    value = "airflow.${var.domain}"
  }
}