feat(oneos): add signed scope API client

This commit is contained in:
lingniu
2026-07-16 18:49:01 +08:00
parent 1243efc7dd
commit 46f2026c03
10 changed files with 555 additions and 41 deletions

View File

@@ -23,10 +23,6 @@ func main() {
} }
func run() error { func run() error {
sourceDSN, err := normalizedDSN("ONEOS_MYSQL_DSN", os.Getenv("ONEOS_MYSQL_DSN"), true)
if err != nil {
return err
}
targetDSN, err := normalizedDSN("MYSQL_DSN", os.Getenv("MYSQL_DSN"), false) targetDSN, err := normalizedDSN("MYSQL_DSN", os.Getenv("MYSQL_DSN"), false)
if err != nil { if err != nil {
return err return err
@@ -34,21 +30,12 @@ func run() error {
timeout := time.Duration(envInt("ONEOS_SCOPE_SYNC_TIMEOUT_SEC", 60)) * time.Second timeout := time.Duration(envInt("ONEOS_SCOPE_SYNC_TIMEOUT_SEC", 60)) * time.Second
ctx, cancel := context.WithTimeout(context.Background(), timeout) ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel() defer cancel()
source, err := openDB(ctx, sourceDSN, 2)
if err != nil {
return fmt.Errorf("open OneOS read-only database: %w", err)
}
defer source.Close()
target, err := openDB(ctx, targetDSN, 4) target, err := openDB(ctx, targetDSN, 4)
if err != nil { if err != nil {
return fmt.Errorf("open vehicle platform database: %w", err) return fmt.Errorf("open vehicle platform database: %w", err)
} }
defer target.Close() defer target.Close()
candidates, err := businessscope.ReadCandidates(ctx, source) snapshot, sourceKind, err := readSnapshot(ctx)
if err != nil {
return err
}
snapshot, err := businessscope.BuildSnapshot(candidates, time.Now())
if err != nil { if err != nil {
return err return err
} }
@@ -64,11 +51,50 @@ func run() error {
if err != nil { if err != nil {
return err return err
} }
log.Printf("OneOS scope sync complete changed=%t candidates=%d accepted=%d rejected=%d version=%s", log.Printf("OneOS scope sync complete source=%s changed=%t candidates=%d accepted=%d rejected=%d version=%s",
result.Changed, snapshot.Candidates, len(snapshot.Items), len(snapshot.Rejections), snapshot.SourceVersion) sourceKind, result.Changed, snapshot.Candidates, len(snapshot.Items), len(snapshot.Rejections), snapshot.SourceVersion)
return nil 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) { func normalizedDSN(name, raw string, requireOneOSDatabase bool) (string, error) {
raw = strings.TrimSpace(raw) raw = strings.TrimSpace(raw)
if raw == "" { if raw == "" {

View File

@@ -0,0 +1,343 @@
package businessscope
import (
"context"
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"io"
"net"
"net/http"
"net/url"
"strconv"
"strings"
"time"
)
const (
apiPageLimit = 500
apiMaxPages = 100
apiMaxRows = 50000
apiMaxBodyBytes = 8 << 20
apiSourcePrefix = "oneos-api-v1:"
)
type APIConfig struct {
URL string
ServiceToken string
SigningSecret string
MaxAttempts int
HTTPClient *http.Client
Now func() time.Time
}
type apiEnvelope struct {
Code int `json:"code"`
Message string `json:"message"`
Data apiSnapshotPage `json:"data"`
RequestID string `json:"requestId"`
}
type apiSnapshotPage struct {
ScopeVersion string `json:"scopeVersion"`
GeneratedAt string `json:"generatedAt"`
Complete bool `json:"complete"`
NextCursor string `json:"nextCursor"`
Items []apiScopeItem `json:"items"`
Rejected []apiRejected `json:"rejected"`
}
type apiScopeItem struct {
VehicleID json.RawMessage `json:"vehicleId"`
VIN string `json:"vin"`
PlateNumber string `json:"plateNumber"`
CustomerID json.RawMessage `json:"customerId"`
CustomerName string `json:"customerName"`
ContractID json.RawMessage `json:"contractId"`
ContractCode string `json:"contractCode"`
ProjectName string `json:"projectName"`
DepartmentID json.RawMessage `json:"departmentId"`
DepartmentName string `json:"departmentName"`
ResponsibleUserID json.RawMessage `json:"responsibleUserId"`
ResponsibleName string `json:"responsibleUserName"`
OperationStatus string `json:"operationStatus"`
ScopeStartAt string `json:"scopeStartAt"`
SourceUpdatedAt string `json:"sourceUpdatedAt"`
}
type apiRejected struct {
VehicleID json.RawMessage `json:"vehicleId"`
VIN string `json:"vin"`
CustomerID json.RawMessage `json:"customerId"`
ContractID json.RawMessage `json:"contractId"`
Reason string `json:"reason"`
ReasonCode string `json:"reasonCode"`
}
func ReadAPISnapshot(ctx context.Context, config APIConfig) (Snapshot, error) {
endpoint, err := url.Parse(strings.TrimSpace(config.URL))
if err != nil || endpoint.Scheme == "" || endpoint.Host == "" {
return Snapshot{}, fmt.Errorf("ONEOS_SCOPE_API_URL must be an absolute HTTP(S) URL")
}
if endpoint.Scheme != "https" && !isPrivateAPIHost(endpoint.Hostname()) {
return Snapshot{}, fmt.Errorf("OneOS scope API must use HTTPS outside private networks")
}
if strings.TrimSpace(config.ServiceToken) == "" || strings.TrimSpace(config.SigningSecret) == "" {
return Snapshot{}, fmt.Errorf("OneOS scope API service token and signing secret are required")
}
if config.MaxAttempts <= 0 {
config.MaxAttempts = 3
}
if config.MaxAttempts > 5 {
config.MaxAttempts = 5
}
if config.HTTPClient == nil {
config.HTTPClient = &http.Client{Timeout: 10 * time.Second}
}
if config.Now == nil {
config.Now = time.Now
}
candidates := make([]Candidate, 0, 1024)
upstreamRejections := make([]Rejection, 0)
cursor := ""
seenCursors := map[string]bool{}
scopeVersion := ""
var generatedAt time.Time
for pageNumber := 0; pageNumber < apiMaxPages; pageNumber++ {
pageURL := *endpoint
query := pageURL.Query()
query.Set("limit", strconv.Itoa(apiPageLimit))
if cursor != "" {
query.Set("cursor", cursor)
} else {
query.Del("cursor")
}
pageURL.RawQuery = query.Encode()
page, err := fetchAPIPage(ctx, config, &pageURL)
if err != nil {
return Snapshot{}, err
}
if !page.Complete {
return Snapshot{}, fmt.Errorf("OneOS scope API returned an incomplete snapshot")
}
if strings.TrimSpace(page.ScopeVersion) == "" {
return Snapshot{}, fmt.Errorf("OneOS scope API omitted scopeVersion")
}
pageGeneratedAt, err := time.Parse(time.RFC3339Nano, page.GeneratedAt)
if err != nil {
return Snapshot{}, fmt.Errorf("OneOS scope API generatedAt is invalid: %w", err)
}
if scopeVersion == "" {
scopeVersion = page.ScopeVersion
generatedAt = pageGeneratedAt
} else if page.ScopeVersion != scopeVersion || !pageGeneratedAt.Equal(generatedAt) {
return Snapshot{}, fmt.Errorf("OneOS scope API pagination changed snapshot version")
}
for _, item := range page.Items {
candidate, err := candidateFromAPI(item, len(candidates)+len(upstreamRejections)+1)
if err != nil {
return Snapshot{}, err
}
candidates = append(candidates, candidate)
}
for _, rejected := range page.Rejected {
upstreamRejections = append(upstreamRejections, rejectionFromAPI(rejected, len(candidates)+len(upstreamRejections)+1))
}
if len(candidates)+len(upstreamRejections) > apiMaxRows {
return Snapshot{}, fmt.Errorf("OneOS scope API exceeded %d rows", apiMaxRows)
}
next := strings.TrimSpace(page.NextCursor)
if next == "" {
break
}
if seenCursors[next] {
return Snapshot{}, fmt.Errorf("OneOS scope API repeated pagination cursor")
}
seenCursors[next] = true
cursor = next
if pageNumber == apiMaxPages-1 {
return Snapshot{}, fmt.Errorf("OneOS scope API exceeded %d pages", apiMaxPages)
}
}
snapshot, err := BuildSnapshot(candidates, generatedAt)
if err != nil {
return Snapshot{}, err
}
snapshot.Candidates += len(upstreamRejections)
snapshot.Rejections = append(snapshot.Rejections, upstreamRejections...)
versionHash := sha256.Sum256([]byte(scopeVersion + "\n" + snapshot.Checksum))
snapshot.SourceVersion = apiSourcePrefix + hex.EncodeToString(versionHash[:])
return snapshot, nil
}
func fetchAPIPage(ctx context.Context, config APIConfig, pageURL *url.URL) (apiSnapshotPage, error) {
var lastErr error
for attempt := 1; attempt <= config.MaxAttempts; attempt++ {
requestID, err := randomHex(16)
if err != nil {
return apiSnapshotPage{}, err
}
timestamp := strconv.FormatInt(config.Now().UTC().Unix(), 10)
canonical := http.MethodGet + "\n" + pageURL.RequestURI() + "\n" + timestamp + "\n" + requestID
mac := hmac.New(sha256.New, []byte(config.SigningSecret))
_, _ = mac.Write([]byte(canonical))
request, err := http.NewRequestWithContext(ctx, http.MethodGet, pageURL.String(), nil)
if err != nil {
return apiSnapshotPage{}, err
}
request.Header.Set("Accept", "application/json")
request.Header.Set("Authorization", "Service "+config.ServiceToken)
request.Header.Set("X-Request-Id", requestID)
request.Header.Set("X-Request-Timestamp", timestamp)
request.Header.Set("X-Request-Signature", hex.EncodeToString(mac.Sum(nil)))
response, err := config.HTTPClient.Do(request)
if err != nil {
lastErr = err
} else {
page, retry, responseErr := decodeAPIResponse(response)
if responseErr == nil {
return page, nil
}
lastErr = responseErr
if !retry {
return apiSnapshotPage{}, responseErr
}
}
if attempt < config.MaxAttempts {
select {
case <-ctx.Done():
return apiSnapshotPage{}, ctx.Err()
case <-time.After(time.Duration(attempt*100) * time.Millisecond):
}
}
}
return apiSnapshotPage{}, fmt.Errorf("OneOS scope API failed after %d attempts: %w", config.MaxAttempts, lastErr)
}
func decodeAPIResponse(response *http.Response) (apiSnapshotPage, bool, error) {
defer response.Body.Close()
retry := response.StatusCode == http.StatusTooManyRequests || response.StatusCode >= 500
if response.StatusCode != http.StatusOK {
_, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 4096))
return apiSnapshotPage{}, retry, fmt.Errorf("OneOS scope API returned HTTP %d", response.StatusCode)
}
var envelope apiEnvelope
decoder := json.NewDecoder(io.LimitReader(response.Body, apiMaxBodyBytes+1))
if err := decoder.Decode(&envelope); err != nil {
return apiSnapshotPage{}, false, fmt.Errorf("decode OneOS scope API response: %w", err)
}
if envelope.Code != 0 {
return apiSnapshotPage{}, false, fmt.Errorf("OneOS scope API returned business code %d", envelope.Code)
}
return envelope.Data, false, nil
}
func candidateFromAPI(item apiScopeItem, row int) (Candidate, error) {
vehicleID, err := requiredInt64ID(item.VehicleID, "vehicleId")
if err != nil {
return Candidate{}, err
}
customerID, err := requiredInt64ID(item.CustomerID, "customerId")
if err != nil {
return Candidate{}, err
}
contractID, err := requiredInt64ID(item.ContractID, "contractId")
if err != nil {
return Candidate{}, err
}
scopeStart, err := time.Parse(time.RFC3339Nano, strings.TrimSpace(item.ScopeStartAt))
if err != nil {
return Candidate{}, fmt.Errorf("OneOS scope API item %d has invalid scopeStartAt", row)
}
var sourceUpdatedAt *time.Time
if value := strings.TrimSpace(item.SourceUpdatedAt); value != "" {
parsed, err := time.Parse(time.RFC3339Nano, value)
if err != nil {
return Candidate{}, fmt.Errorf("OneOS scope API item %d has invalid sourceUpdatedAt", row)
}
sourceUpdatedAt = &parsed
}
return Candidate{
RowNumber: row, VehicleID: vehicleID, VehiclePresent: true,
VIN: item.VIN, PlateNumber: item.PlateNumber,
CustomerID: customerID, CustomerName: item.CustomerName, CustomerPresent: true, CustomerProfileExists: true,
ContractID: contractID, ContractPresent: true, ContractProfileExists: true,
EffectiveCustomerID: customerID, EffectiveCustomerSet: true,
ContractCode: item.ContractCode, ProjectName: item.ProjectName,
DepartmentID: rawIDString(item.DepartmentID), DepartmentName: item.DepartmentName,
ResponsibleUserID: rawIDString(item.ResponsibleUserID), ResponsibleUserName: item.ResponsibleName,
OperationStatus: item.OperationStatus, ScopeStartAt: scopeStart, SourceUpdatedAt: sourceUpdatedAt,
}, nil
}
func rejectionFromAPI(item apiRejected, row int) Rejection {
reason := strings.ToUpper(strings.TrimSpace(firstNonEmptyScope(item.ReasonCode, item.Reason)))
if reason == "" || len(reason) > 64 {
reason = "UPSTREAM_REJECTED"
}
rejection := Rejection{RowNumber: row, VIN: normalizeVIN(item.VIN), ReasonCode: reason}
if value, err := optionalInt64ID(item.VehicleID); err == nil && value > 0 {
rejection.VehicleID = int64Pointer(value)
}
if value, err := optionalInt64ID(item.CustomerID); err == nil && value > 0 {
rejection.CustomerID = int64Pointer(value)
}
if value, err := optionalInt64ID(item.ContractID); err == nil && value > 0 {
rejection.ContractID = int64Pointer(value)
}
return rejection
}
func requiredInt64ID(raw json.RawMessage, name string) (int64, error) {
value, err := optionalInt64ID(raw)
if err != nil || value <= 0 {
return 0, fmt.Errorf("OneOS scope API %s must be a positive integer string", name)
}
return value, nil
}
func optionalInt64ID(raw json.RawMessage) (int64, error) {
value := rawIDString(raw)
if value == "" {
return 0, nil
}
return strconv.ParseInt(value, 10, 64)
}
func rawIDString(raw json.RawMessage) string {
value := strings.TrimSpace(string(raw))
if value == "" || value == "null" {
return ""
}
if strings.HasPrefix(value, `"`) {
var decoded string
if json.Unmarshal(raw, &decoded) == nil {
return strings.TrimSpace(decoded)
}
return ""
}
return value
}
func isPrivateAPIHost(host string) bool {
host = strings.ToLower(strings.TrimSpace(host))
if host == "localhost" {
return true
}
address := net.ParseIP(host)
return address != nil && (address.IsPrivate() || address.IsLoopback())
}
func firstNonEmptyScope(values ...string) string {
for _, value := range values {
if strings.TrimSpace(value) != "" {
return value
}
}
return ""
}

View File

@@ -0,0 +1,99 @@
package businessscope
import (
"context"
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"time"
)
func TestReadAPISnapshotAuthenticatesPaginatesAndPreservesDimensions(t *testing.T) {
const token = "service-token"
const secret = "signing-secret"
now := time.Date(2026, 7, 16, 10, 0, 0, 0, time.UTC)
var requests atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
requests.Add(1)
if r.Header.Get("Authorization") != "Service "+token {
t.Fatalf("missing service token")
}
canonical := r.Method + "\n" + r.URL.RequestURI() + "\n" + r.Header.Get("X-Request-Timestamp") + "\n" + r.Header.Get("X-Request-Id")
mac := hmac.New(sha256.New, []byte(secret))
_, _ = mac.Write([]byte(canonical))
if r.Header.Get("X-Request-Signature") != hex.EncodeToString(mac.Sum(nil)) {
t.Fatalf("invalid request signature")
}
w.Header().Set("Content-Type", "application/json")
if r.URL.Query().Get("cursor") == "" {
fmt.Fprint(w, `{"code":0,"data":{"scopeVersion":"scope-42","generatedAt":"2026-07-16T10:00:00Z","complete":true,"nextCursor":"page-2","items":[{"vehicleId":"10","vin":" lvin0001 ","plateNumber":"沪A00001","customerId":"100","customerName":"客户甲","contractId":"1010","contractCode":"HT-1","projectName":"项目甲","departmentId":"20","departmentName":"运营一部","responsibleUserId":"30","responsibleUserName":"张三","operationStatus":"active","scopeStartAt":"2026-07-01T08:00:00+08:00","sourceUpdatedAt":"2026-07-16T09:59:00Z"}],"rejected":[]}}`)
return
}
fmt.Fprint(w, `{"code":0,"data":{"scopeVersion":"scope-42","generatedAt":"2026-07-16T10:00:00Z","complete":true,"nextCursor":"","items":[{"vehicleId":"11","vin":"LVIN0002","plateNumber":"沪A00002","customerId":"100","customerName":"客户甲","contractId":"1011","contractCode":"HT-2","projectName":"项目甲","departmentId":"20","departmentName":"运营一部","responsibleUserId":"31","responsibleUserName":"李四","operationStatus":"active","scopeStartAt":"2026-07-02T08:00:00+08:00"}],"rejected":[{"vehicleId":"12","vin":"","customerId":"100","contractId":"1012","reasonCode":"VIN_MISSING"}]}}`)
}))
defer server.Close()
snapshot, err := ReadAPISnapshot(context.Background(), APIConfig{
URL: server.URL, ServiceToken: token, SigningSecret: secret,
MaxAttempts: 1, HTTPClient: server.Client(), Now: func() time.Time { return now },
})
if err != nil {
t.Fatal(err)
}
if requests.Load() != 2 || snapshot.Candidates != 3 || len(snapshot.Items) != 2 || len(snapshot.Rejections) != 1 {
t.Fatalf("unexpected API snapshot: requests=%d snapshot=%+v", requests.Load(), snapshot)
}
if !strings.HasPrefix(snapshot.SourceVersion, apiSourcePrefix) || len(snapshot.SourceVersion) > 96 {
t.Fatalf("unexpected source version %q", snapshot.SourceVersion)
}
item := snapshot.Items[0]
if item.VIN != "LVIN0001" || item.CustomerName != "客户甲" || item.DepartmentName != "运营一部" ||
item.ResponsibleUserName != "张三" {
t.Fatalf("business dimensions lost: %+v", item)
}
}
func TestReadAPISnapshotFailsClosedOnVersionDrift(t *testing.T) {
var request int
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
request++
version := "scope-1"
cursor := "next"
if request == 2 {
version = "scope-2"
cursor = ""
}
fmt.Fprintf(w, `{"code":0,"data":{"scopeVersion":%q,"generatedAt":"2026-07-16T10:00:00Z","complete":true,"nextCursor":%q,"items":[],"rejected":[]}}`, version, cursor)
}))
defer server.Close()
_, err := ReadAPISnapshot(context.Background(), APIConfig{
URL: server.URL, ServiceToken: "token", SigningSecret: "secret", MaxAttempts: 1, HTTPClient: server.Client(),
})
if err == nil || !strings.Contains(err.Error(), "changed snapshot version") {
t.Fatalf("version drift must fail closed: %v", err)
}
}
func TestReadAPISnapshotRetriesOnlyTransientHTTPFailure(t *testing.T) {
var requests atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if requests.Add(1) == 1 {
http.Error(w, "temporary", http.StatusServiceUnavailable)
return
}
fmt.Fprint(w, `{"code":0,"data":{"scopeVersion":"scope-1","generatedAt":"2026-07-16T10:00:00Z","complete":true,"nextCursor":"","items":[{"vehicleId":"10","vin":"LVIN0001","customerId":"100","contractId":"1010","scopeStartAt":"2026-07-01T00:00:00+08:00"}],"rejected":[]}}`)
}))
defer server.Close()
snapshot, err := ReadAPISnapshot(context.Background(), APIConfig{
URL: server.URL, ServiceToken: "token", SigningSecret: "secret", MaxAttempts: 2, HTTPClient: server.Client(),
})
if err != nil || requests.Load() != 2 || len(snapshot.Items) != 1 {
t.Fatalf("transient retry failed: requests=%d snapshot=%+v err=%v", requests.Load(), snapshot, err)
}
}

