default_rain_db_path <- function() {
  path <- Sys.getenv(
    "BEAUMONT_RAIN_DB",
    unset = file.path("data", "rain_history.sqlite")
  )

  normalizePath(path, winslash = "/", mustWork = FALSE)
}


get_rain_locations <- function() {
  data.frame(
    location_id = c("vignasses", "la_mure", "la_couchonnier"),
    label = c("Les Vignasses", "La Mure", "La Couchonnier"),
    latitude = c(44.8550665, 44.9167, 45.3722971),
    longitude = c(5.8441789, 5.8000, 5.6387118),
    station_count = c(3L, 3L, 3L),
    stringsAsFactors = FALSE
  )
}


get_rain_location <- function(location_id, locations = get_rain_locations()) {
  location <- locations[locations$location_id == location_id, , drop = FALSE]

  if (!nrow(location)) {
    stop(sprintf("Unknown location_id '%s'.", location_id))
  }

  location[1, , drop = FALSE]
}


load_station_catalog <- function(path = file.path("data", "allstations.csv")) {
  if (!file.exists(path)) {
    stop(sprintf("Station catalog not found at '%s'.", path))
  }

  read.csv(path, stringsAsFactors = FALSE)
}


source_secret_env <- function(secret_path = file.path("data", "secrets")) {
  if (!file.exists(secret_path)) {
    return(NULL)
  }

  secret_env <- new.env(parent = baseenv())
  sys.source(secret_path, envir = secret_env)
  secret_env
}


load_api_headers <- function(
  secret_path = file.path("data", "secrets"),
  token_name = "token4"
) {
  secret_env <- source_secret_env(secret_path = secret_path)
  if (is.null(secret_env)) {
    return(NULL)
  }

  api_key <- secret_env[[token_name]]
  if (is.null(api_key) || !nzchar(api_key)) {
    return(NULL)
  }

  httr::add_headers(
    accept = "*/*",
    apikey = api_key
  )
}


ensure_sqlite_cli <- function() {
  sqlite_bin <- Sys.which("sqlite3")

  if (!nzchar(sqlite_bin)) {
    stop("sqlite3 is required but was not found on PATH.")
  }

  sqlite_bin
}


run_sqlite <- function(args, stdin = "", timeout = 0) {
  sqlite_bin <- ensure_sqlite_cli()

  output <- system2(
    sqlite_bin,
    args = args,
    stdout = TRUE,
    stderr = TRUE,
    stdin = stdin,
    timeout = timeout
  )

  status <- attr(output, "status")
  if (!is.null(status) && status != 0) {
    stop(paste(output, collapse = "\n"))
  }

  output
}


sql_string <- function(value) {
  if (is.null(value) || is.na(value)) {
    return("NULL")
  }

  paste0("'", gsub("'", "''", as.character(value), fixed = TRUE), "'")
}


parse_station_timestamp <- function(x) {
  if (inherits(x, "POSIXt")) {
    return(as.POSIXct(x, tz = "UTC"))
  }

  if (is.numeric(x)) {
    x <- format(x, scientific = FALSE, trim = TRUE)
  }

  x <- trimws(as.character(x))
  x <- sub("\\.0+$", "", x)

  if (!length(x)) {
    return(as.POSIXct(character(), tz = "UTC"))
  }

  if (grepl("T", x[1], fixed = TRUE)) {
    return(as.POSIXct(x, format = "%Y-%m-%dT%H:%M:%SZ", tz = "UTC"))
  }

  as.POSIXct(strptime(x, format = "%Y%m%d%H%M", tz = "UTC"))
}


format_utc_timestamp <- function(x) {
  format(as.POSIXct(x, tz = "UTC"), "%Y-%m-%dT%H:%M:%SZ", tz = "UTC")
}


round_metric_values <- function(values, digits = 2L) {
  values <- suppressWarnings(as.numeric(values))
  round(values, digits = as.integer(digits))
}


celsius_from_kelvin <- function(values) {
  values <- suppressWarnings(as.numeric(values))
  values - 273.15
}


hpa_from_pa <- function(values) {
  values <- suppressWarnings(as.numeric(values))
  values / 100
}


identity_numeric <- function(values) {
  suppressWarnings(as.numeric(values))
}


