locals {
  sources = {
    data_foundation_finance = {
      short_name              = "df-finance"
      type                    = "tidbcloud"
      config_path             = "configs/sources/data_foundation_finance.yaml"
      bucket                  = "gcp-digital-bi-pingcap-df-finance-ingestion"
      location                = "us-west2"
      raw_dataset             = "data_foundation_finance_raw"
      final_dataset           = "data_foundation_finance_final"
      control_dataset         = "data_foundation_finance_control"
      extract_cron            = "0 2 * * *"
      load_cron               = "30 2 * * *"
      timezone                = "UTC"
      scheduler_paused        = false
      snapshot_retention_days = 35
      manifest_retention_days = 365
    }
    nexuscrm = {
      short_name              = "nexuscrm"
      type                    = "tidbcloud"
      config_path             = "configs/sources/nexuscrm.yaml"
      bucket                  = "gcp-digital-bi-pingcap-nexuscrm-ingestion"
      location                = "us-west2"
      raw_dataset             = "nexuscrm_raw"
      final_dataset           = "nexuscrm_final"
      control_dataset         = "nexuscrm_control"
      extract_cron            = "0 1 * * *"
      load_cron               = "30 1 * * *"
      timezone                = "UTC"
      scheduler_paused        = false
      snapshot_retention_days = 7
      manifest_retention_days = 365
    }
    helix = {
      short_name              = "helix-cloud-accounts"
      type                    = "mysql_query"
      config_path             = "configs/sources/helix.yaml"
      bucket                  = "gcp-digital-bi-pingcap-helix-ingestion"
      location                = "us-west2"
      raw_dataset             = "helix_raw"
      final_dataset           = "helix"
      control_dataset         = "helix_control"
      pipeline_cron           = "15 23 * * *"
      timezone                = "UTC"
      scheduler_paused        = false
      snapshot_retention_days = 35
      manifest_retention_days = 365
    }
  }

  tidbcloud_sources = {
    for name, source in local.sources : name => source
    if source.type == "tidbcloud"
  }

  dsn_sources = {
    for name, source in local.sources : name => source
    if contains(["tidbcloud", "mysql_query"], source.type)
  }

  split_sources = {
    for name, source in local.sources : name => source
    if source.type != "mysql_query"
  }

  credentialed_extract_sources = {
    for name, source in local.sources : name => source
    if contains(var.enabled_credentialed_extract_sources, name)
  }

  extract_job_sources = {
    for name, source in local.credentialed_extract_sources : name => source
    if source.type != "mysql_query"
  }

  load_job_sources = local.split_sources

  pipeline_job_sources = {
    for name, source in local.credentialed_extract_sources : name => source
    if source.type == "mysql_query"
  }

  datasets = merge(
    merge([
      for name, source in local.sources : {
        "${name}-raw" = {
          dataset_id = source.raw_dataset
          location   = source.location
        }
        "${name}-control" = {
          dataset_id = source.control_dataset
          location   = source.location
        }
      }
    ]...),
    merge([
      for name, source in local.sources : {
        "${name}-final" = {
          dataset_id = source.final_dataset
          location   = source.location
        }
      }
    ]...),
  )

  required_apis = toset([
    "artifactregistry.googleapis.com",
    "bigquery.googleapis.com",
    "cloudbuild.googleapis.com",
    "cloudresourcemanager.googleapis.com",
    "cloudscheduler.googleapis.com",
    "iam.googleapis.com",
    "iamcredentials.googleapis.com",
    "run.googleapis.com",
    "secretmanager.googleapis.com",
    "storage.googleapis.com",
    "sts.googleapis.com",
  ])

  pipeline_image = coalesce(
    var.pipeline_image,
    "${var.region}-docker.pkg.dev/${var.project_id}/${google_artifact_registry_repository.pipeline.repository_id}/pingcap-datapipelines:${var.pipeline_image_tag}",
  )
}

resource "google_project_service" "required" {
  for_each = local.required_apis

  project            = var.project_id
  service            = each.key
  disable_on_destroy = false
}

resource "google_service_account" "runtime" {
  project      = var.project_id
  account_id   = "pingcap-pipeline-runtime"
  display_name = "PingCAP datapipeline Cloud Run runtime"

  depends_on = [google_project_service.required]
}

