rickub
nandi/gleanpublic⑂ Fork 0
⑂ d5f9944
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 follow distance computation to use SQL queriesUnverified

Julien Robert committed 2026-05-06T09:42:39+02:00 Browse files
d5f9944 parent: d97639c
modified internal/cluster/social.go +51 -43
@@ -3,20 +3,17 @@ package cluster
33 import (
44 "context"
55 "fmt"
6- "iter"
76 )
87
98 const maxFollowDepth = 3
109
11-func chunk[T any](s []T, size int) iter.Seq[[]T] {
12- return func(yield func([]T) bool) {
13- for i := 0; i < len(s); i += size {
14- end := min(i+size, len(s))
15- if !yield(s[i:end]) {
16- return
17- }
18- }
10+func chunk[T any](s []T, size int) [][]T {
11+ var chunks [][]T
12+ for i := 0; i < len(s); i += size {
13+ end := min(i+size, len(s))
14+ chunks = append(chunks, s[i:end])
1915 }
16+ return chunks
2017 }
2118
2219 type followDistance struct {
@@ -30,49 +27,60 @@ func (e *Engine) ComputeFollowDistancesData(ctx context.Context, sources []strin
3027 return nil, nil
3128 }
3229
33- rows, err := e.db.QueryContext(ctx, `SELECT user_did, target_did FROM main.follows WHERE user_did != target_did`)
34- if err != nil {
35- return nil, err
30+ type pair struct {
31+ src, dst string
3632 }
37- defer rows.Close()
33+ distances := make(map[pair]int)
3834
39- adj := make(map[string][]string)
40- for rows.Next() {
41- var src, dst string
42- if err := rows.Scan(&src, &dst); err != nil {
43- return nil, err
44- }
45- adj[src] = append(adj[src], dst)
46- }
47- if err := rows.Err(); err != nil {
48- return nil, err
49- }
50-
51- var result []followDistance
5235 for _, src := range sources {
53- dist := map[string]int{src: 0}
54- queue := []string{src}
55- for len(queue) > 0 {
56- cur := queue[0]
57- queue = queue[1:]
58- d := dist[cur]
59- if d >= maxFollowDepth {
60- continue
61- }
62- for _, next := range adj[cur] {
63- if _, ok := dist[next]; !ok {
64- dist[next] = d + 1
65- queue = append(queue, next)
36+ frontier := []string{src}
37+ reachable := map[string]int{src: 0}
38+
39+ for depth := 0; depth < maxFollowDepth && len(frontier) > 0; depth++ {
40+ var nextLevel []string
41+ for _, batch := range chunk(frontier, 500) {
42+ ph := make([]string, len(batch))
43+ args := make([]any, len(batch))
44+ for i, did := range batch {
45+ ph[i] = "?"
46+ args[i] = did
47+ }
48+
49+ rows, err := e.db.QueryContext(ctx,
50+ fmt.Sprintf(`SELECT target_did FROM main.follows WHERE user_did IN (%s) AND user_did != target_did`, joinPh(ph)),
51+ args...,
52+ )
53+ if err != nil {
54+ return nil, err
55+ }
56+
57+ for rows.Next() {
58+ var dst string
59+ if err := rows.Scan(&dst); err != nil {
60+ rows.Close()
61+ return nil, err
62+ }
63+ if _, ok := reachable[dst]; !ok {
64+ reachable[dst] = depth + 1
65+ nextLevel = append(nextLevel, dst)
66+ }
6667 }
68+ rows.Close()
6769 }
70+ frontier = nextLevel
6871 }
69- for other, d := range dist {
72+
73+ for other, d := range reachable {
7074 if d > 0 {
71- result = append(result, followDistance{userA: src, userB: other, distance: d})
75+ distances[pair{src, other}] = d
7276 }
7377 }
7478 }
7579
80+ var result []followDistance
81+ for k, d := range distances {
82+ result = append(result, followDistance{userA: k.src, userB: k.dst, distance: d})
83+ }
7684 return result, nil
7785 }
7886
@@ -138,7 +146,7 @@ func (e *Engine) ComputeFollowDistances(ctx context.Context) error {
138146 defer func() { _ = tx.Rollback() }()
139147
140148 const sqliteMaxVars = 500
141- for chunk := range chunk(dirtyUsers, sqliteMaxVars) {
149+ for _, chunk := range chunk(dirtyUsers, sqliteMaxVars) {
142150 ph := make([]string, len(chunk))
143151 args := make([]any, len(chunk))
144152 for i, did := range chunk {
@@ -165,7 +173,7 @@ func (e *Engine) ComputeFollowDistances(ctx context.Context) error {
165173 }
166174 }
167175
168- for chunk := range chunk(dirtyUsers, sqliteMaxVars) {
176+ for _, chunk := range chunk(dirtyUsers, sqliteMaxVars) {
169177 ph := make([]string, len(chunk))
170178 args := make([]any, len(chunk))
171179 for i, did := range chunk {
@@ -3,20 +3,17 @@ package cluster
3 import (3 import (
4 "context"4 "context"
5 "fmt"5 "fmt"
6- "iter"
7 )6 )
8 7
9 const maxFollowDepth = 38 const maxFollowDepth = 3
10 9
11-func chunk[T any](s []T, size int) iter.Seq[[]T] {10+func chunk[T any](s []T, size int) [][]T {
12- return func(yield func([]T) bool) {11+ var chunks [][]T
13- for i := 0; i < len(s); i += size {12+ for i := 0; i < len(s); i += size {
14- end := min(i+size, len(s))13+ end := min(i+size, len(s))
15- if !yield(s[i:end]) {14+ chunks = append(chunks, s[i:end])
16- return
17- }
18- }
19 }15 }
16+ return chunks
20 }17 }
21 18
22 type followDistance struct {19 type followDistance struct {
@@ -30,49 +27,60 @@ func (e *Engine) ComputeFollowDistancesData(ctx context.Context, sources []strin
30 return nil, nil27 return nil, nil
31 }28 }
32 29
33- rows, err := e.db.QueryContext(ctx, `SELECT user_did, target_did FROM main.follows WHERE user_did != target_did`)30+ type pair struct {
34- if err != nil {31+ src, dst string
35- return nil, err
36 }32 }
37- defer rows.Close()33+ distances := make(map[pair]int)
38 34
39- adj := make(map[string][]string)
40- for rows.Next() {
41- var src, dst string
42- if err := rows.Scan(&src, &dst); err != nil {
43- return nil, err
44- }
45- adj[src] = append(adj[src], dst)
46- }
47- if err := rows.Err(); err != nil {
48- return nil, err
49- }
50-
51- var result []followDistance
52 for _, src := range sources {35 for _, src := range sources {
53- dist := map[string]int{src: 0}36+ frontier := []string{src}
54- queue := []string{src}37+ reachable := map[string]int{src: 0}
55- for len(queue) > 0 {38+
56- cur := queue[0]39+ for depth := 0; depth < maxFollowDepth && len(frontier) > 0; depth++ {
57- queue = queue[1:]40+ var nextLevel []string
58- d := dist[cur]41+ for _, batch := range chunk(frontier, 500) {
59- if d >= maxFollowDepth {42+ ph := make([]string, len(batch))
60- continue43+ args := make([]any, len(batch))
61- }44+ for i, did := range batch {
62- for _, next := range adj[cur] {45+ ph[i] = "?"
63- if _, ok := dist[next]; !ok {46+ args[i] = did
64- dist[next] = d + 147+ }
65- queue = append(queue, next)48+
49+ rows, err := e.db.QueryContext(ctx,
50+ fmt.Sprintf(`SELECT target_did FROM main.follows WHERE user_did IN (%s) AND user_did != target_did`, joinPh(ph)),
51+ args...,
52+ )
53+ if err != nil {
54+ return nil, err
55+ }
56+
57+ for rows.Next() {
58+ var dst string
59+ if err := rows.Scan(&dst); err != nil {
60+ rows.Close()
61+ return nil, err
62+ }
63+ if _, ok := reachable[dst]; !ok {
64+ reachable[dst] = depth + 1
65+ nextLevel = append(nextLevel, dst)
66+ }
66 }67 }
68+ rows.Close()
67 }69 }
70+ frontier = nextLevel
68 }71 }
69- for other, d := range dist {72+
73+ for other, d := range reachable {
70 if d > 0 {74 if d > 0 {
71- result = append(result, followDistance{userA: src, userB: other, distance: d})75+ distances[pair{src, other}] = d
72 }76 }
73 }77 }
74 }78 }
75 79
80+ var result []followDistance
81+ for k, d := range distances {
82+ result = append(result, followDistance{userA: k.src, userB: k.dst, distance: d})
83+ }
76 return result, nil84 return result, nil
77 }85 }
78 86
@@ -138,7 +146,7 @@ func (e *Engine) ComputeFollowDistances(ctx context.Context) error {
138 defer func() { _ = tx.Rollback() }()146 defer func() { _ = tx.Rollback() }()
139 147
140 const sqliteMaxVars = 500148 const sqliteMaxVars = 500
141- for chunk := range chunk(dirtyUsers, sqliteMaxVars) {149+ for _, chunk := range chunk(dirtyUsers, sqliteMaxVars) {
142 ph := make([]string, len(chunk))150 ph := make([]string, len(chunk))
143 args := make([]any, len(chunk))151 args := make([]any, len(chunk))
144 for i, did := range chunk {152 for i, did := range chunk {
@@ -165,7 +173,7 @@ func (e *Engine) ComputeFollowDistances(ctx context.Context) error {
165 }173 }
166 }174 }
167 175
168- for chunk := range chunk(dirtyUsers, sqliteMaxVars) {176+ for _, chunk := range chunk(dirtyUsers, sqliteMaxVars) {
169 ph := make([]string, len(chunk))177 ph := make([]string, len(chunk))
170 args := make([]any, len(chunk))178 args := make([]any, len(chunk))
171 for i, did := range chunk {179 for i, did := range chunk {
modified internal/db/db.go +1 -0
@@ -340,6 +340,7 @@ var articlesSchema = []string{
340340 `CREATE INDEX IF NOT EXISTS articles.idx_subscriptions_feed ON subscriptions(feed_url)`,
341341 `CREATE INDEX IF NOT EXISTS articles.idx_subscriptions_feed_user ON subscriptions(feed_url, user_did)`,
342342 `CREATE INDEX IF NOT EXISTS articles.idx_subscriptions_user ON subscriptions(user_did)`,
343+ `CREATE INDEX IF NOT EXISTS articles.idx_subscriptions_user_feed ON subscriptions(user_did, feed_url)`,
343344 `CREATE INDEX IF NOT EXISTS articles.idx_subscriptions_uri ON subscriptions(uri)`,
344345 `CREATE INDEX IF NOT EXISTS articles.idx_likes_author_feed ON likes(author_did, feed_url, created_at)`,
345346 `CREATE INDEX IF NOT EXISTS articles.idx_articles_feed ON articles(feed_url)`,
@@ -340,6 +340,7 @@ var articlesSchema = []string{
340 `CREATE INDEX IF NOT EXISTS articles.idx_subscriptions_feed ON subscriptions(feed_url)`,340 `CREATE INDEX IF NOT EXISTS articles.idx_subscriptions_feed ON subscriptions(feed_url)`,
341 `CREATE INDEX IF NOT EXISTS articles.idx_subscriptions_feed_user ON subscriptions(feed_url, user_did)`,341 `CREATE INDEX IF NOT EXISTS articles.idx_subscriptions_feed_user ON subscriptions(feed_url, user_did)`,
342 `CREATE INDEX IF NOT EXISTS articles.idx_subscriptions_user ON subscriptions(user_did)`,342 `CREATE INDEX IF NOT EXISTS articles.idx_subscriptions_user ON subscriptions(user_did)`,
343+ `CREATE INDEX IF NOT EXISTS articles.idx_subscriptions_user_feed ON subscriptions(user_did, feed_url)`,
343 `CREATE INDEX IF NOT EXISTS articles.idx_subscriptions_uri ON subscriptions(uri)`,344 `CREATE INDEX IF NOT EXISTS articles.idx_subscriptions_uri ON subscriptions(uri)`,
344 `CREATE INDEX IF NOT EXISTS articles.idx_likes_author_feed ON likes(author_did, feed_url, created_at)`,345 `CREATE INDEX IF NOT EXISTS articles.idx_likes_author_feed ON likes(author_did, feed_url, created_at)`,
345 `CREATE INDEX IF NOT EXISTS articles.idx_articles_feed ON articles(feed_url)`,346 `CREATE INDEX IF NOT EXISTS articles.idx_articles_feed ON articles(feed_url)`,