get_weather_metric_definitions <- function() {
  list(
    rain_6m = list(
      label = "Rain",
      unit = "mm / 6 min",
      daily_agg = "sum",
      digits = 2L,
      source_name = "DPClim"
    ),
    air_temperature = list(
      label = "Temperature",
      unit = "C",
      daily_agg = "mean",
      digits = 2L,
      source_name = "DPPaquetObs",
      obs_column = "t",
      transform = celsius_from_kelvin
    ),
    dew_point = list(
      label = "Dew point",
      unit = "C",
      daily_agg = "mean",
      digits = 2L,
      source_name = "DPPaquetObs",
      obs_column = "td",
      transform = celsius_from_kelvin
    ),
    humidity = list(
      label = "Humidity",
      unit = "%",
      daily_agg = "mean",
      digits = 1L,
      source_name = "DPPaquetObs",
      obs_column = "u",
      transform = identity_numeric
    ),
    wind_direction = list(
      label = "Wind direction",
      unit = "deg",
      daily_agg = "mean",
      digits = 0L,
      source_name = "DPPaquetObs",
      obs_column = "dd",
      transform = identity_numeric
    ),
    wind_speed = list(
      label = "Wind speed",
      unit = "m/s",
      daily_agg = "mean",
      digits = 2L,
      source_name = "DPPaquetObs",
      obs_column = "ff",
      transform = identity_numeric
    ),
    gust_direction = list(
      label = "Gust direction",
      unit = "deg",
      daily_agg = "mean",
      digits = 0L,
      source_name = "DPPaquetObs",
      obs_column = "dxi10",
      transform = identity_numeric
    ),
    wind_gust = list(
      label = "Wind gust",
      unit = "m/s",
      daily_agg = "max",
      digits = 2L,
      source_name = "DPPaquetObs",
      obs_column = "fxi10",
      transform = identity_numeric
    ),
    soil_temperature_10cm = list(
      label = "Soil temp 10 cm",
      unit = "C",
      daily_agg = "mean",
      digits = 2L,
      source_name = "DPPaquetObs",
      obs_column = "t_10",
      transform = celsius_from_kelvin
    ),
    soil_temperature_20cm = list(
      label = "Soil temp 20 cm",
      unit = "C",
      daily_agg = "mean",
      digits = 2L,
      source_name = "DPPaquetObs",
      obs_column = "t_20",
      transform = celsius_from_kelvin
    ),
    soil_temperature_50cm = list(
      label = "Soil temp 50 cm",
      unit = "C",
      daily_agg = "mean",
      digits = 2L,
      source_name = "DPPaquetObs",
      obs_column = "t_50",
      transform = celsius_from_kelvin
    ),
    soil_temperature_100cm = list(
      label = "Soil temp 100 cm",
      unit = "C",
      daily_agg = "mean",
      digits = 2L,
      source_name = "DPPaquetObs",
      obs_column = "t_100",
      transform = celsius_from_kelvin
    ),
    visibility = list(
      label = "Visibility",
      unit = "m",
      daily_agg = "mean",
      digits = 0L,
      source_name = "DPPaquetObs",
      obs_column = "vv",
      transform = identity_numeric
    ),
    station_pressure = list(
      label = "Station pressure",
      unit = "hPa",
      daily_agg = "mean",
      digits = 1L,
      source_name = "DPPaquetObs",
      obs_column = "pres",
      transform = hpa_from_pa
    ),
    sea_level_pressure = list(
      label = "Sea-level pressure",
      unit = "hPa",
      daily_agg = "mean",
      digits = 1L,
      source_name = "DPPaquetObs",
      obs_column = "pmer",
      transform = hpa_from_pa
    )
  )
}


get_weather_metric <- function(metric_id) {
  definitions <- get_weather_metric_definitions()
  definition <- definitions[[metric_id]]

  if (is.null(definition)) {
    stop(sprintf("Unknown metric_id '%s'.", metric_id))
  }

  definition$metric_id <- metric_id
  definition
}


get_weather_metric_catalog <- function(include_rain = TRUE) {
  definitions <- get_weather_metric_definitions()
  metric_ids <- names(definitions)
  if (!include_rain) {
    metric_ids <- setdiff(metric_ids, "rain_6m")
  }

  data.frame(
    metric_id = metric_ids,
    label = vapply(metric_ids, function(metric_id) definitions[[metric_id]]$label, character(1)),
    unit = vapply(metric_ids, function(metric_id) definitions[[metric_id]]$unit, character(1)),
    daily_agg = vapply(metric_ids, function(metric_id) definitions[[metric_id]]$daily_agg, character(1)),
    source_name = vapply(metric_ids, function(metric_id) definitions[[metric_id]]$source_name, character(1)),
    obs_column = vapply(
      metric_ids,
      function(metric_id) {
        column_name <- definitions[[metric_id]]$obs_column
        if (is.null(column_name)) {
          ""
        } else {
          column_name
        }
      },
      character(1)
    ),
    stringsAsFactors = FALSE
  )
}


empty_weather_cache <- function() {
  data.frame(
    location_id = character(),
    station_id = character(),
    station_name = character(),
    observed_at = character(),
    observed_day = character(),
    metric_id = character(),
    metric_label = character(),
    unit = character(),
    source_name = character(),
    value_num = numeric(),
    fetched_at = character(),
    stringsAsFactors = FALSE
  )
}


empty_metric_query <- function(aggregate = c("raw", "daily", "weekly", "monthly")) {
  aggregate <- match.arg(aggregate)

  if (aggregate != "raw") {
    return(data.frame(
      station_id = character(),
      station_name = character(),
      observed_day = character(),
      metric_id = character(),
      metric_label = character(),
      unit = character(),
      value_num = numeric(),
      stringsAsFactors = FALSE
    ))
  }

  data.frame(
    location_id = character(),
    station_id = character(),
    station_name = character(),
    observed_at = character(),
    observed_day = character(),
    metric_id = character(),
    metric_label = character(),
    unit = character(),
    value_num = numeric(),
    stringsAsFactors = FALSE
  )
}


