add cache preheat

cron/cron.go

@@ -53,8 +53,8 @@ func Start(ctx context.Context) {
 		errC <- err
 	} else {
 		tx.Commit()
+		slog.Info("init task completed", "time", time.Since(startInit).String())
 	}
-	slog.Info("init task completed", "time", time.Since(startInit).String())
 
 	c := cron.New()
 	StartCron(ctx, c, errC)
@@ -70,10 +70,14 @@ func Start(ctx context.Context) {
 func Restart(ctx context.Context) {
 	slog.Info("openalysis service restarted")
 
-	// 1. cron add func error
-	errC := make(chan error, 1)
+	// 1. cache preheat
+	// 2. cron add func error
+	errC := make(chan error, 2)
 
-	// TODO: cache preheating
+	// if cache preheat failed, stop service
+	if err := CachePreheat(ctx, storage.DB); err != nil {
+		errC <- err
+	}
 
 	c := cron.New()
 	StartCron(ctx, c, errC)
@@ -132,7 +136,16 @@ func StartCron(ctx context.Context, c *cron.Cron, errC chan error) {
 }
 
 // map[orgNodeID][]repoNameWithOwner
-var cache map[string][]string
+var cache = make(map[string][]string)
+
+func CachePreheat(ctx context.Context, db *gorm.DB) error {
+	orgRepos, err := storage.QueryOrgRepos(ctx, db)
+	if err != nil {
+		return err
+	}
+	cache = orgRepos
+	return nil
+}
 
 type Count struct {
 	IssueCount       int
@@ -142,8 +155,6 @@ type Count struct {
 }
 
 func InitTask(ctx context.Context, db *gorm.DB) error {
-	// init cache
-	cache = make(map[string][]string)
 	// handle groups
 	for _, group := range config.GlobalConfig.Groups {
 		var groupCount Count

cron/cron_test.go

@@ -17,8 +17,11 @@ package cron
 import (
 	"context"
 	"errors"
+	"fmt"
 	"log"
 	"log/slog"
+	"reflect"
+	"sort"
 	"testing"
 	"time"
 
@@ -48,7 +51,6 @@ func TestUpdateTask(t *testing.T) {
 	graphql.Init()
 	rest.Init()
 
-	cache = make(map[string][]string)
 	for _, group := range config.GlobalConfig.Groups {
 		for _, login := range group.Orgs {
 			org, err := graphql.QueryOrgInfo(context.Background(), login)
@@ -74,7 +76,6 @@ func TestRestart(t *testing.T) {
 	graphql.Init()
 	rest.Init()
 
-	cache = make(map[string][]string)
 	for _, group := range config.GlobalConfig.Groups {
 		for _, login := range group.Orgs {
 			org, err := graphql.QueryOrgInfo(context.Background(), login)
@@ -157,3 +158,73 @@ func TestTransaction(t *testing.T) {
 		time.Sleep(time.Second * 1)
 	}
 }
+
+func TestCachePreheat(t *testing.T) {
+	config.GlobalConfig.ReadInConfig("../default.yaml")
+	storage.Init()
+	graphql.Init()
+	rest.Init()
+
+	latest := make(map[string][]string)
+	for _, group := range config.GlobalConfig.Groups {
+		for _, login := range group.Orgs {
+			org, err := graphql.QueryOrgInfo(context.Background(), login)
+			if err != nil {
+				t.Fatal(err)
+			}
+			repos, err := graphql.QueryRepoNameByOrg(context.Background(), login)
+			if err != nil {
+				t.Fatal(err)
+			}
+			latest[org.ID] = repos
+		}
+	}
+	err := CachePreheat(context.Background(), storage.DB)
+	if err != nil {
+		t.Fatal(err)
+	}
+	//fmt.Println("latest:", latest)
+	for k, v := range latest {
+		fmt.Println(k)
+		fmt.Println(len(v))
+	}
+	//fmt.Println("cache:", cache)
+	fmt.Println()
+	for k, v := range cache {
+		fmt.Println(k)
+		fmt.Println(len(v))
+	}
+	fmt.Println(compareMaps(latest, cache))
+}
+
+func compareMaps(map1, map2 map[string][]string) bool {
+	if len(map1) != len(map2) {
+		return false
+	}
+
+	for key, value1 := range map1 {
+		if value2, ok := map2[key]; !ok {
+			return false
+		} else {
+			sort.Strings(value1)
+			sort.Strings(value2)
+			if !reflect.DeepEqual(value1, value2) {
+				return false
+			}
+		}
+	}
+
+	return true
+}
+
+func TestMap(t *testing.T) {
+	map1 := map[string][]string{
+		"hello": []string{"1", "3", "5"},
+		"world": []string{"2", "4", "6"},
+	}
+	map2 := map[string][]string{
+		"hello": []string{"3", "1", "5"},
+		"world": []string{"2", "4", "6"},
+	}
+	fmt.Println(compareMaps(map1, map2))
+}

storage/storage.go

@@ -497,3 +497,24 @@ func UpdateCursor(ctx context.Context, db *gorm.DB, cursor *model.Cursor) error
 func DeleteCursor(ctx context.Context, db *gorm.DB, repoNodeID string) error {
 	return db.WithContext(ctx).Where("repo_node_id = ?", repoNodeID).Delete(&model.Cursor{}).Error
 }
+
+func QueryOrgRepos(ctx context.Context, db *gorm.DB) (map[string][]string, error) {
+	var groupsOrgs []model.GroupsOrganizations
+	if err := db.WithContext(ctx).Find(&groupsOrgs).Error; err != nil {
+		return nil, err
+	}
+
+	orgRepos := make(map[string][]string)
+	for _, groupOrg := range groupsOrgs {
+		var repos []model.Repository
+		if err := db.WithContext(ctx).Where("owner_node_id = ?", groupOrg.OrgNodeID).Group("node_id").Find(&repos).Error; err != nil {
+			return nil, err
+		}
+
+		for _, repo := range repos {
+			orgRepos[groupOrg.OrgNodeID] = append(orgRepos[groupOrg.OrgNodeID], util.MergeNameWithOwner(repo.Owner, repo.Name))
+		}
+	}
+
+	return orgRepos, nil
+}

util/util.go

@@ -16,6 +16,7 @@ package util
 
 import (
 	"errors"
+	"fmt"
 	"log/slog"
 	"os"
 	"os/signal"
@@ -33,6 +34,10 @@ func SplitNameWithOwner(s string) (string, string) {
 	return parts[0], parts[1]
 }
 
+func MergeNameWithOwner(owner, name string) string {
+	return fmt.Sprintf("%s/%s", owner, name)
+}
+
 func AssembleDSN(host, port, user, password, database string) string {
 	var sb strings.Builder
 	sb.WriteString(user)

util/util_test.go

@@ -19,6 +19,15 @@ import (
 	"testing"
 )
 
+func TestNameWithOwner(t *testing.T) {
+	nameWithOwner := "cloudwego/hertz"
+	owner, name := SplitNameWithOwner(nameWithOwner)
+	fmt.Println("owner:", owner)
+	fmt.Println("name:", name)
+	res := MergeNameWithOwner(owner, name)
+	fmt.Println("res:", res)
+}
+
 func TestIsEmptySlice(t *testing.T) {
 	sli1 := make([]int, 0)
 	var sli2 []int