rickub
nandi/gleanpublic⑂ Fork 0
⑂ 078b9a3
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.

Remove per-user clustering and reduce default interval to 10mUnverified

Julien Robert committed 2026-04-21T17:58:48+02:00 Browse files
078b9a3 parent: 8b00ca2
modified .env.example +1 -1
@@ -3,7 +3,7 @@ GLEAN_DB=glean.db
33 GLEAN_JETSTREAM=wss://jetstream.glean.at
44 GLEAN_PLC_URL=https://didplc.glean.at
55 GLEAN_SYNC_INTERVAL=1h
6-GLEAN_CLUSTER_INTERVAL=6h
6+GLEAN_CLUSTER_INTERVAL=10m
77 GLEAN_COLLECTION_DIR_URL=https://lightrail.microcosm.blue/xrpc/com.atproto.sync.listReposByCollection?collection=at.glean.subscription
88 # Leave empty for localhost OAuth (development)
99 # GLEAN_OAUTH_CLIENT_ID=https://glean.at/oauth/client-metadata
@@ -3,7 +3,7 @@ GLEAN_DB=glean.db
3 GLEAN_JETSTREAM=wss://jetstream.glean.at3 GLEAN_JETSTREAM=wss://jetstream.glean.at
4 GLEAN_PLC_URL=https://didplc.glean.at4 GLEAN_PLC_URL=https://didplc.glean.at
5 GLEAN_SYNC_INTERVAL=1h5 GLEAN_SYNC_INTERVAL=1h
6-GLEAN_CLUSTER_INTERVAL=6h6+GLEAN_CLUSTER_INTERVAL=10m
7 GLEAN_COLLECTION_DIR_URL=https://lightrail.microcosm.blue/xrpc/com.atproto.sync.listReposByCollection?collection=at.glean.subscription7 GLEAN_COLLECTION_DIR_URL=https://lightrail.microcosm.blue/xrpc/com.atproto.sync.listReposByCollection?collection=at.glean.subscription
8 # Leave empty for localhost OAuth (development)8 # Leave empty for localhost OAuth (development)
9 # GLEAN_OAUTH_CLIENT_ID=https://glean.at/oauth/client-metadata9 # GLEAN_OAUTH_CLIENT_ID=https://glean.at/oauth/client-metadata
modified docs/specs.md +1 -1
@@ -714,7 +714,7 @@ For larger scale, move to MinHash + LSH (banded hashing) to approximate Jaccard
714714
715715 ### 7.5 Clustering Engine (Cron)
716716
717-A background goroutine runs on a configurable schedule (`GLEAN_CLUSTER_INTERVAL`, default 6h):
717+A background goroutine runs on a configurable schedule (`GLEAN_CLUSTER_INTERVAL`, default 10m):
718718
719719 1. **Compute feed similarity**: Batch-update the `feed_similarity` table (Jaccard over subscriber sets)
720720 2. **Compute user similarity**: Batch-update the `user_similarity` table (Jaccard over subscription sets, boosted by follow relationships)
@@ -714,7 +714,7 @@ For larger scale, move to MinHash + LSH (banded hashing) to approximate Jaccard
714 714
715 ### 7.5 Clustering Engine (Cron)715 ### 7.5 Clustering Engine (Cron)
716 716
717-A background goroutine runs on a configurable schedule (`GLEAN_CLUSTER_INTERVAL`, default 6h):717+A background goroutine runs on a configurable schedule (`GLEAN_CLUSTER_INTERVAL`, default 10m):
718 718
719 1. **Compute feed similarity**: Batch-update the `feed_similarity` table (Jaccard over subscriber sets)719 1. **Compute feed similarity**: Batch-update the `feed_similarity` table (Jaccard over subscriber sets)
720 2. **Compute user similarity**: Batch-update the `user_similarity` table (Jaccard over subscription sets, boosted by follow relationships)720 2. **Compute user similarity**: Batch-update the `user_similarity` table (Jaccard over subscription sets, boosted by follow relationships)
modified internal/cluster/jaccard.go +0 -308
@@ -79,21 +79,6 @@ func (e *Engine) ComputeArticleRecommendations(ctx context.Context) error {
7979 return tx.Commit()
8080 }
8181
82-func (e *Engine) ComputeForUser(ctx context.Context, userDID string) {
83- if !e.mu.TryLock() {
84- e.logger.Info("skipping ComputeForUser: already in progress", "did", userDID)
85- return
86- }
87- defer e.mu.Unlock()
88-
89- if err := e.ComputeUserSimilarityForUser(ctx, userDID); err != nil {
90- e.logger.Error("per-user similarity failed", "error", err, "did", userDID)
91- }
92- if err := e.ComputeRecommendationsForUser(ctx, userDID); err != nil {
93- e.logger.Error("per-user recommendations failed", "error", err, "did", userDID)
94- }
95-}
96-
9782 func (e *Engine) ComputeFeedSimilarity(ctx context.Context) error {
9883 tx, err := e.db.BeginTx(ctx, nil)
9984 if err != nil {
@@ -416,299 +401,6 @@ func (e *Engine) ComputeUserSimilarity(ctx context.Context) error {
416401 return tx.Commit()
417402 }
418403
419-func (e *Engine) ComputeUserSimilarityForUser(ctx context.Context, userDID string) error {
420- tx, err := e.db.BeginTx(ctx, nil)
421- if err != nil {
422- return err
423- }
424- defer func() { _ = tx.Rollback() }()
425-
426- if _, err := tx.ExecContext(ctx, `DELETE FROM user_similarity WHERE user_a = ? OR user_b = ?`, userDID, userDID); err != nil {
427- return err
428- }
429-
430- _, err = tx.ExecContext(ctx, `
431- INSERT INTO user_similarity (user_a, user_b, jaccard, common_feeds)
432- SELECT
433- MIN(?, s2.user_did),
434- MAX(?, s2.user_did),
435- CAST(COUNT(*) AS REAL) / (
436- (SELECT COUNT(*) FROM subscriptions WHERE user_did = ?) +
437- (SELECT COUNT(*) FROM subscriptions WHERE user_did = s2.user_did) -
438- CAST(COUNT(*) AS REAL)
439- ),
440- COUNT(*)
441- FROM subscriptions s1
442- JOIN subscriptions s2 ON s1.feed_url = s2.feed_url AND s2.user_did != ?
443- WHERE s1.user_did = ?
444- GROUP BY s2.user_did
445- `, userDID, userDID, userDID, userDID, userDID)
446- if err != nil {
447- return err
448- }
449-
450- if _, err := tx.ExecContext(ctx, `
451- CREATE TEMP TABLE IF NOT EXISTS _per_user_likes_count (author_did TEXT PRIMARY KEY, cnt INT)
452- `); err != nil {
453- return err
454- }
455- if _, err := tx.ExecContext(ctx, `DELETE FROM _per_user_likes_count`); err != nil {
456- return err
457- }
458- if _, err := tx.ExecContext(ctx, `
459- INSERT INTO _per_user_likes_count (author_did, cnt)
460- SELECT author_did, COUNT(*) FROM likes
461- WHERE author_did = ? OR author_did IN (
462- SELECT CASE WHEN user_a = ? THEN user_b ELSE user_a END
463- FROM user_similarity
464- WHERE user_a = ? OR user_b = ?
465- )
466- GROUP BY author_did
467- `, userDID, userDID, userDID, userDID); err != nil {
468- return err
469- }
470-
471- if _, err := tx.ExecContext(ctx, `
472- CREATE TEMP TABLE IF NOT EXISTS _per_user_likes_overlap (peer TEXT, common INT)
473- `); err != nil {
474- return err
475- }
476- if _, err := tx.ExecContext(ctx, `DELETE FROM _per_user_likes_overlap`); err != nil {
477- return err
478- }
479- if _, err := tx.ExecContext(ctx, `
480- INSERT INTO _per_user_likes_overlap (peer, common)
481- SELECT l2.author_did, COUNT(*)
482- FROM likes l1
483- JOIN likes l2 ON l1.feed_url = l2.feed_url AND l1.article_url = l2.article_url
484- AND l2.author_did != ?
485- WHERE l1.author_did = ?
486- GROUP BY l2.author_did
487- `, userDID, userDID); err != nil {
488- return err
489- }
490-
491- likesPerUser := fmt.Sprintf(`
492- UPDATE user_similarity SET
493- jaccard = jaccard + %g * CAST(_per_user_likes_overlap.common AS REAL) / NULLIF(
494- (SELECT cnt FROM _per_user_likes_count WHERE author_did = ?) +
495- (SELECT cnt FROM _per_user_likes_count WHERE author_did =
496- CASE WHEN user_similarity.user_a = ? THEN user_similarity.user_b ELSE user_similarity.user_a END
497- ) - CAST(_per_user_likes_overlap.common AS REAL),
498- 0
499- ),
500- common_likes = _per_user_likes_overlap.common
501- FROM _per_user_likes_overlap
502- WHERE user_similarity.user_a = _per_user_likes_overlap.peer
503- OR user_similarity.user_b = _per_user_likes_overlap.peer
504- `, e.config.LikesWeight)
505-
506- if _, err := tx.ExecContext(ctx, likesPerUser, userDID, userDID); err != nil {
507- return err
508- }
509-
510- likesInsert := fmt.Sprintf(`
511- INSERT INTO user_similarity (user_a, user_b, jaccard, common_feeds, common_likes)
512- SELECT sub.user_a, sub.user_b, sub.jaccard, 0, sub.common
513- FROM (
514- SELECT
515- MIN(?, lo.peer) AS user_a,
516- MAX(?, lo.peer) AS user_b,
517- %g * CAST(lo.common AS REAL) / NULLIF(
518- (SELECT cnt FROM _per_user_likes_count WHERE author_did = ?) +
519- (SELECT cnt FROM _per_user_likes_count WHERE author_did = lo.peer) -
520- CAST(lo.common AS REAL),
521- 0
522- ) AS jaccard,
523- lo.common
524- FROM _per_user_likes_overlap lo
525- ) sub WHERE 1
526- ON CONFLICT(user_a, user_b) DO UPDATE SET
527- jaccard = jaccard + excluded.jaccard,
528- common_likes = excluded.common_likes
529- `, e.config.LikesWeight)
530-
531- if _, err := tx.ExecContext(ctx, likesInsert, userDID, userDID, userDID); err != nil {
532- return err
533- }
534-
535- if _, err := tx.ExecContext(ctx, `CREATE TEMP TABLE IF NOT EXISTS _user_tag_overlap (peer TEXT, common INT)`); err != nil {
536- return err
537- }
538- if _, err := tx.ExecContext(ctx, `DELETE FROM _user_tag_overlap`); err != nil {
539- return err
540- }
541-
542- _, err = tx.ExecContext(ctx, `
543- INSERT INTO _user_tag_overlap (peer, common)
544- WITH user_tags AS (
545- SELECT author_did, TRIM(value) AS tag FROM annotations, json_each('["' || REPLACE(tags, ',', '","') || '"]')
546- WHERE tags IS NOT NULL AND tags != ''
547- )
548- SELECT t2.author_did, COUNT(DISTINCT t1.tag)
549- FROM user_tags t1
550- JOIN user_tags t2 ON t1.tag = t2.tag AND t2.author_did != ?
551- WHERE t1.author_did = ?
552- GROUP BY t2.author_did
553- `, userDID, userDID)
554- if err != nil {
555- return err
556- }
557-
558- _, err = tx.ExecContext(ctx, `
559- INSERT OR IGNORE INTO user_similarity (user_a, user_b, jaccard, common_feeds, common_tags)
560- SELECT MIN(?, peer), MAX(?, peer), 0, 0, 0 FROM _user_tag_overlap
561- `, userDID, userDID)
562- if err != nil {
563- return err
564- }
565-
566- if _, err := tx.ExecContext(ctx, `
567- CREATE TEMP TABLE IF NOT EXISTS _per_user_tag_count (author_did TEXT PRIMARY KEY, cnt INT)
568- `); err != nil {
569- return err
570- }
571- if _, err := tx.ExecContext(ctx, `DELETE FROM _per_user_tag_count`); err != nil {
572- return err
573- }
574- if _, err := tx.ExecContext(ctx, `
575- INSERT INTO _per_user_tag_count (author_did, cnt)
576- WITH user_tags AS (
577- SELECT author_did, TRIM(value) AS tag FROM annotations, json_each('["' || REPLACE(tags, ',', '","') || '"]')
578- WHERE tags IS NOT NULL AND tags != ''
579- )
580- SELECT author_did, COUNT(DISTINCT tag) FROM user_tags
581- WHERE author_did = ? OR author_did IN (SELECT peer FROM _user_tag_overlap)
582- GROUP BY author_did
583- `, userDID); err != nil {
584- return err
585- }
586-
587- tagsPerUser := fmt.Sprintf(`
588- UPDATE user_similarity SET
589- jaccard = jaccard + %g * CAST(_user_tag_overlap.common AS REAL) / NULLIF(
590- (SELECT cnt FROM _per_user_tag_count WHERE author_did = ?) +
591- (SELECT cnt FROM _per_user_tag_count WHERE author_did =
592- CASE WHEN user_similarity.user_a = ? THEN user_similarity.user_b ELSE user_similarity.user_a END
593- ) - CAST(_user_tag_overlap.common AS REAL),
594- 0
595- ),
596- common_tags = _user_tag_overlap.common
597- FROM _user_tag_overlap
598- WHERE user_similarity.user_a = _user_tag_overlap.peer
599- OR user_similarity.user_b = _user_tag_overlap.peer
600- `, e.config.TagsWeight)
601-
602- if _, err := tx.ExecContext(ctx, tagsPerUser, userDID, userDID); err != nil {
603- return err
604- }
605-
606- followQuery := fmt.Sprintf(`
607- INSERT INTO user_similarity (user_a, user_b, jaccard, common_feeds, common_likes, common_tags)
608- SELECT
609- MIN(?, peer_did),
610- MAX(?, peer_did),
611- %g,
612- 0, 0, 0
613- FROM (
614- SELECT target_did AS peer_did FROM follows WHERE user_did = ? AND target_did != ?
615- UNION
616- SELECT user_did AS peer_did FROM follows WHERE target_did = ? AND user_did != ?
617- )
618- GROUP BY MIN(?, peer_did), MAX(?, peer_did)
619- ON CONFLICT(user_a, user_b) DO UPDATE SET
620- jaccard = jaccard + %g
621- `, e.config.FollowBoost, e.config.FollowBoost)
622-
623- if _, err := tx.ExecContext(ctx, followQuery, userDID, userDID, userDID, userDID, userDID, userDID, userDID, userDID); err != nil {
624- return err
625- }
626-
627- e.logger.Info("per-user similarity computed", "did", userDID)
628- return tx.Commit()
629-}
630-
631-func (e *Engine) ComputeRecommendationsForUser(ctx context.Context, userDID string) error {
632- tx, err := e.db.BeginTx(ctx, nil)
633- if err != nil {
634- return err
635- }
636- defer func() { _ = tx.Rollback() }()
637-
638- if _, err := tx.ExecContext(ctx, `DELETE FROM user_feed_recommendations WHERE user_did = ?`, userDID); err != nil {
639- return err
640- }
641-
642- recQuery := fmt.Sprintf(`
643- INSERT INTO user_feed_recommendations (user_did, feed_url, score)
644- SELECT ?, s.feed_url, SUM(us.jaccard) AS score
645- FROM user_similarity us
646- JOIN subscriptions s ON s.user_did = CASE
647- WHEN us.user_a = ? THEN us.user_b
648- ELSE us.user_a
649- END
650- WHERE (us.user_a = ? OR us.user_b = ?)
651- AND us.jaccard > %g
652- AND s.feed_url NOT IN (SELECT feed_url FROM subscriptions WHERE user_did = ?)
653- GROUP BY s.feed_url
654- ORDER BY score DESC
655- `, e.config.SimilarityThreshold)
656-
657- if _, err := tx.ExecContext(ctx, recQuery, userDID, userDID, userDID, userDID, userDID); err != nil {
658- return err
659- }
660-
661- if err := tx.Commit(); err != nil {
662- return err
663- }
664-
665- if err := e.computeArticleRecommendationsForUser(ctx, userDID); err != nil {
666- return err
667- }
668-
669- e.logger.Info("per-user recommendations computed", "did", userDID)
670- return nil
671-}
672-
673-func (e *Engine) computeArticleRecommendationsForUser(ctx context.Context, userDID string) error {
674- tx, err := e.db.BeginTx(ctx, nil)
675- if err != nil {
676- return err
677- }
678- defer func() { _ = tx.Rollback() }()
679-
680- if _, err := tx.ExecContext(ctx, `DELETE FROM user_article_recommendations WHERE user_did = ?`, userDID); err != nil {
681- return err
682- }
683-
684- artQuery := fmt.Sprintf(`
685- INSERT INTO user_article_recommendations (user_did, feed_url, article_url, score)
686- SELECT ?, l.feed_url, l.article_url, SUM(us.jaccard) AS score
687- FROM (
688- SELECT us.user_b AS peer, us.jaccard
689- FROM user_similarity us WHERE us.user_a = ? AND us.jaccard > %g
690- UNION ALL
691- SELECT us.user_a AS peer, us.jaccard
692- FROM user_similarity us WHERE us.user_b = ? AND us.jaccard > %g
693- ) us
694- JOIN likes l ON l.author_did = us.peer
695- WHERE NOT EXISTS (
696- SELECT 1 FROM subscriptions sub WHERE sub.user_did = ? AND sub.feed_url = l.feed_url
697- )
698- AND NOT EXISTS (
699- SELECT 1 FROM likes ul WHERE ul.author_did = ? AND ul.feed_url = l.feed_url AND ul.article_url = l.article_url
700- )
701- GROUP BY l.feed_url, l.article_url
702- ORDER BY score DESC
703- `, e.config.SimilarityThreshold, e.config.SimilarityThreshold)
704-
705- if _, err := tx.ExecContext(ctx, artQuery, userDID, userDID, userDID, userDID, userDID); err != nil {
706- return err
707- }
708-
709- return tx.Commit()
710-}
711-
712404 func (e *Engine) ComputeRecommendations(ctx context.Context) error {
713405 tx, err := e.db.BeginTx(ctx, nil)
714406 if err != nil {
@@ -79,21 +79,6 @@ func (e *Engine) ComputeArticleRecommendations(ctx context.Context) error {
79 return tx.Commit()79 return tx.Commit()
80 }80 }
81 81
82-func (e *Engine) ComputeForUser(ctx context.Context, userDID string) {
83- if !e.mu.TryLock() {
84- e.logger.Info("skipping ComputeForUser: already in progress", "did", userDID)
85- return
86- }
87- defer e.mu.Unlock()
88-
89- if err := e.ComputeUserSimilarityForUser(ctx, userDID); err != nil {
90- e.logger.Error("per-user similarity failed", "error", err, "did", userDID)
91- }
92- if err := e.ComputeRecommendationsForUser(ctx, userDID); err != nil {
93- e.logger.Error("per-user recommendations failed", "error", err, "did", userDID)
94- }
95-}
96-
97 func (e *Engine) ComputeFeedSimilarity(ctx context.Context) error {82 func (e *Engine) ComputeFeedSimilarity(ctx context.Context) error {
98 tx, err := e.db.BeginTx(ctx, nil)83 tx, err := e.db.BeginTx(ctx, nil)
99 if err != nil {84 if err != nil {
@@ -416,299 +401,6 @@ func (e *Engine) ComputeUserSimilarity(ctx context.Context) error {
416 return tx.Commit()401 return tx.Commit()
417 }402 }
418 403
419-func (e *Engine) ComputeUserSimilarityForUser(ctx context.Context, userDID string) error {
420- tx, err := e.db.BeginTx(ctx, nil)
421- if err != nil {
422- return err
423- }
424- defer func() { _ = tx.Rollback() }()
425-
426- if _, err := tx.ExecContext(ctx, `DELETE FROM user_similarity WHERE user_a = ? OR user_b = ?`, userDID, userDID); err != nil {
427- return err
428- }
429-
430- _, err = tx.ExecContext(ctx, `
431- INSERT INTO user_similarity (user_a, user_b, jaccard, common_feeds)
432- SELECT
433- MIN(?, s2.user_did),
434- MAX(?, s2.user_did),
435- CAST(COUNT(*) AS REAL) / (
436- (SELECT COUNT(*) FROM subscriptions WHERE user_did = ?) +
437- (SELECT COUNT(*) FROM subscriptions WHERE user_did = s2.user_did) -
438- CAST(COUNT(*) AS REAL)
439- ),
440- COUNT(*)
441- FROM subscriptions s1
442- JOIN subscriptions s2 ON s1.feed_url = s2.feed_url AND s2.user_did != ?
443- WHERE s1.user_did = ?
444- GROUP BY s2.user_did
445- `, userDID, userDID, userDID, userDID, userDID)
446- if err != nil {
447- return err
448- }
449-
450- if _, err := tx.ExecContext(ctx, `
451- CREATE TEMP TABLE IF NOT EXISTS _per_user_likes_count (author_did TEXT PRIMARY KEY, cnt INT)
452- `); err != nil {
453- return err
454- }
455- if _, err := tx.ExecContext(ctx, `DELETE FROM _per_user_likes_count`); err != nil {
456- return err
457- }
458- if _, err := tx.ExecContext(ctx, `
459- INSERT INTO _per_user_likes_count (author_did, cnt)
460- SELECT author_did, COUNT(*) FROM likes
461- WHERE author_did = ? OR author_did IN (
462- SELECT CASE WHEN user_a = ? THEN user_b ELSE user_a END
463- FROM user_similarity
464- WHERE user_a = ? OR user_b = ?
465- )
466- GROUP BY author_did
467- `, userDID, userDID, userDID, userDID); err != nil {
468- return err
469- }
470-
471- if _, err := tx.ExecContext(ctx, `
472- CREATE TEMP TABLE IF NOT EXISTS _per_user_likes_overlap (peer TEXT, common INT)
473- `); err != nil {
474- return err
475- }
476- if _, err := tx.ExecContext(ctx, `DELETE FROM _per_user_likes_overlap`); err != nil {
477- return err
478- }
479- if _, err := tx.ExecContext(ctx, `
480- INSERT INTO _per_user_likes_overlap (peer, common)
481- SELECT l2.author_did, COUNT(*)
482- FROM likes l1
483- JOIN likes l2 ON l1.feed_url = l2.feed_url AND l1.article_url = l2.article_url
484- AND l2.author_did != ?
485- WHERE l1.author_did = ?
486- GROUP BY l2.author_did
487- `, userDID, userDID); err != nil {
488- return err
489- }
490-
491- likesPerUser := fmt.Sprintf(`
492- UPDATE user_similarity SET
493- jaccard = jaccard + %g * CAST(_per_user_likes_overlap.common AS REAL) / NULLIF(
494- (SELECT cnt FROM _per_user_likes_count WHERE author_did = ?) +
495- (SELECT cnt FROM _per_user_likes_count WHERE author_did =
496- CASE WHEN user_similarity.user_a = ? THEN user_similarity.user_b ELSE user_similarity.user_a END
497- ) - CAST(_per_user_likes_overlap.common AS REAL),
498- 0
499- ),
500- common_likes = _per_user_likes_overlap.common
501- FROM _per_user_likes_overlap
502- WHERE user_similarity.user_a = _per_user_likes_overlap.peer
503- OR user_similarity.user_b = _per_user_likes_overlap.peer
504- `, e.config.LikesWeight)
505-
506- if _, err := tx.ExecContext(ctx, likesPerUser, userDID, userDID); err != nil {
507- return err
508- }
509-
510- likesInsert := fmt.Sprintf(`
511- INSERT INTO user_similarity (user_a, user_b, jaccard, common_feeds, common_likes)
512- SELECT sub.user_a, sub.user_b, sub.jaccard, 0, sub.common
513- FROM (
514- SELECT
515- MIN(?, lo.peer) AS user_a,
516- MAX(?, lo.peer) AS user_b,
517- %g * CAST(lo.common AS REAL) / NULLIF(
518- (SELECT cnt FROM _per_user_likes_count WHERE author_did = ?) +
519- (SELECT cnt FROM _per_user_likes_count WHERE author_did = lo.peer) -
520- CAST(lo.common AS REAL),
521- 0
522- ) AS jaccard,
523- lo.common
524- FROM _per_user_likes_overlap lo
525- ) sub WHERE 1
526- ON CONFLICT(user_a, user_b) DO UPDATE SET
527- jaccard = jaccard + excluded.jaccard,
528- common_likes = excluded.common_likes
529- `, e.config.LikesWeight)
530-
531- if _, err := tx.ExecContext(ctx, likesInsert, userDID, userDID, userDID); err != nil {
532- return err
533- }
534-
535- if _, err := tx.ExecContext(ctx, `CREATE TEMP TABLE IF NOT EXISTS _user_tag_overlap (peer TEXT, common INT)`); err != nil {
536- return err
537- }
538- if _, err := tx.ExecContext(ctx, `DELETE FROM _user_tag_overlap`); err != nil {
539- return err
540- }
541-
542- _, err = tx.ExecContext(ctx, `
543- INSERT INTO _user_tag_overlap (peer, common)
544- WITH user_tags AS (
545- SELECT author_did, TRIM(value) AS tag FROM annotations, json_each('["' || REPLACE(tags, ',', '","') || '"]')
546- WHERE tags IS NOT NULL AND tags != ''
547- )
548- SELECT t2.author_did, COUNT(DISTINCT t1.tag)
549- FROM user_tags t1
550- JOIN user_tags t2 ON t1.tag = t2.tag AND t2.author_did != ?
551- WHERE t1.author_did = ?
552- GROUP BY t2.author_did
553- `, userDID, userDID)
554- if err != nil {
555- return err
556- }
557-
558- _, err = tx.ExecContext(ctx, `
559- INSERT OR IGNORE INTO user_similarity (user_a, user_b, jaccard, common_feeds, common_tags)
560- SELECT MIN(?, peer), MAX(?, peer), 0, 0, 0 FROM _user_tag_overlap
561- `, userDID, userDID)
562- if err != nil {
563- return err
564- }
565-
566- if _, err := tx.ExecContext(ctx, `
567- CREATE TEMP TABLE IF NOT EXISTS _per_user_tag_count (author_did TEXT PRIMARY KEY, cnt INT)
568- `); err != nil {
569- return err
570- }
571- if _, err := tx.ExecContext(ctx, `DELETE FROM _per_user_tag_count`); err != nil {
572- return err
573- }
574- if _, err := tx.ExecContext(ctx, `
575- INSERT INTO _per_user_tag_count (author_did, cnt)
576- WITH user_tags AS (
577- SELECT author_did, TRIM(value) AS tag FROM annotations, json_each('["' || REPLACE(tags, ',', '","') || '"]')
578- WHERE tags IS NOT NULL AND tags != ''
579- )
580- SELECT author_did, COUNT(DISTINCT tag) FROM user_tags
581- WHERE author_did = ? OR author_did IN (SELECT peer FROM _user_tag_overlap)
582- GROUP BY author_did
583- `, userDID); err != nil {
584- return err
585- }
586-
587- tagsPerUser := fmt.Sprintf(`
588- UPDATE user_similarity SET
589- jaccard = jaccard + %g * CAST(_user_tag_overlap.common AS REAL) / NULLIF(
590- (SELECT cnt FROM _per_user_tag_count WHERE author_did = ?) +
591- (SELECT cnt FROM _per_user_tag_count WHERE author_did =
592- CASE WHEN user_similarity.user_a = ? THEN user_similarity.user_b ELSE user_similarity.user_a END
593- ) - CAST(_user_tag_overlap.common AS REAL),
594- 0
595- ),
596- common_tags = _user_tag_overlap.common
597- FROM _user_tag_overlap
598- WHERE user_similarity.user_a = _user_tag_overlap.peer
599- OR user_similarity.user_b = _user_tag_overlap.peer
600- `, e.config.TagsWeight)
601-
602- if _, err := tx.ExecContext(ctx, tagsPerUser, userDID, userDID); err != nil {
603- return err
604- }
605-
606- followQuery := fmt.Sprintf(`
607- INSERT INTO user_similarity (user_a, user_b, jaccard, common_feeds, common_likes, common_tags)
608- SELECT
609- MIN(?, peer_did),
610- MAX(?, peer_did),
611- %g,
612- 0, 0, 0
613- FROM (
614- SELECT target_did AS peer_did FROM follows WHERE user_did = ? AND target_did != ?
615- UNION
616- SELECT user_did AS peer_did FROM follows WHERE target_did = ? AND user_did != ?
617- )
618- GROUP BY MIN(?, peer_did), MAX(?, peer_did)
619- ON CONFLICT(user_a, user_b) DO UPDATE SET
620- jaccard = jaccard + %g
621- `, e.config.FollowBoost, e.config.FollowBoost)
622-
623- if _, err := tx.ExecContext(ctx, followQuery, userDID, userDID, userDID, userDID, userDID, userDID, userDID, userDID); err != nil {
624- return err
625- }
626-
627- e.logger.Info("per-user similarity computed", "did", userDID)
628- return tx.Commit()
629-}
630-
631-func (e *Engine) ComputeRecommendationsForUser(ctx context.Context, userDID string) error {
632- tx, err := e.db.BeginTx(ctx, nil)
633- if err != nil {
634- return err
635- }
636- defer func() { _ = tx.Rollback() }()
637-
638- if _, err := tx.ExecContext(ctx, `DELETE FROM user_feed_recommendations WHERE user_did = ?`, userDID); err != nil {
639- return err
640- }
641-
642- recQuery := fmt.Sprintf(`
643- INSERT INTO user_feed_recommendations (user_did, feed_url, score)
644- SELECT ?, s.feed_url, SUM(us.jaccard) AS score
645- FROM user_similarity us
646- JOIN subscriptions s ON s.user_did = CASE
647- WHEN us.user_a = ? THEN us.user_b
648- ELSE us.user_a
649- END
650- WHERE (us.user_a = ? OR us.user_b = ?)
651- AND us.jaccard > %g
652- AND s.feed_url NOT IN (SELECT feed_url FROM subscriptions WHERE user_did = ?)
653- GROUP BY s.feed_url
654- ORDER BY score DESC
655- `, e.config.SimilarityThreshold)
656-
657- if _, err := tx.ExecContext(ctx, recQuery, userDID, userDID, userDID, userDID, userDID); err != nil {
658- return err
659- }
660-
661- if err := tx.Commit(); err != nil {
662- return err
663- }
664-
665- if err := e.computeArticleRecommendationsForUser(ctx, userDID); err != nil {
666- return err
667- }
668-
669- e.logger.Info("per-user recommendations computed", "did", userDID)
670- return nil
671-}
672-
673-func (e *Engine) computeArticleRecommendationsForUser(ctx context.Context, userDID string) error {
674- tx, err := e.db.BeginTx(ctx, nil)
675- if err != nil {
676- return err
677- }
678- defer func() { _ = tx.Rollback() }()
679-
680- if _, err := tx.ExecContext(ctx, `DELETE FROM user_article_recommendations WHERE user_did = ?`, userDID); err != nil {
681- return err
682- }
683-
684- artQuery := fmt.Sprintf(`
685- INSERT INTO user_article_recommendations (user_did, feed_url, article_url, score)
686- SELECT ?, l.feed_url, l.article_url, SUM(us.jaccard) AS score
687- FROM (
688- SELECT us.user_b AS peer, us.jaccard
689- FROM user_similarity us WHERE us.user_a = ? AND us.jaccard > %g
690- UNION ALL
691- SELECT us.user_a AS peer, us.jaccard
692- FROM user_similarity us WHERE us.user_b = ? AND us.jaccard > %g
693- ) us
694- JOIN likes l ON l.author_did = us.peer
695- WHERE NOT EXISTS (
696- SELECT 1 FROM subscriptions sub WHERE sub.user_did = ? AND sub.feed_url = l.feed_url
697- )
698- AND NOT EXISTS (
699- SELECT 1 FROM likes ul WHERE ul.author_did = ? AND ul.feed_url = l.feed_url AND ul.article_url = l.article_url
700- )
701- GROUP BY l.feed_url, l.article_url
702- ORDER BY score DESC
703- `, e.config.SimilarityThreshold, e.config.SimilarityThreshold)
704-
705- if _, err := tx.ExecContext(ctx, artQuery, userDID, userDID, userDID, userDID, userDID); err != nil {
706- return err
707- }
708-
709- return tx.Commit()
710-}
711-
712 func (e *Engine) ComputeRecommendations(ctx context.Context) error {404 func (e *Engine) ComputeRecommendations(ctx context.Context) error {
713 tx, err := e.db.BeginTx(ctx, nil)405 tx, err := e.db.BeginTx(ctx, nil)
714 if err != nil {406 if err != nil {
modified internal/cluster/jaccard_test.go +0 -191
@@ -141,197 +141,6 @@ func TestComputeRecommendations_NoSelfRecommendations(t *testing.T) {
141141 }
142142 }
143143
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 TestComputeUserSimilarityForUser_IncomingFollow(t *testing.T) {
246- ctx := context.Background()
247- database := setupClusterTestDB(t)
248- seedClusterData(t, ctx, database)
249-
250- _, err := database.ExecContext(ctx, `INSERT INTO follows (user_did, target_did) VALUES (?, ?)`, "did:test:bob", "did:test:alice")
251- assert.NilError(t, err)
252-
253- engine := NewEngine(database.DB, slog.Default())
254- assert.NilError(t, engine.ComputeUserSimilarity(ctx))
255-
256- var jaccardBefore float64
257- assert.NilError(t, database.QueryRowContext(ctx,
258- `SELECT jaccard FROM user_similarity WHERE user_a = ? AND user_b = ?`,
259- "did:test:alice", "did:test:bob").Scan(&jaccardBefore))
260-
261- assert.NilError(t, engine.ComputeUserSimilarityForUser(ctx, "did:test:alice"))
262-
263- var jaccardAfter float64
264- assert.NilError(t, database.QueryRowContext(ctx,
265- `SELECT jaccard FROM user_similarity WHERE user_a = ? AND user_b = ?`,
266- "did:test:alice", "did:test:bob").Scan(&jaccardAfter))
267-
268- assert.Assert(t, jaccardAfter > 0, "incoming follow boost should survive per-user recomputation, got %f", jaccardAfter)
269-}
270-
271-func TestComputeRecommendationsForUser(t *testing.T) {
272- ctx := context.Background()
273- database := setupClusterTestDB(t)
274- seedClusterData(t, ctx, database)
275-
276- engine := NewEngine(database.DB, slog.Default())
277- assert.NilError(t, engine.ComputeUserSimilarity(ctx))
278- assert.NilError(t, engine.ComputeRecommendationsForUser(ctx, "did:test:carol"))
279-
280- recs, err := engine.GetFeedRecommendations(ctx, "did:test:carol", 10)
281- assert.NilError(t, err)
282- assert.Assert(t, len(recs) > 0, "carol should get feed recommendations")
283-
284- var found bool
285- for _, r := range recs {
286- if r.FeedURL == "https://a.com/feed" || r.FeedURL == "https://b.com/feed" {
287- found = true
288- }
289- }
290- assert.Assert(t, found, "carol should be recommended feeds from alice")
291-}
292-
293-func TestComputeRecommendationsForUser_MatchesFullCompute(t *testing.T) {
294- ctx := context.Background()
295-
296- readRecs := func(t *testing.T, database *db.DB, did string) map[string]float64 {
297- t.Helper()
298- rows, err := database.QueryContext(ctx,
299- `SELECT feed_url, score FROM user_feed_recommendations WHERE user_did = ? ORDER BY feed_url`, did)
300- assert.NilError(t, err)
301- defer rows.Close()
302- result := map[string]float64{}
303- for rows.Next() {
304- var url string
305- var score float64
306- assert.NilError(t, rows.Scan(&url, &score))
307- result[url] = score
308- }
309- assert.NilError(t, rows.Err())
310- return result
311- }
312-
313- database1 := setupClusterTestDB(t)
314- seedClusterData(t, ctx, database1)
315- engine1 := NewEngine(database1.DB, slog.Default())
316- assert.NilError(t, engine1.ComputeUserSimilarity(ctx))
317- assert.NilError(t, engine1.ComputeRecommendations(ctx))
318- fullRecs := readRecs(t, database1, "did:test:carol")
319-
320- database2 := setupClusterTestDB(t)
321- seedClusterData(t, ctx, database2)
322- engine2 := NewEngine(database2.DB, slog.Default())
323- assert.NilError(t, engine2.ComputeUserSimilarity(ctx))
324- assert.NilError(t, engine2.ComputeRecommendationsForUser(ctx, "did:test:carol"))
325- incrRecs := readRecs(t, database2, "did:test:carol")
326-
327- assert.Equal(t, len(fullRecs), len(incrRecs), "same number of recommendations")
328- for url, score := range fullRecs {
329- incrScore, ok := incrRecs[url]
330- assert.Assert(t, ok, "missing recommendation for %s", url)
331- assert.Equal(t, score, incrScore, "score mismatch for %s", url)
332- }
333-}
334-
335144 func TestLikesBasedSimilarity(t *testing.T) {
336145 ctx := context.Background()
337146 database := setupClusterTestDB(t)
@@ -141,197 +141,6 @@ func TestComputeRecommendations_NoSelfRecommendations(t *testing.T) {
141 }141 }
142 }142 }
143 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 TestComputeUserSimilarityForUser_IncomingFollow(t *testing.T) {
246- ctx := context.Background()
247- database := setupClusterTestDB(t)
248- seedClusterData(t, ctx, database)
249-
250- _, err := database.ExecContext(ctx, `INSERT INTO follows (user_did, target_did) VALUES (?, ?)`, "did:test:bob", "did:test:alice")
251- assert.NilError(t, err)
252-
253- engine := NewEngine(database.DB, slog.Default())
254- assert.NilError(t, engine.ComputeUserSimilarity(ctx))
255-
256- var jaccardBefore float64
257- assert.NilError(t, database.QueryRowContext(ctx,
258- `SELECT jaccard FROM user_similarity WHERE user_a = ? AND user_b = ?`,
259- "did:test:alice", "did:test:bob").Scan(&jaccardBefore))
260-
261- assert.NilError(t, engine.ComputeUserSimilarityForUser(ctx, "did:test:alice"))
262-
263- var jaccardAfter float64
264- assert.NilError(t, database.QueryRowContext(ctx,
265- `SELECT jaccard FROM user_similarity WHERE user_a = ? AND user_b = ?`,
266- "did:test:alice", "did:test:bob").Scan(&jaccardAfter))
267-
268- assert.Assert(t, jaccardAfter > 0, "incoming follow boost should survive per-user recomputation, got %f", jaccardAfter)
269-}
270-
271-func TestComputeRecommendationsForUser(t *testing.T) {
272- ctx := context.Background()
273- database := setupClusterTestDB(t)
274- seedClusterData(t, ctx, database)
275-
276- engine := NewEngine(database.DB, slog.Default())
277- assert.NilError(t, engine.ComputeUserSimilarity(ctx))
278- assert.NilError(t, engine.ComputeRecommendationsForUser(ctx, "did:test:carol"))
279-
280- recs, err := engine.GetFeedRecommendations(ctx, "did:test:carol", 10)
281- assert.NilError(t, err)
282- assert.Assert(t, len(recs) > 0, "carol should get feed recommendations")
283-
284- var found bool
285- for _, r := range recs {
286- if r.FeedURL == "https://a.com/feed" || r.FeedURL == "https://b.com/feed" {
287- found = true
288- }
289- }
290- assert.Assert(t, found, "carol should be recommended feeds from alice")
291-}
292-
293-func TestComputeRecommendationsForUser_MatchesFullCompute(t *testing.T) {
294- ctx := context.Background()
295-
296- readRecs := func(t *testing.T, database *db.DB, did string) map[string]float64 {
297- t.Helper()
298- rows, err := database.QueryContext(ctx,
299- `SELECT feed_url, score FROM user_feed_recommendations WHERE user_did = ? ORDER BY feed_url`, did)
300- assert.NilError(t, err)
301- defer rows.Close()
302- result := map[string]float64{}
303- for rows.Next() {
304- var url string
305- var score float64
306- assert.NilError(t, rows.Scan(&url, &score))
307- result[url] = score
308- }
309- assert.NilError(t, rows.Err())
310- return result
311- }
312-
313- database1 := setupClusterTestDB(t)
314- seedClusterData(t, ctx, database1)
315- engine1 := NewEngine(database1.DB, slog.Default())
316- assert.NilError(t, engine1.ComputeUserSimilarity(ctx))
317- assert.NilError(t, engine1.ComputeRecommendations(ctx))
318- fullRecs := readRecs(t, database1, "did:test:carol")
319-
320- database2 := setupClusterTestDB(t)
321- seedClusterData(t, ctx, database2)
322- engine2 := NewEngine(database2.DB, slog.Default())
323- assert.NilError(t, engine2.ComputeUserSimilarity(ctx))
324- assert.NilError(t, engine2.ComputeRecommendationsForUser(ctx, "did:test:carol"))
325- incrRecs := readRecs(t, database2, "did:test:carol")
326-
327- assert.Equal(t, len(fullRecs), len(incrRecs), "same number of recommendations")
328- for url, score := range fullRecs {
329- incrScore, ok := incrRecs[url]
330- assert.Assert(t, ok, "missing recommendation for %s", url)
331- assert.Equal(t, score, incrScore, "score mismatch for %s", url)
332- }
333-}
334-
335 func TestLikesBasedSimilarity(t *testing.T) {144 func TestLikesBasedSimilarity(t *testing.T) {
336 ctx := context.Background()145 ctx := context.Background()
337 database := setupClusterTestDB(t)146 database := setupClusterTestDB(t)
modified internal/server/server.go +0 -2
@@ -392,7 +392,6 @@ func (s *Server) syncUserInBackground(userDID string, client *atproto.Client) {
392392 if isNewUser {
393393 s.refreshUserFeeds(ctx, userDID)
394394 }
395- s.engine.ComputeForUser(ctx, userDID)
396395 }()
397396 }
398397
@@ -496,7 +495,6 @@ func (s *Server) BackfillFromCollectionDir(ctx context.Context, collectionDirURL
496495 }
497496
498497 s.refreshUserFeeds(ctx, did)
499- s.engine.ComputeForUser(ctx, did)
500498 }
501499
502500 s.logger.Info("collection directory backfill complete")
@@ -392,7 +392,6 @@ func (s *Server) syncUserInBackground(userDID string, client *atproto.Client) {
392 if isNewUser {392 if isNewUser {
393 s.refreshUserFeeds(ctx, userDID)393 s.refreshUserFeeds(ctx, userDID)
394 }394 }
395- s.engine.ComputeForUser(ctx, userDID)
396 }()395 }()
397 }396 }
398 397
@@ -496,7 +495,6 @@ func (s *Server) BackfillFromCollectionDir(ctx context.Context, collectionDirURL
496 }495 }
497 496
498 s.refreshUserFeeds(ctx, did)497 s.refreshUserFeeds(ctx, did)
499- s.engine.ComputeForUser(ctx, did)
500 }498 }
501 499
502 s.logger.Info("collection directory backfill complete")500 s.logger.Info("collection directory backfill complete")
modified main.go +1 -1
@@ -23,7 +23,7 @@ func main() {
2323 dbPath := flag.String("db", envOr("GLEAN_DB", "glean.db"), "database path")
2424 jetstreamURL := flag.String("jetstream", envOr("GLEAN_JETSTREAM", "wss://jetstream.glean.at"), "Jetstream URL")
2525 syncInterval := flag.Duration("sync-interval", envDuration("GLEAN_SYNC_INTERVAL", 1*time.Hour), "PDS sync interval")
26- clusterInterval := flag.Duration("cluster-interval", envDuration("GLEAN_CLUSTER_INTERVAL", 6*time.Hour), "cluster recomputation interval")
26+ clusterInterval := flag.Duration("cluster-interval", envDuration("GLEAN_CLUSTER_INTERVAL", 10*time.Minute), "cluster recomputation interval")
2727 collectionDirURL := flag.String("collection-dir", envOr("GLEAN_COLLECTION_DIR_URL", ""), "collection directory URL for startup backfill")
2828 flag.Parse()
2929
@@ -23,7 +23,7 @@ func main() {
23 dbPath := flag.String("db", envOr("GLEAN_DB", "glean.db"), "database path")23 dbPath := flag.String("db", envOr("GLEAN_DB", "glean.db"), "database path")
24 jetstreamURL := flag.String("jetstream", envOr("GLEAN_JETSTREAM", "wss://jetstream.glean.at"), "Jetstream URL")24 jetstreamURL := flag.String("jetstream", envOr("GLEAN_JETSTREAM", "wss://jetstream.glean.at"), "Jetstream URL")
25 syncInterval := flag.Duration("sync-interval", envDuration("GLEAN_SYNC_INTERVAL", 1*time.Hour), "PDS sync interval")25 syncInterval := flag.Duration("sync-interval", envDuration("GLEAN_SYNC_INTERVAL", 1*time.Hour), "PDS sync interval")
26- clusterInterval := flag.Duration("cluster-interval", envDuration("GLEAN_CLUSTER_INTERVAL", 6*time.Hour), "cluster recomputation interval")26+ clusterInterval := flag.Duration("cluster-interval", envDuration("GLEAN_CLUSTER_INTERVAL", 10*time.Minute), "cluster recomputation interval")
27 collectionDirURL := flag.String("collection-dir", envOr("GLEAN_COLLECTION_DIR_URL", ""), "collection directory URL for startup backfill")27 collectionDirURL := flag.String("collection-dir", envOr("GLEAN_COLLECTION_DIR_URL", ""), "collection directory URL for startup backfill")
28 flag.Parse()28 flag.Parse()
29 29
modified readme.md +1 -1
@@ -43,7 +43,7 @@ Then open `http://localhost:8080`.
4343 | `GLEAN_DB` | `glean.db` | SQLite database path |
4444 | `GLEAN_JETSTREAM` | `wss://jetstream.glean.at` | Jetstream WebSocket URL |
4545 | `GLEAN_SYNC_INTERVAL` | `1h` | PDS sync interval (Go duration: `30m`, `2h30m`, etc.) |
46-| `GLEAN_CLUSTER_INTERVAL` | `6h` | Cluster recomputation interval (Go duration) |
46+| `GLEAN_CLUSTER_INTERVAL` | `10m` | Cluster recomputation interval (Go duration) |
4747 | `GLEAN_PLC_URL` | `https://didplc.glean.at` | PLC directory URL for DID resolution |
4848 | `GLEAN_OAUTH_CLIENT_ID` | _(empty)_ | OAuth client metadata URL (leave empty for localhost dev) |
4949 | `GLEAN_OAUTH_REDIRECT_URL` | _(empty)_ | OAuth redirect URL (leave empty for localhost dev) |
@@ -43,7 +43,7 @@ Then open `http://localhost:8080`.
43 | `GLEAN_DB` | `glean.db` | SQLite database path |43 | `GLEAN_DB` | `glean.db` | SQLite database path |
44 | `GLEAN_JETSTREAM` | `wss://jetstream.glean.at` | Jetstream WebSocket URL |44 | `GLEAN_JETSTREAM` | `wss://jetstream.glean.at` | Jetstream WebSocket URL |
45 | `GLEAN_SYNC_INTERVAL` | `1h` | PDS sync interval (Go duration: `30m`, `2h30m`, etc.) |45 | `GLEAN_SYNC_INTERVAL` | `1h` | PDS sync interval (Go duration: `30m`, `2h30m`, etc.) |
46-| `GLEAN_CLUSTER_INTERVAL` | `6h` | Cluster recomputation interval (Go duration) |46+| `GLEAN_CLUSTER_INTERVAL` | `10m` | Cluster recomputation interval (Go duration) |
47 | `GLEAN_PLC_URL` | `https://didplc.glean.at` | PLC directory URL for DID resolution |47 | `GLEAN_PLC_URL` | `https://didplc.glean.at` | PLC directory URL for DID resolution |
48 | `GLEAN_OAUTH_CLIENT_ID` | _(empty)_ | OAuth client metadata URL (leave empty for localhost dev) |48 | `GLEAN_OAUTH_CLIENT_ID` | _(empty)_ | OAuth client metadata URL (leave empty for localhost dev) |
49 | `GLEAN_OAUTH_REDIRECT_URL` | _(empty)_ | OAuth redirect URL (leave empty for localhost dev) |49 | `GLEAN_OAUTH_REDIRECT_URL` | _(empty)_ | OAuth redirect URL (leave empty for localhost dev) |