252 lines
7.9 KiB
Go
252 lines
7.9 KiB
Go
package main
|
||
|
||
import (
|
||
"bssapp-backend/queries"
|
||
"context"
|
||
"database/sql"
|
||
"log"
|
||
"os"
|
||
"strconv"
|
||
"strings"
|
||
"sync/atomic"
|
||
"time"
|
||
)
|
||
|
||
func startProductPerformanceScheduler(pgDB *sql.DB) {
|
||
enabled := strings.TrimSpace(strings.ToLower(os.Getenv("PRODUCT_PERFORMANCE_ENABLED")))
|
||
if enabled == "0" || enabled == "false" || enabled == "off" {
|
||
log.Println("Product performance scheduler disabled")
|
||
return
|
||
}
|
||
if pgDB == nil {
|
||
return
|
||
}
|
||
|
||
deltaDays := productPerformanceEnvInt("PRODUCT_PERFORMANCE_DELTA_DAYS", 45, 1, 365)
|
||
deltaHHMM := productPerformanceEnvString("PRODUCT_PERFORMANCE_DELTA_HHMM", "02:00")
|
||
deltaTimeoutHours := productPerformanceEnvInt("PRODUCT_PERFORMANCE_DELTA_TIMEOUT_HOURS", 6, 1, 24)
|
||
|
||
fullEnabled := productPerformanceEnvBool("PRODUCT_PERFORMANCE_FULL_ENABLED", true)
|
||
fullWeekday := productPerformanceEnvWeekday("PRODUCT_PERFORMANCE_FULL_WEEKDAY", time.Saturday)
|
||
fullHHMM := productPerformanceEnvString("PRODUCT_PERFORMANCE_FULL_HHMM", "16:00")
|
||
fullTimeoutHours := productPerformanceEnvInt("PRODUCT_PERFORMANCE_FULL_TIMEOUT_HOURS", 18, 1, 48)
|
||
fullStart := productPerformanceEnvDate("PRODUCT_PERFORMANCE_FULL_START_DATE", time.Date(2022, 1, 1, 0, 0, 0, 0, time.Local))
|
||
|
||
resumeOnStartup := productPerformanceEnvBool("PRODUCT_PERFORMANCE_RESUME_ON_STARTUP", true)
|
||
runOnStartup := productPerformanceEnvBool("PRODUCT_PERFORMANCE_RUN_ON_STARTUP", false)
|
||
|
||
var running int32
|
||
runRefresh := func(reason string, req queries.ProductPerformanceRefreshRequest, timeoutHours int) {
|
||
if !atomic.CompareAndSwapInt32(&running, 0, 1) {
|
||
log.Printf("[ProductPerformanceJob] skip (%s): already running", reason)
|
||
return
|
||
}
|
||
defer atomic.StoreInt32(&running, 0)
|
||
|
||
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeoutHours)*time.Hour)
|
||
defer cancel()
|
||
|
||
result, err := queries.RefreshProductPerformance(ctx, pgDB, req)
|
||
if err != nil {
|
||
log.Printf("[ProductPerformanceJob] error (%s): %v", reason, err)
|
||
return
|
||
}
|
||
log.Printf("[ProductPerformanceJob] ok (%s): mode=%s stage=%s start=%s end=%s sales=%d stock=%d kpi=%d duration_ms=%d",
|
||
reason, result.Mode, result.Stage, result.StartDate, result.EndDate, result.SalesRows, result.StockRows, result.KpiRows, result.DurationMS)
|
||
}
|
||
|
||
runDelta := func(reason string) {
|
||
endDate := time.Now()
|
||
runRefresh(reason, queries.ProductPerformanceRefreshRequest{
|
||
Mode: "delta",
|
||
Stage: "all",
|
||
StartDate: endDate.AddDate(0, 0, -deltaDays),
|
||
EndDate: endDate,
|
||
}, deltaTimeoutHours)
|
||
}
|
||
|
||
runFull := func(reason string) {
|
||
runRefresh(reason, queries.ProductPerformanceRefreshRequest{
|
||
Mode: "full",
|
||
Stage: "all",
|
||
StartDate: fullStart,
|
||
EndDate: time.Now(),
|
||
}, fullTimeoutHours)
|
||
}
|
||
|
||
runFullResume := func(reason string) {
|
||
resumeAfter, ok := productPerformanceStockResumeAfter(pgDB, fullStart, time.Now())
|
||
if !ok {
|
||
log.Printf("[ProductPerformanceJob] resume skip (%s): no committed full stock progress found", reason)
|
||
return
|
||
}
|
||
runRefresh(reason, queries.ProductPerformanceRefreshRequest{
|
||
Mode: "full",
|
||
Stage: "stock",
|
||
ResumeAfter: resumeAfter,
|
||
SkipDelete: true,
|
||
StartDate: fullStart,
|
||
EndDate: time.Now(),
|
||
}, fullTimeoutHours)
|
||
}
|
||
|
||
log.Printf("[ProductPerformanceJob] scheduled daily_delta=%s lookback_days=%d weekly_full=%t %s %s full_start=%s resume_on_startup=%t",
|
||
deltaHHMM, deltaDays, fullEnabled, fullWeekday.String(), fullHHMM, fullStart.Format("2006-01-02"), resumeOnStartup)
|
||
|
||
go func() {
|
||
if resumeOnStartup {
|
||
time.Sleep(20 * time.Second)
|
||
runFullResume("startup-resume")
|
||
} else if runOnStartup {
|
||
time.Sleep(20 * time.Second)
|
||
runDelta("startup")
|
||
}
|
||
|
||
for {
|
||
next := productPerformanceNextDaily(time.Now(), deltaHHMM)
|
||
log.Printf("[ProductPerformanceJob] daily delta next_at=%s in=%s", next.Format(time.RFC3339), time.Until(next).Round(time.Second))
|
||
time.Sleep(time.Until(next))
|
||
runDelta("daily-02")
|
||
}
|
||
}()
|
||
|
||
if fullEnabled {
|
||
go func() {
|
||
for {
|
||
next := productPerformanceNextWeekly(time.Now(), fullWeekday, fullHHMM)
|
||
log.Printf("[ProductPerformanceJob] weekly full next_at=%s in=%s", next.Format(time.RFC3339), time.Until(next).Round(time.Second))
|
||
time.Sleep(time.Until(next))
|
||
runFull("weekly-full")
|
||
}
|
||
}()
|
||
}
|
||
}
|
||
|
||
func productPerformanceStockResumeAfter(pgDB *sql.DB, startDate, endDate time.Time) (int, bool) {
|
||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||
defer cancel()
|
||
if err := queries.EnsureProductPerformanceTables(pgDB); err != nil {
|
||
log.Printf("[ProductPerformanceJob] resume inspect failed: %v", err)
|
||
return 0, false
|
||
}
|
||
var stockRows int
|
||
var maxStockDate sql.NullTime
|
||
var maxKPIDate sql.NullTime
|
||
if err := pgDB.QueryRowContext(ctx, `
|
||
SELECT COUNT(*), MAX(stock_date)
|
||
FROM mk_product_performance_stock_daily
|
||
WHERE stock_date BETWEEN $1 AND $2
|
||
`, startDate, endDate).Scan(&stockRows, &maxStockDate); err != nil {
|
||
log.Printf("[ProductPerformanceJob] resume inspect stock failed: %v", err)
|
||
return 0, false
|
||
}
|
||
if stockRows <= 0 {
|
||
return 0, false
|
||
}
|
||
if err := pgDB.QueryRowContext(ctx, `
|
||
SELECT MAX(kpi_date)
|
||
FROM mk_product_performance_kpi_daily
|
||
`).Scan(&maxKPIDate); err != nil {
|
||
log.Printf("[ProductPerformanceJob] resume inspect kpi failed: %v", err)
|
||
return 0, false
|
||
}
|
||
if maxStockDate.Valid && maxKPIDate.Valid && !maxStockDate.Time.After(maxKPIDate.Time) {
|
||
return 0, false
|
||
}
|
||
return stockRows, true
|
||
}
|
||
|
||
func productPerformanceNextDaily(now time.Time, hhmm string) time.Time {
|
||
hour, minute := productPerformanceParseHHMM(hhmm, 2, 0)
|
||
next := time.Date(now.Year(), now.Month(), now.Day(), hour, minute, 0, 0, now.Location())
|
||
if !next.After(now) {
|
||
next = next.AddDate(0, 0, 1)
|
||
}
|
||
return next
|
||
}
|
||
|
||
func productPerformanceNextWeekly(now time.Time, weekday time.Weekday, hhmm string) time.Time {
|
||
hour, minute := productPerformanceParseHHMM(hhmm, 16, 0)
|
||
next := time.Date(now.Year(), now.Month(), now.Day(), hour, minute, 0, 0, now.Location())
|
||
days := (int(weekday) - int(now.Weekday()) + 7) % 7
|
||
next = next.AddDate(0, 0, days)
|
||
if !next.After(now) {
|
||
next = next.AddDate(0, 0, 7)
|
||
}
|
||
return next
|
||
}
|
||
|
||
func productPerformanceParseHHMM(raw string, fallbackHour, fallbackMinute int) (int, int) {
|
||
parts := strings.Split(strings.TrimSpace(raw), ":")
|
||
if len(parts) != 2 {
|
||
return fallbackHour, fallbackMinute
|
||
}
|
||
h, herr := strconv.Atoi(parts[0])
|
||
m, merr := strconv.Atoi(parts[1])
|
||
if herr != nil || merr != nil || h < 0 || h > 23 || m < 0 || m > 59 {
|
||
return fallbackHour, fallbackMinute
|
||
}
|
||
return h, m
|
||
}
|
||
|
||
func productPerformanceEnvString(name, fallback string) string {
|
||
raw := strings.TrimSpace(os.Getenv(name))
|
||
if raw == "" {
|
||
return fallback
|
||
}
|
||
return raw
|
||
}
|
||
|
||
func productPerformanceEnvInt(name string, fallback, min, max int) int {
|
||
raw := strings.TrimSpace(os.Getenv(name))
|
||
if raw == "" {
|
||
return fallback
|
||
}
|
||
n, err := strconv.Atoi(raw)
|
||
if err != nil || n < min || n > max {
|
||
return fallback
|
||
}
|
||
return n
|
||
}
|
||
|
||
func productPerformanceEnvBool(name string, fallback bool) bool {
|
||
raw := strings.TrimSpace(strings.ToLower(os.Getenv(name)))
|
||
if raw == "" {
|
||
return fallback
|
||
}
|
||
return raw == "1" || raw == "true" || raw == "on" || raw == "yes"
|
||
}
|
||
|
||
func productPerformanceEnvDate(name string, fallback time.Time) time.Time {
|
||
raw := strings.TrimSpace(os.Getenv(name))
|
||
if raw == "" {
|
||
return fallback
|
||
}
|
||
t, err := time.ParseInLocation("2006-01-02", raw, time.Local)
|
||
if err != nil {
|
||
return fallback
|
||
}
|
||
return t
|
||
}
|
||
|
||
func productPerformanceEnvWeekday(name string, fallback time.Weekday) time.Weekday {
|
||
switch strings.ToLower(strings.TrimSpace(os.Getenv(name))) {
|
||
case "0", "sunday", "pazar":
|
||
return time.Sunday
|
||
case "1", "monday", "pazartesi":
|
||
return time.Monday
|
||
case "2", "tuesday", "sali", "salı":
|
||
return time.Tuesday
|
||
case "3", "wednesday", "carsamba", "çarşamba":
|
||
return time.Wednesday
|
||
case "4", "thursday", "persembe", "perşembe":
|
||
return time.Thursday
|
||
case "5", "friday", "cuma":
|
||
return time.Friday
|
||
case "6", "saturday", "cumartesi":
|
||
return time.Saturday
|
||
default:
|
||
return fallback
|
||
}
|
||
}
|