resource "google_service_account" "scheduler" {
  project      = var.project_id
  account_id   = "pingcap-pipeline-scheduler"
  display_name = "PingCAP datapipeline Cloud Scheduler invoker"

  depends_on = [google_project_service.required]
}

resource "google_artifact_registry_repository" "pipeline" {
  project       = var.project_id
  location      = var.region
  repository_id = "pingcap-datapipelines"
  description   = "Container images for PingCAP datapipeline Cloud Run jobs"
  format        = "DOCKER"
  labels        = var.labels

  depends_on = [google_project_service.required]
}

resource "google_storage_bucket" "extract" {
  for_each = local.sources

  project                     = var.project_id
  name                        = each.value.bucket
  location                    = each.value.location
  uniform_bucket_level_access = true
  public_access_prevention    = "enforced"
  labels                      = var.labels

  lifecycle_rule {
    action {
      type = "Delete"
    }
    condition {
      age            = each.value.snapshot_retention_days
      matches_prefix = ["snapshots/source=${each.key}/"]
    }
  }

  lifecycle_rule {
    action {
      type = "Delete"
    }
    condition {
      age            = each.value.manifest_retention_days
      matches_prefix = ["manifests/source=${each.key}/", "logs/source=${each.key}/"]
    }
  }

  depends_on = [google_project_service.required]
}

resource "google_bigquery_dataset" "datasets" {
  for_each = local.datasets

  project                    = var.project_id
  dataset_id                 = each.value.dataset_id
  location                   = each.value.location
  labels                     = var.labels
  delete_contents_on_destroy = false

  depends_on = [google_project_service.required]
}

resource "google_project_iam_member" "runtime_bigquery_job_user" {
  project = var.project_id
  role    = "roles/bigquery.jobUser"
  member  = "serviceAccount:${google_service_account.runtime.email}"
}

resource "google_project_iam_member" "runtime_secret_accessor" {
  project = var.project_id
  role    = "roles/secretmanager.secretAccessor"
  member  = "serviceAccount:${google_service_account.runtime.email}"
}

resource "google_storage_bucket_iam_member" "runtime_bucket_admin" {
  for_each = local.sources

  bucket = google_storage_bucket.extract[each.key].name
  role   = "roles/storage.objectAdmin"
  member = "serviceAccount:${google_service_account.runtime.email}"
}

resource "google_bigquery_dataset_iam_member" "runtime_dataset_editor" {
  for_each = google_bigquery_dataset.datasets

  project    = var.project_id
  dataset_id = each.value.dataset_id
  role       = "roles/bigquery.dataEditor"
  member     = "serviceAccount:${google_service_account.runtime.email}"
}

resource "google_service_account" "tidb_export_writer" {
  for_each = local.tidbcloud_sources

  project      = var.project_id
  account_id   = "pc-${each.value.short_name}-gcs-export"
  display_name = "TiDB Cloud GCS export writer for ${each.key}"
}

resource "google_storage_bucket_iam_member" "tidb_export_writer_bucket_admin" {
  for_each = local.tidbcloud_sources

  bucket = google_storage_bucket.extract[each.key].name
  role   = "roles/storage.objectAdmin"
  member = "serviceAccount:${google_service_account.tidb_export_writer[each.key].email}"
}

resource "google_service_account_key" "tidb_export_writer" {
  for_each = local.tidbcloud_sources

  service_account_id = google_service_account.tidb_export_writer[each.key].name
}

resource "google_secret_manager_secret" "tidb_source_dsn" {
  for_each = local.dsn_sources

  project   = var.project_id
  secret_id = "pingcap-${each.value.short_name}-tidb-source-dsn"

  replication {
    auto {}
  }

  depends_on = [google_project_service.required]
}

resource "google_secret_manager_secret" "tidb_public_key" {
  for_each = local.tidbcloud_sources

  project   = var.project_id
  secret_id = "pingcap-${each.value.short_name}-tidb-public-key"

  replication {
    auto {}
  }

  depends_on = [google_project_service.required]
}

resource "google_secret_manager_secret" "tidb_private_key" {
  for_each = local.tidbcloud_sources

  project   = var.project_id
  secret_id = "pingcap-${each.value.short_name}-tidb-private-key"

  replication {
    auto {}
  }

  depends_on = [google_project_service.required]
}

