args <- commandArgs(trailingOnly = FALSE)
file_arg <- grep("^--file=", args, value = TRUE)
analysis_dir <- if (length(file_arg)) dirname(normalizePath(sub("^--file=", "", file_arg[[1]]), mustWork = FALSE)) else file.path(getwd(), "analysis")
source(file.path(analysis_dir, "00_config.R"))

md5_file <- function(path) {
  value <- unname(tools::md5sum(path))
  if (length(value) != 1L || is.na(value)) stop("Could not calculate MD5 for ", path)
  tolower(value)
}

utc_now <- function() format(Sys.time(), tz = "UTC", usetz = TRUE)

replace_file <- function(staged, destination) {
  dir.create(dirname(destination), recursive = TRUE, showWarnings = FALSE)
  backup <- paste0(destination, ".backup-", Sys.getpid())
  if (file.exists(backup)) unlink(backup)
  had_destination <- file.exists(destination)
  if (had_destination && !file.rename(destination, backup)) {
    stop("Could not stage existing file for replacement: ", destination)
  }
  installed <- file.rename(staged, destination)
  if (!installed) {
    if (had_destination) file.rename(backup, destination)
    stop("Could not install validated file: ", destination)
  }
  if (had_destination) unlink(backup)
  invisible(destination)
}

replace_directory <- function(staged, destination) {
  backup <- paste0(destination, ".backup-", Sys.getpid())
  if (dir.exists(backup)) unlink(backup, recursive = TRUE, force = TRUE)
  had_destination <- dir.exists(destination)
  if (had_destination && !file.rename(destination, backup)) {
    stop("Could not stage existing directory for replacement: ", destination)
  }
  installed <- file.rename(staged, destination)
  if (!installed) {
    if (had_destination) file.rename(backup, destination)
    stop("Could not install validated directory: ", destination)
  }
  if (had_destination) unlink(backup, recursive = TRUE, force = TRUE)
  invisible(destination)
}

manifest_path <- file.path(paths$raw, "download_manifest.csv")
previous_manifest <- if (file.exists(manifest_path)) {
  tryCatch(read.csv(manifest_path, stringsAsFactors = FALSE), error = function(error) data.frame())
} else {
  data.frame()
}

prior_retrieval_time <- function(key, observed_md5, destination) {
  if (nrow(previous_manifest) && all(c("key", "md5", "retrieved_utc") %in% names(previous_manifest))) {
    row <- previous_manifest[previous_manifest$key == key & tolower(previous_manifest$md5) == observed_md5, , drop = FALSE]
    if (nrow(row) == 1L && nzchar(row$retrieved_utc)) return(row$retrieved_utc)
  }
  format(file.info(destination)$mtime, tz = "UTC", usetz = TRUE)
}

download_one <- function(key) {
  destination <- file.path(paths$raw, unname(raw_files[[key]]))
  expected <- tolower(unname(expected_source_md5[[key]]))
  observed <- if (file.exists(destination) && file.info(destination)$size > 0) md5_file(destination) else NA_character_

  if (isTRUE(identical(observed, expected))) {
    message("Using hash-verified cached source: ", basename(destination))
    return(list(
      path = destination,
      status = "cached_verified",
      retrieved_utc = prior_retrieval_time(key, observed, destination)
    ))
  }

  if (file.exists(destination)) {
    message("Cached source failed its frozen fingerprint and will not be reused: ", basename(destination))
  } else {
    message("Official source is not cached: ", basename(destination))
  }

  staged <- tempfile(pattern = paste0(".", basename(destination), "-"), tmpdir = paths$raw)
  on.exit(if (file.exists(staged)) unlink(staged), add = TRUE)
  message("Downloading ", key, " from official TIMSS source")
  download.file(urls[[key]], staged, mode = "wb", method = "libcurl", quiet = FALSE)
  if (!file.exists(staged) || file.info(staged)$size <= 0) stop("Downloaded source is empty: ", key)
  downloaded_md5 <- md5_file(staged)
  if (!identical(downloaded_md5, expected)) {
    stop(
      "Official source fingerprint changed for ", key, ". Expected ", expected,
      " but downloaded ", downloaded_md5,
      ". Review the upstream change and update the frozen fingerprint only after validation."
    )
  }
  replace_file(staged, destination)
  list(path = destination, status = "downloaded_verified", retrieved_utc = utc_now())
}

acquired <- lapply(names(urls), download_one)
names(acquired) <- names(urls)
local_files <- vapply(acquired, `[[`, character(1), "path")
observed_md5 <- vapply(local_files, md5_file, character(1))
if (!identical(unname(observed_md5), unname(tolower(expected_source_md5[names(local_files)])))) {
  stop("At least one local source differs from its frozen fingerprint after acquisition")
}

