nandi/glean · 9cacbcf

modified docs/specs.md +17 -6 Viewed

@@ -701,9 +701,9 @@ J(U1, U2) = jaccard_subscriptions + 0.3 * jaccard_likes + 0.2 * jaccard_tags + 0
701	701
702	702	Like overlap uses exponential time decay: `EXP(-0.023 * age_days)` (30-day half-life).
703	703
704		-### 7.4 On-Demand Scoring
	704	+### 7.4 Precomputed + On-Demand Scoring
705	705
706		-Recommendations are computed on-demand at query time, not pre-materialized. This avoids write amplification on every cron run.
	706	+Recommendations are precomputed for all active users on every cron cycle (`GLEAN_CLUSTER_INTERVAL`, default 1h) and stored in the `precomputed_recommendations` table
707	707
708	708	Feed recommendation score (computed in SQL):
709	709
@@ -795,8 +795,9 @@ A background goroutine runs on a configurable schedule (`GLEAN_CLUSTER_INTERVAL`
795	795	6. Compute follow distances: Incremental BFS for dirty users (1-hop through 3-hop from `follows` table)
796	796	7. Compute signal profiles: Per-user category/tag/like summaries
797	797	8. Auto-dismiss stale: Dismiss items shown >=5 times over >5 days without action
798		-9. Prune old impressions: Delete `recommendation_impressions` older than 90 days
	798	+9. Precompute recommendations: For each active user, compute feed, article, and people recommendations and store as JSON in the `precomputed_recommendations` table. This ensures instant load times for all recommendation sections.
799	799	10. DB maintenance: Run `PRAGMA incremental_vacuum` on all 3 databases (users, articles, recs) to reclaim freed pages. Incremental auto-vacuum is enabled via `PRAGMA auto_vacuum = INCREMENTAL` at connection time, so pages freed by impression pruning and other deletions are reclaimed each cycle.
	800	+11. Prune old impressions: Delete `recommendation_impressions` older than 90 days
800	801
801	802	Jetstream ingestion and record indexing happen in a separate persistent goroutine (the Jetstream consumer), not in the cron.
802	803
@@ -828,7 +829,7 @@ CREATE TABLE recommendation_impressions (
828	829
829	830	### 7.12 Computed Recommendation Tables (`<base>_recs`)
830	831
831		-Written exclusively by the cron. No user-facing writes — only reads during on-demand scoring.
	832	+Written exclusively by the cron. Read during recommendation requests (precomputed results served first, on-demand fallback if missing).
832	833
833	834	```sql
834	835	CREATE TABLE feed_similarity (
@@ -879,6 +880,14 @@ CREATE TABLE user_signal_profiles (
879	880	top_tags TEXT,
880	881	updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
881	882	);
	883	+
	884	+CREATE TABLE precomputed_recommendations (
	885	+ user_did TEXT NOT NULL,
	886	+ rec_type TEXT NOT NULL CHECK(rec_type IN ('feed', 'article', 'person')),
	887	+ data TEXT NOT NULL,
	888	+ computed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
	889	+ PRIMARY KEY (user_did, rec_type)
	890	+);
882	891	```
883	892
884	893	### 7.13 Embeddings (recommended)
@@ -1028,7 +1037,8 @@ glean/
1028	1037	│ ├── cluster/
1029	1038	│ │ ├── jaccard.go # Jaccard similarity computation
1030	1039	│ │ ├── article.go # Article + feed embedding computation, vec0 KNN content boost, language detection
1031		-│ │ ├── scoring.go # Feed + people + article recommendation queries (on-demand)
	1040	+│ │ ├── scoring.go # Feed + people + article recommendation queries (precomputed + on-demand fallback)
	1041	+│ │ ├── precompute.go # Precompute all user recommendations (run by cron 4x/day)
1032	1042	│ │ ├── social.go # Incremental follow-distance computation (1-3 hop, dirty-flag)
1033	1043	│ │ ├── weights.go # Bandit-style signal weight auto-tuning
1034	1044	│ │ ├── diversity.go # Post-query domain/category diversity filtering
@@ -1137,7 +1147,8 @@ Cron (every 1h) ──► Cluster Engine
1137	1147
1138	1148	Browser ──GET /dashboard──► Server
1139	1149	│
1140		- ├─► Compute recommendations on-demand (filtered by user's language preferences)
	1150	+ ├─► Read precomputed recommendations from DB (instant)
	1151	+ │ (fallback to on-demand if missing)
1141	1152	├─► Fetch feed metadata
1142	1153	└─◄ Render recommendation cards (htmx)
1143	1154	```

  Like overlap uses exponential time decay: `EXP(-0.023 * age_days)` (30-day half-life).
--### 7.4 On-Demand Scoring
++### 7.4 Precomputed + On-Demand Scoring
--Recommendations are computed **on-demand** at query time, not pre-materialized. This avoids write amplification on every cron run.
++Recommendations are **precomputed** for all active users on every cron cycle (`GLEAN_CLUSTER_INTERVAL`, default 1h) and stored in the `precomputed_recommendations` table
  **Feed recommendation score** (computed in SQL):
 . **Compute follow distances**: Incremental BFS for dirty users (1-hop through 3-hop from `follows` table)
 . **Compute signal profiles**: Per-user category/tag/like summaries
 . **Auto-dismiss stale**: Dismiss items shown >=5 times over >5 days without action
--9. **Prune old impressions**: Delete `recommendation_impressions` older than 90 days
++9. **Precompute recommendations**: For each active user, compute feed, article, and people recommendations and store as JSON in the `precomputed_recommendations` table. This ensures instant load times for all recommendation sections.
 . **DB maintenance**: Run `PRAGMA incremental_vacuum` on all 3 databases (users, articles, recs) to reclaim freed pages. Incremental auto-vacuum is enabled via `PRAGMA auto_vacuum = INCREMENTAL` at connection time, so pages freed by impression pruning and other deletions are reclaimed each cycle.
++11. **Prune old impressions**: Delete `recommendation_impressions` older than 90 days
  Jetstream ingestion and record indexing happen in a separate persistent goroutine (the Jetstream consumer), not in the cron.
  ### 7.12 Computed Recommendation Tables (`<base>_recs`)
--Written exclusively by the cron. No user-facing writes — only reads during on-demand scoring.
++Written exclusively by the cron. Read during recommendation requests (precomputed results served first, on-demand fallback if missing).
  ```sql
  CREATE TABLE feed_similarity (
      top_tags        TEXT,
      updated_at      DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
  );
++
++CREATE TABLE precomputed_recommendations (
++    user_did    TEXT NOT NULL,
++    rec_type    TEXT NOT NULL CHECK(rec_type IN ('feed', 'article', 'person')),
++    data        TEXT NOT NULL,
++    computed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
++    PRIMARY KEY (user_did, rec_type)
++);
  ```
  ### 7.13 Embeddings (recommended)
  │   ├── cluster/
  │   │   ├── jaccard.go             # Jaccard similarity computation
  │   │   ├── article.go             # Article + feed embedding computation, vec0 KNN content boost, language detection
--│   │   ├── scoring.go             # Feed + people + article recommendation queries (on-demand)
++│   │   ├── scoring.go             # Feed + people + article recommendation queries (precomputed + on-demand fallback)
++│   │   ├── precompute.go          # Precompute all user recommendations (run by cron 4x/day)
  │   │   ├── social.go              # Incremental follow-distance computation (1-3 hop, dirty-flag)
  │   │   ├── weights.go             # Bandit-style signal weight auto-tuning
  │   │   ├── diversity.go           # Post-query domain/category diversity filtering
  Browser ──GET /dashboard──► Server
                                  │
--                                ├─► Compute recommendations on-demand (filtered by user's language preferences)
++                                ├─► Read precomputed recommendations from DB (instant)
++                                │   (fallback to on-demand if missing)
                                  ├─► Fetch feed metadata
                                  └─◄ Render recommendation cards (htmx)
  ```

modified go.mod +1 -1 Viewed

@@ -8,7 +8,6 @@ require (
8	8	github.com/bluesky-social/jetstream v0.0.0-20260415170838-8a65de4eda28
9	9	github.com/go-chi/chi/v5 v5.2.5
10	10	github.com/go-chi/cors v1.2.2
11		- github.com/hashicorp/golang-lru/v2 v2.0.7
12	11	github.com/mattn/go-sqlite3 v1.14.22
13	12	github.com/openai/openai-go v1.12.0
14	13	github.com/prometheus/client_golang v1.19.1
@@ -28,6 +27,7 @@ require (
28	27	github.com/google/go-cmp v0.6.0 // indirect
29	28	github.com/google/go-querystring v1.1.0 // indirect
30	29	github.com/gorilla/websocket v1.5.3 // indirect
	30	+ github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
31	31	github.com/ipfs/go-cid v0.4.1 // indirect
32	32	github.com/klauspost/compress v1.17.9 // indirect
33	33	github.com/klauspost/cpuid/v2 v2.2.7 // indirect

  	github.com/bluesky-social/jetstream v0.0.0-20260415170838-8a65de4eda28
  	github.com/go-chi/chi/v5 v5.2.5
  	github.com/go-chi/cors v1.2.2
--	github.com/hashicorp/golang-lru/v2 v2.0.7
  	github.com/mattn/go-sqlite3 v1.14.22
  	github.com/openai/openai-go v1.12.0
  	github.com/prometheus/client_golang v1.19.1
  	github.com/google/go-cmp v0.6.0 // indirect
  	github.com/google/go-querystring v1.1.0 // indirect
  	github.com/gorilla/websocket v1.5.3 // indirect
++	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
  	github.com/ipfs/go-cid v0.4.1 // indirect
  	github.com/klauspost/compress v1.17.9 // indirect
  	github.com/klauspost/cpuid/v2 v2.2.7 // indirect

modified internal/cluster/cron.go +6 -7 Viewed

@@ -10,7 +10,8 @@ import (
10	10	)
11	11
12	12	// Cron periodically runs all cluster engine computations (similarity, embeddings,
13		-// follow distances, signal profiles, auto-dismiss) on a fixed interval.
	13	+// follow distances, signal profiles, auto-dismiss, recommendation precomputation)
	14	+// on a fixed interval.
14	15	type Cron struct {
15	16	engine *Engine
16	17	interval time.Duration
@@ -39,12 +40,6 @@ func (c *Cron) Run(ctx context.Context) error {
39	40	if !c.engine.mu.TryLock() {
40	41	c.logger.Info("skipping computation: already in progress")
41	42	} else {
42		- c.engine.feedCache.Purge()
43		- c.engine.peopleCache.Purge()
44		- c.engine.articleCache.Purge()
45		- c.engine.globalTrendingCache.Purge()
46		- c.engine.personalTrendingCache.Purge()
47		-
48	43	if err := c.engine.ComputeFeedEmbeddings(ctx); err != nil {
49	44	c.engine.logger.Error("feed embeddings failed", "error", err)
50	45	}
@@ -69,6 +64,10 @@ func (c *Cron) Run(ctx context.Context) error {
69	64	if err := c.engine.feedback.AutoDismissStale(ctx, 5, 5); err != nil {
70	65	c.engine.logger.Error("auto dismiss failed", "error", err)
71	66	}
	67	+ if err := c.engine.PrecomputeAllRecommendations(ctx); err != nil {
	68	+ c.engine.logger.Error("recommendation precomputation failed", "error", err)
	69	+ }
	70	+
72	71	c.engine.mu.Unlock()
73	72	}
74	73

  )
  // Cron periodically runs all cluster engine computations (similarity, embeddings,
--// follow distances, signal profiles, auto-dismiss) on a fixed interval.
++// follow distances, signal profiles, auto-dismiss, recommendation precomputation)
++// on a fixed interval.
  type Cron struct {
  	engine   *Engine
  	interval time.Duration
  		if !c.engine.mu.TryLock() {
  			c.logger.Info("skipping computation: already in progress")
  		} else {
--			c.engine.feedCache.Purge()
--			c.engine.peopleCache.Purge()
--			c.engine.articleCache.Purge()
--			c.engine.globalTrendingCache.Purge()
--			c.engine.personalTrendingCache.Purge()
--
  			if err := c.engine.ComputeFeedEmbeddings(ctx); err != nil {
  				c.engine.logger.Error("feed embeddings failed", "error", err)
  			}
  			if err := c.engine.feedback.AutoDismissStale(ctx, 5, 5); err != nil {
  				c.engine.logger.Error("auto dismiss failed", "error", err)
  			}
++			if err := c.engine.PrecomputeAllRecommendations(ctx); err != nil {
++				c.engine.logger.Error("recommendation precomputation failed", "error", err)
++			}
++
  			c.engine.mu.Unlock()
  		}

modified internal/cluster/jaccard.go +8 -25 Viewed

@@ -6,9 +6,6 @@ import (
6	6	"fmt"
7	7	"log/slog"
8	8	"sync"
9		- "time"
10		-
11		- "github.com/hashicorp/golang-lru/v2/expirable"
12	9
13	10	"pkg.rbrt.fr/glean/internal/db"
14	11	"pkg.rbrt.fr/glean/internal/feedback"
@@ -49,31 +46,17 @@ type Engine struct {
49	46	embedder ml.Embedder
50	47	llm ml.TextModel
51	48	feedback *feedback.Service
52		-
53		- feedCache expirable.LRU[string, []FeedRecommendation]
54		- peopleCache expirable.LRU[string, []PersonRecommendation]
55		- articleCache expirable.LRU[string, []ArticleRecommendation]
56		- globalTrendingCache expirable.LRU[string, []db.TrendingItem]
57		- personalTrendingCache expirable.LRU[string, []db.TrendingItem]
58	49	}
59	50
60		-// recCacheSize is the maximum number of recommendations to cache per user.
61		-const recCacheSize = 512
62		-
63		-func NewEngine(sqlDB sql.DB, articles db.ArticleStore, embedder ml.Embedder, llm ml.TextModel, fb feedback.Service, logger slog.Logger, cacheTTL time.Duration, config Config) *Engine {
	51	+func NewEngine(sqlDB sql.DB, articles db.ArticleStore, embedder ml.Embedder, llm ml.TextModel, fb feedback.Service, logger slog.Logger, config Config) *Engine {
64	52	return &Engine{
65		- db: sqlDB,
66		- articles: articles,
67		- logger: logger,
68		- config: config,
69		- embedder: embedder,
70		- llm: llm,
71		- feedback: fb,
72		- feedCache: expirable.NewLRU[string, []*FeedRecommendation](recCacheSize, nil, cacheTTL),
73		- peopleCache: expirable.NewLRU[string, []*PersonRecommendation](recCacheSize, nil, cacheTTL),
74		- articleCache: expirable.NewLRU[string, []*ArticleRecommendation](recCacheSize, nil, cacheTTL),
75		- globalTrendingCache: expirable.NewLRU[string, []*db.TrendingItem](recCacheSize, nil, cacheTTL),
76		- personalTrendingCache: expirable.NewLRU[string, []*db.TrendingItem](recCacheSize, nil, cacheTTL),
	53	+ db: sqlDB,
	54	+ articles: articles,
	55	+ logger: logger,
	56	+ config: config,
	57	+ embedder: embedder,
	58	+ llm: llm,
	59	+ feedback: fb,
77	60	}
78	61	}
79	62

  	"fmt"
  	"log/slog"
  	"sync"
--	"time"
--
--	"github.com/hashicorp/golang-lru/v2/expirable"
  	"pkg.rbrt.fr/glean/internal/db"
  	"pkg.rbrt.fr/glean/internal/feedback"
  	embedder ml.Embedder
  	llm      ml.TextModel
  	feedback *feedback.Service
--
--	feedCache             *expirable.LRU[string, []*FeedRecommendation]
--	peopleCache           *expirable.LRU[string, []*PersonRecommendation]
--	articleCache          *expirable.LRU[string, []*ArticleRecommendation]
--	globalTrendingCache   *expirable.LRU[string, []*db.TrendingItem]
--	personalTrendingCache *expirable.LRU[string, []*db.TrendingItem]
  }
--// recCacheSize is the maximum number of recommendations to cache per user.
++func NewEngine(sqlDB *sql.DB, articles *db.ArticleStore, embedder ml.Embedder, llm ml.TextModel, fb *feedback.Service, logger *slog.Logger, config Config) *Engine {
--const recCacheSize = 512
--
--func NewEngine(sqlDB *sql.DB, articles *db.ArticleStore, embedder ml.Embedder, llm ml.TextModel, fb *feedback.Service, logger *slog.Logger, cacheTTL time.Duration, config Config) *Engine {
  	return &Engine{
--		db:                    sqlDB,
++		db:       sqlDB,
--		articles:              articles,
++		articles: articles,
--		logger:                logger,
++		logger:   logger,
--		config:                config,
++		config:   config,
--		embedder:              embedder,
++		embedder: embedder,
--		llm:                   llm,
++		llm:      llm,
--		feedback:              fb,
++		feedback: fb,
--		feedCache:             expirable.NewLRU[string, []*FeedRecommendation](recCacheSize, nil, cacheTTL),
--		peopleCache:           expirable.NewLRU[string, []*PersonRecommendation](recCacheSize, nil, cacheTTL),
--		articleCache:          expirable.NewLRU[string, []*ArticleRecommendation](recCacheSize, nil, cacheTTL),
--		globalTrendingCache:   expirable.NewLRU[string, []*db.TrendingItem](recCacheSize, nil, cacheTTL),
--		personalTrendingCache: expirable.NewLRU[string, []*db.TrendingItem](recCacheSize, nil, cacheTTL),
  	}
  }

modified internal/cluster/jaccard_test.go +1 -2 Viewed

@@ -8,7 +8,6 @@ import (
8	8	"os"
9	9	"strings"
10	10	"testing"
11		- "time"
12	11
13	12	vec "github.com/asg017/sqlite-vec-go-bindings/cgo"
14	13	"pkg.rbrt.fr/glean/internal/db"
@@ -96,7 +95,7 @@ func seedFollowData(t testing.T, ctx context.Context, dbs db.Store) {
96	95	}
97	96
98	97	func newTestEngine(dbs db.Store) Engine {
99		- return NewEngine(dbs.SQLDB(), dbs.Articles, NewMockEmbedder(8), nil, feedback.NewService(dbs.SQLDB()), slog.Default(), time.Hour, DefaultConfig())
	98	+ return NewEngine(dbs.SQLDB(), dbs.Articles, NewMockEmbedder(8), nil, feedback.NewService(dbs.SQLDB()), slog.Default(), DefaultConfig())
100	99	}
101	100
102	101	func TestComputeFeedSimilarity(t *testing.T) {

  	"os"
  	"strings"
  	"testing"
--	"time"
  	vec "github.com/asg017/sqlite-vec-go-bindings/cgo"
  	"pkg.rbrt.fr/glean/internal/db"
  }
  func newTestEngine(dbs *db.Store) *Engine {
--	return NewEngine(dbs.SQLDB(), dbs.Articles, NewMockEmbedder(8), nil, feedback.NewService(dbs.SQLDB()), slog.Default(), time.Hour, DefaultConfig())
++	return NewEngine(dbs.SQLDB(), dbs.Articles, NewMockEmbedder(8), nil, feedback.NewService(dbs.SQLDB()), slog.Default(), DefaultConfig())
  }
  func TestComputeFeedSimilarity(t *testing.T) {

added internal/cluster/precompute.go +157 -0 Viewed

new file mode 100644
@@ -0,0 +1,157 @@
	1	+package cluster
	2	+
	3	+import (
	4	+ "context"
	5	+ "encoding/json"
	6	+ "time"
	7	+)
	8	+
	9	+type PrecomputedRec struct {
	10	+ RecType string
	11	+ Data string
	12	+}
	13	+
	14	+func (e *Engine) PrecomputeAllRecommendations(ctx context.Context) error {
	15	+ e.logger.Info("starting recommendation precomputation")
	16	+ start := time.Now()
	17	+
	18	+ users, err := e.listActiveUsers(ctx)
	19	+ if err != nil {
	20	+ return err
	21	+ }
	22	+
	23	+ computed := 0
	24	+ for _, did := range users {
	25	+ select {
	26	+ case <-ctx.Done():
	27	+ return ctx.Err()
	28	+ default:
	29	+ }
	30	+
	31	+ if err := e.precomputeForUser(ctx, did); err != nil {
	32	+ e.logger.Warn("precompute failed for user", "did", did, "error", err)
	33	+ continue
	34	+ }
	35	+ computed++
	36	+ }
	37	+
	38	+ e.logger.Info("recommendation precomputation complete",
	39	+ "users", computed,
	40	+ "duration", time.Since(start),
	41	+ )
	42	+ return nil
	43	+}
	44	+
	45	+func (e *Engine) precomputeForUser(ctx context.Context, userDID string) error {
	46	+ feedRecs, err := e.ComputeFeedRecommendationsOnDemand(ctx, userDID, 10)
	47	+ if err != nil {
	48	+ return err
	49	+ }
	50	+ if len(feedRecs) > 0 {
	51	+ normalizeFeedScores(feedRecs)
	52	+ feedRecs = ApplyDiversity(feedRecs, 5)
	53	+ }
	54	+
	55	+ articleRecs, err := e.ComputeArticleRecommendationsOnDemand(ctx, userDID, nil, 10)
	56	+ if err != nil {
	57	+ return err
	58	+ }
	59	+ if len(articleRecs) > 0 {
	60	+ normalizeArticleScores(articleRecs)
	61	+ if len(articleRecs) > 5 {
	62	+ articleRecs = articleRecs[:5]
	63	+ }
	64	+ }
	65	+
	66	+ half := 3
	67	+ inNet, err := e.computePeopleByFollowStatus(ctx, userDID, true, half)
	68	+ if err != nil {
	69	+ return err
	70	+ }
	71	+ outNet, err := e.computePeopleByFollowStatus(ctx, userDID, false, half)
	72	+ if err != nil {
	73	+ return err
	74	+ }
	75	+ peopleRecs := append(inNet, outNet...)
	76	+ if len(peopleRecs) > 0 {
	77	+ normalizePersonScores(peopleRecs)
	78	+ if len(peopleRecs) > 6 {
	79	+ peopleRecs = peopleRecs[:6]
	80	+ }
	81	+ }
	82	+
	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	+ }
	99	+
	100	+ return nil
	101	+}
	102	+
	103	+func (e *Engine) listActiveUsers(ctx context.Context) ([]string, error) {
	104	+ rows, err := e.db.QueryContext(ctx, `
	105	+ SELECT DISTINCT s.user_did FROM articles.subscriptions s
	106	+ UNION
	107	+ SELECT DISTINCT author_did FROM articles.likes
	108	+ `)
	109	+ if err != nil {
	110	+ return nil, err
	111	+ }
	112	+ defer rows.Close()
	113	+
	114	+ var dids []string
	115	+ for rows.Next() {
	116	+ var did string
	117	+ if err := rows.Scan(&did); err != nil {
	118	+ return nil, err
	119	+ }
	120	+ dids = append(dids, did)
	121	+ }
	122	+ return dids, rows.Err()
	123	+}
	124	+
	125	+func (e *Engine) getPrecomputed(ctx context.Context, userDID, recType string) (string, bool) {
	126	+ var data string
	127	+ err := e.db.QueryRowContext(ctx, `
	128	+ SELECT data FROM recs.precomputed_recommendations
	129	+ WHERE user_did = ? AND rec_type = ?
	130	+ `, userDID, recType).Scan(&data)
	131	+ if err != nil {
	132	+ return "", false
	133	+ }
	134	+ return data, true
	135	+}
	136	+
	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)
	148	+ }
	149	+}
	150	+
	151	+func mustJSON(v any) string {
	152	+ b, err := json.Marshal(v)
	153	+ if err != nil {
	154	+ return "null"
	155	+ }
	156	+ return string(b)
	157	+}

++package cluster
++
++import (
++	"context"
++	"encoding/json"
++	"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()
++
++	users, err := e.listActiveUsers(ctx)
++	if err != nil {
++		return err
++	}
++
++	computed := 0
++	for _, did := range users {
++		select {
++		case <-ctx.Done():
++			return ctx.Err()
++		default:
++		}
++
++		if err := e.precomputeForUser(ctx, did); err != nil {
++			e.logger.Warn("precompute failed for user", "did", did, "error", err)
++			continue
++		}
++		computed++
++	}
++
++	e.logger.Info("recommendation precomputation complete",
++		"users", computed,
++		"duration", time.Since(start),
++	)
++	return nil
++}
++
++func (e *Engine) precomputeForUser(ctx context.Context, userDID string) error {
++	feedRecs, err := e.ComputeFeedRecommendationsOnDemand(ctx, userDID, 10)
++	if err != nil {
++		return err
++	}
++	if len(feedRecs) > 0 {
++		normalizeFeedScores(feedRecs)
++		feedRecs = ApplyDiversity(feedRecs, 5)
++	}
++
++	articleRecs, err := e.ComputeArticleRecommendationsOnDemand(ctx, userDID, nil, 10)
++	if err != nil {
++		return err
++	}
++	if len(articleRecs) > 0 {
++		normalizeArticleScores(articleRecs)
++		if len(articleRecs) > 5 {
++			articleRecs = articleRecs[:5]
++		}
++	}
++
++	half := 3
++	inNet, err := e.computePeopleByFollowStatus(ctx, userDID, true, half)
++	if err != nil {
++		return err
++	}
++	outNet, err := e.computePeopleByFollowStatus(ctx, userDID, false, half)
++	if err != nil {
++		return err
++	}
++	peopleRecs := append(inNet, outNet...)
++	if len(peopleRecs) > 0 {
++		normalizePersonScores(peopleRecs)
++		if len(peopleRecs) > 6 {
++			peopleRecs = peopleRecs[:6]
++		}
++	}
++
++	for _, rec := range []PrecomputedRec{
++		{RecType: "feed", Data: mustJSON(feedRecs)},
++		{RecType: "article", Data: mustJSON(articleRecs)},
++		{RecType: "person", Data: mustJSON(peopleRecs)},
++	} {
++		if rec.Data == "null" {
++			continue
++		}
++		if _, 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, rec.RecType, rec.Data); err != nil {
++			e.logger.Warn("failed to store precomputed rec", "did", userDID, "type", rec.RecType, "error", err)
++		}
++	}
++
++	return nil
++}
++
++func (e *Engine) listActiveUsers(ctx context.Context) ([]string, error) {
++	rows, err := e.db.QueryContext(ctx, `
++		SELECT DISTINCT s.user_did FROM articles.subscriptions s
++		UNION
++		SELECT DISTINCT author_did FROM articles.likes
++	`)
++	if err != nil {
++		return nil, err
++	}
++	defer rows.Close()
++
++	var dids []string
++	for rows.Next() {
++		var did string
++		if err := rows.Scan(&did); err != nil {
++			return nil, err
++		}
++		dids = append(dids, did)
++	}
++	return dids, rows.Err()
++}
++
++func (e *Engine) getPrecomputed(ctx context.Context, userDID, recType string) (string, bool) {
++	var data string
++	err := e.db.QueryRowContext(ctx, `
++		SELECT data FROM recs.precomputed_recommendations
++		WHERE user_did = ? AND rec_type = ?
++	`, userDID, recType).Scan(&data)
++	if err != nil {
++		return "", false
++	}
++	return data, true
++}
++
++func (e *Engine) invalidatePrecomputed(ctx context.Context, userDID, recType string) {
++	_, _ = e.db.ExecContext(ctx, `
++		DELETE FROM recs.precomputed_recommendations
++		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)
++	}
++}
++
++func mustJSON(v any) string {
++	b, err := json.Marshal(v)
++	if err != nil {
++		return "null"
++	}
++	return string(b)
++}

modified internal/cluster/scoring.go +22 -47 Viewed

@@ -3,8 +3,8 @@ package cluster
3	3	import (
4	4	"context"
5	5	"database/sql"
	6	+ "encoding/json"
6	7	"fmt"
7		- "strings"
8	8	"time"
9	9
10	10	"pkg.rbrt.fr/glean/internal/db"
@@ -47,15 +47,15 @@ type ArticleRecommendation struct {
47	47	}
48	48
49	49	func (e *Engine) InvalidateFeedCache(userDID string) {
50		- e.feedCache.Remove(userDID)
	50	+ go e.recomputeForUser(context.Background(), userDID, "feed")
51	51	}
52	52
53	53	func (e *Engine) InvalidateArticleCache(userDID string) {
54		- e.articleCache.Remove(userDID)
	54	+ go e.recomputeForUser(context.Background(), userDID, "article")
55	55	}
56	56
57	57	func (e *Engine) InvalidatePeopleCache(userDID string) {
58		- e.peopleCache.Remove(userDID)
	58	+ go e.recomputeForUser(context.Background(), userDID, "person")
59	59	}
60	60
61	61	// GetFeedRecommendations returns feed recommendations for a user. Users with
@@ -63,8 +63,11 @@ func (e *Engine) InvalidatePeopleCache(userDID string) {
63	63	// KNN or graph+popular fallback). Results are min-max normalized and
64	64	// diversity-filtered before returning.
65	65	func (e Engine) GetFeedRecommendations(ctx context.Context, userDID string, limit int) ([]FeedRecommendation, error) {
66		- if entry, ok := e.feedCache.Get(userDID); ok {
67		- return entry, nil
	66	+ if data, ok := e.getPrecomputed(ctx, userDID, "feed"); ok {
	67	+ var recs []*FeedRecommendation
	68	+ if err := json.Unmarshal([]byte(data), &recs); err == nil {
	69	+ return recs, nil
	70	+ }
68	71	}
69	72
70	73	subCount := 0
@@ -84,9 +87,7 @@ func (e *Engine) GetFeedRecommendations(ctx context.Context, userDID string, lim
84	87	}
85	88
86	89	normalizeFeedScores(recs)
87		- result := ApplyDiversity(recs, limit)
88		- e.feedCache.Add(userDID, result)
89		- return result, nil
	90	+ return ApplyDiversity(recs, limit), nil
90	91	}
91	92
92	93	// GetPeopleRecommendations returns similar users based on subscription overlap,
@@ -94,8 +95,11 @@ func (e *Engine) GetFeedRecommendations(ctx context.Context, userDID string, lim
94	95	// limit come from the user's network (followed) and half from outside. When
95	96	// outside-network candidates are scarce, in-network fills the remaining slots.
96	97	func (e Engine) GetPeopleRecommendations(ctx context.Context, userDID string, limit int) ([]PersonRecommendation, error) {
97		- if entry, ok := e.peopleCache.Get(userDID); ok {
98		- return entry, nil
	98	+ if data, ok := e.getPrecomputed(ctx, userDID, "person"); ok {
	99	+ var recs []*PersonRecommendation
	100	+ if err := json.Unmarshal([]byte(data), &recs); err == nil {
	101	+ return recs, nil
	102	+ }
99	103	}
100	104
101	105	half := max(limit/2, 1)
@@ -115,7 +119,6 @@ func (e *Engine) GetPeopleRecommendations(ctx context.Context, userDID string, l
115	119	recs = append(recs, outNet...)
116	120
117	121	normalizePersonScores(recs)
118		- e.peopleCache.Add(userDID, recs)
119	122	return recs, nil
120	123	}
121	124
@@ -124,9 +127,11 @@ func (e *Engine) GetPeopleRecommendations(ctx context.Context, userDID string, l
124	127	// (embedding KNN against user's liked articles), and recency. Scores are
125	128	// min-max normalized.
126	129	func (e Engine) GetArticleRecommendations(ctx context.Context, userDID string, languages []string, limit int) ([]ArticleRecommendation, error) {
127		- artKey := userDID + "\|" + strings.Join(languages, ",")
128		- if entry, ok := e.articleCache.Get(artKey); ok {
129		- return entry, nil
	130	+ if data, ok := e.getPrecomputed(ctx, userDID, "article"); ok {
	131	+ var recs []*ArticleRecommendation
	132	+ if err := json.Unmarshal([]byte(data), &recs); err == nil {
	133	+ return recs, nil
	134	+ }
130	135	}
131	136
132	137	recs, err := e.ComputeArticleRecommendationsOnDemand(ctx, userDID, languages, limit)
@@ -134,47 +139,17 @@ func (e *Engine) GetArticleRecommendations(ctx context.Context, userDID string,
134	139	return nil, err
135	140	}
136	141	normalizeArticleScores(recs)
137		- e.articleCache.Add(artKey, recs)
138	142	return recs, nil
139	143	}
140	144
141	145	func (e Engine) GetGlobalTrending(ctx context.Context, userDID string, limit, offset int) ([]db.TrendingItem, error) {
142		- if offset == 0 {
143		- if entry, ok := e.globalTrendingCache.Get(userDID); ok {
144		- return entry, nil
145		- }
146		- }
147		-
148	146	since := time.Now().AddDate(0, 0, -7).Format(time.RFC3339)
149		- items, err := e.articles.ListTrendingArticles(ctx, userDID, since, limit, offset)
150		- if err != nil {
151		- return nil, err
152		- }
153		-
154		- if offset == 0 {
155		- e.globalTrendingCache.Add(userDID, items)
156		- }
157		- return items, nil
	147	+ return e.articles.ListTrendingArticles(ctx, userDID, since, limit, offset)
158	148	}
159	149
160	150	func (e Engine) GetPersonalTrending(ctx context.Context, userDID string, languages []string, limit, offset int) ([]db.TrendingItem, error) {
161		- key := userDID + "\|" + strings.Join(languages, ",")
162		- if offset == 0 {
163		- if entry, ok := e.personalTrendingCache.Get(key); ok {
164		- return entry, nil
165		- }
166		- }
167		-
168	151	since := time.Now().AddDate(0, 0, -7).Format(time.RFC3339)
169		- items, err := e.articles.ListTrendingArticlesForUser(ctx, userDID, since, languages, limit, offset)
170		- if err != nil {
171		- return nil, err
172		- }
173		-
174		- if offset == 0 {
175		- e.personalTrendingCache.Add(key, items)
176		- }
177		- return items, nil
	152	+ return e.articles.ListTrendingArticlesForUser(ctx, userDID, since, languages, limit, offset)
178	153	}
179	154
180	155	// SignalWeights holds per-signal multipliers used in the recommendation scoring

  import (
  	"context"
  	"database/sql"
++	"encoding/json"
  	"fmt"
--	"strings"
  	"time"
  	"pkg.rbrt.fr/glean/internal/db"
  }
  func (e *Engine) InvalidateFeedCache(userDID string) {
--	e.feedCache.Remove(userDID)
++	go e.recomputeForUser(context.Background(), userDID, "feed")
  }
  func (e *Engine) InvalidateArticleCache(userDID string) {
--	e.articleCache.Remove(userDID)
++	go e.recomputeForUser(context.Background(), userDID, "article")
  }
  func (e *Engine) InvalidatePeopleCache(userDID string) {
--	e.peopleCache.Remove(userDID)
++	go e.recomputeForUser(context.Background(), userDID, "person")
  }
  // GetFeedRecommendations returns feed recommendations for a user. Users with
  // KNN or graph+popular fallback). Results are min-max normalized and
  // diversity-filtered before returning.
  func (e *Engine) GetFeedRecommendations(ctx context.Context, userDID string, limit int) ([]*FeedRecommendation, error) {
--	if entry, ok := e.feedCache.Get(userDID); ok {
++	if data, ok := e.getPrecomputed(ctx, userDID, "feed"); ok {
--		return entry, nil
++		var recs []*FeedRecommendation
++		if err := json.Unmarshal([]byte(data), &recs); err == nil {
++			return recs, nil
++		}
  	}
  	subCount := 0
  	}
  	normalizeFeedScores(recs)
--	result := ApplyDiversity(recs, limit)
++	return ApplyDiversity(recs, limit), nil
--	e.feedCache.Add(userDID, result)
--	return result, nil
  }
  // GetPeopleRecommendations returns similar users based on subscription overlap,
  // limit come from the user's network (followed) and half from outside. When
  // outside-network candidates are scarce, in-network fills the remaining slots.
  func (e *Engine) GetPeopleRecommendations(ctx context.Context, userDID string, limit int) ([]*PersonRecommendation, error) {
--	if entry, ok := e.peopleCache.Get(userDID); ok {
++	if data, ok := e.getPrecomputed(ctx, userDID, "person"); ok {
--		return entry, nil
++		var recs []*PersonRecommendation
++		if err := json.Unmarshal([]byte(data), &recs); err == nil {
++			return recs, nil
++		}
  	}
  	half := max(limit/2, 1)
  	recs = append(recs, outNet...)
  	normalizePersonScores(recs)
--	e.peopleCache.Add(userDID, recs)
  	return recs, nil
  }
  // (embedding KNN against user's liked articles), and recency. Scores are
  // min-max normalized.
  func (e *Engine) GetArticleRecommendations(ctx context.Context, userDID string, languages []string, limit int) ([]*ArticleRecommendation, error) {
--	artKey := userDID + "|" + strings.Join(languages, ",")
++	if data, ok := e.getPrecomputed(ctx, userDID, "article"); ok {
--	if entry, ok := e.articleCache.Get(artKey); ok {
++		var recs []*ArticleRecommendation
--		return entry, nil
++		if err := json.Unmarshal([]byte(data), &recs); err == nil {
++			return recs, nil
++		}
  	}
  	recs, err := e.ComputeArticleRecommendationsOnDemand(ctx, userDID, languages, limit)
  		return nil, err
  	}
  	normalizeArticleScores(recs)
--	e.articleCache.Add(artKey, recs)
  	return recs, nil
  }
  func (e *Engine) GetGlobalTrending(ctx context.Context, userDID string, limit, offset int) ([]*db.TrendingItem, error) {
--	if offset == 0 {
--		if entry, ok := e.globalTrendingCache.Get(userDID); ok {
--			return entry, nil
--		}
--	}
--
  	since := time.Now().AddDate(0, 0, -7).Format(time.RFC3339)
--	items, err := e.articles.ListTrendingArticles(ctx, userDID, since, limit, offset)
++	return e.articles.ListTrendingArticles(ctx, userDID, since, limit, offset)
--	if err != nil {
--		return nil, err
--	}
--
--	if offset == 0 {
--		e.globalTrendingCache.Add(userDID, items)
--	}
--	return items, nil
  }
  func (e *Engine) GetPersonalTrending(ctx context.Context, userDID string, languages []string, limit, offset int) ([]*db.TrendingItem, error) {
--	key := userDID + "|" + strings.Join(languages, ",")
--	if offset == 0 {
--		if entry, ok := e.personalTrendingCache.Get(key); ok {
--			return entry, nil
--		}
--	}
--
  	since := time.Now().AddDate(0, 0, -7).Format(time.RFC3339)
--	items, err := e.articles.ListTrendingArticlesForUser(ctx, userDID, since, languages, limit, offset)
++	return e.articles.ListTrendingArticlesForUser(ctx, userDID, since, languages, limit, offset)
--	if err != nil {
--		return nil, err
--	}
--
--	if offset == 0 {
--		e.personalTrendingCache.Add(key, items)
--	}
--	return items, nil
  }
  // SignalWeights holds per-signal multipliers used in the recommendation scoring

modified internal/db/db.go +8 -0 Viewed

@@ -449,6 +449,14 @@ var recsSchema = []string{
449	449
450	450	`CREATE INDEX IF NOT EXISTS recs.idx_follow_distances_a_dist ON follow_distances(user_a, distance)`,
451	451	`CREATE INDEX IF NOT EXISTS recs.idx_user_similarity_b ON user_similarity(user_b)`,
	452	+
	453	+ `CREATE TABLE IF NOT EXISTS recs.precomputed_recommendations (
	454	+ user_did TEXT NOT NULL,
	455	+ rec_type TEXT NOT NULL CHECK(rec_type IN ('feed', 'article', 'person')),
	456	+ data TEXT NOT NULL,
	457	+ computed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
	458	+ PRIMARY KEY (user_did, rec_type)
	459	+ )`,
452	460	}
453	461
454	462	func NullStr(s string) sql.NullString {

  	`CREATE INDEX IF NOT EXISTS recs.idx_follow_distances_a_dist ON follow_distances(user_a, distance)`,
  	`CREATE INDEX IF NOT EXISTS recs.idx_user_similarity_b ON user_similarity(user_b)`,
++
++	`CREATE TABLE IF NOT EXISTS recs.precomputed_recommendations (
++		user_did    TEXT NOT NULL,
++		rec_type    TEXT NOT NULL CHECK(rec_type IN ('feed', 'article', 'person')),
++		data        TEXT NOT NULL,
++		computed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
++		PRIMARY KEY (user_did, rec_type)
++	)`,
  }
  func NullStr(s string) sql.NullString {

modified main.go +1 -1 Viewed

@@ -101,7 +101,7 @@ func main() {
101	101	})
102	102	}
103	103
104		- engine := cluster.NewEngine(dbs.SQLDB(), dbs.Articles, embedder, llm, feedback.NewService(dbs.SQLDB()), logger, *clusterInterval, cluster.DefaultConfig())
	104	+ engine := cluster.NewEngine(dbs.SQLDB(), dbs.Articles, embedder, llm, feedback.NewService(dbs.SQLDB()), logger, cluster.DefaultConfig())
105	105
106	106	fetcher := feed.NewFetcher(siteFetcher)
107	107	srv := server.New(dbs, clientID, callbackURL, *addr, scheduler, fetcher, engine, logger, []byte(sessionKey), llm)

  		})
  	}
--	engine := cluster.NewEngine(dbs.SQLDB(), dbs.Articles, embedder, llm, feedback.NewService(dbs.SQLDB()), logger, *clusterInterval, cluster.DefaultConfig())
++	engine := cluster.NewEngine(dbs.SQLDB(), dbs.Articles, embedder, llm, feedback.NewService(dbs.SQLDB()), logger, cluster.DefaultConfig())
  	fetcher := feed.NewFetcher(siteFetcher)
  	srv := server.New(dbs, clientID, callbackURL, *addr, scheduler, fetcher, engine, logger, []byte(sessionKey), llm)

modified readme.md +1 -1 Viewed

@@ -19,7 +19,7 @@ Your subscriptions live as records on your PDS. You own them. If Glean goes away
19	19
20	20	## How recommendations work
21	21
22		-Glean looks at what you and other users subscribe to, read, and like to suggest feeds and people you might enjoy.
	22	+Glean looks at what you and other users subscribe to, read, and like to suggest feeds, articles, and people you might enjoy.
23	23
24	24	Feed suggestions come from readers who share your subscriptions. If a lot of people who follow the same blogs as you also follow a blog you haven't seen, that blog shows up as a recommendation. The system also considers which articles you've liked, whether you follow the person on Bluesky, and how popular the feed is overall.
25	25

  ## How recommendations work
--Glean looks at what you and other users subscribe to, read, and like to suggest feeds and people you might enjoy.
++Glean looks at what you and other users subscribe to, read, and like to suggest feeds, articles, and people you might enjoy.
  **Feed suggestions** come from readers who share your subscriptions. If a lot of people who follow the same blogs as you also follow a blog you haven't seen, that blog shows up as a recommendation. The system also considers which articles you've liked, whether you follow the person on Bluesky, and how popular the feed is overall.

Switch to precomputed recommendations with on-demand fallbackUnverified