A lexicon-driven AppView for ATProto.
happyview.dev
backfill
firehose
jetstream
atproto
appview
oauth
lexicon
fix: resolve clippy warnings and formatting issues
+328
-225
+22
-4
.github/workflows/test.yml
+22
-4
.github/workflows/test.yml
···
13
13
test:
14
14
runs-on: ubuntu-latest
15
15
16
+
services:
17
+
postgres:
18
+
image: postgres:16-alpine
19
+
env:
20
+
POSTGRES_USER: happyview
21
+
POSTGRES_PASSWORD: happyview
22
+
POSTGRES_DB: happyview_test
23
+
ports:
24
+
- 5432:5432
25
+
options: >-
26
+
--health-cmd pg_isready
27
+
--health-interval 10s
28
+
--health-timeout 5s
29
+
--health-retries 5
30
+
16
31
steps:
17
32
- name: Checkout repository
18
33
uses: actions/checkout@v4
···
31
46
- name: Build
32
47
run: cargo build --all-targets
33
48
34
-
- name: Run tests
35
-
run: cargo test
49
+
- name: Run unit tests
50
+
run: cargo test --lib
51
+
52
+
- name: Run e2e tests
53
+
env:
54
+
TEST_DATABASE_URL: postgres://happyview:happyview@localhost:5432/happyview_test
55
+
run: cargo test --tests
36
56
37
57
lint:
38
58
runs-on: ubuntu-latest
···
56
76
run: cargo fmt -- --check
57
77
58
78
- name: Clippy
59
-
env:
60
-
SQLX_OFFLINE: true
61
79
run: cargo clippy --all-targets -- -D warnings
+46
-27
src/admin.rs
+46
-27
src/admin.rs
···
6
6
use serde_json::Value;
7
7
use sha2::{Digest, Sha256};
8
8
9
+
use crate::AppState;
9
10
use crate::error::AppError;
10
11
use crate::lexicon::{LexiconType, ParsedLexicon};
11
-
use crate::AppState;
12
12
13
13
// ---------------------------------------------------------------------------
14
14
// Helpers
···
59
59
60
60
// Check admins table first
61
61
let key_hash = hash_api_key(token);
62
-
let found: Option<(String,)> = sqlx::query_as(
63
-
"SELECT id::text FROM admins WHERE api_key_hash = $1",
64
-
)
65
-
.bind(&key_hash)
66
-
.fetch_optional(&state.db)
67
-
.await
68
-
.map_err(|e| AppError::Internal(format!("admin auth query failed: {e}")))?;
62
+
let found: Option<(String,)> =
63
+
sqlx::query_as("SELECT id::text FROM admins WHERE api_key_hash = $1")
64
+
.bind(&key_hash)
65
+
.fetch_optional(&state.db)
66
+
.await
67
+
.map_err(|e| AppError::Internal(format!("admin auth query failed: {e}")))?;
69
68
70
69
if let Some((admin_id,)) = found {
71
70
// Update last_used_at in the background
72
71
let db = state.db.clone();
73
72
let admin_id = admin_id.clone();
74
73
tokio::spawn(async move {
75
-
let _ = sqlx::query(
76
-
"UPDATE admins SET last_used_at = NOW() WHERE id::text = $1",
77
-
)
78
-
.bind(&admin_id)
79
-
.execute(&db)
80
-
.await;
74
+
let _ = sqlx::query("UPDATE admins SET last_used_at = NOW() WHERE id::text = $1")
75
+
.bind(&admin_id)
76
+
.execute(&db)
77
+
.await;
81
78
});
82
79
return Ok(AdminAuth);
83
80
}
84
81
85
82
// Fall back to ADMIN_SECRET env var
86
-
if let Some(ref secret) = state.config.admin_secret {
87
-
if token == secret {
88
-
return Ok(AdminAuth);
89
-
}
83
+
if let Some(ref secret) = state.config.admin_secret
84
+
&& token == secret
85
+
{
86
+
return Ok(AdminAuth);
90
87
}
91
88
92
89
Err(AppError::Auth("invalid admin credentials".into()))
···
181
178
.lexicon_json
182
179
.get("lexicon")
183
180
.and_then(|v| v.as_i64())
184
-
.ok_or_else(|| AppError::BadRequest("lexicon JSON must have a numeric 'lexicon' field".into()))?;
181
+
.ok_or_else(|| {
182
+
AppError::BadRequest("lexicon JSON must have a numeric 'lexicon' field".into())
183
+
})?;
185
184
186
185
if lexicon_version != 1 {
187
186
return Err(AppError::BadRequest(format!(
···
254
253
State(state): State<AppState>,
255
254
_admin: AdminAuth,
256
255
) -> Result<Json<Vec<LexiconSummary>>, AppError> {
256
+
#[allow(clippy::type_complexity)]
257
257
let rows: Vec<(String, i32, Value, bool, chrono::DateTime<chrono::Utc>, chrono::DateTime<chrono::Utc>)> =
258
258
sqlx::query_as(
259
259
"SELECT id, revision, lexicon_json, backfill, created_at, updated_at FROM lexicons ORDER BY id",
···
289
289
_admin: AdminAuth,
290
290
Path(id): Path<String>,
291
291
) -> Result<Json<Value>, AppError> {
292
+
#[allow(clippy::type_complexity)]
292
293
let row: Option<(String, i32, Value, bool, chrono::DateTime<chrono::Utc>, chrono::DateTime<chrono::Utc>)> =
293
294
sqlx::query_as(
294
295
"SELECT id, revision, lexicon_json, backfill, created_at, updated_at FROM lexicons WHERE id = $1",
···
413
414
State(state): State<AppState>,
414
415
_admin: AdminAuth,
415
416
) -> Result<Json<Vec<BackfillJob>>, AppError> {
417
+
#[allow(clippy::type_complexity)]
416
418
let rows: Vec<(
417
419
String,
418
420
Option<String>,
···
435
437
let jobs: Vec<BackfillJob> = rows
436
438
.into_iter()
437
439
.map(
438
-
|(id, collection, did, status, total_repos, processed_repos, total_records, error, started_at, completed_at, created_at)| {
440
+
|(
441
+
id,
442
+
collection,
443
+
did,
444
+
status,
445
+
total_repos,
446
+
processed_repos,
447
+
total_records,
448
+
error,
449
+
started_at,
450
+
completed_at,
451
+
created_at,
452
+
)| {
439
453
BackfillJob {
440
454
id,
441
455
collection,
···
506
520
State(state): State<AppState>,
507
521
_admin: AdminAuth,
508
522
) -> Result<Json<Vec<AdminSummary>>, AppError> {
509
-
let rows: Vec<(String, String, chrono::DateTime<chrono::Utc>, Option<chrono::DateTime<chrono::Utc>>)> =
510
-
sqlx::query_as(
511
-
"SELECT id::text, name, created_at, last_used_at FROM admins ORDER BY created_at",
512
-
)
513
-
.fetch_all(&state.db)
514
-
.await
515
-
.map_err(|e| AppError::Internal(format!("failed to list admins: {e}")))?;
523
+
#[allow(clippy::type_complexity)]
524
+
let rows: Vec<(
525
+
String,
526
+
String,
527
+
chrono::DateTime<chrono::Utc>,
528
+
Option<chrono::DateTime<chrono::Utc>>,
529
+
)> = sqlx::query_as(
530
+
"SELECT id::text, name, created_at, last_used_at FROM admins ORDER BY created_at",
531
+
)
532
+
.fetch_all(&state.db)
533
+
.await
534
+
.map_err(|e| AppError::Internal(format!("failed to list admins: {e}")))?;
516
535
517
536
let admins: Vec<AdminSummary> = rows
518
537
.into_iter()
+1
-1
src/auth/middleware.rs
+1
-1
src/auth/middleware.rs
+16
-9
src/backfill.rs
+16
-9
src/backfill.rs
···
128
128
129
129
let page_count = body.records.len();
130
130
for entry in body.records {
131
-
let rkey = entry.uri.split('/').last().unwrap_or_default().to_string();
131
+
let rkey = entry
132
+
.uri
133
+
.split('/')
134
+
.next_back()
135
+
.unwrap_or_default()
136
+
.to_string();
132
137
records.push((entry.uri, rkey, entry.cid, entry.value));
133
138
}
134
139
···
154
159
job_id: &str,
155
160
) -> Result<(), String> {
156
161
// Fetch the job
157
-
let job: (Option<String>, Option<String>) = sqlx::query_as(
158
-
"SELECT collection, did FROM backfill_jobs WHERE id::text = $1",
159
-
)
160
-
.bind(job_id)
161
-
.fetch_one(db)
162
-
.await
163
-
.map_err(|e| format!("failed to fetch job: {e}"))?;
162
+
let job: (Option<String>, Option<String>) =
163
+
sqlx::query_as("SELECT collection, did FROM backfill_jobs WHERE id::text = $1")
164
+
.bind(job_id)
165
+
.fetch_one(db)
166
+
.await
167
+
.map_err(|e| format!("failed to fetch job: {e}"))?;
164
168
165
169
let (job_collection, job_did) = job;
166
170
···
279
283
.execute(db)
280
284
.await;
281
285
282
-
info!(job = job_id, processed_repos, total_records, "backfill completed");
286
+
info!(
287
+
job = job_id,
288
+
processed_repos, total_records, "backfill completed"
289
+
);
283
290
Ok(())
284
291
}
285
292
+15
-11
src/config.rs
+15
-11
src/config.rs
···
21
21
.ok()
22
22
.and_then(|p| p.parse().ok())
23
23
.unwrap_or(3000),
24
-
database_url: env::var("DATABASE_URL")
25
-
.expect("DATABASE_URL must be set"),
26
-
aip_url: env::var("AIP_URL")
27
-
.expect("AIP_URL must be set"),
24
+
database_url: env::var("DATABASE_URL").expect("DATABASE_URL must be set"),
25
+
aip_url: env::var("AIP_URL").expect("AIP_URL must be set"),
28
26
jetstream_url: env::var("JETSTREAM_URL")
29
27
.unwrap_or_else(|_| "wss://jetstream2.us-west.bsky.network/subscribe".into()),
30
28
admin_secret: env::var("ADMIN_SECRET").ok(),
31
-
relay_url: env::var("RELAY_URL")
32
-
.unwrap_or_else(|_| "https://bsky.network".into()),
33
-
plc_url: env::var("PLC_URL")
34
-
.unwrap_or_else(|_| "https://plc.directory".into()),
29
+
relay_url: env::var("RELAY_URL").unwrap_or_else(|_| "https://bsky.network".into()),
30
+
plc_url: env::var("PLC_URL").unwrap_or_else(|_| "https://plc.directory".into()),
35
31
}
36
32
}
37
33
···
49
45
50
46
unsafe fn clear_env() {
51
47
for key in [
52
-
"HOST", "PORT", "DATABASE_URL", "AIP_URL", "JETSTREAM_URL",
53
-
"ADMIN_SECRET", "RELAY_URL", "PLC_URL",
48
+
"HOST",
49
+
"PORT",
50
+
"DATABASE_URL",
51
+
"AIP_URL",
52
+
"JETSTREAM_URL",
53
+
"ADMIN_SECRET",
54
+
"RELAY_URL",
55
+
"PLC_URL",
54
56
] {
55
-
unsafe { env::remove_var(key); }
57
+
unsafe {
58
+
env::remove_var(key);
59
+
}
56
60
}
57
61
}
58
62
+2
-8
src/error.rs
+2
-8
src/error.rs
···
107
107
108
108
#[test]
109
109
fn display_formats() {
110
-
assert_eq!(
111
-
AppError::Auth("x".into()).to_string(),
112
-
"auth error: x"
113
-
);
110
+
assert_eq!(AppError::Auth("x".into()).to_string(), "auth error: x");
114
111
assert_eq!(
115
112
AppError::BadRequest("y".into()).to_string(),
116
113
"bad request: y"
···
119
116
AppError::Internal("z".into()).to_string(),
120
117
"internal error: z"
121
118
);
122
-
assert_eq!(
123
-
AppError::NotFound("w".into()).to_string(),
124
-
"not found: w"
125
-
);
119
+
assert_eq!(AppError::NotFound("w".into()).to_string(), "not found: w");
126
120
assert_eq!(
127
121
AppError::PdsError(StatusCode::BAD_GATEWAY, Bytes::new()).to_string(),
128
122
"PDS error: 502 Bad Gateway"
+10
-2
src/jetstream.rs
+10
-2
src/jetstream.rs
···
2
2
use serde::Deserialize;
3
3
use serde_json::Value;
4
4
use sqlx::PgPool;
5
-
use std::sync::atomic::{AtomicI64, Ordering};
6
5
use std::sync::Arc;
6
+
use std::sync::atomic::{AtomicI64, Ordering};
7
7
use tokio::sync::watch;
8
8
use tokio_tungstenite::tungstenite::Message;
9
9
···
57
57
58
58
// Connect and process events. If the collection list changes
59
59
// mid-stream, `run` returns so we can reconnect with new filters.
60
-
match run(&db, &jetstream_url, &cursor, &collections, &mut collections_rx).await {
60
+
match run(
61
+
&db,
62
+
&jetstream_url,
63
+
&cursor,
64
+
&collections,
65
+
&mut collections_rx,
66
+
)
67
+
.await
68
+
{
61
69
Ok(()) => {
62
70
tracing::info!("jetstream reconnecting due to collection change");
63
71
}
+31
-10
src/lexicon.rs
+31
-10
src/lexicon.rs
···
43
43
44
44
impl ParsedLexicon {
45
45
/// Parse a lexicon JSON document into a `ParsedLexicon`.
46
-
pub fn parse(raw: Value, revision: i32, target_collection: Option<String>) -> Result<Self, String> {
46
+
pub fn parse(
47
+
raw: Value,
48
+
revision: i32,
49
+
target_collection: Option<String>,
50
+
) -> Result<Self, String> {
47
51
let id = raw
48
52
.get("id")
49
53
.and_then(|v| v.as_str())
···
92
96
#[derive(Debug, Clone)]
93
97
pub struct LexiconRegistry {
94
98
inner: Arc<RwLock<HashMap<String, ParsedLexicon>>>,
99
+
}
100
+
101
+
impl Default for LexiconRegistry {
102
+
fn default() -> Self {
103
+
Self::new()
104
+
}
95
105
}
96
106
97
107
impl LexiconRegistry {
···
277
287
278
288
#[test]
279
289
fn parse_query_lexicon() {
280
-
let parsed = ParsedLexicon::parse(query_lexicon_json(), 2, Some("games.gamesgamesgamesgames.game".into())).unwrap();
290
+
let parsed = ParsedLexicon::parse(
291
+
query_lexicon_json(),
292
+
2,
293
+
Some("games.gamesgamesgamesgames.game".into()),
294
+
)
295
+
.unwrap();
281
296
assert_eq!(parsed.lexicon_type, LexiconType::Query);
282
297
assert!(parsed.parameters.is_some());
283
298
assert!(parsed.output.is_some());
284
-
assert_eq!(parsed.target_collection, Some("games.gamesgamesgamesgames.game".into()));
299
+
assert_eq!(
300
+
parsed.target_collection,
301
+
Some("games.gamesgamesgamesgames.game".into())
302
+
);
285
303
assert_eq!(parsed.revision, 2);
286
304
}
287
305
···
316
334
317
335
#[test]
318
336
fn parse_target_collection_passthrough() {
319
-
let parsed = ParsedLexicon::parse(
320
-
query_lexicon_json(),
321
-
1,
322
-
Some("custom.collection".into()),
323
-
)
324
-
.unwrap();
337
+
let parsed =
338
+
ParsedLexicon::parse(query_lexicon_json(), 1, Some("custom.collection".into()))
339
+
.unwrap();
325
340
assert_eq!(parsed.target_collection, Some("custom.collection".into()));
326
341
}
327
342
···
356
371
reg.upsert(v2).await;
357
372
358
373
assert_eq!(reg.count().await, 1);
359
-
assert_eq!(reg.get("games.gamesgamesgamesgames.game").await.unwrap().revision, 5);
374
+
assert_eq!(
375
+
reg.get("games.gamesgamesgamesgames.game")
376
+
.await
377
+
.unwrap()
378
+
.revision,
379
+
5
380
+
);
360
381
}
361
382
362
383
#[tokio::test]
+12
-3
src/main.rs
+12
-3
src/main.rs
···
1
1
use happyview::config::Config;
2
2
use happyview::lexicon::LexiconRegistry;
3
-
use happyview::{admin, backfill, jetstream, server, AppState};
3
+
use happyview::{AppState, admin, backfill, jetstream, server};
4
4
use tokio::sync::watch;
5
5
use tracing::info;
6
6
···
48
48
collections_tx,
49
49
};
50
50
51
-
jetstream::spawn(state.db.clone(), config.jetstream_url.clone(), collections_rx);
52
-
backfill::spawn_worker(state.db.clone(), state.http.clone(), config.relay_url.clone(), config.plc_url.clone());
51
+
jetstream::spawn(
52
+
state.db.clone(),
53
+
config.jetstream_url.clone(),
54
+
collections_rx,
55
+
);
56
+
backfill::spawn_worker(
57
+
state.db.clone(),
58
+
state.http.clone(),
59
+
config.relay_url.clone(),
60
+
config.plc_url.clone(),
61
+
);
53
62
54
63
let app = server::router(state);
55
64
let addr = config.listen_addr();
+18
-7
src/profile.rs
+18
-7
src/profile.rs
···
36
36
}
37
37
38
38
/// Resolve a full profile for the given DID: DID document -> handle + PDS -> profile record.
39
-
pub async fn resolve_profile(http: &reqwest::Client, plc_url: &str, did: &str) -> Result<Profile, AppError> {
39
+
pub async fn resolve_profile(
40
+
http: &reqwest::Client,
41
+
plc_url: &str,
42
+
did: &str,
43
+
) -> Result<Profile, AppError> {
40
44
let did_doc = resolve_did_document(http, plc_url, did).await?;
41
45
42
46
let handle = did_doc
···
52
56
.map(|s| s.service_endpoint.clone())
53
57
.ok_or_else(|| AppError::NotFound("no PDS endpoint in DID document".into()))?;
54
58
55
-
let (display_name, description, avatar_url) =
56
-
fetch_profile_from_pds(http, &pds_endpoint, did)
57
-
.await
58
-
.unwrap_or((None, None, None));
59
+
let (display_name, description, avatar_url) = fetch_profile_from_pds(http, &pds_endpoint, did)
60
+
.await
61
+
.unwrap_or((None, None, None));
59
62
60
63
Ok(Profile {
61
64
did: did.to_string(),
···
67
70
}
68
71
69
72
/// Resolve the PDS endpoint for a DID by fetching its DID document.
70
-
pub async fn resolve_pds_endpoint(http: &reqwest::Client, plc_url: &str, did: &str) -> Result<String, AppError> {
73
+
pub async fn resolve_pds_endpoint(
74
+
http: &reqwest::Client,
75
+
plc_url: &str,
76
+
did: &str,
77
+
) -> Result<String, AppError> {
71
78
let did_doc = resolve_did_document(http, plc_url, did).await?;
72
79
73
80
did_doc
···
80
87
81
88
/// Fetch a DID document from the PLC directory.
82
89
// TODO: handle did:web:* resolution (fetch https://{domain}/.well-known/did.json)
83
-
async fn resolve_did_document(http: &reqwest::Client, plc_url: &str, did: &str) -> Result<DidDocument, AppError> {
90
+
async fn resolve_did_document(
91
+
http: &reqwest::Client,
92
+
plc_url: &str,
93
+
did: &str,
94
+
) -> Result<DidDocument, AppError> {
84
95
let url = format!("{}/{did}", plc_url.trim_end_matches('/'));
85
96
86
97
let resp = http
+95
-73
src/repo.rs
+95
-73
src/repo.rs
···
5
5
use base64::Engine;
6
6
use p256::pkcs8::EncodePrivateKey;
7
7
use serde::Deserialize;
8
-
use serde_json::{json, Value};
8
+
use serde_json::{Value, json};
9
9
use sha2::{Digest, Sha256};
10
10
11
+
use crate::AppState;
11
12
use crate::auth::Claims;
12
13
use crate::error::AppError;
13
-
use crate::AppState;
14
14
15
15
// ---------------------------------------------------------------------------
16
16
// AT URI parsing
···
110
110
.to_pkcs8_der()
111
111
.map_err(|e| AppError::Internal(format!("PKCS#8 conversion failed: {e}")))?;
112
112
113
-
let encoding_key =
114
-
jsonwebtoken::EncodingKey::from_ec_der(pkcs8_der.as_bytes());
113
+
let encoding_key = jsonwebtoken::EncodingKey::from_ec_der(pkcs8_der.as_bytes());
115
114
116
115
// Public JWK for the header (no private component)
117
116
let public_jwk = jsonwebtoken::jwk::Jwk {
···
212
211
.map_err(|e| AppError::Internal(format!("PDS request failed: {e}")))?;
213
212
214
213
// Retry with nonce if PDS requires it
215
-
if resp.status() == reqwest::StatusCode::UNAUTHORIZED {
216
-
if let Some(nonce) = resp
214
+
if resp.status() == reqwest::StatusCode::UNAUTHORIZED
215
+
&& let Some(nonce) = resp
217
216
.headers()
218
217
.get("dpop-nonce")
219
218
.and_then(|v| v.to_str().ok())
220
-
{
221
-
let nonce = nonce.to_string();
222
-
tracing::debug!("retrying with DPoP nonce");
219
+
{
220
+
let nonce = nonce.to_string();
221
+
tracing::debug!("retrying with DPoP nonce");
223
222
224
-
let dpop = generate_dpop_proof(
225
-
"POST",
226
-
&url,
227
-
&session.dpop_jwk,
228
-
&session.access_token,
229
-
Some(&nonce),
230
-
)?;
223
+
let dpop = generate_dpop_proof(
224
+
"POST",
225
+
&url,
226
+
&session.dpop_jwk,
227
+
&session.access_token,
228
+
Some(&nonce),
229
+
)?;
231
230
232
-
let resp = state
233
-
.http
234
-
.post(&url)
235
-
.header("authorization", format!("DPoP {}", session.access_token))
236
-
.header("dpop", &dpop)
237
-
.json(body)
238
-
.send()
239
-
.await
240
-
.map_err(|e| AppError::Internal(format!("PDS request retry failed: {e}")))?;
231
+
let resp = state
232
+
.http
233
+
.post(&url)
234
+
.header("authorization", format!("DPoP {}", session.access_token))
235
+
.header("dpop", &dpop)
236
+
.json(body)
237
+
.send()
238
+
.await
239
+
.map_err(|e| AppError::Internal(format!("PDS request retry failed: {e}")))?;
241
240
242
-
return Ok(resp);
243
-
}
241
+
return Ok(resp);
244
242
}
245
243
246
244
Ok(resp)
···
271
269
.await
272
270
.map_err(|e| AppError::Internal(format!("PDS uploadBlob failed: {e}")))?;
273
271
274
-
if resp.status() == reqwest::StatusCode::UNAUTHORIZED {
275
-
if let Some(nonce) = resp
272
+
if resp.status() == reqwest::StatusCode::UNAUTHORIZED
273
+
&& let Some(nonce) = resp
276
274
.headers()
277
275
.get("dpop-nonce")
278
276
.and_then(|v| v.to_str().ok())
279
-
{
280
-
let nonce = nonce.to_string();
281
-
tracing::debug!("retrying uploadBlob with DPoP nonce");
277
+
{
278
+
let nonce = nonce.to_string();
279
+
tracing::debug!("retrying uploadBlob with DPoP nonce");
282
280
283
-
let dpop = generate_dpop_proof(
284
-
"POST",
285
-
&url,
286
-
&session.dpop_jwk,
287
-
&session.access_token,
288
-
Some(&nonce),
289
-
)?;
281
+
let dpop = generate_dpop_proof(
282
+
"POST",
283
+
&url,
284
+
&session.dpop_jwk,
285
+
&session.access_token,
286
+
Some(&nonce),
287
+
)?;
290
288
291
-
let resp = state
292
-
.http
293
-
.post(&url)
294
-
.header("authorization", format!("DPoP {}", session.access_token))
295
-
.header("dpop", &dpop)
296
-
.header("content-type", content_type)
297
-
.body(blob)
298
-
.send()
299
-
.await
300
-
.map_err(|e| AppError::Internal(format!("PDS uploadBlob retry failed: {e}")))?;
289
+
let resp = state
290
+
.http
291
+
.post(&url)
292
+
.header("authorization", format!("DPoP {}", session.access_token))
293
+
.header("dpop", &dpop)
294
+
.header("content-type", content_type)
295
+
.body(blob)
296
+
.send()
297
+
.await
298
+
.map_err(|e| AppError::Internal(format!("PDS uploadBlob retry failed: {e}")))?;
301
299
302
-
return forward_pds_response(resp).await;
303
-
}
300
+
return forward_pds_response(resp).await;
304
301
}
305
302
306
303
forward_pds_response(resp).await
···
344
341
.and_then(|l| l.as_str())
345
342
.map(|s| s.to_string());
346
343
347
-
if let Some(cid) = cid {
348
-
if let Some(blob) = item.get_mut("blob") {
349
-
if let Some(obj) = blob.as_object_mut() {
350
-
obj.insert(
351
-
"url".to_string(),
352
-
json!(format!(
353
-
"{pds_base}/xrpc/com.atproto.sync.getBlob?did={did}&cid={cid}"
354
-
)),
355
-
);
356
-
}
357
-
}
344
+
if let Some(cid) = cid
345
+
&& let Some(blob) = item.get_mut("blob")
346
+
&& let Some(obj) = blob.as_object_mut()
347
+
{
348
+
obj.insert(
349
+
"url".to_string(),
350
+
json!(format!(
351
+
"{pds_base}/xrpc/com.atproto.sync.getBlob?did={did}&cid={cid}"
352
+
)),
353
+
);
358
354
}
359
355
}
360
356
}
···
480
476
let point = public.to_encoded_point(false);
481
477
482
478
DpopJwk {
483
-
x: base64::engine::general_purpose::URL_SAFE_NO_PAD
484
-
.encode(point.x().unwrap()),
485
-
y: base64::engine::general_purpose::URL_SAFE_NO_PAD
486
-
.encode(point.y().unwrap()),
487
-
d: base64::engine::general_purpose::URL_SAFE_NO_PAD
488
-
.encode(secret.to_bytes()),
479
+
x: base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(point.x().unwrap()),
480
+
y: base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(point.y().unwrap()),
481
+
d: base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(secret.to_bytes()),
489
482
}
490
483
}
491
484
492
485
#[test]
493
486
fn dpop_proof_produces_valid_jwt_structure() {
494
487
let jwk = test_dpop_jwk();
495
-
let token = generate_dpop_proof("POST", "https://pds.example.com/xrpc/test", &jwk, "access-tok", None).unwrap();
488
+
let token = generate_dpop_proof(
489
+
"POST",
490
+
"https://pds.example.com/xrpc/test",
491
+
&jwk,
492
+
"access-tok",
493
+
None,
494
+
)
495
+
.unwrap();
496
496
497
497
let parts: Vec<&str> = token.split('.').collect();
498
498
assert_eq!(parts.len(), 3, "JWT should have 3 parts");
···
501
501
#[test]
502
502
fn dpop_proof_header_has_correct_fields() {
503
503
let jwk = test_dpop_jwk();
504
-
let token = generate_dpop_proof("POST", "https://pds.example.com/xrpc/test", &jwk, "access-tok", None).unwrap();
504
+
let token = generate_dpop_proof(
505
+
"POST",
506
+
"https://pds.example.com/xrpc/test",
507
+
&jwk,
508
+
"access-tok",
509
+
None,
510
+
)
511
+
.unwrap();
505
512
506
513
let header_b64 = token.split('.').next().unwrap();
507
514
let header_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
···
517
524
#[test]
518
525
fn dpop_proof_claims_have_correct_fields() {
519
526
let jwk = test_dpop_jwk();
520
-
let token = generate_dpop_proof("GET", "https://pds.example.com/xrpc/test", &jwk, "my-access-token", None).unwrap();
527
+
let token = generate_dpop_proof(
528
+
"GET",
529
+
"https://pds.example.com/xrpc/test",
530
+
&jwk,
531
+
"my-access-token",
532
+
None,
533
+
)
534
+
.unwrap();
521
535
522
536
let payload_b64 = token.split('.').nth(1).unwrap();
523
537
let payload_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
···
537
551
#[test]
538
552
fn dpop_proof_includes_nonce_when_provided() {
539
553
let jwk = test_dpop_jwk();
540
-
let token = generate_dpop_proof("POST", "https://pds.example.com/xrpc/test", &jwk, "tok", Some("abc123")).unwrap();
554
+
let token = generate_dpop_proof(
555
+
"POST",
556
+
"https://pds.example.com/xrpc/test",
557
+
&jwk,
558
+
"tok",
559
+
Some("abc123"),
560
+
)
561
+
.unwrap();
541
562
542
563
let payload_b64 = token.split('.').nth(1).unwrap();
543
564
let payload_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
···
552
573
fn dpop_proof_ath_is_sha256_of_access_token() {
553
574
let jwk = test_dpop_jwk();
554
575
let access_token = "test-access-token";
555
-
let token = generate_dpop_proof("POST", "https://example.com", &jwk, access_token, None).unwrap();
576
+
let token =
577
+
generate_dpop_proof("POST", "https://example.com", &jwk, access_token, None).unwrap();
556
578
557
579
let payload_b64 = token.split('.').nth(1).unwrap();
558
580
let payload_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
+3
-2
src/server.rs
+3
-2
src/server.rs
···
4
4
use tower_http::cors::CorsLayer;
5
5
use tower_http::trace::TraceLayer;
6
6
7
+
use crate::AppState;
7
8
use crate::admin;
8
9
use crate::auth::Claims;
9
10
use crate::error::AppError;
10
11
use crate::profile;
11
12
use crate::repo;
12
13
use crate::xrpc;
13
-
use crate::AppState;
14
14
15
15
pub fn router(state: AppState) -> Router {
16
16
Router::new()
···
36
36
State(state): State<AppState>,
37
37
claims: Claims,
38
38
) -> Result<Json<profile::Profile>, AppError> {
39
-
let profile = profile::resolve_profile(&state.http, &state.config.plc_url, claims.did()).await?;
39
+
let profile =
40
+
profile::resolve_profile(&state.http, &state.config.plc_url, claims.did()).await?;
40
41
Ok(Json(profile))
41
42
}
+26
-32
src/xrpc.rs
+26
-32
src/xrpc.rs
···
1
+
use axum::Json;
1
2
use axum::extract::{Path, Query, State};
2
3
use axum::http::StatusCode;
3
4
use axum::response::{IntoResponse, Response};
4
-
use axum::Json;
5
-
use serde_json::{json, Value};
5
+
use serde_json::{Value, json};
6
6
use std::collections::{HashMap, HashSet};
7
7
8
+
use crate::AppState;
8
9
use crate::auth::Claims;
9
10
use crate::error::AppError;
10
11
use crate::lexicon::LexiconType;
11
12
use crate::profile;
12
13
use crate::repo;
13
-
use crate::AppState;
14
14
15
15
// ---------------------------------------------------------------------------
16
16
// Catch-all handler
···
75
75
}
76
76
77
77
// List query: needs a target collection to know what to query
78
-
let collection = lexicon
79
-
.target_collection
80
-
.as_deref()
81
-
.ok_or_else(|| {
82
-
AppError::BadRequest(format!(
83
-
"{method} has no target_collection configured for list queries"
84
-
))
85
-
})?;
78
+
let collection = lexicon.target_collection.as_deref().ok_or_else(|| {
79
+
AppError::BadRequest(format!(
80
+
"{method} has no target_collection configured for list queries"
81
+
))
82
+
})?;
86
83
87
84
let limit: i64 = params
88
85
.get("limit")
···
126
123
let unique_dids: HashSet<&str> = rows.iter().map(|(_, did, _)| did.as_str()).collect();
127
124
let mut pds_map: HashMap<String, String> = HashMap::new();
128
125
for did in unique_dids {
129
-
if let Ok(pds) = profile::resolve_pds_endpoint(&state.http, &state.config.plc_url, did).await {
126
+
if let Ok(pds) =
127
+
profile::resolve_pds_endpoint(&state.http, &state.config.plc_url, did).await
128
+
{
130
129
pds_map.insert(did.to_string(), pds);
131
130
}
132
131
}
···
159
158
async fn handle_get_record(state: &AppState, uri: &str) -> Result<Response, AppError> {
160
159
let did = repo::parse_did_from_at_uri(uri)?;
161
160
162
-
let row: Option<(Value,)> =
163
-
sqlx::query_as("SELECT record FROM records WHERE uri = $1")
164
-
.bind(uri)
165
-
.fetch_optional(&state.db)
166
-
.await
167
-
.map_err(|e| AppError::Internal(format!("DB query failed: {e}")))?;
161
+
let row: Option<(Value,)> = sqlx::query_as("SELECT record FROM records WHERE uri = $1")
162
+
.bind(uri)
163
+
.fetch_optional(&state.db)
164
+
.await
165
+
.map_err(|e| AppError::Internal(format!("DB query failed: {e}")))?;
168
166
169
-
let (mut record,) =
170
-
row.ok_or_else(|| AppError::NotFound("record not found".into()))?;
167
+
let (mut record,) = row.ok_or_else(|| AppError::NotFound("record not found".into()))?;
171
168
172
169
let pds = profile::resolve_pds_endpoint(&state.http, &state.config.plc_url, &did).await?;
173
170
repo::enrich_media_blobs(&mut record, &pds, &did);
···
190
187
input: &Value,
191
188
lexicon: &crate::lexicon::ParsedLexicon,
192
189
) -> Result<Response, AppError> {
193
-
let collection = lexicon
194
-
.target_collection
195
-
.as_deref()
196
-
.ok_or_else(|| {
197
-
AppError::BadRequest(format!(
198
-
"{method} has no target_collection configured"
199
-
))
200
-
})?;
190
+
let collection = lexicon.target_collection.as_deref().ok_or_else(|| {
191
+
AppError::BadRequest(format!("{method} has no target_collection configured"))
192
+
})?;
201
193
202
194
let session = repo::get_atp_session(state, claims.token()).await?;
203
195
···
232
224
"record": record,
233
225
});
234
226
235
-
let resp = repo::pds_post_json_raw(state, session, "com.atproto.repo.createRecord", &pds_body).await?;
227
+
let resp =
228
+
repo::pds_post_json_raw(state, session, "com.atproto.repo.createRecord", &pds_body).await?;
236
229
237
230
if resp.status().is_success() {
238
231
let bytes = resp
···
247
240
pds_result.get("uri").and_then(|v| v.as_str()),
248
241
pds_result.get("cid").and_then(|v| v.as_str()),
249
242
) {
250
-
let rkey = uri.split('/').last().unwrap_or_default();
243
+
let rkey = uri.split('/').next_back().unwrap_or_default();
251
244
let _ = sqlx::query(
252
245
r#"
253
246
INSERT INTO records (uri, did, collection, rkey, record, cid)
···
292
285
293
286
let rkey = uri
294
287
.split('/')
295
-
.last()
288
+
.next_back()
296
289
.ok_or_else(|| AppError::Internal("invalid AT URI".into()))?;
297
290
298
291
// Build record from input, adding $type
···
310
303
"record": record,
311
304
});
312
305
313
-
let resp = repo::pds_post_json_raw(state, session, "com.atproto.repo.putRecord", &pds_body).await?;
306
+
let resp =
307
+
repo::pds_post_json_raw(state, session, "com.atproto.repo.putRecord", &pds_body).await?;
314
308
315
309
if resp.status().is_success() {
316
310
let bytes = resp
+1
-1
tests/common/app.rs
+1
-1
tests/common/app.rs
+1
-3
tests/common/auth.rs
+1
-3
tests/common/auth.rs
···
17
17
pub async fn mock_aip_userinfo(mock_server: &MockServer, did: &str) {
18
18
Mock::given(method("GET"))
19
19
.and(path("/oauth/userinfo"))
20
-
.respond_with(
21
-
ResponseTemplate::new(200).set_body_json(fixtures::userinfo_response(did)),
22
-
)
20
+
.respond_with(ResponseTemplate::new(200).set_body_json(fixtures::userinfo_response(did)))
23
21
.mount(mock_server)
24
22
.await;
25
23
}
+2
-2
tests/common/db.rs
+2
-2
tests/common/db.rs
···
2
2
3
3
/// Connect to the test database using `TEST_DATABASE_URL`.
4
4
pub async fn test_pool() -> PgPool {
5
-
let url = std::env::var("TEST_DATABASE_URL")
6
-
.expect("TEST_DATABASE_URL must be set for e2e tests");
5
+
let url =
6
+
std::env::var("TEST_DATABASE_URL").expect("TEST_DATABASE_URL must be set for e2e tests");
7
7
8
8
let pool = PgPool::connect(&url)
9
9
.await
+1
-1
tests/common/fixtures.rs
+1
-1
tests/common/fixtures.rs
+2
-6
tests/e2e_admin.rs
+2
-6
tests/e2e_admin.rs
···
3
3
use axum::body::Body;
4
4
use axum::http::{Request, StatusCode};
5
5
use http_body_util::BodyExt;
6
-
use serde_json::{json, Value};
6
+
use serde_json::{Value, json};
7
7
use serial_test::serial;
8
8
use tower::ServiceExt;
9
9
···
391
391
// Create a job first
392
392
app.router
393
393
.clone()
394
-
.oneshot(admin_post(
395
-
"/admin/backfill",
396
-
&app.admin_secret,
397
-
&json!({}),
398
-
))
394
+
.oneshot(admin_post("/admin/backfill", &app.admin_secret, &json!({})))
399
395
.await
400
396
.unwrap();
401
397
+24
-23
tests/e2e_xrpc.rs
+24
-23
tests/e2e_xrpc.rs
···
3
3
use axum::body::Body;
4
4
use axum::http::{Request, StatusCode};
5
5
use http_body_util::BodyExt;
6
-
use serde_json::{json, Value};
6
+
use serde_json::{Value, json};
7
7
use serial_test::serial;
8
8
use tower::ServiceExt;
9
9
use wiremock::matchers::{method, path, path_regex};
···
88
88
89
89
/// Seed a record directly into the database.
90
90
async fn seed_record(app: &TestApp, uri: &str, did: &str, collection: &str, record: &Value) {
91
-
let rkey = uri.split('/').last().unwrap_or("1");
91
+
let rkey = uri.split('/').next_back().unwrap_or("1");
92
92
sqlx::query(
93
93
"INSERT INTO records (uri, did, collection, rkey, record, cid) VALUES ($1, $2, $3, $4, $5, $6)",
94
94
)
···
138
138
// Mock PLC directory
139
139
Mock::given(method("GET"))
140
140
.and(path(format!("/{did}")))
141
-
.respond_with(ResponseTemplate::new(200).set_body_json(
142
-
fixtures::did_document(did, &app.mock_server.uri()),
143
-
))
141
+
.respond_with(
142
+
ResponseTemplate::new(200)
143
+
.set_body_json(fixtures::did_document(did, &app.mock_server.uri())),
144
+
)
144
145
.mount(&app.mock_server)
145
146
.await;
146
147
147
148
// Mock PDS getRecord for profile
148
149
Mock::given(method("GET"))
149
150
.and(path("/xrpc/com.atproto.repo.getRecord"))
150
-
.respond_with(
151
-
ResponseTemplate::new(200).set_body_json(fixtures::profile_record()),
152
-
)
151
+
.respond_with(ResponseTemplate::new(200).set_body_json(fixtures::profile_record()))
153
152
.mount(&app.mock_server)
154
153
.await;
155
154
156
155
let resp = app
157
156
.router
158
-
.oneshot(authed_get(
159
-
"/xrpc/app.bsky.actor.getProfile",
160
-
"valid-token",
161
-
))
157
+
.oneshot(authed_get("/xrpc/app.bsky.actor.getProfile", "valid-token"))
162
158
.await
163
159
.unwrap();
164
160
···
226
222
// Mock PLC for PDS resolution
227
223
Mock::given(method("GET"))
228
224
.and(path(format!("/{did}")))
229
-
.respond_with(ResponseTemplate::new(200).set_body_json(
230
-
fixtures::did_document(did, "https://pds.example.com"),
231
-
))
225
+
.respond_with(
226
+
ResponseTemplate::new(200)
227
+
.set_body_json(fixtures::did_document(did, "https://pds.example.com")),
228
+
)
232
229
.mount(&app.mock_server)
233
230
.await;
234
231
···
236
233
.router
237
234
.oneshot(
238
235
Request::builder()
239
-
.uri(&format!(
236
+
.uri(format!(
240
237
"/xrpc/games.gamesgamesgamesgames.listGames?uri={}",
241
238
urlencoding::encode(uri)
242
239
))
···
283
280
// Mock PLC for enrichment
284
281
Mock::given(method("GET"))
285
282
.and(path_regex("/did:plc:.*"))
286
-
.respond_with(ResponseTemplate::new(200).set_body_json(
287
-
fixtures::did_document(did, "https://pds.example.com"),
288
-
))
283
+
.respond_with(
284
+
ResponseTemplate::new(200)
285
+
.set_body_json(fixtures::did_document(did, "https://pds.example.com")),
286
+
)
289
287
.mount(&app.mock_server)
290
288
.await;
291
289
···
326
324
.router
327
325
.oneshot(
328
326
Request::builder()
329
-
.uri(&format!(
327
+
.uri(format!(
330
328
"/xrpc/games.gamesgamesgamesgames.listGames?limit=2&cursor={cursor}"
331
329
))
332
330
.body(Body::empty())
···
350
348
// Mock PLC
351
349
Mock::given(method("GET"))
352
350
.and(path_regex("/did:plc:.*"))
353
-
.respond_with(ResponseTemplate::new(200).set_body_json(
354
-
fixtures::did_document("did:plc:a", "https://pds.example.com"),
355
-
))
351
+
.respond_with(
352
+
ResponseTemplate::new(200).set_body_json(fixtures::did_document(
353
+
"did:plc:a",
354
+
"https://pds.example.com",
355
+
)),
356
+
)
356
357
.mount(&app.mock_server)
357
358
.await;
358
359