package v1beta1 import ( "context" "encoding/json" "fmt" "regexp" "strconv" "time" "github.com/AlekSi/pointer" "github.com/elliotchance/pie/v2" console_api "github.com/tidbcloud/dataflow-service/app/controllers/console" "github.com/tidbcloud/dataflow-service/app/models/backup" "github.com/tidbcloud/dataflow-service/app/models/domain" "github.com/tidbcloud/dataflow-service/app/repo" backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup" "github.com/tidbcloud/dataflow-service/app/repo/entity" "github.com/tidbcloud/dataflow-service/app/utils" pb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/adminapi/v1beta1" dataflowpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow" "github.com/tidbcloud/dataflow-service/internal/constants" "github.com/tidbcloud/dataflow-service/internal/db" model "github.com/tidbcloud/dataflow-service/internal/db/entity" "github.com/tidbcloud/dataflow-service/internal/model/util" "github.com/tidbcloud/dataflow-service/internal/openapi/service" "github.com/tidbcloud/dataflow-service/pkg/account" "github.com/tidbcloud/dataflow-service/pkg/billing" bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper" "github.com/tidbcloud/dataflow-service/pkg/gcs" "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli" "github.com/tidbcloud/dataflow-service/pkg/log" "github.com/tidbcloud/pkgv2/bizerr" mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1" "github.com/tidbcloud/tiinf/sdk/auditlog" "go.uber.org/zap" "google.golang.org/protobuf/types/known/emptypb" "google.golang.org/protobuf/types/known/timestamppb" "gorm.io/gorm" ) type BackupController interface { pb.BackupServiceServer } type BackupControllerImpl struct { pb.UnimplementedBackupServiceServer trans db.TransProvider backupConsoleAPIV2 console_api.BackupControllerV2 backupConsoleAPI console_api.BackupController backupExportRepo backupRepo.BackupExportRepo backupExportModel backup.BackupExportModel backupCreateRouter backup.BackupCreateRouter backupRepo backupRepo.BackupRepo snapshotBackupModel backup.SnapshotBackupModel backupSettingModel backup.BackupSettingModel pitBackupModel backup.PitBackupModel bucketRepo repo.BucketRepo coreCli grpcli.CoreGrpcClient mgmtCli grpcli.MgmtSvcGrpcClient billingHandler billing.BillingHandler } var snapshotScheduleTimeRegexp = regexp.MustCompile(`^([01][0-9]|2[0-3]):[0-5][0-9]$`) const backupLookupPageSize = 20 func NewBackupController( trans db.TransProvider, backupConsoleAPIV2 console_api.BackupControllerV2, backupConsoleAPI console_api.BackupController, backupExportRepo backupRepo.BackupExportRepo, backupExportModel backup.BackupExportModel, backupCreateRouter backup.BackupCreateRouter, backupRepo backupRepo.BackupRepo, snapshotBackupModel backup.SnapshotBackupModel, backupSettingModel backup.BackupSettingModel, pitBackupModel backup.PitBackupModel, bucketRepo repo.BucketRepo, coreCli grpcli.CoreGrpcClient, mgmtCli grpcli.MgmtSvcGrpcClient, billingHandler billing.BillingHandler, ) BackupController { return &BackupControllerImpl{ trans: trans, backupConsoleAPIV2: backupConsoleAPIV2, backupConsoleAPI: backupConsoleAPI, backupExportRepo: backupExportRepo, backupExportModel: backupExportModel, backupCreateRouter: backupCreateRouter, backupRepo: backupRepo, snapshotBackupModel: snapshotBackupModel, backupSettingModel: backupSettingModel, pitBackupModel: pitBackupModel, bucketRepo: bucketRepo, coreCli: coreCli, mgmtCli: mgmtCli, billingHandler: billingHandler, } } // validateListBackupsRequest validates the list backups request func (s *BackupControllerImpl) validateListBackupsRequest( ctx context.Context, req *pb.ListBackupsRequest, ) (uint64, uint64, error) { logger := log.GetLogger(ctx) projectID, err := strconv.ParseUint(req.ProjectId, 10, 64) if err != nil { logger.Warn("invalid project id", zap.Error(err), zap.Any("request", req)) return 0, 0, bizerr_wrapper.NewInvalidInputError("invalid project id") } err = s.checkProjectPermission(ctx, projectID) if err != nil { logger.Warn("failed to check project permission", zap.Error(err), zap.Any("request", req)) return 0, 0, err } clusterID := uint64(0) if req.ClusterId != "" { clusterID, err = strconv.ParseUint(req.ClusterId, 10, 64) if err != nil { logger.Warn("invalid cluster id", zap.Error(err), zap.Any("request", req)) return 0, 0, bizerr_wrapper.NewInvalidInputError("invalid cluster id") } } err = s.checkClusterPermission(ctx, projectID, clusterID) if err != nil { logger.Warn("failed to check cluster permission", zap.Error(err), zap.Any("request", req)) return 0, 0, err } return projectID, clusterID, nil } // convertBackupToProto converts domain backup to protobuf backup func (s *BackupControllerImpl) convertBackupToProto(ctx context.Context, item domain.ListBackupRespItem) *pb.Backup { backup := &pb.Backup{ Id: strconv.FormatUint(item.ID, 10), ClusterId: strconv.FormatUint(item.ClusterID, 10), Name: item.Name, Description: item.Description, CreateTime: timestamppb.New(item.CreatedTime), SizeBytes: item.Size, State: s.convertDomainBackupStatus2pbBackupState(ctx, item.Status), Type: s.convertDomainBackupType2bBackupType(ctx, item.Type), } if item.ExpirationTime != nil { backup.ExpirationTime = timestamppb.New(*item.ExpirationTime) } if item.BackupTs != nil { backup.BackupTso = util.TransToTSO(*item.BackupTs) backupTs := timestamppb.New(time.Unix(*item.BackupTs, 0)) if err := backupTs.CheckValid(); err != nil { log.GetLogger(ctx).Warn("failed to parse backup's BackupTs", zap.Uint64("clusterID", item.ClusterID), zap.Error(err)) } else { backup.BackupTime = backupTs } } backup.Regions = pie.Map(item.Regions, func(region domain.RegionInfo) *pb.RegionInfo { return &pb.RegionInfo{ Name: region.Name, DisplayName: region.DisplayName, } }) return backup } func (s *BackupControllerImpl) ListBackups( ctx context.Context, req *pb.ListBackupsRequest, ) (*pb.ListBackupsResponse, error) { ctx = log.WithTraceID(ctx) logger := log.GetLogger(ctx) projectID, clusterID, err := s.validateListBackupsRequest(ctx, req) if err != nil { return nil, err } pager := utils.NewAipPager() if req.PageSize == 0 { req.PageSize = int32(pager.DefaultPageSize) } err = pager.Normalize(req) if err != nil { logger.Warn("failed to normalize pager", zap.Error(err), zap.Any("request", req)) return nil, err } listBackupReq := &domain.ListBackupReq{ Filter: &domain.ListBackupFilter{ ProjectID: projectID, }, Pager: domain.NewPager( domain.PageSizeOp(int64(pager.PageSize())), domain.PageNumberOp(int64(pager.Offset()/pager.PageSize()+1)), ), } if clusterID != 0 { listBackupReq.Filter.ClusterIDs = []uint64{clusterID} } listBackupResp, err := s.snapshotBackupModel.List(ctx, listBackupReq) if err != nil { logger.Error("failed to list backups", zap.Error(err), zap.Any("request", listBackupReq)) return nil, err } if listBackupResp == nil { return &pb.ListBackupsResponse{}, nil } backups := make([]*pb.Backup, 0, len(listBackupResp.Items)) for _, item := range listBackupResp.Items { scopeType, tableFilter := s.parseBackupScopeFromTableFilter(ctx, item.TableFilter) backups = append(backups, s.convertDomainBackupToProtoWithScope(ctx, item, scopeType, tableFilter)) } return &pb.ListBackupsResponse{ Backups: backups, TotalSize: int32(listBackupResp.Total), NextPageToken: pager.NextPageToken(uint32(listBackupResp.Total)), }, nil } func (s *BackupControllerImpl) checkProjectPermission(ctx context.Context, projectId uint64) error { logger := log.GetLogger(ctx) user := service.NewUserContext(ctx) if user == nil { logger.Warn("user not found") return bizerr_wrapper.NewForbiddenErr("user not found") } err := user.CheckProjectID(model.ID(projectId)) if err != nil { logger.Warn("user does not have permission to access this project", zap.Error(err), zap.Any("projectId", projectId)) return bizerr_wrapper.NewForbiddenErr("user does not have permission to access this project") } return nil } func (s *BackupControllerImpl) convertDomainBackupStatus2pbBackupState( ctx context.Context, status domain.BackupStatus, ) pb.Backup_State { switch status { case domain.BackupStatusRunning: return pb.Backup_RUNNING case domain.BackupStatusSuccess: return pb.Backup_SUCCEEDED case domain.BackupStatusFailed: return pb.Backup_FAILED case domain.BackupStatusSkipped: return pb.Backup_SKIPPED case domain.BackupStatusPending: return pb.Backup_PENDING case domain.BackupStatusTerminated: return pb.Backup_TERMINATED default: log.GetLogger(ctx).Warn("unknown backup status", zap.Any("status", status)) return pb.Backup_STATE_UNSPECIFIED } } func (s *BackupControllerImpl) convertDomainBackupType2bBackupType( ctx context.Context, tp domain.BackupType, ) pb.BackupType { switch tp { case domain.BackupType_Auto: return pb.BackupType_AUTO case domain.BackupType_Manual: return pb.BackupType_MANUAL default: log.GetLogger(ctx).Warn("unknown backup type", zap.Any("type", tp)) return pb.BackupType_BACKUP_TYPE_UNSPECIFIED } } func (s *BackupControllerImpl) checkClusterPermission(ctx context.Context, projectID, clusterID uint64) error { logger := log.GetLogger(ctx) user := service.NewUserContext(ctx) if user == nil { logger.Warn("user not found") return bizerr_wrapper.NewForbiddenErr("user not found") } // clusterID is 0 means all clusters if clusterID == 0 { return nil } cluster, err := s.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, clusterID) if err != nil { return bizerr.New(bizerr.ErrCodeDB, "get cluster %d: %v", clusterID, err) } if cluster.GetProjectId() != projectID || cluster.GetOrganizationId() != uint64(user.OrgID()) { logger.Warn("cluster and request are not matched", zap.Any("cluster_id", clusterID), zap.Any("project_id", projectID), zap.Any("tenant_id", user.OrgID()), ) return bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster is not found") } return nil } func (s *BackupControllerImpl) GetBackupExportProfile( ctx context.Context, req *pb.GetBackupExportProfileRequest, ) (*pb.BackupExportProfile, error) { ctx = log.WithTraceID(ctx) logger := log.GetLogger(ctx) backupID, err := strconv.ParseUint(req.BackupId, 10, 64) if err != nil { logger.Warn("invalid backup id", zap.Error(err), zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("invalid backup id") } backup, err := s.checkBackupResourcePermission(ctx, backupID) if err != nil { logger.Warn("failed to check backup resource permission", zap.Error(err), zap.Any("request", req)) return nil, err } cluster, err := s.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, uint64(backup.ClusterID)) if err != nil { logger.Warn("failed to get cluster", zap.Error(err), zap.Any("request", req), zap.Uint64("cluster id", uint64(backup.ClusterID))) return nil, bizerr.New(bizerr.ErrCodeDB, "failed to get cluster %d: %v", uint64(backup.ClusterID), err) } if cluster.GetCloudProvider() != constants.CloudProviderGCP && cluster.GetCloudProvider() != constants.CloudProviderAWS { logger.Warn(fmt.Sprintf("unsupported cloud provider: %s", cluster.GetCloudProvider()), zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("unsupported cloud provider: %s", cluster.GetCloudProvider()) } profile, err := s.backupExportModel.GetBackupExportProfile(ctx, &domain.GetBackupExportProfileReq{ BackupID: backupID, }) if err != nil { logger.Warn("failed to get backup export profile", zap.Error(err), zap.Any("request", req)) return nil, err } resp := &pb.BackupExportProfile{} // double check here, only support gcp now if profile.GCPServiceAccount != "" { resp.Gcs = &pb.BackupExportProfile_GCSProfile{ ServiceAccount: profile.GCPServiceAccount, } } else if profile.S3RoleArn != "" { resp.S3 = &pb.BackupExportProfile_S3Profile{ RoleArn: profile.S3RoleArn, } } else { logger.Warn("no profile found", zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("no profile found") } return resp, nil } func (s *BackupControllerImpl) checkBackupResourcePermission( ctx context.Context, backID uint64, ) (*entity.Backup, bizerr.BizError) { backup, err := s.backupRepo.Load(ctx, backID) if err != nil { return nil, err } user := service.NewUserContext(ctx) if user == nil { return nil, bizerr_wrapper.NewForbiddenErr("user not found") } if backup.TenantID != entity.ID(user.OrgID()) { return nil, bizerr_wrapper.NewForbiddenErr("you do not have permission to access this resource") } return backup, nil } // validateBackupExportRequest validates the backup export request func (s *BackupControllerImpl) validateBackupExportRequest( ctx context.Context, req *pb.CreateBackupExportRequest, ) (*entity.Backup, error) { backupID, err := strconv.ParseUint(req.BackupId, 10, 64) if err != nil { return nil, bizerr_wrapper.NewInvalidInputError("invalid backup id") } backup, err := s.backupRepo.Load(ctx, backupID) if err != nil { return nil, err } user := service.NewUserContext(ctx) if user == nil { return nil, bizerr_wrapper.NewForbiddenErr("user not found") } if backup.TenantID != entity.ID(user.OrgID()) { return nil, bizerr_wrapper.NewForbiddenErr("you do not have permission to access this resource") } if backup.Status != entity.BackupStatusSuccess { return nil, bizerr_wrapper.NewInvalidInputError("backup is not completed") } if *backup.CleanedUp { return nil, bizerr_wrapper.NewInvalidInputError("backup is already cleaned up") } return backup, nil } // createBackupExportConfig creates the backup export configuration // //nolint:funlen func (s *BackupControllerImpl) createBackupExportConfig( ctx context.Context, req *pb.CreateBackupExportRequest, backup *entity.Backup, ) (*dataflowpb.CreateBackupExportReq, error) { logger := log.GetLogger(ctx) bucket, bizErr := s.bucketRepo.Load(ctx, backup.BucketID.Uint64()) if bizErr != nil { logger.Warn("failed to load bucket", zap.Error(bizErr), zap.Any("request", req)) return nil, bizErr } backupID, err := strconv.ParseUint(req.BackupId, 10, 64) if err != nil { logger.Warn("invalid backup id", zap.Error(err), zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("invalid backup id") } createBackupExportReq := &dataflowpb.CreateBackupExportReq{ BackupId: backupID, } if req.BackupExport == nil { logger.Warn("backup export is nil", zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("backup export is nil") } if req.BackupExport.Destination == nil { logger.Warn("invalid backup export destination", zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("invalid backup export destination") } backupExport := &dataflowpb.BackupExport{} switch bucket.Provider { case constants.CloudProviderGCP: destination := req.BackupExport.Destination.GetGcs() if destination.TargetUri == "" { logger.Warn("gcs target uri is empty", zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("gcs target uri is empty") } backupExport.Storage = &dataflowpb.BackupExport_Gcs{ Gcs: &dataflowpb.BackupExport_GCS{ TargetUri: destination.TargetUri, }, } case constants.CloudProviderAWS: destination := req.BackupExport.Destination.GetS3() if destination.TargetUri == "" || destination.Region == "" { logger.Warn("s3 target parameters empty", zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("s3 target parameters empty") } storage := &dataflowpb.BackupExport_S3_{ S3: &dataflowpb.BackupExport_S3{ TargetUri: destination.TargetUri, Region: destination.Region, }, } if req.DryRun { storage.S3.DryRun = true } backupExport.Storage = storage default: return nil, bizerr_wrapper.NewInvalidInputError("unsupported cloud provider: %s", bucket.Provider) } createBackupExportReq.BackupExport = backupExport return createBackupExportReq, nil } func (s *BackupControllerImpl) checkDestinationUrl( ctx context.Context, backup *entity.Backup, req *pb.CreateBackupExportRequest, ) error { logger := log.GetLogger(ctx) if url := req.BackupExport.Destination.GetGcs(); url != nil { _, err := gcs.NewGcsURL(url.TargetUri) if err != nil { logger.Warn("invalid gcs target uri", zap.Error(err), zap.Any("request", req)) return bizerr_wrapper.NewInvalidInputError("invalid gcs target uri: %v", err) } } return nil } func (s *BackupControllerImpl) CreateBackupExport( ctx context.Context, req *pb.CreateBackupExportRequest, ) (*pb.BackupExport, error) { ctx = log.WithTraceID(ctx) logger := log.GetLogger(ctx) backup, err := s.validateBackupExportRequest(ctx, req) if err != nil { logger.Warn("invalid create backup export request", zap.Error(err), zap.Any("request", req)) return nil, err } createBackupExportReq, err := s.createBackupExportConfig(ctx, req, backup) if err != nil { return nil, err } err = s.checkDestinationUrl(ctx, backup, req) if err != nil { return nil, err } user := account.GetUserFromContext(ctx) var exportId uint64 err = s.trans.Transaction(ctx, func(ctx context.Context) error { id, err := s.backupExportModel.CreateBackupExport(ctx, createBackupExportReq, user) if err != nil { logger.Error("failed to create backup export", zap.Error(err), zap.Any("request", createBackupExportReq)) return err } exportId = id return nil }) if err != nil { return nil, err } resp, err := s.GetBackupExport(ctx, &pb.GetBackupExportRequest{ BackupId: req.BackupId, ExportId: strconv.FormatUint(exportId, 10), }) if err != nil { logger.Error("failed to get backup export", zap.Error(err), zap.Any("request", req)) return nil, err } return resp, nil } func (s *BackupControllerImpl) ListBackupExports( ctx context.Context, req *pb.ListBackupExportsRequest, ) (*pb.ListBackupExportsResponse, error) { ctx = log.WithTraceID(ctx) logger := log.GetLogger(ctx) backupID, err := strconv.ParseUint(req.BackupId, 10, 64) if err != nil { logger.Warn("invalid backup id", zap.Error(err), zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("invalid backup id") } if _, err := s.checkBackupResourcePermission(ctx, backupID); err != nil { logger.Warn("failed to check backup resource permission", zap.Error(err), zap.Any("request", req)) return nil, err } pager := utils.NewAipPager() if req.PageSize == 0 { req.PageSize = int32(pager.DefaultPageSize) } err = pager.Normalize(req) if err != nil { logger.Warn("failed to normalize pager", zap.Error(err), zap.Any("request", req)) return nil, err } filter := &backupRepo.BackupExportFilter{ BackupID: pointer.To(backupID), } // Get total count count, err := s.backupExportRepo.CountByFilter(ctx, filter) if err != nil { logger.Error("failed to count backup exports", zap.Error(err), zap.Any("filter", filter)) return nil, err } // Get list of backup exports backupExports, err := s.backupExportRepo.ListByFilter( ctx, filter, func(tx *gorm.DB) *gorm.DB { return tx.Offset(int(pager.Offset())).Limit(int(pager.PageSize())) }, func(tx *gorm.DB) *gorm.DB { return tx.Order("created_at DESC") }, ) if err != nil { logger.Error("failed to list backup exports", zap.Error(err), zap.Any("filter", filter)) return nil, err } // Convert backup exports to response items := make([]*pb.BackupExport, 0, len(backupExports)) for _, export := range backupExports { item := domain.BackupExportEntity2OpenAPIBackupExport(ctx, &export) items = append(items, item) } resp := &pb.ListBackupExportsResponse{ TotalSize: int32(count), NextPageToken: pager.NextPageToken(uint32(count)), BackupExports: items, } return resp, nil } func (s *BackupControllerImpl) GetBackupExport( ctx context.Context, req *pb.GetBackupExportRequest, ) (*pb.BackupExport, error) { ctx = log.WithTraceID(ctx) logger := log.GetLogger(ctx) backupID, err := strconv.ParseUint(req.BackupId, 10, 64) if err != nil { logger.Warn("invalid backup id", zap.Error(err), zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("invalid backup id") } exportID, err := strconv.ParseUint(req.ExportId, 10, 64) if err != nil { logger.Warn("invalid export id", zap.Error(err), zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("invalid export id") } if _, err := s.checkBackupResourcePermission(ctx, backupID); err != nil { logger.Warn("failed to check backup resource permission", zap.Error(err), zap.Any("request", req)) return nil, err } export, err := s.backupExportRepo.Load(ctx, exportID) if err != nil { logger.Error("failed to load backup export", zap.Error(err), zap.Any("request", req)) return nil, err } if export == nil { logger.Warn("backup export not found", zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("backup export not found") } if export.BackupID != entity.ID(backupID) { logger.Warn("invalid backup id", zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("invalid backup id") } resp := domain.BackupExportEntity2OpenAPIBackupExport(ctx, export) return resp, nil } func (s *BackupControllerImpl) CancelBackupExport( ctx context.Context, req *pb.CancelBackupExportRequest, ) (*emptypb.Empty, error) { ctx = log.WithTraceID(ctx) logger := log.GetLogger(ctx) backupID, err := strconv.ParseUint(req.BackupId, 10, 64) if err != nil { logger.Warn("invalid backup id", zap.Error(err), zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("invalid backup id") } exportID, err := strconv.ParseUint(req.ExportId, 10, 64) if err != nil { logger.Warn("invalid export id", zap.Error(err), zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("invalid export id") } if _, err := s.checkBackupResourcePermission(ctx, backupID); err != nil { logger.Warn("failed to check backup resource permission", zap.Error(err), zap.Any("request", req)) return nil, err } export, bizErr := s.backupExportRepo.Load(ctx, exportID) if bizErr != nil { logger.Error("failed to load backup export", zap.Error(bizErr), zap.Any("request", req)) return nil, bizErr } if export == nil { logger.Warn("backup export not found", zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("backup export not found") } if export.BackupID != entity.ID(backupID) { logger.Warn("invalid backup id", zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("invalid backup id") } if err := s.trans.Transaction(ctx, func(ctx context.Context) error { err := s.backupExportModel.CancelBackupExport(ctx, &dataflowpb.CancelBackupExportReq{ ExportId: exportID, }) if err != nil { logger.Error("failed to cancel backup export", zap.Error(err), zap.Any("request", req)) return err } return nil }); err != nil { return nil, err } return &emptypb.Empty{}, nil } // CreateBackup creates a new backup with multiple granularity levels // //nolint:funlen func (s *BackupControllerImpl) CreateBackup( ctx context.Context, req *pb.CreateBackupRequest, ) (*pb.Backup, error) { ctx = log.WithTraceID(ctx) logger := log.GetLogger(ctx) clusterID, cluster, err := s.validateClusterAccess(ctx, req.ClusterId) if err != nil { return nil, err } user := service.NewUserContext(ctx) // Convert backup scope to table filter tableFilter, scopeType, err := s.convertBackupScopeToTableFilter(ctx, req) if err != nil { logger.Warn("failed to convert backup scope", zap.Error(err), zap.Any("request", req)) return nil, err } // Create backup request createReq := &domain.AsyncCreateBackupReq{ Name: req.Name, Description: req.GetDescription(), ProjectID: cluster.GetProjectId(), ClusterID: clusterID, UserID: uint64(user.UserID()), BackupType: dataflowpb.BackupType_MANUAL, TableFilter: tableFilter, } // Call model layer to create backup var backupID uint64 err = s.trans.Transaction(ctx, func(ctx context.Context) error { resp, bizErr := s.backupCreateRouter.AsyncCreate(ctx, createReq) if bizErr != nil { logger.Error("failed to create backup", zap.Error(bizErr), zap.Any("request", createReq)) return bizErr } backupID = resp.BackupID return nil }) if err != nil { return nil, err } if backupID == 0 { backupID, err = s.waitForCreatedBackupID(ctx, clusterID, createReq.Name) if err != nil { logger.Error("failed to locate created backup after workflow start", zap.Uint64("cluster_id", clusterID), zap.String("backup_name", createReq.Name), zap.Error(err)) return nil, err } } // List the created backup to get full details (use List() to get correct Type field) listResp, bizErr := s.snapshotBackupModel.List(ctx, &domain.ListBackupReq{ Filter: &domain.ListBackupFilter{ BackupIDs: []uint64{backupID}, }, }) if bizErr != nil { logger.Error("failed to list backup", zap.Error(bizErr), zap.Uint64("backup_id", backupID)) return nil, bizErr } if listResp == nil || len(listResp.Items) == 0 { logger.Error("backup not found after creation", zap.Uint64("backup_id", backupID)) return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "backup not found after creation") } // Use the first item from list response (Type field is correctly populated) backupItem := listResp.Items[0] // Convert to proto response with scope information backup := s.convertDomainBackupToProtoWithScope(ctx, backupItem, scopeType, tableFilter) return backup, nil } func (s *BackupControllerImpl) waitForCreatedBackupID( ctx context.Context, clusterID uint64, backupName string, ) (uint64, error) { const ( maxAttempts = 20 waitInterval = 500 * time.Millisecond ) logger := log.GetLogger(ctx) for attempt := 0; attempt < maxAttempts; attempt++ { listResp, bizErr := s.snapshotBackupModel.List(ctx, &domain.ListBackupReq{ Filter: &domain.ListBackupFilter{ ClusterIDs: []uint64{clusterID}, }, Pager: &domain.Pager{ Page: 1, PageSize: backupLookupPageSize, }, }) if bizErr != nil { return 0, bizErr } for _, item := range listResp.Items { if item.Name == backupName { return item.ID, nil } } if attempt < maxAttempts-1 { time.Sleep(waitInterval) } } logger.Warn("backup entity not found after waiting", zap.Uint64("cluster_id", clusterID), zap.String("backup_name", backupName)) return 0, bizerr.New(bizerr.ErrCodeRecordNotFound, "backup not found after workflow start") } // convertBackupScopeToTableFilter converts proto backup scope to table filter strings func (s *BackupControllerImpl) convertBackupScopeToTableFilter( ctx context.Context, req *pb.CreateBackupRequest, ) ([]string, pb.BackupScopeType, error) { logger := log.GetLogger(ctx) scope := req.GetBackupScope() if scope == nil { // No scope specified - default to full backup logger.Info("no backup scope specified, defaulting to full backup") return nil, pb.BackupScopeType_BACKUP_SCOPE_TYPE_FULL, nil } switch scopeDetail := scope.GetBackupScope().(type) { case *pb.BackupScope_Full: // Full backup - no filter needed if scope.GetType() != pb.BackupScopeType_BACKUP_SCOPE_TYPE_FULL { return nil, pb.BackupScopeType_BACKUP_SCOPE_TYPE_UNSPECIFIED, bizerr_wrapper.NewInvalidInputError("backup_scope.type must be FULL when backup_scope.full is specified") } return nil, pb.BackupScopeType_BACKUP_SCOPE_TYPE_FULL, nil case *pb.BackupScope_Database: // Database-level backup - convert to "db.*" format if scope.GetType() != pb.BackupScopeType_BACKUP_SCOPE_TYPE_DATABASE { return nil, pb.BackupScopeType_BACKUP_SCOPE_TYPE_UNSPECIFIED, bizerr_wrapper.NewInvalidInputError("backup_scope.type must be DATABASE when backup_scope.database is specified") } if len(scopeDetail.Database.Databases) == 0 { return nil, pb.BackupScopeType_BACKUP_SCOPE_TYPE_UNSPECIFIED, bizerr_wrapper.NewInvalidInputError("at least one database must be specified") } filters := make([]string, 0, len(scopeDetail.Database.Databases)) for _, db := range scopeDetail.Database.Databases { filters = append(filters, db+".*") } return filters, pb.BackupScopeType_BACKUP_SCOPE_TYPE_DATABASE, nil case *pb.BackupScope_Table: // Table-level backup - use table names directly if scope.GetType() != pb.BackupScopeType_BACKUP_SCOPE_TYPE_TABLE { return nil, pb.BackupScopeType_BACKUP_SCOPE_TYPE_UNSPECIFIED, bizerr_wrapper.NewInvalidInputError("backup_scope.type must be TABLE when backup_scope.table is specified") } if len(scopeDetail.Table.Tables) == 0 { return nil, pb.BackupScopeType_BACKUP_SCOPE_TYPE_UNSPECIFIED, bizerr_wrapper.NewInvalidInputError("at least one table must be specified") } return scopeDetail.Table.Tables, pb.BackupScopeType_BACKUP_SCOPE_TYPE_TABLE, nil case *pb.BackupScope_Filter: // Filter-based backup - use include/exclude rules if scope.GetType() != pb.BackupScopeType_BACKUP_SCOPE_TYPE_FILTER { return nil, pb.BackupScopeType_BACKUP_SCOPE_TYPE_UNSPECIFIED, bizerr_wrapper.NewInvalidInputError("backup_scope.type must be FILTER when backup_scope.filter is specified") } if len(scopeDetail.Filter.IncludeRules) == 0 { return nil, pb.BackupScopeType_BACKUP_SCOPE_TYPE_UNSPECIFIED, bizerr_wrapper.NewInvalidInputError("at least one include rule must be specified") } // Combine include and exclude rules with proper BR format filters := make([]string, 0, len(scopeDetail.Filter.IncludeRules)+len(scopeDetail.Filter.ExcludeRules)) // Add include rules filters = append(filters, scopeDetail.Filter.IncludeRules...) // Add exclude rules with "!" prefix for _, rule := range scopeDetail.Filter.ExcludeRules { filters = append(filters, "!"+rule) } return filters, pb.BackupScopeType_BACKUP_SCOPE_TYPE_FILTER, nil case nil: if scope.GetType() == pb.BackupScopeType_BACKUP_SCOPE_TYPE_UNSPECIFIED { return nil, pb.BackupScopeType_BACKUP_SCOPE_TYPE_UNSPECIFIED, bizerr_wrapper.NewInvalidInputError("backup_scope.type must be specified when backup_scope is provided") } if scope.GetType() == pb.BackupScopeType_BACKUP_SCOPE_TYPE_FULL { return nil, pb.BackupScopeType_BACKUP_SCOPE_TYPE_FULL, nil } return nil, pb.BackupScopeType_BACKUP_SCOPE_TYPE_UNSPECIFIED, bizerr_wrapper.NewInvalidInputError("backup scope payload must match backup_scope.type") default: logger.Warn("unknown backup scope type", zap.Any("scope", scopeDetail), zap.Any("type", scope.GetType())) return nil, pb.BackupScopeType_BACKUP_SCOPE_TYPE_UNSPECIFIED, bizerr_wrapper.NewInvalidInputError("unknown backup scope type") } } // convertDomainBackupToProtoWithScope converts domain backup to proto with scope information func (s *BackupControllerImpl) convertDomainBackupToProtoWithScope( ctx context.Context, item domain.ListBackupRespItem, scopeType pb.BackupScopeType, tableFilter []string, ) *pb.Backup { backup := s.convertBackupToProto(ctx, item) // Add scope information backup.ScopeType = &scopeType if len(tableFilter) > 0 { scopeDetails := &pb.BackupScopeDetails{} switch scopeType { case pb.BackupScopeType_BACKUP_SCOPE_TYPE_DATABASE: // Extract database names from "db.*" format databases := make([]string, 0, len(tableFilter)) for _, filter := range tableFilter { if len(filter) > 2 && filter[len(filter)-2:] == ".*" { databases = append(databases, filter[:len(filter)-2]) } } scopeDetails.Databases = databases case pb.BackupScopeType_BACKUP_SCOPE_TYPE_TABLE: scopeDetails.Tables = tableFilter case pb.BackupScopeType_BACKUP_SCOPE_TYPE_FILTER: includeRules := make([]string, 0) excludeRules := make([]string, 0) for _, filter := range tableFilter { if len(filter) > 0 && filter[0] == '!' { excludeRules = append(excludeRules, filter[1:]) } else { includeRules = append(includeRules, filter) } } scopeDetails.IncludeRules = includeRules scopeDetails.ExcludeRules = excludeRules } backup.ScopeDetails = scopeDetails } return backup } // GetBackup retrieves a specific backup by ID func (s *BackupControllerImpl) GetBackup( ctx context.Context, req *pb.GetBackupRequest, ) (*pb.Backup, error) { ctx = log.WithTraceID(ctx) logger := log.GetLogger(ctx) // Parse backup_id backupID, err := strconv.ParseUint(req.BackupId, 10, 64) if err != nil { logger.Warn("invalid backup id", zap.Error(err), zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("invalid backup id") } // Check permission backup, bizErr := s.checkBackupResourcePermission(ctx, backupID) if bizErr != nil { logger.Warn("failed to check backup resource permission", zap.Error(bizErr), zap.Any("request", req)) return nil, bizErr } if backup == nil { logger.Error("backup not found after permission check", zap.Uint64("backup_id", backupID)) return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "backup not found") } // List backup details - using List() instead of Query() to get correct Type and other fields listResp, bizErr := s.snapshotBackupModel.List(ctx, &domain.ListBackupReq{ Filter: &domain.ListBackupFilter{ BackupIDs: []uint64{backupID}, }, }) if bizErr != nil { logger.Error("failed to list backup", zap.Error(bizErr), zap.Uint64("backup_id", backupID)) return nil, bizErr } if listResp == nil || len(listResp.Items) == 0 { logger.Warn("backup not found", zap.Uint64("backup_id", backupID)) return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "backup not found") } // Use the first item from list response (Type field is correctly populated by List()) backupItem := listResp.Items[0] // Parse scope information from table_filter (still need entity for TableFilter access) scopeType, tableFilter := s.parseBackupScopeFromEntity(ctx, backup) // Convert to proto with scope result := s.convertDomainBackupToProtoWithScope(ctx, backupItem, scopeType, tableFilter) return result, nil } // parseBackupScopeFromEntity extracts scope type and filter from entity func (s *BackupControllerImpl) parseBackupScopeFromEntity( ctx context.Context, backup *entity.Backup, ) (pb.BackupScopeType, []string) { return s.parseBackupScopeFromTableFilter(ctx, backup.TableFilter) } func (s *BackupControllerImpl) parseBackupScopeFromTableFilter( ctx context.Context, tableFilter []string, ) (pb.BackupScopeType, []string) { // If no table filter, it's a full backup if len(tableFilter) == 0 { return pb.BackupScopeType_BACKUP_SCOPE_TYPE_FULL, nil } filters := append([]string(nil), tableFilter...) // Check if it's database-level (all filters end with .*) allDatabaseLevel := true for _, f := range filters { if len(f) < 2 || f[len(f)-2:] != ".*" { allDatabaseLevel = false break } } if allDatabaseLevel { return pb.BackupScopeType_BACKUP_SCOPE_TYPE_DATABASE, filters } // Check if it has exclude rules (contains "!") hasExcludeRules := false for _, f := range filters { if len(f) > 0 && f[0] == '!' { hasExcludeRules = true break } } if hasExcludeRules { return pb.BackupScopeType_BACKUP_SCOPE_TYPE_FILTER, filters } // Default to table-level return pb.BackupScopeType_BACKUP_SCOPE_TYPE_TABLE, filters } // DeleteBackup deletes a specific backup func (s *BackupControllerImpl) DeleteBackup( ctx context.Context, req *pb.DeleteBackupRequest, ) (resp *emptypb.Empty, rErr error) { ctx = log.WithTraceID(ctx) logger := log.GetLogger(ctx) // Parse backup_id backupID, err := strconv.ParseUint(req.BackupId, 10, 64) if err != nil { logger.Warn("invalid backup id", zap.Error(err), zap.Any("request", req)) return nil, bizerr_wrapper.NewInvalidInputError("invalid backup id") } // Check permission and get backup entity backup, bizErr := s.checkBackupResourcePermission(ctx, backupID) if bizErr != nil { logger.Warn("failed to check backup resource permission", zap.Error(bizErr), zap.Any("request", req)) return nil, bizErr } // Record audit log on return (defer to capture error state) defer func() { s.recordAuditLogForDeleteBackup(ctx, backup, rErr) }() // Billing action check if err := s.doBillingActionCheckForBackupDeletion(ctx, uint64(backup.ClusterID), backupID); err != nil { return &emptypb.Empty{}, err } // Delete backup rErr = s.trans.Transaction(ctx, func(ctx context.Context) error { _, bizErr := s.snapshotBackupModel.AsyncDelete(ctx, &domain.AsyncDeleteBackupReq{ BackupID: backupID, }) if bizErr != nil { logger.Error("failed to delete backup", zap.Error(bizErr), zap.Uint64("backup_id", backupID)) return bizErr } return nil }) if rErr != nil { return nil, rErr } return &emptypb.Empty{}, nil } // doBillingActionCheckForBackupDeletion checks if backup deletion is allowed by billing system func (s *BackupControllerImpl) doBillingActionCheckForBackupDeletion( ctx context.Context, clusterID, backupID uint64, ) error { logger := log.GetLogger(ctx) cluster, err := s.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, clusterID) if err != nil { logger.Warn("failed to get cluster for billing check", zap.Uint64("cluster_id", clusterID), zap.Error(err)) return err } checkResult, berr := s.billingHandler.ActionCheck(ctx, cluster.GetOrganizationId(), billing.ActionInfo{ Resource: billing.Resource{ Type: billing.ResourceTypeDedicatedBackup, InstanceID: fmt.Sprint(backupID), Region: cluster.GetRegion(), Provider: cluster.GetCloudProvider(), IsFree: false, }, Action: billing.ActionDelete, }) if berr != nil { logger.Info("failed to check billing action", zap.Error(berr)) return berr } if !checkResult.Pass { return bizerr_wrapper.NewInvalidInputError("billing action check failed: %s", pointer.Get(checkResult.Reason)) } logger.Info("billing action check passed") return nil } // recordAuditLogForDeleteBackup records audit log for backup deletion func (s *BackupControllerImpl) recordAuditLogForDeleteBackup( ctx context.Context, backup *entity.Backup, rErr error, ) { logger := log.GetLogger(ctx) // Query backup schedule to get name and type queryResp, err := s.snapshotBackupModel.Query(ctx, &domain.QueryBackupReq{ Filter: &domain.ListBackupFilter{ BackupIDs: []uint64{uint64(backup.ID)}, }, }) if err != nil || queryResp == nil || queryResp.BackupEntity == nil || queryResp.BackupEntity.BackupSchedule == nil { logger.Warn("failed to submit audit log, query backup error", zap.Uint64("backup_id", uint64(backup.ID)), zap.Error(err)) return } // Prepare audit log details logDetail := &DeleteBackupAuditLogDetail{ BackupName: queryResp.BackupEntity.BackupSchedule.Name, BackupType: string(queryResp.BackupEntity.BackupSchedule.Type), Error: s.convertAuditLogDetailErrMsg(rErr), } details, ierr := s.convertToAuditLogDetail(logDetail) if ierr != nil { logger.Warn("failed to convert to audit log detail", zap.Error(ierr)) return } // Create and submit audit log event event := auditlog.NewEvent(ctx, auditlog.EventDeleteBackup). SetResult(rErr == nil). SetOrgID(uint64(backup.TenantID)). SetProjectID(uint64(backup.ProjectID)). SetDetails(details.Build()) // If OpenAPI call, set API key user := account.GetUserFromContext(ctx) if user != nil && user.Email == "OpenAPI" { event.SetAPIPubKey(user.LastName) } if err := event.Submit(); err != nil { logger.Warn("failed to submit audit log", zap.Error(err)) return } } // DeleteBackupAuditLogDetail represents audit log detail for backup deletion type DeleteBackupAuditLogDetail struct { BackupName string `json:"backup_name,omitempty"` BackupType string `json:"backup_type,omitempty"` Error *string `json:"error,omitempty"` } // convertToAuditLogDetail converts struct to audit log details func (s *BackupControllerImpl) convertToAuditLogDetail(a interface{}) (auditlog.Details, error) { marshal, err := json.Marshal(a) if err != nil { return nil, err } details := auditlog.Details{} if err = json.Unmarshal(marshal, &details); err != nil { return nil, err } return details, nil } // convertAuditLogDetailErrMsg converts error to audit log error message func (s *BackupControllerImpl) convertAuditLogDetailErrMsg(vv error) *string { if vv == nil { return nil } if bz, ok := vv.(bizerr.BizError); ok { return pointer.To(fmt.Sprintf("%d, %s", bz.ErrCode(), bz.ErrMsg())) } return pointer.To(fmt.Sprintf("%s", vv)) } // GetBackupOptions gets available backup options for a cluster // //nolint:funlen func (s *BackupControllerImpl) GetBackupOptions( ctx context.Context, req *pb.GetBackupOptionsRequest, ) (*pb.BackupOptions, error) { ctx = log.WithTraceID(ctx) logger := log.GetLogger(ctx) clusterID, cluster, err := s.validateClusterAccess(ctx, req.ClusterId) if err != nil { return nil, err } // Query backup region options optionsResp, bizErr := s.backupSettingModel.QueryBackupRegionOptions(ctx, &domain.QueryBackupRegionOptionsReq{ ClusterID: clusterID, TenantID: cluster.GetOrganizationId(), ProjectID: cluster.GetProjectId(), }) if bizErr != nil { logger.Error("failed to query backup region options", zap.Error(bizErr), zap.Uint64("cluster_id", clusterID)) return nil, bizErr } // Convert to proto response resp := &pb.BackupOptions{ ClusterId: req.ClusterId, } // Always build DualRegionOptions with at least ClusterRegion if optionsResp.ClusterRegion != nil { resp.DualRegionOptions = &pb.DualRegionOptions{ ClusterRegion: &pb.RegionInfo{ Name: optionsResp.ClusterRegion.Name, DisplayName: optionsResp.ClusterRegion.DisplayName, }, } } // Add region options if available if len(optionsResp.BackupRegionOptions) > 0 { regionOptions := make([]*pb.RegionOption, 0, len(optionsResp.BackupRegionOptions)) for _, opt := range optionsResp.BackupRegionOptions { if opt.Region != nil { regionOptions = append(regionOptions, &pb.RegionOption{ Region: &pb.RegionInfo{ Name: opt.Region.Name, DisplayName: opt.Region.DisplayName, }, }) } } // If DualRegionOptions already exists, add RegionOptions to it if resp.DualRegionOptions != nil { resp.DualRegionOptions.RegionOptions = regionOptions } else { // If ClusterRegion was nil, create DualRegionOptions with just RegionOptions resp.DualRegionOptions = &pb.DualRegionOptions{ RegionOptions: regionOptions, } } } return resp, nil } // ModifyBackupSettings modifies backup settings for a cluster // //nolint:funlen func (s *BackupControllerImpl) ModifyBackupSettings( ctx context.Context, req *pb.ModifyBackupSettingsRequest, ) (resp *pb.BackupSettings, rErr error) { ctx = log.WithTraceID(ctx) logger := log.GetLogger(ctx) // Validate cluster access and permissions clusterID, cluster, err := s.validateClusterAccess(ctx, req.ClusterId) if err != nil { return nil, err } // Query current settings for audit log and change detection beforeSetting, err := s.backupSettingModel.QueryBackupSettingForAPI(ctx, &clusterID) if err != nil { logger.Error("failed to query current backup settings", zap.Error(err)) return nil, err } // Record audit log on return (defer to capture error state) defer func() { s.recordAuditLogForModifyBackupSetting(ctx, cluster, beforeSetting, rErr) }() // Validate parameters if err := s.validateModifyBackupSettingParams(ctx, req); err != nil { logger.Warn("invalid parameters", zap.Error(err)) return nil, err } // Calculate what settings have changed changeInfo, err := s.calculateBackupSettingChangeInfo(ctx, req, beforeSetting) if err != nil { logger.Error("failed to calculate change info", zap.Error(err)) return nil, err } // Check for conflicting changes (PITR + DualRegion simultaneously) if changeInfo.pitrChanged && changeInfo.dualRegionChanged { logger.Warn("attempt to change PITR and dual-region settings simultaneously") return nil, bizerr.New(bizerr.ErrCodeParams, "cannot modify PITR and dual-region settings simultaneously") } // Modify backup settings in a transaction rErr = s.trans.Transaction(ctx, func(ctx context.Context) error { // Modify settings based on what changed if changeInfo.pitrChanged { if err := s.modifyPitrSetting(ctx, clusterID, req.PitrSetting); err != nil { return err } } if changeInfo.dualRegionChanged { if err := s.modifyDualRegionSetting(ctx, clusterID, req.Regions, changeInfo); err != nil { return err } } if changeInfo.snapshotScheduleChanged { if err := s.modifySnapshotBackupSchedule(ctx, clusterID, req.SnapshotBackupSchedule, beforeSetting); err != nil { return err } } return nil }) if rErr != nil { return nil, rErr } // Query updated settings and return return s.GetBackupSettings(ctx, &pb.GetBackupSettingsRequest{ ClusterId: req.ClusterId, }) } // validateModifyBackupSettingParams validates the parameters for ModifyBackupSetting // //nolint:gocognit,gocyclo,nestif func (s *BackupControllerImpl) validateModifyBackupSettingParams( ctx context.Context, req *pb.ModifyBackupSettingsRequest, ) error { logger := log.GetLogger(ctx) // At least one setting must be provided if len(req.Regions) == 0 && req.PitrSetting == nil && req.SnapshotBackupSchedule == nil { logger.Warn("no settings to modify") return bizerr_wrapper.NewInvalidInputError("at least one setting must be provided") } if req.PitrSetting != nil && req.PitrSetting.Enabled == nil { logger.Warn("pitr setting enabled is required when pitr_setting is provided") return bizerr_wrapper.NewInvalidInputError("pitr_setting.enabled is required") } // Validate explicitly provided snapshot backup schedule fields. // Required-field validation for enabled schedules is handled after sparse merge. if req.SnapshotBackupSchedule != nil { if req.SnapshotBackupSchedule.RetentionDays != nil && req.SnapshotBackupSchedule.GetRetentionDays() <= 0 { logger.Warn("invalid retention days") return bizerr_wrapper.NewInvalidInputError("retention_days must be greater than 0") } switch schedule := req.SnapshotBackupSchedule.Schedule.(type) { case *pb.SnapshotBackupSchedule_DailySchedule: if schedule.DailySchedule == nil || schedule.DailySchedule.ScheduleTime == "" { logger.Warn("daily schedule time is required") return bizerr_wrapper.NewInvalidInputError("daily_schedule.schedule_time is required") } if !snapshotScheduleTimeRegexp.MatchString(schedule.DailySchedule.ScheduleTime) { logger.Warn("daily schedule time is invalid") return bizerr_wrapper.NewInvalidInputError("daily_schedule.schedule_time is invalid") } case *pb.SnapshotBackupSchedule_WeeklySchedule: if schedule.WeeklySchedule == nil { logger.Warn("weekly schedule is required") return bizerr_wrapper.NewInvalidInputError("weekly_schedule is required") } if schedule.WeeklySchedule.ScheduleTime == "" { logger.Warn("weekly schedule time is required") return bizerr_wrapper.NewInvalidInputError("weekly_schedule.schedule_time is required") } if !snapshotScheduleTimeRegexp.MatchString(schedule.WeeklySchedule.ScheduleTime) { logger.Warn("weekly schedule time is invalid") return bizerr_wrapper.NewInvalidInputError("weekly_schedule.schedule_time is invalid") } if len(schedule.WeeklySchedule.WeekDays) == 0 { logger.Warn("weekly schedule days are required") return bizerr_wrapper.NewInvalidInputError("weekly_schedule.week_days is required") } } } return nil } // backupSettingChangeInfo tracks what settings have changed type backupSettingChangeInfo struct { dualRegionChanged bool requestDualRegionEnable bool currentDualRegionEnable bool pitrChanged bool requestPitrEnable bool currentPitrEnable bool snapshotScheduleChanged bool } type resolvedSnapshotBackupSchedule struct { Enabled bool RetentionDays int32 ScheduleType domain.SnapshotBackupScheduleType DailyScheduleTime string WeeklySchedule *domain.WeeklySchedule } func hasSnapshotScheduleEnabled(schedule *pb.SnapshotBackupSchedule) bool { return schedule != nil && schedule.Enabled != nil } func hasSnapshotScheduleRetentionDays(schedule *pb.SnapshotBackupSchedule) bool { return schedule != nil && schedule.RetentionDays != nil } //nolint:funlen,gocognit,gocyclo func resolveSnapshotBackupSchedule( schedule *pb.SnapshotBackupSchedule, currentSetting *domain.QueryBackupSettingForAPIResp, ) (*resolvedSnapshotBackupSchedule, error) { resolved := &resolvedSnapshotBackupSchedule{} if currentSetting != nil && currentSetting.SnapshotBackupSchedule != nil { currentSchedule := currentSetting.SnapshotBackupSchedule resolved.Enabled = !currentSchedule.Disable if currentSchedule.RetentionPolicy != nil { resolved.RetentionDays = int32(currentSchedule.RetentionPolicy.RetentionPeriod) } resolved.ScheduleType = currentSchedule.ScheduleType resolved.DailyScheduleTime = currentSchedule.DailyScheduleTime if currentSchedule.WeeklySchedule != nil { resolved.WeeklySchedule = &domain.WeeklySchedule{ ScheduleTime: currentSchedule.WeeklySchedule.ScheduleTime, Days: append([]domain.WeekDay(nil), currentSchedule.WeeklySchedule.Days...), } } } if schedule != nil { if hasSnapshotScheduleEnabled(schedule) { resolved.Enabled = schedule.GetEnabled() } if hasSnapshotScheduleRetentionDays(schedule) { resolved.RetentionDays = schedule.GetRetentionDays() } switch s := schedule.Schedule.(type) { case *pb.SnapshotBackupSchedule_DailySchedule: if s.DailySchedule != nil { resolved.ScheduleType = domain.SnapshotBackupScheduleTypeDaily resolved.DailyScheduleTime = s.DailySchedule.ScheduleTime resolved.WeeklySchedule = nil } case *pb.SnapshotBackupSchedule_WeeklySchedule: if s.WeeklySchedule != nil { resolved.ScheduleType = domain.SnapshotBackupScheduleTypeWeekly resolved.DailyScheduleTime = "" resolved.WeeklySchedule = &domain.WeeklySchedule{ ScheduleTime: s.WeeklySchedule.ScheduleTime, Days: convertProtoWeekDaysToDomainWeekDays(s.WeeklySchedule.WeekDays), } } } } if !resolved.Enabled { return resolved, nil } if resolved.RetentionDays <= 0 { return nil, bizerr_wrapper.NewInvalidInputError("retention_days must be greater than 0") } switch resolved.ScheduleType { case domain.SnapshotBackupScheduleTypeDaily: if resolved.DailyScheduleTime == "" { return nil, bizerr_wrapper.NewInvalidInputError("daily_schedule.schedule_time is required") } if !snapshotScheduleTimeRegexp.MatchString(resolved.DailyScheduleTime) { return nil, bizerr_wrapper.NewInvalidInputError("daily_schedule.schedule_time is invalid") } case domain.SnapshotBackupScheduleTypeWeekly: if resolved.WeeklySchedule == nil { return nil, bizerr_wrapper.NewInvalidInputError("weekly_schedule is required") } if resolved.WeeklySchedule.ScheduleTime == "" { return nil, bizerr_wrapper.NewInvalidInputError("weekly_schedule.schedule_time is required") } if !snapshotScheduleTimeRegexp.MatchString(resolved.WeeklySchedule.ScheduleTime) { return nil, bizerr_wrapper.NewInvalidInputError("weekly_schedule.schedule_time is invalid") } if len(resolved.WeeklySchedule.Days) == 0 { return nil, bizerr_wrapper.NewInvalidInputError("weekly_schedule.week_days is required") } default: return nil, bizerr_wrapper.NewInvalidInputError("schedule is required when enabled") } return resolved, nil } // calculateBackupSettingChangeInfo calculates what settings have changed // //nolint:gocyclo func (s *BackupControllerImpl) calculateBackupSettingChangeInfo( ctx context.Context, req *pb.ModifyBackupSettingsRequest, currentSetting *domain.QueryBackupSettingForAPIResp, ) (*backupSettingChangeInfo, error) { logger := log.GetLogger(ctx) info := &backupSettingChangeInfo{} // Calculate dual region changes if len(req.Regions) > 0 { info.requestDualRegionEnable = len(req.Regions) > 1 info.currentDualRegionEnable = currentSetting.BucketSetting != nil && currentSetting.BucketSetting.DualRegionEnable // Check if regions actually changed currentRegions := s.getCurrentRegionNames(currentSetting) info.dualRegionChanged = !stringSliceEqual(currentRegions, req.Regions) || info.currentDualRegionEnable != info.requestDualRegionEnable } info.currentPitrEnable = currentSetting.PiTRSetting != nil && currentSetting.PiTRSetting.Enable // Handle snapshot schedule changes if req.SnapshotBackupSchedule != nil { info.snapshotScheduleChanged = true if hasSnapshotScheduleEnabled(req.SnapshotBackupSchedule) && !req.SnapshotBackupSchedule.GetEnabled() && info.currentPitrEnable { if req.PitrSetting == nil || req.PitrSetting.Enabled == nil || req.PitrSetting.GetEnabled() { logger.Warn("attempt to disable snapshot backup schedule while PITR remains enabled") return nil, bizerr.New( bizerr.ErrCodeParams, "cannot disable snapshot backup schedule while PITR is enabled; "+ "disable PITR explicitly in the same request or first", ) } } } // Calculate PITR changes if req.PitrSetting != nil { info.requestPitrEnable = req.PitrSetting.GetEnabled() info.pitrChanged = info.requestPitrEnable != info.currentPitrEnable // Validate: cannot enable PITR when snapshot schedule is disabled if req.PitrSetting.GetEnabled() && hasSnapshotScheduleEnabled(req.SnapshotBackupSchedule) && !req.SnapshotBackupSchedule.GetEnabled() { logger.Warn("attempt to enable PITR while disabling snapshot backup schedule") return nil, bizerr.New(bizerr.ErrCodeParams, "cannot enable PITR when snapshot backup schedule is disabled") } } return info, nil } // getCurrentRegionNames extracts region names from current settings func (s *BackupControllerImpl) getCurrentRegionNames( setting *domain.QueryBackupSettingForAPIResp, ) []string { if setting.BucketSetting == nil || len(setting.BucketSetting.Buckets) == 0 { return nil } regionSet := make(map[string]bool) for _, bucket := range setting.BucketSetting.Buckets { for _, region := range bucket.Regions { regionSet[region.Name] = true } } regions := make([]string, 0, len(regionSet)) for region := range regionSet { regions = append(regions, region) } return regions } // stringSliceEqual checks if two string slices contain the same elements (order-independent) func stringSliceEqual(a, b []string) bool { if len(a) != len(b) { return false } aMap := make(map[string]bool) for _, s := range a { aMap[s] = true } for _, s := range b { if !aMap[s] { return false } } return true } // modifyPitrSetting modifies PITR setting func (s *BackupControllerImpl) modifyPitrSetting( ctx context.Context, clusterID uint64, pitrSetting *pb.PITRSetting, ) error { logger := log.GetLogger(ctx) bizErr := s.backupSettingModel.ModifyPitBackupSetting(ctx, &domain.PitSettingModificationReq{ ClusterID: clusterID, Enable: pitrSetting.GetEnabled(), }) if bizErr != nil { logger.Error("failed to modify PITR setting", zap.Error(bizErr)) return bizErr } logger.Info("modified PITR setting successfully", zap.Bool("enabled", pitrSetting.GetEnabled())) return nil } // modifyDualRegionSetting modifies dual-region setting func (s *BackupControllerImpl) modifyDualRegionSetting( ctx context.Context, clusterID uint64, regions []string, changeInfo *backupSettingChangeInfo, ) error { logger := log.GetLogger(ctx) clusterIDPtr := &clusterID _, bizErr := s.backupSettingModel.ModifyRegionSetting(ctx, &domain.RegionSettingModificationReqV2{ ClusterID: clusterIDPtr, Enable: changeInfo.requestDualRegionEnable, Regions: regions, }) if bizErr != nil { logger.Error("failed to modify region setting", zap.Error(bizErr)) return bizErr } logger.Info("modified dual-region setting successfully", zap.Bool("enabled", changeInfo.requestDualRegionEnable), zap.Strings("regions", regions)) return nil } // modifySnapshotBackupSchedule modifies snapshot backup schedule func (s *BackupControllerImpl) modifySnapshotBackupSchedule( ctx context.Context, clusterID uint64, schedule *pb.SnapshotBackupSchedule, currentSetting *domain.QueryBackupSettingForAPIResp, ) error { logger := log.GetLogger(ctx) resolvedSchedule, err := resolveSnapshotBackupSchedule(schedule, currentSetting) if err != nil { logger.Warn("invalid snapshot backup schedule", zap.Error(err)) return err } scheduleReq := &domain.ModifyAutoBackupSettingReq{ ClusterID: clusterID, SnapshotBackupSchedule: domain.SnapshotBackupSchedule{ Disable: !resolvedSchedule.Enabled, }, } // Convert schedule based on oneof field if resolvedSchedule.Enabled { switch resolvedSchedule.ScheduleType { case domain.SnapshotBackupScheduleTypeDaily: scheduleReq.ScheduleType = domain.SnapshotBackupScheduleTypeDaily scheduleReq.DailyScheduleTime = resolvedSchedule.DailyScheduleTime case domain.SnapshotBackupScheduleTypeWeekly: scheduleReq.ScheduleType = domain.SnapshotBackupScheduleTypeWeekly scheduleReq.WeeklySchedule = resolvedSchedule.WeeklySchedule } scheduleReq.RetentionPolicy = &domain.RetentionPolicy{ RetentionPeriod: int(resolvedSchedule.RetentionDays), } } _, bizErr := s.backupSettingModel.ModifyAutoBackupSetting(ctx, scheduleReq) if bizErr != nil { logger.Error("failed to modify snapshot backup schedule", zap.Error(bizErr)) return bizErr } logger.Info("modified snapshot backup schedule successfully", zap.Bool("enabled", resolvedSchedule.Enabled)) return nil } // recordAuditLogForModifyBackupSetting records audit log for backup setting modification func (s *BackupControllerImpl) recordAuditLogForModifyBackupSetting( ctx context.Context, cluster *mgmt.Cluster, beforeSetting *domain.QueryBackupSettingForAPIResp, rErr error, ) { logger := log.GetLogger(ctx) // Prepare audit log details logDetail := &ModifyBackupSettingAuditLogDetail{ Before: s.convertSettingToAuditLogDetailItem(beforeSetting), Error: s.convertAuditLogDetailErrMsg(rErr), } // Query after setting if successful if rErr == nil { clusterID := cluster.GetClusterId() afterSetting, err := s.backupSettingModel.QueryBackupSettingForAPI(ctx, &clusterID) if err != nil { logger.Warn("failed to query backup setting after modification for audit log", zap.Error(err)) } else { logDetail.After = s.convertSettingToAuditLogDetailItem(afterSetting) } } details, ierr := s.convertToAuditLogDetail(logDetail) if ierr != nil { logger.Warn("failed to convert to audit log detail", zap.Error(ierr)) return } // Create and submit audit log event event := auditlog.NewEvent(ctx, auditlog.EventSetAutoBackup). SetResult(rErr == nil). SetOrgID(cluster.GetOrganizationId()). SetProjectID(cluster.GetProjectId()). SetDetails(details.Build()) // If OpenAPI call, set API key user := account.GetUserFromContext(ctx) if user != nil && user.Email == "OpenAPI" { event.SetAPIPubKey(user.LastName) } if err := event.Submit(); err != nil { logger.Warn("failed to submit audit log", zap.Error(err)) return } } // ModifyBackupSettingAuditLogDetail represents audit log detail for backup setting modification type ModifyBackupSettingAuditLogDetail struct { Before *ModifyBackupSettingAuditLogDetailItem `json:"before,omitempty"` After *ModifyBackupSettingAuditLogDetailItem `json:"after,omitempty"` Error *string `json:"error,omitempty"` } // ModifyBackupSettingAuditLogDetailItem represents a single backup setting state type ModifyBackupSettingAuditLogDetailItem struct { SnapshotBackupSchedule *domain.SnapshotBackupSchedule `json:"snapshot_backup_schedule,omitempty"` PiTR *domain.PiTRSetting `json:"pitr,omitempty"` Regions []string `json:"regions,omitempty"` } // convertSettingToAuditLogDetailItem converts setting to audit log detail item func (s *BackupControllerImpl) convertSettingToAuditLogDetailItem( setting *domain.QueryBackupSettingForAPIResp, ) *ModifyBackupSettingAuditLogDetailItem { if setting == nil { return nil } detailItem := &ModifyBackupSettingAuditLogDetailItem{} // Copy snapshot backup schedule detailItem.SnapshotBackupSchedule = setting.SnapshotBackupSchedule if detailItem.SnapshotBackupSchedule != nil { // Remove next backup time for cleaner comparison detailItem.SnapshotBackupSchedule.NextBackupTime = nil } // Copy PITR setting detailItem.PiTR = setting.PiTRSetting // Extract region display names if setting.BucketSetting != nil { for _, bucket := range setting.BucketSetting.Buckets { for _, region := range bucket.Regions { detailItem.Regions = append(detailItem.Regions, region.DisplayName) } } } return detailItem } // validateClusterAccess validates cluster ID and user permissions // Returns clusterID (uint64), cluster object, and error func (s *BackupControllerImpl) validateClusterAccess( ctx context.Context, clusterIDStr string, ) (uint64, *mgmt.Cluster, error) { logger := log.GetLogger(ctx) // Parse cluster_id clusterID, err := strconv.ParseUint(clusterIDStr, 10, 64) if err != nil { logger.Warn("invalid cluster id", zap.Error(err), zap.String("cluster_id_str", clusterIDStr)) return 0, nil, bizerr_wrapper.NewInvalidInputError("invalid cluster id") } // Get cluster cluster, err := s.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, clusterID) if err != nil { logger.Warn("failed to get cluster", zap.Error(err), zap.Uint64("cluster_id", clusterID)) return 0, nil, bizerr.New(bizerr.ErrCodeDB, "get cluster %d: %v", clusterID, err) } // Check user permission user := service.NewUserContext(ctx) if user == nil { logger.Warn("user not found") return 0, nil, bizerr_wrapper.NewForbiddenErr("user not found") } if cluster.GetOrganizationId() != uint64(user.OrgID()) { logger.Warn("user does not have permission to access this cluster", zap.Uint64("cluster_id", clusterID), zap.Uint64("tenant_id", cluster.GetOrganizationId()), zap.Uint64("user_org_id", uint64(user.OrgID()))) return 0, nil, bizerr_wrapper.NewForbiddenErr("user does not have permission to access this cluster") } return clusterID, cluster, nil } // GetBackupSettings gets current backup settings for a cluster func (s *BackupControllerImpl) GetBackupSettings( ctx context.Context, req *pb.GetBackupSettingsRequest, ) (*pb.BackupSettings, error) { ctx = log.WithTraceID(ctx) logger := log.GetLogger(ctx) // Validate cluster access and permissions clusterID, _, err := s.validateClusterAccess(ctx, req.ClusterId) if err != nil { return nil, err } // Query backup settings settingResp, bizErr := s.backupSettingModel.QueryBackupSettingForAPI(ctx, &clusterID) if bizErr != nil { logger.Error("failed to query backup setting", zap.Error(bizErr), zap.Uint64("cluster_id", clusterID)) return nil, bizErr } // Convert to proto response. resp := s.convertDomainBackupSettingToProto(clusterID, settingResp) return resp, nil } // GetBackupPITRTimeRanges gets PITR time ranges for a cluster. func (s *BackupControllerImpl) GetBackupPITRTimeRanges( ctx context.Context, req *pb.GetBackupPITRTimeRangesRequest, ) (*pb.GetBackupPITRTimeRangesResponse, error) { ctx = log.WithTraceID(ctx) logger := log.GetLogger(ctx) clusterID, cluster, err := s.validateClusterAccess(ctx, req.ClusterId) if err != nil { return nil, err } timeRangeResp, bizErr := s.pitBackupModel.ListPiTRTimeRanges(ctx, &domain.PiTRTimeRangeListReq{ OrgId: cluster.GetOrganizationId(), ProjectId: 0, // ProjectId is not used in ListPiTRTimeRanges logic. ClusterId: clusterID, }) if bizErr != nil { logger.Error("failed to get PITR time ranges", zap.Error(bizErr), zap.Uint64("cluster_id", clusterID)) return nil, bizErr } resp := &pb.GetBackupPITRTimeRangesResponse{} if timeRangeResp == nil { return resp, nil } resp.Items = s.convertDomainPiTRRegionTimeRangesToProto(timeRangeResp.Items) resp.Message = timeRangeResp.Msg return resp, nil } // convertDomainBackupSettingToProto converts domain backup setting to proto. // //nolint:funlen func (s *BackupControllerImpl) convertDomainBackupSettingToProto( clusterID uint64, domainSetting *domain.QueryBackupSettingForAPIResp, ) *pb.BackupSettings { setting := &pb.BackupSettings{ ClusterId: strconv.FormatUint(clusterID, 10), } // Convert regions from bucket setting if domainSetting.BucketSetting != nil && len(domainSetting.BucketSetting.Buckets) > 0 { regions := make([]*pb.RegionInfo, 0) for _, bucket := range domainSetting.BucketSetting.Buckets { for _, r := range bucket.Regions { regions = append(regions, &pb.RegionInfo{ Name: r.Name, DisplayName: r.DisplayName, }) } } setting.Regions = regions } // Convert PITR setting if domainSetting.PiTRSetting != nil { pitrSetting := &pb.PITRSetting{ Enabled: pointer.ToBool(domainSetting.PiTRSetting.Enable), } setting.PitrSetting = pitrSetting } // Convert snapshot backup schedule if domainSetting.SnapshotBackupSchedule != nil { schedule := &pb.SnapshotBackupSchedule{ Enabled: pointer.ToBool(!domainSetting.SnapshotBackupSchedule.Disable), } // Convert schedule type and populate oneof schedule field switch domainSetting.SnapshotBackupSchedule.ScheduleType { case domain.SnapshotBackupScheduleTypeDaily: schedule.ScheduleType = pb.SnapshotBackupScheduleType_SNAPSHOT_BACKUP_SCHEDULE_TYPE_DAILY schedule.Schedule = &pb.SnapshotBackupSchedule_DailySchedule{ DailySchedule: &pb.DailySchedule{ ScheduleTime: domainSetting.SnapshotBackupSchedule.DailyScheduleTime, }, } case domain.SnapshotBackupScheduleTypeWeekly: schedule.ScheduleType = pb.SnapshotBackupScheduleType_SNAPSHOT_BACKUP_SCHEDULE_TYPE_WEEKLY if domainSetting.SnapshotBackupSchedule.WeeklySchedule != nil { schedule.Schedule = &pb.SnapshotBackupSchedule_WeeklySchedule{ WeeklySchedule: &pb.WeeklySchedule{ ScheduleTime: domainSetting.SnapshotBackupSchedule.WeeklySchedule.ScheduleTime, WeekDays: convertDomainWeekDaysToProtoWeekDays(domainSetting.SnapshotBackupSchedule.WeeklySchedule.Days), }, } } } // Set retention days from retention policy if domainSetting.SnapshotBackupSchedule.RetentionPolicy != nil { retentionDays := int32(domainSetting.SnapshotBackupSchedule.RetentionPolicy.RetentionPeriod) schedule.RetentionDays = &retentionDays } // Add next backup time if available if domainSetting.SnapshotBackupSchedule.NextBackupTime != nil { schedule.NextBackupTime = timestamppb.New(*domainSetting.SnapshotBackupSchedule.NextBackupTime) } setting.SnapshotBackupSchedule = schedule } // Convert state and actions using the status matrix logic from console API // This considers BackupSetting.Status, AllowActions, and PitBackupSwitch.Status var pitSwitchStatus entity.PitBackupSwitchStatus if domainSetting.PitBackupSwitch != nil { pitSwitchStatus = domainSetting.PitBackupSwitch.Status } consoleStatus, consoleActions := domain.ConvertBackupSettingStatusMatrixToDataflow( domainSetting.Status, domainSetting.AllowActions, pitSwitchStatus, ) // Map console proto enum to v1beta1 proto enum stateValue := s.convertConsoleBackupSettingStatusToV1beta1(consoleStatus) setting.State = &stateValue // Map console action enums to v1beta1 action enums setting.AllowedActions = s.convertConsoleBackupSettingActionsToV1beta1(consoleActions) return setting } func (s *BackupControllerImpl) convertDomainPiTRRegionTimeRangesToProto( items []*domain.PiTRRegionTimeRange, ) []*pb.PITRRegionTimeRange { if len(items) == 0 { return nil } resp := make([]*pb.PITRRegionTimeRange, 0, len(items)) for _, item := range items { if item == nil { continue } regionItem := &pb.PITRRegionTimeRange{} if item.Region != nil { regionItem.Region = &pb.RegionInfo{ Name: item.Region.Name, DisplayName: item.Region.DisplayName, } } if len(item.TimeRanges) > 0 { regionItem.TimeRanges = make([]*pb.PITRTimeRange, 0, len(item.TimeRanges)) for _, tr := range item.TimeRanges { if tr == nil { continue } regionItem.TimeRanges = append(regionItem.TimeRanges, &pb.PITRTimeRange{ StartTime: tr.StartTS, EndTime: tr.EndTS, }) } } resp = append(resp, regionItem) } return resp } // convertConsoleBackupSettingStatusToV1beta1 maps console dataflowpb.BackupSettingStatus to v1beta1 BackupSettingState func (s *BackupControllerImpl) convertConsoleBackupSettingStatusToV1beta1( consoleStatus dataflowpb.BackupSettingStatus, ) pb.BackupSettingState { switch consoleStatus { case dataflowpb.BackupSettingStatus_Creating: return pb.BackupSettingState_BACKUP_SETTING_STATE_CREATING case dataflowpb.BackupSettingStatus_Editing_Region: return pb.BackupSettingState_BACKUP_SETTING_STATE_EDITING_REGION case dataflowpb.BackupSettingStatus_Editing_PiTR: return pb.BackupSettingState_BACKUP_SETTING_STATE_EDITING_PITR case dataflowpb.BackupSettingStatus_Normal: return pb.BackupSettingState_BACKUP_SETTING_STATE_NORMAL case dataflowpb.BackupSettingStatus_Deleted: return pb.BackupSettingState_BACKUP_SETTING_STATE_DELETED case dataflowpb.BackupSettingStatus_Waiting_For_Editing_PiTR: return pb.BackupSettingState_BACKUP_SETTING_STATE_WAITING_FOR_EDITING_PITR default: return pb.BackupSettingState_BACKUP_SETTING_STATE_NORMAL } } // convertConsoleBackupSettingActionsToV1beta1 maps console dataflowpb.BackupSettingActionType to v1beta1 BackupSettingAction func (s *BackupControllerImpl) convertConsoleBackupSettingActionsToV1beta1( consoleActions []dataflowpb.BackupSettingActionType, ) []pb.BackupSettingAction { actions := make([]pb.BackupSettingAction, 0, len(consoleActions)) for _, action := range consoleActions { switch action { case dataflowpb.BackupSettingActionType_Edit_Region: actions = append(actions, pb.BackupSettingAction_BACKUP_SETTING_ACTION_EDIT_REGION) case dataflowpb.BackupSettingActionType_Edit_PiTR: actions = append(actions, pb.BackupSettingAction_BACKUP_SETTING_ACTION_EDIT_PITR) case dataflowpb.BackupSettingActionType_Edit_Schedule: actions = append(actions, pb.BackupSettingAction_BACKUP_SETTING_ACTION_EDIT_SNAPSHOT_BACKUP_SCHEDULE) } } return actions } // convertProtoWeekDaysToDomainWeekDays converts proto dataflowpb.WeekDay to domain WeekDay array // Now that v1beta1 uses the same WeekDay enum as console API, we can use shared conversion func convertProtoWeekDaysToDomainWeekDays(protoWeekDays []dataflowpb.WeekDay) []domain.WeekDay { if len(protoWeekDays) == 0 { return nil } domainDays := make([]domain.WeekDay, 0, len(protoWeekDays)) for _, protoDay := range protoWeekDays { // dataflowpb.WeekDay enum: SUNDAY=0, MONDAY=1, ..., SATURDAY=6 // This matches entity format directly entityDay := int32(protoDay) domainDays = append(domainDays, domain.WeekDayEntity2Domain(entityDay)) } return domainDays } // convertDomainWeekDaysToProtoWeekDays converts domain WeekDay array to proto dataflowpb.WeekDay func convertDomainWeekDaysToProtoWeekDays(domainDays []domain.WeekDay) []dataflowpb.WeekDay { if len(domainDays) == 0 { return nil } protoDays := make([]dataflowpb.WeekDay, 0, len(domainDays)) for _, domainDay := range domainDays { // Convert domain to entity format (0-6) entityDay := domain.WeekdayDomain2Entity(domainDay) // Entity format matches dataflowpb.WeekDay enum directly protoDays = append(protoDays, dataflowpb.WeekDay(entityDay)) } return protoDays }