add LastSuccessfulBackupId retrieval on listing schedules

.env

@@ -3,4 +3,4 @@ S3_ACCESS_KEY = "ydbcp"
 S3_SECRET_KEY = "password"
 S3_BUCKET = "test-bucket"
 S3_REGION = "us-east-1"
-YDB_NAME = "local-ydb"
+YDB_NAME = "local-ydb"

.github/workflows/unit-test.yml

@@ -50,18 +50,6 @@ jobs:
         with:
           create-pdf: false
 
-  comment-pr:
-    runs-on: ubuntu-latest
-    needs: unittest
-    steps:
-      - name: Comment on PR
-        uses: peter-evans/create-or-update-comment@v2
-        with:
-          token: ${{ secrets.GITHUB_TOKEN }}
-          repository: ${{ github.repository }}
-          issue-number: ${{ github.event.pull_request.number }}
-          body: ${{ needs.unittest.outputs.job-summary }}
-
   integration-test:
     runs-on: ubuntu-latest
     steps:
@@ -73,8 +61,16 @@ jobs:
       - name: docker compose up
         run: |
           docker compose up --build -d
-      - name: run integration tests
-        run: docker exec local-ydbcp sh -c './integration'
+      - name: run make_backup tests
+        run: docker exec local-ydbcp sh -c './make_backup'
+      - name: docker compose down
+        run: |
+          docker compose down
+      - name: docker compose up
+        run: |
+          docker compose up -d
+      - name: run list_schedules test
+        run: docker exec local-ydbcp sh -c './list_schedules'
       - name: docker compose down
         run: |
           docker compose down

cmd/integration/list_schedules/main.go

