## ----setup, include = FALSE---------------------------------------------------
knitr::opts_chunk$set(
  collapse = TRUE,
  comment = "#>",
  fig.width = 6,
  fig.height = 4
)
library(joinspy)

## -----------------------------------------------------------------------------
orders <- data.frame(
  customer_id = c("C01", "C02", "C03", "C04"),
  amount = c(100, 200, 150, 300),
  stringsAsFactors = FALSE
)

customers <- data.frame(
  customer_id = c("C01", "C02", "C03", "C04"),
  region = c("East", "West", "East", "North"),
  stringsAsFactors = FALSE
)

# This passes -- keys are clean
stopifnot(key_check(orders, customers, by = "customer_id", warn = FALSE))

## ----error = TRUE-------------------------------------------------------------
try({
orders_dirty <- data.frame(
  customer_id = c("C01", "C02 ", "C03 ", "C04"),
  amount = c(100, 200, 150, 300),
  stringsAsFactors = FALSE
)

stopifnot(key_check(orders_dirty, customers, by = "customer_id", warn = FALSE))
})

## ----error = TRUE-------------------------------------------------------------
try({
if (!key_check(orders_dirty, customers, by = "customer_id", warn = FALSE)) {
  stop("Key quality check failed for orders-customers join. ",
       "Run join_spy() interactively for details.", call. = FALSE)
}
})

## -----------------------------------------------------------------------------
ok <- key_check(orders_dirty, customers, by = "customer_id", warn = FALSE)

if (!ok) {
  repaired <- join_repair(
    orders_dirty, customers,
    by = "customer_id",
    trim_whitespace = TRUE,
    remove_invisible = TRUE
  )
  orders_clean <- repaired$x
  customers_clean <- repaired$y

  # Re-check after repair
  stopifnot(key_check(orders_clean, customers_clean,
                       by = "customer_id", warn = FALSE))
}

## -----------------------------------------------------------------------------
sensors <- data.frame(
  sensor_id = c("S01", "S02", "S03", "S04"),
  location = c("Roof", "Basement", "Lobby", "Garage"),
  stringsAsFactors = FALSE
)

readings <- data.frame(
  sensor_id = c("S01", "S02", "S03", "S05"),
  temperature = c(22.1, 18.5, 21.0, 19.3),
  stringsAsFactors = FALSE
)

# Nothing printed
result <- left_join_spy(sensors, readings, by = "sensor_id", .quiet = TRUE)

## -----------------------------------------------------------------------------
rpt <- last_report()
rpt$match_analysis$match_rate

## -----------------------------------------------------------------------------
rpt <- last_report()
if (rpt$match_analysis$match_rate < 0.95) {
  warning(sprintf(
    "Low match rate (%.1f%%) in sensor join -- check for missing sensor IDs",
    rpt$match_analysis$match_rate * 100
  ))
}

## -----------------------------------------------------------------------------
result1 <- left_join_spy(sensors, readings, by = "sensor_id", .quiet = TRUE)
report1 <- last_report()

# ... later ...
result2 <- inner_join_spy(sensors, readings, by = "sensor_id", .quiet = TRUE)
report2 <- last_report()

## -----------------------------------------------------------------------------
identical(attr(result1, "join_report"), report1)

## -----------------------------------------------------------------------------
rpt <- join_spy(orders_dirty, customers, by = "customer_id")
names(rpt)

## -----------------------------------------------------------------------------
rpt$x_summary$n_duplicates
rpt$match_analysis$match_rate
rpt$expected_rows$left

## -----------------------------------------------------------------------------
length(rpt$issues)
vapply(rpt$issues, function(i) i$type, character(1))

## -----------------------------------------------------------------------------
severities <- vapply(rpt$issues, function(i) i$severity, character(1))
severities
sum(severities == "warning")

## -----------------------------------------------------------------------------
report_gate <- function(rpt, min_match = 0.95) {
  stopifnot(is_join_report(rpt))
  problems <- character(0)
  if (rpt$match_analysis$match_rate < min_match) {
    problems <- c(problems, sprintf(
      "match rate %.0f%% below %.0f%%",
      100 * rpt$match_analysis$match_rate, 100 * min_match
    ))
  }
  sev <- vapply(rpt$issues, function(i) i$severity, character(1))
  if (any(sev == "warning")) {
    problems <- c(problems, sprintf("%d warning-level issue(s)",
                                    sum(sev == "warning")))
  }
  problems
}

report_gate(rpt)

## -----------------------------------------------------------------------------
summary(rpt)

