lti-api/cmd/reflow-adjustments/main.go

package main

import (
	"context"
	"flag"
	"fmt"
	"log"
	"os"
	"sort"
	"strconv"
	"strings"
	"time"

	commonSvc "gitlab.com/mbugroup/lti-api.git/internal/common/service"
	"gitlab.com/mbugroup/lti-api.git/internal/config"
	"gitlab.com/mbugroup/lti-api.git/internal/database"
	entity "gitlab.com/mbugroup/lti-api.git/internal/entities"
	"gitlab.com/mbugroup/lti-api.git/internal/utils/fifo"
	"gorm.io/gorm"
)

type adjustmentRow struct {
	ID                 uint      `gorm:"column:id"`
	ProductWarehouseID uint      `gorm:"column:product_warehouse_id"`
	ProductID          uint      `gorm:"column:product_id"`
	FunctionCode       string    `gorm:"column:function_code"`
	UsageQty           float64   `gorm:"column:usage_qty"`
	PendingQty         float64   `gorm:"column:pending_qty"`
	StockLogIncrease   float64   `gorm:"column:stock_log_increase"`
	StockLogDecrease   float64   `gorm:"column:stock_log_decrease"`
	CreatedAt          time.Time `gorm:"column:created_at"`
}

type routeResolution struct {
	FlagGroupCode string `gorm:"column:flag_group_code"`
	Lane          string `gorm:"column:lane"`
	FunctionCode  string `gorm:"column:function_code"`
	SourceTable   string `gorm:"column:source_table"`
	LegacyTypeKey string `gorm:"column:legacy_type_key"`
}

