lingniu-vehicle-ingest/vehicle-data-platform/apps/api/cmd/oneos-scope-sync/main.go

package main

import (
	"context"
	"database/sql"
	"fmt"
	"log"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/go-sql-driver/mysql"

	"lingniu/vehicle-data-platform/apps/api/internal/businessscope"
)

func main() {
	if err := run(); err != nil {
		log.Printf("OneOS scope sync failed: %v", err)
		os.Exit(1)
	}
}

func run() error {
	targetDSN, err := normalizedDSN("MYSQL_DSN", os.Getenv("MYSQL_DSN"), false)
	if err != nil {
		return err
	}
	timeout := time.Duration(envInt("ONEOS_SCOPE_SYNC_TIMEOUT_SEC", 60)) * time.Second
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	target, err := openDB(ctx, targetDSN, 4)
	if err != nil {
		return fmt.Errorf("open vehicle platform database: %w", err)
	}
	defer target.Close()
	snapshot, sourceKind, err := readSnapshot(ctx)
	if err != nil {
		return err
	}
	maxRejected := envInt("ONEOS_SCOPE_MAX_REJECTED", 100)
	maxRejectedRatio, err := envFloat("ONEOS_SCOPE_MAX_REJECT_RATIO", 0.10)
	if err != nil {
		return err
	}
	if err := businessscope.ValidateSnapshot(snapshot, maxRejected, maxRejectedRatio); err != nil {
		return err
	}
	result, err := businessscope.Publish(ctx, target, snapshot)
	if err != nil {
		return err
	}
	log.Printf("OneOS scope sync complete source=%s changed=%t candidates=%d accepted=%d rejected=%d version=%s",
		sourceKind, result.Changed, snapshot.Candidates, len(snapshot.Items), len(snapshot.Rejections), snapshot.SourceVersion)
	return nil
}

func readSnapshot(ctx context.Context) (businessscope.Snapshot, string, error) {
	mode := strings.ToLower(strings.TrimSpace(os.Getenv("ONEOS_SCOPE_SOURCE")))
	if mode == "" {
		if strings.TrimSpace(os.Getenv("ONEOS_SCOPE_API_URL")) != "" {
			mode = "api"
		} else {
			mode = "database"
		}
	}
	switch mode {
	case "api":
		snapshot, err := businessscope.ReadAPISnapshot(ctx, businessscope.APIConfig{
			URL:           os.Getenv("ONEOS_SCOPE_API_URL"),
			ServiceToken:  os.Getenv("ONEOS_SCOPE_API_SERVICE_TOKEN"),
			SigningSecret: os.Getenv("ONEOS_SCOPE_API_SIGNING_SECRET"),
			MaxAttempts:   envInt("ONEOS_SCOPE_API_MAX_ATTEMPTS", 3),
		})
		return snapshot, mode, err
	case "database":
		sourceDSN, err := normalizedDSN("ONEOS_MYSQL_DSN", os.Getenv("ONEOS_MYSQL_DSN"), true)
		if err != nil {
			return businessscope.Snapshot{}, mode, err
		}
		source, err := openDB(ctx, sourceDSN, 2)
		if err != nil {
			return businessscope.Snapshot{}, mode, fmt.Errorf("open OneOS read-only database: %w", err)
		}
		defer source.Close()
		candidates, err := businessscope.ReadCandidates(ctx, source)
		if err != nil {
			return businessscope.Snapshot{}, mode, err
		}
		snapshot, err := businessscope.BuildSnapshot(candidates, time.Now())
		return snapshot, mode, err
	default:
		return businessscope.Snapshot{}, mode, fmt.Errorf("ONEOS_SCOPE_SOURCE must be api or database")
	}
}

func normalizedDSN(name, raw string, requireOneOSDatabase bool) (string, error) {
	raw = strings.TrimSpace(raw)
	if raw == "" {
		return "", fmt.Errorf("%s is required", name)
	}
	config, err := mysql.ParseDSN(raw)
	if err != nil {
		return "", fmt.Errorf("parse %s: %w", name, err)
	}
	if requireOneOSDatabase && config.DBName != "ln_asset_management" {
		return "", fmt.Errorf("%s must select ln_asset_management", name)
	}
	location, err := time.LoadLocation("Asia/Shanghai")
	if err != nil {
		return "", fmt.Errorf("load Asia/Shanghai timezone: %w", err)
	}
	config.ParseTime = true
	config.Loc = location
	if config.Timeout <= 0 {
		config.Timeout = 10 * time.Second
	}
	if config.ReadTimeout <= 0 {
		config.ReadTimeout = 30 * time.Second
	}
	if config.WriteTimeout <= 0 {
		config.WriteTimeout = 10 * time.Second
	}
	return config.FormatDSN(), nil
}

func openDB(ctx context.Context, dsn string, maxConnections int) (*sql.DB, error) {
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return nil, err
	}
	db.SetMaxOpenConns(maxConnections)
	db.SetMaxIdleConns(1)
	db.SetConnMaxLifetime(5 * time.Minute)
	if err := db.PingContext(ctx); err != nil {
		_ = db.Close()
		return nil, err
	}
	return db, nil
}

func envInt(key string, fallback int) int {
	value := strings.TrimSpace(os.Getenv(key))
	if value == "" {
		return fallback
	}
	parsed, err := strconv.Atoi(value)
	if err != nil {
		return fallback
	}
	return parsed
}

func envFloat(key string, fallback float64) (float64, error) {
	value := strings.TrimSpace(os.Getenv(key))
	if value == "" {
		return fallback, nil
	}
	parsed, err := strconv.ParseFloat(value, 64)
	if err != nil || parsed < 0 || parsed > 1 {
		return 0, fmt.Errorf("%s must be a number between 0 and 1", key)
	}
	return parsed, nil
}