R/safe_output.R

In dsROCrate: 'DataSHIELD' RO-Crate Governance Functions

#' Safe Output details
#'
#' Safe Output details for the RO-Crate.
#'
#' All research outputs are checked to ensure individuals cannot be identified
#' even in the public domain.
#'
#' This means the other four Safes no longer apply.
#'
#' Before outputs are released from the TRE, they are checked to make sure it
#' is reasonably unlikely that any individuals can be identified on publication.
#' Compliance with the Digital Economy Act also requires that the TRE applies
#' methods and standards for output checking that are accredited by the UK
#' Statistics Authority.
#'
#' @param logs_to Upper limit timestamp to filter out the outputs generated
#'     (default: `Sys.time()`, current system time).
#' @param logs_from Lower limit timestamp to filter out the outputs generated
#'     (default: `Sys.time() - 24 * 60 ^ 2`, last 24 hours).
#' @inheritParams init
#'
#' @returns Updated RO-Crate object with Safe Outputs information.
#' @export
#'
#' @source
#' \itemize{
#'  \item Research Data Scotland, 2025. "What is the Five Safes framework?".
#'  <https://www.researchdata.scot/engage-and-learn/data-explainers/what-is-the-five-safes-framework/>
#' }
safe_output <- function(x, ...) {
  UseMethod("safe_output")
}

# @rdname safe_output
#' @export
safe_output.default <- function(x, ...) {
  stop(
    "Unknown class, please try either a file path or",
    " an object with `rocrate` class!"
  )
}

#' @rdname safe_output
#' @export
safe_output.ArmadilloCredentials <- function(x, ...) {
  stop(
    "`safe_output()` for the Armadillo backend is not currently implemented!",
    call. = FALSE
  )
}

#' @rdname safe_output
#' @export
safe_output.character <- function(
  x,
  ...,
  path = attr(x, "path"),
  user = attr(x, "user"),
  logs_to = Sys.time(),
  logs_from = logs_to - 24 * 60^2,
  connection = attr(x, "connection"),
  profile = attr(x, "profile"),
  project = attr(x, "project"),
  resources = attr(x, "resources"),
  tables = attr(x, "tables")
) {
  # attempt loading the RO-Crate
  rocrate <- rocrateR::load_rocrate(x)

  # call method with given `rocrate` object:
  safe_output(
    rocrate,
    connection = connection,
    path = path,
    user = user,
    logs_to = logs_to,
    logs_from = logs_from,
    profile = profile,
    project = project,
    resources = resources,
    tables = tables
  )
}