normalise_rainfall_data <- function(
  raw_data,
  location_id,
  fetched_at = Sys.time()
) {
  if (is.null(raw_data) || !nrow(raw_data)) {
    return(empty_weather_cache())
  }

  required_columns <- c("POSTE", "DATE", "RR6", "Nom_usuel")
  missing_columns <- setdiff(required_columns, names(raw_data))

  if (length(missing_columns)) {
    stop(
      sprintf(
        "Rainfall payload is missing required columns: %s",
        paste(missing_columns, collapse = ", ")
      )
    )
  }

  metric <- get_weather_metric("rain_6m")
  observed_at <- parse_station_timestamp(raw_data$DATE)
  station_id <- as.character(raw_data$POSTE)
  station_name <- trimws(as.character(raw_data$Nom_usuel))
  station_name[!nzchar(station_name)] <- station_id[!nzchar(station_name)]
  fetched_at <- format_utc_timestamp(fetched_at)

  data.frame(
    location_id = location_id,
    station_id = station_id,
    station_name = station_name,
    observed_at = format_utc_timestamp(observed_at),
    observed_day = format(observed_at, "%Y-%m-%d"),
    metric_id = metric$metric_id,
    metric_label = metric$label,
    unit = metric$unit,
    source_name = metric$source_name,
    value_num = round_metric_values(raw_data$RR6, metric$digits),
    fetched_at = fetched_at,
    stringsAsFactors = FALSE
  )
}


normalise_observation_package_data <- function(
  raw_data,
  location_id,
  station_name = NULL,
  fetched_at = Sys.time()
) {
  if (is.null(raw_data) || !nrow(raw_data)) {
    return(empty_weather_cache())
  }

  required_columns <- c("geo_id_insee", "validity_time")
  missing_columns <- setdiff(required_columns, names(raw_data))
  if (length(missing_columns)) {
    stop(
      sprintf(
        "Observation payload is missing required columns: %s",
        paste(missing_columns, collapse = ", ")
      )
    )
  }

  observed_at <- parse_station_timestamp(raw_data$validity_time)
  observed_day <- format(observed_at, "%Y-%m-%d")
  station_id <- as.character(raw_data$geo_id_insee)
  if (is.null(station_name)) {
    station_name <- station_id
  }
  station_name <- rep(as.character(station_name)[1], length.out = nrow(raw_data))
  fetched_at <- format_utc_timestamp(fetched_at)

  definitions <- get_weather_metric_definitions()
  metric_ids <- setdiff(names(definitions), "rain_6m")

  measurement_rows <- lapply(metric_ids, function(metric_id) {
    definition <- definitions[[metric_id]]
    column_name <- definition$obs_column

    if (is.null(column_name) || !column_name %in% names(raw_data)) {
      return(NULL)
    }

    values <- definition$transform(raw_data[[column_name]])
    values <- round_metric_values(values, definition$digits)
    keep <- !is.na(values)
    if (!any(keep)) {
      return(NULL)
    }

    data.frame(
      location_id = location_id,
      station_id = station_id[keep],
      station_name = station_name[keep],
      observed_at = format_utc_timestamp(observed_at[keep]),
      observed_day = observed_day[keep],
      metric_id = metric_id,
      metric_label = definition$label,
      unit = definition$unit,
      source_name = definition$source_name,
      value_num = values[keep],
      fetched_at = fetched_at,
      stringsAsFactors = FALSE
    )
  })

  measurement_rows <- Filter(Negate(is.null), measurement_rows)
  if (!length(measurement_rows)) {
    return(empty_weather_cache())
  }

  do.call(rbind, measurement_rows)
}


sqlite_scalar_query <- function(sql, db_path) {
  script_file <- tempfile(fileext = ".sql")
  on.exit(unlink(script_file), add = TRUE)

  writeLines(c(".bail on", sql), script_file)
  output <- run_sqlite(args = c(db_path), stdin = script_file)

  if (!length(output)) {
    return("")
  }

  utils::tail(output, 1)
}


sqlite_table_exists <- function(db_path, table_name) {
  query <- sprintf(
    paste(
      "SELECT COUNT(*)",
      "FROM sqlite_master",
      "WHERE type = 'table'",
      "  AND name = %s;"
    ),
    sql_string(table_name)
  )

  suppressWarnings(as.integer(sqlite_scalar_query(query, db_path))) > 0
}


migrate_legacy_rainfall_table <- function(db_path) {
  if (!sqlite_table_exists(db_path, "rainfall_observations")) {
    return(invisible(FALSE))
  }

  script_file <- tempfile(fileext = ".sql")
  on.exit(unlink(script_file), add = TRUE)

  writeLines(
    c(
      ".bail on",
      "BEGIN;",
      paste(
        "INSERT OR IGNORE INTO weather_measurements",
        "(location_id, station_id, station_name, observed_at, observed_day, metric_id, metric_label, unit, source_name, value_num, fetched_at)",
        "SELECT location_id, station_id, station_name, observed_at, observed_day,",
        "'rain_6m', 'Rain', 'mm / 6 min', 'DPClim', rain_mm, fetched_at",
        "FROM rainfall_observations;"
      ),
      "COMMIT;"
    ),
    script_file
  )

  invisible(run_sqlite(args = c(db_path), stdin = script_file))
  invisible(TRUE)
}


