Highly ambitious ATProtocol AppView service and sdks
fix infinite loop when fetching sync summary, update ui for sync modals, update start sync with max repos limit
+126
-67
+1
api/src/jobs.rs
+1
api/src/jobs.rs
+1
api/src/models.rs
+1
api/src/models.rs
+32
-3
api/src/sync.rs
+32
-3
api/src/sync.rs
···
212
212
external_collections: Option<&[String]>,
213
213
repos: Option<&[String]>,
214
214
skip_validation: bool,
215
+
max_repos: Option<i32>,
215
216
) -> Result<(i64, i64), SyncError> {
216
217
info!("Starting backfill operation");
217
218
···
252
253
// First, get all repos from primary collections
253
254
let mut primary_repos = std::collections::HashSet::new();
254
255
for collection in &primary_collections {
255
-
match self.get_repos_for_collection(collection, slice_uri).await {
256
+
match self.get_repos_for_collection(collection, slice_uri, max_repos).await {
256
257
Ok(repos) => {
257
258
info!(
258
259
"Found {} repositories for primary collection \"{}\"",
···
671
672
&self,
672
673
collection: &str,
673
674
slice_uri: &str,
675
+
max_repos: Option<i32>,
674
676
) -> Result<Vec<String>, SyncError> {
675
677
let url = format!(
676
678
"{}/xrpc/com.atproto.sync.listReposByCollection",
···
678
680
);
679
681
let mut all_repos = Vec::new();
680
682
let mut cursor: Option<String> = None;
683
+
let mut page_count = 0;
684
+
685
+
// AT Protocol docs: default 500 repos/page, max 2000 repos/page
686
+
// We use 1000 repos/page for efficiency (recommended for large DID lists)
687
+
const REPOS_PER_PAGE: usize = 1000; // Our configured page size
688
+
689
+
// Calculate max pages based on repo limit, with a reasonable safety margin
690
+
let max_pages = if let Some(limit) = max_repos {
691
+
// Add 20% safety margin and ensure at least 5 pages
692
+
((limit as usize * 120 / 100) / REPOS_PER_PAGE).max(5)
693
+
} else {
694
+
25 // Default fallback for unlimited
695
+
};
681
696
682
697
loop {
683
-
let mut query_params = vec![("collection", collection.to_string())];
698
+
page_count += 1;
699
+
if page_count > max_pages {
700
+
warn!(
701
+
"Reached maximum page limit ({}) for collection {} (based on repo limit {:?}, estimated max {} repos at {} per page)",
702
+
max_pages, collection, max_repos, max_pages * REPOS_PER_PAGE, REPOS_PER_PAGE
703
+
);
704
+
break;
705
+
}
706
+
707
+
let mut query_params = vec![
708
+
("collection", collection.to_string()),
709
+
("limit", "1000".to_string()),
710
+
];
684
711
if let Some(ref cursor_value) = cursor {
685
712
query_params.push(("cursor", cursor_value.clone()));
686
713
}
···
708
735
Some(json!({
709
736
"collection": collection,
710
737
"repos_count": all_repos.len(),
711
-
"has_more": true
738
+
"has_more": true,
739
+
"page": page_count
712
740
}))
713
741
);
714
742
}
···
1165
1193
Some(&external_collections),
1166
1194
Some(&[user_did.to_string()]), // Only sync this user's repos
1167
1195
false, // Always validate user collections
1196
+
None, // No limit for user-specific sync
1168
1197
)
1169
1198
.await
1170
1199
};
+28
-36
api/src/xrpc/network/slices/slice/get_sync_summary.rs
+28
-36
api/src/xrpc/network/slices/slice/get_sync_summary.rs
···
35
35
where
36
36
E: de::Error,
37
37
{
38
-
Ok(Some(vec![value.to_string()]))
38
+
// Handle comma-separated strings by splitting them
39
+
let items: Vec<String> = value
40
+
.split(',')
41
+
.map(|s| s.trim().to_string())
42
+
.filter(|s| !s.is_empty())
43
+
.collect();
44
+
45
+
if items.is_empty() {
46
+
Ok(None)
47
+
} else {
48
+
Ok(Some(items))
49
+
}
39
50
}
40
51
41
52
fn visit_seq<A>(self, mut seq: A) -> Result<Self::Value, A::Error>
···
123
134
state.auth_cache.clone(),
124
135
);
125
136
126
-
// Discover repos if not provided
127
-
let all_repos = if let Some(provided_repos) = user_provided_repos {
128
-
provided_repos
137
+
// Discover repos if not provided, and track collection repo counts
138
+
let (all_repos, collection_repo_counts) = if let Some(provided_repos) = user_provided_repos {
139
+
(provided_repos, HashMap::new())
129
140
} else {
130
141
// Discover repos from collections
131
142
let mut discovered_repos = std::collections::HashSet::new();
143
+
let mut counts: HashMap<String, i64> = HashMap::new();
132
144
133
-
// Get repos from primary collections
145
+
// First, get repos ONLY from primary collections
134
146
for collection in &primary_collections {
135
-
match sync_service.get_repos_for_collection(collection, slice_uri).await {
147
+
match sync_service.get_repos_for_collection(collection, slice_uri, Some(applied_limit)).await {
136
148
Ok(repos) => {
149
+
counts.insert(collection.clone(), repos.len() as i64);
137
150
discovered_repos.extend(repos);
138
151
}
139
152
Err(e) => {
140
153
tracing::warn!("Failed to get repos for collection {}: {}", collection, e);
154
+
counts.insert(collection.clone(), 0);
141
155
}
142
156
}
143
157
}
144
158
145
-
// Get repos from external collections
146
-
for collection in &external_collections {
147
-
match sync_service.get_repos_for_collection(collection, slice_uri).await {
148
-
Ok(repos) => {
149
-
discovered_repos.extend(repos);
150
-
}
151
-
Err(e) => {
152
-
tracing::warn!("Failed to get repos for collection {}: {}", collection, e);
153
-
}
154
-
}
155
-
}
156
-
157
-
discovered_repos.into_iter().collect()
159
+
// For external collections, we don't include repo counts since we don't fetch them
160
+
(discovered_repos.into_iter().collect(), counts)
158
161
};
159
162
160
163
let total_repos = all_repos.len() as i64;
161
164
let capped_repos = std::cmp::min(total_repos, applied_limit as i64);
162
165
let would_be_capped = total_repos > applied_limit as i64;
163
166
164
-
// Build collections summary
167
+
// Build collections summary using cached repo counts
165
168
let mut collections_summary = Vec::new();
166
-
let mut collection_repo_counts: HashMap<String, i64> = HashMap::new();
167
169
168
-
// Count repos per collection (this is an approximation)
170
+
// Add primary collections to summary
169
171
for collection in &primary_collections {
170
172
let is_external = !collection.starts_with(&slice_domain);
171
-
let estimated_repos = if let Ok(repos) = sync_service.get_repos_for_collection(collection, slice_uri).await {
172
-
repos.len() as i64
173
-
} else {
174
-
0
175
-
};
173
+
let estimated_repos = *collection_repo_counts.get(collection).unwrap_or(&0);
176
174
177
-
collection_repo_counts.insert(collection.clone(), estimated_repos);
178
175
collections_summary.push(CollectionSummary {
179
176
collection: collection.clone(),
180
177
estimated_repos,
···
182
179
});
183
180
}
184
181
182
+
// Add external collections to summary (no repo counts since we don't fetch them)
185
183
for collection in &external_collections {
186
184
let is_external = !collection.starts_with(&slice_domain);
187
-
let estimated_repos = if let Ok(repos) = sync_service.get_repos_for_collection(collection, slice_uri).await {
188
-
repos.len() as i64
189
-
} else {
190
-
0
191
-
};
192
185
193
-
collection_repo_counts.insert(collection.clone(), estimated_repos);
194
186
collections_summary.push(CollectionSummary {
195
187
collection: collection.clone(),
196
-
estimated_repos,
188
+
estimated_repos: 0, // No count for external collections
197
189
is_external,
198
190
});
199
191
}
···
205
197
would_be_capped,
206
198
applied_limit,
207
199
}))
208
-
}
200
+
}
+7
-1
api/src/xrpc/network/slices/slice/start_sync.rs
+7
-1
api/src/xrpc/network/slices/slice/start_sync.rs
···
34
34
let user_did = user_info.sub;
35
35
let slice_uri = params.slice;
36
36
37
+
// Ensure max_repos is set from config if not provided
38
+
let mut sync_params = params.sync_params;
39
+
if sync_params.max_repos.is_none() {
40
+
sync_params.max_repos = Some(state.config.default_max_sync_repos);
41
+
}
42
+
37
43
let job_id = jobs::enqueue_sync_job(
38
44
&state.database_pool,
39
45
user_did,
40
46
slice_uri.clone(),
41
-
params.sync_params,
47
+
sync_params,
42
48
)
43
49
.await
44
50
.map_err(|e| AppError::Internal(format!("Failed to enqueue sync job: {}", e)))?;
+3
-2
frontend/src/features/slices/sync/templates/fragments/SyncFormModal.tsx
+3
-2
frontend/src/features/slices/sync/templates/fragments/SyncFormModal.tsx
···
17
17
return (
18
18
<Modal
19
19
title="Sync Collections"
20
-
description="Sync entire collections from AT Protocol network to this slice."
20
+
description="Import data from selected collections into your slice."
21
+
size="md"
21
22
>
22
23
<form className="space-y-4">
23
24
<div className="space-y-2">
···
84
85
<div className="flex justify-end gap-3">
85
86
<Button
86
87
type="button"
87
-
variant="secondary"
88
+
variant="outline"
88
89
/* @ts-ignore - Hyperscript attribute */
89
90
_="on click set #modal-container's innerHTML to ''"
90
91
>
+54
-25
frontend/src/features/slices/sync/templates/fragments/SyncSummaryModal.tsx
+54
-25
frontend/src/features/slices/sync/templates/fragments/SyncSummaryModal.tsx
···
1
1
import { Button } from "../../../../../shared/fragments/Button.tsx";
2
2
import { Modal } from "../../../../../shared/fragments/Modal.tsx";
3
+
import { Text } from "../../../../../shared/fragments/Text.tsx";
3
4
import type { NetworkSlicesSliceGetSyncSummaryOutput } from "../../../../../client.ts";
4
5
5
6
interface SyncSummaryModalProps {
···
21
22
<Modal
22
23
title="Sync Summary"
23
24
description="Review what will be synced before starting the operation."
25
+
size="md"
24
26
>
25
27
<div className="space-y-6">
26
28
{/* Summary Stats */}
···
83
85
<h3 className="text-sm font-medium text-zinc-700 dark:text-zinc-300 mb-3">
84
86
Collections Breakdown
85
87
</h3>
86
-
<div className="space-y-2">
87
-
{summary.collectionsSummary.map((collection) => (
88
-
<div
89
-
key={collection.collection}
90
-
className="flex items-center justify-between p-3 bg-white dark:bg-zinc-900 border border-zinc-200 dark:border-zinc-700 rounded-md"
91
-
>
92
-
<div className="flex items-center space-x-3">
93
-
<div
94
-
className={`w-2 h-2 rounded-full ${
95
-
collection.isExternal ? "bg-blue-500" : "bg-green-500"
96
-
}`}
97
-
></div>
98
-
<div>
99
-
<p className="text-sm font-medium text-zinc-900 dark:text-zinc-100">
100
-
{collection.collection}
101
-
</p>
102
-
<p className="text-xs text-zinc-500">
103
-
{collection.isExternal ? "External" : "Primary"}{" "}
104
-
Collection
105
-
</p>
106
-
</div>
88
+
<div className="space-y-4">
89
+
{/* Primary Collections */}
90
+
{summary.collectionsSummary.filter(c => !c.isExternal).length > 0 && (
91
+
<div>
92
+
<h4 className="text-xs font-medium text-zinc-600 dark:text-zinc-400 mb-2">
93
+
Primary Collections
94
+
</h4>
95
+
<div className="space-y-1 ml-4">
96
+
{summary.collectionsSummary
97
+
.filter(c => !c.isExternal)
98
+
.map((collection) => (
99
+
<div
100
+
key={collection.collection}
101
+
className="flex items-center py-1"
102
+
>
103
+
<Text variant="secondary" size="sm">
104
+
{collection.collection}
105
+
</Text>
106
+
<div className="flex-1 border-b border-dotted border-zinc-300 dark:border-zinc-600 mx-2 self-end mb-1"></div>
107
+
<Text variant="secondary" size="sm">
108
+
{collection.estimatedRepos.toLocaleString()} repos
109
+
</Text>
110
+
</div>
111
+
))}
112
+
</div>
113
+
</div>
114
+
)}
115
+
116
+
{/* External Collections */}
117
+
{summary.collectionsSummary.filter(c => c.isExternal).length > 0 && (
118
+
<div>
119
+
<h4 className="text-xs font-medium text-zinc-600 dark:text-zinc-400 mb-2">
120
+
External Collections
121
+
</h4>
122
+
<div className="space-y-1 ml-4">
123
+
{summary.collectionsSummary
124
+
.filter(c => c.isExternal)
125
+
.map((collection) => (
126
+
<div
127
+
key={collection.collection}
128
+
className="flex items-center py-1"
129
+
>
130
+
<Text variant="secondary" size="sm">
131
+
{collection.collection}
132
+
</Text>
133
+
<div className="flex-1 border-b border-dotted border-zinc-300 dark:border-zinc-600 mx-2 self-end mb-1"></div>
134
+
<Text variant="muted" size="sm" className="italic">
135
+
External data
136
+
</Text>
137
+
</div>
138
+
))}
107
139
</div>
108
-
<span className="text-sm font-medium text-zinc-600 dark:text-zinc-400">
109
-
{collection.estimatedRepos.toLocaleString()} repos
110
-
</span>
111
140
</div>
112
-
))}
141
+
)}
113
142
</div>
114
143
</div>
115
144