#' @importFrom utils write.csv
#' @rdname safe_output
#' @export
safe_output.opal <- function(
  x,
  ...,
  rocrate = rocrateR::rocrate_5s(),
  path = NULL,
  user = NULL,
  logs_to = Sys.time(),
  logs_from = logs_to - 24 * 60^2,
  profile = "default",
  project = NULL,
  resources = NULL,
  tables = NULL
) {
  # local bindings
  `@timestamp` <- backend <- logger_name <- safe_people_id <- username <- NULL
  ds_action <- ds_eval <- ds_id <- ds_function <- ds_symbol <- ds_table <- NULL
  asset <- action <- is_placeholder <- kind <- symbol_id <- timestamp <- NULL
  ds_profile <- expr <- fx <- log_id <- r_cmd <- session <- symbol <- NULL
  column <- created_at <- created_by <- id <- uuid <- NULL

  if (is.null(profile)) {
    profile <- "default"
  }

  # create formatted versions of input dates
  logs_from <- parse_utc_limit(logs_from)
  logs_to <- parse_utc_limit(logs_to)

  bad <- c(logs_from = anyNA(logs_from), logs_to = anyNA(logs_to))
  if (any(bad)) {
    stop(
      "Invalid date string identified for:",
      paste0("\n  - ", names(bad)[bad], collapse = ""),
      call. = FALSE
    )
  }
  logs_from_formatted <- fmt_limit(logs_from)
  logs_to_formatted <- fmt_limit(logs_to)

  # validate backend
  validate_backend(x, ...)

  # verify if `user` is NULL, if so, retrieve information from the RO-crate
  if (is.null(user)) {
    # get `author` section from the root (./) entity
    rocrate_author <- rocrate |>
      .get_entity(id = "./") |>
      lapply(getElement, name = "author") |>
      sapply(\(x) x)

    # extract @id attribute(s)
    if (length(rocrate_author) > 1) {
      safe_people_id <- rocrate_author |>
        sapply(getElement, name = "@id") |>
        sapply(unlist)
    } else {
      safe_people_id <- rocrate_author["@id"][[1]]
    }

    # check if Safe People section wasn't found
    if (is.null(safe_people_id)) {
      warning(
        "Safe People section not found (i.e., no author for root entity) ",
        "in the given RO-Crate. \nEither run `dsROCrate::safe_people()` ",
        "or set `user` when calling `dsROCrate::safe_output()`!",
        call. = FALSE
      )

      # return the input RO-Crate
      return(rocrate)
    }

    # retrieve Safe People entity for the current user
    safe_people_information <- safe_people_id |>
      sapply(\(x) .get_entity(rocrate, id = x))

    # update user
    user <- safe_people_information |>
      sapply(getElement, name = "name") |>
      unique()
  }

  # check if the given value for `user` is a list
  if ("list" %in% class(user)) {
    user <- user |>
      purrr::pluck("name") |>
      unique()
  }

  # check if for any reason multiple users were found
  if (length(user) != 1) {
    warning(
      "Error when retrieving the Safe People section in the given ",
      "RO-Crate. ",
      length(user),
      " entries in the 'author' ",
      "section of root (./) entity were found!",
      call. = FALSE
    )

    # return the input RO-Crate
    return(rocrate)
  }

  # initialise symbol registry
  registry <- symbol_registry()

  # parse logs
  userlogs_tbl <- backend_logs(x) |>
    tibble::as_tibble() |>
    dplyr::bind_rows(tibble::tibble(ds_profile = character())) |>
    # filter logs
    dplyr::filter(`@timestamp` >= logs_from, `@timestamp` <= logs_to) |>
    dplyr::filter(logger_name == "datashield.user") |>
    dplyr::filter(ds_profile == profile) |>
    dplyr::filter(username %in% user) |>
    dplyr::mutate(id = paste0("", dplyr::row_number()))

  userlogs <- NULL
  if (nrow(userlogs_tbl) > 0) {
    userlogs <- userlogs_tbl |>
      with(
        sprintf(
          "[%s][%s]%-12s%s",
          level,
          format_utc(`@timestamp`),
          paste0("[", ds_action, "]"),
          message
        )
      )
  }

  # check if any logs were found in the given time frame
  if (length(userlogs) == 0) {
    warning(
      "No logs were found for the following configuration:",
      "\n User: ",
      user,
      "\n Period: ",
      logs_from_formatted,
      " -- ",
      logs_to_formatted,
      call. = FALSE
    )

    # return the input RO-Crate
    return(rocrate)
  }

  log_filename <- paste0(
    format(Sys.time(), "%Y%m%dT%H%M%S"),
    "-dslogs-",
    user,
    ".log"
  )

  # create new data entity for log file
  log_entity <- rocrateR::entity(
    id = basename(log_filename),
    type = "File",
    dateModified = Sys.time(),
    name = basename(log_filename),
    description = paste0(
      "This file contains the raw logs for the user: `",
      user,
      "` , between: ",
      logs_from_formatted,
      " and ",
      logs_to_formatted
    ),
    encodingFormat = "text/plain"
  )

  # update symbol registry
  ## check if Opal 6.0+ is available
  opal_6_available <- tryCatch(
    {
      validate_backend_version(x, minimum = "6.0")
      TRUE
    },
    error = function(e) FALSE
  )
  ## extract 'RESOLVE' operations from the logs
  userlogs_resolve_tbl <- NULL
  if (opal_6_available) {
    userlogs_resolve_tbl <- userlogs_tbl |>
      dplyr::filter(ds_action %in% c("RESOLVE"))
  }
  ## extract 'ASSIGN' operations from the logs
  userlogs_assign_tbl <- userlogs_tbl |>
    dplyr::filter(ds_action %in% c("ASSIGN"))

  ## reshape the logs into a tibble of `symbols`
  symbols_tbl <- if (nrow(userlogs_assign_tbl) == 0) {
    tibble::tibble(
      id = character(),
      symbol = character(),
      kind = character(),
      asset = character(),
      expr = character(),
      created_by = character(),
      created_at = as.POSIXct(character()),
      user = character(),
      action = character(),
      session = character()
    )
  } else {
    seq_len(nrow(userlogs_assign_tbl)) |>
      purrr::map(function(i) {
        # extract log components
        id <- getElement(userlogs_assign_tbl[i, ], "id")
        ds_eval <- getElement(userlogs_assign_tbl[i, ], "ds_eval")
        ds_resource <- getElement(userlogs_assign_tbl[i, ], "ds_resource")
        ds_symbol <- getElement(userlogs_assign_tbl[i, ], "ds_symbol")
        ds_table <- getElement(userlogs_assign_tbl[i, ], "ds_table")

        # evaluate which fields are populated
        is_expr <- !is.null(ds_eval) && !is.na(ds_eval)
        is_resource <- !is.null(ds_resource) && !is.na(ds_resource)
        is_table <- !is.null(ds_table) && !is.na(ds_table)

        tibble::tibble(
          id = id,
          symbol = ds_symbol,
          kind = ifelse(
            is_expr,
            'expression',
            ifelse(
              is_resource,
              'resource',
              ifelse(is_table, 'table', NA_character_)
            )
          ),
          asset = ifelse(
            is_resource,
            ds_resource,
            ifelse(is_table, ds_table, NA_character_)
          ),
          expr = ifelse(is_expr, ds_eval, NA_character_),
          created_by = ifelse(
            is_expr,
            'DSI::datashield.assign.expr',
            ifelse(
              is_resource,
              'DSI::datashield.assign.resource',
              ifelse(is_table, 'DSI::datashield.assign.table', NA_character_)
            )
          ),
          created_at = userlogs_assign_tbl$`@timestamp`[[i]],
          user = userlogs_assign_tbl$username[[i]],
          action = userlogs_assign_tbl$ds_action[[i]],
          session = userlogs_assign_tbl$ds_id[[i]]
        )
      }) |>
      dplyr::bind_rows() |>
      dplyr::distinct()
  }

  ## reshape RESOLVE operations into log entries
  resolve_tbl <- tibble::tibble(
    id = character(),
    timestamp = character(),
    action = character(),
    user = character(),
    r_cmd = character(),
    fx = character(),
    symbol = character(),
    column = character(),
    kind = character(),
    asset = character(),
    expr = character(),
    session = character(),
    backend = character()
  )

  if (!is.null(userlogs_resolve_tbl) && nrow(userlogs_resolve_tbl) > 0) {
    resolve_tbl <- userlogs_resolve_tbl |>
      # insert column placeholders for ds_resource and ds_table
      dplyr::bind_rows(tibble::tibble(
        ds_resource = character(),
        ds_table = character()
      )) |>
      dplyr::transmute(
        id,
        timestamp = format_utc(`@timestamp`),
        action = ds_action,
        user = username,
        r_cmd = dplyr::case_when(
          !is.na(ds_resource) ~ paste0(
            "RESOLVE ",
            ds_resource,
            " -> ",
            ds_symbol
          ),
          !is.na(ds_table) ~ paste0(
            "RESOLVE ",
            ds_table,
            " -> ",
            ds_symbol
          ),
          TRUE ~ NA_character_
        ),
        fx = NA_character_,
        symbol = ds_symbol,
        column = NA_character_,
        kind = dplyr::case_when(
          !is.na(ds_resource) ~ "resource",
          !is.na(ds_table) ~ "table",
          TRUE ~ NA_character_
        ),
        asset = dplyr::case_when(
          !is.na(ds_resource) ~ ds_resource,
          !is.na(ds_table) ~ ds_table,
          TRUE ~ NA_character_
        ),
        expr = NA_character_,
        session = ds_id,
        backend = "OBiBa's Opal"
      )
  }

  ## add symbols to registry
  registry <- symbols_tbl |>
    purrr::pmap(safe_symbol) |>
    purrr::reduce(register_symbol, .init = registry)

  # parse aggregate function calls into list of safe_call objects
  calls_lst <- userlogs_tbl |>
    dplyr::filter((ds_action %in% c("AGGREGATE"))) |>
    # dplyr::filter(!is.na(ds_eval)) |>
    purrr::pmap(function(ds_eval, ...) {
      safe_call(ds_eval, ...) |>
        enrich_call(registry = registry)
    })

  if (length(calls_lst) == 0) {
    calls_symbols_tbl <- tibble::tibble(
      id = character(),
      timestamp = character(),
      action = character(),
      user = character(),
      r_cmd = character(),
      fx = character(),
      symbol = character(),
      column = character(),
      kind = character(),
      asset = character(),
      expr = character(),
      session = character(),
      backend = character()
    )
  } else {
    # convert list of calls into tibble
    calls_tbl <- calls_to_tbl(calls_lst, registry)

    # combine function calls with symbol's registry
    calls_symbols_tbl <- calls_tbl |>
      dplyr::select(-symbol) |>
      dplyr::mutate(
        args = purrr::map(
          args,
          ~ purrr::imap_dfr(.x, function(arg, nm) {
            if (!inherits(arg, "safe_reference")) {
              tibble::tibble(
                argument = nm,
                value = list(arg),
                symbol_id = NA_character_,
                symbol = NA_character_,
                column = NA_character_
              )
            } else {
              tibble::tibble(
                argument = nm,
                value = list(arg),
                symbol_id = arg$symbol_id,
                symbol = arg$symbol,
                column = arg$column
              )
            }
          })
        )
      ) |>
      (\(x) {
        purrr::map2(
          split(x |> dplyr::select(-args), seq_len(nrow(x))),
          x$args,
          dplyr::bind_cols
        )
      })() |>
      purrr::list_c() |>
      dplyr::left_join(
        registry$symbols,
        by = c("symbol_id" = "id"),
        suffix = c("", "_registry")
      ) |>
      dplyr::mutate(
        asset = dplyr::if_else(
          kind == "expression",
          purrr::map_chr(
            symbol_id,
            resolve_symbol_asset,
            registry = registry
          ),
          asset
        )
      ) |>
      # add column with backend
      dplyr::mutate(backend = "OBiBa's Opal") |>
      # subset columns
      dplyr::select(
        id,
        timestamp,
        action,
        user,
        r_cmd,
        fx,
        symbol,
        column,
        kind,
        asset,
        expr,
        # table = ds_table,
        session,
        backend
      )
  }

  # extract session details
  session_tbl <- userlogs_tbl |>
    dplyr::filter((ds_action %in% c("OPEN"))) |>
    dplyr::mutate(
      # format timestamp
      timestamp = format_utc(`@timestamp`),
      # attach session ID, `ds_id`, if `ds_action == 'OPEN'`
      ds_eval = paste0("Open session: ", ds_id),
      # set `ds_function = 'DSI::datashield.login'`
      ds_function = "DSI::datashield.login",
      backend = "OBiBa's Opal"
    ) |>
    dplyr::select(
      id,
      timestamp,
      action = ds_action,
      user = username,
      r_cmd = ds_eval,
      fx = ds_function,
      session = ds_id,
      backend
    )

  # parse symbols to log entries
  assign_tbl <- symbols_tbl |>
    dplyr::transmute(
      id,
      timestamp = format_utc(created_at),
      action = action,
      user,
      r_cmd = dplyr::case_when(
        kind == "table" ~ paste0(symbol, " <- opal[", asset, "]"),
        kind == "resource" ~ paste0(symbol, " <- opal[", asset, "]"),
        kind == "expression" ~ expr,
        TRUE ~ NA_character_
      ),
      fx = created_by,
      symbol,
      column = NA_character_,
      kind,
      asset,
      expr,
      session,
      backend = "OBiBa's Opal"
    )

  # combine the logs
  userlogs_tbl_maps_evals <- dplyr::bind_rows(
    resolve_tbl,
    assign_tbl,
    calls_symbols_tbl,
    session_tbl
  ) |>
    dplyr::distinct() |>
    dplyr::arrange(as.numeric(id), timestamp) |>
    # regenerate unique operation IDs
    (\(.) {
      dplyr::left_join(
        .,
        tibble::tibble(
          id = unique(.$id),
          uuid = uuid::UUIDgenerate(n = length(id))
        ),
        by = "id"
      )
    })() |>
    dplyr::mutate(id = uuid) |>
    dplyr::select(-uuid)

  log_maps_filename <- paste0(
    format(Sys.time(), "%Y%m%dT%H%M%S"),
    "-dslogs-",
    user,
    "_mappings.csv"
  )

  # create new data entity for log file
  log_maps_entity <- rocrateR::entity(
    id = basename(log_maps_filename),
    type = "File",
    dateModified = Sys.time(),
    name = basename(log_maps_filename),
    description = "This file contains mappings and evaluated functions",
    encodingFormat = "text/csv"
  )

  # check if a `path` was not provided, then display warning and store contents
  # inside the RO-Crate entity
  if (is.null(path)) {
    warning(
      "A `path` wasn't provided! The logs will be included in the ",
      "RO-Crate object, under the `content` tag!",
      call. = FALSE
    )
    log_entity$content <- list(userlogs)
    log_maps_entity$content <- list(userlogs_tbl_maps_evals)
  } else {
    # validate if the given path is valid
    if (!dir.exists(path)) {
      warning(
        "The given `path` is not valid! The logs will be included in the ",
        "RO-Crate object, under the `content` tag!",
        call. = FALSE
      )
      log_entity$content <- list(userlogs)
      log_maps_entity$content <- list(userlogs_tbl_maps_evals)
    } else {
      writeLines(userlogs, file.path(path, log_filename))
      write.csv(
        userlogs_tbl_maps_evals,
        file.path(path, log_maps_filename),
        row.names = FALSE
      )
    }
  }

  # add entities to the RO-Crate
  rocrate <- rocrate |>
    rocrateR::add_entity(log_entity) |>
    rocrateR::add_entity(log_maps_entity) |>
    rocrateR::add_entity_value(
      id = "./",
      key = "hasPart",
      value = c(
        getElement(.get_entity(rocrate, id = "./"), "hasPart"),
        list(
          list(`@id` = log_entity$`@id`),
          list(`@id` = log_maps_entity$`@id`)
        )
      ),
      overwrite = TRUE
    )

  # attach input arguments as attributes
  attr(rocrate, "connection") <- x
  attr(rocrate, "path") <- path
  attr(rocrate, "profile") <- profile
  attr(rocrate, "project") <- project
  attr(rocrate, "resources") <- resources
  attr(rocrate, "tables") <- tables
  attr(rocrate, "user") <- user

  # return RO-Crate with the new entity
  return(rocrate)
}

#' @rdname safe_output
#' @export
safe_output.rocrate <- function(
  x,
  ...,
  path = attr(x, "path"),
  user = attr(x, "user"),
  logs_to = Sys.time(),
  logs_from = logs_to - 24 * 60^2,
  connection = attr(x, "connection"),
  profile = attr(x, "profile"),
  project = attr(x, "project"),
  resources = attr(x, "resources"),
  tables = attr(x, "tables")
) {
  # check if the connection was given
  if (is.null(connection)) {
    stop("A `connection` object is required!", call. = FALSE)
  }

  # call method with given `connection` object:
  safe_output(
    connection,
    rocrate = x,
    path = path,
    user = user,
    logs_to = logs_to,
    logs_from = logs_from,
    profile = profile,
    project = project,
    resources = resources,
    tables = tables
  )
}