# Seed a new slot with the pure-R binaries of an existing one.
#
# Usage: Rscript local/seed-pure-r.R DONOR [DONOR...] [--workers=N] [--dry-run]
#   e.g. Rscript local/seed-pure-r.R alpine324 alpine323
#
# Three quarters of the versions a full rebuild targets have no compiled code.
# Building one costs about 20s, nearly all of it dependency installs and
# bookkeeping, and the result is the same R objects every other slot already
# serves: R6 2.6.1 deparses identically on amd64/alpine323, amd64/alpine324 and
# arm64/rhel9. Copying those from a slot of the same OS family and arch saves
# roughly 340 shard-hours per arch and leaves build-all with the compiled rest.
#
# Each candidate is downloaded and checked before it is copied server-side (see
# seed_verdict()): a binary, no compiled code, built by this image's primary R
# minor, and portable across every installed minor by bincraft's own symbol
# inspection. A seeded object then gets the two metadata rows an upload would
# have written: a successful `single_builds` row, so build-all skips it, and an
# `abi_flat_safety` row, without which the per-minor indexes drop it.
#
# Donors are tried in order; the first one with a valid binary wins. Re-running
# is safe: whatever the target already holds is skipped, and the metadata rows
# of earlier copies are re-written idempotently.

source(file.path("local", "seed-helpers.R"))
source(file.path("local", "r-minor-helpers.R"))

args <- commandArgs(trailingOnly = TRUE)
donors <- args[!startsWith(args, "--")]
if (length(donors) == 0L) {
  stop("Usage: seed-pure-r.R DONOR [DONOR...] [--workers=N] [--dry-run]")
}
dry_run <- "--dry-run" %in% args
workers_arg <- sub("^--workers=", "", grep("^--workers=", args, value = TRUE))
workers <- if (length(workers_arg)) as.integer(workers_arg[1L]) else 8L
batch_size <- 400L

local_machine <- Sys.info()[["machine"]]
arch <- if (grepl("arm64|aarch64", local_machine)) "arm64" else "amd64"
target <- bincraft::set_codename(NULL)
platform <- codename_platform(target)
target_minor <- paste(
  R.version$major,
  strsplit(R.version$minor, ".", fixed = TRUE)[[1L]][1L],
  sep = "."
)
for (donor in donors) {
  if (identical(donor, target)) {
    stop(sprintf("Donor %s is the target slot itself", donor))
  }
  if (!identical(codename_family(donor), codename_family(target))) {
    stop(sprintf(
      "Donor %s is not an %s slot; seeding only copies within one OS family",
      donor,
      codename_family(target)
    ))
  }
}
if (is.na(platform)) {
  stop(sprintf("No single_builds platform name for codename %s", target))
}

# Computed once: bincraft would otherwise run nm over every libR per tarball.
symbol_sets <- bincraft:::installed_r_symbol_sets()
if (is.null(symbol_sets)) {
  stop(
    "Fewer than two R minors under /opt/R; ABI portability cannot be ",
    "verified, so nothing is seeded"
  )
}
message(sprintf(
  "Seeding %s/%s (R %s) from %s; ABI checked against R %s",
  arch,
  target,
  target_minor,
  paste(donors, collapse = ", "),
  paste(names(symbol_sets), collapse = ", ")
))

s3fs::s3_file_system(
  aws_access_key_id = Sys.getenv("B2_S3_ACCESS_KEY"),
  aws_secret_access_key = Sys.getenv("B2_S3_SECRET_KEY"),
  endpoint = "https://s3.eu-central-003.backblazeb2.com",
  region_name = "eu-central-003",
  refresh = TRUE
)
slot_dir <- function(codename) {
  sprintf("devxy-rpkgs-binaries/%s/%s/latest/src/contrib", arch, codename)
}
# Keys relative to the contrib dir with unquoted ETags. A listing failure is
# fatal: an empty target listing would mean re-copying everything.
list_slot <- function(codename) {
  info <- s3fs::s3_dir_info(slot_dir(codename), recurse = TRUE)
  uri <- as.character(info$uri)
  is_tarball <- grepl("\\.tar\\.gz$", uri)
  data.frame(
    key = sub(
      sprintf("^(s3://)?%s/", slot_dir(codename)),
      "",
      uri[is_tarball]
    ),
    etag = gsub('^"|"$', "", as.character(info$etag[is_tarball])),
    stringsAsFactors = FALSE
  )
}

cran <- tools::CRAN_package_db()
compiled_pkgs <- cran$Package[tolower(cran$NeedsCompilation) %in% "yes"]
cran_md5 <- bincraft:::cran_source_md5()
exclude <- jsonlite::fromJSON("local/excluded-packages.json")[["package"]]

db_connect <- function() {
  DBI::dbConnect(
    RPostgres::Postgres(),
    dbname = "build_metadata",
    host = "r-binaries.devxy.io",
    port = 15432,
    user = "rpkgs",
    password = Sys.getenv("PGPASS"),
    sslmode = "verify-full",
    sslrootcert = bincraft::postgres_sslrootcert("verify-full")
  )
}

