123 lines
3.6 KiB
Go
123 lines
3.6 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)
|
|
runOnStartup := productPerformanceEnvBool("PRODUCT_PERFORMANCE_DELTA_RUN_ON_STARTUP", false)
|
|
|
|
var running int32
|
|
runDelta := func(reason string) {
|
|
if !atomic.CompareAndSwapInt32(&running, 0, 1) {
|
|
log.Printf("[ProductPerformanceJob] skip (%s): already running", reason)
|
|
return
|
|
}
|
|
defer atomic.StoreInt32(&running, 0)
|
|
|
|
endDate := time.Now()
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(deltaTimeoutHours)*time.Hour)
|
|
defer cancel()
|
|
|
|
result, err := queries.RefreshProductPerformance(ctx, pgDB, queries.ProductPerformanceRefreshRequest{
|
|
Mode: "delta",
|
|
Stage: "all",
|
|
StartDate: endDate.AddDate(0, 0, -deltaDays),
|
|
EndDate: endDate,
|
|
})
|
|
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)
|
|
}
|
|
|
|
log.Printf("[ProductPerformanceJob] scheduled daily_delta=%s lookback_days=%d run_delta_on_startup=%t",
|
|
deltaHHMM, deltaDays, runOnStartup)
|
|
|
|
go func() {
|
|
if runOnStartup {
|
|
time.Sleep(20 * time.Second)
|
|
runDelta("startup-delta")
|
|
}
|
|
|
|
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-delta")
|
|
}
|
|
}()
|
|
}
|
|
|
|
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 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"
|
|
}
|