nandi/glean · 4fa381d

nandi/gleanpublic⑂ Fork 0

⑂ 4fa381d

Refactor precomputation to use helperUnverified

Julien Robert committed 2026-06-05T23:23:36+02:00 Browse files

4fa381d parent: 9cacbcf

modified internal/cluster/precompute.go +23 -32 Viewed

@@ -6,11 +6,6 @@ import (
6	6	"time"
7	7	)
8	8
9		-type PrecomputedRec struct {
10		- RecType string
11		- Data string
12		-}
13		-
14	9	func (e *Engine) PrecomputeAllRecommendations(ctx context.Context) error {
15	10	e.logger.Info("starting recommendation precomputation")
16	11	start := time.Now()
@@ -20,6 +15,8 @@ func (e *Engine) PrecomputeAllRecommendations(ctx context.Context) error {
20	15	return err
21	16	}
22	17
	18	+ e.logger.Info("precomputing recommendations for users", "count", len(users))
	19	+
23	20	computed := 0
24	21	for _, did := range users {
25	22	select {
@@ -43,6 +40,8 @@ func (e *Engine) PrecomputeAllRecommendations(ctx context.Context) error {
43	40	}
44	41
45	42	func (e *Engine) precomputeForUser(ctx context.Context, userDID string) error {
	43	+ start := time.Now()
	44	+
46	45	feedRecs, err := e.ComputeFeedRecommendationsOnDemand(ctx, userDID, 10)
47	46	if err != nil {
48	47	return err
@@ -80,22 +79,17 @@ func (e *Engine) precomputeForUser(ctx context.Context, userDID string) error {
80	79	}
81	80	}
82	81
83		- for _, rec := range []PrecomputedRec{
84		- {RecType: "feed", Data: mustJSON(feedRecs)},
85		- {RecType: "article", Data: mustJSON(articleRecs)},
86		- {RecType: "person", Data: mustJSON(peopleRecs)},
87		- } {
88		- if rec.Data == "null" {
89		- continue
90		- }
91		- if _, err := e.db.ExecContext(ctx, `
92		- INSERT INTO recs.precomputed_recommendations (user_did, rec_type, data, computed_at)
93		- VALUES (?, ?, ?, CURRENT_TIMESTAMP)
94		- ON CONFLICT(user_did, rec_type) DO UPDATE SET data = excluded.data, computed_at = excluded.computed_at
95		- `, userDID, rec.RecType, rec.Data); err != nil {
96		- e.logger.Warn("failed to store precomputed rec", "did", userDID, "type", rec.RecType, "error", err)
97		- }
98		- }
	82	+ e.storePrecomputed(ctx, userDID, "feed", mustJSON(feedRecs))
	83	+ e.storePrecomputed(ctx, userDID, "article", mustJSON(articleRecs))
	84	+ e.storePrecomputed(ctx, userDID, "person", mustJSON(peopleRecs))
	85	+
	86	+ e.logger.Debug("precomputed recommendations for user",
	87	+ "did", userDID,
	88	+ "feeds", len(feedRecs),
	89	+ "articles", len(articleRecs),
	90	+ "people", len(peopleRecs),
	91	+ "duration", time.Since(start),
	92	+ )
99	93
100	94	return nil
101	95	}
@@ -134,18 +128,15 @@ func (e *Engine) getPrecomputed(ctx context.Context, userDID, recType string) (s
134	128	return data, true
135	129	}
136	130
137		-func (e *Engine) invalidatePrecomputed(ctx context.Context, userDID, recType string) {
138		- _, _ = e.db.ExecContext(ctx, `
139		- DELETE FROM recs.precomputed_recommendations
140		- WHERE user_did = ? AND rec_type = ?
141		- `, userDID, recType)
142		-}
143		-
144		-func (e *Engine) recomputeForUser(ctx context.Context, userDID, recType string) {
145		- e.invalidatePrecomputed(ctx, userDID, recType)
146		- if err := e.precomputeForUser(ctx, userDID); err != nil {
147		- e.logger.Warn("recompute failed for user", "did", userDID, "type", recType, "error", err)
	131	+func (e *Engine) storePrecomputed(ctx context.Context, userDID, recType string, data string) {
	132	+ if data == "null" {
	133	+ return
148	134	}
	135	+ _, _ = e.db.ExecContext(ctx, `
	136	+ INSERT INTO recs.precomputed_recommendations (user_did, rec_type, data, computed_at)
	137	+ VALUES (?, ?, ?, CURRENT_TIMESTAMP)
	138	+ ON CONFLICT(user_did, rec_type) DO UPDATE SET data = excluded.data, computed_at = excluded.computed_at
	139	+ `, userDID, recType, data)
149	140	}
150	141
151	142	func mustJSON(v any) string {

  	"time"
  )
--type PrecomputedRec struct {
--	RecType string
--	Data    string
--}
--
  func (e *Engine) PrecomputeAllRecommendations(ctx context.Context) error {
  	e.logger.Info("starting recommendation precomputation")
  	start := time.Now()
  		return err
  	}
++	e.logger.Info("precomputing recommendations for users", "count", len(users))
++
  	computed := 0
  	for _, did := range users {
  		select {
  }
  func (e *Engine) precomputeForUser(ctx context.Context, userDID string) error {
++	start := time.Now()
++
  	feedRecs, err := e.ComputeFeedRecommendationsOnDemand(ctx, userDID, 10)
  	if err != nil {
  		return err
  		}
  	}
--	for _, rec := range []PrecomputedRec{
++	e.storePrecomputed(ctx, userDID, "feed", mustJSON(feedRecs))
--		{RecType: "feed", Data: mustJSON(feedRecs)},
++	e.storePrecomputed(ctx, userDID, "article", mustJSON(articleRecs))
--		{RecType: "article", Data: mustJSON(articleRecs)},
++	e.storePrecomputed(ctx, userDID, "person", mustJSON(peopleRecs))
--		{RecType: "person", Data: mustJSON(peopleRecs)},
++
--	} {
++	e.logger.Debug("precomputed recommendations for user",
--		if rec.Data == "null" {
++		"did", userDID,
--			continue
++		"feeds", len(feedRecs),
--		}
++		"articles", len(articleRecs),
--		if _, err := e.db.ExecContext(ctx, `
++		"people", len(peopleRecs),
--			INSERT INTO recs.precomputed_recommendations (user_did, rec_type, data, computed_at)
++		"duration", time.Since(start),
--			VALUES (?, ?, ?, CURRENT_TIMESTAMP)
++	)
--			ON CONFLICT(user_did, rec_type) DO UPDATE SET data = excluded.data, computed_at = excluded.computed_at
--		`, userDID, rec.RecType, rec.Data); err != nil {
--			e.logger.Warn("failed to store precomputed rec", "did", userDID, "type", rec.RecType, "error", err)
--		}
--	}
  	return nil
  }
  	return data, true
  }
--func (e *Engine) invalidatePrecomputed(ctx context.Context, userDID, recType string) {
++func (e *Engine) storePrecomputed(ctx context.Context, userDID, recType string, data string) {
--	_, _ = e.db.ExecContext(ctx, `
++	if data == "null" {
--		DELETE FROM recs.precomputed_recommendations
++		return
--		WHERE user_did = ? AND rec_type = ?
--	`, userDID, recType)
--}
--
--func (e *Engine) recomputeForUser(ctx context.Context, userDID, recType string) {
--	e.invalidatePrecomputed(ctx, userDID, recType)
--	if err := e.precomputeForUser(ctx, userDID); err != nil {
--		e.logger.Warn("recompute failed for user", "did", userDID, "type", recType, "error", err)
  	}
++	_, _ = e.db.ExecContext(ctx, `
++		INSERT INTO recs.precomputed_recommendations (user_did, rec_type, data, computed_at)
++		VALUES (?, ?, ?, CURRENT_TIMESTAMP)
++		ON CONFLICT(user_did, rec_type) DO UPDATE SET data = excluded.data, computed_at = excluded.computed_at
++	`, userDID, recType, data)
  }
  func mustJSON(v any) string {

modified internal/cluster/scoring.go +24 -5 Viewed

@@ -47,15 +47,15 @@ type ArticleRecommendation struct {
47	47	}
48	48
49	49	func (e *Engine) InvalidateFeedCache(userDID string) {
50		- go e.recomputeForUser(context.Background(), userDID, "feed")
	50	+ go e.precomputeForUser(context.Background(), userDID)
51	51	}
52	52
53	53	func (e *Engine) InvalidateArticleCache(userDID string) {
54		- go e.recomputeForUser(context.Background(), userDID, "article")
	54	+ go e.precomputeForUser(context.Background(), userDID)
55	55	}
56	56
57	57	func (e *Engine) InvalidatePeopleCache(userDID string) {
58		- go e.recomputeForUser(context.Background(), userDID, "person")
	58	+ go e.precomputeForUser(context.Background(), userDID)
59	59	}
60	60
61	61	// GetFeedRecommendations returns feed recommendations for a user. Users with
@@ -70,6 +70,9 @@ func (e *Engine) GetFeedRecommendations(ctx context.Context, userDID string, lim
70	70	}
71	71	}
72	72
	73	+ start := time.Now()
	74	+ e.logger.Info("feed recommendations cache miss, computing on-demand", "did", userDID)
	75	+
73	76	subCount := 0
74	77	_ = e.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM articles.subscriptions WHERE user_did = ?`, userDID).Scan(&subCount)
75	78
@@ -77,7 +80,10 @@ func (e *Engine) GetFeedRecommendations(ctx context.Context, userDID string, lim
77	80	recs, err := e.ColdStartRecommendations(ctx, userDID, limit*2)
78	81	if err == nil && len(recs) > 0 {
79	82	normalizeFeedScores(recs)
80		- return ApplyDiversity(recs, limit), nil
	83	+ result := ApplyDiversity(recs, limit)
	84	+ e.storePrecomputed(ctx, userDID, "feed", mustJSON(result))
	85	+ e.logger.Info("feed recommendations computed (cold-start)", "did", userDID, "count", len(result), "duration", time.Since(start))
	86	+ return result, nil
81	87	}
82	88	}
83	89
@@ -87,7 +93,10 @@ func (e *Engine) GetFeedRecommendations(ctx context.Context, userDID string, lim
87	93	}
88	94
89	95	normalizeFeedScores(recs)
90		- return ApplyDiversity(recs, limit), nil
	96	+ result := ApplyDiversity(recs, limit)
	97	+ e.storePrecomputed(ctx, userDID, "feed", mustJSON(result))
	98	+ e.logger.Info("feed recommendations computed (on-demand)", "did", userDID, "count", len(result), "duration", time.Since(start))
	99	+ return result, nil
91	100	}
92	101
93	102	// GetPeopleRecommendations returns similar users based on subscription overlap,
@@ -102,6 +111,9 @@ func (e *Engine) GetPeopleRecommendations(ctx context.Context, userDID string, l
102	111	}
103	112	}
104	113
	114	+ start := time.Now()
	115	+ e.logger.Info("people recommendations cache miss, computing on-demand", "did", userDID)
	116	+
105	117	half := max(limit/2, 1)
106	118
107	119	inNet, err := e.computePeopleByFollowStatus(ctx, userDID, true, half)
@@ -119,6 +131,8 @@ func (e *Engine) GetPeopleRecommendations(ctx context.Context, userDID string, l
119	131	recs = append(recs, outNet...)
120	132
121	133	normalizePersonScores(recs)
	134	+ e.storePrecomputed(ctx, userDID, "person", mustJSON(recs))
	135	+ e.logger.Info("people recommendations computed (on-demand)", "did", userDID, "count", len(recs), "duration", time.Since(start))
122	136	return recs, nil
123	137	}
124	138
@@ -134,11 +148,16 @@ func (e *Engine) GetArticleRecommendations(ctx context.Context, userDID string,
134	148	}
135	149	}
136	150
	151	+ start := time.Now()
	152	+ e.logger.Info("article recommendations cache miss, computing on-demand", "did", userDID)
	153	+
137	154	recs, err := e.ComputeArticleRecommendationsOnDemand(ctx, userDID, languages, limit)
138	155	if err != nil {
139	156	return nil, err
140	157	}
141	158	normalizeArticleScores(recs)
	159	+ e.storePrecomputed(ctx, userDID, "article", mustJSON(recs))
	160	+ e.logger.Info("article recommendations computed (on-demand)", "did", userDID, "count", len(recs), "duration", time.Since(start))
142	161	return recs, nil
143	162	}
144	163

  }
  func (e *Engine) InvalidateFeedCache(userDID string) {
--	go e.recomputeForUser(context.Background(), userDID, "feed")
++	go e.precomputeForUser(context.Background(), userDID)
  }
  func (e *Engine) InvalidateArticleCache(userDID string) {
--	go e.recomputeForUser(context.Background(), userDID, "article")
++	go e.precomputeForUser(context.Background(), userDID)
  }
  func (e *Engine) InvalidatePeopleCache(userDID string) {
--	go e.recomputeForUser(context.Background(), userDID, "person")
++	go e.precomputeForUser(context.Background(), userDID)
  }
  // GetFeedRecommendations returns feed recommendations for a user. Users with
  		}
  	}
++	start := time.Now()
++	e.logger.Info("feed recommendations cache miss, computing on-demand", "did", userDID)
++
  	subCount := 0
  	_ = e.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM articles.subscriptions WHERE user_did = ?`, userDID).Scan(&subCount)
  		recs, err := e.ColdStartRecommendations(ctx, userDID, limit*2)
  		if err == nil && len(recs) > 0 {
  			normalizeFeedScores(recs)
--			return ApplyDiversity(recs, limit), nil
++			result := ApplyDiversity(recs, limit)
++			e.storePrecomputed(ctx, userDID, "feed", mustJSON(result))
++			e.logger.Info("feed recommendations computed (cold-start)", "did", userDID, "count", len(result), "duration", time.Since(start))
++			return result, nil
  		}
  	}
  	}
  	normalizeFeedScores(recs)
--	return ApplyDiversity(recs, limit), nil
++	result := ApplyDiversity(recs, limit)
++	e.storePrecomputed(ctx, userDID, "feed", mustJSON(result))
++	e.logger.Info("feed recommendations computed (on-demand)", "did", userDID, "count", len(result), "duration", time.Since(start))
++	return result, nil
  }
  // GetPeopleRecommendations returns similar users based on subscription overlap,
  		}
  	}
++	start := time.Now()
++	e.logger.Info("people recommendations cache miss, computing on-demand", "did", userDID)
++
  	half := max(limit/2, 1)
  	inNet, err := e.computePeopleByFollowStatus(ctx, userDID, true, half)
  	recs = append(recs, outNet...)
  	normalizePersonScores(recs)
++	e.storePrecomputed(ctx, userDID, "person", mustJSON(recs))
++	e.logger.Info("people recommendations computed (on-demand)", "did", userDID, "count", len(recs), "duration", time.Since(start))
  	return recs, nil
  }
  		}
  	}
++	start := time.Now()
++	e.logger.Info("article recommendations cache miss, computing on-demand", "did", userDID)
++
  	recs, err := e.ComputeArticleRecommendationsOnDemand(ctx, userDID, languages, limit)
  	if err != nil {
  		return nil, err
  	}
  	normalizeArticleScores(recs)
++	e.storePrecomputed(ctx, userDID, "article", mustJSON(recs))
++	e.logger.Info("article recommendations computed (on-demand)", "did", userDID, "count", len(recs), "duration", time.Since(start))
  	return recs, nil
  }