feat(ydbcp): support ttl for backups

internal/connectors/db/process_result_set.go

@@ -77,6 +77,8 @@ func ReadBackupFromResultSet(res result.Result) (*types.Backup, error) {
 		creator     *string
 		completedAt *time.Time
 		createdAt   *time.Time
+		expireAt    *time.Time
+		expireAtTs  *timestamppb.Timestamp
 	)
 
 	err := res.ScanNamed(
@@ -91,6 +93,7 @@ func ReadBackupFromResultSet(res result.Result) (*types.Backup, error) {
 		named.Optional("status", &status),
 		named.Optional("message", &message),
 		named.Optional("size", &size),
+		named.Optional("expire_at", &expireAt),
 
 		named.Optional("created_at", &createdAt),
 		named.Optional("completed_at", &completedAt),
@@ -100,6 +103,10 @@ func ReadBackupFromResultSet(res result.Result) (*types.Backup, error) {
 		return nil, err
 	}
 
+	if expireAt != nil {
+		expireAtTs = timestamppb.New(*expireAt)
+	}
+
 	return &types.Backup{
 		ID:               backupId,
 		ContainerID:      containerId,
@@ -113,6 +120,7 @@ func ReadBackupFromResultSet(res result.Result) (*types.Backup, error) {
 		Message:          StringOrEmpty(message),
 		AuditInfo:        auditFromDb(creator, createdAt, completedAt),
 		Size:             Int64OrZero(size),
+		ExpireAt:         expireAtTs,
 	}, nil
 }
 
@@ -238,6 +246,14 @@ func ReadOperationFromResultSet(res result.Result) (types.Operation, error) {
 			PathPrefix: pathPrefix,
 			UpdatedAt:  updatedTs,
 		}, nil
+	} else if operationType == string(types.OperationTypeDOB) {
+		return &types.DeleteOutdatedBackupsOperation{
+			ID:        operationId,
+			State:     operationState,
+			Message:   StringOrEmpty(message),
+			Audit:     auditFromDb(creator, createdAt, completedAt),
+			UpdatedAt: updatedTs,
+		}, nil
 	}
 
 	return &types.GenericOperation{ID: operationId}, nil

internal/connectors/db/yql/queries/read.go

@@ -20,7 +20,7 @@ var (
 		"initiated", "created_at", "completed_at",
 		"s3_endpoint", "s3_region", "s3_bucket",
 		"s3_path_prefix", "status", "paths", "message",
-		"size",
+		"size", "expire_at",
 	}
 	AllOperationFields = []string{
 		"id", "type", "container_id", "database", "endpoint", "backup_id",

internal/connectors/db/yql/queries/write.go

@@ -263,6 +263,10 @@ func BuildCreateBackupQuery(b types.Backup, index int) WriteSingleTableQueryImpl
 			d.AddValueParam("$completed_at", table_types.TimestampValueFromTime(b.AuditInfo.CompletedAt.AsTime()))
 		}
 	}
+
+	if b.ExpireAt != nil {
+		d.AddValueParam("$expire_at", table_types.TimestampValueFromTime(b.ExpireAt.AsTime()))
+	}
 	return d
 }
 

internal/server/services/backup/backup_service.go

@@ -217,6 +217,7 @@ func (s *BackupService) MakeBackup(ctx context.Context, req *pb.MakeBackupReques
 		S3Bucket:         s.s3.Bucket,
 		S3PathPrefix:     destinationPrefix,
 		Status:           types.BackupStateRunning,
+		ExpireAt:         nil, // TODO: set ExpireAt based on user backup ttl
 		AuditInfo: &pb.AuditInfo{
 			CreatedAt: now,
 			Creator:   subject,

internal/types/backup.go

@@ -2,6 +2,7 @@ package types
 
 import (
 	"fmt"
+	"google.golang.org/protobuf/types/known/timestamppb"
 
 	pb "ydbcp/pkg/proto/ydbcp/v1alpha1"
 
@@ -36,6 +37,7 @@ type Backup struct {
 	Message          string
 	AuditInfo        *pb.AuditInfo
 	Size             int64
+	ExpireAt         *timestamppb.Timestamp
 }
 
 func (o *Backup) String() string {
@@ -65,7 +67,7 @@ func (o *Backup) Proto() *pb.Backup {
 		Size:     o.Size,
 		Status:   pb.Backup_Status(pb.Backup_Status_value[o.Status]),
 		Message:  o.Message,
-		ExpireAt: nil,
+		ExpireAt: o.ExpireAt,
 	}
 }
 

internal/types/operation.go

@@ -254,6 +254,73 @@ func (o *DeleteBackupOperation) Proto() *pb.Operation {
 	}
 }
 
+type DeleteOutdatedBackupsOperation struct {
+	ID        string
+	State     OperationState
+	Message   string
+	Audit     *pb.AuditInfo
+	UpdatedAt *timestamppb.Timestamp
+}
+
+func (o *DeleteOutdatedBackupsOperation) GetID() string {
+	return o.ID
+}
+func (o *DeleteOutdatedBackupsOperation) SetID(id string) {
+	o.ID = id
+}
+func (o *DeleteOutdatedBackupsOperation) GetContainerID() string {
+	return ""
+}
+func (o *DeleteOutdatedBackupsOperation) GetType() OperationType {
+	return OperationTypeDOB
+}
+func (o *DeleteOutdatedBackupsOperation) SetType(_ OperationType) {
+}
+func (o *DeleteOutdatedBackupsOperation) GetState() OperationState {
+	return o.State
+}
+func (o *DeleteOutdatedBackupsOperation) SetState(s OperationState) {
+	o.State = s
+}
+func (o *DeleteOutdatedBackupsOperation) GetMessage() string {
+	return o.Message
+}
+func (o *DeleteOutdatedBackupsOperation) SetMessage(m string) {
+	o.Message = m
+}
+func (o *DeleteOutdatedBackupsOperation) GetAudit() *pb.AuditInfo {
+	return o.Audit
+}
+func (o *DeleteOutdatedBackupsOperation) GetUpdatedAt() *timestamppb.Timestamp {
+	return o.UpdatedAt
+}
+func (o *DeleteOutdatedBackupsOperation) SetUpdatedAt(t *timestamppb.Timestamp) {
+	o.UpdatedAt = t
+}
+func (o *DeleteOutdatedBackupsOperation) Copy() Operation {
+	copy := *o
+	return &copy
+}
+
+func (o *DeleteOutdatedBackupsOperation) Proto() *pb.Operation {
+	return &pb.Operation{
+		Id:                   o.ID,
+		ContainerId:          "",
+		Type:                 string(OperationTypeDOB),
+		DatabaseName:         "",
+		DatabaseEndpoint:     "",
+		YdbServerOperationId: "",
+		BackupId:             "",
+		SourcePaths:          nil,
+		SourcePathsToExclude: nil,
+		RestorePaths:         nil,
+		Audit:                o.Audit,
+		Status:               o.State.Enum(),
+		Message:              o.Message,
+		UpdatedAt:            o.UpdatedAt,
+	}
+}
+
 type GenericOperation struct {
 	ID          string
 	ContainerID string
@@ -333,6 +400,7 @@ const (
 	OperationTypeTB       = OperationType("TB")
 	OperationTypeRB       = OperationType("RB")
 	OperationTypeDB       = OperationType("DB")
+	OperationTypeDOB      = OperationType("DOB")
 	BackupTimestampFormat = "20060102_150405"
 	S3ForcePathStyle      = true
 )