ensure_weather_db <- function(db_path = default_rain_db_path()) {
  dir.create(dirname(db_path), recursive = TRUE, showWarnings = FALSE)

  script_file <- tempfile(fileext = ".sql")
  on.exit(unlink(script_file), add = TRUE)

  writeLines(
    c(
      ".bail on",
      "PRAGMA journal_mode = WAL;",
      "CREATE TABLE IF NOT EXISTS weather_measurements (",
      "  location_id TEXT NOT NULL,",
      "  station_id TEXT NOT NULL,",
      "  station_name TEXT NOT NULL,",
      "  observed_at TEXT NOT NULL,",
      "  observed_day TEXT NOT NULL,",
      "  metric_id TEXT NOT NULL,",
      "  metric_label TEXT NOT NULL,",
      "  unit TEXT NOT NULL,",
      "  source_name TEXT NOT NULL,",
      "  value_num REAL,",
      "  fetched_at TEXT NOT NULL,",
      "  PRIMARY KEY (location_id, station_id, observed_at, metric_id)",
      ");",
      "CREATE INDEX IF NOT EXISTS idx_weather_location_time",
      "  ON weather_measurements(location_id, observed_at);",
      "CREATE INDEX IF NOT EXISTS idx_weather_location_metric_time",
      "  ON weather_measurements(location_id, metric_id, observed_at);",
      "CREATE INDEX IF NOT EXISTS idx_weather_location_metric_day",
      "  ON weather_measurements(location_id, metric_id, observed_day);"
    ),
    script_file
  )

  invisible(run_sqlite(args = c(db_path), stdin = script_file))
  migrate_legacy_rainfall_table(db_path)
  invisible(db_path)
}


ensure_rain_db <- function(db_path = default_rain_db_path()) {
  ensure_weather_db(db_path = db_path)
}


upsert_weather_measurements <- function(
  weather_data,
  db_path = default_rain_db_path()
) {
  ensure_weather_db(db_path)

  if (is.null(weather_data) || !nrow(weather_data)) {
    return(0L)
  }

  columns <- c(
    "location_id",
    "station_id",
    "station_name",
    "observed_at",
    "observed_day",
    "metric_id",
    "metric_label",
    "unit",
    "source_name",
    "value_num",
    "fetched_at"
  )

  weather_data <- weather_data[, columns, drop = FALSE]

  csv_file <- tempfile(fileext = ".csv")
  script_file <- tempfile(fileext = ".sql")
  on.exit(unlink(c(csv_file, script_file)), add = TRUE)

  utils::write.csv(
    weather_data,
    file = csv_file,
    row.names = FALSE,
    quote = TRUE,
    na = ""
  )

  writeLines(
    c(
      ".bail on",
      "BEGIN;",
      "CREATE TEMP TABLE weather_import (",
      "  location_id TEXT,",
      "  station_id TEXT,",
      "  station_name TEXT,",
      "  observed_at TEXT,",
      "  observed_day TEXT,",
      "  metric_id TEXT,",
      "  metric_label TEXT,",
      "  unit TEXT,",
      "  source_name TEXT,",
      "  value_num TEXT,",
      "  fetched_at TEXT",
      ");",
      ".mode csv",
      sprintf(".import --skip 1 %s weather_import", shQuote(csv_file)),
      paste(
        "INSERT OR REPLACE INTO weather_measurements",
        "(location_id, station_id, station_name, observed_at, observed_day, metric_id, metric_label, unit, source_name, value_num, fetched_at)",
        "SELECT location_id, station_id, station_name, observed_at, observed_day,",
        "metric_id, metric_label, unit, source_name, CAST(NULLIF(value_num, '') AS REAL), fetched_at",
        "FROM weather_import;"
      ),
      "SELECT changes();",
      "DROP TABLE weather_import;",
      "COMMIT;"
    ),
    script_file
  )

  output <- run_sqlite(args = c(db_path), stdin = script_file)
  rows_changed <- suppressWarnings(as.integer(utils::tail(output, 1)))

  if (is.na(rows_changed)) {
    return(nrow(weather_data))
  }

  rows_changed
}


upsert_rainfall_observations <- function(
  rain_data,
  db_path = default_rain_db_path()
) {
  upsert_weather_measurements(
    weather_data = rain_data,
    db_path = db_path
  )
}


read_sqlite_query <- function(sql, db_path = default_rain_db_path()) {
  ensure_weather_db(db_path)

  script_file <- tempfile(fileext = ".sql")
  on.exit(unlink(script_file), add = TRUE)

  writeLines(
    c(
      ".headers on",
      ".mode csv",
      sql
    ),
    script_file
  )

  output <- run_sqlite(args = c(db_path), stdin = script_file)
  if (!length(output)) {
    return(data.frame())
  }

  utils::read.csv(
    text = paste(output, collapse = "\n"),
    stringsAsFactors = FALSE
  )
}


get_location_cache_status <- function(
  location_id,
  db_path = default_rain_db_path()
) {
  status <- read_sqlite_query(
    sprintf(
      paste(
        "SELECT COUNT(*) AS row_count,",
        "COUNT(DISTINCT station_id) AS station_count,",
        "COUNT(DISTINCT metric_id) AS metric_count,",
        "MIN(observed_at) AS earliest_observed_at,",
        "MAX(observed_at) AS latest_observed_at",
        "FROM weather_measurements",
        "WHERE location_id = %s;"
      ),
      sql_string(location_id)
    ),
    db_path = db_path
  )

  if (!nrow(status)) {
    status <- data.frame(
      row_count = 0,
      station_count = 0,
      metric_count = 0,
      earliest_observed_at = "",
      latest_observed_at = "",
      stringsAsFactors = FALSE
    )
  }

  status$row_count <- as.integer(status$row_count)
  status$station_count <- as.integer(status$station_count)
  status$metric_count <- as.integer(status$metric_count)
  status
}


