Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 22 additions & 7 deletions R/dashboard_module_address_entry.R
Original file line number Diff line number Diff line change
Expand Up @@ -142,6 +142,16 @@ observe_address_validation <- function(input, session, db, current_case, jwt, ap

observeEvent(input$address_validate, {
logger::log_debug("Address validation button pressed")

# Check if we have a valid case
case_id <- current_case()
if (is.null(case_id) || is.na(case_id)) {
showModal(modalDialog(
title = "No Case Available",
"No case is currently available for address validation. Please refresh to get a new case."
))
return()
}

address_entered$object <- isolate_address_entered(input)
address_entered$string <- stringify_address_entered(address_entered$object)
Expand Down Expand Up @@ -214,7 +224,7 @@ observe_address_validation <- function(input, session, db, current_case, jwt, ap
} else {
modal_content <- "Could not validate address."

query <- glue::glue_sql('UPDATE "eviction_addresses"."queue" SET attempts = attempts + 1, stopped_at = CURRENT_TIMESTAMP WHERE "case" = {current_case()};', .con = db)
query <- glue::glue_sql('UPDATE "eviction_addresses"."queue" SET attempts = attempts + 1, stopped_at = CURRENT_TIMESTAMP WHERE "case" = {case_id};', .con = db)
logger::log_debug("Query: {query}")

DBI::dbExecute(
Expand All @@ -226,7 +236,7 @@ observe_address_validation <- function(input, session, db, current_case, jwt, ap
} else {
modal_content <- "Bad response from validation server"

query <- glue::glue_sql('UPDATE "eviction_addresses"."queue" SET attempts = attempts + 1, stopped_at = CURRENT_TIMESTAMP WHERE "case" = {current_case()};', .con = db)
query <- glue::glue_sql('UPDATE "eviction_addresses"."queue" SET attempts = attempts + 1, stopped_at = CURRENT_TIMESTAMP WHERE "case" = {case_id};', .con = db)
logger::log_debug("Query: {query}")

DBI::dbExecute(
Expand Down Expand Up @@ -346,14 +356,19 @@ observe_address_submission <- function(input, db, current_case, current_user, ad
observeEvent(input$address_submit, {
logger::log_debug("Address submit button pressed")

current_case <- current_case()
case_id <- current_case()

# Check if we have a valid case
if (is.null(case_id) || is.na(case_id)) {
rlang::abort("Cannot submit address: no valid case is available.")
}

if(is.null(address_validated$object)) {
rlang::abort("Something is wrong. You submitted an address without first validating.")
}

new_row <- tibble::tibble(
case = current_case(),
case = case_id,
street_number = as.character(address_validated$object$streetNumber),
street_direction = as.character(address_validated$object$streetDirection),
street_name = as.character(address_validated$object$streetName),
Expand Down Expand Up @@ -399,7 +414,7 @@ observe_address_submission <- function(input, db, current_case, current_user, ad
logger::log_debug("Wrote new record in 'address' table")

pool::poolWithTransaction(db, function(conn) {
query <- glue::glue_sql('UPDATE "eviction_addresses"."queue" SET success = TRUE, working = FALSE, stopped_at = CURRENT_TIMESTAMP WHERE "case" = {current_case()};', .con = conn)
query <- glue::glue_sql('UPDATE "eviction_addresses"."queue" SET success = TRUE, working = FALSE, stopped_at = CURRENT_TIMESTAMP WHERE "case" = {case_id};', .con = conn)

DBI::dbExecute(
conn = conn,
Expand All @@ -409,7 +424,7 @@ observe_address_submission <- function(input, db, current_case, current_user, ad

# Create a new row in the process_log table associating this address with the current user
new_row <- tibble::tibble(
case = current_case(),
case = case_id,
user = current_user(),
created_at = lubridate::now(tzone = "America/Chicago"),
updated_at = lubridate::now(tzone = "America/Chicago")
Expand All @@ -432,7 +447,7 @@ observe_address_submission <- function(input, db, current_case, current_user, ad
} else {
logger::log_error("Failed to write the new record to table 'address'")

query <- glue::glue_sql('UPDATE "eviction_addresses"."queue" SET attempts = attempts + 1, working = FALSE, success = FALSE, stopped_at = CURRENT_TIMESTAMP WHERE "case" = {current_case()};', .con = db)
query <- glue::glue_sql('UPDATE "eviction_addresses"."queue" SET attempts = attempts + 1, working = FALSE, success = FALSE, stopped_at = CURRENT_TIMESTAMP WHERE "case" = {case_id};', .con = db)

update_res <- DBI::dbExecute(
conn = db,
Expand Down
9 changes: 8 additions & 1 deletion R/dashboard_module_current_documents.R
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,14 @@ currentDocumentsUI <- function(id) {
currentDocumentsServer <- function(id, current_case, db) {
moduleServer(id, function(input, output, session) {
documents <- reactive({
res <- get_documents_by_case(db = db, id = current_case())
case_id <- current_case()

# Return empty data frame if no case is available
if (is.null(case_id) || is.na(case_id)) {
return(data.frame())
}

res <- get_documents_by_case(db = db, id = case_id)
logger::log_debug("Documents retrieved: {res}")
res
})
Expand Down
13 changes: 12 additions & 1 deletion R/dashboard_module_entry_detail.R
Original file line number Diff line number Diff line change
Expand Up @@ -30,8 +30,19 @@ entryDetailUI <- function(id) {
entryDetailServer <- function(id, current_case, total_cases) {
moduleServer(id, function(input, output, session) {
output$current_case_ui <- renderUI({
current_case <- jsonlite::fromJSON(current_case())
case_json <- current_case()
queue <- total_cases()

# Handle empty queue - cases unavailable for processing
if (is.null(case_json) || length(case_json) == 0 || is.na(case_json) || (case_json == "")) {
return(div(
h4("No cases available for processing"),
h4(glue::glue("{queue} total in queue"))
))
}

# Parse JSON and display valid case
current_case <- jsonlite::fromJSON(case_json)
div(
h4(glue::glue("Current case: {current_case$case_number}")),
h4(glue::glue("{queue} cases in queue"))
Expand Down
5 changes: 5 additions & 0 deletions R/dashboard_server.R
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,11 @@ dashboard_server <- function(config) {
documents <- reactive({
current_case <- current_case()

# Return NULL if no case is available (queue is empty)
if (is.null(current_case) || is.na(current_case)) {
return(NULL)
}

res <- get_documents_by_case(db = db, id = current_case)

res
Expand Down
71 changes: 67 additions & 4 deletions R/queue.R
Original file line number Diff line number Diff line change
Expand Up @@ -44,12 +44,11 @@ get_queue_length <- function(db, status = "available") {
#'
#'
clean_queue <- function(db) {
## Remove any queue rows where success is true and it's 2 weeks old
## ? Should we check whether the address is found in the db ?
## Remove any queue rows where success is true

query <- dbplyr::sql("DELETE FROM eviction_addresses.queue WHERE success IS TRUE AND created_at < current_timestamp - interval '14 days';")
query <- dbplyr::sql("DELETE FROM eviction_addresses.queue WHERE success IS TRUE;")
num_deleted_queue_rows <- DBI::dbExecute(db, query)
logger::log_debug("{num_deleted_queue_rows} rows deleted")
logger::log_info("{num_deleted_queue_rows} completed rows deleted from queue")

return()
}
Expand Down Expand Up @@ -114,6 +113,12 @@ get_case_from_queue <- function(db) {
conn,
dbplyr::sql('SELECT q."case" FROM eviction_addresses.queue q LEFT JOIN eviction_addresses."case" c ON q."case" = c."id" LEFT JOIN public."case" pc ON q."case" = pc.id WHERE "success" IS NOT TRUE AND "working" IS NOT TRUE ORDER BY attempts ASC, pc.status DESC, c.date_filed DESC LIMIT 1 FOR UPDATE OF q SKIP LOCKED;')
)

# Return NULL if no cases available in queue
if (nrow(case) == 0) {
return(NULL)
}

query <- glue::glue_sql(
'UPDATE "eviction_addresses"."queue" SET working = TRUE, started_at = CURRENT_TIMESTAMP, stopped_at = NULL WHERE "case" = {case}',
.con = conn
Expand All @@ -126,3 +131,61 @@ get_case_from_queue <- function(db) {
})

}


#' @title Reset Stuck Queue Items
#'
#' @description
#' Resets queue items that have been stuck in 'working' status for over an hour.
#' These are typically cases from crashed or abandoned sessions.
#' Items that exceed max retry attempts (3) are logged but not reset.
#'
#' @param db A database connection
#' @param max_retries Maximum number of retry attempts before giving up on a case. Defaults to 3.
#' @param stale_threshold_hours Number of hours before a working case is considered stuck. Defaults to 1.
#'
#' @returns Nothing
#'
reset_stuck_queue_items <- function(db, max_retries = 3, stale_threshold_hours = 1) {
# Find stuck cases: working=TRUE and started over threshold hours ago
stuck_query <- glue::glue_sql(.con = db, "
SELECT \"case\", attempts
FROM eviction_addresses.queue
WHERE working = TRUE
AND started_at < CURRENT_TIMESTAMP - INTERVAL '{stale_threshold_hours} hours';
")

stuck_cases <- DBI::dbGetQuery(db, stuck_query)

if (nrow(stuck_cases) == 0) {
logger::log_info("No stuck queue items found")
return(invisible(NULL))
}

num_stuck <- nrow(stuck_cases)
num_max_retries <- sum(stuck_cases$attempts >= max_retries)
num_to_reset <- num_stuck - num_max_retries

# Reset cases that haven't exceeded max retries
if (num_to_reset > 0) {
reset_query <- glue::glue_sql(.con = db, "
UPDATE eviction_addresses.queue
SET working = FALSE,
attempts = attempts + 1,
stopped_at = CURRENT_TIMESTAMP
WHERE working = TRUE
AND started_at < CURRENT_TIMESTAMP - INTERVAL '{stale_threshold_hours} hours'
AND attempts < {max_retries};
")

num_reset <- DBI::dbExecute(db, reset_query)
logger::log_info("Reset {num_reset} stuck queue items (out of {num_stuck} total stuck)")
}

# Log cases that hit max retries
if (num_max_retries > 0) {
logger::log_warn("{num_max_retries} queue items reached max retry limit ({max_retries}) and will not be reset")
}

return(invisible(NULL))
}
1 change: 1 addition & 0 deletions R/run_job.R
Original file line number Diff line number Diff line change
Expand Up @@ -16,4 +16,5 @@ run_job <- function(config) {
refresh_cases(db)
refresh_minutes(db)
refresh_queue(db)
reset_stuck_queue_items(db)
}