## -----------------------------------------------------------------------------
products <- data.frame(
  product_id = c("P1", "P2", "P3"),
  name = c("Widget", "Gadget", "Gizmo"),
  stringsAsFactors = FALSE
)

line_items <- data.frame(
  product_id = c("P1", "P1", "P2", "P3", "P3"),
  order_id = c(101, 102, 103, 104, 105),
  stringsAsFactors = FALSE
)

detect_cardinality(products, line_items, by = "product_id")

## -----------------------------------------------------------------------------
result <- join_strict(
  products, line_items,
  by = "product_id",
  type = "left",
  expect = "1:n"
)
nrow(result)

## ----error = TRUE-------------------------------------------------------------
try({
products_bad <- data.frame(
  product_id = c("P1", "P1", "P2", "P3"),
  name = c("Widget", "Widget v2", "Gadget", "Gizmo"),
  stringsAsFactors = FALSE
)

join_strict(
  products_bad, line_items,
  by = "product_id",
  type = "left",
  expect = "1:n"
)
})

## -----------------------------------------------------------------------------
events_a <- data.frame(id = rep(c("E1", "E2"), each = 20), src = "a",
                       stringsAsFactors = FALSE)
events_b <- data.frame(id = rep(c("E1", "E2"), each = 20), src = "b",
                       stringsAsFactors = FALSE)

chk <- check_cartesian(events_a, events_b, by = "id")
chk$expansion_factor

## ----eval = requireNamespace("testthat", quietly = TRUE)----------------------
library(testthat)

test_that("orders join customers cleanly on customer_id", {
  expect_true(key_check(orders, customers, by = "customer_id", warn = FALSE))
})

## ----eval = requireNamespace("testthat", quietly = TRUE)----------------------
test_that("products to line_items is one-to-many", {
  expect_identical(
    detect_cardinality(products, line_items, by = "product_id"),
    "1:n"
  )
})

## ----eval = requireNamespace("testthat", quietly = TRUE)----------------------
test_that("left join is predicted to preserve order rows", {
  rpt <- join_spy(orders, customers, by = "customer_id")
  expect_equal(rpt$expected_rows$left, nrow(orders))
})

## -----------------------------------------------------------------------------
report <- join_spy(sensors, readings, by = "sensor_id")

# Text format -- human-readable
txt_log <- tempfile(fileext = ".log")
log_report(report, txt_log)
cat(readLines(txt_log), sep = "\n")
unlink(txt_log)

## -----------------------------------------------------------------------------
# JSON format -- machine-readable
json_log <- tempfile(fileext = ".json")
log_report(report, json_log)
cat(readLines(json_log), sep = "\n")
unlink(json_log)

## -----------------------------------------------------------------------------
auto_log <- tempfile(fileext = ".log")
set_log_file(auto_log, format = "text")

# These joins are automatically logged
result1 <- left_join_spy(sensors, readings, by = "sensor_id", .quiet = TRUE)
result2 <- inner_join_spy(sensors, readings, by = "sensor_id", .quiet = TRUE)

# Check what got logged
cat(readLines(auto_log), sep = "\n")

# Clean up
set_log_file(NULL)
unlink(auto_log)

## -----------------------------------------------------------------------------
fn_log <- tempfile(fileext = ".log")
previous <- set_log_file(fn_log)

result <- left_join_spy(sensors, readings, by = "sensor_id", .quiet = TRUE)

set_log_file(previous)
unlink(fn_log)

## -----------------------------------------------------------------------------
# Only log if logging is configured
if (!is.null(get_log_file())) {
  message("Logging is active at: ", get_log_file())
}

## -----------------------------------------------------------------------------
json_log <- tempfile(fileext = ".json")
set_log_file(json_log, format = "json")

r1 <- left_join_spy(sensors, readings, by = "sensor_id", .quiet = TRUE)
r2 <- inner_join_spy(orders, customers, by = "customer_id", .quiet = TRUE)

set_log_file(NULL)

## -----------------------------------------------------------------------------
log_lines <- readLines(json_log)
rate_lines <- grep('"match_rate"', log_lines, value = TRUE)
rate_lines

## -----------------------------------------------------------------------------
rates <- as.numeric(sub('.*"match_rate": *([0-9.]+).*', "\\1", rate_lines))
rates
which(rates < 0.95)
unlink(json_log)

## -----------------------------------------------------------------------------
orders_chain <- data.frame(
  order_id = 1:6,
  customer_id = c("C1", "C2", "C2", "C3", "C4", "C4"),
  stringsAsFactors = FALSE
)