View File

@@ -33,6 +33,7 @@ type Candidate struct {
VIN string VIN string
PlateNumber string PlateNumber string
CustomerID int64 CustomerID int64
CustomerName string
CustomerPresent bool CustomerPresent bool
CustomerProfileExists bool CustomerProfileExists bool
ContractID int64 ContractID int64
@@ -42,6 +43,10 @@ type Candidate struct {
EffectiveCustomerSet bool EffectiveCustomerSet bool
ContractCode string ContractCode string
ProjectName string ProjectName string
DepartmentID string
DepartmentName string
ResponsibleUserID string
ResponsibleUserName string
OperationStatus string OperationStatus string
ScopeStartAt time.Time ScopeStartAt time.Time
SourceUpdatedAt *time.Time SourceUpdatedAt *time.Time
@@ -52,9 +57,14 @@ type ScopeItem struct {
VIN string VIN string
PlateNumber string PlateNumber string
CustomerID int64 CustomerID int64
CustomerName string
ContractID int64 ContractID int64
ContractCode string ContractCode string
ProjectName string ProjectName string
DepartmentID string
DepartmentName string
ResponsibleUserID string
ResponsibleUserName string
OperationStatus string OperationStatus string
ScopeStartAt time.Time ScopeStartAt time.Time
SourceUpdatedAt *time.Time SourceUpdatedAt *time.Time
@@ -119,8 +129,10 @@ func BuildSnapshot(candidates []Candidate, generatedAt time.Time) (Snapshot, err
} }
items = append(items, ScopeItem{ items = append(items, ScopeItem{
VehicleID: candidate.VehicleID, VIN: candidate.VIN, PlateNumber: strings.TrimSpace(candidate.PlateNumber), VehicleID: candidate.VehicleID, VIN: candidate.VIN, PlateNumber: strings.TrimSpace(candidate.PlateNumber),
CustomerID: candidate.CustomerID, ContractID: candidate.ContractID, CustomerID: candidate.CustomerID, CustomerName: strings.TrimSpace(candidate.CustomerName), ContractID: candidate.ContractID,
ContractCode: strings.TrimSpace(candidate.ContractCode), ProjectName: strings.TrimSpace(candidate.ProjectName), ContractCode: strings.TrimSpace(candidate.ContractCode), ProjectName: strings.TrimSpace(candidate.ProjectName),
DepartmentID: strings.TrimSpace(candidate.DepartmentID), DepartmentName: strings.TrimSpace(candidate.DepartmentName),
ResponsibleUserID: strings.TrimSpace(candidate.ResponsibleUserID), ResponsibleUserName: strings.TrimSpace(candidate.ResponsibleUserName),
OperationStatus: strings.TrimSpace(candidate.OperationStatus), ScopeStartAt: candidate.ScopeStartAt, OperationStatus: strings.TrimSpace(candidate.OperationStatus), ScopeStartAt: candidate.ScopeStartAt,
SourceUpdatedAt: candidate.SourceUpdatedAt, SourceUpdatedAt: candidate.SourceUpdatedAt,
}) })

View File

@@ -61,8 +61,9 @@ run_id,source_system,source_version,status,candidate_count,accepted_count,reject
return PublishResult{}, fmt.Errorf("clear inactive business scope version: %w", err) return PublishResult{}, fmt.Errorf("clear inactive business scope version: %w", err)
} }
statement, err := tx.PrepareContext(ctx, `INSERT INTO business_customer_vehicle_scope( statement, err := tx.PrepareContext(ctx, `INSERT INTO business_customer_vehicle_scope(
source_version,customer_id,vin,vehicle_id,contract_id,contract_code,plate_number,project_name,operation_status,scope_start_at,source_updated_at,published_at source_version,customer_id,customer_name,vin,vehicle_id,contract_id,contract_code,plate_number,project_name,
) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`) department_id,department_name,responsible_user_id,responsible_user_name,operation_status,scope_start_at,source_updated_at,published_at
) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`)
if err != nil { if err != nil {
return PublishResult{}, fmt.Errorf("prepare business scope insert: %w", err) return PublishResult{}, fmt.Errorf("prepare business scope insert: %w", err)
} }
@@ -72,15 +73,16 @@ source_version,customer_id,vin,vehicle_id,contract_id,contract_code,plate_number
if item.ContractID > 0 { if item.ContractID > 0 {
contractID = item.ContractID contractID = item.ContractID
} }
if _, err := statement.ExecContext(ctx, snapshot.SourceVersion, item.CustomerID, item.VIN, item.VehicleID, contractID, if _, err := statement.ExecContext(ctx, snapshot.SourceVersion, item.CustomerID, item.CustomerName, item.VIN, item.VehicleID, contractID,
item.ContractCode, item.PlateNumber, item.ProjectName, item.OperationStatus, item.ScopeStartAt, item.SourceUpdatedAt, now); err != nil { item.ContractCode, item.PlateNumber, item.ProjectName, item.DepartmentID, item.DepartmentName,
item.ResponsibleUserID, item.ResponsibleUserName, item.OperationStatus, item.ScopeStartAt, item.SourceUpdatedAt, now); err != nil {
return PublishResult{}, fmt.Errorf("insert business scope item: %w", err) return PublishResult{}, fmt.Errorf("insert business scope item: %w", err)
} }
} }
} }
if len(snapshot.Rejections) > 0 { if len(snapshot.Rejections) > 0 {
statement, err := tx.PrepareContext(ctx, `INSERT INTO business_scope_rejection( statement, err := tx.PrepareContext(ctx, `INSERT INTO business_scope_rejection(
run_id,row_number,vehicle_id,vin,customer_id,contract_id,reason_code,created_at run_id,`+"`row_number`"+`,vehicle_id,vin,customer_id,contract_id,reason_code,created_at
) VALUES(?,?,?,?,?,?,?,?)`) ) VALUES(?,?,?,?,?,?,?,?)`)
if err != nil { if err != nil {
return PublishResult{}, fmt.Errorf("prepare scope rejection insert: %w", err) return PublishResult{}, fmt.Errorf("prepare scope rejection insert: %w", err)

View File

@@ -11,8 +11,9 @@ import (
) )
const scopeInsertSQL = `INSERT INTO business_customer_vehicle_scope( const scopeInsertSQL = `INSERT INTO business_customer_vehicle_scope(
source_version,customer_id,vin,vehicle_id,contract_id,contract_code,plate_number,project_name,operation_status,scope_start_at,source_updated_at,published_at source_version,customer_id,customer_name,vin,vehicle_id,contract_id,contract_code,plate_number,project_name,
) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)` department_id,department_name,responsible_user_id,responsible_user_name,operation_status,scope_start_at,source_updated_at,published_at
) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`
func publishSnapshot(version string) Snapshot { func publishSnapshot(version string) Snapshot {
generatedAt := time.Date(2026, 7, 14, 8, 0, 0, 0, time.UTC) generatedAt := time.Date(2026, 7, 14, 8, 0, 0, 0, time.UTC)
@@ -21,8 +22,9 @@ func publishSnapshot(version string) Snapshot {
Checksum: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", Checksum: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
GeneratedAt: generatedAt, Candidates: 1, GeneratedAt: generatedAt, Candidates: 1,
Items: []ScopeItem{{ Items: []ScopeItem{{
VehicleID: 10, VIN: "LVIN0001", PlateNumber: "沪A00001", CustomerID: 100, VehicleID: 10, VIN: "LVIN0001", PlateNumber: "沪A00001", CustomerID: 100, CustomerName: "示例客户",
ContractID: 1010, ContractCode: "HT-1", ProjectName: "项目", OperationStatus: "1", ContractID: 1010, ContractCode: "HT-1", ProjectName: "项目", OperationStatus: "1",
DepartmentID: "20", DepartmentName: "运营一部", ResponsibleUserID: "30", ResponsibleUserName: "负责人",
ScopeStartAt: generatedAt.Add(-time.Hour), ScopeStartAt: generatedAt.Add(-time.Hour),
}}, }},
} }
@@ -47,8 +49,9 @@ func TestPublishRebuildsPreviouslyStoredVersionBeforeActivation(t *testing.T) {
prepared := mock.ExpectPrepare(regexp.QuoteMeta(scopeInsertSQL)) prepared := mock.ExpectPrepare(regexp.QuoteMeta(scopeInsertSQL))
item := snapshot.Items[0] item := snapshot.Items[0]
prepared.ExpectExec().WithArgs( prepared.ExpectExec().WithArgs(
snapshot.SourceVersion, item.CustomerID, item.VIN, item.VehicleID, item.ContractID, snapshot.SourceVersion, item.CustomerID, item.CustomerName, item.VIN, item.VehicleID, item.ContractID,
item.ContractCode, item.PlateNumber, item.ProjectName, item.OperationStatus, item.ScopeStartAt, nil, sqlmock.AnyArg(), item.ContractCode, item.PlateNumber, item.ProjectName, item.DepartmentID, item.DepartmentName,
item.ResponsibleUserID, item.ResponsibleUserName, item.OperationStatus, item.ScopeStartAt, nil, sqlmock.AnyArg(),
).WillReturnResult(sqlmock.NewResult(1, 1)) ).WillReturnResult(sqlmock.NewResult(1, 1))
mock.ExpectExec(`UPDATE business_scope_state SET`). mock.ExpectExec(`UPDATE business_scope_state SET`).
WithArgs(snapshot.SourceVersion, snapshot.Checksum, 1, 1, 0, snapshot.GeneratedAt, sqlmock.AnyArg(), sqlmock.AnyArg()). WithArgs(snapshot.SourceVersion, snapshot.Checksum, 1, 1, 0, snapshot.GeneratedAt, sqlmock.AnyArg(), sqlmock.AnyArg()).

View File

@@ -5,6 +5,7 @@ set -euo pipefail
RELEASE_ID=${1:?release id is required} RELEASE_ID=${1:?release id is required}
ARCHIVE=${2:?web archive is required} ARCHIVE=${2:?web archive is required}
API_BINARY=${3:-} API_BINARY=${3:-}
ONEOS_SCOPE_BINARY=${4:-}
ROOT=${PLATFORM_ROOT:-/opt/lingniu-vehicle-platform} ROOT=${PLATFORM_ROOT:-/opt/lingniu-vehicle-platform}
BASE_URL=${PLATFORM_BASE_URL:-http://127.0.0.1:20300} BASE_URL=${PLATFORM_BASE_URL:-http://127.0.0.1:20300}
SERVICE=${PLATFORM_SERVICE:-lingniu-vehicle-platform} SERVICE=${PLATFORM_SERVICE:-lingniu-vehicle-platform}
@@ -21,6 +22,9 @@ test -f "$ARCHIVE" || { printf 'web archive is missing: %s\n' "$ARCHIVE" >&2; ex
if test -n "$API_BINARY"; then if test -n "$API_BINARY"; then
test -f "$API_BINARY" || { printf 'platform API binary is missing: %s\n' "$API_BINARY" >&2; exit 1; } test -f "$API_BINARY" || { printf 'platform API binary is missing: %s\n' "$API_BINARY" >&2; exit 1; }
fi fi
if test -n "$ONEOS_SCOPE_BINARY"; then
test -f "$ONEOS_SCOPE_BINARY" || { printf 'OneOS scope sync binary is missing: %s\n' "$ONEOS_SCOPE_BINARY" >&2; exit 1; }
fi
test -x "$SCRIPT_DIR/prepare-web-release-tree.sh" || { printf 'release tree helper is missing\n' >&2; exit 1; } test -x "$SCRIPT_DIR/prepare-web-release-tree.sh" || { printf 'release tree helper is missing\n' >&2; exit 1; }
test -f "$SCRIPT_DIR/prune-release-history.py" || { printf 'release pruning helper is missing\n' >&2; exit 1; } test -f "$SCRIPT_DIR/prune-release-history.py" || { printf 'release pruning helper is missing\n' >&2; exit 1; }
test -f "$SCRIPT_DIR/verify-customer-demo.py" || { printf 'customer demo gate is missing\n' >&2; exit 1; } test -f "$SCRIPT_DIR/verify-customer-demo.py" || { printf 'customer demo gate is missing\n' >&2; exit 1; }
@@ -81,6 +85,9 @@ test -f "$next/platform-api" || { printf 'current release is missing platform-ap
if test -n "$API_BINARY"; then if test -n "$API_BINARY"; then
cp "$API_BINARY" "$next/platform-api" cp "$API_BINARY" "$next/platform-api"
fi fi
if test -n "$ONEOS_SCOPE_BINARY"; then
cp "$ONEOS_SCOPE_BINARY" "$next/oneos-scope-sync"
fi
if test -f "$old/lingniu-vehicle-platform.service"; then if test -f "$old/lingniu-vehicle-platform.service"; then
cp "$old/lingniu-vehicle-platform.service" "$next/lingniu-vehicle-platform.service" cp "$old/lingniu-vehicle-platform.service" "$next/lingniu-vehicle-platform.service"
fi fi

View File

@@ -27,22 +27,24 @@ printf '<main>new</main>\n' > "$new_web/index.html"
printf 'window.__LINGNIU_APP_CONFIG__={"amapSecurityServiceHost":"/_AMapService"};\n' > "$new_web/app-config.js" printf 'window.__LINGNIU_APP_CONFIG__={"amapSecurityServiceHost":"/_AMapService"};\n' > "$new_web/app-config.js"
tar -C "$new_web" -czf "$fixture/new.tar.gz" . tar -C "$new_web" -czf "$fixture/new.tar.gz" .
printf 'new platform API\n' > "$fixture/new-platform-api" printf 'new platform API\n' > "$fixture/new-platform-api"
printf 'new OneOS scope sync\n' > "$fixture/new-oneos-scope-sync"
printf '#!/usr/bin/env bash\nexit 0\n' > "$fixture/bin/systemctl" printf '#!/usr/bin/env bash\nexit 0\n' > "$fixture/bin/systemctl"
printf '#!/usr/bin/env bash\nexit 0\n' > "$fixture/bin/curl" printf '#!/usr/bin/env bash\nexit 0\n' > "$fixture/bin/curl"
printf '#!/usr/bin/env bash\nprintf "mock_verify=ok\\n"\n' > "$fixture/bin/verify" printf '#!/usr/bin/env bash\nprintf "mock_verify=ok\\n"\n' > "$fixture/bin/verify"
chmod +x "$fixture/bin/"* chmod +x "$fixture/bin/"*
PLATFORM_ROOT="$root" SYSTEMCTL_BIN="$fixture/bin/systemctl" CURL_BIN="$fixture/bin/curl" VERIFY_WEB_RELEASE_BIN="$fixture/bin/verify" RELEASE_HISTORY_LIMIT=3 "$SCRIPT_DIR/install-web-release.sh" new-release "$fixture/new.tar.gz" "$fixture/new-platform-api" > "$fixture/install.out" PLATFORM_ROOT="$root" SYSTEMCTL_BIN="$fixture/bin/systemctl" CURL_BIN="$fixture/bin/curl" VERIFY_WEB_RELEASE_BIN="$fixture/bin/verify" RELEASE_HISTORY_LIMIT=3 "$SCRIPT_DIR/install-web-release.sh" new-release "$fixture/new.tar.gz" "$fixture/new-platform-api" "$fixture/new-oneos-scope-sync" > "$fixture/install.out"
test "$(readlink -f "$root/current")" = "$(cd "$root/releases/new-release" && pwd -P)" test "$(readlink -f "$root/current")" = "$(cd "$root/releases/new-release" && pwd -P)"
grep -q '^PLATFORM_RELEASE=new-release$' "$root/env/platform.env" grep -q '^PLATFORM_RELEASE=new-release$' "$root/env/platform.env"
test "$(cat "$root/current/platform-api")" = 'new platform API' test "$(cat "$root/current/platform-api")" = 'new platform API'
test "$(cat "$root/current/oneos-scope-sync")" = 'new OneOS scope sync'
test "$(cat "$root/current/alert-evaluator")" = 'old evaluator' test "$(cat "$root/current/alert-evaluator")" = 'old evaluator'
test -f "$root/current/web/assets/new.js" test -f "$root/current/web/assets/new.js"
test -f "$root/current/web/assets/old.js" test -f "$root/current/web/assets/old.js"
test -f "$root/current/web/.compatibility-manifests/1.assets" test -f "$root/current/web/.compatibility-manifests/1.assets"
test -x "$root/current/deploy/verify-customer-demo.py" test -x "$root/current/deploy/verify-customer-demo.py"
test -f "$root/current/deploy/migrations/015_vehicle_source_policy_audit.sql" test -f "$root/current/deploy/migrations/016_business_scope_dimensions.sql"
test ! -e "$root/current/lingniu-vehicle-platform.service" test ! -e "$root/current/lingniu-vehicle-platform.service"
test "$(find "$root/releases" -mindepth 1 -maxdepth 1 -type d | wc -l | tr -d ' ')" = 3 test "$(find "$root/releases" -mindepth 1 -maxdepth 1 -type d | wc -l | tr -d ' ')" = 3
grep -q 'web_release_install=ok release=new-release previous=old-release' "$fixture/install.out" grep -q 'web_release_install=ok release=new-release previous=old-release' "$fixture/install.out"

View File

@@ -35,14 +35,14 @@ CREATE TABLE IF NOT EXISTS business_customer_vehicle_scope (
CREATE TABLE IF NOT EXISTS business_scope_rejection ( CREATE TABLE IF NOT EXISTS business_scope_rejection (
run_id CHAR(32) NOT NULL, run_id CHAR(32) NOT NULL,
row_number INT UNSIGNED NOT NULL, `row_number` INT UNSIGNED NOT NULL,
vehicle_id BIGINT NULL, vehicle_id BIGINT NULL,
vin VARCHAR(64) NOT NULL DEFAULT '', vin VARCHAR(64) NOT NULL DEFAULT '',
customer_id BIGINT NULL, customer_id BIGINT NULL,
contract_id BIGINT NULL, contract_id BIGINT NULL,
reason_code VARCHAR(64) NOT NULL, reason_code VARCHAR(64) NOT NULL,
created_at DATETIME(3) NOT NULL, created_at DATETIME(3) NOT NULL,
PRIMARY KEY (run_id, row_number), PRIMARY KEY (run_id, `row_number`),
INDEX idx_business_scope_rejection_reason (reason_code, created_at), INDEX idx_business_scope_rejection_reason (reason_code, created_at),
CONSTRAINT fk_business_scope_rejection_run CONSTRAINT fk_business_scope_rejection_run
FOREIGN KEY (run_id) REFERENCES business_scope_sync_run(run_id) FOREIGN KEY (run_id) REFERENCES business_scope_sync_run(run_id)

View File

@@ -0,0 +1,20 @@
ALTER TABLE business_customer_vehicle_scope
ADD COLUMN customer_name VARCHAR(255) NOT NULL DEFAULT '' AFTER customer_id;
ALTER TABLE business_customer_vehicle_scope
ADD COLUMN department_id VARCHAR(64) NOT NULL DEFAULT '' AFTER project_name;
ALTER TABLE business_customer_vehicle_scope
ADD COLUMN department_name VARCHAR(255) NOT NULL DEFAULT '' AFTER department_id;
ALTER TABLE business_customer_vehicle_scope
ADD COLUMN responsible_user_id VARCHAR(64) NOT NULL DEFAULT '' AFTER department_name;
ALTER TABLE business_customer_vehicle_scope
ADD COLUMN responsible_user_name VARCHAR(255) NOT NULL DEFAULT '' AFTER responsible_user_id;
CREATE INDEX idx_business_scope_department
ON business_customer_vehicle_scope(source_version, department_id, vin);
CREATE INDEX idx_business_scope_responsible
ON business_customer_vehicle_scope(source_version, responsible_user_id, vin);