plumber.R

library(plumber)
library(dplyr)
library(DBI)
library(RPostgres)
library(httr2)
library(jsonlite)

source("00_db_helper.R")
# source("backend/00_db_helper.R")

# Baca statistik training sekali saja saat API start
# train_stats <- NULL
if (file.exists("models/train_statistics.rds")) {
  train_stats <- readRDS("models/train_statistics.rds")
  # print(train_stats)
} else {
  con <- connect_supabase()
  train_stats <- dbGetQuery(con, "SELECT * FROM mlops.train_statistics ORDER BY create_date DESC LIMIT 1")
  DBI::dbDisconnect(con)
}

#* @filter cors
function(req, res) {
  res$setHeader("Access-Control-Allow-Origin", "*")
  res$setHeader("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
  res$setHeader("Access-Control-Allow-Headers", "Content-Type, Authorization")

  if (req$REQUEST_METHOD == "OPTIONS") {
    res$status <- 200
    return(list())
  }
  plumber::forward()
}

#* Mengambil ringkasan performa model
#* @param window:int Ukuran window data
#* @get /performance
function(window = 100) {
  con <- connect_supabase()
  db_res <- dbGetQuery(con,  
    "SELECT predicted_value FROM mlops.predictions ORDER BY timestamp DESC LIMIT $1", 
    list(as.integer(window))
  )
  
  total_db_res <- dbGetQuery(con, "SELECT count(predicted_value) as total FROM mlops.predictions")
  dbDisconnect(con)
  
  if(nrow(db_res) == 0) {
    return(list(total = 0, mean_pred = NA, sd_pred = NA))
  }
  
  list(
    total = total_db_res$total,
    window = window, 
    mean_pred = mean(db_res$predicted_value, na.rm = TRUE),
    sd_pred = sd(db_res$predicted_value, na.rm = TRUE),
    min_pred = min(db_res$predicted_value, na.rm = TRUE),
    max_pred = max(db_res$predicted_value, na.rm = TRUE)
  )
}

#* Mengambil data logs dan drift yang sudah diproses
#* @param window:int Ukuran window data
#* @param drift_feature:str Fitur yang ingin dihitung Z-Score nya
#* @get /logs-drift
function(window = 100) {
  con <- connect_supabase()
  query_sql <- "
  WITH exploded_data AS (
    SELECT 
      request_id,
      timestamp,
      variant, version, row_id, 
      predicted_value,
      (input_features->>'displ')::numeric AS displ,
      (input_features->>'year')::numeric AS year,
      (input_features->>'cyl')::numeric AS cyl,
      (input_features->>'class')::text as class, 
      status, 
      COALESCE(error_message::text, '') AS error_message
    FROM mlops.predictions
    WHERE status = 'SUCCESS'
    ORDER BY timestamp DESC
    LIMIT %d
  )
  SELECT 
    *,
    ABS(displ - %f) / %f AS z_displ,
    ABS(year - %f) / %f AS z_year,
    ABS(cyl - %f) / %f AS z_cyl
  FROM exploded_data;
  "
  # print(train_stats)
  query_filled <- sprintf(
    query_sql,
    as.integer(window),
    train_stats$displ_mean, train_stats$displ_sd,
    train_stats$year_mean, train_stats$year_sd,
    train_stats$cyl_mean, train_stats$cyl_sd
  )
  # print(query_filled)

  final_data <- dbGetQuery(con, query_filled)
  # print("final_data:")
  # print(head(final_data))
  # print(str(final_data))
  
  # Jika query kosong, langsung kembalikan data frame kosong
  if(nrow(final_data) == 0) {
    dbDisconnect(con)
    return(list(data = data.frame()))
  }

  actuals_db <- tryCatch({
    dbGetQuery(con, "SELECT request_id, row_id, actual_value FROM mlops.actuals;")
  }, error = function(e) { 
    data.frame(request_id = character(), row_id = character(), actual_value = numeric()) 
  })
  dbDisconnect(con)
  
  
  # Gabung Actual
  # print(nrow(actuals_db) > 0) 
  # print("row_id" %in% names(final_data))
  if (nrow(actuals_db) > 0 && "row_id" %in% names(final_data)) {
    actuals_db$request_id <- as.character(actuals_db$request_id)
    actuals_db$row_id <- as.character(actuals_db$row_id)
    final_data <- final_data %>% 
      left_join(actuals_db, by = c("request_id", "row_id"))
    # print(final_data)
  } else {
    final_data$actual_value <- NA_real_
  }
  # print("final_data:")
  # print(nrow(final_data))
  # print(str(final_data))
    
  return(final_data)
}