#  ~/git/orch/src/Discover_Weekly/debugging/create_spotify_user_counts_table supportFns (wrong).r

create_counts_table_name_from_colsToPull <- function(colsToPull, uid_columns=c("anonymized_uid", "user_id", "uid"), preface="by") {
    words <- colsToPull %>% unname() %>% tolower() %>% setdiff(., uid_columns)
    if (!length(words))
      return("daily_totals")
    else {
      if (any(wh.playlist <- grepl("^(playlist_|source_uri$)", words)))
        words <- words[!wh.playlist] %>% c("playlist")
      if (any(wh.upc <- grepl("^(album_code|upc$)", words)))
        words <- words[!wh.upc] %>% c("upc", .)
      sprintf("%s_%s", preface, pasteC(words, C="_"))
    }
}

create_spotify_user_counts_table <- function(
      colsToPull
    , colsWithaggFunc = c(date_first_streamed = "friday_weekof(min(first_tmstamp::date))", date_last_streamed = "friday_weekof(max(last_tmstamp::date))")
    , dateCol.nm="date"
    , colsToAgg  = c("streams")
    , tbl_out = "..auto.."
    , aggFunc = "sum"
    , tbl_in = "user_minmax_timestamps_detailed_test"
    , schema = "spot_user_counts"
    , confirm_streams = TRUE
    , uid_columns=c("anonymized_uid", "user_id", "uid")
    , minDate = "friday_weekof('2015-01-01'::date)"
    , having_clause = if (length(minDate) == 1) sprintf("date_first_streamed > %s", ifelse(grepl("^\\d", minDate), sprintf("'%s'", minDate), minDate))
    , streams_expected = makeQry(tbl=tbl_in, schema=schema, colsToAgg="streams", aggFunc="sum", limit=NULL) %>% sfQry(verbose=FALSE) %>% unlist()
    , where__for_debugging=NULL
    , verbose.qry=FALSE
) {
  if (aggFunc != "sum")
    warning("Aggregation may be incorrect when using aggFunc other than 'sum'")

  if (!any(uid_columns %in% colsToPull))
    warning("There is no USER ID column in colsToPull.  Was this intentional?\n\nHINT: Not having a user_id column is going to cause all of the dates to be the first by group, regardless of user")

  if (!is.null(where__for_debugging))
    message("REMINDER:  The WHERE clause should only be used for debugging. It will apply to the inner table")

  if (tbl_out == "..auto..") 
    tbl_out <- create_counts_table_name_from_colsToPull(colsToPull=colsToPull, uid_columns=uid_columns)


  if (grepl("_test$", tbl_in, ignore.case=TRUE) && !grepl("_test", tbl_out, ignore.case=TRUE)) {
    verboseMsg(TRUE, "Pasting '_test' to name of tbl_out")
    tbl_out %<>% paste0("_TEST")
  }

  ## ------  INNER QUERY    ---------------------------------
    qry.inner <- makeQry(tbl=tbl_in, schema=schema, colsToPull=colsToPull, colsToAgg=colsToAgg, colsWithaggFunc=colsWithaggFunc, aggFunc=aggFunc, having=having_clause, where=where__for_debugging, order=NULL, limit=NULL)


  ## ------  COUNTS QUERY    ---------------------------------
    ## combine the agg'd columns with colsToPull, but remove the "last streamed" column, the last_updated and, most importantly, the user
    colsToPull.counts <- c(colsWithaggFunc, colsToPull) %>% colNamesFromVector() %>% setdiff(., c("date_last_streamed", "last_updated", "last_tmstamp", uid_columns)) %>% tolower %>%
                          {setNames(obj=., nm=ifelse(. == "date_first_streamed", "date", .))} # %>% c(cumulative_new_users= default_cum_user_count)

    ## Each row is one user
    colsWithaggFunc.counts <- c(new_users = "count(*)")

    qry.counts <- makeQry(tbl=qry.inner, schema=NULL, colsToPull=colsToPull.counts, colsToAgg=colsToAgg, colsWithaggFunc=colsWithaggFunc.counts, aggFunc=aggFunc, order=colsToPull.counts, limit=NULL)


  ## ------  CUMULATIVE SUM QUERY    ---------------------------------
    ## Will agg the original columns, plus aggregate new_users
    ## Preface colsToAgg with T2, and colsToPull with T1

  colsToAgg.cumsum  <- c(if (aggFunc == "sum") colsToAgg, "new_users") %>% { setNames(obj=paste0("T2.", .), nm=paste0("cumulative_", colNamesFromVector(.))) }
  colsToPull.cumsum <- c(colsToPull.counts, colsWithaggFunc.counts, colsToAgg) %>% colNamesFromVector # %>% paste0("T1.", .)
  with.cumsum <- c(CTE_Counts=qry.counts)
  ON_AND.cumsum <- ifelse (length(colsToAgg.cumsum) <= 1, "", {
    # colsToPull.cumsum %>% setdiff("date") %>% sprintf("equal_null(T1.%s, T2.%1$s)", .) %>% paste(" AND ", .) %>% pasteC

    ## DOESN'T WORK.  BUGGY
    ## ------------------------------------------------- ##
    frmt.null_joiner <- "equal_null(T1.%s, T2.%1$s)" ## equal_null() is buggy, thus using NVL()
    colsToPull.cumsum %>% setdiff("date") %>% sprintf(frmt.null_joiner, .) %>% paste(" AND ", .) %>% pasteC
    ## ------------------------------------------------- ##

    ## TEMP FIX
    ## ------------------------------------------------- ##
    frmt.null_joiner.numeric <- "NVL(T1.%s, -123.1230321) = NVL(T2.%1$s, -123.1230321)"
    frmt.null_joiner.string  <- "NVL(T1.%s, 'dummy_dummy_value___xyz123') = NVL(T2.%1$s, 'dummy_dummy_value___xyz123')"
    ## Which columns to use the numeric version. Technically UPC is numeric and album_code is string; but should only use album_code anyway
    tmp.known_numeric_cols <- c("streams", "new_users", "playlist_is_unknown", "user_birthyear", "length_seconds", "row_id")
    colsToPull.cumsum %>% setdiff("date") %>% {sprintf(ifelse(. %in% tmp.known_numeric_cols, frmt.null_joiner.numeric, frmt.null_joiner.string), .)} %>% paste(" AND ", .) %>% pasteC
    ## ------------------------------------------------- ##

  })
  join_statement.cumsum <- sprintf("\n INNER JOIN %s T2\n   ON T1.date >= T2.date  %s", names(with.cumsum), ON_AND.cumsum)
  tbl.cumsum <- c(T1=names(with.cumsum))

  ## CREATE AN ACCELERATION SCORE
  ## Only do so if the only colsToAgg is "streams"
  ## This could be expanded to other colsToAgg if needed, so long as the aggFunc is "SUM"
  if (isTRUE(all("streams" == colsToAgg))) {
    cols <- c("streams", "new_users")

    #    make_accel_cols <- function(col, dateCol="date", ctp)  {
    #      ctp %<>% setdiff(dateCol) %>% setdiff(cols)
    #      LG <- sprintf("(LAG(T1.%s, 1, NULL) OVER (PARTITION BY %s ORDER BY T1.%s ASC))", col, commaSep(paste0("T1.", ctp)), dateCol)
    #
    #      diffCol        <- sprintf("(T1.%2$s - %1$s)", LG, col)                                %>% setNames(obj=., nm=paste0(col, "_diff_in_streams"))
    #      percCol        <- sprintf("(T1.%2$s - %1$s)/%1$s", LG, col)                           %>% setNames(obj=., nm=paste0(col, "_perc_increase"))
    #      accelCol       <- sprintf("score_accel_before_after(%1$s, T1.%2$s)", LG, col)         %>% setNames(obj=., nm=paste0(col, "_accel_score"))
    #      scaledAccelCol <- sprintf("scaled_score_accel_before_after(%1$s, T1.%2$s)", LG, col)  %>% setNames(obj=., nm=paste0(col, "_accel_score_scaled"))
    #
    #      return(c(diffCol, percCol, accelCol, scaledAccelCol))
    #    }
    addl_cols <- lapply(unname(cols), make_accel_cols, orderby=dateCol.nm, partitionby=setdiff(colsToPull.cumsum, cols)) %>% unlist(recursive=FALSE, use.names=TRUE)
  } else {
    addl_cols <- NULL
  }

  qry.cumsum <- makeQry(with=with.cumsum, tbl=tbl.cumsum, colsToPull=colsToPull.cumsum %>% paste0("T1.", .) %>% c(addl_cols), colsToAgg=colsToAgg.cumsum, join=join_statement.cumsum, aggFunc="sum", prependCols.with.tbl=FALSE, nogroupby=TRUE, limit=NULL) %>% paste0(., "\nGROUP BY ", commaSep(paste0("T1.", colsToPull.cumsum)))

  # if (verbose.qry)
  #   catn(verboseQry(qry.cumsum, max.width=150, max.lines=55, add.dots=FALSE))

  browser(expr=inDebugMode(c("user_counts", "create_spotify_user_counts_table")), text="in create_spotify_user_counts_table() right before calling sfPopulateTable()")
  ##  catn(verboseQry(qry.cumsum))

  sfPopulateTable(tbl=tbl_out, schema=schema, qry=qry.cumsum, over=TRUE, verbose.rows=TRUE, verbose=verbose.qry)

  if (confirm_streams) {
    tryCatch({
      streams_users <- makeQry(tbl=tbl_out, schema=schema, colsToAgg=c("streams", "new_users"), aggFunc="sum", limit=NULL) %>% sfQry(verbose=FALSE) %>% unlist()
      streams <- streams_users[["streams"]]
      new_users <- streams_users[["new_users"]]
      msg <- sprintf("Total number of streams %s (%s).  Total New Users is %s", ifelse(streams == streams_expected, "confirmed", "is *WRONG*"), formnumb(streams, round=FALSE), formnumb(new_users, round=-3))
      verboseMsg(TRUE, msg, func="message", minw=100)
    }, error=function(e) {
      warning("Was unable to confirm streams due to an error. The error was caught and execution unaffected. Caught error message: \n\t", e$message, "\n", call.=FALSE);
    })
  }

  return(tbl_out)
}


