# snowflake_check_20150707.r

# screen -xRR auto_dbconcile
# cd ~/git; git pull
# Rscript -e 'source("~/git/orch/src/dbconcile/snowflake_check_20150707.r")'

## This should run at 7pm, 1am, 7am EST
# 0 1,7,19 * * * Rscript -e 'source("~/git/orch/src/dbconcile/snowflake_check_20150707.r")'
## Everyday at 6am, clear the file to mark once-a-day success
# 0 6 * * * rm ~/git/orch/src/dbconcile/success_email/success_email.marker_file
fresh()

## ------------------------------------------------------------------------------ ##
##                 FOR MANUAL RUNS                                                ##
## ------------------------------------------------------------------------------ ##

  ## ONE LINE RUN
  #   setScience(proj="dbconcile", subProj="Snowflake", quiet=TRUE); setGitBranchToSystem(); .g();  .us();  source(src.p("snowflake_check_20150707.r"))

  if (FALSE) 
  {
    setGitBranchToSystem(); .g()
    if (FALSE)
      source("~/git/orch/src/dbconcile/snowflake_check_20150707.r")
  }
## ------------------------------------------------------------------------------ ##
## ------------------------------------------------------------------------------ ##



## TODO:  Set up a "recovered" subject.  This requires a sepearte file-marker per table


## Whether to use 'processeddaytime' as dateCol for fact_analytics
## if FALSE will use getDateColForTBL(), ie  dayid
use_processeddaytime_for_fa <- TRUE

## When debugiing, for interactive, set to TRUE
dont_sink <- FALSE ## TRUE

## DBI from rstats-DB is not playing nicely with RPostgreSQL
## and the generic DBI is not playing nicely with RPostgres from rstats-db
## ----- UPDATE 2015-11-24 this is no longer needed
#        try(install.packages("DBI"))

## -------------------------- ##
##        MANUAL FLAGS        ##
## -------------------------- ##
  ## Number of cores for parallel processing
  cores_to_use <- 8
## -------------------------- ##

setScience("dbconcile", subProj="Snowflake", create=FALSE, subl=FALSE, load=FALSE)
.us();  setPrompt(TRUE)

## This file is hardcoded in another cron job
file.email_marker_last_failed <- "~/git/orch/src/dbconcile/success_email/last_run_failed.marker_file"
file.email_marker <- "~/git/orch/src/dbconcile/success_email/success_email.marker_file"
dir.create(dirname(file.email_marker), recursive=TRUE, showWarnings=FALSE)


library(data.table)
try(library(rJava))
try(library(mailR))
#-X  library(foreach)
#-X  library(doParallel)


## SUBJECT STARTING PHRASES
OK_subj <- "Snowflake data confirmed for"
recovered_subj <- "[Recovered] Snowflake data confirmed for"  ## UNUSED FOR NOW
outofsync_subj <- "Snowflake is out of sync for"
error_subj <- "ERROR checking Snowflake data for"

## REDSHIFT SETTINGS TO TEST AGAINST
cluster <- 4

## SNOWFLAKE SETTINGS TO TEST AGAINST
wh <- ifelse(interactive(), getSnowflakeWH(), "CRON_JOBS_SMALL") %>% {ifelse(. == "", "LOOKER_WH_LARGE", .)}
dbname <- "prod"
schema <- "production"

## TABLES TO CHECK FOR SYNCH
## TODO:  Add dim tables
tbls <- c("fact_sales", "fact_analytics")

#-X  ## PARALLEL FAILS FOR SNOWFLAKE CONNECTION
#-X  ## Number of cores
#-X  # registerDoParallel(cores=cores_to_use)

what_compared <- sprintf("[Snowflake (%s) <--> Redshift (%02i)]", dbname, cluster)


# email(to="RSAPORTA@GMAIL.com", subj=sprintf("STARTING: %s", what_compared))

## SINK OUTPUT TO LOGFILE
if (!dont_sink) {
  cat("\n\n\n", "Starting ON: ", timeStamp(human=TRUE), " ~~~~~~~~ \n", sep="")
  sinkfile <- newLogFile(name="snowflake_redshift_compare")
  sinkOn(file=sinkfile, zarchive.old.files=TRUE)
  # on.exit(sinkOff(), add=TRUE)
} else {
  sinkfile <- "<LOG FILE WAS NOT USED FOR THIS RUN>"
}

# if (FALSE)
# emailGroup_if_DB_out_of_sync <- c(
#     "Data Analytics Team"="DataRequests@TheOrchard.com"
#   , "Pasha Katsev <pkatsev@theorchard.com>"
#   , "Paulo Kuong <paulo@theorchard.com>"
#   , "Myron Machado <mmachado@theorchard.com>"
# )

## TODO: unique_preserve_names
emailGroup_if_DB_out_of_sync <- uniqueByNamesAndValues(c(
    "Data Analytics Team"="DataRequests@TheOrchard.com"
  , "Alexey Zinger" = "azinger@theorchard.com"
  , "John Penner" = "jpenner@theorchard.com"
))
emailGroup_if_OK <- uniqueByNamesAndValues(c(
    "Data Analytics Team"="DataRequests@TheOrchard.com"
  , "John Penner" = "jpenner@theorchard.com"
  , if (file.exists(file.email_marker_last_failed))
    emailGroup_if_DB_out_of_sync
))
emailGroup_if_ERR <- uniqueByNamesAndValues(c(
    "Data Analytics Team"="DataRequests@TheOrchard.com"
))

send_email_when_ok <- file.exists(file.email_marker_last_failed) || !file.exists(file.email_marker) || interactive()

## CONFIRM THAT WE CAN SWITCH BACK AND FORTH BETWEEN WAREHOUSES
check_snowflake_redshift_switch(cluster=cluster, wh=wh)