describe_cache_status <- function(
  location_id,
  db_path = default_rain_db_path(),
  locations = get_rain_locations()
) {
  location <- get_rain_location(location_id, locations = locations)
  status <- get_location_cache_status(location_id, db_path = db_path)

  if (!status$row_count[1]) {
    return(
      paste(
        sprintf("Location: %s", location$label[1]),
        sprintf("Database: %s", db_path),
        "No cached weather measurements yet.",
        sep = "\n"
      )
    )
  }

  paste(
    sprintf("Location: %s", location$label[1]),
    sprintf("Database: %s", db_path),
    sprintf("Cached rows: %s", format(status$row_count[1], big.mark = ",")),
    sprintf("Stations: %s", status$station_count[1]),
    sprintf("Metrics: %s", status$metric_count[1]),
    sprintf("Latest observation: %s", status$latest_observed_at[1]),
    sep = "\n"
  )
}


get_daily_aggregate_expression <- function(metric_id) {
  metric <- get_weather_metric(metric_id)

  switch(
    metric$daily_agg,
    sum = "ROUND(SUM(COALESCE(value_num, 0)), 2)",
    max = "ROUND(MAX(value_num), 2)",
    min = "ROUND(MIN(value_num), 2)",
    "ROUND(AVG(value_num), 2)"
  )
}


get_time_bucket_expression <- function(aggregate = c("daily", "weekly", "monthly")) {
  aggregate <- match.arg(aggregate)

  switch(
    aggregate,
    daily = "observed_day",
    weekly = paste(
      "date(",
      "observed_day,",
      "'-' || ((CAST(strftime('%w', observed_day) AS integer) + 6) % 7) || ' days'",
      ")"
    ),
    "date(observed_day, 'start of month')"
  )
}


query_cached_metric <- function(
  location_id,
  metric_id,
  start_date,
  end_date = Sys.Date(),
  aggregate = c("raw", "daily", "weekly", "monthly"),
  db_path = default_rain_db_path()
) {
  aggregate <- match.arg(aggregate)
  metric <- get_weather_metric(metric_id)

  start_date <- as.Date(start_date)
  end_date <- as.Date(end_date)
  if (start_date > end_date) {
    return(empty_metric_query(aggregate = aggregate))
  }

  start_at <- format_utc_timestamp(as.POSIXct(start_date, tz = "UTC"))
  end_at <- format_utc_timestamp(as.POSIXct(end_date + 1, tz = "UTC"))

  sql <- if (aggregate != "raw") {
    period_expression <- get_time_bucket_expression(aggregate)
    sprintf(
      paste(
        "SELECT station_id, station_name, %s AS observed_day, metric_id, metric_label, unit,",
        "%s AS value_num",
        "FROM weather_measurements",
        "WHERE location_id = %s",
        "  AND metric_id = %s",
        "  AND observed_at >= %s",
        "  AND observed_at < %s",
        "GROUP BY station_id, station_name, %s, metric_id, metric_label, unit",
        "ORDER BY observed_day, station_name;"
      ),
      period_expression,
      get_daily_aggregate_expression(metric_id),
      sql_string(location_id),
      sql_string(metric$metric_id),
      sql_string(start_at),
      sql_string(end_at),
      period_expression
    )
  } else {
    sprintf(
      paste(
        "SELECT location_id, station_id, station_name, observed_at, observed_day, metric_id, metric_label, unit, value_num",
        "FROM weather_measurements",
        "WHERE location_id = %s",
        "  AND metric_id = %s",
        "  AND observed_at >= %s",
        "  AND observed_at < %s",
        "ORDER BY observed_at, station_name;"
      ),
      sql_string(location_id),
      sql_string(metric$metric_id),
      sql_string(start_at),
      sql_string(end_at)
    )
  }

  result <- read_sqlite_query(sql, db_path = db_path)
  if (!nrow(result)) {
    return(empty_metric_query(aggregate = aggregate))
  }

  result
}


query_cached_rainfall <- function(
  location_id,
  start_date,
  end_date = Sys.Date(),
  aggregate = c("raw", "daily", "weekly", "monthly"),
  db_path = default_rain_db_path()
) {
  aggregate <- match.arg(aggregate)
  result <- query_cached_metric(
    location_id = location_id,
    metric_id = "rain_6m",
    start_date = start_date,
    end_date = end_date,
    aggregate = aggregate,
    db_path = db_path
  )

  if ("value_num" %in% names(result)) {
    names(result)[names(result) == "value_num"] <- "rain_mm"
  }

  result
}


query_latest_metric_values <- function(
  location_id,
  metric_id,
  db_path = default_rain_db_path()
) {
  metric <- get_weather_metric(metric_id)

  sql <- sprintf(
    paste(
      "SELECT m.station_id, m.station_name, m.observed_at, m.metric_id, m.metric_label, m.unit, m.value_num",
      "FROM weather_measurements AS m",
      "INNER JOIN (",
      "  SELECT station_id, MAX(observed_at) AS observed_at",
      "  FROM weather_measurements",
      "  WHERE location_id = %s",
      "    AND metric_id = %s",
      "  GROUP BY station_id",
      ") AS latest",
      "ON latest.station_id = m.station_id",
      "AND latest.observed_at = m.observed_at",
      "WHERE m.location_id = %s",
      "  AND m.metric_id = %s",
      "ORDER BY m.station_name;"
    ),
    sql_string(location_id),
    sql_string(metric$metric_id),
    sql_string(location_id),
    sql_string(metric$metric_id)
  )

  result <- read_sqlite_query(sql, db_path = db_path)
  if (!nrow(result)) {
    return(data.frame(
      station_id = character(),
      station_name = character(),
      observed_at = character(),
      metric_id = character(),
      metric_label = character(),
      unit = character(),
      value_num = numeric(),
      stringsAsFactors = FALSE
    ))
  }

  result
}