func main() {
	var (
		idsRaw                 string
		apply                  bool
		asOfCreatedAt          bool
		compensateMissingAlloc bool
	)

	flag.StringVar(&idsRaw, "ids", "", "Comma-separated adjustment IDs (required), example: 1,2")
	flag.BoolVar(&apply, "apply", false, "Apply changes. If false, run as dry-run")
	flag.BoolVar(&asOfCreatedAt, "as-of-created-at", true, "Use adjustment created_at as reflow AsOf boundary")
	flag.BoolVar(&compensateMissingAlloc, "compensate-missing-alloc", true, "When active allocations are missing and usage_qty > 0, temporarily add back usage_qty before reflow")
	flag.Parse()

	ids, err := parseIDs(idsRaw)
	if err != nil {
		log.Fatalf("invalid --ids: %v", err)
	}
	if len(ids) == 0 {
		log.Fatal("--ids is required")
	}

	ctx := context.Background()
	db := database.Connect(config.DBHost, config.DBName)
	fifoStockV2Svc := commonSvc.NewFifoStockV2Service(db, nil)

	adjustments, err := loadAdjustments(ctx, db, ids)
	if err != nil {
		log.Fatalf("failed to load adjustments: %v", err)
	}
	if len(adjustments) == 0 {
		log.Fatal("no adjustments found for provided IDs")
	}

	sort.Slice(adjustments, func(i, j int) bool {
		return adjustments[i].ID < adjustments[j].ID
	})

	fmt.Printf("Mode: %s\n", modeLabel(apply))
	fmt.Printf("Adjustments loaded: %d\n\n", len(adjustments))

	success := 0
	failed := 0
	skipped := 0

	for _, adj := range adjustments {
		if strings.TrimSpace(adj.FunctionCode) == "" {
			fmt.Printf("SKIP adj=%d reason=function_code empty\n", adj.ID)
			skipped++
			continue
		}

		route, err := resolveRouteByFunctionCode(ctx, db, adj.ProductID, strings.ToUpper(strings.TrimSpace(adj.FunctionCode)))
		if err != nil {
			fmt.Printf("FAIL adj=%d error=resolve route: %v\n", adj.ID, err)
			failed++
			continue
		}
		if route.Lane != "USABLE" {
			fmt.Printf("SKIP adj=%d reason=lane=%s (not USABLE)\n", adj.ID, route.Lane)
			skipped++
			continue
		}

		desiredQty := adj.UsageQty + adj.PendingQty
		desiredQtySource := "usage+pending"
		if desiredQty <= 0 && adj.StockLogDecrease > 0 {
			desiredQty = adj.StockLogDecrease
			desiredQtySource = "stock_log.decrease"
		}
		if desiredQty <= 0 {
			fmt.Printf(
				"SKIP adj=%d reason=no usable qty (usage=%.3f pending=%.3f stock_log.decrease=%.3f)\n",
				adj.ID,
				adj.UsageQty,
				adj.PendingQty,
				adj.StockLogDecrease,
			)
			skipped++
			continue
		}

		usableType := fifo.UsableKeyAdjustmentOut.String()
		if route.SourceTable == "adjustment_stocks" && strings.TrimSpace(route.LegacyTypeKey) != "" {
			usableType = strings.TrimSpace(route.LegacyTypeKey)
		}

		activeAllocationCount, err := countActiveAllocations(ctx, db, usableType, adj.ID)
		if err != nil {
			fmt.Printf("FAIL adj=%d error=count allocations: %v\n", adj.ID, err)
			failed++
			continue
		}

		compensateQty := adj.UsageQty
		if compensateQty <= 0 && desiredQtySource == "stock_log.decrease" {
			compensateQty = adj.StockLogDecrease
		}
		shouldCompensate := compensateMissingAlloc && activeAllocationCount == 0 && compensateQty > 0

		reflowReq := commonSvc.FifoStockV2ReflowRequest{
			FlagGroupCode:      route.FlagGroupCode,
			ProductWarehouseID: adj.ProductWarehouseID,
			Usable: commonSvc.FifoStockV2Ref{
				ID:            adj.ID,
				LegacyTypeKey: usableType,
				FunctionCode:  route.FunctionCode,
			},
			DesiredQty:     desiredQty,
			IdempotencyKey: fmt.Sprintf("manual-adjustment-reflow-%d-%d", adj.ID, time.Now().UnixNano()),
		}
		if asOfCreatedAt {
			asOf := adj.CreatedAt
			reflowReq.AsOf = &asOf
		}

		fmt.Printf(
			"PLAN adj=%d pw=%d product=%d function=%s group=%s desired=%.3f source=%s active_alloc=%d compensate=%t\n",
			adj.ID,
			adj.ProductWarehouseID,
			adj.ProductID,
			route.FunctionCode,
			route.FlagGroupCode,
			desiredQty,
			desiredQtySource,
			activeAllocationCount,
			shouldCompensate,
		)

		if !apply {
			skipped++
			continue
		}

		err = db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
			if shouldCompensate {
				if err := tx.Table("product_warehouses").
					Where("id = ?", adj.ProductWarehouseID).
					Update("qty", gorm.Expr("COALESCE(qty,0) + ?", compensateQty)).Error; err != nil {
					return fmt.Errorf("compensate product_warehouse qty: %w", err)
				}
			}

			reflowReq.Tx = tx
			res, err := fifoStockV2Svc.Reflow(ctx, reflowReq)
			if err != nil {
				return err
			}

			fmt.Printf(
				"DONE adj=%d rollback=%.3f allocate=%.3f pending=%.3f\n",
				adj.ID,
				res.Rollback.ReleasedQty,
				res.Allocate.AllocatedQty,
				res.Allocate.PendingQty,
			)
			return nil
		})
		if err != nil {
			fmt.Printf("FAIL adj=%d error=%v\n", adj.ID, err)
			failed++
			continue
		}

		success++
	}

	fmt.Println()
	fmt.Printf("Summary: success=%d failed=%d skipped=%d\n", success, failed, skipped)
	if failed > 0 {
		os.Exit(1)
	}
}

func modeLabel(apply bool) string {
	if apply {
		return "APPLY"
	}
	return "DRY-RUN"
}

