implement fifo-v2 to transfer stock pakan

internal/common/service/fifo_stock_v2/gather.go

@@ -199,18 +199,18 @@ func (s *fifoStockV2Service) buildGatherSubquery(rule routeRule, trait traitRule
 	}
 
 	if rule.ScopeSQL != nil && strings.TrimSpace(*rule.ScopeSQL) != "" {
-		whereParts = append(whereParts, fmt.Sprintf("(%s)", strings.TrimSpace(*rule.ScopeSQL)))
+		whereParts = append(whereParts, fmt.Sprintf("(%s)", normalizeScopeSQL(*rule.ScopeSQL)))
 	}
 
 	subquery := fmt.Sprintf(`
 		SELECT
-			? AS source_table,
-			? AS legacy_type_key,
-			? AS function_code,
+			?::text AS source_table,
+			?::text AS legacy_type_key,
+			?::text AS function_code,
 			src.%s AS source_id,
 			src.%s AS product_warehouse_id,
 			%s AS sort_at,
-			? AS sort_priority,
+			?::int AS sort_priority,
 			%s AS quantity,
 			%s AS used_quantity,
 			%s AS pending_quantity,

internal/common/service/fifo_stock_v2/scope_sql.go

@@ -0,0 +1,100 @@
+package fifo_stock_v2
+
+import "strings"
+
+func normalizeScopeSQL(scopeSQL string) string {
+	scopeSQL = strings.TrimSpace(scopeSQL)
+	if scopeSQL == "" {
+		return scopeSQL
+	}
+
+	var out strings.Builder
+	out.Grow(len(scopeSQL) + 16)
+
+	inSingleQuote := false
+	inDoubleQuote := false
+
+	for i := 0; i < len(scopeSQL); {
+		ch := scopeSQL[i]
+
+		if inSingleQuote {
+			out.WriteByte(ch)
+			i++
+			if ch == '\'' {
+				if i < len(scopeSQL) && scopeSQL[i] == '\'' {
+					out.WriteByte(scopeSQL[i])
+					i++
+				} else {
+					inSingleQuote = false
+				}
+			}
+			continue
+		}
+
+		if inDoubleQuote {
+			out.WriteByte(ch)
+			i++
+			if ch == '"' {
+				inDoubleQuote = false
+			}
+			continue
+		}
+
+		if ch == '\'' {
+			inSingleQuote = true
+			out.WriteByte(ch)
+			i++
+			continue
+		}
+
+		if ch == '"' {
+			inDoubleQuote = true
+			out.WriteByte(ch)
+			i++
+			continue
+		}
+
+		if isIdentifierStart(ch) {
+			start := i
+			i++
+			for i < len(scopeSQL) && isIdentifierPart(scopeSQL[i]) {
+				i++
+			}
+
+			token := scopeSQL[start:i]
+			if strings.EqualFold(token, "deleted_at") && !hasAliasQualifier(scopeSQL, start) {
+				out.WriteString("src.deleted_at")
+			} else {
+				out.WriteString(token)
+			}
+			continue
+		}
+
+		out.WriteByte(ch)
+		i++
+	}
+
+	return out.String()
+}
+
+func hasAliasQualifier(scopeSQL string, tokenStart int) bool {
+	for i := tokenStart - 1; i >= 0; i-- {
+		switch scopeSQL[i] {
+		case ' ', '\t', '\n', '\r':
+			continue
+		case '.':
+			return true
+		default:
+			return false
+		}
+	}
+	return false
+}
+
+func isIdentifierStart(ch byte) bool {
+	return (ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z') || ch == '_'
+}
+
+func isIdentifierPart(ch byte) bool {
+	return isIdentifierStart(ch) || (ch >= '0' && ch <= '9')
+}