split_sync_ranges <- function(start_date, end_date, chunk_days = 120L) {
  start_date <- as.Date(start_date)
  end_date <- as.Date(end_date)
  chunk_days <- as.integer(chunk_days)

  if (start_date > end_date) {
    return(data.frame(
      start_date = as.Date(character()),
      end_date = as.Date(character())
    ))
  }

  ranges <- vector("list", length = 0L)
  current_start <- start_date

  while (current_start <= end_date) {
    current_end <- min(current_start + chunk_days - 1L, end_date)
    ranges[[length(ranges) + 1L]] <- data.frame(
      start_date = current_start,
      end_date = current_end
    )
    current_start <- current_end + 1L
  }

  do.call(rbind, ranges)
}


get_sync_start_date <- function(
  location_id,
  db_path = default_rain_db_path(),
  end_date = Sys.Date(),
  initial_backfill_days = 21L,
  overlap_days = 1L,
  metric_id = "rain_6m"
) {
  metric <- get_weather_metric(metric_id)
  end_date <- as.Date(end_date)
  requested_start_date <- end_date - as.integer(initial_backfill_days) + 1L

  latest_data <- read_sqlite_query(
    sprintf(
      paste(
        "SELECT MIN(observed_at) AS earliest_observed_at,",
        "MAX(observed_at) AS latest_observed_at",
        "FROM weather_measurements",
        "WHERE location_id = %s",
        "  AND metric_id = %s;"
      ),
      sql_string(location_id),
      sql_string(metric$metric_id)
    ),
    db_path = db_path
  )

  earliest_observed_at <- if (nrow(latest_data)) latest_data$earliest_observed_at[1] else ""
  latest_observed_at <- if (nrow(latest_data)) latest_data$latest_observed_at[1] else ""
  if (is.na(latest_observed_at) || !nzchar(latest_observed_at)) {
    return(requested_start_date)
  }

  earliest_observed_day <- as.Date(earliest_observed_at, format = "%Y-%m-%dT%H:%M:%SZ")
  latest_observed_day <- as.Date(latest_observed_at, format = "%Y-%m-%dT%H:%M:%SZ")
  if (!is.na(earliest_observed_day) && earliest_observed_day > requested_start_date) {
    return(requested_start_date)
  }
  if (latest_observed_day > end_date) {
    return(end_date + 1L)
  }

  latest_observed_day - as.integer(overlap_days)
}


get_nearby_stations_for_location <- function(
  location_id,
  allstations = load_station_catalog(),
  locations = get_rain_locations()
) {
  location <- get_rain_location(location_id, locations = locations)

  getIdFromCoords(
    coords = c(location$latitude[1], location$longitude[1]),
    stations = allstations,
    N = location$station_count[1]
  )
}


fetch_rainfall_from_api <- function(
  location_id,
  start_date,
  end_date,
  headers,
  allstations = load_station_catalog(),
  locations = get_rain_locations(),
  btw_station_sleep = 0.25,
  within_station_sleep = 1
) {
  if (is.null(headers)) {
    stop("API credentials are not available.")
  }

  location <- get_rain_location(location_id, locations = locations)

  getAllFromCoord(
    coord = c(location$latitude[1], location$longitude[1]),
    start_date = format(as.Date(start_date), "%Y-%m-%d"),
    end_date = format(as.Date(end_date), "%Y-%m-%d"),
    allstations = allstations,
    N = location$station_count[1],
    headers = headers,
    btw_station_sleep = btw_station_sleep,
    within_statio_sleep = within_station_sleep
  )
}


fetch_observations_from_api <- function(
  location_id,
  headers,
  allstations = load_station_catalog(),
  locations = get_rain_locations(),
  btw_station_sleep = 0.25
) {
  if (is.null(headers)) {
    stop("API credentials are not available.")
  }

  nearby_stations <- get_nearby_stations_for_location(
    location_id = location_id,
    allstations = allstations,
    locations = locations
  )

  measurement_rows <- lapply(seq_len(nrow(nearby_stations)), function(row_index) {
    station_id <- as.character(nearby_stations$Id_station[row_index])
    station_name <- as.character(nearby_stations$Nom_usuel[row_index])

    raw_data <- tryCatch(
      getStationPaquet(id_station = station_id, headers = headers),
      error = function(error) {
        message(conditionMessage(error))
        NULL
      }
    )

    Sys.sleep(btw_station_sleep)

    normalise_observation_package_data(
      raw_data = raw_data,
      location_id = location_id,
      station_name = station_name
    )
  })

  measurement_rows <- Filter(function(x) !is.null(x) && nrow(x), measurement_rows)
  if (!length(measurement_rows)) {
    return(empty_weather_cache())
  }

  do.call(rbind, measurement_rows)
}


