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

Refactor clustering and add subscription deduplicationUnverified

Julien Robert committed 2026-04-21T13:52:41+02:00 Browse files
cf462e3 parent: a57ff73
modified internal/atproto/stream_handler.go +6 -1
@@ -4,6 +4,7 @@ import (
44 "context"
55 "database/sql"
66 "encoding/json"
7+ "errors"
78 "log/slog"
89 "time"
910
@@ -56,7 +57,11 @@ func (h *StreamDBHandler) handleSubscription(ctx context.Context, event *Event)
5657
5758 f := &db.Feed{FeedURL: rec.FeedURL, Title: db.NullStr(rec.Title)}
5859 _ = h.db.UpsertFeed(ctx, f)
59- return h.db.CreateSubscription(ctx, event.DID, rec.FeedURL, rec.Title, rec.Category, event.URI, event.CID)
60+ err = h.db.CreateSubscription(ctx, event.DID, rec.FeedURL, rec.Title, rec.Category, event.URI, event.CID)
61+ if errors.Is(err, db.ErrDuplicateSubscription) {
62+ return nil
63+ }
64+ return err
6065
6166 case "delete":
6267 parsed, ok := ParseRecordURI(event.URI)
@@ -4,6 +4,7 @@ import (
4 "context"4 "context"
5 "database/sql"5 "database/sql"
6 "encoding/json"6 "encoding/json"
7+ "errors"
7 "log/slog"8 "log/slog"
8 "time"9 "time"
9 10
@@ -56,7 +57,11 @@ func (h *StreamDBHandler) handleSubscription(ctx context.Context, event *Event)
56 57
57 f := &db.Feed{FeedURL: rec.FeedURL, Title: db.NullStr(rec.Title)}58 f := &db.Feed{FeedURL: rec.FeedURL, Title: db.NullStr(rec.Title)}
58 _ = h.db.UpsertFeed(ctx, f)59 _ = h.db.UpsertFeed(ctx, f)
59- return h.db.CreateSubscription(ctx, event.DID, rec.FeedURL, rec.Title, rec.Category, event.URI, event.CID)60+ err = h.db.CreateSubscription(ctx, event.DID, rec.FeedURL, rec.Title, rec.Category, event.URI, event.CID)
61+ if errors.Is(err, db.ErrDuplicateSubscription) {
62+ return nil
63+ }
64+ return err
60 65
61 case "delete":66 case "delete":
62 parsed, ok := ParseRecordURI(event.URI)67 parsed, ok := ParseRecordURI(event.URI)
modified internal/atproto/sync.go +6 -1
@@ -14,6 +14,7 @@ package atproto
1414 import (
1515 "context"
1616 "encoding/json"
17+ "errors"
1718 "log/slog"
1819 "time"
1920
@@ -97,7 +98,11 @@ func (s *Sync) reconcileSubscription(ctx context.Context, userDID, uri, cid stri
9798 return nil
9899 }
99100
100- return s.db.CreateSubscription(ctx, userDID, rec.FeedURL, rec.Title, rec.Category, uri, cid)
101+ err = s.db.CreateSubscription(ctx, userDID, rec.FeedURL, rec.Title, rec.Category, uri, cid)
102+ if errors.Is(err, db.ErrDuplicateSubscription) {
103+ return nil
104+ }
105+ return err
101106 }
102107
103108 func (s *Sync) reconcileLike(ctx context.Context, userDID, uri, cid string, value json.RawMessage) error {
@@ -14,6 +14,7 @@ package atproto
14 import (14 import (
15 "context"15 "context"
16 "encoding/json"16 "encoding/json"
17+ "errors"
17 "log/slog"18 "log/slog"
18 "time"19 "time"
19 20
@@ -97,7 +98,11 @@ func (s *Sync) reconcileSubscription(ctx context.Context, userDID, uri, cid stri
97 return nil98 return nil
98 }99 }
99 100
100- return s.db.CreateSubscription(ctx, userDID, rec.FeedURL, rec.Title, rec.Category, uri, cid)101+ err = s.db.CreateSubscription(ctx, userDID, rec.FeedURL, rec.Title, rec.Category, uri, cid)
102+ if errors.Is(err, db.ErrDuplicateSubscription) {
103+ return nil
104+ }
105+ return err
101 }106 }
102 107
103 func (s *Sync) reconcileLike(ctx context.Context, userDID, uri, cid string, value json.RawMessage) error {108 func (s *Sync) reconcileLike(ctx context.Context, userDID, uri, cid string, value json.RawMessage) error {
modified internal/cluster/cron.go +1 -11
@@ -23,17 +23,7 @@ func (c *Cron) Run(ctx context.Context) error {
2323 c.logger.Info("starting similarity computation")
2424 start := time.Now()
2525
26- if err := c.engine.ComputeFeedSimilarity(ctx); err != nil {
27- c.logger.Error("feed similarity failed", "error", err)
28- }
29-
30- if err := c.engine.ComputeUserSimilarity(ctx); err != nil {
31- c.logger.Error("user similarity failed", "error", err)
32- }
33-
34- if err := c.engine.ComputeRecommendations(ctx); err != nil {
35- c.logger.Error("recommendations failed", "error", err)
36- }
26+ c.engine.ComputeAll(ctx)
3727
3828 metrics.ClusterRuns.Inc()
3929 metrics.ClusterDuration.Observe(time.Since(start).Seconds())
@@ -23,17 +23,7 @@ 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- if err := c.engine.ComputeFeedSimilarity(ctx); err != nil {26+ c.engine.ComputeAll(ctx)
27- c.logger.Error("feed similarity failed", "error", err)
28- }
29-
30- if err := c.engine.ComputeUserSimilarity(ctx); err != nil {
31- c.logger.Error("user similarity failed", "error", err)
32- }
33-
34- if err := c.engine.ComputeRecommendations(ctx); err != nil {
35- c.logger.Error("recommendations failed", "error", err)
36- }
37 27
38 metrics.ClusterRuns.Inc()28 metrics.ClusterRuns.Inc()
39 metrics.ClusterDuration.Observe(time.Since(start).Seconds())29 metrics.ClusterDuration.Observe(time.Since(start).Seconds())
modified internal/cluster/jaccard.go +20 -0
@@ -4,11 +4,31 @@ import (
44 "context"
55 "database/sql"
66 "log/slog"
7+ "sync"
78 )
89
910 type Engine struct {
1011 db *sql.DB
1112 logger *slog.Logger
13+ mu sync.Mutex
14+}
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+ }
1232 }
1333
1434 func (e *Engine) ComputeArticleRecommendations(ctx context.Context) error {
@@ -4,11 +4,31 @@ import (
4 "context"4 "context"
5 "database/sql"5 "database/sql"
6 "log/slog"6 "log/slog"
7+ "sync"
7 )8 )
8 9
9 type Engine struct {10 type Engine struct {
10 db *sql.DB11 db *sql.DB
11 logger *slog.Logger12 logger *slog.Logger
13+ mu sync.Mutex
14+}
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+ }
12 }32 }
13 33
14 func (e *Engine) ComputeArticleRecommendations(ctx context.Context) error {34 func (e *Engine) ComputeArticleRecommendations(ctx context.Context) error {
added internal/cluster/jaccard_test.go +142 -0
new file mode 100644
@@ -0,0 +1,142 @@
1+package cluster
2+
3+import (
4+ "context"
5+ "os"
6+ "testing"
7+
8+ "pkg.rbrt.fr/glean/internal/db"
9+
10+ "gotest.tools/v3/assert"
11+ "log/slog"
12+)
13+
14+func setupClusterTestDB(t *testing.T) *db.DB {
15+ t.Helper()
16+ f, err := os.CreateTemp("", "glean-cluster-test-*.db")
17+ assert.NilError(t, err)
18+ assert.NilError(t, f.Close())
19+ path := f.Name()
20+ t.Cleanup(func() { _ = os.Remove(path) })
21+
22+ database, err := db.Open(path)
23+ assert.NilError(t, err)
24+ t.Cleanup(func() { _ = database.Close() })
25+ return database
26+}
27+
28+func seedClusterData(t *testing.T, ctx context.Context, database *db.DB) {
29+ t.Helper()
30+
31+ users := []struct{ did, handle string }{
32+ {"did:test:alice", "alice"},
33+ {"did:test:bob", "bob"},
34+ {"did:test:carol", "carol"},
35+ }
36+ for _, u := range users {
37+ _, err := database.ExecContext(ctx, `INSERT INTO users (did, handle) VALUES (?, ?)`, u.did, u.handle)
38+ assert.NilError(t, err)
39+ }
40+
41+ feeds := []struct{ url, title string }{
42+ {"https://a.com/feed", "Feed A"},
43+ {"https://b.com/feed", "Feed B"},
44+ {"https://c.com/feed", "Feed C"},
45+ {"https://d.com/feed", "Feed D"},
46+ }
47+ for _, f := range feeds {
48+ _, err := database.ExecContext(ctx, `INSERT INTO feeds (feed_url, title, site_url, description, feed_type) VALUES (?, ?, ?, '', 'rss')`, f.url, f.title, f.url)
49+ assert.NilError(t, err)
50+ }
51+
52+ subs := []struct{ user, feed string }{
53+ {"did:test:alice", "https://a.com/feed"},
54+ {"did:test:alice", "https://b.com/feed"},
55+ {"did:test:alice", "https://c.com/feed"},
56+ {"did:test:bob", "https://a.com/feed"},
57+ {"did:test:bob", "https://b.com/feed"},
58+ {"did:test:bob", "https://d.com/feed"},
59+ {"did:test:carol", "https://c.com/feed"},
60+ }
61+ for _, s := range subs {
62+ _, err := database.ExecContext(ctx, `INSERT INTO subscriptions (user_did, feed_url) VALUES (?, ?)`, s.user, s.feed)
63+ assert.NilError(t, err)
64+ }
65+}
66+
67+func TestComputeFeedSimilarity(t *testing.T) {
68+ ctx := context.Background()
69+ database := setupClusterTestDB(t)
70+ seedClusterData(t, ctx, database)
71+
72+ engine := NewEngine(database.DB, slog.Default())
73+ err := engine.ComputeFeedSimilarity(ctx)
74+ assert.NilError(t, err)
75+
76+ var count int
77+ err = database.QueryRowContext(ctx, `SELECT COUNT(*) FROM feed_similarity`).Scan(&count)
78+ assert.NilError(t, err)
79+ assert.Assert(t, count > 0, "expected feed similarity pairs")
80+}
81+
82+func TestComputeUserSimilarity(t *testing.T) {
83+ ctx := context.Background()
84+ database := setupClusterTestDB(t)
85+ seedClusterData(t, ctx, database)
86+
87+ engine := NewEngine(database.DB, slog.Default())
88+ err := engine.ComputeUserSimilarity(ctx)
89+ assert.NilError(t, err)
90+
91+ var count int
92+ err = database.QueryRowContext(ctx, `SELECT COUNT(*) FROM user_similarity`).Scan(&count)
93+ assert.NilError(t, err)
94+ assert.Assert(t, count > 0, "expected user similarity pairs")
95+}
96+
97+func TestComputeRecommendations_GeneratesFeedRecsForNewUser(t *testing.T) {
98+ ctx := context.Background()
99+ database := setupClusterTestDB(t)
100+ seedClusterData(t, ctx, database)
101+
102+ engine := NewEngine(database.DB, slog.Default())
103+ assert.NilError(t, engine.ComputeFeedSimilarity(ctx))
104+ assert.NilError(t, engine.ComputeUserSimilarity(ctx))
105+ assert.NilError(t, engine.ComputeRecommendations(ctx))
106+
107+ recs, err := engine.GetFeedRecommendations(ctx, "did:test:carol", 10)
108+ assert.NilError(t, err)
109+ assert.Assert(t, len(recs) > 0, "carol should get feed recommendations from similar users")
110+
111+ var found bool
112+ for _, r := range recs {
113+ if r["feed_url"] == "https://a.com/feed" || r["feed_url"] == "https://b.com/feed" {
114+ found = true
115+ }
116+ }
117+ assert.Assert(t, found, "carol should be recommended feeds she doesn't subscribe to")
118+}
119+
120+func TestComputeRecommendations_NoSelfRecommendations(t *testing.T) {
121+ ctx := context.Background()
122+ database := setupClusterTestDB(t)
123+ seedClusterData(t, ctx, database)
124+
125+ engine := NewEngine(database.DB, slog.Default())
126+ assert.NilError(t, engine.ComputeFeedSimilarity(ctx))
127+ assert.NilError(t, engine.ComputeUserSimilarity(ctx))
128+ assert.NilError(t, engine.ComputeRecommendations(ctx))
129+
130+ recs, err := engine.GetFeedRecommendations(ctx, "did:test:alice", 10)
131+ assert.NilError(t, err)
132+
133+ subscribedFeeds := map[string]bool{
134+ "https://a.com/feed": true,
135+ "https://b.com/feed": true,
136+ "https://c.com/feed": true,
137+ }
138+ for _, r := range recs {
139+ assert.Assert(t, !subscribedFeeds[r["feed_url"].(string)],
140+ "should not recommend a feed the user already subscribes to")
141+ }
142+}
new file mode 100644
@@ -0,0 +1,142 @@
1+package cluster
2+
3+import (
4+ "context"
5+ "os"
6+ "testing"
7+
8+ "pkg.rbrt.fr/glean/internal/db"
9+
10+ "gotest.tools/v3/assert"
11+ "log/slog"
12+)
13+
14+func setupClusterTestDB(t *testing.T) *db.DB {
15+ t.Helper()
16+ f, err := os.CreateTemp("", "glean-cluster-test-*.db")
17+ assert.NilError(t, err)
18+ assert.NilError(t, f.Close())
19+ path := f.Name()
20+ t.Cleanup(func() { _ = os.Remove(path) })
21+
22+ database, err := db.Open(path)
23+ assert.NilError(t, err)
24+ t.Cleanup(func() { _ = database.Close() })
25+ return database
26+}
27+
28+func seedClusterData(t *testing.T, ctx context.Context, database *db.DB) {
29+ t.Helper()
30+
31+ users := []struct{ did, handle string }{
32+ {"did:test:alice", "alice"},
33+ {"did:test:bob", "bob"},
34+ {"did:test:carol", "carol"},
35+ }
36+ for _, u := range users {
37+ _, err := database.ExecContext(ctx, `INSERT INTO users (did, handle) VALUES (?, ?)`, u.did, u.handle)
38+ assert.NilError(t, err)
39+ }
40+
41+ feeds := []struct{ url, title string }{
42+ {"https://a.com/feed", "Feed A"},
43+ {"https://b.com/feed", "Feed B"},
44+ {"https://c.com/feed", "Feed C"},
45+ {"https://d.com/feed", "Feed D"},
46+ }
47+ for _, f := range feeds {
48+ _, err := database.ExecContext(ctx, `INSERT INTO feeds (feed_url, title, site_url, description, feed_type) VALUES (?, ?, ?, '', 'rss')`, f.url, f.title, f.url)
49+ assert.NilError(t, err)
50+ }
51+
52+ subs := []struct{ user, feed string }{
53+ {"did:test:alice", "https://a.com/feed"},
54+ {"did:test:alice", "https://b.com/feed"},
55+ {"did:test:alice", "https://c.com/feed"},
56+ {"did:test:bob", "https://a.com/feed"},
57+ {"did:test:bob", "https://b.com/feed"},
58+ {"did:test:bob", "https://d.com/feed"},
59+ {"did:test:carol", "https://c.com/feed"},
60+ }
61+ for _, s := range subs {
62+ _, err := database.ExecContext(ctx, `INSERT INTO subscriptions (user_did, feed_url) VALUES (?, ?)`, s.user, s.feed)
63+ assert.NilError(t, err)
64+ }
65+}
66+
67+func TestComputeFeedSimilarity(t *testing.T) {
68+ ctx := context.Background()
69+ database := setupClusterTestDB(t)
70+ seedClusterData(t, ctx, database)
71+
72+ engine := NewEngine(database.DB, slog.Default())
73+ err := engine.ComputeFeedSimilarity(ctx)
74+ assert.NilError(t, err)
75+
76+ var count int
77+ err = database.QueryRowContext(ctx, `SELECT COUNT(*) FROM feed_similarity`).Scan(&count)
78+ assert.NilError(t, err)
79+ assert.Assert(t, count > 0, "expected feed similarity pairs")
80+}
81+
82+func TestComputeUserSimilarity(t *testing.T) {
83+ ctx := context.Background()
84+ database := setupClusterTestDB(t)
85+ seedClusterData(t, ctx, database)
86+
87+ engine := NewEngine(database.DB, slog.Default())
88+ err := engine.ComputeUserSimilarity(ctx)
89+ assert.NilError(t, err)
90+
91+ var count int
92+ err = database.QueryRowContext(ctx, `SELECT COUNT(*) FROM user_similarity`).Scan(&count)
93+ assert.NilError(t, err)
94+ assert.Assert(t, count > 0, "expected user similarity pairs")
95+}
96+
97+func TestComputeRecommendations_GeneratesFeedRecsForNewUser(t *testing.T) {
98+ ctx := context.Background()
99+ database := setupClusterTestDB(t)
100+ seedClusterData(t, ctx, database)
101+
102+ engine := NewEngine(database.DB, slog.Default())
103+ assert.NilError(t, engine.ComputeFeedSimilarity(ctx))
104+ assert.NilError(t, engine.ComputeUserSimilarity(ctx))
105+ assert.NilError(t, engine.ComputeRecommendations(ctx))
106+
107+ recs, err := engine.GetFeedRecommendations(ctx, "did:test:carol", 10)
108+ assert.NilError(t, err)
109+ assert.Assert(t, len(recs) > 0, "carol should get feed recommendations from similar users")
110+
111+ var found bool
112+ for _, r := range recs {
113+ if r["feed_url"] == "https://a.com/feed" || r["feed_url"] == "https://b.com/feed" {
114+ found = true
115+ }
116+ }
117+ assert.Assert(t, found, "carol should be recommended feeds she doesn't subscribe to")
118+}
119+
120+func TestComputeRecommendations_NoSelfRecommendations(t *testing.T) {
121+ ctx := context.Background()
122+ database := setupClusterTestDB(t)
123+ seedClusterData(t, ctx, database)
124+
125+ engine := NewEngine(database.DB, slog.Default())
126+ assert.NilError(t, engine.ComputeFeedSimilarity(ctx))
127+ assert.NilError(t, engine.ComputeUserSimilarity(ctx))
128+ assert.NilError(t, engine.ComputeRecommendations(ctx))
129+
130+ recs, err := engine.GetFeedRecommendations(ctx, "did:test:alice", 10)
131+ assert.NilError(t, err)
132+
133+ subscribedFeeds := map[string]bool{
134+ "https://a.com/feed": true,
135+ "https://b.com/feed": true,
136+ "https://c.com/feed": true,
137+ }
138+ for _, r := range recs {
139+ assert.Assert(t, !subscribedFeeds[r["feed_url"].(string)],
140+ "should not recommend a feed the user already subscribes to")
141+ }
142+}
modified internal/db/feed.go +9 -2
@@ -3,10 +3,13 @@ package db
33 import (
44 "context"
55 "database/sql"
6+ "errors"
67 "strings"
78 "time"
89 )
910
11+var ErrDuplicateSubscription = errors.New("already subscribed to this feed")
12+
1013 type Feed struct {
1114 FeedURL string
1215 Title sql.NullString
@@ -135,13 +138,17 @@ func (db *DB) DecrementSubscriberCount(ctx context.Context, feedURL string) erro
135138 }
136139
137140 func (db *DB) CreateSubscription(ctx context.Context, userDID, feedURL, title, category, uri, cid string) error {
138- _, err := db.ExecContext(ctx, `
139- INSERT INTO subscriptions (user_did, feed_url, title, category, uri, cid)
141+ result, err := db.ExecContext(ctx, `
142+ INSERT OR IGNORE INTO subscriptions (user_did, feed_url, title, category, uri, cid)
140143 VALUES (?, ?, ?, ?, ?, ?)
141144 `, userDID, feedURL, nilIfEmpty(title), category, uriOrNil(category, uri), uriOrNil(category, cid))
142145 if err != nil {
143146 return err
144147 }
148+ n, _ := result.RowsAffected()
149+ if n == 0 {
150+ return ErrDuplicateSubscription
151+ }
145152 return db.IncrementSubscriberCount(ctx, feedURL)
146153 }
147154
@@ -3,10 +3,13 @@ package db
3 import (3 import (
4 "context"4 "context"
5 "database/sql"5 "database/sql"
6+ "errors"
6 "strings"7 "strings"
7 "time"8 "time"
8 )9 )
9 10
11+var ErrDuplicateSubscription = errors.New("already subscribed to this feed")
12+
10 type Feed struct {13 type Feed struct {
11 FeedURL string14 FeedURL string
12 Title sql.NullString15 Title sql.NullString
@@ -135,13 +138,17 @@ func (db *DB) DecrementSubscriberCount(ctx context.Context, feedURL string) erro
135 }138 }
136 139
137 func (db *DB) CreateSubscription(ctx context.Context, userDID, feedURL, title, category, uri, cid string) error {140 func (db *DB) CreateSubscription(ctx context.Context, userDID, feedURL, title, category, uri, cid string) error {
138- _, err := db.ExecContext(ctx, `141+ result, err := db.ExecContext(ctx, `
139- INSERT INTO subscriptions (user_did, feed_url, title, category, uri, cid)142+ INSERT OR IGNORE INTO subscriptions (user_did, feed_url, title, category, uri, cid)
140 VALUES (?, ?, ?, ?, ?, ?)143 VALUES (?, ?, ?, ?, ?, ?)
141 `, userDID, feedURL, nilIfEmpty(title), category, uriOrNil(category, uri), uriOrNil(category, cid))144 `, userDID, feedURL, nilIfEmpty(title), category, uriOrNil(category, uri), uriOrNil(category, cid))
142 if err != nil {145 if err != nil {
143 return err146 return err
144 }147 }
148+ n, _ := result.RowsAffected()
149+ if n == 0 {
150+ return ErrDuplicateSubscription
151+ }
145 return db.IncrementSubscriberCount(ctx, feedURL)152 return db.IncrementSubscriberCount(ctx, feedURL)
146 }153 }
147 154
modified internal/server/feeds_handler.go +13 -1
@@ -4,6 +4,7 @@ import (
44 "context"
55 "database/sql"
66 "encoding/json"
7+ "errors"
78 "net/http"
89 "time"
910
@@ -112,6 +113,15 @@ func (s *Server) handleAddFeed(w http.ResponseWriter, r *http.Request) {
112113 }
113114
114115 if err := s.db.CreateSubscription(r.Context(), user.DID, feedURL, feedTitle, category, subURI, subCID); err != nil {
116+ if errors.Is(err, db.ErrDuplicateSubscription) {
117+ subs, _ := s.db.ListSubscriptions(r.Context(), user.DID, "", 100, 0)
118+ s.render(w, r, "feed_list.html", map[string]any{
119+ "User": user,
120+ "Subscriptions": subs,
121+ "Error": "Already subscribed to this feed.",
122+ })
123+ return
124+ }
115125 s.logger.Error("failed to create subscription", "error", err)
116126 http.Error(w, err.Error(), http.StatusInternalServerError)
117127 return
@@ -231,7 +241,9 @@ func (s *Server) handleOPMLUpload(w http.ResponseWriter, r *http.Request) {
231241 }
232242
233243 if subErr := s.db.CreateSubscription(r.Context(), user.DID, fu.URL, fu.Title, fu.Category, subURI, subCID); subErr != nil {
234- s.logger.Error("failed to create subscription", "error", subErr)
244+ if !errors.Is(subErr, db.ErrDuplicateSubscription) {
245+ s.logger.Error("failed to create subscription", "error", subErr)
246+ }
235247 continue
236248 }
237249 added++
@@ -4,6 +4,7 @@ import (
4 "context"4 "context"
5 "database/sql"5 "database/sql"
6 "encoding/json"6 "encoding/json"
7+ "errors"
7 "net/http"8 "net/http"
8 "time"9 "time"
9 10
@@ -112,6 +113,15 @@ func (s *Server) handleAddFeed(w http.ResponseWriter, r *http.Request) {
112 }113 }
113 114
114 if err := s.db.CreateSubscription(r.Context(), user.DID, feedURL, feedTitle, category, subURI, subCID); err != nil {115 if err := s.db.CreateSubscription(r.Context(), user.DID, feedURL, feedTitle, category, subURI, subCID); err != nil {
116+ if errors.Is(err, db.ErrDuplicateSubscription) {
117+ subs, _ := s.db.ListSubscriptions(r.Context(), user.DID, "", 100, 0)
118+ s.render(w, r, "feed_list.html", map[string]any{
119+ "User": user,
120+ "Subscriptions": subs,
121+ "Error": "Already subscribed to this feed.",
122+ })
123+ return
124+ }
115 s.logger.Error("failed to create subscription", "error", err)125 s.logger.Error("failed to create subscription", "error", err)
116 http.Error(w, err.Error(), http.StatusInternalServerError)126 http.Error(w, err.Error(), http.StatusInternalServerError)
117 return127 return
@@ -231,7 +241,9 @@ func (s *Server) handleOPMLUpload(w http.ResponseWriter, r *http.Request) {
231 }241 }
232 242
233 if subErr := s.db.CreateSubscription(r.Context(), user.DID, fu.URL, fu.Title, fu.Category, subURI, subCID); subErr != nil {243 if subErr := s.db.CreateSubscription(r.Context(), user.DID, fu.URL, fu.Title, fu.Category, subURI, subCID); subErr != nil {
234- s.logger.Error("failed to create subscription", "error", subErr)244+ if !errors.Is(subErr, db.ErrDuplicateSubscription) {
245+ s.logger.Error("failed to create subscription", "error", subErr)
246+ }
235 continue247 continue
236 }248 }
237 added++249 added++
modified internal/server/server.go +49 -2
@@ -22,6 +22,7 @@ import (
2222 "github.com/bluesky-social/indigo/atproto/syntax"
2323
2424 "pkg.rbrt.fr/glean/internal/atproto"
25+ "pkg.rbrt.fr/glean/internal/cluster"
2526 "pkg.rbrt.fr/glean/internal/db"
2627 "pkg.rbrt.fr/glean/internal/feed"
2728 "pkg.rbrt.fr/glean/internal/metrics"
@@ -62,12 +63,13 @@ type Server struct {
6263 oauthStore *db.OAuthStore
6364 fetcher *feed.Fetcher
6465 scheduler *feed.Scheduler
66+ engine *cluster.Engine
6567 scraper *scraper.Scraper
6668 clientID string
6769 callbackURL string
6870 }
6971
70-func New(database *db.DB, clientID, callbackURL, addr string, scheduler *feed.Scheduler, logger *slog.Logger) *Server {
72+func New(database *db.DB, clientID, callbackURL, addr string, scheduler *feed.Scheduler, engine *cluster.Engine, logger *slog.Logger) *Server {
7173 oauthStore := db.NewOAuthStore(database)
7274
7375 var config oauth.ClientConfig
@@ -91,6 +93,7 @@ func New(database *db.DB, clientID, callbackURL, addr string, scheduler *feed.Sc
9193 oauthStore: oauthStore,
9294 fetcher: feed.NewFetcher(),
9395 scheduler: scheduler,
96+ engine: engine,
9497 scraper: scraper.New(logger),
9598 clientID: clientID,
9699 callbackURL: callbackURL,
@@ -373,15 +376,59 @@ func (s *Server) pdsClientForUser(r *http.Request) *atproto.Client {
373376
374377 func (s *Server) syncUserInBackground(userDID string, client *atproto.Client) {
375378 go func() {
376- ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
379+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
377380 defer cancel()
381+
382+ isNewUser := false
383+ if count, err := s.db.GetSubscriptionCount(ctx, userDID); err == nil && count == 0 {
384+ isNewUser = true
385+ }
386+
378387 sync := atproto.NewSync(s.db, client, s.logger)
379388 if err := sync.Run(ctx, userDID); err != nil {
380389 s.logger.Error("background sync failed", "error", err, "did", userDID)
381390 }
391+
392+ if !isNewUser {
393+ return
394+ }
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)
382399 }()
383400 }
384401
402+func (s *Server) refreshUserFeeds(ctx context.Context, userDID string) {
403+ subs, err := s.db.ListSubscriptions(ctx, userDID, "", 1000, 0)
404+ if err != nil {
405+ s.logger.Error("failed to list subscriptions for initial fetch", "error", err, "did", userDID)
406+ return
407+ }
408+
409+ seen := make(map[string]bool, len(subs))
410+ for _, sub := range subs {
411+ if seen[sub.FeedURL] {
412+ continue
413+ }
414+ seen[sub.FeedURL] = true
415+
416+ f, err := s.db.GetFeed(ctx, sub.FeedURL)
417+ if err != nil {
418+ continue
419+ }
420+ s.scheduler.FetchFeed(ctx, &feed.Feed{
421+ URL: f.FeedURL,
422+ Title: f.Title.String,
423+ SiteURL: f.SiteURL.String,
424+ Description: f.Description.String,
425+ Type: f.FeedType.String,
426+ ETag: f.Etag.String,
427+ LastModified: f.LastModified.String,
428+ })
429+ }
430+}
431+
385432 func (s *Server) PeriodicSync(ctx context.Context, interval time.Duration) {
386433 ticker := time.NewTicker(interval)
387434 defer ticker.Stop()
@@ -22,6 +22,7 @@ import (
22 "github.com/bluesky-social/indigo/atproto/syntax"22 "github.com/bluesky-social/indigo/atproto/syntax"
23 23
24 "pkg.rbrt.fr/glean/internal/atproto"24 "pkg.rbrt.fr/glean/internal/atproto"
25+ "pkg.rbrt.fr/glean/internal/cluster"
25 "pkg.rbrt.fr/glean/internal/db"26 "pkg.rbrt.fr/glean/internal/db"
26 "pkg.rbrt.fr/glean/internal/feed"27 "pkg.rbrt.fr/glean/internal/feed"
27 "pkg.rbrt.fr/glean/internal/metrics"28 "pkg.rbrt.fr/glean/internal/metrics"
@@ -62,12 +63,13 @@ type Server struct {
62 oauthStore *db.OAuthStore63 oauthStore *db.OAuthStore
63 fetcher *feed.Fetcher64 fetcher *feed.Fetcher
64 scheduler *feed.Scheduler65 scheduler *feed.Scheduler
66+ engine *cluster.Engine
65 scraper *scraper.Scraper67 scraper *scraper.Scraper
66 clientID string68 clientID string
67 callbackURL string69 callbackURL string
68 }70 }
69 71
70-func New(database *db.DB, clientID, callbackURL, addr string, scheduler *feed.Scheduler, logger *slog.Logger) *Server {72+func New(database *db.DB, clientID, callbackURL, addr string, scheduler *feed.Scheduler, engine *cluster.Engine, logger *slog.Logger) *Server {
71 oauthStore := db.NewOAuthStore(database)73 oauthStore := db.NewOAuthStore(database)
72 74
73 var config oauth.ClientConfig75 var config oauth.ClientConfig
@@ -91,6 +93,7 @@ func New(database *db.DB, clientID, callbackURL, addr string, scheduler *feed.Sc
91 oauthStore: oauthStore,93 oauthStore: oauthStore,
92 fetcher: feed.NewFetcher(),94 fetcher: feed.NewFetcher(),
93 scheduler: scheduler,95 scheduler: scheduler,
96+ engine: engine,
94 scraper: scraper.New(logger),97 scraper: scraper.New(logger),
95 clientID: clientID,98 clientID: clientID,
96 callbackURL: callbackURL,99 callbackURL: callbackURL,
@@ -373,15 +376,59 @@ func (s *Server) pdsClientForUser(r *http.Request) *atproto.Client {
373 376
374 func (s *Server) syncUserInBackground(userDID string, client *atproto.Client) {377 func (s *Server) syncUserInBackground(userDID string, client *atproto.Client) {
375 go func() {378 go func() {
376- ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)379+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
377 defer cancel()380 defer cancel()
381+
382+ isNewUser := false
383+ if count, err := s.db.GetSubscriptionCount(ctx, userDID); err == nil && count == 0 {
384+ isNewUser = true
385+ }
386+
378 sync := atproto.NewSync(s.db, client, s.logger)387 sync := atproto.NewSync(s.db, client, s.logger)
379 if err := sync.Run(ctx, userDID); err != nil {388 if err := sync.Run(ctx, userDID); err != nil {
380 s.logger.Error("background sync failed", "error", err, "did", userDID)389 s.logger.Error("background sync failed", "error", err, "did", userDID)
381 }390 }
391+
392+ if !isNewUser {
393+ return
394+ }
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)
382 }()399 }()
383 }400 }
384 401
402+func (s *Server) refreshUserFeeds(ctx context.Context, userDID string) {
403+ subs, err := s.db.ListSubscriptions(ctx, userDID, "", 1000, 0)
404+ if err != nil {
405+ s.logger.Error("failed to list subscriptions for initial fetch", "error", err, "did", userDID)
406+ return
407+ }
408+
409+ seen := make(map[string]bool, len(subs))
410+ for _, sub := range subs {
411+ if seen[sub.FeedURL] {
412+ continue
413+ }
414+ seen[sub.FeedURL] = true
415+
416+ f, err := s.db.GetFeed(ctx, sub.FeedURL)
417+ if err != nil {
418+ continue
419+ }
420+ s.scheduler.FetchFeed(ctx, &feed.Feed{
421+ URL: f.FeedURL,
422+ Title: f.Title.String,
423+ SiteURL: f.SiteURL.String,
424+ Description: f.Description.String,
425+ Type: f.FeedType.String,
426+ ETag: f.Etag.String,
427+ LastModified: f.LastModified.String,
428+ })
429+ }
430+}
431+
385 func (s *Server) PeriodicSync(ctx context.Context, interval time.Duration) {432 func (s *Server) PeriodicSync(ctx context.Context, interval time.Duration) {
386 ticker := time.NewTicker(interval)433 ticker := time.NewTicker(interval)
387 defer ticker.Stop()434 defer ticker.Stop()
modified internal/tmpl/partials/feed-list.html +1 -0
@@ -1,4 +1,5 @@
11 {{define "feed-list.html"}}
2+{{if .Error}}<div class="px-4 py-2 text-sm text-spot-red bg-spot-red/10 border-b border-spot-red/30">{{.Error}}</div>{{end}}
23 {{range .Subscriptions}}
34 <div class="px-5 py-4 flex items-center justify-between hover:bg-spot-hover-50 transition rounded-xl">
45 <div class="min-w-0">
@@ -1,4 +1,5 @@
1 {{define "feed-list.html"}}1 {{define "feed-list.html"}}
2+{{if .Error}}<div class="px-4 py-2 text-sm text-spot-red bg-spot-red/10 border-b border-spot-red/30">{{.Error}}</div>{{end}}
2 {{range .Subscriptions}}3 {{range .Subscriptions}}
3 <div class="px-5 py-4 flex items-center justify-between hover:bg-spot-hover-50 transition rounded-xl">4 <div class="px-5 py-4 flex items-center justify-between hover:bg-spot-hover-50 transition rounded-xl">
4 <div class="min-w-0">5 <div class="min-w-0">
modified main.go +3 -2
@@ -43,9 +43,10 @@ func main() {
4343 storeAdapter := db.NewFeedStoreAdapter(database)
4444 scheduler := feed.NewScheduler(storeAdapter, logger)
4545
46- srv := server.New(database, clientID, callbackURL, *addr, scheduler, logger)
47-
4846 engine := cluster.NewEngine(database.DB, logger)
47+
48+ srv := server.New(database, clientID, callbackURL, *addr, scheduler, engine, logger)
49+
4950 cron := cluster.NewCron(engine, *clusterInterval, logger)
5051
5152 handler := atproto.NewStreamDBHandler(database, logger)
@@ -43,9 +43,10 @@ func main() {
43 storeAdapter := db.NewFeedStoreAdapter(database)43 storeAdapter := db.NewFeedStoreAdapter(database)
44 scheduler := feed.NewScheduler(storeAdapter, logger)44 scheduler := feed.NewScheduler(storeAdapter, logger)
45 45
46- srv := server.New(database, clientID, callbackURL, *addr, scheduler, logger)
47-
48 engine := cluster.NewEngine(database.DB, logger)46 engine := cluster.NewEngine(database.DB, logger)
47+
48+ srv := server.New(database, clientID, callbackURL, *addr, scheduler, engine, logger)
49+
49 cron := cluster.NewCron(engine, *clusterInterval, logger)50 cron := cluster.NewCron(engine, *clusterInterval, logger)
50 51
51 handler := atproto.NewStreamDBHandler(database, logger)52 handler := atproto.NewStreamDBHandler(database, logger)