rickub
nandi/gleanpublic⑂ Fork 0
⑂ f988072
Branches
claude/modal-container-min-76527a main View all branches → View all tags →
Commits
⬇ Clone ▾
git clone https://git.rickub.com/nandi/glean.git
git clone ssh://git@rickub.com/nandi/glean.git

Host key fingerprint (ed25519): SHA256:iycHnxEyq0Q7uyVpB7JlznP0G7JrTPXLYRcAU5CSLhc — verify it before your first connect.

Support per-user similarity computation and handle resolutionUnverified

Julien Robert committed 2026-04-21T14:45:07+02:00 Browse files
f988072 parent: 450d849
modified internal/cluster/cron.go +14 -1
@@ -23,7 +23,20 @@ func (c *Cron) Run(ctx context.Context) error {
2323 c.logger.Info("starting similarity computation")
2424 start := time.Now()
2525
26- c.engine.ComputeAll(ctx)
26+ if !c.engine.mu.TryLock() {
27+ c.logger.Info("skipping computation: already in progress")
28+ } else {
29+ if err := c.engine.ComputeFeedSimilarity(ctx); err != nil {
30+ c.engine.logger.Error("feed similarity failed", "error", err)
31+ }
32+ if err := c.engine.ComputeUserSimilarity(ctx); err != nil {
33+ c.engine.logger.Error("user similarity failed", "error", err)
34+ }
35+ if err := c.engine.ComputeRecommendations(ctx); err != nil {
36+ c.engine.logger.Error("recommendations failed", "error", err)
37+ }
38+ c.engine.mu.Unlock()
39+ }
2740
2841 metrics.ClusterRuns.Inc()
2942 metrics.ClusterDuration.Observe(time.Since(start).Seconds())
@@ -23,7 +23,20 @@ func (c *Cron) Run(ctx context.Context) error {
23 c.logger.Info("starting similarity computation")23 c.logger.Info("starting similarity computation")
24 start := time.Now()24 start := time.Now()
25 25
26- c.engine.ComputeAll(ctx)26+ if !c.engine.mu.TryLock() {
27+ c.logger.Info("skipping computation: already in progress")
28+ } else {
29+ if err := c.engine.ComputeFeedSimilarity(ctx); err != nil {
30+ c.engine.logger.Error("feed similarity failed", "error", err)
31+ }
32+ if err := c.engine.ComputeUserSimilarity(ctx); err != nil {
33+ c.engine.logger.Error("user similarity failed", "error", err)
34+ }
35+ if err := c.engine.ComputeRecommendations(ctx); err != nil {
36+ c.engine.logger.Error("recommendations failed", "error", err)
37+ }
38+ c.engine.mu.Unlock()
39+ }
27 40
28 metrics.ClusterRuns.Inc()41 metrics.ClusterRuns.Inc()
29 metrics.ClusterDuration.Observe(time.Since(start).Seconds())42 metrics.ClusterDuration.Observe(time.Since(start).Seconds())
modified internal/cluster/jaccard.go +148 -18
@@ -13,24 +13,6 @@ type Engine struct {
1313 mu sync.Mutex
1414 }
1515
16-func (e *Engine) ComputeAll(ctx context.Context) {
17- if !e.mu.TryLock() {
18- e.logger.Info("skipping ComputeAll: already in progress")
19- return
20- }
21- defer e.mu.Unlock()
22-
23- if err := e.ComputeFeedSimilarity(ctx); err != nil {
24- e.logger.Error("feed similarity failed", "error", err)
25- }
26- if err := e.ComputeUserSimilarity(ctx); err != nil {
27- e.logger.Error("user similarity failed", "error", err)
28- }
29- if err := e.ComputeRecommendations(ctx); err != nil {
30- e.logger.Error("recommendations failed", "error", err)
31- }
32-}
33-
3416 func (e *Engine) ComputeArticleRecommendations(ctx context.Context) error {
3517 tx, err := e.db.BeginTx(ctx, nil)
3618 if err != nil {
@@ -73,6 +55,21 @@ func (e *Engine) ComputeArticleRecommendations(ctx context.Context) error {
7355 return tx.Commit()
7456 }
7557
58+func (e *Engine) ComputeForUser(ctx context.Context, userDID string) {
59+ if !e.mu.TryLock() {
60+ e.logger.Info("skipping ComputeForUser: already in progress", "did", userDID)
61+ return
62+ }
63+ defer e.mu.Unlock()
64+
65+ if err := e.ComputeUserSimilarityForUser(ctx, userDID); err != nil {
66+ e.logger.Error("per-user similarity failed", "error", err, "did", userDID)
67+ }
68+ if err := e.ComputeRecommendationsForUser(ctx, userDID); err != nil {
69+ e.logger.Error("per-user recommendations failed", "error", err, "did", userDID)
70+ }
71+}
72+
7673 func NewEngine(db *sql.DB, logger *slog.Logger) *Engine {
7774 return &Engine{db: db, logger: logger}
7875 }
@@ -161,6 +158,139 @@ func (e *Engine) ComputeUserSimilarity(ctx context.Context) error {
161158 return tx.Commit()
162159 }
163160
161+func (e *Engine) ComputeUserSimilarityForUser(ctx context.Context, userDID string) error {
162+ tx, err := e.db.BeginTx(ctx, nil)
163+ if err != nil {
164+ return err
165+ }
166+ defer func() { _ = tx.Rollback() }()
167+
168+ if _, err := tx.ExecContext(ctx, `DELETE FROM user_similarity WHERE user_a = ? OR user_b = ?`, userDID, userDID); err != nil {
169+ return err
170+ }
171+
172+ _, err = tx.ExecContext(ctx, `
173+ INSERT INTO user_similarity (user_a, user_b, jaccard, common_feeds)
174+ SELECT
175+ MIN(?, s2.user_did),
176+ MAX(?, s2.user_did),
177+ CAST(COUNT(*) AS REAL) / (
178+ (SELECT COUNT(*) FROM subscriptions WHERE user_did = ?) +
179+ (SELECT COUNT(*) FROM subscriptions WHERE user_did = s2.user_did) -
180+ CAST(COUNT(*) AS REAL)
181+ ),
182+ COUNT(*)
183+ FROM subscriptions s1
184+ JOIN subscriptions s2 ON s1.feed_url = s2.feed_url AND s2.user_did != ?
185+ WHERE s1.user_did = ?
186+ GROUP BY s2.user_did
187+ HAVING COUNT(*) > 0
188+ `, userDID, userDID, userDID, userDID, userDID)
189+ if err != nil {
190+ return err
191+ }
192+
193+ _, err = tx.ExecContext(ctx, `
194+ INSERT INTO user_similarity (user_a, user_b, jaccard, common_feeds)
195+ SELECT
196+ MIN(?, f.target_did),
197+ MAX(?, f.target_did),
198+ 0.5,
199+ 0
200+ FROM follows f
201+ WHERE f.user_did = ? AND f.target_did != ?
202+ GROUP BY MIN(?, f.target_did), MAX(?, f.target_did)
203+ ON CONFLICT(user_a, user_b) DO UPDATE SET
204+ jaccard = jaccard + 0.5
205+ `, userDID, userDID, userDID, userDID, userDID, userDID)
206+ if err != nil {
207+ return err
208+ }
209+
210+ e.logger.Info("per-user similarity computed", "did", userDID)
211+ return tx.Commit()
212+}
213+
214+func (e *Engine) ComputeRecommendationsForUser(ctx context.Context, userDID string) error {
215+ tx, err := e.db.BeginTx(ctx, nil)
216+ if err != nil {
217+ return err
218+ }
219+ defer func() { _ = tx.Rollback() }()
220+
221+ if _, err := tx.ExecContext(ctx, `DELETE FROM user_feed_recommendations WHERE user_did = ?`, userDID); err != nil {
222+ return err
223+ }
224+
225+ _, err = tx.ExecContext(ctx, `
226+ INSERT INTO user_feed_recommendations (user_did, feed_url, score)
227+ SELECT ?, s.feed_url, SUM(us.jaccard) AS score
228+ FROM user_similarity us
229+ JOIN subscriptions s ON s.user_did = CASE
230+ WHEN us.user_a = ? THEN us.user_b
231+ ELSE us.user_a
232+ END
233+ WHERE (us.user_a = ? OR us.user_b = ?)
234+ AND us.jaccard > 0.2
235+ AND s.feed_url NOT IN (SELECT feed_url FROM subscriptions WHERE user_did = ?)
236+ GROUP BY s.feed_url
237+ ORDER BY score DESC
238+ `, userDID, userDID, userDID, userDID, userDID)
239+ if err != nil {
240+ return err
241+ }
242+
243+ if err := tx.Commit(); err != nil {
244+ return err
245+ }
246+
247+ if err := e.computeArticleRecommendationsForUser(ctx, userDID); err != nil {
248+ return err
249+ }
250+
251+ e.logger.Info("per-user recommendations computed", "did", userDID)
252+ return nil
253+}
254+
255+func (e *Engine) computeArticleRecommendationsForUser(ctx context.Context, userDID string) error {
256+ tx, err := e.db.BeginTx(ctx, nil)
257+ if err != nil {
258+ return err
259+ }
260+ defer func() { _ = tx.Rollback() }()
261+
262+ if _, err := tx.ExecContext(ctx, `DELETE FROM user_article_recommendations WHERE user_did = ?`, userDID); err != nil {
263+ return err
264+ }
265+
266+ _, err = tx.ExecContext(ctx, `
267+ INSERT INTO user_article_recommendations (user_did, feed_url, article_url, score)
268+ SELECT ?, l.feed_url, l.article_url, SUM(us.jaccard) AS score
269+ FROM (
270+ SELECT us.user_b AS peer, us.jaccard
271+ FROM user_similarity us WHERE us.user_a = ? AND us.jaccard > 0.2
272+ UNION ALL
273+ SELECT us.user_a AS peer, us.jaccard
274+ FROM user_similarity us WHERE us.user_b = ? AND us.jaccard > 0.2
275+ ) us
276+ JOIN likes l ON l.author_did = us.peer
277+ WHERE NOT EXISTS (
278+ SELECT 1 FROM subscriptions sub WHERE sub.user_did = ? AND sub.feed_url = l.feed_url
279+ )
280+ AND NOT EXISTS (
281+ SELECT 1 FROM likes ul WHERE ul.author_did = ? AND ul.feed_url = l.feed_url AND ul.article_url = l.article_url
282+ )
283+ GROUP BY l.feed_url, l.article_url
284+ HAVING COUNT(*) > 0
285+ ORDER BY score DESC
286+ `, userDID, userDID, userDID, userDID, userDID)
287+ if err != nil {
288+ return err
289+ }
290+
291+ return tx.Commit()
292+}
293+
164294 func (e *Engine) ComputeRecommendations(ctx context.Context) error {
165295 tx, err := e.db.BeginTx(ctx, nil)
166296 if err != nil {
@@ -13,24 +13,6 @@ type Engine struct {
13 mu sync.Mutex13 mu sync.Mutex
14 }14 }
15 15
16-func (e *Engine) ComputeAll(ctx context.Context) {
17- if !e.mu.TryLock() {
18- e.logger.Info("skipping ComputeAll: already in progress")
19- return
20- }
21- defer e.mu.Unlock()
22-
23- if err := e.ComputeFeedSimilarity(ctx); err != nil {
24- e.logger.Error("feed similarity failed", "error", err)
25- }
26- if err := e.ComputeUserSimilarity(ctx); err != nil {
27- e.logger.Error("user similarity failed", "error", err)
28- }
29- if err := e.ComputeRecommendations(ctx); err != nil {
30- e.logger.Error("recommendations failed", "error", err)
31- }
32-}
33-
34 func (e *Engine) ComputeArticleRecommendations(ctx context.Context) error {16 func (e *Engine) ComputeArticleRecommendations(ctx context.Context) error {
35 tx, err := e.db.BeginTx(ctx, nil)17 tx, err := e.db.BeginTx(ctx, nil)
36 if err != nil {18 if err != nil {
@@ -73,6 +55,21 @@ func (e *Engine) ComputeArticleRecommendations(ctx context.Context) error {
73 return tx.Commit()55 return tx.Commit()
74 }56 }
75 57
58+func (e *Engine) ComputeForUser(ctx context.Context, userDID string) {
59+ if !e.mu.TryLock() {
60+ e.logger.Info("skipping ComputeForUser: already in progress", "did", userDID)
61+ return
62+ }
63+ defer e.mu.Unlock()
64+
65+ if err := e.ComputeUserSimilarityForUser(ctx, userDID); err != nil {
66+ e.logger.Error("per-user similarity failed", "error", err, "did", userDID)
67+ }
68+ if err := e.ComputeRecommendationsForUser(ctx, userDID); err != nil {
69+ e.logger.Error("per-user recommendations failed", "error", err, "did", userDID)
70+ }
71+}
72+
76 func NewEngine(db *sql.DB, logger *slog.Logger) *Engine {73 func NewEngine(db *sql.DB, logger *slog.Logger) *Engine {
77 return &Engine{db: db, logger: logger}74 return &Engine{db: db, logger: logger}
78 }75 }
@@ -161,6 +158,139 @@ func (e *Engine) ComputeUserSimilarity(ctx context.Context) error {
161 return tx.Commit()158 return tx.Commit()
162 }159 }
163 160
161+func (e *Engine) ComputeUserSimilarityForUser(ctx context.Context, userDID string) error {
162+ tx, err := e.db.BeginTx(ctx, nil)
163+ if err != nil {
164+ return err
165+ }
166+ defer func() { _ = tx.Rollback() }()
167+
168+ if _, err := tx.ExecContext(ctx, `DELETE FROM user_similarity WHERE user_a = ? OR user_b = ?`, userDID, userDID); err != nil {
169+ return err
170+ }
171+
172+ _, err = tx.ExecContext(ctx, `
173+ INSERT INTO user_similarity (user_a, user_b, jaccard, common_feeds)
174+ SELECT
175+ MIN(?, s2.user_did),
176+ MAX(?, s2.user_did),
177+ CAST(COUNT(*) AS REAL) / (
178+ (SELECT COUNT(*) FROM subscriptions WHERE user_did = ?) +
179+ (SELECT COUNT(*) FROM subscriptions WHERE user_did = s2.user_did) -
180+ CAST(COUNT(*) AS REAL)
181+ ),
182+ COUNT(*)
183+ FROM subscriptions s1
184+ JOIN subscriptions s2 ON s1.feed_url = s2.feed_url AND s2.user_did != ?
185+ WHERE s1.user_did = ?
186+ GROUP BY s2.user_did
187+ HAVING COUNT(*) > 0
188+ `, userDID, userDID, userDID, userDID, userDID)
189+ if err != nil {
190+ return err
191+ }
192+
193+ _, err = tx.ExecContext(ctx, `
194+ INSERT INTO user_similarity (user_a, user_b, jaccard, common_feeds)
195+ SELECT
196+ MIN(?, f.target_did),
197+ MAX(?, f.target_did),
198+ 0.5,
199+ 0
200+ FROM follows f
201+ WHERE f.user_did = ? AND f.target_did != ?
202+ GROUP BY MIN(?, f.target_did), MAX(?, f.target_did)
203+ ON CONFLICT(user_a, user_b) DO UPDATE SET
204+ jaccard = jaccard + 0.5
205+ `, userDID, userDID, userDID, userDID, userDID, userDID)
206+ if err != nil {
207+ return err
208+ }
209+
210+ e.logger.Info("per-user similarity computed", "did", userDID)
211+ return tx.Commit()
212+}
213+
214+func (e *Engine) ComputeRecommendationsForUser(ctx context.Context, userDID string) error {
215+ tx, err := e.db.BeginTx(ctx, nil)
216+ if err != nil {
217+ return err
218+ }
219+ defer func() { _ = tx.Rollback() }()
220+
221+ if _, err := tx.ExecContext(ctx, `DELETE FROM user_feed_recommendations WHERE user_did = ?`, userDID); err != nil {
222+ return err
223+ }
224+
225+ _, err = tx.ExecContext(ctx, `
226+ INSERT INTO user_feed_recommendations (user_did, feed_url, score)
227+ SELECT ?, s.feed_url, SUM(us.jaccard) AS score
228+ FROM user_similarity us
229+ JOIN subscriptions s ON s.user_did = CASE
230+ WHEN us.user_a = ? THEN us.user_b
231+ ELSE us.user_a
232+ END
233+ WHERE (us.user_a = ? OR us.user_b = ?)
234+ AND us.jaccard > 0.2
235+ AND s.feed_url NOT IN (SELECT feed_url FROM subscriptions WHERE user_did = ?)
236+ GROUP BY s.feed_url
237+ ORDER BY score DESC
238+ `, userDID, userDID, userDID, userDID, userDID)
239+ if err != nil {
240+ return err
241+ }
242+
243+ if err := tx.Commit(); err != nil {
244+ return err
245+ }
246+
247+ if err := e.computeArticleRecommendationsForUser(ctx, userDID); err != nil {
248+ return err
249+ }
250+
251+ e.logger.Info("per-user recommendations computed", "did", userDID)
252+ return nil
253+}
254+
255+func (e *Engine) computeArticleRecommendationsForUser(ctx context.Context, userDID string) error {
256+ tx, err := e.db.BeginTx(ctx, nil)
257+ if err != nil {
258+ return err
259+ }
260+ defer func() { _ = tx.Rollback() }()
261+
262+ if _, err := tx.ExecContext(ctx, `DELETE FROM user_article_recommendations WHERE user_did = ?`, userDID); err != nil {
263+ return err
264+ }
265+
266+ _, err = tx.ExecContext(ctx, `
267+ INSERT INTO user_article_recommendations (user_did, feed_url, article_url, score)
268+ SELECT ?, l.feed_url, l.article_url, SUM(us.jaccard) AS score
269+ FROM (
270+ SELECT us.user_b AS peer, us.jaccard
271+ FROM user_similarity us WHERE us.user_a = ? AND us.jaccard > 0.2
272+ UNION ALL
273+ SELECT us.user_a AS peer, us.jaccard
274+ FROM user_similarity us WHERE us.user_b = ? AND us.jaccard > 0.2
275+ ) us
276+ JOIN likes l ON l.author_did = us.peer
277+ WHERE NOT EXISTS (
278+ SELECT 1 FROM subscriptions sub WHERE sub.user_did = ? AND sub.feed_url = l.feed_url
279+ )
280+ AND NOT EXISTS (
281+ SELECT 1 FROM likes ul WHERE ul.author_did = ? AND ul.feed_url = l.feed_url AND ul.article_url = l.article_url
282+ )
283+ GROUP BY l.feed_url, l.article_url
284+ HAVING COUNT(*) > 0
285+ ORDER BY score DESC
286+ `, userDID, userDID, userDID, userDID, userDID)
287+ if err != nil {
288+ return err
289+ }
290+
291+ return tx.Commit()
292+}
293+
164 func (e *Engine) ComputeRecommendations(ctx context.Context) error {294 func (e *Engine) ComputeRecommendations(ctx context.Context) error {
165 tx, err := e.db.BeginTx(ctx, nil)295 tx, err := e.db.BeginTx(ctx, nil)
166 if err != nil {296 if err != nil {
modified internal/cluster/jaccard_test.go +165 -0
@@ -140,3 +140,168 @@ func TestComputeRecommendations_NoSelfRecommendations(t *testing.T) {
140140 "should not recommend a feed the user already subscribes to")
141141 }
142142 }
143+
144+func TestComputeUserSimilarityForUser(t *testing.T) {
145+ ctx := context.Background()
146+ database := setupClusterTestDB(t)
147+ seedClusterData(t, ctx, database)
148+
149+ engine := NewEngine(database.DB, slog.Default())
150+
151+ assert.NilError(t, engine.ComputeUserSimilarityForUser(ctx, "did:test:alice"))
152+ assert.NilError(t, engine.ComputeUserSimilarityForUser(ctx, "did:test:bob"))
153+ assert.NilError(t, engine.ComputeUserSimilarityForUser(ctx, "did:test:carol"))
154+
155+ var count int
156+ assert.NilError(t, database.QueryRowContext(ctx, `SELECT COUNT(*) FROM user_similarity`).Scan(&count))
157+ assert.Equal(t, count, 2, "alice-bob and alice-carol share feeds")
158+}
159+
160+func TestComputeUserSimilarityForUser_MatchesFullCompute(t *testing.T) {
161+ ctx := context.Background()
162+
163+ type simPair struct {
164+ userA, userB string
165+ jaccard float64
166+ commonFeeds int
167+ }
168+
169+ readAll := func(t *testing.T, database *db.DB) []simPair {
170+ t.Helper()
171+ rows, err := database.QueryContext(ctx, `SELECT user_a, user_b, jaccard, common_feeds FROM user_similarity ORDER BY user_a, user_b`)
172+ assert.NilError(t, err)
173+ defer rows.Close()
174+ var result []simPair
175+ for rows.Next() {
176+ var p simPair
177+ assert.NilError(t, rows.Scan(&p.userA, &p.userB, &p.jaccard, &p.commonFeeds))
178+ result = append(result, p)
179+ }
180+ assert.NilError(t, rows.Err())
181+ return result
182+ }
183+
184+ database1 := setupClusterTestDB(t)
185+ seedClusterData(t, ctx, database1)
186+ engine1 := NewEngine(database1.DB, slog.Default())
187+ assert.NilError(t, engine1.ComputeUserSimilarity(ctx))
188+ fullPairs := readAll(t, database1)
189+
190+ database2 := setupClusterTestDB(t)
191+ seedClusterData(t, ctx, database2)
192+ engine2 := NewEngine(database2.DB, slog.Default())
193+ assert.NilError(t, engine2.ComputeUserSimilarityForUser(ctx, "did:test:alice"))
194+ assert.NilError(t, engine2.ComputeUserSimilarityForUser(ctx, "did:test:bob"))
195+ assert.NilError(t, engine2.ComputeUserSimilarityForUser(ctx, "did:test:carol"))
196+ incrPairs := readAll(t, database2)
197+
198+ assert.Equal(t, len(fullPairs), len(incrPairs), "same number of pairs")
199+ for i := range fullPairs {
200+ assert.Equal(t, fullPairs[i].userA, incrPairs[i].userA)
201+ assert.Equal(t, fullPairs[i].userB, incrPairs[i].userB)
202+ assert.Equal(t, fullPairs[i].commonFeeds, incrPairs[i].commonFeeds)
203+ }
204+}
205+
206+func TestComputeUserSimilarityForUser_UpdatesOnSubscriptionChange(t *testing.T) {
207+ ctx := context.Background()
208+ database := setupClusterTestDB(t)
209+ seedClusterData(t, ctx, database)
210+
211+ engine := NewEngine(database.DB, slog.Default())
212+ assert.NilError(t, engine.ComputeUserSimilarity(ctx))
213+
214+ _, err := database.ExecContext(ctx, `INSERT INTO subscriptions (user_did, feed_url) VALUES (?, ?)`, "did:test:carol", "https://a.com/feed")
215+ assert.NilError(t, err)
216+
217+ assert.NilError(t, engine.ComputeUserSimilarityForUser(ctx, "did:test:carol"))
218+
219+ var jaccard float64
220+ var common int
221+ assert.NilError(t, database.QueryRowContext(ctx,
222+ `SELECT jaccard, common_feeds FROM user_similarity WHERE user_a = ? AND user_b = ?`,
223+ "did:test:alice", "did:test:carol").Scan(&jaccard, &common))
224+ assert.Equal(t, common, 2, "alice and carol now share feed A and feed C")
225+}
226+
227+func TestComputeUserSimilarityForUser_WithFollowBoost(t *testing.T) {
228+ ctx := context.Background()
229+ database := setupClusterTestDB(t)
230+ seedClusterData(t, ctx, database)
231+
232+ _, err := database.ExecContext(ctx, `INSERT INTO follows (user_did, target_did) VALUES (?, ?)`, "did:test:alice", "did:test:bob")
233+ assert.NilError(t, err)
234+
235+ engine := NewEngine(database.DB, slog.Default())
236+ assert.NilError(t, engine.ComputeUserSimilarityForUser(ctx, "did:test:alice"))
237+
238+ var jaccard float64
239+ assert.NilError(t, database.QueryRowContext(ctx,
240+ `SELECT jaccard FROM user_similarity WHERE user_a = ? AND user_b = ?`,
241+ "did:test:alice", "did:test:bob").Scan(&jaccard))
242+ assert.Assert(t, jaccard > 0.6, "follow boost should add 0.5 to subscription jaccard, got %f", jaccard)
243+}
244+
245+func TestComputeRecommendationsForUser(t *testing.T) {
246+ ctx := context.Background()
247+ database := setupClusterTestDB(t)
248+ seedClusterData(t, ctx, database)
249+
250+ engine := NewEngine(database.DB, slog.Default())
251+ assert.NilError(t, engine.ComputeUserSimilarity(ctx))
252+ assert.NilError(t, engine.ComputeRecommendationsForUser(ctx, "did:test:carol"))
253+
254+ recs, err := engine.GetFeedRecommendations(ctx, "did:test:carol", 10)
255+ assert.NilError(t, err)
256+ assert.Assert(t, len(recs) > 0, "carol should get feed recommendations")
257+
258+ var found bool
259+ for _, r := range recs {
260+ if r["feed_url"] == "https://a.com/feed" || r["feed_url"] == "https://b.com/feed" {
261+ found = true
262+ }
263+ }
264+ assert.Assert(t, found, "carol should be recommended feeds from alice")
265+}
266+
267+func TestComputeRecommendationsForUser_MatchesFullCompute(t *testing.T) {
268+ ctx := context.Background()
269+
270+ readRecs := func(t *testing.T, database *db.DB, did string) map[string]float64 {
271+ t.Helper()
272+ rows, err := database.QueryContext(ctx,
273+ `SELECT feed_url, score FROM user_feed_recommendations WHERE user_did = ? ORDER BY feed_url`, did)
274+ assert.NilError(t, err)
275+ defer rows.Close()
276+ result := map[string]float64{}
277+ for rows.Next() {
278+ var url string
279+ var score float64
280+ assert.NilError(t, rows.Scan(&url, &score))
281+ result[url] = score
282+ }
283+ assert.NilError(t, rows.Err())
284+ return result
285+ }
286+
287+ database1 := setupClusterTestDB(t)
288+ seedClusterData(t, ctx, database1)
289+ engine1 := NewEngine(database1.DB, slog.Default())
290+ assert.NilError(t, engine1.ComputeUserSimilarity(ctx))
291+ assert.NilError(t, engine1.ComputeRecommendations(ctx))
292+ fullRecs := readRecs(t, database1, "did:test:carol")
293+
294+ database2 := setupClusterTestDB(t)
295+ seedClusterData(t, ctx, database2)
296+ engine2 := NewEngine(database2.DB, slog.Default())
297+ assert.NilError(t, engine2.ComputeUserSimilarity(ctx))
298+ assert.NilError(t, engine2.ComputeRecommendationsForUser(ctx, "did:test:carol"))
299+ incrRecs := readRecs(t, database2, "did:test:carol")
300+
301+ assert.Equal(t, len(fullRecs), len(incrRecs), "same number of recommendations")
302+ for url, score := range fullRecs {
303+ incrScore, ok := incrRecs[url]
304+ assert.Assert(t, ok, "missing recommendation for %s", url)
305+ assert.Equal(t, score, incrScore, "score mismatch for %s", url)
306+ }
307+}
@@ -140,3 +140,168 @@ func TestComputeRecommendations_NoSelfRecommendations(t *testing.T) {
140 "should not recommend a feed the user already subscribes to")140 "should not recommend a feed the user already subscribes to")
141 }141 }
142 }142 }
143+
144+func TestComputeUserSimilarityForUser(t *testing.T) {
145+ ctx := context.Background()
146+ database := setupClusterTestDB(t)
147+ seedClusterData(t, ctx, database)
148+
149+ engine := NewEngine(database.DB, slog.Default())
150+
151+ assert.NilError(t, engine.ComputeUserSimilarityForUser(ctx, "did:test:alice"))
152+ assert.NilError(t, engine.ComputeUserSimilarityForUser(ctx, "did:test:bob"))
153+ assert.NilError(t, engine.ComputeUserSimilarityForUser(ctx, "did:test:carol"))
154+
155+ var count int
156+ assert.NilError(t, database.QueryRowContext(ctx, `SELECT COUNT(*) FROM user_similarity`).Scan(&count))
157+ assert.Equal(t, count, 2, "alice-bob and alice-carol share feeds")
158+}
159+
160+func TestComputeUserSimilarityForUser_MatchesFullCompute(t *testing.T) {
161+ ctx := context.Background()
162+
163+ type simPair struct {
164+ userA, userB string
165+ jaccard float64
166+ commonFeeds int
167+ }
168+
169+ readAll := func(t *testing.T, database *db.DB) []simPair {
170+ t.Helper()
171+ rows, err := database.QueryContext(ctx, `SELECT user_a, user_b, jaccard, common_feeds FROM user_similarity ORDER BY user_a, user_b`)
172+ assert.NilError(t, err)
173+ defer rows.Close()
174+ var result []simPair
175+ for rows.Next() {
176+ var p simPair
177+ assert.NilError(t, rows.Scan(&p.userA, &p.userB, &p.jaccard, &p.commonFeeds))
178+ result = append(result, p)
179+ }
180+ assert.NilError(t, rows.Err())
181+ return result
182+ }
183+
184+ database1 := setupClusterTestDB(t)
185+ seedClusterData(t, ctx, database1)
186+ engine1 := NewEngine(database1.DB, slog.Default())
187+ assert.NilError(t, engine1.ComputeUserSimilarity(ctx))
188+ fullPairs := readAll(t, database1)
189+
190+ database2 := setupClusterTestDB(t)
191+ seedClusterData(t, ctx, database2)
192+ engine2 := NewEngine(database2.DB, slog.Default())
193+ assert.NilError(t, engine2.ComputeUserSimilarityForUser(ctx, "did:test:alice"))
194+ assert.NilError(t, engine2.ComputeUserSimilarityForUser(ctx, "did:test:bob"))
195+ assert.NilError(t, engine2.ComputeUserSimilarityForUser(ctx, "did:test:carol"))
196+ incrPairs := readAll(t, database2)
197+
198+ assert.Equal(t, len(fullPairs), len(incrPairs), "same number of pairs")
199+ for i := range fullPairs {
200+ assert.Equal(t, fullPairs[i].userA, incrPairs[i].userA)
201+ assert.Equal(t, fullPairs[i].userB, incrPairs[i].userB)
202+ assert.Equal(t, fullPairs[i].commonFeeds, incrPairs[i].commonFeeds)
203+ }
204+}
205+
206+func TestComputeUserSimilarityForUser_UpdatesOnSubscriptionChange(t *testing.T) {
207+ ctx := context.Background()
208+ database := setupClusterTestDB(t)
209+ seedClusterData(t, ctx, database)
210+
211+ engine := NewEngine(database.DB, slog.Default())
212+ assert.NilError(t, engine.ComputeUserSimilarity(ctx))
213+
214+ _, err := database.ExecContext(ctx, `INSERT INTO subscriptions (user_did, feed_url) VALUES (?, ?)`, "did:test:carol", "https://a.com/feed")
215+ assert.NilError(t, err)
216+
217+ assert.NilError(t, engine.ComputeUserSimilarityForUser(ctx, "did:test:carol"))
218+
219+ var jaccard float64
220+ var common int
221+ assert.NilError(t, database.QueryRowContext(ctx,
222+ `SELECT jaccard, common_feeds FROM user_similarity WHERE user_a = ? AND user_b = ?`,
223+ "did:test:alice", "did:test:carol").Scan(&jaccard, &common))
224+ assert.Equal(t, common, 2, "alice and carol now share feed A and feed C")
225+}
226+
227+func TestComputeUserSimilarityForUser_WithFollowBoost(t *testing.T) {
228+ ctx := context.Background()
229+ database := setupClusterTestDB(t)
230+ seedClusterData(t, ctx, database)
231+
232+ _, err := database.ExecContext(ctx, `INSERT INTO follows (user_did, target_did) VALUES (?, ?)`, "did:test:alice", "did:test:bob")
233+ assert.NilError(t, err)
234+
235+ engine := NewEngine(database.DB, slog.Default())
236+ assert.NilError(t, engine.ComputeUserSimilarityForUser(ctx, "did:test:alice"))
237+
238+ var jaccard float64
239+ assert.NilError(t, database.QueryRowContext(ctx,
240+ `SELECT jaccard FROM user_similarity WHERE user_a = ? AND user_b = ?`,
241+ "did:test:alice", "did:test:bob").Scan(&jaccard))
242+ assert.Assert(t, jaccard > 0.6, "follow boost should add 0.5 to subscription jaccard, got %f", jaccard)
243+}
244+
245+func TestComputeRecommendationsForUser(t *testing.T) {
246+ ctx := context.Background()
247+ database := setupClusterTestDB(t)
248+ seedClusterData(t, ctx, database)
249+
250+ engine := NewEngine(database.DB, slog.Default())
251+ assert.NilError(t, engine.ComputeUserSimilarity(ctx))
252+ assert.NilError(t, engine.ComputeRecommendationsForUser(ctx, "did:test:carol"))
253+
254+ recs, err := engine.GetFeedRecommendations(ctx, "did:test:carol", 10)
255+ assert.NilError(t, err)
256+ assert.Assert(t, len(recs) > 0, "carol should get feed recommendations")
257+
258+ var found bool
259+ for _, r := range recs {
260+ if r["feed_url"] == "https://a.com/feed" || r["feed_url"] == "https://b.com/feed" {
261+ found = true
262+ }
263+ }
264+ assert.Assert(t, found, "carol should be recommended feeds from alice")
265+}
266+
267+func TestComputeRecommendationsForUser_MatchesFullCompute(t *testing.T) {
268+ ctx := context.Background()
269+
270+ readRecs := func(t *testing.T, database *db.DB, did string) map[string]float64 {
271+ t.Helper()
272+ rows, err := database.QueryContext(ctx,
273+ `SELECT feed_url, score FROM user_feed_recommendations WHERE user_did = ? ORDER BY feed_url`, did)
274+ assert.NilError(t, err)
275+ defer rows.Close()
276+ result := map[string]float64{}
277+ for rows.Next() {
278+ var url string
279+ var score float64
280+ assert.NilError(t, rows.Scan(&url, &score))
281+ result[url] = score
282+ }
283+ assert.NilError(t, rows.Err())
284+ return result
285+ }
286+
287+ database1 := setupClusterTestDB(t)
288+ seedClusterData(t, ctx, database1)
289+ engine1 := NewEngine(database1.DB, slog.Default())
290+ assert.NilError(t, engine1.ComputeUserSimilarity(ctx))
291+ assert.NilError(t, engine1.ComputeRecommendations(ctx))
292+ fullRecs := readRecs(t, database1, "did:test:carol")
293+
294+ database2 := setupClusterTestDB(t)
295+ seedClusterData(t, ctx, database2)
296+ engine2 := NewEngine(database2.DB, slog.Default())
297+ assert.NilError(t, engine2.ComputeUserSimilarity(ctx))
298+ assert.NilError(t, engine2.ComputeRecommendationsForUser(ctx, "did:test:carol"))
299+ incrRecs := readRecs(t, database2, "did:test:carol")
300+
301+ assert.Equal(t, len(fullRecs), len(incrRecs), "same number of recommendations")
302+ for url, score := range fullRecs {
303+ incrScore, ok := incrRecs[url]
304+ assert.Assert(t, ok, "missing recommendation for %s", url)
305+ assert.Equal(t, score, incrScore, "score mismatch for %s", url)
306+ }
307+}
modified internal/server/profile_handler.go +22 -1
@@ -2,12 +2,33 @@ package server
22
33 import (
44 "net/http"
5+ "strings"
56
67 "github.com/go-chi/chi/v5"
8+
9+ "pkg.rbrt.fr/glean/internal/atproto"
710 )
811
912 func (s *Server) handleProfile(w http.ResponseWriter, r *http.Request) {
10- did := chi.URLParam(r, "did")
13+ param := chi.URLParam(r, "did")
14+
15+ var did string
16+ if strings.HasPrefix(param, "did:") {
17+ did = param
18+ } else {
19+ profileUser, err := s.db.GetUserByHandle(r.Context(), param)
20+ if err == nil {
21+ did = profileUser.DID
22+ } else {
23+ resolved, err := atproto.ResolveHandle(r.Context(), param)
24+ if err != nil {
25+ http.Error(w, "handle not found", http.StatusNotFound)
26+ return
27+ }
28+ did = resolved
29+ }
30+ }
31+
1132 profileUser, err := s.db.GetUser(r.Context(), did)
1233 if err != nil {
1334 http.Error(w, err.Error(), http.StatusNotFound)
@@ -2,12 +2,33 @@ package server
2 2
3 import (3 import (
4 "net/http"4 "net/http"
5+ "strings"
5 6
6 "github.com/go-chi/chi/v5"7 "github.com/go-chi/chi/v5"
8+
9+ "pkg.rbrt.fr/glean/internal/atproto"
7 )10 )
8 11
9 func (s *Server) handleProfile(w http.ResponseWriter, r *http.Request) {12 func (s *Server) handleProfile(w http.ResponseWriter, r *http.Request) {
10- did := chi.URLParam(r, "did")13+ param := chi.URLParam(r, "did")
14+
15+ var did string
16+ if strings.HasPrefix(param, "did:") {
17+ did = param
18+ } else {
19+ profileUser, err := s.db.GetUserByHandle(r.Context(), param)
20+ if err == nil {
21+ did = profileUser.DID
22+ } else {
23+ resolved, err := atproto.ResolveHandle(r.Context(), param)
24+ if err != nil {
25+ http.Error(w, "handle not found", http.StatusNotFound)
26+ return
27+ }
28+ did = resolved
29+ }
30+ }
31+
11 profileUser, err := s.db.GetUser(r.Context(), did)32 profileUser, err := s.db.GetUser(r.Context(), did)
12 if err != nil {33 if err != nil {
13 http.Error(w, err.Error(), http.StatusNotFound)34 http.Error(w, err.Error(), http.StatusNotFound)
modified internal/server/server.go +3 -6
@@ -389,13 +389,10 @@ func (s *Server) syncUserInBackground(userDID string, client *atproto.Client) {
389389 s.logger.Error("background sync failed", "error", err, "did", userDID)
390390 }
391391
392- if !isNewUser {
393- return
392+ if isNewUser {
393+ s.refreshUserFeeds(ctx, userDID)
394394 }
395-
396- // if the user is new, but has value from the PDS, we should backfill and refetch their data.
397- s.refreshUserFeeds(ctx, userDID)
398- s.engine.ComputeAll(ctx)
395+ s.engine.ComputeForUser(ctx, userDID)
399396 }()
400397 }
401398
@@ -389,13 +389,10 @@ func (s *Server) syncUserInBackground(userDID string, client *atproto.Client) {
389 s.logger.Error("background sync failed", "error", err, "did", userDID)389 s.logger.Error("background sync failed", "error", err, "did", userDID)
390 }390 }
391 391
392- if !isNewUser {392+ if isNewUser {
393- return393+ s.refreshUserFeeds(ctx, userDID)
394 }394 }
395-395+ s.engine.ComputeForUser(ctx, userDID)
396- // if the user is new, but has value from the PDS, we should backfill and refetch their data.
397- s.refreshUserFeeds(ctx, userDID)
398- s.engine.ComputeAll(ctx)
399 }()396 }()
400 }397 }
401 398