fix: make streams and nobori fetching work

Author: madonuko

db/db.go

@@ -2,11 +2,11 @@ package db
 
 import (
 	"os"
+	"strings"
 	"time"
 
-	"github.com/google/uuid"
 	"github.com/terrapkg/gura/util"
-	"gorm.io/datatypes"
+	dt "gorm.io/datatypes"
 	"gorm.io/driver/postgres"
 	"gorm.io/gorm"
 	"moul.io/zapgorm2"
@@ -28,18 +28,44 @@ const (
 
 // Streams are grouped packages with the same upstream
 type Stream struct {
-	ID      uuid.UUID `gorm:"type:uuid;default:gen_random_uuid();primaryKey" json:"id"`
+	ID      dt.UUID   `gorm:"type:uuid;default:gen_random_uuid();primaryKey" json:"id"`
 	LastChk time.Time `json:"last_chk"`
 	LastUpd time.Time `json:"last_upd"`
 	Fetch   string    `json:"fetch"`
 	Forge   ForgeType `json:"forge"`
 	Ver     string    `json:"ver"`
-	Mirrors string    `json:"mirrors"` // comma-separated list of mirrors
+
+	Mirrors []StreamMirror `json:"mirrors"` // comma-separated list of mirrors
+}
+
+func (s *Stream) BeforeCreate(tx *gorm.DB) (err error) {
+	if s.ID.IsEmpty() {
+		s.ID = dt.NewUUIDv4()
+	}
+	return nil
+}
+
+type StreamMirror struct {
+	ID       dt.UUID `gorm:"type:uuid" json:"id"`
+	StreamID dt.UUID `gorm:"type:uuid" json:"streamid"`
+	Stream   Stream  `json:"-"`
+	Mirror   string  `json:"mirror"` // url without `http(s)://` and trailing slash
+}
+
+func (m *StreamMirror) BeforeCreate(tx *gorm.DB) error {
+	if m.ID.IsEmpty() {
+		m.ID = dt.NewUUIDv4()
+	}
+	return nil
+}
+func (m *StreamMirror) BeforeSave(tx *gorm.DB) error {
+	m.Mirror = strings.TrimSuffix(m.Mirror, "/")
+	return nil
 }
 
 // Pkg is the package model. Uses UUID primary key instead of gorm.Model's uint.
 type Pkg struct {
-	ID        datatypes.UUID `gorm:"type:uuid;default:gen_random_uuid();primaryKey" json:"id"`
+	ID        dt.UUID        `gorm:"type:uuid;default:gen_random_uuid();primaryKey" json:"id"`
 	CreatedAt time.Time      `gorm:"autoCreateTime" json:"created_at"`
 	UpdatedAt time.Time      `gorm:"autoUpdateTime" json:"updated_at"`
 	DeletedAt gorm.DeletedAt `json:"deleted_at"`
@@ -53,10 +79,17 @@ type Pkg struct {
 	// Foreign key relationship referencing Repo.ID (string).
 	Repo Repo `gorm:"constraint:OnDelete:CASCADE;foreignKey:RepoID;references:ID" json:"-"`
 
-	Meta datatypes.JSON `gorm:"type:jsonb;default:'{}'" json:"meta"`
+	Meta dt.JSON `gorm:"type:jsonb;default:'{}'" json:"meta"`
 
-	StreamID *uuid.UUID `json:"stream_id"`
-	Stream   Stream     `json:"-"`
+	StreamID *dt.UUID `gorm:"type:uuid" json:"stream_id"`
+	Stream   Stream   `json:"-"`
+}
+
+func (s *Pkg) BeforeCreate(tx *gorm.DB) (err error) {
+	if s.ID.IsEmpty() {
+		s.ID = dt.NewUUIDv4()
+	}
+	return nil
 }
 
 // Repo represents a package repository. Its ID is a string.
@@ -73,6 +106,7 @@ var l = util.SetupLog("db")
 
 // SetupDB initializes the global DB connection and runs migrations.
 func SetupDB() {
+	l.Info("setting up db")
 	var err error
 	dsn := os.Getenv("GURA_DSN")
 	if dsn == "" {
@@ -85,5 +119,6 @@ func SetupDB() {
 	util.MaybeSuicide(l, "cannot open db", err)
 
 	// AutoMigrate in an order that respects foreign keys.
-	util.MaybeSuicide(l, "auto migrate failed", DB.AutoMigrate(&Repo{}, &Pkg{}, &Stream{}))
+	util.MaybeSuicide(l, "auto migrate failed", DB.AutoMigrate(&Repo{}, &Pkg{}, &Stream{}, &StreamMirror{}))
+	l.Info("db ready")
 }

db/pkg.go

@@ -44,15 +44,15 @@ func SubmitPkg(pkg *Pkg) (*Pkg, error) {
 	return pkg, nil
 }
 
-func (p *Pkg) Update() error {
-	if err := DB.Save(p).Error; err != nil {
+func (s *Pkg) Update() error {
+	if err := DB.Save(s).Error; err != nil {
 		return err
 	}
 	return nil
 }
 
-func (p *Pkg) Delete() error {
-	if err := DB.Delete(p).Error; err != nil {
+func (s *Pkg) Delete() error {
+	if err := DB.Delete(s).Error; err != nil {
 		return err
 	}
 	return nil

kudari/kudari.go

@@ -22,24 +22,25 @@ var l = util.SetupLog("kudari")
 const fetchRepoTimer = 3e10 // 30s
 
 func fetch(repo db.Repo) {
-	defer func() {
-		time.Sleep(fetchRepoTimer)
-		go fetch(repo)
-	}()
 	l.Info("fetching repository", zap.String("repoID", repo.ID))
 	switch repo.Type {
 	case db.Rpm:
 		rpmFetch(repo)
 	}
 	n, err := gorm.G[db.Repo](db.DB).Where("id = ?", repo.ID).Update(context.Background(), "upd_at", time.Now())
 	if util.Yeet(l, "error while updating upd_at", err, zap.String("repoID", repo.ID), zap.Error(err)) {
-		return
+		goto next
 	}
 	if n != 1 {
 		l.DPanic("bug: mut upd_at", zap.Int("n", n), zap.String("repoID", repo.ID))
-		return
+		goto next
 	}
 	l.Info("done fetching repository", zap.String("repoID", repo.ID))
+
+next:
+	time.Sleep(fetchRepoTimer)
+	go fetch(repo)
+
 }
 
 func FetchLoop() {

kudari/rpm.go

@@ -167,8 +167,7 @@ func rpmFetch(repo db.Repo) {
 	packages := util.MergeSortedDedup(allSlices, rpmCompare)
 
 	l.Info("processing packages", zap.String("repoID", repo.ID), zap.Int("count", len(packages)))
-	var newpkgs []*db.Pkg
-	updated, unchanged, lastIdx := 0, 0, 0
+	updated, unchanged, lastIdx, newpkgs := 0, 0, 0, 0
 	walked := make([]bool, len(pkgs))
 	tx := db.DB.Begin()
 	for _, p := range packages {
@@ -189,17 +188,18 @@ func rpmFetch(repo db.Repo) {
 			pkgs[n].FullVer = fullver
 			pkgs[n].Ver = p.Version.Ver
 			util.MaybeSuicide(l, "Meta.UnmarshalJSON", pkgs[n].Meta.UnmarshalJSON(rpm2MetaJSON(p)))
-			nobori.RegPkg(tx, &pkgs[n])
 			updated++
 		} else {
-			newpkgs = append(newpkgs, &db.Pkg{
+			p := &db.Pkg{
 				Name:    p.Name,
 				FullVer: rpmFullVer(p),
 				Ver:     p.Version.Ver,
 				Arch:    p.Arch,
 				RepoID:  repo.ID,
 				Meta:    rpm2MetaJSON(p),
-			})
+			}
+			util.Yeet(l, "cannot RegPkg", nobori.RegPkg(tx, p), zap.Any("pkg", p))
+			newpkgs++
 		}
 	}
 	var deletes []uuid.UUID
@@ -209,15 +209,12 @@ func rpmFetch(repo db.Repo) {
 		}
 	}
 	tx.Delete(&db.Pkg{}, "id IN (?)", deletes)
-	if newpkgs != nil {
-		tx.CreateInBatches(newpkgs, 5000)
-	}
 	tx.Commit()
 	l.Info("package update summary",
 		zap.String("repoID", repo.ID),
 		zap.Int("unchanged", unchanged),
 		zap.Int("updated", updated),
-		zap.Int("added", len(newpkgs)),
+		zap.Int("added", newpkgs),
 		zap.Int("deleted", len(deletes)),
 	)
 }

nobori/TRACE.go

@@ -23,6 +23,7 @@ func UpTrace(p db.Pkg, url_ch chan string) {
 	default:
 		l.DPanic("unreachable in uptrace")
 	}
+	close(url_ch)
 }
 
 func rpmTrace(p db.Pkg, url_ch chan string) {

nobori/github.go

@@ -23,10 +23,12 @@ import (
 	"go.uber.org/zap"
 )
 
+const GH_WARN_DUR = 500 // ms
+
 // ? https://docs.github.com/en/rest/using-the-rest-api/rate-limits-for-the-rest-api?apiVersion=2022-11-28#about-secondary-rate-limits
 // no more than 100 parallel requests
-var ghRtPool = make(chan db.Stream)
-var ghQlPool = make(chan db.Stream)
+var ghRtPool = make(chan db.Stream, 100)
+var ghQlPool = make(chan db.Stream, 100)
 var ghTokRt []GHToken // GitHub REST Tokens
 var ghTokQl []GHToken // GitHub GraphQL Tokens
 var ghRtTokIdx int = 0
@@ -73,9 +75,9 @@ func (token *GHToken) waitForFish() {
 // Update token quota and reset time from HTTP response headers
 func (token *GHToken) updTok(h http.Header) {
 	q, err := strconv.ParseInt(h.Get("x-ratelimit-remaining"), 10, 16)
-	util.MaybeSuicide(ghl, "strconv x-ratelimit-remaining", err)
+	util.MaybeSuicide(ghl, "strconv", err, zap.String("x-ratelimit-remaining", h.Get("x-ratelimit-remaining")))
 	r, err := strconv.ParseInt(h.Get("x-ratelimit-reset"), 10, 64)
-	util.MaybeSuicide(ghl, "strconv x-ratelimit-reset", err)
+	util.MaybeSuicide(ghl, "strconv", err, zap.String("x-ratelimit-reset", h.Get("x-ratelimit-reset")))
 	token.quota = int16(q)
 	token.reset = time.Unix(r, 0)
 }
@@ -161,18 +163,16 @@ func ghFillTokRt() *GHToken {
 //
 // Return the token with the highest quota.
 func ghFillTokQl() *GHToken {
-	return ghInitTokenPool(&ghTokQl, func(token string) (GHToken, error) {
-		var qlcli = ql.NewClient("https://api.github.com/graphql", http.DefaultClient).WithRequestModifier(
-			func(r *http.Request) {
-				r.Header.Add("Authorization", "Bearer "+token)
-			})
+	return ghInitTokenPool(&ghTokQl, func(token string) (tok GHToken, err error) {
+		tok.key = token
+		qlcli := tok.qlcli()
 		var q struct {
 			RateLimit struct {
 				Remaining int16
 				ResetAt   string
 			}
 		}
-		err := qlcli.Query(context.Background(), &q, nil)
+		err = qlcli.Query(context.Background(), &q, nil)
 		if err != nil {
 			return GHToken{}, xerrors.Newf("query RateLimit -> give up token %s: %w", token, err)
 		}
@@ -196,7 +196,7 @@ func GhSwim() chan struct{} {
 	ready := make(chan struct{}, 1)
 	go func() {
 		for len(ghTokRt) == 0 || len(ghTokQl) == 0 {
-			time.Sleep(20 * time.Millisecond)
+			time.Sleep(1 * time.Millisecond)
 		}
 		ready <- struct{}{}
 	}()
@@ -269,18 +269,36 @@ func ghRtReleaseCall(stream *db.Stream, repo string, token *GHToken) []string {
 		return []string{}
 	}
 	req.Header.Add("Authorization", "Bearer "+token.key)
+	l.Debug("ghRtReleaseCall", zap.String("repo", repo))
+	t0 := time.Now()
 	resp, err := http.DefaultClient.Do(req)
+	if t := time.Since(t0); t.Milliseconds() > GH_WARN_DUR {
+		l.Warn("took " + t.String())
+	}
 	if util.Yeet(ghl, "resp fail", err, zap.String("fetch", stream.Fetch)) {
 		return []string{}
 	}
+	if resp.StatusCode != http.StatusOK {
+		body, err := io.ReadAll(resp.Body)
+		if err != nil {
+			l.Error("can't read body", zap.String("fetch", stream.Fetch), zap.Error(err))
+			body = []byte{}
+		}
+		l.Error("bad status", zap.String("status", resp.Status), zap.String("fetch", stream.Fetch), zap.ByteString("body", body))
+		return []string{}
+	}
 	token.updTok(resp.Header)
 	buf, err := io.ReadAll(resp.Body)
 	util.MaybeSuicide(ghl, "can't read buf", err)
-	var v []struct {
-		tag_name string
+	type Resp struct {
+		Tag string `json:"tag_name"`
 	}
-	json.Unmarshal(buf, &v)
-	return util.SliceMap(v, func(r struct{ tag_name string }) string { return r.tag_name })
+
+	var v []Resp
+	if util.Yeet(l, "can't unmarshal", json.Unmarshal(buf, &v), zap.ByteString("resp", buf)) {
+		return []string{}
+	}
+	return util.SliceMap(v, func(r Resp) string { return r.Tag })
 }
 
 // Fetch latest GitHub release using the REST API
@@ -320,16 +338,22 @@ func ghQlTagCall(prefix, owner, name string, len int, qlcli ql.Client, token *GH
 			} `graphql:"refs(refPrefix: \"refs/tags/\", last: $len, orderBy: {field: TAG_COMMIT_DATE, direction: ASC}, query: $prefix)"`
 		} `graphql:"repository(owner: $owner, name: $name)"`
 	}
-	var headers http.Header
-	err := qlcli.Query(context.Background(), &q, map[string]any{
+	headers := http.Header{}
+	l.Debug("ghQlTagCall", zap.String("repo", owner+"/"+name))
+	t0 := time.Now()
+	err := qlcli.Query(context.TODO(), &q, map[string]any{
 		"prefix": prefix,
 		"owner":  owner,
 		"name":   name,
 		"len":    len,
 	}, ql.BindResponseHeaders(&headers))
+	if t := time.Since(t0); t.Milliseconds() > GH_WARN_DUR {
+		l.Warn("took " + t.String())
+	}
 	if util.Yeet(ghl, "ql_tag_call", err) {
 		return nil
 	}
+
 	tags := make([]string, 0, len)
 	for _, edge := range q.Repository.Refs.Edges {
 		tags = append(tags, edge.Node.Name)
@@ -353,7 +377,7 @@ func ghQlTag(stream *db.Stream, remain string, token *GHToken, qlcli ql.Client)
 	}
 }
 
-const GH_STRM_HDLR_TEST_QL_MAX = 100
+const GH_STRM_HDLR_TEST_QL_MAX = 20
 
 func GhStrmHdlr(pkg db.Pkg, url string) *db.Stream {
 	strm := &db.Stream{
@@ -377,7 +401,8 @@ func GhStrmHdlr(pkg db.Pkg, url string) *db.Stream {
 		l.Warn("Found releases, but cannot determine ver/prefix",
 			zap.Strings("releases", releases),
 			zap.String("repo", repo),
-			zap.String("pkgid", pkg.ID.String()))
+			zap.String("pkgid", pkg.ID.String()),
+			zap.String("pkgv", pkg.Ver))
 	}
 
 	tokenql := &ghTokQl[ghQlTokIdx]

nobori/nobori.go

@@ -58,6 +58,7 @@ func schedule(stream db.Stream) {
 //
 // Return when nobori is ready.
 func StartFetchLoop() {
+	l.Info("preparing fetch loop")
 	ready_chs := util.SliceMap(swimmers, func(swimmer func() chan struct{}) chan struct{} { return swimmer() })
 	var strms []db.Stream
 	r := db.DB.Find(&strms)
@@ -66,6 +67,7 @@ func StartFetchLoop() {
 	util.SliceEach(ready_chs, func(ch chan struct{}) { <-ch })
 
 	go fetchLoop(strms)
+	l.Info("nobori ready")
 }
 
 // Main loop for upstream metadata fetching