@@ -0,0 +1,304 @@
+package main
+
+import (
+	"context"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/durationpb"
+	"google.golang.org/protobuf/types/known/timestamppb"
+	"log"
+	"time"
+	"ydbcp/internal/config"
+	"ydbcp/internal/connectors/db"
+	"ydbcp/internal/connectors/db/yql/queries"
+
+	"ydbcp/internal/types"
+	pb "ydbcp/pkg/proto/ydbcp/v1alpha1"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+const (
+	containerID      = "abcde"
+	databaseName     = "/local"
+	ydbcpEndpoint    = "localhost:50051"
+	databaseEndpoint = "grpcs://local-ydb:2135"
+	connectionString = "grpcs://local-ydb:2135/local"
+)
+
+var (
+	threePM = time.Date(2024, 01, 01, 15, 0, 0, 0, time.UTC)
+	fourPM  = time.Date(2024, 01, 01, 16, 0, 0, 0, time.UTC)
+	fivePM  = time.Date(2024, 01, 01, 17, 0, 0, 0, time.UTC)
+)
+
+func BackupsToInsert() ([]types.Backup, error) {
+	audit1 := &pb.AuditInfo{
+		Creator:     containerID,
+		CreatedAt:   timestamppb.New(threePM),
+		CompletedAt: timestamppb.New(fourPM),
+	}
+	audit2 := &pb.AuditInfo{
+		Creator:     containerID,
+		CreatedAt:   timestamppb.New(fourPM),
+		CompletedAt: timestamppb.New(fivePM),
+	}
+	scheduleId1 := "1"
+	scheduleId2 := "2"
+	scheduleId3 := "3"
+	scheduleId4 := "4"
+	return []types.Backup{
+		{
+			ID:               "1",
+			ContainerID:      containerID,
+			DatabaseName:     databaseName,
+			DatabaseEndpoint: databaseEndpoint,
+			Status:           types.BackupStateAvailable,
+			Message:          "for schedule 1",
+			AuditInfo:        audit1,
+			Size:             10,
+			ScheduleID:       &scheduleId1,
+		},
+		{
+			ID:               "2",
+			ContainerID:      containerID,
+			DatabaseName:     databaseName,
+			DatabaseEndpoint: databaseEndpoint,
+			Status:           types.BackupStateAvailable,
+			Message:          "for schedule 1",
+			AuditInfo:        audit2,
+			Size:             10,
+			ScheduleID:       &scheduleId1,
+		},
+		{
+			ID:               "3",
+			ContainerID:      containerID,
+			DatabaseName:     databaseName,
+			DatabaseEndpoint: databaseEndpoint,
+			Status:           types.BackupStateAvailable,
+			Message:          "not for schedule",
+			AuditInfo:        audit2,
+			Size:             10,
+			ScheduleID:       nil,
+		},
+		{
+			ID:               "4",
+			ContainerID:      containerID,
+			DatabaseName:     databaseName,
+			DatabaseEndpoint: databaseEndpoint,
+			Status:           types.BackupStateAvailable,
+			Message:          "for schedule 2",
+			AuditInfo:        audit1,
+			Size:             10,
+			ScheduleID:       &scheduleId2,
+		},
+		{
+			ID:               "5",
+			ContainerID:      containerID,
+			DatabaseName:     databaseName,
+			DatabaseEndpoint: databaseEndpoint,
+			Status:           types.BackupStateDeleted,
+			Message:          "for schedule 2",
+			AuditInfo:        audit2,
+			Size:             10,
+			ScheduleID:       &scheduleId2,
+		},
+		{
+			ID:               "6",
+			ContainerID:      containerID,
+			DatabaseName:     databaseName,
+			DatabaseEndpoint: databaseEndpoint,
+			Status:           types.BackupStateAvailable,
+			Message:          "for schedule 3",
+			AuditInfo:        audit1,
+			Size:             10,
+			ScheduleID:       &scheduleId3,
+		},
+		{
+			ID:               "7",
+			ContainerID:      containerID,
+			DatabaseName:     databaseName,
+			DatabaseEndpoint: databaseEndpoint,
+			Status:           types.BackupStateDeleted,
+			Message:          "for schedule 3",
+			AuditInfo:        audit2,
+			Size:             10,
+			ScheduleID:       &scheduleId3,
+		},
+		{
+			ID:               "8",
+			ContainerID:      containerID,
+			DatabaseName:     databaseName,
+			DatabaseEndpoint: databaseEndpoint,
+			Status:           types.BackupStateDeleted,
+			Message:          "for schedule 4",
+			AuditInfo:        audit2,
+			Size:             10,
+			ScheduleID:       &scheduleId4,
+		},
+	}, nil
+}
+
+func SchedulesToInsert() []types.BackupSchedule {
+	return []types.BackupSchedule{
+		{
+			ID:               "1",
+			ContainerID:      containerID,
+			DatabaseName:     databaseName,
+			DatabaseEndpoint: databaseEndpoint,
+			Name:             "schedule 1",
+			Active:           true,
+			ScheduleSettings: &pb.BackupScheduleSettings{
+				SchedulePattern:        &pb.BackupSchedulePattern{Crontab: "* * * * * *"},
+				RecoveryPointObjective: durationpb.New(time.Hour),
+			},
+		},
+		{
+			ID:               "2",
+			ContainerID:      containerID,
+			DatabaseName:     databaseName,
+			DatabaseEndpoint: databaseEndpoint,
+			Name:             "schedule 2",
+			Active:           true,
+			ScheduleSettings: &pb.BackupScheduleSettings{
+				SchedulePattern:        &pb.BackupSchedulePattern{Crontab: "* * * * * *"},
+				RecoveryPointObjective: durationpb.New(time.Minute * 15),
+			},
+		},
+		{
+			ID:               "3",
+			ContainerID:      containerID,
+			DatabaseName:     databaseName,
+			DatabaseEndpoint: databaseEndpoint,
+			Name:             "schedule 3",
+			Active:           true,
+			ScheduleSettings: &pb.BackupScheduleSettings{
+				SchedulePattern: &pb.BackupSchedulePattern{Crontab: "* * * * * *"},
+			},
+		},
+		{
+			ID:               "4",
+			ContainerID:      containerID,
+			DatabaseName:     databaseName,
+			DatabaseEndpoint: databaseEndpoint,
+			Name:             "schedule 4",
+			Active:           true,
+			ScheduleSettings: &pb.BackupScheduleSettings{
+				SchedulePattern:        &pb.BackupSchedulePattern{Crontab: "* * * * * *"},
+				RecoveryPointObjective: durationpb.New(time.Minute * 15),
+			},
+		},
+	}
+}
+
+func main() {
+	ctx := context.Background()
+	var opts []grpc.DialOption
+	opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	conn, err := grpc.NewClient(ydbcpEndpoint, opts...)
+	if err != nil {
+		log.Panicln("failed to dial")
+	}
+	defer func(conn *grpc.ClientConn) {
+		err := conn.Close()
+		if err != nil {
+			log.Panicln("failed to close connection")
+		}
+	}(conn)
+	ydbConn, err := db.NewYdbConnector(
+		ctx,
+		config.YDBConnectionConfig{
+			ConnectionString:   connectionString,
+			Insecure:           true,
+			Discovery:          false,
+			DialTimeoutSeconds: 10,
+		},
+	)
+	if err != nil {
+		log.Panicf("failed to create ydb connector: %v", err)
+	}
+	backups, err := BackupsToInsert()
+	if err != nil {
+		log.Panicf("failed to create backups to insert: %v", err)
+	}
+	for _, b := range backups {
+		err = ydbConn.ExecuteUpsert(ctx, queries.NewWriteTableQuery().WithCreateBackup(b))
+		if err != nil {
+			log.Panicf("failed to insert backup: %v", err)
+		}
+	}
+
+	schedulesToInsert := SchedulesToInsert()
+	for _, s := range schedulesToInsert {
+		err = ydbConn.ExecuteUpsert(ctx, queries.NewWriteTableQuery().WithCreateBackupSchedule(s))
+		if err != nil {
+			log.Panicf("failed to insert schedule: %v", err)
+		}
+	}
+
+	scheduleClient := pb.NewBackupScheduleServiceClient(conn)
+	schedules, err := scheduleClient.ListBackupSchedules(
+		context.Background(), &pb.ListBackupSchedulesRequest{
+			ContainerId:      containerID,
+			DatabaseNameMask: "%",
+		},
+	)
+	if err != nil {
+		log.Panicf("failed to list backup schedules: %v", err)
+	}
+
+	if len(schedules.Schedules) != 4 {
+		log.Panicln("did not get all the schedules")
+	}
+	for _, s := range schedules.Schedules {
+		switch s.Id {
+		case "1":
+			{
+				if s.LastSuccessfulBackupInfo.BackupId != "2" || s.LastSuccessfulBackupInfo.RecoveryPoint.AsTime() != fivePM {
+					log.Panicf(
+						"Expected BackupID = 2, RecoveryPoint = %s, got %s for scheduleID %s", fivePM.String(),
+						s.LastSuccessfulBackupInfo.String(),
+						s.Id,
+					)
+				}
+			}
+		case "2":
+			{
+				if s.LastSuccessfulBackupInfo.BackupId != "4" || s.LastSuccessfulBackupInfo.RecoveryPoint.AsTime() != fourPM {
+					log.Panicf(
+						"Expected BackupID = 4, RecoveryPoint = %s, got %s for scheduleID %s", fourPM.String(),
+						s.LastSuccessfulBackupInfo.String(),
+						s.Id,
+					)
+
+				}
+			}
+		case "3":
+			{
+				info := &pb.ScheduledBackupInfo{
+					BackupId:      "6",
+					RecoveryPoint: timestamppb.New(fourPM),
+				}
+				if !proto.Equal(info, s.LastSuccessfulBackupInfo) {
+					log.Panicf(
+						"Expected %s, got %s for scheduleID %s", info.String(), s.LastSuccessfulBackupInfo.String(),
+						s.Id,
+					)
+				}
+			}
+		case "4":
+			{
+				if s.LastSuccessfulBackupInfo != nil {
+					log.Panicf(
+						"Expected nil, got %s for scheduleID %s", s.LastSuccessfulBackupInfo.String(),
+						s.Id,
+					)
+				}
+			}
+		default:
+			{
+				log.Panicf("unexpected schedule id: %s", s.Id)
+			}
+		}
+	}
+}

dockerfile

@@ -17,7 +17,10 @@ RUN go mod download
 RUN go build -o . ./cmd/ydbcp/main.go
 
 # Build integration test app
-RUN go build -o ./integration ./cmd/integration/main.go
+RUN go build -o ./make_backup ./cmd/integration/make_backup/main.go
+
+# Build integration test app
+RUN go build -o ./list_schedules ./cmd/integration/list_schedules/main.go
 
 # Command to run the executable
 CMD ["./main", "--config=local_config.yaml"]

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/gorhill/cronexpr v0.0.0-20180427100037-88b0669f7d75
 	github.com/jonboulle/clockwork v0.3.0
 	github.com/stretchr/testify v1.8.1
+	github.com/undefinedlabs/go-mpatch v1.0.7
 	github.com/ydb-platform/ydb-go-genproto v0.0.0-20240821162910-6cb364b2ccc8
 	github.com/ydb-platform/ydb-go-sdk/v3 v3.75.2
 	go.uber.org/automaxprocs v1.5.3

go.sum

@@ -82,6 +82,8 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+github.com/undefinedlabs/go-mpatch v1.0.7 h1:943FMskd9oqfbZV0qRVKOUsXQhTLXL0bQTVbQSpzmBs=
+github.com/undefinedlabs/go-mpatch v1.0.7/go.mod h1:TyJZDQ/5AgyN7FSLiBJ8RO9u2c6wbtRvK827b6AVqY4=
 github.com/ydb-platform/ydb-go-genproto v0.0.0-20240821162910-6cb364b2ccc8 h1:ZWxYw6L51aNAMLbTpC/VbXP0rcnvsCAJqx7EI/CjWmc=
 github.com/ydb-platform/ydb-go-genproto v0.0.0-20240821162910-6cb364b2ccc8/go.mod h1:Er+FePu1dNUieD+XTMDduGpQuCPssK5Q4BjF+IIXJ3I=
 github.com/ydb-platform/ydb-go-sdk/v3 v3.75.2 h1:thrbvktqKA6LFZTnZrGuQi8LQVel1J2dDfoQFsgvcYs=

internal/connectors/db/mock.go

@@ -195,6 +195,18 @@ func (c *MockDBConnector) GetBackup(
 	return types.Backup{}, fmt.Errorf("backup not found, id %s", backupID)
 }
 
+func (c *MockDBConnector) GetSchedule(
+	_ context.Context, scheduleID string,
+) (types.BackupSchedule, error) {
+	c.guard.Lock()
+	defer c.guard.Unlock()
+
+	if schedule, exist := c.backupSchedules[scheduleID]; exist {
+		return schedule, nil
+	}
+	return types.BackupSchedule{}, fmt.Errorf("backupSchedule not found, id %s", scheduleID)
+}
+
 func (c *MockDBConnector) SelectOperations(
 	_ context.Context, _ queries.ReadTableQuery,
 ) ([]types.Operation, error) {