sync_location_rainfall <- function(
  location_id,
  db_path = default_rain_db_path(),
  headers = load_api_headers(),
  allstations = load_station_catalog(),
  locations = get_rain_locations(),
  start_date = NULL,
  end_date = Sys.Date(),
  initial_backfill_days = 21L,
  overlap_days = 1L,
  chunk_days = 120L,
  btw_station_sleep = 0.25,
  within_station_sleep = 1
) {
  ensure_weather_db(db_path)

  if (is.null(headers)) {
    stop("API credentials are not available. Add token4 to data/secrets.")
  }

  end_date <- as.Date(end_date)
  if (is.null(start_date)) {
    start_date <- get_sync_start_date(
      location_id = location_id,
      db_path = db_path,
      end_date = end_date,
      initial_backfill_days = initial_backfill_days,
      overlap_days = overlap_days,
      metric_id = "rain_6m"
    )
  }

  start_date <- as.Date(start_date)
  if (start_date > end_date) {
    return(data.frame(
      location_id = location_id,
      start_date = "",
      end_date = as.character(end_date),
      rows_fetched = 0L,
      rows_written = 0L,
      status = "up_to_date",
      message = "",
      stringsAsFactors = FALSE
    ))
  }

  ranges <- split_sync_ranges(
    start_date = start_date,
    end_date = end_date,
    chunk_days = chunk_days
  )

  rows_fetched <- 0L
  rows_written <- 0L

  for (range_index in seq_len(nrow(ranges))) {
    raw_data <- fetch_rainfall_from_api(
      location_id = location_id,
      start_date = ranges$start_date[range_index],
      end_date = ranges$end_date[range_index],
      headers = headers,
      allstations = allstations,
      locations = locations,
      btw_station_sleep = btw_station_sleep,
      within_station_sleep = within_station_sleep
    )

    weather_data <- normalise_rainfall_data(
      raw_data = raw_data,
      location_id = location_id
    )

    rows_fetched <- rows_fetched + nrow(weather_data)
    rows_written <- rows_written + upsert_weather_measurements(
      weather_data = weather_data,
      db_path = db_path
    )
  }

  data.frame(
    location_id = location_id,
    start_date = as.character(start_date),
    end_date = as.character(end_date),
    rows_fetched = rows_fetched,
    rows_written = rows_written,
    status = "ok",
    message = "",
    stringsAsFactors = FALSE
  )
}


sync_location_observations <- function(
  location_id,
  db_path = default_rain_db_path(),
  headers = load_api_headers(),
  allstations = load_station_catalog(),
  locations = get_rain_locations(),
  btw_station_sleep = 0.25
) {
  ensure_weather_db(db_path)

  if (is.null(headers)) {
    stop("API credentials are not available. Add token4 to data/secrets.")
  }

  weather_data <- fetch_observations_from_api(
    location_id = location_id,
    headers = headers,
    allstations = allstations,
    locations = locations,
    btw_station_sleep = btw_station_sleep
  )

  rows_written <- upsert_weather_measurements(
    weather_data = weather_data,
    db_path = db_path
  )

  data.frame(
    location_id = location_id,
    rows_fetched = nrow(weather_data),
    rows_written = rows_written,
    status = "ok",
    message = "",
    stringsAsFactors = FALSE
  )
}


sync_location_weather <- function(
  location_id,
  db_path = default_rain_db_path(),
  headers = load_api_headers(),
  allstations = load_station_catalog(),
  locations = get_rain_locations(),
  start_date = NULL,
  end_date = Sys.Date(),
  initial_backfill_days = 21L,
  overlap_days = 1L,
  chunk_days = 120L,
  btw_station_sleep = 0.25,
  within_station_sleep = 1
) {
  rain_result <- sync_location_rainfall(
    location_id = location_id,
    db_path = db_path,
    headers = headers,
    allstations = allstations,
    locations = locations,
    start_date = start_date,
    end_date = end_date,
    initial_backfill_days = initial_backfill_days,
    overlap_days = overlap_days,
    chunk_days = chunk_days,
    btw_station_sleep = btw_station_sleep,
    within_station_sleep = within_station_sleep
  )

  observation_result <- sync_location_observations(
    location_id = location_id,
    db_path = db_path,
    headers = headers,
    allstations = allstations,
    locations = locations,
    btw_station_sleep = btw_station_sleep
  )

  status <- get_location_cache_status(location_id, db_path = db_path)

  data.frame(
    location_id = location_id,
    start_date = rain_result$start_date[1],
    end_date = rain_result$end_date[1],
    rain_rows_fetched = rain_result$rows_fetched[1],
    rain_rows_written = rain_result$rows_written[1],
    observation_rows_fetched = observation_result$rows_fetched[1],
    observation_rows_written = observation_result$rows_written[1],
    rows_written = rain_result$rows_written[1] + observation_result$rows_written[1],
    latest_observed_at = status$latest_observed_at[1],
    status = if (identical(rain_result$status[1], "error") || identical(observation_result$status[1], "error")) "error" else "ok",
    message = paste(
      c(rain_result$message[1], observation_result$message[1]),
      collapse = " "
    ),
    stringsAsFactors = FALSE
  )
}


sync_all_weather_locations <- function(
  location_ids = get_rain_locations()$location_id,
  db_path = default_rain_db_path(),
  headers = load_api_headers(),
  allstations = load_station_catalog(),
  locations = get_rain_locations(),
  start_date = NULL,
  end_date = Sys.Date(),
  initial_backfill_days = 21L,
  overlap_days = 1L,
  chunk_days = 120L,
  btw_station_sleep = 0.25,
  within_station_sleep = 1
) {
  results <- lapply(location_ids, function(location_id) {
    tryCatch(
      sync_location_weather(
        location_id = location_id,
        db_path = db_path,
        headers = headers,
        allstations = allstations,
        locations = locations,
        start_date = start_date,
        end_date = end_date,
        initial_backfill_days = initial_backfill_days,
        overlap_days = overlap_days,
        chunk_days = chunk_days,
        btw_station_sleep = btw_station_sleep,
        within_station_sleep = within_station_sleep
      ),
      error = function(error) {
        data.frame(
          location_id = location_id,
          start_date = if (is.null(start_date)) "" else as.character(as.Date(start_date)),
          end_date = as.character(as.Date(end_date)),
          rain_rows_fetched = 0L,
          rain_rows_written = 0L,
          observation_rows_fetched = 0L,
          observation_rows_written = 0L,
          rows_written = 0L,
          latest_observed_at = "",
          status = "error",
          message = conditionMessage(error),
          stringsAsFactors = FALSE
        )
      }
    )
  })

  do.call(rbind, results)
}