## EXAMPLE
##
#   ----------------------------------------------------------
#   
#   WITH unnamed_cte AS (
#      SELECT date_first_streamed AS date, isrc, source, playlist_is_unknown, playlist_uri
#           , SUM(streams) AS streams
#           , COUNT(*) AS new_users, seq8() AS row_number
#      FROM   
#      (
#         SELECT   isrc, anonymized_uid, source, playlist_is_unknown, playlist_uri
#           , SUM(streams) AS streams
#           , MIN(first_tmstamp::date) AS date_first_streamed
#           , MAX(last_tmstamp::date) AS date_last_streamed
#          FROM   spot_user_counts.user_minmax_timestamps_detailed_test
#         -- DEBUGGING
#          WHERE isrc = 'TR0061500165' AND first_tmstamp < '2015-08-30'
#          GROUP BY   1, 2, 3, 4, 5  
#      )
#      GROUP BY   1, 2, 3, 4, 5
#      ORDER BY   date,isrc,source,playlist_is_unknown,playlist_uri
#   )
#   
#   SELECT 
#       T1.date
#     ---------
#     , T1.isrc
#     , T1.source
#     , T1.playlist_is_unknown
#     , T1.playlist_uri
#     ---------
#     , T1.streams
#     , T1.new_users
#     , sum(T2.new_users) AS cusers
#     , sum(T2.streams) AS cstreams
#   FROM       unnamed_cte T1
#   FULL JOIN  unnamed_cte T2 
#   ON T1.date >= T2.date
#    AND equal_null(T1.isrc, T2.isrc)
#    AND equal_null(T1.source, T2.source)
#    AND equal_null(T1.playlist_is_unknown, T2.playlist_is_unknown)
#    AND equal_null(T1.playlist_uri, T2.playlist_uri)
#   WHERE T1.isrc = 'TR0061500165' AND T1.date < '2015-08-30'
#   GROUP BY T1.date, T1.isrc, T1.source, T1.playlist_is_unknown, T1.playlist_uri, T1.streams, T1.new_users
#   order by source asc, 1, 2, 3
#   ;
#   ----------------------------------------------------------