resource "google_secret_manager_secret" "tidb_gcs_writer_key" {
  for_each = local.tidbcloud_sources

  project   = var.project_id
  secret_id = "pingcap-${each.value.short_name}-tidb-gcs-writer-key"

  replication {
    auto {}
  }

  depends_on = [google_project_service.required]
}

resource "google_secret_manager_secret_version" "tidb_gcs_writer_key" {
  for_each = local.tidbcloud_sources

  secret      = google_secret_manager_secret.tidb_gcs_writer_key[each.key].id
  secret_data = google_service_account_key.tidb_export_writer[each.key].private_key
}

resource "google_cloud_run_v2_job" "extract" {
  for_each = local.extract_job_sources

  project             = var.project_id
  name                = "pingcap-${each.value.short_name}-extract"
  location            = var.region
  labels              = var.labels
  deletion_protection = false

  template {
    template {
      service_account = google_service_account.runtime.email
      timeout         = each.value.type == "tidbcloud" ? "7200s" : "3600s"
      max_retries     = 0

      containers {
        image = local.pipeline_image
        args  = ["extract", "--source", "/app/${each.value.config_path}"]

        resources {
          limits = {
            cpu    = "2"
            memory = "4Gi"
          }
        }

        env {
          name  = "GOOGLE_CLOUD_PROJECT"
          value = var.project_id
        }

        dynamic "env" {
          for_each = each.value.type == "tidbcloud" ? [
            {
              name   = "TIDB_SOURCE_DSN"
              secret = google_secret_manager_secret.tidb_source_dsn[each.key].secret_id
            },
            {
              name   = "TIDB_CLOUD_PUBLIC_KEY"
              secret = google_secret_manager_secret.tidb_public_key[each.key].secret_id
            },
            {
              name   = "TIDB_CLOUD_PRIVATE_KEY"
              secret = google_secret_manager_secret.tidb_private_key[each.key].secret_id
            },
            {
              name   = "TIDB_EXPORT_GCS_SERVICE_ACCOUNT_KEY"
              secret = google_secret_manager_secret.tidb_gcs_writer_key[each.key].secret_id
            },
          ] : []

          content {
            name = env.value.name
            value_source {
              secret_key_ref {
                secret  = env.value.secret
                version = "latest"
              }
            }
          }
        }
      }
    }
  }

  depends_on = [
    google_bigquery_dataset.datasets,
    google_project_iam_member.runtime_bigquery_job_user,
    google_project_iam_member.runtime_secret_accessor,
    google_storage_bucket_iam_member.runtime_bucket_admin,
  ]

  lifecycle {
    # Image tag is owned by the GitHub Actions deploy (gcloud run jobs update),
    # not Terraform. Without this, apply would revert the running job to the
    # var.pipeline_image_tag default and fight the CD pipeline.
    ignore_changes = [
      template[0].template[0].containers[0].image,
    ]
  }
}

resource "google_cloud_run_v2_job" "load" {
  for_each = local.load_job_sources

  project             = var.project_id
  name                = "pingcap-${each.value.short_name}-load"
  location            = var.region
  labels              = var.labels
  deletion_protection = false

  template {
    template {
      service_account = google_service_account.runtime.email
      timeout         = each.key == "nexuscrm" ? "7200s" : "3600s"
      max_retries     = 0

      containers {
        image = local.pipeline_image
        args  = ["load", "--source", "/app/${each.value.config_path}", "--all-unloaded"]

        resources {
          limits = {
            cpu    = "2"
            memory = "4Gi"
          }
        }

        env {
          name  = "GOOGLE_CLOUD_PROJECT"
          value = var.project_id
        }
      }
    }
  }

  depends_on = [
    google_bigquery_dataset.datasets,
    google_project_iam_member.runtime_bigquery_job_user,
    google_storage_bucket_iam_member.runtime_bucket_admin,
  ]

  lifecycle {
    # Image tag is owned by the GitHub Actions deploy (gcloud run jobs update),
    # not Terraform. Without this, apply would revert the running job to the
    # var.pipeline_image_tag default and fight the CD pipeline.
    ignore_changes = [
      template[0].template[0].containers[0].image,
    ]
  }
}