sync_all_rain_locations <- function(
  location_ids = get_rain_locations()$location_id,
  db_path = default_rain_db_path(),
  headers = load_api_headers(),
  allstations = load_station_catalog(),
  locations = get_rain_locations(),
  start_date = NULL,
  end_date = Sys.Date(),
  initial_backfill_days = 21L,
  overlap_days = 1L,
  chunk_days = 120L,
  btw_station_sleep = 0.25,
  within_station_sleep = 1
) {
  sync_all_weather_locations(
    location_ids = location_ids,
    db_path = db_path,
    headers = headers,
    allstations = allstations,
    locations = locations,
    start_date = start_date,
    end_date = end_date,
    initial_backfill_days = initial_backfill_days,
    overlap_days = overlap_days,
    chunk_days = chunk_days,
    btw_station_sleep = btw_station_sleep,
    within_station_sleep = within_station_sleep
  )
}


compute_plot_limits <- function(values) {
  values <- suppressWarnings(as.numeric(values))
  values <- values[is.finite(values)]

  if (!length(values)) {
    return(c(0, 1))
  }

  lower_limit <- min(values)
  upper_limit <- max(values)

  if (identical(lower_limit, upper_limit)) {
    padding <- if (upper_limit == 0) 1 else abs(upper_limit) * 0.15
    return(c(lower_limit - padding, upper_limit + padding))
  }

  padding <- (upper_limit - lower_limit) * 0.1
  c(lower_limit - padding, upper_limit + padding)
}


get_metric_plot_y_label <- function(
  metric_id,
  view = c("raw", "daily", "weekly", "monthly")
) {
  view <- match.arg(view)
  metric <- get_weather_metric(metric_id)

  if (!identical(metric_id, "rain_6m") || identical(view, "raw")) {
    return(sprintf("%s (%s)", metric$label, metric$unit))
  }

  switch(
    view,
    daily = "Daily rain total (mm)",
    weekly = "Weekly rain total (mm)",
    monthly = "Monthly rain total (mm)"
  )
}


plot_cached_metric <- function(
  metric_data,
  metric_id,
  view = c("raw", "daily", "weekly", "monthly"),
  main = NULL
) {
  view <- match.arg(view)
  metric <- get_weather_metric(metric_id)

  if (is.null(metric_data) || !nrow(metric_data)) {
    stop(sprintf("No cached %s data available.", tolower(metric$label)))
  }

  station_names <- unique(as.character(metric_data$station_name))
  colours <- grDevices::hcl.colors(length(station_names), "Dark 3")
  names(colours) <- station_names

  y_limits <- compute_plot_limits(metric_data$value_num)
  y_label <- get_metric_plot_y_label(metric_id = metric_id, view = view)

  if (view != "raw") {
    metric_data$observed_day <- as.Date(metric_data$observed_day)

    plot(
      range(metric_data$observed_day, na.rm = TRUE),
      y_limits,
      type = "n",
      xlab = "",
      ylab = y_label,
      main = main
    )

    for (station_name in station_names) {
      station_data <- metric_data[metric_data$station_name == station_name, , drop = FALSE]
      lines(
        station_data$observed_day,
        station_data$value_num,
        type = "o",
        col = colours[[station_name]],
        lwd = 2,
        pch = 19
      )
    }
  } else {
    metric_data$observed_at <- as.POSIXct(metric_data$observed_at, tz = "UTC")

    plot(
      range(metric_data$observed_at, na.rm = TRUE),
      y_limits,
      type = "n",
      xlab = "",
      ylab = y_label,
      main = main
    )

    for (station_name in station_names) {
      station_data <- metric_data[metric_data$station_name == station_name, , drop = FALSE]
      lines(
        station_data$observed_at,
        station_data$value_num,
        col = colours[[station_name]],
        lwd = 1.5
      )
      points(
        station_data$observed_at,
        station_data$value_num,
        pch = 19,
        col = grDevices::adjustcolor(colours[[station_name]], 0.55),
        cex = 0.85
      )
    }
  }

  legend(
    "topleft",
    legend = station_names,
    col = colours,
    lwd = 2,
    pch = 19,
    bty = "n"
  )
}


plot_cached_rainfall <- function(
  rain_data,
  view = c("raw", "daily", "weekly", "monthly"),
  hide_zero = TRUE,
  main = NULL
) {
  view <- match.arg(view)

  if (is.null(rain_data) || !nrow(rain_data)) {
    stop("No rainfall data available.")
  }

  if (hide_zero) {
    rain_data <- rain_data[is.na(rain_data$rain_mm) | rain_data$rain_mm > 0, , drop = FALSE]
  }

  if (!nrow(rain_data)) {
    stop("No rainfall data left after removing zero values.")
  }

  rain_data$value_num <- rain_data$rain_mm

  plot_cached_metric(
    metric_data = rain_data,
    metric_id = "rain_6m",
    view = view,
    main = main
  )
}