# The rows an upload would have written for these objects. Idempotent, so a
# re-run can repair rows a failed write left out. Pure-R objects reference no R
# symbols, hence an empty `unsupported`.
record_seeded <- function(packages, versions) {
  if (length(packages) == 0L) {
    return(invisible(TRUE))
  }
  ok <- tryCatch(
    {
      con <- db_connect()
      on.exit(DBI::dbDisconnect(con), add = TRUE)
      bincraft:::ensure_flat_safety_table(con, "abi_flat_safety")
      DBI::dbWithTransaction(con, {
        DBI::dbWriteTable(
          con,
          "seed_rows",
          data.frame(
            name = packages,
            tag = versions,
            stringsAsFactors = FALSE
          ),
          temporary = TRUE,
          overwrite = TRUE
        )
        DBI::dbExecute(
          con,
          paste(
            "INSERT INTO single_builds (name, tag, platform, arch,",
            "error_occurred, error_text, size, timestamp, duration, removed,",
            "r_version)",
            "SELECT s.name, s.tag, $1, $2, FALSE, NULL, NULL, now(), NULL,",
            "FALSE, $3 FROM seed_rows s WHERE NOT EXISTS (SELECT 1 FROM",
            "single_builds b WHERE b.name = s.name AND b.tag = s.tag",
            "AND b.platform = $1 AND b.arch = $2 AND b.r_version = $3)"
          ),
          params = list(platform, arch, target_minor)
        )
        DBI::dbExecute(
          con,
          paste(
            "INSERT INTO abi_flat_safety (package, version, codename, arch,",
            "safe, unsupported) SELECT s.name, s.tag, $1, $2, TRUE, ''",
            "FROM seed_rows s ON CONFLICT (package, version, codename, arch)",
            "DO UPDATE SET safe = EXCLUDED.safe,",
            "unsupported = EXCLUDED.unsupported, checked_at = now()"
          ),
          params = list(target, arch)
        )
      })
      TRUE
    },
    error = function(e) {
      message(sprintf(
        "WARNING: could not record %d seeded object(s): %s. Re-run to repair.",
        length(packages),
        conditionMessage(e)
      ))
      FALSE
    }
  )
  invisible(ok)
}

target_listing <- list_slot(target)
message(sprintf("Target %s holds %d tarball(s)", target, nrow(target_listing)))

future::plan(future::multisession, workers = workers)
tally <- integer()
count <- function(reasons) {
  for (r in reasons) {
    tally[[r]] <<- (if (is.na(tally[r])) 0L else tally[[r]]) + 1L
  }
}

for (donor in donors) {
  donor_listing <- list_slot(donor)

  # Byte-identical copies an earlier run made: make sure their rows exist.
  same <- merge(donor_listing, target_listing, by = c("key", "etag"))
  same <- same[!per_minor_key(same$key), , drop = FALSE]
  parsed <- split_tarball_name(basename(same$key))
  ours <- !is.na(parsed$package) & !parsed$package %in% compiled_pkgs
  if (any(ours) && !dry_run) {
    record_seeded(parsed$package[ours], parsed$version[ours])
  }

  candidates <- seed_candidates(
    donor_listing$key,
    donor_listing$etag,
    target_listing$key,
    cran$Package,
    compiled_pkgs,
    cran_md5,
    exclude
  )
  message(sprintf(
    "%s: %d tarball(s), %d candidate(s) after the metadata filters",
    donor,
    nrow(donor_listing),
    nrow(candidates)
  ))
  if (dry_run || nrow(candidates) == 0L) {
    next
  }

  batches <- split(
    seq_len(nrow(candidates)),
    ceiling(seq_len(nrow(candidates)) / batch_size)
  )
  for (b in seq_along(batches)) {
    batch <- candidates[batches[[b]], , drop = FALSE]
    dl_dir <- tempfile("seed_")
    dir.create(dl_dir)
    local_files <- file.path(dl_dir, basename(batch$key))
    downloaded <- tryCatch(
      {
        s3fs::s3_file_download(
          file.path(slot_dir(donor), batch$key),
          local_files,
          overwrite = TRUE
        )
        TRUE
      },
      error = function(e) {
        message(sprintf(
          "Batch %d: download failed (%s); leaving it to build-all",
          b,
          conditionMessage(e)
        ))
        FALSE
      }
    )
    if (!downloaded) {
      count(rep("download-failed", nrow(batch)))
      unlink(dl_dir, recursive = TRUE)
      next
    }

    verdicts <- future.apply::future_lapply(
      seq_len(nrow(batch)),
      function(i) {
        seed_verdict(
          local_files[i],
          batch$package[i],
          batch$version[i],
          target_minor,
          function(t) bincraft:::tarball_abi_verdict(t, symbol_sets)
        )
      },
      future.seed = NULL
    )
    unlink(dl_dir, recursive = TRUE)
    seed <- vapply(verdicts, function(v) isTRUE(v$seed), logical(1L))
    count(vapply(verdicts, function(v) v$reason, ""))

    if (any(seed)) {
      copied <- tryCatch(
        {
          s3fs::s3_file_copy(
            paste0("s3://", file.path(slot_dir(donor), batch$key[seed])),
            paste0("s3://", file.path(slot_dir(target), batch$key[seed])),
            overwrite = TRUE
          )
          TRUE
        },
        error = function(e) {
          message(sprintf(
            "Batch %d: copy failed (%s); leaving it to build-all",
            b,
            conditionMessage(e)
          ))
          FALSE
        }
      )
      if (copied) {
        record_seeded(batch$package[seed], batch$version[seed])
        target_listing <- rbind(
          target_listing,
          data.frame(key = batch$key[seed], etag = NA_character_)
        )
      } else {
        count(rep("copy-failed", sum(seed)))
        tally[["seeded"]] <- tally[["seeded"]] - sum(seed)
      }
    }
    message(sprintf(
      "%s batch %d/%d: %d of %d seeded",
      donor,
      b,
      length(batches),
      sum(seed),
      nrow(batch)
    ))
  }
}

message(
  "Seed summary: ",
  paste(sprintf("%s=%d", names(tally), tally), collapse = ", ")
)