package diagnostics import ( "context" "encoding/json" "sort" "strings" "time" "data-terminal-cloud/internal/audit" "data-terminal-cloud/internal/idempotency" "data-terminal-cloud/internal/identity" "github.com/google/uuid" "github.com/jackc/pgx/v5" ) type PilotWeeklySnapshotInput struct { CandidateVersion string `json:"candidate_version"` BuildDigest string `json:"build_digest"` WeekStart string `json:"week_start"` Timezone string `json:"timezone"` Reason string `json:"reason"` } type PilotWeeklyContentReview struct { StoriesGenerated int64 `json:"stories_generated"` StoriesReviewTouched int64 `json:"stories_review_touched"` StoriesManuallyRevised int64 `json:"stories_manually_revised"` ManualRevisionRateAvailable bool `json:"manual_revision_rate_available"` ManualRevisionRateBasisPoints int64 `json:"manual_revision_rate_basis_points"` StoriesConfirmed int64 `json:"stories_confirmed"` StoriesRejected int64 `json:"stories_rejected"` StoryConfirmationRateAvailable bool `json:"story_confirmation_rate_available"` StoryConfirmationRateBasisPoints int64 `json:"story_confirmation_rate_basis_points"` MemoriesCreated int64 `json:"memories_created"` MemoryRevisions int64 `json:"memory_revisions"` CurrentMemoryIndexPending int64 `json:"current_memory_index_pending"` CurrentMemoryIndexFailed int64 `json:"current_memory_index_failed"` } type PilotWeeklyCountByType struct { Type string `json:"type"` Count int64 `json:"count"` } type PilotWeeklyCountByStatus struct { Status string `json:"status"` Count int64 `json:"count"` } type PilotWeeklyPrivacyState struct { RequestsCreatedDuringWeek int64 `json:"requests_created_during_week"` ByType []PilotWeeklyCountByType `json:"by_type"` ByStatus []PilotWeeklyCountByStatus `json:"by_status"` CurrentOpen int64 `json:"current_open"` CurrentPartialFailed int64 `json:"current_partial_failed"` } type PilotWeeklyFirmwareVersion struct { Model string `json:"model"` FirmwareVersion string `json:"firmware_version"` DeviceCount int64 `json:"device_count"` } type PilotWeeklyConfigState struct { EligibleDevices int64 `json:"eligible_devices"` Aligned int64 `json:"aligned"` Drifted int64 `json:"drifted"` Unreported int64 `json:"unreported"` ActiveRollouts int64 `json:"active_rollouts"` PausedRollouts int64 `json:"paused_rollouts"` } type PilotWeeklyExternalEvidence struct { Code string `json:"code"` Status string `json:"status"` Reason string `json:"reason"` } type PilotWeeklySnapshot struct { ID string `json:"id"` GeneratedAt time.Time `json:"generated_at"` TenantID string `json:"tenant_id"` CandidateVersion string `json:"candidate_version"` BuildDigest string `json:"build_digest"` WeekStart string `json:"week_start"` Timezone string `json:"timezone"` WindowStart time.Time `json:"window_start"` WindowEnd time.Time `json:"window_end"` IsComplete bool `json:"is_complete"` RedactionPolicy string `json:"redaction_policy"` ContentReview PilotWeeklyContentReview `json:"content_review"` Privacy PilotWeeklyPrivacyState `json:"privacy"` FirmwareVersions []PilotWeeklyFirmwareVersion `json:"firmware_versions"` Config PilotWeeklyConfigState `json:"config"` ExternalEvidence []PilotWeeklyExternalEvidence `json:"external_evidence"` Disclaimer string `json:"disclaimer"` } var pilotPrivacyTypes = []string{"ACCESS", "CORRECTION", "CONSENT_WITHDRAWAL", "DELETE"} var pilotPrivacyStatuses = []string{"SUBMITTED", "VERIFYING", "EXECUTING", "COMPLETED", "PARTIAL_FAILED"} func (s *Service) ExportPilotWeeklySnapshot(ctx context.Context, principal identity.Principal, tenantPublicID, key, requestID string, input PilotWeeklySnapshotInput) (PilotWeeklySnapshot, bool, error) { normalizePilotWeeklySnapshotInput(&input) _, windowStart, windowEnd, err := validatePilotWeeklySnapshotInput(input, time.Now().UTC()) if err != nil { return PilotWeeklySnapshot{}, false, ErrInvalid } a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_operator", "ops_admin") if err != nil { return PilotWeeklySnapshot{}, false, err } requestBody, _ := json.Marshal(input) result, err := s.idem.Execute(ctx, idempotency.Request{ TenantID: &a.tenantID, ActorKey: "ops:" + a.userPublicID, Scope: "POST /api/v1/ops/diagnostics/pilot-weekly-snapshots", Key: key, RequestHash: idempotency.Hash(requestBody), ExpiresAfter: 24 * time.Hour, }, func(operationContext context.Context, tx pgx.Tx) (idempotency.Result, error) { snapshot, buildErr := buildPilotWeeklySnapshot(operationContext, tx, tenantPublicID, a.tenantID, input, windowStart, windowEnd) if buildErr != nil { return idempotency.Result{}, buildErr } if auditErr := s.audit.Record(operationContext, tx, audit.Event{ TenantID: &a.tenantID, ActorType: "user", ActorPublicID: &a.userPublicID, Action: "ops.diagnostics.pilot_weekly_snapshot.export", ResourceType: "pilot_weekly_snapshot", ResourceID: &snapshot.ID, RequestID: requestID, Purpose: input.Reason, Result: "success", Detail: map[string]any{"candidate_version": input.CandidateVersion, "build_digest": input.BuildDigest, "week_start": input.WeekStart, "timezone": input.Timezone, "window_start": windowStart, "window_end": windowEnd, "is_complete": snapshot.IsComplete, "redaction_policy": snapshot.RedactionPolicy}, }); auditErr != nil { return idempotency.Result{}, auditErr } encoded, _ := json.Marshal(snapshot) return idempotency.Result{StatusCode: 201, Body: encoded, ResourceID: &snapshot.ID}, nil }) if err != nil { return PilotWeeklySnapshot{}, false, err } var snapshot PilotWeeklySnapshot if err = json.Unmarshal(result.Body, &snapshot); err != nil { return PilotWeeklySnapshot{}, false, err } return snapshot, !result.Replayed, nil } func normalizePilotWeeklySnapshotInput(input *PilotWeeklySnapshotInput) { input.CandidateVersion = strings.TrimSpace(input.CandidateVersion) input.BuildDigest = strings.ToLower(strings.TrimSpace(input.BuildDigest)) input.WeekStart = strings.TrimSpace(input.WeekStart) input.Timezone = strings.TrimSpace(input.Timezone) input.Reason = strings.TrimSpace(input.Reason) } func validatePilotWeeklySnapshotInput(input PilotWeeklySnapshotInput, now time.Time) (*time.Location, time.Time, time.Time, error) { if !acceptanceCandidatePattern.MatchString(input.CandidateVersion) || !acceptanceDigestPattern.MatchString(input.BuildDigest) || input.WeekStart == "" || input.Timezone == "" || len([]rune(input.Timezone)) > 80 || input.Reason == "" || len([]rune(input.Reason)) > 255 { return nil, time.Time{}, time.Time{}, ErrInvalid } location, err := time.LoadLocation(input.Timezone) if err != nil { return nil, time.Time{}, time.Time{}, ErrInvalid } weekStart, err := time.ParseInLocation("2006-01-02", input.WeekStart, location) if err != nil || weekStart.Format("2006-01-02") != input.WeekStart || weekStart.Weekday() != time.Monday { return nil, time.Time{}, time.Time{}, ErrInvalid } today := now.In(location) todayStart := time.Date(today.Year(), today.Month(), today.Day(), 0, 0, 0, 0, location) if weekStart.After(todayStart) { return nil, time.Time{}, time.Time{}, ErrInvalid } return location, weekStart.UTC(), weekStart.AddDate(0, 0, 7).UTC(), nil } func buildPilotWeeklySnapshot(ctx context.Context, tx pgx.Tx, tenantPublicID, tenantInternalID string, input PilotWeeklySnapshotInput, windowStart, windowEnd time.Time) (PilotWeeklySnapshot, error) { generatedAt := time.Now().UTC() snapshot := PilotWeeklySnapshot{ ID: uuid.NewString(), GeneratedAt: generatedAt, TenantID: tenantPublicID, CandidateVersion: input.CandidateVersion, BuildDigest: input.BuildDigest, WeekStart: input.WeekStart, Timezone: input.Timezone, WindowStart: windowStart, WindowEnd: windowEnd, IsComplete: !generatedAt.Before(windowEnd), RedactionPolicy: "ops-pilot-weekly-minimal-v1", Privacy: PilotWeeklyPrivacyState{ByType: []PilotWeeklyCountByType{}, ByStatus: []PilotWeeklyCountByStatus{}}, FirmwareVersions: []PilotWeeklyFirmwareVersion{}, ExternalEvidence: []PilotWeeklyExternalEvidence{{Code: "complaints", Status: "REQUIRED", Reason: "当前平台没有权威投诉台账;归档时必须附客服或CRM的脱敏周度投诉证据。"}}, Disclaimer: "本周报只汇总平台内可证明的周度活动与生成时刻状态;投诉、真实用户体验、供应商账单和业务结论必须附外部证据,不得由本周报推断。", } if err := tx.QueryRow(ctx, `WITH review_actions AS ( SELECT resource_id,action FROM audit_event WHERE tenant_id=$1::uuid AND occurred_at >= $2 AND occurred_at < $3 AND result='success' AND action IN ('story.revise','sharing.story.edit','story.confirm','story.reject')) SELECT (SELECT count(*) FROM story_draft WHERE tenant_id=$1::uuid AND created_at >= $2 AND created_at < $3), count(DISTINCT resource_id) FILTER(WHERE resource_id IS NOT NULL), count(DISTINCT resource_id) FILTER(WHERE resource_id IS NOT NULL AND action IN ('story.revise','sharing.story.edit')), count(*) FILTER(WHERE action='story.confirm'), count(*) FILTER(WHERE action='story.reject') FROM review_actions`, tenantInternalID, windowStart, windowEnd).Scan( &snapshot.ContentReview.StoriesGenerated, &snapshot.ContentReview.StoriesReviewTouched, &snapshot.ContentReview.StoriesManuallyRevised, &snapshot.ContentReview.StoriesConfirmed, &snapshot.ContentReview.StoriesRejected, ); err != nil { return PilotWeeklySnapshot{}, err } if snapshot.ContentReview.StoriesReviewTouched > 0 { snapshot.ContentReview.ManualRevisionRateAvailable = true snapshot.ContentReview.ManualRevisionRateBasisPoints = rateBasisPoints(snapshot.ContentReview.StoriesManuallyRevised, snapshot.ContentReview.StoriesReviewTouched) } decisions := snapshot.ContentReview.StoriesConfirmed + snapshot.ContentReview.StoriesRejected if decisions > 0 { snapshot.ContentReview.StoryConfirmationRateAvailable = true snapshot.ContentReview.StoryConfirmationRateBasisPoints = rateBasisPoints(snapshot.ContentReview.StoriesConfirmed, decisions) } if err := tx.QueryRow(ctx, `SELECT count(*) FILTER(WHERE mv.version_no=1 AND mv.confirmed_at >= $2 AND mv.confirmed_at < $3), count(*) FILTER(WHERE mv.version_no>1 AND mv.confirmed_at >= $2 AND mv.confirmed_at < $3), count(DISTINCT mie.id) FILTER(WHERE mie.status='PENDING' AND cm.status NOT IN ('PENDING_DELETE','DELETED')), count(DISTINCT mie.id) FILTER(WHERE mie.status='FAILED' AND cm.status NOT IN ('PENDING_DELETE','DELETED')) FROM curated_memory cm LEFT JOIN memory_version mv ON mv.memory_id=cm.id LEFT JOIN memory_index_entry mie ON mie.memory_version_id=cm.current_version_id AND mie.index_type='pgvector' WHERE cm.tenant_id=$1::uuid`, tenantInternalID, windowStart, windowEnd).Scan( &snapshot.ContentReview.MemoriesCreated, &snapshot.ContentReview.MemoryRevisions, &snapshot.ContentReview.CurrentMemoryIndexPending, &snapshot.ContentReview.CurrentMemoryIndexFailed, ); err != nil { return PilotWeeklySnapshot{}, err } privacyTypeCounts := map[string]int64{} privacyStatusCounts := map[string]int64{} rows, err := tx.Query(ctx, `SELECT request_type,status,count(*) FROM privacy_request WHERE tenant_id=$1::uuid AND created_at >= $2 AND created_at < $3 GROUP BY request_type,status`, tenantInternalID, windowStart, windowEnd) if err != nil { return PilotWeeklySnapshot{}, err } for rows.Next() { var requestType, status string var count int64 if err = rows.Scan(&requestType, &status, &count); err != nil { rows.Close() return PilotWeeklySnapshot{}, err } privacyTypeCounts[requestType] += count privacyStatusCounts[status] += count snapshot.Privacy.RequestsCreatedDuringWeek += count } if err = rows.Err(); err != nil { rows.Close() return PilotWeeklySnapshot{}, err } rows.Close() for _, requestType := range pilotPrivacyTypes { snapshot.Privacy.ByType = append(snapshot.Privacy.ByType, PilotWeeklyCountByType{Type: requestType, Count: privacyTypeCounts[requestType]}) } for _, status := range pilotPrivacyStatuses { snapshot.Privacy.ByStatus = append(snapshot.Privacy.ByStatus, PilotWeeklyCountByStatus{Status: status, Count: privacyStatusCounts[status]}) } if err = tx.QueryRow(ctx, `SELECT count(*) FILTER(WHERE status<>'COMPLETED'),count(*) FILTER(WHERE status='PARTIAL_FAILED') FROM privacy_request WHERE tenant_id=$1::uuid`, tenantInternalID).Scan(&snapshot.Privacy.CurrentOpen, &snapshot.Privacy.CurrentPartialFailed); err != nil { return PilotWeeklySnapshot{}, err } rows, err = tx.Query(ctx, `SELECT d.model,COALESCE(dt.firmware_version,'UNREPORTED'),count(*) FROM device_binding db JOIN device d ON d.id=db.device_id AND d.tenant_id=db.tenant_id LEFT JOIN device_twin dt ON dt.device_id=d.id AND dt.tenant_id=d.tenant_id WHERE db.tenant_id=$1::uuid AND db.unbound_at IS NULL AND d.status='registered' GROUP BY d.model,COALESCE(dt.firmware_version,'UNREPORTED') ORDER BY d.model,COALESCE(dt.firmware_version,'UNREPORTED')`, tenantInternalID) if err != nil { return PilotWeeklySnapshot{}, err } for rows.Next() { var item PilotWeeklyFirmwareVersion if err = rows.Scan(&item.Model, &item.FirmwareVersion, &item.DeviceCount); err != nil { rows.Close() return PilotWeeklySnapshot{}, err } snapshot.FirmwareVersions = append(snapshot.FirmwareVersions, item) } if err = rows.Err(); err != nil { rows.Close() return PilotWeeklySnapshot{}, err } rows.Close() if err = tx.QueryRow(ctx, `WITH eligible AS ( SELECT dt.desired_revision,dt.reported_revision FROM device_binding db JOIN device d ON d.id=db.device_id AND d.tenant_id=db.tenant_id LEFT JOIN device_twin dt ON dt.device_id=d.id AND dt.tenant_id=d.tenant_id WHERE db.tenant_id=$1::uuid AND db.unbound_at IS NULL AND d.status='registered') SELECT count(*),count(*) FILTER(WHERE reported_revision IS NOT NULL AND reported_revision=desired_revision), count(*) FILTER(WHERE reported_revision IS NOT NULL AND reported_revision<>desired_revision), count(*) FILTER(WHERE reported_revision IS NULL), (SELECT count(*) FROM runtime_config_rollout WHERE tenant_id=$1::uuid AND status='ACTIVE'), (SELECT count(*) FROM runtime_config_rollout WHERE tenant_id=$1::uuid AND status='PAUSED') FROM eligible`, tenantInternalID).Scan( &snapshot.Config.EligibleDevices, &snapshot.Config.Aligned, &snapshot.Config.Drifted, &snapshot.Config.Unreported, &snapshot.Config.ActiveRollouts, &snapshot.Config.PausedRollouts, ); err != nil { return PilotWeeklySnapshot{}, err } sort.Slice(snapshot.FirmwareVersions, func(i, j int) bool { if snapshot.FirmwareVersions[i].Model == snapshot.FirmwareVersions[j].Model { return snapshot.FirmwareVersions[i].FirmwareVersion < snapshot.FirmwareVersions[j].FirmwareVersion } return snapshot.FirmwareVersions[i].Model < snapshot.FirmwareVersions[j].Model }) return snapshot, nil } func rateBasisPoints(numerator, denominator int64) int64 { if numerator <= 0 || denominator <= 0 { return 0 } return (numerator*10000 + denominator/2) / denominator }