compare_ok <- emptylist(tbls)
processed_lessthan <- today() - 1

# maxDate = qMaxDate(tbl, snowflake=TRUE, dbname=dbname, simplify=TRUE)
maxDate = NULL

cat(" --- outside for loop --- \n")

for (tbl in rev(tbls)) {
#-X  ## PARALLEL FAILS FOR SNOWFLAKE CONNECTION
#-X  # foreach (tbl=rev(tbls)) %dopar% {

  ## use 'processeddaytime' for fact_analytics, otherwise the default for the table
  dateCol <- if (use_processeddaytime_for_fa && tbl == "fact_analytics") "processeddaytime" else getDateColForTBL(tbl, schema=schema)

  message ("            ~~~~~~~~~  CHECKING '", tbl, "' with dateCol = '", dateCol, "' ~~~~~~~~~~~~")
  timestart_string <- timeStamp("Query began: ", human=TRUE, sep=" ") %>% paste("EST")
  compare_ok[[tbl]] <- try(compare_snowflake_to_redshift(tbl=tbl, dateCol=dateCol, cluster=cluster, processed_lessthan=processed_lessthan, maxDate=maxDate, refresh_connex=TRUE, wh=wh, dbname=dbname, schema=schema))

  ## ------------ CHECK ERROR --------------- ##
  if (!isErr(compare_ok[[tbl]]) && !("all_row_values_are_equal" %in% names(compare_ok[[tbl]]) && compare_ok[[tbl]][, all(all_row_values_are_equal)])) {
    warning("\n", pasteR("~", 52), "\n\t", tbl, " does NOT match\n", pasteR("~", 52), call.=FALSE)

    subject <- sprintf("%s %s %s", outofsync_subj, tbl, what_compared)

    Body <- sprintf("comparing %s\nTable '%s' appears to be out-of-sync\n%s\n\n%s",
      what_compared, tbl
      , ifelse (tbl == "fact_analytics" && exists("processed_lessthan") && !is.null(processed_lessthan), sprintf("Checked for processdaytime < \"%s\"", processed_lessthan), "")
      , timestart_string)

    if (!is.null(err.msg <- attr(compare_ok[[tbl]], "err.msg", exact=TRUE)))
      Body %<>% paste0("\n\n", err.msg, "\n")

    if (!isErr(compare_ok[[tbl]]) && is.data.table(compare_ok[[tbl]])) {
      f.out <- writeDT(compare_ok[[tbl]], base.file.name = sprintf("snowflake_redshift_comp_for_%s", tbl))
    } else
      f.out <- c()

    # ## IF TEST 
    # subject <- "TEST EMAIL for snowflake sync"
    # Body %<>% paste("THIS IS JUST A TEST EMAIL\nTHIS IS JUST A TEST EMAIL\nTHIS IS JUST A TEST EMAIL\n\nThis is the email that will go out if snowflake falls out of sync from Redshift\nFor now, We will use processdaytime < today - 24 hours\n\n", .)

    email(
      to=emailGroup_if_DB_out_of_sync
      , subject = subject
      # , from = c("Ricks RBox AutoEmail" = "thesam@email.theorchard.com")
      , body = Body
      , attachments = f.out
    )
  } else if (isErr(compare_ok[[tbl]])) { # // end of IF ERROR

    f.saved <- saveImageTo()
    email(
        to=emailGroup_if_ERR
      , subject = sprintf("%s %s %s", error_subj, tbl, what_compared)
      , body = sprintf("UNSURE ABOUT %s %s\n\nEither the data has errors or Rick's script encountered an error.\nRick needs to check the log files.\n\n      LOG FILE: '%s'\n    IMAGE FILE: '%s'", tbl, what_compared, sinkfile, f.saved)
    )

  } else { # // end of IF ERROR

      ## ONLY EMAIL "SUCCESS" per day
      ## This file is cleared via a cron job
      if (send_email_when_ok) {
        email(
          to=emailGroup_if_OK
          , subject = sprintf("%s %s %s", OK_subj, tbl, what_compared)
          , body = sprintf("All OK for %s %s\n\n%s", tbl, what_compared, timestart_string)
        )
        system(sprintf("touch %s", file.email_marker))
      } else {
        # email(
        #   to=c("Data Analytics Team"="DataRequests@TheOrchard.com")
        #   , subject = sprintf("Snowflake data confirmed for %s %s", tbl, what_compared)
        #   , body = sprintf("All OK for %s %s\n\n%s", tbl, what_compared, timestart_string)
        # )
        message(sprintf("All OK for %s %s - no email being sent", tbl, what_compared))
      }
  }

  cat("tbl '", tbl, "' ", what_compared, " is: ")
  print (compare_ok[[tbl]])
}

all_ok <- sapply(compare_ok, function(DT) !isErr(DT) && "all_row_values_are_equal" %in% names(DT) && DT[, all(all_row_values_are_equal)])

if (all(unlist(all_ok))) {
  cat("\n", timestart_string, "\n ~~~~~  ALL OK ", timeStamp(human=TRUE), " ~~~~~~~~ \n")

  if (file.exists(file.email_marker_last_failed))
    unlink(file.email_marker_last_failed)
  if (file.exists(file.email_marker_last_failed))
    warning("Was unable to remove file marker '", file.email_marker_last_failed, "'")
}

## STOP SINKING OUTPUT TO LOGFILE
if (!dont_sink)
   sinkOff()


cat("\n", timestart_string, "\n ~~~~~  ALL ", ifelse(all(unlist(all_ok)), "", "*NOT* ") ,"OK ", timeStamp(human=TRUE), " ~~~~~~~~ \n", sep="")