sync mirror component

OpenCSGs · Dec 18, 2024 · 67007de · 67007de
1 parent 001c56c
commit 67007de
Show file tree

Hide file tree

Showing 7 changed files with 534 additions and 46 deletions.
diff --git a/_mocks/opencsg.com/csghub-server/builder/store/database/mock_MirrorStore.go b/_mocks/opencsg.com/csghub-server/builder/store/database/mock_MirrorStore.go
diff --git a/api/handler/mirror.go b/api/handler/mirror.go
@@ -123,7 +123,8 @@ func (h *MirrorHandler) Index(ctx *gin.Context) {
 		httpbase.UnauthorizedError(ctx, component.ErrUserNotFound)
 		return
 	}
-	repos, total, err := h.mc.Index(ctx, currentUser, per, page)
+	search := ctx.Query("search")
+	repos, total, err := h.mc.Index(ctx, currentUser, per, page, search)
 	if err != nil {
 		slog.Error("failed to get mirror repos", slog.Any("error", err))
 		httpbase.ServerError(ctx, err)

diff --git a/builder/store/database/mirror.go b/builder/store/database/mirror.go
@@ -3,6 +3,7 @@ package database
 import (
 	"context"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/uptrace/bun"
@@ -31,7 +32,8 @@ type MirrorStore interface {
 	Finished(ctx context.Context) ([]Mirror, error)
 	ToSyncRepo(ctx context.Context) ([]Mirror, error)
 	ToSyncLfs(ctx context.Context) ([]Mirror, error)
-	IndexWithPagination(ctx context.Context, per, page int) (mirrors []Mirror, count int, err error)
+	IndexWithPagination(ctx context.Context, per, page int, search string) (mirrors []Mirror, count int, err error)
+	StatusCount(ctx context.Context) ([]MirrorStatusCount, error)
 	UpdateMirrorAndRepository(ctx context.Context, mirror *Mirror, repo *Repository) error
 }
 
@@ -76,6 +78,11 @@ type Mirror struct {
 	times
 }
 
+type MirrorStatusCount struct {
+	Status types.MirrorTaskStatus `bun:"status"`
+	Count  int                    `bun:"count"`
+}
+
 func (s *mirrorStoreImpl) IsExist(ctx context.Context, repoID int64) (exists bool, err error) {
 	var mirror Mirror
 	exists, err = s.db.Operator.Core.
@@ -90,7 +97,8 @@ func (s *mirrorStoreImpl) IsRepoExist(ctx context.Context, repoType types.Reposi
 	exists, err = s.db.Operator.Core.
 		NewSelect().
 		Model(&repo).
-		Where("git_path=?", fmt.Sprintf("%ss_%s/%s", repoType, namespace, name)).
+		Where("path=?", fmt.Sprintf("%s/%s", namespace, name)).
+		Where("repository_type=?", repoType).
 		Exists(ctx)
 	return
 }
@@ -273,7 +281,13 @@ func (s *mirrorStoreImpl) ToSyncRepo(ctx context.Context) ([]Mirror, error) {
 	var mirrors []Mirror
 	err := s.db.Operator.Core.NewSelect().
 		Model(&mirrors).
-		Where("next_execution_timestamp < ? or status in (?,?,?)", time.Now(), types.MirrorIncomplete, types.MirrorFailed, types.MirrorWaiting).
+		Where(
+			"next_execution_timestamp < ? or status in (?,?,?,?)",
+			time.Now(),
+			types.MirrorIncomplete,
+			types.MirrorFailed,
+			types.MirrorWaiting,
+			types.MirrorRunning).
 		Scan(ctx)
 	if err != nil {
 		return nil, err
@@ -293,11 +307,17 @@ func (s *mirrorStoreImpl) ToSyncLfs(ctx context.Context) ([]Mirror, error) {
 	return mirrors, nil
 }
 
-func (s *mirrorStoreImpl) IndexWithPagination(ctx context.Context, per, page int) (mirrors []Mirror, count int, err error) {
+func (s *mirrorStoreImpl) IndexWithPagination(ctx context.Context, per, page int, search string) (mirrors []Mirror, count int, err error) {
 	q := s.db.Operator.Core.NewSelect().
 		Model(&mirrors).
 		Relation("Repository").
 		Relation("MirrorSource")
+	if search != "" {
+		q = q.Where("LOWER(mirror.source_url) like ? or LOWER(mirror.local_repo_path) like ?",
+			fmt.Sprintf("%%%s%%", strings.ToLower(search)),
+			fmt.Sprintf("%%%s%%", strings.ToLower(search)),
+		)
+	}
 	count, err = q.Count(ctx)
 	if err != nil {
 		return
@@ -313,6 +333,17 @@ func (s *mirrorStoreImpl) IndexWithPagination(ctx context.Context, per, page int
 	return
 }
 
+func (s *mirrorStoreImpl) StatusCount(ctx context.Context) ([]MirrorStatusCount, error) {
+	var statusCounts []MirrorStatusCount
+	err := s.db.Operator.Core.NewSelect().
+		Model((*Mirror)(nil)).
+		Column("status").
+		ColumnExpr("COUNT(*) AS count").
+		Group("status").
+		Scan(ctx, &statusCounts)
+	return statusCounts, err
+}
+
 func (s *mirrorStoreImpl) UpdateMirrorAndRepository(ctx context.Context, mirror *Mirror, repo *Repository) error {
 	err := s.db.Operator.Core.RunInTx(ctx, nil, func(ctx context.Context, tx bun.Tx) error {
 		_, err := tx.NewUpdate().Model(mirror).WherePK().Exec(ctx)

diff --git a/builder/store/database/mirror_test.go b/builder/store/database/mirror_test.go
@@ -51,6 +51,7 @@ func TestMirrorStore_CRUD(t *testing.T) {
 		RepositoryType: types.ModelRepo,
 		GitPath:        "models_ns/n",
 		Name:           "repo",
+		Path:           "ns/n",
 	}
 	err = db.Core.NewInsert().Model(repo).Scan(ctx, repo)
 	require.Nil(t, err)
@@ -194,7 +195,7 @@ func TestMirrorStore_ToSync(t *testing.T) {
 	for _, m := range ms {
 		names = append(names, m.Interval)
 	}
-	require.ElementsMatch(t, []string{"m1", "m3", "m6", "m7"}, names)
+	require.ElementsMatch(t, []string{"m1", "m3", "m5", "m6", "m7"}, names)
 
 	ms, err = store.ToSyncLfs(ctx)
 	require.Nil(t, err)
@@ -222,7 +223,7 @@ func TestMirrorStore_IndexWithPagination(t *testing.T) {
 		require.Nil(t, err)
 	}
 
-	ms, count, err := store.IndexWithPagination(ctx, 10, 1)
+	ms, count, err := store.IndexWithPagination(ctx, 10, 1, "foo")
 	require.Nil(t, err)
 	names := []string{}
 	for _, m := range ms {
@@ -250,4 +251,12 @@ func TestMirrorStore_StatusCount(t *testing.T) {
 		require.Nil(t, err)
 	}
 
+	cs, err := store.StatusCount(ctx)
+	require.Nil(t, err)
+	require.Equal(t, 2, len(cs))
+	require.ElementsMatch(t, []database.MirrorStatusCount{
+		{types.MirrorFailed, 2},
+		{types.MirrorFinished, 1},
+	}, cs)
+
 }
diff --git a/common/types/mirror.go b/common/types/mirror.go
@@ -135,3 +135,8 @@ type Mirror struct {
 type MirrorSource struct {
 	SourceName string `json:"source_name"`
 }
+
+type MirrorStatusCount struct {
+	Status MirrorTaskStatus
+	Count  int
+}