resource "google_cloud_run_v2_job" "pipeline" {
  for_each = local.pipeline_job_sources

  project             = var.project_id
  name                = "pingcap-${each.value.short_name}"
  location            = var.region
  labels              = var.labels
  deletion_protection = false

  template {
    template {
      service_account = google_service_account.runtime.email
      timeout         = "3600s"
      max_retries     = 0

      containers {
        image = local.pipeline_image
        args  = ["run-pipeline", "--source", "/app/${each.value.config_path}"]

        resources {
          limits = {
            cpu    = "2"
            memory = "4Gi"
          }
        }

        env {
          name  = "GOOGLE_CLOUD_PROJECT"
          value = var.project_id
        }

        env {
          name = "TIDB_SOURCE_DSN"
          value_source {
            secret_key_ref {
              secret  = google_secret_manager_secret.tidb_source_dsn[each.key].secret_id
              version = "latest"
            }
          }
        }
      }
    }
  }

  depends_on = [
    google_bigquery_dataset.datasets,
    google_project_iam_member.runtime_bigquery_job_user,
    google_project_iam_member.runtime_secret_accessor,
    google_storage_bucket_iam_member.runtime_bucket_admin,
  ]

  lifecycle {
    ignore_changes = [
      template[0].template[0].containers[0].image,
    ]
  }
}

resource "google_project_iam_member" "scheduler_run_developer" {
  project = var.project_id
  role    = "roles/run.developer"
  member  = "serviceAccount:${google_service_account.scheduler.email}"
}

resource "google_service_account_iam_member" "scheduler_act_as_runtime" {
  service_account_id = google_service_account.runtime.name
  role               = "roles/iam.serviceAccountUser"
  member             = "serviceAccount:${google_service_account.scheduler.email}"
}

resource "google_cloud_scheduler_job" "extract" {
  for_each = local.extract_job_sources

  project     = var.project_id
  region      = var.region
  name        = "pingcap-${each.value.short_name}-extract"
  description = "Run ${each.key} extract into GCS"
  schedule    = each.value.extract_cron
  time_zone   = each.value.timezone
  paused      = each.value.scheduler_paused

  http_target {
    http_method = "POST"
    uri         = "https://run.googleapis.com/v2/projects/${var.project_id}/locations/${var.region}/jobs/${google_cloud_run_v2_job.extract[each.key].name}:run"

    oauth_token {
      service_account_email = google_service_account.scheduler.email
      scope                 = "https://www.googleapis.com/auth/cloud-platform"
    }
  }

  depends_on = [
    google_cloud_run_v2_job.extract,
    google_project_iam_member.scheduler_run_developer,
    google_service_account_iam_member.scheduler_act_as_runtime,
  ]
}

resource "google_cloud_scheduler_job" "load" {
  for_each = local.load_job_sources

  project     = var.project_id
  region      = var.region
  name        = "pingcap-${each.value.short_name}-load"
  description = "Load unloaded ${each.key} manifests from GCS into BigQuery"
  schedule    = each.value.load_cron
  time_zone   = each.value.timezone
  paused      = each.value.scheduler_paused

  http_target {
    http_method = "POST"
    uri         = "https://run.googleapis.com/v2/projects/${var.project_id}/locations/${var.region}/jobs/${google_cloud_run_v2_job.load[each.key].name}:run"

    oauth_token {
      service_account_email = google_service_account.scheduler.email
      scope                 = "https://www.googleapis.com/auth/cloud-platform"
    }
  }

  depends_on = [
    google_cloud_run_v2_job.load,
    google_project_iam_member.scheduler_run_developer,
    google_service_account_iam_member.scheduler_act_as_runtime,
  ]
}

resource "google_cloud_scheduler_job" "pipeline" {
  for_each = local.pipeline_job_sources

  project     = var.project_id
  region      = var.region
  name        = "pingcap-${each.value.short_name}"
  description = "Extract and load the ${each.key} snapshot into BigQuery"
  schedule    = each.value.pipeline_cron
  time_zone   = each.value.timezone
  paused      = each.value.scheduler_paused

  http_target {
    http_method = "POST"
    uri         = "https://run.googleapis.com/v2/projects/${var.project_id}/locations/${var.region}/jobs/${google_cloud_run_v2_job.pipeline[each.key].name}:run"

    oauth_token {
      service_account_email = google_service_account.scheduler.email
      scope                 = "https://www.googleapis.com/auth/cloud-platform"
    }
  }

  depends_on = [
    google_cloud_run_v2_job.pipeline,
    google_project_iam_member.scheduler_run_developer,
    google_service_account_iam_member.scheduler_act_as_runtime,
  ]
}