customers_chain <- data.frame(
  customer_id = c("C1", "C2", "C3", "C4"),
  region_id = c("R1", "R1", "R2", "R3"),
  stringsAsFactors = FALSE
)

regions <- data.frame(
  region_id = c("R1", "R2"),
  region_name = c("North", "South"),
  stringsAsFactors = FALSE
)

## -----------------------------------------------------------------------------
chain <- analyze_join_chain(
  tables = list(orders = orders_chain, customers = customers_chain,
                regions = regions),
  joins = list(
    list(left = "orders", right = "customers", by = "customer_id"),
    list(left = "result", right = "regions", by = "region_id")
  )
)

## -----------------------------------------------------------------------------
chain[[2]]$report$match_analysis$match_rate
chain[[2]]$report$match_analysis$left_only_keys

## -----------------------------------------------------------------------------
vapply(chain, function(s) length(s$report$issues), integer(1))

## -----------------------------------------------------------------------------
# Simulate a large dataset
set.seed(42)
big_orders <- data.frame(
  customer_id = sample(paste0("C", sprintf("%04d", 1:5000)), 50000, replace = TRUE),
  amount = round(runif(50000, 10, 500), 2),
  stringsAsFactors = FALSE
)

big_customers <- data.frame(
  customer_id = paste0("C", sprintf("%04d", 1:6000)),
  region = sample(c("North", "South", "East", "West"), 6000, replace = TRUE),
  stringsAsFactors = FALSE
)

# Full analysis
system.time(report_full <- join_spy(big_orders, big_customers, by = "customer_id"))

# Sampled analysis
system.time(report_sampled <- join_spy(big_orders, big_customers,
                                        by = "customer_id", sample = 5000))

## -----------------------------------------------------------------------------
report_sampled$sampling

## -----------------------------------------------------------------------------
# ============================================================
# Nightly order enrichment pipeline
# ============================================================

# --- Setup logging ---
pipeline_log <- tempfile(fileext = ".log")
set_log_file(pipeline_log, format = "text")

# --- Load data (simulated) ---
orders <- data.frame(
  order_id = 1:6,
  customer_id = c("C001", "C002 ", "C003", "C003", "C004", "C005"),
  amount = c(150, 230, 89, 410, 320, 175),
  stringsAsFactors = FALSE
)

customers <- data.frame(
  customer_id = c("C001", "C002", "C003", "C004", "C005", "C006"),
  name = c("Acme Corp", "Globex", "Initech", "Umbrella", "Soylent", "Wonka"),
  tier = c("gold", "silver", "gold", "bronze", "silver", "gold"),
  stringsAsFactors = FALSE
)

# --- Gate 1: key quality assertion ---
keys_ok <- key_check(orders, customers, by = "customer_id", warn = FALSE)

if (!keys_ok) {
  message("Key issues detected -- attempting repair")
  repaired <- join_repair(
    orders, customers,
    by = "customer_id",
    trim_whitespace = TRUE,
    remove_invisible = TRUE
  )
  orders <- repaired$x
  customers <- repaired$y
}

# --- Gate 2: cardinality check ---
card <- detect_cardinality(orders, customers, by = "customer_id")
if (card == "n:m") {
  set_log_file(NULL)
  unlink(pipeline_log)
  stop("Unexpected n:m cardinality in orders-customers join", call. = FALSE)
}

# --- Join (with auto-logging via *_join_spy) ---
enriched <- left_join_spy(orders, customers, by = "customer_id", .quiet = TRUE)

# --- Gate 3: row count sanity check ---
# A left join should never lose rows from the left table
if (nrow(enriched) < nrow(orders)) {
  set_log_file(NULL)
  unlink(pipeline_log)
  stop("Row count decreased after left join -- possible data corruption",
       call. = FALSE)
}

# --- Output ---
message(sprintf("Pipeline complete: %d enriched orders", nrow(enriched)))
head(enriched)

# --- Review the log ---
if (file.exists(pipeline_log)) {
  cat(readLines(pipeline_log), sep = "\n")
}

# --- Cleanup ---
set_log_file(NULL)
unlink(pipeline_log)

## -----------------------------------------------------------------------------
t_merge <- system.time(
  merge(big_orders, big_customers, by = "customer_id", all.x = TRUE)
)
t_spy <- system.time(
  left_join_spy(big_orders, big_customers, by = "customer_id", .quiet = TRUE)
)
t_check <- system.time(
  key_check(big_orders, big_customers, by = "customer_id", warn = FALSE)
)

rbind(merge = t_merge, left_join_spy = t_spy, key_check = t_check)[, 1:3]