slices.network
Highly ambitious ATProtocol AppView service and sdks
refactor handlers into api folder
chadtmiller.com
+12
api/src/api/mod.rs
+12
api/src/api/mod.rs
+146
api/src/api/sync.rs
+146
api/src/api/sync.rs
···
···
1
+
use crate::AppState;
2
+
use crate::auth;
3
+
use crate::jobs;
4
+
use crate::models::BulkSyncParams;
5
+
use axum::{
6
+
extract::State,
7
+
http::{HeaderMap, StatusCode},
8
+
response::Json,
9
+
};
10
+
use serde::{Deserialize, Serialize};
11
+
use tracing::{info, warn};
12
+
use uuid::Uuid;
13
+
14
+
#[derive(Debug, Deserialize)]
15
+
#[serde(rename_all = "camelCase")]
16
+
pub struct SyncRequest {
17
+
#[serde(flatten)]
18
+
pub params: BulkSyncParams,
19
+
pub slice: String,
20
+
}
21
+
22
+
#[derive(Debug, Serialize)]
23
+
#[serde(rename_all = "camelCase")]
24
+
pub struct SyncJobResponse {
25
+
pub success: bool,
26
+
pub job_id: Option<Uuid>,
27
+
pub message: String,
28
+
}
29
+
30
+
pub async fn sync(
31
+
State(state): State<AppState>,
32
+
headers: HeaderMap,
33
+
axum::extract::Json(request): axum::extract::Json<SyncRequest>,
34
+
) -> Result<Json<SyncJobResponse>, StatusCode> {
35
+
let token = auth::extract_bearer_token(&headers)?;
36
+
let user_info = auth::verify_oauth_token(&token, &state.config.auth_base_url).await?;
37
+
38
+
let user_did = user_info.sub;
39
+
let slice_uri = request.slice;
40
+
41
+
match jobs::enqueue_sync_job(&state.database_pool, user_did, slice_uri, request.params).await {
42
+
Ok(job_id) => Ok(Json(SyncJobResponse {
43
+
success: true,
44
+
job_id: Some(job_id),
45
+
message: format!("Sync job {} enqueued successfully", job_id),
46
+
})),
47
+
Err(e) => {
48
+
tracing::error!("Failed to enqueue sync job: {}", e);
49
+
Ok(Json(SyncJobResponse {
50
+
success: false,
51
+
job_id: None,
52
+
message: format!("Failed to enqueue sync job: {}", e),
53
+
}))
54
+
}
55
+
}
56
+
}
57
+
58
+
#[derive(Deserialize)]
59
+
#[serde(rename_all = "camelCase")]
60
+
pub struct SyncUserCollectionsRequest {
61
+
pub slice: String,
62
+
#[serde(default = "default_timeout")]
63
+
pub timeout_seconds: u64,
64
+
}
65
+
66
+
fn default_timeout() -> u64 {
67
+
30
68
+
}
69
+
70
+
pub async fn sync_user_collections(
71
+
State(state): State<AppState>,
72
+
headers: HeaderMap,
73
+
Json(request): Json<SyncUserCollectionsRequest>,
74
+
) -> Result<Json<crate::sync::SyncUserCollectionsResult>, (StatusCode, Json<serde_json::Value>)> {
75
+
let token = auth::extract_bearer_token(&headers).map_err(|e| {
76
+
(
77
+
StatusCode::UNAUTHORIZED,
78
+
Json(serde_json::json!({
79
+
"error": "AuthenticationRequired",
80
+
"message": format!("Bearer token required: {}", e)
81
+
})),
82
+
)
83
+
})?;
84
+
85
+
let user_info = auth::verify_oauth_token(&token, &state.config.auth_base_url)
86
+
.await
87
+
.map_err(|e| {
88
+
(
89
+
StatusCode::UNAUTHORIZED,
90
+
Json(serde_json::json!({
91
+
"error": "InvalidToken",
92
+
"message": format!("Token verification failed: {}", e)
93
+
})),
94
+
)
95
+
})?;
96
+
97
+
let user_did = user_info.did.unwrap_or(user_info.sub);
98
+
99
+
info!(
100
+
"🔄 Starting user collections sync for {} on slice {} (timeout: {}s)",
101
+
user_did, request.slice, request.timeout_seconds
102
+
);
103
+
104
+
if request.timeout_seconds > 300 {
105
+
return Err((
106
+
StatusCode::BAD_REQUEST,
107
+
Json(serde_json::json!({
108
+
"error": "InvalidTimeout",
109
+
"message": "Maximum timeout is 300 seconds (5 minutes)"
110
+
})),
111
+
));
112
+
}
113
+
114
+
let sync_service =
115
+
crate::sync::SyncService::new(state.database.clone(), state.config.relay_endpoint.clone());
116
+
117
+
match sync_service
118
+
.sync_user_collections(&user_did, &request.slice, request.timeout_seconds)
119
+
.await
120
+
{
121
+
Ok(result) => {
122
+
if result.timed_out {
123
+
info!(
124
+
"⏰ Sync timed out for user {}, suggesting async job",
125
+
user_did
126
+
);
127
+
} else {
128
+
info!(
129
+
"✅ Sync completed for user {}: {} repos, {} records",
130
+
user_did, result.repos_processed, result.records_synced
131
+
);
132
+
}
133
+
Ok(Json(result))
134
+
}
135
+
Err(e) => {
136
+
warn!("❌ Sync failed for user {}: {}", user_did, e);
137
+
Err((
138
+
StatusCode::INTERNAL_SERVER_ERROR,
139
+
Json(serde_json::json!({
140
+
"error": "SyncFailed",
141
+
"message": format!("Sync operation failed: {}", e)
142
+
})),
143
+
))
144
+
}
145
+
}
146
+
}
api/src/handler_get_actors.rs
api/src/api/actors.rs
api/src/handler_get_actors.rs
api/src/api/actors.rs
api/src/handler_get_records.rs
api/src/api/records.rs
api/src/handler_get_records.rs
api/src/api/records.rs
api/src/handler_jetstream_status.rs
api/src/api/jetstream.rs
api/src/handler_jetstream_status.rs
api/src/api/jetstream.rs
api/src/handler_jobs.rs
api/src/api/jobs.rs
api/src/handler_jobs.rs
api/src/api/jobs.rs
api/src/handler_logs.rs
api/src/api/logs.rs
api/src/handler_logs.rs
api/src/api/logs.rs
api/src/handler_oauth_clients.rs
api/src/api/oauth.rs
api/src/handler_oauth_clients.rs
api/src/api/oauth.rs
api/src/handler_openapi_spec.rs
api/src/api/openapi.rs
api/src/handler_openapi_spec.rs
api/src/api/openapi.rs
api/src/handler_sparkline.rs
api/src/api/sparkline.rs
api/src/handler_sparkline.rs
api/src/api/sparkline.rs
api/src/handler_stats.rs
api/src/api/stats.rs
api/src/handler_stats.rs
api/src/api/stats.rs
-58
api/src/handler_sync.rs
-58
api/src/handler_sync.rs
···
1
-
use crate::AppState;
2
-
use crate::auth;
3
-
use crate::jobs;
4
-
use crate::models::BulkSyncParams;
5
-
use axum::{
6
-
extract::State,
7
-
http::{HeaderMap, StatusCode},
8
-
response::Json,
9
-
};
10
-
use serde::{Deserialize, Serialize};
11
-
use uuid::Uuid;
12
-
13
-
#[derive(Debug, Deserialize)]
14
-
#[serde(rename_all = "camelCase")]
15
-
pub struct SyncRequest {
16
-
#[serde(flatten)]
17
-
pub params: BulkSyncParams,
18
-
pub slice: String, // The slice URI
19
-
}
20
-
21
-
#[derive(Debug, Serialize)]
22
-
#[serde(rename_all = "camelCase")]
23
-
pub struct SyncJobResponse {
24
-
pub success: bool,
25
-
pub job_id: Option<Uuid>,
26
-
pub message: String,
27
-
}
28
-
29
-
/// Start a sync job (enqueue it for background processing)
30
-
pub async fn sync(
31
-
State(state): State<AppState>,
32
-
headers: HeaderMap,
33
-
axum::extract::Json(request): axum::extract::Json<SyncRequest>,
34
-
) -> Result<Json<SyncJobResponse>, StatusCode> {
35
-
// Extract and verify authentication
36
-
let token = auth::extract_bearer_token(&headers)?;
37
-
let user_info = auth::verify_oauth_token(&token, &state.config.auth_base_url).await?;
38
-
39
-
let user_did = user_info.sub;
40
-
let slice_uri = request.slice;
41
-
42
-
// Enqueue the sync job with authenticated user information
43
-
match jobs::enqueue_sync_job(&state.database_pool, user_did, slice_uri, request.params).await {
44
-
Ok(job_id) => Ok(Json(SyncJobResponse {
45
-
success: true,
46
-
job_id: Some(job_id),
47
-
message: format!("Sync job {} enqueued successfully", job_id),
48
-
})),
49
-
Err(e) => {
50
-
tracing::error!("Failed to enqueue sync job: {}", e);
51
-
Ok(Json(SyncJobResponse {
52
-
success: false,
53
-
job_id: None,
54
-
message: format!("Failed to enqueue sync job: {}", e),
55
-
}))
56
-
}
57
-
}
58
-
}
···
-108
api/src/handler_sync_user_collections.rs
-108
api/src/handler_sync_user_collections.rs
···
1
-
use axum::{
2
-
extract::State,
3
-
http::{HeaderMap, StatusCode},
4
-
response::Json,
5
-
};
6
-
use serde::Deserialize;
7
-
use tracing::{info, warn};
8
-
9
-
use crate::AppState;
10
-
use crate::auth::{extract_bearer_token, verify_oauth_token};
11
-
use crate::sync::{SyncService, SyncUserCollectionsResult};
12
-
13
-
#[derive(Deserialize)]
14
-
#[serde(rename_all = "camelCase")]
15
-
pub struct SyncUserCollectionsRequest {
16
-
pub slice: String,
17
-
#[serde(default = "default_timeout")]
18
-
pub timeout_seconds: u64,
19
-
}
20
-
21
-
fn default_timeout() -> u64 {
22
-
30 // 30 second default timeout for login scenarios
23
-
}
24
-
25
-
/// Handler for network.slices.slice.syncUserCollections
26
-
/// Synchronously syncs external collections for the authenticated user with timeout protection
27
-
/// Automatically discovers external collections based on the slice's domain configuration
28
-
pub async fn sync_user_collections(
29
-
State(state): State<AppState>,
30
-
headers: HeaderMap,
31
-
Json(request): Json<SyncUserCollectionsRequest>,
32
-
) -> Result<Json<SyncUserCollectionsResult>, (StatusCode, Json<serde_json::Value>)> {
33
-
// Extract and verify OAuth token
34
-
let token = extract_bearer_token(&headers).map_err(|e| {
35
-
(
36
-
StatusCode::UNAUTHORIZED,
37
-
Json(serde_json::json!({
38
-
"error": "AuthenticationRequired",
39
-
"message": format!("Bearer token required: {}", e)
40
-
})),
41
-
)
42
-
})?;
43
-
44
-
let user_info = verify_oauth_token(&token, &state.config.auth_base_url)
45
-
.await
46
-
.map_err(|e| {
47
-
(
48
-
StatusCode::UNAUTHORIZED,
49
-
Json(serde_json::json!({
50
-
"error": "InvalidToken",
51
-
"message": format!("Token verification failed: {}", e)
52
-
})),
53
-
)
54
-
})?;
55
-
56
-
let user_did = user_info.did.unwrap_or(user_info.sub);
57
-
58
-
info!(
59
-
"🔄 Starting user collections sync for {} on slice {} (timeout: {}s)",
60
-
user_did, request.slice, request.timeout_seconds
61
-
);
62
-
63
-
// Validate timeout (max 5 minutes for sync operations)
64
-
if request.timeout_seconds > 300 {
65
-
return Err((
66
-
StatusCode::BAD_REQUEST,
67
-
Json(serde_json::json!({
68
-
"error": "InvalidTimeout",
69
-
"message": "Maximum timeout is 300 seconds (5 minutes)"
70
-
})),
71
-
));
72
-
}
73
-
74
-
// Create sync service
75
-
let sync_service =
76
-
SyncService::new(state.database.clone(), state.config.relay_endpoint.clone());
77
-
78
-
// Perform timeout-protected sync with auto-discovered external collections
79
-
match sync_service
80
-
.sync_user_collections(&user_did, &request.slice, request.timeout_seconds)
81
-
.await
82
-
{
83
-
Ok(result) => {
84
-
if result.timed_out {
85
-
info!(
86
-
"⏰ Sync timed out for user {}, suggesting async job",
87
-
user_did
88
-
);
89
-
} else {
90
-
info!(
91
-
"✅ Sync completed for user {}: {} repos, {} records",
92
-
user_did, result.repos_processed, result.records_synced
93
-
);
94
-
}
95
-
Ok(Json(result))
96
-
}
97
-
Err(e) => {
98
-
warn!("❌ Sync failed for user {}: {}", user_did, e);
99
-
Err((
100
-
StatusCode::INTERNAL_SERVER_ERROR,
101
-
Json(serde_json::json!({
102
-
"error": "SyncFailed",
103
-
"message": format!("Sync operation failed: {}", e)
104
-
})),
105
-
))
106
-
}
107
-
}
108
-
}
···
api/src/handler_upload_blob.rs
api/src/api/upload_blob.rs
api/src/handler_upload_blob.rs
api/src/api/upload_blob.rs
api/src/handler_xrpc_dynamic.rs
api/src/api/xrpc_dynamic.rs
api/src/handler_xrpc_dynamic.rs
api/src/api/xrpc_dynamic.rs
+21
-35
api/src/main.rs
+21
-35
api/src/main.rs
···
1
mod actor_resolver;
2
mod atproto_extensions;
3
mod auth;
4
mod database;
5
mod errors;
6
-
mod handler_get_actors;
7
-
mod handler_get_records;
8
-
mod handler_jetstream_status;
9
-
mod handler_jobs;
10
-
mod handler_logs;
11
-
mod handler_oauth_clients;
12
-
mod handler_openapi_spec;
13
-
mod handler_sparkline;
14
-
mod handler_stats;
15
-
mod handler_sync;
16
-
mod handler_sync_user_collections;
17
-
mod handler_upload_blob;
18
-
mod handler_xrpc_dynamic;
19
mod jetstream;
20
mod jobs;
21
mod logging;
···
281
"#
282
}),
283
)
284
-
// AT Protocol blob upload endpoint (must come before wildcard routes)
285
.route(
286
"/xrpc/com.atproto.repo.uploadBlob",
287
-
post(handler_upload_blob::upload_blob),
288
)
289
-
// XRPC endpoints
290
.route(
291
"/xrpc/network.slices.slice.startSync",
292
-
post(handler_sync::sync),
293
)
294
.route(
295
"/xrpc/network.slices.slice.syncUserCollections",
296
-
post(handler_sync_user_collections::sync_user_collections),
297
)
298
.route(
299
"/xrpc/network.slices.slice.getJobStatus",
300
-
get(handler_jobs::get_job_status),
301
)
302
.route(
303
"/xrpc/network.slices.slice.getJobHistory",
304
-
get(handler_jobs::get_slice_job_history),
305
)
306
.route(
307
"/xrpc/network.slices.slice.getJobLogs",
308
-
get(handler_logs::get_sync_job_logs_handler),
309
)
310
.route(
311
"/xrpc/network.slices.slice.getJetstreamLogs",
312
-
get(handler_logs::get_jetstream_logs_handler),
313
)
314
.route(
315
"/xrpc/network.slices.slice.stats",
316
-
post(handler_stats::stats),
317
)
318
.route(
319
"/xrpc/network.slices.slice.getSparklines",
320
-
post(handler_sparkline::batch_sparkline),
321
)
322
.route(
323
"/xrpc/network.slices.slice.getSliceRecords",
324
-
post(handler_get_records::get_records),
325
)
326
.route(
327
"/xrpc/network.slices.slice.openapi",
328
-
get(handler_openapi_spec::get_openapi_spec),
329
)
330
.route(
331
"/xrpc/network.slices.slice.getJetstreamStatus",
332
-
get(handler_jetstream_status::get_jetstream_status),
333
)
334
.route(
335
"/xrpc/network.slices.slice.getActors",
336
-
post(handler_get_actors::get_actors),
337
)
338
-
// OAuth client management endpoints
339
.route(
340
"/xrpc/network.slices.slice.createOAuthClient",
341
-
post(handler_oauth_clients::create_oauth_client),
342
)
343
.route(
344
"/xrpc/network.slices.slice.getOAuthClients",
345
-
get(handler_oauth_clients::get_oauth_clients),
346
)
347
.route(
348
"/xrpc/network.slices.slice.updateOAuthClient",
349
-
post(handler_oauth_clients::update_oauth_client),
350
)
351
.route(
352
"/xrpc/network.slices.slice.deleteOAuthClient",
353
-
post(handler_oauth_clients::delete_oauth_client),
354
)
355
// Dynamic collection-specific XRPC endpoints (wildcard routes must come last)
356
.route(
357
"/xrpc/*method",
358
-
get(handler_xrpc_dynamic::dynamic_xrpc_handler),
359
)
360
.route(
361
"/xrpc/*method",
362
-
post(handler_xrpc_dynamic::dynamic_xrpc_post_handler),
363
)
364
.layer(TraceLayer::new_for_http())
365
.layer(CorsLayer::permissive())
···
1
mod actor_resolver;
2
+
mod api;
3
mod atproto_extensions;
4
mod auth;
5
mod database;
6
mod errors;
7
mod jetstream;
8
mod jobs;
9
mod logging;
···
269
"#
270
}),
271
)
272
+
// XRPC endpoints
273
.route(
274
"/xrpc/com.atproto.repo.uploadBlob",
275
+
post(api::upload_blob::upload_blob),
276
)
277
.route(
278
"/xrpc/network.slices.slice.startSync",
279
+
post(api::sync::sync),
280
)
281
.route(
282
"/xrpc/network.slices.slice.syncUserCollections",
283
+
post(api::sync::sync_user_collections),
284
)
285
.route(
286
"/xrpc/network.slices.slice.getJobStatus",
287
+
get(api::jobs::get_job_status),
288
)
289
.route(
290
"/xrpc/network.slices.slice.getJobHistory",
291
+
get(api::jobs::get_slice_job_history),
292
)
293
.route(
294
"/xrpc/network.slices.slice.getJobLogs",
295
+
get(api::logs::get_sync_job_logs_handler),
296
)
297
.route(
298
"/xrpc/network.slices.slice.getJetstreamLogs",
299
+
get(api::logs::get_jetstream_logs_handler),
300
)
301
.route(
302
"/xrpc/network.slices.slice.stats",
303
+
post(api::stats::stats),
304
)
305
.route(
306
"/xrpc/network.slices.slice.getSparklines",
307
+
post(api::sparkline::batch_sparkline),
308
)
309
.route(
310
"/xrpc/network.slices.slice.getSliceRecords",
311
+
post(api::records::get_records),
312
)
313
.route(
314
"/xrpc/network.slices.slice.openapi",
315
+
get(api::openapi::get_openapi_spec),
316
)
317
.route(
318
"/xrpc/network.slices.slice.getJetstreamStatus",
319
+
get(api::jetstream::get_jetstream_status),
320
)
321
.route(
322
"/xrpc/network.slices.slice.getActors",
323
+
post(api::actors::get_actors),
324
)
325
.route(
326
"/xrpc/network.slices.slice.createOAuthClient",
327
+
post(api::oauth::create_oauth_client),
328
)
329
.route(
330
"/xrpc/network.slices.slice.getOAuthClients",
331
+
get(api::oauth::get_oauth_clients),
332
)
333
.route(
334
"/xrpc/network.slices.slice.updateOAuthClient",
335
+
post(api::oauth::update_oauth_client),
336
)
337
.route(
338
"/xrpc/network.slices.slice.deleteOAuthClient",
339
+
post(api::oauth::delete_oauth_client),
340
)
341
// Dynamic collection-specific XRPC endpoints (wildcard routes must come last)
342
.route(
343
"/xrpc/*method",
344
+
get(api::xrpc_dynamic::dynamic_xrpc_handler),
345
)
346
.route(
347
"/xrpc/*method",
348
+
post(api::xrpc_dynamic::dynamic_xrpc_post_handler),
349
)
350
.layer(TraceLayer::new_for_http())
351
.layer(CorsLayer::permissive())