/*
/* Copyright 2018-2026 contributors to the OpenLineage project
/* SPDX-License-Identifier: Apache-2.0
*/

package io.openlineage.spark.api.naming;

import io.openlineage.spark.api.OpenLineageContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;

/**
 * Provides the job name in the AWS Glue environment. Glue generates unique values for
 * "spark.app.name" property with every run. This provider is able to detect if the application name
 * was autogenerated or provided by the user. If the "spark.app.name" property was autogenerated,
 * the value of job name is used.
 */
@Slf4j
class AwsGlueApplicationJobNameProvider implements ApplicationJobNameProvider {

  /** Glue version - "2.0" or "3.0" or "4.0". */
  private static final String VERSION_PROPERTY = "spark.glue.GLUE_VERSION";

  /**
   * Job name as provided by the user when the job was created. It can include spaces, international
   * characters, special characters etc. For example "Send data from marketing tables to HDFS - V2".
   */
  private static final String JOB_NAME_PROPERTY = "spark.glue.JOB_NAME";

  /**
   * The autogenerated unique run identifier. It has the format "jr_<64 ascii characters>" For
   * example "jr_27897ad718cfca473928fdd53d57bec6d351c4fe530830212cbd1c74e9c84c2e".
   */
  private static final String JOB_RUN_ID_PROPERTY = "spark.glue.JOB_RUN_ID";

  @Override
  public boolean isDefinedAt(OpenLineageContext openLineageContext) {
    boolean isGlueEnvironment =
        openLineageContext
            .getSparkContext()
            .map(sc -> sc.getConf().contains(VERSION_PROPERTY))
            .orElse(false);
    if (isGlueEnvironment) {
      log.debug("The application is deployed in Glue environment.");
    } else {
      log.debug("The environment is not Glue environment. Skipping the provider");
    }
    return isGlueEnvironment;
  }

  @Override
  public String getJobName(OpenLineageContext openLineageContext) {
    SparkContext sparkContext = openLineageContext.getSparkContext().get();
    String appName = sparkContext.appName();
    SparkConf sparkConf = sparkContext.getConf();
    String jobName = sparkConf.get(JOB_NAME_PROPERTY);
    if (appName != null) {
      if (isAppNameGeneratedByGlue(appName, sparkContext)) {
        log.debug(
            "The [spark.app.name] property was autogenerated by Glue. Using the job name for the application name.");
        return jobName;
      } else {
        log.debug("Using [spark.app.name] property for the application name.");
        return appName;
      }
    } else {
      /*
       The "spark.app.name" property should always be set either by the developer or automatically
       by AWS Glue, but it is technically possible to erase it. In such scenario we use the Glue job name.
      */
      log.debug(
          "The [spark.app.name] property was not specified. Using the job name for the application name.");
      return jobName;
    }
  }

  private boolean isAppNameGeneratedByGlue(String appName, SparkContext sparkContext) {
    /*
     The structure of an autogenerated "spark.app.name property" is "nativespark-<job name>-<job run id>"
     without any normalization (it can contain spaces and special characters)
    */
    String appNameAsWouldBeGeneratedByGlue =
        "nativespark-"
            + sparkContext.getConf().get(JOB_NAME_PROPERTY)
            + "-"
            + sparkContext.getConf().get(JOB_RUN_ID_PROPERTY);
    return appName.equals(appNameAsWouldBeGeneratedByGlue);
  }
}