manifest <- data.frame(
  key = names(local_files),
  filename = basename(local_files),
  url = unname(urls[names(local_files)]),
  bytes = unname(file.info(local_files)$size),
  expected_md5 = unname(expected_source_md5[names(local_files)]),
  md5 = unname(observed_md5),
  cache_status = vapply(acquired, `[[`, character(1), "status"),
  retrieved_utc = vapply(acquired, `[[`, character(1), "retrieved_utc"),
  verified_utc = utc_now(),
  stringsAsFactors = FALSE
)
manifest_stage <- tempfile(pattern = ".download-manifest-", tmpdir = paths$raw)
write.csv(manifest, manifest_stage, row.names = FALSE)
replace_file(manifest_stage, manifest_path)
message("Wrote hash-verified source manifest: ", manifest_path)

extracted_file_manifest <- function(directory, key, archive_path, archive_md5) {
  files <- sort(list.files(directory, recursive = TRUE, full.names = TRUE, all.files = FALSE, no.. = TRUE))
  files <- files[file.info(files)$isdir %in% FALSE]
  if (!length(files)) stop("Archive produced no files: ", archive_path)
  prefix <- paste0(normalizePath(directory, mustWork = TRUE), .Platform$file.sep)
  relative <- substring(normalizePath(files, mustWork = TRUE), nchar(prefix) + 1L)
  data.frame(
    source_key = key,
    archive_filename = basename(archive_path),
    archive_md5 = archive_md5,
    relative_path = relative,
    bytes = unname(file.info(files)$size),
    md5 = unname(tools::md5sum(files)),
    stringsAsFactors = FALSE
  )
}

extraction_cache_valid <- function(directory, key, archive_path, archive_md5) {
  marker <- file.path(directory, ".extraction_manifest.csv")
  if (!dir.exists(directory) || !file.exists(marker)) return(FALSE)
  recorded <- tryCatch(read.csv(marker, stringsAsFactors = FALSE), error = function(error) data.frame())
  required <- c("source_key", "archive_filename", "archive_md5", "relative_path", "bytes", "md5")
  if (!nrow(recorded) || !all(required %in% names(recorded))) return(FALSE)
  if (any(recorded$source_key != key) || any(tolower(recorded$archive_md5) != archive_md5)) return(FALSE)
  current <- tryCatch(
    extracted_file_manifest(directory, key, archive_path, archive_md5),
    error = function(error) data.frame()
  )
  if (!nrow(current) || !identical(recorded$relative_path, current$relative_path)) return(FALSE)
  identical(as.numeric(recorded$bytes), as.numeric(current$bytes)) &&
    identical(tolower(recorded$md5), tolower(current$md5))
}

refresh_extraction <- function(key, destination) {
  archive_path <- local_files[[key]]
  archive_md5 <- observed_md5[[key]]
  if (extraction_cache_valid(destination, key, archive_path, archive_md5)) {
    recorded <- read.csv(file.path(destination, ".extraction_manifest.csv"), stringsAsFactors = FALSE)
    message("Using source- and content-verified extraction cache: ", basename(destination))
    return(data.frame(
      source_key = key,
      archive_filename = basename(archive_path),
      archive_md5 = archive_md5,
      extracted_files = nrow(recorded),
      extracted_bytes = sum(recorded$bytes),
      cache_status = "cached_verified",
      verified_utc = utc_now(),
      stringsAsFactors = FALSE
    ))
  }

  parent <- dirname(destination)
  dir.create(parent, recursive = TRUE, showWarnings = FALSE)
  staged <- tempfile(pattern = paste0(".", basename(destination), "-"), tmpdir = parent)
  dir.create(staged, recursive = TRUE, showWarnings = FALSE)
  on.exit(if (dir.exists(staged)) unlink(staged, recursive = TRUE, force = TRUE), add = TRUE)
  message("Refreshing extraction cache from verified archive: ", basename(archive_path))
  utils::unzip(archive_path, exdir = staged, overwrite = TRUE)
  extracted <- extracted_file_manifest(staged, key, archive_path, archive_md5)
  write.csv(extracted, file.path(staged, ".extraction_manifest.csv"), row.names = FALSE)
  replace_directory(staged, destination)
  data.frame(
    source_key = key,
    archive_filename = basename(archive_path),
    archive_md5 = archive_md5,
    extracted_files = nrow(extracted),
    extracted_bytes = sum(extracted$bytes),
    cache_status = "refreshed_verified",
    verified_utc = utc_now(),
    stringsAsFactors = FALSE
  )
}

extract_map <- list(
  data_r_g45 = file.path(paths$raw, "r_g45"),
  data_r_g89 = file.path(paths$raw, "r_g89"),
  r_programs = file.path(paths$raw, "r_programs")
)
extraction_summary <- do.call(rbind, lapply(names(extract_map), function(key) {
  refresh_extraction(key, extract_map[[key]])
}))
rownames(extraction_summary) <- NULL
extraction_summary_path <- file.path(paths$raw, "extraction_manifest.csv")
extraction_summary_stage <- tempfile(pattern = ".extraction-summary-", tmpdir = paths$raw)
write.csv(extraction_summary, extraction_summary_stage, row.names = FALSE)
replace_file(extraction_summary_stage, extraction_summary_path)
message("Wrote extraction lineage summary: ", extraction_summary_path)
