shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(history): scope export tasks to owners

Browse Source
This commit is contained in:
lingniu
2026-07-16 17:00:33 +08:00
parent a2722e4afd
commit 38b056f5f1
18 changed files with 528 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

223
vehicle-data-platform/apps/api/internal/platform/service.go
View File

@@ -2612,7 +2612,15 @@ func rawMetricMetadata(key string) (string, string) {
return label, unit
}
func (s *Service) CreateHistoryExport(request HistoryExportRequest) (HistoryExportJob, error) {
func (s *Service) CreateHistoryExport(ctx context.Context, request HistoryExportRequest) (HistoryExportJob, error) {
principal, ok := PrincipalFromContext(ctx)
if !ok {
return HistoryExportJob{}, clientError{Code: "EXPORT_OWNER_REQUIRED", Message: "导出任务需要已认证账号"}
}
ownerID := historyExportOwnerID(principal)
if ownerID == "" {
return HistoryExportJob{}, clientError{Code: "EXPORT_OWNER_REQUIRED", Message: "当前账号缺少可持久化的导出身份"}
}
request.Keywords = normalizedKeywords(request.Keywords)
if len(request.Keywords) == 0 || len(request.Keywords) > 5 {
return HistoryExportJob{}, clientError{Code: "EXPORT_SCOPE_INVALID", Message: "导出任务需要 1 至 5 台车辆"}
@@ -2640,12 +2648,32 @@ func (s *Service) CreateHistoryExport(request HistoryExportRequest) (HistoryExpo
if len(request.Metrics) > 32 {
return HistoryExportJob{}, clientError{Code: "EXPORT_METRIC_LIMIT_EXCEEDED", Message: "单次导出最多支持 32 个指标"}
}
scopes, err := s.resolveHistoryExportScopes(ctx, request)
if err != nil {
return HistoryExportJob{}, err
}
identifier, err := randomExportID()
if err != nil {
return HistoryExportJob{}, err
}
now := time.Now().UTC().Format(time.RFC3339)
job := &HistoryExportJob{ID: identifier, Name: "历史数据_" + time.Now().Format("20060102_150405"), Status: "queued", Progress: 0, Format: request.Format, Category: request.Category, Keywords: append([]string(nil), request.Keywords...), CreatedAt: now, UpdatedAt: now, Evidence: "单并发流式任务；最多 1,000,000 行；完成文件原子发布"}
vehicleVINs := make([]string, 0, len(scopes))
for _, scope := range scopes {
vehicleVINs = append(vehicleVINs, scope.VIN)
}
creator := firstNonEmpty(strings.TrimSpace(principal.Username), strings.TrimSpace(principal.Name), "account")
job := &HistoryExportJob{
ID: identifier, Name: "历史数据_" + time.Now().Format("20060102_150405") + "_" + safeExportNamePart(creator),
Status: "queued", Progress: 0, Format: request.Format, Category: request.Category, Protocol: request.Protocol,
Keywords: append([]string(nil), request.Keywords...), Metrics: append([]string(nil), request.Metrics...),
VehicleVINs: vehicleVINs, VehicleScopes: append([]HistoryExportVehicleScope(nil), scopes...),
DateFrom: request.DateFrom, DateTo: request.DateTo,
OwnerID: ownerID, OwnerSubjectID: principal.SubjectID, OwnerName: principal.Name, OwnerUsername: principal.Username,
OwnerRole: principal.Role, OwnerUserType: principal.UserType, AuthProvider: principal.AuthProvider,
CustomerRef: principal.CustomerRef, TenantRef: principal.TenantRef,
CreatedAt: now, UpdatedAt: now,
Evidence: "账号归属与车辆/时间 Scope 已固化；单并发流式任务；最多 1,000,000 行；完成文件原子发布",
}
s.exportsMu.Lock()
s.exports[job.ID] = job
if err := s.persistHistoryExportsLocked(); err != nil {
@@ -2654,15 +2682,18 @@ func (s *Service) CreateHistoryExport(request HistoryExportRequest) (HistoryExpo
return HistoryExportJob{}, fmt.Errorf("persist export job: %w", err)
}
s.exportsMu.Unlock()
go s.runHistoryExport(job.ID, request)
go s.runHistoryExport(job.ID, request, scopes)
return copyHistoryExportJob(job), nil
}
func (s *Service) ListHistoryExports() []HistoryExportJob {
func (s *Service) ListHistoryExports(ctx context.Context) []HistoryExportJob {
principal, _ := PrincipalFromContext(ctx)
s.exportsMu.RLock()
result := make([]HistoryExportJob, 0, len(s.exports))
for _, job := range s.exports {
result = append(result, copyHistoryExportJob(job))
if canAccessHistoryExport(principal, job) {
result = append(result, copyHistoryExportJob(job))
}
}
s.exportsMu.RUnlock()
sort.Slice(result, func(i, j int) bool { return result[i].CreatedAt > result[j].CreatedAt })
@@ -2672,10 +2703,11 @@ func (s *Service) ListHistoryExports() []HistoryExportJob {
return result
}
func (s *Service) HistoryExportFile(id string) (string, string, error) {
func (s *Service) HistoryExportFile(ctx context.Context, id string) (string, string, error) {
principal, _ := PrincipalFromContext(ctx)
s.exportsMu.RLock()
job := s.exports[id]
if job == nil {
if job == nil || !canAccessHistoryExport(principal, job) {
s.exportsMu.RUnlock()
return "", "", clientError{Code: "EXPORT_NOT_FOUND", Message: "导出任务不存在"}
}
@@ -2685,15 +2717,154 @@ func (s *Service) HistoryExportFile(id string) (string, string, error) {
if copy.Status != "completed" || path == "" {
return "", "", clientError{Code: "EXPORT_NOT_READY", Message: "导出文件尚未生成"}
}
if err := s.ensureHistoryExportScopeActive(ctx, copy); err != nil {
return "", "", err
}
return path, copy.Name + ".csv", nil
}
func (s *Service) runHistoryExport(id string, request HistoryExportRequest) {
func historyExportOwnerID(principal Principal) string {
provider := strings.TrimSpace(principal.AuthProvider)
if provider == "" {
provider = "unknown"
}
if subject := strings.TrimSpace(principal.SubjectID); subject != "" {
return provider + ":subject:" + subject
}
if username := strings.TrimSpace(principal.Username); username != "" {
return provider + ":username:" + strings.ToLower(username)
}
if name := strings.TrimSpace(principal.Name); name != "" {
return provider + ":name:" + strings.ToLower(name)
}
return ""
}
func canAccessHistoryExport(principal Principal, job *HistoryExportJob) bool {
if job == nil {
return false
}
if principal.Role == "admin" || principal.UserType == "admin" {
return true
}
ownerID := historyExportOwnerID(principal)
return ownerID != "" && job.OwnerID != "" && ownerID == job.OwnerID
}
func safeExportNamePart(value string) string {
value = strings.TrimSpace(value)
if value == "" {
return "account"
}
var builder strings.Builder
for _, char := range value {
switch {
case char >= 'a' && char <= 'z', char >= 'A' && char <= 'Z', char >= '0' && char <= '9', char == '-', char == '_':
builder.WriteRune(char)
default:
builder.WriteRune('_')
}
if builder.Len() >= 32 {
break
}
}
result := strings.Trim(builder.String(), "_")
if result == "" {
return "account"
}
return result
}
func (s *Service) resolveHistoryExportScopes(ctx context.Context, request HistoryExportRequest) ([]HistoryExportVehicleScope, error) {
scopes := make([]HistoryExportVehicleScope, 0, len(request.Keywords))
seen := map[string]bool{}
for _, keyword := range request.Keywords {
vin, err := s.resolveVehicleVIN(ctx, keyword, request.Protocol)
if err != nil {
return nil, err
}
vin = strings.ToUpper(strings.TrimSpace(vin))
if vin == "" || seen[vin] {
continue
}
if err := authorizeVehicleVIN(ctx, vin); err != nil {
return nil, err
}
scoped, err := applyPrincipalHistoryTimeScope(ctx, vin, url.Values{
"dateFrom": {request.DateFrom},
"dateTo": {request.DateTo},
})
if err != nil {
return nil, err
}
if request.Category == "mileage" {
fromDay := strings.Split(scoped.Get("dateFrom"), "T")[0]
toDay := strings.Split(scoped.Get("dateTo"), "T")[0]
if err := authorizeVehicleDailyEvidenceDate(ctx, vin, fromDay); err != nil {
return nil, err
}
if err := authorizeVehicleDailyEvidenceDate(ctx, vin, toDay); err != nil {
return nil, err
}
}
seen[vin] = true
scopes = append(scopes, HistoryExportVehicleScope{VIN: vin, DateFrom: scoped.Get("dateFrom"), DateTo: scoped.Get("dateTo")})
}
if len(scopes) == 0 {
return nil, clientError{Code: "EXPORT_SCOPE_INVALID", Message: "导出任务未解析到可授权车辆"}
}
return scopes, nil
}
type historyExportAuthorizationStore interface {
HistoryExportScopeActive(context.Context, HistoryExportJob) (bool, error)
}
func (s *Service) ensureHistoryExportScopeActive(ctx context.Context, job HistoryExportJob) error {
if job.OwnerUserType != "customer" {
return nil
}
if job.OwnerSubjectID == "" {
return clientError{Code: "EXPORT_SCOPE_REVOKED", Message: "客户导出缺少可复核的账号标识，任务已停止"}
}
store, ok := s.store.(historyExportAuthorizationStore)
if !ok {
return clientError{Code: "EXPORT_SCOPE_REVOKED", Message: "当前数据存储无法复核客户授权，任务已停止"}
}
active, err := store.HistoryExportScopeActive(ctx, job)
if err != nil {
return err
}
if !active {
return clientError{Code: "EXPORT_SCOPE_REVOKED", Message: "账号状态或车辆授权已变化，导出任务/下载已停止"}
}
return nil
}
func (s *Service) historyExportJob(id string) (HistoryExportJob, bool) {
s.exportsMu.RLock()
defer s.exportsMu.RUnlock()
job := s.exports[id]
if job == nil {
return HistoryExportJob{}, false
}
return copyHistoryExportJob(job), true
}
func (s *Service) runHistoryExport(id string, request HistoryExportRequest, scopes []HistoryExportVehicleScope) {
s.exportSlots <- struct{}{}
defer func() { <-s.exportSlots }()
s.updateHistoryExport(id, func(job *HistoryExportJob) { job.Status = "running"; job.Progress = 1 })
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
defer cancel()
job, ok := s.historyExportJob(id)
if !ok {
return
}
if err := s.ensureHistoryExportScopeActive(ctx, job); err != nil {
s.failHistoryExport(id, err)
return
}
store, ok := s.store.(interface {
HistoryExportCount(context.Context, HistoryExportStoreQuery) (int64, error)
HistoryExportBatch(context.Context, HistoryExportStoreQuery, HistoryExportCursor, int) ([]HistoryDataRow, HistoryExportCursor, error)
@@ -2702,23 +2873,10 @@ func (s *Service) runHistoryExport(id string, request HistoryExportRequest) {
s.failHistoryExport(id, errors.New("当前数据存储不支持流式历史导出"))
return
}
resolvedVINs := make([]string, 0, len(request.Keywords))
seenVINs := map[string]bool{}
for _, keyword := range request.Keywords {
vin, err := s.resolveVehicleVIN(ctx, keyword, request.Protocol)
if err != nil {
s.failHistoryExport(id, err)
return
}
if vin = strings.TrimSpace(vin); vin != "" && !seenVINs[vin] {
seenVINs[vin] = true
resolvedVINs = append(resolvedVINs, vin)
}
}
queries := make([]HistoryExportStoreQuery, 0, len(resolvedVINs))
queries := make([]HistoryExportStoreQuery, 0, len(scopes))
var totalRows int64
for _, vin := range resolvedVINs {
query := HistoryExportStoreQuery{Category: request.Category, VIN: vin, Protocol: request.Protocol, DateFrom: request.DateFrom, DateTo: request.DateTo, Metrics: append([]string(nil), request.Metrics...)}
for _, scope := range scopes {
query := HistoryExportStoreQuery{Category: request.Category, VIN: scope.VIN, Protocol: request.Protocol, DateFrom: scope.DateFrom, DateTo: scope.DateTo, Metrics: append([]string(nil), request.Metrics...)}
count, err := store.HistoryExportCount(ctx, query)
if err != nil {
s.failHistoryExport(id, err)
@@ -2758,7 +2916,7 @@ func (s *Service) runHistoryExport(id string, request HistoryExportRequest) {
}
buffered := bufio.NewWriterSize(file, 1<<20)
writer := csv.NewWriter(buffered)
if err := writeHistoryExportMetadata(writer, request, columns); err != nil {
if err := writeHistoryExportMetadata(writer, request, job, columns); err != nil {
s.failHistoryExport(id, err)
return
}
@@ -2766,6 +2924,14 @@ func (s *Service) runHistoryExport(id string, request HistoryExportRequest) {
for _, query := range queries {
cursor := HistoryExportCursor{}
for {
currentJob, exists := s.historyExportJob(id)
if !exists {
return
}
if err := s.ensureHistoryExportScopeActive(ctx, currentJob); err != nil {
s.failHistoryExport(id, err)
return
}
rows, nextCursor, err := store.HistoryExportBatch(ctx, query, cursor, 5000)
if err != nil {
s.failHistoryExport(id, err)
@@ -2950,6 +3116,9 @@ func copyHistoryExportJob(job *HistoryExportJob) HistoryExportJob {
}
copy := *job
copy.Keywords = append([]string(nil), job.Keywords...)
copy.Metrics = append([]string(nil), job.Metrics...)
copy.VehicleVINs = append([]string(nil), job.VehicleVINs...)
copy.VehicleScopes = append([]HistoryExportVehicleScope(nil), job.VehicleScopes...)
return copy
}
@@ -2991,9 +3160,11 @@ func historyExportColumns(category string, requested []string) []HistoryMetricDe
return selected
}
func writeHistoryExportMetadata(writer *csv.Writer, request HistoryExportRequest, columns []HistoryMetricDefinition) error {
func writeHistoryExportMetadata(writer *csv.Writer, request HistoryExportRequest, job HistoryExportJob, columns []HistoryMetricDefinition) error {
rows := [][]string{
{"导出元数据", "查询开始", request.DateFrom, "查询结束", request.DateTo, "车辆", strings.Join(request.Keywords, "、"), "数据类型", request.Category, "协议", firstNonEmpty(request.Protocol, "全部")},
{"导出审计", "创建账号", firstNonEmpty(job.OwnerUsername, job.OwnerName), "显示名称", job.OwnerName, "角色", firstNonEmpty(job.OwnerUserType, job.OwnerRole), "客户", job.CustomerRef, "创建时间", job.CreatedAt, "生成时间", time.Now().UTC().Format(time.RFC3339)},
{"车辆 Scope", strings.Join(job.VehicleVINs, "、"), "授权窗口", job.DateFrom + " 至 " + job.DateTo},
{"指标与单位", strings.Join(exportMetricLabels(columns), "；")},
append([]string{"设备时间", "服务时间", "车牌", "VIN", "协议", "数据质量", "质量原因", "证据ID"}, exportMetricLabels(columns)...),
}