func parseIDs(raw string) ([]uint, error) {
	raw = strings.TrimSpace(raw)
	if raw == "" {
		return nil, nil
	}
	parts := strings.Split(raw, ",")
	ids := make([]uint, 0, len(parts))
	seen := map[uint]struct{}{}

	for _, part := range parts {
		part = strings.TrimSpace(part)
		if part == "" {
			continue
		}
		value, err := strconv.ParseUint(part, 10, 64)
		if err != nil {
			return nil, fmt.Errorf("invalid id %q", part)
		}
		if value == 0 {
			return nil, fmt.Errorf("id must be > 0: %q", part)
		}
		id := uint(value)
		if _, ok := seen[id]; ok {
			continue
		}
		seen[id] = struct{}{}
		ids = append(ids, id)
	}

	return ids, nil
}

func loadAdjustments(ctx context.Context, db *gorm.DB, ids []uint) ([]adjustmentRow, error) {
	var rows []adjustmentRow
	err := db.WithContext(ctx).
		Table("adjustment_stocks a").
		Select(`
				a.id,
				a.product_warehouse_id,
				pw.product_id,
				a.function_code,
				COALESCE(a.usage_qty, 0) AS usage_qty,
				COALESCE(a.pending_qty, 0) AS pending_qty,
				COALESCE((
					SELECT sl.increase
					FROM stock_logs sl
					WHERE sl.loggable_type = 'ADJUSTMENT'
						AND sl.loggable_id = a.id
					ORDER BY sl.id DESC
					LIMIT 1
				), 0) AS stock_log_increase,
				COALESCE((
					SELECT sl.decrease
					FROM stock_logs sl
					WHERE sl.loggable_type = 'ADJUSTMENT'
						AND sl.loggable_id = a.id
					ORDER BY sl.id DESC
					LIMIT 1
				), 0) AS stock_log_decrease,
				a.created_at
			`).
		Joins("JOIN product_warehouses pw ON pw.id = a.product_warehouse_id").
		Where("a.id IN ?", ids).
		Find(&rows).Error
	if err != nil {
		return nil, err
	}
	return rows, nil
}

func resolveRouteByFunctionCode(ctx context.Context, db *gorm.DB, productID uint, functionCode string) (*routeResolution, error) {
	var rows []routeResolution
	err := db.WithContext(ctx).
		Table("fifo_stock_v2_route_rules rr").
		Select("rr.flag_group_code, rr.lane, rr.function_code, rr.source_table, rr.legacy_type_key").
		Joins("JOIN fifo_stock_v2_flag_groups fg ON fg.code = rr.flag_group_code AND fg.is_active = TRUE").
		Where("rr.is_active = TRUE").
		Where("rr.function_code = ?", functionCode).
		Where(`
			EXISTS (
				SELECT 1
				FROM flags f
				JOIN fifo_stock_v2_flag_members fm ON fm.flag_name = f.name AND fm.is_active = TRUE
				WHERE f.flagable_type = ?
					AND f.flagable_id = ?
					AND fm.flag_group_code = rr.flag_group_code
			)
		`, entity.FlagableTypeProduct, productID).
		Order("CASE WHEN rr.source_table = 'adjustment_stocks' THEN 0 ELSE 1 END ASC").
		Order("rr.id ASC").
		Find(&rows).Error
	if err != nil {
		return nil, err
	}
	if len(rows) == 0 {
		return nil, fmt.Errorf("no route found for product_id=%d function_code=%s", productID, functionCode)
	}

	selected := rows[0]
	for _, row := range rows {
		if row.Lane != selected.Lane {
			return nil, fmt.Errorf("ambiguous lane for product_id=%d function_code=%s", productID, functionCode)
		}
	}

	selected.FunctionCode = functionCode
	return &selected, nil
}

func countActiveAllocations(ctx context.Context, db *gorm.DB, usableType string, usableID uint) (int64, error) {
	var count int64
	err := db.WithContext(ctx).
		Table("stock_allocations").
		Where("usable_type = ? AND usable_id = ?", usableType, usableID).
		Where("status = ?", entity.StockAllocationStatusActive).
		Count(&count).Error
	if err != nil {
		return 0, err
	}
	return count, nil
}