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

Add collectiondir (`com.atproto.sync.listReposByCollection`) endpoint support to backfill on startupUnverified

Julien Robert committed 2026-04-21T17:07:26+02:00 Browse files
e7fd77c parent: fd9ed81
modified .env.example +1 -0
@@ -4,6 +4,7 @@ GLEAN_JETSTREAM=wss://jetstream.glean.at
44 GLEAN_PLC_URL=https://didplc.glean.at
55 GLEAN_SYNC_INTERVAL=1h
66 GLEAN_CLUSTER_INTERVAL=6h
7+GLEAN_COLLECTION_DIR_URL=https://lightrail.microcosm.blue/xrpc/com.atproto.sync.listReposByCollection?collection=at.glean.subscription
78 # Leave empty for localhost OAuth (development)
89 # GLEAN_OAUTH_CLIENT_ID=https://glean.at/oauth/client-metadata
910 # GLEAN_OAUTH_REDIRECT_URL=https://glean.at/auth/callback
@@ -4,6 +4,7 @@ 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=6h
7+GLEAN_COLLECTION_DIR_URL=https://lightrail.microcosm.blue/xrpc/com.atproto.sync.listReposByCollection?collection=at.glean.subscription
7 # Leave empty for localhost OAuth (development)8 # Leave empty for localhost OAuth (development)
8 # GLEAN_OAUTH_CLIENT_ID=https://glean.at/oauth/client-metadata9 # GLEAN_OAUTH_CLIENT_ID=https://glean.at/oauth/client-metadata
9 # GLEAN_OAUTH_REDIRECT_URL=https://glean.at/auth/callback10 # GLEAN_OAUTH_REDIRECT_URL=https://glean.at/auth/callback
modified internal/atproto/client.go +4 -0
@@ -17,6 +17,10 @@ func NewClient(api *atclient.APIClient) *Client {
1717 return &Client{api: api}
1818 }
1919
20+func NewUnauthenticatedClient(pdsURL string) *Client {
21+ return &Client{api: atclient.NewAPIClient(pdsURL)}
22+}
23+
2024 func (c *Client) CreateRecord(ctx context.Context, did, collection string, record any) (string, string, error) {
2125 input := map[string]any{
2226 "repo": did,
@@ -17,6 +17,10 @@ func NewClient(api *atclient.APIClient) *Client {
17 return &Client{api: api}17 return &Client{api: api}
18 }18 }
19 19
20+func NewUnauthenticatedClient(pdsURL string) *Client {
21+ return &Client{api: atclient.NewAPIClient(pdsURL)}
22+}
23+
20 func (c *Client) CreateRecord(ctx context.Context, did, collection string, record any) (string, string, error) {24 func (c *Client) CreateRecord(ctx context.Context, did, collection string, record any) (string, string, error) {
21 input := map[string]any{25 input := map[string]any{
22 "repo": did,26 "repo": did,
added internal/atproto/collectiondir.go +67 -0
new file mode 100644
@@ -0,0 +1,67 @@
1+package atproto
2+
3+import (
4+ "context"
5+ "encoding/json"
6+ "fmt"
7+ "io"
8+ "net/http"
9+ "time"
10+)
11+
12+// https://github.com/bluesky-social/indigo/tree/main/cmd/collectiondir
13+type listReposByCollectionResponse struct {
14+ Repos []struct {
15+ DID string `json:"did"`
16+ } `json:"repos"`
17+ Cursor string `json:"cursor"`
18+}
19+
20+func FetchSubscriberDIDs(ctx context.Context, baseURL string) ([]string, error) {
21+ var allDIDs []string
22+ cursor := ""
23+ client := &http.Client{Timeout: 30 * time.Second}
24+
25+ for {
26+ url := baseURL
27+ if cursor != "" {
28+ url = fmt.Sprintf("%s&cursor=%s", baseURL, cursor)
29+ }
30+
31+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
32+ if err != nil {
33+ return nil, fmt.Errorf("building request: %w", err)
34+ }
35+
36+ resp, err := client.Do(req)
37+ if err != nil {
38+ return nil, fmt.Errorf("fetching collection dir: %w", err)
39+ }
40+
41+ body, err := io.ReadAll(resp.Body)
42+ resp.Body.Close()
43+ if err != nil {
44+ return nil, fmt.Errorf("reading response: %w", err)
45+ }
46+
47+ if resp.StatusCode != http.StatusOK {
48+ return nil, fmt.Errorf("collection dir returned %d: %s", resp.StatusCode, string(body))
49+ }
50+
51+ var result listReposByCollectionResponse
52+ if err := json.Unmarshal(body, &result); err != nil {
53+ return nil, fmt.Errorf("parsing response: %w", err)
54+ }
55+
56+ for _, r := range result.Repos {
57+ allDIDs = append(allDIDs, r.DID)
58+ }
59+
60+ if result.Cursor == "" || len(result.Repos) == 0 {
61+ break
62+ }
63+ cursor = result.Cursor
64+ }
65+
66+ return allDIDs, nil
67+}
new file mode 100644
@@ -0,0 +1,67 @@
1+package atproto
2+
3+import (
4+ "context"
5+ "encoding/json"
6+ "fmt"
7+ "io"
8+ "net/http"
9+ "time"
10+)
11+
12+// https://github.com/bluesky-social/indigo/tree/main/cmd/collectiondir
13+type listReposByCollectionResponse struct {
14+ Repos []struct {
15+ DID string `json:"did"`
16+ } `json:"repos"`
17+ Cursor string `json:"cursor"`
18+}
19+
20+func FetchSubscriberDIDs(ctx context.Context, baseURL string) ([]string, error) {
21+ var allDIDs []string
22+ cursor := ""
23+ client := &http.Client{Timeout: 30 * time.Second}
24+
25+ for {
26+ url := baseURL
27+ if cursor != "" {
28+ url = fmt.Sprintf("%s&cursor=%s", baseURL, cursor)
29+ }
30+
31+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
32+ if err != nil {
33+ return nil, fmt.Errorf("building request: %w", err)
34+ }
35+
36+ resp, err := client.Do(req)
37+ if err != nil {
38+ return nil, fmt.Errorf("fetching collection dir: %w", err)
39+ }
40+
41+ body, err := io.ReadAll(resp.Body)
42+ resp.Body.Close()
43+ if err != nil {
44+ return nil, fmt.Errorf("reading response: %w", err)
45+ }
46+
47+ if resp.StatusCode != http.StatusOK {
48+ return nil, fmt.Errorf("collection dir returned %d: %s", resp.StatusCode, string(body))
49+ }
50+
51+ var result listReposByCollectionResponse
52+ if err := json.Unmarshal(body, &result); err != nil {
53+ return nil, fmt.Errorf("parsing response: %w", err)
54+ }
55+
56+ for _, r := range result.Repos {
57+ allDIDs = append(allDIDs, r.DID)
58+ }
59+
60+ if result.Cursor == "" || len(result.Repos) == 0 {
61+ break
62+ }
63+ cursor = result.Cursor
64+ }
65+
66+ return allDIDs, nil
67+}
added internal/atproto/collectiondir_test.go +97 -0
new file mode 100644
@@ -0,0 +1,97 @@
1+package atproto
2+
3+import (
4+ "context"
5+ "encoding/json"
6+ "net/http"
7+ "net/http/httptest"
8+ "testing"
9+
10+ "gotest.tools/v3/assert"
11+)
12+
13+func TestFetchSubscriberDIDs(t *testing.T) {
14+ page1 := struct {
15+ Repos []struct {
16+ DID string `json:"did"`
17+ } `json:"repos"`
18+ Cursor string `json:"cursor"`
19+ }{
20+ Repos: []struct {
21+ DID string `json:"did"`
22+ }{
23+ {DID: "did:plc:aaa"},
24+ {DID: "did:plc:bbb"},
25+ },
26+ Cursor: "nextpage",
27+ }
28+ page2 := struct {
29+ Repos []struct {
30+ DID string `json:"did"`
31+ } `json:"repos"`
32+ Cursor string `json:"cursor"`
33+ }{
34+ Repos: []struct {
35+ DID string `json:"did"`
36+ }{
37+ {DID: "did:plc:ccc"},
38+ },
39+ Cursor: "",
40+ }
41+
42+ callCount := 0
43+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
44+ callCount++
45+ assert.Equal(t, r.URL.Query().Get("collection"), "at.glean.subscription")
46+
47+ w.Header().Set("Content-Type", "application/json")
48+ if callCount == 1 {
49+ assert.Equal(t, r.URL.Query().Get("cursor"), "")
50+ json.NewEncoder(w).Encode(page1)
51+ } else {
52+ assert.Equal(t, r.URL.Query().Get("cursor"), "nextpage")
53+ json.NewEncoder(w).Encode(page2)
54+ }
55+ }))
56+ defer server.Close()
57+
58+ url := server.URL + "/xrpc/com.atproto.sync.listReposByCollection?collection=at.glean.subscription"
59+ dids, err := FetchSubscriberDIDs(context.Background(), url)
60+ assert.NilError(t, err)
61+ assert.Equal(t, callCount, 2)
62+ assert.DeepEqual(t, dids, []string{"did:plc:aaa", "did:plc:bbb", "did:plc:ccc"})
63+}
64+
65+func TestFetchSubscriberDIDs_EmptyResponse(t *testing.T) {
66+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
67+ w.Header().Set("Content-Type", "application/json")
68+ json.NewEncoder(w).Encode(map[string]any{"repos": []any{}})
69+ }))
70+ defer server.Close()
71+
72+ url := server.URL + "/xrpc/com.atproto.sync.listReposByCollection?collection=at.glean.subscription"
73+ dids, err := FetchSubscriberDIDs(context.Background(), url)
74+ assert.NilError(t, err)
75+ assert.Equal(t, len(dids), 0)
76+}
77+
78+func TestFetchSubscriberDIDs_ServerError(t *testing.T) {
79+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
80+ w.WriteHeader(http.StatusInternalServerError)
81+ w.Write([]byte("internal error"))
82+ }))
83+ defer server.Close()
84+
85+ url := server.URL + "/xrpc/com.atproto.sync.listReposByCollection?collection=at.glean.subscription"
86+ _, err := FetchSubscriberDIDs(context.Background(), url)
87+ assert.Assert(t, err != nil)
88+}
89+
90+func TestFetchSubscriberDIDs_CancelledContext(t *testing.T) {
91+ ctx, cancel := context.WithCancel(context.Background())
92+ cancel()
93+
94+ url := "http://localhost/xrpc/com.atproto.sync.listReposByCollection?collection=at.glean.subscription"
95+ _, err := FetchSubscriberDIDs(ctx, url)
96+ assert.Assert(t, err != nil)
97+}
new file mode 100644
@@ -0,0 +1,97 @@
1+package atproto
2+
3+import (
4+ "context"
5+ "encoding/json"
6+ "net/http"
7+ "net/http/httptest"
8+ "testing"
9+
10+ "gotest.tools/v3/assert"
11+)
12+
13+func TestFetchSubscriberDIDs(t *testing.T) {
14+ page1 := struct {
15+ Repos []struct {
16+ DID string `json:"did"`
17+ } `json:"repos"`
18+ Cursor string `json:"cursor"`
19+ }{
20+ Repos: []struct {
21+ DID string `json:"did"`
22+ }{
23+ {DID: "did:plc:aaa"},
24+ {DID: "did:plc:bbb"},
25+ },
26+ Cursor: "nextpage",
27+ }
28+ page2 := struct {
29+ Repos []struct {
30+ DID string `json:"did"`
31+ } `json:"repos"`
32+ Cursor string `json:"cursor"`
33+ }{
34+ Repos: []struct {
35+ DID string `json:"did"`
36+ }{
37+ {DID: "did:plc:ccc"},
38+ },
39+ Cursor: "",
40+ }
41+
42+ callCount := 0
43+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
44+ callCount++
45+ assert.Equal(t, r.URL.Query().Get("collection"), "at.glean.subscription")
46+
47+ w.Header().Set("Content-Type", "application/json")
48+ if callCount == 1 {
49+ assert.Equal(t, r.URL.Query().Get("cursor"), "")
50+ json.NewEncoder(w).Encode(page1)
51+ } else {
52+ assert.Equal(t, r.URL.Query().Get("cursor"), "nextpage")
53+ json.NewEncoder(w).Encode(page2)
54+ }
55+ }))
56+ defer server.Close()
57+
58+ url := server.URL + "/xrpc/com.atproto.sync.listReposByCollection?collection=at.glean.subscription"
59+ dids, err := FetchSubscriberDIDs(context.Background(), url)
60+ assert.NilError(t, err)
61+ assert.Equal(t, callCount, 2)
62+ assert.DeepEqual(t, dids, []string{"did:plc:aaa", "did:plc:bbb", "did:plc:ccc"})
63+}
64+
65+func TestFetchSubscriberDIDs_EmptyResponse(t *testing.T) {
66+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
67+ w.Header().Set("Content-Type", "application/json")
68+ json.NewEncoder(w).Encode(map[string]any{"repos": []any{}})
69+ }))
70+ defer server.Close()
71+
72+ url := server.URL + "/xrpc/com.atproto.sync.listReposByCollection?collection=at.glean.subscription"
73+ dids, err := FetchSubscriberDIDs(context.Background(), url)
74+ assert.NilError(t, err)
75+ assert.Equal(t, len(dids), 0)
76+}
77+
78+func TestFetchSubscriberDIDs_ServerError(t *testing.T) {
79+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
80+ w.WriteHeader(http.StatusInternalServerError)
81+ w.Write([]byte("internal error"))
82+ }))
83+ defer server.Close()
84+
85+ url := server.URL + "/xrpc/com.atproto.sync.listReposByCollection?collection=at.glean.subscription"
86+ _, err := FetchSubscriberDIDs(context.Background(), url)
87+ assert.Assert(t, err != nil)
88+}
89+
90+func TestFetchSubscriberDIDs_CancelledContext(t *testing.T) {
91+ ctx, cancel := context.WithCancel(context.Background())
92+ cancel()
93+
94+ url := "http://localhost/xrpc/com.atproto.sync.listReposByCollection?collection=at.glean.subscription"
95+ _, err := FetchSubscriberDIDs(ctx, url)
96+ assert.Assert(t, err != nil)
97+}
modified internal/db/user.go +18 -0
@@ -54,6 +54,24 @@ func (db *DB) GetUserByHandle(ctx context.Context, handle string) (*User, error)
5454 return u, nil
5555 }
5656
57+func (db *DB) ListUserDIDs(ctx context.Context) (map[string]bool, error) {
58+ rows, err := db.QueryContext(ctx, `SELECT did FROM users`)
59+ if err != nil {
60+ return nil, err
61+ }
62+ defer rows.Close()
63+
64+ dids := make(map[string]bool)
65+ for rows.Next() {
66+ var did string
67+ if err := rows.Scan(&did); err != nil {
68+ return nil, err
69+ }
70+ dids[did] = true
71+ }
72+ return dids, rows.Err()
73+}
74+
5775 func (db *DB) ListUsers(ctx context.Context) ([]*User, error) {
5876 rows, err := db.QueryContext(ctx, `
5977 SELECT did, handle, display_name, avatar_url, indexed_at, updated_at
@@ -54,6 +54,24 @@ func (db *DB) GetUserByHandle(ctx context.Context, handle string) (*User, error)
54 return u, nil54 return u, nil
55 }55 }
56 56
57+func (db *DB) ListUserDIDs(ctx context.Context) (map[string]bool, error) {
58+ rows, err := db.QueryContext(ctx, `SELECT did FROM users`)
59+ if err != nil {
60+ return nil, err
61+ }
62+ defer rows.Close()
63+
64+ dids := make(map[string]bool)
65+ for rows.Next() {
66+ var did string
67+ if err := rows.Scan(&did); err != nil {
68+ return nil, err
69+ }
70+ dids[did] = true
71+ }
72+ return dids, rows.Err()
73+}
74+
57 func (db *DB) ListUsers(ctx context.Context) ([]*User, error) {75 func (db *DB) ListUsers(ctx context.Context) ([]*User, error) {
58 rows, err := db.QueryContext(ctx, `76 rows, err := db.QueryContext(ctx, `
59 SELECT did, handle, display_name, avatar_url, indexed_at, updated_at77 SELECT did, handle, display_name, avatar_url, indexed_at, updated_at
modified internal/server/server.go +62 -0
@@ -440,6 +440,68 @@ func (s *Server) PeriodicSync(ctx context.Context, interval time.Duration) {
440440 }
441441 }
442442
443+func (s *Server) BackfillFromCollectionDir(ctx context.Context, collectionDirURL string) {
444+ if collectionDirURL == "" {
445+ return
446+ }
447+
448+ s.logger.Info("backfilling from collection directory", "url", collectionDirURL)
449+
450+ dids, err := atproto.FetchSubscriberDIDs(ctx, collectionDirURL)
451+ if err != nil {
452+ s.logger.Error("failed to fetch subscriber DIDs", "error", err)
453+ return
454+ }
455+
456+ existing, err := s.db.ListUserDIDs(ctx)
457+ if err != nil {
458+ s.logger.Error("failed to list existing users", "error", err)
459+ return
460+ }
461+
462+ var missing []string
463+ for _, did := range dids {
464+ if !existing[did] {
465+ missing = append(missing, did)
466+ }
467+ }
468+
469+ s.logger.Info("collection directory backfill", "total", len(dids), "missing", len(missing))
470+
471+ for _, did := range missing {
472+ if ctx.Err() != nil {
473+ return
474+ }
475+
476+ handle := did
477+ if ident, err := atproto.ResolveIdentity(ctx, did); err == nil {
478+ handle = ident.Handle.String()
479+ }
480+
481+ if _, err := s.db.CreateUser(ctx, did, handle, "", ""); err != nil {
482+ s.logger.Error("failed to create user during backfill", "error", err, "did", did)
483+ continue
484+ }
485+
486+ pdsURL, err := atproto.ResolvePDSEndpoint(ctx, did)
487+ if err != nil {
488+ s.logger.Error("failed to resolve PDS for backfill", "error", err, "did", did)
489+ continue
490+ }
491+
492+ client := atproto.NewUnauthenticatedClient(pdsURL)
493+ sync := atproto.NewSync(s.db, client, s.logger)
494+ if err := sync.Run(ctx, did); err != nil {
495+ s.logger.Error("backfill sync failed", "error", err, "did", did)
496+ }
497+
498+ s.refreshUserFeeds(ctx, did)
499+ s.engine.ComputeForUser(ctx, did)
500+ }
501+
502+ s.logger.Info("collection directory backfill complete")
503+}
504+
443505 func (s *Server) runSyncAll(ctx context.Context) {
444506 users, err := s.db.ListUsers(ctx)
445507 if err != nil {
@@ -440,6 +440,68 @@ func (s *Server) PeriodicSync(ctx context.Context, interval time.Duration) {
440 }440 }
441 }441 }
442 442
443+func (s *Server) BackfillFromCollectionDir(ctx context.Context, collectionDirURL string) {
444+ if collectionDirURL == "" {
445+ return
446+ }
447+
448+ s.logger.Info("backfilling from collection directory", "url", collectionDirURL)
449+
450+ dids, err := atproto.FetchSubscriberDIDs(ctx, collectionDirURL)
451+ if err != nil {
452+ s.logger.Error("failed to fetch subscriber DIDs", "error", err)
453+ return
454+ }
455+
456+ existing, err := s.db.ListUserDIDs(ctx)
457+ if err != nil {
458+ s.logger.Error("failed to list existing users", "error", err)
459+ return
460+ }
461+
462+ var missing []string
463+ for _, did := range dids {
464+ if !existing[did] {
465+ missing = append(missing, did)
466+ }
467+ }
468+
469+ s.logger.Info("collection directory backfill", "total", len(dids), "missing", len(missing))
470+
471+ for _, did := range missing {
472+ if ctx.Err() != nil {
473+ return
474+ }
475+
476+ handle := did
477+ if ident, err := atproto.ResolveIdentity(ctx, did); err == nil {
478+ handle = ident.Handle.String()
479+ }
480+
481+ if _, err := s.db.CreateUser(ctx, did, handle, "", ""); err != nil {
482+ s.logger.Error("failed to create user during backfill", "error", err, "did", did)
483+ continue
484+ }
485+
486+ pdsURL, err := atproto.ResolvePDSEndpoint(ctx, did)
487+ if err != nil {
488+ s.logger.Error("failed to resolve PDS for backfill", "error", err, "did", did)
489+ continue
490+ }
491+
492+ client := atproto.NewUnauthenticatedClient(pdsURL)
493+ sync := atproto.NewSync(s.db, client, s.logger)
494+ if err := sync.Run(ctx, did); err != nil {
495+ s.logger.Error("backfill sync failed", "error", err, "did", did)
496+ }
497+
498+ s.refreshUserFeeds(ctx, did)
499+ s.engine.ComputeForUser(ctx, did)
500+ }
501+
502+ s.logger.Info("collection directory backfill complete")
503+}
504+
443 func (s *Server) runSyncAll(ctx context.Context) {505 func (s *Server) runSyncAll(ctx context.Context) {
444 users, err := s.db.ListUsers(ctx)506 users, err := s.db.ListUsers(ctx)
445 if err != nil {507 if err != nil {
modified main.go +4 -0
@@ -24,6 +24,7 @@ func main() {
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")
2626 clusterInterval := flag.Duration("cluster-interval", envDuration("GLEAN_CLUSTER_INTERVAL", 6*time.Hour), "cluster recomputation interval")
27+ collectionDirURL := flag.String("collection-dir", envOr("GLEAN_COLLECTION_DIR_URL", ""), "collection directory URL for startup backfill")
2728 flag.Parse()
2829
2930 atproto.InitIdentity(envOr("GLEAN_PLC_URL", "https://didplc.glean.at"))
@@ -68,6 +69,9 @@ func main() {
6869 go func() {
6970 srv.PeriodicSync(ctx, *syncInterval)
7071 }()
72+ go func() {
73+ srv.BackfillFromCollectionDir(ctx, *collectionDirURL)
74+ }()
7175 go func() {
7276 if err := jetstream.Start(ctx); err != nil && ctx.Err() == nil {
7377 logger.Error("jetstream error", "error", err)
@@ -24,6 +24,7 @@ func main() {
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", 6*time.Hour), "cluster recomputation interval")
27+ collectionDirURL := flag.String("collection-dir", envOr("GLEAN_COLLECTION_DIR_URL", ""), "collection directory URL for startup backfill")
27 flag.Parse()28 flag.Parse()
28 29
29 atproto.InitIdentity(envOr("GLEAN_PLC_URL", "https://didplc.glean.at"))30 atproto.InitIdentity(envOr("GLEAN_PLC_URL", "https://didplc.glean.at"))
@@ -68,6 +69,9 @@ func main() {
68 go func() {69 go func() {
69 srv.PeriodicSync(ctx, *syncInterval)70 srv.PeriodicSync(ctx, *syncInterval)
70 }()71 }()
72+ go func() {
73+ srv.BackfillFromCollectionDir(ctx, *collectionDirURL)
74+ }()
71 go func() {75 go func() {
72 if err := jetstream.Start(ctx); err != nil && ctx.Err() == nil {76 if err := jetstream.Start(ctx); err != nil && ctx.Err() == nil {
73 logger.Error("jetstream error", "error", err)77 logger.Error("jetstream error", "error", err)