Fix product performance grouped JSON build

2026-07-05 02:58:06 +03:00
parent 480cad0d6f
commit 48bde09c2c
4 changed files with 562 additions and 70 deletions
@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"log"
+	"os"
 	"sort"
 	"strings"
 	"time"
@@ -39,16 +40,19 @@ type ProductPerformanceRefreshRequest struct {
 }

 type ProductPerformanceRefreshResult struct {
-	Mode       string `json:"mode"`
-	Stage      string `json:"stage"`
-	StartDate  string `json:"start_date"`
-	EndDate    string `json:"end_date"`
-	SalesRows  int    `json:"sales_rows"`
-	StockRows  int    `json:"stock_rows"`
-	KpiRows    int    `json:"kpi_rows"`
-	DurationMS int64  `json:"duration_ms"`
+	Mode         string `json:"mode"`
+	Stage        string `json:"stage"`
+	StartDate    string `json:"start_date"`
+	EndDate      string `json:"end_date"`
+	SalesRows    int    `json:"sales_rows"`
+	StockRows    int    `json:"stock_rows"`
+	KpiRows      int    `json:"kpi_rows"`
+	SnapshotRows int    `json:"snapshot_rows"`
+	DurationMS   int64  `json:"duration_ms"`
 }

+type productPerformanceSnapshotBypassKey struct{}
+
 func productPerformanceStockQuery() string {
 	return `
 ;WITH ActiveWarehouses AS (
@@ -561,6 +565,23 @@ WHERE btrim(askili_yan) = '-'`,
 		`ALTER TABLE mk_product_performance_kpi_daily ADD COLUMN IF NOT EXISTS unit_profit_base_90d NUMERIC(18,6) NOT NULL DEFAULT 0`,
 		`ALTER TABLE mk_product_performance_kpi_daily ADD COLUMN IF NOT EXISTS unit_profit_base_180d NUMERIC(18,6) NOT NULL DEFAULT 0`,
 		`ALTER TABLE mk_product_performance_kpi_daily ADD COLUMN IF NOT EXISTS market_count_90d INTEGER NOT NULL DEFAULT 0`,
+		`
+CREATE TABLE IF NOT EXISTS mk_product_performance_report_snapshot (
+  report_key TEXT NOT NULL,
+  row_order  INTEGER NOT NULL,
+  payload    JSONB NOT NULL,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  CONSTRAINT pk_mk_product_performance_report_snapshot PRIMARY KEY (report_key, row_order)
+)`,
+		`CREATE INDEX IF NOT EXISTS ix_mk_product_perf_report_snapshot_key ON mk_product_performance_report_snapshot (report_key, row_order)`,
+		`
+CREATE TABLE IF NOT EXISTS mk_product_performance_report_snapshot_meta (
+  report_key TEXT PRIMARY KEY,
+  row_count  INTEGER NOT NULL DEFAULT 0,
+  refreshed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  duration_ms BIGINT NOT NULL DEFAULT 0
+)`,
 	}
 	for _, stmt := range stmts {
 		if _, err := pg.Exec(stmt); err != nil {
@@ -570,6 +591,264 @@ WHERE btrim(askili_yan) = '-'`,
 	return nil
 }

+func productPerformanceSnapshotBypass(ctx context.Context) context.Context {
+	return context.WithValue(ctx, productPerformanceSnapshotBypassKey{}, true)
+}
+
+func productPerformanceUseSnapshot(ctx context.Context) bool {
+	v, _ := ctx.Value(productPerformanceSnapshotBypassKey{}).(bool)
+	return !v
+}
+
+func productPerformanceLiveFallbackEnabled() bool {
+	raw := strings.TrimSpace(strings.ToLower(os.Getenv("PRODUCT_PERFORMANCE_LIVE_FALLBACK")))
+	return raw == "1" || raw == "true" || raw == "on" || raw == "yes"
+}
+
+func productPerformanceSnapshotKey(parts ...string) string {
+	cleaned := make([]string, 0, len(parts))
+	for _, part := range parts {
+		part = strings.ToLower(strings.TrimSpace(part))
+		if part != "" {
+			cleaned = append(cleaned, part)
+		}
+	}
+	return strings.Join(cleaned, ":")
+}
+
+func loadProductPerformanceSnapshotRows[T any](ctx context.Context, pg *sql.DB, reportKey string, limit int) ([]T, bool, error) {
+	if !productPerformanceUseSnapshot(ctx) || pg == nil || strings.TrimSpace(reportKey) == "" {
+		return nil, false, nil
+	}
+	if limit <= 0 || limit > 50000 {
+		limit = 50000
+	}
+	rows, err := pg.QueryContext(ctx, `
+SELECT payload
+FROM mk_product_performance_report_snapshot
+WHERE report_key = $1
+ORDER BY row_order
+LIMIT $2
+`, reportKey, limit)
+	if err != nil {
+		return nil, false, err
+	}
+	defer rows.Close()
+
+	out := make([]T, 0, limit)
+	for rows.Next() {
+		var raw []byte
+		if err := rows.Scan(&raw); err != nil {
+			return nil, false, err
+		}
+		var item T
+		if err := json.Unmarshal(raw, &item); err != nil {
+			return nil, false, err
+		}
+		out = append(out, item)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, false, err
+	}
+	if len(out) > 0 {
+		return out, true, nil
+	}
+	var exists bool
+	if err := pg.QueryRowContext(ctx, `
+SELECT EXISTS (
+  SELECT 1
+  FROM mk_product_performance_report_snapshot_meta
+  WHERE report_key = $1
+)
+`, reportKey).Scan(&exists); err != nil {
+		return nil, false, err
+	}
+	if !exists && !productPerformanceLiveFallbackEnabled() {
+		return out, true, nil
+	}
+	return out, exists, nil
+}
+
+func loadProductPerformanceSnapshotItem[T any](ctx context.Context, pg *sql.DB, reportKey string) (T, bool, error) {
+	var zero T
+	rows, ok, err := loadProductPerformanceSnapshotRows[T](ctx, pg, reportKey, 1)
+	if err != nil || !ok || len(rows) == 0 {
+		return zero, ok, err
+	}
+	return rows[0], true, nil
+}
+
+func saveProductPerformanceSnapshotRows[T any](ctx context.Context, pg *sql.DB, reportKey string, rows []T, started time.Time) error {
+	if pg == nil || strings.TrimSpace(reportKey) == "" {
+		return nil
+	}
+	tx, err := pg.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	if _, err := tx.ExecContext(ctx, `DELETE FROM mk_product_performance_report_snapshot WHERE report_key = $1`, reportKey); err != nil {
+		return err
+	}
+	stmt, err := tx.PrepareContext(ctx, `
+INSERT INTO mk_product_performance_report_snapshot (report_key, row_order, payload, updated_at)
+VALUES ($1, $2, $3, now())
+`)
+	if err != nil {
+		return err
+	}
+	defer stmt.Close()
+	for i, row := range rows {
+		raw, err := json.Marshal(row)
+		if err != nil {
+			return err
+		}
+		if _, err := stmt.ExecContext(ctx, reportKey, i, raw); err != nil {
+			return err
+		}
+	}
+	if _, err := tx.ExecContext(ctx, `
+INSERT INTO mk_product_performance_report_snapshot_meta (report_key, row_count, refreshed_at, duration_ms)
+VALUES ($1, $2, now(), $3)
+ON CONFLICT (report_key) DO UPDATE SET
+  row_count = EXCLUDED.row_count,
+  refreshed_at = EXCLUDED.refreshed_at,
+  duration_ms = EXCLUDED.duration_ms
+`, reportKey, len(rows), time.Since(started).Milliseconds()); err != nil {
+		return err
+	}
+	return tx.Commit()
+}
+
+func RebuildProductPerformanceReportSnapshots(ctx context.Context, pg *sql.DB) (int, error) {
+	if err := EnsureProductPerformanceTables(pg); err != nil {
+		return 0, err
+	}
+	ctx = productPerformanceSnapshotBypass(ctx)
+	total := 0
+
+	save := func(reportKey string, rows any, started time.Time) error {
+		switch v := rows.(type) {
+		case []models.ProductPerformanceSummary:
+			total += len(v)
+			return saveProductPerformanceSnapshotRows(ctx, pg, reportKey, v, started)
+		case []models.ProductPerformanceRow:
+			total += len(v)
+			return saveProductPerformanceSnapshotRows(ctx, pg, reportKey, v, started)
+		case []models.ProductPerformanceGeneralRow:
+			total += len(v)
+			return saveProductPerformanceSnapshotRows(ctx, pg, reportKey, v, started)
+		case []models.ProductPerformanceOrderAnalysisRow:
+			total += len(v)
+			return saveProductPerformanceSnapshotRows(ctx, pg, reportKey, v, started)
+		case []models.ProductPerformanceOrderGroupRow:
+			total += len(v)
+			return saveProductPerformanceSnapshotRows(ctx, pg, reportKey, v, started)
+		case []models.ProductPerformanceOrderProductCustomerRow:
+			total += len(v)
+			return saveProductPerformanceSnapshotRows(ctx, pg, reportKey, v, started)
+		case []models.ProductPerformanceOrderMarketDetailRow:
+			total += len(v)
+			return saveProductPerformanceSnapshotRows(ctx, pg, reportKey, v, started)
+		case []models.ProductPerformanceMarketRow:
+			total += len(v)
+			return saveProductPerformanceSnapshotRows(ctx, pg, reportKey, v, started)
+		case []models.ProductPerformanceCountryRow:
+			total += len(v)
+			return saveProductPerformanceSnapshotRows(ctx, pg, reportKey, v, started)
+		case []models.ProductPerformanceCustomerRow:
+			total += len(v)
+			return saveProductPerformanceSnapshotRows(ctx, pg, reportKey, v, started)
+		case []models.ProductPerformanceSalesBreakdownRow:
+			total += len(v)
+			return saveProductPerformanceSnapshotRows(ctx, pg, reportKey, v, started)
+		default:
+			return fmt.Errorf("unsupported product performance snapshot payload %s", reportKey)
+		}
+	}
+
+	started := time.Now()
+	if summary, err := GetProductPerformanceSummary(ctx, pg); err != nil {
+		return total, err
+	} else if err := save(productPerformanceSnapshotKey("summary"), []models.ProductPerformanceSummary{summary}, started); err != nil {
+		return total, err
+	}
+	started = time.Now()
+	if rows, _, err := ListProductPerformance(ctx, pg, ProductPerformanceFilters{Limit: 50000, Page: 1, SortBy: "performance_score", Descending: true}); err != nil {
+		return total, err
+	} else if err := save(productPerformanceSnapshotKey("products"), rows, started); err != nil {
+		return total, err
+	}
+	started = time.Now()
+	if rows, err := ListProductPerformanceGeneral(ctx, pg, 50000); err != nil {
+		return total, err
+	} else if err := save(productPerformanceSnapshotKey("general"), rows, started); err != nil {
+		return total, err
+	}
+	started = time.Now()
+	if rows, err := ListProductPerformanceOrderAnalysis(ctx, pg, 50000); err != nil {
+		return total, err
+	} else if err := save(productPerformanceSnapshotKey("orders"), rows, started); err != nil {
+		return total, err
+	}
+	for _, mode := range []string{"market", "customer"} {
+		started = time.Now()
+		rows, err := ListProductPerformanceOrderGroups(ctx, pg, mode, 50000)
+		if err != nil {
+			return total, err
+		}
+		if err := save(productPerformanceSnapshotKey("order-groups", mode), rows, started); err != nil {
+			return total, err
+		}
+	}
+	started = time.Now()
+	if rows, err := ListProductPerformanceOrderProductCustomers(ctx, pg, 50000); err != nil {
+		return total, err
+	} else if err := save(productPerformanceSnapshotKey("order-product-customers"), rows, started); err != nil {
+		return total, err
+	}
+	started = time.Now()
+	if rows, err := ListProductPerformanceOrderMarketDetails(ctx, pg, 50000); err != nil {
+		return total, err
+	} else if err := save(productPerformanceSnapshotKey("order-market-details"), rows, started); err != nil {
+		return total, err
+	}
+	started = time.Now()
+	if rows, err := ListProductPerformanceMarkets(ctx, pg, 50000); err != nil {
+		return total, err
+	} else if err := save(productPerformanceSnapshotKey("markets"), rows, started); err != nil {
+		return total, err
+	}
+	started = time.Now()
+	if rows, err := ListProductPerformanceCountries(ctx, pg, 50000); err != nil {
+		return total, err
+	} else if err := save(productPerformanceSnapshotKey("countries"), rows, started); err != nil {
+		return total, err
+	}
+	for _, mode := range []string{"market_customer", "country_customer", "market_country_customer"} {
+		started = time.Now()
+		rows, err := ListProductPerformanceCustomers(ctx, pg, mode, 50000)
+		if err != nil {
+			return total, err
+		}
+		if err := save(productPerformanceSnapshotKey("customers", mode), rows, started); err != nil {
+			return total, err
+		}
+	}
+	for _, mode := range []string{"color_yaka_market_customer", "product_country_segment_market_customer", "market_customer_product", "country_segment_market_customer_product"} {
+		started = time.Now()
+		rows, err := ListProductPerformanceSalesBreakdown(ctx, pg, mode, 50000)
+		if err != nil {
+			return total, err
+		}
+		if err := save(productPerformanceSnapshotKey("sales-breakdown", mode), rows, started); err != nil {
+			return total, err
+		}
+	}
+	return total, nil
+}
+
 func RefreshProductPerformance(ctx context.Context, pg *sql.DB, req ProductPerformanceRefreshRequest) (ProductPerformanceRefreshResult, error) {
 	started := time.Now()
 	stage := productPerformanceRefreshStage(req.Stage)
@@ -689,6 +968,7 @@ func RefreshProductPerformance(ctx context.Context, pg *sql.DB, req ProductPerfo
 	}

 	var kpiRows int
+	var snapshotRows int
 	if shouldRun("kpi") {
 		if err := runStage("kpi", func(tx *sql.Tx) error {
 			log.Printf("[ProductPerformanceRefresh] kpi rebuild start")
@@ -702,20 +982,28 @@ func RefreshProductPerformance(ctx context.Context, pg *sql.DB, req ProductPerfo
 		}); err != nil {
 			return ProductPerformanceRefreshResult{}, err
 		}
+		log.Printf("[ProductPerformanceRefresh] report snapshots rebuild start")
+		rows, err := RebuildProductPerformanceReportSnapshots(ctx, pg)
+		if err != nil {
+			return ProductPerformanceRefreshResult{}, err
+		}
+		snapshotRows = rows
+		log.Printf("[ProductPerformanceRefresh] report snapshots rebuild done rows=%d elapsed=%s", snapshotRows, time.Since(started).Round(time.Second))
 	} else {
 		log.Printf("[ProductPerformanceRefresh] kpi skipped stage=%s", stage)
 	}

 	log.Printf("[ProductPerformanceRefresh] refresh done total_elapsed=%s", time.Since(started).Round(time.Second))
 	return ProductPerformanceRefreshResult{
-		Mode:       mode,
-		Stage:      stage,
-		StartDate:  req.StartDate.Format("2006-01-02"),
-		EndDate:    req.EndDate.Format("2006-01-02"),
-		SalesRows:  salesRows,
-		StockRows:  stockRows,
-		KpiRows:    kpiRows,
-		DurationMS: time.Since(started).Milliseconds(),
+		Mode:         mode,
+		Stage:        stage,
+		StartDate:    req.StartDate.Format("2006-01-02"),
+		EndDate:      req.EndDate.Format("2006-01-02"),
+		SalesRows:    salesRows,
+		StockRows:    stockRows,
+		KpiRows:      kpiRows,
+		SnapshotRows: snapshotRows,
+		DurationMS:   time.Since(started).Milliseconds(),
 	}, nil
 }

@@ -1168,6 +1456,13 @@ func ListProductPerformance(ctx context.Context, pg *sql.DB, f ProductPerformanc
 	if page <= 0 {
 		page = 1
 	}
+	if !productPerformanceHasServerFilters(f) {
+		if rows, ok, err := loadProductPerformanceSnapshotRows[models.ProductPerformanceRow](ctx, pg, productPerformanceSnapshotKey("products"), limit); err != nil {
+			return nil, 0, err
+		} else if ok {
+			return rows, len(rows), nil
+		}
+	}
 	where, args := productPerformanceWhere(f)
 	countQuery := `SELECT COUNT(*) FROM mk_product_performance_kpi_daily WHERE kpi_date = (SELECT MAX(kpi_date) FROM mk_product_performance_kpi_daily)` + where
 	var total int
@@ -1237,6 +1532,11 @@ func GetProductPerformanceSummary(ctx context.Context, pg *sql.DB) (models.Produ
 	if err := EnsureProductPerformanceTables(pg); err != nil {
 		return models.ProductPerformanceSummary{}, err
 	}
+	if summary, ok, err := loadProductPerformanceSnapshotItem[models.ProductPerformanceSummary](ctx, pg, productPerformanceSnapshotKey("summary")); err != nil {
+		return models.ProductPerformanceSummary{}, err
+	} else if ok {
+		return summary, nil
+	}
 	var s models.ProductPerformanceSummary
 	err := pg.QueryRowContext(ctx, `
 WITH Latest AS (
@@ -1307,6 +1607,11 @@ func ListProductPerformanceGeneral(ctx context.Context, pg *sql.DB, limit int) (
 	} else if limit > 50000 {
 		limit = 50000
 	}
+	if rows, ok, err := loadProductPerformanceSnapshotRows[models.ProductPerformanceGeneralRow](ctx, pg, productPerformanceSnapshotKey("general"), limit); err != nil {
+		return nil, err
+	} else if ok {
+		return rows, nil
+	}
 	rows, err := pg.QueryContext(ctx, `
 WITH Bounds AS (
  SELECT
@@ -1615,6 +1920,11 @@ func ListProductPerformanceOrderAnalysis(ctx context.Context, pg *sql.DB, limit
 	} else if limit > 50000 {
 		limit = 50000
 	}
+	if rows, ok, err := loadProductPerformanceSnapshotRows[models.ProductPerformanceOrderAnalysisRow](ctx, pg, productPerformanceSnapshotKey("orders"), limit); err != nil {
+		return nil, err
+	} else if ok {
+		return rows, nil
+	}

 	rows, err := db.MssqlDB.QueryContext(ctx, `
 WITH OpenOrderLines AS (
@@ -1917,6 +2227,11 @@ func ListProductPerformanceOrderGroups(ctx context.Context, pg *sql.DB, breakdow
 	} else if limit > 50000 {
 		limit = 50000
 	}
+	if rows, ok, err := loadProductPerformanceSnapshotRows[models.ProductPerformanceOrderGroupRow](ctx, pg, productPerformanceSnapshotKey("order-groups", mode), limit); err != nil {
+		return nil, err
+	} else if ok {
+		return rows, nil
+	}

 	rows, err := db.MssqlDB.QueryContext(ctx, `
 WITH OpenOrderLines AS (
@@ -2153,6 +2468,11 @@ func ListProductPerformanceOrderProductCustomers(ctx context.Context, pg *sql.DB
 	} else if limit > 50000 {
 		limit = 50000
 	}
+	if rows, ok, err := loadProductPerformanceSnapshotRows[models.ProductPerformanceOrderProductCustomerRow](ctx, pg, productPerformanceSnapshotKey("order-product-customers"), limit); err != nil {
+		return nil, err
+	} else if ok {
+		return rows, nil
+	}
 	rows, err := db.MssqlDB.QueryContext(ctx, `
 WITH OpenOrderLines AS (
  SELECT
@@ -2291,6 +2611,11 @@ func ListProductPerformanceOrderMarketDetails(ctx context.Context, pg *sql.DB, l
 	} else if limit > 50000 {
 		limit = 50000
 	}
+	if rows, ok, err := loadProductPerformanceSnapshotRows[models.ProductPerformanceOrderMarketDetailRow](ctx, pg, productPerformanceSnapshotKey("order-market-details"), limit); err != nil {
+		return nil, err
+	} else if ok {
+		return rows, nil
+	}
 	rows, err := db.MssqlDB.QueryContext(ctx, `
 WITH OpenOrderLines AS (
  SELECT
@@ -2414,6 +2739,11 @@ func ListProductPerformanceMarkets(ctx context.Context, pg *sql.DB, limit int) (
 	} else if limit > 50000 {
 		limit = 50000
 	}
+	if rows, ok, err := loadProductPerformanceSnapshotRows[models.ProductPerformanceMarketRow](ctx, pg, productPerformanceSnapshotKey("markets"), limit); err != nil {
+		return nil, err
+	} else if ok {
+		return rows, nil
+	}
 	rows, err := pg.QueryContext(ctx, `
 SELECT
  market_key,
@@ -2472,6 +2802,11 @@ func ListProductPerformanceCountries(ctx context.Context, pg *sql.DB, limit int)
 	} else if limit > 50000 {
 		limit = 50000
 	}
+	if rows, ok, err := loadProductPerformanceSnapshotRows[models.ProductPerformanceCountryRow](ctx, pg, productPerformanceSnapshotKey("countries"), limit); err != nil {
+		return nil, err
+	} else if ok {
+		return rows, nil
+	}
 	rows, err := pg.QueryContext(ctx, `
 SELECT
  customer_country,
@@ -2549,6 +2884,11 @@ func ListProductPerformanceCustomers(ctx context.Context, pg *sql.DB, breakdown
 		selectMarket = "market_key"
 		groupCols = append(groupCols, "market_key")
 	}
+	if rows, ok, err := loadProductPerformanceSnapshotRows[models.ProductPerformanceCustomerRow](ctx, pg, productPerformanceSnapshotKey("customers", mode), limit); err != nil {
+		return nil, err
+	} else if ok {
+		return rows, nil
+	}

 	query := fmt.Sprintf(`
 SELECT
@@ -2709,6 +3049,12 @@ func ListProductPerformanceSalesBreakdown(ctx context.Context, pg *sql.DB, break
 		groupCols = []string{"CASE WHEN btrim(COALESCE(s.urun_ilk_grubu,'')) = '-' THEN '' WHEN upper(translate(btrim(COALESCE(s.urun_ilk_grubu,'')), 'İŞĞÜÖÇ', 'ISGUOC')) IN ('YETISKIN', 'YETISKIN/GARSON', 'GARSON') THEN '' ELSE s.urun_ilk_grubu END", "s.color_code", "s.yaka_kodu", "CASE WHEN btrim(COALESCE(s.askili_yan,'')) = '-' THEN '' ELSE s.askili_yan END", "s.kategori", "s.urun_ana_grubu", "s.urun_alt_grubu", "s.product_code", "s.customer_country", "s.market_key", "s.customer_segment", "s.customer_code"}
 	}

+	if rows, ok, err := loadProductPerformanceSnapshotRows[models.ProductPerformanceSalesBreakdownRow](ctx, pg, productPerformanceSnapshotKey("sales-breakdown", mode), limit); err != nil {
+		return nil, err
+	} else if ok {
+		return rows, nil
+	}
+
 	selectExpr := func(name string) string {
 		if selected[name] {
 			return fields[name] + " AS " + name
@@ -4404,6 +4750,16 @@ func productPerformanceWhere(f ProductPerformanceFilters) (string, []any) {
 	return strings.Join(parts, ""), args
 }

+func productPerformanceHasServerFilters(f ProductPerformanceFilters) bool {
+	return strings.TrimSpace(f.Search) != "" ||
+		strings.TrimSpace(f.ProductCode) != "" ||
+		strings.TrimSpace(f.MarketKey) != "" ||
+		strings.TrimSpace(f.Kategori) != "" ||
+		strings.TrimSpace(f.Seri) != "" ||
+		strings.TrimSpace(f.Bucket) != "" ||
+		f.Page > 1
+}
+
 func productPerformanceOrderBy(sortBy string, desc bool) string {
 	allowed := map[string]string{
 		"product_code":           "product_code",